#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ CVIN 批量登记工具（零依赖，只需 Python 3.8+ 与 cryptography） 用法： 1) 把要登记的序列号写进 serials.txt，每行一个 2) 改下面 CONFIG 四行 3) python3 cvin_onboard.py 它会为每台设备： 算 CVIN → 提交声明 → 生成并保存设备密钥 → 签名上报遥测 → 打印证书链接 注意：device_keys.json 是设备私钥，只生成一次。丢了就再也无法为这台设备 做签名证明（服务端会以 409 拒绝换密钥），务必备份。 """ import base64, hashlib, json, os, re, sys, time, urllib.request, urllib.error # ======== 改这里 ======== CONFIG = { "base": "https://cvin.org", # 服务地址 "operator": "your-cloud-name", # 你的运营方名称（会公开在登记记录里） "region": "apac-sg", # 机房/区域代号 "serial_file": "serials.txt", # 序列号清单 "api_key": "", # 超过 12 台请填运营方密钥；留空走自助限频 } MFR, MODEL, FORM, MEM = "nv", "h100", "sxm", "80g" # 型号四要素，按你的卡型改 # ======================== # 也可以用环境变量覆盖，方便放进 CI：CVIN_BASE / CVIN_OPERATOR / CVIN_API_KEY / CVIN_SERIALS for _k, _env in (("base", "CVIN_BASE"), ("operator", "CVIN_OPERATOR"), ("api_key", "CVIN_API_KEY"), ("serial_file", "CVIN_SERIALS")): if os.environ.get(_env): CONFIG[_k] = os.environ[_env] B36 = "0123456789abcdefghijklmnopqrstuvwxyz" KEYFILE = "device_keys.json" # ---------- 1. 算 CVIN（与 cvin.org 服务端逐字节一致） ---------- def norm_serial(s): s = str(s).strip().lower() s = re.sub(r"[\s_\-./:#]+", "", s) return re.sub(r"^(sn|s/n|serial|serialno)", "", s) def to_base36(n): if n == 0: return "0" out = [] while n: n, r = divmod(n, 36) out.append(B36[r]) return "".join(reversed(out)) def serial_hash(serial, length=16): h = hashlib.sha256(norm_serial(serial).encode("utf-8")).hexdigest() return to_base36(int(h, 16))[:length].ljust(length, "0") def encode_cvin(serial, mfr=MFR, model=MODEL, form=FORM, mem=MEM): tok = lambda v: re.sub(r"[\s_]+", "", str(v).strip().lower()) parts = [tok(x) for x in (mfr, model, form, mem)] for p in parts: if not re.fullmatch(r"[a-z0-9]+", p): raise ValueError("型号四要素只能用小写字母和数字：" + p) return "cvin:%s.%s.%s.%s:%s" % (parts[0], parts[1], parts[2], parts[3], serial_hash(serial)) # ---------- 2. 签名原文（与服务端 canon+sha256 一致） ---------- def canon(v): if v is None or not isinstance(v, (dict, list)): return json.dumps(v, separators=(",", ":"), ensure_ascii=False) if isinstance(v, list): return "[" + ",".join(canon(x) for x in v) + "]" return "{" + ",".join( json.dumps(k, ensure_ascii=False) + ":" + canon(v[k]) for k in sorted(v) ) + "}" def message(cvin, observed_at, telemetry): return hashlib.sha256( canon({"cvin": cvin, "observed_at": observed_at, "telemetry": telemetry}).encode("utf-8") ).hexdigest() # ---------- 3. HTTP ---------- def post(path, body): headers = {"content-type": "application/json", "user-agent": "cvin-onboard/1.0"} if CONFIG.get("api_key"): headers["x-api-key"] = CONFIG["api_key"] req = urllib.request.Request( CONFIG["base"] + path, data=json.dumps(body).encode("utf-8"), headers=headers, method="POST", ) try: with urllib.request.urlopen(req, timeout=20) as r: return r.status, json.loads(r.read() or b"{}") except urllib.error.HTTPError as e: raw = e.read() try: return e.code, json.loads(raw or b"{}") except Exception: return e.code, {"error": raw.decode("utf-8", "replace")[:200]} def post_retry(path, body, tries=4): """限频（429）不是错误，是让你慢一点。按 retry_after_s 等待后重试。""" for i in range(tries): st, r = post(path, body) if st != 429: return st, r wait = int(r.get("retry_after_s") or 30) + 1 print(" 触发限频，等待 %ds 后重试（%d/%d）" % (wait, i + 1, tries)) time.sleep(wait) return st, r # ---------- 4. 主流程 ---------- def main(): from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey from cryptography.hazmat.primitives import serialization if not os.path.exists(CONFIG["serial_file"]): sys.exit("找不到 %s，请先写入序列号（每行一个）" % CONFIG["serial_file"]) serials = [l.strip() for l in open(CONFIG["serial_file"], encoding="utf-8") if l.strip()] keys = json.load(open(KEYFILE)) if os.path.exists(KEYFILE) else {} print("共 %d 台待登记，已有 %d 把设备密钥\n" % (len(serials), len(keys))) ok = fail = 0 for i, sn in enumerate(serials, 1): cvin = encode_cvin(sn) tag = "[%d/%d] %s" % (i, len(serials), cvin) # 4.1 声明（写入免费；同一 CVIN 同来源 10 分钟内去重） st, r = post_retry("/v1/declare", { "cvin": cvin, "operator": CONFIG["operator"], "region": CONFIG["region"], "source_url": "%s://%s" % ("https", CONFIG["operator"]), }) if st == 409 and r.get("error") == "duplicate_write": print("%s 已登记过（10 分钟内去重），继续做签名证明" % tag) elif st not in (201, 200): print("%s 声明失败 %s %s" % (tag, st, r.get("detail") or r.get("error"))) fail += 1 time.sleep(0.3) continue # 4.2 设备密钥（只生成一次，必须落盘备份） if cvin not in keys: sk = Ed25519PrivateKey.generate() priv = sk.private_bytes( serialization.Encoding.PEM, serialization.PrivateFormat.PKCS8, serialization.NoEncryption(), ).decode() pub = sk.public_key().public_bytes( serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo ).decode() keys[cvin] = {"private": priv, "public": pub} json.dump(keys, open(KEYFILE, "w"), indent=1) priv_pem, pub_pem = keys[cvin]["private"], keys[cvin]["public"] # 4.3 签名上报 sk = serialization.load_pem_private_key(priv_pem.encode(), password=None) observed_at = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) telemetry = {"hours_power_on": 0, "ecc_errors": 0, "temp_c": 0} # ← 换成真实采集值 sig = base64.b64encode(sk.sign(message(cvin, observed_at, telemetry).encode())).decode() st2, r2 = post_retry("/v1/attest", { "cvin": cvin, "observed_at": observed_at, "telemetry": telemetry, "device_pubkey": pub_pem, "signature": sig, "operator": CONFIG["operator"], }) if st2 in (200, 201) and r2.get("verified"): print("%s ✓ 已登记，身份 L2，证书 %s/r/%s" % (tag, CONFIG["base"], cvin)) ok += 1 elif st2 == 409: print("%s 密钥不匹配：这台设备此前绑定了另一把密钥，本轮跳过（%s）" % (tag, r2.get("error"))) fail += 1 else: print("%s 签名上报失败 %s %s" % (tag, st2, r2.get("detail") or r2.get("error"))) fail += 1 time.sleep(0.4) # 自助写入限频 3 次/分钟；带密钥后无此约束 print("\n完成：成功 %d / 失败 %d" % (ok, fail)) if keys: print("设备私钥已存 %s —— 务必备份，丢了无法重建（服务端不允许换密钥）" % KEYFILE) if __name__ == "__main__": main()