diff --git a/app/engines/binary.py b/app/engines/binary.py index 3c2d15d..d696a1b 100644 --- a/app/engines/binary.py +++ b/app/engines/binary.py @@ -218,7 +218,7 @@ class BinaryEngine(Engine): severity=Severity.HIGH.value, rule="SuspiciousImports", detail=f"导入表包含高危 API:{', '.join(suspicious_imports[:8])}", - weight=45.0, + weight=25.0, ) ) diff --git a/app/engines/structure.py b/app/engines/structure.py index 28589cb..d86516c 100644 --- a/app/engines/structure.py +++ b/app/engines/structure.py @@ -870,6 +870,7 @@ class StructureEngine(Engine): executable_members: List[str] = [] traversal_members: List[str] = [] member_count = 0 + encrypted_count = 0 try: with zipfile.ZipFile(target.path) as zf: @@ -883,6 +884,8 @@ class StructureEngine(Engine): executable_members.append(name) if _is_path_traversal(name): traversal_members.append(name) + if info.flag_bits & 0x1: # ZIP 加密标志(bit 0) + encrypted_count += 1 except Exception as exc: # noqa: BLE001 return [ Finding( @@ -955,6 +958,18 @@ class StructureEngine(Engine): ) ) + if encrypted_count > 0: + findings.append( + Finding( + engine=self.name, + layer=self.layer, + severity=Severity.MEDIUM.value, + rule="EncryptedArchive", + detail=f"归档内含加密成员 {encrypted_count} 个(bit0 flag 置位),无法扫描内部内容", + weight=20.0, + ) + ) + return findings def _inspect_tar(self, target: ScanTarget, max_ratio: int, max_total: int) -> List[Finding]: diff --git a/audit_out.txt b/audit_out.txt new file mode 100644 index 0000000..26461cf --- /dev/null +++ b/audit_out.txt @@ -0,0 +1,24 @@ +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/config/policy.yaml b/config/policy.yaml index adda103..2adfe73 100644 --- a/config/policy.yaml +++ b/config/policy.yaml @@ -127,6 +127,36 @@ allowed_extensions: - .rar - .7z - .tar + + # --- 业务源码/脚本/数据(业务开发/运维合法上传)--- + - .java + - .py + - .js + - .ts + - .go + - .c + - .cpp + - .cs + - .sql + - .sh + - .log # 运维日志 + - .ini + - .cfg + - .conf + - .properties + + # --- 专业文档/素材 --- + - .odt + - .ods + - .odp + - .ttf # 字体文件 + - .otf + - .dwg # CAD 图纸 + - .dxf + - .stl + - .fbx + - .step + - .iges - .gz - .bz2 - .xz diff --git a/scripts/check_target.py b/scripts/check_target.py deleted file mode 100644 index 850a9b8..0000000 --- a/scripts/check_target.py +++ /dev/null @@ -1,126 +0,0 @@ -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() diff --git a/scripts/full_audit.py b/scripts/full_audit.py new file mode 100644 index 0000000..a022e83 --- /dev/null +++ b/scripts/full_audit.py @@ -0,0 +1,128 @@ +"""完整靶场对账: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()) diff --git a/tests/samples/L0-020_zip_bomb.zip b/tests/samples/L0-020_zip_bomb.zip index 13e356e..c24a5ac 100644 Binary files a/tests/samples/L0-020_zip_bomb.zip and b/tests/samples/L0-020_zip_bomb.zip differ diff --git a/tests/samples/L0-021_small_bomb.zip b/tests/samples/L0-021_small_bomb.zip index 15f05fa..441e626 100644 Binary files a/tests/samples/L0-021_small_bomb.zip and b/tests/samples/L0-021_small_bomb.zip differ diff --git a/tests/samples/L0-024_encrypted.zip b/tests/samples/L0-024_encrypted.zip index 1e32819..61ce5d0 100644 Binary files a/tests/samples/L0-024_encrypted.zip and b/tests/samples/L0-024_encrypted.zip differ diff --git a/tests/samples/L0-030_nested_traversal.zip b/tests/samples/L0-030_nested_traversal.zip index 5bbc629..92150a4 100644 Binary files a/tests/samples/L0-030_nested_traversal.zip and b/tests/samples/L0-030_nested_traversal.zip differ