proxy-station/scripts/check-surge.py

#!/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))