80 lines
2.6 KiB
Python
80 lines
2.6 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""回放测试：把今天所有会话的真实 Bash 命令灌进 bash-output-guard，统计误拦率。
|
||
|
||
只打印「摘要 + DENY 明细」，不 dump 全量命令。
|
||
"""
|
||
import collections
|
||
import glob
|
||
import importlib.util
|
||
import io
|
||
import json
|
||
import os
|
||
|
||
GUARD = r"D:\github\dsh_shenxian\dsh-server-docs\scripts\bash-output-guard.py"
|
||
PROJ = r"E:\ProgramData\.workbuddy\projects\e-ProgramData-AIProject-aliyun-dsh-server"
|
||
|
||
sp = importlib.util.spec_from_file_location("g", GUARD)
|
||
g = importlib.util.module_from_spec(sp)
|
||
sp.loader.exec_module(g)
|
||
|
||
cmds = []
|
||
for f in sorted(glob.glob(os.path.join(PROJ, "*.jsonl"))):
|
||
sid = os.path.basename(f)[:8]
|
||
for ln in io.open(f, encoding="utf-8", errors="replace"):
|
||
if '"function_call"' not in ln or '"Bash"' not in ln:
|
||
continue
|
||
try:
|
||
o = json.loads(ln)
|
||
except ValueError:
|
||
continue
|
||
if o.get("type") != "function_call" or o.get("name") != "Bash":
|
||
continue
|
||
try:
|
||
a = json.loads(o.get("arguments") or "{}")
|
||
except ValueError:
|
||
continue
|
||
c = a.get("command")
|
||
if isinstance(c, str) and c.strip():
|
||
cmds.append((sid, c))
|
||
|
||
by_sid = collections.Counter(s for s, _ in cmds)
|
||
print("=== 样本 ===")
|
||
print(" Bash 命令 %d 条 ｜ 会话 %d 个" % (len(cmds), len(by_sid)))
|
||
for s, n in by_sid.most_common():
|
||
print(" %s %d 条" % (s, n))
|
||
|
||
den, safehit = [], 0
|
||
for sid, c in cmds:
|
||
if g.SAFE.search(c): # 真链路里 main() 先过 SAFE 就放行
|
||
safehit += 1
|
||
continue
|
||
why, fix = g.reason_for(c, hard=False)
|
||
if why:
|
||
den.append((sid, why, c))
|
||
|
||
print()
|
||
print("=== soft 档（默认）判定 ===")
|
||
print(" DENY = %d / %d = **%.2f%%** （被 SAFE 限流救回 %d 条）"
|
||
% (len(den), len(cmds), 100.0 * len(den) / max(len(cmds), 1), safehit))
|
||
for w, n in collections.Counter(w for _, w, _ in den).most_common():
|
||
print(" %-22s %d" % (w, n))
|
||
print()
|
||
print("=== DENY 明细（前 30 条 · 截 108 字符）===")
|
||
for sid, why, c in den[:30]:
|
||
print(" [%s] %-20s %s" % (sid, why, c.replace("\n", " ")[:108]))
|
||
if not den:
|
||
print(" （无 DENY —— 全部放行）")
|
||
|
||
print()
|
||
print("=== hard 档（仅参考）===")
|
||
denh = []
|
||
for sid, c in cmds:
|
||
if g.SAFE.search(c):
|
||
continue
|
||
why, _ = g.reason_for(c, hard=True)
|
||
if why:
|
||
denh.append((sid, why, c))
|
||
print(" DENY = %d / %d = %.2f%%" % (len(denh), len(cmds), 100.0 * len(denh) / max(len(cmds), 1)))
|
||
for w, n in collections.Counter(w for _, w, _ in denh).most_common():
|
||
print(" %-22s %d" % (w, n))
|