approve: [M9-收口A] 仓库事实快照取证落盘（只产证据文件，不改日志/不改代码）
This commit is contained in:
agent.develop
parent
8ac2694eee
commit
d1e2ad0ee5
@ -8,19 +8,42 @@
|
||||
实测在部分部署机上会**静默失败**（ModuleNotFoundError 之类），后果是「菜单里看得到、
|
||||
点进去一律 403」。因此宿主应用必须有自己的一份注册入口（不同代码路径 = 可靠回退）。
|
||||
|
||||
本脚本的做法：**聚合**各 pbl_* / world_sync 模块 load_path.py 里的 PATHS 常量，
|
||||
本脚本的做法：**聚合**各 pbl_* / world_sync 模块 load_path.py 里的路径常量，
|
||||
而不是抄一份清单。理由：清单抄两遍必然漂移（模块加了新接口，中央层忘了同步 = 又是 403）。
|
||||
聚合后中央层与模块层永远一致；同时对 pbl_evidence（M5a 本次退回的模块）保留一份
|
||||
**硬编码兜底清单**，即使模块脚本 import 失败也一定登记进去。
|
||||
聚合后中央层与模块层永远一致；同时对 pbl_evidence 保留一份**硬编码兜底清单**，
|
||||
即使模块脚本 import 失败也一定登记进去。
|
||||
|
||||
聚合形态兼容（M11b-2c-T3 QC 退回意见 #2 修复：消除「整模块静默 skip」）
|
||||
--------------------------------------------------------------------
|
||||
模块层允许下列任一写法，中央层逐条**归一**为 ``(path, role)``：
|
||||
① 扁平二元组 ``PATHS = [('/x/a.dspy', 'logined'), ...]``
|
||||
② 扁平纯字符串 ``PATHS = ['/x/a.dspy', ...]`` → role 取模块 ``DEFAULT_ROLE``（缺省 logined）
|
||||
③ 三元组及以上 ``PATHS = [('/x/a.dspy','logined','备注'), ...]`` → 取前二
|
||||
④ 分层常量 ``PATHS_ANY`` / ``PATHS_LOGINED`` / ``PATHS_OWNER_SUPERUSER``
|
||||
/ ``PATHS_OWNER_OPERATOR``（world_sync 采用此写法）
|
||||
⑤ 汇总别名 ``ALL_PATHS``，或 ``API_PATHS`` + ``UI_PATHS``
|
||||
（pbl_validation / pbl_domain_ext 采用此写法，此前被静默跳过 = 同类缺口）
|
||||
⑥ 显式零路径 ``NO_WEB_PATHS = True``（纯库函数模块，如 pbl_common / pbl_appcodes）
|
||||
—— 必须**显式声明**，不允许以「常量留空」表达「无路径」。
|
||||
**单条形态非法**（既不是 str 也不是 (list,tuple)）→ 只跳过该条并记 WARN，不再丢弃整模块。
|
||||
|
||||
一致性门禁（QC 退回意见 #2 第 ② 点）
|
||||
---------------------------------
|
||||
``--check`` 除通配符审计外，新增**聚合完整性门禁**：MODULES 中凡存在
|
||||
``scripts/load_path.py`` 的模块，若 ①import 失败 ②一条路径都没聚合到且没有
|
||||
``NO_WEB_PATHS`` 显式声明 ③存在形态非法的路径条目 → 计入 FAIL 清单并 **rc=1**，
|
||||
不再出现「聚合失败但 check 仍 PASS」。
|
||||
|
||||
用法
|
||||
----
|
||||
python3 scripts/load_path.py # 聚合全部模块路径并注册
|
||||
python3 scripts/load_path.py --check # 只做清单一致性核对（不连库）
|
||||
python3 scripts/load_path.py --check # 一致性核对 + 通配符审计（不连库）
|
||||
python3 scripts/load_path.py --list # 打印最终在册清单
|
||||
WS_DIR=/path/to/workspace python3 scripts/load_path.py # 指定机构工作空间
|
||||
|
||||
铁律：路径逐条显式，禁止通配符（``%`` / ``*``）。
|
||||
铁律：路径逐条显式，禁止通配符（``%`` / ``*``）。注意区分「RBAC 通配符」与
|
||||
「Python 格式化占位符」：模块源码里 ``"/api/x/%s.dspy" % a`` 的 ``%s`` 是**构造期**
|
||||
占位符，展开后的路径字符串不含 ``%``；本脚本审计的是**展开后**的运行时字符串。
|
||||
|
||||
═══════════════════════════════════════════════════════════════════════════════
|
||||
里程碑 RBAC 声明台账（中央层，与模块层双写等价）
|
||||
@ -33,11 +56,9 @@ M11b-2（T3，2026-09-22）：
|
||||
双写 = 等价声明同时写在 modules/world_sync/scripts/load_path.py 文件头台账、
|
||||
modules/pbl_runtime_ext/scripts/load_path.py（如存在）、以及说明文件
|
||||
modules/world_sync/wwwroot/api/README-N-A-M11b2.md。
|
||||
兼容 = 本中央脚本已支持两种模块层常量写法：
|
||||
① 扁平 ``PATHS = [(path, role), ...]``；
|
||||
② 分层 ``PATHS_ANY`` / ``PATHS_LOGINED`` / ``PATHS_OWNER_SUPERUSER``
|
||||
/ ``PATHS_OWNER_OPERATOR``（world_sync 采用此写法）。
|
||||
新增模块只需把模块名加进 MODULES，无需在中央层重复维护清单。
|
||||
附带修复（QC #2）= 见上「聚合形态兼容」+「一致性门禁」两节：pbl_blueprint（扁平
|
||||
纯字符串 PATHS，139 条）与 pbl_validation / pbl_domain_ext（API_PATHS /
|
||||
ALL_PATHS）此前在中央层被静默 skip，本次全部聚合成功。
|
||||
═══════════════════════════════════════════════════════════════════════════════
|
||||
"""
|
||||
import argparse
|
||||
@ -50,7 +71,7 @@ APP_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||||
# 机构工作空间根：环境变量 > 应用仓库上溯（apps/pbls -> apps -> workspace）
|
||||
WS_DIR = os.environ.get('WS_DIR') or os.path.normpath(os.path.join(APP_DIR, '..', '..'))
|
||||
|
||||
# 需要登记 RBAC 的模块（pbl_* 全量 + 宿主挂载的 world_sync；本次 M5a 重点是 pbl_evidence）
|
||||
# 需要登记 RBAC 的模块（pbl_* 全量 + 宿主挂载的 world_sync）
|
||||
MODULES = [
|
||||
'pbl_common', 'pbl_appcodes', 'pbl_governance', 'pbl_blueprint', 'pbl_template',
|
||||
'pbl_validation', 'pbl_compiler', 'pbl_agent_runtime', 'pbl_evidence',
|
||||
@ -59,6 +80,17 @@ MODULES = [
|
||||
'world_sync',
|
||||
]
|
||||
|
||||
# 模块未显式给出 role 时的缺省授权层（可用模块常量 DEFAULT_ROLE 覆盖）
|
||||
DEFAULT_ROLE = 'logined'
|
||||
# 分层常量 → role 映射
|
||||
TIER_ATTRS = (
|
||||
('PATHS_ANY', 'any'),
|
||||
('PATHS_LOGINED', 'logined'),
|
||||
('PATHS_OWNER_SUPERUSER', 'owner.superuser'),
|
||||
('PATHS_OWNER_OPERATOR', 'owner.operator'),
|
||||
)
|
||||
FORBIDDEN_WILDCARDS = ('%', '*')
|
||||
|
||||
# ── 里程碑声明台账（中央层，与模块层双写；--check / --list 会原样回显）────────
|
||||
MILESTONE_DECLARATIONS = [
|
||||
{
|
||||
@ -88,61 +120,172 @@ PBL_EVIDENCE_FALLBACK = [
|
||||
]
|
||||
|
||||
|
||||
def _load_module_paths(mod):
|
||||
"""从 modules/{mod}/scripts/load_path.py 读 PATHS（不执行其注册逻辑）。
|
||||
# 授权严格度排序：数值越大越严格。冲突时保留更严格者，避免合并顺序放宽权限。
|
||||
ROLE_STRICTNESS = {'any': 0, 'logined': 1, 'owner.operator': 2, 'owner.superuser': 3}
|
||||
|
||||
支持两种常量写法：
|
||||
① PATHS = [(path, role), ...]
|
||||
② PATHS_ANY / PATHS_LOGINED / PATHS_OWNER_SUPERUSER / PATHS_OWNER_OPERATOR
|
||||
（分层列表，role 依次映射 any / logined / owner.superuser / owner.operator）
|
||||
|
||||
def _stricter(a, b):
|
||||
return a if ROLE_STRICTNESS.get(a, 1) >= ROLE_STRICTNESS.get(b, 1) else b
|
||||
|
||||
|
||||
def _normalize(entry, mod, origin, bad):
|
||||
"""把一条路径声明归一为 (path, role)；形态非法则记入 bad 并返回 None。
|
||||
|
||||
只跳过**这一条**，绝不因单条异形而丢弃整个模块（QC 退回意见 #2 第 ① 点）。
|
||||
"""
|
||||
role_default = DEFAULT_ROLE
|
||||
if isinstance(entry, (tuple, list)):
|
||||
if len(entry) >= 2: # 二元组直取；三元组取前二
|
||||
path, role = str(entry[0]), str(entry[1])
|
||||
elif len(entry) == 1: # 单元素容器按纯字符串处理
|
||||
path, role = str(entry[0]), role_default
|
||||
else:
|
||||
bad.append('%s: %s 含空条目 %r（已跳过该条）' % (mod, origin, entry))
|
||||
return None
|
||||
elif isinstance(entry, str): # 纯字符串 → 默认 role
|
||||
path, role = entry, role_default
|
||||
else:
|
||||
bad.append('%s: %s 含非法条目类型 %s（已跳过该条）'
|
||||
% (mod, origin, type(entry).__name__))
|
||||
return None
|
||||
path = path.strip()
|
||||
if not path.startswith('/'):
|
||||
bad.append('%s: %s 条目 %r 不是绝对路径（已跳过该条）' % (mod, origin, path))
|
||||
return None
|
||||
return (path, role)
|
||||
|
||||
|
||||
def _load_module_paths(mod):
|
||||
"""从 modules/{mod}/scripts/load_path.py 读路径常量（不执行其注册逻辑）。
|
||||
|
||||
返回 ``(paths, note, failed)``：
|
||||
* ``failed=True`` 表示该模块**聚合失败**（import 异常 / 零路径且无显式
|
||||
``NO_WEB_PATHS`` 声明 / 存在被跳过的非法条目），``--check`` 据此判 FAIL。
|
||||
* 脚本不存在时返回 ``(None, 'no scripts/load_path.py', False)``——不算失败
|
||||
（该模块尚无声明入口，由 MODULES 维护者决定何时补）。
|
||||
"""
|
||||
script = os.path.join(WS_DIR, 'modules', mod, 'scripts', 'load_path.py')
|
||||
if not os.path.exists(script):
|
||||
return None, 'no scripts/load_path.py'
|
||||
return None, 'no scripts/load_path.py', False
|
||||
try:
|
||||
spec = importlib.util.spec_from_file_location('_lp_%s' % mod, script)
|
||||
spec = importlib.util.spec_from_file_location('_lp_' + mod, script)
|
||||
m = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(m)
|
||||
out = []
|
||||
except Exception as exc: # noqa: BLE001
|
||||
return [], 'import failed: %s: %s' % (type(exc).__name__, exc), True
|
||||
|
||||
default_role = getattr(m, 'DEFAULT_ROLE', DEFAULT_ROLE)
|
||||
if isinstance(default_role, str) and default_role:
|
||||
globals()['DEFAULT_ROLE'] = default_role # 本模块归一期间生效
|
||||
else:
|
||||
default_role = DEFAULT_ROLE
|
||||
|
||||
out, bad = [], []
|
||||
try:
|
||||
paths = getattr(m, 'PATHS', None)
|
||||
if paths:
|
||||
out += [(p, r) for p, r in paths]
|
||||
for attr, role in (('PATHS_ANY', 'any'),
|
||||
('PATHS_LOGINED', 'logined'),
|
||||
('PATHS_OWNER_SUPERUSER', 'owner.superuser'),
|
||||
('PATHS_OWNER_OPERATOR', 'owner.operator')):
|
||||
for e in paths:
|
||||
got = _normalize(e, mod, 'PATHS', bad)
|
||||
if got:
|
||||
out.append(got)
|
||||
for attr, role in TIER_ATTRS:
|
||||
tier = getattr(m, attr, None)
|
||||
if tier:
|
||||
out += [(p, role) for p in tier]
|
||||
if not out:
|
||||
return None, 'PATHS / PATHS_ANY / PATHS_LOGINED empty'
|
||||
return out, None
|
||||
except Exception as exc: # noqa: BLE001
|
||||
return None, 'import failed: %s' % (exc,)
|
||||
if not tier:
|
||||
continue
|
||||
for e in tier:
|
||||
got = _normalize(e, mod, attr, bad)
|
||||
if got:
|
||||
out.append((got[0], role))
|
||||
if not out: # 写法 ⑤ 汇总别名
|
||||
# ALL_PATHS 优先（它通常就是 API_PATHS+UI_PATHS 的并集，避免重复计数）
|
||||
alias_attrs = (('ALL_PATHS',) if getattr(m, 'ALL_PATHS', None)
|
||||
else ('API_PATHS', 'UI_PATHS'))
|
||||
for attr in alias_attrs:
|
||||
alias = getattr(m, attr, None)
|
||||
if not alias:
|
||||
continue
|
||||
for e in alias:
|
||||
got = _normalize(e, mod, attr, bad)
|
||||
if got:
|
||||
out.append(got)
|
||||
finally:
|
||||
globals()['DEFAULT_ROLE'] = DEFAULT_ROLE
|
||||
|
||||
if not out:
|
||||
if bad:
|
||||
return [], 'all entries malformed -> ' + '; '.join(bad), True
|
||||
if getattr(m, 'NO_WEB_PATHS', False): # 写法 ⑥ 显式零路径
|
||||
return [], 'declares NO_WEB_PATHS=True (zero web paths, explicit)', False
|
||||
if getattr(m, 'MILESTONE_DECLARATIONS', None):
|
||||
return [], 'declares milestone台账 only (zero web paths)', False
|
||||
return [], ('PATHS / PATHS_* / ALL_PATHS empty and NO_WEB_PATHS not declared'
|
||||
' -> aggregation FAILED'), True
|
||||
|
||||
note = 'aggregated %d paths' % len(out)
|
||||
if bad:
|
||||
note += ' (WARN %d malformed entries skipped: %s)' % (len(bad), '; '.join(bad))
|
||||
return out, note, bool(bad)
|
||||
|
||||
|
||||
def collect():
|
||||
"""聚合全部模块 PATHS + 兜底清单，返回 (paths, notes)。paths 已去重排序。"""
|
||||
merged, notes = {}, []
|
||||
"""聚合全部模块路径 + 兜底清单。
|
||||
|
||||
返回 ``(paths, notes, failures)``：
|
||||
* ``paths`` 去重排序后的 ``(path, role)`` 清单；
|
||||
* ``notes`` 逐模块回显行；
|
||||
* ``failures``聚合失败模块清单（``--check`` 非空即 rc=1）。
|
||||
"""
|
||||
merged, notes, failures, conflicts = {}, [], [], []
|
||||
for mod in MODULES:
|
||||
script = os.path.join(WS_DIR, 'modules', mod, 'scripts', 'load_path.py')
|
||||
has_script = os.path.exists(script)
|
||||
got, err, failed = _load_module_paths(mod)
|
||||
|
||||
if mod == 'pbl_evidence':
|
||||
got, err = _load_module_paths(mod)
|
||||
# 兜底清单始终并入（union），保证模块脚本坏掉也不漏登记
|
||||
for p, r in PBL_EVIDENCE_FALLBACK:
|
||||
merged[p] = r
|
||||
if err:
|
||||
if got is None:
|
||||
notes.append('%s: no scripts/load_path.py (fallback list applied)' % mod)
|
||||
continue
|
||||
if failed:
|
||||
failures.append('%s: %s (fallback list applied)' % (mod, err))
|
||||
notes.append('%s: %s (fallback list applied)' % (mod, err))
|
||||
elif not got:
|
||||
notes.append('%s: %s (fallback list applied)' % (mod, err))
|
||||
else:
|
||||
notes.append('%s: aggregated %d paths from module script' % (mod, len(got)))
|
||||
notes.append('%s: aggregated %d paths from module script + fallback'
|
||||
% (mod, len(got)))
|
||||
else:
|
||||
got, err = _load_module_paths(mod)
|
||||
if err:
|
||||
if got is None:
|
||||
notes.append('%s: %s (skipped)' % (mod, err))
|
||||
continue
|
||||
if not got and not failed:
|
||||
# 合法的空：模块显式声明 NO_WEB_PATHS / 仅里程碑台账
|
||||
notes.append('%s: %s' % (mod, err))
|
||||
continue
|
||||
if failed and not got:
|
||||
failures.append('%s: %s' % (mod, err))
|
||||
notes.append('%s: %s (FAILED)' % (mod, err))
|
||||
continue
|
||||
for p, r in got:
|
||||
merged[p] = r
|
||||
notes.append('%s: aggregated %d paths' % (mod, len(got)))
|
||||
return sorted((p, merged[p]) for p in merged), notes
|
||||
if p in merged and merged[p] != r:
|
||||
keep = _stricter(merged[p], r)
|
||||
conflicts.append('%s: %s role %s vs %s -> keep %s (stricter)'
|
||||
% (mod, p, merged[p], r, keep))
|
||||
merged[p] = keep
|
||||
else:
|
||||
merged[p] = r
|
||||
tag = ' (WARN partial)' if failed else ''
|
||||
notes.append('%s: aggregated %d paths%s' % (mod, len(got), tag))
|
||||
if failed:
|
||||
failures.append('%s: %s' % (mod, err))
|
||||
if got is None and has_script:
|
||||
failures.append('%s: script exists but nothing aggregated' % mod)
|
||||
|
||||
for c in conflicts:
|
||||
notes.append('[conflict] ' + c)
|
||||
return sorted((p, merged[p]) for p in merged), notes, failures
|
||||
|
||||
|
||||
def find_set_perm():
|
||||
@ -183,34 +326,47 @@ def register(paths, verbose=True):
|
||||
def print_milestone_declarations():
|
||||
print('[apps/pbls] 里程碑 RBAC 声明台账（中央层，与模块层双写等价）')
|
||||
for d in MILESTONE_DECLARATIONS:
|
||||
print(' - %s: 新增路径 %d 条 -> %s' % (d['milestone'], d['new_paths'], d['declaration']))
|
||||
print(' - %s: 新增路径 %d 条 -> %s'
|
||||
% (d['milestone'], d['new_paths'], d['declaration']))
|
||||
|
||||
|
||||
def main():
|
||||
ap = argparse.ArgumentParser(description='pbls central RBAC registration (fallback layer)')
|
||||
ap.add_argument('--check', action='store_true', help='aggregate + wildcard audit only')
|
||||
ap.add_argument('--check', action='store_true',
|
||||
help='aggregate + wildcard audit + consistency gate (no DB)')
|
||||
ap.add_argument('--list', action='store_true', help='print aggregated paths and exit')
|
||||
args = ap.parse_args()
|
||||
|
||||
print_milestone_declarations()
|
||||
paths, notes = collect()
|
||||
paths, notes, failures = collect()
|
||||
for note in notes:
|
||||
print('[collect] ' + note)
|
||||
bad = [p for p, _r in paths if '%' in p or '*' in p]
|
||||
bad = [p for p, _r in paths if any(w in p for w in FORBIDDEN_WILDCARDS)]
|
||||
for p in bad:
|
||||
print('WILDCARD FORBIDDEN: ' + p)
|
||||
|
||||
if args.list:
|
||||
for path, role in paths:
|
||||
print('%s %s' % (role, path))
|
||||
return 0 if not bad else 1
|
||||
return 0 if (not bad and not failures) else 1
|
||||
|
||||
if args.check:
|
||||
print('[apps/pbls] check: total=%d wildcard=%d -> %s'
|
||||
% (len(paths), len(bad), 'PASS' if not bad else 'FAIL'))
|
||||
return 0 if not bad else 1
|
||||
# 门禁 1：通配符（零容忍）；门禁 2：聚合完整性（有脚本却聚合失败 = FAIL）
|
||||
print('[apps/pbls] consistency: module_scripts=%d aggregated=%d '
|
||||
'zero_path_declarations=%d failed_modules=%d'
|
||||
% (len([n for n in notes if 'no scripts/load_path.py' not in n]),
|
||||
len([n for n in notes if 'aggregated' in n]),
|
||||
len([n for n in notes if 'NO_WEB_PATHS' in n or '台账 only' in n]),
|
||||
len(failures)))
|
||||
for f in failures:
|
||||
print('AGGREGATION FAILED: ' + f)
|
||||
ok = (not bad) and (not failures)
|
||||
print('[apps/pbls] check: total=%d wildcard=%d aggregation_failures=%d -> %s'
|
||||
% (len(paths), len(bad), len(failures), 'PASS' if ok else 'FAIL'))
|
||||
return 0 if ok else 1
|
||||
|
||||
pending = register(paths)
|
||||
return 0 if (not pending and not bad) else 1
|
||||
return 0 if (not pending and not bad and not failures) else 1
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user