159 lines
6.9 KiB
Python
159 lines
6.9 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""pbl_evidence RBAC 路径注册（硬门禁 6.6 / QC 退回意见 #1 修复）。
|
||
|
||
约定
|
||
----
|
||
- 路径 = 模块自动路由 ``/pbl_evidence/api/<契约>.dspy``，不带端口、不带 /wss 前缀；
|
||
- 角色 ``logined`` = 登录即可访问；本模块无 ``any``（免登录）资源——所有接口都要登录态，
|
||
证据写入是服务端权威行为，绝不允许匿名调用；
|
||
- 由 apps/pbls/build.sh 的 RBAC 步骤调用 ``register()``；
|
||
rbac CLI（set_role_perm.py）不在位时逐条打印 PENDING 清单并以非零码退出（不静默跳过）。
|
||
|
||
铁律（module-development-spec / QC 硬门禁）
|
||
------------------------------------------
|
||
1. **禁止任何通配符**：每条路径逐条显式列出，``%`` / ``*`` 一个都不允许出现在 PATHS 里
|
||
（通配符会让未注册的新接口被误放行，等于 RBAC 形同虚设）。
|
||
2. **清单与磁盘一一对应**：wwwroot 下每个 .dspy / .ui 都必须在册；新增文件必须同步
|
||
追加条目。用 ``python3 scripts/load_path.py --check`` 做自愈核对：
|
||
磁盘有文件但未在册 → 打印 MISSING 并非零退出；在册但文件不存在 → 打印 STALE。
|
||
3. **双层回退**：本脚本是「模块层」。中央宿主 ``apps/pbls/scripts/load_path.py``
|
||
另有一份等价清单（不同代码路径，可靠回退）——模块层 set_role_perm 静默失败时，
|
||
中央层仍能把权限写进去。两层都必须登记本模块全部路径。
|
||
|
||
用法
|
||
----
|
||
python3 scripts/load_path.py # 注册（rbac CLI 不在位则打印 PENDING）
|
||
python3 scripts/load_path.py --check # 只做清单与磁盘的一致性核对
|
||
python3 scripts/load_path.py --list # 打印在册清单
|
||
RBAC_SET_PERM=/path/set_role_perm.py python3 scripts/load_path.py
|
||
"""
|
||
import argparse
|
||
import os
|
||
import subprocess
|
||
import sys
|
||
|
||
MODULE = 'pbl_evidence'
|
||
WWWROOT = os.path.normpath(
|
||
os.path.join(os.path.dirname(os.path.abspath(__file__)), '..', 'wwwroot'))
|
||
|
||
# ── (path, role) 清单：逐条显式，禁通配符 ─────────────────────────────────
|
||
# 覆盖 wwwroot 下全部 9 个 .dspy + 模块入口 index.ui。M5a 主能力
|
||
# pbl_evidence_collect_from_events.dspy 与统计口 pbl_evidence_stats.dspy 在此登记，
|
||
# 漏登记 = 部署后点击一律 403，整条批量采集链路不可用（QC 退回意见 #1 的实测后果）。
|
||
PATHS = [
|
||
# 模块入口页（wwwroot/index.ui）
|
||
('/pbl_evidence/index.ui', 'logined'),
|
||
# 产出物 pbl_artifact CRUD（5 条）
|
||
('/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_runtime_event 采集（M5a 主能力）
|
||
('/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'),
|
||
]
|
||
|
||
STATIC_SUFFIXES = ('.dspy', '.ui', '.js', '.css', '.wss')
|
||
|
||
|
||
def find_set_perm():
|
||
"""定位 rbac 的 set_role_perm.py：环境变量 > 常见 Sage 根 > PATH。"""
|
||
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 on_disk_paths():
|
||
"""扫描 wwwroot 真实文件，换算成路由路径，用于与 PATHS 一致性核对。"""
|
||
got = set()
|
||
if not os.path.isdir(WWWROOT):
|
||
return got
|
||
for base, dirs, files in os.walk(WWWROOT):
|
||
dirs[:] = [d for d in dirs if d not in ('__pycache__', 'node_modules')]
|
||
for fn in files:
|
||
if not fn.endswith(STATIC_SUFFIXES):
|
||
continue
|
||
rel = os.path.relpath(os.path.join(base, fn), WWWROOT).replace(os.sep, '/')
|
||
got.add('/%s/%s' % (MODULE, rel))
|
||
return got
|
||
|
||
|
||
def check():
|
||
"""清单 vs 磁盘一致性核对，返回 (missing, stale, wildcard) 三个列表。"""
|
||
registered = set(p for p, _role in PATHS)
|
||
on_disk = on_disk_paths()
|
||
missing = sorted(on_disk - registered)
|
||
stale = sorted(registered - on_disk)
|
||
wildcard = sorted(p for p, _role in PATHS if '%' in p or '*' in p)
|
||
return missing, stale, wildcard
|
||
|
||
|
||
def register(verbose=True):
|
||
"""逐条调用 set_role_perm.py 注册；返回未成功（待注册）的路径列表。"""
|
||
tool = find_set_perm()
|
||
if not tool:
|
||
# rbac CLI 不在位：不静默跳过，逐条打印待注册清单。部署机由 build.sh 在装有
|
||
# rbac 的 venv 中重跑；中央宿主 apps/pbls/scripts/load_path.py 为第二层回退。
|
||
if verbose:
|
||
print('[%s] set_role_perm.py not found '
|
||
'(check RBAC_SET_PERM / sage root / PATH)' % MODULE)
|
||
return [p for p, _role 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('[%s] rbac paths: total=%d ok=%d pending=%d'
|
||
% (MODULE, 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 main():
|
||
ap = argparse.ArgumentParser(description='%s RBAC path registration' % MODULE)
|
||
ap.add_argument('--check', action='store_true', help='verify PATHS against wwwroot files')
|
||
ap.add_argument('--list', action='store_true', help='print registered paths and exit')
|
||
args = ap.parse_args()
|
||
|
||
if args.list:
|
||
for path, role in PATHS:
|
||
print('%s %s' % (role, path))
|
||
return 0
|
||
|
||
if args.check:
|
||
missing, stale, wildcard = check()
|
||
print('[%s] PATHS entries: %d' % (MODULE, len(PATHS)))
|
||
for p in missing:
|
||
print('MISSING (on disk, not registered): ' + p)
|
||
for p in stale:
|
||
print('STALE (registered, file absent): ' + p)
|
||
for p in wildcard:
|
||
print('WILDCARD FORBIDDEN: ' + p)
|
||
ok = not missing and not stale and not wildcard
|
||
print('[%s] check: %s' % (MODULE, 'PASS' if ok else 'FAIL'))
|
||
return 0 if ok else 1
|
||
|
||
pending = register()
|
||
return 0 if not pending else 1
|
||
|
||
|
||
if __name__ == '__main__':
|
||
sys.exit(main())
|