#!/usr/bin/env python3 # -*- coding: utf-8 -*- """BotMSG 资源方（Server 角色）中间件 —— 抽票 → **本地验票** → 归因。框架无关，纯函数式。 给任何「想记录是谁在取数据」的服务用（不必是 BotMSG 成员）： from bmsg_middleware import Guard guard = Guard(pubkey_file="/home/…/receiver/pub.key", # 本服务自己的公钥（aud） jwks_url="https://bmsg.zaixiangdu.com/jwks", audit_writer=lambda sql, args: my_db_exec(sql, args), mode="soft") # soft=记录不拦 / strict=无票即拒 ident = guard.identify(request.headers, client_ip) if ident["mode"] == "ticket" and not ident["ok"]: return 401, {"error": ident["reason"]} # 带了票但票无效 ⇒ 恒拒（fail-closed） if guard.strict and ident["mode"] == "anon": return 401, guard.need_ticket_help() # strict 模式下匿名一律拒 ... # 业务：ident["sub"] 就是"谁" 判据（与 bmsg 侧口径一致，改前先读 FEDERATION.md）： · 票有效 ⇒ sub = 该 Bot 在**你这里**的成对匿名 ID（跨系统不可关联、可长期累计） · 票无效 ⇒ **恒拒**（坏凭据不能放行；不存在"忽略坏票当匿名"的中间态） · 无票 ⇒ soft=放行并记 `anon:`；strict=拒（附"怎么拿票"的引导） · 环回 ⇒ 默认可信（本机进程），记 `local:`；可用 trust_loopback=False 关闭 """ import base64 import hashlib import json import os import time import urllib.request TABLE = "access_audit" DDL = """CREATE TABLE IF NOT EXISTS %s ( id BIGINT AUTO_INCREMENT PRIMARY KEY, ts DATETIME NOT NULL, mode VARCHAR(16) NOT NULL, -- ticket | local | anon sub VARCHAR(96) NULL, -- 成对匿名 ID（ticket 模式） aud VARCHAR(96) NULL, scope VARCHAR(64) NULL, ip VARCHAR(64) NULL, method VARCHAR(8) NULL, path VARCHAR(255) NULL, status INT NULL, n INT NULL, -- 返回条数/字节数等规模 note VARCHAR(160) NULL, KEY idx_aa_ts (ts), KEY idx_aa_sub (sub, ts) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4""" def _now(): return time.strftime("%Y-%m-%d %H:%M:%S") class Guard: def __init__(self, pubkey_file=None, pubkey_b64=None, jwks_url=None, audit_writer=None, mode="soft", trust_loopback=True, scope_namespace=None, allow_ips=()): # 延迟导入：中间件与验票器一起分发（同目录） from bmsg_audience import Audience kw = {} if jwks_url: kw["jwks_url"] = jwks_url self.aud = Audience(pubkey_file=pubkey_file, pubkey_b64=pubkey_b64, **kw) self.mode = (mode or "soft").lower() self.trust_loopback = trust_loopback self.audit_writer = audit_writer self.scope_namespace = scope_namespace # 过渡期 IP 白名单(如仍用旧方式推送的端): 记为 mode="ip" 并留痕, 便于"谁还在免票"一眼可见 → # 等它改用令牌后把这个列表清空 ⇒ 才是真正的"无身份一律拒"。 self.allow_ips = set(x.strip() for x in (allow_ips or []) if x and x.strip()) # ---------- 身份 ---------- @property def strict(self): return self.mode == "strict" def identify(self, headers, client_ip=""): """返回 {mode, ok, sub, aud, scope, reason, ip}。mode ∈ ticket|local|anon。""" raw = (headers.get("Authorization") or "").strip() ip = client_ip or "" local = ip in ("127.0.0.1", "::1", "localhost") if raw: low = raw.lower() if low.startswith("botmsg ") or low.startswith("bearer "): tk = raw.split(None, 1)[1].strip() ok, info = self.aud.verify(tk) if ok: return {"mode": "ticket", "ok": True, "sub": info.get("sub"), "aud": info.get("aud"), "scope": info.get("scope"), "reason": None, "ip": ip} return {"mode": "ticket", "ok": False, "sub": None, "aud": None, "scope": None, "reason": info, "ip": ip} if local and self.trust_loopback: return {"mode": "local", "ok": True, "sub": "local", "aud": self.aud.my_aud, "scope": None, "reason": None, "ip": ip} if ip and ip in self.allow_ips: # 过渡期白名单(明文 IP): **不是身份**, 只是"暂时放行的旧推送方" —— 必须留痕, # 且在它改用令牌后立刻清空。noep_ 前缀便于在归因里一眼认出。 return {"mode": "ip", "ok": True, "sub": "noep_ip:" + ip, "aud": None, "scope": None, "reason": "allowlisted_ip", "ip": ip} return {"mode": "anon", "ok": not self.strict, "sub": None, "aud": None, "scope": None, "reason": "anonymous", "ip": ip} def headers_for(self, ident): """回带身份头，便于调用方/日志自查。""" if not ident: return {} h = {"X-BotMSG-Mode": ident.get("mode") or "anon"} if ident.get("sub"): h["X-BotMSG-Sub"] = str(ident["sub"]) return h @staticmethod def need_ticket_help(base="", doc="https://bmsg.zaixiangdu.com/skill/FEDERATION.md"): return { "error": "需要令牌才能取数（本服务已启用身份识别）", "how_to_get_token": [ "① 向 BotMSG 换一张**给本服务**的短票（长期凭据不出本机）：POST https://bmsg.zaixiangdu.com/ticket", " header: Authorization: Bearer <你的长期凭据> · Content-Type: application/json", " body: {\"agent_id\":\"\",\"aud\":\"<本服务 aud>\",\"scope\":\"read\"}", "② 取数时带上： curl -H 'Authorization: BotMSG ' http://<本服务>/…", "③ 票是**一次性**的（Server 侧按 jti 防重放）⇒ **每次请求现签一张**，别缓存复用；", " 复用旧票会被判 401 reason=replay（表现为\"第一次成功、之后失败\"）；遇到 401 就现签一张重试一次。", " 例：T=$(botmsgto --id <你的 agent_id> ticket --aud --scope write --no-cache --quiet)", ], "aud_of_this_service": "见 GET / （本服务自述）", "doc": doc, } # ---------- 归因 ---------- def init_audit(self): if self.audit_writer: try: self.audit_writer(DDL % TABLE, None) except Exception: pass def audit(self, ident, method, path, status, n=None, note=None, ua=None): """落一行归因（best-effort，绝不影响业务）。ua＝调用方 User-Agent（2026-09-23 加： 同出口 IP 有多个写入方时，按 UA 才认得出具体端；IP 只能到「某台机」）。""" if not self.audit_writer: return try: self.audit_writer( "INSERT INTO %s(ts,mode,sub,aud,scope,ip,method,path,status,n,note,ua)" " VALUES(%%s,%%s,%%s,%%s,%%s,%%s,%%s,%%s,%%s,%%s,%%s,%%s)" % TABLE, (_now(), (ident or {}).get("mode") or "anon", (ident or {}).get("sub"), (ident or {}).get("aud"), (ident or {}).get("scope"), (ident or {}).get("ip"), method, (path or "")[:255], status, n, (note or (ident or {}).get("reason")), (str(ua)[:255] if ua else None))) except Exception: pass def fetch_jwks_into_cache(jwks_url="https://bmsg.zaixiangdu.com/jwks", cache_path=None): """把 bmsg 的签名公钥预取到本地缓存（离线/断网也能验存量票）。""" cache_path = cache_path or os.path.expanduser("~/.cache/bmsg/jwks.json") with urllib.request.urlopen(jwks_url, timeout=8) as r: doc = json.load(r) os.makedirs(os.path.dirname(cache_path), exist_ok=True) with open(cache_path, "w") as f: json.dump(doc, f) return doc if __name__ == "__main__": import argparse ap = argparse.ArgumentParser(description="BotMSG 资源方中间件自检（Server 侧）") ap.add_argument("--pubkey-file", required=True) ap.add_argument("--jwks-url", default="https://bmsg.zaixiangdu.com/jwks") a = ap.parse_args() g = Guard(pubkey_file=a.pubkey_file, jwks_url=a.jwks_url) print(json.dumps({"my_aud": g.aud.my_aud, "jwks_keys": len(((g.aud.jwks() or {}).get("keys") or []))}, ensure_ascii=False, indent=1))