#!/usr/bin/env python3
# SPDX-License-Identifier: GPL-3.0-or-later
"""Host-only regression tests for the Phase-0.9A anti-brick design."""

from __future__ import annotations

import argparse
import hashlib
import importlib.util
import json
import sys
from pathlib import Path
from types import ModuleType
from typing import Any, Callable


IMMUTABLE = {
    "docs/runtime/phase-0.8-read-only-preflight.md": (
        "3fbe086175a6048176075f447ec1482074928e3b5282db97ea2169395fe1d508"
    ),
    "manifests/runtime/phase-0.8-read-only-preflight.json": (
        "47d7f452f8799979fe99b3e6d56859f03544112725bf7e5b349eba5ed81b3322"
    ),
    "tests/test_phase08_preflight.py": (
        "8a4ad7c70de28ffe3148fd3fd1f68c36a872c53c691c9068e1ff163970863c48"
    ),
}

DENIED_SHA256 = (
    "4be1c17b4964f2b68c39b5145bc4af4619c32512d60269ecf5c39728b390fa63"
)

TEMPLATES = {
    "docs/approvals/phase-0.9-observation-template.md": (
        "PHASE09_OBSERVATION_TEMPLATE",
        "observation",
    ),
    "docs/approvals/phase-0.9-backup-creation-template.md": (
        "PHASE09_BACKUP_CREATION_TEMPLATE",
        "backup_creation",
    ),
    "docs/approvals/phase-0.9-staging-template.md": (
        "PHASE09_STAGING_TEMPLATE",
        "staging",
    ),
    "docs/approvals/phase-0.9-switch-template.md": (
        "PHASE09_SWITCH_TEMPLATE",
        "live_switch",
    ),
    "docs/approvals/phase-0.9-one-shot-execution-template.md": (
        "PHASE09_ONE_SHOT_EXECUTION_TEMPLATE",
        "one_shot_execution",
    ),
}

FALSE_AUTH_FIELDS = (
    "authorized",
    "installation_authorized",
    "execution_authorized",
    "transfer_authorized",
    "lifecycle_authorized",
    "automatic_retry",
)


def sha256(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as stream:
        for block in iter(lambda: stream.read(1024 * 1024), b""):
            digest.update(block)
    return digest.hexdigest()


def load_json(path: Path) -> dict[str, Any]:
    result = json.loads(path.read_text(encoding="utf-8"))
    if not isinstance(result, dict):
        raise RuntimeError(f"{path}: expected JSON object")
    return result


def extract_contract(path: Path, name: str) -> dict[str, Any]:
    text = path.read_text(encoding="utf-8")
    begin = f"<!-- BEGIN {name} -->"
    end = f"<!-- END {name} -->"
    if text.count(begin) != 1 or text.count(end) != 1:
        raise RuntimeError(f"{path}: expected one {name} contract")
    fenced = text.split(begin, 1)[1].split(end, 1)[0].strip()
    if not fenced.startswith("```json\n") or not fenced.endswith("\n```"):
        raise RuntimeError(f"{path}: malformed JSON fence")
    result = json.loads(fenced[len("```json\n") : -len("\n```")])
    if not isinstance(result, dict):
        raise RuntimeError(f"{path}: expected object in contract")
    return result


def load_simulator(root: Path) -> ModuleType:
    path = root / "tools/simulate_phase09_transaction.py"
    spec = importlib.util.spec_from_file_location("phase09_simulator", path)
    if spec is None or spec.loader is None:
        raise RuntimeError("could not load Phase-0.9A simulator")
    module = importlib.util.module_from_spec(spec)
    sys.modules[spec.name] = module
    previous = sys.dont_write_bytecode
    sys.dont_write_bytecode = True
    try:
        spec.loader.exec_module(module)
    finally:
        sys.dont_write_bytecode = previous
    return module


def require(condition: bool, message: str) -> None:
    if not condition:
        raise RuntimeError(message)


def require_blocked(report: dict[str, Any], scenario: str) -> None:
    require(
        report["state"] in {"BLOCKED", "ROLLBACK_REQUIRED"},
        f"{scenario}: state did not fail closed: {report['state']}",
    )
    require(
        report["target_execution_performed"] is False,
        f"{scenario}: target execution was claimed",
    )
    require(
        report["automatic_retry"] is False,
        f"{scenario}: automatic retry was enabled",
    )
    require(
        report["crash_invariant"]
        in {
            "A_OLD_LIVE_COMPLETE",
            "B_NEW_LIVE_COMPLETE_AND_BACKUP_INTACT",
            "C_REJECTED_UNSAFE_OR_UNPROVEN",
        },
        f"{scenario}: crash invariant missing",
    )


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--root", type=Path, required=True)
    args = parser.parse_args()
    root = args.root.resolve()
    manifest = load_json(
        root / "manifests/runtime/phase-0.9-anti-brick-design.json"
    )
    denylist = load_json(root / "manifests/artifact-denylist.json")
    simulator = load_simulator(root)
    passed: list[str] = []

    def check(name: str, test: Callable[[], None]) -> None:
        test()
        passed.append(name)

    def immutable_evidence() -> None:
        for relative, expected in IMMUTABLE.items():
            require(sha256(root / relative) == expected, f"immutable drift: {relative}")

    check("immutable Phase-0.8 evidence", immutable_evidence)

    check(
        "design-only top-level state",
        lambda: require(
            manifest["status"] == "DESIGN_ONLY"
            and manifest["phase"] == "PHASE_0_9A_OFFLINE_ANTI_BRICK"
            and manifest["historical_status"] == "READ_ONLY_PREFLIGHT_BLOCKED",
            "design or historical status changed",
        ),
    )

    def false_activity_and_authority() -> None:
        for field in (
            "hardware_observed",
            "ps5_connected",
            "device_write_performed",
            "files_transferred",
            "target_execution_performed",
            "target_artifact_created",
            "installation_authorized",
            "lifecycle_authorized",
            "execution_authorized",
            "transfer_authorized",
            "automatic_retry",
            "lifecycle_probe_in_scope",
            "graphics_in_scope",
            "retroarch_in_scope",
            "host_simulation_is_hardware_evidence",
        ):
            require(manifest.get(field) is False, f"{field} is not false")

    check("all activity and authorization fields false", false_activity_and_authority)

    def nested_authorizations_false() -> None:
        controlled = {
            "authorized",
            "installation_authorized",
            "execution_authorized",
            "transfer_authorized",
            "lifecycle_authorized",
            "automatic_retry",
        }

        def visit(value: Any, location: str) -> None:
            if isinstance(value, dict):
                for key, child in value.items():
                    if key in controlled:
                        require(child is False, f"{location}.{key} is not false")
                    visit(child, f"{location}.{key}")
            elif isinstance(value, list):
                for index, child in enumerate(value):
                    visit(child, f"{location}[{index}]")

        visit(manifest, "manifest")

    check("nested authorization and retry fields false", nested_authorizations_false)

    def permanent_denylist() -> None:
        entries = denylist["entries"]
        require(denylist["fail_closed"] is True, "denylist is not fail closed")
        require(len(entries) == 1, "denylist entry count changed")
        entry = entries[0]
        require(entry["sha256"] == DENIED_SHA256, "denied hash changed")
        require(entry["status"] == "BLOCKED", "denied hash not blocked")
        require(entry["permanent"] is True, "denied hash not permanent")
        require(entry["execution_eligible"] is False, "denied hash eligible")
        require(
            manifest["permanent_denylist_binding"]
            == {
                "sha256": DENIED_SHA256,
                "status": "BLOCKED",
                "permanent": True,
                "execution_eligible": False,
            },
            "manifest denylist binding changed",
        )

    check("permanent denylist remains exact", permanent_denylist)

    check(
        "firmware and stock identities remain unproven",
        lambda: require(
            manifest["firmware_runtime_behavior"] == "UNPROVEN"
            and manifest["stock_identification"] == "reference_only",
            "firmware or stock evidence was promoted",
        ),
    )
    check(
        "Payload Manager backup remains hard blocker",
        lambda: require(
            manifest["payload_manager_backup"] == "HARD_BLOCKER",
            "Payload Manager backup hard blocker changed",
        ),
    )

    def threat_model_complete() -> None:
        items = manifest["threat_model_items"]
        require(len(items) == 56, f"expected 56 threat items, got {len(items)}")
        ids = {item["id"] for item in items}
        require(len(ids) == len(items), "duplicate threat item ID")
        require(
            {item["profile"] for item in items}
            == {
                "A_WRONG_TARGET",
                "B_WRONG_PREIMAGE",
                "C_BACKUP_FAILURE",
                "D_WRITE_POWER_LOSS",
                "E_PROCESS_LIFECYCLE",
                "F_ROLLBACK_FAILURE",
                "G_OPERATOR_ERROR",
            },
            "threat profiles incomplete",
        )
        require(
            all(
                item["severity"] in {"CATASTROPHIC", "HIGH", "MEDIUM", "LOW"}
                and item["reason"]
                for item in items
            ),
            "threat severity or reason missing",
        )

    check("complete classified threat model", threat_model_complete)

    def invariants_complete() -> None:
        invariants = manifest["anti_brick_invariants"]
        require(len(invariants) == 20, "anti-brick invariant count changed")
        require(
            [item["id"] for item in invariants]
            == [f"AB-{number:03d}" for number in range(1, 21)],
            "anti-brick invariant IDs changed",
        )

    check("AB-001 through AB-020 present", invariants_complete)
    check(
        "transaction state set exact",
        lambda: require(
            tuple(manifest["transaction_states"]) == simulator.STATES,
            "manifest and simulator state sets differ",
        ),
    )

    def transition_skip_rejected() -> None:
        machine = simulator.StateMachine()
        try:
            machine.transition("LIVE_OBJECTS_VERIFIED")
        except simulator.TransitionError:
            return
        raise RuntimeError("state machine accepted an approval-gate skip")

    check("authorization states cannot be skipped", transition_skip_rejected)
    check(
        "no retry, autoload or combined transition",
        lambda: require(
            {
                "automatic_retry",
                "autoload",
                "combined_component_installation",
                "lifecycle_transition",
                "graphics_transition",
                "retroarch_transition",
            }.issubset(set(manifest["forbidden_transitions"])),
            "forbidden transition set incomplete",
        ),
    )

    def templates_fail_closed() -> None:
        for relative, (marker, action) in TEMPLATES.items():
            template = extract_contract(root / relative, marker)
            require(template["template_only"] is True, f"{relative}: not template")
            require(
                template["template_action"] == action,
                f"{relative}: wrong action binding",
            )
            for field in FALSE_AUTH_FIELDS:
                require(template[field] is False, f"{relative}: {field} not false")
            required = template["required_fields"]
            require(required, f"{relative}: required fields absent")
            require(
                all(value is None for value in required.values()),
                f"{relative}: request data was prefilled",
            )
            require(
                template["fixed_exclusions"]["automatic_retry"] is True,
                f"{relative}: retry exclusion missing",
            )
            require(
                template["fixed_exclusions"]["autoload"] is True,
                f"{relative}: autoload exclusion missing",
            )
            require(
                template["fixed_exclusions"]["lifecycle_probe"] is True,
                f"{relative}: lifecycle exclusion missing",
            )

    check("all five templates are false and unfilled", templates_fail_closed)

    def exact_live_identity_required() -> None:
        for fault in (
            "missing_live_path",
            "missing_mount_id",
            "missing_object_id",
            "wrong_size",
            "wrong_preimage_hash",
        ):
            report = simulator.simulate_transaction(
                "hardened_elfldr", [fault], prove_virtual_switch_model=True
            )
            require_blocked(report, fault)
            require(report["virtual_writes"] == [], f"{fault}: write occurred")

    check("no write without exact full live identity", exact_live_identity_required)

    check(
        "reference-only stock hash never authorizes write",
        lambda: require_blocked(
            simulator.simulate_transaction(
                "hardened_elfldr",
                ["reference_only_preimage"],
                prove_virtual_switch_model=True,
            ),
            "reference_only_preimage",
        ),
    )

    def staging_needs_backup() -> None:
        for fault in (
            "backup_same_object_as_live",
            "short_backup_write",
            "backup_hash_mismatch",
            "backup_not_reopened",
        ):
            report = simulator.simulate_transaction(
                "controlled_payload_manager",
                [fault],
                prove_virtual_switch_model=True,
            )
            require_blocked(report, fault)
            require(
                "virtual_candidate_stage" not in report["virtual_writes"],
                f"{fault}: staging occurred without verified backup",
            )

    check("no staging without separate reopened backup", staging_needs_backup)

    def separate_approvals() -> None:
        for fault in (
            "authorization_missing",
            "authorization_wrong_hash",
            "authorization_expired",
        ):
            require_blocked(
                simulator.simulate_transaction(
                    "hardened_elfldr", [fault], prove_virtual_switch_model=True
                ),
                fault,
            )

    check("missing mismatched or expired approval stops", separate_approvals)

    check(
        "no switch while atomicity is unproven",
        lambda: require_blocked(
            simulator.simulate_transaction("hardened_elfldr"),
            "default unproven platform",
        ),
    )
    check(
        "directory durability unknown stops",
        lambda: require_blocked(
            simulator.simulate_transaction(
                "hardened_elfldr",
                ["directory_durability_unknown"],
                prove_virtual_switch_model=True,
            ),
            "directory_durability_unknown",
        ),
    )

    def synthetic_happy_path_stops_before_execution() -> None:
        report = simulator.simulate_transaction(
            "hardened_elfldr", prove_virtual_switch_model=True
        )
        require(
            report["status"] == "DESIGN_MODEL_STOP_BEFORE_EXECUTION",
            "synthetic model did not stop before execution",
        )
        require(
            report["state"] == "MANUAL_EXECUTION_NOT_AUTHORIZED",
            "execution approval gate was bypassed",
        )
        require(
            report["crash_invariant"]
            == "B_NEW_LIVE_COMPLETE_AND_BACKUP_INTACT",
            "synthetic old/new invariant failed",
        )
        require(report["target_execution_performed"] is False, "execution claimed")

    check("post-switch model stops before execution", synthetic_happy_path_stops_before_execution)

    def no_retry_or_autoload() -> None:
        for fault in ("retry_active", "autoload_active", "timeout"):
            report = simulator.simulate_transaction(
                "hardened_elfldr", [fault], prove_virtual_switch_model=True
            )
            require_blocked(report, fault)
            require(report["automatic_retry"] is False, f"{fault}: retry enabled")
            require(report["autoload"] is False, f"{fault}: autoload enabled")

    check("autoload retry and timeout always stop", no_retry_or_autoload)

    check(
        "component transactions cannot be combined",
        lambda: require(
            set(simulator.COMPONENTS)
            == {"hardened_elfldr", "controlled_payload_manager"}
            and manifest["component_order"]["combined_install_all"] is False,
            "combined transaction surface exists",
        ),
    )
    check(
        "second component waits for acceptance or rollback",
        lambda: require_blocked(
            simulator.simulate_transaction(
                "controlled_payload_manager",
                ["second_component_before_first_accepted"],
                prove_virtual_switch_model=True,
            ),
            "second_component_before_first_accepted",
        ),
    )
    check(
        "lifecycle probe excluded from candidate set",
        lambda: require_blocked(
            simulator.simulate_transaction(
                "hardened_elfldr",
                ["lifecycle_probe_candidate"],
                prove_virtual_switch_model=True,
            ),
            "lifecycle_probe_candidate",
        ),
    )
    check(
        "graphics SDL and RetroArch inactive",
        lambda: require(
            manifest["graphics_in_scope"] is False
            and manifest["retroarch_in_scope"] is False
            and manifest["lifecycle_probe_in_scope"] is False,
            "later-phase scope became active",
        ),
    )

    def power_loss_contract_complete() -> None:
        boundaries = manifest["power_loss_boundaries"]
        require(len(boundaries) == 14, "power-loss boundary count changed")
        require(
            [item["id"] for item in boundaries]
            == list(simulator.POWER_LOSS_BOUNDARIES),
            "power-loss boundary order or identity changed",
        )
        require(
            all(item["result"] == "UNPROVEN" for item in boundaries),
            "a PS5 power-loss boundary was promoted",
        )
        require(
            any(item["result"] in {"UNSAFE", "UNPROVEN"} for item in boundaries),
            "unproven boundaries no longer block",
        )

    check("all power-loss boundaries explicit and blocking", power_loss_contract_complete)

    def virtual_power_loss_invariant() -> None:
        for boundary in simulator.POWER_LOSS_BOUNDARIES:
            report = simulator.simulate_power_loss_boundary(
                "hardened_elfldr",
                boundary,
                prove_virtual_switch_model=True,
            )
            require(
                report["crash_invariant"]
                in {
                    "A_OLD_LIVE_COMPLETE",
                    "B_NEW_LIVE_COMPLETE_AND_BACKUP_INTACT",
                },
                f"{boundary}: virtual model lost old/new invariant",
            )
            require(report["automatic_start"] is False, f"{boundary}: auto start")
            require(report["automatic_retry"] is False, f"{boundary}: retry")

    check("virtual power-loss old-or-new invariant", virtual_power_loss_invariant)

    check(
        "in-place overwrite forbidden",
        lambda: require_blocked(
            simulator.simulate_transaction(
                "hardened_elfldr",
                ["in_place_overwrite"],
                prove_virtual_switch_model=True,
            ),
            "in_place_overwrite",
        ),
    )
    check(
        "two-step rename gap forbidden",
        lambda: require_blocked(
            simulator.simulate_transaction(
                "hardened_elfldr",
                ["two_step_rename_gap"],
                prove_virtual_switch_model=True,
            ),
            "two_step_rename_gap",
        ),
    )

    def identity_race_rejected() -> None:
        for fault in ("symlink_substitution", "object_swap_after_preflight"):
            require_blocked(
                simulator.simulate_transaction(
                    "hardened_elfldr", [fault], prove_virtual_switch_model=True
                ),
                fault,
            )

    check("symlink and object-swap races rejected", identity_race_rejected)
    check(
        "active target rejected",
        lambda: require_blocked(
            simulator.simulate_transaction(
                "hardened_elfldr",
                ["target_process_active"],
                prove_virtual_switch_model=True,
            ),
            "target_process_active",
        ),
    )
    check(
        "candidate mismatch rejected",
        lambda: require_blocked(
            simulator.simulate_transaction(
                "controlled_payload_manager",
                ["candidate_hash_mismatch"],
                prove_virtual_switch_model=True,
            ),
            "candidate_hash_mismatch",
        ),
    )
    check(
        "wrong component mapping rejected",
        lambda: require_blocked(
            simulator.simulate_transaction(
                "controlled_payload_manager",
                ["wrong_component_artifact_mapping"],
                prove_virtual_switch_model=True,
            ),
            "wrong_component_artifact_mapping",
        ),
    )
    check(
        "unknown firmware rejected",
        lambda: require_blocked(
            simulator.simulate_transaction(
                "hardened_elfldr",
                ["unknown_firmware"],
                prove_virtual_switch_model=True,
            ),
            "unknown_firmware",
        ),
    )
    check(
        "live hash mismatch never promotes backup",
        lambda: require(
            simulator.simulate_transaction(
                "hardened_elfldr",
                ["wrong_preimage_hash"],
                prove_virtual_switch_model=True,
            )["virtual_writes"]
            == [],
            "mismatched live hash reached backup creation",
        ),
    )
    check(
        "recovery dependency on replaced component rejected",
        lambda: require_blocked(
            simulator.simulate_transaction(
                "controlled_payload_manager",
                ["recovery_depends_on_replaced_component"],
                prove_virtual_switch_model=True,
            ),
            "recovery_depends_on_replaced_component",
        ),
    )

    def failed_post_switch_goes_to_rollback_required() -> None:
        report = simulator.simulate_transaction(
            "hardened_elfldr",
            ["live_verification_failure"],
            prove_virtual_switch_model=True,
        )
        require(
            report["state"] == "ROLLBACK_REQUIRED",
            "failed live verification did not require rollback",
        )
        require(report["target_execution_performed"] is False, "execution claimed")

    check("post-switch failure requires rollback", failed_post_switch_goes_to_rollback_required)

    def rollback_mismatch_catastrophic() -> None:
        report = simulator.simulate_transaction(
            "hardened_elfldr",
            ["rollback_hash_mismatch"],
            prove_virtual_switch_model=True,
        )
        require(report["state"] == "BLOCKED", "rollback mismatch not blocked")
        require(report["risk"] == "CATASTROPHIC", "rollback risk not catastrophic")
        require(
            report["blockers"][0]["code"] == "ROLLBACK_VERIFY_FAILED",
            "rollback mismatch blocker changed",
        )

    check("rollback hash mismatch catastrophic blocked", rollback_mismatch_catastrophic)

    def complete_fault_suite() -> None:
        for component in simulator.COMPONENTS:
            suite = simulator.run_fault_suite(component)
            require(
                len(suite["fault_results"]) == len(simulator.FAULTS),
                f"{component}: fault suite incomplete",
            )
            require(
                len(suite["power_loss_results"])
                == len(simulator.POWER_LOSS_BOUNDARIES),
                f"{component}: power-loss suite incomplete",
            )
            require(suite["hardware_evidence"] is False, "hardware proof claimed")
            require(suite["device_write_performed"] is False, "device write claimed")
            require(
                suite["target_execution_performed"] is False,
                "target execution claimed",
            )
            require(suite["automatic_retry"] is False, "suite retry enabled")

    check("all declared faults injected for both components", complete_fault_suite)

    def simulator_has_no_io_or_target_surface() -> None:
        source = (
            root / "tools/simulate_phase09_transaction.py"
        ).read_text(encoding="utf-8")
        prohibited = (
            "import socket",
            "import subprocess",
            "import requests",
            "urllib",
            "ctypes",
            "os.system",
            ".write_text(",
            ".write_bytes(",
            "open(",
            "9021",
            "8084",
            "8085",
        )
        for token in prohibited:
            require(token not in source, f"simulator contains prohibited surface: {token}")

    check("simulator cannot perform filesystem network or target I/O", simulator_has_no_io_or_target_surface)

    def docs_keep_platform_unproven() -> None:
        docs = "\n".join(
            (root / relative).read_text(encoding="utf-8")
            for relative in (
                "docs/runtime/phase-0.9-anti-brick-threat-model.md",
                "docs/runtime/phase-0.9-installation-transaction-design.md",
                "docs/runtime/phase-0.9-recovery-and-rollback-contract.md",
            )
        )
        require(
            "BLOCKER: NO PROVEN POWER-LOSS-SAFE SWITCH" in docs,
            "power-loss-safe switch blocker missing",
        )
        require(
            "Host simulation is not hardware evidence" in docs
            or "Host simulation is not hardware" in docs,
            "host/hardware evidence boundary missing",
        )
        for forbidden in ("READY_FOR_INSTALLATION", "DEPLOYMENT_READY"):
            require(forbidden not in docs, f"forbidden positive status: {forbidden}")

    check("documentation never promotes host model to PS5 proof", docs_keep_platform_unproven)

    require(len(passed) >= 28, "too few Phase-0.9A guardrail tests")
    print(
        "Phase-0.9A host-only anti-brick tests passed: "
        f"{len(passed)} guardrails, {len(simulator.FAULTS)} fault types x "
        f"{len(simulator.COMPONENTS)} components, "
        f"{len(simulator.POWER_LOSS_BOUNDARIES)} power-loss boundaries; "
        "hardware evidence not claimed"
    )
    return 0


if __name__ == "__main__":
    raise SystemExit(main())