#!/usr/bin/env python3 """Read-only, fail-closed deployment preflight probes. No probe mutates a remote service. Missing configuration or inaccessible targets are BLOCKED, never PASS. Output is deliberately limited to redacted metadata. """ from __future__ import annotations import argparse import datetime as dt import json import os import re import socket import ssl import urllib.error import urllib.parse import urllib.request from pathlib import Path from typing import Any ROOT = Path(__file__).resolve().parent STATUSES = {"PASS", "FAIL", "BLOCKED"} SENSITIVE = re.compile(r"(?i)(secret|token|password|authorization|cookie|key|credential)") def now() -> str: return dt.datetime.now(dt.timezone.utc).isoformat() def safe_target(value: str | None) -> str | None: if not value: return None parsed = urllib.parse.urlsplit(value) if parsed.scheme and parsed.hostname: port = f":{parsed.port}" if parsed.port else "" return f"{parsed.scheme}://{parsed.hostname}{port}{parsed.path}" return value if not SENSITIVE.search(value) else "[REDACTED]" def scrub(value: Any, key: str = "") -> Any: if SENSITIVE.search(key): return "[REDACTED]" if isinstance(value, dict): return {k: scrub(v, k) for k, v in value.items()} if isinstance(value, list): return [scrub(v, key) for v in value] if isinstance(value, str): value = re.sub(r"(?i)(bearer|basic)\s+\S+", r"\1 [REDACTED]", value) return safe_target(value) if "://" in value else value return value def result(probe_id: str, status: str, summary: str, *, target: str | None = None, evidence: dict[str, Any] | None = None, remediation: str | None = None) -> dict[str, Any]: if status not in STATUSES: raise ValueError(status) return scrub({"probe_id": probe_id, "status": status, "checked_at": now(), "target": safe_target(target), "summary": summary, "evidence": evidence or {}, "remediation": remediation}) def https_probe(p: dict[str, Any]) -> dict[str, Any]: target = p.get("target") or os.getenv(p.get("target_env", "")) if not target: return result(p["id"], "BLOCKED", "Probe URL is not configured.", remediation=f"Set {p.get('target_env', 'the target URL')} for an authorized read-only endpoint.") req = urllib.request.Request(target, method="GET", headers={"User-Agent": "pdf2html-preflight/1"}) try: with urllib.request.urlopen(req, timeout=8) as response: code = response.status status = "PASS" if 200 <= code < 400 else "FAIL" return result(p["id"], status, "HTTPS endpoint responded." if status == "PASS" else "HTTPS endpoint returned an unhealthy status.", target=target, evidence={"http_status": code}) except urllib.error.HTTPError as exc: return result(p["id"], "FAIL", "HTTPS endpoint returned an error status.", target=target, evidence={"http_status": exc.code}) except (OSError, urllib.error.URLError) as exc: return result(p["id"], "BLOCKED", "HTTPS endpoint could not be reached.", target=target, evidence={"error_type": type(exc).__name__}, remediation="Verify read-only network access and routing.") def dns_tls_probe(p: dict[str, Any]) -> dict[str, Any]: host = p["host"] try: addresses = sorted({row[4][0] for row in socket.getaddrinfo(host, 443, type=socket.SOCK_STREAM)}) context = ssl.create_default_context() with socket.create_connection((host, 443), timeout=8) as raw: with context.wrap_socket(raw, server_hostname=host) as secured: cert = secured.getpeercert() version = secured.version() return result(p["id"], "PASS", "DNS resolved and the TLS certificate validated.", target=f"https://{host}", evidence={"address_count": len(addresses), "tls_version": version, "certificate_not_after": cert.get("notAfter")}) except (OSError, ssl.SSLError) as exc: return result(p["id"], "BLOCKED", "DNS/TLS verification could not complete.", target=f"https://{host}", evidence={"error_type": type(exc).__name__}, remediation="Verify resolver/network access before diagnosing deployment state.") def cors_probe(p: dict[str, Any]) -> dict[str, Any]: target = os.getenv(p["target_env"]) if not target: return result(p["id"], "BLOCKED", "CORS/CSRF probe endpoint is not configured.", remediation=f"Set {p['target_env']} to a safe endpoint that supports OPTIONS.") origin = p["origin"] req = urllib.request.Request(target, method="OPTIONS", headers={ "Origin": origin, "Access-Control-Request-Method": "POST", "Access-Control-Request-Headers": "content-type,x-csrf-token"}) try: with urllib.request.urlopen(req, timeout=8) as response: allow_origin = response.headers.get("Access-Control-Allow-Origin") credentials = response.headers.get("Access-Control-Allow-Credentials", "").lower() vary = response.headers.get("Vary", "") ok = allow_origin == origin and allow_origin != "*" and credentials == "true" return result(p["id"], "PASS" if ok else "FAIL", "Exact credentialed CORS policy observed." if ok else "Required exact credentialed CORS policy was not observed.", target=target, evidence={"http_status": response.status, "allow_origin": allow_origin, "allow_credentials": credentials, "vary_origin": "origin" in vary.lower(), "csrf_header_preflighted": "x-csrf-token" in response.headers.get("Access-Control-Allow-Headers", "").lower()}) except urllib.error.HTTPError as exc: return result(p["id"], "FAIL", "CORS preflight returned an error status.", target=target, evidence={"http_status": exc.code}) except (OSError, urllib.error.URLError) as exc: return result(p["id"], "BLOCKED", "CORS preflight could not be reached.", target=target, evidence={"error_type": type(exc).__name__}) def evidence_probe(p: dict[str, Any]) -> dict[str, Any]: path = os.getenv(p["evidence_env"]) if not path: return result(p["id"], "BLOCKED", "Authorized redacted evidence was not supplied.", remediation=f"Set {p['evidence_env']} to a redacted JSON evidence file.") try: data = json.loads(Path(path).read_text()) except (OSError, json.JSONDecodeError): return result(p["id"], "BLOCKED", "Evidence file could not be read as JSON.") if data.get("probe_id") != p["id"] or data.get("status") not in STATUSES: return result(p["id"], "FAIL", "Evidence identity or status is invalid.") if data["status"] == "PASS" and not data.get("checked_at"): return result(p["id"], "FAIL", "PASS evidence has no verification timestamp.") return result(p["id"], data["status"], "Imported authorized redacted evidence.", evidence={"checked_at": data.get("checked_at"), "method": data.get("method"), "assertions": data.get("assertions", {})}) def secret_names_probe(p: dict[str, Any], contract: dict[str, Any]) -> dict[str, Any]: names = [item["name"] for item in contract["required_secrets"]] missing = [name for name in names if name not in os.environ] # Presence only: values are never read or emitted. status = "PASS" if not missing else "BLOCKED" return result(p["id"], status, "All required secret names are present." if not missing else "Required secret names are not all present.", evidence={"required_count": len(names), "present_count": len(names) - len(missing), "missing_names": missing}, remediation="Provision the named secrets in the documented runtime scopes." if missing else None) def run(contract: dict[str, Any]) -> dict[str, Any]: handlers = {"https": https_probe, "dns_tls": dns_tls_probe, "cors": cors_probe, "evidence": evidence_probe} results = [] for probe in contract["probes"]: if probe["kind"] == "secret_names": item = secret_names_probe(probe, contract) else: item = handlers[probe["kind"]](probe) results.append(item) overall = "PASS" if all(x["status"] == "PASS" for x in results) else ("FAIL" if any(x["status"] == "FAIL" for x in results) else "BLOCKED") return {"contract_version": contract["contract_version"], "generated_at": now(), "overall_status": overall, "results": results, "redaction": {"secrets_included": False, "credentials_included": False, "response_bodies_included": False}} def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("--contract", type=Path, default=ROOT / "deployment-contract.json") parser.add_argument("--output", type=Path) args = parser.parse_args() report = run(json.loads(args.contract.read_text())) text = json.dumps(report, indent=2) + "\n" if args.output: args.output.write_text(text) else: print(text, end="") return 0 if report["overall_status"] == "PASS" else 2 if __name__ == "__main__": raise SystemExit(main())