chore: 清理对账临时脚本

This commit is contained in:
2026-10-29 12:02:49 +08:00
parent 2daa3fcd3c
commit 43b4db85e7
2 changed files with 0 additions and 152 deletions

View File

@@ -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 <module>
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'

View File

@@ -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())