modify: 授权
This commit is contained in:
+5
-11
@@ -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"
|
||||
|
||||
+147
-6
@@ -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}=<base64 公钥> 或 "
|
||||
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)
|
||||
|
||||
@@ -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())
|
||||
Reference in New Issue
Block a user