05fed0e25c
Publish REPO-001 through REPO-008 as domestic routes, add the public AI discovery API, and demote Singapore paths to historical backup.
314 lines
10 KiB
Python
Executable File
314 lines
10 KiB
Python
Executable File
#!/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()
|