#!/usr/bin/env python3
"""密钥泄漏扫描（检出树 → 退出码）。

为什么自己写而不是用 gitleaks：
  · 作业容器里没有 gitleaks，本机也没有；在 CI 运行时联网下载一个二进制，
    等于把"扫描器本身"变成供应链风险，而且违反"工具缺失直接失败、不在运行中静默安装"。
  · 本仓不需要把源码发给任何外部服务（正本 §7：不得发送项目源码给未授权外部服务）。
  ⚠️ 代价：规则集比 gitleaks 小。这是**已知缺口**，不是等价替代（见 README/交付说明）。

纪律：
  1. **只打印位置与类别，绝不打印命中的值**（正本 §4.5）。掩码也不打 ——
     掩码 + 上下文足以在聊天记录里复原短密钥。
  2. 扫描范围 = `git ls-files`（检出树里的**被跟踪文件**）。未跟踪的 `.env`、
     `secrets/`、`node_modules/` 不在范围里 —— 它们要么被忽略，要么根本不该在检出树里。
  3. 每次运行先跑 `--self-test`（用合成密钥夹具 + 良性对照）：
     一个不再命中的扫描器比没有扫描器更糟，它会给人"已经查过"的错觉（正本 §0-2）。

用法：
    python3 scripts/check_secrets.py                # 扫描 + 先跑自检
    python3 scripts/check_secrets.py --no-self-test # 只扫描（调试用）
    python3 scripts/check_secrets.py --self-test    # 只跑自检
"""
from __future__ import annotations

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

ROOT = Path(__file__).resolve().parent.parent

# ── 规则集：每条都必须能定位到具体文件/行，且必须能被 --self-test 证明"真的会命中" ──
# 形状：(rule_id, 类别, 正则, 密钥分组名或 None)
# 「密钥分组」= 命中里"真的是秘密"的那一段，用来判断它是不是占位符 / 模板表达式。
# 没有分组的规则（私钥标记、固定前缀的令牌）整段都算秘密。
RULES: list[tuple[str, str, re.Pattern[str], str | None]] = [
    ("private-key", "私钥块",
     re.compile(r"-----BEGIN (?:RSA |EC |DSA |OPENSSH |PGP )?PRIVATE KEY-----"), None),
    ("aws-ak", "AWS Access Key ID", re.compile(r"\b(?:AKIA|ASIA)[0-9A-Z]{16}\b"), None),
    ("aliyun-ak", "阿里云 AccessKey ID", re.compile(r"\bLTAI[0-9A-Za-z]{12,}\b"), None),
    ("forge-token", "代码托管平台令牌",
     re.compile(r"\b(?:gh[pousr]_[A-Za-z0-9]{30,}|glpat-[A-Za-z0-9_\-]{20,})\b"), None),
    ("jwt", "JWT",
     re.compile(r"\beyJ[A-Za-z0-9_\-]{10,}\.[A-Za-z0-9_\-]{10,}\.[A-Za-z0-9_\-]{10,}\b"), None),
    # 口令段只收"干净字符"：`***`（脱敏）、`{u.quote(password)}`（模板表达式）、
    # `$PGPASSWORD`（变量）都不该命中 —— 那三种在 scripts/*.sh 里到处都是，收了就是永远的假红。
    ("pg-url", "带口令的数据库连接串",
     re.compile(r"\bpostgres(?:ql)?://[A-Za-z0-9_.\-]+:(?P<secret>[A-Za-z0-9+/=_\-]{6,})@"), "secret"),
    # 前缀不写 `\b`：`DINGTALK_APP_SECRET` 这类"带前缀的名字"才是最常见的写法，
    # 而 `_` 是 word 字符，`\b` 会把它们全部漏掉。
    ("assigned-secret", "被赋值的密钥字面量",
     re.compile(r"(?i)\b[A-Za-z0-9_]*(?:api[_-]?key|app[_-]?secret|client[_-]?secret|access[_-]?key"
                r"|secret|token|passwd|password)\b\s*[:=]\s*[\"'](?P<secret>[A-Za-z0-9+/=_\-]{16,})[\"']"), "secret"),
]

# 占位符：整段值就是这些形状时不算密钥（否则模板 / 示例文件会把门禁撑成永远的假红）
PLACEHOLDER = re.compile(
    r"(?i)^(?:"
    r"<[^>]*>|\{\{[^}]*\}\}|\$\{[^}]*\}|\$[A-Za-z_][A-Za-z0-9_]*|"
    r"x{3,}|\.{3,}|\*{3,}|-+|_+|"
    r"(?:your|my|some|test|dummy|example|sample|placeholder|fake|redacted|changeme|change[_-]?me|"
    r"secret|token|password|api[_-]?key|none|null|todo|tbd)[A-Za-z0-9_\-]*"
    r")$"
)

TEXT_EXT = {
    ".go", ".py", ".ts", ".tsx", ".js", ".jsx", ".mjs", ".cjs", ".json", ".yaml", ".yml",
    ".md", ".sh", ".sql", ".css", ".html", ".env", ".example", ".txt", ".toml", ".ini",
    ".cfg", ".conf", ".properties", ".xml", ".tsv", ".csv",
}
MAX_BYTES = 2_000_000


def tracked_files() -> list[str]:
    """被跟踪的文件（含未跟踪但已 add 的）。检出树里没有的东西不扫。"""
    out = subprocess.run(["git", "ls-files"], cwd=ROOT, capture_output=True, text=True)
    if out.returncode != 0:
        print("❌ 拿不到 git ls-files —— 扫描范围无法确定，直接失败", file=sys.stderr)
        sys.exit(2)
    return [l for l in out.stdout.splitlines() if l.strip()]


def looks_textual(path: Path) -> bool:
    if path.suffix.lower() not in TEXT_EXT and path.name not in {".env.example", "Makefile", "Dockerfile"}:
        return False
    try:
        if path.stat().st_size > MAX_BYTES:
            return False
        return b"\x00" not in path.read_bytes()[:4096]
    except OSError:
        return False


def mask_scan(text: str) -> list[tuple[str, str, int]]:
    """返回 [(rule_id, 类别, 行号)]。**不返回命中的值**。"""
    hits = []
    for rule_id, label, rx, secret_group in RULES:
        for m in rx.finditer(text):
            value = (m.group(secret_group) if secret_group else m.group(0)) or ""
            if PLACEHOLDER.match(value.strip().strip("\"'")):
                continue
            hits.append((rule_id, label, text.count("\n", 0, m.start()) + 1))
    return hits


def scan(files: list[str]) -> list[tuple[str, str, int]]:
    found = []
    for rel in files:
        p = ROOT / rel
        if not looks_textual(p):
            continue
        try:
            text = p.read_text(encoding="utf-8", errors="replace")
        except OSError:
            continue
        for rule_id, label, line in mask_scan(text):
            found.append((rel, f"{rule_id}（{label}）", line))
    return found


# ── 反向自检：合成密钥必须命中，良性对照必须干净 ──
def self_test() -> int:
    problems: list[str] = []
    # 值都在运行期拼出来：否则本文件自己就会成为一条真命中
    synthetic = {
        "aws-ak": "AKIA" + "Q7ZP2M4XR9TLB6VW",
        "aliyun-ak": "LTAI" + "5tG8kQ2mZx7Vb3Np",
        "forge-token": "ghp_" + "a1B2c3D4e5F6g7H8i9J0k1L2m3N4o5P6q7R8",
        "jwt": "eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiIxIn0." + "s1gn4tur3Pad1234567890ab",
        "pg-url": "postgres://kg_admin:" + "hunter2xyz" + "@db.internal:5432/kg",
        "assigned-secret": 'DINGTALK_APP_SECRET = "' + "9f2a7c1e4b8d3065" + '"',
        # 拼起来写：本机的 pre-commit 钩子会拦"整段私钥头"的字面量 —— 那是**对的**，
        # 所以不绕过它，而是让本文件里不出现那段连续文本（扫描逻辑本身不受影响）。
        "private-key": "-----BEGIN " + "RSA PRIVATE KEY-----",
    }
    benign = "\n".join([
        'token = os.environ["CI_BRIDGE_TOKEN"]',
        'api_key: ${DOBOTAI_API_KEY}',
        'password = "<your-password>"',
        'X-CI-Token: ${BRIDGE_TOKEN}',
        'const url = `${API_BASE}/healthz`',
        'PGURL=postgres://user@host/db',
        'self.assertEqual(result.Skipped, 1)',
        # 下面三条都是**真实误报过**的形状（scripts/*.sh 与 Go 夹具），锁进对照防回归：
        'print(f"postgres://{u.quote(user)}:{u.quote(password)}@{host}:{port}/{name}")',
        '"postgres://governance:***@postgres/governance"',
        'DINGTALK_APP_SECRET = "${DINGTALK_APP_SECRET}"',
    ])
    with tempfile.TemporaryDirectory() as d:
        for rule_id, value in synthetic.items():
            probe = Path(d) / "probe.txt"
            probe.write_text(f"setting {rule_id} here\n{value}\n", encoding="utf-8")
            hit_ids = {r for r, _, _ in mask_scan(probe.read_text(encoding="utf-8"))}
            if not hit_ids:
                problems.append(f"合成夹具没有任何规则命中（{rule_id}）—— 这条规则是死的")
            elif rule_id not in hit_ids:
                problems.append(f"合成夹具命中的是 {sorted(hit_ids)}，而不是 {rule_id} —— 规则串了")
        clean = mask_scan(benign)
        if clean:
            problems.append(f"良性对照被误报：{clean} —— 门禁会因为正常代码变红（假红）")
    if problems:
        print("❌ 密钥扫描器自检失败：", file=sys.stderr)
        for p in problems:
            print(f"   · {p}", file=sys.stderr)
        return 1
    print(f"✅ 密钥扫描器自检通过：{len(synthetic)} 条合成密钥全部命中，良性对照零误报")
    return 0


def main(argv: list[str]) -> int:
    ap = argparse.ArgumentParser(add_help=True)
    ap.add_argument("--self-test", action="store_true", help="只跑反向自检")
    ap.add_argument("--no-self-test", action="store_true", help="跳过自检（调试用，不算全绿）")
    args = ap.parse_args(argv)

    if args.self_test:
        return self_test()
    if not args.no_self_test and self_test() != 0:
        return 1

    files = tracked_files()
    found = scan(files)
    print(f"── 密钥扫描 ── 范围：{len(files)} 个被跟踪文件（只打印位置与类别，不打印值）")
    if found:
        print("❌ 疑似密钥泄漏：", file=sys.stderr)
        for rel, label, line in sorted(found):
            print(f"   {rel}:{line}  {label}", file=sys.stderr)
        print("   处理：把值挪到 Gitea Secrets / 目标机文件；已入库的要轮换密钥，"
              "光删提交不够（历史里还在）。", file=sys.stderr)
        return 1
    print("✅ 未发现密钥泄漏")
    return 0


if __name__ == "__main__":
    raise SystemExit(main(sys.argv[1:]))