diff --git a/.docker/kerberos-kdc/Dockerfile b/.docker/kerberos-kdc/Dockerfile new file mode 100644 index 00000000..af50cb07 --- /dev/null +++ b/.docker/kerberos-kdc/Dockerfile @@ -0,0 +1,10 @@ +FROM rockylinux:9 + +RUN yum install -y ca-certificates krb5-server krb5-libs krb5-workstation + +EXPOSE 88 749 + +RUN touch /config.sh +# Overwritten via the docker-compose volume mount -- see tests/integration_tests/kerberos_conf/kerberos_image_config.sh + +ENTRYPOINT ["/bin/bash", "/config.sh"] diff --git a/.github/workflows/on_push.yml b/.github/workflows/on_push.yml index 3a5c0966..98194653 100644 --- a/.github/workflows/on_push.yml +++ b/.github/workflows/on_push.yml @@ -289,9 +289,56 @@ jobs: COMPOSE_PROJECT_NAME: clickhouse-connect-ci run: docker compose -f docker-compose.yml down --volumes --remove-orphans + kerberos-integration-test: + runs-on: ubuntu-latest + needs: bare-import-test + name: Kerberos Integration Tests + steps: + - name: Checkout + uses: actions/checkout@v6 + - name: Start ClickHouse 25.8 in Docker + env: + CLICKHOUSE_CONNECT_TEST_CH_VERSION: '25.8' + COMPOSE_PROJECT_NAME: clickhouse-connect-ci + run: docker compose -f docker-compose.yml up -d --wait clickhouse + - name: Set up Python 3.12 + uses: actions/setup-python@v6 + with: + python-version: '3.12' + - name: Install pip + run: python -m pip install --upgrade pip + - name: Install system Kerberos packages + run: | + sudo apt-get update + sudo DEBIAN_FRONTEND=noninteractive apt-get install -y gcc python3-dev libkrb5-dev krb5-user + - name: Add Kerberos ClickHouse instance to /etc/hosts + run: sudo echo "127.0.0.1 server1.clickhouse.test" | sudo tee -a /etc/hosts + - name: Install Test Dependencies + run: | + pip install -r tests/test_requirements.txt + pip install "pyspnego[kerberos]>=0.10" + - name: Build cython extensions + run: python setup.py build_ext --inplace + - name: "Add distribution info" + run: pip install -e . --no-deps + - name: Run tests + env: + CLICKHOUSE_CONNECT_TEST_KERBEROS: '1' + run: pytest --dist=loadgroup tests/integration_tests/test_kerberos.py + - name: Stop Kerberos services + if: ${{ always() && hashFiles('docker-compose.yml') != '' }} + env: + COMPOSE_PROJECT_NAME: clickhouse-connect-kerberos + run: docker compose -f docker-compose.yml --profile kerberos down --volumes --remove-orphans + - name: Stop ClickHouse + if: ${{ always() && hashFiles('docker-compose.yml') != '' }} + env: + COMPOSE_PROJECT_NAME: clickhouse-connect-ci + run: docker compose -f docker-compose.yml down --volumes --remove-orphans + check-secret: runs-on: ubuntu-latest - needs: [bare-import-test, tests, pandas-3x-compat-test, sqlalchemy-1x-compat-test] + needs: [bare-import-test, tests, pandas-3x-compat-test, sqlalchemy-1x-compat-test, kerberos-integration-test] outputs: has_secrets: ${{ steps.has_secrets.outputs.HAS_SECRETS }} steps: diff --git a/CHANGELOG.md b/CHANGELOG.md index 95b56365..0c2d53f3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,6 +12,10 @@ - SQLAlchemy ClickHouse type literals containing percent signs now compile safely alongside remaining bound parameters, and server-side parameter mode preserves consecutive percent signs in literals. Closes [#966](https://github.com/ClickHouse/clickhouse-connect/issues/966). - SQLAlchemy now applies ClickHouse backslash escaping to generic `literal_binds` strings, string `DEFAULT`, `MATERIALIZED`, `ALIAS`, and `TTL` clauses, CREATE comments, and Alembic table and column comment operations. Backslash values now round-trip verbatim instead of being reinterpreted or terminating a quoted literal. ClickHouse-native literal processors and percent handling are unchanged. If custom `TypeDecorator.process_literal_param` or `UserDefinedType` code pre-escaped backslashes as a workaround, remove that workaround because the dialect now applies ClickHouse escaping. Closes [#975](https://github.com/ClickHouse/clickhouse-connect/issues/975). +### Improvements + +- Added Kerberos authentication through the HTTP `Negotiate` scheme for both clients via the new `use_kerberos` and `kerberos_hostname_override` connection parameters, using the current process's Kerberos credential cache. Kerberos support is experimental. This requires the new `kerberos` extra: `pip install clickhouse-connect[kerberos]`. Because ClickHouse authenticates each HTTP request independently rather than caching authentication for a session, each authenticated request attempt uses a fresh preemptive Kerberos token and validates the successful authenticated response token to complete mutual authentication. Closes [#128](https://github.com/ClickHouse/clickhouse-connect/issues/128). + ## 1.7.1, 2026-08-12 ### Bug Fixes diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index c5f7aaf8..389da0f7 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -56,9 +56,9 @@ python setup.py develop ### Add /etc/hosts entry -Required for TLS tests. +Required for TLS and Kerberos tests. The generated certificates assume TLS requests use `server1.clickhouse.test` as the hostname. -See [test_tls.py](tests/integration_tests/test_tls.py) for more details. +See [test_tls.py](tests/integration_tests/test_tls.py) and [test_kerberos.py](tests/integration_tests/test_kerberos.py) for more details. ```bash sudo -- sh -c "echo 127.0.0.1 server1.clickhouse.test >> /etc/hosts" @@ -120,6 +120,50 @@ Additionally, the TLS ClickHouse instance should be running (see [docker-compose CLICKHOUSE_CONNECT_TEST_TLS=1 pytest tests/integration_tests/test_tls.py ``` +### Run the Kerberos integration tests + +These tests require the `CLICKHOUSE_CONNECT_TEST_KERBEROS` environment variable to be set to `1`; otherwise, they will be skipped. +Unlike the other test instances, the Kerberos KDC and ClickHouse instance (the `kerberos_kdc` and `kerberos_clickhouse` services in +[docker-compose.yml](docker-compose.yml)) are behind a `kerberos` Compose profile rather than started by a plain `docker compose up -d`, +since they also need an extra host-side step (obtaining a real Kerberos ticket) that Docker Compose cannot do for you. This walks +through setting them up from scratch. + +Install the system Kerberos client and development packages (needed to build the `gssapi`/`krb5` Python packages): + +```bash +# Debian/Ubuntu +sudo apt-get install gcc python3-dev libkrb5-dev krb5-user + +# CentOS/RHEL/Fedora +sudo dnf install gcc python3-devel krb5-devel krb5-workstation + +# Arch Linux +sudo pacman -S gcc krb5 +``` + +Make sure you've added the `server1.clickhouse.test` `/etc/hosts` entry from +["Add /etc/hosts entry"](#add-etchosts-entry) above. + +The rest (starting a KDC and a Kerberos-configured ClickHouse instance, obtaining a ticket, and tearing it all +back down afterward) is handled automatically by a fixture in +[`test_kerberos.py`](tests/integration_tests/test_kerberos.py), via +[`kerberos_manage.py`](tests/integration_tests/kerberos_manage.py), which uses the fixtures vendored under +[`tests/integration_tests/kerberos_conf`](tests/integration_tests/kerberos_conf). + +Run from the repo root: + +```bash +CLICKHOUSE_CONNECT_TEST_KERBEROS=1 pytest --dist=loadgroup tests/integration_tests/test_kerberos.py +``` + +To stand up (or tear down) the same environment by hand, outside of pytest -- for example, to poke at it manually +with `curl --negotiate` -- run: + +```bash +python -m tests.integration_tests.kerberos_manage setup +python -m tests.integration_tests.kerberos_manage teardown +``` + ### Running the integration tests with ClickHouse Cloud If you want to run the tests using your ClickHouse Cloud instance instead of the local ClickHouse instance running in Docker, you will need a few additional environment variables. diff --git a/README.md b/README.md index 42aefec8..467d64dc 100644 --- a/README.md +++ b/README.md @@ -90,6 +90,56 @@ pip install clickhouse-connect[async] Then create a client with `clickhouse_connect.get_async_client()`. See the [run_async example](./examples/run_async.py) for more details. +### Kerberos Authentication + +Kerberos authentication is experimental. + +ClickHouse Connect supports Kerberos authentication through the HTTP `Negotiate` scheme against a ClickHouse server +configured for Kerberos. Each authenticated request attempt uses a fresh preemptive Kerberos token and validates the +successful authenticated response token to complete mutual authentication. + +Install the system Kerberos client and development packages first (needed to build the `gssapi`/`krb5` Python +packages): + +```bash +# Debian/Ubuntu +sudo apt-get install gcc python3-dev libkrb5-dev krb5-user + +# CentOS/RHEL/Fedora +sudo dnf install gcc python3-devel krb5-devel krb5-workstation + +# Arch Linux +sudo pacman -S gcc krb5 +``` + +Then install the optional dependency: + +```bash +pip install clickhouse-connect[kerberos] +``` + +Obtain a Kerberos ticket first, for example by running `kinit`. The client uses the current process's Kerberos +credential cache. It does not acquire a ticket from a username and password. + +Create a synchronous client with `use_kerberos=True`: + +```python +import clickhouse_connect + +client = clickhouse_connect.get_client(host="localhost", use_kerberos=True) +``` + +The same options work with `await clickhouse_connect.get_async_client(...)`. Install both extras with +`pip install "clickhouse-connect[async,kerberos]"` for async use. + +If the hostname you connect through doesn't match the server's Kerberos service principal name (for example, +connecting via an IP address or load balancer), pass `kerberos_hostname_override` with the principal's actual +hostname. + +Do not combine Kerberos authentication with `username`, `password`, `access_token`, `token_provider`, or +`client_cert`. Each authenticated HTTP request attempt sends a fresh preemptive token. Every successful authenticated +response must contain a `WWW-Authenticate: Negotiate ` response header that completes mutual authentication. + ### Complete Documentation The documentation for ClickHouse Connect has moved to diff --git a/clickhouse_connect/driver/__init__.py b/clickhouse_connect/driver/__init__.py index 880ce2fa..e93ef242 100644 --- a/clickhouse_connect/driver/__init__.py +++ b/clickhouse_connect/driver/__init__.py @@ -204,6 +204,13 @@ def create_client( :param token_provider: A callable returning a JWT access token (ClickHouse Cloud feature). Called for the initial token and again to refresh it whenever the server rejects the current one. Should not be set if `access_token` or `username`/`password` are used. + :param use_kerberos: Use Kerberos authentication through the HTTP `Negotiate` scheme against a ClickHouse server + configured for Kerberos, using the current process's Kerberos credential cache (equivalent to having run `kinit` beforehand). + Requires the kerberos extra: pip install clickhouse-connect[kerberos]. Cannot be combined with `access_token`, + `token_provider`, `username`/`password`, or `client_cert`. + :param kerberos_hostname_override: Override the hostname used to build the Kerberos service principal name + (`HTTP/`). Defaults to `host`. Useful when connecting via an IP address or a load balancer whose + name does not match the ClickHouse server's Kerberos principal. :param database: The default database for the connection. If not set, ClickHouse Connect will use the default database for username. :param interface: Must be http, https, or chdb. Defaults to http, or to https if port is set to 8443 or 443. @@ -364,6 +371,13 @@ async def create_async_client( :param token_provider: A callable returning a JWT access token. Called for the initial token and again to refresh it whenever the server rejects the current one. Because multiple in-flight requests may each trigger a refresh concurrently, the callable must be safe to invoke in parallel. + :param use_kerberos: Use Kerberos authentication through the HTTP `Negotiate` scheme against a ClickHouse server + configured for Kerberos, using the current process's Kerberos credential cache (equivalent to having run `kinit` beforehand). + Requires the kerberos extra: pip install clickhouse-connect[kerberos]. Cannot be combined with `access_token`, + `token_provider`, `username`/`password`, or `client_cert`. + :param kerberos_hostname_override: Override the hostname used to build the Kerberos service principal name + (`HTTP/`). Defaults to `host`. Useful when connecting via an IP address or a load balancer whose + name does not match the ClickHouse server's Kerberos principal. :param database: The default database for the connection. If not set, ClickHouse Connect will use the default database for username. :param interface: Must be http or https. Defaults to http, or to https if port is set to 8443 or 443 diff --git a/clickhouse_connect/driver/_backend/http_async.py b/clickhouse_connect/driver/_backend/http_async.py index 3159be67..30198fbb 100644 --- a/clickhouse_connect/driver/_backend/http_async.py +++ b/clickhouse_connect/driver/_backend/http_async.py @@ -38,6 +38,7 @@ from clickhouse_connect.driver._backend.models import Capabilities, CommandExecution, QueryExecution, QueryRuntime from clickhouse_connect.driver.common import ShowClickHouseErrors, dict_copy from clickhouse_connect.driver.exceptions import OperationalError, ProgrammingError +from clickhouse_connect.driver.kerberos import KerberosAuthContext from clickhouse_connect.driver.streaming import start_streaming_response if TYPE_CHECKING: @@ -159,6 +160,8 @@ def __init__( server_host_name: str | None, token_provider: Callable[[], str | Awaitable[str]] | None, autogenerate_query_id: bool, + use_kerberos: bool = False, + kerberos_hostname: str | None = None, read_format: str = "Native", form_encode_query_params: bool = False, ): @@ -172,6 +175,8 @@ def __init__( self.proxy_url = proxy_url self.server_host_name = server_host_name self.token_provider = token_provider + self.use_kerberos = use_kerberos + self.kerberos_hostname = kerberos_hostname self.autogenerate_query_id = autogenerate_query_id self.read_format = read_format self.form_encode_query_params = form_encode_query_params @@ -474,9 +479,22 @@ async def request( lease.acquire() lease_released = False try: + kerberos_context = None + if self.use_kerberos: + assert self.kerberos_hostname is not None + kerberos_context = KerberosAuthContext(self.kerberos_hostname) + req_headers["Authorization"] = kerberos_context.authorization_header # Construct full URL (aiohttp doesn't have base_url) url = self._base_url - request_kwargs = {"method": method, "url": url, "params": final_params, "headers": req_headers} + request_method = "POST" if self.use_kerberos and method == "GET" else method + request_kwargs: dict[str, Any] = { + "method": request_method, + "url": url, + "params": final_params, + "headers": req_headers, + } + if self.use_kerberos: + request_kwargs["allow_redirects"] = False if self.server_host_name and self.ssl_context is not None: request_kwargs["ssl"] = self.ssl_context request_kwargs["server_hostname"] = self.server_host_name @@ -507,6 +525,12 @@ async def request( response = await session.request(**request_kwargs) if 200 <= response.status < 300 and not response.headers.get(ex_header): + if kerberos_context is not None: + try: + kerberos_context.validate_response(response.headers.get("WWW-Authenticate")) + except Exception: + response.close() + raise # Caller releases lease after consuming the body. response._lease_release = _one_shot(lease.release) # type: ignore[attr-defined] lease_released = True diff --git a/clickhouse_connect/driver/_backend/http_sync.py b/clickhouse_connect/driver/_backend/http_sync.py index 45d9ef31..dad1ced7 100644 --- a/clickhouse_connect/driver/_backend/http_sync.py +++ b/clickhouse_connect/driver/_backend/http_sync.py @@ -38,6 +38,7 @@ from clickhouse_connect.driver.common import ShowClickHouseErrors, dict_copy from clickhouse_connect.driver.exceptions import OperationalError, ProgrammingError from clickhouse_connect.driver.httputil import ResponseSource, all_managers, check_conn_expiration, get_response_data +from clickhouse_connect.driver.kerberos import KerberosAuthContext if TYPE_CHECKING: from clickhouse_connect.driver._backend.contracts import SyncBackend @@ -78,6 +79,8 @@ def __init__( server_host_name: str | None, token_provider: Callable[[], str] | None, autogenerate_query_id: bool, + use_kerberos: bool = False, + kerberos_hostname: str | None = None, http_retries: int = 1, read_format: str = "Native", form_encode_query_params: bool = False, @@ -94,6 +97,8 @@ def __init__( self.timeout = timeout self.server_host_name = server_host_name self.token_provider = token_provider + self.use_kerberos = use_kerberos + self.kerberos_hostname = kerberos_hostname self.autogenerate_query_id = autogenerate_query_id self.http_retries = http_retries self.read_format = read_format @@ -306,7 +311,12 @@ def request( final_params["query_id"] = str(uuid.uuid4()) url = f"{self._base_url}?{urlencode(final_params)}" - kwargs: dict[str, Any] = {"headers": headers, "timeout": self.timeout, "retries": self.http_retries, "preload_content": not stream} + kwargs: dict[str, Any] = { + "headers": headers, + "timeout": self.timeout, + "retries": 0 if self.use_kerberos else self.http_retries, + "preload_content": not stream, + } if self.server_host_name: kwargs["assert_same_host"] = False kwargs["headers"].update({"Host": self.server_host_name}) @@ -328,6 +338,11 @@ def request( # throw an error instead, but in most cases this more helpful error will be thrown first self._active_session = query_session try: + kerberos_context = None + if self.use_kerberos: + assert self.kerberos_hostname is not None + kerberos_context = KerberosAuthContext(self.kerberos_hostname) + headers["Authorization"] = kerberos_context.authorization_header response: HTTPResponse = cast(HTTPResponse, cast(PoolManager, self.http).request(method, url, **kwargs)) except HTTPError as ex: # Always allow at least one retry on a clean connection error so a single stale @@ -360,11 +375,18 @@ def request( if query_session: self._active_session = None # Make sure we always clear this if 200 <= response.status < 300 and not response.headers.get(ex_header): + if kerberos_context is not None: + try: + kerberos_context.validate_response(response.headers.get("WWW-Authenticate")) + except Exception: + response.close() + raise return response if response.status in retryable_http_statuses: if attempts > retries: self.error_handler(response, True) logger.debug("Retrying requests with status code %d", response.status) + response.close() elif self.token_provider and not auth_retried and response.headers.get(ex_header) == auth_failed_ex_code: body = kwargs.get("body") if retry_body is None and not (body is None or isinstance(body, (bytes, bytearray, str))): diff --git a/clickhouse_connect/driver/asyncclient.py b/clickhouse_connect/driver/asyncclient.py index bc4c715e..48ad8cab 100644 --- a/clickhouse_connect/driver/asyncclient.py +++ b/clickhouse_connect/driver/asyncclient.py @@ -59,6 +59,7 @@ from clickhouse_connect.driver.exceptions import DataError, ProgrammingError from clickhouse_connect.driver.external import ExternalData from clickhouse_connect.driver.insert import InsertContext +from clickhouse_connect.driver.kerberos import check_kerberos from clickhouse_connect.driver.options import check_arrow, check_numpy, check_pandas, check_polars from clickhouse_connect.driver.query import ( QueryContext, @@ -156,6 +157,8 @@ def __init__( form_encode_query_params: bool = False, rename_response_column: str | None = None, headers: dict[str, str] | None = None, + use_kerberos: bool | str = False, + kerberos_hostname_override: str | None = None, ): """ Async HTTP Client using aiohttp. Initialization is handled via _initialize(). @@ -176,9 +179,18 @@ def __init__( # The initial token from token_provider is resolved in _initialize() + use_kerberos = coerce_bool(use_kerberos) + kerberos_hostname: str | None = None + + # Priority: kerberos > access_token > mutual TLS > basic auth # Auth headers follow the sync client: mutual TLS headers are set # independently, and a bearer token wins over basic auth. - if client_cert and (tls_mode is None or tls_mode == "mutual"): + if use_kerberos: + if access_token or token_provider or username or password or client_cert: + raise ProgrammingError("Cannot combine use_kerberos with access_token, token_provider, username/password, or client_cert") + check_kerberos() + kerberos_hostname = kerberos_hostname_override or host + elif client_cert and (tls_mode is None or tls_mode == "mutual"): if not username: raise ProgrammingError("username parameter is required for Mutual TLS authentication") self.headers["X-ClickHouse-User"] = username @@ -278,6 +290,8 @@ def __init__( proxy_url=proxy_url, server_host_name=server_host_name, token_provider=token_provider, + use_kerberos=use_kerberos, + kerberos_hostname=kerberos_hostname, autogenerate_query_id=(common.get_setting("autogenerate_query_id") if autogenerate_query_id is None else autogenerate_query_id), read_format="Native", form_encode_query_params=form_encode_query_params, diff --git a/clickhouse_connect/driver/httpclient.py b/clickhouse_connect/driver/httpclient.py index 97d8fba2..dc274ef4 100644 --- a/clickhouse_connect/driver/httpclient.py +++ b/clickhouse_connect/driver/httpclient.py @@ -40,6 +40,7 @@ get_pool_manager, get_proxy_manager, ) +from clickhouse_connect.driver.kerberos import check_kerberos from clickhouse_connect.driver.query import TzMode, TzSource from clickhouse_connect.driver.transform import NativeTransform @@ -106,6 +107,8 @@ def __init__( form_encode_query_params: bool = False, rename_response_column: str | None = None, headers: dict[str, str] | None = None, + use_kerberos: bool | str = False, + kerberos_hostname_override: str | None = None, ): """ Create an HTTP ClickHouse Connect client @@ -151,12 +154,20 @@ def __init__( else: pool = default_pool_manager() - if token_provider: - access_token = token_provider() - if access_token: - client_headers["Authorization"] = f"Bearer {access_token}" - elif (not client_cert or tls_mode in ("strict", "proxy")) and username: - client_headers["Authorization"] = "Basic " + b64encode(f"{username}:{password}".encode()).decode() + use_kerberos = coerce_bool(use_kerberos) + kerberos_hostname: str | None = None + if use_kerberos: + if access_token or token_provider or username or password or client_cert: + raise ProgrammingError("Cannot combine use_kerberos with access_token, token_provider, username/password, or client_cert") + check_kerberos() + kerberos_hostname = kerberos_hostname_override or host + else: + if token_provider: + access_token = token_provider() + if access_token: + client_headers["Authorization"] = f"Bearer {access_token}" + elif (not client_cert or tls_mode in ("strict", "proxy")) and username: + client_headers["Authorization"] = "Basic " + b64encode(f"{username}:{password}".encode()).decode() self._reported_libs: set[str] = set() client_headers["User-Agent"] = common.build_client_name(client_name) @@ -197,6 +208,8 @@ def __init__( timeout=Timeout(connect=connect_timeout, read=send_receive_timeout), server_host_name=server_host_name, token_provider=token_provider, + use_kerberos=use_kerberos, + kerberos_hostname=kerberos_hostname, # allow to override the global autogenerate_query_id setting via the constructor params autogenerate_query_id=(common.get_setting("autogenerate_query_id") if autogenerate_query_id is None else autogenerate_query_id), read_format="Native", diff --git a/clickhouse_connect/driver/kerberos.py b/clickhouse_connect/driver/kerberos.py new file mode 100644 index 00000000..cf13a93f --- /dev/null +++ b/clickhouse_connect/driver/kerberos.py @@ -0,0 +1,70 @@ +import base64 +import binascii +from typing import Protocol, cast + +from clickhouse_connect.driver.exceptions import OperationalError +from clickhouse_connect.driver.options import check_spnego + + +class _SpnegoContext(Protocol): + @property + def complete(self) -> bool: ... + + def step(self, in_token: bytes | None = None) -> bytes | None: ... + + +class _SpnegoExceptions(Protocol): + SpnegoError: type[Exception] + + +class _SpnegoModule(Protocol): + exceptions: _SpnegoExceptions + + def client(self, *, hostname: str, service: str, protocol: str) -> _SpnegoContext: ... + + +def check_kerberos() -> _SpnegoModule: + """Return pyspnego when Kerberos support is installed.""" + return cast(_SpnegoModule, check_spnego()) + + +class KerberosAuthContext: + """Kerberos client context for one HTTP request attempt.""" + + def __init__(self, hostname: str, service: str = "HTTP") -> None: + self._spnego = check_kerberos() + try: + self._context = cast( + _SpnegoContext, + self._spnego.client(hostname=hostname, service=service, protocol="kerberos"), + ) + token = self._context.step() + except (self._spnego.exceptions.SpnegoError, ImportError) as ex: + raise OperationalError(f"Kerberos negotiation failed: {ex}") from ex + if not token: + raise OperationalError("Kerberos negotiation failed: no client authentication token was produced") + self.authorization_header = "Negotiate " + base64.b64encode(token).decode() + + def validate_response(self, authenticate_header: str | None) -> None: + """Consume the server AP-REP token and require mutual authentication.""" + if authenticate_header is None: + raise OperationalError("Kerberos mutual authentication failed: successful response is missing the WWW-Authenticate header") + + scheme, separator, encoded_token = authenticate_header.partition(" ") + encoded_token = encoded_token.strip() + if scheme.lower() != "negotiate" or not separator or not encoded_token: + raise OperationalError("Kerberos mutual authentication failed: WWW-Authenticate must contain a Negotiate response token") + try: + token = base64.b64decode(encoded_token, validate=True) + except (binascii.Error, ValueError) as ex: + raise OperationalError( + "Kerberos mutual authentication failed: WWW-Authenticate contains an invalid Negotiate response token" + ) from ex + if not token: + raise OperationalError("Kerberos mutual authentication failed: WWW-Authenticate contains an empty Negotiate response token") + try: + self._context.step(token) + except self._spnego.exceptions.SpnegoError as ex: + raise OperationalError(f"Kerberos mutual authentication failed: {ex}") from ex + if not self._context.complete: + raise OperationalError("Kerberos mutual authentication failed: the server response did not complete the context") diff --git a/clickhouse_connect/driver/options.py b/clickhouse_connect/driver/options.py index e39da43d..2e293ec3 100644 --- a/clickhouse_connect/driver/options.py +++ b/clickhouse_connect/driver/options.py @@ -1,11 +1,11 @@ from clickhouse_connect.driver.exceptions import NotSupportedError # Attributes resolved lazily by __getattr__ / _resolve_* functions: -# np, pd, arrow, pl, pd_time_test +# np, pd, arrow, pl, pd_time_test, spnego _PANDAS_ATTRS = frozenset({"pd", "pd_time_test"}) -_ALL_LAZY = frozenset({"np", "arrow", "pl"}) | _PANDAS_ATTRS +_ALL_LAZY = frozenset({"np", "arrow", "pl", "spnego"}) | _PANDAS_ATTRS def _pd_time_test(arr_or_dtype): @@ -67,6 +67,17 @@ def _resolve_polars(): globals()["pl"] = None +def _resolve_spnego(): + if "spnego" in globals(): + return + try: + import spnego + + globals()["spnego"] = spnego + except ImportError: + globals()["spnego"] = None + + def __getattr__(name): if name in _PANDAS_ATTRS: _resolve_pandas() @@ -76,6 +87,8 @@ def __getattr__(name): _resolve_arrow() elif name == "pl": _resolve_polars() + elif name == "spnego": + _resolve_spnego() else: raise AttributeError(f"module {__name__!r} has no attribute {name!r}") return globals()[name] @@ -115,3 +128,11 @@ def check_polars(): if pl: return pl raise NotSupportedError("Polars package is not installed") + + +def check_spnego(): + _resolve_spnego() + spnego = globals()["spnego"] + if spnego: + return spnego + raise NotSupportedError("Kerberos authentication support is not installed. Install with: pip install clickhouse-connect[kerberos]") diff --git a/docker-compose.yml b/docker-compose.yml index d268f03d..ee01c53a 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -34,3 +34,44 @@ services: - './.docker/clickhouse/single_node_tls/config.xml:/etc/clickhouse-server/config.xml' - './.docker/clickhouse/single_node_tls/users.xml:/etc/clickhouse-server/users.xml' - './.docker/clickhouse/single_node_tls/docker_related_config.xml:/etc/clickhouse-server/config.d/docker_related_config.xml' + # Both kerberos services are opt-in (profiles: [kerberos]), started only by + # tests/integration_tests/kerberos_manage.py (see test_kerberos.py's kerberos_env fixture), + # since they need an extra host-side step (obtaining a real Kerberos ticket) that Docker Compose + # cannot do for you. See "Run the Kerberos integration tests" in CONTRIBUTING.md. + kerberos_kdc: + build: + context: ./ + dockerfile: .docker/kerberos-kdc/Dockerfile + container_name: 'clickhouse-connect-kerberos-kdc' + profiles: ['kerberos'] + ports: + - '8088:88/udp' + - '8088:88/tcp' + volumes: + - './tests/integration_tests/kerberos_conf/kerberos_image_config.sh:/config.sh:ro' + - 'kerberos_kdc_keytabs:/tmp/keytab' + - 'kerberos_kdc_data:/var/kerberos/krb5kdc' + + kerberos_clickhouse: + image: 'clickhouse/clickhouse-server:${CLICKHOUSE_CONNECT_TEST_CH_VERSION-25.8-alpine}' + container_name: 'clickhouse-connect-kerberos-clickhouse-server' + profiles: ['kerberos'] + depends_on: + - kerberos_kdc + environment: + CLICKHOUSE_SKIP_USER_SETUP: 1 + ports: + - '8124:8123' + ulimits: + nofile: + soft: 262144 + hard: 262144 + volumes: + - 'kerberos_kdc_keytabs:/tmp/keytab:ro' + - './tests/integration_tests/kerberos_conf/kerberos_config.xml:/etc/clickhouse-server/config.d/kerberos_config.xml:ro' + - './tests/integration_tests/kerberos_conf/kerberos_users.xml:/etc/clickhouse-server/users.d/kerberos_users.xml:ro' + - './tests/integration_tests/kerberos_conf/kerberos_server_krb5.conf:/etc/krb5.conf:ro' + +volumes: + kerberos_kdc_keytabs: + kerberos_kdc_data: diff --git a/docs/driver-api.mdx b/docs/driver-api.mdx index c92c4269..eb7230ee 100644 --- a/docs/driver-api.mdx +++ b/docs/driver-api.mdx @@ -28,6 +28,8 @@ Use `clickhouse_connect.get_client` to create a synchronous `Client`, or install | `password` | str | `""` | Password for `username`. Do not combine user/password authentication with token authentication. | | `access_token` | str or None | `None` | ClickHouse Cloud JWT access token. Mutually exclusive with `token_provider` and user/password authentication. | | `token_provider` | callable or None | `None` | Callable that supplies a JWT initially and after an authentication rejection. An async provider may be used with `get_async_client`. | +| `use_kerberos` | bool or boolean string | `False` | Use Kerberos authentication through HTTP `Negotiate`. Supported by the synchronous and async HTTP clients. | +| `kerberos_hostname_override` | str or None | `None` | Hostname used in the Kerberos service principal when it differs from `host`. The default principal is `HTTP/`. | | `database` | str or None | User default | Default database. Passing `None` requests the server default for the user. | | `secure` | bool or str | `False` | Enable HTTPS/TLS. `interface="https"` also selects HTTPS, as does port 443 or 8443 when `interface` is not set. | | `dsn` | str or None | `None` | Connection URL. Explicit keyword arguments take precedence over values parsed from the DSN. Percent-encode reserved characters in credentials and database names. | @@ -54,6 +56,43 @@ Use `clickhouse_connect.get_client` to create a synchronous `Client`, or install The async factory also accepts `connector_limit=100`, `connector_limit_per_host=20`, and `keepalive_timeout=30.0` to configure its aiohttp connection pool. It does not accept `pool_mgr`. The synchronous chDB backend accepts `path` and `chdb_options`; see [Embedded chDB backend](#embedded-chdb-backend). +### Kerberos authentication {#kerberos-authentication} + +Kerberos authentication is experimental. + +Kerberos authentication is available for both `get_client` and `get_async_client`. Install the `kerberos` extra and the Kerberos client libraries required by your operating system: + +```bash +pip install "clickhouse-connect[kerberos]" +``` + +For the async client, install both optional dependencies with `pip install "clickhouse-connect[async,kerberos]"`. + +Obtain a Kerberos ticket before creating the client, for example by running `kinit`. ClickHouse Connect uses the current process's Kerberos credential cache. It does not acquire a ticket from the `username` or `password` arguments. + +```python +import clickhouse_connect + +client = clickhouse_connect.get_client( + host="clickhouse.example.com", + use_kerberos=True, +) +``` + +The default service principal is `HTTP/`. When the connection hostname is an IP address, load balancer, or proxy that differs from the ClickHouse server principal, set `kerberos_hostname_override`: + +```python +client = clickhouse_connect.get_client( + host="127.0.0.1", + use_kerberos=True, + kerberos_hostname_override="clickhouse.example.com", +) +``` + +Each authenticated HTTP request attempt sends a fresh preemptive Kerberos token. The client also requires a `WWW-Authenticate: Negotiate ` header on every successful authenticated response and validates that token to complete mutual authentication. It raises `OperationalError` if the response token is missing, malformed, rejected, or does not complete the Kerberos context. + +Do not combine `use_kerberos=True` with `username`, `password`, `access_token`, `token_provider`, or `client_cert`. Kerberos authentication supports the HTTP and HTTPS clients only. It does not support the chDB backend. + ### HTTPS/TLS arguments {#httpstls-arguments} | Parameter | Type | Default | Description | diff --git a/docs/index.mdx b/docs/index.mdx index 46aaeca7..60fb9f76 100644 --- a/docs/index.mdx +++ b/docs/index.mdx @@ -60,6 +60,7 @@ pip install "clickhouse-connect[polars]" # Polars pip install "clickhouse-connect[sqlalchemy]" # SQLAlchemy dialect pip install "clickhouse-connect[alembic]" # SQLAlchemy and Alembic pip install "clickhouse-connect[chdb]" # Embedded chDB backend +pip install "clickhouse-connect[kerberos]" # Experimental Kerberos authentication pip install "clickhouse-connect[tzdata]" # IANA time zones on minimal systems ``` diff --git a/pyproject.toml b/pyproject.toml index 0a96625e..9acf172a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -45,6 +45,7 @@ module = [ "orjson.*", "pandas.*", "pyarrow.*", + "spnego.*", "tzlocal.*", "ujson.*", "clickhouse_connect.driverc.*", diff --git a/setup.py b/setup.py index 0a83088c..bbfad96a 100644 --- a/setup.py +++ b/setup.py @@ -82,6 +82,7 @@ def run_setup(try_c: bool = True): "tzdata": ["tzdata"], "async": ["aiohttp>=3.9.0"], "chdb": ["chdb>=4.1.7"], + "kerberos": ["pyspnego[kerberos]>=0.10"], }, tests_require=["pytest"], entry_points={ diff --git a/tests/integration_tests/kerberos_conf/kerberos_client_krb5.conf b/tests/integration_tests/kerberos_conf/kerberos_client_krb5.conf new file mode 100644 index 00000000..4eb7eb33 --- /dev/null +++ b/tests/integration_tests/kerberos_conf/kerberos_client_krb5.conf @@ -0,0 +1,11 @@ +[libdefaults] + default_realm = TEST.CLICKHOUSE.TECH + dns_lookup_realm = false + dns_lookup_kdc = false + rdns = false + +[realms] + TEST.CLICKHOUSE.TECH = { + kdc = 127.0.0.1:8088 + admin_server = 127.0.0.1:8088 + } diff --git a/tests/integration_tests/kerberos_conf/kerberos_config.xml b/tests/integration_tests/kerberos_conf/kerberos_config.xml new file mode 100644 index 00000000..abd6d27a --- /dev/null +++ b/tests/integration_tests/kerberos_conf/kerberos_config.xml @@ -0,0 +1,6 @@ + + + TEST.CLICKHOUSE.TECH + /tmp/keytab/server1.clickhouse.test.keytab + + diff --git a/tests/integration_tests/kerberos_conf/kerberos_image_config.sh b/tests/integration_tests/kerberos_conf/kerberos_image_config.sh new file mode 100644 index 00000000..fb9dd4da --- /dev/null +++ b/tests/integration_tests/kerberos_conf/kerberos_image_config.sh @@ -0,0 +1,116 @@ +#!/bin/bash + + +set -x # trace + +: "${REALM:=TEST.CLICKHOUSE.TECH}" +: "${DOMAIN_REALM:=test.clickhouse.com}" +: "${KERB_MASTER_KEY:=masterkey}" +: "${KERB_ADMIN_USER:=admin}" +: "${KERB_ADMIN_PASS:=admin}" + +create_config() { + : "${KDC_ADDRESS:=$(hostname -f)}" + + cat>/etc/krb5.conf</var/kerberos/krb5kdc/kdc.conf< /var/kerberos/krb5kdc/kadm5.acl +} + +create_keytabs() { + rm /tmp/keytab/*.keytab + + kadmin.local -q "addprinc -randkey kuser@${REALM}" + kadmin.local -q "ktadd -norandkey -k /tmp/keytab/kuser.keytab kuser@${REALM}" + + kadmin.local -q "addprinc -randkey HTTP/server1.clickhouse.test@${REALM}" + kadmin.local -q "ktadd -norandkey -k /tmp/keytab/server1.clickhouse.test.keytab HTTP/server1.clickhouse.test@${REALM}" + + chmod g+r /tmp/keytab/kuser.keytab + chmod g+r /tmp/keytab/server1.clickhouse.test.keytab +} + +main() { + + if [ ! -f /kerberos_initialized ]; then + create_config + create_db + create_admin_user + start_kdc + + touch /kerberos_initialized + fi + + if [ ! -f /var/kerberos/krb5kdc/principal ]; then + while true; do sleep 1000; done + else + start_kdc + create_keytabs + tail -F /var/log/kerberos/krb5kdc.log + fi + +} + +[[ "$0" == "${BASH_SOURCE[0]}" ]] && main "$@" diff --git a/tests/integration_tests/kerberos_conf/kerberos_server_krb5.conf b/tests/integration_tests/kerberos_conf/kerberos_server_krb5.conf new file mode 100644 index 00000000..b9c1566f --- /dev/null +++ b/tests/integration_tests/kerberos_conf/kerberos_server_krb5.conf @@ -0,0 +1,22 @@ +[logging] + default = FILE:/var/log/kerberos/krb5libs.log + kdc = FILE:/var/log/kerberos/krb5kdc.log + admin_server = FILE:/var/log/kerberos/kadmind.log + +[libdefaults] + default_realm = TEST.CLICKHOUSE.TECH + dns_lookup_realm = false + dns_lookup_kdc = false + ticket_lifetime = 15s + renew_lifetime = 15s + forwardable = true + +[realms] + TEST.CLICKHOUSE.TECH = { + kdc = kerberos_kdc + admin_server = kerberos_kdc + } + +[domain_realm] + .test.clickhouse.com = TEST.CLICKHOUSE.TECH + test.clickhouse.com = TEST.CLICKHOUSE.TECH diff --git a/tests/integration_tests/kerberos_conf/kerberos_users.xml b/tests/integration_tests/kerberos_conf/kerberos_users.xml new file mode 100644 index 00000000..33d658e6 --- /dev/null +++ b/tests/integration_tests/kerberos_conf/kerberos_users.xml @@ -0,0 +1,19 @@ + + + + + + + + + TEST.CLICKHOUSE.TECH + + 1 + + ::/0 + + default + default + + + diff --git a/tests/integration_tests/kerberos_manage.py b/tests/integration_tests/kerberos_manage.py new file mode 100644 index 00000000..9d502716 --- /dev/null +++ b/tests/integration_tests/kerberos_manage.py @@ -0,0 +1,89 @@ +"""Stands up (and tears down) a real Kerberos KDC and Kerberos-configured ClickHouse instance +(the "kerberos_kdc" and "kerberos_clickhouse" services in docker-compose.yml) for +test_kerberos.py. See "Run the Kerberos integration tests" in CONTRIBUTING.md for prerequisites. + +Used automatically by test_kerberos.py's kerberos_env fixture when CLICKHOUSE_CONNECT_TEST_KERBEROS +is set. Can also be run by hand for manual/exploratory use: + + python -m tests.integration_tests.kerberos_manage setup + python -m tests.integration_tests.kerberos_manage teardown +""" + +import os +import subprocess +import sys +import tempfile +import time + +from tests.helpers import PROJECT_ROOT_DIR + +COMPOSE_PROJECT_NAME = "clickhouse-connect-kerberos" +KDC_CONTAINER = f"{COMPOSE_PROJECT_NAME}-kdc" +LOCAL_KEYTAB_PATH = os.path.join(tempfile.gettempdir(), "kuser.keytab") +CONTAINER_KEYTAB_PATH = "/tmp/keytab/kuser.keytab" +KRB5_CONFIG_PATH = str(PROJECT_ROOT_DIR / "tests/integration_tests/kerberos_conf/kerberos_client_krb5.conf") +CLICKHOUSE_HOST = "server1.clickhouse.test" +CLICKHOUSE_PORT = 8124 + + +def _compose_env(): + env = os.environ.copy() + env["COMPOSE_PROJECT_NAME"] = COMPOSE_PROJECT_NAME + return env + + +def _compose(*args): + compose_file = str(PROJECT_ROOT_DIR / "docker-compose.yml") + subprocess.run(["docker", "compose", "-f", compose_file, "--profile", "kerberos", *args], env=_compose_env(), check=True) + + +def setup(): + _compose("up", "-d", "--wait", "kerberos_kdc", "kerberos_clickhouse") + + for _ in range(30): + result = subprocess.run( + ["docker", "exec", KDC_CONTAINER, "test", "-f", CONTAINER_KEYTAB_PATH], + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + if result.returncode == 0: + break + time.sleep(1) + else: + raise RuntimeError(f"The KDC did not provision keytabs within 30s. Check 'docker logs {KDC_CONTAINER}' for what went wrong.") + + subprocess.run(["docker", "cp", f"{KDC_CONTAINER}:{CONTAINER_KEYTAB_PATH}", LOCAL_KEYTAB_PATH], check=True) + + os.environ["KRB5_CONFIG"] = KRB5_CONFIG_PATH + subprocess.run(["kinit", "-k", "-t", LOCAL_KEYTAB_PATH, "kuser@TEST.CLICKHOUSE.TECH"], check=True) + + _wait_for_kerberos_auth() + + +def _wait_for_kerberos_auth(): + # docker compose --wait only confirms the container's own healthcheck (the HTTP port + # responding), which can pass a moment before ClickHouse's Kerberos acceptor context is fully + # ready -- causing the very first real auth attempt to occasionally fail. Retrying an actual + # negotiated request here (via curl, already relying on the kinit ticket obtained above, same + # as kinit itself relies on external krb5 tooling) avoids handing pytest a cold first attempt. + url = f"http://{CLICKHOUSE_HOST}:{CLICKHOUSE_PORT}/?query=SELECT+1" + deadline = time.monotonic() + 10 + while True: + result = subprocess.run(["curl", "-fsS", "--negotiate", "-u", ":", url], stdout=subprocess.PIPE, stderr=subprocess.DEVNULL) + if result.returncode == 0 and result.stdout.strip() == b"1": + return + if time.monotonic() >= deadline: + raise RuntimeError(f"Kerberos auth against {CLICKHOUSE_HOST}:{CLICKHOUSE_PORT} did not succeed within 10s.") + time.sleep(0.5) + + +def teardown(): + _compose("down", "--volumes") + + +if __name__ == "__main__": + actions = {"setup": setup, "teardown": teardown} + if len(sys.argv) != 2 or sys.argv[1] not in actions: + print(f"usage: python -m tests.integration_tests.kerberos_manage {{{'|'.join(actions)}}}", file=sys.stderr) + sys.exit(1) + actions[sys.argv[1]]() diff --git a/tests/integration_tests/test_kerberos.py b/tests/integration_tests/test_kerberos.py new file mode 100644 index 00000000..4cf4a4f7 --- /dev/null +++ b/tests/integration_tests/test_kerberos.py @@ -0,0 +1,151 @@ +import os +from unittest.mock import patch + +import aiohttp +import pytest +from urllib3.poolmanager import PoolManager + +from clickhouse_connect.driver.common import coerce_bool +from clickhouse_connect.driver.exceptions import DatabaseError +from clickhouse_connect.driver.options import spnego +from tests.integration_tests import kerberos_manage + + +def _kerberos_enabled(): + return coerce_bool(os.environ.get("CLICKHOUSE_CONNECT_TEST_KERBEROS", "False")) + + +# All tests here must land on the same xdist worker. The dedicated CI job uses --dist=loadgroup: +# kerberos_env below does real docker compose/kinit side effects that must run exactly once, not +# once per worker. +pytestmark = [ + pytest.mark.xdist_group(name="kerberos"), + pytest.mark.skipif(not _kerberos_enabled(), reason="CLICKHOUSE_CONNECT_TEST_KERBEROS is False"), + pytest.mark.skipif(spnego is None, reason="kerberos option not installed"), +] + + +@pytest.fixture(scope="module", autouse=True) +def kerberos_env(): + kerberos_manage.setup() + yield + kerberos_manage.teardown() + + +def test_basic_kerberos_auth(client_factory, call): + client = client_factory( + host=kerberos_manage.CLICKHOUSE_HOST, + port=kerberos_manage.CLICKHOUSE_PORT, + username="", + password="", + database="default", + use_kerberos=True, + ) + assert call(client.command, "SELECT currentUser()") == "kuser" + + +def test_kerberos_hostname_override(client_factory, call): + client = client_factory( + host="localhost", + port=kerberos_manage.CLICKHOUSE_PORT, + username="", + password="", + database="default", + use_kerberos=True, + kerberos_hostname_override=kerberos_manage.CLICKHOUSE_HOST, + ) + assert call(client.command, "SELECT currentUser()") == "kuser" + + +def test_kerberos_insert_and_query(client_factory, call): + client = client_factory( + host=kerberos_manage.CLICKHOUSE_HOST, + port=kerberos_manage.CLICKHOUSE_PORT, + username="", + password="", + database="default", + use_kerberos=True, + ) + call(client.command, "CREATE TABLE IF NOT EXISTS default.krb_int_test (id UInt32, name String) ENGINE = Memory") + try: + call(client.insert, "default.krb_int_test", [[13, "alpha"], [79, "beta"]], column_names=["id", "name"]) + result = call(client.query, "SELECT id, name FROM default.krb_int_test ORDER BY id") + assert result.result_rows == [(13, "alpha"), (79, "beta")] + finally: + call(client.command, "DROP TABLE IF EXISTS default.krb_int_test") + + +def _patch_auth_headers(on_header): + """Patch the real sync/async transport calls to observe (and optionally rewrite, via + in-place mutation) the Authorization header of each outgoing request. Returns the two patch + context managers to enter together; the real request always goes through underneath.""" + real_sync_request = PoolManager.request + real_async_request = aiohttp.ClientSession.request + + def sync_wrapper(self, method, url, *args, **kwargs): + headers = dict(kwargs.get("headers") or {}) + on_header(headers) + return real_sync_request(self, method, url, *args, **{**kwargs, "headers": headers}) + + async def async_wrapper(self, method, url, *args, **kwargs): + headers = dict(kwargs.get("headers") or {}) + on_header(headers) + return await real_async_request(self, method, url, *args, **{**kwargs, "headers": headers}) + + return patch.object(PoolManager, "request", new=sync_wrapper), patch.object(aiohttp.ClientSession, "request", new=async_wrapper) + + +def test_kerberos_multiple_sequential_requests(client_factory, call): + # ClickHouse authenticates each HTTP request independently (no session carryover), so a + # fresh Negotiate token must be generated and accepted on every single request, not just + # the first one on a connection. Headers are captured at the actual transport call + # (urllib3/aiohttp), so this confirms what was truly sent over the wire. + client = client_factory( + host=kerberos_manage.CLICKHOUSE_HOST, + port=kerberos_manage.CLICKHOUSE_PORT, + username="", + password="", + database="default", + use_kerberos=True, + ) + headers_sent = [] + + def _record(headers): + auth = headers.get("Authorization") + if auth: + headers_sent.append(auth) + + sync_patch, async_patch = _patch_auth_headers(_record) + with sync_patch, async_patch: + for _ in range(3): + assert call(client.command, "SELECT currentUser()") == "kuser" + + assert len(headers_sent) == 3 + assert len(set(headers_sent)) == 3, "expected a fresh Negotiate header per request, got a reused one" + + +def test_kerberos_rejects_reused_header(client_factory, call): + # Complements test_kerberos_multiple_sequential_requests, proving the guarantee from the + # other direction: ClickHouse actually rejects a stale/reused Negotiate header rather than + # merely happening to receive a fresh one each time. + client = client_factory( + host=kerberos_manage.CLICKHOUSE_HOST, + port=kerberos_manage.CLICKHOUSE_PORT, + username="", + password="", + database="default", + use_kerberos=True, + ) + captured = {"value": None} + + def _force_reuse(headers): + if captured["value"] is None: + captured["value"] = headers.get("Authorization") + else: + headers["Authorization"] = captured["value"] + + sync_patch, async_patch = _patch_auth_headers(_force_reuse) + with sync_patch, async_patch: + assert call(client.command, "SELECT currentUser()") == "kuser" + with pytest.raises(DatabaseError): + call(client.command, "SELECT currentUser()") diff --git a/tests/test_requirements.txt b/tests/test_requirements.txt index 5b634340..455b4a58 100644 --- a/tests/test_requirements.txt +++ b/tests/test_requirements.txt @@ -25,4 +25,3 @@ pyjwt[crypto]==2.10.1 pre-commit==4.3.0 ruff==0.15.8 mypy==2.1.0 - \ No newline at end of file diff --git a/tests/unit_tests/test_driver/test_kerberos.py b/tests/unit_tests/test_driver/test_kerberos.py new file mode 100644 index 00000000..84e1549c --- /dev/null +++ b/tests/unit_tests/test_driver/test_kerberos.py @@ -0,0 +1,400 @@ +"""Unit tests for the use_kerberos auth mode (sync and async). + +These avoid a live KDC and a live server: pyspnego itself is mocked (or, for the +"package not installed" cases, left as the real module-level None), so the tests +exercise validation and request/header wiring rather than a real GSSAPI handshake. +""" + +import base64 +from inspect import signature +from unittest.mock import MagicMock, call, patch + +import pytest + +import clickhouse_connect.driver as drv +import clickhouse_connect.driver.kerberos as kerberos_module +import clickhouse_connect.driver.options as options_module +from clickhouse_connect.driver import create_async_client, create_client +from clickhouse_connect.driver._backend.http_async import HttpAsyncBackend +from clickhouse_connect.driver._backend.http_sync import HttpSyncBackend +from clickhouse_connect.driver.asyncclient import AsyncClient +from clickhouse_connect.driver.client import Client +from clickhouse_connect.driver.exceptions import NotSupportedError, OperationalError, ProgrammingError +from clickhouse_connect.driver.httpclient import HttpClient + + +@pytest.fixture +def fake_spnego(): + """Patch clickhouse_connect.driver.options.spnego with a working fake module.""" + fake = MagicMock() + fake.client.return_value.step.return_value = b"fake-token" + with patch.object(options_module, "spnego", fake, create=True): + yield fake + + +class TestKerberosValidation: + def test_rejects_use_kerberos_with_username(self, fake_spnego): + with pytest.raises(ProgrammingError): + create_client(username="user_1", use_kerberos=True) + + def test_rejects_use_kerberos_with_password(self, fake_spnego): + with pytest.raises(ProgrammingError): + create_client(password="secret", use_kerberos=True) + + def test_rejects_use_kerberos_with_access_token(self, fake_spnego): + with pytest.raises(ProgrammingError): + create_client(access_token="t", use_kerberos=True) + + def test_rejects_use_kerberos_with_token_provider(self, fake_spnego): + with pytest.raises(ProgrammingError): + create_client(use_kerberos=True, token_provider=lambda: "t") + + def test_rejects_use_kerberos_with_client_cert(self, fake_spnego): + with pytest.raises(ProgrammingError): + create_client(use_kerberos=True, client_cert="cert.pem") + + @pytest.mark.asyncio + async def test_async_rejects_use_kerberos_with_username(self, fake_spnego): + with pytest.raises(ProgrammingError): + await create_async_client(username="user_1", use_kerberos=True) + + @pytest.mark.asyncio + async def test_async_rejects_use_kerberos_with_client_cert(self, fake_spnego): + with pytest.raises(ProgrammingError): + await create_async_client(use_kerberos=True, client_cert="cert.pem") + + def test_missing_pyspnego_raises_not_supported(self): + with patch.object(options_module, "spnego", None, create=True): + with pytest.raises(NotSupportedError): + create_client(interface="http", host="h", port=8123, use_kerberos=True) + + @pytest.mark.asyncio + async def test_async_missing_pyspnego_raises_not_supported(self): + with patch.object(options_module, "spnego", None, create=True): + with pytest.raises(NotSupportedError): + await create_async_client(interface="http", host="h", port=8123, use_kerberos=True) + + +class _RecordingClient: + """Stand-in mirroring the leading client signature so generic_args routing works.""" + + def __init__( + self, + interface=None, + host=None, + port=None, + username=None, + password=None, + database=None, + access_token=None, + token_provider=None, + use_kerberos=None, + kerberos_hostname_override=None, + settings=None, + **kwargs, + ): + self.host = host + self.use_kerberos = use_kerberos + self.kerberos_hostname_override = kerberos_hostname_override + self.extra = kwargs + self.server_tz = None + + def _add_integration_tag(self, name): + pass + + async def _initialize(self): + pass + + +class TestKerberosConstruction: + def test_direct_create_client(self): + with patch.object(drv, "HttpClient", _RecordingClient): + client = create_client(interface="http", host="h", port=8123, use_kerberos=True) + assert client.use_kerberos is True + + def test_create_client_via_generic_args(self): + with patch.object(drv, "HttpClient", _RecordingClient): + client = create_client(interface="http", host="h", port=8123, generic_args={"use_kerberos": True}) + assert client.use_kerberos is True + + def test_kerberos_hostname_override_reaches_client(self): + with patch.object(drv, "HttpClient", _RecordingClient): + client = create_client( + interface="http", host="h", port=8123, use_kerberos=True, kerberos_hostname_override="chnode1.example.com" + ) + assert client.kerberos_hostname_override == "chnode1.example.com" + + @pytest.mark.asyncio + async def test_create_async_client_via_generic_args(self): + with patch("clickhouse_connect.driver.asyncclient.AsyncClient", _RecordingClient): + client = await create_async_client(interface="http", host="h", port=8123, generic_args={"use_kerberos": True}) + assert client.use_kerberos is True + + def test_httpclient_signature_preserves_positional_parameters(self): + parameters = list(signature(HttpClient).parameters) + assert parameters[8] == "compress" + assert parameters[-2:] == ["use_kerberos", "kerberos_hostname_override"] + + def test_asyncclient_signature_preserves_positional_parameters(self): + parameters = list(signature(AsyncClient).parameters) + assert parameters[8] == "compress" + assert parameters[-2:] == ["use_kerberos", "kerberos_hostname_override"] + + def test_real_http_client_sets_kerberos_hostname_from_host(self, fake_spnego): + with patch.object(Client, "_init_common_settings"): + client = HttpClient(interface="http", host="chnode1", port=8123, username="", password="", database=None, use_kerberos=True) + assert client._backend.use_kerberos is True + assert client._backend.kerberos_hostname == "chnode1" + assert "Authorization" not in client._backend.headers + + def test_real_http_client_kerberos_hostname_override(self, fake_spnego): + with patch.object(Client, "_init_common_settings"): + client = HttpClient( + interface="http", + host="chnode1", + port=8123, + username="", + password="", + database=None, + use_kerberos=True, + kerberos_hostname_override="chnode1.example.com", + ) + assert client._backend.kerberos_hostname == "chnode1.example.com" + + def test_direct_httpclient_rejects_kerberos_with_password(self, fake_spnego): + with pytest.raises(ProgrammingError): + HttpClient("http", "chnode1", 8123, "", "secret", None, use_kerberos=True) + + def test_direct_httpclient_rejects_provider_without_calling_it(self, fake_spnego): + provider = MagicMock(return_value="token") + + with pytest.raises(ProgrammingError): + HttpClient("http", "chnode1", 8123, "", "", None, token_provider=provider, use_kerberos=True) + + provider.assert_not_called() + + def test_direct_asyncclient_rejects_kerberos_with_password(self, fake_spnego): + with pytest.raises(ProgrammingError): + AsyncClient("http", "chnode1", 8123, "", "secret", None, use_kerberos=True) + + +class TestKerberosAuthContext: + def test_builds_kerberos_negotiate_header(self, fake_spnego): + context = kerberos_module.KerberosAuthContext("chnode1.example.com") + + assert context.authorization_header == "Negotiate " + base64.b64encode(b"fake-token").decode() + fake_spnego.client.assert_called_once_with( + hostname="chnode1.example.com", + service="HTTP", + protocol="kerberos", + ) + + def test_validates_server_token_with_same_context(self, fake_spnego): + spnego_context = fake_spnego.client.return_value + spnego_context.step.side_effect = [b"request-token", None] + spnego_context.complete = True + context = kerberos_module.KerberosAuthContext("chnode1.example.com") + + context.validate_response("Negotiate " + base64.b64encode(b"response-token").decode()) + + assert spnego_context.step.call_args_list == [call(), call(b"response-token")] + + @pytest.mark.parametrize( + "authenticate_header", + [None, "", "Basic abc", "Negotiate", "Negotiate !!!"], + ) + def test_rejects_missing_or_malformed_server_token(self, fake_spnego, authenticate_header): + context = kerberos_module.KerberosAuthContext("chnode1.example.com") + + with pytest.raises(OperationalError, match="mutual authentication failed"): + context.validate_response(authenticate_header) + + def test_rejects_incomplete_mutual_authentication(self, fake_spnego): + fake_spnego.client.return_value.complete = False + context = kerberos_module.KerberosAuthContext("chnode1.example.com") + + with pytest.raises(OperationalError, match="did not complete"): + context.validate_response("Negotiate " + base64.b64encode(b"response-token").decode()) + + def test_missing_pyspnego(self): + with patch.object(options_module, "spnego", None, create=True): + with pytest.raises(NotSupportedError): + kerberos_module.KerberosAuthContext("host") + + def test_negotiation_failure_preserves_pyspnego_message(self): + class _FakeSpnegoError(Exception): + pass + + fake = MagicMock() + fake.exceptions.SpnegoError = _FakeSpnegoError + fake.client.return_value.step.side_effect = _FakeSpnegoError("credential cache is unavailable") + + with patch.object(options_module, "spnego", fake, create=True): + with pytest.raises(OperationalError, match="credential cache is unavailable") as exc_info: + kerberos_module.KerberosAuthContext("chnode1.example.com") + + assert exc_info.value.__cause__ is not None + + def test_missing_system_kerberos_support_is_operational_error(self): + fake = MagicMock() + fake.exceptions.SpnegoError = RuntimeError + fake.client.side_effect = ImportError("GSSAPI support is unavailable") + + with patch.object(options_module, "spnego", fake, create=True): + with pytest.raises(OperationalError, match="GSSAPI support is unavailable"): + kerberos_module.KerberosAuthContext("chnode1.example.com") + + +def _response(status=200, authenticate_header="Negotiate response-token"): + response = MagicMock() + response.status = status + response.headers = {"WWW-Authenticate": authenticate_header} if authenticate_header is not None else {} + return response + + +def _kerberos_attempt(header): + context = MagicMock() + context.authorization_header = header + return context + + +def _build_sync_kerberos_backend(hostname="chnode1.example.com"): + return HttpSyncBackend( + url="http://localhost:8123", + pool_manager=MagicMock(), + owns_pool_manager=False, + headers={}, + params={}, + timeout=None, + server_host_name=None, + token_provider=None, + autogenerate_query_id=False, + use_kerberos=True, + kerberos_hostname=hostname, + ) + + +class TestSyncKerberosRequest: + def test_retry_uses_fresh_context_and_validates_success(self): + backend = _build_sync_kerberos_backend() + responses = iter([_response(503), _response()]) + request_kwargs = [] + + def request(method, url, **kwargs): + request_kwargs.append(dict(kwargs, headers=dict(kwargs["headers"]))) + return next(responses) + + backend.http.request.side_effect = request + first_context = _kerberos_attempt("Negotiate request-1") + second_context = _kerberos_attempt("Negotiate request-2") + + with patch( + "clickhouse_connect.driver._backend.http_sync.KerberosAuthContext", + side_effect=[first_context, second_context], + ) as context_factory: + response = backend.request(b"SELECT 13", {}, retries=1) + + assert response.status == 200 + assert [kwargs["headers"]["Authorization"] for kwargs in request_kwargs] == [ + "Negotiate request-1", + "Negotiate request-2", + ] + assert [kwargs["retries"] for kwargs in request_kwargs] == [0, 0] + assert context_factory.call_args_list == [call("chnode1.example.com"), call("chnode1.example.com")] + first_context.validate_response.assert_not_called() + second_context.validate_response.assert_called_once_with("Negotiate response-token") + + def test_success_without_server_token_fails(self): + backend = _build_sync_kerberos_backend() + backend.http.request.return_value = _response(authenticate_header=None) + context = _kerberos_attempt("Negotiate request-token") + context.validate_response.side_effect = OperationalError("missing server token") + + with patch("clickhouse_connect.driver._backend.http_sync.KerberosAuthContext", return_value=context): + with pytest.raises(OperationalError, match="missing server token"): + backend.request(b"SELECT 13", {}) + + backend.http.request.return_value.close.assert_called_once() + + +class _FakeAsyncLease: + def __init__(self, session): + self.session = session + self.inflight = 0 + + def acquire(self): + self.inflight += 1 + + def release(self): + self.inflight -= 1 + + +class _FakeAsyncSession: + def __init__(self, responses): + self._responses = iter(responses) + self.closed = False + self.headers = {} + self.request_kwargs = [] + + async def request(self, **kwargs): + self.request_kwargs.append(dict(kwargs, headers=dict(kwargs["headers"]))) + return next(self._responses) + + +def _build_async_kerberos_backend(responses, hostname="chnode1.example.com", use_kerberos=True): + backend = HttpAsyncBackend( + url="http://localhost:8123", + headers={}, + client_settings={}, + timeout=None, + connector_kwargs={}, + ssl_context=None, + proxy_url=None, + server_host_name=None, + token_provider=None, + autogenerate_query_id=False, + use_kerberos=use_kerberos, + kerberos_hostname=hostname if use_kerberos else None, + ) + session = _FakeAsyncSession(responses) + backend.session_lease = _FakeAsyncLease(session) + return backend, session + + +class TestAsyncKerberosRequest: + @pytest.mark.asyncio + @pytest.mark.parametrize( + ("use_kerberos", "expected_method"), + [(False, "GET"), (True, "POST")], + ) + async def test_get_method_replay_protection(self, use_kerberos, expected_method): + backend, session = _build_async_kerberos_backend([_response()], use_kerberos=use_kerberos) + context = _kerberos_attempt("Negotiate request-token") + + with patch("clickhouse_connect.driver._backend.http_async.KerberosAuthContext", return_value=context): + await backend.request(b"", {}, method="GET") + + assert session.request_kwargs[0]["method"] == expected_method + assert ("allow_redirects" in session.request_kwargs[0]) is use_kerberos + + @pytest.mark.asyncio + async def test_retry_uses_fresh_context_and_validates_success(self): + backend, session = _build_async_kerberos_backend([_response(503), _response()]) + first_context = _kerberos_attempt("Negotiate request-1") + second_context = _kerberos_attempt("Negotiate request-2") + + with patch( + "clickhouse_connect.driver._backend.http_async.KerberosAuthContext", + side_effect=[first_context, second_context], + ) as context_factory: + response = await backend.request(b"SELECT 13", {}, retries=1) + + assert response.status == 200 + assert [kwargs["headers"]["Authorization"] for kwargs in session.request_kwargs] == [ + "Negotiate request-1", + "Negotiate request-2", + ] + assert all(kwargs["allow_redirects"] is False for kwargs in session.request_kwargs) + assert context_factory.call_args_list == [call("chnode1.example.com"), call("chnode1.example.com")] + first_context.validate_response.assert_not_called() + second_context.validate_response.assert_called_once_with("Negotiate response-token")