feat: 合并边缘节点运维并完善 Surge 配置安全

迁入通用节点部署工具、离线交付页生成器和共用技能，隔离真实配置与交付产物。统一 Codex 与 Claude Code 开发约定，补充节点迁移与运维文档。完善 FINAL、策略组引用、输入校验、Surge 字段编码与 AI DNS 例外保护，增加隔离测试和订阅验证工具。验证：全量类型检查、44 项单测、前端构建、脚本语法、技能入口与凭据隔离检查均通过。
2026-10-03 23:19:59 +08:00
parent 7358008936
commit 5326c572a2
93 changed files with 6780 additions and 320 deletions
--- a/scripts/check-surge.py
+++ b/scripts/check-surge.py
@@ -0,0 +1,141 @@
+#!/usr/bin/env python3
+"""Read-only Surge profile validation. Credential-bearing files stay in a private temp dir."""
+import argparse
+import concurrent.futures
+import csv
+import hashlib
+import os
+from pathlib import Path
+import re
+import shutil
+import subprocess
+import tempfile
+
+
+def fields(line):
+    return [x.strip() for x in next(csv.reader([line], skipinitialspace=True, escapechar='\\'))]
+
+
+def sections(text):
+    result = {}
+    section = None
+    for number, raw in enumerate(text.splitlines(), 1):
+        line = raw.strip()
+        if line.startswith('[') and line.endswith(']'):
+            section = line[1:-1]
+        elif section and line and not line.startswith(('#', ';', '//')):
+            result.setdefault(section, []).append((number, line))
+    return result
+
+
+def native_check(cli, profile):
+    result = subprocess.run([cli, '--check', str(profile)], capture_output=True, text=True, timeout=60)
+    if result.returncode:
+        # Native diagnostics may echo node passwords; do not print them to a shared log.
+        raise RuntimeError('Surge CLI 校验失败；请在本地对该文件运行 surge-cli --check 查看详情（输出可能含凭据）')
+
+
+def domain_match(kind, value, host):
+    host, value = host.lower(), value.lower()
+    if kind == 'DOMAIN':
+        return host == value
+    if kind == 'DOMAIN-SUFFIX':
+        return host == value or host.endswith('.' + value)
+    if kind == 'DOMAIN-KEYWORD':
+        return value in host
+    return False
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument('profile', type=Path)
+    parser.add_argument('--fetch-rulesets', action='store_true', help='下载远程规则集，替换为本地快照再校验，并检查域名遮蔽')
+    parser.add_argument('--cli', default='surge-cli')
+    args = parser.parse_args()
+    if not shutil.which(args.cli):
+        raise RuntimeError('找不到 surge-cli；请在安装 Surge 的 macOS 上校验，不能以其他客户端校验替代')
+    text = args.profile.read_text(encoding='utf-8-sig')
+    parsed = sections(text)
+    rules = [(number, fields(line)) for number, line in parsed.get('Rule', [])]
+    finals = [r for r in rules if r[1][0] == 'FINAL']
+    if len(finals) != 1 or not rules or rules[-1][1][0] != 'FINAL':
+        raise RuntimeError('必须恰有一条 FINAL 且位于规则末尾')
+    for number, line in parsed.get('Proxy Group', []):
+        if re.search(r',\s*url\s*=', line):
+            raise RuntimeError(f'配置第 {number} 行含 Surge 无效的组级 url=')
+    for number, line in parsed.get('General', []):
+        if re.match(r'(?:http-api|external-controller-access)\s*=\s*["\']?(?:surge|clashconnectrules)@', line, re.I):
+            raise RuntimeError(f'配置第 {number} 行使用公开控制口令')
+    if any(re.match(r'skip-server-cert-verify\s*=\s*["\']?(?:true|1)["\']?\s*(?:$|[#;]|//)', line, re.I) for _, line in parsed.get('MITM', [])):
+        raise RuntimeError('MITM 关闭了上游证书验证')
+    native_check(args.cli, args.profile)
+    print('官方 Surge CLI：OK；FINAL、测速参数、控制口令与 MITM 检查通过')
+    if not args.fetch_rulesets:
+        return
+
+    urls = {}
+    for _, parts in rules:
+        if parts[0] in ('RULE-SET', 'DOMAIN-SET') and parts[1].startswith(('https://', 'http://')):
+            urls[parts[1]] = parts[0]
+    for _, line in parsed.get('Host', []):
+        match = re.match(r'(RULE-SET|DOMAIN-SET):(https?://\S+)\s+=', line)
+        if match:
+            urls[match[2]] = match[1]
+    env = {k: v for k, v in os.environ.items() if k.lower() not in ('http_proxy', 'https_proxy', 'all_proxy')}
+    with tempfile.TemporaryDirectory(prefix='proxy-station-surge-check-') as directory:
+        root = Path(directory)
+
+        def fetch(item):
+            url, kind = item
+            file = root / (hashlib.sha256(url.encode()).hexdigest() + '.txt')
+            result = subprocess.run(['curl', '--silent', '--show-error', '--location', '--max-time', '25', '--max-filesize', '10000000', '--output', str(file), '--write-out', '%{http_code}', url], capture_output=True, text=True, timeout=30, env=env)
+            if result.returncode or result.stdout != '200':
+                raise RuntimeError('规则集下载失败（URL 未输出，避免泄露私有资源 token）')
+            data = file.read_text(encoding='utf-8-sig')
+            if re.search(r'<(?:!DOCTYPE|html)', data[:200], re.I):
+                raise RuntimeError('规则集返回 HTML')
+            entries = []
+            for line in data.splitlines():
+                line = line.strip()
+                if not line or line.startswith(('#', ';', '//')):
+                    continue
+                parts = fields(line) if kind == 'RULE-SET' else ['DOMAIN-SUFFIX' if line.startswith('.') else 'DOMAIN', line.lstrip('.')]
+                if parts[0] == 'FINAL':
+                    raise RuntimeError('外部规则集不允许含 FINAL')
+                entries.append(parts)
+            return url, file, entries
+
+        with concurrent.futures.ThreadPoolExecutor(max_workers=8) as pool:
+            snapshots = list(pool.map(fetch, urls.items()))
+        local = text
+        coverage = {}
+        for url, file, entries in snapshots:
+            local = local.replace(url, str(file))
+            coverage[url] = entries
+        profile = root / 'profile.conf'
+        profile.write_text(local)
+        profile.chmod(0o600)
+        native_check(args.cli, profile)
+        print(f'远程规则集 {len(snapshots)}/{len(urls)} 下载成功；本地快照官方 CLI：OK')
+        warnings = 0
+        for index, (number, current) in enumerate(rules):
+            if current[0] not in ('DOMAIN', 'DOMAIN-SUFFIX') or len(current) < 3:
+                continue
+            for previous_number, previous in rules[:index]:
+                if len(previous) < 3 or previous[2] == current[2]:
+                    continue
+                entries = coverage.get(previous[1], []) if previous[0] in ('RULE-SET', 'DOMAIN-SET') else [previous]
+                if any(len(entry) >= 2 and domain_match(entry[0], entry[1], current[1]) for entry in entries):
+                    print(f'需审阅：第 {number} 行域名规则的样例被第 {previous_number} 行的不同策略覆盖（仅静态域名判断）')
+                    warnings += 1
+                    break
+        print(f'静态域名遮蔽提示：{warnings}；不模拟 IP／进程／协议匹配或设备缓存')
+
+
+if __name__ == '__main__':
+    try:
+        main()
+    except subprocess.TimeoutExpired:
+        raise SystemExit('校验或规则集下载超时；URL 与配置内容未输出')
+    except (RuntimeError, OSError, csv.Error) as error:
+        raise SystemExit(str(error))