Only load certificates into existing context if changes are detected

[sethmlarson]

I've created this as a potential alternate implementation of #213, but instead of using a lock and to preserve the behavior of getting new certificates potentially per handshake we load certificates into an empty SSLContext and compare with the existing SSLContext, only then reloading the certificates.

My suspicion is that updating certificate stores mid-application is rare, but we still want to preserve this behavior because it's what the other two backends do automatically for us. If this seems good I can add test cases.

cc @notatallshaw

[notatallshaw]

Disclaimer: I used codex's Daybreak Blue to to drive this review. I then went through a few rounds of MREs and adjudication to confirm the findings. There is one typo and two independent refresh bugs.

[notatallshaw]

Matching certificate sets do not prove that ctx has registered SSL_CERT_DIR; OpenSSL loads certificates from that directory lazily. I reproduced this with root A in SSL_CERT_FILE, root B in a hashed SSL_CERT_DIR, and root A preloaded into ctx. Main verifies a server signed by B. This branch rejects it until set_default_verify_paths() is called explicitly.

One possible fix is to store the automatic source state on the context. This is the implementation I tested; it replaces the current _configure_context() and adds import hashlib:

Example implementation

_AUTOMATIC_SOURCE_STATE_ATTR = "_truststore_openssl_automatic_source_state"
_MISSING_SOURCE_STATE = object()


class _FileState(typing.NamedTuple):
    path: str
    digest: bytes


class _CASourceState(typing.NamedTuple):
    cafile: _FileState | None
    capath: str | None
    fallback: tuple[_FileState, ...]


@contextlib.contextmanager
def _configure_context(ctx: ssl.SSLContext) -> typing.Iterator[None]:
    source_state = _ca_source_state()
    if (
        getattr(ctx, _AUTOMATIC_SOURCE_STATE_ATTR, _MISSING_SOURCE_STATE)
        != source_state
    ):
        _load_ca_certs(ctx)
        setattr(ctx, _AUTOMATIC_SOURCE_STATE_ATTR, source_state)

    yield


def _file_state(path: str) -> _FileState | None:
    try:
        with open(path, "rb") as file:
            digest = hashlib.sha256()
            while chunk := file.read(128 * 1024):
                digest.update(chunk)
    except OSError:
        return None
    return _FileState(path, digest.digest())


def _fallback_source_state() -> tuple[_FileState, ...]:
    states = []
    for candidate in _CA_FILE_CANDIDATES:
        if os.path.isdir(candidate):
            for name in sorted(os.listdir(candidate)):
                if _HASHED_CERT_FILENAME_RE.match(name):
                    state = _file_state(os.path.join(candidate, name))
                    if state is not None:
                        states.append(state)
        elif os.path.isfile(candidate):
            state = _file_state(candidate)
            if state is not None:
                states.append(state)
    return tuple(states)


def _ca_source_state() -> _CASourceState:
    defaults = ssl.get_default_verify_paths()
    if defaults.cafile or (
        defaults.capath and _capath_contains_certs(defaults.capath)
    ):
        return _CASourceState(
            cafile=_file_state(defaults.cafile) if defaults.cafile else None,
            capath=defaults.capath,
            fallback=(),
        )
    return _CASourceState(
        cafile=None,
        capath=None,
        fallback=_fallback_source_state(),
    )

This version stores the state only after _load_ca_certs() succeeds. I'm not attached to this shape; the important parts are performing the first load and distinguishing changed eager sources from unchanged ones.

[notatallshaw]

Certificate sets do not change when only the CRL changes. I reproduced this with a bundle containing a CA and an empty CRL, then replaced it with the same CA and a newer CRL that revokes the server certificate. Main rejects the next handshake. This branch accepts it until an explicit reload.

A content digest is one way to include CRL-only changes in the source state (with import hashlib):

def _file_state(path: str) -> tuple[str, bytes] | None:
    try:
        with open(path, "rb") as file:
            digest = hashlib.sha256()
            while chunk := file.read(128 * 1024):
                digest.update(chunk)
    except OSError:
        return None
    return path, digest.digest()

This caught a CRL replacement even when the file's inode, size, and modification time were unchanged. The fingerprint does not have to use SHA-256, but it needs to change when the CRL bytes change; a CRL count alone would still miss replacements.

[notatallshaw]

Track whether automatic roots have been loaded. A context may already contain an application CA while OpenSSL's directory-backed system roots remain lazy and absent from get_ca_certs(), so the certificate-set comparison cannot establish that the initial load happened.

Suggested change

	_CA_CERTS_LOADED_ATTR = "_truststore_ca_certs_loaded"
	@contextlib.contextmanager

[notatallshaw]

Load roots once before comparing sets, and keep the probe on the underlying context type.

Suggested change

	ctx_certs = set(ctx.get_ca_certs(binary_form=True))
	new_certs = set() # Unused if ctx_certs is empty.
	if ctx_certs:
	# Load all certs into an empty SSLContext.
	new_ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
	_load_ca_certs(new_ctx)
	new_certs.update(new_ctx.get_ca_certs(binary_form=True))
	# If the new certificates aren't all in the current
	# context, then load all certificates again into
	# the existing context.
	if not ctx_certs or not new_certs.issubset(ctx_certs):
	_load_ca_certs(ctx)
	should_load = not getattr(ctx, _CA_CERTS_LOADED_ATTR, False)
	if not should_load:
	ctx_certs = set(ctx.get_ca_certs(binary_form=True))
	# Load all certs into an empty SSLContext.
	new_ctx = type(ctx)(ssl.PROTOCOL_TLS_CLIENT)
	_load_ca_certs(new_ctx)
	new_certs = set(new_ctx.get_ca_certs(binary_form=True))
	should_load = not new_certs.issubset(ctx_certs)
	# If the new certificates aren't all in the current
	# context, then load all certificates again into
	# the existing context.
	if should_load:
	_load_ca_certs(ctx)
	setattr(ctx, _CA_CERTS_LOADED_ATTR, True)

[notatallshaw]

Added further possible fixes for CI.