#!/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 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",
}
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)

    probe_ids = [item.get("id") for item in contract.get("probes", [])]
    check(set(probe_ids) == REQUIRED_PROBES and len(probe_ids) == len(set(probe_ids)), "probe coverage or uniqueness is invalid", 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)
    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())