Custom Backend to record HTTPS certificate.

[dalf]

I would like to record the HTTPS certificate.

One way is to have a custom Backend, but the January solution can't be used anymore: encode/httpx#782 (comment)

A partial update:

from ssl import SSLContext
from typing import Optional

import asyncio
import httpx
import uvloop
from httpcore._backends.auto import AutoBackend
from httpcore._backends.asyncio import SocketStream
from httpcore._async.connection_pool import AsyncConnectionPool
from httpcore._types import TimeoutDict


class CustomBackend(AutoBackend):
    async def open_tcp_stream(
        self,
        hostname: bytes,
        port: int,
        ssl_context: Optional[SSLContext],
        timeout: TimeoutDict,
    ) -> SocketStream:
        value = await super().open_tcp_stream(hostname, port, ssl_context, timeout)
        # use value.stream_reader._transport.get_extra_info('ssl_object')
        return value


class CustomAsyncConnectionPool(AsyncConnectionPool):

    def __init__(
        self,
        ssl_context: SSLContext = None,
        max_connections: int = None,
        max_keepalive: int = None,
        keepalive_expiry: float = None,
        http2: bool = False,
    ):
        super().__init__(ssl_context, max_connections, max_keepalive, keepalive_expiry, http2)
        self._backend = CustomBackend()


async def main():
    dispatch = CustomAsyncConnectionPool(http2=True)
    async with httpx.AsyncClient(dispatch=dispatch) as client:
        response = await client.get('https://github.com/')


loop = asyncio.get_event_loop()
loop.run_until_complete(main())

Unfortunately, in the request method, AsyncHTTPConnection is instanciated without the Backend argument:

httpcore/httpcore/_async/connection_pool.py

Lines 143 to 145 in 59eb0a3

	connection = AsyncHTTPConnection(
	origin=origin, http2=self._http2, ssl_context=self._ssl_context,
	)

And the Backend is instanciated again:

httpcore/httpcore/_async/connection.py

Lines 20 to 40 in 59eb0a3

	def __init__(
	self,
	origin: Origin,
	http2: bool = False,
	ssl_context: SSLContext = None,
	socket: AsyncSocketStream = None,
	):
	self.origin = origin
	self.http2 = http2
	self.ssl_context = SSLContext() if ssl_context is None else ssl_context
	self.socket = socket
	if self.http2:
	self.ssl_context.set_alpn_protocols(["http/1.1", "h2"])
	self.connection: Union[None, AsyncHTTP11Connection, AsyncHTTP2Connection] = None
	self.is_http11 = False
	self.is_http2 = False
	self.connect_failed = False
	self.expires_at: Optional[float] = None
	self.backend = AutoBackend()

I guess the purpose is to match the SyncHTTPConnection constructor signature.

Some solutions:

• add an optional backend argument to the AsyncHTTPConnection constructor.
• add an AsyncBackend factory.
• out of scope, at least for now, fall back to AutoBackend monkey patch.

[florimondmanca]

out of scope, at least for now, fall back to AutoBackend monkey patch.

Seems like the most viable approach to me, and it should work pretty okay?

Quick note that I think you should specifically target AsyncioBackend instead, since AutoBackend is using it in the asyncio case, i.e.:

from httpcore._backends.asyncio import AsyncioBackend

open_tcp_stream = AsyncioBackend.open_tcp_stream

async def my_open_tcp_stream(*args, **kwargs):
    socket = await open_tcp_stream(*args, **kwargs)
    # Use `socket.stream_reader._transport.get_extra_info('ssl_object')`
    return socket

AsyncioBackend.open_tcp_stream = my_open_tcp_stream

[dalf]

Seems like the most viable approach to me, and it should work pretty okay?

Yes, the monkey patch works as expected. Thank you.

I was hoping to implement the "Advanced connection options" at the same time (for the local_addr support), but this is a different issue.