#!/usr/bin/env python3
"""测试通道守门（M169，r2 复审后重写）：**官方发布产物里不许有 `update-test-source` feature**。

那条 feature 让 `core::update::job::default_source` 可以被两个环境变量指向任意本地目录，而更新链路
的终点是**替换用户盘上的 exe**。默认关 + 编译期不存在是唯一可靠的边界，所以「谁也没在流水线里把它
打开」必须是自动断言，不能靠注释和记性。

四条断言，任何一条不成立即 exit 1（CI 里红在 `.github/workflows/release-guard.yml`）：

1. **workflow 与复合 action 里不许打开它**——唯一例外是 `.github/workflows/ci.yml` 里名为
   `update-e2e` 的那一步，且该步必须同时带 `--test update_e2e_test`（见 `ALLOWED_STEP`）。
   文件名 / 步骤名 / 命令行**三者任一不符即红**（r1 复审口径；job 名不参与判据——改 job 名不该
   解锁新能力，能力只取决于那一行命令）。为什么要开这个口子（r1 复审 P2-3）：E2E 不带 feature 时
   两个目标是 0 例，而「任何 workflow 都不许出现这个字面量」会把唯一合法的 CI 调用点也堵死——
   守卫强度与回归可见性会互相吃掉。口子开得极窄：一个文件、一个步骤名、一个参数。
2. **不许「打开所有 feature」**：`--all-features` 与 cargo hack 的 `--feature-powerset` /
   `--each-feature` / `--all` 都会把测试通道一起打开。
3. **不许设成默认**：任何 `Cargo.toml` 里它只许作为**自己的声明**出现；进 `default` 或别的
   feature 的依赖数组即红（这条专治「扫 workflow 看不见」的开法）。
4. **扫描面逐项列举**（`WORKFLOW_GLOBS` / `SCRIPT_GLOBS` / `BUILD_ENTRY_GLOBS`）：含
   `.github/actions/**`、`packaging/**` 的**子目录**脚本、`Makefile`/`justfile`/`Taskfile` 这类第二
   构建入口（r1 复审 P2-2 的漏检来源）。扩扫描面 = 往那三个元组加一行 + 补一条 self-test 向量。

注释剥离是**引号感知**的：`- run: echo "#"; cargo build --features update-test-source` 里第一个
`#` 在引号内，不是注释起点（粗暴 `#.*$` 会把真命令一起吃掉，也是 P2-2 漏检之一）。

# 抓得住什么 / 抓不住什么

| 抓得住 | 抓不住 |
|---|---|
| workflow / 复合 action / 打包脚本 / Makefile 里打开它（含 `--features=`、引号、多行续行写法） | 变量间接引用（`--features "$FEAT"`），或 feature 名写在守门不扫的文件里 |
| `--all-features` / `cargo hack --feature-powerset|--each-feature|--all` 一把梭 | 有人把 feature 改名后再接回发布：名字一变就脱离扫描表（改名要过 review，但守门看不见） |
| 把 feature 塞进 `default` 或别的 feature 的依赖数组 | `build.rs` 里 spawn `cargo`、或第三方工具链自带的构建入口 |
| allowlist 被放宽：文件名 / 步骤名 / `--test` 参数任一不符 | allowlist 被滥用：别的步骤改名成 `update-e2e` 且带 `--test update_e2e_test` 却干别的事（扫描只看字符串；merge review 会看到 ci.yml 那一行） |
| 大小写/下划线变体不算漏检：cargo feature 名敏感，写错直接编不过（r1 复审同判） | |

`--self-test` 是这条守门的变异自测：每条断言的**真实回归**注入临时副本，断言该红的红、该绿的绿
（尤其 allowlist 六个向量与 P2-2 的七条绕过）。守门自己抓不住回归时它会先红。
"""

from __future__ import annotations

import argparse
import re
import sys
import tempfile
from pathlib import Path

#: 被守的 feature 名（改名就得同步改本文件与各 `Cargo.toml`）。
FEATURE = "update-test-source"

#: 仓库根（本文件在 `packaging/` 下）与本文件自身（扫描时跳过：它必须写出 feature 名）。
ROOT = Path(__file__).resolve().parent.parent
GUARD = Path(__file__).resolve()

#: 扫描面（逐个列举，不用 `**` 全仓：那会把 `target/` 拖进来）。
WORKFLOW_GLOBS = (
    ".github/workflows/**/*.yml",
    ".github/workflows/**/*.yaml",
    ".github/actions/**/*.yml",
    ".github/actions/**/*.yaml",
)
SCRIPT_GLOBS = ("packaging/**/*.py", "packaging/**/*.sh", "packaging/**/*.ps1")
BUILD_ENTRY_GLOBS = (
    "Makefile",
    "justfile",
    "Justfile",
    "Taskfile.yml",
    "Taskfile.yaml",
    "crates/*/Makefile",
    "packaging/Makefile",
)
MANIFEST_GLOBS = ("Cargo.toml", "crates/*/Cargo.toml")

#: 唯一允许出现该 feature 的地方：**一个文件 + 一个步骤名 + 一个必须同时在场的参数**。
ALLOWED_STEP = {
    "workflow": ".github/workflows/ci.yml",
    "name": "update-e2e",
    "must_contain": "--test update_e2e_test",
}

#: 「打开全部 feature」的开关（后三条是 cargo hack 的走法）。
ALL_FEATURES = (
    re.compile(r"--all-features\b"),
    re.compile(r"--feature-powerset\b"),
    re.compile(r"--each-feature\b"),
    re.compile(r"\bcargo\s+hack\b[^\n]*\s--all\b"),
)

#: Cargo.toml 里「自己的声明」长这样：行首 `update-test-source = [`。
DECLARATION = re.compile(rf"^\s*{re.escape(FEATURE)}\s*=\s*\[")
#: workflow 的一个步骤从 `- name: <x>` 开始。
STEP_NAME = re.compile(r"^\s*-\s*name:\s*(.+?)\s*$")


def strip_comment(line: str) -> str:
    """引号感知地剥注释：`#` 只有「行首或空白之后」且不在引号内时才开始注释。

    YAML 的单引号 `''`、双引号 `\\` 转义这里不模拟——本守门只判断「引号有没有闭合」，够用，
    且不会把 `echo "#"; cargo build --features …` 当成注释。
    """
    quote = None
    for index, char in enumerate(line):
        if quote is not None:
            if char == quote:
                quote = None
        elif char in "\"'":
            quote = char
        elif char == "#" and (index == 0 or line[index - 1].isspace()):
            return line[:index]
    return line


def dangerous_flag(code: str) -> str | None:
    """这一行有没有「打开全部 feature」的开关（返回命中的开关，便于写清原因）。"""
    for pattern in ALL_FEATURES:
        match = pattern.search(code)
        if match:
            return match.group(0)
    return None


def _lines(path: Path) -> list[tuple[int, str]]:
    return list(enumerate(path.read_text(encoding="utf-8").splitlines(), start=1))


def split_steps(lines: list[tuple[int, str]]) -> list[tuple[str | None, list[tuple[int, str]]]]:
    """按 `- name: <step>` 把 workflow 切成「步骤名 + 该步骤的行」。

    没有 YAML 解析器（stdlib 不带、也不许引第三方），用「最近的 `- name:` 行」当步骤边界：GitHub
    workflow 就是这个形状，而本守门只关心「某个 feature 字面量落在哪一步里」。第一个 step 之前的
    行归到 `None` 块（含 `on:` / `env:` 这类全局段）。
    """
    blocks: list[tuple[str | None, list[tuple[int, str]]]] = []
    name: str | None = None
    body: list[tuple[int, str]] = []
    for number, line in lines:
        match = STEP_NAME.match(line)
        if match:
            if body:
                blocks.append((name, body))
            name = match.group(1).strip("'\"")
            body = []
        body.append((number, line))
    if body:
        blocks.append((name, body))
    return blocks


def _allowlist_reason(where: str, name: str | None, body: list[tuple[int, str]]) -> str | None:
    """`None` = 这一步被放行；否则返回「为什么红」的一句话。"""
    if where != ALLOWED_STEP["workflow"]:
        return (
            f"`{FEATURE}` 只许出现在 `{ALLOWED_STEP['workflow']}` 的 "
            f"`{ALLOWED_STEP['name']}` 步骤里（这里是 `{where}`）"
        )
    if name != ALLOWED_STEP["name"]:
        return (
            f"`{where}` 里只有名为 `{ALLOWED_STEP['name']}` 的步骤可以用 `{FEATURE}`"
            f"（这一步叫 `{name}`）"
        )
    joined = "\n".join(strip_comment(line) for _number, line in body)
    if ALLOWED_STEP["must_contain"] not in joined:
        return f"`{where}` 的 `{ALLOWED_STEP['name']}` 步骤少了 `{ALLOWED_STEP['must_contain']}`"
    return None


def scan_workflow(path: Path, root: Path) -> list[str]:
    """扫 workflow / 复合 action：feature 只许出现在 allowlist 点名的那一步里。"""
    problems: list[str] = []
    where = path.relative_to(root).as_posix()
    for name, body in split_steps(_lines(path)):
        code_lines = [(number, strip_comment(line)) for number, line in body]
        for number, code in code_lines:
            flag = dangerous_flag(code)
            if flag:
                problems.append(
                    f"::error file={where}:{number}::{where} 用了 `{flag}`："
                    f"它会把 `{FEATURE}` 一起打开"
                )
        hits = [(number, code) for number, code in code_lines if FEATURE in code]
        if not hits:
            continue
        reason = _allowlist_reason(where, name, body)
        if reason is None:
            continue
        for number, _code in hits:
            problems.append(
                f"::error file={where}:{number}::{reason}；"
                "官方发布产物里不许有测试通道（它能把更新源指向任意本地目录）"
            )
    return problems


def scan_code_file(path: Path, root: Path) -> list[str]:
    """扫脚本 / 第二构建入口：没有任何 allowlist，字面量或开关出现即红。"""
    problems: list[str] = []
    where = path.relative_to(root).as_posix()
    for number, line in _lines(path):
        code = strip_comment(line)
        if not code:
            continue
        if FEATURE in code:
            problems.append(
                f"::error file={where}:{number}::{where} 打开了 `{FEATURE}`："
                "官方发布产物里不许有测试通道（它能把更新源指向任意本地目录）"
            )
        flag = dangerous_flag(code)
        if flag:
            problems.append(
                f"::error file={where}:{number}::{where} 用了 `{flag}`："
                f"它会把 `{FEATURE}` 一起打开"
            )
    return problems


def scan_manifest(path: Path, root: Path) -> list[str]:
    """扫 Cargo.toml：feature 只许出现在自己的声明行上（不许进 default / 别的 feature）。"""
    problems: list[str] = []
    where = path.relative_to(root).as_posix()
    for number, line in _lines(path):
        code = strip_comment(line).strip()
        if FEATURE not in code or DECLARATION.match(code):
            continue
        problems.append(
            f"::error file={where}:{number}::{where} 把 `{FEATURE}` 写进了别的 feature 列表："
            "只许它是自己的声明（`update-test-source = [...]`），不许进 `default` 或别的 feature"
        )
    return problems


def scan(root: Path) -> list[str]:
    """扫整个仓库，返回所有问题（空 = 绿）。"""
    problems: list[str] = []
    for pattern, scanner in (
        (WORKFLOW_GLOBS, scan_workflow),
        (SCRIPT_GLOBS, scan_code_file),
        (BUILD_ENTRY_GLOBS, scan_code_file),
        (MANIFEST_GLOBS, scan_manifest),
    ):
        for path in sorted({path for glob in pattern for path in root.glob(glob)}):
            if scanner is scan_code_file and path.resolve() == GUARD:
                continue
            try:
                problems += scanner(path, root)
            except OSError as error:
                problems.append(f"::error file={path}::读不到 {path}：{error}")
    return problems


# ────────────────────────── 变异自测 ──────────────────────────

def _wf(*steps: str) -> str:
    """拼一个最小 workflow（job 名与真实 `ci.yml` 一致；**job 名不参与判据**——判据锚在步骤名上，
    改 job 名不该解锁新能力。这里用真名只是让自测基线长成真文件的形状）。"""
    return "jobs:\n  update-e2e:\n    steps:\n" + "".join(steps)


#: 允许的那一步（基线与若干向量共用它）。
ALLOWED_STEP_YAML = (
    "      - uses: actions/checkout@v4\n"
    "      - name: update-e2e\n"
    "        if: runner.os == 'Windows'\n"
    "        run: |\n"
    f"          cargo test --release -p unlucky-cli --features {FEATURE} \\\n"
    "            --test update_e2e_test --test update_e2e_tamper_test\n"
)

BASELINE: dict[str, str] = {
    ".github/workflows/ci.yml": _wf(*[ALLOWED_STEP_YAML]),
    ".github/workflows/release-build.yml": _wf("      - run: cargo build --release\n"),
    ".github/workflows/release-publish.yml": _wf("      - run: echo publish\n"),
    "packaging/make_artifacts.py": "def build():\n    return 'cargo build --release'\n",
    "packaging/tests/build.sh": "#!/bin/sh\ncargo build --release\n",
    "Makefile": "build:\n\tcargo build --release\n",
    "Cargo.toml": "[workspace]\nmembers = []\n",
    "crates/core/Cargo.toml": (
        f"[features]\n{FEATURE} = []\n# 注释里提到 update-test-source 不算违规\n"
    ),
}

_LEAK = f"      - run: cargo build --release --features {FEATURE}\n"
_HACK = "      - run: cargo hack {mode} build\n"

#: 向量 =（标签、覆盖/新增文件、期望）。期望绿的两条是**防误杀**。
VECTORS: list[tuple[str, dict[str, str], str]] = [
    ("allowlist ①：ci.yml 的 update-e2e 一步", {}, "green"),
    ("allowlist ②：同样的步骤放进 release-build.yml",
     {".github/workflows/release-build.yml": _wf(*[ALLOWED_STEP_YAML])}, "red"),
    ("allowlist ③：步骤名改成 update-e2e-r2",
     {".github/workflows/ci.yml": _wf(ALLOWED_STEP_YAML.replace("name: update-e2e", "name: update-e2e-r2"))},
     "red"),
    ("allowlist ④：少了 --test update_e2e_test",
     {".github/workflows/ci.yml": _wf(ALLOWED_STEP_YAML.replace("--test update_e2e_test ", ""))}, "red"),
    ("allowlist ⑤：发布工作流里另起一步带该 feature",
     {".github/workflows/release-publish.yml": _wf("      - name: leak\n", _LEAK)}, "red"),
    ("allowlist ⑥：ci.yml 里另一步（不叫 update-e2e）带该 feature",
     {".github/workflows/ci.yml": _wf(*[ALLOWED_STEP_YAML, "      - name: another\n", _LEAK])}, "red"),
    ("P2-2 ①：`#` 在引号内、真命令在后面",
     {".github/workflows/release-build.yml": _wf(f'      - run: echo "#"; cargo build --features {FEATURE}\n')},
     "red"),
    ("P2-2 ②：cargo hack --feature-powerset",
     {".github/workflows/release-build.yml": _wf(_HACK.format(mode="--feature-powerset"))}, "red"),
    ("P2-2 ③：cargo hack --each-feature",
     {".github/workflows/release-build.yml": _wf(_HACK.format(mode="--each-feature"))}, "red"),
    ("P2-2 ④：packaging 子目录脚本",
     {"packaging/tests/build.sh": f"#!/bin/sh\ncargo build --features {FEATURE}\n"}, "red"),
    ("P2-2 ⑤：.github/actions/** 复合 action",
     {".github/actions/rebuild/action.yml":
      "runs:\n  using: composite\n  steps:\n"
      f"    - run: cargo build --features {FEATURE}\n      shell: bash\n"}, "red"),
    ("P2-2 ⑥：Makefile 里的第二构建入口",
     {"Makefile": f"build:\n\tcargo build --release --features {FEATURE}\n"}, "red"),
    ("P2-2 ⑦：workflow 子目录",
     {".github/workflows/nested/extra.yml": _wf(_LEAK)}, "red"),
    ("P2-2 ⑧（防误杀）：注释里提到该 feature",
     {".github/workflows/release-build.yml":
      _wf(f"      # 这里不许加 {FEATURE}\n", "      - run: cargo build --release\n")}, "green"),
    ("Cargo ①：default 把 feature 带上",
     {"crates/core/Cargo.toml": BASELINE["crates/core/Cargo.toml"] + 'default = ["update-test-source"]\n'},
     "red"),
    ("Cargo ②：别的 feature 依赖它",
     {"crates/core/Cargo.toml": BASELINE["crates/core/Cargo.toml"] + 'release = ["update-test-source"]\n'},
     "red"),
]


def _scan_case(files: dict[str, str]) -> list[str]:
    with tempfile.TemporaryDirectory(prefix="ull-test-source-guard-") as temp:
        root = Path(temp)
        for relative, text in files.items():
            path = root / relative
            path.parent.mkdir(parents=True, exist_ok=True)
            path.write_text(text, encoding="utf-8")
        return scan(root)


def self_test() -> int:
    failures: list[str] = []
    baseline = _scan_case(BASELINE)
    if baseline:
        failures.append(f"基线应当绿，实际红了：{baseline}")
    for label, changes, expect in VECTORS:
        files = dict(BASELINE)
        files.update(changes)
        hits = _scan_case(files)
        if expect == "red" and not hits:
            failures.append(f"{label}：注入了真实回归，守门却没红（抓不住）")
        if expect == "green" and hits:
            failures.append(f"{label}：应当放行，实际红了（误杀）：{hits}")
    if failures:
        for failure in failures:
            print(f"::error::{failure}", file=sys.stderr)
        return 1
    reds = sum(1 for _label, _changes, expect in VECTORS if expect == "red")
    print(f"OK: 守门自测通过（{reds} 条注入全红，{len(VECTORS) - reds + 1} 条基线/放行保持绿）")
    return 0


def main() -> int:
    parser = argparse.ArgumentParser(description=f"守门：官方发布产物里不许有 {FEATURE}")
    parser.add_argument("--self-test", action="store_true", help="注入变异自测（该红的红、该绿的绿）")
    parser.add_argument("--root", type=Path, default=ROOT, help="仓根（自测用）")
    args = parser.parse_args()

    if args.self_test:
        return self_test()

    problems = scan(args.root)
    if problems:
        for problem in problems:
            print(problem, file=sys.stderr)
        return 1
    print(f"OK: 流水线与打包脚本都没有打开 {FEATURE}（默认关，发布产物不含测试通道）")
    return 0


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