chongming/chongming-core
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

modify: 授权

Browse Source
This commit is contained in:
chakcy
2026-09-21 14:03:47 +08:00
parent 66e82c02e7
commit 8a3dc8036c
19 changed files with 2826 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files
test/test_license_build.py
+224
View File
@@ -0,0 +1,224 @@
"""构建期授权守护测试：公钥注入 + 私钥泄漏守卫。
对齐 `test_build_plugin.py` 对"插件包不得包含 secret_key"的同类断言 ——
**签发私钥绝不能进入部署产物**：它是授权的唯一信任根，一旦随包发布，
任何人都能签发授权，机器码绑定形同虚设。
运行方式：`python test/test_license_build.py`，或 `uv run pytest test/test_license_build.py`。
"""
import os
import sys
import tempfile
import tomllib
from pathlib import Path
ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(ROOT))
sys.path.insert(0, str(ROOT / "utils" / "cli" / "src"))
from chongming_cli.build import ( # noqa: E402
ALLOW_NO_KEY_ENV,
LICENSE_PUBKEY_ENV,
_resolve_license_public_key,
_set_license_pubkey,
_stage_license_guard,
)
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)
def test_pubkey_injection_is_section_scoped():
"""公钥只写进 [license] 段，注释与其它段不受影响。"""
src = (ROOT / "public" / "config.toml").read_text(encoding="utf-8")
out = _set_license_pubkey(src, "AAAABBBB=")
body = tomllib.loads(out)
check("注入后 TOML 可解析", body["license"]["public_key"] == "AAAABBBB=",
str(body["license"].get("public_key")))
# 只改了 [license] 段内那一行
lic_start = out.index("[license]")
prod_start = out.index("[production]")
check("公钥位于 [license] 段内",
'public_key = "AAAABBBB="' in out[lic_start:prod_start])
check("[production] 段未被误改", "public_key" not in out[prod_start:])
check("注释保留", "# 机器码授权" in out)
check("其它段结构完整", "[production.module_system]" in out)
check("[license] 段仍保留 path/min_match",
body["license"]["path"] == "license.lic" and body["license"]["min_match"] == 3)
diff = [i for i, (a, b) in enumerate(zip(src.splitlines(), out.splitlines())) if a != b]
check("仅一行发生变化", len(diff) == 1, f"变化行号={diff}")
def test_pubkey_resolution_order():
"""公钥解析优先级：环境变量 > 文件环境变量 > 默认文件 license_public.key。
必须在**空的临时目录**里跑：默认回退路径 ``license_public.key`` 是相对 CWD 的，
而厂商在项目根放过这个文件是正常用法（也正是构建取公钥的正规途径），
测试不能依赖它"不存在"。
"""
tmp = Path(tempfile.mkdtemp())
key_file = tmp / "pk.txt"
key_file.write_text("FROMFILE", encoding="utf-8")
old_cwd = os.getcwd()
old_env = dict(os.environ)
try:
os.chdir(tmp) # 空目录：既无 license_public.key 也无其它干扰
for var in (LICENSE_PUBKEY_ENV, "CHONGMING_LICENSE_PUBKEY_FILE"):
os.environ.pop(var, None)
check("三者都无时返回空", _resolve_license_public_key() == "")
# 默认文件回退（构建按此取公钥，之前漏测）
(tmp / "license_public.key").write_text("FROMDEFAULT", encoding="utf-8")
check("回退到默认文件 license_public.key",
_resolve_license_public_key() == "FROMDEFAULT")
os.environ["CHONGMING_LICENSE_PUBKEY_FILE"] = str(key_file)
check("文件环境变量优先于默认文件",
_resolve_license_public_key() == "FROMFILE")
os.environ[LICENSE_PUBKEY_ENV] = "FROMENV"
check("环境变量优先于文件", _resolve_license_public_key() == "FROMENV")
finally:
os.chdir(old_cwd)
os.environ.clear()
os.environ.update(old_env)
def test_guard_detects_private_key():
"""守护能检出私钥（按内容，与扩展名无关，含跨块边界）。"""
tmp = Path(tempfile.mkdtemp())
build = tmp / "build"
(build / "sub").mkdir(parents=True)
old_cwd = os.getcwd()
try:
os.chdir(tmp)
(build / "ok.txt").write_text("正常内容\n", encoding="utf-8")
check("干净产物通过", _stage_license_guard() == "未发现私钥泄漏")
# 证书类 .pem 是合法的构建内容，不应被误判
(build / "cert.pem").write_text(
"-----BEGIN CERTIFICATE-----\nMIIB\n-----END CERTIFICATE-----\n", encoding="utf-8"
)
check("合法证书 .pem 不误判", _stage_license_guard() == "未发现私钥泄漏")
# 私钥改名藏进普通文件 -> 按内容仍能检出
leak = build / "sub" / "innocent.txt"
leak.write_text("-----BEGIN PRIVATE KEY-----\nMIIEvQ\n", encoding="utf-8")
try:
_stage_license_guard()
check("按内容检出改名的私钥", False, "未检出")
except RuntimeError as exc:
check("按内容检出改名的私钥", "innocent.txt" in str(exc))
leak.unlink()
# 跨 1MB 块边界的私钥
(build / "big.bin").write_bytes(
b"x" * (1 << 20) + b"-----BEGIN RSA PRIVATE KEY-----" + b"y" * (1 << 20)
)
try:
_stage_license_guard()
check("检出跨块边界的私钥", False, "未检出")
except RuntimeError as exc:
check("检出跨块边界的私钥", "big.bin" in str(exc))
(build / "big.bin").unlink()
# 文件名叫 license_private* 的一律拦下
(build / "license_private.pem").write_text("(truncated)", encoding="utf-8")
try:
_stage_license_guard()
check("按文件名拦下授权私钥文件", False, "未检出")
except RuntimeError as exc:
check("按文件名拦下授权私钥文件", "license_private" in str(exc))
finally:
os.chdir(old_cwd)
def test_build_fails_without_pubkey():
"""打包时缺少公钥 -> 构建失败（避免产出拒绝所有插件的砖头包）。"""
import chongming_cli.build as build_mod
tmp = Path(tempfile.mkdtemp())
old_cwd = os.getcwd()
old_env = dict(os.environ)
try:
# 复刻 build/ 最小结构：_stage_prepare_build_dir 需要 public/ 与 config.toml
(tmp / "public" / "utils").mkdir(parents=True)
(tmp / "public" / "main.py").write_text("", encoding="utf-8")
(tmp / "public" / "server.py").write_text("", encoding="utf-8")
(tmp / "public" / "config.toml").write_text(
(ROOT / "public" / "config.toml").read_text(encoding="utf-8"), encoding="utf-8"
)
os.chdir(tmp)
for var in (LICENSE_PUBKEY_ENV, "CHONGMING_LICENSE_PUBKEY_FILE", ALLOW_NO_KEY_ENV):
os.environ.pop(var, None)
try:
build_mod._stage_prepare_build_dir({}, "pyinstaller") # type: ignore[arg-type]
check("缺少公钥时构建失败", False, "未抛异常")
except RuntimeError as exc:
check("缺少公钥时构建失败", LICENSE_PUBKEY_ENV in str(exc))
# 显式放行时才允许无公钥构建
os.environ[ALLOW_NO_KEY_ENV] = "1"
build_mod._stage_prepare_build_dir({}, "pyinstaller") # type: ignore[arg-type]
built = tomllib.loads((tmp / "build" / "config.toml").read_text(encoding="utf-8"))
check("显式放行时构建成功且公钥为空", built["license"]["public_key"] == "")
# 提供公钥后应注入到产物配置
os.environ[LICENSE_PUBKEY_ENV] = "INJECTEDKEY="
build_mod._stage_prepare_build_dir({}, "pyinstaller") # type: ignore[arg-type]
built = tomllib.loads((tmp / "build" / "config.toml").read_text(encoding="utf-8"))
check("公钥被注入产物配置", built["license"]["public_key"] == "INJECTEDKEY=")
check("产物配置仍含 min_match", built["license"]["min_match"] == 3)
finally:
os.chdir(old_cwd)
os.environ.clear()
os.environ.update(old_env)
def test_mbank_signing_is_not_a_trust_root():
""".mbank 验签用随包发布的对称密钥 —— 不能作为授权信任根。
本测试固化这一认知：只要 sqlite_vfs 的 sign 仍是 HMAC，就说明它只能算防篡改，
授权必须依赖非对称签名（chongming_license 的 Ed25519）。
"""
import inspect
from sqlite_vfs import core as vfs_core
src = inspect.getsource(vfs_core.SQLiteVFS.sign)
check("确认 .mbank 签名为 HMAC（对称）", "hmac.new" in src,
"若已改为非对称，可更新本测试与文档说明")
check("确认验证端同样是对称 HMAC",
"hmac.new" in inspect.getsource(vfs_core.SQLiteVFS.verify_signature))
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())
test/test_license_gate.py
+377
View File
@@ -0,0 +1,377 @@
"""机器码授权门禁测试（网关侧）。
核心验证目标 —— 未授权插件被拒绝时，**网关必须仍能正常启动**：
1. 门禁收口在 `_discover_plugins()`：未授权插件不出现在返回列表中，
且 `denied_plugin_types()` 能给出拒绝原因；
2. **契约校验不硬失败**：这是最容易踩的坑 —— `validate_contracts()` 会把
"TOML 路由没有对应 @rpc_method 注册"视为硬性错误并终止启动。若把"不加载
未授权插件"实现成"跳过 import 但保留在插件列表里"，该插件的 TOML 路由就会
变成孤儿 → 设备起不来。本测试确保收口位置正确。
3. 路由描述符门禁：未授权类型的 external 路由描述符被忽略（防"改 load_mode
成 external 绕过门禁"）；
4. 管理接口门禁：未授权插件的扩缩容/重启/配置接口一律 403；
5. 未启用校验时（开发/测试）全部放行。
运行方式：`python test/test_license_gate.py`，或 `uv run pytest test/test_license_gate.py`。
"""
import asyncio
import base64
import json
import os
import sys
import tempfile
from datetime import date, timedelta
from pathlib import Path
ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(ROOT))
# 必须在导入网关模块前设置：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
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 = 3) -> 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 test_gate_denies_unlicensed_plugins():
"""未授权插件被剔除，且原因可查。"""
tmp = Path(tempfile.mkdtemp())
pub, priv = _make_keypair()
_use_license(pub, _issue(tmp, priv, plugins=["ip_sound_column"]))
scanned = sorted(p["type"] for p in _scan_plugins())
allowed = sorted(p["type"] for p in _discover_plugins())
denied = denied_plugin_types()
check("盘扫到全部插件", scanned == ["aibox_sophon_adapter", "ip_sound_column"],
f"scanned={scanned}")
check("只放行已授权插件", allowed == ["ip_sound_column"], f"allowed={allowed}")
check("未授权插件给出拒绝原因",
"aibox_sophon_adapter" in denied
and "白名单" in denied["aibox_sophon_adapter"],
str(denied))
def test_contract_validation_survives_denial():
"""关键：有插件被拒时契约校验不得抛异常（否则设备起不来）。"""
tmp = Path(tempfile.mkdtemp())
pub, priv = _make_keypair()
_use_license(pub, _issue(tmp, priv, plugins=["ip_sound_column"]))
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))
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))
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=["ip_sound_column"]))
denied = licensing.denied_types({p["type"] for p in _scan_plugins()})
check("denied_types 含未授权插件", "aibox_sophon_adapter" 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": "aibox_sophon_adapter",
"subject": "aibox_sophon_adapter.alarm_report", "method": "POST",
"path": "/aibox_sophon_adapter/alarm", "params": {},
}
await registry._handle_sample(
"chongming/routes/aibox_sophon_adapter/alarm_report",
__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))
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=["ip_sound_column"]))
try:
_ensure_licensed("aibox_sophon_adapter")
check("未授权插件的管理操作被拒", False, "未抛 403")
except HTTPException as exc:
check("未授权插件的管理操作被拒", exc.status_code == 403,
f"status={exc.status_code}")
try:
_ensure_licensed("ip_sound_column")
check("已授权插件的管理操作放行", True)
except HTTPException as exc: # noqa: BLE001
check("已授权插件的管理操作放行", False, f"status={exc.status_code}")
def test_expired_license_denies_all():
"""过期授权 -> 全部拒绝。"""
tmp = Path(tempfile.mkdtemp())
pub, priv = _make_keypair()
_use_license(pub, _issue(tmp, priv, plugins=["ip_sound_column", "aibox_sophon_adapter"],
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=[]))
licensing.reset_manager()
os.environ.pop("CHONGMING_LICENSE_MODE", None)
get_config()["license"]["enabled"] = False
try:
allowed = sorted(p["type"] for p in _discover_plugins())
check("未启用校验时全部放行",
allowed == ["aibox_sophon_adapter", "ip_sound_column"], str(allowed))
check("未启用校验时无拒绝记录", denied_plugin_types() == {},
str(denied_plugin_types()))
finally:
os.environ["CHONGMING_LICENSE_MODE"] = "enforce"
get_config()["license"]["enabled"] = False
licensing.reset_manager()
def test_http_endpoints_and_upload_safety():
"""HTTP 层：门禁、信息接口、以及"坏文件不覆盖现有授权"。
这一步覆盖最容易在现场出事的行为 —— 误传一个坏 license 文件把设备搞成未授权。
"""
import base64 as _b64
import json as _json
import tempfile as _tf
from datetime import date as _date, timedelta as _td
import orjson
from cryptography.hazmat.primitives import serialization as _ser
from fastapi.testclient import TestClient
from chongming_jwt import JWT
from chongming_license import collect_fingerprints as _collect
from chongming.app import app
tmp = Path(_tf.mkdtemp())
priv = Ed25519PrivateKey.generate()
pub = _b64.b64encode(
priv.public_key().public_bytes(
encoding=_ser.Encoding.Raw, format=_ser.PublicFormat.Raw
)
).decode("ascii")
src = _collect()
payload = {
"v": 1, "lic_id": "L-HTTP-1", "customer": "HTTP 测试",
"machine_code": machine_code(src), "sources": src,
"plugins": ["ip_sound_column"], "not_before": None,
"expires_at": (_date.today() + _td(days=30)).isoformat(),
"issued_at": "2026-09-20T00:00:00Z", "note": "",
}
lic = tmp / "license.lic"
lic.write_text(_json.dumps({
"payload": payload, "alg": "ed25519", "kid": "test",
"sig": sign_payload(payload, priv),
}), encoding="utf-8")
_use_license(pub, str(lic))
sec = get_config()["security"]
def _token(superuser: bool) -> str:
return JWT(sec["secret_key"], sec["algorithm"]).encode(
orjson.dumps({"sub": "u1", "username": "admin",
"is_superuser": superuser}).decode(),
expires_in_seconds=3600,
)
client = TestClient(app)
auth = {"Authorization": f"Bearer {_token(True)}"}
r = client.get("/api/v1/license/info")
check("无 token 时拒绝访问", r.status_code == 401, f"status={r.status_code}")
r = client.get("/api/v1/license/info",
headers={"Authorization": f"Bearer {_token(False)}"})
check("非超管拒绝访问", r.status_code == 403, f"status={r.status_code}")
r = client.get("/api/v1/license/info", headers=auth)
check("超管可读取授权信息", r.status_code == 200, f"status={r.status_code}")
info = r.json()
check("/license/info 含机器码", info["status"]["machine_code"] != "",
info["status"]["machine_code"])
check("/license/info 含申请码", info["request_code"].startswith("CM1-"),
info["request_code"][:16])
check("未授权插件在 denied 中", "aibox_sophon_adapter" in info["plugins"]["denied"],
str(list(info["plugins"]["denied"])))
check("来源明细含 weak 标注",
any(s["weak"] for s in info["sources_detail"]),
str([(s["name"], s["weak"]) for s in info["sources_detail"]]))
# 签名错误的文件：必须拒绝，且不得覆盖现有授权
tampered = dict(payload, plugins=["aibox_sophon_adapter"])
r = client.post(
"/api/v1/license/file", headers=auth,
files={"file": ("bad.lic", _json.dumps(
{"payload": tampered, "alg": "ed25519", "kid": "test", "sig": "AAAA"}
), "application/json")},
)
check("签名错误的文件被拒绝", r.status_code == 422, f"status={r.status_code}")
check("被拒文件未覆盖现有授权",
_json.loads(lic.read_text(encoding="utf-8"))["payload"]["lic_id"] == "L-HTTP-1")
# 非 JSON 文件
r = client.post("/api/v1/license/file", headers=auth,
files={"file": ("bad.lic", "not json at all", "application/json")})
check("非 JSON 文件被拒绝", r.status_code == 422, f"status={r.status_code}")
# 合法文件：安装成功，并提示需重启加载
payload2 = dict(payload, lic_id="L-HTTP-2",
plugins=["ip_sound_column", "aibox_sophon_adapter"])
r = client.post(
"/api/v1/license/file", headers=auth,
files={"file": ("good.lic", _json.dumps(
{"payload": payload2, "alg": "ed25519", "kid": "test",
"sig": sign_payload(payload2, priv)}
), "application/json")},
)
check("合法文件导入成功", r.status_code == 200, f"status={r.status_code}")
if r.status_code == 200:
body = r.json()
check("导入后授权插件更新",
body["licensed_plugins"] == ["aibox_sophon_adapter", "ip_sound_column"],
str(body["licensed_plugins"]))
check("导入后提示需重启加载",
body["restart_required"] and "aibox_sophon_adapter" in body["pending_plugins"],
str(body["pending_plugins"]))
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())
test/test_license_registry.py
+327
View File
@@ -0,0 +1,327 @@
"""已签发授权台账测试（厂商侧 Registry）。
重点验证：
1. 由 license 文档派生记录（唯一写入口，不手工拼记录）
2. 状态判定（永久/有效/即将到期/已过期/未生效）与剩余天数
3. 检索：客户 / 机器码 / 插件 / 关键字 / 状态
4. **续期复用台账里的来源指纹** —— 台账最大的价值：不必再向客户索取申请码
5. 重新导入不覆盖人工维护的备注
6. 导出 .lic（按客户分子目录）与 CSV（UTF-8 BOM，Excel 中文不乱码）
7. 结构版本不匹配时明确报错，而不是让查询诡异失败
运行：`python test/test_license_registry.py`，或 `uv run pytest test/test_license_registry.py`。
"""
import csv
import sqlite3
import sys
import tempfile
from datetime import date, timedelta
from pathlib import Path
ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(ROOT))
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey # noqa: E402
from chongming_license import ( # noqa: E402
Registry,
RegistryError,
build_license,
build_payload,
machine_code,
public_key_b64,
public_key_fingerprint,
record_from_license,
)
failures: list[str] = []
def check(name, cond, detail=""):
print(f"[{'PASS' if cond else 'FAIL'}] {name}" + (f" — {detail}" if detail else ""))
if not cond:
failures.append(name)
PRIV = Ed25519PrivateKey.generate()
PUB = public_key_b64(PRIV)
FP = public_key_fingerprint(PUB)
SRC = {"board_serial": "sn123456789", "emmc_cid": "1a2b3c4d", "mac": "aabbccddeeff"}
CODE = machine_code(SRC)
def make_doc(*, customer="某某科技", days=300, plugins=("aibox_sophon_adapter",),
note="", machine=None, sources=None, lic_id="", not_before=None):
payload = build_payload(
machine_code=machine or CODE,
plugins=list(plugins),
sources=SRC if sources is None else sources,
customer=customer,
note=note,
lic_id=lic_id,
not_before=not_before,
expires_at=(date.today() + timedelta(days=days)) if days is not None else None,
)
return build_license(payload, PRIV, kid="vendor-2026")
def test_record_derivation():
"""记录完全由 license 文档派生，查询列与原始文档不会不一致。"""
doc = make_doc(customer="派生测试", note="合同 X", plugins=["a", "b"])
rec = record_from_license(doc, public_key_fp=FP, source="test")
payload = doc["payload"]
check("lic_id 取自载荷", rec.lic_id == payload["lic_id"], rec.lic_id)
check("机器码大写归一", rec.machine_code == payload["machine_code"].upper())
check("插件列表解析", rec.plugins == ["a", "b"], str(rec.plugins))
check("来源指纹解析", rec.sources == SRC, str(sorted(rec.sources)))
check("kid 取自信封", rec.kid == "vendor-2026")
check("公钥指纹记录在案", rec.public_key_fp == FP)
check("可还原完整文档", rec.license_doc()["sig"] == doc["sig"])
# 日期只保留日期部分：否则 SQL 里的字符串比较会得出错误排序
doc2 = make_doc()
doc2["payload"]["expires_at"] = "2027-09-01T12:34:56Z"
check("日期裁剪到日粒度",
record_from_license(doc2).expires_at == "2027-09-01",
str(record_from_license(doc2).expires_at))
# 缺 lic_id 必须报错，而不是静默写一条垃圾记录
try:
record_from_license({"payload": {"machine_code": "X"}})
check("缺 lic_id 时报错", False, "未抛异常")
except RegistryError:
check("缺 lic_id 时报错", True)
def test_status_and_days_left():
"""五种状态的判定（含边界：到期当天仍有效）。"""
tmp = Path(tempfile.mkdtemp())
with Registry(tmp / "r.db") as reg:
reg.save(make_doc(lic_id="L-FOREVER", days=None), public_key_fp=FP)
reg.save(make_doc(lic_id="L-ACTIVE", days=300), public_key_fp=FP)
reg.save(make_doc(lic_id="L-EXPIRING", days=10), public_key_fp=FP)
reg.save(make_doc(lic_id="L-EXPIRED", days=-1), public_key_fp=FP)
reg.save(make_doc(lic_id="L-TODAY", days=0), public_key_fp=FP)
reg.save(make_doc(lic_id="L-FUTURE", days=100,
not_before=(date.today() + timedelta(days=30)).isoformat()),
public_key_fp=FP)
got = {r.lic_id: r.status() for r in reg.query()}
check("无到期日 → forever", got["L-FOREVER"] == "forever", got["L-FOREVER"])
check("远期 → active", got["L-ACTIVE"] == "active", got["L-ACTIVE"])
check("30 天内 → expiring", got["L-EXPIRING"] == "expiring", got["L-EXPIRING"])
check("已过 → expired", got["L-EXPIRED"] == "expired", got["L-EXPIRED"])
check("到期当天仍有效（含当天）", got["L-TODAY"] == "expiring", got["L-TODAY"])
check("未到生效日 → not_yet_valid",
got["L-FUTURE"] == "not_yet_valid", got["L-FUTURE"])
check("剩余天数为负表示已过期",
reg.get("L-EXPIRED").days_left() < 0,
str(reg.get("L-EXPIRED").days_left()))
check("永久授权剩余天数为 None",
reg.get("L-FOREVER").days_left() is None)
st = reg.stats()
check("概览计数一致",
st == {"total": 6, "expired": 1, "expiring": 2, "active": 1,
"forever": 1, "not_yet_valid": 1}, str(st))
def test_query_filters():
"""检索：客户 / 机器码 / 插件 / 关键字 / 状态，以及排序。"""
tmp = Path(tempfile.mkdtemp())
with Registry(tmp / "r.db") as reg:
reg.save(make_doc(lic_id="L-A", customer="阿尔法科技", days=300,
plugins=["doubao"], note="HT-001"), public_key_fp=FP)
reg.save(make_doc(lic_id="L-B", customer="贝塔公司", days=10,
plugins=["ip_sound_column"], note="HT-002"), public_key_fp=FP)
reg.save(make_doc(lic_id="L-C", customer="贝塔公司", days=200,
plugins=["doubao_extended"], note="HT-003"), public_key_fp=FP)
other = "AAAA-BBBB-CCCC-DDDD-EEEE-FFFF"
reg.save(make_doc(lic_id="L-D", customer="伽马", days=300, machine=other,
sources={"mac": "aabbccddeeff"}), public_key_fp=FP)
check("按客户模糊匹配", {r.lic_id for r in reg.query(customer="贝塔")} == {"L-B", "L-C"})
# L-C 装的是 doubao_extended，查 doubao 时不应被命中（前缀同名不算同一插件）
check("按插件匹配不误伤前缀同名",
{r.lic_id for r in reg.query(plugin="doubao")} == {"L-A"},
str([r.lic_id for r in reg.query(plugin="doubao")]))
check("按插件匹配到前缀同名者本身",
{r.lic_id for r in reg.query(plugin="doubao_extended")} == {"L-C"})
check("按机器码匹配", {r.lic_id for r in reg.query(machine_code=CODE[:16])}
== {"L-A", "L-B", "L-C"})
check("按备注关键字匹配", {r.lic_id for r in reg.query(keyword="HT-002")} == {"L-B"})
check("大小写不敏感（机器码）",
{r.lic_id for r in reg.query(machine_code=other.lower())} == {"L-D"})
order = [r.lic_id for r in reg.query()]
check("按到期日升序（急的在前）", order[0] == "L-B", str(order))
# 某台机器的全部历史授权
check("按机器码取全部历史",
len(reg.find_by_machine_code(CODE)) == 3,
str(len(reg.find_by_machine_code(CODE))))
check("limit 生效", len(reg.query(limit=2)) == 2)
def test_renew_reuses_sources_without_request_code():
"""核心能力：续期复用台账里的指纹，无需再向客户索取申请码。"""
tmp = Path(tempfile.mkdtemp())
with Registry(tmp / "r.db") as reg:
old = reg.save(make_doc(lic_id="L-OLD", customer="续期客户", days=5,
plugins=["aibox_sophon_adapter", "ip_sound_column"],
note="合同 HT-100"), public_key_fp=FP)
check("原授权即将到期", old.status() == "expiring", old.status())
# 模拟现场：客户已联系不上，只有台账记录可用
renewed_doc = make_doc(lic_id="L-NEW", customer=old.customer, days=365,
plugins=old.plugins, note=old.note,
machine=old.machine_code, sources=old.sources)
new = reg.save(renewed_doc, public_key_fp=FP, source="renew")
check("续期成功", new.status() == "active", new.status())
check("指纹与原件一致（未索取新申请码）", new.sources == old.sources)
check("机器码不变", new.machine_code == old.machine_code)
check("新旧授权都在台账（可追溯）",
{r.lic_id for r in reg.find_by_machine_code(old.machine_code)}
== {"L-OLD", "L-NEW"})
def test_note_preserved_on_reimport():
"""重新导入同一份 license 不应覆盖人工维护的备注。"""
tmp = Path(tempfile.mkdtemp())
with Registry(tmp / "r.db") as reg:
doc = make_doc(lic_id="L-N", note="原始备注")
reg.save(doc, public_key_fp=FP)
check("备注已入库", reg.get("L-N").note == "原始备注")
reg.update_note("L-N", "人工补充：已电话确认续期")
check("备注可更新", reg.get("L-N").note == "人工补充：已电话确认续期")
created = reg.get("L-N").created_at
reg.save(doc, public_key_fp=FP) # 重新导入（载荷里还是旧备注）
after = reg.get("L-N")
check("重新导入不覆盖人工备注",
after.note == "人工补充：已电话确认续期", after.note)
check("创建时间保持不变", after.created_at == created, after.created_at)
check("更新时间被推进", after.updated_at >= created)
def test_export_and_csv():
"""导出 .lic（可按客户分目录）与 CSV（UTF-8 BOM，Excel 不乱码）。"""
tmp = Path(tempfile.mkdtemp())
with Registry(tmp / "r.db") as reg:
reg.save(make_doc(lic_id="L-X1", customer="甲方/科技", days=100), public_key_fp=FP)
reg.save(make_doc(lic_id="L-X2", customer="乙方", days=-3), public_key_fp=FP)
files = reg.export_licenses(tmp / "arc", subdir_by_customer=True)
check("导出数量正确", len(files) == 2, str([f.name for f in files]))
check("客户名含非法字符时安全替换",
any("甲方" in str(f) for f in files), str([str(f.parent.name) for f in files]))
# 导出的文件应能被校验端直接读取（验证导出内容不是"看着像"而已）
# 注意顺序按到期日升序：files[0] 是已过期的 L-X2，files[1] 是有效的 L-X1
from chongming_license import read_license
valid_file = next(f for f in files if "L-X1" in f.name)
expired_file = next(f for f in files if "L-X2" in f.name)
st = read_license(valid_file, PUB, sources=SRC)
check("导出的有效授权仍可验签通过", st.ok, st.message)
st2 = read_license(expired_file, PUB, sources=SRC)
check("导出的已过期授权被判定为过期（到期日完整保留）",
not st2.ok and st2.code == "expired", f"{st2.code} / {st2.message}")
check("导出内容可原样还原",
valid_file.read_text(encoding="utf-8").lstrip().startswith("{"))
csv_path = reg.export_csv(tmp / "台账.csv")
with csv_path.open(encoding="utf-8-sig", newline="") as f:
rows = list(csv.reader(f))
check("CSV 含表头 + 两条数据", len(rows) == 3, str(len(rows)))
check("CSV 表头含状态与到期日",
"状态" in rows[0] and "到期日期" in rows[0], str(rows[0]))
check("CSV 记录了状态文案",
any("已过期" in r for r in rows[1:]), str(rows[1:]))
check("CSV 以 UTF-8 BOM 写入（Excel 中文不乱码）",
csv_path.read_bytes()[:3] == b"\xef\xbb\xbf")
def test_schema_version_mismatch_is_explicit():
"""结构版本不匹配时明确报错，而不是让后续查询诡异失败。"""
tmp = Path(tempfile.mkdtemp())
path = tmp / "r.db"
with Registry(path) as reg:
reg.save(make_doc(lic_id="L-V"), public_key_fp=FP)
# 手工把版本改成未来值，模拟"被更新版本的工具写过"
conn = sqlite3.connect(str(path))
conn.execute("PRAGMA user_version = 99")
conn.commit()
conn.close()
try:
Registry(path)
check("版本不匹配时拒绝打开", False, "未抛异常")
except RegistryError as exc:
check("版本不匹配时拒绝打开", "99" in str(exc), str(exc).splitlines()[0])
# 已存在但 user_version=0 的库（旧文件）应被平滑接管
path2 = tmp / "old.db"
conn = sqlite3.connect(str(path2))
conn.executescript(
"CREATE TABLE issued_licenses (lic_id TEXT PRIMARY KEY, machine_code TEXT, "
"customer TEXT, note TEXT, plugins_json TEXT, sources_json TEXT, not_before TEXT, "
"expires_at TEXT, issued_at TEXT, kid TEXT, public_key_fp TEXT, license_json TEXT, "
"source TEXT, created_at TEXT, updated_at TEXT);"
)
conn.commit()
conn.close()
with Registry(path2) as reg2:
check("user_version=0 的旧库被平滑接管", reg2.stats()["total"] == 0)
def test_concurrent_open():
"""两个实例同时打开同一库（多个 GUI 窗口 / CLI 并行）不应互相报错。"""
tmp = Path(tempfile.mkdtemp())
path = tmp / "r.db"
a = Registry(path)
b = Registry(path)
try:
a.save(make_doc(lic_id="L-C1", customer="并发A"), public_key_fp=FP)
b.save(make_doc(lic_id="L-C2", customer="并发B"), public_key_fp=FP)
check("两实例各自写入后互相可见",
{r.lic_id for r in a.query()} == {"L-C1", "L-C2"},
str([r.lic_id for r in a.query()]))
check("B 能看到 A 的记录", b.get("L-C1") is not None)
finally:
a.close()
b.close()
def test_delete():
tmp = Path(tempfile.mkdtemp())
with Registry(tmp / "r.db") as reg:
reg.save(make_doc(lic_id="L-DEL"), public_key_fp=FP)
check("删除成功", reg.delete("L-DEL") is True)
check("删除后查不到", reg.get("L-DEL") is None)
check("删除不存在的返回 False", reg.delete("L-DEL") is False)
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())
test/test_license_verify.py
+358
View File
@@ -0,0 +1,358 @@
"""chongming-license 库层测试：指纹、机器码、验签、容错匹配、时钟防线。
运行方式：`python test/test_license_verify.py`，或 `uv run pytest test/test_license_verify.py`。
"""
import base64
import json
import sys
import tempfile
from datetime import date, datetime, timedelta, timezone
from pathlib import Path
ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(ROOT))
from cryptography.hazmat.primitives import serialization # noqa: E402
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey # noqa: E402
from chongming_license import ( # noqa: E402
REASON_CLOCK_ROLLBACK,
REASON_EXPIRED,
REASON_MACHINE_MISMATCH,
REASON_NOT_YET_VALID,
REASON_OK,
REASON_PLUGIN_NOT_LICENSED,
REASON_SIGNATURE_INVALID,
ClockGuard,
LicenseManager,
MemoryClockStore,
RequestCodeError,
decode_request_code,
encode_request_code,
machine_code,
match_fingerprints,
normalize,
read_license,
sign_payload,
)
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)
# ---------------------------------------------------------------------------
# 夹具
# ---------------------------------------------------------------------------
# 模拟真实盒子的指纹：4 个可信来源 + 1 个 weak（同镜像刷机会重复的 machine-id）
HW = {
"board_serial": "sn123456789",
"emmc_cid": "1a2b3c4d",
"cpu_serial": "0000deadbeef",
"mac": "aabbccddeeff",
"machine_id": "sharedbyallunits",
}
def _keypair():
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 _write_license(tmp: Path, priv, sources=None, plugins=None, **over) -> str:
src = HW if sources is None else sources
payload = {
"v": 1,
"lic_id": "L-TEST",
"customer": "测试客户",
# sources 为空表示"只签机器码"的授权（无逐来源容错），此时机器码必须由调用方给定
"machine_code": over.pop("machine_code") if not src else machine_code(src),
"sources": src,
"plugins": plugins if plugins is not None else ["aibox_sophon_adapter"],
"not_before": None,
"expires_at": (date.today() + timedelta(days=365)).isoformat(),
"issued_at": "2026-09-20T00:00:00Z",
"note": "",
}
payload.update(over)
doc = {"payload": payload, "alg": "ed25519", "kid": "test",
"sig": sign_payload(payload, priv)}
path = tmp / "license.lic"
path.write_text(json.dumps(doc), encoding="utf-8")
return str(path)
# ---------------------------------------------------------------------------
# 指纹与机器码
# ---------------------------------------------------------------------------
def test_normalize_rejects_placeholders():
"""占位符/全 0/全 f/过短一律视为无效。"""
for bad in ["", " ", "unknown", "None", "0000000000000000", "ffffffffffff",
"ab", "00000000-0000-0000-0000-000000000000"]:
check(f"normalize 拒绝 {bad!r}", normalize(bad) == "", f"got {normalize(bad)!r}")
check("normalize 保留合法值", normalize("SN-1234-ABCD") == "sn1234abcd",
normalize("SN-1234-ABCD"))
check("normalize 分隔符差异归一", normalize("AA:BB:CC:DD:EE:FF") == normalize("aabbccddeeff"))
def test_machine_code_deterministic_and_weak_excluded():
"""机器码确定，且不受 weak 来源影响。"""
a = machine_code(HW)
b = machine_code(dict(reversed(list(HW.items()))))
check("机器码与来源顺序无关", a == b)
check("机器码分组格式", len(a.split("-")) == 6 and len(a.replace("-", "")) == 24, a)
other = dict(HW, machine_id="anothervalue")
check("weak 来源变化不影响机器码", machine_code(other) == a)
strong_changed = dict(HW, mac="112233445566")
check("可信来源变化会改变机器码", machine_code(strong_changed) != a)
# ---------------------------------------------------------------------------
# 容错匹配
# ---------------------------------------------------------------------------
def test_tolerance_single_component_swap():
"""换一块网卡 / 换一颗 eMMC 仍通过；整盘拷到别的机器不通过。"""
same = match_fingerprints(HW, dict(HW))
check("同机全命中", same.ok and same.matched == 4 and same.total == 4, same.reason)
nic = match_fingerprints(HW, dict(HW, mac="112233445566"))
check("换网卡仍通过（3/4）", nic.ok, nic.reason)
emmc = match_fingerprints(HW, dict(HW, emmc_cid="ffffffff"))
check("换 eMMC 仍通过（3/4）", emmc.ok, emmc.reason)
two = match_fingerprints(HW, dict(HW, mac="112233445566", emmc_cid="ffffffff"))
check("同时换网卡与 eMMC 不通过（2/4）", not two.ok, two.reason)
copied = match_fingerprints(
HW,
{"board_serial": "x1", "emmc_cid": "x2", "cpu_serial": "x3", "mac": "x4",
"machine_id": "sharedbyallunits"},
)
check("整盘拷贝到另一台机器不通过（weak 来源不得计票）",
not copied.ok and copied.matched == 0, copied.reason)
check("判定未降级", not copied.degraded, str(copied.degraded))
def test_tolerance_degraded_when_only_weak():
"""仅有 weak 来源时降级判定，并显式标记。"""
r = match_fingerprints({"machine_id": "shared"}, {"machine_id": "shared"})
check("仅 weak 来源时降级通过", r.ok and r.degraded, r.reason)
# ---------------------------------------------------------------------------
# 验签与判定
# ---------------------------------------------------------------------------
def test_sign_verify_roundtrip_and_tamper():
tmp = Path(tempfile.mkdtemp())
pub, priv = _keypair()
path = _write_license(tmp, priv)
st = read_license(path, pub, sources=HW)
check("正常授权校验通过", st.ok and st.code == REASON_OK, st.message)
check("解析出授权插件", st.licensed_plugins == {"aibox_sophon_adapter"},
str(st.licensed_plugins))
# 篡改白名单（典型的"自己加一个插件"）→ 签名失效
doc = json.loads(Path(path).read_text())
doc["payload"]["plugins"] = ["aibox_sophon_adapter", "doubao"]
bad = tmp / "tampered.lic"
bad.write_text(json.dumps(doc))
st2 = read_license(str(bad), pub, sources=HW)
check("篡改白名单被识破", not st2.ok and st2.code == REASON_SIGNATURE_INVALID,
st2.message)
# 换公钥（非本厂商签发）
other_pub, _ = _keypair()
st3 = read_license(path, other_pub, sources=HW)
check("非本厂商公钥验签失败",
not st3.ok and st3.code == REASON_SIGNATURE_INVALID, st3.message)
# 字段重排序不影响验签（签名覆盖规范化 JSON）
doc = json.loads(Path(path).read_text())
doc["payload"] = dict(reversed(list(doc["payload"].items())))
reordered = tmp / "reordered.lic"
reordered.write_text(json.dumps(doc))
st4 = read_license(str(reordered), pub, sources=HW)
check("载荷字段重排序不影响验签", st4.ok, st4.message)
def test_expiry_and_not_before():
tmp = Path(tempfile.mkdtemp())
pub, priv = _keypair()
expired = _write_license(
tmp, priv, expires_at=(date.today() - timedelta(days=1)).isoformat()
)
st = read_license(expired, pub, sources=HW)
check("过期授权被拒", not st.ok and st.code == REASON_EXPIRED, st.message)
# 到期当天仍有效（含首尾当天）
today = _write_license(tmp, priv, expires_at=date.today().isoformat())
st2 = read_license(today, pub, sources=HW)
check("到期当天仍然有效", st2.ok, st2.message)
future = _write_license(
tmp, priv, not_before=(date.today() + timedelta(days=5)).isoformat()
)
st3 = read_license(future, pub, sources=HW)
check("未生效授权被拒", not st3.ok and st3.code == REASON_NOT_YET_VALID, st3.message)
never = _write_license(tmp, priv, expires_at=None)
st4 = read_license(never, pub, sources=HW)
check("无到期日视为永久有效", st4.ok, st4.message)
def test_machine_mismatch_and_plugin_whitelist():
tmp = Path(tempfile.mkdtemp())
pub, priv = _keypair()
path = _write_license(tmp, priv, plugins=["ip_sound_column"])
st = read_license(path, pub, sources=HW)
check("本机授权有效", st.ok, st.message)
mgr = LicenseManager(license_path=path, public_key=pub, min_match=3,
clock_store=MemoryClockStore(), collector=lambda: HW)
check("白名单内插件放行", mgr.check_plugin("ip_sound_column").allowed)
denied = mgr.check_plugin("doubao")
check("白名单外插件拒绝",
not denied.allowed and denied.code == REASON_PLUGIN_NOT_LICENSED, denied.message)
other_machine = dict(HW, mac="999999999999", cpu_serial="888888888888")
st2 = read_license(path, pub, sources=other_machine)
check("机器码不符拒绝",
not st2.ok and st2.code == REASON_MACHINE_MISMATCH, st2.message)
def test_machine_code_only_license_fallback():
"""只签机器码（无逐来源指纹）时退化为精确匹配。"""
tmp = Path(tempfile.mkdtemp())
pub, priv = _keypair()
path = _write_license(tmp, priv, sources={}, machine_code=machine_code(HW))
st = read_license(path, pub, sources=HW)
check("仅机器码的授权：本机通过", st.ok, st.message)
st2 = read_license(path, pub, sources=dict(HW, mac="000000000001"))
check("仅机器码的授权：异机拒绝", not st2.ok and st2.code == REASON_MACHINE_MISMATCH,
st2.message)
# ---------------------------------------------------------------------------
# 申请码
# ---------------------------------------------------------------------------
def test_request_code_roundtrip():
code = encode_request_code(HW)
check("申请码前缀", code.startswith("CM1-"), code[:12])
data = decode_request_code(code)
check("申请码解析出全部来源", data["sources"] == HW, str(data["sources"]))
check("申请码含机器码", data["machine_code"] == machine_code(HW))
wrapped = code[:30] + "\n " + code[30:] + " "
check("申请码容忍折行与空白", decode_request_code(wrapped)["sources"] == HW)
try:
decode_request_code(code[:-1] + ("0" if code[-1] != "0" else "1"))
check("申请码校验和可检出抄写错误", False, "未抛异常")
except RequestCodeError as exc:
check("申请码校验和可检出抄写错误", True, str(exc))
try:
decode_request_code("not-a-code")
check("非法申请码被拒", False, "未抛异常")
except RequestCodeError:
check("非法申请码被拒", True)
# ---------------------------------------------------------------------------
# 时钟防线
# ---------------------------------------------------------------------------
def test_clock_guard():
store = MemoryClockStore()
guard = ClockGuard(store, tolerance_seconds=3600)
now = datetime(2026, 9, 20, 12, 0, tzinfo=timezone.utc)
check("首次检查通过并建立水位", guard.check(now).ok)
later = now + timedelta(hours=5)
check("时间前进正常", guard.check(later).ok)
# 回退 30 分钟：容差内，放行
slight = later - timedelta(minutes=30)
check("容差内回退放行", guard.check(slight).ok)
# 回退 10 小时：超过容差，拒绝
rollback = later - timedelta(hours=10)
r = guard.check(rollback)
check("明显回拨被检出", not r.ok and r.rollback_seconds > 0, r.reason)
# 水位未被回退覆盖
check("回拨后水位未被拉低", guard.check(later).ok)
def test_clock_guard_reseeds_when_far_ahead():
"""水位远超当前时间（RTC 异常）时重新起算，避免把设备永久卡死。"""
store = MemoryClockStore()
guard = ClockGuard(store, tolerance_seconds=3600, reseed_if_ahead_days=90)
guard.check(datetime(2027, 1, 1, tzinfo=timezone.utc))
r = guard.check(datetime(2026, 9, 20, tzinfo=timezone.utc))
check("大幅领先判为时钟异常而非攻击", r.ok and "异常" in r.reason, r.reason)
def test_manager_blocks_on_clock_rollback():
"""时钟回拨时管理器的判定整体中止。"""
tmp = Path(tempfile.mkdtemp())
pub, priv = _keypair()
path = _write_license(tmp, priv, plugins=["ip_sound_column"])
store = MemoryClockStore()
far_future = datetime.now(timezone.utc) + timedelta(days=2)
ClockGuard(store, tolerance_seconds=3600).check(far_future)
mgr = LicenseManager(license_path=path, public_key=pub, clock_store=store,
collector=lambda: HW)
st = mgr.reload()
check("时钟回拨时授权判定中止",
not st.ok and st.code == REASON_CLOCK_ROLLBACK, st.message)
check("时钟回拨时插件被拒", not mgr.check_plugin("ip_sound_column").allowed)
def test_manager_disabled_mode_reports_reality():
"""enforce=False（开发/测试）时放行，但状态仍如实反映真实校验结果。"""
tmp = Path(tempfile.mkdtemp())
pub, priv = _keypair()
path = _write_license(tmp, priv, plugins=[])
mgr = LicenseManager(license_path=path, public_key=pub, enforce=False,
collector=lambda: HW)
check("未启用时插件放行", mgr.check_plugin("doubao").allowed)
check("未启用时仍如实报告授权有效", mgr.status.ok, mgr.status.message)
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())