chakcy
93fed12d14
三处收口，未授权插件不加载、不暴露端点、不可管理：
- 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 项：门禁/契约存活/描述符拦截/管理接口/上传安全/烘焙三态，自造夹具直接运行）。
626 lines
25 KiB
Python
626 lines
25 KiB
Python
"""机器码授权门禁测试（网关侧）。
|
||
|
||
核心验证目标 —— 未授权插件被拒绝时，**网关必须仍能正常启动**：
|
||
|
||
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())
|