From f4a319337aaca83d027ca41ab6388066c0f63470 Mon Sep 17 00:00:00 2001 From: chakcy <947105045@qq.com> Date: Mon, 21 Sep 2026 11:00:37 +0800 Subject: [PATCH] first commit --- .gitignore | 72 +++++ README.md | 91 ++++++ pyproject.toml | 25 ++ src/chongming_license/__init__.py | 366 ++++++++++++++++++++++ src/chongming_license/clock.py | 159 ++++++++++ src/chongming_license/issue.py | 126 ++++++++ src/chongming_license/machine.py | 496 ++++++++++++++++++++++++++++++ src/chongming_license/models.py | 91 ++++++ src/chongming_license/registry.py | 496 ++++++++++++++++++++++++++++++ src/chongming_license/verify.py | 493 +++++++++++++++++++++++++++++ 10 files changed, 2415 insertions(+) create mode 100644 .gitignore create mode 100644 README.md create mode 100644 pyproject.toml create mode 100644 src/chongming_license/__init__.py create mode 100644 src/chongming_license/clock.py create mode 100644 src/chongming_license/issue.py create mode 100644 src/chongming_license/machine.py create mode 100644 src/chongming_license/models.py create mode 100644 src/chongming_license/registry.py create mode 100644 src/chongming_license/verify.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..c8f0442 --- /dev/null +++ b/.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 diff --git a/README.md b/README.md new file mode 100644 index 0000000..3634e7c --- /dev/null +++ b/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": "" +} +``` + +签名覆盖 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 需要厂商私钥，而公钥可明文放置 +（换公钥只会验签失败，无法据以伪造）。 diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..9d6ddda --- /dev/null +++ b/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/" diff --git a/src/chongming_license/__init__.py b/src/chongming_license/__init__.py new file mode 100644 index 0000000..bb10c79 --- /dev/null +++ b/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 diff --git a/src/chongming_license/clock.py b/src/chongming_license/clock.py new file mode 100644 index 0000000..7a7dcaa --- /dev/null +++ b/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)) diff --git a/src/chongming_license/issue.py b/src/chongming_license/issue.py new file mode 100644 index 0000000..7363738 --- /dev/null +++ b/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 diff --git a/src/chongming_license/machine.py b/src/chongming_license/machine.py new file mode 100644 index 0000000..21dd78a --- /dev/null +++ b/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--<校验和>``）。 + + 机器码是哈希、无法反推各来源值，而容错匹配需要逐来源原始值 —— 因此首次 + 申请授权时客户应回传**申请码**（一段单行文本），而不是只报机器码。 + + 申请码**不是**安全凭据：它只是公开硬件标识的传输编码，不参与验签。 + 客户伪造申请码无利可图 —— 签出来的授权只会绑定到他所声称的那台硬件。 + """ + 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, + ) diff --git a/src/chongming_license/models.py b/src/chongming_license/models.py new file mode 100644 index 0000000..4a266f6 --- /dev/null +++ b/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": "" + } + +签名覆盖的是 **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) diff --git a/src/chongming_license/registry.py b/src/chongming_license/registry.py new file mode 100644 index 0000000..754daa2 --- /dev/null +++ b/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``；可选按客户名分子目录， + 便于直接打包发给对应客户。 + """ + 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 diff --git a/src/chongming_license/verify.py b/src/chongming_license/verify.py new file mode 100644 index 0000000..36849ec --- /dev/null +++ b/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 ""), + )