#!/usr/bin/env python3 """Validate the deployment preflight contract without network access.""" from __future__ import annotations import importlib.util import json import os import sys import tempfile from pathlib import Path from unittest.mock import MagicMock, patch ROOT = Path(__file__).resolve().parent CONTRACT_PATH = ROOT / "deployment-contract.json" TEMPLATE_PATH = ROOT / "evidence-template.json" REQUIRED_PROBES = { "cloudflare-frontend", "cloudpanel-api-route", "cloudpanel-worker", "postgres-platform-db", "private-s3-lifecycle", "dns-tls-origin", "split-origin-session", "required-secret-names", } PROBE_SCHEMAS = { "cloudflare-frontend": {"kind": "https", "target": "https://tools.press.zone/"}, "cloudpanel-api-route": {"kind": "https", "target_env": "PDF2HTML_API_HEALTH_URL", "expected_origin": "https://api.press.zone"}, "cloudpanel-worker": {"kind": "evidence", "evidence_env": "PDF2HTML_WORKER_EVIDENCE"}, "postgres-platform-db": {"kind": "evidence", "evidence_env": "PDF2HTML_POSTGRES_EVIDENCE"}, "private-s3-lifecycle": {"kind": "evidence", "evidence_env": "PDF2HTML_S3_LIFECYCLE_EVIDENCE"}, "dns-tls-origin": {"kind": "dns_tls", "host": "api.press.zone"}, "split-origin-session": {"kind": "cors", "target_env": "PDF2HTML_API_CORS_PROBE_URL", "expected_origin": "https://api.press.zone", "origin": "https://tools.press.zone", "method": "POST"}, "required-secret-names": {"kind": "secret_names"}, } REQUIRED_SECRETS = { "DATABASE_URL", "SESSION_SECRET", "S3_ENDPOINT", "S3_REGION", "S3_BUCKET", "S3_ACCESS_KEY_ID", "S3_SECRET_ACCESS_KEY", "STRIPE_SECRET_KEY", "STRIPE_WEBHOOK_SECRET", } def check(condition: bool, message: str, errors: list[str]) -> None: if not condition: errors.append(message) def load_probe(): spec = importlib.util.spec_from_file_location("preflight_probe", ROOT / "probe.py") assert spec and spec.loader module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module def main() -> int: errors: list[str] = [] try: contract = json.loads(CONTRACT_PATH.read_text()) template = json.loads(TEMPLATE_PATH.read_text()) except (OSError, json.JSONDecodeError) as exc: print(f"FAIL: contract files are unreadable: {type(exc).__name__}") return 1 policy = contract.get("policy", {}) check(policy.get("read_only") is True, "policy must be read-only", errors) check(policy.get("fail_closed") is True, "policy must fail closed", errors) check(policy.get("unknown_access_status") == "BLOCKED", "unknown access must be BLOCKED", errors) check(set(policy.get("allowed_statuses", [])) == {"PASS", "FAIL", "BLOCKED"}, "status vocabulary is invalid", errors) check(set(contract.get("source_requirements", [])) == {"plan-0.6", "readiness-B7"}, "source traceability is incomplete", errors) deployment = contract.get("deployment", {}) check(deployment.get("frontend", {}).get("origin") == "https://tools.press.zone", "frontend origin is wrong", errors) check(deployment.get("api", {}).get("origin") == "https://api.press.zone", "API origin is wrong", errors) check(deployment.get("worker", {}).get("separate_process") is True, "worker separation is not recorded", errors) check(deployment.get("database", {}).get("access") == "@platform-modules/db", "database adapter contract is wrong", errors) storage = deployment.get("object_storage", {}) check(storage.get("kind") == "private-s3-compatible", "private S3 contract is missing", errors) check(storage.get("source_ttl_hours") == storage.get("zip_ttl_hours") == 48, "48-hour lifecycle is missing", errors) session = deployment.get("session", {}) check(session.get("cors", {}).get("allowed_origins") == ["https://tools.press.zone"], "CORS allowlist must be exact", errors) check(session.get("cors", {}).get("wildcard") is False, "wildcard credentialed CORS is forbidden", errors) check(session.get("cors", {}).get("allow_credentials") is True, "credentialed CORS is not recorded", errors) check(session.get("csrf", {}).get("required_for_cookie_authenticated_mutations") is True, "CSRF requirement is missing", errors) check(session.get("cookie", {}).get("secure") is True and session.get("cookie", {}).get("same_site") is True, "secure same-site cookie contract is missing", errors) probes = contract.get("probes", []) probe_ids = [item.get("id") for item in probes] check(set(probe_ids) == REQUIRED_PROBES and len(probe_ids) == len(set(probe_ids)), "probe coverage or uniqueness is invalid", errors) for item in probes: probe_id = item.get("id") expected = PROBE_SCHEMAS.get(probe_id) if expected is None: continue check(item.get("required") is True, f"probe {probe_id} must be required", errors) check(set(item) == {"id", "required", *expected}, f"probe {probe_id} has missing or unexpected configuration keys", errors) for key, value in expected.items(): check(item.get(key) == value, f"probe {probe_id} has invalid {key}", errors) secret_names = {item.get("name") for item in contract.get("required_secrets", [])} check(secret_names == REQUIRED_SECRETS, "required secret-name manifest is incomplete", errors) check(all(set(item) == {"name", "scope"} for item in contract.get("required_secrets", [])), "secret manifest may contain names/scopes only", errors) check(template.get("overall_status") == "BLOCKED", "blank evidence must default to BLOCKED", errors) redaction = template.get("redaction", {}) check(all(redaction.get(key) is False for key in ("secrets_included", "credentials_included", "response_bodies_included")), "evidence template redaction flags are unsafe", errors) try: probe = load_probe() empty_env = {"PATH": os.environ.get("PATH", "")} with patch.dict(os.environ, empty_env, clear=True): blocked = probe.secret_names_probe({"id": "required-secret-names"}, contract) check(blocked["status"] == "BLOCKED", "missing secret access did not report BLOCKED", errors) evidence = probe.evidence_probe({"id": "private-s3-lifecycle", "evidence_env": "UNSET"}) check(evidence["status"] == "BLOCKED", "missing evidence did not report BLOCKED", errors) redacted = probe.scrub({"Authorization": "Bearer exposed", "url": "https://user:pass@example.test/path?token=bad"}) check(redacted["Authorization"] == "[REDACTED]", "authorization redaction failed", errors) check("user" not in redacted["url"] and "pass" not in redacted["url"] and "token" not in redacted["url"], "URL credential/query redaction failed", errors) with tempfile.TemporaryDirectory() as directory: path = Path(directory) / "evidence.json" path.write_text(json.dumps({"probe_id": "cloudpanel-worker", "status": "PASS"})) with patch.dict(os.environ, {"EVIDENCE": str(path)}, clear=True): invalid = probe.evidence_probe({"id": "cloudpanel-worker", "evidence_env": "EVIDENCE"}) check(invalid["status"] == "FAIL", "un-timestamped PASS evidence was accepted", errors) path.write_text(json.dumps({"probe_id": "private-s3-lifecycle", "status": "PASS", "checked_at": "not-a-date", "method": None, "assertions": {}})) with patch.dict(os.environ, {"EVIDENCE": str(path)}, clear=True): invalid = probe.evidence_probe({"id": "private-s3-lifecycle", "evidence_env": "EVIDENCE"}) check(invalid["status"] == "FAIL", "malformed unstructured S3 PASS evidence was accepted", errors) response = MagicMock() response.__enter__.return_value = response response.status = 204 response.geturl.return_value = "https://api.press.zone/cors" response.headers = { "Access-Control-Allow-Origin": "https://tools.press.zone", "Access-Control-Allow-Credentials": "true", "Access-Control-Allow-Methods": "POST", "Access-Control-Allow-Headers": "content-type", } opener = MagicMock() opener.open.return_value = response cors_config = PROBE_SCHEMAS["split-origin-session"] | {"id": "split-origin-session", "required": True} with patch.dict(os.environ, {"PDF2HTML_API_CORS_PROBE_URL": "https://api.press.zone/cors"}, clear=True), \ patch.object(probe.urllib.request, "build_opener", return_value=opener): cors_result = probe.cors_probe(cors_config) check(cors_result["status"] == "FAIL", "CORS without x-csrf-token allow-header was accepted", errors) with patch.dict(os.environ, {"PDF2HTML_API_HEALTH_URL": "http://unrelated.invalid/ok"}, clear=True): route_result = probe.https_probe(PROBE_SCHEMAS["cloudpanel-api-route"] | {"id": "cloudpanel-api-route", "required": True}) check(route_result["status"] == "FAIL", "non-HTTPS unrelated API health URL was accepted", errors) except Exception as exc: # validator must turn internal issues into a clear failure errors.append(f"probe self-test raised {type(exc).__name__}: {exc}") if errors: for error in errors: print(f"FAIL: {error}") print(f"Contract validation failed with {len(errors)} error(s).") return 1 print(f"PASS: deployment contract validated ({len(REQUIRED_PROBES)} probes, {len(REQUIRED_SECRETS)} required secret names).") print("PASS: fail-closed and redaction self-tests passed.") return 0 if __name__ == "__main__": raise SystemExit(main())