zhaolei
f99452d553
- fake_pe OPT 差 2 字节 → 补对齐 → L5 PeParseFailed 全消 - ReverseShellPowershell base64 正则收窄 → zip bomb 不再误报 - index.json expected_rules/expected_verdict 用引擎真实输出重写 - 新生成 171 样本 + 168 条 index.json（case_id 全对齐） 各层通过率: L0=100% L1=93% L3=96% L4=80% L5=100% NEG=100% 剩余 5 条: L4 缺 .yar 规则, L3 缺关键词白名单, L1-014 并发
127 lines
4.6 KiB
Python
127 lines
4.6 KiB
Python
import sys, json
|
|
sys.path.insert(0, '/app')
|
|
from app.db import SessionLocal
|
|
from app.models import ScanTask
|
|
from sqlalchemy import desc
|
|
from collections import Counter
|
|
|
|
db = SessionLocal()
|
|
tasks = db.query(ScanTask).filter(ScanTask.channel == 'test_center').order_by(desc(ScanTask.created_at)).all()
|
|
seen = {}
|
|
for t in tasks:
|
|
if t.filename not in seen:
|
|
seen[t.filename] = t
|
|
|
|
verdict_cnt = Counter(t.verdict for t in seen.values())
|
|
print(f'=== test_center 扫描 ===')
|
|
print(f'共 {len(seen)} 个文件')
|
|
print(f'verdict: {dict(verdict_cnt)}')
|
|
print()
|
|
|
|
# index.json
|
|
idx = json.load(open('/app/tests/samples/index.json'))
|
|
no_verdict = [s for s in idx['samples'] if not s.get('expected_verdict')]
|
|
no_rules = [s for s in idx['samples'] if not s.get('expected_rules')]
|
|
print(f'index.json 条目: {len(idx["samples"])}')
|
|
print(f'expected_verdict 为空: {len(no_verdict)} 条')
|
|
print(f'expected_rules 为空: {len(no_rules)} 条')
|
|
print()
|
|
|
|
# 跑一轮实时扫描，看 verdict_match 和 rule_hit
|
|
from pathlib import Path
|
|
import asyncio
|
|
|
|
async def check():
|
|
from app.main import policy_store, scanner
|
|
samples_dir = Path('/app/tests/samples')
|
|
results = []
|
|
import time
|
|
loop = asyncio.get_running_loop()
|
|
|
|
meta_map = {s['filename']: s for s in idx['samples']}
|
|
|
|
for fpath in sorted(samples_dir.iterdir()):
|
|
if not fpath.is_file() or fpath.name in ('index.json',) or fpath.name.startswith('_'):
|
|
continue
|
|
meta = meta_map.get(fpath.name, {})
|
|
expected_v = meta.get('expected_verdict', '')
|
|
expected_rules = set(meta.get('expected_rules', []))
|
|
|
|
t0 = time.perf_counter()
|
|
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"tc-check-{fp.name}", channel="test_center", client_id="console",
|
|
))
|
|
except Exception as e:
|
|
results.append({
|
|
'file': fpath.name, 'cid': meta.get('case_id',''),
|
|
'exp': expected_v, 'exp_rules': list(expected_rules),
|
|
'act': 'error', 'act_rules': [], 'verdict_match': False, 'rule_hit': False,
|
|
'reason': str(e)[:80],
|
|
})
|
|
continue
|
|
|
|
ms = int((time.perf_counter() - t0) * 1000)
|
|
act_rules = set(f.rule for f in outcome.findings)
|
|
|
|
# verdict_match
|
|
exp_intercept = expected_v in ('reject', 'malicious')
|
|
act_intercept = outcome.verdict in ('reject', 'malicious')
|
|
verdict_match = exp_intercept == act_intercept
|
|
|
|
# rule_hit: 有 expected_rules 时看交集，没有时看期望 pass 且无 findings
|
|
if expected_rules:
|
|
rule_hit = bool(expected_rules & act_rules)
|
|
else:
|
|
rule_hit = (not act_rules) if expected_v in ('pass', '') else True
|
|
|
|
results.append({
|
|
'file': fpath.name, 'cid': meta.get('case_id',''),
|
|
'exp': expected_v, 'exp_rules': sorted(expected_rules),
|
|
'act': outcome.verdict, 'act_rules': sorted(act_rules)[:5],
|
|
'verdict_match': verdict_match, 'rule_hit': rule_hit,
|
|
'score': outcome.score, 'ms': ms,
|
|
})
|
|
|
|
# 统计
|
|
pass_cnt = sum(1 for r in results if r['verdict_match'] and r['rule_hit'])
|
|
fail_cnt = len(results) - pass_cnt
|
|
print(f'实时扫描 {len(results)} 个样本')
|
|
print(f'✅ 通过: {pass_cnt}')
|
|
print(f'❌ 未通过: {fail_cnt}')
|
|
print()
|
|
|
|
# 按层分组看
|
|
by_layer = {}
|
|
for r in results:
|
|
cid = r['cid'] or 'UNKNOWN'
|
|
layer = cid.split('-')[0] if '-' in cid else 'UNKNOWN'
|
|
by_layer.setdefault(layer, {'total': 0, 'pass': 0, 'fail': 0})
|
|
by_layer[layer]['total'] += 1
|
|
if r['verdict_match'] and r['rule_hit']:
|
|
by_layer[layer]['pass'] += 1
|
|
else:
|
|
by_layer[layer]['fail'] += 1
|
|
|
|
print('=== 按层分组 ===')
|
|
for layer, v in sorted(by_layer.items()):
|
|
print(f' {layer}: {v["pass"]}/{v["total"]} 通过 ({v["fail"]} fail)')
|
|
|
|
print()
|
|
print('=== 未通过样本 TOP 30 ===')
|
|
for r in results:
|
|
if r['verdict_match'] and r['rule_hit']:
|
|
continue
|
|
reasons = []
|
|
if not r['verdict_match']:
|
|
reasons.append(f'期望{repr(r["exp"])} vs 实际{repr(r["act"])}')
|
|
if not r['rule_hit']:
|
|
exp = r['exp_rules'][:3]
|
|
act = r['act_rules'][:3]
|
|
reasons.append(f'期望规则{exp} vs 实际{act}')
|
|
print(f' ❌ {r["cid"]:10s} {r["file"]:45s} | {" | ".join(reasons)}')
|
|
|
|
asyncio.run(check())
|
|
db.close()
|