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

#!/usr/bin/env python3
"""
光湖语言系统 · pre-push-clean 插件 v1.0
Guanghu Language System · Pre-Push Cleaner

每次 git push 前自动运行 · 扫描待推送内容 · 敏感信息自动转乱码
装法: cp pre-push-clean.py .git/hooks/pre-push && chmod +x .git/hooks/pre-push

设计哲学:
  ⊢ 人格体会下意识把密码/邮箱写进代码 → 不改这个习惯
  ⊢ 在 push 之前自动扫描 → 自动替换 → 敏感信息不出本地
  ⊢ 不阻塞工作流 → 替换 → amend → 继续 push
  ⊢ 人格体只需要正常 git push · 其他全自动
"""

import os
import sys
import re
import subprocess

# ═══════════════════════════════════════════════════════
# 敏感模式库（与服务器端 pre-receive-guard 保持同步）
# ═══════════════════════════════════════════════════════
PATTERNS = [
    # 任何 QQ 邮箱
    (r'[a-zA-Z0-9._%+-]+@qq\.com', 'QQ邮箱地址', 'EMAIL_REDACTED@qq.com'),

    # 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'),

    # 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'),

    # 密码明文赋值
    (r'(password|passwd|pwd|secret)\s*[:=]\s*["\']([^"\']{3,})["\']',
     '密码明文赋值', lambda m: f'{m.group(1)}="REDACTED_PASSWORD"'),

]


def load_private_patterns():
    """从仓库外私有文件加载精确禁入值；绝不把值写入源码或日志。"""
    path = os.environ.get(
        'GUANGHU_FORBIDDEN_IDENTIFIERS_FILE',
        os.path.expanduser(
            '~/Documents/guanghulab-local-secrets/code-guard/forbidden-identifiers'
        ),
    )
    patterns = []
    try:
        with open(path, 'r', encoding='utf-8') as handle:
            for line in handle:
                value = line.strip()
                if value and not value.startswith('#'):
                    patterns.append((re.escape(value), '私有禁入标识', 'PRIVATE_VALUE_REDACTED'))
    except FileNotFoundError:
        pass
    return patterns

# 文件类型白名单
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',
}

# 排除文件（不扫描）
EXCLUDE_FILES = {
    '.git/HEAD', '.git/index', '.git/config',
    'package-lock.json', 'yarn.lock', 'pnpm-lock.yaml',
    'poetry.lock', 'Cargo.lock', 'Gemfile.lock',
}

# 是否启用（可通过环境变量关闭）
ENABLED = os.environ.get('PRE_PUSH_CLEAN_ENABLED', '1') == '1'


def is_text_file(path):
    _, ext = os.path.splitext(path)
    return ext.lower() in TEXT_EXTENSIONS


def is_excluded(path):
    for ex in EXCLUDE_FILES:
        if ex in path:
            return True
    return False


def scan_file(filepath):
    """扫描单个文件，返回 (clean_content, findings)"""
    try:
        with open(filepath, 'r', encoding='utf-8', errors='ignore') as f:
            content = f.read()
    except (IOError, PermissionError):
        return None, []

    findings = []
    new_content = content

    for pattern, name, replacement in PATTERNS + load_private_patterns():
        matches = list(re.finditer(pattern, content, re.IGNORECASE))
        for m in matches:
            matched_text = m.group(0)
            # 跳过已经是占位符的
            if 'REDACTED' in matched_text:
                continue
            repl = replacement(m) if callable(replacement) else replacement
            findings.append({
                'file': filepath,
                'pattern_name': name,
            })
            # 替换（注意：用 re.sub 而不是简单替换，避免破坏后续匹配位置）
            new_content = new_content.replace(matched_text, repl, 1)

    return new_content if findings else None, findings


def get_changed_files():
    """获取本次 push 涉及的所有文件"""
    files = set()

    # 方法1: git diff --cached（暂存区的文件）
    try:
        r = subprocess.run(
            ['git', '-c', 'core.quotepath=false', 'diff', '--cached', '--name-only', '--diff-filter=ACM'],
            capture_output=True, text=True, timeout=5
        )
        if r.returncode == 0:
            for f in r.stdout.strip().split('\n'):
                f = f.strip()
                if f:
                    files.add(f)
    except Exception:
        pass

    # 方法2: 从 stdin 读取 pre-push 传入的 refs
    try:
        for line in sys.stdin.read().strip().split('\n'):
            if not line.strip():
                continue
            parts = line.split()
            if len(parts) >= 4:
                local_ref, local_sha, remote_ref, remote_sha = parts[0], parts[1], parts[2], parts[3]
                if remote_sha != '0' * 40:
                    r = subprocess.run(
                        ['git', '-c', 'core.quotepath=false', 'diff', '--name-only', '--diff-filter=ACM',
                         remote_sha, local_sha],
                        capture_output=True, text=True, timeout=5
                    )
                    if r.returncode == 0:
                        for f in r.stdout.strip().split('\n'):
                            f = f.strip()
                            if f:
                                files.add(f)
    except Exception:
        pass

    # 如果没有文件（空 push 或新分支），扫描整个工作树中已跟踪的文件
    if not files:
        try:
            r = subprocess.run(
                ['git', '-c', 'core.quotepath=false', 'ls-files'],
                capture_output=True, text=True, timeout=5
            )
            if r.returncode == 0:
                for f in r.stdout.strip().split('\n'):
                    f = f.strip()
                    if f:
                        files.add(f)
        except Exception:
            pass

    return list(files)


def main():
    # --check 模式：仅验证插件是否正常工作
    if '--check' in sys.argv:
        print('✅ pre-push-clean v1.0 已就绪 · 每次 git push 前自动扫描')
        print(f'   敏感模式: {len(PATTERNS)} 种')
        print(f'   状态: {"🟢 启用" if ENABLED else "🔴 已禁用 (PRE_PUSH_CLEAN_ENABLED=0)"}')
        sys.exit(0)

    if not ENABLED:
        sys.exit(0)

    # 第二道本地门禁：持续记忆必须挂在可恢复的导航图上。
    guard = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'navigation-memory-guard.py')
    if os.path.isfile(guard):
        result = subprocess.run([sys.executable, guard], timeout=15)
        if result.returncode != 0:
            sys.exit(result.returncode)

    repo_root = subprocess.run(
        ['git', 'rev-parse', '--show-toplevel'],
        capture_output=True, text=True, timeout=5
    ).stdout.strip()

    if not repo_root:
        sys.exit(0)

    os.chdir(repo_root)

    # 获取变更文件
    files = get_changed_files()
    if not files:
        sys.exit(0)

    all_findings = []
    modified_files = {}

    # 扫描每个文件
    for filepath in files:
        full_path = os.path.join(repo_root, filepath)
        if not os.path.isfile(full_path):
            continue
        if not is_text_file(filepath):
            continue
        if is_excluded(filepath):
            continue

        clean_content, findings = scan_file(full_path)
        if findings:
            all_findings.extend(findings)
            modified_files[filepath] = clean_content

    # ── 无敏感信息 → 放行 ──
    if not all_findings:
        sys.exit(0)

    # ── 有敏感信息 → 自动清理 ──
    print()
    print('╔══════════════════════════════════════╗')
    print('║  🛡️  pre-push-clean · 敏感信息清理   ║')
    print('╚══════════════════════════════════════╝')
    print()
    print(f'  共 {len(all_findings)} 处敏感信息，已自动替换为乱码:')
    print()

    for f in all_findings[:20]:
        print(f'  📄 {f["file"]}')
        print(f'     ⚠️  {f["pattern_name"]}')
        print('     ✗  原文已隐藏')
        print('     →  已替换为安全占位符')
        print()

    if len(all_findings) > 20:
        print(f'  ... 还有 {len(all_findings) - 20} 处（仅显示前20）')
        print()

    # 写入清理后的文件
    for filepath, content in modified_files.items():
        full_path = os.path.join(repo_root, filepath)
        try:
            with open(full_path, 'w', encoding='utf-8') as f:
                f.write(content)
        except (IOError, PermissionError) as e:
            print(f'  ❌ 无法写入 {filepath}: {e}', file=sys.stderr)

    # 将清理后的修改加入暂存区
    for filepath in modified_files:
        subprocess.run(
            ['git', 'add', filepath],
            capture_output=True, timeout=5
        )

    # 获取当前 HEAD commit message
    try:
        r = subprocess.run(
            ['git', 'log', '-1', '--format=%B'],
            capture_output=True, text=True, timeout=5
        )
        old_msg = r.stdout.strip()
    except Exception:
        old_msg = ''

    # 检查是否已有 [SEC-CLEAN] 标记
    if '[SEC-CLEAN]' not in old_msg:
        new_msg = f'{old_msg}\n\n[SEC-CLEAN] · pre-push-clean v1.0 · {len(all_findings)}处敏感信息已自动转乱码'
        # 写临时文件给 git commit --amend
        msg_file = os.path.join(repo_root, '.git', 'PRE_PUSH_CLEAN_MSG')
        with open(msg_file, 'w') as f:
            f.write(new_msg)
        subprocess.run(
            ['git', 'commit', '--amend', '--no-edit', '--file', msg_file],
            capture_output=True, timeout=10
        )
        try:
            os.remove(msg_file)
        except Exception:
            pass

    print(f'  ✅ 已自动清理 · 推送继续')
    print(f'  📧 建议通知冰朔: {len(all_findings)}处敏感信息已转乱码')
    print()

    sys.exit(0)


if __name__ == '__main__':
    main()