#!/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)")
EVIDENCE_MAX_AGE = dt.timedelta(hours=24)
EVIDENCE_FUTURE_SKEW = dt.timedelta(minutes=5)
EVIDENCE_SCHEMAS: dict[str, dict[str, Any]] = {
    "cloudpanel-worker": {
        "method": "read-only health/status inspection",
        "assertions": {"separate_process": True},
    },
    "postgres-platform-db": {
        "method": "read-only @platform-modules/db connectivity probe",
        "assertions": {"connected": True, "access": "@platform-modules/db"},
    },
    "private-s3-lifecycle": {
        "method": "read-only bucket policy and lifecycle inspection",
        "assertions": {"private": True, "source_ttl_hours": 48, "zip_ttl_hours": 48},
    },
}


class NoRedirect(urllib.request.HTTPRedirectHandler):
    def redirect_request(self, req, fp, code, msg, headers, newurl):
        return None


def origin_matches(value: str, expected: str) -> bool:
    try:
        actual = urllib.parse.urlsplit(value)
        wanted = urllib.parse.urlsplit(expected)
        actual_port = actual.port or (443 if actual.scheme == "https" else None)
        wanted_port = wanted.port or (443 if wanted.scheme == "https" else None)
    except (TypeError, ValueError):
        return False
    return (actual.scheme == "https" and actual.username is None and actual.password is None
            and actual.hostname == wanted.hostname and actual_port == wanted_port)


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.")
    expected_origin = p.get("expected_origin") or p.get("target")
    if not expected_origin or not origin_matches(target, expected_origin):
        return result(p["id"], "FAIL", "Probe URL is outside the required HTTPS origin.", target=target)
    req = urllib.request.Request(target, method="GET", headers={"User-Agent": "pdf2html-preflight/1"})
    try:
        with urllib.request.build_opener(NoRedirect).open(req, timeout=8) as response:
            code = response.status
            final_url = response.geturl()
            origin_ok = origin_matches(final_url, expected_origin)
            status = "PASS" if 200 <= code < 300 and origin_ok else "FAIL"
            return result(p["id"], status, "HTTPS endpoint responded on the required origin." if status == "PASS" else "HTTPS endpoint did not remain healthy on the required origin.",
                          target=target, evidence={"http_status": code, "final_origin_valid": origin_ok})
    except urllib.error.HTTPError as exc:
        return result(p["id"], "FAIL", "HTTPS endpoint returned an error or redirect status.", target=target,
                      evidence={"http_status": exc.code})
    except (OSError, urllib.error.URLError, ValueError) 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"]
    method = p["method"]
    if not origin_matches(target, p["expected_origin"]):
        return result(p["id"], "FAIL", "CORS probe URL is outside the required HTTPS API origin.", target=target)
    req = urllib.request.Request(target, method="OPTIONS", headers={
        "Origin": origin, "Access-Control-Request-Method": method,
        "Access-Control-Request-Headers": "content-type,x-csrf-token"})
    try:
        with urllib.request.build_opener(NoRedirect).open(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", "")
            allow_headers = {value.strip().lower() for value in response.headers.get("Access-Control-Allow-Headers", "").split(",")}
            allow_methods = {value.strip().upper() for value in response.headers.get("Access-Control-Allow-Methods", "").split(",")}
            csrf_ok = "x-csrf-token" in allow_headers
            method_ok = method.upper() in allow_methods
            final_origin_ok = origin_matches(response.geturl(), p["expected_origin"])
            ok = (200 <= response.status < 300 and allow_origin == origin and allow_origin != "*"
                  and credentials == "true" and csrf_ok and method_ok and final_origin_ok)
            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": csrf_ok,
                                                   "requested_method_preflighted": method_ok,
                                                   "final_origin_valid": final_origin_ok})
    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":
        checked_at = data.get("checked_at")
        try:
            if not isinstance(checked_at, str) or not checked_at.endswith(("Z", "+00:00")):
                raise ValueError("timestamp is not explicitly UTC")
            parsed_at = dt.datetime.fromisoformat(checked_at.replace("Z", "+00:00"))
            current = dt.datetime.now(dt.timezone.utc)
            if parsed_at.tzinfo != dt.timezone.utc or parsed_at < current - EVIDENCE_MAX_AGE or parsed_at > current + EVIDENCE_FUTURE_SKEW:
                raise ValueError("timestamp is stale or outside the allowed clock skew")
        except (TypeError, ValueError):
            return result(p["id"], "FAIL", "PASS evidence has an invalid or stale UTC verification timestamp.")
        schema = EVIDENCE_SCHEMAS.get(p["id"])
        assertions = data.get("assertions")
        if (not schema or data.get("method") != schema["method"] or not isinstance(assertions, dict)
                or any(key not in assertions or type(assertions[key]) is not type(expected)
                       or assertions[key] != expected
                       for key, expected in schema["assertions"].items())):
            return result(p["id"], "FAIL", "PASS evidence does not satisfy the required method and assertions.")
    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())