80 lines
3.2 KiB
Python
80 lines
3.2 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""pbl_runtime_ext RBAC 路径注册（硬门禁 6.6 / QC #11 / M11b QC #8）。
|
||
|
||
约定：
|
||
- 路径 = 模块自动路由 `/pbl_runtime_ext/api/<契约>.dspy`，不带端口、不带 /wss 前缀；
|
||
- 角色 `logined` = 登录即可访问；本模块 8 个契约全部要求登录（含写接口），
|
||
**禁止通配符**（module-development-spec 硬规定：每条路径显式列出）；
|
||
- 由 apps/pbls/build.sh 第 8 步调用 `register()`；rbac CLI 不在位时打印 PENDING 清单
|
||
（返回 False，不静默跳过、不冒充成功）。
|
||
|
||
维护规则：wwwroot/api/ 下 .dspy 增删必须同步本清单 —— `check_paths()` 会在
|
||
自检阶段比对磁盘文件与 PATHS，缺登记或多登记都判 FAIL。
|
||
"""
|
||
import os
|
||
import subprocess
|
||
import sys
|
||
|
||
MODULE = 'pbl_runtime_ext'
|
||
HERE = os.path.dirname(os.path.abspath(__file__))
|
||
WWWROOT_API = os.path.join(os.path.dirname(HERE), 'wwwroot', 'api')
|
||
|
||
# (path, role) —— M11a 6 个 + M11b 广播 2 个，共 8 个显式路径
|
||
PATHS = [
|
||
('/pbl_runtime_ext/api/pbl_runtime_event_append.dspy', 'logined'),
|
||
('/pbl_runtime_ext/api/pbl_runtime_event_poll.dspy', 'logined'),
|
||
('/pbl_runtime_ext/api/pbl_entity_state_get.dspy', 'logined'),
|
||
('/pbl_runtime_ext/api/pbl_entity_state_apply.dspy', 'logined'),
|
||
('/pbl_runtime_ext/api/pbl_world_broadcast.dspy', 'logined'),
|
||
('/pbl_runtime_ext/api/pbl_session_member_add.dspy', 'logined'),
|
||
('/pbl_runtime_ext/api/pbl_runtime_broadcast_pull.dspy', 'logined'),
|
||
('/pbl_runtime_ext/api/pbl_runtime_broadcast_stats.dspy', 'logined'),
|
||
]
|
||
|
||
|
||
def on_disk():
|
||
"""磁盘上真实存在的 .dspy 文件名集合（小写，用于比对）。"""
|
||
try:
|
||
return set(f for f in os.listdir(WWWROOT_API) if f.endswith('.dspy'))
|
||
except OSError:
|
||
return set()
|
||
|
||
|
||
def check_paths():
|
||
"""清单 ↔ 磁盘一致性自检。返回 (ok, missing_in_list, stale_in_list)。"""
|
||
listed = set(os.path.basename(p) for p, _ in PATHS)
|
||
disk = on_disk()
|
||
missing = sorted(disk - listed) # 有文件没登记 → 上线必 403
|
||
stale = sorted(listed - disk) # 登记了但文件不存在 → 死路径
|
||
return (not missing and not stale), missing, stale
|
||
|
||
|
||
def register():
|
||
tool = os.environ.get('RBAC_SET_PERM', 'set_role_perm.py')
|
||
done, missing = 0, []
|
||
for path, role in PATHS:
|
||
cmd = [os.environ.get('PY', 'python3'), tool, role, path]
|
||
try:
|
||
rc = subprocess.call(cmd, stdout=subprocess.DEVNULL,
|
||
stderr=subprocess.DEVNULL)
|
||
except Exception: # noqa: BLE001 工具不在位也不能崩，记 PENDING
|
||
rc = 1
|
||
if rc == 0:
|
||
done += 1
|
||
else:
|
||
missing.append((path, role))
|
||
print('[%s] rbac paths: total=%d ok=%d pending=%d'
|
||
% (MODULE, len(PATHS), done, len(missing)))
|
||
for path, role in missing:
|
||
print(' PENDING %-12s %s' % (role, path))
|
||
ok_list, absent, stale = check_paths()
|
||
if not ok_list:
|
||
print(' PATHS-DISK-MISMATCH missing_in_list=%s stale_in_list=%s'
|
||
% (absent, stale))
|
||
return len(missing) == 0 and ok_list
|
||
|
||
|
||
if __name__ == '__main__':
|
||
sys.exit(0 if register() else 1)
|