DataOps-platform/tests/integration/test_phase3_wp04_edge_runtime.py

"""Real image/container smoke for the pull-only P3-WP04 edge runtime."""

from __future__ import annotations

import hashlib
import json
import os
import shutil
import sqlite3
import ssl
import subprocess
import threading
import time
from datetime import UTC, datetime, timedelta
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path

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

import app.edge_gateway.runtime as edge_runtime
from app.core.edge_gateway.contracts import (
    EdgeTaskContract,
    SignedTaskEnvelope,
    canonical_sha256,
    canonical_timestamp,
)
from app.core.edge_gateway.policy import EdgePolicyError
from app.edge_gateway.bootstrap import (
    build_server_crl_ssl_context,
    validate_server_crl,
)

ROOT = Path(__file__).resolve().parents[2]
IMAGE = "dataops-edge:wp04-runtime-smoke"


def test_runtime_runner_executes_closed_engineering_self_check_only():
    deadline = canonical_timestamp(
        (datetime.now(UTC) + timedelta(minutes=2)).isoformat().replace("+00:00", "Z")
    )
    task = EdgeTaskContract.from_mapping({
        "task_id": "runtime-self-check-1",
        "gateway_id": "11111111-1111-4111-8111-111111111111",
        "environment": "production",
        "network_zone": "zone-a",
        "purpose": "quality-evaluation",
        "classification": "evidence",
        "task_type": "quality",
        "contract_version": 1,
        "deadline_at": deadline,
        "attempt": 1,
        "idempotency_key": "runtime-self-check-1",
        "policy_digest": "b" * 64,
    })
    result = edge_runtime.build_runtime_runner().execute(task, lambda: False)
    assert result == {
        "classification": "evidence",
        "payload": {
            "check_count": 1,
            "passed_count": 1,
            "scope": "edge_runtime",
            "status": "passed",
        },
        "local_artifact_digest": None,
        "local_artifact_ref": None,
    }
    with pytest.raises(EdgePolicyError):
        edge_runtime.build_runtime_runner().execute(
            EdgeTaskContract.from_mapping({
                **task.to_mapping(),
                "task_type": "collect",
                "purpose": "governed-inventory",
                "classification": "desensitized_metadata",
            }),
            lambda: False,
        )


def test_runtime_session_rebuilds_tls_context_when_server_crl_changes(tmp_path):
    _fingerprint, _context, ca_key, ca, _server = _materials(tmp_path)
    ca_path = tmp_path / "certificates" / "ca.pem"
    crl_path = tmp_path / "certificates" / "control-plane-server.crl.pem"
    session = edge_runtime.ServerCrlSession(str(ca_path), str(crl_path))
    initial = session.server_crl_digest
    now = datetime.now(UTC)
    replacement = (
        x509.CertificateRevocationListBuilder().issuer_name(ca.subject)
        .last_update(now - timedelta(seconds=1))
        .next_update(now + timedelta(hours=1))
        .sign(ca_key, hashes.SHA256())
    )
    _write(crl_path, replacement.public_bytes(serialization.Encoding.PEM), 0o644)
    assert session.refresh_server_crl() is True
    assert session.server_crl_digest != initial
    assert session.refresh_server_crl() is False


def _write(path: Path, value: bytes, mode: int = 0o600) -> None:
    path.write_bytes(value)
    path.chmod(mode)


def _materials(root: Path):
    now = datetime.now(UTC)
    ca_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
    ca_name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "wp04-runtime-ca")])
    ca = (
        x509.CertificateBuilder()
        .subject_name(ca_name).issuer_name(ca_name).public_key(ca_key.public_key())
        .serial_number(x509.random_serial_number())
        .not_valid_before(now - timedelta(minutes=1))
        .not_valid_after(now + timedelta(days=7))
        .add_extension(x509.BasicConstraints(ca=True, path_length=0), critical=True)
        .sign(ca_key, hashes.SHA256())
    )

    def issue(name: str, *, server: bool):
        key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
        builder = (
            x509.CertificateBuilder()
            .subject_name(x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, name)]))
            .issuer_name(ca.subject).public_key(key.public_key())
            .serial_number(x509.random_serial_number())
            .not_valid_before(now - timedelta(minutes=1))
            .not_valid_after(now + timedelta(days=1))
            .add_extension(
                x509.ExtendedKeyUsage([
                    ExtendedKeyUsageOID.SERVER_AUTH if server
                    else ExtendedKeyUsageOID.CLIENT_AUTH
                ]), critical=True,
            )
        )
        if server:
            builder = builder.add_extension(
                x509.SubjectAlternativeName([
                    x509.DNSName("host.docker.internal"), x509.DNSName("localhost")
                ]), critical=False,
            )
        return key, builder.sign(ca_key, hashes.SHA256())

    server_key, server = issue("host.docker.internal", server=True)
    client_key, client = issue("wp04-edge-client", server=False)
    crl = (
        x509.CertificateRevocationListBuilder().issuer_name(ca.subject)
        .last_update(now - timedelta(minutes=1)).next_update(now + timedelta(hours=2))
        .sign(ca_key, hashes.SHA256())
    )
    certificates = root / "certificates"
    secrets = root / "secrets"
    certificates.mkdir()
    secrets.mkdir()
    _write(certificates / "ca.pem", ca.public_bytes(serialization.Encoding.PEM), 0o644)
    _write(certificates / "client.pem", client.public_bytes(serialization.Encoding.PEM), 0o644)
    _write(
        certificates / "control-plane-server.crl.pem",
        crl.public_bytes(serialization.Encoding.PEM), 0o644,
    )
    _write(
        secrets / "client.key",
        client_key.private_bytes(
            serialization.Encoding.PEM, serialization.PrivateFormat.PKCS8,
            serialization.NoEncryption(),
        ),
    )
    _write(
        root / "server.pem", server.public_bytes(serialization.Encoding.PEM), 0o644
    )
    _write(
        root / "server.key",
        server_key.private_bytes(
            serialization.Encoding.PEM, serialization.PrivateFormat.PKCS8,
            serialization.NoEncryption(),
        ),
    )
    context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
    context.load_cert_chain(str(root / "server.pem"), str(root / "server.key"))
    context.load_verify_locations(str(certificates / "ca.pem"))
    context.verify_mode = ssl.CERT_REQUIRED
    return client.fingerprint(hashes.SHA256()).hex(), context, ca_key, ca, server


class _ControlHandler(BaseHTTPRequestHandler):
    paths: list[str] = []
    accepted_events: list[dict[str, object]] = []
    task_private_key: ed25519.Ed25519PrivateKey | None = None
    task_sent = False

    @classmethod
    def reset(cls, private_key: ed25519.Ed25519PrivateKey) -> None:
        cls.paths = []
        cls.accepted_events = []
        cls.task_private_key = private_key
        cls.task_sent = False

    def do_POST(self):  # noqa: N802
        length = int(self.headers.get("Content-Length", "0"))
        self.paths.append(self.path)
        request_body = json.loads(self.rfile.read(length) or b"{}")
        if self.path.endswith("/reconcile"):
            data = {
                "cancelled_task_ids": [], "cancel_next_cursor": None,
                "release_offers": [], "release_next_cursor": None,
                "release_baseline": None,
            }
        elif self.path.endswith("/tasks/pull"):
            if self.task_sent:
                data = {"task": None}
            else:
                now = datetime.now(UTC)
                task = {
                    "task_id": "runtime-self-check-1",
                    "gateway_id": "11111111-1111-4111-8111-111111111111",
                    "environment": "production",
                    "network_zone": "zone-a",
                    "purpose": "quality-evaluation",
                    "classification": "evidence",
                    "task_type": "quality",
                    "contract_version": 1,
                    "deadline_at": canonical_timestamp(
                        (now + timedelta(minutes=2)).isoformat().replace("+00:00", "Z")
                    ),
                    "attempt": 1,
                    "idempotency_key": "runtime-self-check-1",
                    "policy_digest": "b" * 64,
                }
                unsigned = {
                    "task": task,
                    "authority_key_id": "task-key-1",
                    "signature_algorithm": "Ed25519",
                    "contract_digest": canonical_sha256(task),
                    "gateway_id": task["gateway_id"],
                    "environment": task["environment"],
                    "network_zone": task["network_zone"],
                    "policy_digest": task["policy_digest"],
                    "purpose": task["purpose"],
                    "issued_at": canonical_timestamp(
                        (now - timedelta(seconds=1)).isoformat().replace("+00:00", "Z")
                    ),
                    "expires_at": task["deadline_at"],
                }
                assert self.task_private_key is not None
                envelope = {
                    **unsigned,
                    "signature": self.task_private_key.sign(
                        SignedTaskEnvelope.canonical_unsigned_bytes(unsigned)
                    ).hex(),
                }
                data = {
                    "task": task,
                    "signed_task_envelope": envelope,
                    "lease_token": "dopl_runtime-smoke-lease",
                    "lease_expires_at": canonical_timestamp(
                        (now + timedelta(seconds=90)).isoformat().replace("+00:00", "Z")
                    ),
                }
                self.task_sent = True
        elif self.path.endswith("/events"):
            event = request_body["event"]
            lease_token = request_body["lease_token"]
            self.accepted_events.append(event)
            data = {
                "event_id": event["event_id"],
                "event_digest": canonical_sha256(event),
                "remote_lease_digest": hashlib.sha256(
                    lease_token.encode("utf-8")
                ).hexdigest(),
                "received_at": canonical_timestamp(
                    datetime.now(UTC).isoformat().replace("+00:00", "Z")
                ),
                "status": "accepted",
            }
        else:
            data = {"status": "active"}
        encoded = json.dumps({"code": 200, "message": "ok", "data": data}).encode()
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(encoded)))
        self.end_headers()
        self.wfile.write(encoded)

    def log_message(self, _format, *_args):
        return


def _wait_for(path: Path, status: str, timeout: float = 30) -> dict:
    deadline = time.monotonic() + timeout
    while time.monotonic() < deadline:
        try:
            value = json.loads(path.read_text())
            if value.get("status") == status:
                return value
        except (OSError, json.JSONDecodeError):
            pass
        time.sleep(0.25)
    raise AssertionError(f"health did not reach {status}")


@pytest.mark.skipif(shutil.which("docker") is None, reason="Docker is required")
def test_real_edge_image_mtls_pull_degraded_and_clean_sigterm(tmp_path):
    fingerprint, ssl_context, ca_key, ca, server_certificate = _materials(tmp_path)
    task_private_key = ed25519.Ed25519PrivateKey.generate()
    _ControlHandler.reset(task_private_key)
    server = ThreadingHTTPServer(("0.0.0.0", 0), _ControlHandler)
    server.socket = ssl_context.wrap_socket(server.socket, server_side=True)
    thread = threading.Thread(target=server.serve_forever, daemon=True)
    thread.start()
    port = server.server_address[1]
    for name in ("config", "queue", "artifacts", "releases", "signing", "health"):
        (tmp_path / name).mkdir()
    credential = "dopg_runtime_smoke_credential"
    _write(tmp_path / "secrets" / "control-credential", credential.encode())
    signing_key = task_private_key.public_key().public_bytes(
        serialization.Encoding.Raw, serialization.PublicFormat.Raw
    ).hex()
    _write(
        tmp_path / "signing" / "task-public-keys.json",
        json.dumps({"task-key-1": signing_key}).encode(), 0o644,
    )
    _write(
        tmp_path / "signing" / "release-public-keys.json",
        json.dumps({"release-key-1": signing_key}).encode(), 0o644,
    )
    config = {
        "allowed_control_hosts": ["host.docker.internal"],
        "allowed_control_origins": [f"https://host.docker.internal:{port}"],
        "allowed_proxy_hosts": [],
        "allowed_proxy_origins": [],
        "artifact_root": "/var/lib/dataops-edge/artifacts",
        "ca_bundle_path": "/run/edge/certificates/ca.pem",
        "certificate_sha256": fingerprint,
        "client_certificate_path": "/run/edge/certificates/client.pem",
        "client_private_key_path": "/run/edge/secrets/client.key",
        "control_url": f"https://host.docker.internal:{port}",
        "environment": "production",
        "gateway_id": "11111111-1111-4111-8111-111111111111",
        "generation": 1, "network_zone": "zone-a",
        "policy_digest": "b" * 64, "proxy_url": None,
        "queue_path": "/var/lib/dataops-edge/queue/edge-queue-v2.sqlite3",
        "task_authority_clock_skew_seconds": 30, "version": "1.0.0",
    }
    _write(tmp_path / "config" / "edge.json", json.dumps(config).encode(), 0o644)
    build = subprocess.run(
        ["docker", "build", "-f", "deploy/edge/Dockerfile.edge", "-t", IMAGE, "."],
        cwd=ROOT, capture_output=True, text=True, check=False,
    )
    assert build.returncode == 0, build.stderr[-4000:]
    command = [
        "docker", "run", "-d", "--add-host", "host.docker.internal:host-gateway",
        "--user", f"{os.getuid()}:{os.getgid()}",
    ]
    for host, container, mode in (
        ("config", "/run/edge/config", "ro"), ("queue", "/var/lib/dataops-edge/queue", "rw"),
        ("artifacts", "/var/lib/dataops-edge/artifacts", "rw"),
        ("releases", "/var/lib/dataops-edge/releases", "rw"),
        ("secrets", "/run/edge/secrets", "ro"),
        ("certificates", "/run/edge/certificates", "ro"),
        ("signing", "/run/edge/signing", "ro"), ("health", "/run/edge/tmp", "rw"),
    ):
        command.extend(["-v", f"{tmp_path / host}:{container}:{mode}"])
    command.extend([
        "-e", "EDGE_CONFIG_PATH=/run/edge/config/edge.json",
        "-e", "EDGE_CREDENTIAL_FILE=/run/edge/secrets/control-credential",
        "-e", "EDGE_CLIENT_CERTIFICATE_PATH=/run/edge/certificates/client.pem",
        "-e", "EDGE_CLIENT_PRIVATE_KEY_PATH=/run/edge/secrets/client.key",
        "-e", "EDGE_CA_BUNDLE_PATH=/run/edge/certificates/ca.pem",
        "-e", "EDGE_SERVER_CRL_PATH=/run/edge/certificates/control-plane-server.crl.pem",
        "-e", "EDGE_TASK_SIGNING_KEYS_PATH=/run/edge/signing/task-public-keys.json",
        "-e", "EDGE_RELEASE_SIGNING_KEYS_PATH=/run/edge/signing/release-public-keys.json",
        "-e", "EDGE_HEALTH_PATH=/run/edge/tmp/health.json",
        "-e", "EDGE_POLL_INTERVAL_SECONDS=1", IMAGE,
    ])
    started = subprocess.run(command, capture_output=True, text=True, check=False)
    assert started.returncode == 0, started.stderr
    container_id = started.stdout.strip()
    health_path = tmp_path / "health" / "health.json"
    try:
        healthy = _wait_for(health_path, "healthy", 40)
        assert credential not in json.dumps(healthy)
        assert any(path.endswith("/heartbeat") for path in _ControlHandler.paths)
        assert any(path.endswith("/reconcile") for path in _ControlHandler.paths)
        assert any(path.endswith("/tasks/pull") for path in _ControlHandler.paths)
        deadline = time.monotonic() + 15
        while not _ControlHandler.accepted_events and time.monotonic() < deadline:
            time.sleep(0.25)
        if len(_ControlHandler.accepted_events) != 1:
            queue_path = tmp_path / "queue" / "edge-queue-v2.sqlite3"
            with sqlite3.connect(queue_path) as connection:
                event_rows = connection.execute(
                    "SELECT status,attempt_count,error_code,available_at,lease_expires_at,event_json "
                    "FROM edge_outbound_events"
                ).fetchall()
            diagnostic = subprocess.run(
                ["docker", "logs", container_id],
                capture_output=True,
                text=True,
                check=False,
            )
            probe = subprocess.run(
                [
                    "docker", "exec", container_id, "python", "-c",
                    "import json,sqlite3;"
                    "from app.edge_gateway.runtime import build_runtime_from_environment;"
                    "a,_,_=build_runtime_from_environment();"
                    "c=sqlite3.connect(a.queue.db_path);"
                    "e=json.loads(c.execute('select event_json from edge_outbound_events limit 1').fetchone()[0]);"
                    "t=a.queue.recover_remote_lease('runtime-self-check-1');"
                    "print('lease',len(t),__import__('hashlib').sha256(t.encode()).hexdigest() == __import__('hashlib').sha256(b'dopl_runtime-smoke-lease').hexdigest());"
                    "\ntry: a.transport.send_event(e,t)"
                    "\nexcept Exception as x: print(type(x).__name__,str(x),type(x.__cause__).__name__ if x.__cause__ else '',repr(x.__cause__))",
                ],
                capture_output=True,
                text=True,
                check=False,
            )
            raise AssertionError(
                f"event chain incomplete paths={_ControlHandler.paths!r} "
                f"health={health_path.read_text()!r} events={event_rows!r} logs="
                f"{(diagnostic.stdout + diagnostic.stderr)[-4000:]!r} "
                f"probe={(probe.stdout + probe.stderr)[-4000:]!r}"
            )
        assert _ControlHandler.accepted_events[0]["classification"] == "evidence"
        assert _ControlHandler.accepted_events[0]["payload"] == {
            "check_count": 1,
            "passed_count": 1,
            "scope": "edge_runtime",
            "status": "passed",
        }
        inspect = subprocess.run(
            ["docker", "inspect", container_id], capture_output=True, text=True, check=True
        )
        record = json.loads(inspect.stdout)[0]
        assert not record["HostConfig"]["PortBindings"]
        now = datetime.now(UTC)
        revoked = (
            x509.RevokedCertificateBuilder()
            .serial_number(server_certificate.serial_number)
            .revocation_date(now - timedelta(seconds=1))
            .build()
        )
        replacement_crl = (
            x509.CertificateRevocationListBuilder().issuer_name(ca.subject)
            .last_update(now - timedelta(seconds=1))
            .next_update(now + timedelta(hours=1))
            .add_revoked_certificate(revoked)
            .sign(ca_key, hashes.SHA256())
        )
        _write(
            tmp_path / "certificates" / "control-plane-server.crl.pem",
            replacement_crl.public_bytes(serialization.Encoding.PEM),
            0o644,
        )
        degraded = _wait_for(health_path, "degraded", 15)
        assert credential not in json.dumps(degraded)
        running = subprocess.run(
            ["docker", "inspect", "-f", "{{.State.Running}}", container_id],
            capture_output=True, text=True, check=True,
        )
        assert running.stdout.strip() == "true"
        logs = subprocess.run(
            ["docker", "logs", container_id], capture_output=True, text=True, check=True
        )
        assert credential not in logs.stdout + logs.stderr
        stopped = subprocess.run(
            ["docker", "stop", "--time", "5", container_id],
            capture_output=True, text=True, check=False,
        )
        assert stopped.returncode == 0, stopped.stderr
        exit_code = subprocess.run(
            ["docker", "inspect", "-f", "{{.State.ExitCode}}", container_id],
            capture_output=True, text=True, check=True,
        )
        assert exit_code.stdout.strip() == "0"
    finally:
        server.shutdown()
        server.server_close()
        subprocess.run(["docker", "rm", "-f", container_id], capture_output=True)


def test_server_crl_context_checks_leaf_and_rejects_stale_crl(tmp_path):
    _fingerprint, _context, ca_key, ca, _server = _materials(tmp_path)
    ca_path = tmp_path / "certificates" / "ca.pem"
    crl_path = tmp_path / "certificates" / "control-plane-server.crl.pem"
    context = build_server_crl_ssl_context(str(ca_path), str(crl_path))
    assert context.verify_flags & ssl.VERIFY_CRL_CHECK_LEAF
    now = datetime.now(UTC)
    stale = (
        x509.CertificateRevocationListBuilder().issuer_name(ca.subject)
        .last_update(now - timedelta(days=2)).next_update(now - timedelta(days=1))
        .sign(ca_key, hashes.SHA256())
    )
    _write(crl_path, stale.public_bytes(serialization.Encoding.PEM), 0o644)
    with pytest.raises(ValueError, match="not fresh"):
        validate_server_crl(str(ca_path), str(crl_path))