diff --git a/pyproject.toml b/pyproject.toml index 98eae0d..c528a91 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,16 +4,11 @@ requires = ["hatchling"] [project] name = "chongming-cli" -version = "0.3.0" -description = "chongming 命令脚本集合：构建打包 / 进程插件打包 / 插件脚手架 / 模型生成" +version = "0.3.1" +description = "chongming 命令脚本集合：构建打包 / 进程插件打包 / 插件脚手架 / 模型生成 / 机器码授权签发" readme = "README.md" requires-python = ">=3.12" -dependencies = [ - "chongming-config", - "module-bank", - "sqlite-vfs", - "tomli", -] +dependencies = ["chongming-config", "chongming-license", "module-bank", "sqlite-vfs", "tomli"] [tool.hatch.build.targets.wheel] packages = ["src/chongming_cli"] @@ -22,12 +17,11 @@ packages = ["src/chongming_cli"] url = "http://124.71.68.6:3000/api/packages/chongming/pypi/simple/" [dependency-groups] -dev = [ - "twine>=7.0.0", -] +dev = ["twine"] [project.scripts] build = "chongming_cli.build:build" build_plugin = "chongming_cli.build_plugin:build_plugin" gen_models = "chongming_cli.gen_models:gen_models" gen_plugin = "chongming_cli.gen_plugin:gen_plugin" +chongming-license = "chongming_cli.license:license_cmd" diff --git a/src/chongming_cli/build.py b/src/chongming_cli/build.py index 6945f78..94594bd 100644 --- a/src/chongming_cli/build.py +++ b/src/chongming_cli/build.py @@ -19,6 +19,7 @@ obfuscate PyArmor 混淆（仅 pyinstaller 模式） pack PyInstaller/Nuitka 打包 clean 清理临时文件 + license_guard 授权私钥泄漏守卫 vue Vue 前端构建 可传多个: --only mbank,vue 或 --only mbank --only vue @@ -27,7 +28,9 @@ uv run build nuitka --only mbank,prepare """ +import os import platform +import re import shutil import subprocess import sys @@ -211,12 +214,23 @@ def _parse_args() -> tuple[str, set[str]]: def _pack_mbank(output_path: str, source_dir: str, package_name: str, obfuscate: bool) -> str: """打包单个 mbank 目标，返回文件大小描述""" + # 每次从空文件开始：旧包里的历史版本既让体积膨胀，也会留下重名/空的版本 + Path(output_path).unlink(missing_ok=True) + key = Encryption.generate_key() + key_path = Path(output_path).with_suffix(".key") + key_path.write_text(key, encoding="utf-8") + packer = PythonToSQLite(output_path, key, pyarmor_obfuscate=obfuscate) - packer.pack_directory(source_dir, package_name) - packer.verify_package_structure() - packer.packer_key(key) - packer.delete_source_code(None) + try: + packer.pack_directory(source_dir, package_name) + packer.verify_package_structure() + packer.delete_source_code(None) + packer.sign_version("my_key", key_id="ci-2026") + # 给"当前这一版（已签名的）"起名，不派生工作版本 —— 加载端跟最新走就是它 + packer.vfs.rename_snapshot(packer.vfs.snapshot_id, "v1.0.0") + finally: + packer.close() return _format_size(output_path) @@ -280,16 +294,139 @@ def _stage_prepare_build_dir(config: Config, mode: str) -> str: for f in ("main.py", "server.py", "config.toml"): shutil.copy(f"public/{f}", f"build/{f}") - # 注入 JWT 密钥 - jwt_key = Encryption.generate_key() cfg_path = build_path / "config.toml" content = cfg_path.read_text(encoding="utf-8") + + # 注入 JWT 密钥 + jwt_key = Encryption.generate_key() content = content.replace("your-secret-key-change-in-production", jwt_key) + + # 注入厂商授权公钥（机器码授权） + pubkey = _resolve_license_public_key() + if pubkey: + content = _set_license_pubkey(content, pubkey) + elif os.environ.get(ALLOW_NO_KEY_ENV) == "1": + # 公钥缺失时打包产物会拒绝**全部**插件（没人能证明被授权）= 砖头设备， + # 故默认让构建失败；仅内部演示允许显式放行。 + print( + f" ⚠️ 未提供授权公钥且设置了 {ALLOW_NO_KEY_ENV}=1：" + "本产物无法校验任何授权（所有插件都不会加载），仅限内部演示" + ) + else: + raise RuntimeError( + "缺少厂商授权公钥：打包产物会拒绝加载所有插件（设备变砖）。\n" + f" 请用环境变量 {LICENSE_PUBKEY_ENV}= 或 " + f"{LICENSE_PUBKEY_FILE_ENV}=<文件路径> 提供，\n" + f" 或把公钥写入 {DEFAULT_LICENSE_PUBKEY_FILE}" + "（由 `uv run chongming-license keygen` 输出）。\n" + f" 仅内部演示无授权构建可设 {ALLOW_NO_KEY_ENV}=1。" + ) cfg_path.write_text(content, encoding="utf-8") return "资源准备完成" +# ── 授权公钥注入与私钥泄漏守卫 ───────────────────────────────────────────── +# +# 签发私钥是授权的唯一信任根：一旦随产物发布，任何人拿到它就能签发授权， +# 整套机器码绑定形同虚设。因此构建期必须提供**公钥**，且产物里**不得出现私钥** +# （与 test_build_plugin.py 对插件包 secret_key 的同类断言对齐）。 + +LICENSE_PUBKEY_ENV = "CHONGMING_LICENSE_PUBKEY" +LICENSE_PUBKEY_FILE_ENV = "CHONGMING_LICENSE_PUBKEY_FILE" +ALLOW_NO_KEY_ENV = "CHONGMING_LICENSE_ALLOW_NO_KEY" +DEFAULT_LICENSE_PUBKEY_FILE = "license_public.key" + +# 私钥 PEM 头（覆盖 PKCS#8 / 加密 PKCS#8 / RSA / EC / OpenSSH） +_PRIVATE_KEY_MARKERS = ( + b"-----BEGIN PRIVATE KEY-----", + b"-----BEGIN ENCRYPTED PRIVATE KEY-----", + b"-----BEGIN RSA PRIVATE KEY-----", + b"-----BEGIN EC PRIVATE KEY-----", + b"-----BEGIN OPENSSH PRIVATE KEY-----", +) + + +def _resolve_license_public_key() -> str: + """按 环境变量 > 环境变量指定文件 > 默认文件 的顺序读取厂商授权公钥。""" + env_key = (os.environ.get(LICENSE_PUBKEY_ENV) or "").strip() + if env_key: + return env_key + + candidates = [] + if os.environ.get(LICENSE_PUBKEY_FILE_ENV): + candidates.append(Path(os.environ[LICENSE_PUBKEY_FILE_ENV])) + candidates.append(Path(DEFAULT_LICENSE_PUBKEY_FILE)) + for path in candidates: + if path.is_file(): + return path.read_text(encoding="utf-8").strip() + return "" + + +def _set_license_pubkey(text: str, pubkey: str) -> str: + """把公钥写进 [license] 段的 public_key，保留其余注释与排版。 + + 按段定位而非全局替换：config.toml 注释很多，整文件重新序列化会丢注释， + 全局字符串替换又可能误伤其它段。 + """ + lines = text.splitlines() + out: list[str] = [] + in_license = False + done = False + for line in lines: + stripped = line.strip() + if stripped.startswith("[") and stripped.endswith("]"): + in_license = stripped == "[license]" + if in_license and not done and re.match(r"^\s*public_key\s*=", line): + out.append(f'public_key = "{pubkey}"') + done = True + continue + out.append(line) + if not done: + raise RuntimeError( + "config.toml 中未找到 [license] 段的 public_key 项；" + "请确认模板已包含该段（先同步 public/config.toml）" + ) + return "\n".join(out) + ("\n" if text.endswith("\n") else "") + + +def _stage_license_guard() -> str: + """构建产物守卫：确认 build/ 内不含任何私钥。""" + build_path = Path("build") + if not build_path.exists(): + return "无 build 目录，跳过" + + offenders: list[str] = [] + for path in build_path.rglob("*"): + if not path.is_file(): + continue + # 按内容判定为主（与扩展名无关，改名藏不住私钥）； + # 另外拦一下名字就写明是授权私钥的文件（即便内容为空/被截断）。 + # 不按 .pem 扩展名一刀切 —— 证书类 .pem 是合法的构建内容。 + if path.name.startswith("license_private"): + offenders.append(f"{path}（授权私钥文件）") + continue + try: + with path.open("rb") as f: + tail = b"" + while chunk := f.read(1 << 20): + window = tail + chunk + if any(m in window for m in _PRIVATE_KEY_MARKERS): + offenders.append(f"{path}（内容含私钥 PEM）") + break + # 保留尾巴，避免标记被块边界截断而漏检 + tail = window[-64:] + except OSError: + continue + + if offenders: + raise RuntimeError( + "构建产物中发现私钥，已终止构建（授权私钥泄漏 = 任何人都能签发授权）：\n - " + + "\n - ".join(offenders) + ) + return "未发现私钥泄漏" + + # ── 阶段 3: PyArmor 混淆 ─────────────────────────────────────────────────── @@ -593,6 +730,10 @@ async def main(): if should_run("clean"): await run_step("清理临时文件", _stage_cleanup, mode) + # 阶段 6: 授权私钥泄漏守卫（产物落地后、分发前） + if should_run("license_guard"): + await run_step("授权私钥守卫", _stage_license_guard) + # 阶段 7: Vue 前端构建 if should_run("vue"): await run_step("Vue 前端构建", _stage_build_vue) diff --git a/src/chongming_cli/license.py b/src/chongming_cli/license.py new file mode 100644 index 0000000..f806817 --- /dev/null +++ b/src/chongming_cli/license.py @@ -0,0 +1,722 @@ +"""license 签发与管理工具（厂商侧 / 设备侧诊断） + +用法:: + + # ① 设备侧：导出申请码（机器码 + 各硬件来源指纹），交给客户回传 + uv run chongming-license show # 机器码 + uv run chongming-license show --sources # 连同各来源明细（现场核对用） + uv run chongming-license show --request-code # 单行申请码，供客户粘贴 + + # ② 厂商侧：一次性生成密钥对（私钥离线保管，绝不进仓库/构建产物） + uv run chongming-license keygen -o license_private.pem + + # ③ 厂商侧：签发 + uv run chongming-license issue --request-code CM1-xxx-yyy \\ + --plugins aibox_sophon_adapter,ip_sound_column \\ + --customer 某某科技 --days 365 -k license_private.pem -o license.lic + + # ④ 校验（设备侧诊断 / 厂商侧发货前自检） + uv run chongming-license verify license.lic --key <厂商公钥> + uv run chongming-license inspect license.lic # 只看内容，不验签 + + 退出码：0 成功/校验通过，1 失败（便于脚本判断）。 + +注意：本命令属于 **chongming-cli**（根仓的 dev 依赖），不随部署产物发布 —— +这正是刻意安排：签发私钥只存在于厂商的构建/商务机器上，设备端不携带签发能力。 +设备端展示机器码请走网关管理接口 ``GET /api/v1/license/info`` 与前端「授权」页。 +""" +from __future__ import annotations + +import argparse +import json +import sys +from datetime import date, datetime, timezone +from pathlib import Path + +from chongming_license import ( + DEFAULT_EXPIRING_DAYS, + DEFAULT_MIN_MATCH, + SOURCES, + Registry, + RegistryError, + default_registry_path, + LicenseManager, + LicenseStatus, + MemoryClockStore, + collect_report, + decode_request_code, + generate_keypair, + encode_request_code, + load_private_key, + machine_code, + build_license, + build_payload, + public_key_b64, + public_key_fingerprint, + read_license, + resolve_expiry, + utc_today, + write_license, + write_private_key, +) + +DEFAULT_KEY_PATH = "license_private.pem" + + +# --------------------------------------------------------------------------- +# 输出小工具 +# --------------------------------------------------------------------------- +def _ok(msg: str) -> None: + print(f"✓ {msg}") + + +def _err(msg: str) -> None: + print(f"✗ {msg}", file=sys.stderr) + + +def _warn(msg: str) -> None: + print(f"⚠ {msg}", file=sys.stderr) + + +def _today() -> date: + """判定基准日期用 **UTC**，与校验端（chongming_license.verify）保持一致。 + + 若签发端用本地日期、校验端用 UTC，在时区偏移跨日的时段会差一天， + 导致"刚签发即过期"或"到期后仍可用一天"。 + """ + return datetime.now(timezone.utc).date() + + +# --------------------------------------------------------------------------- +# keygen +# --------------------------------------------------------------------------- +def _cmd_keygen(args) -> int: + out = Path(args.output) + if out.exists() and not args.force: + _err(f"{out} 已存在；确要覆盖请加 --force（覆盖后旧 license 将无法再签新授权，" + "已发出去的授权仍可用旧公钥校验）") + return 1 + + private_key, pub_b64 = generate_keypair() + try: + out = write_private_key(private_key, out, overwrite=True) + except Exception as exc: + _err(f"写入私钥失败：{exc}") + return 1 + + print() + _ok(f"私钥已写入 {out}（权限 0600）") + print() + print("厂商公钥（填入部署产物的 [license] 段）：") + print(f" public_key = \"{pub_b64}\"") + print() + print(f"公钥指纹：{public_key_fingerprint(pub_b64)}") + print() + _warn("私钥是本授权的唯一信任根：请离线保管并做好备份，切勿提交仓库或打进部署产物。") + _warn("私钥丢失 = 无法再签发新授权；私钥泄漏 = 任何人都能签发授权。") + if "kid" in vars(args) and args.kid: + print(f"建议签发时使用 --kid {args.kid}") + return 0 + + +# --------------------------------------------------------------------------- +# show +# --------------------------------------------------------------------------- +def _cmd_show(args) -> int: + report = collect_report() + + if args.json: + print(json.dumps({"machine_code": machine_code(report.sources), **report.as_dict()}, + ensure_ascii=False, indent=2)) + return 0 + + if args.request_code: + print(encode_request_code(report.sources)) + return 0 + + print(f"机器码: {machine_code(report.sources)}") + if report.warnings: + print() + for w in report.warnings: + _warn(w) + + if args.sources: + print() + print(f"{'来源':<14}{'状态':<8}{'值':<40}说明") + print("-" * 96) + for d in report.details: + state = "已采集" if d.present else "缺失" + if d.weak and d.present: + state = "已采集*" + value = d.value[:38] + "…" if len(d.value) > 38 else d.value + print(f"{d.name:<14}{state:<8}{value:<40}{d.detail}") + print() + print("* 标记为 weak 的来源不参与判定阈值（同镜像批量刷机时可能全机相同）") + return 0 + + +# --------------------------------------------------------------------------- +# issue +# --------------------------------------------------------------------------- +def _load_sources_for_issue(args) -> tuple[dict, str]: + """从三种入参之一得到 (sources, machine_code)。 + + 优先级：申请码 > 来源 JSON > 单一机器码。前两者可签出**带容错**的授权， + 仅给机器码只能签出精确匹配的授权（失去换网卡/换 eMMC 的容错）。 + """ + if args.request_code: + data = decode_request_code(args.request_code) + return data["sources"], data["machine_code"] + if args.sources_json: + raw = json.loads(Path(args.sources_json).read_text(encoding="utf-8")) + sources = raw.get("sources", raw) if isinstance(raw, dict) else {} + if not isinstance(sources, dict) or not sources: + raise ValueError(f"{args.sources_json} 中未找到 sources 字段") + sources = {str(k): str(v) for k, v in sources.items()} + return sources, machine_code(sources) + if args.machine_code: + return {}, args.machine_code.strip().upper() + raise ValueError("必须提供 --request-code / --sources-json / --machine-code 之一") + + +def _cmd_issue(args) -> int: + try: + sources, code = _load_sources_for_issue(args) + except Exception as exc: + _err(f"无法确定签发对象：{exc}") + return 1 + + try: + plugins = [p.strip() for p in (args.plugins or "").split(",") if p.strip()] + except Exception: + plugins = [] + if not plugins: + _err("必须用 --plugins 指定授权插件（逗号分隔）；不授权任何插件的 license 没有意义") + return 1 + + # 有效期：--days 与 --expires 二选一，都不给则永久有效 + try: + expires = resolve_expiry(days=args.days, expires=args.expires or None) + except ValueError as exc: + _err(f"{exc}") + return 1 + if expires and date.fromisoformat(expires) < utc_today(): + _warn(f"到期日期 {expires} 已过，签出的授权立即失效") + + try: + private_key = load_private_key(args.key) + except Exception as exc: + _err(f"读取私钥失败：{exc}") + return 1 + + # 载荷拼装与签名一律走 chongming_license.issue：与可视化工具共用同一实现， + # 避免两边 schema 漂移（载荷是签名覆盖的字节，字段差异会直接导致验签失败） + try: + payload = build_payload( + machine_code=code, + plugins=plugins, + sources=sources, + customer=args.customer or "", + lic_id=args.lic_id or "", + not_before=args.not_before or None, + expires_at=expires, + note=args.note or "", + ) + except ValueError as exc: + _err(f"{exc}") + return 1 + license_doc = build_license(payload, private_key, kid=args.kid or "") + lic_id = payload["lic_id"] + + out = write_license(license_doc, args.output) + + pub_b64 = public_key_b64(private_key) + recorded = "" + if not args.no_registry: + # 签发即入库：否则授权文件会散落在邮件/下载目录里，续期时既不知道给哪台 + # 机器签过，也拿不到那台机器的指纹（只能再找客户要申请码） + reg = _open_registry(args) + if reg is not None: + with reg: + try: + reg.save(license_doc, + public_key_fp=public_key_fingerprint(pub_b64), + source="issue") + recorded = str(reg.path) + except RegistryError as exc: + _warn(f"签发成功但记入台账失败：{exc}") + + print() + _ok(f"已签发 {out}") + print(f" 授权编号 {lic_id}") + print(f" 客户 {payload['customer'] or '(未填)'}") + print(f" 机器码 {code}") + if sources: + weak = {s.name for s in SOURCES if s.weak} + strong = sorted(k for k in sources if k not in weak) + weak_present = sorted(k for k in sources if k in weak) + desc = ", ".join(strong) or "无" + if weak_present: + desc += f"（不参与判定：{', '.join(weak_present)}）" + print(f" 指纹来源 {desc}；容错阈值 {args.min_match}") + if not strong: + _warn("授权内没有任何可信来源，判定将只依赖易重复的来源（安全性不足）") + else: + print(" 指纹来源 无（仅机器码精确匹配，无容错）") + print(f" 授权插件 {plugins}") + print(f" 有效期 {payload['not_before'] or '立即生效'} → {expires or '永久'}") + print(f" 公钥指纹 {public_key_fingerprint(pub_b64)}") + if recorded: + print(f" 已入台账 {recorded}") + print() + print("部署：把该文件放到网关工作目录（默认 ./license.lic，或 [license] path 指定）后重启网关。") + if recorded: + print("续期：`chongming-license renew --lic-id " + f"{lic_id} --days 365`（复用台账指纹，无需客户再报申请码）") + return 0 + + +# --------------------------------------------------------------------------- +# inspect / verify +# --------------------------------------------------------------------------- +def _resolve_public_key(args) -> str: + """取校验用公钥：--key > --public-key-file > 根 config.toml 的 [license] public_key。 + + 根配置缺失（在项目根之外执行本命令）时返回空串而非抛异常 —— + `inspect` 在无公钥时的语义本就是"只看内容、不验签"。 + """ + if args.key: + return args.key + if args.public_key_file: + return Path(args.public_key_file).read_text(encoding="utf-8").strip() + try: + from chongming_config import get_config + + return str(get_config().get("license", {}).get("public_key", "") or "") + except Exception: # noqa: BLE001 - 读不到配置（无 config.toml 等）按"未提供公钥"处理 + return "" + + +def _resolve_sources(args) -> dict | None: + """校验用的"本机指纹"：显式来源 JSON、来源码，或现场采集。""" + if args.sources_json: + raw = json.loads(Path(args.sources_json).read_text(encoding="utf-8")) + sources = raw.get("sources", raw) if isinstance(raw, dict) else {} + return {str(k): str(v) for k, v in sources.items()} + if args.request_code: + return decode_request_code(args.request_code)["sources"] + return None # None = read_license 内部现场采集 + + +def _print_status(status: LicenseStatus, verbose: bool = True) -> None: + if status.ok: + _ok(status.message) + else: + _err(f"{status.message}（原因码 {status.code}）") + if not verbose: + return + print(f" 本机机器码 {status.machine_code}") + if status.match: + print(f" 指纹比对 {status.match.reason}") + if status.payload: + p = status.payload + print(f" 授权编号 {p.lic_id}") + if p.customer: + print(f" 客户 {p.customer}") + if p.issued_at: + print(f" 签发时间 {p.issued_at}") + print(f" 有效期 {p.not_before or '立即生效'} → {p.expires_at or '永久'}") + print(f" 授权插件 {sorted(p.plugins_normalized()) or '无'}") + if p.note: + print(f" 备注 {p.note}") + if status.public_key_fingerprint: + print(f" 公钥指纹 {status.public_key_fingerprint}") + + +def _cmd_inspect(args) -> int: + """只看内容，不验签（不提供 --key 时）。""" + try: + doc = json.loads(Path(args.license_file).read_text(encoding="utf-8")) + except Exception as exc: + _err(f"读取失败：{exc}") + return 1 + pub = _resolve_public_key(args) + if not pub: + print(json.dumps(doc, ensure_ascii=False, indent=2)) + print() + _warn("未提供公钥（--key / --public-key-file / 根配置 [license]），未做验签") + return 0 + status = read_license(args.license_file, pub, min_match=args.min_match, + sources=_resolve_sources(args), today=_today()) + _print_status(status) + return 0 if status.ok else 1 + + +def _cmd_verify(args) -> int: + pub = _resolve_public_key(args) + if not pub: + _err("未提供厂商公钥：请用 --key / --public-key-file，或在根 config.toml 配置 [license] public_key") + return 1 + # 设备侧真实校验：走 LicenseManager（含时钟回拨检测与插件白名单判定） + mgr = LicenseManager( + license_path=args.license_file, + public_key=pub, + min_match=args.min_match, + clock_store=MemoryClockStore(), # 一次性进程，水位无长期意义 + enforce=True, + ) + try: + status = mgr.reload() + except Exception as exc: + _err(f"校验异常：{exc}") + return 1 + _print_status(status) + + if args.plugin: + print() + decision = mgr.check_plugin(args.plugin) + (_ok if decision.allowed else _err)(decision.message) + if not decision.allowed: + return 1 + return 0 if status.ok else 1 + + +# --------------------------------------------------------------------------- +# 台账（list / renew / export） +# --------------------------------------------------------------------------- +def _open_registry(args): + """打开台账库（路径：--db > CHONGMING_LICENSE_DB > ~/.chongming/licenses.db）。""" + path = Path(args.db).expanduser() if getattr(args, "db", "") else default_registry_path() + try: + return Registry(path) + except RegistryError as exc: + _err(f"{exc}") + return None + + +def _display_width(text: str) -> int: + """字符串的终端显示宽度（中日韩字符占 2 列）。 + + 直接用 ``str.ljust`` 对齐含中文的表格会错位 —— Python 按字符数计算， + 终端却按列宽渲染，中文客户名会把后面所有列往右推。 + """ + import unicodedata + + return sum(2 if unicodedata.east_asian_width(ch) in ("W", "F") else 1 for ch in text) + + +def _pad(text: str, width: int) -> str: + """按显示宽度左对齐补齐（超宽则不截断，原样输出）。""" + text = str(text) + return text + " " * max(0, width - _display_width(text)) + + +def _print_records(records, *, title: str = "") -> None: + if title: + print(f"\n{title}") + if not records: + print(" （无记录）") + return + print(" " + _pad("授权编号", 20) + _pad("客户", 16) + _pad("机器码", 30) + + _pad("状态", 16) + _pad("到期", 12) + "插件") + print(" " + "-" * 108) + for r in records: + left = r.days_left() + status = r.status_label() + (f"({left}d)" if left is not None else "") + customer = (r.customer or "(未填)")[:14] + print(" " + _pad(r.lic_id, 20) + _pad(customer, 16) + + _pad(r.machine_code, 30) + _pad(status, 16) + + _pad(r.expires_at or "永久", 12) + ",".join(r.plugins)) + + +def _cmd_list(args) -> int: + """列出台账中的授权（默认按到期日升序，最急的排最前）。""" + reg = _open_registry(args) + if reg is None: + return 1 + with reg: + records = reg.query( + customer=args.customer or None, + machine_code=args.machine_code or None, + plugin=args.plugin or None, + status=args.status or None, + keyword=args.keyword or None, + expiring_days=args.days, + limit=args.limit or None, + ) + if args.json: + print(json.dumps( + [{ + "lic_id": r.lic_id, "customer": r.customer, "machine_code": r.machine_code, + "plugins": r.plugins, "status": r.status(), "status_label": r.status_label(), + "days_left": r.days_left(), "not_before": r.not_before, + "expires_at": r.expires_at, "issued_at": r.issued_at, + "kid": r.kid, "public_key_fp": r.public_key_fp, "note": r.note, + } for r in records], + ensure_ascii=False, indent=2, + )) + return 0 + + st = reg.stats(expiring_days=args.days) + print(f"台账库: {reg.path}") + print(f"总计 {st['total']} 条 —— 有效 {st['active']}、即将到期 {st['expiring']}、" + f"已过期 {st['expired']}、永久 {st['forever']}、未生效 {st['not_yet_valid']}") + _print_records(records, title=f"符合条件 {len(records)} 条：") + return 0 + + +def _cmd_renew(args) -> int: + """续期：复用台账里的机器指纹重新签发，**无需再向客户索取申请码**。 + + 这是台账最大的价值：现场设备续期时常常联系不上客户，而指纹台账里已有。 + """ + try: + private_key = load_private_key(args.key) + except Exception as exc: + _err(f"读取私钥失败：{exc}") + return 1 + + reg = _open_registry(args) + if reg is None: + return 1 + with reg: + old = reg.get(args.lic_id) + if old is None: + _err(f"台账中没有授权编号 {args.lic_id}；用 `chongming-license list` 查看现有记录") + return 1 + if not old.sources and not args.request_code and not args.sources_json: + _err(f"{old.lic_id} 未记录逐来源指纹（当初只按机器码精确匹配签发），" + "续期前请向客户重新索取申请码后用 issue 签发") + return 1 + + if args.days is None and not args.expires: + _err("续期需指定 --days 或 --expires（都不给等于保持原有效期，没有意义）") + return 1 + try: + expires = resolve_expiry(days=args.days, expires=args.expires or None) + except ValueError as exc: + _err(f"{exc}") + return 1 + + # 指纹来源：显式给了新申请码/来源文件就用新的，否则复用台账里的（续期常态） + if args.request_code or args.sources_json: + try: + sources, machine = _load_sources_for_issue(args) + except Exception as exc: + _err(f"无法解析新的来源指纹：{exc}") + return 1 + else: + sources, machine = old.sources, old.machine_code + + plugins = ([p.strip() for p in args.plugins.split(",") if p.strip()] + if args.plugins else list(old.plugins)) + try: + payload = build_payload( + machine_code=machine, + plugins=plugins, + sources=sources, + customer=args.customer if args.customer is not None else old.customer, + note=args.note if args.note else old.note, + expires_at=expires, + ) + except ValueError as exc: + _err(f"{exc}") + return 1 + if args.keep_lic_id: + payload["lic_id"] = old.lic_id + doc = build_license(payload, private_key, kid=args.kid or old.kid) + + out = Path(args.output) if args.output else Path(f"license_{payload['lic_id']}.lic") + write_license(doc, out) + new_rec = reg.save( + doc, + public_key_fp=public_key_fingerprint(public_key_b64(private_key)), + source="renew", + ) + # 在旧记录上留一句指向新授权，日后从任一编号都能串起来 + if not args.keep_lic_id: + link = f"已续期 → {new_rec.lic_id}" + if link not in old.note: + reg.update_note(old.lic_id, (old.note + "；" if old.note else "") + link) + + print() + _ok(f"已续期并记入台账：{out}") + if not args.keep_lic_id: + print(f" 原授权 {old.lic_id}（{old.expires_at or '永久'}，{old.status_label()}）") + print(f" 新授权 {new_rec.lic_id}（{new_rec.expires_at or '永久'}）") + print(f" 客户 {new_rec.customer or '(未填)'}") + print(f" 机器码 {new_rec.machine_code}") + print(f" 授权插件 {plugins}") + print(f" 指纹来源 {sorted(sources) or '（无，仅机器码精确匹配）'}") + print(f" 台账库 {reg.path}") + print() + print("说明：指纹复用自台账，**无需客户再提供申请码**；把新 .lic 给客户导入即可。") + return 0 + + +def _cmd_export(args) -> int: + """导出台账：.lic 文件（可按客户分目录）+ 可选 CSV 清单。""" + reg = _open_registry(args) + if reg is None: + return 1 + with reg: + records = reg.query( + customer=args.customer or None, + status=args.status or None, + expiring_days=args.days, + ) + if not records: + _warn("没有符合条件的记录，未导出任何文件") + return 0 + files = reg.export_licenses(args.out, records, subdir_by_customer=args.by_customer) + _ok(f"已导出 {len(files)} 个 .lic 到 {args.out}" + + ("（按客户分子目录）" if args.by_customer else "")) + if args.csv: + csv_path = reg.export_csv(args.csv, records) + _ok(f"已导出清单 {csv_path}（UTF-8 BOM，Excel 中文不乱码）") + st = reg.stats(expiring_days=args.days) + print(f" 台账库 {reg.path}（共 {st['total']} 条，本次导出 {len(files)} 条）") + return 0 + + +# --------------------------------------------------------------------------- +# 入口 +# --------------------------------------------------------------------------- +def _add_common_source_args(p: argparse.ArgumentParser) -> None: + p.add_argument("--sources-json", help="来源 JSON 文件（`show --json` 的导出，或含 sources 字段）") + p.add_argument("--request-code", help="申请码（`show --request-code` 的输出）") + p.add_argument("--machine-code", help="仅机器码（失去换硬件容错，不建议）") + + +def _add_db_arg(p: argparse.ArgumentParser) -> None: + p.add_argument("--db", default="", + help=f"台账库路径（默认 {default_registry_path()}，" + "也可用环境变量 CHONGMING_LICENSE_DB）") + + +def license_cli(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser( + prog="chongming-license", + description="机器码授权工具：设备侧导出申请码，厂商侧签发/续期/台账/校验", + formatter_class=argparse.RawDescriptionHelpFormatter, + epilog=__doc__, + ) + sub = parser.add_subparsers(dest="command", required=True) + + # keygen + p_keygen = sub.add_parser("keygen", help="生成 Ed25519 密钥对（厂商侧，一次性）") + p_keygen.add_argument("-o", "--output", default=DEFAULT_KEY_PATH, + help=f"私钥输出路径（默认 {DEFAULT_KEY_PATH}）") + p_keygen.add_argument("--kid", default="", help="建议的密钥标识，仅提示用途") + p_keygen.add_argument("--force", action="store_true", help="覆盖已存在的私钥文件") + p_keygen.set_defaults(func=_cmd_keygen) + + # show + p_show = sub.add_parser("show", help="导出本机机器码 / 申请码（设备侧）") + p_show.add_argument("--sources", action="store_true", help="打印各硬件来源明细（现场核对用）") + p_show.add_argument("--json", action="store_true", help="以 JSON 输出全部内容") + p_show.add_argument("--request-code", action="store_true", help="只输出申请码（单行，供客户粘贴回传）") + p_show.set_defaults(func=_cmd_show) + + # issue + p_issue = sub.add_parser("issue", help="签发 license（厂商侧）") + _add_common_source_args(p_issue) + p_issue.add_argument("--plugins", required=True, + help="授权插件 type，逗号分隔，如 aibox_sophon_adapter,ip_sound_column") + p_issue.add_argument("-k", "--key", default=DEFAULT_KEY_PATH, help=f"私钥路径（默认 {DEFAULT_KEY_PATH}）") + p_issue.add_argument("-o", "--output", default="license.lic", help="输出路径（默认 license.lic）") + p_issue.add_argument("--customer", default="", help="客户名称") + p_issue.add_argument("--lic-id", default="", help="授权编号（默认自动生成）") + p_issue.add_argument("--kid", default="", help="密钥标识（写入文件，仅提示用途）") + p_issue.add_argument("--not-before", default="", help="生效日期 ISO8601（默认立即生效）") + p_issue.add_argument("--expires", default="", help="到期日期 ISO8601，含当天") + p_issue.add_argument("--days", type=int, default=None, help="有效期天数（与 --expires 二选一）") + p_issue.add_argument("--note", default="", help="备注（合同号等）") + p_issue.add_argument("--min-match", type=int, default=DEFAULT_MIN_MATCH, + help=f"指纹容错阈值（默认 {DEFAULT_MIN_MATCH}；仅影响签发时的提示文字）") + p_issue.add_argument("--no-registry", action="store_true", + help="不记入台账（默认签发即入库，避免日后找不到当初给哪台机器签过）") + _add_db_arg(p_issue) + p_issue.set_defaults(func=_cmd_issue) + + # inspect + p_inspect = sub.add_parser("inspect", help="查看 license 内容（可选验签）") + p_inspect.add_argument("license_file") + p_inspect.add_argument("--key", default="", help="厂商公钥（base64 或 PEM）") + p_inspect.add_argument("--public-key-file", default="", help="从文件读取公钥") + p_inspect.add_argument("--min-match", type=int, default=DEFAULT_MIN_MATCH, help="指纹容错阈值") + _add_common_source_args(p_inspect) + p_inspect.set_defaults(func=_cmd_inspect) + + # verify + p_verify = sub.add_parser("verify", help="按本机（或指定的）环境完整校验 license") + p_verify.add_argument("license_file") + p_verify.add_argument("--key", default="", help="厂商公钥（base64 或 PEM）") + p_verify.add_argument("--public-key-file", default="", help="从文件读取公钥") + p_verify.add_argument("--min-match", type=int, default=DEFAULT_MIN_MATCH, help="指纹容错阈值") + p_verify.add_argument("--plugin", default="", help="额外判定某插件是否在授权白名单内") + _add_common_source_args(p_verify) + p_verify.set_defaults(func=_cmd_verify) + + # list + p_list = sub.add_parser("list", help="列出台账中的授权（按到期日升序，最急的在前）") + p_list.add_argument("--customer", default="", help="按客户名模糊筛选") + p_list.add_argument("--machine-code", default="", help="按机器码筛选（大小写不敏感）") + p_list.add_argument("--plugin", default="", help="按授权插件筛选") + p_list.add_argument("--status", default="", + choices=["", "expired", "expiring", "active", "forever", "not_yet_valid"], + help="按状态筛选") + p_list.add_argument("--keyword", default="", help="关键字：匹配授权编号/机器码/客户/备注") + p_list.add_argument("--days", type=int, default=DEFAULT_EXPIRING_DAYS, + help=f"「即将到期」的窗口天数（默认 {DEFAULT_EXPIRING_DAYS}）") + p_list.add_argument("--limit", type=int, default=0, help="最多显示多少条（0=不限）") + p_list.add_argument("--json", action="store_true", help="以 JSON 输出（便于脚本处理）") + _add_db_arg(p_list) + p_list.set_defaults(func=_cmd_list) + + # renew + p_renew = sub.add_parser( + "renew", help="续期：复用台账里的机器指纹重签，无需客户再报申请码" + ) + p_renew.add_argument("--lic-id", required=True, help="要续期的授权编号（list 可查）") + p_renew.add_argument("--days", type=int, default=None, help="新的有效期天数") + p_renew.add_argument("--expires", default="", help="新的到期日 ISO8601（与 --days 二选一）") + p_renew.add_argument("--plugins", default="", help="更换授权插件（逗号分隔，默认沿用原白名单）") + p_renew.add_argument("--customer", default=None, help="更换客户名（默认沿用）") + p_renew.add_argument("--note", default="", help="更换备注（默认沿用）") + p_renew.add_argument("--keep-lic-id", action="store_true", + help="沿用原授权编号（默认生成新编号，新旧都留台账便于追溯）") + p_renew.add_argument("-k", "--key", default=DEFAULT_KEY_PATH, help=f"私钥路径（默认 {DEFAULT_KEY_PATH}）") + p_renew.add_argument("-o", "--output", default="", help="输出路径（默认 license_<新编号>.lic）") + p_renew.add_argument("--kid", default="", help="密钥标识（默认沿用原记录）") + p_renew.add_argument("--sources-json", help="改用新的来源 JSON 文件（换机器时用）") + p_renew.add_argument("--request-code", help="改用新的申请码（换机器时用）") + p_renew.add_argument("--machine-code", help="改用新的机器码（换机器时用）") + _add_db_arg(p_renew) + p_renew.set_defaults(func=_cmd_renew) + + # export + p_export = sub.add_parser("export", help="导出台账：.lic 文件 + 可选 CSV 清单") + p_export.add_argument("--out", required=True, help=".lic 输出目录") + p_export.add_argument("--csv", default="", help="同时导出 CSV 清单到该路径") + p_export.add_argument("--by-customer", action="store_true", help="按客户名分子目录") + p_export.add_argument("--customer", default="", help="只导出某客户（模糊匹配）") + p_export.add_argument("--status", default="", + choices=["", "expired", "expiring", "active", "forever", "not_yet_valid"], + help="只导出某状态") + p_export.add_argument("--days", type=int, default=DEFAULT_EXPIRING_DAYS, + help=f"「即将到期」的窗口天数（默认 {DEFAULT_EXPIRING_DAYS}）") + _add_db_arg(p_export) + p_export.set_defaults(func=_cmd_export) + + args = parser.parse_args(argv) + try: + return int(args.func(args) or 0) + except KeyboardInterrupt: + _err("已中断") + return 1 + + +def license_cmd() -> None: + """console script 入口（uv run chongming-license）。""" + sys.exit(license_cli())