359 lines
14 KiB
Python
359 lines
14 KiB
Python
"""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())
|