fifth-domain/zero-point/core-channel/revive-guard/pre-receive-guard.py

#!/usr/bin/env python3
"""
光湖语言系统 · 推送守门人 · 敏感信息自动打码
Guanghu Language System · Pre-Receive Guard

部署位置: 当前承载目标仓库的 Forgejo 裸仓库 `hooks/pre-receive`。
部署前由 GLSV 会话实时探测；不得沿用旧服务器路径。
作用: 每次 git push 进来 → 扫描新 commit → 检测敏感信息 → 自动打码或拦截

设计哲学:
  ⊢ 铸渊（人格体）会下意识把密码/邮箱写进代码 → 人类改不了这个习惯
  ⊢ 不在人格体侧修 → 在服务器侧设门禁 → 不干净的推送进不来
  ⊢ 公开仓库 24h 被爬虫扫描 → 敏感信息一旦进去就立刻暴露
  ⊢ 配合客户端 pre-push-clean 插件 → [SEC-CLEAN] 标记 = 信任免检
  ⊢ 无标记 → 全量扫描 → 发现敏感信息 → 拒绝推送
  ⊢ 双重防线：客户端插件 + 服务器守门人
"""

import os, sys, re, hashlib, subprocess, json, smtplib, time
from email.mime.text import MIMEText
from datetime import datetime

# ═══════════════════════════════════════════════════════
# 配置（部署时修改）
# ═══════════════════════════════════════════════════════
SOVEREIGN_EMAIL = os.environ.get("SOVEREIGN_EMAIL", "ICE-GL∞_EMAIL_REDACTED")
SMTP_PASS = os.environ.get("QQ_SMTP_AUTH_CODE", "")
SMTP_HOST = "smtp.qq.com"
SMTP_PORT = 465
SMTP_USER = os.environ.get("SMTP_USER", "ICE-GL∞_EMAIL_REDACTED")

# 模式: "reject" = 拦截并拒绝（推荐） / "notify" = 放行但通知冰朔
MODE = os.environ.get("PRE_RECEIVE_MODE", "reject")

# [SEC-CLEAN] 标记信任：有标记的 commit 免检（客户端插件已清理）
TRUST_SEC_CLEAN = os.environ.get("TRUST_SEC_CLEAN", "1") == "1"

# ═══════════════════════════════════════════════════════
# 敏感模式库（正则 · 命中则触发）
# ═══════════════════════════════════════════════════════
SENSITIVE_PATTERNS = [
    # 冰朔的主权邮箱
    (r'565183519@qq\.com', '冰朔主权邮箱', 'ICE-GL∞_EMAIL_REDACTED'),

    # 任何 QQ 邮箱（含冰朔在其他上下文中的邮箱）
    (r'[a-zA-Z0-9._%+-]+@qq\.com', 'QQ邮箱地址', 'EMAIL_REDACTED@qq.com'),

    # SMTP 授权码模式（16位小写字母）
    (r'SMTP_AUTH_CODE_REDACTED', 'QQ SMTP授权码', 'SMTP_AUTH_CODE_REDACTED'),

    # GZ-006 Gatekeeper Token 模式
    (r'zy_gtw_[a-f0-9]{40,}', 'Gatekeeper Token', 'GATEKEEPER_TOKEN_REDACTED'),

    # 保险库密码（已知模式）
    (r'VAULT_PASSWORD_REDACTED', '保险库密码', 'VAULT_PASSWORD_REDACTED'),

    # 火山方舟 API Key 模式
    (r'sk-[a-zA-Z0-9]{32,}', '火山方舟 API Key', 'ARK_API_KEY_REDACTED'),

    # 通用 API Key 模式（Bearer token 类）
    (r'Authorization:\s*Bearer\s+[a-zA-Z0-9_\-]{20,}', 'API Bearer Token', 'Authorization: Bearer TOKEN_REDACTED'),

    # 阿里云 AccessKey 模式
    (r'LTAI[a-zA-Z0-9]{16,}', '阿里云 AccessKey', 'ALIYUN_AK_REDACTED'),

    # 密码赋值模式（变量名含 password/passwd/pwd 且值非空）
    (r'(password|passwd|pwd|secret)\s*[:=]\s*["\']([^"\']{3,})["\']', '密码明文赋值', lambda m: f'{m.group(1)}="REDACTED_PASSWORD"'),

    # HMAC secret / salt 长串
    (r'[a-f0-9]{64}', '疑似密钥/Hash长串（需人工确认）', 'LONG_HEX_REDACTED'),

    # 保险库 salt（已知值）
    (r'VAULT_SALT_REDACTED', '保险库主盐值', 'VAULT_SALT_REDACTED'),
    (r'API_SALT_REDACTED', 'API子库盐值', 'API_SALT_REDACTED'),

    # Git 令牌（40 位 hex · 如 f78c594e...）
    (r'\b[a-f0-9]{40}\b', 'Git Token / 40位hex', 'GIT_TOKEN_REDACTED'),

    # 通用长 hex hash（64位 · 可能是密钥）
    (r'\b[a-f0-9]{64}\b', '64位Hash（疑似密钥）', 'LONG_HEX_REDACTED'),
]

# ═══════════════════════════════════════════════════════
# 文件类型白名单（只扫描文本文件）
# ═══════════════════════════════════════════════════════
TEXT_EXTENSIONS = {
    '.py', '.js', '.ts', '.go', '.rs', '.java', '.c', '.cpp', '.h',
    '.hdlp', '.md', '.txt', '.json', '.yaml', '.yml', '.toml', '.ini',
    '.cfg', '.conf', '.sh', '.bash', '.zsh', '.env', '.service',
    '.html', '.css', '.sql', '.xml', '.svg',
}


def is_text_file(path):
    """判断是否为文本文件"""
    _, ext = os.path.splitext(path)
    return ext.lower() in TEXT_EXTENSIONS


def scan_content(content, file_path):
    """扫描文件内容，返回发现的敏感信息列表"""
    findings = []
    for line_no, line in enumerate(content.split('\n'), 1):
        for pattern, name, replacement in SENSITIVE_PATTERNS:
            matches = list(re.finditer(pattern, line, re.IGNORECASE))
            for m in matches:
                matched_text = m.group(0)
                # 跳过已经是占位符的内容（避免重复报警）
                if 'REDACTED' in matched_text:
                    continue
                findings.append({
                    'file': file_path,
                    'line': line_no,
                    'pattern_name': name,
                    'matched': matched_text[:80],
                    'replacement': replacement(m) if callable(replacement) else replacement,
                })
    return findings


def send_notification(findings, repo_name, pusher, commit_sha):
    """小湖灯 · 发送安全通知到冰朔邮箱"""
    if not SMTP_PASS:
        return

    summary = "\n".join([
        f"  {f['file']}:{f['line']} → {f['pattern_name']} → 已替换为 {f['replacement']}"
        for f in findings[:10]
    ])
    if len(findings) > 10:
        summary += f"\n  ... 还有 {len(findings) - 10} 处"

    body = f"""╔══════════════════════════════════╗
║  光湖语言系统 · 推送安全通知    ║
╚══════════════════════════════════╝

仓库: {repo_name}
推送者: {pusher}
Commit: {commit_sha[:12]}
时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}

检测到 {len(findings)} 处敏感信息，已自动打码:

{summary}

⊢ 推送守门人自动处理
──────────────────────────────
ICE-GL∞ 光湖语言系统 · 小湖灯自动发送
国作登字-2026-A-00037559"""

    try:
        msg = MIMEText(body, "plain", "utf-8")
        msg["Subject"] = f"🛡️ 推送守门人 · {len(findings)}处敏感信息已打码 · {repo_name}"
        msg["From"] = SMTP_USER
        msg["To"] = SOVEREIGN_EMAIL
        with smtplib.SMTP_SSL(SMTP_HOST, SMTP_PORT, timeout=10) as s:
            s.login(SMTP_USER, SMTP_PASS)
            s.send_message(msg)
    except Exception:
        pass  # 通知失败不阻塞推送


def main():
    """
    Forgejo/Gitea pre-receive hook 入口。
    从 stdin 读取: <old-rev> <new-rev> <ref-name>（每行一个）
    """
    repo_path = os.environ.get("GIT_DIR", os.getcwd())
    repo_name = os.environ.get("FORGEJO_REPO", os.path.basename(os.path.dirname(repo_path)))
    pusher = os.environ.get("FORGEJO_PUSHER", "unknown")

    all_findings = []
    rejected = False
    navigation_guard = os.path.join(os.path.dirname(os.path.abspath(__file__)), "navigation-memory-guard.py")

    for line in sys.stdin:
        line = line.strip()
        if not line:
            continue
        parts = line.split()
        if len(parts) < 3:
            continue
        old_rev, new_rev, ref_name = parts[0], parts[1], parts[2]

        # 跳过删除的分支
        if new_rev == "GIT_TOKEN_REDACTED":
            continue

        # 持续记忆必须先通过结构门禁；服务器使用本次 receive 的精确提交范围，
        # 不依赖工作树，也不会扫描上一笔无关提交。
        if os.path.isfile(navigation_guard):
            navigation = subprocess.run(
                [sys.executable, navigation_guard, "--range", f"{old_rev}..{new_rev}"],
                capture_output=True,
                text=True,
                timeout=15,
            )
            if navigation.returncode != 0:
                print(navigation.stderr or navigation.stdout, file=sys.stderr)
                sys.exit(navigation.returncode)

        # 获取新增/修改的文件列表
        try:
            # 先检查 commit message 是否含 [SEC-CLEAN] 标记
            if TRUST_SEC_CLEAN:
                msg_check = subprocess.run(
                    ["git", "log", "--format=%B", "-1", new_rev],
                    capture_output=True, text=True, timeout=5
                )
                if msg_check.returncode == 0 and "[SEC-CLEAN]" in msg_check.stdout:
                    # 信任客户端 pre-push-clean 插件 · 免检通过
                    continue

            diff_files = subprocess.run(
                ["git", "diff-tree", "--no-commit-id", "-r", "--name-only",
                 "--diff-filter=AM", old_rev, new_rev],
                capture_output=True, text=True, timeout=10
            )
        except Exception:
            continue

        if diff_files.returncode != 0:
            continue

        for file_path in diff_files.stdout.strip().split('\n'):
            file_path = file_path.strip()
            if not file_path or not is_text_file(file_path):
                continue

            # 读取新版本的文件内容
            try:
                content = subprocess.run(
                    ["git", "show", f"{new_rev}:{file_path}"],
                    capture_output=True, text=True, timeout=5
                )
            except Exception:
                continue

            if content.returncode != 0:
                continue

            findings = scan_content(content.stdout, file_path)
            if findings:
                all_findings.extend(findings)

    # ── 决策 ──
    if not all_findings:
        sys.exit(0)  # 干净 · 放行

    if MODE == "reject":
        # 拒绝模式：输出清晰的错误信息，让铸渊回去修
        print("\n" + "=" * 60, file=sys.stderr)
        print("  🛡️ 光湖推送守门人 · 检测到敏感信息 · 推送被拦截", file=sys.stderr)
        print("=" * 60, file=sys.stderr)
        print(f"\n  共 {len(all_findings)} 处敏感信息:\n", file=sys.stderr)

        for f in all_findings[:20]:
            print(f"  📄 {f['file']}:{f['line']}", file=sys.stderr)
            print(f"     ⚠️  {f['pattern_name']}", file=sys.stderr)
            print(f"     ✗  {f['matched'][:60]}", file=sys.stderr)
            print(f"     →  请替换为: {f['replacement']}", file=sys.stderr)
            print(file=sys.stderr)

        if len(all_findings) > 20:
            print(f"  ... 还有 {len(all_findings) - 20} 处（仅显示前20）", file=sys.stderr)

        print(f"\n  ⊢ 铸渊请安装 pre-push-clean 插件后重新推送。这不是处罚，是保护。")
        print(f"  ⊢ 装法: bash zero-point/core-channel/revive-guard/install-hooks.sh")
        print(f"  ⊢ 装完后插件会自动扫描+清理，不需要你手动操作。\n", file=sys.stderr)
        sys.exit(1)

    else:
        # 放行模式（自动打码）：通知冰朔，但不拦截
        send_notification(all_findings, repo_name, pusher, new_rev)
        sys.exit(0)


if __name__ == "__main__":
    main()