128 lines
4.6 KiB
Python
128 lines
4.6 KiB
Python
|
#!/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}")
|