"""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))