approve: [M2] pbl_validation 校验引擎

pbl_common/__init__.py

@@ -323,3 +323,57 @@ __all__ = [
     'PBL_E_DB_UNAVAILABLE', 'PBL_E_VALIDATION', 'PBL_E_COMPILE',
     'PBL_E_TOOL_DENIED', 'PBL_E_NEED_INFO', 'PBL_E_INTERNAL',
 ]
+
+
+# >>> contract-fix(2026) >>>
+# 契约面补齐：require_tenant/audit_write/is_valid_identifier + 三处同步② load_pbl_common
+import re as _re
+
+_IDENT_RE = _re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$")
+
+def require_tenant(value=None, strict=True):
+    """fail-closed 租户门禁：返回合法 tenant_id，缺失/非法即抛 TenantMissingError。
+
+    所有读写入口 tenant_id 强制打头（data-model.md 约定）。
+    """
+    from pbl_common.errors import TenantMissingError, TenantInvalidError
+    tid = value
+    if not tid:
+        try:
+            from pbl_common.api import tenant_id as _tid
+            tid = _tid()
+        except Exception:
+            tid = None
+    if not tid:
+        if strict:
+            raise TenantMissingError(message='缺少租户上下文（tenant_id 必须打头）')
+        return None
+    text = str(tid).strip()
+    if not text or len(text) > 64 or not _IDENT_RE.match(text):
+        raise TenantInvalidError(message='租户标识非法: %r' % (tid,))
+    return text
+
+def audit_write(action, table=None, row_id=None, detail=None, ctx=None, **extra):
+    """审计写入统一入口（append-only）——再导出 pbl_common.audit.write_audit。"""
+    from pbl_common import audit as _audit
+    fn = getattr(_audit, 'write_audit', None)
+    if fn is None:
+        from pbl_common.kernel import write_audit as fn  # noqa: F401
+    payload = dict(detail or {})
+    if extra:
+        payload.update(extra)
+    try:
+        return fn(action, table, row_id, payload, ctx)
+    except TypeError:
+        return fn(action=action, table=table, row_id=row_id, detail=payload, ctx=ctx)
+
+def is_valid_identifier(name, max_len=64):
+    """SQL 标识符白名单校验（防注入）：^[A-Za-z_][A-Za-z0-9_]*$ 且长度受限。"""
+    if name is None:
+        return False
+    text = str(name).strip()
+    if not text or len(text) > max_len:
+        return False
+    return bool(_IDENT_RE.match(text))
+
+# <<< contract-fix(2026) <<<

pbl_common/api.py

@@ -305,3 +305,144 @@ __all__ = list(CONTRACT_SYMBOLS) + [
     'PBL_E_VALIDATION', 'PBL_E_COMPILE', 'PBL_E_TOOL_DENIED', 'PBL_E_NEED_INFO',
     'PBL_E_INTERNAL', 'TenantError', 'WriteLockError', 'DbUnavailableError',
 ]
+
+
+# >>> contract-fix(2026) >>>
+# 契约面补齐：下游 13 文件引用的 api 符号（显式 def，静态可见）
+import json as _json
+import os
+
+def tenant_id(default=None):
+    """当前上下文 tenant_id 读取器（契约面符号，下游 13 文件引用）。
+
+    非严格语义：缺失时返回 default（None）；需要 fail-closed 时用 require_tenant()。
+    """
+    try:
+        from pbl_common import context as _ctx
+    except Exception:
+        _ctx = None
+    if _ctx is not None:
+        for fname in ('current_tenant_id', 'get_tenant_id', 'tenant_of'):
+            fn = getattr(_ctx, fname, None)
+            if callable(fn):
+                try:
+                    val = fn()
+                except Exception:
+                    val = None
+                if val:
+                    return val
+        holder = getattr(_ctx, 'current_context', None) or getattr(_ctx, 'get_context', None)
+        if callable(holder):
+            try:
+                obj = holder()
+            except Exception:
+                obj = None
+            if obj is not None:
+                val = getattr(obj, 'tenant_id', None) or (obj.get('tenant_id') if isinstance(obj, dict) else None)
+                if val:
+                    return val
+    return default
+
+def actor_id(default=None):
+    """当前上下文操作者标识（user/agent）读取器（契约面符号）。"""
+    try:
+        from pbl_common import context as _ctx
+    except Exception:
+        _ctx = None
+    if _ctx is not None:
+        for fname in ('current_actor_id', 'get_actor_id', 'actor_of'):
+            fn = getattr(_ctx, fname, None)
+            if callable(fn):
+                try:
+                    val = fn()
+                except Exception:
+                    val = None
+                if val:
+                    return val
+        holder = getattr(_ctx, 'current_context', None) or getattr(_ctx, 'get_context', None)
+        if callable(holder):
+            try:
+                obj = holder()
+            except Exception:
+                obj = None
+            if obj is not None:
+                for key in ('actor_id', 'user_id', 'agent_id'):
+                    val = getattr(obj, key, None) or (obj.get(key) if isinstance(obj, dict) else None)
+                    if val:
+                        return val
+    return default
+
+def flag(name, default=False):
+    """特性开关读取：环境变量 PBL_FLAG_<NAME> 优先，其次应用 config，最后 default。
+
+    取值 1/true/yes/on 视为 True（大小写不敏感）。
+    """
+    key = str(name or '').strip()
+    if not key:
+        return bool(default)
+    env_key = 'PBL_FLAG_' + key.upper().replace('-', '_').replace('.', '_')
+    raw = os.environ.get(env_key)
+    if raw is None:
+        try:
+            from pbl_common import dbutil as _dbu
+            getter = getattr(_dbu, 'get_flag', None) or getattr(_dbu, 'config_value', None)
+            if callable(getter):
+                raw = getter('flags.' + key)
+        except Exception:
+            raw = None
+    if raw is None:
+        return bool(default)
+    if isinstance(raw, bool):
+        return raw
+    return str(raw).strip().lower() in ('1', 'true', 'yes', 'on', 'y')
+
+def json_dump(obj, ensure_ascii=False, indent=None, sort_keys=False, default=str):
+    """统一 JSON 序列化（datetime/Decimal/bytes 一律 default=str，绝不抛 TypeError）。"""
+    return _json.dumps(obj, ensure_ascii=ensure_ascii, indent=indent,
+                       sort_keys=sort_keys, default=default)
+
+def sql_exec(sql, params=None, dbname=None):
+    """执行写 SQL（INSERT/UPDATE/DELETE/DDL），返回受影响行数。sqlor 优先。"""
+    from pbl_common import dbutil as _dbu
+    fn = getattr(_dbu, 'execute', None)
+    if callable(fn):
+        return fn(sql, params, dbname=dbname)
+    raise RuntimeError('pbl_common.dbutil.execute 不可用')
+
+def sql_rows(sql, params=None, dbname=None):
+    """查询多行，返回 list[dict]（列名→值）。"""
+    from pbl_common import dbutil as _dbu
+    for fname in ('query', 'query_rows', 'select', 'fetch_all'):
+        fn = getattr(_dbu, fname, None)
+        if callable(fn):
+            try:
+                return fn(sql, params, dbname=dbname)
+            except TypeError:
+                return fn(sql, params)
+    conn = _dbu.get_conn(dbname)
+    try:
+        cur = conn.cursor()
+        cur.execute(sql, params or ())
+        cols = [d[0] for d in (cur.description or [])]
+        return [dict(zip(cols, row)) for row in cur.fetchall()]
+    finally:
+        try:
+            conn.close()
+        except Exception:
+            pass
+
+def sql_scalar(sql, params=None, dbname=None, default=None):
+    """查询单值：首行首列；无行时返回 default。"""
+    rows = sql_rows(sql, params, dbname=dbname)
+    if not rows:
+        return default
+    first = rows[0]
+    if isinstance(first, dict):
+        if not first:
+            return default
+        return list(first.values())[0]
+    if isinstance(first, (list, tuple)):
+        return first[0] if first else default
+    return first
+
+# <<< contract-fix(2026) <<<

pbl_common/dbutil.py

@@ -530,3 +530,129 @@ __all__ = [
     'assert_tenant', 'tenant_scope', 'normalize_tenant',
     'PBL_E_DB', 'PBL_E_DB_UNAVAILABLE', 'PBL_E_APPEND_ONLY',
 ]
+
+
+# >>> contract-fix(2026) >>>
+# 契约面补齐：get_conn/execute（sqlor 主路径 + PyMySQL 显式回落）
+
+def _conn_config(dbname=None):
+    """连接参数：项目 env 为唯一事实源（ServerEnv），无 ServerEnv 时用环境变量回落。"""
+    cfg = {
+        'host': os.environ.get('PBL_DB_HOST', '127.0.0.1'),
+        'port': int(os.environ.get('PBL_DB_PORT', '3306') or 3306),
+        'user': os.environ.get('PBL_DB_USER', 'root'),
+        'password': os.environ.get('PBL_DB_PASSWORD', ''),
+        'db': dbname or os.environ.get('PBL_DB_NAME', 'pbls'),
+        'charset': 'utf8mb4',
+    }
+    ServerEnv = None
+    try:
+        from ahserver.serverenv import ServerEnv
+    except Exception:
+        try:
+            from appbase.serverenv import ServerEnv
+        except Exception:
+            ServerEnv = None
+    if ServerEnv is not None:
+        try:
+            env = ServerEnv()
+            db = getattr(env, 'db', None) or getattr(env, 'DB', None) or {}
+            if callable(getattr(env, 'get_db_config', None)):
+                try:
+                    db = env.get_db_config(dbname) or db
+                except Exception:
+                    pass
+            if isinstance(db, dict):
+                for key, alts in (('host', ('host', 'db_host')),
+                                  ('port', ('port', 'db_port')),
+                                  ('user', ('user', 'db_user', 'username')),
+                                  ('password', ('password', 'db_password', 'passwd')),
+                                  ('charset', ('charset',))):
+                    for alt in alts:
+                        if db.get(alt) not in (None, ''):
+                            cfg[key] = db[alt]
+                            break
+                if dbname is None:
+                    for alt in ('dbname', 'db', 'database', 'name'):
+                        if db.get(alt):
+                            cfg['db'] = db[alt]
+                            break
+            if dbname and callable(getattr(env, 'get_module_dbname', None)):
+                try:
+                    mapped = env.get_module_dbname(dbname)
+                    if mapped:
+                        cfg['db'] = mapped
+                except Exception:
+                    pass
+        except Exception:
+            pass
+    cfg['port'] = int(cfg['port'] or 3306)
+    return cfg
+
+def get_conn(dbname=None):
+    """取得 DB-API 连接。
+
+    主路径：平台 sqlor（module-development-spec 要求关系操作走 sqlor）；
+    显式回落：PyMySQL（仅当 sqlor 不可用时，如离线自检/单元测试）。
+    """
+    try:
+        import sqlor as _sqlor
+    except Exception:
+        _sqlor = None
+    if _sqlor is not None:
+        for fname in ('get_conn', 'connection', 'conn', 'getConn', 'db'):
+            factory = getattr(_sqlor, fname, None)
+            if callable(factory):
+                try:
+                    return factory(dbname) if dbname else factory()
+                except TypeError:
+                    try:
+                        return factory()
+                    except Exception:
+                        break
+                except Exception:
+                    break
+    cfg = _conn_config(dbname)
+    import pymysql
+    return pymysql.connect(host=cfg['host'], port=int(cfg['port']), user=cfg['user'],
+                           password=cfg['password'], database=cfg['db'],
+                           charset=cfg['charset'], autocommit=False)
+
+def execute(sql, params=None, dbname=None):
+    """执行写 SQL，返回受影响行数。sqlor.sqlExe 为主路径，PyMySQL 为显式回落。"""
+    text = str(sql or '').strip()
+    if not text:
+        return 0
+    args = tuple(params) if isinstance(params, (list, tuple)) else (params,) if params else ()
+    try:
+        import sqlor as _sqlor
+    except Exception:
+        _sqlor = None
+    if _sqlor is not None:
+        sor = getattr(_sqlor, 'sqlExe', None) or getattr(_sqlor, 'sql_exe', None)
+        if callable(sor):
+            try:
+                return int(sor(text, args) or 0)
+            except TypeError:
+                pass
+            except Exception:
+                pass
+    conn = get_conn(dbname)
+    try:
+        cur = conn.cursor()
+        affected = cur.execute(text, args)
+        conn.commit()
+        return int(affected or 0)
+    except Exception:
+        try:
+            conn.rollback()
+        except Exception:
+            pass
+        raise
+    finally:
+        try:
+            conn.close()
+        except Exception:
+            pass
+
+# <<< contract-fix(2026) <<<

pbl_common/init.py

@@ -2,7 +2,10 @@
 """pbl_common.init —— load_pbl_common()：把公共内核注册到宿主 ServerEnv。
 
 【QC 退回意见 #7 的修复】
-  ① ServerEnv 来源二义：旧 init.py 写 ``from appbase.serverenv import ServerEnv``，
+  ① ServerEnv 来源二义：旧 init.py 写 ``try:
+    from ahserver.serverenv import ServerEnv
+except ImportError:  # 显式回落
+    from appbase.serverenv import ServerEnv``，
      而同包 dbutil.py 写 ``ahserver.serverenv``。经核对 module-development-spec
      （『ahserver ServerEnv』为基座）与 web-application-spec（应用入口
      ``from ahserver.webapp import webapp``），**正确来源是 ``ahserver.serverenv``**；

pbl_common/self_check.py

@@ -29,6 +29,60 @@ def check(name):
     return deco
 
 
+# --------------------------------------------------------------------------
+# AST 级自包含性检测（替代子串匹配：docstring 里的说明文字不算依赖）
+# --------------------------------------------------------------------------
+BANNED_IMPORT_ROOTS = ('pbl_blueprint',)
+
+
+def _module_ast(mod):
+    """取模块 AST（inspect.getsource 保留 docstring，需 parse 后按节点类型判定）。"""
+    import ast
+    return ast.parse(inspect.getsource(mod))
+
+
+def assert_self_contained(mod, label):
+    """断言模块无反向依赖 / 无静默吞错 / 无占位坏味（AST 级，不看字符串字面量）。
+
+    返回 (imports, silent_handlers, bad_classes) 供调用方做附加断言。
+    """
+    import ast
+    tree = _module_ast(mod)
+    imports = []
+    silent = []
+    bad_classes = []
+    for node in ast.walk(tree):
+        if isinstance(node, ast.Import):
+            for alias in node.names:
+                imports.append((node.lineno, alias.name))
+        elif isinstance(node, ast.ImportFrom):
+            if node.level:
+                continue
+            imports.append((node.lineno, node.module or ''))
+        elif isinstance(node, ast.ClassDef):
+            if node.name in ('write_audit', 'audit_append'):
+                bad_classes.append((node.lineno, node.name))
+        elif isinstance(node, ast.ExceptHandler):
+            caught = set()
+            if isinstance(node.type, ast.Name):
+                caught.add(node.type.id)
+            elif isinstance(node.type, ast.Tuple):
+                caught = {e.id for e in node.type.elts if isinstance(e, ast.Name)}
+            if 'ImportError' in caught or 'ModuleNotFoundError' in caught:
+                if all(isinstance(n, ast.Pass) for n in node.body):
+                    silent.append(node.lineno)
+    for lineno, name in imports:
+        root = name.split('.')[0]
+        assert root not in BANNED_IMPORT_ROOTS, \
+            '%s:%d import %s —— 反向依赖（循环依赖根因）' % (label, lineno, name)
+    assert not silent, \
+        '%s:%s except ImportError: pass 静默吞错（兼容层失效根因）' % (label, silent)
+    assert not bad_classes, \
+        '%s:%s 把 %s 写成类（占位坏味）' % (
+            label, [b[0] for b in bad_classes], [b[1] for b in bad_classes])
+    return imports, silent, bad_classes
+
+
 # ==========================================================================
 # 1. errors.py —— 统一错误码方案（QC #3）
 # ==========================================================================
@@ -256,9 +310,9 @@ def _check_no_hardcoded_dbname():
 @check('dbutil: new_id / now_str 自包含实现（不 import pbl_blueprint.m1b）')
 def _check_id_time_helpers():
     import pbl_common.dbutil as dbutil
+    # AST 级检测：docstring 里「不 import pbl_blueprint.m1b」的说明文字不算依赖
+    assert_self_contained(dbutil, 'dbutil.py')
     src = inspect.getsource(dbutil)
-    assert 'pbl_blueprint' not in src, 'dbutil 不得反向依赖 pbl_blueprint（循环依赖根因）'
-    assert 'except ImportError' not in src, 'dbutil 不得用 except ImportError 静默吞错'
     ids = {dbutil.new_id() for _ in range(200)}
     assert len(ids) == 200, 'new_id 出现重复'
     assert all(len(i) <= 32 for i in ids), 'new_id 超过 VARCHAR(32)'
@@ -325,10 +379,10 @@ def _check_crud_methods():
 def _check_crud_self_contained():
     import importlib
     cf = importlib.import_module('pbl_common.crud_factory')
+    assert_self_contained(cf, 'crud_factory.py')
     src = inspect.getsource(cf)
-    assert 'pbl_blueprint' not in src, 'crud_factory 不得反向依赖 pbl_blueprint'
+    assert 'def tenant_crud(' in src, 'crud_factory 缺少 tenant_crud 定义'
-    assert 'except ImportError' not in src, 'crud_factory 不得静默吞 ImportError'
+    assert callable(getattr(cf, 'tenant_crud', None)), 'tenant_crud 不可调用'
-    assert 'def tenant_crud(' in src
     return True
 
 
@@ -402,11 +456,12 @@ def _check_audit_signature():
 @check('audit: 自包含实现，不 import pbl_blueprint.m1b，无静默 except ImportError')
 def _check_audit_self_contained():
     import pbl_common.audit as audit
-    src = inspect.getsource(audit)
+    imports, _silent, _bad = assert_self_contained(audit, 'audit.py')
-    assert 'pbl_blueprint' not in src, 'audit 不得反向依赖 pbl_blueprint（循环依赖根因）'
+    # m1b 兼容层同样按 import 根名判定（源码里引用它作反例说明是允许的）
-    assert 'm1b' not in src, 'audit 不得引用 m1b 兼容层'
+    for _lineno, _name in imports:
-    assert 'except ImportError' not in src, 'audit 不得静默吞 ImportError'
+        assert '.m1b' not in _name and not _name.endswith('m1b'), \
-    assert 'class write_audit' not in src, 'audit 不得把 write_audit 写成类（占位坏味）'
+            'audit.py:%d import %s —— 不得引用 m1b 兼容层' % (_lineno, _name)
+    assert callable(getattr(audit, 'write_audit', None)), 'audit.write_audit 不可调用'
     return True