zhaolei
16f134e328
核心功能: - 6 层检测管线: L0 结构分析 → L1 ClamAV → L2 文档深度检测 → L3 关键词匹配 → L4 YARA → L5 PE/二进制 - FastAPI + PostgreSQL + Redis + Docker Compose 架构 - 客户端 HMAC-SHA256 签名鉴权 + 限流 + 幂等性 - 控制台 UI (登录/Dashboard/诊断中心/策略管理/任务列表/测试中心/产品介绍页) - 三层诊断定时器 (error_scan 30s / db_aggregate 30s / container_check 5min) 并发与稳定性修复: - 诊断定时器 Redis 时间槽位锁 (多 worker 同一槽位只有一个执行) - JSONL 存储 fcntl 跨进程排他锁 + 容错解码 + 原子重写 - 移除 SQLite fallback, 统一 PostgreSQL - API_CLIENTS 脱敏为环境变量, .env 已 .gitignore
114 lines
3.8 KiB
Python
114 lines
3.8 KiB
Python
import sys, os, time, hashlib, tempfile
|
||
sys.path.insert(0, '.')
|
||
# DB_URL 由 .env 或 compose environment 提供
|
||
os.environ.setdefault('MESHSAFE_MASTER_KEY', 'Dt0ktRs8SjEH8VxWZJDX125NV1qiII88HoiXZPhWers=')
|
||
os.environ.setdefault('MESHSAFE_POLICY_FILE', 'config/policy.yaml')
|
||
os.environ.setdefault('MESHSAFE_ADMIN_PASSWORD', 'admin123')
|
||
|
||
# 等 uvicorn 起来
|
||
for _ in range(15):
|
||
try:
|
||
import requests
|
||
r = requests.get('http://127.0.0.1:8443/health', timeout=2)
|
||
if r.status_code == 200:
|
||
break
|
||
except Exception:
|
||
pass
|
||
time.sleep(0.5)
|
||
else:
|
||
print('❌ uvicorn 没起来')
|
||
sys.exit(1)
|
||
print('✅ uvicorn 已就绪')
|
||
|
||
import requests
|
||
from app.services.scanner import Scanner
|
||
from app.services.policy import PolicyStore
|
||
from app.db import SessionLocal
|
||
from app.models import ScanTask, ScanFinding, IdempotencyRecord
|
||
|
||
ps = PolicyStore(policy_file='config/policy.yaml')
|
||
scanner = Scanner(ps)
|
||
|
||
# ---- 1. 控制台登录 + 页面 ----
|
||
s = requests.Session()
|
||
r = s.post('http://127.0.0.1:8443/console/login', data={'username': 'admin', 'password': 'admin123'})
|
||
print(f'\n1. login: {r.status_code}')
|
||
|
||
r = s.get('http://127.0.0.1:8443/console/')
|
||
print(f'2. dashboard: {r.status_code}')
|
||
|
||
r = s.get('http://127.0.0.1:8443/console/policy')
|
||
print(f'3. policy: {r.status_code} yara_tab={"panel-yara" in r.text}')
|
||
|
||
r = s.get('http://127.0.0.1:8443/console/tasks')
|
||
print(f'4. tasks: {r.status_code}')
|
||
|
||
r = requests.get('http://127.0.0.1:8443/health')
|
||
d = r.json()
|
||
print(f'5. /health: status={d["status"]} yara_ok={d["engines"]["yara"]["ok"]} clamav_stale={d["engines"]["clamav"].get("stale_warning")}')
|
||
|
||
# ---- 2. 正常扫描（首次）----
|
||
tmp1 = tempfile.NamedTemporaryFile(delete=False, suffix='.php')
|
||
tmp1.write(b'<?php eval($_POST["cmd"]); echo "ok"; ?>')
|
||
tmp1.close()
|
||
|
||
t1 = time.perf_counter()
|
||
outcome1 = scanner.scan_file(
|
||
file_path=tmp1.name, original_filename='shell.php',
|
||
task_id='e2e-idem-001', channel='e2e', client_id='test',
|
||
)
|
||
first_ms = int((time.perf_counter() - t1) * 1000)
|
||
|
||
os.unlink(tmp1.name)
|
||
|
||
print(f'\n6. 首次扫描: verdict={outcome1.verdict} score={outcome1.score} findings={len(outcome1.findings)} took={first_ms}ms idem={outcome1.meta.get("idempotent", False)}')
|
||
|
||
# 验证 idempotency_records 表写入
|
||
db = SessionLocal()
|
||
idem_count = db.query(IdempotencyRecord).count()
|
||
print(f' idempotency_records 表行数: {idem_count}')
|
||
db.close()
|
||
|
||
# ---- 3. 幂等二次扫描（同内容 + 同策略）----
|
||
tmp2 = tempfile.NamedTemporaryFile(delete=False, suffix='.php')
|
||
tmp2.write(b'<?php eval($_POST["cmd"]); echo "ok"; ?>') # 完全相同内容
|
||
tmp2.close()
|
||
|
||
t2 = time.perf_counter()
|
||
outcome2 = scanner.scan_file(
|
||
file_path=tmp2.name, original_filename='shell.php',
|
||
task_id='e2e-idem-002', channel='e2e', client_id='test',
|
||
)
|
||
idem_ms = int((time.perf_counter() - t2) * 1000)
|
||
|
||
os.unlink(tmp2.name)
|
||
|
||
print(f'\n7. 幂等扫描: verdict={outcome2.verdict} findings={len(outcome2.findings)} took={idem_ms}ms idem={outcome2.meta.get("idempotent", False)}')
|
||
speedup = first_ms / max(idem_ms, 1)
|
||
print(f' 提速: {speedup:.1f}x（幂等跳过了引擎扫描）')
|
||
|
||
assert outcome2.meta.get('idempotent') is True, '❌ 幂等未命中'
|
||
assert idem_ms < first_ms, '❌ 幂等应该更快'
|
||
print(' ✅ 幂等生效')
|
||
|
||
# ---- 4. 限流（内存模式，client_id=test）----
|
||
# 扫描接口需要签名，这里用 memory 限流直接测底层
|
||
from app.security.ratelimit import _memory_check
|
||
|
||
# 填满窗口
|
||
for _ in range(60):
|
||
_memory_check('test:ratelimit_test', 60)
|
||
|
||
ok = _memory_check('test:ratelimit_test', 60)
|
||
print(f'\n8. 限流测试（61/60）: {"通过" if ok else "被拒绝 ✅"}')
|
||
|
||
assert ok is False, '❌ 限流应拒绝第 61 次'
|
||
print(' ✅ 限流生效')
|
||
|
||
# 清内存窗口，给后续 run 不受影响
|
||
import app.security.ratelimit as rl
|
||
with rl._MEM_LOCK:
|
||
rl._MEM_BUCKET.clear()
|
||
|
||
print('\n=== E2E 全部通过 ✅ ===')
|