Lazy instantiation of ssl_context in a Client or Transport

[kafonek]

Checklist

• There are no similar issues or pull requests for this yet.
• I discussed this idea on the community chat

This topic probably touches on -

• Mount API syntax - Mount API #977
• Maybe the Transport API - Finalising the Transport API for 1.0. #1274

Is your feature related to a problem? Please describe.

In some situations, we need to prompt a user for a password just prior to creating an ssl.SSLContext. For instance, if a user has password-protected PKCS12 certificates, we use pypki2 or requests_pkcs12 to decrypt the certificate and build an ssl.SSLContext. While I do not have an example on hand for HSM (Hardware Security Module, PKCS11 format, "Smart Cards" for example), these will probably fit into the same use case.

Normally, the flow in something like a Jupyter Notebook would not be a problem - context = pypki2config.ssl_context() -> prompts for password -> client = httpx.Client(verify=context). The wrinkle is that we have a development environment where some domains require a PKI (PKCS12) authenticated request, while others don't. I'll call them a list of audited domains and clear domains.

The behavior I want is for a user to create a client = InternalClient() object, a subclass of httpx.Client that knows about which domains need a PKI-sourced ssl_context and which need a bare httpx.create_ssl_context(). I would like to lazily instantiate those contexts so that there is no password prompt at client init, and may never be a password prompt if the user does not make a request to an audited domain.

Our solution right now is to use a subclassed requests.Session object, session = InternalSession(). We override .send (although .get_adapter could have worked too) to create/mount the right adapter at the time of the request.

Describe the solution you would like.

There are probably many ways to make our use-case work. Here are some options that come to mind -

• Expose a client.mount method (@tomchristie was against this in 977), which we would invoke while overriding .send or something
• Have the mounts dictionary values be a Transport or a callback that returns a Transport at the time the Transport is needed
• Create a .get_transport(url) method that can be overridden (ala .get_adapter in requests). The default behavior would be to perform URL matching from the .get_transports() dictionary
• Change the httpcore.Transport ssl_context init variable to accept a callback, and that callback wouldn't be executed until needed (such as when ._open_socket is invoked)

Thanks so much!

[tomchristie]

So, some options here...

1. Undoubtedly you'd be able to do what you need here with a custom transport - the request method could dispatch between either the standard transport or the password transport, eg...

class HelloWorldTransport(httpcore.SyncHTTPTransport):
    def request(self, method, url, headers=None, stream=None, timeout=None):
        if self.use_password_transport(url):
            if not hasattr(self, 'password_transport'):
                self.password_transport = self.init_password_transport(...)
            return self.password_transport(method, url, headers=headers, stream=stream, timeout=timeout)

        if not hasattr(self, 'standard_transport'):
            self.standard_transport = self.init_standard_transport(...)
        self.standard_transport(method, url, headers=headers, stream=stream, timeout=timeout)

    def init_standard_transport(self):
        ...

    def init_password_transport(self):
        ...

Note that the transport API likely will change with the next median version bump, so you'd want to be pinned to httpx 0.16.*.

2. You could override _transport_for_url. It's a private method so you'd want to make sure you're pinning httpx exactly, and doubly checking that the signature stays the same whenever you upgrade. Actually a perfectly reasonable approach.

3. You could wait for Add support for Mount API #1362 to land. Although that depends on if you know ahead of time what set of hostnames need the password prompt transport.

[kafonek]

Thanks @tomchristie ! I totally missed _transport_for_url or misunderstood it. That works for the short term. I'll watch that method for any changes, and keep on top of #1362 to see if it's a better long term solution.

For anyone else that ends up with the same use-case, this is generally what I'm using right now:

import pypki2config

class Client(httpx.Client):
    def _transport_for_url(self, url):
        if url.host in AUDITED_DOMAINS:
            ctx = pypki2config.ssl_context()
            return self._init_transport(verify=ctx)
        else:
            ctx = ssl.create_default_context()
            return self._init_transport(verify=ctx)

### and the same override code for AsyncClient