headroom/tests/test_ssl_context.py

"""Unit tests for headroom.proxy.ssl_context.find_ca_bundle.

Covers:
- Returns None when no env var is set
- Returns a path string 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 NODE_EXTRA_CA_CERTS points to a valid PEM file
- The SSLContext is additive: default/system roots are preserved (#998)
- 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)
"""

from __future__ import annotations

import os
import ssl

import pytest

from headroom.proxy.ssl_context import find_ca_bundle

# Minimal self-signed CA certificate (PEM) used only to verify that
# load_verify_locations accepts the file.  Generated offline; never used
# for real TLS handshakes in these tests.
_SELF_SIGNED_CA_PEM = b"""\
-----BEGIN CERTIFICATE-----
MIIDFzCCAf+gAwIBAgIUWP49K8QzU5B68/BZSmeqPCDaBoQwDQYJKoZIhvcNAQEL
BQAwGzEZMBcGA1UEAwwQaGVhZHJvb20tdGVzdC1jYTAeFw0yNjA2MDgxNDIwMzFa
Fw0zNjA2MDUxNDIwMzFaMBsxGTAXBgNVBAMMEGhlYWRyb29tLXRlc3QtY2EwggEi
MA0GCSqGSIb3DQEBAQUAA4IBDwAwggEKAoIBAQCvTqYZXAhet9yw1n4cFeC8HosC
1Od/bibXyW7ko7aOuuzUT7B9l7MwDfgrE2mjHecoSe2qbknFcv6hxbYojh4J7C8r
UPgCA2QTtU3pBxQdwO156YAOmFPuBFPb19NAErOVlnHCU+NXCVSsE5y+AJjM161S
W0HnZgO8OADZHBs5jSAGDE3ymMw+8xpuvRKJnuvK0Tcu6bOqOTMbnggwmPBZBBLW
PrurPTN0vV9C2oyHA1tXgEJyYtEPoMfaqyE80GxYeUujt9EQWrLp+3k8ufB/yJ1b
DaSrH0GZYx2HUn0p1mqWzXcKZrSrL1o+38gCmCivG0movXt6z1tUly8mTGz/AgMB
AAGjUzBRMB0GA1UdDgQWBBTyJ8OWE/bpWbKM3SB52P+9DhGN/TAfBgNVHSMEGDAW
gBTyJ8OWE/bpWbKM3SB52P+9DhGN/TAPBgNVHRMBAf8EBTADAQH/MA0GCSqGSIb3
DQEBCwUAA4IBAQAb44h2gg9wWU5todvwSXVAlBb/WZD1l/NG2PeTsGoH7xqmfgq9
DxV6tvoIuDlu6OKz071ljSqRh0Mesh1ma1cj6snsc/jqgsakSlcOpOCsrTCvw2DB
2oTztHnO4PiZAPtuKiawhVQpJfEna9/xOkbalazecSGngtSzd/oIJEXe299hE1/1
Tfx2hBGZ0UogmREaXFi099rmaueZ0HIBn51b3kYqc7of5TI0fHwSHF4GdXXs2OZi
6EVQWhKx5nQbklTYP5/ge9olEIsMdGqJEiz7WfSC6QBBgvoYyH596GiSGRZcX67p
kF9agIt8Q8t/2kviMn2roInGTwTyPYOEQV0m
-----END CERTIFICATE-----
"""


@pytest.fixture()
def ca_pem_file(tmp_path):
    """Write the self-signed CA PEM to a temp file and return its path."""
    p = tmp_path / "ca.pem"
    p.write_bytes(_SELF_SIGNED_CA_PEM)
    return str(p)


def _clean_env(monkeypatch):
    """Remove all three CA-bundle env vars so tests start from a clean state."""
    for var in ("SSL_CERT_FILE", "REQUESTS_CA_BUNDLE", "NODE_EXTRA_CA_CERTS"):
        monkeypatch.delenv(var, raising=False)


class TestFindCaBundleNoEnvVars:
    def test_returns_none_when_no_env_var_set(self, monkeypatch):
        _clean_env(monkeypatch)
        assert find_ca_bundle() is None


class TestFindCaBundleWithValidPem:
    def test_ssl_cert_file_returns_path(self, monkeypatch, ca_pem_file):
        _clean_env(monkeypatch)
        monkeypatch.setenv("SSL_CERT_FILE", ca_pem_file)
        ctx = find_ca_bundle()
        assert isinstance(ctx, str)
        assert os.path.isfile(ctx)

    def test_requests_ca_bundle_returns_path(self, monkeypatch, ca_pem_file):
        _clean_env(monkeypatch)
        monkeypatch.setenv("REQUESTS_CA_BUNDLE", ca_pem_file)
        ctx = find_ca_bundle()
        assert isinstance(ctx, str)
        assert os.path.isfile(ctx)

    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)."""
        _clean_env(monkeypatch)
        monkeypatch.setenv("NODE_EXTRA_CA_CERTS", ca_pem_file)
        ctx = find_ca_bundle()
        assert isinstance(ctx, ssl.SSLContext)

    def test_node_extra_ca_certs_is_additive(self, monkeypatch, ca_pem_file):
        """The SSLContext must contain default/system roots plus the extra cert (#998)."""
        _clean_env(monkeypatch)
        monkeypatch.setenv("NODE_EXTRA_CA_CERTS", ca_pem_file)
        ctx = find_ca_bundle()
        assert isinstance(ctx, ssl.SSLContext)
        stats = ctx.cert_store_stats()
        # The default trust store has dozens of CAs; if only the test cert
        # were loaded (replacement), x509_ca would be 1.
        assert stats["x509_ca"] > 1


class TestFindCaBundlePriority:
    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."""
        _clean_env(monkeypatch)
        # Two distinct files so we can identify which was loaded.
        pem1 = tmp_path / "first.pem"
        pem2 = tmp_path / "second.pem"
        pem1.write_bytes(_SELF_SIGNED_CA_PEM)
        pem2.write_bytes(_SELF_SIGNED_CA_PEM)

        monkeypatch.setenv("SSL_CERT_FILE", str(pem1))
        monkeypatch.setenv("REQUESTS_CA_BUNDLE", str(pem2))

        # Both files are valid; we cannot easily inspect which CA was loaded
        # into the context, but we can verify the function returns a path
        # (not None) and that it is tied to SSL_CERT_FILE by temporarily
        # making REQUESTS_CA_BUNDLE point to a nonexistent path.
        monkeypatch.setenv("REQUESTS_CA_BUNDLE", "/nonexistent/path.pem")
        ctx = find_ca_bundle()
        # SSL_CERT_FILE still valid → should return a path
        assert isinstance(ctx, str)
        assert os.path.isfile(ctx)

    def test_ssl_cert_file_beats_node_extra_ca_certs(self, monkeypatch, tmp_path):
        """SSL_CERT_FILE takes precedence over NODE_EXTRA_CA_CERTS."""
        _clean_env(monkeypatch)
        pem = tmp_path / "ca.pem"
        pem.write_bytes(_SELF_SIGNED_CA_PEM)

        monkeypatch.setenv("SSL_CERT_FILE", str(pem))
        monkeypatch.setenv("NODE_EXTRA_CA_CERTS", "/nonexistent/node.pem")

        ctx = find_ca_bundle()
        assert isinstance(ctx, str)
        assert os.path.isfile(ctx)

    def test_requests_ca_bundle_beats_node_extra_ca_certs(self, monkeypatch, tmp_path):
        """REQUESTS_CA_BUNDLE is used before NODE_EXTRA_CA_CERTS."""
        _clean_env(monkeypatch)
        pem = tmp_path / "ca.pem"
        pem.write_bytes(_SELF_SIGNED_CA_PEM)

        monkeypatch.setenv("SSL_CERT_FILE", "/nonexistent/ssl.pem")
        monkeypatch.setenv("REQUESTS_CA_BUNDLE", str(pem))
        monkeypatch.setenv("NODE_EXTRA_CA_CERTS", "/nonexistent/node.pem")

        ctx = find_ca_bundle()
        assert isinstance(ctx, str)
        assert os.path.isfile(ctx)


class TestFindCaBundleNonexistentPaths:
    def test_nonexistent_path_is_skipped(self, monkeypatch):
        _clean_env(monkeypatch)
        monkeypatch.setenv("SSL_CERT_FILE", "/nonexistent/path/ca.pem")
        assert find_ca_bundle() is None

    def test_all_nonexistent_returns_none(self, monkeypatch):
        _clean_env(monkeypatch)
        monkeypatch.setenv("SSL_CERT_FILE", "/no/such/file1.pem")
        monkeypatch.setenv("REQUESTS_CA_BUNDLE", "/no/such/file2.pem")
        monkeypatch.setenv("NODE_EXTRA_CA_CERTS", "/no/such/file3.pem")
        assert find_ca_bundle() is None

    def test_first_nonexistent_falls_through_to_valid(self, monkeypatch, ca_pem_file):
        """When the first env var path is missing, the next valid one is used."""
        _clean_env(monkeypatch)
        monkeypatch.setenv("SSL_CERT_FILE", "/nonexistent/ssl.pem")
        monkeypatch.setenv("REQUESTS_CA_BUNDLE", ca_pem_file)

        ctx = find_ca_bundle()
        assert ctx == ca_pem_file