awoooi/scripts/security/awooop-controlled-automation-copy-guard.py

#!/usr/bin/env python3
"""Guard AwoooP public copy against legacy manual-gate defaults.

This guard is static and read-only. It prevents AwoooP pages and serialized
message payloads from reintroducing old manual-gate wording as the default
state for low / medium / high controlled automation.
"""

from __future__ import annotations

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


TEXT_FILES = [
    Path("apps/web/messages/zh-TW.json"),
    Path("apps/web/messages/en.json"),
]

AWOOOP_SOURCE_ROOT = Path("apps/web/src/app/[locale]/awooop")
ALERTS_ROUTE = AWOOOP_SOURCE_ROOT / "alerts" / "page.tsx"

FORBIDDEN_FRAGMENTS = [
    "待人工決策",
    "等待人工決策",
    "阻塞與人工閘門",
    "人工接手",
    "人工決策佇列",
    "人工關卡",
    "人工 Gate",
    "人工 gate",
    "人工閘門",
    "人工升級",
    "待 owner 複核",
    "未批准不會執行",
    "等 負責人審查",
    "owner review",
    "owner packet",
    "manual gate",
    "manual handoff",
]

REQUIRED_FRAGMENTS = [
    "待 AI 受控決策",
    "阻塞與 AI 受控隊列",
    "AI 處置包與工作項",
    "受控執行邊界",
    "受控授權閘門",
    "controlled gate",
    "controlled review",
    "AI 受控 Gate",
    "AI 受控補齊：NO_ACTION",
    "等待 AI 受控補齊",
    "受控驗收流程",
    "負責人脫敏證據收件",
    "Controlled Review",
    "AI 受控 review",
]

AWOOOP_LIVE_FORBIDDEN_FRAGMENTS = [
    "人工 Gate",
    "人工介入",
    "待人工",
    "等待人工",
    "人工回覆",
    "人工收件",
    "人工判斷",
    "人工交接",
    "人工方案",
    "人工確認無需動作",
    "人工：{human}",
    "需要人工={human}",
    "人工覆核",
    "人工檢查",
    "Owner review",
    "Owner Review",
    "負責人審查",
]

AWOOOP_ALLOWED_LEGACY_PATH_PREFIXES = [
    "awooop.approvals.legacyHitl.",
]


def _iter_guarded_files(root: Path) -> list[Path]:
    files = [root / path for path in TEXT_FILES]
    source_root = root / AWOOOP_SOURCE_ROOT
    files.extend(sorted(source_root.rglob("*.tsx")))
    return files


def _collect_awooop_message_violations(path: Path, root: Path) -> list[str]:
    data = json.loads(path.read_text(encoding="utf-8"))
    awooop = data.get("awooop")
    if not isinstance(awooop, dict):
        return [f"{path.relative_to(root)}: missing awooop namespace"]

    violations: list[str] = []

    def walk(value: Any, parts: list[str]) -> None:
        if isinstance(value, dict):
            for key, child in value.items():
                walk(child, [*parts, key])
            return
        if isinstance(value, list):
            for index, child in enumerate(value):
                walk(child, [*parts, str(index)])
            return
        if not isinstance(value, str):
            return

        dotted = ".".join(parts)
        if any(dotted.startswith(prefix) for prefix in AWOOOP_ALLOWED_LEGACY_PATH_PREFIXES):
            return
        for fragment in AWOOOP_LIVE_FORBIDDEN_FRAGMENTS:
            if fragment in value:
                relative = path.relative_to(root)
                violations.append(f"{relative}:{dotted}: forbidden live AwoooP copy {fragment!r}")

    walk(awooop, ["awooop"])
    return violations


def _collect_forbidden_line_violations(path: Path, root: Path, text: str) -> list[str]:
    violations: list[str] = []
    for line_number, line in enumerate(text.splitlines(), start=1):
        for fragment in FORBIDDEN_FRAGMENTS:
            if fragment in line:
                relative = path.relative_to(root)
                violations.append(f"{relative}:{line_number}: forbidden {fragment!r}")
    return violations


def validate(root: Path) -> None:
    root = root.resolve()
    violations: list[str] = []
    guarded_text = []

    for path in _iter_guarded_files(root):
        if not path.exists():
            violations.append(f"{path.relative_to(root)}: missing guarded file")
            continue
        text = path.read_text(encoding="utf-8")
        guarded_text.append(text)
        if path.name.endswith(".json"):
            violations.extend(_collect_awooop_message_violations(path, root))
        violations.extend(_collect_forbidden_line_violations(path, root, text))

    alerts_route = root / ALERTS_ROUTE
    if not alerts_route.exists():
        violations.append(f"{ALERTS_ROUTE}: missing AwoooP Alerts route")
    else:
        route_text = alerts_route.read_text(encoding="utf-8")
        if "#ai-alert-card-delivery-readback" not in route_text:
            violations.append(f"{ALERTS_ROUTE}: missing ai-alert-card-delivery-readback redirect target")

    combined_text = "\n".join(guarded_text)
    for fragment in REQUIRED_FRAGMENTS:
        if fragment not in combined_text:
            violations.append(f"messages/source: missing required controlled automation text {fragment!r}")

    if violations:
        formatted = "\n".join(violations[:40])
        raise SystemExit(f"BLOCKED awooop_controlled_automation_copy_guard:\n{formatted}")


def main() -> int:
    parser = argparse.ArgumentParser(description="Validate AwoooP controlled automation public copy.")
    parser.add_argument("--root", default=".")
    args = parser.parse_args()
    validate(Path(args.root))
    print("AWOOOP_CONTROLLED_AUTOMATION_COPY_GUARD_OK")
    return 0


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