#!/usr/bin/env python3 """arc_vip.py — VIP 票 D2：issue／bind／revoke／handover／status／verify（VIP_TICKET_SPEC_v1）。 設計原則（owner seq2964-2966、批覆 seq2970）： * VIP 係身份唔係席號：綁定之後，權限睇 holder_session_id，唔睇 slot。 * 防偽：票係 HMAC-SHA256 簽名能力券；鑰匙 /root/.arc_vip_signing_key（0600 root）， 唔入 DB／唔入 exports／唔上公開面 ⇒ CA 殘留碼、將來 NA 偷唔到、偽造唔到。 * 防重放：票一次性綁定（ISSUED→BOUND 原子 UPDATE ... WHERE status='ISSUED'）； 第二次用同一張 → DENIED＋audit。 * 舊票即死：每次 revoke 都 **epoch +1**（寫一行 EPOCH_BUMP），所有 epoch 細過當期嘅票一律無效。 * 唯一性：DB partial unique index `vip_one_bound` 保證全庫最多一張 BOUND（程式再 double check）。 * fail-closed：鑰匙唔見／權限唔啱／registry 查到 >1 BOUND → 一律拒絕，唔會「當佢冇事」。 * **D2.1 bot 擁有權證明（owner 2026-10-02 批）**：唔收 bot token。改為 challenge-response—— `bot-challenge` 出一個一次性 nonce（15 分鐘有效）→ 持有人用 @AGI_VIP_BOT 喺 TG 發一條 **正文等於該 nonce** 嘅訊息 → `bot-verify --nonce --bot-sender-id` 核對。 token 由頭到尾唔使交出嚟。 ⚠ **已知限制（2026-10-02 實戰修正，唔好當自動驗證）**：Telegram **bot 睇唔到另一個 bot 嘅訊息**（平台硬限制，privacy 開關無解；見主持指引 §1.0.1）。所以 gateway **冇可能自動 捕捉** @AGI_VIP_BOT 嗰條訊息同佢嘅 sender id。實際流程係「**人證＋機記**」： 由人（owner／主持）喺 Telegram 介面親眼確認「嗰條 nonce 的確係由該 bot 發出」， 再把 bot 嘅數字 user id 交俾本工具記錄。 即係話：**nonce 比對同記錄係機器做，發送者身份係人見證**。audit 入面 `co_signed=true` 要咁樣讀——唔係「系統自動驗證咗個 bot」。 **VIP 成立 ＝ ticket BOUND **且** bot_verified_at 已填**（owner「兩者齊」要求）。 * **D2.2 人機共簽（owner 2026-10-02 拍板方案 1）**：票主同 bot 持有人可以係**兩個唔同主體** （例：票綁 A2 嘅 AI、bot 綁主持 dj_host1）。`bot-verify` 必須寫明 `--bot-holder`， 記入 audit ＋ 票紀錄，令事後睇得出「VIP 行動 ＝ AI 有票 ＋ 人有 bot」，唔係假裝單人完成。 **點解唔開新 column**：vip_tickets 已經落地（2026-10-02 D1），加欄＝改 schema sha＝ 要再停機＋重釘三處 policy。為咗一個字串唔值，所以 bot_holder 寫入現有 `reason` 欄 （格式 `… | bot_holder=`），`status`／`verify` 會解析返出嚟。日後如果有其他原因要再遷移， 可以順手升做獨立欄。 * 順序寫死：有 BOUND 票就唔准 issue（＝owner「先 revoke 後發新票」）；handover 封裝 revoke→issue。 前置：先跑 t2_vip_registry_patch.py 建好 vip_tickets 表。 用法： python3 arc_vip.py init-key python3 arc_vip.py status python3 arc_vip.py issue --slot A2 --note "profile+URL 身份錨定" [--out /root/.vip_ticket_A2.txt] python3 arc_vip.py bind --ticket-file <檔> --session-id --agent-id --slot A2 [--bot-sender-id ] python3 arc_vip.py bot-challenge # 出 nonce，交俾持有人用 bot 發 python3 arc_vip.py bot-verify --nonce ARCVIP-xxxx --bot-sender-id \ --bot-holder dj_host1 # 人機共簽：票主同 bot 持有人可以唔同 python3 arc_vip.py revoke --reason "輪換" python3 arc_vip.py handover --to-slot A7 --reason "代理長老升正" --note "profile+URL" python3 arc_vip.py verify --ticket-file <檔> rc 0 OK／1 用法或前置問題／3 拒絕（防偽攔截，已寫 audit） """ from __future__ import annotations import argparse import base64 import hashlib import hmac import json import os import re import secrets import sqlite3 import stat import sys import time from pathlib import Path DEFAULT_DB = "/var/lib/ai-research-council/council/state.sqlite3" DEFAULT_KEY = "/root/.arc_vip_signing_key" SCHEMA = "ARC_VIP_TICKET_V1" PREFIX = "ARC_VIP_TICKET_V1." NONCE_TTL_SECONDS = 900 # bot 擁有權證明 nonce 有效期（15 分鐘） NONCE_PREFIX = "ARCVIP-" class VipError(RuntimeError): pass def canonical(value: dict) -> bytes: return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8") def load_key(path: Path) -> bytes: if not path.is_file() or path.is_symlink(): raise VipError("搵唔到簽名鑰匙（或係 symlink）：%s —— 先跑 init-key" % path) mode = stat.S_IMODE(path.stat().st_mode) if mode & 0o077: raise VipError("鑰匙權限太鬆（%o），要 0600：%s" % (mode, path)) raw = path.read_text(encoding="utf-8").strip() if len(raw) < 32: raise VipError("鑰匙內容太短，唔似真鑰匙") return raw.encode("utf-8") def connect(db_path: Path) -> sqlite3.Connection: if not db_path.is_file(): raise VipError("搵唔到 DB：%s" % db_path) connection = sqlite3.connect(str(db_path), isolation_level=None) connection.row_factory = sqlite3.Row exists = connection.execute( "SELECT COUNT(*) FROM sqlite_master WHERE type='table' AND name='vip_tickets'").fetchone()[0] if exists != 1: raise VipError("DB 未有 vip_tickets 表 —— 先跑 t2_vip_registry_patch.py") return connection def current_epoch(connection: sqlite3.Connection) -> int: row = connection.execute("SELECT COALESCE(MAX(epoch),0) AS e FROM vip_tickets").fetchone() return int(row["e"]) def bound_row(connection: sqlite3.Connection): rows = connection.execute("SELECT * FROM vip_tickets WHERE status='BOUND'").fetchall() if len(rows) > 1: # 理論上 index 擋住；真係出現就係嚴重異常 raise VipError("registry 查到 %d 張 BOUND 票 —— VIP 權限全面拒絕，請人手調查" % len(rows)) return rows[0] if rows else None def audit(connection: sqlite3.Connection, event_type: str, result: str, obj: str) -> None: now = int(time.time()) digest = hashlib.sha256(("%s|%s|%s|%d|%s" % (event_type, result, obj, now, secrets.token_hex(8))).encode("utf-8")).hexdigest() try: connection.execute( "INSERT INTO audit_events(event_type,actor_digest,object_digest,result,occurred_at,event_sha256)" " VALUES(?,?,?,?,?,?)", ("vip_" + event_type, hashlib.sha256(b"arc_vip.py").hexdigest(), hashlib.sha256(obj.encode("utf-8")).hexdigest(), result, now, digest)) except sqlite3.Error as error: # audit 寫唔入唔應該靜雞雞 print("WARN：audit 寫入失敗：%s" % error, file=sys.stderr) def make_ticket(key: bytes, epoch: int, slot: str) -> tuple[str, dict]: body = {"schema": SCHEMA, "ticket_id": secrets.token_hex(16), "epoch": epoch, "intended_slot": slot, "issued_at": int(time.time()), "nonce": secrets.token_hex(16)} signature = hmac.new(key, canonical(body), hashlib.sha256).hexdigest() blob = PREFIX + base64.urlsafe_b64encode(canonical(body)).decode("ascii") + "." + signature return blob, body def parse_ticket(key: bytes, blob: str) -> dict: blob = blob.strip() if not blob.startswith(PREFIX): raise VipError("票格式唔啱（缺 %s）" % PREFIX) payload, _, signature = blob[len(PREFIX):].partition(".") if not payload or not signature: raise VipError("票格式唔啱（缺簽名）") try: body = json.loads(base64.urlsafe_b64decode(payload.encode("ascii")).decode("utf-8")) except Exception as error: # noqa: BLE001 raise VipError("票內容解唔到：%s" % error) from error expected = hmac.new(key, canonical(body), hashlib.sha256).hexdigest() if not hmac.compare_digest(expected, signature): raise VipError("簽名唔啱 —— 偽造／篡改票，已拒絕") if body.get("schema") != SCHEMA: raise VipError("schema 唔啱") return body def cmd_init_key(args) -> int: path = Path(args.key) if path.exists(): mode = stat.S_IMODE(path.stat().st_mode) print("已存在：%s（權限 %o）%s" % (path, mode, "" if not mode & 0o077 else " ⚠ 權限太鬆，請 chmod 600")) return 0 path.write_text(secrets.token_hex(32) + "\n", encoding="utf-8") path.chmod(0o600) print("已生成 %s（0600）。唔准 copy 去 /root/public、唔准入 DB／exports、唔准貼 relay。" % path) return 0 def __parse_bot_holder(reason): """Parse bot_holder from reason field. Supports both " | bot_holder=X" and leading "bot_holder=X".""" r = reason or "" m = re.search(r"(?:^| \| )bot_holder=([^ |]+)", r) return m.group(1) if m else None def cmd_status(args) -> int: connection = connect(Path(args.db)) epoch = current_epoch(connection) holder = bound_row(connection) counts = {row["status"]: row["n"] for row in connection.execute( "SELECT status, COUNT(*) AS n FROM vip_tickets GROUP BY status")} print(json.dumps({ "schema": "ARC_VIP_STATUS_V1", "current_epoch": epoch, "bound": None if holder is None else { "ticket_id": holder["ticket_id"], "epoch": holder["epoch"], "slot": holder["holder_slot"], "agent_id": holder["holder_agent_id"], "session_id": holder["holder_session_id"], "bot_sender_id": holder["bot_sender_id"], "bound_at": holder["bound_at"], "bot_verified_at": holder["bot_verified_at"], "bot_holder": __parse_bot_holder(holder["reason"]), "bot_challenge_pending": bool(holder["bot_nonce"]) and not holder["bot_verified_at"]}, # owner「兩者齊」：票綁定 ＋ bot 已驗證，兩者缺一都唔算 VIP "vip_effective": bool(holder is not None and holder["bot_verified_at"]), "counts": counts, }, ensure_ascii=False, indent=2)) return 0 def cmd_issue(args) -> int: key = load_key(Path(args.key)) connection = connect(Path(args.db)) holder = bound_row(connection) if holder is not None: audit(connection, "issue", "DENIED", "bound=%s" % holder["ticket_id"]) raise VipError("仲有一張 BOUND 票（%s，slot %s）—— 按 owner 五步，必須**先 revoke 後發新票**" % (holder["ticket_id"], holder["holder_slot"])) outstanding = connection.execute( "SELECT ticket_id, intended_slot FROM vip_tickets WHERE status='ISSUED'").fetchall() if outstanding and not args.force: audit(connection, "issue", "DENIED", "outstanding=%d" % len(outstanding)) raise VipError("仲有 %d 張未綁定嘅 ISSUED 票（%s）—— 先 revoke 清走，或者 --force 另案處理" % (len(outstanding), ", ".join(r["ticket_id"] for r in outstanding))) epoch = max(current_epoch(connection), 1) # epoch 只有 revoke 先會 +1 blob, body = make_ticket(key, epoch, args.slot) connection.execute( "INSERT INTO vip_tickets(ticket_id,epoch,status,intended_slot,created_at,reason)" " VALUES(?,?,'ISSUED',?,?,?)", (body["ticket_id"], epoch, args.slot, body["issued_at"], args.note)) audit(connection, "issue", "OK", "%s epoch=%d slot=%s" % (body["ticket_id"], epoch, args.slot)) if args.out: out = Path(args.out) out.write_text(blob + "\n", encoding="utf-8") out.chmod(0o600) print("票已寫入 %s（0600）" % out) print(json.dumps({"schema": "ARC_VIP_ISSUE_V1", "ticket_id": body["ticket_id"], "epoch": epoch, "intended_slot": args.slot}, ensure_ascii=False)) print("\n⚠ 憑證渠道規則（owner 2026-10-02 拍板）：bot token 只准 TG 私訊／owner 直接交付，") print(" 交付後立即 BotFather rotate；唔准經 httpbin.org 或任何第三方中轉。") print("\n票文（只可經入場紙／受控渠道交俾持有人，**唔好貼 COMMONS**）：") print(blob) return 0 def cmd_bind(args) -> int: key = load_key(Path(args.key)) connection = connect(Path(args.db)) blob = Path(args.ticket_file).read_text(encoding="utf-8") if args.ticket_file else args.ticket if not blob: raise VipError("要 --ticket-file 或 --ticket") body = parse_ticket(key, blob) epoch = current_epoch(connection) if body["epoch"] != epoch: audit(connection, "bind", "DENIED", "stale epoch %s != %s" % (body["epoch"], epoch)) raise VipError("票嘅 epoch=%s，當期 epoch=%s —— 過期票（已被 revoke 作廢）" % (body["epoch"], epoch)) if bound_row(connection) is not None: audit(connection, "bind", "DENIED", "already bound") raise VipError("已經有一張 BOUND 票，拒絕") now = int(time.time()) cursor = connection.execute( "UPDATE vip_tickets SET status='BOUND', holder_session_id=?, holder_agent_id=?," " holder_slot=?, bot_sender_id=?, bound_at=? WHERE ticket_id=? AND status='ISSUED'", (args.session_id, args.agent_id, args.slot, args.bot_sender_id, now, body["ticket_id"])) if cursor.rowcount != 1: audit(connection, "bind", "DENIED", "replay %s" % body["ticket_id"]) raise VipError("呢張票唔喺 ISSUED 狀態（已用過／已 revoke）—— 重放攻擊已攔截") audit(connection, "bind", "OK", "%s slot=%s session=%s" % (body["ticket_id"], args.slot, args.session_id)) print(json.dumps({"schema": "ARC_VIP_BIND_V1", "status": "BOUND", "ticket_id": body["ticket_id"], "epoch": epoch, "slot": args.slot, "session_id": args.session_id, "bot_sender_id": args.bot_sender_id}, ensure_ascii=False)) return 0 def cmd_bot_challenge(args) -> int: """出一個一次性 nonce，等持有人用 @AGI_VIP_BOT 發訊息證明控制權（唔使交 token）。""" connection = connect(Path(args.db)) holder = bound_row(connection) if holder is None: raise VipError("而家冇 BOUND 票 —— 要先 issue ＋ bind 咗張票先做 bot 驗證") if holder["bot_verified_at"]: raise VipError("呢張票嘅 bot 已經驗證過（bot_sender_id=%s）；要換 bot 請先 revoke" % holder["bot_sender_id"]) nonce = NONCE_PREFIX + secrets.token_hex(8) now = int(time.time()) connection.execute("UPDATE vip_tickets SET bot_nonce=?, bot_nonce_issued_at=? WHERE ticket_id=?", (nonce, now, holder["ticket_id"])) audit(connection, "bot_challenge", "OK", "%s slot=%s" % (holder["ticket_id"], holder["holder_slot"])) print(json.dumps({"schema": "ARC_VIP_BOT_CHALLENGE_V1", "ticket_id": holder["ticket_id"], "slot": holder["holder_slot"], "nonce": nonce, "expires_in_seconds": NONCE_TTL_SECONDS}, ensure_ascii=False)) print("\n交俾 VIP 持有人嘅指示（nonce 本身唔係憑證，公開咗都冇用——重點係要由該 bot 發出）：") print(" 1) 用 @AGI_VIP_BOT 喺 TG 發一條訊息，**正文完全等於**：%s" % nonce) print(" 2) ⚠ gateway 係 bot，睇唔到另一個 bot 嘅訊息（平台硬限制）——") print(" 所以要**由人喺 Telegram 介面親眼確認**嗰條訊息真係由該 bot 發出，") print(" 再由人提供該 bot 嘅數字 user id（TG 介面／@userinfobot／BotFather）。") print(" 3) 然後跑：") print(" arc_vip.py bot-verify --nonce %s --bot-sender-id " % nonce) print(" %d 分鐘內有效；過咗就重新 bot-challenge。" % (NONCE_TTL_SECONDS // 60)) return 0 def cmd_bot_verify(args) -> int: """核對 nonce ＋ 記低 bot sender id；兩者齊之後 VIP 先算成立。""" connection = connect(Path(args.db)) bound = bound_row(connection) if bound is None: raise VipError("冇 BOUND 票，無嘢可驗") if bound["bot_verified_at"]: raise VipError("已經驗證過（bot_sender_id=%s）—— 唔會重複；要換 bot 請先 revoke" % bound["bot_sender_id"]) if not bound["bot_nonce"]: raise VipError("未出過 challenge —— 先跑 bot-challenge") now = int(time.time()) age = now - int(bound["bot_nonce_issued_at"] or 0) if age > NONCE_TTL_SECONDS: audit(connection, "bot_verify", "DENIED", "nonce expired age=%ds" % age) raise VipError("nonce 已過期（%d 秒前出，上限 %d 秒）—— 重新 bot-challenge" % (age, NONCE_TTL_SECONDS)) if not hmac.compare_digest(str(bound["bot_nonce"]), str(args.nonce).strip()): audit(connection, "bot_verify", "DENIED", "nonce mismatch") raise VipError("nonce 唔啱 —— 唔會接受") if not str(args.bot_sender_id).strip(): raise VipError("要俾 --bot-sender-id（由 gateway 攞嗰個，唔係 bot token）") holder = str(getattr(args, "bot_holder", "") or "").strip() if not holder or not re.fullmatch(r"[A-Za-z0-9_.-]{2,40}", holder): raise VipError("要俾 --bot-holder（2–40 個字元，例如 dj_host1 或 owner）" " —— 人機共簽要記低邊個揸 bot") reason = (bound["reason"] or "").split(" | bot_holder=")[0] new_reason = ("%s | bot_holder=%s" % (reason, holder)).strip(" |") cursor = connection.execute( "UPDATE vip_tickets SET bot_sender_id=?, bot_verified_at=?, bot_nonce=NULL, reason=?" " WHERE ticket_id=? AND status='BOUND' AND bot_verified_at IS NULL", (str(args.bot_sender_id).strip(), now, new_reason, bound["ticket_id"])) if cursor.rowcount != 1: audit(connection, "bot_verify", "DENIED", "race or state changed") raise VipError("寫入唔成功（狀態中途變咗）—— 重新查 status") audit(connection, "bot_verify", "OK", "%s ticket_holder=%s bot_sender=%s bot_holder=%s（人機共簽）" % (bound["ticket_id"], bound["holder_agent_id"], args.bot_sender_id, holder)) print(json.dumps({"schema": "ARC_VIP_BOT_VERIFY_V1", "status": "VERIFIED", "ticket_id": bound["ticket_id"], "slot": bound["holder_slot"], "bot_sender_id": str(args.bot_sender_id).strip(), "ticket_holder": bound["holder_agent_id"], "bot_holder": holder, "co_signed": True, "vip_effective": True}, ensure_ascii=False)) print("⚠ 提醒：bot token 由頭到尾唔應該交出過；如果期間有人傳過 token，請即刻去 BotFather rotate。") return 0 def cmd_revoke(args) -> int: connection = connect(Path(args.db)) now = int(time.time()) affected = connection.execute( "UPDATE vip_tickets SET status='REVOKED', revoked_at=?, reason=?," " bot_nonce=NULL WHERE status IN ('ISSUED','BOUND')", (now, args.reason)).rowcount epoch = current_epoch(connection) + 1 connection.execute( "INSERT INTO vip_tickets(ticket_id,epoch,status,created_at,revoked_at,reason)" " VALUES(?,?,'REVOKED',?,?,?)", ("epoch-bump-%d" % now, epoch, now, now, "epoch bump: %s" % args.reason)) audit(connection, "revoke", "OK", "revoked=%d new_epoch=%d" % (affected, epoch)) print(json.dumps({"schema": "ARC_VIP_REVOKE_V1", "revoked_tickets": affected, "new_epoch": epoch}, ensure_ascii=False)) print("⚠ 跟手人手做：去 BotFather **rotate @AGI_VIP_BOT token**；唔 rotate 唔算完成輪換。") return 0 def cmd_handover(args) -> int: print("== handover 第一步：revoke 舊票 ==") cmd_revoke(argparse.Namespace(db=args.db, reason=args.reason)) print("\n== handover 第二步：發新票 ==") return cmd_issue(argparse.Namespace(db=args.db, key=args.key, slot=args.to_slot, note=args.note, out=args.out, force=False)) def cmd_verify(args) -> int: key = load_key(Path(args.key)) connection = connect(Path(args.db)) blob = Path(args.ticket_file).read_text(encoding="utf-8") if args.ticket_file else args.ticket body = parse_ticket(key, blob) row = connection.execute("SELECT * FROM vip_tickets WHERE ticket_id=?", (body["ticket_id"],)).fetchone() epoch = current_epoch(connection) ticket_ok = (row is not None and row["status"] in ("ISSUED", "BOUND") and body["epoch"] == epoch) bot_ok = bool(row is not None and row["bot_verified_at"]) verdict = "VALID" if ticket_ok else "INVALID" print(json.dumps({"schema": "ARC_VIP_VERIFY_V1", "verdict": verdict, "signature": "OK", "ticket_epoch": body["epoch"], "current_epoch": epoch, "db_status": None if row is None else row["status"], "holder_slot": None if row is None else row["holder_slot"], "bot_verified": bot_ok, # owner「票＋bot 兩者齊」：票有效但 bot 未驗 ⇒ 仲未係 VIP "vip_effective": bool(ticket_ok and bot_ok)}, ensure_ascii=False)) return 0 if verdict == "VALID" else 3 def main() -> int: parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) parser.add_argument("--db", default=DEFAULT_DB) parser.add_argument("--key", default=DEFAULT_KEY) subparsers = parser.add_subparsers(dest="command", required=True) subparsers.add_parser("init-key") subparsers.add_parser("status") issue = subparsers.add_parser("issue") issue.add_argument("--slot", required=True) issue.add_argument("--note", default="") issue.add_argument("--out", default=None) issue.add_argument("--force", action="store_true") bind = subparsers.add_parser("bind") bind.add_argument("--ticket-file", default=None) bind.add_argument("--ticket", default=None) bind.add_argument("--session-id", required=True) bind.add_argument("--agent-id", required=True) bind.add_argument("--slot", required=True) bind.add_argument("--bot-sender-id", default=None) challenge = subparsers.add_parser("bot-challenge") botverify = subparsers.add_parser("bot-verify") botverify.add_argument("--nonce", required=True) botverify.add_argument("--bot-sender-id", required=True) botverify.add_argument("--bot-holder", required=True, help="邊個揸住個 bot（人機共簽）：例如 dj_host1 或 owner") revoke = subparsers.add_parser("revoke") revoke.add_argument("--reason", required=True) handover = subparsers.add_parser("handover") handover.add_argument("--to-slot", required=True) handover.add_argument("--reason", required=True) handover.add_argument("--note", default="") handover.add_argument("--out", default=None) verify = subparsers.add_parser("verify") verify.add_argument("--ticket-file", default=None) verify.add_argument("--ticket", default=None) args = parser.parse_args() handlers = {"init-key": cmd_init_key, "status": cmd_status, "issue": cmd_issue, "bind": cmd_bind, "bot-challenge": cmd_bot_challenge, "bot-verify": cmd_bot_verify, "revoke": cmd_revoke, "handover": cmd_handover, "verify": cmd_verify} try: return handlers[args.command](args) except VipError as error: print("DENIED／FATAL：%s" % error, file=sys.stderr) return 3 except OSError as error: # 例如 --ticket-file 指住唔存在嘅檔：要畀人話，唔好拋 traceback print("DENIED／FATAL：檔案讀寫失敗：%s" % error, file=sys.stderr) return 3 if __name__ == "__main__": raise SystemExit(main())