ss-tools/backend/tests/integration/fixtures/pki.py

# #region Test.Pki.PKIFixtures [C:3] [TYPE Module] [SEMANTICS test,fixtures,pki,tls,certificates]
# @BRIEF Session-scoped PKI certificate generation and isolated trust-store installation.
# @RELATION DEPENDS_ON -> [EXT:cryptography]
# @POST 3-tier PKI (Root CA → Intermediate CA → Server cert) generated once per session.
# @SIDE_EFFECT Generates RSA 2048-bit keys in memory.
# @RATIONALE
#   PKI generation is computationally expensive (RSA key generation).
#   Session-scoped to avoid regenerating keys for every test.
#   Uses OpenSSL-compatible hash-symlink capath strategy for CA trust,
#   avoiding system-wide update-ca-certificates mutations.
# @REJECTED
#   System-wide update-ca-certificates rejected — requires root, mutates host state,
#   and may not be reversible cleanly in CI environments.
#   certifi-only trust rejected — cannot trust custom CAs without patching.
import datetime
import os
from pathlib import Path
import pytest
import shutil
import subprocess
import tempfile


# #region Test.Pki.CaChain [C:2] [TYPE Fixture]
# @BRIEF Session-scoped — generates 3-tier PKI (Root CA → Intermediate CA → Server cert).
# @POST Returns dict with root_crt, intermediate_crt, server_crt, server_key,
#   server_key_encrypted, server_key_passphrase, fullchain PEM strings.
@pytest.fixture(scope="session")
def ca_chain():
    """Generate 3-tier PKI: Root CA → Intermediate CA → Server certificate (SAN: localhost)."""
    import ipaddress

    from cryptography import x509
    from cryptography.hazmat.primitives import hashes, serialization
    from cryptography.hazmat.primitives.asymmetric import rsa
    from cryptography.x509.oid import NameOID

    def _gen_key():
        return rsa.generate_private_key(public_exponent=65537, key_size=2048)

    def _gen_cert_pem(
        subject_name, issuer_name, subject_key, issuer_key,
        is_ca=False, san_dns=None, san_ip=None,
    ):
        now = datetime.datetime.now(datetime.UTC)
        builder = x509.CertificateBuilder()
        builder = builder.subject_name(
            x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, subject_name)])
        )
        builder = builder.issuer_name(
            x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, issuer_name)])
        )
        builder = builder.not_valid_before(now - datetime.timedelta(hours=1))
        builder = builder.not_valid_after(now + datetime.timedelta(days=365))
        builder = builder.serial_number(x509.random_serial_number())
        builder = builder.public_key(subject_key.public_key())

        if is_ca:
            builder = builder.add_extension(
                x509.BasicConstraints(ca=True, path_length=None),
                critical=True,
            )
            builder = builder.add_extension(
                x509.KeyUsage(
                    key_cert_sign=True,
                    crl_sign=True,
                    digital_signature=False,
                    content_commitment=False,
                    key_encipherment=False,
                    data_encipherment=False,
                    key_agreement=False,
                    encipher_only=False,
                    decipher_only=False,
                ),
                critical=True,
            )
        else:
            builder = builder.add_extension(
                x509.BasicConstraints(ca=False, path_length=None),
                critical=True,
            )
            sans: list[x509.GeneralName] = []
            if san_dns:
                sans.append(x509.DNSName(san_dns))
            if san_ip:
                sans.append(x509.IPAddress(san_ip))
            builder = builder.add_extension(
                x509.SubjectAlternativeName(sans),
                critical=False,
            )

        # Subject Key Identifier
        builder = builder.add_extension(
            x509.SubjectKeyIdentifier.from_public_key(subject_key.public_key()),
            critical=False,
        )
        # Authority Key Identifier
        if subject_name != issuer_name:
            builder = builder.add_extension(
                x509.AuthorityKeyIdentifier.from_issuer_public_key(issuer_key.public_key()),
                critical=False,
            )

        cert = builder.sign(issuer_key, hashes.SHA256())
        return cert.public_bytes(serialization.Encoding.PEM).decode()

    root_key = _gen_key()
    root_crt = _gen_cert_pem("Test Root CA", "Test Root CA", root_key, root_key, is_ca=True)

    intermediate_key = _gen_key()
    intermediate_crt = _gen_cert_pem(
        "Test Intermediate CA",
        "Test Root CA",
        intermediate_key,
        root_key,
        is_ca=True,
    )

    server_key = _gen_key()
    server_crt = _gen_cert_pem(
        "localhost",
        "Test Intermediate CA",
        server_key,
        intermediate_key,
        is_ca=False,
        san_dns="localhost",
        san_ip=ipaddress.IPv4Address("127.0.0.1"),
    )

    fullchain = server_crt + "\n" + intermediate_crt
    server_key_passphrase = "test-passphrase-12345"
    server_key_encrypted = server_key.private_bytes(
        encoding=serialization.Encoding.PEM,
        format=serialization.PrivateFormat.TraditionalOpenSSL,
        encryption_algorithm=serialization.BestAvailableEncryption(
            server_key_passphrase.encode()
        ),
    ).decode()

    return {
        "root_crt": root_crt,
        "root_key": root_key.private_bytes(
            encoding=serialization.Encoding.PEM,
            format=serialization.PrivateFormat.TraditionalOpenSSL,
            encryption_algorithm=serialization.NoEncryption(),
        ).decode(),
        "intermediate_crt": intermediate_crt,
        "intermediate_key": intermediate_key.private_bytes(
            encoding=serialization.Encoding.PEM,
            format=serialization.PrivateFormat.TraditionalOpenSSL,
            encryption_algorithm=serialization.NoEncryption(),
        ).decode(),
        "server_crt": server_crt,
        "server_key": server_key.private_bytes(
            encoding=serialization.Encoding.PEM,
            format=serialization.PrivateFormat.TraditionalOpenSSL,
            encryption_algorithm=serialization.NoEncryption(),
        ).decode(),
        "server_key_encrypted": server_key_encrypted,
        "server_key_passphrase": server_key_passphrase,
        "fullchain": fullchain,
    }


# #endregion Test.Pki.CaChain


# #region Test.Pki.InstallCustomCa [C:2] [TYPE Fixture]
# @BRIEF Function-scoped — installs Test Root CA into an isolated OpenSSL capath via SSL_CERT_DIR.
# @POST Root CA accessible via ssl.create_default_context(capath=...).
# @SIDE_EFFECT Creates temp directory with OpenSSL hash symlinks; sets SSL_CERT_DIR env var.
# @RATIONALE Uses only SSL_CERT_DIR with a temporary directory containing OpenSSL hash
#   symlinks. This avoids mutating the host's system trust store (/usr/local/share/ca-certificates/)
#   and running update-ca-certificates (which requires root). Cleanup is deterministic:
#   env var removed and temp directory deleted.
# @REJECTED System-wide update-ca-certificates rejected — requires root, mutates host state.
#   certifi patching rejected — would break non-test code paths.
@pytest.fixture
def install_custom_ca(ca_chain):
    """Install the custom Root CA into an isolated OpenSSL capath via SSL_CERT_DIR.

    Strategy: Create a temporary directory, write the Root CA PEM, compute the
    OpenSSL hash, create a hash symlink, and set SSL_CERT_DIR to point to it.

    This is the ONLY supported strategy — no system-wide mutations.
    """
    ca_pem = ca_chain["root_crt"]
    cert_dir = Path(tempfile.mkdtemp(prefix="test_ca_ssl_"))
    ca_verify_path = None

    try:
        ca_path = cert_dir / "test_ss_tools_ca.pem"
        ca_path.write_text(ca_pem)

        # Create hash symlink for OpenSSL capath
        result = subprocess.run(
            ["openssl", "x509", "-hash", "-noout"],
            input=ca_pem,
            capture_output=True,
            text=True,
            timeout=10,
        )
        cert_hash = result.stdout.strip()
        symlink_path = cert_dir / f"{cert_hash}.0"
        if symlink_path.exists():
            symlink_path.unlink()
        symlink_path.symlink_to(ca_path.name)

        os.environ["SSL_CERT_DIR"] = str(cert_dir)
        ca_verify_path = str(cert_dir)

        yield {
            "ca_verify_path": ca_verify_path,
            "used_update_ca": False,
            "ca_pem": ca_pem,
        }
    finally:
        if "SSL_CERT_DIR" in os.environ:
            del os.environ["SSL_CERT_DIR"]
        if cert_dir and cert_dir.exists():
            shutil.rmtree(str(cert_dir), ignore_errors=True)


# #endregion Test.Pki.InstallCustomCa
# #endregion Test.Pki.PKIFixtures