first commit

.gitignore

@@ -0,0 +1,72 @@
+/target
+
+# Byte-compiled / optimized / DLL files
+__pycache__/
+.pytest_cache/
+*.py[cod]
+
+# C extensions
+*.so
+
+# Distribution / packaging
+.Python
+.venv/
+env/
+bin/
+build/
+develop-eggs/
+dist/
+eggs/
+lib/
+lib64/
+parts/
+sdist/
+var/
+include/
+man/
+venv/
+*.egg-info/
+.installed.cfg
+*.egg
+
+# Installer logs
+pip-log.txt
+pip-delete-this-directory.txt
+pip-selfcheck.json
+
+# Unit test / coverage reports
+htmlcov/
+.tox/
+.coverage
+.cache
+nosetests.xml
+coverage.xml
+
+# Translations
+*.mo
+
+# Mr Developer
+.mr.developer.cfg
+.project
+.pydevproject
+
+# Rope
+.ropeproject
+
+# Django stuff:
+*.log
+*.pot
+
+.DS_Store
+
+# Sphinx documentation
+docs/_build/
+
+# PyCharm
+.idea/
+
+# VSCode
+.vscode/
+
+# Pyenv
+.python-version

README.md

@@ -0,0 +1,91 @@
+# chongming-license
+
+机器码授权共享包：硬件指纹采集、license 载荷格式、Ed25519 验签与插件白名单判定。
+网关与厂商侧签发工具共用同一份实现 —— 尤其是**规范化 JSON 只有一处实现**，
+避免"自己签的自己验不过"。
+
+## 设计要点
+
+- **纯离线**：设备用内置公钥验签，运行期零联网。与现有 air-gapped 部署形态
+  （Zenoh localhost、Caddy `tls internal`、无 phone-home）一致。
+- **非对称**：Ed25519。注意根仓 `resources/*.key` 那套 `.mbank` 验签是
+  **HMAC 对称**且密钥与银行同目录发布，只能算防篡改/混淆，**不能**作为厂商
+  信任根 —— 授权必须用非对称，公钥可安全内置。
+- **N-of-M 容错**：license 里存各来源的原始指纹，校验按"命中数达阈值"判定，
+  而非把所有来源哈希成一个值 —— 否则换一块网卡就直接失效。
+  展示给客户的仍是单个 `machine_code`。
+- **weak 来源**：`/etc/machine-id` 在同镜像批量刷机场景下**所有机器相同**，
+  故标记 `weak` 且**不参与判定阈值**（只作诊断参考），避免它给攻击者白送一票。
+
+## 采集来源
+
+| 来源 | 位置 | 说明 |
+| --- | --- | --- |
+| `board_serial` | `/proc/device-tree/serial-number`、DMI `board_serial` | 逐机唯一，最稳定 |
+| `emmc_cid` | `/sys/block/mmcblk*/device/cid` | 芯片出厂唯一，不随系统重装变化 |
+| `cpu_serial` | `/proc/cpuinfo` 的 `Serial` | ARM 常见；部分型号全 0（会被剔除） |
+| `mac` | `/sys/class/net/*/address` | 取字典序最小者：插入/移除其它网卡不改变它 |
+| `product_uuid` | `/sys/class/dmi/id/product_uuid` | x86 有效，ARM 通常缺失 |
+| `machine_id` | `/etc/machine-id` | ⚠ weak，见上 |
+
+## 文件格式
+
+```json
+{
+  "payload": {
+    "v": 1,
+    "lic_id": "L2026-0001",
+    "customer": "某某科技",
+    "machine_code": "A3F9-1B77-9E04-2C8D-6F31-0A55",
+    "sources": {"board_serial": "...", "emmc_cid": "...", "mac": "..."},
+    "plugins": ["aibox_sophon_adapter", "ip_sound_column"],
+    "not_before": "2026-09-01",
+    "expires_at": "2027-09-01",
+    "issued_at": "2026-09-01T10:00:00Z",
+    "note": ""
+  },
+  "alg": "ed25519",
+  "kid": "vendor-2026",
+  "sig": "<base64，覆盖 payload 的规范化 JSON>"
+}
+```
+
+签名覆盖 payload **原始字节的规范化形式**（`canonical_json`：键排序、无空白），
+因此载荷内不允许浮点数（float 文本表示存在多种合法写法）。校验顺序是
+先验签、后解析模型 —— 反过来会让默认值填充/字段顺序改变字节导致验签失败。
+
+## 用法
+
+网关侧：
+
+```python
+from chongming_license import LicenseManager
+
+mgr = LicenseManager(
+    license_path="license.lic",
+    public_key=config["license"]["public_key"],
+    clock_store=cache,          # chongming_cache.Cache，存时间高水位
+    enforce=is_packaged(),      # 打包产物强制校验；开发/测试放行
+)
+decision = mgr.check_plugin("aibox_sophon_adapter")
+if not decision.allowed:
+    ...                          # 不加载该插件
+```
+
+厂商侧（`chongming-cli`，私钥只在这边）：
+
+```bash
+uv run chongming-license keygen -o license_private.pem   # 一次性生成密钥对
+uv run chongming-license issue --machine-code <码> --sources-json <采集导出> \
+    --plugins aibox_sophon_adapter,ip_sound_column \
+    --customer 某某科技 --expires 2027-09-01 -k license_private.pem -o license.lic
+```
+
+## 诚实的边界
+
+离线 + root 的天花板是**阻止随手拷贝**，不是阻止逆向：有 root 者可改二进制
+跳过校验，或 `LD_PRELOAD` 伪造指纹读取。`clock` 模块的时间回拨检测同理 ——
+删掉缓存文件即失效，它只是抬高随手改时间的门槛。
+
+真正难以绕过的是**签名**：伪造 license 需要厂商私钥，而公钥可明文放置
+（换公钥只会验签失败，无法据以伪造）。

pyproject.toml

@@ -0,0 +1,25 @@
+[build-system]
+build-backend = "hatchling.build"
+requires = ["hatchling"]
+
+[project]
+name = "chongming-license"
+version = "0.1.0"
+description = "chongming 机器码授权：硬件指纹采集、license 载荷与 Ed25519 验签、插件白名单判定（网关与签发工具共用）"
+readme = "README.md"
+requires-python = ">=3.12"
+dependencies = [
+    "cryptography",
+    "orjson",
+    "pydantic",
+]
+
+# 不在此声明 [dependency-groups] dev：本包已加入根 workspace（members），
+# 其 dev group 会并入 workspace 解析，而根仓的测试统一放在 test/ 下运行。
+# 若日后需要独立开发，再按 utils/auth 的写法单独加上。
+
+[tool.hatch.build.targets.wheel]
+packages = ["src/chongming_license"]
+
+[[tool.uv.index]]
+url = "http://124.71.68.6:3000/api/packages/chongming/pypi/simple/"

src/chongming_license/__init__.py

@@ -0,0 +1,366 @@
+"""chongming-license —— 机器码授权共享包。
+
+纯离线授权：厂商用**私钥**签发 ``license.lic``（绑定机器指纹 + 插件白名单 +
+有效期），设备用内置**公钥**验签，无需任何联网。
+
+使用方式（网关侧）::
+
+    from chongming_license import LicenseManager
+
+    mgr = LicenseManager(
+        license_path="license.lic",
+        public_key=config["license"]["public_key"],
+        clock_store=cache,
+        enforce=is_packaged(),          # 打包产物强制校验，开发/测试放行
+    )
+    decision = mgr.check_plugin("aibox_sophon_adapter")
+    if not decision.allowed:
+        ...                             # 不加载该插件
+
+两侧职责边界：
+  - 本包（网关与签发工具共用）：指纹采集、载荷格式、规范化与验签、判定、时钟防线
+  - 签发工具（``chongming-cli license``，仅厂商侧）：生成密钥对、签发 license
+  - 私钥绝不进入部署产物（构建脚本有守护检查）
+"""
+from typing import Dict
+
+from .clock import (
+    CLOCK_HWM_KEY,
+    DEFAULT_CLOCK_TOLERANCE_SECONDS,
+    DEFAULT_RESEED_IF_AHEAD_DAYS,
+    ClockCheck,
+    ClockGuard,
+    MemoryClockStore,
+)
+from .issue import (
+    build_license,
+    build_payload,
+    make_license_id,
+    resolve_expiry,
+    utc_today,
+    write_license,
+)
+from .machine import (
+    DEFAULT_MIN_MATCH,
+    MACHINE_CODE_HEX_LEN,
+    REQUEST_CODE_PREFIX,
+    SOURCES,
+    FingerprintReport,
+    FingerprintUnavailableError,
+    MatchResult,
+    RawValue,
+    RequestCodeError,
+    SourceDetail,
+    SourceSpec,
+    collect_fingerprints,
+    collect_report,
+    decode_request_code,
+    encode_request_code,
+    machine_code,
+    match_fingerprints,
+    normalize,
+)
+from .models import (
+    LICENSE_VERSION,
+    MACHINE_CODE_GROUP,
+    SIGNATURE_ALGORITHM,
+    LicenseFile,
+    LicensePayload,
+)
+from .registry import (
+    DEFAULT_EXPIRING_DAYS,
+    SCHEMA_VERSION,
+    STATUS_LABELS,
+    LicenseRecord,
+    Registry,
+    RegistryError,
+    default_registry_path,
+    record_from_license,
+)
+from .verify import (
+    REASON_CLOCK_ROLLBACK,
+    REASON_DISABLED,
+    REASON_EXPIRED,
+    REASON_FINGERPRINT_UNAVAILABLE,
+    REASON_LICENSE_MALFORMED,
+    REASON_LICENSE_MISSING,
+    REASON_MACHINE_MISMATCH,
+    REASON_NOT_YET_VALID,
+    REASON_NO_PUBLIC_KEY,
+    REASON_OK,
+    REASON_PLUGIN_NOT_LICENSED,
+    REASON_SIGNATURE_INVALID,
+    REASON_TEXT,
+    REASON_UNSUPPORTED_ALG,
+    LicenseError,
+    LicenseMalformedError,
+    LicenseStatus,
+    PluginDecision,
+    PublicKeyError,
+    SignatureInvalidError,
+    canonical_json,
+    decision_for_plugin,
+    generate_keypair,
+    load_private_key,
+    load_public_key,
+    public_key_b64,
+    public_key_fingerprint,
+    public_key_pem,
+    read_license,
+    sign_payload,
+    verify_signature,
+    write_private_key,
+)
+
+__all__ = [
+    # 门面
+    "LicenseManager",
+    # 签发（厂商侧）
+    "build_license",
+    "build_payload",
+    "make_license_id",
+    "resolve_expiry",
+    "utc_today",
+    "write_license",
+    # 机器指纹
+    "SOURCES",
+    "DEFAULT_MIN_MATCH",
+    "MACHINE_CODE_HEX_LEN",
+    "REQUEST_CODE_PREFIX",
+    "FingerprintReport",
+    "FingerprintUnavailableError",
+    "MatchResult",
+    "RawValue",
+    "RequestCodeError",
+    "SourceDetail",
+    "SourceSpec",
+    "collect_fingerprints",
+    "collect_report",
+    "decode_request_code",
+    "encode_request_code",
+    "machine_code",
+    "match_fingerprints",
+    "normalize",
+    # 载荷格式
+    "LICENSE_VERSION",
+    "MACHINE_CODE_GROUP",
+    "SIGNATURE_ALGORITHM",
+    "LicenseFile",
+    "LicensePayload",
+    # 验签与判定
+    "LicenseError",
+    "LicenseMalformedError",
+    "LicenseStatus",
+    "PluginDecision",
+    "PublicKeyError",
+    "SignatureInvalidError",
+    "REASON_TEXT",
+    "REASON_OK",
+    "REASON_DISABLED",
+    "REASON_LICENSE_MISSING",
+    "REASON_LICENSE_MALFORMED",
+    "REASON_SIGNATURE_INVALID",
+    "REASON_UNSUPPORTED_ALG",
+    "REASON_NO_PUBLIC_KEY",
+    "REASON_MACHINE_MISMATCH",
+    "REASON_NOT_YET_VALID",
+    "REASON_EXPIRED",
+    "REASON_CLOCK_ROLLBACK",
+    "REASON_PLUGIN_NOT_LICENSED",
+    "REASON_FINGERPRINT_UNAVAILABLE",
+    "canonical_json",
+    "decision_for_plugin",
+    "generate_keypair",
+    "load_private_key",
+    "load_public_key",
+    "public_key_b64",
+    "public_key_fingerprint",
+    "public_key_pem",
+    "read_license",
+    "sign_payload",
+    "verify_signature",
+    "write_private_key",
+    # 时钟防线
+    "CLOCK_HWM_KEY",
+    "DEFAULT_CLOCK_TOLERANCE_SECONDS",
+    "DEFAULT_RESEED_IF_AHEAD_DAYS",
+    "ClockCheck",
+    "ClockGuard",
+    "MemoryClockStore",
+    # 已签发台账（厂商侧）
+    "DEFAULT_EXPIRING_DAYS",
+    "SCHEMA_VERSION",
+    "STATUS_LABELS",
+    "LicenseRecord",
+    "Registry",
+    "RegistryError",
+    "default_registry_path",
+    "record_from_license",
+]
+
+
+class LicenseManager:
+    """授权门面：读一次 license，回答"某插件能否运行"。
+
+    结果缓存到 :meth:`reload` 之前，因此启动期对每个插件调用 ``check_plugin``
+    只付出一次文件读取与验签开销。替换 license 文件后调用 :meth:`reload` 即可
+    生效（管理接口导入新文件时会调用）。
+
+    Args:
+        license_path: license.lic 路径
+        public_key: 厂商公钥（base64 裸 32 字节或 PEM）；为空则视为未配置
+        min_match: 指纹命中阈值
+        clock_store: 时钟水位存储（``chongming_cache.Cache`` 或等价对象）
+        enforce: 是否真正执行拒绝。False 时 ``check_plugin`` 一律放行，
+                 但 :attr:`status` 仍如实反映校验结果（开发/测试模式）
+        collector: 指纹采集函数，None = 现场采集（测试注入用）
+        logger: 可选的 ``logging.Logger``，用于输出启动横幅
+    """
+
+    def __init__(
+        self,
+        *,
+        license_path: str = "license.lic",
+        public_key: str = "",
+        min_match: int = DEFAULT_MIN_MATCH,
+        clock_store=None,
+        enforce: bool = True,
+        collector=None,
+        logger=None,
+    ) -> None:
+        self.license_path = license_path
+        self.public_key = public_key or ""
+        self.min_match = max(1, int(min_match))
+        self.enforce = bool(enforce)
+        self._collector = collector
+        self._logger = logger
+        self._clock_guard = ClockGuard(clock_store)
+        self._status: LicenseStatus | None = None
+
+    # -- 校验与缓存 ---------------------------------------------------------
+    def _collect(self):
+        """采集本机指纹；未注入 collector 时走现场采集。"""
+        if self._collector is not None:
+            return self._collector()
+        return collect_fingerprints()
+
+    def collect_sources(self) -> Dict[str, str]:
+        """采集本机指纹来源（公开接口；管理接口导入 license 时复用同一份采集）。"""
+        return self._collect()
+
+    def reload(self) -> LicenseStatus:
+        """重新读取并校验 license（丢弃缓存）。"""
+        # 先过时钟防线：检出回拨时不再采信有效期
+        clock = self._clock_guard.check()
+        if not clock.ok:
+            self._status = LicenseStatus(
+                ok=False,
+                code=REASON_CLOCK_ROLLBACK,
+                message=f"{REASON_TEXT[REASON_CLOCK_ROLLBACK]}：{clock.reason}",
+                license_path=str(self.license_path),
+                public_key_fingerprint=public_key_fingerprint(self.public_key),
+                warnings=[clock.reason],
+            )
+            return self._status
+
+        try:
+            sources = self._collect()
+        except Exception as exc:
+            sources = {}
+            self._status = LicenseStatus(
+                ok=False,
+                code=REASON_FINGERPRINT_UNAVAILABLE,
+                message=f"{REASON_TEXT[REASON_FINGERPRINT_UNAVAILABLE]}：{exc}",
+                license_path=str(self.license_path),
+                public_key_fingerprint=public_key_fingerprint(self.public_key),
+                warnings=[str(exc)],
+            )
+            return self._status
+
+        try:
+            report = collect_report()
+            warnings = list(report.warnings)
+        except Exception:
+            warnings = []
+
+        status = read_license(
+            self.license_path,
+            self.public_key,
+            min_match=self.min_match,
+            sources=sources,
+        )
+        # 采集侧告警（无效值、来源不足等）并入状态，便于管理页一次看全
+        status.warnings = warnings + list(status.warnings)
+        if clock.reason:
+            status.warnings.append(clock.reason)
+        self._status = status
+        return status
+
+    @property
+    def status(self) -> LicenseStatus:
+        """当前校验结论（首次访问时自动校验）。"""
+        if self._status is None:
+            return self.reload()
+        return self._status
+
+    # -- 对外判定 -----------------------------------------------------------
+    def check_plugin(self, plugin_type: str) -> PluginDecision:
+        """该插件是否获授权运行。
+
+        ``enforce=False``（开发/测试）时始终放行，但仍可在 :attr:`status`
+        看到真实的校验结论。
+        """
+        status = self.status
+        if not self.enforce:
+            return PluginDecision(
+                allowed=True,
+                code=REASON_DISABLED,
+                plugin_type=plugin_type,
+                message=f"{REASON_TEXT[REASON_DISABLED]}（真实校验结果：{status.code}）",
+            )
+        return decision_for_plugin(status, plugin_type)
+
+    def denied_types(self, plugin_types) -> set[str]:
+        """从给定插件类型中筛出**未获授权**的那些（当前不允许加载的集合）。"""
+        return {t for t in plugin_types if not self.check_plugin(t).allowed}
+
+    def describe(self) -> dict:
+        """给管理接口/前端展示的完整信息（含每个来源的采集明细）。"""
+        status = self.status
+        try:
+            report = collect_report()
+            details = [d.as_dict() for d in report.details]
+            warnings = list(report.warnings)
+        except Exception as exc:
+            details, warnings = [], [str(exc)]
+        return {
+            "enforce": self.enforce,
+            "min_match": self.min_match,
+            "status": status.as_dict(),
+            "sources_detail": details,
+            "warnings": list(dict.fromkeys(warnings + status.warnings)),
+        }
+
+    # -- 日志 ---------------------------------------------------------------
+    def banner(self) -> list[str]:
+        """启动日志行（由调用方按自己的日志格式输出）。"""
+        status = self.status
+        flag = "✓" if status.ok else "✗"
+        if not self.enforce:
+            lines = [
+                f"[license] 校验未启用（开发/测试模式），本机机器码 {status.machine_code}",
+                f"[license] 当前授权文件真实校验结果：{status.message}",
+            ]
+        else:
+            lines = [f"[license] {flag} {status.message}（本机机器码 {status.machine_code}）"]
+            if status.payload:
+                p = status.payload
+                lines.append(
+                    f"[license] 编号 {p.lic_id}"
+                    + (f"，客户 {p.customer}" if p.customer else "")
+                    + (f"，到期 {p.expires_at}" if p.expires_at else "，无到期日")
+                    + f"，授权插件 {sorted(status.licensed_plugins) or '无'}"
+                )
+        if status.public_key_fingerprint:
+            lines.append(f"[license] 厂商公钥指纹 {status.public_key_fingerprint}")
+        return lines

src/chongming_license/clock.py

@@ -0,0 +1,159 @@
+"""系统时间回拨检测（离线授权的必要配套）。
+
+纯离线方案里"有效期"依赖本机时钟，而本机时钟对 root 是可改的：把时间调回
+授权期内，过期的 license 就能一直用。这里用**高水位标记**做最基础的一道防线 ——
+把观测到过的最大时间持久化，之后若时间显著早于该水位，即判定被回拨。
+
+诚实的边界：攻击者若删掉缓存文件（或整库恢复快照），水位就没了。因此这是
+"抬高随手改时间的门槛"，不是不可绕过的防线；真正难绕过的是 license 有效期
+本身（由厂商私钥签名，无法伪造）。所以本模块**只用于中止校验并告警**，
+不参与签名与机器绑定判定。
+
+存储接口（duck typing）：``get(key, default)`` / ``set(key, value)``
+—— 直接传 ``chongming_cache.Cache`` 即可；测试可传 :class:`MemoryClockStore`。
+注意**不要给该键设置 TTL**：水位一旦过期被淘汰，防线即失效。
+"""
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from typing import Any, Dict, Protocol
+
+# 高水位键名（与 cache 内其它键同库共存）
+CLOCK_HWM_KEY = "license:clock_hwm"
+
+# 容差（秒）：允许的回退幅度，默认 1 天。
+# 取较大值是为了容忍 NTP 校时、RTC 初始不准等**正常**回退而不误判；对 license
+# 而言（有效期以月/年计）1 天的作弊窗口没有实际意义。
+DEFAULT_CLOCK_TOLERANCE_SECONDS = 86400
+
+# 水位领先当前时间超过该天数时，视为时钟本身不可信（如首次上电 RTC 混乱、
+# 出厂预置了错误年份），记录告警并重新起算，避免把设备永久卡死。
+DEFAULT_RESEED_IF_AHEAD_DAYS = 90
+
+# 记录到存储里的时间格式（ISO8601，便于运维直接查看 cache.db）
+_STORED_FORMAT = "%Y-%m-%dT%H:%M:%SZ"
+
+
+class ClockStore(Protocol):
+    """时钟水位的最小存储接口（chongming_cache.Cache 天然满足）。"""
+
+    def get(self, key: str, default: Any = None) -> Any: ...
+    def set(self, key: str, value: Any) -> None: ...
+
+
+class MemoryClockStore:
+    """内存存储（测试用；进程重启即丢失水位）。"""
+
+    def __init__(self) -> None:
+        self._data: Dict[str, Any] = {}
+
+    def get(self, key: str, default: Any = None) -> Any:
+        return self._data.get(key, default)
+
+    def set(self, key: str, value: Any) -> None:
+        self._data[key] = value
+
+
+@dataclass
+class ClockCheck:
+    """时钟检查结论。"""
+
+    ok: bool
+    reason: str = ""
+    rollback_seconds: float = 0.0
+    high_water_mark: datetime | None = None
+
+    def as_dict(self) -> dict:
+        return {
+            "ok": self.ok,
+            "reason": self.reason,
+            "rollback_seconds": self.rollback_seconds,
+            "high_water_mark": (
+                self.high_water_mark.strftime(_STORED_FORMAT) if self.high_water_mark else None
+            ),
+        }
+
+
+class ClockGuard:
+    """时间回拨检测器。
+
+    Args:
+        store: 持久化存储（chongming_cache.Cache 或等价对象）；None = 不检测
+        tolerance_seconds: 允许的回退幅度
+        reseed_if_ahead_days: 水位领先当前时间超过该天数时重新起算
+    """
+
+    def __init__(
+        self,
+        store: ClockStore | None = None,
+        *,
+        tolerance_seconds: float = DEFAULT_CLOCK_TOLERANCE_SECONDS,
+        reseed_if_ahead_days: float = DEFAULT_RESEED_IF_AHEAD_DAYS,
+    ) -> None:
+        self._store = store
+        self.tolerance = max(0.0, float(tolerance_seconds))
+        self.reseed_if_ahead_days = max(0.0, float(reseed_if_ahead_days))
+
+    def _read_hwm(self) -> datetime | None:
+        raw = self._store.get(CLOCK_HWM_KEY) if self._store else None
+        if not raw:
+            return None
+        try:
+            return datetime.strptime(str(raw), _STORED_FORMAT).replace(tzinfo=timezone.utc)
+        except ValueError:
+            # 存储被人工改坏：丢弃水位重新起算，不让它卡死设备
+            return None
+
+    def _write_hwm(self, when: datetime) -> None:
+        if self._store:
+            self._store.set(CLOCK_HWM_KEY, when.strftime(_STORED_FORMAT))
+
+    def check(self, now: datetime | None = None) -> ClockCheck:
+        """检查并推进水位。
+
+        Returns:
+            ClockCheck：``ok=False`` 表示检出明显回拨（应中止授权校验）
+        """
+        if self._store is None:
+            return ClockCheck(ok=True, reason="未配置时钟水位存储，跳过回拨检测")
+
+        current = now or datetime.now(timezone.utc)
+        if current.tzinfo is None:
+            current = current.replace(tzinfo=timezone.utc)
+
+        hwm = self._read_hwm()
+        if hwm is None:
+            self._write_hwm(current)
+            return ClockCheck(ok=True, high_water_mark=current)
+
+        skew = (hwm - current).total_seconds()
+
+        # 时钟大幅领先：视为 RTC/时区异常而非攻击，重新起算并告警
+        if skew > self.reseed_if_ahead_days * 86400:
+            self._write_hwm(current)
+            return ClockCheck(
+                ok=True,
+                reason=(
+                    f"检测到系统时间比历史水位晚 {skew / 86400:.1f} 天，"
+                    "判为时钟异常并重新起算水位（请检查 RTC/时区/NTP）"
+                ),
+                high_water_mark=current,
+            )
+
+        # 明显回拨：中止授权校验
+        if skew > self.tolerance:
+            return ClockCheck(
+                ok=False,
+                reason=(
+                    f"系统时间比历史水位早 {skew / 3600:.1f} 小时"
+                    f"（超过容差 {self.tolerance / 3600:.1f} 小时）"
+                ),
+                rollback_seconds=skew,
+                high_water_mark=hwm,
+            )
+
+        # 正常或容差内的小幅回退：推进水位（不推进到 current，保持单调）
+        if current > hwm:
+            self._write_hwm(current)
+        return ClockCheck(ok=True, high_water_mark=max(hwm, current))

src/chongming_license/issue.py

@@ -0,0 +1,126 @@
+"""授权签发（厂商侧共用实现）。
+
+字段拼装刻意收在这里，而不是让 CLI 与 GUI 各自拼一遍：载荷 schema 只要有两处
+实现，就迟早出现"两边字段不一致 → 签出来的文件另一侧读不懂"的问题。签发端的
+规范化与校验端共用 ``canonical_json``（见 :mod:`chongming_license.verify`），
+所以**必须**用本模块构造载荷，不要手写 dict。
+"""
+from __future__ import annotations
+
+import json
+import secrets
+from datetime import date, datetime, timedelta, timezone
+from pathlib import Path
+from typing import Any, Dict, Iterable, Mapping
+
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+
+from .models import LICENSE_VERSION, SIGNATURE_ALGORITHM
+from .verify import sign_payload
+
+
+def utc_today() -> date:
+    """判定基准日期用 **UTC**，与校验端保持一致。
+
+    若签发端用本地日期、校验端用 UTC，在时区偏移跨日的时段会差一天
+    （出现"刚签发即过期"或"到期后仍可用一天"）。
+    """
+    return datetime.now(timezone.utc).date()
+
+
+def make_license_id(when: date | None = None) -> str:
+    """自动生成授权编号：``L<日期>-<6位随机>``（仅便于对账，不参与校验）。"""
+    ref = when or utc_today()
+    return f"L{ref.strftime('%Y%m%d')}-{secrets.token_hex(3).upper()}"
+
+
+def resolve_expiry(
+    *,
+    days: int | None = None,
+    expires: str | date | None = None,
+) -> str | None:
+    """把「N 天」或「指定日期」统一成 ``YYYY-MM-DD``；两者都空 = 永久（None）。"""
+    if days is not None and expires:
+        raise ValueError("--days 与 --expires 不能同时使用")
+    if days is not None:
+        return (utc_today() + timedelta(days=int(days))).isoformat()
+    if not expires:
+        return None
+    text = expires.isoformat() if isinstance(expires, date) else str(expires).strip()
+    return date.fromisoformat(text[:10]).isoformat()
+
+
+def build_payload(
+    *,
+    machine_code: str,
+    plugins: Iterable[str],
+    sources: Mapping[str, str] | None = None,
+    customer: str = "",
+    lic_id: str = "",
+    not_before: str | date | None = None,
+    expires_at: str | date | None = None,
+    note: str = "",
+    issued_at: str | None = None,
+) -> Dict[str, Any]:
+    """构造 license 载荷（签发前可预览）。
+
+    Args:
+        machine_code: 机器码。由申请码/来源指纹推导，或客户直接报来的短码
+        plugins: 授权插件 type 列表（去空白、去重、保持顺序）
+        sources: 各来源原始指纹；为空表示"仅按机器码精确匹配"（无换硬件容错）
+        not_before / expires_at: ISO8601 日期，含首尾当天；None = 立即生效 / 永久
+    """
+    clean_plugins = [p.strip() for p in plugins if str(p).strip()]
+    if not clean_plugins:
+        raise ValueError("授权插件列表为空：不授权任何插件的 license 没有意义")
+    if not machine_code.strip():
+        raise ValueError("缺少机器码：无法确定授权对象")
+
+    def _date_str(v: str | date | None) -> str | None:
+        if v is None:
+            return None
+        return v.isoformat() if isinstance(v, date) else (str(v).strip()[:10] or None)
+
+    return {
+        "v": LICENSE_VERSION,
+        "lic_id": lic_id or make_license_id(),
+        "customer": customer,
+        "machine_code": machine_code.strip().upper(),
+        "sources": dict(sources or {}),
+        "plugins": clean_plugins,
+        "not_before": _date_str(not_before),
+        "expires_at": _date_str(expires_at),
+        "issued_at": issued_at or datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
+        "note": note,
+    }
+
+
+def build_license(
+    payload_raw: Mapping[str, Any],
+    private_key: Ed25519PrivateKey,
+    *,
+    kid: str = "",
+) -> Dict[str, Any]:
+    """用私钥对载荷签名，返回完整的 license 文档（可 JSON 序列化）。
+
+    签名覆盖载荷的规范化 JSON，因此载荷内**不允许浮点数**（float 的文本表示
+    有多种合法写法，会破坏字节级一致性）。
+    """
+    payload = dict(payload_raw)
+    return {
+        "payload": payload,
+        "alg": SIGNATURE_ALGORITHM,
+        "kid": kid,
+        "sig": sign_payload(payload, private_key),
+    }
+
+
+def write_license(doc: Mapping[str, Any], path: str | Path) -> Path:
+    """把 license 文档写成易读的 JSON 文件（保证父目录存在）。"""
+    target = Path(path)
+    target.parent.mkdir(parents=True, exist_ok=True)
+    target.write_text(
+        json.dumps(dict(doc), ensure_ascii=False, indent=2),
+        encoding="utf-8",
+    )
+    return target

src/chongming_license/machine.py

@@ -0,0 +1,496 @@
+"""硬件指纹采集与机器码生成。
+
+目标平台是无外网的嵌入式边缘盒子（如 Sophon BM1684），因此指纹必须来自
+**本机可读的硬件标识**，且在重启、升级、（同机型）换机之间保持可预期。
+
+采集来源（按可信度排序）::
+
+    board_serial   设备树 / DMI 主板序列号      —— 逐机唯一，最稳定
+    emmc_cid       eMMC CID（芯片出厂唯一）     —— 逐机唯一，不随系统重装变化
+    cpu_serial     /proc/cpuinfo 的 Serial     —— ARM 常见；部分型号全 0（会被剔除）
+    mac            物理网卡 MAC（取字典序最小） —— 逐机唯一；换网卡会变
+    product_uuid   DMI product_uuid            —— x86 有效，ARM 通常缺失
+    machine_id     /etc/machine-id             —— ⚠ weak：同镜像批量刷机时所有机器相同
+
+``weak`` 标记的来源**不参与判定阈值**，只作诊断参考 —— 这正是为了避开
+"/etc/machine-id 在批量刷机场景下全机相同"这个陷阱：若把 machine_id 计入
+阈值，攻击者只要把镜像整体拷到另一台机器就能拿到 1 票；不计入则它毫无
+帮助。参见过滤后若可用来源不足，会自动降级把 weak 来源纳入并给出告警。
+
+**容错匹配**：license 里保存的是各来源的**原始指纹**，校验时按 N-of-M
+判定（默认 3/4），而不是把所有来源哈希成一个值 —— 否则换一块网卡或换
+一颗 eMMC 就直接失效。展示给客户的仍是单个 ``machine_code``。
+"""
+from __future__ import annotations
+
+import base64
+import hashlib
+import os
+import re
+from dataclasses import dataclass, field
+from glob import glob
+from typing import Callable, Dict, List
+
+import orjson
+
+# 机器码位数（十六进制字符数）。128 bit 足以避免碰撞，且抄写长度可接受。
+MACHINE_CODE_HEX_LEN = 24
+
+# 默认容错阈值：命中数 >= min_match 即视为同一台机器
+DEFAULT_MIN_MATCH = 3
+
+# 来源值最小长度（过短不可能是有效身份标识，多半是占位符）
+_MIN_VALUE_LEN = 4
+
+# 常见占位/无效值（规范化后比较）
+_PLACEHOLDERS = frozenset({
+    "unknown", "none", "null", "nil", "na", "n", "notspecified", "notavailable",
+    "systemserialnumber", "systemversion", "defaultstring", "tobefilledbyoem",
+    "oem", "oemstring", "nonesuch", "invalid", "uninitialized",
+})
+
+# 虚拟网卡名字前缀兜底（优先用 sysfs 的 device 符号链接判定物理性）
+_VIRTUAL_IF_PREFIXES = (
+    "docker", "veth", "br-", "virbr", "tun", "tap", "bond", "dummy",
+    "sit", "ip6", "wg", "zt", "tailscale", "vmnet", "ham",
+)
+
+# 一律排除的接口名前缀：回环及其别名。WSL 等环境存在名为 loopback0 的接口，
+# 它带有 device 符号链接、能骗过"物理网卡"判定，因此名字层面的排除必须
+# **无条件生效**，不能只作为兜底。
+_ALWAYS_EXCLUDED_PREFIXES = ("lo", "loopback")
+
+_NON_ALNUM = re.compile(r"[^0-9a-z]+")
+
+
+class FingerprintUnavailableError(RuntimeError):
+    """本机采集不到任何可用硬件指纹（无法生成机器码）。"""
+
+
+@dataclass(frozen=True)
+class RawValue:
+    """一个来源的原始读取结果。
+
+    value:  参与归一化与比对的值
+    detail: 仅用于展示的附加说明（如"来自 eth0"），不参与指纹计算
+    """
+
+    value: str
+    detail: str = ""
+
+
+@dataclass(frozen=True)
+class SourceSpec:
+    """一个指纹来源的声明。"""
+
+    name: str
+    label: str
+    reader: Callable[[], RawValue | None]
+    weak: bool = False
+
+
+def normalize(value: str | None) -> str:
+    """归一化来源值：小写、仅保留 [0-9a-z]。
+
+    签发端与校验端共用本函数，因此结果一致；分隔符差异
+    （``AA:BB:..`` / ``aa-bb-..``）不会造成误判。
+    无效值（空、占位符、全 0、全 f、过短）返回空字符串。
+    """
+    if not value:
+        return ""
+    cleaned = _NON_ALNUM.sub("", value.strip().lower())
+    if len(cleaned) < _MIN_VALUE_LEN:
+        return ""
+    if cleaned in _PLACEHOLDERS:
+        return ""
+    # 全 0 / 全 f：未烧录或读失败的典型表现
+    if set(cleaned) <= {"0"} or set(cleaned) <= {"f"}:
+        return ""
+    return cleaned
+
+
+def _read_text(path: str) -> str | None:
+    """读文本文件；不存在/无权限/异常一律返回 None（单个来源失败不影响整体）。"""
+    try:
+        with open(path, "rb") as f:
+            raw = f.read()
+    except OSError:
+        return None
+    # 设备树属性文件常以 NUL 结尾
+    return raw.replace(b"\x00", b"").decode("utf-8", errors="ignore").strip()
+
+
+def _read_first_line(text: str | None, *keys: str) -> str | None:
+    """在 ``key: value`` 形式的文本中取首个匹配键的值（如 /proc/cpuinfo）。"""
+    if not text:
+        return None
+    wanted = tuple(k.lower() for k in keys)
+    for line in text.splitlines():
+        if ":" not in line:
+            continue
+        k, _, v = line.partition(":")
+        if k.strip().lower() in wanted and v.strip():
+            return v.strip()
+    return None
+
+
+def _read_board_serial() -> RawValue | None:
+    """主板序列号：优先设备树（ARM），其次 DMI（x86）。"""
+    for path in (
+        "/proc/device-tree/serial-number",
+        "/sys/firmware/devicetree/base/serial-number",
+    ):
+        v = _read_text(path)
+        if v:
+            return RawValue(v, detail=path)
+    v = _read_text("/sys/class/dmi/id/board_serial")
+    if v:
+        return RawValue(v, detail="dmi/board_serial")
+    return None
+
+
+def _read_emmc_cid() -> RawValue | None:
+    """eMMC 芯片 CID / 序列号（出厂唯一，不随系统重装变化）。"""
+    for pattern, note in (
+        ("/sys/block/mmcblk*/device/cid", "cid"),
+        ("/sys/block/mmcblk*/device/serial", "serial"),
+    ):
+        for path in sorted(glob(pattern)):
+            v = _read_text(path)
+            if v:
+                return RawValue(v, detail=f"{path} ({note})")
+    return None
+
+
+def _read_cpu_serial() -> RawValue | None:
+    """ARM 平台 /proc/cpuinfo 的 Serial 字段（部分型号输出全 0，归一化时剔除）。"""
+    v = _read_first_line(_read_text("/proc/cpuinfo"), "serial")
+    return RawValue(v, detail="/proc/cpuinfo Serial") if v else None
+
+
+def _physical_interfaces() -> List[str]:
+    """枚举物理网卡名。
+
+    先无条件排除回环类接口（lo / loopback0 等），再用 sysfs 的 ``device``
+    符号链接判定物理性（物理网卡有、虚拟网卡没有）。若一个都没识别出来
+    （部分 ARM 平台无该链接），退回名字前缀黑名单。
+    """
+    base = "/sys/class/net"
+    try:
+        names = sorted(os.listdir(base))
+    except OSError:
+        return []
+    candidates = [
+        n for n in names if not n.startswith(_ALWAYS_EXCLUDED_PREFIXES)
+    ]
+    physical = [
+        n for n in candidates
+        if os.path.isdir(os.path.join(base, n, "device"))
+    ]
+    if physical:
+        return physical
+    return [n for n in candidates if not n.startswith(_VIRTUAL_IF_PREFIXES)]
+
+
+def _read_mac() -> RawValue | None:
+    """物理网卡 MAC：取字典序最小者。
+
+    取最小值而非"第一块网卡"，是为了让指纹在**插入/移除其它网卡**（USB 网卡、
+    调试用的额外接口）以及**接口重命名**（eth0 ↔ enp1s0）时保持稳定 —— 只有
+    最小值那块网卡本身被更换才会变化。
+    """
+    base = "/sys/class/net"
+    found: Dict[str, str] = {}
+    for name in _physical_interfaces():
+        mac = normalize(_read_text(os.path.join(base, name, "address")))
+        if len(mac) == 12:  # MAC 定长 6 字节
+            found.setdefault(mac, name)
+    if not found:
+        return None
+    mac = min(found)
+    return RawValue(mac, detail=f"来自 {found[mac]}")
+
+
+def _read_product_uuid() -> RawValue | None:
+    """DMI product_uuid（x86 逐机唯一；ARM 通常不存在）。"""
+    v = _read_text("/sys/class/dmi/id/product_uuid")
+    return RawValue(v, detail="dmi/product_uuid") if v else None
+
+
+def _read_machine_id() -> RawValue | None:
+    """systemd machine-id。
+
+    ⚠ weak 来源：以同一份系统镜像批量刷机时，若未在首次启动重新生成，
+    所有机器的该值完全相同。因此只作诊断参考，不计入判定阈值。
+    """
+    for path in ("/etc/machine-id", "/var/lib/dbus/machine-id"):
+        v = _read_text(path)
+        if v:
+            return RawValue(v, detail=f"{path}（可能因同镜像刷机而重复）")
+    return None
+
+
+# 来源清单（顺序即展示顺序，也是筛选不足时的优先取舍顺序）
+SOURCES: tuple[SourceSpec, ...] = (
+    SourceSpec("board_serial", "主板序列号（设备树/DMI）", _read_board_serial),
+    SourceSpec("emmc_cid", "eMMC 芯片 CID", _read_emmc_cid),
+    SourceSpec("cpu_serial", "CPU 序列号（/proc/cpuinfo）", _read_cpu_serial),
+    SourceSpec("mac", "物理网卡 MAC（字典序最小）", _read_mac),
+    SourceSpec("product_uuid", "DMI product_uuid", _read_product_uuid),
+    SourceSpec("machine_id", "systemd machine-id", _read_machine_id, weak=True),
+)
+
+_BY_NAME = {s.name: s for s in SOURCES}
+
+
+@dataclass
+class SourceDetail:
+    """单个来源的采集明细（供 ``--sources`` / 前端诊断展示）。"""
+
+    name: str
+    label: str
+    weak: bool
+    present: bool
+    value: str = ""
+    raw: str = ""
+    detail: str = ""
+
+    def as_dict(self) -> dict:
+        return {
+            "name": self.name,
+            "label": self.label,
+            "weak": self.weak,
+            "present": self.present,
+            "value": self.value,
+            "raw": self.raw,
+            "detail": self.detail,
+        }
+
+
+@dataclass
+class FingerprintReport:
+    """一次完整采集的结果。"""
+
+    sources: Dict[str, str] = field(default_factory=dict)
+    details: List[SourceDetail] = field(default_factory=list)
+    warnings: List[str] = field(default_factory=list)
+
+    @property
+    def authoritative(self) -> Dict[str, str]:
+        """可信来源（非 weak）的指纹子集，判定阈值的基数。"""
+        return {k: v for k, v in self.sources.items() if not _BY_NAME[k].weak}
+
+    def as_dict(self) -> dict:
+        return {
+            "sources": dict(self.sources),
+            "details": [d.as_dict() for d in self.details],
+            "warnings": list(self.warnings),
+        }
+
+
+def collect_report() -> FingerprintReport:
+    """采集全部来源并生成明细报告（含诊断告警）。"""
+    report = FingerprintReport()
+    for spec in SOURCES:
+        raw_value: RawValue | None = None
+        try:
+            raw_value = spec.reader()
+        except Exception as exc:  # 单个来源异常绝不拖垮整体采集
+            report.warnings.append(f"来源 {spec.name} 采集异常：{exc}")
+            raw_value = None
+
+        normalized = normalize(raw_value.value) if raw_value else ""
+        detail = SourceDetail(
+            name=spec.name,
+            label=spec.label,
+            weak=spec.weak,
+            present=bool(normalized),
+            value=normalized,
+            raw=(raw_value.value if raw_value else ""),
+            detail=(raw_value.detail if raw_value else ""),
+        )
+        report.details.append(detail)
+        if normalized:
+            report.sources[spec.name] = normalized
+        elif raw_value:
+            # 读到了值但被判为无效（占位/全 0/过短）—— 这是最需要现场排查的情况
+            report.warnings.append(
+                f"来源 {spec.name} 读取到无效值 {raw_value.value!r}，已忽略"
+            )
+
+    if not report.sources:
+        report.warnings.append(
+            "未采集到任何可用硬件指纹：请检查 /proc/device-tree/serial-number、"
+            "/sys/block/mmcblk*/device/cid、/proc/cpuinfo、/sys/class/net 是否可读"
+        )
+    elif len(report.authoritative) < DEFAULT_MIN_MATCH:
+        report.warnings.append(
+            f"可信来源仅 {len(report.authoritative)} 个（< {DEFAULT_MIN_MATCH}），"
+            "判定将降级纳入 machine_id 等易重复来源，建议现场确认硬件序列号是否可读"
+        )
+    if "machine_id" in report.sources:
+        report.warnings.append(
+            "已采集 machine-id：同镜像批量刷机时该值可能全机相同，故不参与判定阈值"
+        )
+    return report
+
+
+def collect_fingerprints() -> Dict[str, str]:
+    """采集本机硬件指纹（来源名 -> 归一化值）。"""
+    return collect_report().sources
+
+
+def machine_code(sources: Dict[str, str]) -> str:
+    """由指纹子集生成展示用机器码。
+
+    只使用**可信来源**（非 weak），并以 ``来源名=值`` 排序后拼接再哈希：
+    同类设备间只要可信来源有任何差异，机器码就不同。
+
+    注意机器码是**单向哈希**，无法反推出各来源指纹。而容错匹配需要逐来源的
+    原始值，所以申请授权时要用 :func:`encode_request_code` 导出申请码，
+    只报机器码则只能签出"按哈希精确匹配"的授权（失去容错，见 read_license）。
+    """
+    usable = {k: v for k, v in sources.items() if not _BY_NAME[k].weak} if sources else {}
+    if not usable:
+        usable = dict(sources)  # 降级：仅有 weak 来源时仍给出机器码并另行告警
+    if not usable:
+        raise FingerprintUnavailableError("未采集到任何硬件指纹，无法生成机器码")
+
+    payload = "\n".join(f"{k}={usable[k]}" for k in sorted(usable))
+    digest = hashlib.sha256(payload.encode("utf-8")).hexdigest().upper()
+    code = digest[:MACHINE_CODE_HEX_LEN]
+    n = 4
+    return "-".join(code[i:i + n] for i in range(0, len(code), n))
+
+
+# ---------------------------------------------------------------------------
+# 申请码：把各来源指纹打包成一段可抄写的文本，供客户回传厂商签发
+# ---------------------------------------------------------------------------
+REQUEST_CODE_PREFIX = "CM1"
+_CHECKSUM_LEN = 8
+
+
+class RequestCodeError(ValueError):
+    """申请码格式错误（抄写不全、被截断或校验和不符）。"""
+
+
+def encode_request_code(sources: Dict[str, str] | None = None) -> str:
+    """把本机指纹打包成申请码（``CM1-<base64url>-<校验和>``）。
+
+    机器码是哈希、无法反推各来源值，而容错匹配需要逐来源原始值 —— 因此首次
+    申请授权时客户应回传**申请码**（一段单行文本），而不是只报机器码。
+
+    申请码**不是**安全凭据：它只是公开硬件标识的传输编码，不参与验签。
+    客户伪造申请码无利可图 —— 签出来的授权只会绑定到他所声称的那台硬件。
+    """
+    data = sources if sources is not None else collect_fingerprints()
+    if not data:
+        raise FingerprintUnavailableError("未采集到任何硬件指纹，无法生成申请码")
+    raw = orjson.dumps({"sources": dict(sorted(data.items())), "machine_code": machine_code(data)})
+    body = base64.urlsafe_b64encode(raw).decode("ascii").rstrip("=")
+    checksum = hashlib.sha256(body.encode("ascii")).hexdigest()[:_CHECKSUM_LEN]
+    return f"{REQUEST_CODE_PREFIX}-{body}-{checksum}"
+
+
+def decode_request_code(text: str) -> dict:
+    """解析申请码，返回 ``{"sources": {...}, "machine_code": "..."}``。
+
+    容忍空白与换行（客户从邮件/截图抄回时常被折行）。
+    """
+    cleaned = "".join(str(text or "").split())
+    if not cleaned:
+        raise RequestCodeError("申请码为空")
+    parts = cleaned.split("-")
+    if len(parts) != 3 or parts[0] != REQUEST_CODE_PREFIX:
+        raise RequestCodeError(
+            f"申请码应以 {REQUEST_CODE_PREFIX}- 开头且由三段构成，实际：{cleaned[:24]}…"
+        )
+    _, body, checksum = parts
+    expected = hashlib.sha256(body.encode("ascii")).hexdigest()[:_CHECKSUM_LEN]
+    if checksum.lower() != expected:
+        raise RequestCodeError("申请码校验和不符（可能抄写不全或被截断），请重新导出")
+    try:
+        padded = body + "=" * (-len(body) % 4)
+        raw = base64.urlsafe_b64decode(padded)
+        data = orjson.loads(raw)
+    except Exception as exc:
+        raise RequestCodeError(f"申请码内容无法解析：{exc}") from exc
+    sources = data.get("sources")
+    if not isinstance(sources, dict) or not sources:
+        raise RequestCodeError("申请码内不含任何指纹来源")
+    return {
+        "sources": {str(k): str(v) for k, v in sources.items()},
+        "machine_code": str(data.get("machine_code", "")),
+    }
+
+
+
+@dataclass
+class MatchResult:
+    """机器指纹比对结果。"""
+
+    ok: bool
+    matched: int
+    total: int
+    reason: str = ""
+    matched_sources: List[str] = field(default_factory=list)
+    mismatched_sources: List[str] = field(default_factory=list)
+    # 降级标记：可信来源不足，本次判定纳入了 weak 来源（安全性下降，需告警）
+    degraded: bool = False
+
+    def as_dict(self) -> dict:
+        return {
+            "ok": self.ok,
+            "matched": self.matched,
+            "total": self.total,
+            "reason": self.reason,
+            "matched_sources": list(self.matched_sources),
+            "mismatched_sources": list(self.mismatched_sources),
+            "degraded": self.degraded,
+        }
+
+
+def match_fingerprints(
+    licensed: Dict[str, str],
+    current: Dict[str, str],
+    min_match: int = DEFAULT_MIN_MATCH,
+) -> MatchResult:
+    """按 N-of-M 容错比对 license 记录与本机采集的指纹。
+
+    只比对**双方都有**的来源（license 签发时某来源缺失、或本机现在读不到该
+    来源，都不计入分母），命中数达到 ``min_match`` 即判定为同一台机器。
+    这样换一块网卡 / 换一颗 eMMC 只会少一票，不会直接失效。
+
+    **weak 来源不参与判定**：``machine_id`` 这类同镜像刷机会重复的值若能计票，
+    等于给"整盘拷贝到另一台机器"白送一票。仅当双方都只采集到 weak 来源时
+    才降级纳入（``degraded=True``），此时安全性与"只靠 machine_id 绑定"等同，
+    调用方应告警。
+    """
+    both = set(licensed) & set(current)
+    if not both:
+        return MatchResult(
+            ok=False, matched=0, total=0,
+            reason="license 记录与本机无任何共同指纹来源",
+        )
+
+    strong = {k for k in both if not _BY_NAME[k].weak}
+    degraded = not strong
+    common = sorted(strong or both)
+
+    matched = [k for k in common if licensed[k] == current[k]]
+    mismatched = [k for k in common if licensed[k] != current[k]]
+
+    # 阈值不能超过可比较的来源数，否则永远无法通过（例如 license 只记录了 2 个来源）
+    threshold = max(1, min(min_match, len(common)))
+    ok = len(matched) >= threshold
+    suffix = "；仅 weak 来源可用，判定已降级" if degraded else ""
+    reason = (
+        f"指纹命中 {len(matched)}/{len(common)}（阈值 {threshold}）{suffix}"
+        if ok else
+        f"指纹命中不足：{len(matched)}/{len(common)}，低于阈值 {threshold}"
+        f"（不匹配来源：{', '.join(mismatched)}）{suffix}"
+    )
+    return MatchResult(
+        ok=ok, matched=len(matched), total=len(common), reason=reason,
+        matched_sources=matched, mismatched_sources=mismatched, degraded=degraded,
+    )

src/chongming_license/models.py

@@ -0,0 +1,91 @@
+"""license 载荷与文件格式。
+
+磁盘格式（``license.lic``）::
+
+    {
+      "payload": {
+        "v": 1,
+        "lic_id": "L2026-0001",
+        "customer": "某某科技",
+        "machine_code": "A3F9-1B77-9E04-2C8D-6F31-0A55",
+        "sources": {"board_serial": "...", "emmc_cid": "...", "mac": "..."},
+        "plugins": ["aibox_sophon_adapter", "ip_sound_column"],
+        "not_before": "2026-09-01",
+        "expires_at": "2027-09-01",
+        "issued_at": "2026-09-01T10:00:00Z",
+        "note": ""
+      },
+      "alg": "ed25519",
+      "kid": "vendor-2026",
+      "sig": "<base64 Ed25519 签名，覆盖 payload 的规范化 JSON>"
+    }
+
+签名覆盖的是 **payload 原始字节的规范化形式**（``canonical_json``，键排序、
+无多余空白），而不是 pydantic 重新序列化的结果 —— 否则任何字段顺序变化或
+默认值填充都会改变字节导致验签失败。因此验签必须在解析模型之前完成。
+"""
+from __future__ import annotations
+
+from typing import Any, Dict, List
+
+from pydantic import BaseModel, ConfigDict, Field
+
+# license 载荷版本。新增**可选**字段不必升版本（签名覆盖原始字节，
+# 旧版本代码会忽略未知字段）；字段语义变更或变为必填时才升。
+LICENSE_VERSION = 1
+
+# 签名算法。目前只支持 Ed25519（非对称，公钥可安全内置/外发）。
+SIGNATURE_ALGORITHM = "ed25519"
+
+# 机器码展示分组长度（4 字符一组，便于人工抄写/口述）
+MACHINE_CODE_GROUP = 4
+
+
+class LicensePayload(BaseModel):
+    """license 载荷：机器绑定 + 插件白名单 + 有效期。
+
+    只允许 str/int/bool/list/dict（不含浮点），以保证 ``canonical_json``
+    在签发端与校验端产生完全相同的字节。
+    """
+
+    model_config = ConfigDict(extra="ignore")
+
+    v: int = Field(default=LICENSE_VERSION, description="载荷版本")
+    lic_id: str = Field(..., description="授权编号（厂商内部唯一）")
+    customer: str = Field(default="", description="被授权客户名称（仅展示/审计）")
+
+    machine_code: str = Field(..., description="机器码（人工核对用，判定以 sources 为准）")
+    sources: Dict[str, str] = Field(
+        default_factory=dict,
+        description="签发时采集到的各硬件来源指纹，用于 N-of-M 容错匹配",
+    )
+
+    plugins: List[str] = Field(
+        default_factory=list,
+        description="被授权的插件 type 列表（空列表 = 不授权任何插件）",
+    )
+
+    not_before: str | None = Field(default=None, description="生效日期 ISO8601（含当天），None = 立即生效")
+    expires_at: str | None = Field(default=None, description="到期日期 ISO8601（含当天），None = 永不过期")
+    issued_at: str | None = Field(default=None, description="签发时间（审计用）")
+    note: str = Field(default="", description="备注（合同号等，仅展示）")
+
+    def plugins_normalized(self) -> set[str]:
+        """归一化插件白名单：去空白、去空项。"""
+        return {str(p).strip() for p in self.plugins if str(p).strip()}
+
+
+class LicenseFile(BaseModel):
+    """``license.lic`` 的完整结构（信封 + 载荷 + 签名）。"""
+
+    model_config = ConfigDict(extra="ignore")
+
+    payload: LicensePayload
+    alg: str = Field(default=SIGNATURE_ALGORITHM, description="签名算法")
+    kid: str = Field(default="", description="签名密钥标识（轮换用，仅提示）")
+    sig: str = Field(default="", description="base64 签名值（覆盖 payload 的规范化 JSON）")
+
+    @staticmethod
+    def from_raw(raw: Dict[str, Any]) -> "LicenseFile":
+        """从已解析的 JSON 字典构造（payload 交给 pydantic 校验）。"""
+        return LicenseFile.model_validate(raw)

src/chongming_license/registry.py

@@ -0,0 +1,496 @@
+"""已签发授权台账（厂商侧）。
+
+解决的问题：签出去的 license 会散落在邮件、聊天记录、下载目录里。等到要续期时，
+既不知道当初给哪台机器签过，也拿不到那台机器的硬件指纹 —— 只能让客户重新导出
+申请码，现场还可能已经联系不上人。
+
+台账把每次签发**连同机器指纹一起**存下来，于是：
+
+- 按客户 / 机器码 / 插件 / 有效期检索
+- 到期前能主动列出"该续期的"清单
+- **续期无需再向客户索取申请码** —— 直接复用台账里的指纹重新签发（这是最大价值）
+- 原样导出当初签发的文件（审计 / 补发）
+- 识别"哪份授权是轮换前的旧公钥签的"（密钥轮换后需要重签）
+
+存储是单文件 SQLite（纯标准库）：无服务、可拷贝、可以直接丢进备份。
+
+**为什么放在 chongming-license 库而不是 GUI 包**：CLI 也要能批量 `list` / `renew` /
+`export`（几十上百台的续期适合脚本化），而 CLI 不能去 import 带 PySide6 的 GUI 包。
+代价是网关产物里会多带这个纯标准库模块（无第三方依赖，可忽略）。
+
+**注意**：本库文件含客户设备的硬件指纹与授权信息，属厂商内部数据，不宜外发。
+"""
+from __future__ import annotations
+
+import csv
+import json
+import os
+import sqlite3
+from dataclasses import dataclass, field
+from datetime import date, timedelta
+from pathlib import Path
+from threading import RLock
+from typing import Any, Dict, Iterable, List, Mapping, Sequence
+
+from .issue import utc_today
+
+# 台账库结构版本。变更时请同时提供迁移或导出方案（本模块不做自动迁移）。
+SCHEMA_VERSION = 1
+
+# 默认"即将到期"窗口（天）
+DEFAULT_EXPIRING_DAYS = 30
+
+# 记录状态码 -> 中文说明
+STATUS_LABELS = {
+    "forever": "永久有效",
+    "active": "有效",
+    "expiring": "即将到期",
+    "expired": "已过期",
+    "not_yet_valid": "未生效",
+}
+
+_SCHEMA = """
+CREATE TABLE IF NOT EXISTS issued_licenses (
+    lic_id          TEXT PRIMARY KEY,
+    machine_code    TEXT NOT NULL,
+    customer        TEXT NOT NULL DEFAULT '',
+    note            TEXT NOT NULL DEFAULT '',
+    plugins_json    TEXT NOT NULL,
+    sources_json    TEXT NOT NULL,
+    not_before      TEXT,
+    expires_at      TEXT,
+    issued_at       TEXT NOT NULL DEFAULT '',
+    kid             TEXT NOT NULL DEFAULT '',
+    public_key_fp   TEXT NOT NULL DEFAULT '',
+    license_json    TEXT NOT NULL,
+    source          TEXT NOT NULL DEFAULT '',
+    created_at      TEXT NOT NULL,
+    updated_at      TEXT NOT NULL
+);
+CREATE INDEX IF NOT EXISTS idx_lic_machine ON issued_licenses(machine_code);
+CREATE INDEX IF NOT EXISTS idx_lic_customer ON issued_licenses(customer);
+CREATE INDEX IF NOT EXISTS idx_lic_expires ON issued_licenses(expires_at);
+"""
+
+
+class RegistryError(Exception):
+    """台账读写错误（含结构版本不匹配）。"""
+
+
+def default_registry_path() -> Path:
+    """默认台账库位置：``~/.chongming/licenses.db``。
+
+    刻意放在用户主目录而非项目内：台账含客户设备指纹，且**必须**能跨项目、
+    跨工具长期留存（放在仓库里容易被 .gitignore 忽略、被清理脚本删掉、
+    或在换机器时随项目一起丢掉）。可用 ``CHONGMING_LICENSE_DB`` 环境变量或
+    ``--db`` 参数覆盖。
+    """
+    env = os.environ.get("CHONGMING_LICENSE_DB", "").strip()
+    if env:
+        return Path(env).expanduser()
+    return Path.home() / ".chongming" / "licenses.db"
+
+
+def _now_iso() -> str:
+    from datetime import datetime, timezone
+
+    return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
+
+
+def _clip_date(value: Any) -> str | None:
+    """只保留 ISO8601 的日期部分（校验端同样只读前 10 字符）。
+
+    统一在这里裁剪，避免台账里混入 "2027-09-01T00:00:00Z" 这类写法 ——
+    否则按日期排序/比较（SQL 里是字符串比较）会得出错误结果。
+    """
+    if value is None:
+        return None
+    text = str(value).strip()
+    return text[:10] or None
+
+
+@dataclass
+class LicenseRecord:
+    """台账中的一条签发记录。"""
+
+    lic_id: str
+    machine_code: str
+    customer: str = ""
+    note: str = ""
+    plugins: List[str] = field(default_factory=list)
+    sources: Dict[str, str] = field(default_factory=dict)
+    not_before: str | None = None
+    expires_at: str | None = None
+    issued_at: str = ""
+    kid: str = ""
+    public_key_fp: str = ""
+    license_json: str = ""
+    source: str = ""
+    created_at: str = ""
+    updated_at: str = ""
+
+    # ---- 派生信息 ----------------------------------------------------
+    def status(self, today: date | None = None,
+               expiring_days: int = DEFAULT_EXPIRING_DAYS) -> str:
+        """状态：expired / not_yet_valid / expiring / active / forever。"""
+        ref = today or utc_today()
+        if self.not_before:
+            try:
+                if date.fromisoformat(self.not_before) > ref:
+                    return "not_yet_valid"
+            except ValueError:
+                pass
+        if not self.expires_at:
+            return "forever"
+        try:
+            expires = date.fromisoformat(self.expires_at)
+        except ValueError:
+            return "active"
+        if expires < ref:
+            return "expired"
+        if expires <= ref + timedelta(days=expiring_days):
+            return "expiring"
+        return "active"
+
+    def status_label(self, today: date | None = None,
+                     expiring_days: int = DEFAULT_EXPIRING_DAYS) -> str:
+        return STATUS_LABELS.get(self.status(today, expiring_days), "未知")
+
+    def days_left(self, today: date | None = None) -> int | None:
+        """剩余天数；无到期日返回 None，已过期返回负数。"""
+        if not self.expires_at:
+            return None
+        try:
+            return (date.fromisoformat(self.expires_at) - (today or utc_today())).days
+        except ValueError:
+            return None
+
+    def license_doc(self) -> Dict[str, Any]:
+        """还原完整 license 文档（即当初签发的文件内容）。"""
+        return json.loads(self.license_json) if self.license_json else {}
+
+    def summary(self) -> str:
+        """一行摘要（日志/CSV 用）。"""
+        plugins = ",".join(self.plugins)
+        return (
+            f"{self.lic_id} | {self.customer or '(未填客户)'} | {self.machine_code} "
+            f"| {plugins} | {self.not_before or '立即'} → {self.expires_at or '永久'}"
+        )
+
+
+def record_from_license(
+    license_doc: Mapping[str, Any],
+    *,
+    public_key_fp: str = "",
+    source: str = "",
+) -> LicenseRecord:
+    """由 license 文档派生台账记录（**唯一的写入口**）。
+
+    所有查询用的列都从文档派生，因此不会与 license_json 不一致；
+    调用方不要手工拼 LicenseRecord 去写库。
+    """
+    payload = dict(license_doc.get("payload") or {})
+    lic_id = str(payload.get("lic_id") or "").strip()
+    if not lic_id:
+        raise RegistryError("license 缺少 lic_id，无法入库")
+    now = _now_iso()
+    return LicenseRecord(
+        lic_id=lic_id,
+        machine_code=str(payload.get("machine_code") or "").strip().upper(),
+        customer=str(payload.get("customer") or ""),
+        note=str(payload.get("note") or ""),
+        plugins=[str(p) for p in (payload.get("plugins") or [])],
+        sources={str(k): str(v) for k, v in (payload.get("sources") or {}).items()},
+        not_before=_clip_date(payload.get("not_before")),
+        expires_at=_clip_date(payload.get("expires_at")),
+        issued_at=str(payload.get("issued_at") or ""),
+        kid=str(license_doc.get("kid") or ""),
+        public_key_fp=public_key_fp,
+        license_json=json.dumps(dict(license_doc), ensure_ascii=False),
+        source=source,
+        created_at=now,
+        updated_at=now,
+    )
+
+
+class Registry:
+    """已签发授权台账（SQLite 单文件）。
+
+    与其他实例并发访问安全（WAL + busy_timeout），因此可以同时开多个 GUI 窗口。
+    """
+
+    def __init__(self, path: str | Path) -> None:
+        self.path = Path(path)
+        self.path.parent.mkdir(parents=True, exist_ok=True)
+        self._lock = RLock()
+        self._conn = sqlite3.connect(str(self.path), check_same_thread=False)
+        self._conn.row_factory = sqlite3.Row
+        self._conn.execute("PRAGMA journal_mode=WAL")
+        self._conn.execute("PRAGMA busy_timeout=5000")
+        self._conn.executescript(_SCHEMA)
+        self._conn.commit()
+        self._check_schema_version()
+
+    # ---- 生命周期 ----------------------------------------------------
+    def _check_schema_version(self) -> None:
+        """结构版本不匹配时明确报错，而不是让后续查询以诡异方式失败。"""
+        current = self._conn.execute("PRAGMA user_version").fetchone()[0]
+        if current == 0:
+            self._conn.execute(f"PRAGMA user_version = {SCHEMA_VERSION}")
+            self._conn.commit()
+        elif current != SCHEMA_VERSION:
+            raise RegistryError(
+                f"台账库结构版本为 {current}，本程序支持 {SCHEMA_VERSION}：{self.path}\n"
+                "该库由更新/更旧版本的工具创建。请先导出台账（export）后换用匹配的版本。"
+            )
+
+    def close(self) -> None:
+        with self._lock:
+            self._conn.close()
+
+    def __enter__(self) -> "Registry":
+        return self
+
+    def __exit__(self, *exc) -> None:
+        """退出 ``with`` 时**关闭连接**。
+
+        ``with Registry(path) as reg:`` 是"用完即关"的语义，适合 CLI 这类
+        「打开 → 操作 → 结束」的短生命周期用法。
+
+        ⚠️ 长生命周期场景（例如 GUI 里被多个页面共享的台账）**不要**用 ``with``
+        —— 它会把连接提前关掉，之后任何操作都报
+        ``Cannot operate on a closed database``。那种场景应由持有连接的一方
+        （见 GUI 的 ``RegistryHolder``）显式管理生命周期。
+        """
+        self.close()
+
+    # ---- 写 ----------------------------------------------------------
+    def save(self, license_doc: Mapping[str, Any], *,
+             public_key_fp: str = "", source: str = "",
+             overwrite_note: bool = False) -> LicenseRecord:
+        """写入/覆盖一条记录（按 lic_id 幂等）。
+
+        **备注的处理**：载荷里的 `note` 是签发时写进**签名内容**的字段，而台账的
+        `note` 是厂商自己的人工批注（"已电话确认续期"、"设备已返修"…）。二者同名但
+        不同用途，因此覆盖已有记录时**默认保留台账里的人工批注**，不被重新导入冲掉；
+        需要让载荷备注覆盖批注时显式传 ``overwrite_note=True``。
+        """
+        record = record_from_license(license_doc, public_key_fp=public_key_fp, source=source)
+        with self._lock:
+            old = self.get(record.lic_id)
+            if old is not None:
+                if not overwrite_note:
+                    record.note = old.note
+                record.created_at = old.created_at
+                record.updated_at = _now_iso()
+            self._conn.execute(
+                """
+                INSERT INTO issued_licenses (
+                    lic_id, machine_code, customer, note, plugins_json, sources_json,
+                    not_before, expires_at, issued_at, kid, public_key_fp, license_json,
+                    source, created_at, updated_at
+                ) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)
+                ON CONFLICT(lic_id) DO UPDATE SET
+                    machine_code=excluded.machine_code,
+                    customer=excluded.customer,
+                    note=excluded.note,
+                    plugins_json=excluded.plugins_json,
+                    sources_json=excluded.sources_json,
+                    not_before=excluded.not_before,
+                    expires_at=excluded.expires_at,
+                    issued_at=excluded.issued_at,
+                    kid=excluded.kid,
+                    public_key_fp=excluded.public_key_fp,
+                    license_json=excluded.license_json,
+                    source=excluded.source,
+                    updated_at=excluded.updated_at
+                """,
+                (
+                    record.lic_id, record.machine_code, record.customer, record.note,
+                    json.dumps(record.plugins, ensure_ascii=False),
+                    json.dumps(record.sources, ensure_ascii=False),
+                    record.not_before, record.expires_at, record.issued_at, record.kid,
+                    record.public_key_fp, record.license_json, record.source,
+                    record.created_at, record.updated_at,
+                ),
+            )
+            self._conn.commit()
+        return record
+
+    def update_note(self, lic_id: str, note: str) -> bool:
+        """只改备注（台账里唯一允许手工编辑、且不会被重新导入覆盖的字段）。"""
+        with self._lock:
+            cur = self._conn.execute(
+                "UPDATE issued_licenses SET note=?, updated_at=? WHERE lic_id=?",
+                (note, _now_iso(), lic_id),
+            )
+            self._conn.commit()
+            return cur.rowcount > 0
+
+    def delete(self, lic_id: str) -> bool:
+        with self._lock:
+            cur = self._conn.execute("DELETE FROM issued_licenses WHERE lic_id=?", (lic_id,))
+            self._conn.commit()
+            return cur.rowcount > 0
+
+    # ---- 读 ----------------------------------------------------------
+    @staticmethod
+    def _row_to_record(row: sqlite3.Row) -> LicenseRecord:
+        return LicenseRecord(
+            lic_id=row["lic_id"],
+            machine_code=row["machine_code"],
+            customer=row["customer"],
+            note=row["note"],
+            plugins=json.loads(row["plugins_json"] or "[]"),
+            sources=json.loads(row["sources_json"] or "{}"),
+            not_before=row["not_before"],
+            expires_at=row["expires_at"],
+            issued_at=row["issued_at"],
+            kid=row["kid"],
+            public_key_fp=row["public_key_fp"],
+            license_json=row["license_json"],
+            source=row["source"],
+            created_at=row["created_at"],
+            updated_at=row["updated_at"],
+        )
+
+    def get(self, lic_id: str) -> LicenseRecord | None:
+        with self._lock:
+            row = self._conn.execute(
+                "SELECT * FROM issued_licenses WHERE lic_id=?", (lic_id,)
+            ).fetchone()
+        return self._row_to_record(row) if row else None
+
+    def query(
+        self,
+        *,
+        customer: str | None = None,
+        machine_code: str | None = None,
+        plugin: str | None = None,
+        status: str | None = None,
+        keyword: str | None = None,
+        expiring_days: int = DEFAULT_EXPIRING_DAYS,
+        today: date | None = None,
+        limit: int | None = None,
+    ) -> List[LicenseRecord]:
+        """按条件检索，按到期日升序（无到期日的排最后）。
+
+        Args:
+            status: expired / expiring / active / forever / not_yet_valid
+            keyword: 模糊匹配 授权编号 / 机器码 / 客户 / 备注
+
+        状态筛选刻意放在 Python 侧（用 :meth:`LicenseRecord.status`）而不是写 SQL 条件：
+        状态有优先级（未生效 > 已过期 > 即将到期 > 其余），SQL 里各条件会互相重叠
+        （例如"未生效且到期日在 30 天内"会被 active 和 expiring 同时命中），
+        导致 stats() 各项之和不等于总数。复用同一个 status() 才是唯一自洽的做法，
+        台账规模（厂商手里几百到几千条）也完全不需要把它下推到 SQL。
+        """
+        ref = today or utc_today()
+        where: List[str] = []
+        args: List[Any] = []
+
+        if customer:
+            where.append("customer LIKE ?")
+            args.append(f"%{customer}%")
+        if machine_code:
+            # 机器码统一大写存储，这里同样大写以便大小写不敏感
+            where.append("machine_code LIKE ?")
+            args.append(f"%{machine_code.strip().upper()}%")
+        if plugin:
+            # 插件列表是 JSON 数组文本，用带引号的子串匹配避免 "doubao" 命中 "doubao_extended"
+            where.append("plugins_json LIKE ?")
+            args.append(f'%"{plugin}"%')
+        if keyword:
+            like = f"%{keyword}%"
+            where.append("(lic_id LIKE ? OR machine_code LIKE ? OR customer LIKE ? OR note LIKE ?)")
+            args.extend([like, like, like, like])
+
+        sql = "SELECT * FROM issued_licenses"
+        if where:
+            sql += " WHERE " + " AND ".join(where)
+        # 到期日升序；NULL（永久）用哨兵排到最后
+        sql += " ORDER BY COALESCE(expires_at, '9999-12-31') ASC, lic_id ASC"
+
+        with self._lock:
+            rows = self._conn.execute(sql, args).fetchall()
+        records = [self._row_to_record(r) for r in rows]
+
+        if status:
+            records = [r for r in records
+                       if r.status(today=ref, expiring_days=expiring_days) == status]
+        if limit:
+            records = records[:int(limit)]
+        return records
+
+    def find_by_machine_code(self, machine_code: str) -> List[LicenseRecord]:
+        """某台机器的全部授权（含历史，用于续期前确认上次签了什么）。"""
+        return self.query(machine_code=machine_code)
+
+    def stats(self, *, expiring_days: int = DEFAULT_EXPIRING_DAYS,
+              today: date | None = None) -> Dict[str, int]:
+        """台账概览计数。"""
+        return {
+            "total": len(self.query(today=today)),
+            "expired": len(self.query(status="expired", today=today)),
+            "expiring": len(self.query(status="expiring", expiring_days=expiring_days,
+                                      today=today)),
+            "active": len(self.query(status="active", expiring_days=expiring_days,
+                                     today=today)),
+            "forever": len(self.query(status="forever", today=today)),
+            "not_yet_valid": len(self.query(status="not_yet_valid", today=today)),
+        }
+
+    # ---- 导出 --------------------------------------------------------
+    def export_licenses(
+        self,
+        out_dir: str | Path,
+        records: Sequence[LicenseRecord] | None = None,
+        *,
+        subdir_by_customer: bool = False,
+    ) -> List[Path]:
+        """把台账里的授权原样导出为 .lic 文件（归档 / 补发）。
+
+        文件名形如 ``<lic_id>_<机器码>.lic``；可选按客户名分子目录，
+        便于直接打包发给对应客户。
+        """
+        base = Path(out_dir)
+        base.mkdir(parents=True, exist_ok=True)
+        written: List[Path] = []
+        for rec in (records if records is not None else self.query()):
+            target_dir = base
+            if subdir_by_customer:
+                # 客户名可能含路径分隔符等非法字符，做一次保守替换
+                safe = "".join(c if c.isalnum() or c in " ._-" else "_"
+                               for c in (rec.customer or "未填客户")).strip() or "未填客户"
+                target_dir = base / safe
+                target_dir.mkdir(parents=True, exist_ok=True)
+            path = target_dir / f"{rec.lic_id}_{rec.machine_code or 'nomachine'}.lic"
+            path.write_text(rec.license_json, encoding="utf-8")
+            written.append(path)
+        return written
+
+    def export_csv(self, path: str | Path,
+                   records: Iterable[LicenseRecord] | None = None,
+                   *, today: date | None = None) -> Path:
+        """导出台账清单为 CSV（Excel 可直接打开，用于对账/商务统计）。
+
+        以 UTF-8 BOM 写入：否则 Excel 在中文 Windows 下会按 GBK 解码而乱码。
+        """
+        target = Path(path)
+        target.parent.mkdir(parents=True, exist_ok=True)
+        rows = list(records if records is not None else self.query())
+        with target.open("w", encoding="utf-8-sig", newline="") as f:
+            writer = csv.writer(f)
+            writer.writerow([
+                "授权编号", "客户", "机器码", "授权插件", "状态", "剩余天数",
+                "生效日期", "到期日期", "签发时间", "密钥标识", "公钥指纹", "备注",
+            ])
+            for rec in rows:
+                writer.writerow([
+                    rec.lic_id, rec.customer, rec.machine_code,
+                    ",".join(rec.plugins), rec.status_label(today=today),
+                    "" if rec.days_left(today=today) is None else rec.days_left(today=today),
+                    rec.not_before or "", rec.expires_at or "", rec.issued_at,
+                    rec.kid, rec.public_key_fp, rec.note,
+                ])
+        return target

src/chongming_license/verify.py

@@ -0,0 +1,493 @@
+"""license 验签与判定。
+
+**关键设计**：签名覆盖 payload 的规范化 JSON（``canonical_json``），而签发
+（``sign_payload``）与校验（``verify_signature``）共用同一个函数 —— 否则两端
+序列化差异会导致"自己签的自己验不过"。因此本模块同时包含签名与验签逻辑，
+是刻意为之：规范化只有一处实现。
+
+非对称密钥选型：Ed25519（``cryptography``），公钥可安全内置进网关配置或
+随包发布。注意根仓 ``resources/*.key`` 那套 ``.mbank`` 验签是 **HMAC 对称**
+且密钥与银行同目录发布，只能算防篡改/混淆，不能作为厂商信任根 —— 授权必须
+用非对称。
+"""
+from __future__ import annotations
+
+import base64
+import binascii
+import hashlib
+import json
+import os
+import stat
+from dataclasses import dataclass, field
+from datetime import date, datetime, timezone
+from pathlib import Path
+from typing import Any, Dict, List
+
+import orjson
+from cryptography.exceptions import InvalidSignature
+from cryptography.hazmat.primitives import serialization
+from cryptography.hazmat.primitives.asymmetric.ed25519 import (
+    Ed25519PrivateKey,
+    Ed25519PublicKey,
+)
+
+from .machine import (
+    DEFAULT_MIN_MATCH,
+    FingerprintUnavailableError,
+    MatchResult,
+    collect_fingerprints,
+    machine_code,
+    match_fingerprints,
+)
+from .models import LICENSE_VERSION, SIGNATURE_ALGORITHM, LicenseFile, LicensePayload
+
+# ---------------------------------------------------------------------------
+# 判定原因码（前端/日志按码分支，文案可改）
+# ---------------------------------------------------------------------------
+REASON_OK = "ok"
+REASON_LICENSE_MISSING = "license_missing"
+REASON_LICENSE_MALFORMED = "license_malformed"
+REASON_SIGNATURE_INVALID = "signature_invalid"
+REASON_UNSUPPORTED_ALG = "unsupported_alg"
+REASON_NO_PUBLIC_KEY = "no_public_key"
+REASON_MACHINE_MISMATCH = "machine_mismatch"
+REASON_NOT_YET_VALID = "not_yet_valid"
+REASON_EXPIRED = "expired"
+REASON_CLOCK_ROLLBACK = "clock_rollback"
+REASON_PLUGIN_NOT_LICENSED = "plugin_not_licensed"
+REASON_FINGERPRINT_UNAVAILABLE = "fingerprint_unavailable"
+REASON_DISABLED = "disabled"
+
+# 原因码 -> 中文说明（日志与前端共用）
+REASON_TEXT = {
+    REASON_OK: "授权有效",
+    REASON_LICENSE_MISSING: "未找到授权文件",
+    REASON_LICENSE_MALFORMED: "授权文件格式错误或内容损坏",
+    REASON_SIGNATURE_INVALID: "授权文件签名校验失败（可能被篡改或非本厂商签发）",
+    REASON_UNSUPPORTED_ALG: "授权文件使用了不支持的签名算法",
+    REASON_NO_PUBLIC_KEY: "未配置厂商公钥，无法校验授权",
+    REASON_MACHINE_MISMATCH: "机器码不匹配（授权文件不是为本机签发的）",
+    REASON_NOT_YET_VALID: "授权尚未生效",
+    REASON_EXPIRED: "授权已过期",
+    REASON_CLOCK_ROLLBACK: "检测到系统时间被回拨，授权校验中止",
+    REASON_PLUGIN_NOT_LICENSED: "该插件不在授权白名单内",
+    REASON_FINGERPRINT_UNAVAILABLE: "本机无法采集硬件指纹",
+    REASON_DISABLED: "授权校验未启用（开发/测试模式）",
+}
+
+
+class LicenseError(Exception):
+    """license 校验相关的基类异常。"""
+
+
+class LicenseMalformedError(LicenseError):
+    """文件不是合法 JSON / 缺少必要字段。"""
+
+
+class SignatureInvalidError(LicenseError):
+    """签名校验失败。"""
+
+
+class PublicKeyError(LicenseError):
+    """公钥缺失或格式无法解析。"""
+
+
+# ---------------------------------------------------------------------------
+# 规范化与签名
+# ---------------------------------------------------------------------------
+def canonical_json(payload: Any) -> bytes:
+    """payload 的规范化字节：键排序、无多余空白。
+
+    签发与校验**必须**共用本函数（见模块文档）。载荷内不允许浮点数
+    （float 的文本表示存在多种合法写法，会破坏字节级一致性）。
+    """
+    return orjson.dumps(payload, option=orjson.OPT_SORT_KEYS)
+
+
+def load_public_key(key: str) -> Ed25519PublicKey:
+    """解析公钥：支持 base64 裸 32 字节，或 PEM。
+
+    优先按 base64 裸字节解析，因为配置里放一行 base64 最便于运维；
+    解析不出再当作 PEM（带 ``-----BEGIN`` 头）。
+    """
+    if not key or not key.strip():
+        raise PublicKeyError("未提供公钥")
+    text = key.strip()
+
+    if "-----BEGIN" in text:
+        try:
+            return serialization.load_pem_public_key(text.encode("utf-8"))  # type: ignore[return-value]
+        except Exception as exc:
+            raise PublicKeyError(f"PEM 公钥解析失败：{exc}") from exc
+
+    try:
+        raw = base64.b64decode(text, validate=True)
+    except (binascii.Error, ValueError) as exc:
+        raise PublicKeyError(f"公钥既不是合法 PEM 也不是合法 base64：{exc}") from exc
+    if len(raw) != 32:
+        raise PublicKeyError(
+            f"Ed25519 公钥应为 32 字节，实际 {len(raw)} 字节（base64 内容有误？）"
+        )
+    return Ed25519PublicKey.from_public_bytes(raw)
+
+
+def public_key_fingerprint(key: str) -> str:
+    """公钥指纹（短十六进制），用于日志确认"当前内置的是哪把公钥"。"""
+    try:
+        raw = load_public_key(key).public_bytes(
+            encoding=serialization.Encoding.Raw,
+            format=serialization.PublicFormat.Raw,
+        )
+    except PublicKeyError:
+        return ""
+    return hashlib.sha256(raw).hexdigest()[:16]
+
+
+def verify_signature(
+    payload_raw: Dict[str, Any],
+    sig_b64: str,
+    public_key: str,
+    alg: str = SIGNATURE_ALGORITHM,
+) -> None:
+    """校验 payload 的签名；失败抛 :class:`SignatureInvalidError`。
+
+    ``payload_raw`` 必须是**从文件中直接解析出的原始字典**，不能是 pydantic
+    重新序列化的结果（默认值填充/字段顺序变化都会改变字节）。
+    """
+    if alg != SIGNATURE_ALGORITHM:
+        raise SignatureInvalidError(f"不支持的签名算法：{alg}（仅支持 {SIGNATURE_ALGORITHM}）")
+    if not sig_b64:
+        raise SignatureInvalidError("缺少签名值")
+    try:
+        sig = base64.b64decode(sig_b64, validate=True)
+    except (binascii.Error, ValueError) as exc:
+        raise SignatureInvalidError(f"签名值不是合法 base64：{exc}") from exc
+
+    pub = load_public_key(public_key)
+    try:
+        pub.verify(sig, canonical_json(payload_raw))
+    except InvalidSignature as exc:
+        raise SignatureInvalidError("签名校验失败") from exc
+
+
+def sign_payload(
+    payload_raw: Dict[str, Any],
+    private_key: Ed25519PrivateKey,
+) -> str:
+    """对 payload 签名，返回 base64 签名值（厂商侧签发工具用）。"""
+    return base64.b64encode(private_key.sign(canonical_json(payload_raw))).decode("ascii")
+
+
+def load_private_key(path: str | Path) -> Ed25519PrivateKey:
+    """从 PEM 文件读取 Ed25519 私钥（厂商侧，私钥绝不进部署产物）。"""
+    data = Path(path).read_bytes()
+    key = serialization.load_pem_private_key(data, password=None)
+    if not isinstance(key, Ed25519PrivateKey):
+        raise LicenseError(f"{path} 不是 Ed25519 私钥")
+    return key
+
+
+def public_key_b64(private_key: Ed25519PrivateKey) -> str:
+    """私钥对应的公钥，base64 裸 32 字节 —— 可直接填进 ``[license] public_key``。"""
+    raw = private_key.public_key().public_bytes(
+        encoding=serialization.Encoding.Raw,
+        format=serialization.PublicFormat.Raw,
+    )
+    return base64.b64encode(raw).decode("ascii")
+
+
+def public_key_pem(private_key: Ed25519PrivateKey) -> str:
+    """私钥对应的公钥 PEM（便于贴进工单/邮件，或改用 PEM 形式配置）。"""
+    return private_key.public_key().public_bytes(
+        encoding=serialization.Encoding.PEM,
+        format=serialization.PublicFormat.SubjectPublicKeyInfo,
+    ).decode("ascii")
+
+
+def generate_keypair() -> tuple[Ed25519PrivateKey, str]:
+    """生成 Ed25519 密钥对，返回 ``(私钥, 公钥 base64)``。"""
+    private_key = Ed25519PrivateKey.generate()
+    return private_key, public_key_b64(private_key)
+
+
+def write_private_key(
+    private_key: Ed25519PrivateKey,
+    path: str | Path,
+    *,
+    overwrite: bool = False,
+) -> Path:
+    """把私钥写为**不加密**的 PKCS#8 PEM，文件权限置 ``0600``。
+
+    不加密是有意为之：这个私钥进不了制品、只存在厂商的签发机器上，加口令会让
+    自动化签发（批量签几百台设备）变得不可用，而防护收益有限（口令多半就写在
+    同一个脚本里）。真正的防线是**文件权限 + 不落地到产物 + 离线保管**。
+
+    Args:
+        overwrite: False 时目标已存在则抛 :class:`LicenseError`
+    """
+    target = Path(path)
+    if target.exists() and not overwrite:
+        raise LicenseError(f"{target} 已存在（确要覆盖请显式允许）")
+    target.parent.mkdir(parents=True, exist_ok=True)
+    target.write_bytes(
+        private_key.private_bytes(
+            encoding=serialization.Encoding.PEM,
+            format=serialization.PrivateFormat.PKCS8,
+            encryption_algorithm=serialization.NoEncryption(),
+        )
+    )
+    # 私钥文件权限收到 0600：同机器上的其它用户不应能读到
+    os.chmod(target, stat.S_IRUSR | stat.S_IWUSR)
+    return target
+
+
+# ---------------------------------------------------------------------------
+# 日期
+# ---------------------------------------------------------------------------
+def _parse_date(value: str | None) -> date | None:
+    """解析 ISO8601 日期（容忍 "2027-09-01" / "2027-09-01T00:00:00Z" 两种写法）。"""
+    if not value:
+        return None
+    text = str(value).strip()
+    if not text:
+        return None
+    try:
+        return date.fromisoformat(text[:10])
+    except ValueError as exc:
+        raise LicenseMalformedError(f"日期格式无法解析：{value!r}") from exc
+
+
+def _today() -> date:
+    return datetime.now(timezone.utc).date()
+
+
+# ---------------------------------------------------------------------------
+# 判定结果
+# ---------------------------------------------------------------------------
+@dataclass
+class LicenseStatus:
+    """整份 license 的校验结论（与"哪个插件被授权"无关）。"""
+
+    ok: bool
+    code: str
+    message: str
+    payload: LicensePayload | None = None
+    license_path: str = ""
+    machine_code: str = ""
+    match: MatchResult | None = None
+    public_key_fingerprint: str = ""
+    sources: Dict[str, str] = field(default_factory=dict)
+    warnings: List[str] = field(default_factory=list)
+
+    @property
+    def licensed_plugins(self) -> set[str]:
+        return self.payload.plugins_normalized() if self.payload else set()
+
+    def as_dict(self) -> dict:
+        """给 /license/info 的可序列化结构。"""
+        return {
+            "ok": self.ok,
+            "code": self.code,
+            "message": self.message,
+            "license_path": self.license_path,
+            "machine_code": self.machine_code,
+            "public_key_fingerprint": self.public_key_fingerprint,
+            "sources": dict(self.sources),
+            "warnings": list(self.warnings),
+            "match": self.match.as_dict() if self.match else None,
+            "payload": self.payload.model_dump() if self.payload else None,
+            "licensed_plugins": sorted(self.licensed_plugins),
+        }
+
+
+@dataclass
+class PluginDecision:
+    """单个插件是否被授权运行的结论。"""
+
+    allowed: bool
+    code: str
+    message: str
+    plugin_type: str = ""
+
+    def as_dict(self) -> dict:
+        return {
+            "allowed": self.allowed,
+            "code": self.code,
+            "message": self.message,
+            "plugin_type": self.plugin_type,
+        }
+
+
+# ---------------------------------------------------------------------------
+# 整份 license 的读取与校验
+# ---------------------------------------------------------------------------
+def read_license(
+    license_path: str | Path,
+    public_key: str,
+    *,
+    min_match: int = DEFAULT_MIN_MATCH,
+    sources: Dict[str, str] | None = None,
+    today: date | None = None,
+) -> LicenseStatus:
+    """读取并校验 license 文件：签名 → 机器指纹 → 有效期。
+
+    调用方负责时钟回拨检查（见 :mod:`chongming_license.clock`），本函数只做
+    纯函数式的判定，便于测试。
+
+    Args:
+        license_path: license.lic 路径
+        public_key: 厂商公钥（base64 裸字节或 PEM）
+        min_match: 指纹命中阈值
+        sources: 本机指纹；None = 现场采集（测试可注入）
+        today: 判定基准日期；None = 当前 UTC 日期
+
+    Returns:
+        LicenseStatus：``ok`` 表示这份 license 对本机有效（与插件白名单无关）
+    """
+    path = Path(license_path)
+    key_fp = public_key_fingerprint(public_key)
+
+    # 1. 本机指纹
+    if sources is None:
+        try:
+            sources = collect_fingerprints()
+        except Exception as exc:
+            return LicenseStatus(
+                ok=False, code=REASON_FINGERPRINT_UNAVAILABLE,
+                message=f"{REASON_TEXT[REASON_FINGERPRINT_UNAVAILABLE]}：{exc}",
+                license_path=str(path), public_key_fingerprint=key_fp,
+            )
+    try:
+        current_code = machine_code(sources)
+    except FingerprintUnavailableError as exc:
+        return LicenseStatus(
+            ok=False, code=REASON_FINGERPRINT_UNAVAILABLE,
+            message=f"{REASON_TEXT[REASON_FINGERPRINT_UNAVAILABLE]}：{exc}",
+            license_path=str(path), public_key_fingerprint=key_fp,
+        )
+
+    def _fail(code: str, extra: str = "", **kw) -> LicenseStatus:
+        msg = REASON_TEXT.get(code, code)
+        return LicenseStatus(
+            ok=False, code=code,
+            message=f"{msg}{('：' + extra) if extra else ''}",
+            license_path=str(path), machine_code=current_code,
+            public_key_fingerprint=key_fp, sources=dict(sources or {}), **kw,
+        )
+
+    # 2. 公钥（未配置时后续无法验签，提前给出明确原因）
+    if not key_fp:
+        return _fail(REASON_NO_PUBLIC_KEY)
+
+    # 3. 读文件
+    if not path.is_file():
+        return _fail(REASON_LICENSE_MISSING, str(path))
+    try:
+        raw = json.loads(path.read_bytes())
+    except (OSError, ValueError) as exc:
+        return _fail(REASON_LICENSE_MALFORMED, str(exc))
+    if not isinstance(raw, dict) or "payload" not in raw:
+        return _fail(REASON_LICENSE_MALFORMED, "缺少 payload 字段")
+
+    payload_raw = raw.get("payload")
+    if not isinstance(payload_raw, dict):
+        return _fail(REASON_LICENSE_MALFORMED, "payload 不是对象")
+
+    # 4. 验签（必须先于模型解析：签名覆盖的是原始字节）
+    alg = str(raw.get("alg", SIGNATURE_ALGORITHM))
+    if alg != SIGNATURE_ALGORITHM:
+        return _fail(REASON_UNSUPPORTED_ALG, alg)
+    try:
+        verify_signature(payload_raw, str(raw.get("sig", "")), public_key, alg)
+    except SignatureInvalidError as exc:
+        return _fail(REASON_SIGNATURE_INVALID, str(exc))
+
+    # 5. 解析载荷（签名已验证，此后的内容可信）
+    try:
+        lic = LicenseFile.from_raw(raw)
+        payload = lic.payload
+    except Exception as exc:
+        return _fail(REASON_LICENSE_MALFORMED, str(exc))
+
+    if payload.v > LICENSE_VERSION:
+        return _fail(
+            REASON_LICENSE_MALFORMED,
+            f"载荷版本 {payload.v} 高于本程序支持的 {LICENSE_VERSION}，请升级网关",
+        )
+
+    # 6. 机器指纹比对
+    #    license 记录了逐来源指纹 → N-of-M 容错匹配；
+    #    只记录了机器码（客户仅报了一个短码）→ 退化为哈希精确匹配。
+    if payload.sources:
+        match = match_fingerprints(payload.sources, sources, min_match=min_match)
+    else:
+        same = bool(payload.machine_code) and payload.machine_code == current_code
+        match = MatchResult(
+            ok=same, matched=1 if same else 0, total=1,
+            matched_sources=["machine_code"] if same else [],
+            mismatched_sources=[] if same else ["machine_code"],
+            reason=(
+                "授权按机器码精确匹配（未记录逐来源指纹，无容错）"
+                if same else
+                f"机器码不符（授权 {payload.machine_code}，本机 {current_code}）"
+            ),
+        )
+    if not match.ok:
+        return LicenseStatus(
+            ok=False, code=REASON_MACHINE_MISMATCH,
+            message=f"{REASON_TEXT[REASON_MACHINE_MISMATCH]}：{match.reason}",
+            payload=payload, license_path=str(path), machine_code=current_code,
+            match=match, public_key_fingerprint=key_fp, sources=dict(sources),
+        )
+
+    # 7. 有效期（含首尾当天）
+    ref = today or _today()
+    try:
+        not_before = _parse_date(payload.not_before)
+        expires_at = _parse_date(payload.expires_at)
+    except LicenseMalformedError as exc:
+        return _fail(REASON_LICENSE_MALFORMED, str(exc), payload=payload, match=match)
+    if not_before and ref < not_before:
+        return LicenseStatus(
+            ok=False, code=REASON_NOT_YET_VALID,
+            message=f"{REASON_TEXT[REASON_NOT_YET_VALID]}（生效日期 {not_before}）",
+            payload=payload, license_path=str(path), machine_code=current_code,
+            match=match, public_key_fingerprint=key_fp, sources=dict(sources),
+        )
+    if expires_at and ref > expires_at:
+        return LicenseStatus(
+            ok=False, code=REASON_EXPIRED,
+            message=f"{REASON_TEXT[REASON_EXPIRED]}（到期日期 {expires_at}）",
+            payload=payload, license_path=str(path), machine_code=current_code,
+            match=match, public_key_fingerprint=key_fp, sources=dict(sources),
+        )
+
+    return LicenseStatus(
+        ok=True, code=REASON_OK,
+        message=REASON_TEXT[REASON_OK],
+        payload=payload, license_path=str(path), machine_code=current_code,
+        match=match, public_key_fingerprint=key_fp, sources=dict(sources),
+    )
+
+
+def decision_for_plugin(status: LicenseStatus, plugin_type: str) -> PluginDecision:
+    """由整份 license 的结论推导单个插件的授权结论。"""
+    def deny(code: str, message: str = "") -> PluginDecision:
+        # status.message 已含原因文案（如"未配置厂商公钥，无法校验授权"），
+        # 优先用它，避免把原因码文案再拼一遍造成"…：…"的重复
+        detail = message or REASON_TEXT.get(code, code)
+        return PluginDecision(
+            allowed=False, code=code, plugin_type=plugin_type,
+            message=f"插件 {plugin_type} 未获授权：{detail}",
+        )
+
+    if not status.ok:
+        # 整份 license 无效（缺失/过期/机器码不符…）→ 一律拒绝
+        return deny(status.code, status.message)
+    if plugin_type not in status.licensed_plugins:
+        return deny(REASON_PLUGIN_NOT_LICENSED)
+    return PluginDecision(
+        allowed=True, code=REASON_OK, plugin_type=plugin_type,
+        message=f"插件 {plugin_type} 已授权"
+        + (f"（到期 {status.payload.expires_at}）" if status.payload and status.payload.expires_at else ""),
+    )