# -*- coding: utf-8 -*- """md_cg · 版本层（蜂群互联 · 层0） 设计依据：`docs/swarm/蜂群互联_v0.1.md` §2.3 / §3.1 / §10。 版本更新权 = 蜂群**唯一核心权限**（严格唯一，仅 designer）。本模块只做 「声明 + 校验」两件事，不做跨节点证据交换（§9.1 属存在约束，留待 v0.3）： ① 声明 —— `~/.mdcg/theory.json` 记录 theory_version / accepted_versions / declared_by / declared_at，并以 declaration_hash 封缄（防意外损坏）。 ② 校验 —— MCP 启动时 `check()`；不合法即 `theory_ok=False`，主体降级为只读 （写操作全拒，仅保留 theory 修复入口），**不拒绝启动**。 fail-closed 与令牌的差别（有意为之）： · 令牌非法 → **拒绝启动**：身份不可信，不能降级为可用； · 版本非法 → **降级只读**：身份可信但公理状态未知，禁止改写事实层。 存储与 `_tokens.json` 同目录约定（0600）。零第三方依赖。 """ from __future__ import annotations import argparse import hashlib import hmac import json import os import sys import time THEORY_ENV = "MDCG_THEORY" THEORY_FILE_ENV = "MDCG_THEORY_FILE" DEFAULT_THEORY_DIR = os.path.join(os.path.expanduser("~"), ".mdcg") DEFAULT_THEORY_FILE = os.path.join(DEFAULT_THEORY_DIR, "theory.json") SCHEMA = 1 #: 与《智能论》当前版本对齐；仅为**默认声明值**，权威来源是声明文件本身。 PROTOCOL_VERSION = "3.4" #: 被 declaration_hash 封缄的字段（改动其一即校验失败） SEALED_FIELDS = ("theory_version", "accepted_versions", "declared_by", "declared_at") #: theory_ok=False 时仍可执行的 op（唯一修复入口） ESCAPE_OPS = ("theory",) class TheoryError(Exception): """版本层声明非法 / 被篡改 / 越权声明。""" # -------------------------------------------------------------------------- # 存储 # -------------------------------------------------------------------------- # 生效条件：path 为真值时返回 path，否则取 os.environ.get(THEORY_FILE_ENV)，该值为空串/未设置时回落模块常量 DEFAULT_THEORY_FILE。 def theory_file(path: str = None) -> str: return path or os.environ.get(THEORY_FILE_ENV) or DEFAULT_THEORY_FILE # 生效条件：传入可被 json 序列化的 payload（dict）时返回 json.dumps(payload, sort_keys=True, ensure_ascii=False, separators=(",", ":"))。 def _canon(payload: dict) -> str: """规范化 JSON：键排序 + 紧凑分隔符（哈希可复现的唯一前提）。""" return json.dumps(payload, sort_keys=True, ensure_ascii=False, separators=(",", ":")) # 生效条件：payload 可被 _canon 序列化时返回 hashlib.sha256(_canon(payload).encode("utf-8")).hexdigest() 的十六进制摘要。 def declaration_hash(payload: dict) -> str: return hashlib.sha256(_canon(payload).encode("utf-8")).hexdigest() # 生效条件：对 SEALED_FIELDS 中每个 k 用 decl.get(k) 取值（缺键即 None），返回同集合长度的 dict。 def _payload(decl: dict) -> dict: return {k: decl.get(k) for k in SEALED_FIELDS} # 生效条件：以 dict(decl) 为副本 d，写入 d["declaration_hash"]=declaration_hash(_payload(d))，返回该副本（不改动入参 decl）。 def seal(decl: dict) -> dict: """封缄：补上 declaration_hash。""" d = dict(decl) d["declaration_hash"] = declaration_hash(_payload(d)) return d # 生效条件：p=theory_file(path) 后，os.path.exists(p) 为假返回 None；open/json.load 抛 OSError 或 ValueError 也返回 None；否则 d 是 dict 时返回 d，不是 dict 返回 None。 def load(path: str = None): p = theory_file(path) if not os.path.exists(p): return None try: with open(p, encoding="utf-8") as f: d = json.load(f) except (OSError, ValueError): return None return d if isinstance(d, dict) else None # 生效条件：p=theory_file(path)，先 os.makedirs(os.path.dirname(p) or ".")，把 decl 以 ensure_ascii=False, indent=1 写入 p+".tmp" 再 os.replace 到 p，os.chmod(p, 0o600) 抛 OSError 时忽略，最终返回 p。 def save(decl: dict, path: str = None) -> str: p = theory_file(path) os.makedirs(os.path.dirname(p) or ".", exist_ok=True) tmp = p + ".tmp" with open(tmp, "w", encoding="utf-8") as f: json.dump(decl, f, ensure_ascii=False, indent=1) os.replace(tmp, p) try: os.chmod(p, 0o600) except OSError: pass return p # -------------------------------------------------------------------------- # 声明与校验 # -------------------------------------------------------------------------- # 生效条件：v=(version or PROTOCOL_VERSION).strip()，v 为空串时抛 TheoryError；accepted 为假值（None/空）时 acc=[v]，否则 list(accepted) 并剔除 strip 后为空的项，v 不在 acc 时 insert(0, v)；declared_by 假值回落 "designer"，declared_at=float(now if now is not None else time.time())，返回含 SCHEMA 的 dict。 def make_declaration(version: str = None, accepted=None, declared_by: str = "designer", now: float = None) -> dict: """构造（未封缄的）版本声明。accepted 默认只含自身版本。""" v = (version or PROTOCOL_VERSION).strip() if not v: raise TheoryError("theory_version 不能为空") acc = list(accepted) if accepted else [v] acc = [str(x).strip() for x in acc if str(x).strip()] if v not in acc: acc.insert(0, v) return {"schema": SCHEMA, "theory_version": v, "accepted_versions": acc, "declared_by": (declared_by or "designer").strip(), "declared_at": float(now if now is not None else time.time())} # 生效条件：d=decl if decl is not None else load(theory_file(path))；d 为假时返回 theory_ok=True、auto=True、version=PROTOCOL_VERSION 的状态；d.get("schema")!=SCHEMA、hmac.compare_digest(str(d.get("declaration_hash") or ""), declaration_hash(_payload(d))) 不通过、v=d.get("theory_version") 为假、或 v 不在 acc=[str(x) for x in (d.get("accepted_versions") or [])] 时各自返回同一个 base 且 theory_ok=False，其余情况 base 更新为 theory_ok=True、reason="ok"。 def check(path: str = None, decl: dict = None) -> dict: """校验版本声明 → 状态字典（永不抛异常，供启动路径安全调用）。 theory_ok=False 的四种成因互斥且各自可归因： ① 未声明（文件缺失） ② schema 不符 ③ 封缄不匹配（被篡改） ④ 版本不在认可集合 """ p = theory_file(path) d = decl if decl is not None else load(p) base = {"source": p, "theory_ok": False, "version": None, "auto": False, "accepted_versions": [], "declared_by": None, "declared_at": None, "declaration_hash": None, "schema": SCHEMA, "reason": ""} if not d: # 未声明 ≠ 违规：既有部署升级后不应被立刻降级只读。 # 按默认版本放行，并由启动路径 ensure() 落盘声明（声明常态化）。 base.update({"theory_ok": True, "auto": True, "version": PROTOCOL_VERSION, "accepted_versions": [PROTOCOL_VERSION], "reason": f"未声明版本层（按默认版本 {PROTOCOL_VERSION} 运行）"}) return base if d.get("schema") != SCHEMA: base["reason"] = f"声明 schema 不符：{d.get('schema')!r}（期望 {SCHEMA}）" return base if not hmac.compare_digest(str(d.get("declaration_hash") or ""), declaration_hash(_payload(d))): base["reason"] = "声明被篡改（declaration_hash 不匹配）" return base v = d.get("theory_version") acc = [str(x) for x in (d.get("accepted_versions") or [])] base.update({"version": v, "accepted_versions": acc, "declared_by": d.get("declared_by"), "declared_at": d.get("declared_at"), "declaration_hash": d.get("declaration_hash")}) if not v: base["reason"] = "声明缺少 theory_version" return base if v not in acc: base["reason"] = f"版本 {v} 不在认可集合 {acc}" return base base.update({"theory_ok": True, "reason": "ok"}) return base # 生效条件：p=theory_file(path)，仅当 load(p) 为 None 时尝试 declare(PROTOCOL_VERSION, actor="system:autoinit", path=p, append_audit=False) 并吞掉 TheoryError/OSError，最终返回 check(p)。 def ensure(path: str = None) -> dict: """启动路径：无声明则落盘默认声明（声明常态化），再返回校验状态。 落盘失败**不阻断启动**——退回 `check()` 的判定（此时为未声明态，放行）。 """ p = theory_file(path) if load(p) is None: try: declare(PROTOCOL_VERSION, actor="system:autoinit", path=p, append_audit=False) except (TheoryError, OSError): pass return check(p) # 生效条件：version=None/空串时在 make_declaration 内回落 PROTOCOL_VERSION（仅回落结果 strip 后仍为空才抛 TheoryError），seal 后若 append_audit 为真值则把 (old or {}).get("audit") 或 [] 追加一条（仅 old 为真时追加）写入 decl["audit"]，save(decl, path) 后返回 {'ok': True, 'declaration': decl, 'status': check(path)}。 def declare(version: str = None, accepted=None, actor: str = "designer", path: str = None, append_audit: bool = True) -> dict: """写入版本声明（版本更新权的唯一物理落点）。""" old = load(path) decl = make_declaration(version, accepted=accepted, declared_by=actor) decl = seal(decl) if append_audit: hist = list((old or {}).get("audit") or []) if old: hist.append({"at": time.time(), "from": old.get("theory_version"), "to": decl["theory_version"], "by": actor}) decl["audit"] = hist save(decl, path) return {"ok": True, "declaration": decl, "status": check(path)} # 生效条件：以 path 调 load、check、theory_file，返回 {"declaration": load(path), "status": check(path), "theory_file": theory_file(path)}。 def show(path: str = None) -> dict: return {"declaration": load(path), "status": check(path), "theory_file": theory_file(path)} # 生效条件：返回含模块常量 SCHEMA、PROTOCOL_VERSION、list(SEALED_FIELDS)、list(ESCAPE_OPS) 及 theory_file() 与 THEORY_ENV/THEORY_FILE_ENV 的自描述 dict。 def catalog() -> dict: """版本层自描述（供 whoami / service_info 暴露）。""" return {"schema": SCHEMA, "theory_file": theory_file(), "protocol_version": PROTOCOL_VERSION, "sealed_fields": list(SEALED_FIELDS), "escape_ops": list(ESCAPE_OPS), "env": {"version": THEORY_ENV, "file": THEORY_FILE_ENV}} # -------------------------------------------------------------------------- # CLI # -------------------------------------------------------------------------- # 生效条件：把 json.dumps(obj, ensure_ascii=False, indent=1)+"\n" 写入 sys.stdout，无返回值。 def _print(obj): sys.stdout.write(json.dumps(obj, ensure_ascii=False, indent=1) + "\n") # 生效条件：argv 交给 ap.parse_args（argv 为 None 时 argparse 读 sys.argv），子解析器 required=True 缺 cmd 时由 argparse 报错退出；a.cmd 为 declare/check/ensure/show/catalog 之一且未抛 TheoryError 时打印并返回 0，抛出 TheoryError 时写 stderr 返回 2。 def main(argv=None): ap = argparse.ArgumentParser( prog="python -m md_cg.theory", description="版本层（蜂群互联层0）：版本声明与校验（版本更新权唯一）") ap.add_argument("--theory-file", default=None, help="声明文件路径（默认 ~/.mdcg/theory.json）") sub = ap.add_subparsers(dest="cmd", required=True) p_d = sub.add_parser("declare", help="声明版本（仅设计者）") p_d.add_argument("--version", default=None, help=f"默认 {PROTOCOL_VERSION}") p_d.add_argument("--accept", default=None, help="认可版本集合，逗号分隔") p_d.add_argument("--actor", default="designer") sub.add_parser("check", help="校验声明（打印 theory_ok）") sub.add_parser("ensure", help="无声明则落盘默认声明（启动路径用）") sub.add_parser("show", help="打印声明与校验状态") sub.add_parser("catalog", help="版本层自描述") a = ap.parse_args(argv) try: if a.cmd == "declare": acc = [x for x in (a.accept or "").split(",") if x.strip()] or None _print(declare(a.version, accepted=acc, actor=a.actor, path=a.theory_file)) elif a.cmd == "check": _print(check(a.theory_file)) elif a.cmd == "ensure": _print(ensure(a.theory_file)) elif a.cmd == "show": _print(show(a.theory_file)) elif a.cmd == "catalog": _print(catalog()) except TheoryError as e: sys.stderr.write(f"[theory] {e}\n") return 2 return 0 if __name__ == "__main__": sys.exit(main())