feat: 集成 chongming-license 机器码授权门禁（打包可选、授权版不可关）

三处收口，未授权插件不加载、不暴露端点、不可管理： - core/licensing：授权管理器单例 + 构建期烘焙决策（chongming/_build_flags.py，随代码编译、部署侧不可改）；开发/测试默认放行，CHONGMING_LICENSE_MODE 可临时开 - lifespan_setup：_scan_plugins（盘扫）与 _discover_plugins（授权收口）分离，未授权插件连同 TOML 路由/MQTT 主题/副本数一起从启动流程消失（契约校验因此不会硬失败），启动打印授权横幅 - route_registry.set_denied_types：未授权类型经 Zenoh 推来的描述符一律忽略（否则把 load_mode 改成 external 即可绕过） - worker_admin：扩缩容/重启/读改配置对未授权插件一律 403，实例盘点附 licensed 标注 - routes/licensing：/license/{info,reload,file}（仅超管），导入先验签后原子落盘，坏文件不覆盖现有授权；导入不即时加载插件，如实告知需重启打包可选：不加 --license 出"非授权版"（不校验、插件照常加载）；授权版强制校验且公钥随代码内置（现场换 config.toml 公钥自签也无效）。未烘焙的旧打包产物沿用 "打包即强制"的旧语义；烘焙标记只在产物内生效，不影响源码运行。补依赖 greenlet（chongming-auth→database 的 sqlalchemy asyncio import 期即要求）与 python-multipart（上传端点注册路由即要求）；新增 test/test_license_gate.py （52 项：门禁/契约存活/描述符拦截/管理接口/上传安全/烘焙三态，自造夹具直接运行）。
2026-10-14 10:57:09 +08:00
parent dd1e6ed846
commit 93fed12d14
12 changed files with 1456 additions and 15 deletions
@@ -0,0 +1,625 @@
+"""机器码授权门禁测试（网关侧）。
+
+核心验证目标 —— 未授权插件被拒绝时，**网关必须仍能正常启动**：
+
+1. 门禁收口在 `_discover_plugins()`：未授权插件不出现在返回列表中，
+   且 `denied_plugin_types()` 能给出拒绝原因；
+2. **契约校验不硬失败**：这是最容易踩的坑 —— `validate_contracts()` 会把
+   "TOML 路由没有对应 @rpc_method 注册"视为硬性错误并终止启动。若把"不加载
+   未授权插件"实现成"跳过 import 但保留在插件列表里"，该插件的 TOML 路由就会
+   变成孤儿 → 设备起不来。本测试确保收口位置正确（并反向验证夹具确实会硬失败，
+   防止测试因夹具写错而静默通过）；
+3. 路由描述符门禁：未授权类型的 external 路由描述符被忽略（防"改 load_mode
+   成 external 绕过门禁"）；
+4. 管理接口门禁：未授权插件的扩缩容/重启/配置接口一律 403，实例盘点如实标注；
+5. 授权管理接口：超管门禁、机器码/申请码输出，以及"导入坏文件不覆盖现有授权"；
+6. 有效期与开关：过期授权不放行任何插件；未启用校验时（开发/测试）全部放行。
+
+测试在临时工作目录里自造夹具（config.toml + config/workers/**），不依赖宿主
+项目文件。运行方式：`python test/test_license_gate.py`。
+"""
+import asyncio
+import base64
+import io
+import json
+import os
+import sys
+import tempfile
+import types
+from contextlib import contextmanager
+from datetime import date, timedelta
+from pathlib import Path
+from types import SimpleNamespace
+
+ROOT = Path(__file__).resolve().parent.parent
+sys.path.insert(0, str(ROOT / "src"))
+
+# ---- 夹具工作目录 ---------------------------------------------------------
+# 网关全部相对路径（config.toml / config/workers/** / license.lic / 日志 / 缓存）
+# 均以 cwd 为基准，故先切到自造的临时目录，再导入网关模块。
+WORK = Path(tempfile.mkdtemp(prefix="gw-license-"))
+os.chdir(WORK)
+
+(WORK / "config.toml").write_text(
+    """
+[default]
+app.debug = false
+app.description = "license 门禁测试夹具"
+app.name = "gw-license-test"
+app.version = "0.0.1"
+env = "development"
+prefix = "/api/v1"
+
+[cache]
+backend = "sqlite"
+
+[security]
+algorithm = "HS256"
+secret_key = "test-secret"
+
+[license]
+enabled = true
+path = "license.lic"
+public_key = ""
+min_match = 1
+
+[development]
+debug = true
+env = "development"
+
+[development.server]
+access_log = false
+host = "127.0.0.1"
+port = 8000
+reload = false
+workers = 1
+
+[development.logging]
+file = "logs"
+format = "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
+level = "WARNING"
+""",
+    encoding="utf-8",
+)
+
+# 两个插件：alpha 在授权白名单内，beta 不在。beta 声明一条 TOML HTTP 路由而
+# 没有任何 @rpc_method —— 若它未被门禁过滤，契约校验将硬失败终止启动。
+for ptype in ("alpha", "beta"):
+    d = WORK / "config" / "workers" / ptype
+    d.mkdir(parents=True)
+    (d / "config.toml").write_text(
+        f"""
+[workers]
+type = "{ptype}"
+module = "plugins.{ptype}"
+factory = "create_workers"
+cleanup = "cleanup"
+count = 1
+
+[worker]
+name = "{ptype}"
+version = "0.1.0"
+
+[worker.registration]
+router_prefix = "/{ptype}"
+""",
+        encoding="utf-8",
+    )
+(WORK / "config" / "workers" / "beta" / "ping.toml").write_text(
+    '[POST]\nsummary = "beta 端点（无 @rpc_method）"\n', encoding="utf-8"
+)
+
+# 必须在导入网关模块前设置：licensing 的 enforce 在首次建管理器时读取
+os.environ["CHONGMING_LICENSE_MODE"] = "enforce"
+
+from chongming_config import get_config  # noqa: E402
+from cryptography.hazmat.primitives import serialization  # noqa: E402
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey  # noqa: E402
+from chongming_license import collect_fingerprints, machine_code, sign_payload  # noqa: E402
+
+import chongming  # noqa: E402
+from chongming.app.core import licensing  # noqa: E402
+from chongming.app.core.lifespan_setup import (  # noqa: E402
+    _discover_plugins,
+    _scan_plugins,
+    denied_plugin_types,
+)
+
+failures: list[str] = []
+
+
+def check(name, cond, detail=""):
+    status = "PASS" if cond else "FAIL"
+    print(f"[{status}] {name}" + (f" — {detail}" if detail else ""))
+    if not cond:
+        failures.append(name)
+
+
+# ---------------------------------------------------------------------------
+# 测试夹具：临时密钥对 + license 文件
+# ---------------------------------------------------------------------------
+def _make_keypair() -> tuple[str, Ed25519PrivateKey]:
+    priv = Ed25519PrivateKey.generate()
+    pub = base64.b64encode(
+        priv.public_key().public_bytes(
+            encoding=serialization.Encoding.Raw,
+            format=serialization.PublicFormat.Raw,
+        )
+    ).decode("ascii")
+    return pub, priv
+
+
+def _issue(tmpdir: Path, priv: Ed25519PrivateKey, plugins: list[str],
+           expires_days: int = 365, sources: dict | None = None) -> str:
+    src = sources if sources is not None else collect_fingerprints()
+    payload = {
+        "v": 1,
+        "lic_id": "L-TEST-0001",
+        "customer": "测试客户",
+        "machine_code": machine_code(src),
+        "sources": src,
+        "plugins": plugins,
+        "not_before": None,
+        "expires_at": (date.today() + timedelta(days=expires_days)).isoformat(),
+        "issued_at": "2026-09-20T00:00:00Z",
+        "note": "",
+    }
+    doc = {
+        "payload": payload,
+        "alg": "ed25519",
+        "kid": "test",
+        "sig": sign_payload(payload, priv),
+    }
+    path = tmpdir / "license.lic"
+    path.write_text(json.dumps(doc), encoding="utf-8")
+    return str(path)
+
+
+def _use_license(public_key: str, path: str, min_match: int = 1) -> None:
+    """把测试用的公钥/路径写进运行期配置并重建管理器。
+
+    直接改 `get_config()` 返回的 dict（chongming_config 缓存同一对象）。
+    """
+    cfg = get_config()
+    cfg.setdefault("license", {})
+    cfg["license"]["public_key"] = public_key
+    cfg["license"]["path"] = path
+    cfg["license"]["min_match"] = min_match
+    licensing.reset_manager()
+
+
+def _request_stub() -> SimpleNamespace:
+    """管理接口直调用的 Request 桩：只需要 app.state（inline/external 类型观测）。"""
+    return SimpleNamespace(app=SimpleNamespace(state=SimpleNamespace()))
+
+
+@contextmanager
+def _baked_build_flags(enforce: bool, public_key: str = ""):
+    """模拟打包脚本烘焙的 ``chongming/_build_flags.py``（编译进产物那份）。
+
+    直接塞进 sys.modules／父包属性：读取端是静态 import，等价于产物内已编译模块。
+    """
+    module = types.ModuleType("chongming._build_flags")
+    module.LICENSE_ENFORCE = enforce
+    module.LICENSE_PUBLIC_KEY = public_key
+    prev_mod = sys.modules.get("chongming._build_flags")
+    had_attr = hasattr(chongming, "_build_flags")
+    sys.modules["chongming._build_flags"] = module
+    chongming._build_flags = module
+    licensing.reset_manager()
+    try:
+        yield module
+    finally:
+        if prev_mod is None:
+            sys.modules.pop("chongming._build_flags", None)
+        else:
+            sys.modules["chongming._build_flags"] = prev_mod
+        if not had_attr and hasattr(chongming, "_build_flags"):
+            del chongming._build_flags
+        licensing.reset_manager()
+
+
+@contextmanager
+def _packaged(value: bool = True):
+    """模拟运行在打包产物中（Nuitka/PyInstaller 的 sys.frozen/__compiled__）。"""
+    original = licensing.is_packaged
+    licensing.is_packaged = lambda: value
+    licensing.reset_manager()
+    try:
+        yield
+    finally:
+        licensing.is_packaged = original
+        licensing.reset_manager()
+
+
+# ---------------------------------------------------------------------------
+# 1. 插件发现门禁
+# ---------------------------------------------------------------------------
+def test_gate_denies_unlicensed_plugins():
+    """未授权插件被剔除，且原因可查。"""
+    tmp = Path(tempfile.mkdtemp())
+    pub, priv = _make_keypair()
+    _use_license(pub, _issue(tmp, priv, plugins=["alpha"]))
+
+    scanned = sorted(p["type"] for p in _scan_plugins())
+    allowed = sorted(p["type"] for p in _discover_plugins())
+    denied = denied_plugin_types()
+
+    check("盘扫到全部插件", scanned == ["alpha", "beta"], f"scanned={scanned}")
+    check("只放行已授权插件", allowed == ["alpha"], f"allowed={allowed}")
+    check("未授权插件给出拒绝原因",
+          "beta" in denied and "白名单" in denied["beta"], str(denied))
+
+
+def test_contract_validation_survives_denial():
+    """关键：有插件被拒时契约校验不得抛异常（否则设备起不来）。"""
+    tmp = Path(tempfile.mkdtemp())
+    pub, priv = _make_keypair()
+    _use_license(pub, _issue(tmp, priv, plugins=["alpha"]))
+
+    from chongming.app.core.contracts import validate_contracts
+
+    inline = [p for p in _discover_plugins() if p.get("load_mode", "inline") != "external"]
+    try:
+        validate_contracts(inline)
+        check("被拒插件的 TOML 路由不会导致契约校验硬失败", True)
+    except Exception as exc:  # noqa: BLE001
+        check("被拒插件的 TOML 路由不会导致契约校验硬失败", False, repr(exc))
+
+    # 反向验证：夹具确实构成"硬失败"场景 —— 若 beta 未被剔除，校验必然抛错。
+    # 没有这一步，夹具写错（如路由文件缺失）会让上一条检查静默通过。
+    beta_entry = [p for p in _scan_plugins() if p["type"] == "beta"]
+    try:
+        validate_contracts(beta_entry)
+        check("夹具自身有效：beta 被纳入时会硬失败", False, "未抛异常")
+    except RuntimeError as exc:
+        check("夹具自身有效：beta 被纳入时会硬失败", True, str(exc)[:60])
+
+
+def test_all_denied_still_validates():
+    """极端情况：完全未授权（无 license 文件）时启动流程仍不硬失败。"""
+    tmp = Path(tempfile.mkdtemp())
+    pub, _priv = _make_keypair()
+    _use_license(pub, str(tmp / "not-exists.lic"))
+
+    from chongming.app.core.contracts import validate_contracts
+
+    allowed = _discover_plugins()
+    check("无授权文件时不放行任何插件", allowed == [], str(allowed))
+    try:
+        validate_contracts([])
+        check("完全未授权时契约校验不抛异常", True)
+    except Exception as exc:  # noqa: BLE001
+        check("完全未授权时契约校验不抛异常", False, repr(exc))
+
+
+# ---------------------------------------------------------------------------
+# 2. 路由描述符门禁（external 插件）
+# ---------------------------------------------------------------------------
+def test_route_registry_denies_descriptors():
+    """未授权类型的路由描述符被忽略（防改 load_mode 绕过）。"""
+    from fastapi import FastAPI
+
+    from chongming.app.core.route_registry import RouteRegistry
+
+    tmp = Path(tempfile.mkdtemp())
+    pub, priv = _make_keypair()
+    _use_license(pub, _issue(tmp, priv, plugins=["alpha"]))
+
+    denied = licensing.denied_types({p["type"] for p in _scan_plugins()})
+    check("denied_types 含未授权插件", "beta" in denied, str(denied))
+
+    async def run():
+        registry = RouteRegistry(FastAPI(), worker_manager=None, prefix="/api/v1")
+        registry.set_denied_types(denied)
+        descriptor = {
+            "schema_version": 1, "type": "beta",
+            "subject": "beta.ping", "method": "POST",
+            "path": "/beta/ping", "params": {},
+        }
+        await registry._handle_sample(
+            "chongming/routes/beta/ping",
+            __import__("zenoh").SampleKind.PUT,
+            json.dumps(descriptor).encode(),
+        )
+        return registry._groups, registry.route_types
+
+    groups, route_types = asyncio.run(run())
+    check("未授权类型的描述符未进入路由表", groups == {}, f"groups={list(groups)}")
+    check("未授权类型不出现在 route_types 中", route_types == [], str(route_types))
+
+
+# ---------------------------------------------------------------------------
+# 3. 管理接口门禁
+# ---------------------------------------------------------------------------
+def test_admin_gate_403_for_denied():
+    """未授权插件的管理操作一律 403。"""
+    from fastapi import HTTPException
+
+    from chongming.app.routes.worker_admin import _ensure_licensed
+
+    tmp = Path(tempfile.mkdtemp())
+    pub, priv = _make_keypair()
+    _use_license(pub, _issue(tmp, priv, plugins=["alpha"]))
+
+    try:
+        _ensure_licensed("beta")
+        check("未授权插件的管理操作被拒", False, "未抛 403")
+    except HTTPException as exc:
+        check("未授权插件的管理操作被拒", exc.status_code == 403, f"status={exc.status_code}")
+
+    try:
+        _ensure_licensed("alpha")
+        check("已授权插件的管理操作放行", True)
+    except HTTPException as exc:
+        check("已授权插件的管理操作放行", False, f"status={exc.status_code}")
+
+
+def test_list_workers_annotates_licensed():
+    """实例盘点对每个副本标注 licensed，并汇总拒绝类型（供前端禁用按钮）。"""
+    from chongming.app.routes.worker_admin import list_workers
+
+    tmp = Path(tempfile.mkdtemp())
+    pub, priv = _make_keypair()
+    _use_license(pub, _issue(tmp, priv, plugins=["alpha"]))
+
+    class _StubManager:
+        def list_workers(self):
+            return [
+                {"worker_type": "alpha", "worker_id": "worker-0-1", "alive": True},
+                {"worker_type": "beta", "worker_id": "worker-0-2", "alive": True},
+            ]
+
+    request = _request_stub()
+    request.app.state.worker_manager = _StubManager()
+    body = asyncio.run(list_workers(request, _admin={"is_superuser": True}))
+
+    by_type = {w["worker_type"]: w for w in body["workers"]}
+    check("已授权副本标注 licensed=True", by_type["alpha"]["licensed"] is True)
+    check("未授权副本标注 licensed=False", by_type["beta"]["licensed"] is False)
+    check("响应标明门禁启用", body["license_enforced"] is True)
+    check("响应汇总拒绝类型", body["denied_plugin_types"] == ["beta"],
+          str(body["denied_plugin_types"]))
+
+
+# ---------------------------------------------------------------------------
+# 4. 有效期与开关
+# ---------------------------------------------------------------------------
+def test_expired_license_denies_all():
+    """过期授权 -> 全部拒绝。"""
+    tmp = Path(tempfile.mkdtemp())
+    pub, priv = _make_keypair()
+    _use_license(pub, _issue(tmp, priv, plugins=["alpha", "beta"], expires_days=-1))
+
+    allowed = _discover_plugins()
+    check("过期授权不放行任何插件", allowed == [], str(allowed))
+    check("拒绝原因为过期",
+          all("过期" in m for m in denied_plugin_types().values()),
+          str(denied_plugin_types()))
+
+
+def test_disabled_mode_allows_everything():
+    """未启用校验时（非打包态的默认行为）全部放行。"""
+    tmp = Path(tempfile.mkdtemp())
+    pub, priv = _make_keypair()
+    _use_license(pub, _issue(tmp, priv, plugins=[]))
+    os.environ.pop("CHONGMING_LICENSE_MODE", None)
+    get_config()["license"]["enabled"] = False
+    licensing.reset_manager()
+    try:
+        allowed = sorted(p["type"] for p in _discover_plugins())
+        check("未启用校验时全部放行", allowed == ["alpha", "beta"], str(allowed))
+        check("未启用校验时无拒绝记录", denied_plugin_types() == {},
+              str(denied_plugin_types()))
+    finally:
+        os.environ["CHONGMING_LICENSE_MODE"] = "enforce"
+        get_config()["license"]["enabled"] = True
+        licensing.reset_manager()
+
+
+# ---------------------------------------------------------------------------
+# 5. 授权管理接口（直调端点协程，不需要 TestClient）
+# ---------------------------------------------------------------------------
+def test_license_endpoints_guard_and_upload():
+    """权限门禁、机器码/申请码、以及"坏文件不覆盖现有授权"。"""
+    from chongming.app.routes.licensing import (
+        license_info,
+        license_upload,
+        require_superuser,
+    )
+    from fastapi import HTTPException, UploadFile
+
+    tmp = Path(tempfile.mkdtemp())
+    pub, priv = _make_keypair()
+    lic_path = _issue(tmp, priv, plugins=["alpha"])
+    _use_license(pub, lic_path)
+
+    # 权限门禁：非超管 403，超管放行
+    try:
+        require_superuser({"is_superuser": False})
+        check("非超管访问被拒", False, "未抛 403")
+    except HTTPException as exc:
+        check("非超管访问被拒", exc.status_code == 403, f"status={exc.status_code}")
+    check("超管访问放行", require_superuser({"is_superuser": True})["is_superuser"] is True)
+
+    # 授权信息：机器码 / 申请码 / 插件全景 / 来源明细
+    info = asyncio.run(license_info(_request_stub(), _admin={"is_superuser": True}))
+    check("info 含机器码", info["status"]["machine_code"] != "")
+    check("info 含申请码", info["request_code"].startswith("CM1-"),
+          info["request_code"][:16])
+    check("info：未授权插件在 denied 中", "beta" in info["plugins"]["denied"],
+          str(list(info["plugins"]["denied"])))
+    check("info：已授权插件在 licensed 中", info["plugins"]["licensed"] == ["alpha"],
+          str(info["plugins"]["licensed"]))
+    check("info：来源明细含 weak 标注",
+          any(s["weak"] for s in info["sources_detail"]),
+          str([(s["name"], s["weak"]) for s in info["sources_detail"]]))
+    check("info：标明判定来源", info["enforce_source"] in
+          ("build", "packaging", "env", "config"), info["enforce_source"])
+
+    # 签名错误的文件：必须拒绝，且不得覆盖现有授权
+    tampered = {
+        "v": 1, "lic_id": "L-TAMPER", "customer": "x",
+        "machine_code": "AAAA-BBBB", "sources": {}, "plugins": ["alpha", "beta"],
+        "not_before": None, "expires_at": None, "issued_at": None, "note": "",
+    }
+    bad = UploadFile(
+        filename="bad.lic",
+        file=io.BytesIO(json.dumps(
+            {"payload": tampered, "alg": "ed25519", "kid": "test", "sig": "AAAA"}
+        ).encode()),
+    )
+    try:
+        asyncio.run(license_upload(_request_stub(), file=bad, _admin={"is_superuser": True}))
+        check("签名错误的文件被拒绝", False, "未抛 422")
+    except HTTPException as exc:
+        check("签名错误的文件被拒绝", exc.status_code == 422, f"status={exc.status_code}")
+    check("被拒文件未覆盖现有授权",
+          json.loads(Path(lic_path).read_text(encoding="utf-8"))["payload"]["lic_id"]
+          == "L-TEST-0001")
+
+    # 非 JSON 文件
+    not_json = UploadFile(filename="bad.lic", file=io.BytesIO(b"not json at all"))
+    try:
+        asyncio.run(license_upload(_request_stub(), file=not_json, _admin={"is_superuser": True}))
+        check("非 JSON 文件被拒绝", False, "未抛 422")
+    except HTTPException as exc:
+        check("非 JSON 文件被拒绝", exc.status_code == 422, f"status={exc.status_code}")
+
+    # 合法文件：安装成功，并提示需重启加载新授权的插件
+    src = collect_fingerprints()
+    payload2 = {
+        "v": 1, "lic_id": "L-TEST-0002", "customer": "测试客户",
+        "machine_code": machine_code(src), "sources": src,
+        "plugins": ["alpha", "beta"], "not_before": None,
+        "expires_at": (date.today() + timedelta(days=365)).isoformat(),
+        "issued_at": "2026-09-20T00:00:00Z", "note": "",
+    }
+    good = UploadFile(
+        filename="good.lic",
+        file=io.BytesIO(json.dumps(
+            {"payload": payload2, "alg": "ed25519", "kid": "test",
+             "sig": sign_payload(payload2, priv)}
+        ).encode()),
+    )
+    # 模拟运行中的网关：alpha 已在启动期加载（inline），beta 此前未授权未加载
+    running = _request_stub()
+    running.app.state.inline_types = ["alpha"]
+    body = asyncio.run(license_upload(running, file=good, _admin={"is_superuser": True}))
+    check("合法文件导入成功", body["success"] is True)
+    check("导入后授权插件更新",
+          body["licensed_plugins"] == ["alpha", "beta"], str(body["licensed_plugins"]))
+    check("导入后提示需重启加载",
+          body["restart_required"] and body["pending_plugins"] == ["beta"],
+          str(body["pending_plugins"]))
+
+
+# ---------------------------------------------------------------------------
+# 6. 构建期烘焙开关（"打包可选、授权版不可关"）
+# ---------------------------------------------------------------------------
+def test_build_baked_licensed_artifact():
+    """授权版产物：烘焙 enforce=True + 公钥 → 强制校验，现场改配置关不掉、换不了公钥。"""
+    tmp = Path(tempfile.mkdtemp())
+    baked_pub, baked_priv = _make_keypair()        # 产物内置公钥（厂商签发方）
+    other_pub, _other_priv = _make_keypair()       # 现场试图替换成的"自己的公钥"
+    lic = _issue(tmp, baked_priv, plugins=["alpha"])
+
+    with _baked_build_flags(True, baked_pub), _packaged(True):
+        _use_license(other_pub, lic)               # 配置里被换成别的公钥
+        get_config()["license"]["enabled"] = False  # 并试图关掉校验
+        os.environ[licensing.ENV_MODE] = "off"
+        try:
+            check("授权版产物强制校验", licensing.should_enforce() is True)
+            check("判定来源为构建期", licensing.enforce_source() == "build")
+            allowed = sorted(p["type"] for p in _discover_plugins())
+            check("授权版产物按内置公钥判定（换 config 公钥/自签均无效）",
+                  allowed == ["alpha"], str(allowed))
+            check("未授权插件仍被拒", "beta" in denied_plugin_types())
+        finally:
+            get_config()["license"]["enabled"] = True
+            os.environ[licensing.ENV_MODE] = "enforce"
+            licensing.reset_manager()
+
+
+def test_build_baked_unlicensed_artifact():
+    """无授权版产物：烘焙 enforce=False → 不校验（打包不配 license 也是合法选项）。"""
+    with _baked_build_flags(False), _packaged(True):
+        get_config()["license"]["enabled"] = True  # 配置试图开启也不生效（没有可校验的公钥）
+        try:
+            check("非授权版产物不校验", licensing.should_enforce() is False)
+            check("判定来源为构建期", licensing.enforce_source() == "build")
+            allowed = sorted(p["type"] for p in _discover_plugins())
+            check("非授权版产物全部插件照常加载", allowed == ["alpha", "beta"], str(allowed))
+            check("非授权版产物无拒绝记录", denied_plugin_types() == {},
+                  str(denied_plugin_types()))
+        finally:
+            get_config()["license"]["enabled"] = True
+
+
+def test_unbaked_packaged_keeps_enforcing():
+    """未烘焙（旧打包脚本产物）：沿用"打包即强制"的既有语义。"""
+    with _packaged(True):
+        check("未烘焙打包产物仍强制校验", licensing.should_enforce() is True)
+        check("判定来源为打包形态", licensing.enforce_source() == "packaging")
+
+
+def test_baked_flags_ignored_outside_packaging():
+    """烘焙标记只对产物生效：源码运行（含可编辑安装）不受上次打包结果影响。"""
+    pub, priv = _make_keypair()
+    tmp = Path(tempfile.mkdtemp())
+    lic = _issue(tmp, priv, plugins=["alpha"])
+
+    with _baked_build_flags(True, pub), _packaged(False):
+        _use_license(pub, lic)                     # 源码运行仍由配置/环境变量决定
+        check("源码运行不受烘焙标记影响（判定来源非 build）",
+              licensing.enforce_source() != "build", licensing.enforce_source())
+        os.environ[licensing.ENV_MODE] = "off"
+        try:
+            check("源码运行按环境变量/配置放行", licensing.should_enforce() is False)
+        finally:
+            os.environ[licensing.ENV_MODE] = "enforce"
+            licensing.reset_manager()
+
+
+def test_write_build_flags_source():
+    """write_build_flags 生成的模块源码可执行且值正确；非法公钥被拒绝。"""
+    from chongming.app.core.licensing import write_build_flags
+
+    pub, _priv = _make_keypair()
+    target = Path(tempfile.mkdtemp()) / "pkg" / "_build_flags.py"
+    path = write_build_flags(enforce=True, public_key=pub, target=target)
+    ns: dict = {}
+    exec(compile(path.read_text(encoding="utf-8"), str(path), "exec"), ns)
+    check("生成文件落盘于指定位置", path == target)
+    check("烘焙 enforce=True", ns["LICENSE_ENFORCE"] is True)
+    check("烘焙公钥", ns["LICENSE_PUBLIC_KEY"] == pub)
+    check("记录公钥指纹", bool(ns["PUBLIC_KEY_FINGERPRINT"]))
+
+    path2 = write_build_flags(enforce=False, target=target)
+    ns2: dict = {}
+    exec(compile(path2.read_text(encoding="utf-8"), str(path2), "exec"), ns2)
+    check("无授权版烘焙 enforce=False 且无公钥",
+          ns2["LICENSE_ENFORCE"] is False and ns2["LICENSE_PUBLIC_KEY"] == "")
+
+    try:
+        write_build_flags(enforce=True, public_key="not-a-key", target=target)
+        check("非法公钥被拒绝", False, "未抛异常")
+    except ValueError as exc:
+        check("非法公钥被拒绝", "公钥无效" in str(exc), str(exc)[:60])
+
+
+def main():
+    tests = [v for k, v in sorted(globals().items()) if k.startswith("test_")]
+    for fn in tests:
+        print(f"\n--- {fn.__name__} ---")
+        fn()
+    print("\n" + "=" * 56)
+    if failures:
+        print(f"❌ {len(failures)} 项未通过: {failures}")
+        return 1
+    print("✅ 全部通过")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())