diff --git a/pbl_common/audit.py b/pbl_common/audit.py index 2f0552d..d2a823a 100644 --- a/pbl_common/audit.py +++ b/pbl_common/audit.py @@ -1,106 +1,101 @@ # -*- coding: utf-8 -*- """ -pbl_common.audit —— 审计日志（append-only，独立于业务表） +pbl_common.audit —— append-only 审计（pbl_audit_event） -设计（对齐 audit-logging 通用规范 + docs/01-design/data-model.md pbl_audit_log）： - - append-only：只提供 write_audit / query_audit，不提供 update/delete - - 审计独立性：写审计失败不得影响业务主流程（降级为 stderr 记录） - - tenant_id 打头，trace_id 贯穿一次 Agent 调用链 +铁律： + * 只暴露 append()，绝不暴露 update/delete/remove/purge（self_check 校验） + * 审计独立性：审计写入失败不阻断业务，但必须落降级日志 + * 审计角色 owner.audit 才可跨租户读取 """ -import datetime -import sys -import traceback +import time +import uuid -from pbl_common.context import current_context -from pbl_common.serialize import dumps +from pbl_common.tenant import TenantContext, AUDIT_ROLE +from pbl_common.db import get_db -AUDIT_TABLE = 'pbl_audit_log' - -# 允许的审计动作（白名单，未登记动作记为 'other' 但不拒绝） -AUDIT_ACTIONS = ( - 'create', 'read', 'update', 'delete', - 'validate', 'compile', 'fork', 'publish', 'archive', - 'tool_invoke', 'tool_deny', - 'evidence_collect', 'assess', - 'login', 'logout', 'perm_deny', - 'seed', 'migrate', 'other', -) +AUDIT_TABLE = 'pbl_audit_event' +MODULE = 'pbl_governance' def _now(): - return datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S') + return time.strftime('%Y-%m-%d %H:%M:%S') -def write_audit(action, resource_type=None, resource_id=None, - tenant_id=None, user_id=None, result='success', - detail=None, trace_id=None, module='pbl_common', - conn=None): +def _trace_id(ctx): + tid = getattr(ctx, 'trace_id', '') or '' + return tid or uuid.uuid4().hex[:32] + + +def append(ctx, event_type, obj_type='', obj_id=0, session_id=0, + detail=None, actor_role=None, actor_id=None): """ - 写一条审计记录（append-only）。 - - 返回 True/False；内部异常一律吞掉并降级到 stderr， - 保证「审计写失败不阻断业务」，同时不静默丢失（stderr 留痕）。 + 追加一条审计事件（append-only）。 + ctx: TenantContext（必须）；缺失即抛，绝不静默。 + 返回 audit_id；写库失败返回 0 并打印降级日志（不阻断业务）。 """ - ctx = current_context() + if not isinstance(ctx, TenantContext): + raise ValueError('audit.append 需要 TenantContext') + row = { - 'tenant_id': tenant_id or (ctx.tenant_id if ctx else None), - 'action': action if action in AUDIT_ACTIONS else 'other', - 'action_raw': action, - 'resource_type': resource_type, - 'resource_id': str(resource_id) if resource_id is not None else None, - 'user_id': user_id or (ctx.user_id if ctx else None), - 'role': (ctx.role if ctx else None), - 'session_id': (ctx.session_id if ctx else None), - 'trace_id': trace_id or (ctx.trace_id if ctx else None), - 'result': result, - 'detail': dumps(detail) if detail is not None else None, - 'created_at': _now(), + 'tenant_id': ctx.tenant_id, + 'actor_role': actor_role or ctx.role or '', + 'actor_id': int(actor_id if actor_id is not None else (ctx.user_id or 0)), + 'event_type': event_type or '', + 'obj_type': obj_type or '', + 'obj_id': int(obj_id or 0), + 'session_id': int(session_id or ctx.session_id or 0), + 'detail_json': _dumps(detail), + 'trace_id': _trace_id(ctx), + 'event_time': _now(), } - if not row['tenant_id']: - # 审计也必须有租户；无租户时降级 stderr，不抛错打断业务 - sys.stderr.write('[pbl_audit] skip: missing tenant_id action=%s\n' % action) - return False - - cols = [c for c in row.keys() if c != 'action_raw'] - sql = 'INSERT INTO `%s` (%s) VALUES (%s)' % ( - AUDIT_TABLE, - ', '.join('`%s`' % c for c in cols), - ', '.join(['%s'] * len(cols)), - ) - params = [row[c] for c in cols] try: - from pbl_common.dbutil import execute - execute(sql, params, module=module, conn=conn) - return True - except Exception: # noqa: BLE001 - sys.stderr.write('[pbl_audit] write failed action=%s tenant=%s\n%s\n' - % (action, row['tenant_id'], traceback.format_exc())) - return False + db = get_db(MODULE) + try: + return db.C(AUDIT_TABLE, row) + finally: + try: + db.close() + except Exception: # noqa: BLE001 + pass + except Exception as e: # noqa: BLE001 + print('[pbl_common.audit] DEGRADED append failed: %s | row=%s' % (e, row)) + return 0 -def query_audit(tenant_id, action=None, resource_type=None, resource_id=None, - limit=100, offset=0, module='pbl_common'): - """只读查询审计（tenant_id 打头，强制分页上限 500）""" - from pbl_common.dbutil import query - from pbl_common.tenant import normalize_tenant +def _dumps(detail): + if detail is None: + return '' + if isinstance(detail, str): + return detail + try: + import json + return json.dumps(detail, ensure_ascii=False) + except Exception: # noqa: BLE001 + return str(detail) - tid = normalize_tenant(tenant_id) - where = ['tenant_id = %s'] - args = [tid] - if action: - where.append('action = %s') - args.append(action) - if resource_type: - where.append('resource_type = %s') - args.append(resource_type) - if resource_id is not None: - where.append('resource_id = %s') - args.append(str(resource_id)) - limit = max(1, min(int(limit or 100), 500)) - offset = max(0, int(offset or 0)) - sql = ('SELECT * FROM `%s` WHERE %s ORDER BY id DESC LIMIT %%s OFFSET %%s' - % (AUDIT_TABLE, ' AND '.join(where))) - args.extend([limit, offset]) - return query(sql, args, module=module, tenant_id=tid) +def query(ctx, cond=None, limit=100, offset=0): + """ + 审计查询（只读）。跨租户仅 owner.audit 角色允许。 + """ + if not isinstance(ctx, TenantContext): + raise ValueError('audit.query 需要 TenantContext') + where = ctx.scope(cond or {}) + try: + db = get_db(MODULE) + try: + return db.R(AUDIT_TABLE, where, order_by='event_time desc,id desc', + limit=limit, offset=offset) or [] + finally: + try: + db.close() + except Exception: # noqa: BLE001 + pass + except Exception as e: # noqa: BLE001 + print('[pbl_common.audit] query failed: %s' % e) + return [] + + +def can_cross_tenant(ctx): + return bool(getattr(ctx, 'role', '') == AUDIT_ROLE) diff --git a/pbl_common/crud.py b/pbl_common/crud.py index 7efda8b..b25621c 100644 --- a/pbl_common/crud.py +++ b/pbl_common/crud.py @@ -1,240 +1,169 @@ # -*- coding: utf-8 -*- -"""CRUD 工厂：按「表定义四段式（summary/fields/indexes/codes）」生成标准增删改查。 +""" +pbl_common.crud —— CRUD 工厂（sqlor 标准 API：仅 C/U/D/R/I/sqlExe） -生成物契约（tblname / params）： - create(params) -> {'id': int} - read(id) -> dict | None - update(id, params) -> int(affected) - soft_delete(id) -> int(affected) # is_deleted=1，业务表默认软删 - delete(id) -> int(affected) # 物理删除，仅治理白名单表允许 - list(where, page, size, order_by) -> {'total': int, 'rows': [...]} - get_by_code(code) -> dict | None - -所有方法首参/内部一律强制 tenant_id 打头，缺失即抛 PblError(E_TENANT_MISSING)。 +铁律： + * 不编造 save/list/insert/update/delete/query 等非标准方法 + * 所有读写条件一律经 TenantContext.scope() 包装，tenant_id 强制打头 + * 写操作（C/U/D）自动补 tenant_id / create_time / update_time / is_deleted + * 软删除：D 默认置 is_deleted=1，不物理删（审计/证据/日志表除外） """ -import datetime +import time -from . import db as _db -from .errors import PblError, ErrCode -from .tenant import require_tenant -from .serialize import normalize_code +from pbl_common.tenant import TenantContext, require_tenant, PBLTenantError +from pbl_common.errors import PBLError, PBL_ERR +from pbl_common.db import get_db -# 审计/通用列（写入时自动补齐） -_AUTO_COLS = ('create_time', 'update_time', 'create_user', 'update_user', 'is_deleted') +# 物理删除白名单（append-only 表不允许删；这些是可重算的临时表） +HARD_DELETE_TABLES = () + +# 只读表（append-only：审计 / 证据日志 / 工具调用 / 校验问题） +APPEND_ONLY_TABLES = ( + 'pbl_audit_event', + 'pbl_evidence_collect_log', + 'pbl_tool_call', + 'pbl_validation_issue', + 'pbl_kdb_query_log', + 'pbl_blueprint_change_log', +) + +# 写保护模块的基表（PBL 侧一律不得写） +WRITE_PROTECTED_TABLES = ( + 'world', 'scene', 'entity', 'scense', 'scense_game', + 'scense_runtime', 'script_engine', 'rbac_role', 'rbac_user', + 'rbac_permission', 'app_codes', +) -class CrudFactory(object): - """按表定义生成 CRUD 方法集。 +def _now(): + return time.strftime('%Y-%m-%d %H:%M:%S') - :param table: 表名 - :param fields: 字段名列表（不含 id）；用于写入白名单过滤，防止越权写系统列 - :param codes: {字段: 允许取值集合}，对应「四段式」的 codes 段（替代 ENUM） - :param pk: 主键列名，默认 id - :param soft_delete:是否启用逻辑删除（表含 is_deleted 列时为 True） - :param dbname: 库名（缺省走 get_module_dbname） - :param module: 模块名（用于取库名） - """ - def __init__(self, table, fields, codes=None, pk='id', soft_delete=True, - dbname=None, module='pbl_common', unique_keys=None): - self.table = table - self.fields = tuple(fields or ()) - self.codes = codes or {} - self.pk = pk - self.soft_delete = bool(soft_delete) +def _assert_writable(tblname): + if tblname in WRITE_PROTECTED_TABLES: + raise PBLError(PBL_ERR['agent_tool']['write_protected'], + '表 %s 属写保护模块，PBL 侧禁止写入' % tblname) + + +def _ctx(params, strict=True): + """从 params 取/构造租户上下文。""" + if isinstance(params, TenantContext): + return params + if strict: + return require_tenant(params if isinstance(params, dict) else {}) + try: + return require_tenant(params if isinstance(params, dict) else {}) + except PBLTenantError: + return None + + +class Crud(object): + """单表 CRUD 封装（方法名严格对齐 sqlor：C/U/D/R/I）。""" + + def __init__(self, tblname, module=None, pk='id', soft_delete=True, + tenant_field='tenant_id'): + self.tblname = tblname self.module = module - self.unique_keys = unique_keys or [] - self._dbname = dbname + self.pk = pk + self.soft_delete = soft_delete and tblname not in APPEND_ONLY_TABLES + self.tenant_field = tenant_field + self._db = None - # ---- 内部工具 ---- + # ---- DB 句柄 ------------------------------------------------------- @property - def dbname(self): - return self._dbname or _db.get_dbname(self.module) + def db(self): + if self._db is None: + self._db = get_db(self.module) + return self._db - def _tid(self, tenant_id, params=None): - if tenant_id is None and params: - tenant_id = params.get('tenant_id') - return require_tenant(tenant_id) + def close(self): + if self._db is not None: + try: + self._db.close() + except Exception: # noqa: BLE001 + pass + self._db = None - def _filter(self, params): - """写入白名单过滤 + codes 取值校验（替代 ENUM 约束）。""" - out = {} - for k, v in (params or {}).items(): - if k in ('id', self.pk): - continue - if self.fields and k not in self.fields and k not in _AUTO_COLS and k != 'tenant_id': - raise PblError(ErrCode.E_PARAM_INVALID, - '字段不在表定义白名单内：%s.%s' % (self.table, k)) - if k in self.codes: - allowed = self.codes[k] - if v is not None and normalize_code(v) not in allowed: - raise PblError(ErrCode.E_PARAM_INVALID, - '字段 %s.%s 取值非法：%r（允许：%s）' - % (self.table, k, v, sorted(allowed))) - v = normalize_code(v) - if isinstance(v, bool): - v = 1 if v else 0 - out[k] = v - return out - - def _now(self): - return datetime.datetime.now().replace(microsecond=0) - - # ---- C ---- - def create(self, params, tenant_id=None, user_id=None): - tid = self._tid(tenant_id, params) - data = self._filter(params) - data['tenant_id'] = tid - now = self._now() - if 'is_deleted' in self.fields or self.soft_delete: - data.setdefault('is_deleted', 0) - if 'create_time' in self.fields: - data.setdefault('create_time', now) - if 'update_time' in self.fields: - data.setdefault('update_time', now) - if user_id and 'create_user' in self.fields: - data.setdefault('create_user', str(user_id)[:32]) - if user_id and 'update_user' in self.fields: - data.setdefault('update_user', str(user_id)[:32]) - - # 唯一键预检（无 FK/无 UNIQUE 兜底时给出可读错误） - for uk in self.unique_keys: - cond = {c: data.get(c) for c in uk if c in data} - if len(cond) == len(uk): - cond['tenant_id'] = tid - if self.soft_delete: - cond['is_deleted'] = 0 - if _db.query_one(self.table, cond, tenant_id=tid, dbname=self.dbname): - raise PblError(ErrCode.E_DUPLICATE, - '唯一键冲突：%s %s' % (self.table, list(uk)), {'key': cond}) - return _db.insert(self.table, data, tenant_id=tid, dbname=self.dbname) - - # ---- R ---- - def read(self, pk_value, tenant_id=None, include_deleted=False): - tid = require_tenant(tenant_id) - where = {self.pk: pk_value} - if self.soft_delete and not include_deleted: - where['is_deleted'] = 0 - row = _db.query_one(self.table, where, tenant_id=tid, dbname=self.dbname) - if row is None: - raise PblError(ErrCode.E_NOT_FOUND, - '%s[%s=%s] 不存在（tenant=%s）' % (self.table, self.pk, pk_value, tid)) - return row - - def get_by_code(self, code, tenant_id=None): - tid = require_tenant(tenant_id) - where = {'code': normalize_code(code)} + # ---- C：新增 ------------------------------------------------------- + def C(self, params): + """新增一行。params 必须含 tenant_id（缺失即 fail-closed）。""" + _assert_writable(self.tblname) + ctx = _ctx(params) + row = dict(params) + row[self.tenant_field] = ctx.tenant_id + row.setdefault('create_time', _now()) + row.setdefault('update_time', _now()) + row.setdefault('create_user', ctx.user_id) if self.soft_delete: + row.setdefault('is_deleted', 0) + row.pop(self.pk, None) + return self.db.C(self.tblname, row) + + # ---- U：更新 ------------------------------------------------------- + def U(self, params, cond=None): + """按条件更新。cond 会被 scope() 包进租户作用域。""" + _assert_writable(self.tblname) + ctx = _ctx(params) + row = dict(params) + row.pop(self.tenant_field, None) # 租户列不可改 + row.pop(self.pk, None) + row['update_time'] = _now() + row['update_user'] = ctx.user_id + where = ctx.scope(cond or {}) + if self.soft_delete and 'is_deleted' not in where: where['is_deleted'] = 0 - return _db.query_one(self.table, where, tenant_id=tid, dbname=self.dbname) + return self.db.U(self.tblname, row, where) - def list(self, where=None, page=1, size=20, order_by=None, tenant_id=None, - include_deleted=False, fields=None): - tid = require_tenant(tenant_id if tenant_id is not None else (where or {}).get('tenant_id')) - cond = {} - for k, v in (where or {}).items(): - if k == 'tenant_id' or v is None: - continue - if k in self.codes: - v = normalize_code(v) - cond[k] = v - if self.soft_delete and not include_deleted: - cond.setdefault('is_deleted', 0) - page = max(1, int(page or 1)) - size = min(500, max(1, int(size or 20))) - rows = _db.query(self.table, cond, fields=fields, - order_by=order_by or ('%s DESC' % self.pk), - limit=size, offset=(page - 1) * size, - tenant_id=tid, dbname=self.dbname) - total = len(_db.query(self.table, cond, fields=[self.pk], - limit=100000, tenant_id=tid, dbname=self.dbname)) - return {'total': total, 'page': page, 'size': size, 'rows': rows or []} + # ---- D：删除（默认软删）------------------------------------------- + def D(self, params, cond=None): + _assert_writable(self.tblname) + if self.tblname in APPEND_ONLY_TABLES: + raise PBLError(PBL_ERR['agent_tool']['write_protected'], + '表 %s 为 append-only，禁止删除' % self.tblname) + ctx = _ctx(params) + where = ctx.scope(cond or {}) + if not self.soft_delete or self.tblname in HARD_DELETE_TABLES: + return self.db.D(self.tblname, where) + return self.db.U(self.tblname, + {'is_deleted': 1, 'update_time': _now(), + 'update_user': ctx.user_id}, + where) - # ---- U ---- - def update(self, pk_value, params, tenant_id=None, user_id=None, expect_version=None): - tid = require_tenant(tenant_id if tenant_id is not None else (params or {}).get('tenant_id')) - data = self._filter(params) - data.pop('tenant_id', None) - if not data: - raise PblError(ErrCode.E_PARAM_MISSING, '无可更新字段') - if 'update_time' in self.fields: - data['update_time'] = self._now() - if user_id and 'update_user' in self.fields: - data['update_user'] = str(user_id)[:32] - - where = {self.pk: pk_value, 'tenant_id': tid} - if self.soft_delete: + # ---- R：查询（单行/列表）------------------------------------------ + def R(self, params, cond=None, fields=None, order_by=None, + limit=None, offset=0, one=False): + """租户作用域查询。one=True 返回单行 dict，否则返回 list。""" + ctx = _ctx(params) + where = ctx.scope(cond or {}) + if self.soft_delete and 'is_deleted' not in where: where['is_deleted'] = 0 - if expect_version is not None: - # 乐观锁：版本不符即拒绝（E_VERSION_STALE），防止并发覆盖 - cur = _db.query_one(self.table, where, tenant_id=tid, dbname=self.dbname) - if cur is None: - raise PblError(ErrCode.E_NOT_FOUND, '%s[%s] 不存在' % (self.table, pk_value)) - if int(cur.get('version', 0)) != int(expect_version): - raise PblError(ErrCode.E_VERSION_STALE, - '版本已过期：期望 %s，实际 %s' % (expect_version, cur.get('version')), - {'current': cur.get('version')}) - if 'version' in self.fields: - data['version'] = int(cur.get('version', 0)) + 1 - return _db.update(self.table, data, where, tenant_id=tid, dbname=self.dbname) + rows = self.db.R(self.tblname, where, fields=fields, + order_by=order_by, limit=limit, offset=offset) + rows = rows or [] + if one: + return rows[0] if rows else None + return rows - # ---- D ---- - def soft_delete(self, pk_value, tenant_id=None, user_id=None): - tid = require_tenant(tenant_id) - if not self.soft_delete: - raise PblError(ErrCode.E_PARAM_INVALID, '%s 无 is_deleted 列，不支持软删' % self.table) - data = {'is_deleted': 1} - if 'update_time' in self.fields: - data['update_time'] = self._now() - if user_id and 'update_user' in self.fields: - data['update_user'] = str(user_id)[:32] - return _db.update(self.table, data, {self.pk: pk_value}, tenant_id=tid, dbname=self.dbname) - - def delete(self, pk_value, tenant_id=None): - """物理删除：默认走软删；确需物理删除请显式 hard=True 且仅限治理白名单表。""" - tid = require_tenant(tenant_id) - if self.soft_delete: - return self.soft_delete(pk_value, tenant_id=tid) - return _db.delete(self.table, {self.pk: pk_value}, tenant_id=tid, dbname=self.dbname) - - def exists(self, where, tenant_id=None): - tid = require_tenant(tenant_id if tenant_id is not None else (where or {}).get('tenant_id')) - cond = {k: v for k, v in (where or {}).items() if k != 'tenant_id' and v is not None} - return _db.query_one(self.table, cond, tenant_id=tid, dbname=self.dbname) is not None + # ---- I：原生 SQL（只读统计/复杂联查）------------------------------ + def I(self, sql, args=None, params=None): + """ + 执行原生 SELECT。调用方必须自行把 tenant_id 放进 WHERE 首条件； + 非 SELECT 语句一律拒绝（写操作请走 C/U/D）。 + """ + s = (sql or '').strip().lower() + if not s.startswith('select'): + raise PBLError(PBL_ERR['agent_tool']['write_protected'], + 'I() 仅允许 SELECT，写操作请用 C/U/D') + if params is not None: + ctx = _ctx(params) + if 'tenant_id' not in s: + raise PBLError(PBL_ERR['tenant']['missing'], + '原生 SQL 必须显式带 tenant_id 条件') + args = tuple([ctx.tenant_id] + list(args or ())) + return self.db.I(sql, args or ()) -def make_crud(table_def, module='pbl_common', dbname=None): - """由「四段式表定义」dict 生成 CrudFactory。 - - table_def 形如： - {'summary': '蓝图聚合根', - 'fields': [{'name': 'tenant_id', 'type': 'varchar(32)'}, ...], - 'indexes': [{'name': 'ix_x', 'cols': ['tenant_id', 'status']}], - 'codes': {'status': ['draft', 'published']}} - """ - fields = [] - for f in (table_def.get('fields') or []): - name = f.get('name') if isinstance(f, dict) else f - if name: - fields.append(name) - codes = {} - for k, v in (table_def.get('codes') or {}).items(): - codes[k] = set(v) - unique_keys = [] - for ix in (table_def.get('indexes') or []): - if ix.get('unique'): - cols = ix.get('cols') or [] - cols = [c for c in cols if c != 'tenant_id'] - if cols: - unique_keys.append(tuple(cols)) - return CrudFactory( - table=table_def.get('table') or table_def.get('name'), - fields=fields, - codes=codes, - pk=table_def.get('pk', 'id'), - soft_delete='is_deleted' in fields, - dbname=dbname, - module=module, - unique_keys=unique_keys, - ) +def make_crud(tblname, module=None, **kw): + """工厂：make_crud('pbl_blueprint') -> Crud 实例（含 C/U/D/R/I）。""" + return Crud(tblname, module=module, **kw) diff --git a/pbl_common/db.py b/pbl_common/db.py index 759c702..c02f275 100644 --- a/pbl_common/db.py +++ b/pbl_common/db.py @@ -1,303 +1,160 @@ # -*- coding: utf-8 -*- -"""DB 适配层：统一走 ServerEnv().get_module_dbname('模块名') 取库名（禁止硬编码 DBNAME）， -底层复用平台 sqlor（sor.C/U/D/R/I/sqlExe），并补充「单事务多语句」能力供 pbl_runtime_ext 使用。 +""" +pbl_common.db —— DB 适配层（方言锁定 mariadb） -设计要点： - * get_dbname(module) —— 唯一库名入口，ServerEnv 未注入时回落环境变量 PBLS_DBNAME，再回落 'pbls' - * tx() —— 上下文管理器，保证「事件 + 状态 + 广播登记」在同一事务内提交/回滚 - * 所有查询自动补 tenant_id 条件（tenant_first），杜绝跨租户读写 +硬约束（QC 6.4）： + * 主键 BIGINT NOT NULL AUTO_INCREMENT + * 禁止 BIGSERIAL / SERIAL / nextval / IDENTITY（postgres 方言） + * 禁止 ENUM / SET（枚举走 pbl_appcodes + VARCHAR 存码值） + * 禁止 TIMESTAMP（时间列一律 DATETIME） + * 禁止 FOREIGN KEY / REFERENCES（关联应用层维护） + * 业务表首列 tenant_id BIGINT NOT NULL（pbl_tenant 自身除外） + +取库名一律 ServerEnv().get_module_dbname(module)，禁止硬编码 DBNAME。 """ -import os -import threading +DIALECT = 'mariadb' +APP_NAME = 'pbls' -from .errors import PblError, ErrCode -from .tenant import require_tenant +FORBIDDEN_TOKENS = ('BIGSERIAL', 'SERIAL', 'NEXTVAL', 'ENUM(', 'TIMESTAMP', + 'FOREIGN KEY', 'REFERENCES', 'IDENTITY') -_ENV = None # ServerEnv 实例（由 init.load_pbl_common(env) 注入） -_ENV_LOCK = threading.Lock() +# 抽象类型 -> mariadb 物理类型 +TYPE_MAP = { + 'pk': 'BIGINT NOT NULL AUTO_INCREMENT', + 'bigint': 'BIGINT NOT NULL DEFAULT 0', + 'int': 'INT NOT NULL DEFAULT 0', + 'tinyint': 'TINYINT NOT NULL DEFAULT 0', + 'bool': 'TINYINT NOT NULL DEFAULT 0', + 'varchar': 'VARCHAR(%s) NOT NULL DEFAULT \'\'', + 'text': 'TEXT', + 'longtext': 'LONGTEXT', + 'datetime': 'DATETIME NULL', + 'date': 'DATE NULL', + 'decimal': 'DECIMAL(%s) NOT NULL DEFAULT 0', +} -DEFAULT_DBNAME = 'pbls' +_TENANT_EXEMPT_TABLES = ('pbl_tenant',) -def set_env(env): - """由 pbl_common/init.py 在应用挂载时调用，注入 ServerEnv。""" - global _ENV - with _ENV_LOCK: - _ENV = env - return env +def get_dialect(): + """当前方言（固定 mariadb，与 env 的 db.engine 对齐）。""" + return DIALECT -def get_env(): - return _ENV - - -def get_dbname(module_name='pbl_common'): - """模块 → 库名。优先 ServerEnv.get_module_dbname，其次环境变量，最后默认库。""" - env = _ENV - if env is not None: +def get_module_dbname(module_name): + """模块 -> 库名。一律走 ServerEnv，失败回落应用主库名（不返回 None）。""" + try: + from appbase.env import ServerEnv + env = ServerEnv() fn = getattr(env, 'get_module_dbname', None) if callable(fn): - try: - name = fn(module_name) - if name: - return name - except Exception: # noqa: BLE001 —— 映射异常不阻断，回落 - pass - return os.environ.get('PBLS_DBNAME') or DEFAULT_DBNAME + name = fn(module_name) + if name: + return name + except Exception: # noqa: BLE001 + pass + return APP_NAME -def _sor(dbname=None): - """取 sqlor 句柄（平台标准 API：sor.C/U/D/R/I/sqlExe）。""" - try: - import sqlor - except Exception as e: # noqa: BLE001 - raise PblError(ErrCode.E_DB_ERROR, 'sqlor 不可用：%s' % e) - db = dbname or get_dbname() - factory = getattr(sqlor, 'Sqlor', None) or getattr(sqlor, 'sqlor', None) - if factory is None: - # 部分版本直接暴露模块级函数 - return sqlor - try: - return factory(db) - except TypeError: - return factory(dbname=db) +def get_db(module_name=None): + """取 sqlor DB 句柄（标准 API 仅 C/U/D/R/I/sqlExe）。""" + import sqlor + return sqlor.getDB(get_module_dbname(module_name or APP_NAME)) -def _sqlor_kwargs(params): - """sqlor 参数规范化：None 值剔除，bool → int。""" - out = {} - for k, v in (params or {}).items(): - if v is None: - continue - if isinstance(v, bool): - v = 1 if v else 0 - out[k] = v - return out - - -# --------------------------------------------------------------------------- -# 基础读写（tenant 强制打头） -# --------------------------------------------------------------------------- -def insert(table, params, tenant_id=None, dbname=None): - """C —— 新增。params 必须含 tenant_id（或显式传入），否则拒绝。""" - tid = require_tenant(tenant_id if tenant_id is not None else (params or {}).get('tenant_id')) - data = dict(params or {}) - data['tenant_id'] = tid - try: - return _sor(dbname).C(table, _sqlor_kwargs(data)) - except PblError: - raise - except Exception as e: # noqa: BLE001 - raise PblError(ErrCode.E_DB_ERROR, 'insert %s 失败：%s' % (table, e)) - - -def update(table, params, where, tenant_id=None, dbname=None): - """U —— 更新。where 自动注入 tenant_id，禁止跨租户更新。""" - tid = require_tenant(tenant_id if tenant_id is not None else (where or {}).get('tenant_id')) - cond = dict(where or {}) - cond['tenant_id'] = tid - try: - return _sor(dbname).U(table, _sqlor_kwargs(params), _sqlor_kwargs(cond)) - except PblError: - raise - except Exception as e: # noqa: BLE001 - raise PblError(ErrCode.E_DB_ERROR, 'update %s 失败：%s' % (table, e)) - - -def delete(table, where, tenant_id=None, dbname=None): - """D —— 删除（物理删除仅限治理白名单表；业务表请用逻辑删除 is_deleted=1）。""" - tid = require_tenant(tenant_id if tenant_id is not None else (where or {}).get('tenant_id')) - cond = dict(where or {}) - cond['tenant_id'] = tid - try: - return _sor(dbname).D(table, _sqlor_kwargs(cond)) - except PblError: - raise - except Exception as e: # noqa: BLE001 - raise PblError(ErrCode.E_DB_ERROR, 'delete %s 失败：%s' % (table, e)) - - -def get(table, where, tenant_id=None, dbname=None): - """R —— 单条查询。""" - tid = require_tenant(tenant_id if tenant_id is not None else (where or {}).get('tenant_id')) - cond = dict(where or {}) - cond['tenant_id'] = tid - try: - return _sor(dbname).R(table, _sqlor_kwargs(cond)) - except PblError: - raise - except Exception as e: # noqa: BLE001 - raise PblError(ErrCode.E_DB_ERROR, 'get %s 失败：%s' % (table, e)) - - -def query(table, where=None, fields=None, order_by=None, limit=None, offset=None, - tenant_id=None, dbname=None): - """I —— 列表查询（tenant_id 强制作为首个条件）。""" - tid = require_tenant(tenant_id if tenant_id is not None else (where or {}).get('tenant_id')) - cond = {'tenant_id': tid} - for k, v in (where or {}).items(): - if k == 'tenant_id': - continue - cond[k] = v - kw = _sqlor_kwargs(cond) - if fields: - kw['_fields'] = fields if isinstance(fields, str) else ','.join(fields) - if order_by: - kw['_orderby'] = order_by - if limit is not None: - kw['_limit'] = int(limit) - if offset: - kw['_offset'] = int(offset) - try: - rows = _sor(dbname).I(table, kw) - except PblError: - raise - except Exception as e: # noqa: BLE001 - raise PblError(ErrCode.E_DB_ERROR, 'query %s 失败：%s' % (table, e)) - return rows or [] - - -def query_one(table, where=None, tenant_id=None, dbname=None): - rows = query(table, where=where, limit=1, tenant_id=tenant_id, dbname=dbname) - return rows[0] if rows else None - - -def execute(sql, args=None, dbname=None): - """sqlExe —— 原生 SQL（DDL/批量/复杂联查）。调用方必须自行保证 SQL 内含 tenant_id 条件。""" - if 'tenant_id' not in (sql or ''): - raise PblError(ErrCode.E_TENANT_MISSING, - 'execute() 的 SQL 必须显式包含 tenant_id 条件（租户强制打头）') - try: - return _sor(dbname).sqlExe(sql, args or ()) - except PblError: - raise - except Exception as e: # noqa: BLE001 - raise PblError(ErrCode.E_DB_ERROR, 'execute 失败：%s' % e) - - -# --------------------------------------------------------------------------- -# 单事务（pbl_runtime_ext：事件 + 状态 + 广播登记 必须同事务） -# --------------------------------------------------------------------------- -class _Tx(object): - """事务句柄：内部缓存语句，commit 时按序执行，任一失败整体回滚。""" - - def __init__(self, dbname=None): - self.dbname = dbname or get_dbname() - self._ops = [] # [(kind, table, a, b)] - self._conn = None - self.committed = False - self.rolled_back = False - - # -- 事务内操作登记（延迟执行，保证原子性由 DB 事务兜底）-- - def insert(self, table, params, tenant_id=None): - tid = require_tenant(tenant_id if tenant_id is not None else (params or {}).get('tenant_id')) - data = dict(params or {}) - data['tenant_id'] = tid - self._ops.append(('C', table, _sqlor_kwargs(data), None)) - return len(self._ops) - - def update(self, table, params, where, tenant_id=None): - tid = require_tenant(tenant_id if tenant_id is not None else (where or {}).get('tenant_id')) - cond = dict(where or {}) - cond['tenant_id'] = tid - self._ops.append(('U', table, _sqlor_kwargs(params), _sqlor_kwargs(cond))) - return len(self._ops) - - def execute(self, sql, args=None): - if 'tenant_id' not in (sql or ''): - raise PblError(ErrCode.E_TENANT_MISSING, '事务内 SQL 必须含 tenant_id 条件') - self._ops.append(('X', sql, args or (), None)) - return len(self._ops) - - @property - def op_count(self): - return len(self._ops) - - # -- 提交 / 回滚 -- - def commit(self): - if self.committed or self.rolled_back: - raise PblError(ErrCode.E_TX_FAILED, '事务已结束，不可重复提交') - sor = _sor(self.dbname) - conn = self._acquire_conn(sor) - try: - if conn is not None: - conn.begin() - for kind, a, b, c in self._ops: - if kind == 'C': - sor.C(a, b) - elif kind == 'U': - sor.U(a, b, c) - elif kind == 'X': - sor.sqlExe(a, b) - else: - raise PblError(ErrCode.E_TX_FAILED, '未知事务操作：%s' % kind) - if conn is not None: - conn.commit() - self.committed = True - return len(self._ops) - except PblError: - self._safe_rollback(conn) - raise - except Exception as e: # noqa: BLE001 - self._safe_rollback(conn) - raise PblError(ErrCode.E_TX_FAILED, '事务提交失败已回滚：%s' % e) - - def rollback(self): - if self.committed: - return False - self._safe_rollback(self._conn) - self.rolled_back = True - self._ops = [] - return True - - # -- 内部 -- - def _acquire_conn(self, sor): - """尽力取底层连接以启用真实 DB 事务；取不到则退化为「全成功才落库」的延迟执行模型。""" - for attr in ('conn', 'connection', '_conn', '_connection', 'db'): - c = getattr(sor, attr, None) - if c is not None and hasattr(c, 'commit'): - self._conn = c - return c - return None - - def _safe_rollback(self, conn): - self.rolled_back = True - self._ops = [] - if conn is not None: - try: - conn.rollback() - except Exception: # noqa: BLE001 - pass - - -class tx(object): - """事务上下文管理器。 - - 用法： - with tx() as t: - t.insert('pbl_runtime_event', {...}) - t.update('pbl_runtime_state', {...}, {...}) - # 退出无异常 → 自动 commit；抛异常 → 自动 rollback +def ddl_column_sql(col_name, abstype, size=None): """ - - def __init__(self, dbname=None): - self._t = _Tx(dbname) - - def __enter__(self): - return self._t - - def __exit__(self, exc_type, exc_val, exc_tb): - if exc_type is None: - self._t.commit() - else: - self._t.rollback() - return False + 抽象类型 -> mariadb 列定义片段。 + 返回形如：`id BIGINT NOT NULL AUTO_INCREMENT` + """ + t = (abstype or '').lower() + if t not in TYPE_MAP: + raise ValueError('未知抽象类型：%s（列 %s）' % (abstype, col_name)) + tpl = TYPE_MAP[t] + if '%s' in tpl: + if t == 'varchar': + tpl = tpl % (int(size or 200)) + elif t == 'decimal': + tpl = tpl % (size or '18,4') + sql = '%s %s' % (col_name, tpl) + _assert_dialect_safe(sql) + return sql -# 兼容别名（部分模块按 query/execute 直接调用） -R = get -I = query -C = insert -U = update -D = delete +def _assert_dialect_safe(sql): + up = (sql or '').upper() + for tok in FORBIDDEN_TOKENS: + if tok in up: + raise ValueError('DDL 含禁用 token %s（方言必须为 mariadb）：%s' % (tok, sql)) + + +def build_create_table(table, columns, indexes=None, comment=''): + """ + 生成 CREATE TABLE IF NOT EXISTS（mariadb 方言，幂等）。 + columns: [{'name','type','size','tenant':bool}]，首列建议 tenant_id + indexes: [{'name','cols':[...],'unique':bool}] + """ + _assert_dialect_safe(table) + lines = [] + has_pk = False + has_tenant = any(c.get('name') == 'tenant_id' for c in columns) + + if table not in _TENANT_EXEMPT_TABLES and not has_tenant: + # 业务表强制 tenant_id 打头 + lines.append(ddl_column_sql('tenant_id', 'bigint')) + + for c in columns: + name = c.get('name') + if name == 'tenant_id' and table not in _TENANT_EXEMPT_TABLES: + continue + lines.append(ddl_column_sql(name, c.get('type'), c.get('size'))) + if (c.get('type') or '').lower() == 'pk': + has_pk = True + + if not has_pk: + lines.insert(0, ddl_column_sql('id', 'pk')) + has_pk = True + + lines.append('PRIMARY KEY (id)') + + for idx in (indexes or []): + cols = ', '.join(idx.get('cols') or []) + if not cols: + continue + kw = 'UNIQUE KEY' if idx.get('unique') else 'KEY' + lines.append('%s %s (%s)' % (kw, idx.get('name'), cols)) + + # 租户表默认带租户索引 + if table not in _TENANT_EXEMPT_TABLES: + names = [i.get('name') for i in (indexes or [])] + if 'idx_%s_tenant' % table not in names: + lines.append('KEY idx_%s_tenant (tenant_id)' % table) + + body = ',\n '.join(lines) + sql = ( + 'CREATE TABLE IF NOT EXISTS %s (\n %s\n) ' + 'ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci' + % (table, body) + ) + if comment: + sql += " COMMENT='%s'" % comment.replace("'", '') + _assert_dialect_safe(sql) + return sql + ';' + + +def now_sql(): + """mariadb 取当前时间（DATETIME，不用 TIMESTAMP）。""" + return 'NOW()' + + +def upsert_sql(table, unique_cols, update_cols): + """mariadb 幂等 upsert：INSERT ... ON DUPLICATE KEY UPDATE。""" + _assert_dialect_safe(table) + cols = list(unique_cols) + [c for c in update_cols if c not in unique_cols] + placeholders = ', '.join(['%s'] * len(cols)) + sets = ', '.join(['%s=VALUES(%s)' % (c, c) for c in update_cols]) + sql = 'INSERT INTO %s (%s) VALUES (%s)' % (table, ', '.join(cols), placeholders) + if sets: + sql += ' ON DUPLICATE KEY UPDATE %s' % sets + return sql diff --git a/pbl_common/errors.py b/pbl_common/errors.py index 8046133..5c5951a 100644 --- a/pbl_common/errors.py +++ b/pbl_common/errors.py @@ -1,186 +1,164 @@ # -*- coding: utf-8 -*- """ -pbl_common.errors —— 统一错误码与异常类型（fail-closed） +pbl_common.errors —— PBL 统一错误码（6 大类，码值全局唯一） -错误码格式：PBL-{域}-{4位序号} -所有对外接口只抛 PblError 子类，禁止裸 Exception 穿透到 dspy 层。 +约定： + * 码值格式 PBL-<类别缩写>-<三位序号>，全局唯一（self_check 校验） + * 所有对外接口失败一律返回 {'ok': False, 'err_code': ..., 'err_msg': ...} + * fail-closed：未登记错误码不允许对外抛出裸异常 """ +PBL_ERR = { + # ---- 1) 租户上下文/隔离 ------------------------------------------- + 'tenant': { + 'missing': 'PBL-TEN-001', + 'invalid': 'PBL-TEN-002', + 'cross_tenant_read': 'PBL-TEN-003', + 'cross_tenant_write': 'PBL-TEN-004', + 'quota_exceeded': 'PBL-TEN-005', + 'not_activated': 'PBL-TEN-006', + }, + # ---- 2) 蓝图聚合根 ------------------------------------------------- + 'blueprint': { + 'not_found': 'PBL-BP-001', + 'version_conflict': 'PBL-BP-002', + 'subobj_type_unknown':'PBL-BP-003', + 'subobj_ref_broken': 'PBL-BP-004', + 'status_illegal': 'PBL-BP-005', + 'fork_denied': 'PBL-BP-006', + 'template_invalid': 'PBL-BP-007', + 'delta_mismatch': 'PBL-BP-008', + }, + # ---- 3) 校验引擎（14 维 + 5 级质量状态）--------------------------- + 'validation': { + 'rule_not_found': 'PBL-VAL-001', + 'dim_unknown': 'PBL-VAL-002', + 'fatal_issue': 'PBL-VAL-003', + 'quality_below_gate': 'PBL-VAL-004', + 'run_timeout': 'PBL-VAL-005', + }, + # ---- 4) 编译器 ----------------------------------------------------- + 'compiler': { + 'not_compilable': 'PBL-CMP-001', + 'gd_hash_conflict': 'PBL-CMP-002', + 'job_failed': 'PBL-CMP-003', + 'scense_ref_missing': 'PBL-CMP-004', + }, + # ---- 5) 工具裁决（fail-closed，22 工具 = 13 启用 + 9 禁用）--------- + 'agent_tool': { + 'unregistered': 'PBL-TOOL-001', + 'disabled': 'PBL-TOOL-002', + 'write_protected': 'PBL-TOOL-003', + 'invalid_args': 'PBL-TOOL-004', + 'need_confirm': 'PBL-TOOL-005', + 'rate_limited': 'PBL-TOOL-006', + 'session_invalid': 'PBL-TOOL-007', + 'exec_error': 'PBL-TOOL-008', + 'timeout': 'PBL-TOOL-009', + }, + # ---- 6) 证据采集（幂等）------------------------------------------- + 'evidence': { + 'idem_conflict': 'PBL-EVI-001', + 'type_unknown': 'PBL-EVI-002', + 'payload_invalid': 'PBL-EVI-003', + 'artifact_missing': 'PBL-EVI-004', + 'collect_denied': 'PBL-EVI-005', + }, +} -class ErrorCode(object): - """错误码常量表（与 docs/01-design/agent-tool-contract.md 对齐）""" +# 类别 -> 缩写（供 self_check / 文档生成使用） +CATEGORY_PREFIX = { + 'tenant': 'TEN', + 'blueprint': 'BP', + 'validation': 'VAL', + 'compiler': 'CMP', + 'agent_tool': 'TOOL', + 'evidence': 'EVI', +} - # 租户域 - TENANT_MISSING = 'PBL-TENANT-0001' - TENANT_INVALID = 'PBL-TENANT-0002' - TENANT_MISMATCH = 'PBL-TENANT-0003' - - # 参数域 - PARAM_MISSING = 'PBL-PARAM-0001' - PARAM_INVALID = 'PBL-PARAM-0002' - PARAM_CODE_UNREGISTERED = 'PBL-PARAM-0003' - - # 数据域 - NOT_FOUND = 'PBL-DATA-0001' - CONFLICT = 'PBL-DATA-0002' - DUPLICATE = 'PBL-DATA-0003' - - # 权限域 - FORBIDDEN = 'PBL-AUTH-0001' - WRITE_PROTECTED = 'PBL-AUTH-0002' - TOOL_DISABLED = 'PBL-AUTH-0003' - - # 校验域 - VALIDATION_FAILED = 'PBL-VALID-0001' - QUALITY_INSUFFICIENT = 'PBL-VALID-0002' - - # 编译域 - COMPILE_FAILED = 'PBL-COMP-0001' - COMPILE_INCOMPLETE = 'PBL-COMP-0002' - - # 运行时域 - RUNTIME_EVENT_FAILED = 'PBL-RUN-0001' - RUNTIME_STATE_ILLEGAL = 'PBL-RUN-0002' - - # 系统域 - DB_ERROR = 'PBL-SYS-0001' - INTERNAL = 'PBL-SYS-0009' - - -ERR_TENANT_MISSING = ErrorCode.TENANT_MISSING -ERR_TENANT_INVALID = ErrorCode.TENANT_INVALID -ERR_PARAM_INVALID = ErrorCode.PARAM_INVALID -ERR_NOT_FOUND = ErrorCode.NOT_FOUND -ERR_CONFLICT = ErrorCode.CONFLICT -ERR_FORBIDDEN = ErrorCode.FORBIDDEN -ERR_TOOL_DISABLED = ErrorCode.TOOL_DISABLED -ERR_WRITE_PROTECTED = ErrorCode.WRITE_PROTECTED - -# HTTP 状态映射（dspy 层据此返回） -CODE_TO_HTTP = { - ErrorCode.TENANT_MISSING: 400, - ErrorCode.TENANT_INVALID: 400, - ErrorCode.TENANT_MISMATCH: 403, - ErrorCode.PARAM_MISSING: 400, - ErrorCode.PARAM_INVALID: 400, - ErrorCode.PARAM_CODE_UNREGISTERED: 400, - ErrorCode.NOT_FOUND: 404, - ErrorCode.CONFLICT: 409, - ErrorCode.DUPLICATE: 409, - ErrorCode.FORBIDDEN: 403, - ErrorCode.WRITE_PROTECTED: 403, - ErrorCode.TOOL_DISABLED: 403, - ErrorCode.VALIDATION_FAILED: 422, - ErrorCode.QUALITY_INSUFFICIENT: 422, - ErrorCode.COMPILE_FAILED: 422, - ErrorCode.COMPILE_INCOMPLETE: 422, - ErrorCode.RUNTIME_EVENT_FAILED: 500, - ErrorCode.RUNTIME_STATE_ILLEGAL: 409, - ErrorCode.DB_ERROR: 500, - ErrorCode.INTERNAL: 500, +DEFAULT_MSG = { + 'PBL-TEN-001': 'tenant_id 缺失，拒绝执行', + 'PBL-TEN-002': 'tenant_id 非法', + 'PBL-TEN-003': '跨租户读取被拒绝', + 'PBL-TEN-004': '跨租户写入被拒绝', + 'PBL-TEN-005': '租户配额超限', + 'PBL-TEN-006': '租户未启用', + 'PBL-BP-001': '蓝图不存在', + 'PBL-BP-002': '蓝图版本冲突', + 'PBL-BP-003': '未知子对象类型', + 'PBL-BP-004': '子对象引用断裂', + 'PBL-BP-005': '蓝图状态不允许该操作', + 'PBL-BP-006': '蓝图 fork 被拒绝', + 'PBL-BP-007': '模板非法', + 'PBL-BP-008': 'change_delta 与快照不一致', + 'PBL-VAL-001': '校验规则不存在', + 'PBL-VAL-002': '未知校验维度', + 'PBL-VAL-003': '存在致命校验问题', + 'PBL-VAL-004': '质量等级未达门禁', + 'PBL-VAL-005': '校验执行超时', + 'PBL-CMP-001': '蓝图不可编译', + 'PBL-CMP-002': 'Game Definition 哈希冲突', + 'PBL-CMP-003': '编译任务失败', + 'PBL-CMP-004': 'scense 引用缺失', + 'PBL-TOOL-001': '工具未注册', + 'PBL-TOOL-002': '工具已禁用', + 'PBL-TOOL-003': '目标模块写保护，操作被拦截', + 'PBL-TOOL-004': '工具参数非法', + 'PBL-TOOL-005': '需人工确认后执行', + 'PBL-TOOL-006': '工具调用被限流', + 'PBL-TOOL-007': 'Agent 会话无效', + 'PBL-TOOL-008': '工具执行错误', + 'PBL-TOOL-009': '工具执行超时', + 'PBL-EVI-001': '证据幂等键冲突', + 'PBL-EVI-002': '未知证据类型', + 'PBL-EVI-003': '证据载荷非法', + 'PBL-EVI-004': '证据产物缺失', + 'PBL-EVI-005': '证据采集被拒绝', } -class PblError(Exception): - """PBL 统一业务异常基类""" +class PBLError(Exception): + """PBL 业务异常基类。""" - default_code = ErrorCode.INTERNAL - default_message = '内部错误' - - def __init__(self, message=None, code=None, detail=None, http_status=None): - self.code = code or self.default_code - self.message = message or self.default_message + def __init__(self, err_code, err_msg=None, detail=None): + self.err_code = err_code + self.err_msg = err_msg or DEFAULT_MSG.get(err_code, err_code) self.detail = detail or {} - self.http_status = http_status or CODE_TO_HTTP.get(self.code, 500) - super(PblError, self).__init__('%s: %s' % (self.code, self.message)) + super(PBLError, self).__init__(self.err_msg) def to_dict(self): - """序列化为对外 JSON 结构（dspy 层直接返回）""" - out = { - 'ok': False, - 'error': { - 'code': self.code, - 'message': self.message, - 'http_status': self.http_status, - }, - } - if self.detail: - out['error']['detail'] = self.detail - return out - - def __repr__(self): - return 'PblError(code=%r, message=%r)' % (self.code, self.message) + return {'ok': False, 'err_code': self.err_code, + 'err_msg': self.err_msg, 'detail': self.detail} -class TenantMissingError(PblError): - default_code = ErrorCode.TENANT_MISSING - default_message = '缺少 tenant_id（所有 PBL 读写必须 tenant_id 打头）' +def err(category, key, msg=None, detail=None): + """按 类别+键 构造 PBLError。未登记的键直接抛 KeyError（fail-closed）。""" + code = PBL_ERR[category][key] + return PBLError(code, msg, detail) -class TenantInvalidError(PblError): - default_code = ErrorCode.TENANT_INVALID - default_message = 'tenant_id 非法' +def fail(category, key, msg=None, detail=None): + """返回统一失败响应体（不抛异常）。""" + code = PBL_ERR[category][key] + return {'ok': False, 'err_code': code, + 'err_msg': msg or DEFAULT_MSG.get(code, code), + 'detail': detail or {}} -class ParamInvalidError(PblError): - default_code = ErrorCode.PARAM_INVALID - default_message = '参数非法' +def ok(data=None, **extra): + """返回统一成功响应体。""" + out = {'ok': True, 'err_code': '', 'err_msg': ''} + if data is not None: + out['data'] = data + out.update(extra) + return out -class NotFoundError(PblError): - default_code = ErrorCode.NOT_FOUND - default_message = '记录不存在' - - -class ConflictError(PblError): - default_code = ErrorCode.CONFLICT - default_message = '数据冲突' - - -class ForbiddenError(PblError): - default_code = ErrorCode.FORBIDDEN - default_message = '无权限' - - -class WriteProtectedError(PblError): - """引用模块写保护违规（rbac/world/scene/entity/scense/scense_runtime/script_engine）""" - default_code = ErrorCode.WRITE_PROTECTED - default_message = '目标为写保护引用模块基表，禁止写入（扩展请走 pbl_*_ext）' - - -class ToolDisabledError(PblError): - """Agent 工具 fail-closed：未启用工具一律拒绝""" - default_code = ErrorCode.TOOL_DISABLED - default_message = 'Agent 工具未启用，fail-closed 拒绝执行' - - -class ValidationFailedError(PblError): - default_code = ErrorCode.VALIDATION_FAILED - default_message = '蓝图校验未通过' - - -class CompileFailedError(PblError): - default_code = ErrorCode.COMPILE_FAILED - default_message = '蓝图编译失败' - - -class DbError(PblError): - default_code = ErrorCode.DB_ERROR - default_message = '数据库操作失败' - - -# 写保护模块清单（与 projects/pbls/env/test.json write_protected_modules 同值） -WRITE_PROTECTED_MODULES = ( - 'rbac', 'world', 'scene', 'entity', - 'scense', 'scense_runtime', 'script_engine', -) - - -def assert_not_write_protected(module_name, table_name=None): - """写保护断言：命中引用模块即抛 WriteProtectedError""" - if module_name in WRITE_PROTECTED_MODULES: - raise WriteProtectedError( - message='模块 %s 为写保护引用模块，禁止写入' % module_name, - detail={'module': module_name, 'table': table_name}, - ) - return True +def all_codes(): + """扁平化全部错误码，供文档/前端映射使用。""" + out = [] + for cat, kv in PBL_ERR.items(): + for key, code in kv.items(): + out.append({'category': cat, 'key': key, 'code': code, + 'msg': DEFAULT_MSG.get(code, '')}) + return out diff --git a/pbl_common/init.py b/pbl_common/init.py index f1ae4d9..7c4d55e 100644 --- a/pbl_common/init.py +++ b/pbl_common/init.py @@ -1,113 +1,186 @@ # -*- coding: utf-8 -*- -""" -pbl_common.init —— 模块挂载入口（load_pbl_common） +"""pbl_common 模块挂载入口（load_pbl_common）。 -挂载职责（module-development-spec）： - 1. 幂等建公共表（pbl_audit_log / pbl_seed_record） - 2. 注册 API 契约到 ServerEnv（供其它 pbl_* 模块取用） - 3. 注册 RBAC 权限路径（若 rbac 已挂载） - 4. 建表失败不阻断启动（降级 stderr），保证应用可起、界面可访问 +三处同步注册（module-development-spec）： + 1. 本文件 load_pbl_common() + 2. apps/pbls/app/pbls.py LOAD_ORDER 含 'pbl_common'（Wave0 首位） + 3. apps/pbls/conf/config.json load_order / module_dbname 含 pbl_common + +pbl_common 零自有表：只提供租户上下文 / DB 适配 / 错误码 / 审计 / CRUD 工厂。 +审计写入落在 pbl_governance.pbl_audit_log（跨模块共享，经 get_module_dbname 解析库名）。 """ -import sys - -from pbl_common import api as pbl_common_api -from pbl_common.tables import ensure_tables, TABLES +from pbl_common.kernel import ( + CrudFactory, + DbAdapter, + PblError, + assert_tenant_match, + check_in, + check_len, + clear_tenant_context, + current_tenant_id, + db, + fail, + from_request, + get_tenant_context, + idem_key, + json_dumps_canonical, + make_crud, + new_code, + new_trace_id, + now_str, + ok, + require, + set_tenant_context, + sha256_text, + write_audit, +) MODULE_NAME = 'pbl_common' +MODULE_TABLES = () # 零自有表 +MODULE_VERSION = '1.0.0' -# 对外暴露的契约接口名（其它模块通过 ServerEnv().pbl_common.xxx 调用） -CONTRACT_FUNCS = ( - 'health', 'self_check', - 'build_context', 'bind_context', 'unbind_context', 'current_context', - 'require_tenant', 'require_context', 'context_scope', - 'normalize_tenant', 'assert_tenant', 'tenant_scope', 'with_tenant', - 'get_dbname', 'query', 'query_one', 'execute', 'insert', 'transaction', - 'make_crud', 'write_audit', 'query_audit', - 'dumps', 'loads', 'to_jsonable', 'parse_json_column', - 'PblError', 'ErrorCode', 'assert_not_write_protected', -) - -# RBAC 权限路径（公共内核只读接口） -RBAC_PATHS = ( - '/api/pbl_common_health.dspy', - '/api/pbl_common_self_check.dspy', -) +_loaded = False -def _register_contract(env): - """把契约接口挂到 ServerEnv，供其它 pbl_* 模块统一取用""" - contract = {} - for name in CONTRACT_FUNCS: - fn = getattr(pbl_common_api, name, None) - if fn is not None: - contract[name] = fn - try: - env.pbl_common = contract - except Exception: # noqa: BLE001 - # ServerEnv 可能是 __slots__ 对象，退化为字典挂载 - try: - setattr(env, 'modules', getattr(env, 'modules', {})) - env.modules[MODULE_NAME] = contract - except Exception: # noqa: BLE001 - pass - return contract +def load_pbl_common(): + """挂载 pbl_common：注册内核单例与公共 API 到 ServerEnv。 - -def _register_rbac(env): - """注册 RBAC 权限（rbac 未挂载时静默跳过）""" - registered = [] - try: - perm = getattr(env, 'register_perm', None) or getattr(env, 'add_perm', None) - if callable(perm): - for path in RBAC_PATHS: - try: - perm('logined', path) - registered.append(path) - except Exception: # noqa: BLE001 - pass - except Exception: # noqa: BLE001 - pass - return registered - - -def load_pbl_common(env=None): + 幂等：重复调用只生效一次（应用重启/热加载安全）。 """ - 挂载 pbl_common（应用 init() 中第一个业务模块）。 + global _loaded + if _loaded: + return True - env 为 None 时自取 ServerEnv()（兼容单模块测试）。 - 返回 {'module','tables','contract','rbac','ok'} - """ - if env is None: - try: - from ahserver.serverenv import ServerEnv - env = ServerEnv() - except Exception: # noqa: BLE001 - env = None + from appbase.serverenv import ServerEnv - result = { - 'module': MODULE_NAME, - 'tables': [], - 'contract': [], - 'rbac': [], - 'ok': True, + env = ServerEnv() + adapter = db(MODULE_NAME) + + # 公共内核挂到 ServerEnv，供其余 14 个 pbl_* 模块直接取用（避免各自 import 路径分叉） + env.pbl_common = { + 'version': MODULE_VERSION, + 'db': adapter, + 'crud_factory': make_crud, + 'tenant': { + 'set': set_tenant_context, + 'get': get_tenant_context, + 'current_id': current_tenant_id, + 'clear': clear_tenant_context, + 'from_request': from_request, + 'assert_match': assert_tenant_match, + }, + 'errors': { + 'PblError': PblError, + 'fail': fail, + }, + 'utils': { + 'new_code': new_code, + 'sha256_text': sha256_text, + 'json_dumps_canonical': json_dumps_canonical, + 'idem_key': idem_key, + 'now_str': now_str, + 'new_trace_id': new_trace_id, + }, + 'validators': { + 'require': require, + 'check_len': check_len, + 'check_in': check_in, + }, + 'audit': write_audit, + 'response': ok, + 'append_only_tables': sorted(DbAdapter.APPEND_ONLY_TABLES), + 'write_protected_modules': sorted(DbAdapter.WRITE_PROTECTED_MODULES), } - # 1) 幂等建表（失败降级，不阻断启动） + # 启动期自检：租户上下文缺失必须 fail-closed（不依赖 DB，纯逻辑断言） + _self_test_fail_closed() + + _loaded = True + return True + + +def _self_test_fail_closed(): + """内核自检：无租户上下文时读写一律拒绝（fail-closed 门禁的单元级证据）。""" + clear_tenant_context() try: - result['tables'] = ensure_tables(module=MODULE_NAME) - except Exception as e: # noqa: BLE001 - result['ok'] = False - result['error'] = 'ensure_tables 失败：%s' % e - sys.stderr.write('[pbl_common] ensure_tables failed: %s\n' % e) + current_tenant_id(required=True) + except PblError as exc: + assert exc.code == 'PBL_E_TENANT_MISSING', '错误码应为 PBL_E_TENANT_MISSING，实际 %s' % exc.code + else: + raise AssertionError('fail-closed 失效：无租户上下文未抛 PBL_E_TENANT_MISSING') - # 2) 注册契约 - if env is not None: - contract = _register_contract(env) - result['contract'] = sorted(contract.keys()) - # 3) 注册 RBAC - result['rbac'] = _register_rbac(env) + ctx = set_tenant_context('t_selfcheck', user_code='u_selfcheck', role_code='teacher') + assert ctx.tenant_id == 't_selfcheck' + assert current_tenant_id() == 't_selfcheck' - sys.stdout.write('[pbl_common] loaded tables=%s contract=%d\n' - % (result['tables'] or sorted(TABLES.keys()), len(result['contract']))) - return result + # 租户不匹配必须拒绝 + try: + assert_tenant_match({'tenant_id': 't_other'}, 't_selfcheck') + except PblError as exc: + assert exc.code == 'PBL_E_TENANT_MISMATCH' + else: + raise AssertionError('越权防护失效：跨租户行未被拒绝') + + # append-only 表拒绝 update/delete + adapter = DbAdapter(module_name=MODULE_NAME, sor=_FakeSor()) + for op in ('U', 'D'): + try: + adapter._guard_append_only('pbl_audit_log', op) + except PblError as exc: + assert exc.code == 'PBL_E_APPEND_ONLY' + else: + raise AssertionError('append-only 防护失效：%s 未被拒绝' % op) + + # 写保护模块拒绝 C/U/D + try: + adapter._guard_write_protected('world', 'C') + except PblError as exc: + assert exc.code == 'PBL_E_WRITE_PROTECTED' + else: + raise AssertionError('写保护失效：world 写入未被拒绝') + + # 幂等键稳定性（同输入同输出，证据幂等依赖） + assert idem_key('t', 'bp', 'artifact', 'src', 'h1') == idem_key('t', 'bp', 'artifact', 'src', 'h1') + assert idem_key('t', 'bp', 'artifact', 'src', 'h1') != idem_key('t', 'bp', 'artifact', 'src', 'h2') + + # 规范化 JSON 稳定性（确定性编译依赖） + assert sha256_text({'b': 2, 'a': 1}) == sha256_text({'a': 1, 'b': 2}) + + clear_tenant_context() + + +class _FakeSor(object): + """自检用 sqlor 桩：只记录调用，不触库。""" + + def __init__(self): + self.calls = [] + + def C(self, table, data): + self.calls.append(('C', table, data)) + return 1 + + def U(self, table, where, data): + self.calls.append(('U', table, where, data)) + return 1 + + def D(self, table, where): + self.calls.append(('D', table, where)) + return 1 + + def R(self, table, where, **kwargs): + self.calls.append(('R', table, where, kwargs)) + return [] + + def I(self, table, rows): + self.calls.append(('I', table, rows)) + return len(rows or []) + + def sqlExe(self, sql, args=None): + self.calls.append(('sqlExe', sql, args)) + return [] + + +# 别名：兼容 load_{module} 与 init 两种解析路径 +load_module = load_pbl_common +init = load_pbl_common diff --git a/pbl_common/kernel.py b/pbl_common/kernel.py new file mode 100644 index 0000000..9c20fc4 --- /dev/null +++ b/pbl_common/kernel.py @@ -0,0 +1,570 @@ +# -*- coding: utf-8 -*- +"""pbl_common 公共内核：租户上下文 / DB 适配 / 错误码 / 审计 / CRUD 工厂。 + +设计对齐：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、无物理删除 +""" + +import hashlib +import json +import threading +import time +import uuid + +# --------------------------------------------------------------------------- +# 错误码（统一前缀 PBL_E_，HTTP 语义由 api 层映射） +# --------------------------------------------------------------------------- + +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: '内部错误', +} + + +class PblError(Exception): + """PBL 统一业务异常：code + message + detail。""" + + def __init__(self, code, message=None, detail=None, http_status=400): + super(PblError, self).__init__(message or ERROR_MESSAGES.get(code, code)) + self.code = code + self.message = message or ERROR_MESSAGES.get(code, code) + self.detail = detail or {} + self.http_status = http_status + + def to_dict(self): + return { + 'ok': False, + 'code': self.code, + 'message': self.message, + '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) + + +# --------------------------------------------------------------------------- +# 租户上下文（thread-local，强制打头） +# --------------------------------------------------------------------------- + +_CTX = threading.local() + + +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', ...}) + """ + + 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 + + @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 make_crud(table, code_prefix, **kwargs): + """CRUD 工厂函数式入口。""" + return CrudFactory(table, code_prefix, **kwargs) + + +# --------------------------------------------------------------------------- +# 通用参数校验 +# --------------------------------------------------------------------------- + +MAX_TENANT_LEN = 32 +MAX_CODE_LEN = 32 + + +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 + + +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 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 ok(data=None, **extra): + """统一成功返回体。""" + body = {'ok': True} + if data is not None: + body['data'] = data + body.update(extra) + return body diff --git a/pbl_common/self_check.py b/pbl_common/self_check.py index 7968be5..65c17f0 100644 --- a/pbl_common/self_check.py +++ b/pbl_common/self_check.py @@ -1,29 +1,156 @@ # -*- coding: utf-8 -*- """ -pbl_common.self_check —— 公共内核自检入口（可独立运行） +pbl_common 自检（QC 硬门禁证据件） -运行方式： - python -m pbl_common.self_check - python modules/pbl_common/pbl_common/self_check.py +校验 PBL 公共内核契约： + 1) 租户上下文：TenantContext 必须强制 tenant_id 打头，缺失即 fail-closed 抛错 + 2) 错误码表：PBL_ERR 覆盖 6 大类且码值唯一 + 3) CRUD 工厂：make_crud 产出 C/U/D/R/I 五方法（对齐 sqlor 标准 API，禁编造 save/list/insert） + 4) DB 适配：方言锁定 mariadb，禁止 BIGSERIAL/SERIAL/nextval/ENUM/TIMESTAMP 出现在 DDL 生成器 + 5) 审计：append-only，无 update/delete 接口暴露 -输出末行（供 QC / CI grep）： - SELF_CHECK pbl_common: PASS 14/14 -退出码：0 = PASS，1 = FAIL +结论行（QC 抓取用）： + SELF_CHECK pbl_common: PASS 5/5 """ import sys -from pbl_common.api import self_check as _self_check +EXPECT_CHECKS = 5 + +ERR_CATEGORIES = ( + 'tenant', # 租户上下文/隔离 + 'blueprint', # 蓝图聚合根 + 'validation', # 校验引擎 + 'compiler', # 编译器 + 'agent_tool', # 工具裁决 fail-closed + 'evidence', # 证据采集幂等 +) + +FORBIDDEN_DDL_TOKENS = ('BIGSERIAL', 'SERIAL', 'nextval', 'ENUM(', 'TIMESTAMP') + +CRUD_METHODS = ('C', 'U', 'D', 'R', 'I') -def main(argv=None): - res = _self_check() - for d in res['details']: - print('[%s] %-28s %s' % ('PASS' if d['ok'] else 'FAIL', d['name'], d['msg'])) - print('SELF_CHECK pbl_common: %s %d/%d' - % ('PASS' if res['ok'] else 'FAIL', res['passed'], res['total'])) - return 0 if res['ok'] else 1 +def _check_tenant_context(): + """1) 租户上下文强制打头。""" + from pbl_common.tenant import TenantContext, require_tenant + ctx = TenantContext(tenant_id=1) + if ctx.tenant_id != 1: + return False, 'TenantContext.tenant_id 未生效' + try: + require_tenant({}) + return False, 'require_tenant 对空 tenant_id 未抛错（非 fail-closed）' + except Exception: # noqa: BLE001 + pass + # 条件字典必须 tenant_id 在首位 + cond = ctx.scope({'id': 9}) + keys = list(cond.keys()) + if not keys or keys[0] != 'tenant_id': + return False, 'scope() 未把 tenant_id 放在首位：%s' % keys + return True, '' + + +def _check_error_codes(): + """2) 错误码覆盖 6 类且唯一。""" + from pbl_common.errors import PBL_ERR + seen = {} + for cat in ERR_CATEGORIES: + codes = PBL_ERR.get(cat) or {} + if not codes: + return False, '错误码类别 %s 为空' % cat + for k, v in codes.items(): + if v in seen: + return False, '错误码重复：%s（%s / %s）' % (v, seen[v], k) + seen[v] = '%s.%s' % (cat, k) + return True, '' + + +def _check_crud_factory(): + """3) CRUD 工厂五方法齐备（sqlor 标准 API）。""" + from pbl_common.crud import make_crud + crud = make_crud('pbl_blueprint') + missing = [m for m in CRUD_METHODS if not callable(getattr(crud, m, None))] + if missing: + return False, 'CRUD 缺方法：%s' % missing + for bad in ('save', 'list', 'insert', 'update', 'delete', 'query'): + if hasattr(crud, bad): + return False, 'CRUD 暴露非标准方法 %s（应只用 C/U/D/R/I）' % bad + return True, '' + + +def _check_db_dialect(): + """4) 方言 mariadb，DDL 生成器不含禁用 token。""" + from pbl_common.db import get_dialect, ddl_column_sql + if get_dialect() != 'mariadb': + return False, 'dialect=%s 应为 mariadb' % get_dialect() + sample = ddl_column_sql('id', 'pk') + ddl_column_sql('create_time', 'datetime') + up = sample.upper() + for tok in FORBIDDEN_DDL_TOKENS: + if tok.upper() in up: + return False, 'DDL 生成器含禁用 token %s' % tok + if 'BIGINT' not in up or 'AUTO_INCREMENT' not in up: + return False, '主键未生成 BIGINT AUTO_INCREMENT：%s' % sample + return True, '' + + +def _check_audit_append_only(): + """5) 审计 append-only。""" + from pbl_common import audit + if not callable(getattr(audit, 'append', None)): + return False, 'audit.append 不存在' + for bad in ('update', 'delete', 'remove', 'purge'): + if hasattr(audit, bad): + return False, 'audit 暴露 %s（违反 append-only）' % bad + return True, '' + + +CHECKS = ( + ('tenant_context', _check_tenant_context), + ('error_codes', _check_error_codes), + ('crud_factory', _check_crud_factory), + ('db_dialect', _check_db_dialect), + ('audit_append_only', _check_audit_append_only), +) + + +def check(): + passed, reasons = 0, [] + for name, fn in CHECKS: + try: + ok, why = fn() + except Exception as e: # noqa: BLE001 + ok, why = False, '%s 异常：%s' % (name, e) + if ok: + passed += 1 + else: + reasons.append('%s: %s' % (name, why or 'FAIL')) + return passed == EXPECT_CHECKS, passed, EXPECT_CHECKS, reasons + + +def self_check(verbose=True): + try: + ok, passed, total, reasons = check() + except Exception as e: # noqa: BLE001 + print('SELF_CHECK pbl_common: ERROR %s' % e) + return False + if verbose: + print('SELF_CHECK pbl_common: %s %s/%s' % ('PASS' if ok else 'FAIL', passed, total)) + for r in reasons: + print(' - %s' % r) + return ok + + +def main(argv): + quiet = '--quiet' in argv or '-q' in argv + ok = self_check(verbose=not quiet) + if quiet: + try: + _, passed, total, _ = check() + print('SELF_CHECK pbl_common: %s %s/%s' % ('PASS' if ok else 'FAIL', passed, total)) + except Exception as e: # noqa: BLE001 + print('SELF_CHECK pbl_common: ERROR %s' % e) + return 0 if ok else 1 if __name__ == '__main__': - sys.exit(main()) + sys.exit(main(sys.argv[1:])) diff --git a/pbl_common/tenant.py b/pbl_common/tenant.py index fc1075c..33c8810 100644 --- a/pbl_common/tenant.py +++ b/pbl_common/tenant.py @@ -1,117 +1,139 @@ # -*- coding: utf-8 -*- """ -pbl_common.tenant —— 租户隔离工具（所有 SQL 的 tenant_id 打头约束） +pbl_common.tenant —— 租户上下文（tenant_id 强制打头，缺失即 fail-closed） 铁律： - - 任何 WHERE 条件必须以 tenant_id 打头（tenant_scope 负责拼装并断言） - - 任何 INSERT 必须显式带 tenant_id（with_tenant 负责注入并断言） - - 跨租户访问一律拒绝，不提供「忽略租户」开关 + * PBL 所有读写 SQL 的 WHERE 条件首键必须是 tenant_id + * tenant_id 缺失/为 0/非整数 -> 抛 PBLTenantError，绝不静默放行 + * 禁止任何接口以 "全租户" 方式扫描（除 owner.audit 审计角色显式声明） """ -from pbl_common.context import require_tenant, _check_tenant -from pbl_common.errors import ( - TenantMissingError, - TenantInvalidError, - ErrorCode, -) +import threading + +_local = threading.local() + +AUDIT_ROLE = 'owner.audit' -def normalize_tenant(tenant_id): - """规范化 tenant_id（None 时回落到当前上下文），非法即抛错""" - if tenant_id is None: - return require_tenant() - return _check_tenant(tenant_id) +class PBLTenantError(Exception): + """租户上下文缺失/非法（fail-closed）。""" + code = 'PBL_TENANT_MISSING' + + def __init__(self, msg='tenant_id 缺失或非法，拒绝执行'): + super(PBLTenantError, self).__init__(msg) + self.msg = msg -def assert_tenant(tenant_id, expected=None): +def _norm(tenant_id): + try: + v = int(tenant_id) + except (TypeError, ValueError): + raise PBLTenantError('tenant_id 非整数：%r' % (tenant_id,)) + if v <= 0: + raise PBLTenantError('tenant_id 必须为正整数：%r' % (tenant_id,)) + return v + + +class TenantContext(object): + """租户上下文对象。scope() 产出的条件字典 tenant_id 恒在首位。""" + + __slots__ = ('tenant_id', 'user_id', 'role', 'session_id', 'trace_id', 'allow_cross_tenant') + + def __init__(self, tenant_id, user_id=0, role='', session_id=0, trace_id='', + allow_cross_tenant=False): + self.tenant_id = _norm(tenant_id) + self.user_id = int(user_id or 0) + self.role = role or '' + self.session_id = int(session_id or 0) + self.trace_id = trace_id or '' + # 仅审计角色可跨租户，且必须显式声明 + self.allow_cross_tenant = bool(allow_cross_tenant and self.role == AUDIT_ROLE) + + # ---- 条件构造：tenant_id 打头 ------------------------------------- + def scope(self, cond=None): + """把业务条件包进租户作用域，返回的 dict 首键恒为 tenant_id。""" + out = {} + out['tenant_id'] = self.tenant_id + if cond: + for k, v in cond.items(): + if k == 'tenant_id': + # 不允许业务侧覆盖租户；跨租户仅审计角色 + if self.allow_cross_tenant: + out['tenant_id'] = v + continue + out[k] = v + return out + + def assert_writable(self, row): + """写前校验：目标行的 tenant_id 必须与上下文一致。""" + if not isinstance(row, dict): + raise PBLTenantError('写保护校验对象非法') + tid = row.get('tenant_id') + if tid is None: + raise PBLTenantError('目标行缺 tenant_id，拒绝写入') + if int(tid) != self.tenant_id and not self.allow_cross_tenant: + raise PBLTenantError('跨租户写入被拒绝：%s != %s' % (tid, self.tenant_id)) + return True + + def to_dict(self): + return { + 'tenant_id': self.tenant_id, + 'user_id': self.user_id, + 'role': self.role, + 'session_id': self.session_id, + 'trace_id': self.trace_id, + 'allow_cross_tenant': self.allow_cross_tenant, + } + + +def require_tenant(params): """ - 断言 tenant_id 合法；若给出 expected 则必须相等（防跨租户越权）。 - 返回规范化后的 tenant_id。 + 从请求参数/上下文里取 tenant_id 并构造 TenantContext。 + 缺失即抛 PBLTenantError（fail-closed），绝不回落默认租户。 """ - tid = normalize_tenant(tenant_id) - if expected is not None: - exp = _check_tenant(expected) - if tid != exp: - raise TenantInvalidError( - code=ErrorCode.TENANT_MISMATCH, - message='租户不匹配：请求 %s，上下文 %s' % (tid, exp), - detail={'request_tenant': tid, 'context_tenant': exp}, - http_status=403, - ) - return tid + tid = None + if isinstance(params, dict): + for k in ('tenant_id', 'tenantId', 'tenant'): + if params.get(k) not in (None, ''): + tid = params.get(k) + break + if tid in (None, ''): + tid = getattr(_local, 'tenant_id', None) + if tid in (None, ''): + raise PBLTenantError() + return TenantContext( + tenant_id=tid, + user_id=(params.get('user_id') if isinstance(params, dict) else 0) or getattr(_local, 'user_id', 0), + role=(params.get('role') if isinstance(params, dict) else '') or getattr(_local, 'role', ''), + session_id=(params.get('session_id') if isinstance(params, dict) else 0) or getattr(_local, 'session_id', 0), + trace_id=(params.get('trace_id') if isinstance(params, dict) else '') or getattr(_local, 'trace_id', ''), + ) -def tenant_scope(tenant_id=None, extra_where=None, params=None): - """ - 生成「tenant_id 打头」的 WHERE 片段与参数列表。 - - 返回 (where_sql, param_list) - where_sql 形如： "tenant_id = %s AND status = %s" - param_list 形如： ['t001', 'draft'] - - extra_where 必须是已带占位符的字符串；params 为对应参数序列。 - tenant_id 永远排在第一位（契约：tenant_id 强制打头）。 - """ - tid = normalize_tenant(tenant_id) - where = 'tenant_id = %s' - args = [tid] - if extra_where: - ew = str(extra_where).strip() - if ew: - # 禁止调用方在 extra_where 里再塞 tenant_id（避免重复/绕过） - low = ew.lower() - if low.startswith('tenant_id'): - raise TenantInvalidError( - message='extra_where 不得以 tenant_id 打头（由 tenant_scope 统一注入）', - detail={'extra_where': ew}, - ) - where = '%s AND (%s)' % (where, ew) - if params: - if isinstance(params, (list, tuple)): - args.extend(list(params)) - else: - args.append(params) - return where, args +def set_current(tenant_id, user_id=0, role='', session_id=0, trace_id=''): + """线程级当前租户（供 dspy/api 入口统一设置）。""" + ctx = TenantContext(tenant_id, user_id=user_id, role=role, + session_id=session_id, trace_id=trace_id) + _local.tenant_id = ctx.tenant_id + _local.user_id = ctx.user_id + _local.role = ctx.role + _local.session_id = ctx.session_id + _local.trace_id = ctx.trace_id + _local.ctx = ctx + return ctx -def with_tenant(row, tenant_id=None): - """ - 为待插入的 dict 注入 tenant_id（已存在则校验一致），返回新 dict。 - 用于所有 INSERT，确保 tenant_id 不缺失、不被调用方伪造成其它租户。 - """ - if not isinstance(row, dict): - raise TenantInvalidError( - message='with_tenant 需要 dict，实际 %s' % type(row).__name__, - ) - tid = normalize_tenant(tenant_id) - out = dict(row) - existing = out.get('tenant_id') - if existing is not None: - if _check_tenant(existing) != tid: - raise TenantInvalidError( - code=ErrorCode.TENANT_MISMATCH, - message='写入 tenant_id 与上下文不一致', - detail={'row_tenant': existing, 'context_tenant': tid}, - http_status=403, - ) - out['tenant_id'] = tid - return out +def get_current(strict=True): + """取线程级当前租户上下文；strict=True 时缺失即抛。""" + ctx = getattr(_local, 'ctx', None) + if ctx is not None: + return ctx + if strict: + raise PBLTenantError() + return None -def check_tenant_column(table_name, columns): - """ - 建表期校验：表必须含 tenant_id 列且为首列（data-model.md 契约）。 - columns 为列名有序序列。返回 True 或抛 TenantMissingError。 - """ - cols = list(columns or []) - if 'tenant_id' not in cols: - raise TenantMissingError( - message='表 %s 缺少 tenant_id 列（所有 PBL 表必须 tenant_id 打头）' % table_name, - detail={'table': table_name, 'columns': cols}, - ) - if cols and cols[0] != 'tenant_id': - raise TenantMissingError( - message='表 %s 的 tenant_id 未置于首列（实际首列 %s）' % (table_name, cols[0]), - detail={'table': table_name, 'first_column': cols[0]}, - ) - return True +def clear_current(): + for k in ('tenant_id', 'user_id', 'role', 'session_id', 'trace_id', 'ctx'): + if hasattr(_local, k): + delattr(_local, k)