#!/usr/bin/env python3
"""檢查公開前端不可洩漏內網拓樸、原始 namespace 或工作視窗內容。"""

from __future__ import annotations

import argparse
import json
import re
import subprocess
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Any


PRIVATE_IP_PATTERN = re.compile(
    r"\b(?:10(?:\.\d{1,3}){3}|192\.168(?:\.\d{1,3}){2}|172\.(?:1[6-9]|2\d|3[01])(?:\.\d{1,3}){2})\b"
)
TAIPEI = timezone(timedelta(hours=8))

RETIRED_PUBLIC_TOPOLOGY_KEYS = {
    "NEXT_PUBLIC_HOST_IPS",
    "NEXT_PUBLIC_K8S_VIP_INFO",
}

ENV_EXAMPLE_PATHS = (
    Path("apps/web/.env.example"),
)

PUBLIC_SURFACE_ROOTS = (
    Path("apps/web/src"),
    Path("apps/web/messages"),
)

PUBLIC_SURFACE_SUFFIXES = {
    ".json",
    ".mdx",
    ".ts",
    ".tsx",
}

PUBLIC_SURFACE_FORBIDDEN_PATTERNS: tuple[tuple[str, re.Pattern[str]], ...] = (
    ("raw_personal_owner_namespace", re.compile(r"\bowenhytsai\b", re.IGNORECASE)),
    ("raw_external_owner_namespace", re.compile(r"\bnexu-io\b", re.IGNORECASE)),
    ("raw_blocked_waiting_state", re.compile(r"\bblocked_waiting_[A-Za-z0-9_]+\b")),
    ("raw_blockers_counter", re.compile(r"\bblockers=\d+\b")),
    ("codex_delegation_payload", re.compile(r"\bcodex_delegation\b", re.IGNORECASE)),
    ("codex_source_thread_id", re.compile(r"\bsource_thread_id\b", re.IGNORECASE)),
    ("approval_chat_phrase", re.compile(r"批准！")),
    ("work_window_plaintext", re.compile(r"工作視窗")),
    ("in_app_browser_transcript", re.compile(r"\bIn app browser\b", re.IGNORECASE)),
    ("codex_request_transcript", re.compile(r"\bMy request for Codex\b", re.IGNORECASE)),
    ("work_window_transcript", re.compile(r"\bwork window transcript\b", re.IGNORECASE)),
    ("internal_rfc1918_ip", PRIVATE_IP_PATTERN),
)

ALLOWED_PUBLIC_SURFACE_MATCHES: set[tuple[str, str]] = {
    ("apps/web/src/lib/api-client.ts", "work_window_transcript"),
    ("apps/web/src/app/[locale]/governance/tabs/automation-inventory-tab.tsx", "work_window_transcript"),
}


def _active_assignment(line: str) -> tuple[str, str] | None:
    stripped = line.strip()
    if not stripped or stripped.startswith("#") or "=" not in stripped:
        return None
    key, value = stripped.split("=", 1)
    return key.strip(), value.strip()


def _git_short_sha(root: Path) -> str:
    try:
        result = subprocess.run(
            ["git", "rev-parse", "--short", "HEAD"],
            cwd=root,
            check=True,
            capture_output=True,
            text=True,
        )
        return result.stdout.strip()
    except Exception:
        return "unknown"


def _iter_public_surface_files(root: Path) -> list[Path]:
    files: list[Path] = []
    for surface_root in PUBLIC_SURFACE_ROOTS:
        absolute_root = root / surface_root
        if not absolute_root.exists():
            continue
        for path in absolute_root.rglob("*"):
            if path.is_file() and path.suffix in PUBLIC_SURFACE_SUFFIXES:
                files.append(path)
    return sorted(files)


def _line_excerpt(line: str, start: int, end: int) -> str:
    prefix_start = max(start - 36, 0)
    suffix_end = min(end + 36, len(line))
    excerpt = line[prefix_start:suffix_end].strip()
    return excerpt if len(excerpt) <= 120 else excerpt[:117] + "..."


def build_report(root: Path, generated_at: str | None = None) -> dict[str, Any]:
    generated = generated_at or datetime.now(TAIPEI).isoformat(timespec="seconds")
    errors: list[str] = []
    env_violation_count = 0
    public_surface_files = _iter_public_surface_files(root)
    public_surface_matches: list[dict[str, Any]] = []
    public_surface_violations: list[dict[str, Any]] = []
    allowlisted_match_count = 0

    for relative_path in ENV_EXAMPLE_PATHS:
        path = root / relative_path
        if not path.exists():
            continue

        for line_number, line in enumerate(path.read_text(encoding="utf-8").splitlines(), start=1):
            assignment = _active_assignment(line)
            if assignment is None:
                continue

            key, value = assignment
            if key in RETIRED_PUBLIC_TOPOLOGY_KEYS:
                errors.append(f"{relative_path}:{line_number}: 已停用公開前端拓樸 env key：{key}")
                env_violation_count += 1
            if PRIVATE_IP_PATTERN.search(value):
                errors.append(f"{relative_path}:{line_number}: env 範例不得包含內網 IP：{key}")
                env_violation_count += 1
            if key.startswith("NEXT_PUBLIC_") and not value:
                errors.append(f"{relative_path}:{line_number}: NEXT_PUBLIC_* 範例需明確使用公網入口或安全預設值：{key}")
                env_violation_count += 1

    for path in public_surface_files:
        relative_path = path.relative_to(root).as_posix()
        text = path.read_text(encoding="utf-8", errors="replace")
        for line_number, line in enumerate(text.splitlines(), start=1):
            for pattern_id, pattern in PUBLIC_SURFACE_FORBIDDEN_PATTERNS:
                for match in pattern.finditer(line):
                    item = {
                        "path": relative_path,
                        "line": line_number,
                        "pattern_id": pattern_id,
                        "excerpt": _line_excerpt(line, match.start(), match.end()),
                    }
                    public_surface_matches.append(item)
                    if (relative_path, pattern_id) in ALLOWED_PUBLIC_SURFACE_MATCHES:
                        allowlisted_match_count += 1
                        continue
                    public_surface_violations.append(item)
                    errors.append(
                        f"{relative_path}:{line_number}: 前台敏感資訊命中 {pattern_id}，需改成脫敏名稱或遮罩器例外"
                    )

    return {
        "schema_version": "public_frontend_sensitive_surface_guard_v1",
        "generated_at": generated,
        "git_commit": _git_short_sha(root),
        "status": "pass" if not errors else "blocked",
        "mode": "repo_source_scan_no_runtime_no_secret_collection",
        "guarded_paths": [path.as_posix() for path in PUBLIC_SURFACE_ROOTS],
        "env_example_paths": [path.as_posix() for path in ENV_EXAMPLE_PATHS],
        "forbidden_patterns": [pattern_id for pattern_id, _ in PUBLIC_SURFACE_FORBIDDEN_PATTERNS],
        "allowed_matches": [
            {"path": path, "pattern_id": pattern_id}
            for path, pattern_id in sorted(ALLOWED_PUBLIC_SURFACE_MATCHES)
        ],
        "summary": {
            "env_example_file_count": sum(1 for path in ENV_EXAMPLE_PATHS if (root / path).exists()),
            "public_surface_file_count": len(public_surface_files),
            "forbidden_pattern_count": len(PUBLIC_SURFACE_FORBIDDEN_PATTERNS),
            "raw_match_count": len(public_surface_matches),
            "allowlisted_match_count": allowlisted_match_count,
            "violation_count": len(public_surface_violations),
            "env_violation_count": env_violation_count,
            "runtime_gate_count": 0,
            "action_button_count": 0,
        },
        "execution_boundaries": {
            "runtime_execution_authorized": False,
            "frontend_deploy_authorized": False,
            "production_deploy_authorized": False,
            "secret_value_collection_allowed": False,
            "raw_payload_storage_allowed": False,
            "internal_namespace_public_display_allowed": False,
            "work_window_transcript_public_display_allowed": False,
            "internal_ip_public_display_allowed": False,
            "action_buttons_allowed": False,
            "not_authorization": True,
        },
        "public_surface_matches": public_surface_matches,
        "public_surface_violations": public_surface_violations,
        "operator_interpretation": [
            "此 guard 只掃描 repo 內前端 source / messages 與 env example，不讀 production bundle、不部署、不收 secret。",
            "遮罩器中的 banned phrase 測試 pattern 允許列在 allowlist；產品文案、表格、API payload 與 i18n 不允許顯示 raw namespace、工作視窗逐字內容、raw blocker 狀態或內網 IP。",
            "violation_count 維持 0 才能視為 source-control 防洩漏檢查通過；仍不代表 production smoke、runtime approval 或 owner response accepted。",
        ],
    }


def validate(root: Path) -> None:
    report = build_report(root)
    errors = [
        f"{item['path']}:{item['line']}: 前台敏感資訊命中 {item['pattern_id']}，需改成脫敏名稱或遮罩器例外"
        for item in report["public_surface_violations"]
    ]
    if report["summary"]["env_violation_count"]:
        errors.append(f"env_violation_count={report['summary']['env_violation_count']}")

    if errors:
        raise SystemExit("BLOCKED public frontend sensitive surface guard:\n" + "\n".join(f"- {error}" for error in errors))


def main() -> None:
    parser = argparse.ArgumentParser(description="檢查公開前端不可洩漏內網拓樸、原始 namespace 或工作視窗內容")
    parser.add_argument("--root", default=".", help="repository root")
    parser.add_argument("--output", help="寫出 JSON 報告")
    parser.add_argument("--generated-at", help="固定報告時間，供 committed snapshot 使用")
    args = parser.parse_args()
    root = Path(args.root).resolve()
    report = build_report(root, args.generated_at)
    if args.output:
        output = Path(args.output)
        output.parent.mkdir(parents=True, exist_ok=True)
        output.write_text(json.dumps(report, ensure_ascii=False, indent=2, sort_keys=True) + "\n", encoding="utf-8")
    validate(root)
    summary = report["summary"]
    print(
        "OK public frontend sensitive surface guard "
        f"files={summary['public_surface_file_count']} "
        f"patterns={summary['forbidden_pattern_count']} "
        f"allowlisted={summary['allowlisted_match_count']} "
        f"violations={summary['violation_count']} "
        f"runtime_gate={summary['runtime_gate_count']}"
    )


if __name__ == "__main__":
    main()
