#!/usr/bin/env python3
"""驗證 Package / Docker 供應鏈 owner policy gate 維持只讀邊界。

本 guard 只讀取 repo 內的 package / Docker 供應鏈 baseline 與 owner
policy gate snapshot，不安裝套件、不連外查 CVE / license、不產生 lockfile、
不 pull / build / push image，也不修改 workflow、registry 或 production。
"""

from __future__ import annotations

import argparse
import json
from pathlib import Path
from typing import Any


EXPECTED_BASELINE_GAPS = [
    "python_lockfile_absent",
    "docker_base_images_not_all_digest_pinned",
    "docker_copy_from_images_not_all_digest_pinned",
    "compose_images_not_all_digest_pinned",
    "requirements_unpinned_entries_present",
]

EXPECTED_REQUEST_IDS = [
    "supply_chain_owner_policy:package_manager_lockfile_owner",
    "supply_chain_owner_policy:python_lockfile_policy",
    "supply_chain_owner_policy:requirements_pin_policy",
    "supply_chain_owner_policy:dockerfile_digest_pin_policy",
    "supply_chain_owner_policy:compose_image_digest_pin_policy",
    "supply_chain_owner_policy:cve_license_sbom_window",
]

EXPECTED_OWNER_FIELDS = [
    "package_manager_policy",
    "lockfile_owner",
    "python_lock_policy",
    "docker_base_image_policy",
    "compose_image_policy",
    "registry_owner",
    "cve_scan_window",
    "rollback_owner",
]

FALSE_BOUNDARY_KEYS = [
    "package_installation_allowed",
    "package_upgrade_allowed",
    "lockfile_write_allowed",
    "requirements_pin_change_allowed",
    "external_cve_lookup_allowed",
    "external_license_lookup_allowed",
    "sbom_generation_allowed",
    "docker_pull_allowed",
    "docker_build_allowed",
    "docker_push_allowed",
    "image_tag_change_allowed",
    "image_digest_pin_change_allowed",
    "registry_login_allowed",
    "workflow_modification_authorized",
    "production_deploy_authorized",
    "runtime_execution_authorized",
    "action_buttons_allowed",
    "secret_value_collection_allowed",
]


def fail(message: str) -> None:
    raise SystemExit(f"BLOCKED {message}")


def load_json(path: Path) -> dict[str, Any]:
    return json.loads(path.read_text(encoding="utf-8"))


def assert_equal(label: str, actual: Any, expected: Any) -> None:
    if actual != expected:
        fail(f"{label}: expected {expected!r}, got {actual!r}")


def assert_path_exists(root: Path, relative_path: str) -> None:
    if not (root / relative_path).exists():
        fail(f"path missing: {relative_path}")


def assert_text_contains(label: str, text: str, expected: str) -> None:
    if expected not in text:
        fail(f"{label}: missing {expected!r}")


def assert_false_boundaries(label: str, data: dict[str, Any]) -> None:
    boundaries = data.get("execution_boundaries", {})
    for key in FALSE_BOUNDARY_KEYS:
        assert_equal(f"{label}.execution_boundaries.{key}", boundaries.get(key), False)
    assert_equal(f"{label}.execution_boundaries.runtime_gate_count", boundaries.get("runtime_gate_count"), 0)
    assert_equal(f"{label}.execution_boundaries.not_authorization", boundaries.get("not_authorization"), True)


def validate_baseline(root: Path) -> dict[str, Any]:
    baseline_path = root / "docs/security/package-supply-chain-baseline.snapshot.json"
    assert_path_exists(root, "docs/security/package-supply-chain-baseline.snapshot.json")
    baseline = load_json(baseline_path)
    summary = baseline.get("summary", {})

    assert_equal("baseline.schema_version", baseline.get("schema_version"), "package_supply_chain_baseline_v1")
    assert_equal("baseline.status", baseline.get("status"), "repo_only_inventory_ready_needs_owner_policy")
    assert_equal("baseline.mode", baseline.get("mode"), "repo_snapshot_only_no_install_no_network_no_cve_scan")
    assert_equal("baseline.summary.package_json_count", summary.get("package_json_count"), 6)
    assert_equal("baseline.summary.pyproject_count", summary.get("pyproject_count"), 4)
    assert_equal("baseline.summary.requirements_file_count", summary.get("requirements_file_count"), 2)
    assert_equal("baseline.summary.requirements_unpinned_entry_count", summary.get("requirements_unpinned_entry_count"), 26)
    assert_equal("baseline.summary.python_lockfile_count", summary.get("python_lockfile_count"), 0)
    assert_equal("baseline.summary.docker_base_digest_pinned_count", summary.get("docker_base_digest_pinned_count"), 0)
    assert_equal(
        "baseline.summary.compose_digest_pinned_image_ref_count",
        summary.get("compose_digest_pinned_image_ref_count"),
        0,
    )
    assert_equal("baseline.summary.gap_count", summary.get("gap_count"), 5)
    assert_equal("baseline.gaps", baseline.get("gaps"), EXPECTED_BASELINE_GAPS)
    assert_equal("baseline.owner_response_received_count", summary.get("owner_response_received_count"), 0)
    assert_equal("baseline.owner_response_accepted_count", summary.get("owner_response_accepted_count"), 0)
    assert_equal("baseline.runtime_gate_count", summary.get("runtime_gate_count"), 0)
    assert_equal("baseline.action_button_count", summary.get("action_button_count"), 0)

    for relative_path in baseline.get("lockfiles", []):
        assert_path_exists(root, relative_path)
    for key in ["package_json_manifests", "pyproject_manifests", "requirements_files", "dockerfiles", "compose_files"]:
        for item in baseline.get(key, []):
            assert_path_exists(root, item["path"])
    return baseline


def validate_policy_gate(root: Path, baseline: dict[str, Any]) -> None:
    gate_path = root / "docs/security/package-supply-chain-owner-policy-gate.snapshot.json"
    doc_path = root / "docs/security/PACKAGE-SUPPLY-CHAIN-OWNER-POLICY-GATE.md"
    assert_path_exists(root, "docs/security/package-supply-chain-owner-policy-gate.snapshot.json")
    assert_path_exists(root, "docs/security/PACKAGE-SUPPLY-CHAIN-OWNER-POLICY-GATE.md")
    gate = load_json(gate_path)
    summary = gate.get("summary", {})

    assert_equal("policy_gate.schema_version", gate.get("schema_version"), "package_supply_chain_owner_policy_gate_v1")
    assert_equal("policy_gate.status", gate.get("status"), "draft_waiting_owner_policy_response")
    assert_equal("policy_gate.mode", gate.get("mode"), "repo_snapshot_policy_gate_no_install_no_network_no_cve_scan")
    assert_equal("policy_gate.baseline_ref", gate.get("baseline_ref"), "docs/security/package-supply-chain-baseline.snapshot.json")
    assert_equal("policy_gate.summary.baseline_gap_count", summary.get("baseline_gap_count"), baseline["summary"]["gap_count"])
    assert_equal("policy_gate.summary.owner_policy_request_count", summary.get("owner_policy_request_count"), 6)
    assert_equal("policy_gate.summary.c0_policy_request_count", summary.get("c0_policy_request_count"), 2)
    assert_equal("policy_gate.summary.write_capable_policy_request_count", summary.get("write_capable_policy_request_count"), 6)
    assert_equal("policy_gate.summary.required_owner_field_count", summary.get("required_owner_field_count"), 8)
    assert_equal("policy_gate.summary.reviewer_check_count", summary.get("reviewer_check_count"), 12)
    assert_equal("policy_gate.summary.blocked_action_count", summary.get("blocked_action_count"), 20)
    assert_equal("policy_gate.summary.owner_policy_request_sent_count", summary.get("owner_policy_request_sent_count"), 0)
    assert_equal("policy_gate.summary.owner_response_received_count", summary.get("owner_response_received_count"), 0)
    assert_equal("policy_gate.summary.owner_response_accepted_count", summary.get("owner_response_accepted_count"), 0)
    assert_equal("policy_gate.summary.owner_response_rejected_count", summary.get("owner_response_rejected_count"), 0)
    assert_equal("policy_gate.summary.runtime_gate_count", summary.get("runtime_gate_count"), 0)
    assert_equal("policy_gate.summary.action_button_count", summary.get("action_button_count"), 0)
    assert_equal("policy_gate.baseline_gaps", gate.get("baseline_gaps"), EXPECTED_BASELINE_GAPS)
    assert_equal("policy_gate.required_owner_fields", gate.get("required_owner_fields"), EXPECTED_OWNER_FIELDS)
    assert_equal("policy_gate.owner_policy_request_ids", [item["request_id"] for item in gate["owner_policy_requests"]], EXPECTED_REQUEST_IDS)

    for item in gate["owner_policy_requests"]:
        assert_equal(f"{item['request_id']}.status", item.get("status"), "draft_waiting_owner_policy_response")
        assert_equal(f"{item['request_id']}.required_owner_fields", item.get("required_owner_fields"), EXPECTED_OWNER_FIELDS)
        assert_equal(f"{item['request_id']}.request_sent", item.get("request_sent"), False)
        assert_equal(f"{item['request_id']}.owner_response_received", item.get("owner_response_received"), False)
        assert_equal(f"{item['request_id']}.owner_response_accepted", item.get("owner_response_accepted"), False)
        assert_equal(f"{item['request_id']}.runtime_gate_open", item.get("runtime_gate_open"), False)
        if not item.get("baseline_gap_refs"):
            fail(f"{item['request_id']}.baseline_gap_refs: expected non-empty list")
        for ref in item.get("evidence_refs", []):
            assert_path_exists(root, ref)
        if not item.get("blocked_until"):
            fail(f"{item['request_id']}.blocked_until: expected non-empty list")

    assert_false_boundaries("policy_gate", gate)

    doc = doc_path.read_text(encoding="utf-8")
    for text in [
        "owner policy gate",
        "Python lockfile",
        "requirements pinning",
        "Docker digest pinning",
        "CVE / license / SBOM",
        "runtime gate",
        "0 / false",
        "不 install、不 upgrade、不 pull image、不 build image、不 push image",
    ]:
        assert_text_contains("policy_gate.doc", doc, text)


def validate(root: Path) -> None:
    baseline = validate_baseline(root)
    validate_policy_gate(root, baseline)


def main() -> None:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "--root",
        default=Path(__file__).resolve().parents[2],
        type=Path,
        help="Repository root. Defaults to the current script's repository.",
    )
    args = parser.parse_args()
    validate(args.root.resolve())
    print("PACKAGE_SUPPLY_CHAIN_OWNER_POLICY_GUARD_OK")


if __name__ == "__main__":
    main()
