234 lines
11 KiB
Python
234 lines
11 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 下全部 12 个 .dspy + 模块入口 index.ui（与磁盘一一对应，
|
||
# `python3 scripts/load_path.py --check` 做自愈核对）。M5a 主能力
|
||
# pbl_evidence_collect_from_events.dspy 与统计口 pbl_evidence_stats.dspy 在此登记，
|
||
# CRUD 适配三端点 pbl_artifact_options / pbl_evidence_update / pbl_evidence_delete
|
||
# （json/pbl_evidence.json 的 editable 三 URL + alters.artifact_id.dataurl 所指）也必须登记，
|
||
# 漏登记 = 部署后点击一律 403，整条批量采集/编辑/删除链路不可用（QC 退回意见 #1/#3 的实测后果）。
|
||
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'),
|
||
# ── M5b 证据聚合与查询契约（幂等采集包装 / 聚合 / 分页查询 / 主体档案）
|
||
# 新增 .dspy 必须同步登记，漏登记 = 部署后 403（load_path.py --check 会报 MISSING）
|
||
('/pbl_evidence/api/pbl_evidence_collect_idempotent.dspy', 'logined'),
|
||
('/pbl_evidence/api/pbl_evidence_aggregate.dspy', 'logined'),
|
||
('/pbl_evidence/api/pbl_evidence_query.dspy', 'logined'),
|
||
('/pbl_evidence/api/pbl_evidence_subject_summary.dspy', 'logined'),
|
||
# CRUD 框架适配端点（json/pbl_evidence.json editable 三 URL + artifact_id 下拉）
|
||
('/pbl_evidence/api/pbl_artifact_options.dspy', 'logined'),
|
||
('/pbl_evidence/api/pbl_evidence_update.dspy', 'logined'),
|
||
('/pbl_evidence/api/pbl_evidence_delete.dspy', 'logined'),
|
||
# ── CRUD 生成物见下方 GENERATED_PATHS（xls2ui 产物，build 时才落盘）──
|
||
]
|
||
|
||
# xls2ui 从 json/pbl_artifact.json / json/pbl_evidence.json 生成的 CRUD 目录
|
||
# （alias = tblname = pbl_artifact / pbl_evidence）。每个 alias 6 条：目录 +
|
||
# index.ui + get_/add_/update_/delete_ 端点，逐条显式、禁通配符。这些文件是
|
||
# **构建产物**（.gitignore 排除、不入库），所以只出现在部署环境的 wwwroot 下；
|
||
# 注册权限时必须一并登记，否则 CRUD 列表页/新增/编辑/删除一律 403。
|
||
CRUD_ALIASES = ('pbl_artifact', 'pbl_evidence')
|
||
GENERATED_PATHS = []
|
||
for _alias in CRUD_ALIASES:
|
||
GENERATED_PATHS.append(('/pbl_evidence/%s' % _alias, 'logined'))
|
||
GENERATED_PATHS.append(('/pbl_evidence/%s/index.ui' % _alias, 'logined'))
|
||
for _op in ('get', 'add', 'update', 'delete'):
|
||
GENERATED_PATHS.append(
|
||
('/pbl_evidence/%s/%s_%s.dspy' % (_alias, _op, _alias), 'logined'))
|
||
|
||
# 注册用的全量清单 = 手写端点 + CRUD 生成物（顺序即登记顺序）
|
||
PATHS = PATHS + GENERATED_PATHS
|
||
|
||
# ── permission / rolepermission 幂等 INSERT（供部署说明与手工兜底使用）──────
|
||
# 与 PATHS 一一对应：rbac 三表链路 permission.permcode = 路径本身，
|
||
# rolepermission 关联 role.rolecode='logined'。列名以 models 真实字段为准
|
||
# （permission: permcode/name；role: rolecode；rolepermission: roleid/permid），
|
||
# 用 INSERT ... SELECT 反查 id，避免硬编码主键。
|
||
def permission_inserts():
|
||
"""返回与 PATHS 等价的幂等 SQL（角色固定 logined）。"""
|
||
out = []
|
||
for path, role in PATHS:
|
||
out.append(
|
||
"INSERT INTO `permission` (`id`, `permcode`, `name`) "
|
||
"SELECT REPLACE(UUID(),'-',''), '%(p)s', '%(p)s' "
|
||
"WHERE NOT EXISTS (SELECT 1 FROM `permission` WHERE `permcode`='%(p)s');"
|
||
% {'p': path})
|
||
out.append(
|
||
"INSERT INTO `rolepermission` (`id`, `roleid`, `permid`) "
|
||
"SELECT REPLACE(UUID(),'-',''), r.`id`, p.`id` FROM `role` r, `permission` p "
|
||
"WHERE r.`rolecode`='%(rl)s' AND p.`permcode`='%(p)s' "
|
||
"AND NOT EXISTS (SELECT 1 FROM `rolepermission` rp "
|
||
"WHERE rp.`roleid`=r.`id` AND rp.`permid`=p.`id`);"
|
||
% {'rl': role, 'p': path})
|
||
return out
|
||
|
||
def print_permission_sql():
|
||
"""--sql 子命令：打印 INSERT 清单（部署时可直接管道进 mysql）。"""
|
||
for stmt in permission_inserts():
|
||
print(stmt)
|
||
|
||
|
||
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(include_generated=False):
|
||
"""清单 vs 磁盘一致性核对，返回 (missing, stale, wildcard) 三个列表。
|
||
|
||
CRUD 生成物（GENERATED_PATHS）是 build.sh 里 xls2ui 的产物，开发仓库中被
|
||
.gitignore 排除、不落盘，因此默认不计入 stale（否则开发环境永远误报）；
|
||
部署后跑 `--check --generated` 才把它们纳入核对。手写端点与磁盘必须一一对应。
|
||
"""
|
||
handwritten = set(p for p, _r in PATHS) - set(p for p, _r in GENERATED_PATHS)
|
||
registered = set(p for p, _role in PATHS) if include_generated else handwritten
|
||
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('--generated', action='store_true',
|
||
help='with --check: also reconcile xls2ui-generated CRUD paths '
|
||
'(run this on the deployment host, after build.sh)')
|
||
ap.add_argument('--list', action='store_true', help='print registered paths and exit')
|
||
ap.add_argument('--sql', action='store_true',
|
||
help='print idempotent permission/rolepermission INSERTs and exit')
|
||
args = ap.parse_args()
|
||
|
||
if args.sql:
|
||
print_permission_sql()
|
||
return 0
|
||
|
||
if args.list:
|
||
for path, role in PATHS:
|
||
print('%s %s' % (role, path))
|
||
return 0
|
||
|
||
if args.check:
|
||
missing, stale, wildcard = check(include_generated=args.generated)
|
||
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())
|