diff --git a/audit_out.txt b/audit_out.txt deleted file mode 100644 index 26461cf..0000000 --- a/audit_out.txt +++ /dev/null @@ -1,24 +0,0 @@ -2026-10-29 11:58:57 [INFO] root - 日志系统初始化完成：级别=INFO，日志文件=/var/log/meshsafe/meshsafe.log（按 10 MB 轮转，保留 5 份） -2026-10-29 11:58:57 [INFO] app.services.policy - DB 加载策略失败（回退 YAML）: Multiple rows were found when one or none was required -docker : Traceback (most recent call last): -At line:15 char:1849 -+ ... RM = 'dumb'; docker exec meshsafe-api python3 /tmp/fa.py 2>&1 | Out-F ... -+ ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ - + CategoryInfo : NotSpecified: (Traceback (most recent call last)::String) [], RemoteException - + FullyQualifiedErrorId : NativeCommandError - - File "/tmp/fa.py", line 128, in - asyncio.run(run()) - File "/usr/local/lib/python3.12/asyncio/runners.py", line 195, in run - return runner.run(main) - ^^^^^^^^^^^^^^^^ - File "/usr/local/lib/python3.12/asyncio/runners.py", line 118, in run - return self._loop.run_until_complete(task) - ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ - File "/usr/local/lib/python3.12/asyncio/base_events.py", line 691, in run_until_complete - return future.result() - ^^^^^^^^^^^^^^^ - File "/tmp/fa.py", line 16, in run - with open('/tmp/yongli.csv', encoding='utf-8') as f: - ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ -FileNotFoundError: [Errno 2] No such file or directory: '/tmp/yongli.csv' diff --git a/scripts/full_audit.py b/scripts/full_audit.py deleted file mode 100644 index a022e83..0000000 --- a/scripts/full_audit.py +++ /dev/null @@ -1,128 +0,0 @@ -"""完整靶场对账：CSV 原始预期 vs 引擎实际输出。 - -逐样本打印：case_id / 文件名 / CSV 预期（裁决+关键规则）/ 引擎实际（裁决+规则）/ 差距分类 -""" -import sys, json, csv, asyncio, time -from pathlib import Path - -sys.path.insert(0, '/app') - -async def run(): - from app.main import scanner - loop = asyncio.get_running_loop() - - # 读 CSV - csv_rows = {} - with open('/tmp/yongli.csv', encoding='utf-8') as f: - for row in csv.DictReader(f): - csv_rows[row['用例ID'].strip()] = row - - # 读 index.json（用于归类 L7 旧样本等） - idx = json.load(open('/tmp/regen_v2.json')) if Path('/tmp/regen_v2.json').exists() else {} - # 先从宿主机拷 - print('=== 开始全量靶场对账 ===\n') - - # 扫描所有文件 - from pathlib import Path as P - samples_dir = P('/app/tests/samples') - results = [] - - files = sorted(p for p in samples_dir.iterdir() - if p.is_file() and p.name not in ('index.json',) and not p.name.startswith('_')) - - # 先用 regen_v2 的 index 拿 case_id - old_idx = json.loads(samples_dir.joinpath('index.json').read_text()) - meta_map = {s['filename']: s for s in old_idx['samples']} - - for fpath in files: - meta = meta_map.get(fpath.name, {}) - cid = meta.get('case_id', '') - - # CSV 原始预期 - csv_exp = csv_rows.get(cid, {}) - csv_verdict = csv_exp.get('预期结果', '') - csv_name = csv_exp.get('用例名称', '') - - # 引擎实际 - try: - outcome = await loop.run_in_executor(None, lambda fp=fpath: scanner.scan_file( - file_path=str(fp), original_filename=fp.name, - task_id=f"chk-{fp.name}", channel="audit", client_id="console", - )) - except Exception as e: - results.append({ - 'cid': cid, 'file': fpath.name, 'layer': meta.get('category', ''), - 'csv_verdict': csv_verdict, 'csv_name': csv_name, - 'eng_verdict': 'error', 'eng_rules': [], 'score': 0, 'reason': str(e), - }) - continue - - eng_rules = sorted({f.rule for f in outcome.findings}) - results.append({ - 'cid': cid, 'file': fpath.name, 'layer': meta.get('category', ''), - 'csv_verdict': csv_verdict, 'csv_name': csv_name, - 'eng_verdict': outcome.verdict, 'eng_rules': eng_rules, - 'score': outcome.score, - }) - - # 分类差距 - def classify(r): - csv_v = r['csv_verdict'] - eng_v = r['eng_verdict'] - eng_reject = eng_v in ('reject', 'malicious') - eng_pass = eng_v in ('pass', 'clean') - - if not csv_v: - return 'NO_CSV' - # CSV 预期 reject - if any(k in csv_v for k in ['拦截', '告警', '高危', '拒绝', '命中']) and eng_pass: - return 'CSV想拦→引擎放' - # CSV 预期 pass - if any(k in csv_v for k in ['无风险', '无告警', '无命中', '正常解析']) and eng_reject: - return 'CSV想放→引擎拦' - # verdict 类型不匹配 - if '拦截' in csv_v and eng_v == 'suspicious': - return 'CSV要拦截→引擎仅可疑' - if '低危' in csv_v and eng_reject: - return 'CSV说低危→引擎拦截' - if '低危' in csv_v and eng_pass: - return 'CSV说低危→引擎放（合理，低危=放行）' - return 'OK' - - # 打印汇总 - from collections import Counter, defaultdict - gap_cnt = Counter() - gap_by_layer = defaultdict(Counter) - for r in results: - gap = classify(r) - gap_cnt[gap] += 1 - gap_by_layer[r['layer']][gap] += 1 - - print('=== 总体分类 ===') - for gap, cnt in gap_cnt.most_common(): - print(f' {gap:20s}: {cnt}') - print() - - print('=== 按层 × 分类 ===') - for layer in sorted(set(r['layer'] for r in results)): - total = sum(1 for r in results if r['layer'] == layer) - ok = sum(1 for r in results if r['layer'] == layer and classify(r) == 'OK') - print(f' {layer:5s}: {ok}/{total} OK') - for gap, cnt in gap_by_layer[layer].most_common(): - if gap != 'OK': - print(f' └ {gap}: {cnt}') - print() - - # 逐类打印差距详情 - for gap_type in ['CSV想拦→引擎放', 'CSV想放→引擎拦', 'CSV要拦截→引擎仅可疑', 'CSV说低危→引擎拦截', 'NO_CSV']: - items = [r for r in results if classify(r) == gap_type] - if not items: - continue - print(f'=== {gap_type} ({len(items)} 条) ===') - for r in items: - print(f" {r['cid']:10s} [{r['layer']:3s}] {r['file']:45s}") - print(f" CSV预期: {r['csv_verdict'][:80]}") - print(f" 引擎输出: verdict={r['eng_verdict']:10s} rules={r['eng_rules'][:5]}") - print() - -asyncio.run(run())