218 lines
10 KiB
Python
218 lines
10 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""pbls 应用（中央宿主）RBAC 注册 —— 双层回退的第二层。
|
||
|
||
背景（module-development-spec「Dual-Layer RBAC」/ QC 退回意见 #1）
|
||
----------------------------------------------------------------
|
||
模块自带的 ``modules/{mod}/scripts/load_path.py`` 直接调 set_role_perm.py，
|
||
实测在部分部署机上会**静默失败**（ModuleNotFoundError 之类），后果是「菜单里看得到、
|
||
点进去一律 403」。因此宿主应用必须有自己的一份注册入口（不同代码路径 = 可靠回退）。
|
||
|
||
本脚本的做法：**聚合**各 pbl_* / world_sync 模块 load_path.py 里的 PATHS 常量，
|
||
而不是抄一份清单。理由：清单抄两遍必然漂移（模块加了新接口，中央层忘了同步 = 又是 403）。
|
||
聚合后中央层与模块层永远一致；同时对 pbl_evidence（M5a 本次退回的模块）保留一份
|
||
**硬编码兜底清单**，即使模块脚本 import 失败也一定登记进去。
|
||
|
||
用法
|
||
----
|
||
python3 scripts/load_path.py # 聚合全部模块路径并注册
|
||
python3 scripts/load_path.py --check # 只做清单一致性核对（不连库）
|
||
python3 scripts/load_path.py --list # 打印最终在册清单
|
||
WS_DIR=/path/to/workspace python3 scripts/load_path.py # 指定机构工作空间
|
||
|
||
铁律：路径逐条显式，禁止通配符（``%`` / ``*``）。
|
||
|
||
═══════════════════════════════════════════════════════════════════════════════
|
||
里程碑 RBAC 声明台账（中央层，与模块层双写等价）
|
||
═══════════════════════════════════════════════════════════════════════════════
|
||
M11b-2（T3，2026-09-22）：
|
||
结论 = 无新路径登记（中央层同样零新增条目）。
|
||
理由 = M11b-2 里程碑（world_sync 列级差异比对 + pbl_runtime_ext 单事务事件/状态写入）
|
||
只交付 **Python 库函数** 与 models/*.json 表定义，**不新增任何
|
||
.dspy / .ui / .js / .css 可访问路径**，不经 HTTP 路由，故 RBAC 零登记项。
|
||
双写 = 等价声明同时写在 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，无需在中央层重复维护清单。
|
||
═══════════════════════════════════════════════════════════════════════════════
|
||
"""
|
||
import argparse
|
||
import importlib.util
|
||
import os
|
||
import subprocess
|
||
import sys
|
||
|
||
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）
|
||
MODULES = [
|
||
'pbl_common', 'pbl_appcodes', 'pbl_governance', 'pbl_blueprint', 'pbl_template',
|
||
'pbl_validation', 'pbl_compiler', 'pbl_agent_runtime', 'pbl_evidence',
|
||
'pbl_assessment', 'pbl_kdb_ext', 'pbl_domain_ext', 'pbl_scense_ext',
|
||
'pbl_analytics', 'pbl_runtime_ext',
|
||
'world_sync',
|
||
]
|
||
|
||
# ── 里程碑声明台账（中央层，与模块层双写；--check / --list 会原样回显）────────
|
||
MILESTONE_DECLARATIONS = [
|
||
{
|
||
'milestone': 'M11b-2',
|
||
'new_paths': 0,
|
||
'declaration': (
|
||
'M11b-2 本里程碑不新增 .dspy/.ui/.js/.css 路径，仅提供库函数'
|
||
'（world_sync / pbl_runtime_ext 的 *.py 与 models/*.json），'
|
||
'故无 RBAC 登记项；禁通配符（%/*），'
|
||
'如有路径须按 any/logined/owner.* 分层逐条登记（中央层与模块层双写）。'
|
||
),
|
||
},
|
||
]
|
||
|
||
# ── pbl_evidence 硬编码兜底清单（与模块层一致；模块脚本 import 失败时由此登记）────
|
||
PBL_EVIDENCE_FALLBACK = [
|
||
('/pbl_evidence/index.ui', 'logined'),
|
||
('/pbl_evidence/api/pbl_artifact_create.dspy', 'logined'),
|
||
('/pbl_evidence/api/pbl_artifact_read.dspy', 'logined'),
|
||
('/pbl_evidence/api/pbl_artifact_update.dspy', 'logined'),
|
||
('/pbl_evidence/api/pbl_artifact_delete.dspy', 'logined'),
|
||
('/pbl_evidence/api/pbl_artifact_list.dspy', 'logined'),
|
||
('/pbl_evidence/api/pbl_evidence_collect.dspy', 'logined'),
|
||
('/pbl_evidence/api/pbl_evidence_collect_from_events.dspy', 'logined'),
|
||
('/pbl_evidence/api/pbl_evidence_list.dspy', 'logined'),
|
||
('/pbl_evidence/api/pbl_evidence_stats.dspy', 'logined'),
|
||
]
|
||
|
||
|
||
def _load_module_paths(mod):
|
||
"""从 modules/{mod}/scripts/load_path.py 读 PATHS（不执行其注册逻辑）。
|
||
|
||
支持两种常量写法：
|
||
① PATHS = [(path, role), ...]
|
||
② PATHS_ANY / PATHS_LOGINED / PATHS_OWNER_SUPERUSER / PATHS_OWNER_OPERATOR
|
||
（分层列表，role 依次映射 any / logined / owner.superuser / owner.operator）
|
||
"""
|
||
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'
|
||
try:
|
||
spec = importlib.util.spec_from_file_location('_lp_%s' % mod, script)
|
||
m = importlib.util.module_from_spec(spec)
|
||
spec.loader.exec_module(m)
|
||
out = []
|
||
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')):
|
||
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,)
|
||
|
||
|
||
def collect():
|
||
"""聚合全部模块 PATHS + 兜底清单，返回 (paths, notes)。paths 已去重排序。"""
|
||
merged, notes = {}, []
|
||
for mod in MODULES:
|
||
if mod == 'pbl_evidence':
|
||
got, err = _load_module_paths(mod)
|
||
# 兜底清单始终并入（union），保证模块脚本坏掉也不漏登记
|
||
for p, r in PBL_EVIDENCE_FALLBACK:
|
||
merged[p] = r
|
||
if err:
|
||
notes.append('%s: %s (fallback list applied)' % (mod, err))
|
||
else:
|
||
notes.append('%s: aggregated %d paths from module script' % (mod, len(got)))
|
||
else:
|
||
got, err = _load_module_paths(mod)
|
||
if err:
|
||
notes.append('%s: %s (skipped)' % (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
|
||
|
||
|
||
def find_set_perm():
|
||
env_tool = os.environ.get('RBAC_SET_PERM')
|
||
if env_tool and os.path.exists(env_tool):
|
||
return env_tool
|
||
for root in (os.path.expanduser('~/repos/sage'), os.path.expanduser('~/test/sage'),
|
||
'/d/ymq/repos/sage', '/opt/sage'):
|
||
cand = os.path.join(root, 'py3', 'bin', 'set_role_perm.py')
|
||
if os.path.exists(cand):
|
||
return cand
|
||
from shutil import which
|
||
return which('set_role_perm.py')
|
||
|
||
|
||
def register(paths, verbose=True):
|
||
tool = find_set_perm()
|
||
if not tool:
|
||
if verbose:
|
||
print('[apps/pbls] set_role_perm.py not found -> %d paths PENDING '
|
||
'(rerun inside the app venv on the deploy host)' % len(paths))
|
||
return [p for p, _r in paths]
|
||
py = sys.executable or 'python3'
|
||
pending = []
|
||
for path, role in paths:
|
||
rc = subprocess.call([py, tool, role, path],
|
||
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
|
||
if rc != 0:
|
||
pending.append((path, role))
|
||
if verbose:
|
||
print('[apps/pbls] rbac paths: total=%d ok=%d pending=%d'
|
||
% (len(paths), len(paths) - len(pending), len(pending)))
|
||
for path, role in pending:
|
||
print(' PENDING %-8s %s' % (role, path))
|
||
return [p for p, _r in pending]
|
||
|
||
|
||
def print_milestone_declarations():
|
||
print('[apps/pbls] 里程碑 RBAC 声明台账（中央层，与模块层双写等价）')
|
||
for d in MILESTONE_DECLARATIONS:
|
||
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('--list', action='store_true', help='print aggregated paths and exit')
|
||
args = ap.parse_args()
|
||
|
||
print_milestone_declarations()
|
||
paths, notes = collect()
|
||
for note in notes:
|
||
print('[collect] ' + note)
|
||
bad = [p for p, _r in paths if '%' in p or '*' in p]
|
||
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
|
||
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
|
||
|
||
pending = register(paths)
|
||
return 0 if (not pending and not bad) else 1
|
||
|
||
|
||
if __name__ == '__main__':
|
||
sys.exit(main())
|