From 9759be47bbc742bc8aa365b6ccdba70ee5b47022 Mon Sep 17 00:00:00 2001 From: host Date: Fri, 6 Nov 2026 22:38:19 +0000 Subject: [PATCH] =?UTF-8?q?feat(chv):=20=E5=AF=B9=E9=BD=90=20Muse=20P12?= =?UTF-8?q?=E2=80=94=E2=80=94=E5=AE=A1=E8=AE=A1=E8=BD=A8=E8=BF=B9=E4=B8=8E?= =?UTF-8?q?=E6=B4=BB=E5=8A=A8=E8=A7=86=E5=9B=BE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 统一审计记录模型（时间/主体/动作/对象/结果/原因/关联任务/ 关联审批）+ 追加写 JSONL + SHA-256 哈希链（改写/删除/调序可 被 verify 发现）+ 字段级白名单清洗（密钥/token 值/对话原文 绝不进审计）；事件分类覆盖任务派发、VM 生命周期、同步、审批、 支付、配额、记忆、后台、推理/遥测；from_legacy_row 承接现有 audit_log 行；导出选 JSON。 活动视图后端：计划登记（已做 vs 计划做，流转受控）+ 已批准 权限登记（接 P5 describe 形状）+ activity_summary 纯函数； 结构化审批卡数据（风险分级 + 接受/拒绝明确动作 + 五档授权 选项，永久档须显式 opt-in）+ card_matches_channel 校验卡与 P5 通道载荷一致。 只做后端 + 数据接口，前端视图留接口；P3/P5 不在本基线， 接线走 dict 契约，集成由 round-11 合并；审计持久化先落独立 JSONL，待 P7 状态库就绪后迁移。真机验证排队在 round-11 之后。 新测试 43/43，全量 cloud 套件 1910 OK（skipped=8）； mypy 在改动文件上干净。 --- cloud/chv/activity.py | 274 ++++++++++++++++++ cloud/chv/approval_card.py | 172 +++++++++++ cloud/chv/audit_trail.py | 343 ++++++++++++++++++++++ cloud/tests/test_chv_muse_p12.py | 471 +++++++++++++++++++++++++++++++ docs/chv-muse-p12-gap.md | 95 +++++++ 5 files changed, 1355 insertions(+) create mode 100644 cloud/chv/activity.py create mode 100644 cloud/chv/approval_card.py create mode 100644 cloud/chv/audit_trail.py create mode 100644 cloud/tests/test_chv_muse_p12.py create mode 100644 docs/chv-muse-p12-gap.md diff --git a/cloud/chv/activity.py b/cloud/chv/activity.py new file mode 100644 index 0000000..c519441 --- /dev/null +++ b/cloud/chv/activity.py @@ -0,0 +1,274 @@ +"""活动视图后端（feat-chv-muse-p12，纯标准库 + atomic_file）。 + +接 Muse 官方要求（报告 §2.2 第 7 步、§2.6）：完整审计轨迹展示 +已做与计划做的事；头像下方显示正在做什么的摘要，点头像可见 +完整活动日志与已批准权限；审批是严格能力，绑定连接器/目的地 +与用例。 + +本模块只给后端 + 数据接口，前端视图（客户端 UI）是另一侧的 +工作，契约见各函数 docstring。 + +三件东西： + +- ``PlanStore``：计划登记——已做 vs 计划做。状态只认 + pending/running/done/cancelled，流转 pending→running/ + cancelled、running→done/cancelled，终态不可再动，非法流转 + 抛 ValueError。落 JSON 文件，原子写（0600）。 +- ``PermissionRegistry``：已批准权限登记。接 P5 + ``ScopedCapability.describe()`` 的 dict 形状（capability_id / + tier / connector / host / use_case / session_id / task_id / + methods / max_uses / uses_left / expires_at / revoked），不 + import P5 模块——P5 不在本基线里，集成时由 round-11 合并。 + ``list_active()`` 只返回未撤销、未过期、次数未用完的授权。 +- ``activity_summary()``：纯函数，把计划列表 + 近期审计记录 + + 有效授权拼成活动摘要：正在做 / 计划中 / 已做 / 近期记录 / + 已批准权限数。正在做来自计划登记与审计记录，不是实时进程 + 探针；VM 内真实执行状态以执行器为准。 +""" + +import json +import os +import time + +from cloud.chv.atomic_file import atomic_write_text + +# 计划状态与合法流转（终态 done/cancelled 不可再动）。 +PLAN_STATUSES = ("pending", "running", "done", "cancelled") +_PLAN_TRANSITIONS: dict[str, tuple[str, ...]] = { + "pending": ("running", "cancelled"), + "running": ("done", "cancelled"), + "done": (), + "cancelled": (), +} + + +def _load_json(path: str) -> dict: + """读 JSON 文件：不存在或坏掉返回空 dict（调用方重建）。""" + if not os.path.exists(path): + return {} + try: + with open(path, "r", encoding="utf-8") as f: + data = json.load(f) + except (ValueError, OSError): + return {} + return data if isinstance(data, dict) else {} + + +def _save_json(path: str, data: dict) -> None: + """原子写 JSON 文件（0600，写失败原样抛错）。""" + text = json.dumps(data, ensure_ascii=False, indent=2) + atomic_write_text(path, text, mode=0o600) + + +class PlanStore: + """计划登记：一个项目一串计划，状态流转受控。""" + + def __init__(self, path: str) -> None: + self.path = path + self._data = _load_json(path) + + def add_plan(self, project_id: str, title: str, + task_id: str = "", + ts: float | None = None) -> dict: + """登记一条计划（初始 pending），返回该计划。""" + if not project_id: + raise ValueError("project_id 不能为空") + if not title: + raise ValueError("title 不能为空") + now = ts if ts is not None else time.time() + plan = { + "plan_id": "%d-%s" % (int(now * 1000), + str(len(self._plans(project_id)))), + "project_id": project_id, + "title": title, + "task_id": task_id, + "status": "pending", + "created_at": now, + "updated_at": now, + } + self._data.setdefault(project_id, []).append(plan) + _save_json(self.path, self._data) + return dict(plan) + + def set_status(self, project_id: str, plan_id: str, + status: str, ts: float | None = None) -> dict: + """改计划状态：非法流转或未知 plan_id 抛 ValueError。""" + if status not in PLAN_STATUSES: + raise ValueError("未知计划状态：%s" % status) + for plan in self._plans(project_id): + if plan["plan_id"] != plan_id: + continue + allowed = _PLAN_TRANSITIONS[plan["status"]] + if status not in allowed: + raise ValueError( + "计划 %s 不能从 %s 到 %s" + % (plan_id, plan["status"], status)) + plan["status"] = status + plan["updated_at"] = ts if ts is not None else time.time() + _save_json(self.path, self._data) + return dict(plan) + raise ValueError("未知计划：%s" % plan_id) + + def list_plans(self, project_id: str, status: str = "") -> list: + """该项目的计划（副本，按创建时间排序，可按状态过滤）。""" + plans = [dict(p) for p in self._plans(project_id)] + plans.sort(key=lambda p: p.get("created_at", 0.0)) + if status: + plans = [p for p in plans if p.get("status") == status] + return plans + + def _plans(self, project_id: str) -> list: + """该项目的计划列表（内部引用，调用方不得外泄）。""" + items = self._data.get(project_id, []) + return items if isinstance(items, list) else [] + + +class PermissionRegistry: + """已批准权限登记：接 P5 能力描述，不存任何凭证值。 + + 授权只含绑定（连接器/主机/用例/档位/有效期/次数余量）， + 与 P5 ``ScopedCapability.describe()`` 的字段一一对应。 + """ + + def __init__(self, path: str) -> None: + self.path = path + self._data = _load_json(path) + + def record_grant(self, project_id: str, capability: dict, + ts: float | None = None) -> dict: + """登记一条已签发的授权，返回该授权（副本）。 + + capability 是 P5 describe() 形状的 dict；缺 + capability_id/tier/connector/host/use_case 任一即 + 抛 ValueError。同 capability_id 重复登记视为更新。 + """ + if not project_id: + raise ValueError("project_id 不能为空") + required = ("capability_id", "tier", "connector", "host", + "use_case") + for key in required: + if not capability.get(key): + raise ValueError("能力描述缺字段：%s" % key) + now = ts if ts is not None else time.time() + grant = { + "capability_id": str(capability["capability_id"]), + "project_id": project_id, + "tier": str(capability["tier"]), + "connector": str(capability["connector"]), + "host": str(capability["host"]), + "use_case": str(capability["use_case"]), + "session_id": capability.get("session_id") or "", + "task_id": capability.get("task_id") or "", + "methods": sorted(capability.get("methods") or []), + "max_uses": capability.get("max_uses"), + "uses_left": capability.get("uses_left"), + "expires_at": capability.get("expires_at"), + "revoked": bool(capability.get("revoked", False)), + "granted_at": now, + } + grants = self._grants(project_id) + for index, old in enumerate(grants): + if old["capability_id"] == grant["capability_id"]: + grants[index] = grant + break + else: + grants.append(grant) + self._data.setdefault(project_id, grants) + _save_json(self.path, self._data) + return dict(grant) + + def note_use(self, project_id: str, capability_id: str) -> bool: + """记一次使用：次数余量减一；未知 id 或已无余量返回 False。""" + for grant in self._grants(project_id): + if grant["capability_id"] != capability_id: + continue + if grant.get("revoked"): + return False + left = grant.get("uses_left") + if left is None: + return True + if left <= 0: + return False + grant["uses_left"] = left - 1 + _save_json(self.path, self._data) + return True + return False + + def revoke(self, project_id: str, capability_id: str) -> bool: + """撤销一条授权：未知 id 返回 False，已撤销也返回 False。""" + for grant in self._grants(project_id): + if grant["capability_id"] != capability_id: + continue + if grant.get("revoked"): + return False + grant["revoked"] = True + _save_json(self.path, self._data) + return True + return False + + def list_active(self, project_id: str, + now: float | None = None) -> list: + """该项目当前有效的授权：未撤销、未过期、次数未用完。""" + at = now if now is not None else time.time() + active = [dict(g) for g in self._grants(project_id) + if _grant_active(g, at)] + active.sort(key=lambda g: g.get("granted_at", 0.0)) + return active + + def list_all(self, project_id: str) -> list: + """该项目全部授权（含已撤销/已过期，副本，按签发排序）。""" + grants = [dict(g) for g in self._grants(project_id)] + grants.sort(key=lambda g: g.get("granted_at", 0.0)) + return grants + + def _grants(self, project_id: str) -> list: + """该项目的授权列表（内部引用，调用方不得外泄）。""" + items = self._data.get(project_id, []) + return items if isinstance(items, list) else [] + + +def _grant_active(grant: dict, now: float) -> bool: + """单条授权是否当前有效（未撤销 + 未过期 + 有余量）。""" + if grant.get("revoked"): + return False + expires_at = grant.get("expires_at") + if expires_at is not None and float(expires_at) <= now: + return False + left = grant.get("uses_left") + if left is not None and int(left) <= 0: + return False + return True + + +def activity_summary(plans: list, recent_records: list, + grants: list) -> dict: + """拼活动摘要（纯函数）：正在做 / 计划中 / 已做 / 近期 / 授权。 + + 入参是调用方已按项目过滤好的列表：plans 为计划 dict 列表， + recent_records 为近期审计记录（倒序），grants 为有效授权。 + 返回只含描述性字段，不含任何密钥与对话原文。 + """ + running = [p for p in plans if p.get("status") == "running"] + pending = [p for p in plans if p.get("status") == "pending"] + done = [p for p in plans if p.get("status") == "done"] + current = running[0]["title"] if running else "" + recent = [{"ts": r.get("ts"), "action": r.get("action"), + "category": r.get("category"), + "result": r.get("result"), + "object_type": r.get("object_type"), + "object_id": r.get("object_id")} + for r in recent_records] + return { + "current": current, + "running": running, + "planned": pending, + "done": done, + "recent": recent, + "granted_permissions": grants, + "counts": { + "running": len(running), + "planned": len(pending), + "done": len(done), + "granted": len(grants), + }, + } diff --git a/cloud/chv/approval_card.py b/cloud/chv/approval_card.py new file mode 100644 index 0000000..f152f36 --- /dev/null +++ b/cloud/chv/approval_card.py @@ -0,0 +1,172 @@ +"""结构化审批卡数据（feat-chv-muse-p12，叶子模块，纯标准库）。 + +接 Muse 官方要求（报告 §2.6）：确定性 UI——结构化审批卡 +（明确接受/拒绝动作）；审批是严格能力，绑定特定连接器/ +目的地与用例，支持一次性、会话级、任务级、时限级、永久级 +授权，Sentinel 决定向用户呈现哪几种授权类型；审批 UI 不在 +对话内，Sentinel 直接向客户端发审批请求，用户答复直接回 +Sentinel（防止对话内容伪造审批）。 + +本模块只做卡的**数据**：把 P3 ``ApprovalQueue`` 的 pending +记录 dict 转成客户端可呈现的结构化卡。不 import P3/P5 模块 +——它们不在本基线里，集成时由 round-11 合并；只按 dict +契约接线。 + +卡数据（纯数据 dict，可直接 JSON 序列化）:: + {"pending_id": str, "action": str, "target": {...}, + "reasons": [str, ...], "risk": "low|medium|high", + "actions": [{"id": "accept", "label": "接受"}, + {"id": "deny", "label": "拒绝"}], + "grant_options": [{"tier": str, "label": str, + "requires_explicit_opt_in": bool}, ...], + "expires_at": float | None} + +- ``risk`` 分级（确定性规则，不冒充模型分类器）：写方法 + （POST/PUT/PATCH/DELETE）+ 敏感路径（/admin、/settings、 + /login、token 相关）或私有主机 → high；写方法但目标不 + 敏感 → medium；其余只读 → low。 +- ``grant_options`` 的档位列表由调用方传入（Sentinel 决定 + 呈现哪几种）；不传默认五档全列，永久档永远标注 + ``requires_explicit_opt_in: true``（P5 口径：永久默认拒绝， + 须显式 opt-in）。 +- ``card_matches_channel()``：校验卡数据与 P5 对话外通道载荷 + （``ApprovalRequest.payload()``）一致——pending_id 相同且 + 目标字段（连接器/方法/scheme/主机/端口/路径）全等；不一 + 致即客户端不该呈现该卡。 +""" + +import ipaddress + +# 五档授权类型（P3/P5 口径）：一次/会话/任务/时限/永久。 +TIERS = ("one-shot", "session", "task", "timed", "permanent") + +TIER_LABELS = { + "one-shot": "仅这一次", + "session": "本次会话", + "task": "本任务期间", + "timed": "限时授权", + "permanent": "永久授权", +} + +# 目标描述字段：卡与通道载荷比对只认这些键。 +TARGET_KEYS = ("connector", "method", "scheme", "host", "port", + "path") + +_WRITE_METHODS = ("POST", "PUT", "PATCH", "DELETE") + +_SENSITIVE_PATHS = ("/admin", "/settings", "/login", "/token", + "/oauth", "/credential") + +# 通道审批默认有效期（秒）：15 分钟，与 P5 通道一致。 +DEFAULT_TTL_SECONDS = 900 + + +def _is_private_host(host: str) -> bool: + """主机是否私有/本地：IP 段判断，localhost 视为私有。""" + if not host: + return False + lowered = host.lower() + if lowered in ("localhost", "localhost.localdomain"): + return True + try: + return ipaddress.ip_address(host).is_private + except ValueError: + return False + + +def _is_sensitive_path(path: str) -> bool: + """路径是否敏感：命中敏感前缀即敏感。""" + lowered = (path or "").lower() + return any(lowered.startswith(p) for p in _SENSITIVE_PATHS) + + +def risk_level(queue_record: dict) -> str: + """审批风险分级：high/medium/low（确定性规则）。 + + 写方法 +（敏感路径或私有主机）→ high；写方法 → medium； + 其余只读 → low。 + """ + method = str(queue_record.get("method", "")).upper() + if method not in _WRITE_METHODS: + return "low" + if _is_sensitive_path(str(queue_record.get("path", ""))): + return "high" + if _is_private_host(str(queue_record.get("host", ""))): + return "high" + return "medium" + + +def _target_from_record(queue_record: dict) -> dict: + """队列记录里只取目标描述字段，别的什么都不带出卡。""" + return {k: queue_record[k] for k in TARGET_KEYS + if k in queue_record} + + +def _grant_options(tiers: tuple | list | None) -> list: + """档位选项：未知档位丢弃，永久档永远标注须显式 opt-in。""" + wanted = list(tiers) if tiers is not None else list(TIERS) + options = [] + for tier in wanted: + if tier not in TIER_LABELS: + continue + options.append({ + "tier": tier, + "label": TIER_LABELS[tier], + "requires_explicit_opt_in": tier == "permanent", + }) + return options + + +def build_approval_card(queue_record: dict, + tiers: tuple | list | None = None, + ttl_seconds: int = DEFAULT_TTL_SECONDS + ) -> dict: + """P3 队列记录转结构化审批卡数据。 + + queue_record 是 ``ApprovalQueue.record()`` 记入的 dict： + 目标描述（connector/method/scheme/host/port/path）+ + pending_id + reasons + ts。缺 pending_id 抛 ValueError。 + expires_at 由记录 ts + 有效期算出；无 ts 则为 None。 + """ + pending_id = queue_record.get("pending_id") + if not pending_id: + raise ValueError("队列记录缺 pending_id") + target = _target_from_record(queue_record) + connector = str(target.get("connector", "")) + method = str(target.get("method", "")).upper() + action = "%s %s" % (connector, method) if connector else method + ts = queue_record.get("ts") + expires_at = None + if isinstance(ts, (int, float)): + expires_at = float(ts) + ttl_seconds + return { + "pending_id": str(pending_id), + "action": action, + "target": target, + "reasons": list(queue_record.get("reasons") or []), + "risk": risk_level(queue_record), + "actions": [ + {"id": "accept", "label": "接受"}, + {"id": "deny", "label": "拒绝"}, + ], + "grant_options": _grant_options(tiers), + "expires_at": expires_at, + } + + +def card_matches_channel(card: dict, channel_payload: dict) -> bool: + """卡数据与 P5 通道载荷是否一致（pending_id + 目标全等）。 + + 任一目标字段不一致、或 pending_id 不同，即返回 False。 + 客户端只应呈现校验通过的卡。 + """ + if card.get("pending_id") != channel_payload.get("pending_id"): + return False + card_target = card.get("target") or {} + channel_target = channel_payload.get("target") or {} + for key in TARGET_KEYS: + if key not in card_target and key not in channel_target: + continue + if card_target.get(key) != channel_target.get(key): + return False + return True diff --git a/cloud/chv/audit_trail.py b/cloud/chv/audit_trail.py new file mode 100644 index 0000000..6513b69 --- /dev/null +++ b/cloud/chv/audit_trail.py @@ -0,0 +1,343 @@ +"""统一审计轨迹（feat-chv-muse-p12，叶子模块，纯标准库）。 + +接 Muse 官方要求（报告 §2.2 第 7 步、§2.6）：完整审计轨迹展示 +已做与计划做的事；审批走确定性 UI；用户可检查、编辑、下载 +VM 内所有文件。 + +现状收敛：基线里审计是散的——``AuthStore.audit()`` 只追加 +``(user_id, project_id, action, detail, created_at)`` 到 SQLite +``audit_log``，``Worker._audit()`` 是 best-effort 包装， +``worker_adapter.make_chv_audit_callback`` 把 CHV 事件转成 +``"chv_vm_" + event``。本模块给它们一个统一记录模型与一条 +追加写 JSONL 轨迹，不改现有调用方。 + +记录模型（每条都是纯数据 dict，可直接 JSON 序列化）:: + {"seq": int, "ts": float, "project_id": str, "actor": str, + "category": str, "action": str, "object_type": str, + "object_id": str, "result": str, "reason": str, + "task_id": str, "approval_id": str, "detail": dict, + "prev_hash": str, "hash": str} + +- ``category`` 只认 ``AUDIT_CATEGORIES``：任务派发、VM 生命周期、 + 同步、审批、支付（P10）、配额（P11）、记忆操作（P7 的忘记）、 + 后台任务（P8）、推理/遥测（P9）、安全事件。 +- ``detail`` 只含描述性字段：``sanitize_detail()`` 按字段级白名 + 单清洗——密钥、token 值、对话原文绝不进审计。清洗是字段名 + 黑名单 + ``key=value`` 模式脱敏，不是内容理解：密钥若藏在 + 自由文本里且不带标记，清洗器看不出来，调用方仍须只传描述 + 性字段（与 P3/P5 的口径一致）。 +- 哈希链：每条记录的 ``hash`` = SHA-256(prev_hash + 规范 JSON)， + 首条 prev_hash 为 ``GENESIS_HASH``。``verify()`` 能发现改写/ + 删除/调序。诚实边界：这是**篡改可发现**，不是防篡改——能写 + 文件的人重算整条链就能掩盖改动；真防篡改要外部锚定（远端 + 日志/签名时间戳），不在本轨。 +- project_id 强制：``list_records()``/``export_json()`` 必须传 project_id， + 只返回该项目记录；跨项目读取返回空，不抛错（与现有 + ``list_audit`` 行为一致，不泄露“存在别的项目记录”）。 +- 持久化：先落独立 JSONL 文件（0600），待 P7 状态库就绪后迁移； + ``from_legacy_row()`` 把现有 ``audit_log`` 行转成统一记录 + （标 ``legacy: true``，本身无哈希链，append 时重新入链）。 +""" + +import hashlib +import json +import os +import re +import time + +# 审计事件分类：P1～P11 各轨的事件都落进这里，不另起表。 +AUDIT_CATEGORIES = frozenset(( + "task_dispatch", "vm_lifecycle", "gitea_sync", "approval", + "payment", "quota", "memory", "background", "inference", + "telemetry", "security", +)) + +# 创世前哈希：首条记录的 prev_hash。 +GENESIS_HASH = "0" * 64 + +# 字段名里出现这些子串（小写比对）的键一律丢弃：值疑似密钥。 +_SECRET_KEY_HINTS = ( + "token", "secret", "password", "passwd", "api_key", "apikey", + "credential", "private_key", "bearer", "cookie", "auth", + "session", +) + +# 自由文本里的 ``key=value`` / ``key: value`` 模式：值脱敏。 +_SECRET_VALUE_RE = re.compile( + r"(?i)(token|secret|password|passwd|api[-_]?key|bearer" + r"|authorization)\s*[:=]\s*[^,\s;\"']+") + +_MAX_STRING_LEN = 4096 + + +def _is_secret_key(key: str) -> bool: + """字段名是否疑似密钥（小写子串比对）。""" + lowered = key.lower() + return any(hint in lowered for hint in _SECRET_KEY_HINTS) + + +def _sanitize_string(value: str) -> str: + """自由文本脱敏：密钥模式的值打码，超长截断。""" + redacted = _SECRET_VALUE_RE.sub(r"\1=***", value) + if len(redacted) > _MAX_STRING_LEN: + return redacted[:_MAX_STRING_LEN] + "...(truncated)" + return redacted + + +def sanitize_detail(detail: dict | None) -> dict: + """递归清洗 detail：密钥字段丢弃，自由文本脱敏。 + + dict 键疑似密钥即整键丢弃；list 逐项清洗；str 脱敏； + 其它标量原样保留。返回新 dict，不改入参。 + """ + if not detail: + return {} + cleaned: dict = {} + for key, value in detail.items(): + if _is_secret_key(str(key)): + continue + cleaned[str(key)] = _sanitize_value(value) + return cleaned + + +def _sanitize_value(value: object) -> object: + """清洗单个值：dict/list/str 递归，其它原样。""" + if isinstance(value, dict): + return {str(k): _sanitize_value(v) + for k, v in value.items() if not _is_secret_key(str(k))} + if isinstance(value, list): + return [_sanitize_value(item) for item in value] + if isinstance(value, str): + return _sanitize_string(value) + return value + + +def _canonical(record: dict) -> str: + """记录的规范 JSON（排序键、紧凑分隔、原文编码）。""" + return json.dumps(record, sort_keys=True, separators=(",", ":"), + ensure_ascii=False) + + +def compute_hash(prev_hash: str, record: dict) -> str: + """一条记录的链哈希：SHA-256(prev_hash + 规范 JSON)。""" + body = prev_hash + "\n" + _canonical(record) + return hashlib.sha256(body.encode("utf-8")).hexdigest() + + +def from_legacy_row(row: dict) -> dict: + """现有 audit_log 行转统一记录（标 legacy，无哈希链）。 + + 入参是 ``AuthStore.list_audit()`` 的行 dict： + id/user_id/project_id/action/detail/created_at。 + detail 若是 str 则原样包进 ``{"detail": ...}`` 再清洗； + 若是 dict 直接清洗。返回的记录由调用方经 ``append()`` + 重新入链，seq/hash 由 ``append()`` 补。 + """ + detail = row.get("detail") + if isinstance(detail, str): + detail = {"detail": detail} + if not isinstance(detail, dict): + detail = {} + return { + "project_id": str(row.get("project_id", "")), + "actor": str(row.get("user_id", "")), + "category": "security", + "action": str(row.get("action", "")), + "object_type": "", + "object_id": str(row.get("id", "")), + "result": "ok", + "reason": "legacy audit_log row", + "task_id": "", + "approval_id": "", + "detail": sanitize_detail(detail), + "ts": float(row.get("created_at", 0.0) or 0.0), + "legacy": True, + } + + +class AuditTrail: + """追加写 JSONL 审计轨迹：哈希链 + 项目隔离 + 导出。 + + 文件不存在即创建（0600）；每条 append 都 O_APPEND 写 + + fsync，写失败原样抛错（审计失败不静默丢，由调用方决定 + 是否 best-effort）。读路径全部只读，不改文件。 + """ + + def __init__(self, path: str) -> None: + self.path = path + self._last_hash = GENESIS_HASH + self._seq = 0 + self._load_tail() + + def _load_tail(self) -> None: + """启动时扫一遍文件，恢复末尾 seq 与链哈希。""" + if not os.path.exists(self.path): + return + with open(self.path, "r", encoding="utf-8") as f: + for line in f: + line = line.strip() + if not line: + continue + try: + record = json.loads(line) + except ValueError: + continue + seq = record.get("seq") + if isinstance(seq, int): + self._seq = seq + chain_hash = record.get("hash") + if isinstance(chain_hash, str): + self._last_hash = chain_hash + + def append(self, project_id: str, actor: str, category: str, + action: str, object_type: str = "", + object_id: str = "", result: str = "ok", + reason: str = "", task_id: str = "", + approval_id: str = "", + detail: dict | None = None, + ts: float | None = None) -> dict: + """追加一条审计记录，返回该记录（含 seq 与链哈希）。""" + if not project_id: + raise ValueError("project_id 不能为空") + if category not in AUDIT_CATEGORIES: + raise ValueError("未知审计分类：%s" % category) + seq = self._seq + 1 + record = { + "seq": seq, + "ts": ts if ts is not None else time.time(), + "project_id": project_id, + "actor": actor, + "category": category, + "action": action, + "object_type": object_type, + "object_id": object_id, + "result": result, + "reason": reason, + "task_id": task_id, + "approval_id": approval_id, + "detail": sanitize_detail(detail), + "prev_hash": self._last_hash, + } + chain_hash = compute_hash(self._last_hash, record) + record["hash"] = chain_hash + line = _canonical(record) + "\n" + fd = os.open(self.path, os.O_WRONLY | os.O_CREAT | os.O_APPEND, + 0o600) + try: + os.write(fd, line.encode("utf-8")) + os.fsync(fd) + finally: + os.close(fd) + self._seq = seq + self._last_hash = chain_hash + return dict(record) + + def list_records(self, project_id: str, category: str = "", + action: str = "", task_id: str = "", + since: float = 0.0, until: float = 0.0, + limit: int = 100, offset: int = 0) -> list: + """按时间倒序返回该项目的审计记录（副本，可分页过滤）。 + + 方法名不用 ``list``：类内同名方法会遮住内置 ``list`` + 类型，导致返回注解被 mypy 认成方法本身。 + project_id 为空或不匹配一律返回空列表。limit <= 0 + 表示不限制条数。 + """ + if not project_id: + return [] + rows = self._read_all() + kept = [r for r in rows if r.get("project_id") == project_id] + kept = _filter_rows(kept, category, action, task_id, + since, until) + kept.sort(key=lambda r: (r.get("ts", 0.0), r.get("seq", 0)), + reverse=True) + if offset: + kept = kept[offset:] + if limit > 0: + kept = kept[:limit] + return [dict(r) for r in kept] + + def verify(self) -> tuple[bool, int]: + """校验整条哈希链：返回 (是否通过, 首个坏行下标)。 + + 通过时下标为 -1。坏行包括：解析失败、seq 不连续、 + prev_hash 对不上、hash 重算不符。 + """ + rows = self._read_raw() + expected_prev = GENESIS_HASH + expected_seq = 1 + for index, raw in enumerate(rows): + try: + record = json.loads(raw) + except ValueError: + return False, index + if not _chain_ok(record, expected_prev, expected_seq): + return False, index + expected_prev = record["hash"] + expected_seq += 1 + return True, -1 + + def export_json(self, project_id: str, dest: str) -> int: + """导出该项目全部审计记录为 JSON 数组，返回导出条数。 + + 导出内容与 ``list_records()`` 同款清洗后记录（含哈希链 + 字段，不含任何原始 detail 密钥——清洗在 append 时已做）。 + """ + rows = self.list_records(project_id, limit=0) + rows.sort(key=lambda r: (r.get("ts", 0.0), r.get("seq", 0))) + with open(dest, "w", encoding="utf-8") as f: + json.dump(rows, f, ensure_ascii=False, indent=2) + os.chmod(dest, 0o600) + return len(rows) + + def _read_all(self) -> list: + """读全部可解析记录（跳过坏行，坏行由 verify 发现）。""" + rows = [] + for raw in self._read_raw(): + try: + record = json.loads(raw) + except ValueError: + continue + if isinstance(record, dict): + rows.append(record) + return rows + + def _read_raw(self) -> list: + """读全部非空行原文（verify 用）。""" + if not os.path.exists(self.path): + return [] + with open(self.path, "r", encoding="utf-8") as f: + return [line.strip() for line in f if line.strip()] + + +def _filter_rows(rows: list, category: str, action: str, + task_id: str, since: float, until: float) -> list: + """list() 的过滤：分类/动作/任务/时间窗口，全空即不过滤。""" + kept = rows + if category: + kept = [r for r in kept if r.get("category") == category] + if action: + kept = [r for r in kept if r.get("action") == action] + if task_id: + kept = [r for r in kept if r.get("task_id") == task_id] + if since: + kept = [r for r in kept if r.get("ts", 0.0) >= since] + if until: + kept = [r for r in kept if r.get("ts", 0.0) <= until] + return kept + + +def _chain_ok(record: dict, expected_prev: str, + expected_seq: int) -> bool: + """单条记录的链校验：seq 连续 + prev 对上 + hash 重算符。""" + if not isinstance(record, dict): + return False + if record.get("seq") != expected_seq: + return False + if record.get("prev_hash") != expected_prev: + return False + claimed = record.get("hash") + if not isinstance(claimed, str): + return False + body = {k: v for k, v in record.items() if k != "hash"} + return compute_hash(expected_prev, body) == claimed diff --git a/cloud/tests/test_chv_muse_p12.py b/cloud/tests/test_chv_muse_p12.py new file mode 100644 index 0000000..c965c34 --- /dev/null +++ b/cloud/tests/test_chv_muse_p12.py @@ -0,0 +1,471 @@ +"""对齐 Muse P12 测试：审计轨迹与活动视图。 + +只用标准库，不启动真实 VM，不联网，不碰生产文件： +审计轨迹、计划登记、权限登记一律落在临时目录。 +覆盖：清洗白名单、哈希链验签、项目隔离、旧行迁移、 +导出、计划流转、授权有效期/撤销/次数、审批卡风险与 +通道一致性——含否定测试。 +""" + +import json +import os +import sys +import tempfile +import unittest +from pathlib import Path + +try: + from cloud.chv.activity import ( + PermissionRegistry, + PlanStore, + activity_summary, + ) + from cloud.chv.approval_card import ( + build_approval_card, + card_matches_channel, + risk_level, + ) + from cloud.chv.audit_trail import ( + AUDIT_CATEGORIES, + AuditTrail, + from_legacy_row, + sanitize_detail, + ) +except ImportError: # discover -s cloud/tests 时把项目根目录补进 sys.path + sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + from cloud.chv.activity import ( + PermissionRegistry, + PlanStore, + activity_summary, + ) + from cloud.chv.approval_card import ( + build_approval_card, + card_matches_channel, + risk_level, + ) + from cloud.chv.audit_trail import ( + AUDIT_CATEGORIES, + AuditTrail, + from_legacy_row, + sanitize_detail, + ) + +PROJECT = "proj-p12" +OTHER = "proj-other" + + +def _capability(cap_id="cap-1", tier="task", uses_left=3, + expires_at=None, revoked=False): + """P5 ScopedCapability.describe() 形状的 dict。""" + return { + "capability_id": cap_id, "tier": tier, + "connector": "gitea", "host": "gitea.tose.top", + "use_case": "sync", "session_id": "s1", "task_id": "t1", + "methods": ["GET", "POST"], "max_uses": 3, + "uses_left": uses_left, "expires_at": expires_at, + "revoked": revoked, + } + + +def _queue_record(**overrides): + """P3 ApprovalQueue.record() 形状的 dict。""" + record = { + "pending_id": "pend-abc123", "connector": "gitea", + "method": "GET", "scheme": "https", + "host": "gitea.tose.top", "port": 443, "path": "/api/v1/repos", + "reasons": ["host_not_allowlisted"], "ts": 1000.0, + } + record.update(overrides) + return record + + +class SanitizeTest(unittest.TestCase): + def test_secret_keys_dropped(self): + cleaned = sanitize_detail({ + "action": "sync", + "gitea_token": "gto_abcdef", + "api_key": "sk-123", + "password": "hunter2", + "nested": {"secret": "x", "repo": "r"}, + }) + self.assertEqual(cleaned, + {"action": "sync", "nested": {"repo": "r"}}) + + def test_secret_value_in_string_redacted(self): + cleaned = sanitize_detail( + {"detail": "failed with token=abc123xyz and ok"}) + self.assertEqual(cleaned["detail"], + "failed with token=*** and ok") + + def test_benign_fields_kept(self): + cleaned = sanitize_detail( + {"repo": "admin/chv-sync-test", "count": 3, + "tags": ["a", "b"]}) + self.assertEqual(cleaned, {"repo": "admin/chv-sync-test", + "count": 3, "tags": ["a", "b"]}) + + def test_empty_and_none_detail(self): + self.assertEqual(sanitize_detail({}), {}) + self.assertEqual(sanitize_detail(None), {}) + + +class AuditTrailTest(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.path = os.path.join(self.tmp.name, "audit.jsonl") + self.trail = AuditTrail(self.path) + + def tearDown(self): + self.tmp.cleanup() + + def test_append_and_list_roundtrip(self): + rec = self.trail.append( + PROJECT, "worker", "vm_lifecycle", "vm_boot", + object_type="vm", object_id="chv-1", ts=1000.0, + detail={"image": "lite10"}) + self.assertEqual(rec["seq"], 1) + self.assertEqual(rec["prev_hash"], "0" * 64) + self.assertEqual(len(rec["hash"]), 64) + rows = self.trail.list_records(PROJECT) + self.assertEqual(len(rows), 1) + self.assertEqual(rows[0]["action"], "vm_boot") + + def test_append_sanitizes_detail(self): + self.trail.append( + PROJECT, "worker", "approval", "ask", + detail={"token": "gto_secret", "host": "example.invalid"}, + ts=1000.0) + rows = self.trail.list_records(PROJECT) + self.assertNotIn("token", rows[0]["detail"]) + self.assertEqual(rows[0]["detail"]["host"], "example.invalid") + + def test_unknown_category_rejected(self): + with self.assertRaises(ValueError): + self.trail.append(PROJECT, "worker", "nope", "x") + + def test_empty_project_rejected(self): + with self.assertRaises(ValueError): + self.trail.append("", "worker", "approval", "x") + + def test_project_isolation(self): + self.trail.append(PROJECT, "worker", "approval", "ask", + ts=1000.0) + self.trail.append(OTHER, "worker", "approval", "ask", + ts=1001.0) + self.assertEqual(len(self.trail.list_records(PROJECT)), 1) + self.assertEqual(len(self.trail.list_records(OTHER)), 1) + self.assertEqual(self.trail.list_records(""), []) + self.assertEqual(self.trail.list_records("proj-missing"), []) + + def test_list_filters_and_pagination(self): + for index in range(5): + self.trail.append( + PROJECT, "worker", "vm_lifecycle", "vm_boot", + task_id="t-%d" % (index % 2), ts=1000.0 + index) + self.assertEqual(len(self.trail.list_records(PROJECT, task_id="t-0")), + 3) + self.assertEqual( + len(self.trail.list_records(PROJECT, since=1003.0)), 2) + self.assertEqual( + len(self.trail.list_records(PROJECT, limit=2)), 2) + self.assertEqual( + len(self.trail.list_records(PROJECT, limit=2, offset=2)), 2) + rows = self.trail.list_records(PROJECT) + self.assertGreater(rows[0]["ts"], rows[-1]["ts"]) + + def test_verify_passes_then_detects_tamper(self): + for index in range(3): + self.trail.append(PROJECT, "worker", "approval", "ask", + ts=1000.0 + index) + ok, bad = self.trail.verify() + self.assertTrue(ok) + self.assertEqual(bad, -1) + with open(self.path, "r", encoding="utf-8") as f: + lines = f.readlines() + tampered = json.loads(lines[1]) + tampered["action"] = "allow" + lines[1] = json.dumps(tampered) + "\n" + with open(self.path, "w", encoding="utf-8") as f: + f.writelines(lines) + ok, bad = self.trail.verify() + self.assertFalse(ok) + self.assertEqual(bad, 1) + + def test_verify_detects_deletion(self): + for index in range(3): + self.trail.append(PROJECT, "worker", "approval", "ask", + ts=1000.0 + index) + with open(self.path, "r", encoding="utf-8") as f: + lines = f.readlines() + with open(self.path, "w", encoding="utf-8") as f: + f.writelines([lines[0], lines[2]]) + ok, bad = self.trail.verify() + self.assertFalse(ok) + self.assertEqual(bad, 1) + + def test_reload_restores_chain(self): + self.trail.append(PROJECT, "worker", "approval", "ask", + ts=1000.0) + second = AuditTrail(self.path) + rec = second.append(PROJECT, "worker", "approval", "allow", + ts=1001.0) + self.assertEqual(rec["seq"], 2) + ok, _ = second.verify() + self.assertTrue(ok) + + def test_export_json(self): + self.trail.append(PROJECT, "worker", "approval", "ask", + detail={"token": "nope"}, ts=1000.0) + dest = os.path.join(self.tmp.name, "export.json") + count = self.trail.export_json(PROJECT, dest) + self.assertEqual(count, 1) + with open(dest, "r", encoding="utf-8") as f: + rows = json.load(f) + self.assertEqual(len(rows), 1) + self.assertNotIn("token", rows[0]["detail"]) + self.assertEqual(os.stat(dest).st_mode & 0o777, 0o600) + self.assertEqual(self.trail.export_json(OTHER, dest), 0) + + def test_all_categories_accepted(self): + for category in sorted(AUDIT_CATEGORIES): + self.trail.append(PROJECT, "worker", category, "probe", + ts=1000.0) + self.assertEqual(len(self.trail.list_records(PROJECT, limit=0)), + len(AUDIT_CATEGORIES)) + + def test_legacy_row_conversion(self): + row = {"id": 7, "user_id": "system", "project_id": PROJECT, + "action": "worker_exec_start", + "detail": {"token": "gto_x", "task": "t1"}, + "created_at": 1000.0} + converted = from_legacy_row(row) + self.assertTrue(converted["legacy"]) + self.assertEqual(converted["action"], "worker_exec_start") + self.assertEqual(converted["object_id"], "7") + self.assertNotIn("token", converted["detail"]) + self.assertEqual(converted["detail"]["task"], "t1") + + +class PlanStoreTest(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.path = os.path.join(self.tmp.name, "plans.json") + self.store = PlanStore(self.path) + + def tearDown(self): + self.tmp.cleanup() + + def test_add_and_list(self): + plan = self.store.add_plan(PROJECT, "建 round-11 镜像", + task_id="t1", ts=1000.0) + self.assertEqual(plan["status"], "pending") + self.assertEqual(len(self.store.list_plans(PROJECT)), 1) + + def test_valid_transitions(self): + plan = self.store.add_plan(PROJECT, "同步仓库", ts=1000.0) + moved = self.store.set_status( + PROJECT, plan["plan_id"], "running", ts=1001.0) + self.assertEqual(moved["status"], "running") + done = self.store.set_status( + PROJECT, plan["plan_id"], "done", ts=1002.0) + self.assertEqual(done["status"], "done") + + def test_illegal_transition_rejected(self): + plan = self.store.add_plan(PROJECT, "跑测试", ts=1000.0) + with self.assertRaises(ValueError): + self.store.set_status(PROJECT, plan["plan_id"], "done") + self.store.set_status(PROJECT, plan["plan_id"], "cancelled") + with self.assertRaises(ValueError): + self.store.set_status(PROJECT, plan["plan_id"], "running") + + def test_unknown_plan_rejected(self): + with self.assertRaises(ValueError): + self.store.set_status(PROJECT, "nope", "running") + + def test_empty_title_rejected(self): + with self.assertRaises(ValueError): + self.store.add_plan(PROJECT, "") + + def test_list_filter_by_status(self): + first = self.store.add_plan(PROJECT, "甲", ts=1000.0) + self.store.add_plan(PROJECT, "乙", ts=1001.0) + self.store.set_status(PROJECT, first["plan_id"], "running") + self.assertEqual( + len(self.store.list_plans(PROJECT, "running")), 1) + self.assertEqual( + len(self.store.list_plans(PROJECT, "pending")), 1) + + def test_persistence_roundtrip(self): + self.store.add_plan(PROJECT, "甲", ts=1000.0) + second = PlanStore(self.path) + self.assertEqual(len(second.list_plans(PROJECT)), 1) + + +class PermissionRegistryTest(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.path = os.path.join(self.tmp.name, "grants.json") + self.registry = PermissionRegistry(self.path) + + def tearDown(self): + self.tmp.cleanup() + + def test_grant_and_list_active(self): + self.registry.record_grant(PROJECT, _capability(), ts=1000.0) + active = self.registry.list_active(PROJECT, now=1001.0) + self.assertEqual(len(active), 1) + self.assertEqual(active[0]["connector"], "gitea") + self.assertEqual(active[0]["uses_left"], 3) + + def test_missing_field_rejected(self): + bad = _capability() + del bad["host"] + with self.assertRaises(ValueError): + self.registry.record_grant(PROJECT, bad) + + def test_expired_grant_not_active(self): + self.registry.record_grant( + PROJECT, _capability(expires_at=1000.0), ts=900.0) + self.assertEqual(self.registry.list_active(PROJECT, now=1001.0), + []) + + def test_revoked_grant_not_active(self): + self.registry.record_grant(PROJECT, _capability(), ts=1000.0) + self.assertTrue(self.registry.revoke(PROJECT, "cap-1")) + self.assertFalse(self.registry.revoke(PROJECT, "cap-1")) + self.assertEqual(self.registry.list_active(PROJECT, now=1001.0), + []) + self.assertEqual(len(self.registry.list_all(PROJECT)), 1) + + def test_uses_exhausted_not_active(self): + self.registry.record_grant(PROJECT, _capability(uses_left=1), + ts=1000.0) + self.assertTrue(self.registry.note_use(PROJECT, "cap-1")) + self.assertFalse(self.registry.note_use(PROJECT, "cap-1")) + self.assertEqual(self.registry.list_active(PROJECT, now=1001.0), + []) + + def test_unlimited_uses_never_exhaust(self): + self.registry.record_grant( + PROJECT, _capability(uses_left=None), ts=1000.0) + for _ in range(5): + self.assertTrue(self.registry.note_use(PROJECT, "cap-1")) + self.assertEqual(len(self.registry.list_active(PROJECT)), 1) + + def test_unknown_capability_use_revoke_false(self): + self.assertFalse(self.registry.note_use(PROJECT, "nope")) + self.assertFalse(self.registry.revoke(PROJECT, "nope")) + + def test_regrant_updates(self): + self.registry.record_grant(PROJECT, _capability(), ts=1000.0) + self.registry.record_grant( + PROJECT, _capability(uses_left=9), ts=1001.0) + active = self.registry.list_active(PROJECT, now=1002.0) + self.assertEqual(len(active), 1) + self.assertEqual(active[0]["uses_left"], 9) + + +class ActivitySummaryTest(unittest.TestCase): + def test_summary_shape_and_counts(self): + plans = [ + {"title": "跑测试", "status": "running"}, + {"title": "建镜像", "status": "pending"}, + {"title": "同步", "status": "done"}, + ] + recent = [{"ts": 1000.0, "action": "vm_boot", + "category": "vm_lifecycle", "result": "ok", + "object_type": "vm", "object_id": "chv-1", + "detail": {"token": "must-not-leak"}}] + grants = [_capability()] + summary = activity_summary(plans, recent, grants) + self.assertEqual(summary["current"], "跑测试") + self.assertEqual(summary["counts"], + {"running": 1, "planned": 1, "done": 1, + "granted": 1}) + self.assertEqual(len(summary["recent"]), 1) + self.assertNotIn("detail", summary["recent"][0]) + + def test_no_running_current_empty(self): + summary = activity_summary([], [], []) + self.assertEqual(summary["current"], "") + self.assertEqual(summary["counts"]["granted"], 0) + + +class ApprovalCardTest(unittest.TestCase): + def test_read_get_low_risk(self): + card = build_approval_card(_queue_record()) + self.assertEqual(card["risk"], "low") + self.assertEqual(card["action"], "gitea GET") + self.assertEqual(card["expires_at"], 1900.0) + self.assertEqual([a["id"] for a in card["actions"]], + ["accept", "deny"]) + self.assertEqual( + [o["tier"] for o in card["grant_options"]], + ["one-shot", "session", "task", "timed", "permanent"]) + + def test_permanent_always_opt_in(self): + card = build_approval_card(_queue_record()) + permanent = [o for o in card["grant_options"] + if o["tier"] == "permanent"][0] + self.assertTrue(permanent["requires_explicit_opt_in"]) + others = [o for o in card["grant_options"] + if o["tier"] != "permanent"] + self.assertTrue(all( + not o["requires_explicit_opt_in"] for o in others)) + + def test_write_to_admin_high_risk(self): + card = build_approval_card(_queue_record( + method="POST", path="/admin/settings")) + self.assertEqual(card["risk"], "high") + + def test_write_to_private_host_high_risk(self): + card = build_approval_card(_queue_record( + method="DELETE", host="192.168.1.5", path="/api/v1/x")) + self.assertEqual(card["risk"], "high") + + def test_write_public_medium_risk(self): + card = build_approval_card(_queue_record( + method="POST", path="/api/v1/repos")) + self.assertEqual(card["risk"], "medium") + + def test_missing_pending_id_rejected(self): + record = _queue_record() + del record["pending_id"] + with self.assertRaises(ValueError): + build_approval_card(record) + + def test_tiers_filtered_to_sentinel_choice(self): + card = build_approval_card(_queue_record(), + tiers=["one-shot", "task", "bogus"]) + self.assertEqual([o["tier"] for o in card["grant_options"]], + ["one-shot", "task"]) + + def test_risk_level_function(self): + self.assertEqual(risk_level(_queue_record()), "low") + self.assertEqual(risk_level(_queue_record( + method="PUT", path="/api/v1/x")), "medium") + + def test_card_matches_channel(self): + card = build_approval_card(_queue_record()) + payload = { + "pending_id": "pend-abc123", + "nonce": "deadbeef", + "target": {"connector": "gitea", "method": "GET", + "scheme": "https", "host": "gitea.tose.top", + "port": 443, "path": "/api/v1/repos"}, + "reasons": ["host_not_allowlisted"], + "expires_at": 1900.0, + } + self.assertTrue(card_matches_channel(card, payload)) + + def test_card_mismatch_channel(self): + card = build_approval_card(_queue_record()) + payload = {"pending_id": "pend-other", "target": {}} + self.assertFalse(card_matches_channel(card, payload)) + payload = {"pending_id": "pend-abc123", + "target": {"host": "evil.example.invalid"}} + self.assertFalse(card_matches_channel(card, payload)) + + +if __name__ == "__main__": + unittest.main() diff --git a/docs/chv-muse-p12-gap.md b/docs/chv-muse-p12-gap.md new file mode 100644 index 0000000..c9ec492 --- /dev/null +++ b/docs/chv-muse-p12-gap.md @@ -0,0 +1,95 @@ +# 对齐 Muse P12 差距表：审计轨迹与活动视图 + +依据：`~/workspace/research_notes/meta-muse-ai-architecture-20261101-0755/report.md` +（官方新闻室、安全文 2026-09-08、设计文为准；第三方拆解只作参考， +不得把逆向细节写成官方架构）。基线：06c3e17。 + +## 1. Muse 官方要求（报告原文口径） + +| # | 要求 | 来源 | +|---|------|------| +| R1 | 完整审计轨迹（audit trail）展示已做与计划做的事 | 新闻室、设计文（报告 §2.2 第 7 步） | +| R2 | 头像下方显示正在做什么的摘要，点头像可见完整活动日志与已批准权限 | 设计文（任务简报口径；报告未逐字展开，列为设计文要求） | +| R3 | 确定性 UI：结构化审批卡（明确接受/拒绝动作）与凭证安全存储 | 设计文（报告 §2.6） | +| R4 | 审批请求由 Sentinel 直接发到客户端 UI（不是对话内），用户答复直接回 Sentinel | 安全文（报告 §2.6） | +| R5 | 用户可自由检查、编辑、下载 VM 内所有文件，包括 Muse 关于用户的记忆 | 安全文（报告 §3 凭证与数据政策） | + +## 2. 现状（基线 06c3e17） + +- 审计是散的：`cloud/auth.py` 的 `AuthStore.audit()` 只追加 + `(user_id, project_id, action, detail, created_at)` 到 SQLite + `audit_log` 表，`list_audit()` 只按项目倒序取；`cloud/worker.py` + 的 `Worker._audit()` 是 best-effort 包装；`worker_adapter.py` 的 + `make_chv_audit_callback` 把 CHV 事件转成 `(project_id, + "chv_vm_" + event, detail)`。没有统一记录模型，没有“已做/计划 + 做”汇总，没有活动视图，没有已批准权限清单，没有导出。 +- 审批侧：P3（分支 feat-chv-muse-p3，提交 a90cf78）有 + `ApprovalQueue`（内存态 pending 记录：目标描述 + 原因 + ts）； + P5（分支 feat-chv-muse-p5，提交 25bcd35）有对话外审批通道 + `approval_channel.py`（nonce + 15 分钟过期 + HMAC-SHA256 签名， + 答复只能用一次）与 `ScopedCapability.describe()`（能力绑定描述）。 + 但 P3/P5 都不在本基线里，本轨不能 import 它们，只能按 dict + 契约接线，集成时由 round-11 合并。 +- 状态存储：P7（feat-chv-muse-p7）正在做状态分离，本轨审计轨迹 + 的持久化先落独立 JSONL 文件 + 明确标注待迁移到 P7 存储， + 不重复造 P7 的库。 + +## 3. 本轨改什么 + +- `cloud/chv/audit_trail.py`（新，纯标准库）：统一审计记录模型 + （时间/主体/动作/对象/结果/原因/关联任务/关联审批/项目）； + 追加写 JSONL + SHA-256 哈希链（改写/删除/调序可被 `verify()` + 发现）；字段级白名单清洗（密钥、token 值、对话原文绝不进 + 审计，否定测试钉住）；事件分类覆盖任务派发、VM 生命周期、 + 同步、审批、支付（P10）、配额（P11）、记忆操作（P7 的忘记）、 + 后台任务（P8）、推理/遥测（P9）；`from_legacy_row()` 把现有 + `audit_log` 行转成统一记录（标 `legacy: true`，无哈希链）； + 导出选 JSON（不做 CSV，见 §5）。 +- `cloud/chv/activity.py`（新，纯标准库）：计划登记（已做 vs + 计划做：pending/running/done/cancelled，非法流转拒绝）； + 已批准权限登记（接 P5 `describe()` 形状：连接器/主机/用例/ + 档位/有效期/次数余量/撤销）；`activity_summary()` 纯函数拼出 + 正在做摘要 + 近期已做 + 计划清单 + 已批准权限数。 +- `cloud/chv/approval_card.py`（新，纯标准库）：结构化审批卡 + 数据（动作/目标/原因/风险分级/接受与拒绝两个明确动作/五档 + 授权类型选项，永久档标注须显式 opt-in）；`card_matches_channel()` + 校验卡数据与 P5 通道载荷一致（pending_id + 目标字段全等）。 +- 前端视图不做：本轨只给后端 + 数据接口，客户端 UI 是另一侧的 + 工作，契约见各模块 docstring。 + +## 4. 诚实边界 + +- 哈希链是**篡改可发现**，不是防篡改：能写文件的人重算整条链 + 就能掩盖改动。真防篡改要外部锚定（远端日志/签名时间戳）， + 不在本轨。 +- 审计清洗是字段名黑名单 + `key=value` 模式脱敏，不是内容理解： + 密钥若藏在自由文本字段里且不带 `token=`/`password=` 等标记， + 清洗器看不出来。调用方仍须只传描述性字段（P3/P5 的口径）。 +- 活动视图的“正在做什么”来自计划登记与审计记录，不是实时 + 进程探针；VM 内真实执行状态以执行器为准。 +- 审批卡只做数据，不做签发：档位可选项由 Sentinel 决定（P5 口 + 径），本轨只呈现调用方传进来的档位列表。 + +## 5. 决定记录 + +- 导出选 JSON 不选 CSV：JSON 保留哈希链与嵌套结构，CSV 会丢 + 字段；任务简报允许二选一。 +- 审计持久化先用独立 JSONL 文件（`audit_trail.jsonl`，0600）， + 待 P7 状态库就绪后迁移，迁移接口是 `from_legacy_row()` 同款 + 转换（本文件行 → P7 记录）。 +- project_id 强制：`list()`/`export()` 必须传 project_id，只返 + 回该项目记录；跨项目读取返回空，不抛错（与现有 `list_audit` + 行为一致，不泄露“存在别的项目记录”这一事实）。 + +## 6. 真机待验证清单（未做，排队在 round-11 之后） + +1. 审计 JSONL 在真实 VM 生命周期（wake/pause/resume/shutdown） + 下的追加与 `verify()` 表现。 +2. 活动摘要与真实任务派发/完成事件的对应（正在做是否及时变 + 更、计划状态流转是否与 worker 终态一致）。 +3. 已批准权限清单与 P5 五档能力在 wake/同步路径下的真实表现 + （签发→使用→过期→撤销全程）。 +4. 审批卡数据与 P5 对话外通道在真机上的端到端一致（卡上点的 + 接受/拒绝与通道答复同一 pending_id）。 +5. 审计文件落盘权限（0600）与多进程追加（worker + host.py） + 下的完整性。