Issue with ssl.ca.location schema registry client option after migration to httpx

[pkwarren]

We're using the schema registry client in confluent_kafka 2.8.0 and encountering an issue when talking to a schema registry with self-signed certificates. In previous releases, the ssl.ca.location parameter worked as a str, but it appears after the migration to httpx this is used as the httpx verify option directly.

In https://www.python-httpx.org/advanced/ssl/#configuring-client-instances, the docs for httpx state that the verify parameter should be set as follows to use a custom CA store:

# Use an explicitly configured certificate store.
ctx = ssl.create_default_context(cafile="path/to/certs.pem")  # Either cafile or capath.
client = httpx.Client(verify=ctx)

I tried the following:

• Setting ssl.ca.location to a str: FAILED.
• Setting ssl.ca.location to a ssl context as documented above: FAILED. See the error message below for details.
• Setting ssl.ca.location=False: SUCCESS.

         .venv/lib/python3.12/site-packages/confluent_kafka/schema_registry/protobuf.py:563: in __call__
            references = self._resolve_dependencies(
        .venv/lib/python3.12/site-packages/confluent_kafka/schema_registry/protobuf.py:522: in _resolve_dependencies
            self._registry.register_schema(subject, schema)
        .venv/lib/python3.12/site-packages/confluent_kafka/schema_registry/schema_registry_client.py:651: in register_schema
            registered_schema = self.register_schema_full_response(subject_name, schema, normalize_schemas)
        .venv/lib/python3.12/site-packages/confluent_kafka/schema_registry/schema_registry_client.py:683: in register_schema_full_response
            response = self._rest_client.post(
        .venv/lib/python3.12/site-packages/confluent_kafka/schema_registry/schema_registry_client.py:216: in post
            return self.send_request(url, method='POST', body=body)
        .venv/lib/python3.12/site-packages/confluent_kafka/schema_registry/schema_registry_client.py:273: in send_request
            raise e
        .venv/lib/python3.12/site-packages/confluent_kafka/schema_registry/schema_registry_client.py:262: in send_request
            response = self.send_http_request(
        .venv/lib/python3.12/site-packages/confluent_kafka/schema_registry/schema_registry_client.py:316: in send_http_request
            response = self.session.request(
        .venv/lib/python3.12/site-packages/httpx/_client.py:825: in request
            return self.send(request, auth=auth, follow_redirects=follow_redirects)
        .venv/lib/python3.12/site-packages/httpx/_client.py:914: in send
            response = self._send_handling_auth(
        .venv/lib/python3.12/site-packages/httpx/_client.py:942: in _send_handling_auth
            response = self._send_handling_redirects(
        .venv/lib/python3.12/site-packages/httpx/_client.py:979: in _send_handling_redirects
            response = self._send_single_request(request)
        .venv/lib/python3.12/site-packages/httpx/_client.py:1014: in _send_single_request
            response = transport.handle_request(request)
        .venv/lib/python3.12/site-packages/httpx/_transports/default.py:249: in handle_request
            with map_httpcore_exceptions():
        /usr/lib/python3.12/contextlib.py:158: in __exit__
            self.gen.throw(value)
        _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
        
            @contextlib.contextmanager
            def map_httpcore_exceptions() -> typing.Iterator[None]:
                global HTTPCORE_EXC_MAP
                if len(HTTPCORE_EXC_MAP) == 0:
                    HTTPCORE_EXC_MAP = _load_httpcore_exceptions()
                try:
                    yield
                except Exception as exc:
                    mapped_exc = None
            
                    for from_exc, to_exc in HTTPCORE_EXC_MAP.items():
                        if not isinstance(exc, from_exc):
                            continue
                        # We want to map to the most specific exception we can find.
                        # Eg if `exc` is an `httpcore.ReadTimeout`, we want to map to
                        # `httpx.ReadTimeout`, not just `httpx.TimeoutException`.
                        if mapped_exc is None or issubclass(to_exc, mapped_exc):
                            mapped_exc = to_exc
            
                    if mapped_exc is None:  # pragma: no cover
                        raise
            
                    message = str(exc)
        >           raise mapped_exc(message) from exc
        E           httpx.ConnectError: [SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed: unable to get local issuer certificate (_ssl.c:1000)

[marcinciesinski]

I was fighting with this issue entire day and I could not find the resolution (those error messages are to laconic !)

And finally, when I read this comment I tried to pass ssl context instead of str path and this helps me to overcome this problem as @pkwarren suggested:

ca_context = ssl.create_default_context(cafile=properties.kafka_ca_cert)

schema_registry_configuration = {
    "url": properties.kafka_schema_registry_url,
    "ssl.ca.location": ca_context,
    "ssl.certificate.location": properties.kafka_schema_registry_ssl_cert,
    "ssl.key.location": properties.kafka_schema_registry_ssl_key,
}

So for me this way of creating the ssl context works.

PS: Remember that ca file in pem format might require to have entire CA chain in it. (that's why you probably got unable to get local issuer certificate (_ssl.c:1000)