DataOps-platform/tests/integration/test_phase3_wp04_nginx_mtls.py

"""Actual Nginx mTLS handshake test using only temporary certificates."""

from __future__ import annotations

import ipaddress
import shutil
import socket
import subprocess
import time
import uuid
from datetime import UTC, datetime, timedelta
from pathlib import Path

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

ROOT = Path(__file__).resolve().parents[2]


def _certificate(
    common_name: str,
    *,
    issuer_certificate=None,
    issuer_key=None,
    client: bool = False,
):
    key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
    name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, common_name)])
    issuer_certificate = issuer_certificate or None
    issuer_key = issuer_key or key
    now = datetime.now(UTC)
    builder = (
        x509.CertificateBuilder()
        .subject_name(name)
        .issuer_name(issuer_certificate.subject if issuer_certificate else name)
        .public_key(key.public_key())
        .serial_number(x509.random_serial_number())
        .not_valid_before(now - timedelta(minutes=1))
        .not_valid_after(now + timedelta(minutes=10))
        .add_extension(x509.BasicConstraints(ca=issuer_certificate is None, path_length=None), critical=True)
    )
    if issuer_certificate is not None:
        usage = ExtendedKeyUsageOID.CLIENT_AUTH if client else ExtendedKeyUsageOID.SERVER_AUTH
        builder = builder.add_extension(x509.ExtendedKeyUsage([usage]), critical=False)
        if not client:
            builder = builder.add_extension(
                x509.SubjectAlternativeName(
                    [
                        x509.DNSName("localhost"),
                        x509.IPAddress(ipaddress.ip_address("127.0.0.1")),
                    ]
                ),
                critical=False,
            )
    return key, builder.sign(issuer_key, hashes.SHA256())


def _write_material(path: Path, name: str, key, certificate):
    key_path = path / f"{name}.key"
    certificate_path = path / f"{name}.crt"
    key_path.write_bytes(
        key.private_bytes(
            serialization.Encoding.PEM,
            serialization.PrivateFormat.PKCS8,
            serialization.NoEncryption(),
        )
    )
    certificate_path.write_bytes(certificate.public_bytes(serialization.Encoding.PEM))
    return certificate_path, key_path


@pytest.mark.integration
def test_actual_nginx_rejects_missing_wrong_and_forged_client_certificate(tmp_path):
    if shutil.which("docker") is None or shutil.which("curl") is None:
        pytest.skip("Docker and curl are required for the actual Nginx mTLS test")
    image = subprocess.run(
        ["docker", "image", "inspect", "nginx:alpine"],
        capture_output=True,
        check=False,
    )
    if image.returncode != 0:
        pytest.skip("the local nginx:alpine image is required; this test never pulls")

    ca_key, ca_certificate = _certificate("DataOps test client CA")
    server_key, server_certificate = _certificate(
        "localhost", issuer_certificate=ca_certificate, issuer_key=ca_key
    )
    client_key, client_certificate = _certificate(
        "gateway-valid",
        issuer_certificate=ca_certificate,
        issuer_key=ca_key,
        client=True,
    )
    revoked_key, revoked_certificate = _certificate(
        "gateway-revoked",
        issuer_certificate=ca_certificate,
        issuer_key=ca_key,
        client=True,
    )
    wrong_ca_key, wrong_ca_certificate = _certificate("Wrong client CA")
    wrong_key, wrong_certificate = _certificate(
        "gateway-wrong",
        issuer_certificate=wrong_ca_certificate,
        issuer_key=wrong_ca_key,
        client=True,
    )
    ca_path, _ = _write_material(tmp_path, "client-ca", ca_key, ca_certificate)
    server_path, server_key_path = _write_material(
        tmp_path, "server", server_key, server_certificate
    )
    client_path, client_key_path = _write_material(
        tmp_path, "client", client_key, client_certificate
    )
    wrong_path, wrong_key_path = _write_material(
        tmp_path, "wrong", wrong_key, wrong_certificate
    )
    revoked_path, revoked_key_path = _write_material(
        tmp_path, "revoked", revoked_key, revoked_certificate
    )
    now = datetime.now(UTC)
    revoked_entry = (
        x509.RevokedCertificateBuilder()
        .serial_number(revoked_certificate.serial_number)
        .revocation_date(now - timedelta(seconds=30))
        .build()
    )
    crl = (
        x509.CertificateRevocationListBuilder()
        .issuer_name(ca_certificate.subject)
        .last_update(now - timedelta(minutes=1))
        .next_update(now + timedelta(minutes=10))
        .add_revoked_certificate(revoked_entry)
        .sign(private_key=ca_key, algorithm=hashes.SHA256())
    )

    production = (ROOT / "deploy/docker/edge-mtls-nginx.conf").read_text()
    production = production.replace("http://backend:5500", "http://127.0.0.1:8080")
    upstream = """
server {
    listen 8080;
    location / {
        add_header X-Seen-Verify $http_x_dataops_edge_client_verify always;
        add_header X-Seen-Cert $http_x_dataops_edge_client_cert always;
        return 204;
    }
}
"""
    (tmp_path / "nginx.conf").write_text(production + upstream)
    (tmp_path / "server.crt").write_bytes(server_path.read_bytes())
    (tmp_path / "server.key").write_bytes(server_key_path.read_bytes())
    (tmp_path / "client-ca.pem").write_bytes(ca_path.read_bytes())
    (tmp_path / "client-ca.crl").write_bytes(
        crl.public_bytes(serialization.Encoding.PEM)
    )

    valid_crl = (tmp_path / "client-ca.crl").read_bytes()
    for invalid_crl in (None, b"not-a-crl"):
        if invalid_crl is None:
            (tmp_path / "client-ca.crl").unlink()
        else:
            (tmp_path / "client-ca.crl").write_bytes(invalid_crl)
        invalid = subprocess.run(
            [
                "docker", "run", "--rm",
                "--volume", f"{tmp_path / 'nginx.conf'}:/etc/nginx/conf.d/default.conf:ro",
                "--volume", f"{tmp_path}:/etc/nginx/edge-mtls:ro",
                "nginx:alpine", "nginx", "-t",
            ],
            capture_output=True,
            check=False,
        )
        assert invalid.returncode != 0
        (tmp_path / "client-ca.crl").write_bytes(valid_crl)

    expired_crl = (
        x509.CertificateRevocationListBuilder()
        .issuer_name(ca_certificate.subject)
        .last_update(now - timedelta(minutes=10))
        .next_update(now - timedelta(minutes=1))
        .sign(private_key=ca_key, algorithm=hashes.SHA256())
    )
    (tmp_path / "client-ca.crl").write_bytes(
        expired_crl.public_bytes(serialization.Encoding.PEM)
    )
    with socket.socket() as expired_probe:
        expired_probe.bind(("127.0.0.1", 0))
        expired_port = expired_probe.getsockname()[1]
    expired_name = f"dataops-edge-expired-crl-{uuid.uuid4().hex[:10]}"
    subprocess.run(
        [
            "docker", "run", "--detach", "--rm", "--name", expired_name,
            "--publish", f"127.0.0.1:{expired_port}:8443",
            "--volume", f"{tmp_path / 'nginx.conf'}:/etc/nginx/conf.d/default.conf:ro",
            "--volume", f"{tmp_path}:/etc/nginx/edge-mtls:ro",
            "nginx:alpine",
        ],
        check=True,
        capture_output=True,
    )
    try:
        time.sleep(0.2)
        expired = subprocess.run(
            [
                "curl", "--silent", "--show-error", "--max-time", "3",
                "--cacert", str(ca_path), "--cert", str(client_path),
                "--key", str(client_key_path), "--output", "/dev/null",
                "--write-out", "%{http_code}",
                f"https://127.0.0.1:{expired_port}/api/datasource/edge/test",
            ],
            capture_output=True,
            text=True,
            check=False,
        )
        assert expired.returncode != 0 or expired.stdout in {"400", "495", "496"}
    finally:
        subprocess.run(
            ["docker", "stop", expired_name], check=False, capture_output=True
        )
    (tmp_path / "client-ca.crl").write_bytes(valid_crl)

    with socket.socket() as probe:
        probe.bind(("127.0.0.1", 0))
        port = probe.getsockname()[1]
    name = f"dataops-edge-mtls-{uuid.uuid4().hex[:12]}"
    subprocess.run(
        [
            "docker", "run", "--detach", "--rm", "--name", name,
            "--publish", f"127.0.0.1:{port}:8443",
            "--volume", f"{tmp_path / 'nginx.conf'}:/etc/nginx/conf.d/default.conf:ro",
            "--volume", f"{tmp_path}:/etc/nginx/edge-mtls:ro",
            "nginx:alpine",
        ],
        check=True,
        capture_output=True,
    )
    try:
        for _ in range(40):
            try:
                with socket.create_connection(("127.0.0.1", port), timeout=0.2):
                    break
            except OSError:
                time.sleep(0.05)
        else:
            raise AssertionError("temporary Nginx did not start")

        endpoint = f"https://127.0.0.1:{port}/api/datasource/edge/test"
        for certificate_path, key_path in (
            (None, None),
            (wrong_path, wrong_key_path),
            (revoked_path, revoked_key_path),
        ):
            command = [
                "curl", "--silent", "--show-error", "--max-time", "3",
                "--cacert", str(ca_path), "--output", "/dev/null",
                "--write-out", "%{http_code}", endpoint,
            ]
            if certificate_path is not None:
                command[1:1] = [
                    "--cert", str(certificate_path), "--key", str(key_path)
                ]
            rejected = subprocess.run(command, capture_output=True, text=True, check=False)
            assert rejected.returncode != 0 or rejected.stdout in {"400", "495", "496"}

        valid = subprocess.run(
            [
                "curl", "--silent", "--show-error", "--max-time", "3",
                "--cacert", str(ca_path), "--cert", str(client_path),
                "--key", str(client_key_path), "--dump-header", "-",
                "--output", "/dev/null",
                "--header", "X-DataOps-Edge-Client-Verify: FORGED",
                "--header", "X-DataOps-Edge-Client-Cert: forged",
                endpoint,
            ],
            capture_output=True,
            text=True,
            check=False,
        )
        assert valid.returncode == 0, valid.stderr
        assert "HTTP/1.1 204" in valid.stdout
        assert "X-Seen-Verify: SUCCESS" in valid.stdout
        assert "X-Seen-Cert: -----BEGIN%20CERTIFICATE-----" in valid.stdout
        assert "X-Seen-Cert: forged" not in valid.stdout
    finally:
        subprocess.run(
            ["docker", "stop", name],
            check=False,
            capture_output=True,
        )