fix(proxy): build SSL contexts for custom CA bundles (#1134)

## Description

Build explicit `ssl.SSLContext` objects for custom CA bundles configured
through `SSL_CERT_FILE` and `REQUESTS_CA_BUNDLE`. Python 3.13 / newer
OpenSSL can reject some enterprise/private PKI roots that platform TLS
stacks accept, for example roots without a `keyUsage` extension. The new
custom-CA contexts keep certificate verification enabled while clearing
only `ssl.VERIFY_X509_STRICT`.

Closes #

## Type of Change

- [x] Bug fix (non-breaking change that fixes an issue)
- [ ] New feature (non-breaking change that adds functionality)
- [ ] Breaking change (fix or feature that would cause existing
functionality to change)
- [ ] Documentation update
- [ ] Performance improvement
- [ ] Code refactoring (no functional changes)

## Changes Made

- Build replacement `SSLContext` objects for `SSL_CERT_FILE` and
`REQUESTS_CA_BUNDLE` instead of passing raw CA bundle paths to httpx.
- Clear only `ssl.VERIFY_X509_STRICT` for operator-provided custom CA
contexts; certificate verification, hostname verification, expiry
checks, and chain validation stay enabled.
- Preserve `NODE_EXTRA_CA_CERTS` additive behavior by loading the extra
CA bundle on top of the default/system trust store.
- Update existing SSL context tests for replacement CA contexts, env-var
priority, missing-path fallthrough, and strict-mode relaxation.
- Add an Unreleased changelog entry for the proxy bug fix.

## Testing

- [x] Unit tests pass (`pytest`)
- [x] Linting passes (`ruff check .`)
- [ ] Type checking passes (`mypy headroom`)
- [x] New tests added for new functionality
- [x] Manual testing performed

### Test Output

```text
$ PYTHONPATH=. .venv/bin/python -m pytest tests/test_ssl_context.py -q
collected 12 items

tests/test_ssl_context.py ............                                   [100%]

12 passed, 1 warning in 0.11s
```

```text
$ uv tool run ruff check headroom/proxy/ssl_context.py tests/test_ssl_context.py CHANGELOG.md
All checks passed!

$ uv tool run ruff format --check headroom/proxy/ssl_context.py tests/test_ssl_context.py
2 files already formatted
```

## Real Behavior Proof

- Environment: macOS, Python 3.13.2, Headroom checkout on this branch,
`SSL_CERT_FILE` / `REQUESTS_CA_BUNDLE` / `NODE_EXTRA_CA_CERTS` pointed
at an enterprise/private PKI CA bundle. Upstream HTTPS endpoint name is
redacted.
- Exact command / steps: Ran a Python smoke script that imports
`headroom.proxy.ssl_context.find_ca_bundle()`, passes the returned
verifier into `httpx.AsyncClient(verify=...)`, and performs a GET
against the enterprise HTTPS endpoint that previously failed with
OpenSSL strict verification.
- Observed result: The request used an `SSLContext`, strict X.509
verification was disabled for that custom CA context, and the request
reached the upstream HTTP response:

```text
verify_type SSLContext
strict_enabled False
status 302
```

- Not tested: Full `uv run pytest`, `uv sync --extra dev`, and `mypy
headroom` in this local environment. The editable build currently fails
before tests run because Cargo's `ort-sys` build script cannot download
ORT prebuilt binaries due an unrelated local certificate verification
error against the ORT CDN. No dependency or lockfile changes are
included in this PR.

## Review Readiness

- [x] I have performed a self-review
- [x] This PR is ready for human review

## Checklist

- [x] My code follows the project's style guidelines
- [x] I have performed a self-review of my code
- [x] I have commented my code, particularly in hard-to-understand areas
- [ ] I have made corresponding changes to the documentation
- [x] My changes generate no new warnings
- [x] I have added tests that prove my fix is effective or that my
feature works
- [x] New and existing unit tests pass locally with my changes
- [x] I have updated the CHANGELOG.md if applicable

## Screenshots (if applicable)

N/A.

## Additional Notes

No dependencies or lockfiles changed. Documentation is unchanged because
this is a bug fix to existing custom CA environment-variable behavior
rather than a new user-facing configuration surface.

Signed-off-by: Mark Phelps <209477+markphelps@users.noreply.github.com>
Co-authored-by: JD Davis <mxjerrett@gmail.com>
This commit is contained in:
Mark Phelps 2026-06-22 19:48:19 -04:00 committed by GitHub
parent 978ffa0a6a
commit 561ba17ec2
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
3 changed files with 107 additions and 40 deletions

View file

@ -28,6 +28,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
### Bug Fixes ### Bug Fixes
* **proxy:** build SSL contexts for custom CA bundles so enterprise/private PKI roots work with Python/OpenSSL strict verification.
* **proxy:** route Codex OAuth image generation and edit requests through the ChatGPT Codex image backend, while preserving OpenAI API-key image passthrough ([#1215](https://github.com/chopratejas/headroom/pull/1215)). * **proxy:** route Codex OAuth image generation and edit requests through the ChatGPT Codex image backend, while preserving OpenAI API-key image passthrough ([#1215](https://github.com/chopratejas/headroom/pull/1215)).
* **wrap (codex):** keep RTK guidance in the global Codex `AGENTS.md` instead of modifying the shared project `AGENTS.md` ([#1235](https://github.com/chopratejas/headroom/issues/1235)). * **wrap (codex):** keep RTK guidance in the global Codex `AGENTS.md` instead of modifying the shared project `AGENTS.md` ([#1235](https://github.com/chopratejas/headroom/issues/1235)).
* **proxy:** enable SSO credential resolution in the native Bedrock route via the `aws-config` `sso` feature flag, making the credential chain match what `docs/bedrock.md` already documented ([#999](https://github.com/chopratejas/headroom/pull/999)). * **proxy:** enable SSO credential resolution in the native Bedrock route via the `aws-config` `sso` feature flag, making the credential chain match what `docs/bedrock.md` already documented ([#999](https://github.com/chopratejas/headroom/pull/999)).

View file

@ -27,16 +27,47 @@ _REPLACEMENT_CA_VARS = (
) )
def find_ca_bundle() -> str | ssl.SSLContext | None: def _relax_x509_strict_for_custom_ca(ctx: ssl.SSLContext, *, path: str) -> ssl.SSLContext:
"""Relax OpenSSL strict-mode checks for an operator-provided CA bundle.
Python 3.13 / newer OpenSSL can reject some enterprise or private PKI
roots that platform TLS stacks accept, for example roots without a
keyUsage extension. Clearing only ``VERIFY_X509_STRICT`` keeps certificate
verification, hostname verification, expiry checks, and chain validation
enabled while making custom CA bundles usable in those environments.
"""
strict_flag = getattr(ssl, "VERIFY_X509_STRICT", 0)
if strict_flag and ctx.verify_flags & strict_flag:
ctx.verify_flags &= ~strict_flag
logger.info("event=ssl_x509_strict_disabled_for_custom_ca path=%s", path)
return ctx
def _replacement_ca_context(path: str) -> ssl.SSLContext:
"""Build a replacement trust-store context from a CA bundle path."""
ctx = ssl.create_default_context(cafile=path)
ctx.set_alpn_protocols(["h2", "http/1.1"])
return _relax_x509_strict_for_custom_ca(ctx, path=path)
def _additive_ca_context(path: str) -> ssl.SSLContext:
"""Build an additive trust-store context from a CA bundle path."""
ctx = ssl.create_default_context()
ctx.load_verify_locations(cafile=path)
ctx.set_alpn_protocols(["h2", "http/1.1"])
return _relax_x509_strict_for_custom_ca(ctx, path=path)
def find_ca_bundle() -> ssl.SSLContext | None:
"""Return a CA verification target for httpx's ``verify=`` parameter. """Return a CA verification target for httpx's ``verify=`` parameter.
``SSL_CERT_FILE`` and ``REQUESTS_CA_BUNDLE`` use **replacement** ``SSL_CERT_FILE`` and ``REQUESTS_CA_BUNDLE`` use **replacement**
semantics: the returned path becomes the *only* trust store. semantics: the returned context trusts that bundle as its trust store.
``NODE_EXTRA_CA_CERTS`` uses **additive** semantics (matching Node.js): ``NODE_EXTRA_CA_CERTS`` uses **additive** semantics (matching Node.js):
an ``ssl.SSLContext`` is returned that contains the default/system the returned context contains the default/system roots *plus* the extra
roots *plus* the extra certificate, so public upstreams stay reachable certificate, so public upstreams stay reachable when the extra bundle
when the extra bundle contains only a private/internal root. contains only a private/internal root.
Returns ``None`` when no env var is set (or all paths are missing), Returns ``None`` when no env var is set (or all paths are missing),
which signals to the caller to use httpx's default TLS verification. which signals to the caller to use httpx's default TLS verification.
@ -49,7 +80,7 @@ def find_ca_bundle() -> str | ssl.SSLContext | None:
var, var,
path, path,
) )
return path return _replacement_ca_context(path)
if path and not os.path.isfile(path): if path and not os.path.isfile(path):
logger.warning( logger.warning(
"event=ssl_ca_bundle_missing env_var=%s path=%r (skipped)", "event=ssl_ca_bundle_missing env_var=%s path=%r (skipped)",
@ -59,14 +90,11 @@ def find_ca_bundle() -> str | ssl.SSLContext | None:
node_path = os.environ.get("NODE_EXTRA_CA_CERTS") node_path = os.environ.get("NODE_EXTRA_CA_CERTS")
if node_path and os.path.isfile(node_path): if node_path and os.path.isfile(node_path):
ctx = ssl.create_default_context()
ctx.load_verify_locations(cafile=node_path)
ctx.set_alpn_protocols(["h2", "http/1.1"])
logger.info( logger.info(
"event=ssl_ca_bundle_loaded env_var=NODE_EXTRA_CA_CERTS path=%s additive=true", "event=ssl_ca_bundle_loaded env_var=NODE_EXTRA_CA_CERTS path=%s additive=true",
node_path, node_path,
) )
return ctx return _additive_ca_context(node_path)
if node_path and not os.path.isfile(node_path): if node_path and not os.path.isfile(node_path):
logger.warning( logger.warning(
"event=ssl_ca_bundle_missing env_var=NODE_EXTRA_CA_CERTS path=%r (skipped)", "event=ssl_ca_bundle_missing env_var=NODE_EXTRA_CA_CERTS path=%r (skipped)",

View file

@ -2,21 +2,22 @@
Covers: Covers:
- Returns None when no env var is set - Returns None when no env var is set
- Returns a path string when SSL_CERT_FILE points to a valid PEM file - Returns an ssl.SSLContext when SSL_CERT_FILE points to a valid PEM file
- Returns a path string when REQUESTS_CA_BUNDLE points to a valid PEM file - Returns an ssl.SSLContext when REQUESTS_CA_BUNDLE points to a valid PEM file
- Replacement contexts relax OpenSSL VERIFY_X509_STRICT for custom CA bundles
- Returns an ssl.SSLContext when NODE_EXTRA_CA_CERTS points to a valid PEM file - Returns an ssl.SSLContext when NODE_EXTRA_CA_CERTS points to a valid PEM file
- The SSLContext is additive: default/system roots are preserved (#998) - The NODE_EXTRA_CA_CERTS SSLContext is additive: default/system roots are preserved (#998)
- Priority order: SSL_CERT_FILE beats REQUESTS_CA_BUNDLE beats NODE_EXTRA_CA_CERTS - Priority order: SSL_CERT_FILE beats REQUESTS_CA_BUNDLE beats NODE_EXTRA_CA_CERTS
- Nonexistent paths are skipped (returns None if all paths are missing) - Nonexistent paths are skipped (returns None if all paths are missing)
""" """
from __future__ import annotations from __future__ import annotations
import os
import ssl import ssl
import pytest import pytest
from headroom.proxy import ssl_context
from headroom.proxy.ssl_context import find_ca_bundle from headroom.proxy.ssl_context import find_ca_bundle
# Minimal self-signed CA certificate (PEM) used only to verify that # Minimal self-signed CA certificate (PEM) used only to verify that
@ -59,6 +60,19 @@ def _clean_env(monkeypatch):
monkeypatch.delenv(var, raising=False) monkeypatch.delenv(var, raising=False)
class FakeSSLContext:
def __init__(self, verify_flags: int = 0) -> None:
self.verify_flags = verify_flags
self.loaded_cafile: str | None = None
self.alpn_protocols: list[str] | None = None
def load_verify_locations(self, *, cafile: str) -> None:
self.loaded_cafile = cafile
def set_alpn_protocols(self, protocols: list[str]) -> None:
self.alpn_protocols = protocols
class TestFindCaBundleNoEnvVars: class TestFindCaBundleNoEnvVars:
def test_returns_none_when_no_env_var_set(self, monkeypatch): def test_returns_none_when_no_env_var_set(self, monkeypatch):
_clean_env(monkeypatch) _clean_env(monkeypatch)
@ -66,19 +80,37 @@ class TestFindCaBundleNoEnvVars:
class TestFindCaBundleWithValidPem: class TestFindCaBundleWithValidPem:
def test_ssl_cert_file_returns_path(self, monkeypatch, ca_pem_file): def test_ssl_cert_file_returns_ssl_context(self, monkeypatch, ca_pem_file):
_clean_env(monkeypatch) _clean_env(monkeypatch)
monkeypatch.setenv("SSL_CERT_FILE", ca_pem_file) monkeypatch.setenv("SSL_CERT_FILE", ca_pem_file)
ctx = find_ca_bundle() ctx = find_ca_bundle()
assert isinstance(ctx, str) assert isinstance(ctx, ssl.SSLContext)
assert os.path.isfile(ctx)
def test_requests_ca_bundle_returns_path(self, monkeypatch, ca_pem_file): def test_requests_ca_bundle_returns_ssl_context(self, monkeypatch, ca_pem_file):
_clean_env(monkeypatch) _clean_env(monkeypatch)
monkeypatch.setenv("REQUESTS_CA_BUNDLE", ca_pem_file) monkeypatch.setenv("REQUESTS_CA_BUNDLE", ca_pem_file)
ctx = find_ca_bundle() ctx = find_ca_bundle()
assert isinstance(ctx, str) assert isinstance(ctx, ssl.SSLContext)
assert os.path.isfile(ctx)
def test_replacement_ca_context_relaxes_x509_strict(self, monkeypatch, ca_pem_file):
_clean_env(monkeypatch)
monkeypatch.setenv("SSL_CERT_FILE", ca_pem_file)
strict_flag = 0x20
created_context = FakeSSLContext(verify_flags=strict_flag | 0x100)
def fake_create_default_context(*, cafile: str | None = None):
assert cafile == ca_pem_file
return created_context
monkeypatch.setattr(ssl_context.ssl, "VERIFY_X509_STRICT", strict_flag, raising=False)
monkeypatch.setattr(ssl_context.ssl, "create_default_context", fake_create_default_context)
ctx = find_ca_bundle()
assert ctx is created_context
assert created_context.verify_flags & strict_flag == 0
assert created_context.verify_flags & 0x100
assert created_context.alpn_protocols == ["h2", "http/1.1"]
def test_node_extra_ca_certs_returns_ssl_context(self, monkeypatch, ca_pem_file): def test_node_extra_ca_certs_returns_ssl_context(self, monkeypatch, ca_pem_file):
"""NODE_EXTRA_CA_CERTS returns an SSLContext, not a bare path (#998).""" """NODE_EXTRA_CA_CERTS returns an SSLContext, not a bare path (#998)."""
@ -103,51 +135,56 @@ class TestFindCaBundlePriority:
def test_ssl_cert_file_beats_requests_ca_bundle(self, monkeypatch, tmp_path): def test_ssl_cert_file_beats_requests_ca_bundle(self, monkeypatch, tmp_path):
"""SSL_CERT_FILE is used first even when REQUESTS_CA_BUNDLE is also set.""" """SSL_CERT_FILE is used first even when REQUESTS_CA_BUNDLE is also set."""
_clean_env(monkeypatch) _clean_env(monkeypatch)
# Two distinct files so we can identify which was loaded.
pem1 = tmp_path / "first.pem" pem1 = tmp_path / "first.pem"
pem2 = tmp_path / "second.pem" pem2 = tmp_path / "second.pem"
pem1.write_bytes(_SELF_SIGNED_CA_PEM) pem1.write_bytes(_SELF_SIGNED_CA_PEM)
pem2.write_bytes(_SELF_SIGNED_CA_PEM) pem2.write_bytes(_SELF_SIGNED_CA_PEM)
monkeypatch.setenv("SSL_CERT_FILE", str(pem1)) monkeypatch.setenv("SSL_CERT_FILE", str(pem1))
monkeypatch.setenv("REQUESTS_CA_BUNDLE", str(pem2)) monkeypatch.setenv("REQUESTS_CA_BUNDLE", str(pem2))
created_context = FakeSSLContext()
# Both files are valid; we cannot easily inspect which CA was loaded def fake_create_default_context(*, cafile: str | None = None):
# into the context, but we can verify the function returns a path assert cafile == str(pem1)
# (not None) and that it is tied to SSL_CERT_FILE by temporarily return created_context
# making REQUESTS_CA_BUNDLE point to a nonexistent path.
monkeypatch.setenv("REQUESTS_CA_BUNDLE", "/nonexistent/path.pem") monkeypatch.setattr(ssl_context.ssl, "create_default_context", fake_create_default_context)
ctx = find_ca_bundle()
# SSL_CERT_FILE still valid → should return a path assert find_ca_bundle() is created_context
assert isinstance(ctx, str)
assert os.path.isfile(ctx)
def test_ssl_cert_file_beats_node_extra_ca_certs(self, monkeypatch, tmp_path): def test_ssl_cert_file_beats_node_extra_ca_certs(self, monkeypatch, tmp_path):
"""SSL_CERT_FILE takes precedence over NODE_EXTRA_CA_CERTS.""" """SSL_CERT_FILE takes precedence over NODE_EXTRA_CA_CERTS."""
_clean_env(monkeypatch) _clean_env(monkeypatch)
pem = tmp_path / "ca.pem" pem = tmp_path / "ca.pem"
pem.write_bytes(_SELF_SIGNED_CA_PEM) pem.write_bytes(_SELF_SIGNED_CA_PEM)
monkeypatch.setenv("SSL_CERT_FILE", str(pem)) monkeypatch.setenv("SSL_CERT_FILE", str(pem))
monkeypatch.setenv("NODE_EXTRA_CA_CERTS", "/nonexistent/node.pem") monkeypatch.setenv("NODE_EXTRA_CA_CERTS", "/nonexistent/node.pem")
created_context = FakeSSLContext()
ctx = find_ca_bundle() def fake_create_default_context(*, cafile: str | None = None):
assert isinstance(ctx, str) assert cafile == str(pem)
assert os.path.isfile(ctx) return created_context
monkeypatch.setattr(ssl_context.ssl, "create_default_context", fake_create_default_context)
assert find_ca_bundle() is created_context
def test_requests_ca_bundle_beats_node_extra_ca_certs(self, monkeypatch, tmp_path): def test_requests_ca_bundle_beats_node_extra_ca_certs(self, monkeypatch, tmp_path):
"""REQUESTS_CA_BUNDLE is used before NODE_EXTRA_CA_CERTS.""" """REQUESTS_CA_BUNDLE is used before NODE_EXTRA_CA_CERTS."""
_clean_env(monkeypatch) _clean_env(monkeypatch)
pem = tmp_path / "ca.pem" pem = tmp_path / "ca.pem"
pem.write_bytes(_SELF_SIGNED_CA_PEM) pem.write_bytes(_SELF_SIGNED_CA_PEM)
monkeypatch.setenv("SSL_CERT_FILE", "/nonexistent/ssl.pem") monkeypatch.setenv("SSL_CERT_FILE", "/nonexistent/ssl.pem")
monkeypatch.setenv("REQUESTS_CA_BUNDLE", str(pem)) monkeypatch.setenv("REQUESTS_CA_BUNDLE", str(pem))
monkeypatch.setenv("NODE_EXTRA_CA_CERTS", "/nonexistent/node.pem") monkeypatch.setenv("NODE_EXTRA_CA_CERTS", "/nonexistent/node.pem")
created_context = FakeSSLContext()
ctx = find_ca_bundle() def fake_create_default_context(*, cafile: str | None = None):
assert isinstance(ctx, str) assert cafile == str(pem)
assert os.path.isfile(ctx) return created_context
monkeypatch.setattr(ssl_context.ssl, "create_default_context", fake_create_default_context)
assert find_ca_bundle() is created_context
class TestFindCaBundleNonexistentPaths: class TestFindCaBundleNonexistentPaths:
@ -170,4 +207,5 @@ class TestFindCaBundleNonexistentPaths:
monkeypatch.setenv("REQUESTS_CA_BUNDLE", ca_pem_file) monkeypatch.setenv("REQUESTS_CA_BUNDLE", ca_pem_file)
ctx = find_ca_bundle() ctx = find_ca_bundle()
assert ctx == ca_pem_file
assert isinstance(ctx, ssl.SSLContext)