From ed19f79dae9565d29c47f7d5f013742293e1731b Mon Sep 17 00:00:00 2001 From: "agent.develop" Date: Wed, 16 Sep 2026 15:54:45 +0800 Subject: [PATCH] =?UTF-8?q?deliver:=20=E4=BA=A4=E4=BB=98=E6=94=B6=E5=8F=A3?= =?UTF-8?q?=EF=BC=88=E5=BC=95=E6=93=8E=E4=BB=A3=E4=B8=BA=E6=8F=90=E4=BA=A4?= =?UTF-8?q?=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- pbl_common/__init__.py | 229 +++++++---- pbl_common/kernel.py | 889 +++++++++++++++++------------------------ 2 files changed, 511 insertions(+), 607 deletions(-) diff --git a/pbl_common/__init__.py b/pbl_common/__init__.py index 236bb51..dc5a824 100644 --- a/pbl_common/__init__.py +++ b/pbl_common/__init__.py @@ -1,93 +1,160 @@ # -*- coding: utf-8 -*- -""" -pbl_common —— PBL 公共内核（租户上下文 / DB 适配 / 错误码 / 审计 / CRUD 工厂） +"""pbl_common —— PBL Agent OS 公共内核包（M0） -所有 pbl_* 模块的依赖底座，必须在 app/pbls.py 的 init() 中最先 load。 -铁律： - - 所有读写 tenant_id 强制打头，缺失即抛 PBL-TENANT-0001（fail-closed） - - 取库名一律 ServerEnv().get_module_dbname('模块名')，禁止硬编码 DBNAME - - DB 方言 mariadb（BIGINT AUTO_INCREMENT），无 FK / ENUM / TIMESTAMP +导出：PblError / ERR / TenantContext / require_tenant / get_dbname / + audit_write / crud_factory / WRITE_PROTECTED_TABLES / is_valid_identifier + 以及 _self_test_fail_closed()（六项 fail-closed 断言，可独立运行）。 + +自测运行方式： + cd modules/pbl_common && python -c "import pbl_common; print(pbl_common.self_test())" + 或 python modules/pbl_common/pbl_common/__init__.py +期望输出末行： + SELF_TEST pbl_common: PASS 6/6 """ -from pbl_common.errors import ( +from .kernel import ( # noqa: F401 PblError, - ErrorCode, - ERR_TENANT_MISSING, - ERR_TENANT_INVALID, - ERR_PARAM_INVALID, - ERR_NOT_FOUND, - ERR_CONFLICT, - ERR_FORBIDDEN, - ERR_TOOL_DISABLED, - ERR_WRITE_PROTECTED, -) -from pbl_common.context import ( + ERR, TenantContext, - build_context, - bind_context, - unbind_context, - current_context, require_tenant, - require_context, -) -from pbl_common.tenant import ( - tenant_scope, - assert_tenant, - normalize_tenant, - with_tenant, -) -from pbl_common.dbutil import ( get_dbname, - get_conn, - query, - query_one, - execute, - insert, - transaction, + audit_write, + build_audit_record, + crud_factory, + is_valid_identifier, + WRITE_PROTECTED_TABLES, + MODULE_DB_KEYS, + IDEMPOTENT_CODES, ) -from pbl_common.serialize import ( - to_jsonable, - dumps, - loads, - datetime_to_str, -) -from pbl_common.crud_factory import ( - make_crud, - CrudBase, -) -from pbl_common.tables import ( - TABLES, - ensure_tables, - ddl_of, -) -from pbl_common.audit import ( - write_audit, - AUDIT_ACTIONS, -) - -__version__ = '1.0.0' __all__ = [ - # errors - 'PblError', 'ErrorCode', - 'ERR_TENANT_MISSING', 'ERR_TENANT_INVALID', 'ERR_PARAM_INVALID', - 'ERR_NOT_FOUND', 'ERR_CONFLICT', 'ERR_FORBIDDEN', - 'ERR_TOOL_DISABLED', 'ERR_WRITE_PROTECTED', - # context - 'TenantContext', 'build_context', 'bind_context', 'unbind_context', - 'current_context', 'require_tenant', 'require_context', - # tenant - 'tenant_scope', 'assert_tenant', 'normalize_tenant', 'with_tenant', - # db - 'get_dbname', 'get_conn', 'query', 'query_one', 'execute', 'insert', 'transaction', - # serialize - 'to_jsonable', 'dumps', 'loads', 'datetime_to_str', - # crud - 'make_crud', 'CrudBase', - # tables - 'TABLES', 'ensure_tables', 'ddl_of', - # audit - 'write_audit', 'AUDIT_ACTIONS', - # meta - '__version__', + "PblError", "ERR", "TenantContext", "require_tenant", "get_dbname", + "audit_write", "build_audit_record", "crud_factory", "is_valid_identifier", + "WRITE_PROTECTED_TABLES", "MODULE_DB_KEYS", "IDEMPOTENT_CODES", + "_self_test_fail_closed", "self_test", ] + +__version__ = "1.0.0" + + +def _self_test_fail_closed(): + """六项 fail-closed 断言（纯逻辑，不连库、不依赖平台环境）。 + + A1 无租户上下文且未显式传 tenant_id -> 抛 PBL_E_TENANT（不得返回 None/空） + A2 tenant_id 为空串/空白/非法字符 -> 抛 PBL_E_TENANT + A3 上下文内 require_tenant() 返回上下文 tenant_id；显式入参优先于上下文 + A4 嵌套上下文退出后恢复上层，栈不泄漏 + A5 crud_factory 对写保护基表（如 world/entity/scense）-> 抛 PBL_E_WRITE_LOCK + A6 crud_factory 对非法表名/非法 order_by -> 抛 PBL_E_PARAM； + 审计记录缺 actor/action -> 抛 PBL_E_PARAM + + 返回 (passed, total, failures)；任一断言不成立即视为失败（fail-closed）。 + """ + failures = [] + total = 6 + + def expect_pbl_error(fn, code, tag): + try: + fn() + except PblError as e: + if e.code != code: + failures.append("%s: 期望 %s 实得 %s" % (tag, code, e.code)) + return + except Exception as e: # noqa + failures.append("%s: 抛出非 PblError(%s): %s" % (tag, type(e).__name__, e)) + return + failures.append("%s: 未抛错（fail-open 违规）" % tag) + + # A1 无上下文 -> 必须抛错 + def a1(): + require_tenant() + expect_pbl_error(a1, "PBL_E_TENANT", "A1.no_context") + + # A2 非法 tenant_id + for bad in ("", " ", None, "t#01", 123): + def a2(b=bad): + require_tenant(b) + expect_pbl_error(a2, "PBL_E_TENANT", "A2.bad_tenant(%r)" % (bad,)) + + # A3 上下文取值 + 显式优先 + try: + with TenantContext("T_A", actor="u1"): + got = require_tenant() + if got != "T_A": + failures.append("A3.context: 期望 T_A 实得 %r" % (got,)) + got2 = require_tenant("T_B") + if got2 != "T_B": + failures.append("A3.explicit_priority: 期望 T_B 实得 %r" % (got2,)) + except Exception as e: # noqa + failures.append("A3: 异常 %s" % e) + + # A4 嵌套与栈恢复 + try: + with TenantContext("T_OUT"): + with TenantContext("T_IN"): + if require_tenant() != "T_IN": + failures.append("A4.nested: 内层未生效") + if require_tenant() != "T_OUT": + failures.append("A4.restore: 外层未恢复") + if TenantContext.peek() is not None: + failures.append("A4.leak: 退出后上下文栈未清空") + try: + require_tenant() + failures.append("A4.after_exit: 退出后仍可取到租户（fail-open）") + except PblError: + pass + except Exception as e: # noqa + failures.append("A4: 异常 %s" % e) + + # A5 写保护基表 + for tbl in ("world", "entity", "scense", "script_engine", "rbac_user"): + def a5(t=tbl): + crud_factory("pbl_domain_ext", t) + expect_pbl_error(a5, "PBL_E_WRITE_LOCK", "A5.write_protected(%s)" % tbl) + + # A6 非法表名 / order_by / 审计必填 + expect_pbl_error(lambda: crud_factory("pbl_blueprint", "pbl_blueprint; DROP TABLE x"), + "PBL_E_PARAM", "A6.bad_table") + expect_pbl_error(lambda: crud_factory("pbl_blueprint", "pbl_blueprint", + order_by="id; delete from t"), + "PBL_E_PARAM", "A6.bad_order_by") + expect_pbl_error(lambda: build_audit_record("T_A", None, "create"), + "PBL_E_PARAM", "A6.audit_missing_actor") + expect_pbl_error(lambda: build_audit_record("T_A", "u1", " "), + "PBL_E_PARAM", "A6.audit_missing_action") + + passed = total - (1 if failures else 0) * 0 + # 断言项按 6 项计：只要有失败即逐项计失败 + if failures: + passed = 0 + else: + passed = total + return passed, total, failures + + +def self_test(verbose=True): + """运行六项 fail-closed 自测并打印结果；返回退出码 0/1。""" + passed, total, failures = _self_test_fail_closed() + if verbose: + print("=" * 68) + print("SELF_TEST pbl_common —— fail-closed 六项断言") + print("-" * 68) + for i, name in enumerate(("A1.no_context", "A2.bad_tenant", "A3.context_priority", + "A4.nested_restore", "A5.write_protected", + "A6.param_and_audit"), 1): + hit = [f for f in failures if f.startswith(name.split(".")[0])] + print("[%s] %s%s" % ("FAIL" if hit else "PASS", name, + (" -> " + "; ".join(hit)) if hit else "")) + print("-" * 68) + if failures: + for f in failures: + print(" FAIL detail: %s" % f) + print("SELF_TEST pbl_common: FAIL %d/%d" % (passed, total)) + return 1 + print("SELF_TEST pbl_common: PASS %d/%d" % (passed, total)) + return 0 + + +if __name__ == "__main__": + import sys + sys.exit(self_test()) diff --git a/pbl_common/kernel.py b/pbl_common/kernel.py index 9c20fc4..7988db6 100644 --- a/pbl_common/kernel.py +++ b/pbl_common/kernel.py @@ -1,570 +1,407 @@ # -*- coding: utf-8 -*- -"""pbl_common 公共内核：租户上下文 / DB 适配 / 错误码 / 审计 / CRUD 工厂。 +"""pbl_common.kernel —— PBL Agent OS 公共内核（M0） -设计对齐：projects/pbls/docs/01-design/modules/pbl_common.md -铁律： - * 所有读写 tenant_id 强制打头；缺失租户上下文一律 fail-closed 拒绝（PBL_E_TENANT_MISSING） - * 库名只经 ServerEnv().get_module_dbname('模块名') 获取，禁止硬编码 DBNAME - * DB 方言 mariadb：BIGINT AUTO_INCREMENT 主键 + xxx_code 业务主键；无 FK / 无 ENUM / 无原生 TIMESTAMP - * append-only 表无 update_time、无物理删除 +职责（对齐 docs/01-design/modules/pbl_common.md）： + 1. 租户上下文 TenantContext：tenant_id 强制打头，缺失/空值一律 fail-closed 抛错； + 2. DB 适配 get_dbname()：库名只允许来自 ServerEnv().get_module_dbname()，禁止硬编码； + 3. 错误码 ERR：统一 PBL 错误码注册表（业务码 + HTTP 语义映射）； + 4. 审计 audit_write()：append-only 审计写入，缺 tenant_id/actor/action 直接拒绝； + 5. CRUD 工厂 crud_factory()：基于 sqlor 标准 API（sor.C/U/D/R/I/sqlExe）生成 + tenant 隔离的增删改查函数，表名/字段名做标识符白名单校验，杜绝 SQL 拼接注入。 + +设计铁律： + - fail-closed：任何前置条件不满足 -> 抛 PblError，绝不静默降级、绝不返回空结果冒充成功； + - 零外部副作用：本模块 import 时不连库、不读网络，纯逻辑可单测； + - 写保护：引用模块（rbac/world/scene/entity/scense/scense_runtime/script_engine） + 的基表不得经本工厂写入，WRITE_PROTECTED_TABLES 命中即拒绝。 """ -import hashlib -import json +import re import threading -import time -import uuid -# --------------------------------------------------------------------------- -# 错误码（统一前缀 PBL_E_，HTTP 语义由 api 层映射） -# --------------------------------------------------------------------------- +__all__ = [ + "PblError", + "ERR", + "TenantContext", + "require_tenant", + "get_dbname", + "audit_write", + "crud_factory", + "WRITE_PROTECTED_TABLES", + "is_valid_identifier", +] -ERR_TENANT_MISSING = 'PBL_E_TENANT_MISSING' -ERR_TENANT_MISMATCH = 'PBL_E_TENANT_MISMATCH' -ERR_PARAM_INVALID = 'PBL_E_PARAM_INVALID' -ERR_NOT_FOUND = 'PBL_E_NOT_FOUND' -ERR_DUPLICATE = 'PBL_E_DUPLICATE' -ERR_LOCK_HELD = 'PBL_E_LOCK_HELD' -ERR_VERSION_CONFLICT = 'PBL_E_VERSION_CONFLICT' -ERR_VALIDATION_BLOCKED = 'PBL_E_VALIDATION_BLOCKED' -ERR_COMPILE_PRECONDITION = 'PBL_E_COMPILE_PRECONDITION' -ERR_TOOL_DENIED = 'PBL_E_TOOL_DENIED' -ERR_WRITE_PROTECTED = 'PBL_E_WRITE_PROTECTED' -ERR_APPEND_ONLY = 'PBL_E_APPEND_ONLY' -ERR_DB_UNAVAILABLE = 'PBL_E_DB_UNAVAILABLE' -ERR_INTERNAL = 'PBL_E_INTERNAL' -ERROR_MESSAGES = { - ERR_TENANT_MISSING: '租户上下文缺失，请求被拒绝（fail-closed）', - ERR_TENANT_MISMATCH: '租户上下文与数据归属不一致', - ERR_PARAM_INVALID: '参数不合法', - ERR_NOT_FOUND: '对象不存在', - ERR_DUPLICATE: '对象已存在（唯一键冲突）', - ERR_LOCK_HELD: '蓝图编辑锁被他人持有', - ERR_VERSION_CONFLICT: '乐观锁版本冲突', - ERR_VALIDATION_BLOCKED: '存在 blocker 级校验发现，禁止进入下一阶段', - ERR_COMPILE_PRECONDITION: '编译前置条件不满足（校验未通过）', - ERR_TOOL_DENIED: 'Agent 工具调用被 fail-closed 裁决拒绝', - ERR_WRITE_PROTECTED: '目标为写保护复用域对象，禁止写入', - ERR_APPEND_ONLY: 'append-only 表禁止更新/删除', - ERR_DB_UNAVAILABLE: '数据库不可用', - ERR_INTERNAL: '内部错误', +# -------------------------------------------------------------------------- +# 1. 错误码注册表 +# -------------------------------------------------------------------------- +ERR = { + # 通用 + "PBL_OK": (0, "ok"), + "PBL_E_PARAM": (40001, "参数缺失或非法"), + "PBL_E_TENANT": (40002, "tenant_id 缺失或非法（fail-closed 拒绝）"), + "PBL_E_AUTH": (40101, "未认证或凭据无效"), + "PBL_E_FORBIDDEN": (40301, "无权限执行该操作"), + "PBL_E_WRITE_LOCK": (40302, "目标为写保护对象，禁止写入"), + "PBL_E_NOT_FOUND": (40401, "对象不存在"), + "PBL_E_CONFLICT": (40901, "唯一键冲突或状态冲突"), + "PBL_E_VALIDATION": (42201, "蓝图/对象校验未通过"), + "PBL_E_TOOL_DISABLED": (40303, "Agent 工具已禁用（fail-closed）"), + "PBL_E_DB": (50001, "数据库适配失败：库名映射不可用"), + "PBL_E_INTERNAL": (50000, "内部错误"), } +# 唯一键冲突 -> 幂等语义（采集/注入类接口据此转 PASS 而非报错） +IDEMPOTENT_CODES = ("PBL_E_CONFLICT",) + class PblError(Exception): - """PBL 统一业务异常：code + message + detail。""" + """PBL 统一业务异常。 - def __init__(self, code, message=None, detail=None, http_status=400): - super(PblError, self).__init__(message or ERROR_MESSAGES.get(code, code)) + 属性： + code —— ERR 注册表键名（如 'PBL_E_TENANT'） + errno —— 数字错误码 + msg —— 人类可读信息 + detail—— 附加上下文（dict，可序列化） + """ + + def __init__(self, code, msg=None, detail=None): + if code not in ERR: + # 未注册错误码本身即缺陷，fail-closed 暴露而非吞掉 + code = "PBL_E_INTERNAL" + msg = msg or "未注册错误码" self.code = code - self.message = message or ERROR_MESSAGES.get(code, code) - self.detail = detail or {} - self.http_status = http_status + self.errno, self.default_msg = ERR[code] + self.msg = msg or self.default_msg + self.detail = detail if isinstance(detail, dict) else {} + super(PblError, self).__init__("[%s/%s] %s" % (code, self.errno, self.msg)) def to_dict(self): return { - 'ok': False, - 'code': self.code, - 'message': self.message, - 'detail': self.detail, + "ok": False, + "code": self.code, + "errno": self.errno, + "msg": self.msg, + "detail": self.detail, } -def fail(code, message=None, detail=None, http_status=400): - """抛出 PblError 的快捷函数。""" - raise PblError(code, message=message, detail=detail, http_status=http_status) +# -------------------------------------------------------------------------- +# 2. 租户上下文（tenant_id 强制打头） +# -------------------------------------------------------------------------- +_IDENT_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]{0,63}$") +_TENANT_RE = re.compile(r"^[A-Za-z0-9_\-]{1,64}$") -# --------------------------------------------------------------------------- -# 租户上下文（thread-local，强制打头） -# --------------------------------------------------------------------------- - -_CTX = threading.local() +def is_valid_identifier(name): + """表名/字段名合法性：仅允许字母数字下划线且不以数字开头，长度 <=64。""" + return bool(name) and isinstance(name, str) and bool(_IDENT_RE.match(name)) class TenantContext(object): - """一次请求内的租户上下文快照。""" - - __slots__ = ('tenant_id', 'user_code', 'role_code', 'trace_id', 'client_ip', 'session_code') - - def __init__(self, tenant_id, user_code=None, role_code=None, - trace_id=None, client_ip=None, session_code=None): - self.tenant_id = tenant_id - self.user_code = user_code - self.role_code = role_code - self.trace_id = trace_id or new_trace_id() - self.client_ip = client_ip - self.session_code = session_code - - def as_dict(self): - return { - 'tenant_id': self.tenant_id, - 'user_code': self.user_code, - 'role_code': self.role_code, - 'trace_id': self.trace_id, - 'client_ip': self.client_ip, - 'session_code': self.session_code, - } - - -def new_trace_id(): - """生成链路追踪 ID。""" - return 'trc_%s' % uuid.uuid4().hex[:24] - - -def set_tenant_context(tenant_id, **kwargs): - """设置当前线程租户上下文；tenant_id 为空直接 fail-closed。""" - if not tenant_id or not str(tenant_id).strip(): - fail(ERR_TENANT_MISSING, http_status=403) - ctx = TenantContext(str(tenant_id).strip(), **kwargs) - _CTX.current = ctx - return ctx - - -def get_tenant_context(required=True): - """取当前租户上下文；required=True 且缺失时 fail-closed。""" - ctx = getattr(_CTX, 'current', None) - if ctx is None and required: - fail(ERR_TENANT_MISSING, http_status=403) - return ctx - - -def current_tenant_id(required=True): - """取当前 tenant_id 字符串。""" - ctx = get_tenant_context(required=required) - return ctx.tenant_id if ctx else None - - -def clear_tenant_context(): - """请求结束时清理，避免线程复用串租户。""" - _CTX.current = None - - -def from_request(params, required=True): - """从 api 入参构造租户上下文（api 层统一入口）。""" - params = params or {} - tenant_id = params.get('tenant_id') or params.get('tenantId') - if not tenant_id and required: - fail(ERR_TENANT_MISSING, http_status=403) - if not tenant_id: - return None - return set_tenant_context( - tenant_id, - user_code=params.get('user_code') or params.get('operator'), - role_code=params.get('role_code'), - trace_id=params.get('trace_id'), - client_ip=params.get('client_ip'), - session_code=params.get('session_code'), - ) - - -def assert_tenant_match(row, tenant_id=None): - """校验数据行归属当前租户，不一致即拒绝（防越权读）。""" - tid = tenant_id or current_tenant_id() - row_tid = (row or {}).get('tenant_id') - if row_tid is None: - return row - if str(row_tid) != str(tid): - fail(ERR_TENANT_MISMATCH, detail={'row_tenant': row_tid, 'ctx_tenant': tid}, http_status=403) - return row - - -# --------------------------------------------------------------------------- -# 标识与哈希工具 -# --------------------------------------------------------------------------- - -def new_code(prefix): - """生成业务主键：{prefix}_{22位hex}，全局唯一、可读前缀。""" - return '%s_%s' % (prefix, uuid.uuid4().hex[:22]) - - -def sha256_text(text): - """对文本做 sha256（确定性编译/证据幂等共用）。""" - if text is None: - text = '' - if not isinstance(text, bytes): - text = json_dumps_canonical(text) if not isinstance(text, str) else text - text = text.encode('utf-8') - return hashlib.sha256(text).hexdigest() - - -def json_dumps_canonical(obj): - """规范化 JSON 序列化：键排序、无多余空格、非 ASCII 保留——保证同输入同 hash。""" - return json.dumps(obj, sort_keys=True, ensure_ascii=False, separators=(',', ':'), default=str) - - -def idem_key(*parts): - """证据幂等键：各段拼接后 sha256，长度固定 128 内（列宽 VARCHAR(128)）。""" - joined = '|'.join('' if p is None else str(p) for p in parts) - return 'ik_%s' % sha256_text(joined)[:64] - - -# --------------------------------------------------------------------------- -# DB 适配（sqlor 标准 API：仅 sor.C/U/D/R/I/sqlExe） -# --------------------------------------------------------------------------- - -DIALECT = 'mariadb' - - -class DbAdapter(object): - """sqlor 之上的租户安全适配层。 - - * 库名经 ServerEnv().get_module_dbname(module) 解析，禁止硬编码 - * 所有查询自动注入 tenant_id 条件（强制打头） - * append-only 表拒绝 update/delete - """ - - APPEND_ONLY_TABLES = frozenset([ - 'pbl_audit_log', 'pbl_blueprint_version', 'pbl_subobject_review', 'pbl_blueprint_publish', - 'pbl_validation_finding', 'pbl_compile_log', 'pbl_agent_message', 'pbl_agent_tool_call', - 'pbl_agent_arbitration', 'pbl_evidence_idem', 'pbl_analytics_snapshot', - ]) - - WRITE_PROTECTED_MODULES = frozenset([ - 'rbac', 'accounting', 'apppublic', 'sqlor', 'ahserver', 'appbase', - 'world', 'scene', 'entity', 'script_engine', 'scense_runtime', 'scense', 'scense_game', - ]) - - def __init__(self, module_name='pbl_common', sor=None): - self.module_name = module_name - self._sor = sor - - # -- 连接解析 ----------------------------------------------------------- - def sor(self): - """惰性解析 sqlor 句柄；未注入时从 ServerEnv 取。""" - if self._sor is not None: - return self._sor - try: - from appbase.serverenv import ServerEnv - env = ServerEnv() - sor = getattr(env, 'sor', None) - if sor is None: - dbname = env.get_module_dbname(self.module_name) - sor = env.get_sor(dbname) if hasattr(env, 'get_sor') else None - self._sor = sor - except Exception as exc: - fail(ERR_DB_UNAVAILABLE, detail={'reason': str(exc)}, http_status=503) - if self._sor is None: - fail(ERR_DB_UNAVAILABLE, detail={'reason': 'sor 未就绪'}, http_status=503) - return self._sor - - def dbname(self): - """当前模块库名（唯一合法来源：ServerEnv.get_module_dbname）。""" - from appbase.serverenv import ServerEnv - return ServerEnv().get_module_dbname(self.module_name) - - # -- 租户条件 ----------------------------------------------------------- - def _with_tenant(self, where, tenant_id=None): - """把 tenant_id 合并进 where 字典（强制打头）。""" - tid = tenant_id or current_tenant_id() - merged = dict(where or {}) - merged['tenant_id'] = tid - return merged - - def _guard_append_only(self, table, op): - if op in ('U', 'D') and table in self.APPEND_ONLY_TABLES: - fail(ERR_APPEND_ONLY, detail={'table': table, 'op': op}, http_status=403) - - def _guard_write_protected(self, module_name, op): - if op in ('C', 'U', 'D') and module_name in self.WRITE_PROTECTED_MODULES: - fail(ERR_WRITE_PROTECTED, detail={'module': module_name, 'op': op}, http_status=403) - - # -- 标准 CRUD（sqlor 五法） ------------------------------------------- - def insert(self, table, row, tenant_id=None): - """新增：自动补 tenant_id / create_time / 业务主键（若表约定 xxx_code）。""" - data = dict(row or {}) - data['tenant_id'] = tenant_id or current_tenant_id() - data.setdefault('create_time', now_str()) - if table not in self.APPEND_ONLY_TABLES: - data.setdefault('update_time', data['create_time']) - return self.sor().C(table, data) - - def update(self, table, where, row, tenant_id=None): - """更新：where 强制含 tenant_id；append-only 表拒绝。""" - self._guard_append_only(table, 'U') - data = dict(row or {}) - data.pop('tenant_id', None) # 租户列不可改 - data.pop('id', None) # 物理主键不可改 - data['update_time'] = now_str() - return self.sor().U(table, self._with_tenant(where, tenant_id), data) - - def delete(self, table, where, tenant_id=None): - """删除：where 强制含 tenant_id；append-only 表拒绝（无物理删除）。""" - self._guard_append_only(table, 'D') - return self.sor().D(table, self._with_tenant(where, tenant_id)) - - def get(self, table, where, tenant_id=None): - """单行读取并校验归属。""" - rows = self.sor().R(table, self._with_tenant(where, tenant_id), limit=1) - row = rows[0] if rows else None - return assert_tenant_match(row, tenant_id or current_tenant_id()) if row else None - - def query(self, table, where=None, order_by=None, limit=100, offset=0, tenant_id=None): - """列表读取：tenant_id 强制注入。""" - return self.sor().R( - table, - self._with_tenant(where, tenant_id), - order_by=order_by, - limit=int(limit or 100), - offset=int(offset or 0), - ) or [] - - def count(self, table, where=None, tenant_id=None): - """计数。""" - rows = self.sor().sqlExe( - 'SELECT COUNT(1) AS cnt FROM %s WHERE tenant_id=%%s' % table, - [tenant_id or current_tenant_id()], - ) - return int((rows[0].get('cnt') if rows else 0) or 0) - - def sql(self, sql, args=None): - """原生 SQL 执行（仅 mariadb 方言；调用方自行保证 tenant_id 条件）。""" - return self.sor().sqlExe(sql, list(args or [])) - - def transaction(self): - """单事务上下文（M11a 事件+状态同事务写入依赖此能力）。""" - return DbTransaction(self.sor()) - - -class DbTransaction(object): - """单事务封装：with 块内全部语句同事务，异常整体回滚（原子性门禁）。""" - - def __init__(self, sor): - self._sor = sor - self._conn = None - self._statements = [] - - def __enter__(self): - self._conn = self._sor.begin() if hasattr(self._sor, 'begin') else None - return self - - def execute(self, sql, args=None): - """事务内执行一条语句。""" - self._statements.append((sql, list(args or []))) - if self._conn is not None: - return self._conn.execute(sql, list(args or [])) - return self._sor.sqlExe(sql, list(args or [])) - - def insert(self, table, row): - cols = list(row.keys()) - placeholders = ', '.join(['%s'] * len(cols)) - sql = 'INSERT INTO %s (%s) VALUES (%s)' % (table, ', '.join(cols), placeholders) - return self.execute(sql, [row[c] for c in cols]) - - def __exit__(self, exc_type, exc_val, exc_tb): - if exc_type is None: - if self._conn is not None and hasattr(self._conn, 'commit'): - self._conn.commit() - return False - if self._conn is not None and hasattr(self._conn, 'rollback'): - self._conn.rollback() - return False - - -def now_str(): - """当前时间字符串（DATETIME 格式，禁用原生 TIMESTAMP）。""" - return time.strftime('%Y-%m-%d %H:%M:%S', time.localtime()) - - -_db_singleton = {} - - -def db(module_name='pbl_common', sor=None): - """取模块级 DbAdapter 单例。""" - key = (module_name, id(sor)) - if key not in _db_singleton: - _db_singleton[key] = DbAdapter(module_name=module_name, sor=sor) - return _db_singleton[key] - - -# --------------------------------------------------------------------------- -# 审计（写 pbl_audit_log，append-only；owner.audit 角色独立只读） -# --------------------------------------------------------------------------- - -AUDIT_ACTIONS = { - 'create', 'update', 'delete', 'read', 'validate', 'compile', 'publish', - 'tool_allow', 'tool_deny', 'login', 'logout', 'seed', 'broadcast', 'score', -} - - -def write_audit(action_code, object_type=None, object_code=None, result='success', - detail=None, tenant_id=None, actor_type=None, actor_code=None, - adapter=None): - """追加一条审计记录（append-only，永不更新/删除）。 - - 审计失败不阻断主流程（降级为 stderr 记录），但必须留痕。 - """ - ctx = get_tenant_context(required=False) - tid = tenant_id or (ctx.tenant_id if ctx else None) - if not tid: - # 无租户上下文的审计（如系统启动自检）落 __system__ 租户 - tid = '__system__' - row = { - 'audit_code': new_code('aud'), - 'tenant_id': tid, - 'actor_type': actor_type or (ctx.role_code if ctx and ctx.role_code else 'system'), - 'actor_code': actor_code or (ctx.user_code if ctx else None), - 'action_code': action_code if action_code in AUDIT_ACTIONS else 'read', - 'object_type': object_type, - 'object_code': object_code, - 'result': result, - 'detail': json_dumps_canonical(detail) if detail is not None else None, - 'trace_id': ctx.trace_id if ctx else new_trace_id(), - 'client_ip': ctx.client_ip if ctx else None, - 'create_time': now_str(), - } - try: - adp = adapter or db('pbl_governance') - return adp.sor().C('pbl_audit_log', row) - except Exception as exc: # pragma: no cover - 审计降级 - import sys - sys.stderr.write('[pbl_common.audit] degraded: %s | row=%s\n' % (exc, row)) - return None - - -# --------------------------------------------------------------------------- -# CRUD 工厂（各 pbl_* 模块复用，统一租户/主键/乐观锁/审计语义） -# --------------------------------------------------------------------------- - -class CrudFactory(object): - """按表生成 list/get/create/update/remove 五法，统一注入租户与审计。 + """线程局部租户上下文。 用法： - bp_crud = CrudFactory('pbl_blueprint', code_prefix='bp', code_field='bp_code', - module='pbl_blueprint', optimistic=True, version_field='cur_version') - bp_crud.create({'bp_name': 'xxx', ...}) + with TenantContext(tenant_id, actor='u001'): + rows = api.list_blueprints() + 退出 with 自动恢复上层上下文（支持嵌套）。 """ - def __init__(self, table, code_prefix, code_field='code', module=None, - optimistic=False, version_field='version_no', append_only=False, - adapter=None, search_fields=None): - self.table = table - self.code_prefix = code_prefix - self.code_field = code_field - self.module = module or 'pbl_common' - self.optimistic = optimistic - self.version_field = version_field - self.append_only = append_only - self.search_fields = search_fields or [] - self._adapter = adapter + _local = threading.local() + def __init__(self, tenant_id, actor=None, roles=None): + self.tenant_id = _check_tenant(tenant_id) + self.actor = actor + self.roles = roles or [] + self._prev = None + + # -- 栈式读写 -------------------------------------------------------- + @classmethod + def current(cls): + return getattr(cls._local, "stack", None) + + @classmethod + def peek(cls): + stack = getattr(cls._local, "stack", None) + return stack[-1] if stack else None + + def __enter__(self): + stack = getattr(self._local, "stack", None) + if stack is None: + stack = [] + self._local.stack = stack + self._prev = len(stack) + stack.append(self) + return self + + def __exit__(self, exc_type, exc, tb): + stack = getattr(self._local, "stack", None) + if stack is not None and self._prev is not None: + del stack[self._prev:] + return False + + # -- 便捷属性 -------------------------------------------------------- @property - def adp(self): - return self._adapter or db(self.module) - - def list(self, where=None, order_by=None, limit=100, offset=0, tenant_id=None): - """分页列表（tenant_id 强制）。""" - return self.adp.query(self.table, where=where, order_by=order_by, - limit=limit, offset=offset, tenant_id=tenant_id) - - def get(self, code=None, where=None, tenant_id=None): - """按业务主键或条件取单行。""" - cond = dict(where or {}) - if code: - cond[self.code_field] = code - if not cond: - fail(ERR_PARAM_INVALID, detail={'reason': 'code 或 where 至少给一个'}) - return self.adp.get(self.table, cond, tenant_id=tenant_id) - - def create(self, row, tenant_id=None, audit=True): - """新增：自动生成业务主键、租户打头、乐观锁初值。""" - data = dict(row or {}) - data.setdefault(self.code_field, new_code(self.code_prefix)) - if self.optimistic: - data.setdefault(self.version_field, 1) - if self.append_only: - data.pop('update_time', None) - result = self.adp.insert(self.table, data, tenant_id=tenant_id) - if audit: - write_audit('create', object_type=self.table, object_code=data[self.code_field], - detail={'fields': sorted(data.keys())}, tenant_id=tenant_id, - adapter=self.adp) - return {'ok': True, self.code_field: data[self.code_field], 'id': result} - - def update(self, code, row, expect_version=None, tenant_id=None, audit=True): - """更新：append-only 拒绝；乐观锁版本不匹配拒绝。""" - if self.append_only: - fail(ERR_APPEND_ONLY, detail={'table': self.table}, http_status=403) - data = dict(row or {}) - data.pop(self.code_field, None) - where = {self.code_field: code} - if self.optimistic: - current = self.get(code=code, tenant_id=tenant_id) - if not current: - fail(ERR_NOT_FOUND, detail={'table': self.table, self.code_field: code}, http_status=404) - cur_ver = int(current.get(self.version_field) or 0) - if expect_version is not None and int(expect_version) != cur_ver: - fail(ERR_VERSION_CONFLICT, - detail={'current': cur_ver, 'expect': int(expect_version)}, http_status=409) - data[self.version_field] = cur_ver + 1 - where[self.version_field] = cur_ver - affected = self.adp.update(self.table, where, data, tenant_id=tenant_id) - if self.optimistic and not affected: - fail(ERR_VERSION_CONFLICT, detail={'table': self.table}, http_status=409) - if audit: - write_audit('update', object_type=self.table, object_code=code, - detail={'fields': sorted(data.keys())}, tenant_id=tenant_id, - adapter=self.adp) - return {'ok': True, 'affected': affected} - - def remove(self, code, tenant_id=None, audit=True): - """删除：append-only 表拒绝物理删除。""" - if self.append_only: - fail(ERR_APPEND_ONLY, detail={'table': self.table}, http_status=403) - affected = self.adp.delete(self.table, {self.code_field: code}, tenant_id=tenant_id) - if audit: - write_audit('delete', object_type=self.table, object_code=code, - tenant_id=tenant_id, adapter=self.adp) - return {'ok': True, 'affected': affected} + def actor_id(self): + return self.actor or "system" -def make_crud(table, code_prefix, **kwargs): - """CRUD 工厂函数式入口。""" - return CrudFactory(table, code_prefix, **kwargs) +def _check_tenant(tenant_id): + """tenant_id 校验：None/空串/非字符串/含非法字符 -> PBL_E_TENANT。""" + if tenant_id is None: + raise PblError("PBL_E_TENANT", "tenant_id 为 None") + if not isinstance(tenant_id, str): + raise PblError("PBL_E_TENANT", "tenant_id 必须是字符串", {"got": type(tenant_id).__name__}) + tid = tenant_id.strip() + if not tid: + raise PblError("PBL_E_TENANT", "tenant_id 为空串/空白") + if not _TENANT_RE.match(tid): + raise PblError("PBL_E_TENANT", "tenant_id 含非法字符", {"tenant_id": tid}) + return tid -# --------------------------------------------------------------------------- -# 通用参数校验 -# --------------------------------------------------------------------------- +def require_tenant(explicit=None): + """取当前 tenant_id：显式入参优先，其次上下文；两者皆无 -> fail-closed 抛错。 -MAX_TENANT_LEN = 32 -MAX_CODE_LEN = 32 + 所有对外契约接口的第一行都应调用本函数，确保 tenant_id 强制打头。 + """ + if explicit is not None and str(explicit).strip() != "": + return _check_tenant(explicit) + ctx = TenantContext.peek() + if ctx is None: + raise PblError("PBL_E_TENANT", "无租户上下文且未显式传入 tenant_id") + return ctx.tenant_id -def require(params, *fields): - """必填校验：缺任一字段即 PBL_E_PARAM_INVALID。""" - params = params or {} - missing = [f for f in fields if params.get(f) in (None, '')] - if missing: - fail(ERR_PARAM_INVALID, detail={'missing': missing}) - return params +# -------------------------------------------------------------------------- +# 3. DB 适配（库名只来自 ServerEnv，禁止硬编码） +# -------------------------------------------------------------------------- +#: 模块名 -> 库名逻辑键；实际库名由 ServerEnv().get_module_dbname() 解析。 +#: 这里只登记「模块名」，绝不登记真实库名字符串。 +MODULE_DB_KEYS = ( + "pbl_common", + "pbl_appcodes", + "pbl_blueprint", + "pbl_validation", + "pbl_compiler", + "pbl_agent_runtime", + "pbl_evidence", + "pbl_assessment", + "pbl_kdb_ext", + "pbl_domain_ext", + "pbl_scense_ext", + "pbl_runtime_ext", + "pbl_template", + "pbl_analytics", + "pbl_governance", +) + +#: 写保护基表（引用模块所有）：任何写入路径命中即拒绝（G6/Q-OPEN-3） +WRITE_PROTECTED_TABLES = frozenset( + [ + "rbac_user", "rbac_role", "rbac_permission", "rbac_user_role", + "world", "world_snapshot", "world_sync", "world_sync_task", "world_sync_log", + "scene", "scene_import", + "entity", "entity_import", + "scense", "scense_game", + "script_engine", "script_engine_log", + "kdb_doc", "kdb_chunk", "research_paper", + ] +) -def check_len(value, field, max_len, allow_none=False): - """长度校验（对齐 DDL 列宽）。""" - if value is None: - if allow_none: - return value - fail(ERR_PARAM_INVALID, detail={'field': field, 'reason': 'not null'}) - if len(str(value)) > max_len: - fail(ERR_PARAM_INVALID, detail={'field': field, 'max_len': max_len}) - return value +def get_dbname(module): + """解析模块库名。 + + 规则（fail-closed）： + - module 必须在 MODULE_DB_KEYS 白名单内； + - 必须能从 ServerEnv 取到 get_module_dbname；取不到 -> PBL_E_DB； + - 严禁返回硬编码库名。 + """ + if not module or not isinstance(module, str): + raise PblError("PBL_E_PARAM", "module 名非法") + if module not in MODULE_DB_KEYS: + raise PblError("PBL_E_PARAM", "模块未登记库名映射", {"module": module}) + try: + from appbase.serverenv import ServerEnv # noqa: 平台基础模块 + except Exception: + try: + from appbase import ServerEnv # noqa: 兼容不同导出层级 + except Exception as e: + raise PblError("PBL_E_DB", "ServerEnv 不可用，无法解析库名", {"err": str(e)}) + try: + env = ServerEnv() + fn = getattr(env, "get_module_dbname", None) + if not callable(fn): + raise PblError("PBL_E_DB", "ServerEnv 未挂载 get_module_dbname") + dbname = fn(module) + except PblError: + raise + except Exception as e: + raise PblError("PBL_E_DB", "get_module_dbname 调用失败", {"module": module, "err": str(e)}) + if not dbname or not isinstance(dbname, str): + raise PblError("PBL_E_DB", "库名映射为空", {"module": module}) + return dbname -def check_in(value, field, allowed, allow_none=False): - """枚举校验：值必须落在 appcodes_kv 字典内（DDL 无 ENUM，约束在应用层）。""" - if value is None and allow_none: - return value - if value not in allowed: - fail(ERR_PARAM_INVALID, detail={'field': field, 'allowed': sorted(allowed), 'got': value}) - return value +def _get_sor(module): + """取 sqlor 句柄（延迟 import，便于无 DB 环境跑纯逻辑自测）。""" + try: + import sqlor + except Exception as e: + raise PblError("PBL_E_DB", "sqlor 不可用", {"err": str(e)}) + dbname = get_dbname(module) + try: + return sqlor.sor(dbname) if hasattr(sqlor, "sor") else sqlor.Sor(dbname) + except Exception as e: + raise PblError("PBL_E_DB", "sqlor 连接构造失败", {"dbname_key": module, "err": str(e)}) -def ok(data=None, **extra): - """统一成功返回体。""" - body = {'ok': True} - if data is not None: - body['data'] = data - body.update(extra) - return body +# -------------------------------------------------------------------------- +# 4. 审计（append-only） +# -------------------------------------------------------------------------- +AUDIT_TABLE = "pbl_audit_log" +AUDIT_REQUIRED = ("tenant_id", "actor", "action") + + +def build_audit_record(tenant_id, actor, action, obj_type=None, obj_id=None, + before=None, after=None, result="ok", detail=None): + """构造审计记录（纯函数，不落库）——供自测与调用方复用。 + + fail-closed：tenant_id/actor/action 任一缺失即抛 PBL_E_PARAM。 + """ + rec = { + "tenant_id": _check_tenant(tenant_id), + "actor": (actor or "").strip() if isinstance(actor, str) else actor, + "action": (action or "").strip() if isinstance(action, str) else action, + "obj_type": obj_type, + "obj_id": obj_id, + "before": before, + "after": after, + "result": result or "ok", + "detail": detail, + } + for k in AUDIT_REQUIRED: + v = rec.get(k) + if v is None or (isinstance(v, str) and v.strip() == ""): + raise PblError("PBL_E_PARAM", "审计必填字段缺失: %s" % k) + return rec + + +def audit_write(module, tenant_id, actor, action, **kw): + """写一条审计记录（append-only，不做 UPDATE/DELETE）。""" + rec = build_audit_record(tenant_id, actor, action, **kw) + sor = _get_sor(module) + try: + return sor.C(AUDIT_TABLE, rec) + except Exception as e: + raise PblError("PBL_E_DB", "审计写入失败", {"err": str(e)}) + + +# -------------------------------------------------------------------------- +# 5. CRUD 工厂（tenant 隔离 + 写保护 + 标识符校验） +# -------------------------------------------------------------------------- +def crud_factory(module, tblname, search_fields=None, order_by="id desc", + immutable_fields=None): + """生成一组 tenant 隔离的 CRUD 函数。 + + 参数： + module —— 库名映射所属模块（须在 MODULE_DB_KEYS） + tblname —— 目标表（须为合法标识符，且不在 WRITE_PROTECTED_TABLES） + search_fields —— 关键字模糊搜索字段白名单 + order_by —— 默认排序（须为合法 'field [asc|desc]' 形式） + immutable_fields—— 更新时禁止修改的字段（如 tenant_id/created_at） + + 返回 dict：create / get / list / update / delete / count + """ + if not is_valid_identifier(tblname): + raise PblError("PBL_E_PARAM", "表名非法", {"tblname": tblname}) + if tblname in WRITE_PROTECTED_TABLES: + raise PblError("PBL_E_WRITE_LOCK", "写保护基表禁止经 CRUD 工厂操作", + {"tblname": tblname}) + if not order_by or not re.match(r"^[A-Za-z_][A-Za-z0-9_]{0,63}(\s+(asc|desc))?$", + str(order_by).strip(), re.I): + raise PblError("PBL_E_PARAM", "order_by 非法", {"order_by": order_by}) + + search_fields = tuple(search_fields or ()) + for f in search_fields: + if not is_valid_identifier(f): + raise PblError("PBL_E_PARAM", "search_fields 含非法字段", {"field": f}) + immutable_fields = frozenset(immutable_fields or ("tenant_id",)) + for f in immutable_fields: + if not is_valid_identifier(f): + raise PblError("PBL_E_PARAM", "immutable_fields 含非法字段", {"field": f}) + + def _clean(data, tenant_id, forbid_immutable=False): + if not isinstance(data, dict): + raise PblError("PBL_E_PARAM", "数据必须是 dict") + out = {} + for k, v in data.items(): + if not is_valid_identifier(k): + raise PblError("PBL_E_PARAM", "字段名非法", {"field": k}) + if forbid_immutable and k in immutable_fields: + raise PblError("PBL_E_PARAM", "字段不可修改", {"field": k}) + out[k] = v + out["tenant_id"] = tenant_id # tenant_id 强制打头/强制覆盖 + return out + + def create(data, tenant_id=None): + tid = require_tenant(tenant_id) + return _get_sor(module).C(tblname, _clean(data, tid)) + + def get(pk, tenant_id=None): + tid = require_tenant(tenant_id) + rows = _get_sor(module).R(tblname, {"tenant_id": tid, "id": pk}, limit=1) + if not rows: + raise PblError("PBL_E_NOT_FOUND", "记录不存在", {"tblname": tblname, "id": pk}) + return rows[0] + + def list(where=None, page=1, page_size=20, keyword=None, tenant_id=None): + tid = require_tenant(tenant_id) + cond = {"tenant_id": tid} + if isinstance(where, dict): + for k, v in where.items(): + if not is_valid_identifier(k): + raise PblError("PBL_E_PARAM", "查询字段非法", {"field": k}) + cond[k] = v + kw = None + if keyword and search_fields: + kw = {"fields": list(search_fields), "value": keyword} + try: + page = max(1, int(page)) + page_size = min(500, max(1, int(page_size))) + except (TypeError, ValueError): + raise PblError("PBL_E_PARAM", "分页参数非法") + return _get_sor(module).R(tblname, cond, order=order_by, + page=page, page_size=page_size, keyword=kw) + + def count(where=None, tenant_id=None): + tid = require_tenant(tenant_id) + cond = {"tenant_id": tid} + if isinstance(where, dict): + for k, v in where.items(): + if not is_valid_identifier(k): + raise PblError("PBL_E_PARAM", "查询字段非法", {"field": k}) + cond[k] = v + rows = _get_sor(module).R(tblname, cond) + return len(rows or []) + + def update(pk, data, tenant_id=None): + tid = require_tenant(tenant_id) + payload = _clean(data, tid, forbid_immutable=True) + return _get_sor(module).U(tblname, {"tenant_id": tid, "id": pk}, payload) + + def delete(pk, tenant_id=None): + tid = require_tenant(tenant_id) + return _get_sor(module).D(tblname, {"tenant_id": tid, "id": pk}) + + return { + "create": create, + "get": get, + "list": list, + "count": count, + "update": update, + "delete": delete, + "tblname": tblname, + "module": module, + }