catalog: 96 个假必填参数改为选填并新增参数审计脚本
- 全量对比 catalog 字段与实现实际读取(audit-catalog-params.py),A 类假必填 74→0(96 个字段/72 组件) - 最大族:web×31、designated_obj×23、accuracy2×3、emails×3 等 - Screenlock/ScreenUnlock 清掉无意义必填并补 label - 隐藏参数(实现读取但目录未定义)审计为 0
This commit is contained in:
1 parent
5ef797dcc0
commit
de6165e76e
2 files changed
+223
-96
No files matched your search
@@ -0,0 +1,127 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
"""审计能力参数 v2:catalog 字段 vs 实现读取。
|
||||
|
||||
策略:
|
||||
- 直接读取:handler 函数体内 text(args,["k"]) / args.k / pick(args,[...]) 等模式
|
||||
- 间接读取:body 中把 args 整体传给函数(xxx(args)),或必填 key 出现在同文件其他位置(兜底)
|
||||
- A 类(假必填)只在「直接+间接+全文兜底都读不到」时报告,保证高置信
|
||||
- B 类(隐藏参数)只报直接读取的 key(保守)
|
||||
"""
|
||||
import json
|
||||
import re
|
||||
import os
|
||||
|
||||
ROOT = r"D:\code\PRA-only-demo"
|
||||
CAP_DIRS = [
|
||||
os.path.join(ROOT, "packages", "rpa-core", "src", "capabilities"),
|
||||
os.path.join(ROOT, "src", "main", "capabilities"),
|
||||
]
|
||||
|
||||
with open(os.path.join(ROOT, "resources", "catalog.json"), encoding="utf-8") as f:
|
||||
catalog = {c["id"]: c for c in json.load(f)["components"]}
|
||||
|
||||
src_blobs = []
|
||||
for d in CAP_DIRS:
|
||||
for fn in os.listdir(d):
|
||||
if fn.endswith(".ts") and not fn.endswith(".test.ts"):
|
||||
with open(os.path.join(d, fn), encoding="utf-8") as f:
|
||||
src_blobs.append((fn, f.read()))
|
||||
|
||||
def find_register_bodies(src):
|
||||
out = []
|
||||
for m in re.finditer(r'register\(\s*(\[[^\]]*\]|"[A-Za-z_.0-9]+")\s*,\s*(async)?\s*\(', src):
|
||||
ids = re.findall(r'"([A-Za-z_.0-9]+)"', m.group(1))
|
||||
i = m.end() - 1
|
||||
depth = 0
|
||||
start_body = None
|
||||
j = i
|
||||
while j < len(src):
|
||||
ch = src[j]
|
||||
if ch == "(":
|
||||
depth += 1
|
||||
elif ch == ")":
|
||||
depth -= 1
|
||||
if depth == 0:
|
||||
break
|
||||
elif ch == "{" and start_body is None and depth >= 2:
|
||||
start_body = j
|
||||
j += 1
|
||||
if start_body is None:
|
||||
body = src[i + 1:j]
|
||||
else:
|
||||
d2 = 0
|
||||
k = start_body
|
||||
while k < len(src):
|
||||
if src[k] == "{":
|
||||
d2 += 1
|
||||
elif src[k] == "}":
|
||||
d2 -= 1
|
||||
if d2 == 0:
|
||||
break
|
||||
k += 1
|
||||
body = src[start_body:k]
|
||||
out.append((ids, body))
|
||||
return out
|
||||
|
||||
PAT_DIRECT = [
|
||||
re.compile(r'\w+\(\s*args\s*,\s*\[([^\]]*)\)'),
|
||||
re.compile(r'pick\(\s*args\s*,\s*\[([^\]]*)\)'),
|
||||
re.compile(r'\bargs\.([a-zA-Z_][a-zA-Z0-9_]*)'),
|
||||
re.compile(r"args\[['\"]([^'\"]+)['\"]\]"),
|
||||
]
|
||||
INTERNAL = {"nodeId", "componentId", "timeoutMs", "time_out", "timeout", "evidenceDir"}
|
||||
|
||||
info = {} # cid -> dict(direct=set, indirect=bool)
|
||||
for fn, src in src_blobs:
|
||||
for ids, body in find_register_bodies(src):
|
||||
direct = set()
|
||||
for pat in PAT_DIRECT:
|
||||
for mm in pat.finditer(body):
|
||||
g = mm.group(1)
|
||||
if g is None:
|
||||
continue
|
||||
if "," in g or '"' in g or "'" in g:
|
||||
direct.update(re.findall(r"['\"]([^'\"]+)['\"]", g))
|
||||
else:
|
||||
direct.add(g)
|
||||
# args 整体传走 → 间接(keys 未知)
|
||||
indirect = bool(re.search(r'\(\s*args\s*[,)]|\bargs\s*\)', body)) and not direct
|
||||
# selectorOf(args) 这种:args 作为参数出现在调用里
|
||||
if re.search(r'\w+\(\s*args\s*\)|\w+\(\s*ctx\s*,\s*node\s*,\s*args', body):
|
||||
indirect = True
|
||||
for cid in ids:
|
||||
rec = info.setdefault(cid, {"direct": set(), "indirect": False, "file": fn})
|
||||
rec["direct"] |= direct
|
||||
rec["indirect"] = rec["indirect"] or indirect
|
||||
|
||||
A, B = [], []
|
||||
for cid, comp in sorted(catalog.items()):
|
||||
rec = info.get(cid)
|
||||
if not rec:
|
||||
continue
|
||||
fields = comp.get("fields") or []
|
||||
cat_keys = {f["key"] for f in fields}
|
||||
required = {f["key"] for f in fields if f.get("required")}
|
||||
direct = rec["direct"]
|
||||
if rec["indirect"]:
|
||||
# keys 未知:不判 A(无法确定),不判 B
|
||||
continue
|
||||
a = {k for k in required if k not in direct}
|
||||
# 兜底:全目录源码找 key(带引号形式,或 .key 词边界的直接访问)
|
||||
if a:
|
||||
all_src = "\n".join(s for _, s in src_blobs)
|
||||
a = {k for k in a if not (re.search(r'["\']' + re.escape(k) + r'["\']', all_src) or re.search(r'\.' + re.escape(k) + r'\b', all_src))}
|
||||
if a:
|
||||
A.append((cid, sorted(a)))
|
||||
b = {k for k in direct if k not in cat_keys and k not in INTERNAL and len(k) > 2}
|
||||
if b:
|
||||
B.append((cid, sorted(b)))
|
||||
|
||||
print(f"=== A 类:必填但实现全链路都不读(高置信假必填) {len(A)} 个 ===")
|
||||
for cid, keys in A:
|
||||
print(f" {cid}: {keys}")
|
||||
print()
|
||||
print(f"=== B 类:实现直接读取但目录未定义(隐藏参数,保守列表) {len(B)} 个 ===")
|
||||
for cid, keys in B:
|
||||
print(f" {cid}: {keys}")
|
||||
Reference in new issue
Block a user