merge(feat-chv-muse-p3): P5 起点——authd 替身凭证 + Sentinel 最小实现

chv-guest/authd.py

@@ -0,0 +1,325 @@
+#!/usr/bin/env python3
+"""chv-authd：客内凭证守护进程（feat-chv-muse-p3）。
+
+对齐 Muse 官方安全文：authd 保管凭证（含 token，存于用户 VM
+而非集中基础设施），代理只见替身令牌；凭证存入运行时单元之外
+的安全凭证存储，代理不可见，仅在需要时注入。
+
+本进程跑在客内、运行时单元之外（root 身份的 systemd 服务，
+见 chv-authd.service；P2 的 nspawn 单元落地后，codex 在单元内，
+本进程在单元外——单元内 root 映射为非特权用户，读不到本进程的
+socket 与 seed 文件）。它只做三件事：
+
+1. 启动时读 /etc/chv/authd-seed（0600 root，seed 投递），把
+   GITEA_TOKEN 持在内存里（cloud.chv.authd.CredentialVault），
+   除此之外不落盘、不进日志；
+2. 在 Unix socket /run/chv-authd/authd.sock 上接受 git credential
+   helper 的请求（SO_PEERCRED 只放行 root 与 chv 用户）；
+3. 每次放凭证前先过 Sentinel 评估（cloud.chv.sentinel，主机/
+   方法/路径/解析后 IP/SSRF），再按次释放（单次、300 秒 TTL），
+   全程审计（审计里无值）。
+
+只用标准库。cloud/chv/authd.py 与 cloud/chv/sentinel.py 是叶子
+模块（只 import 标准库），镜像构建时随本文件一起拷进
+/usr/local/lib/chv-authd/（见 docs/chv-muse-p3-gap.md §2 的接入
+说明），本文件把该目录加进 sys.path 后 import。
+"""
+
+import argparse
+import json
+import os
+import socket
+import socketserver
+import struct
+import sys
+import threading
+
+DEFAULT_LIB_DIR = "/usr/local/lib/chv-authd"
+DEFAULT_SEED_PATH = "/etc/chv/authd-seed"
+DEFAULT_SOCKET_PATH = "/run/chv-authd/authd.sock"
+DEFAULT_LOG_PATH = "/var/log/chv-authd.log"
+
+# seed 文件里允许的键（与 cloud.chv.gitea_sync.GITEA_SYNC_KEYS
+# 同款白名单；未知键一律丢掉，调用方拼错的键绝不静默进保险库）。
+SEED_KEYS = frozenset({
+    "GITEA_URL", "GITEA_USER", "GITEA_TOKEN",
+    "SYNC_REPO", "SYNC_BRANCH", "GITEA_EMAIL",
+})
+
+# SO_PEERCRED 放行的 uid：root + chv（git 以 chv 身份跑 helper）。
+# 可用环境变量 CHV_AUTHD_ALLOWED_UIDS 覆盖（逗号分隔）。
+DEFAULT_ALLOWED_UIDS = (0, 1000)
+
+
+def _load_seed(path):
+    """读 seed 文件（KEY=VALUE，每行一项）→ 白名单后的 dict。
+    文件不存在即 FileNotFoundError（调用方 fail-fast，绝不带着
+    “无凭证”的 authd 继续跑、让 git 静默拿不到凭证）。"""
+    values = {}
+    with open(path) as fh:
+        for line in fh:
+            line = line.strip()
+            if not line or line.startswith("#") or "=" not in line:
+                continue
+            key, value = line.split("=", 1)
+            key = key.strip()
+            if key in SEED_KEYS and value.strip():
+                values[key] = value.strip()
+    return values
+
+
+def _audit(log_path, record):
+    """审计落盘（JSON 行，0600）：只含绑定与 verdict，不含值。
+    落盘失败只记 stderr，绝不炸守护进程（审计失败不改变放行
+    决策是 worker 层的既定语义，见 executor._emit）。"""
+    line = json.dumps(record, sort_keys=True)
+    try:
+        with open(log_path, "a") as fh:
+            fh.write(line + "\n")
+    except OSError as exc:
+        print("chv-authd: audit 落盘失败：%s" % exc, file=sys.stderr)
+    print("chv-authd: %s" % line, file=sys.stderr)
+
+
+def _operation_for_path(path):
+    """git 智能 HTTP 路径 → 操作（helper 不知道 git 操作，守护
+    进程按路径推断；Sentinel 的 L7 规则已把路径限定在这三条内，
+    推断不到的一律拒绝）：
+    - .../git-receive-pack → push（写）
+    - .../git-upload-pack → pull（读）
+    - .../info/refs → pull（读，clone 也走这条）"""
+    if path.endswith(".git/git-receive-pack"):
+        return "POST", "push"
+    if path.endswith(".git/git-upload-pack"):
+        return "POST", "pull"
+    if path.endswith(".git/info/refs"):
+        return "GET", "pull"
+    return None, None
+
+
+def _repo_from_path(path):
+    """从 git 路径取仓库名：/admin/repo.git/info/refs → admin/repo。
+    取不出即 None（调用方拒绝）。"""
+    parts = [p for p in (path or "").split("/") if p]
+    if len(parts) < 2:
+        return None
+    repo = parts[1]
+    if not repo.endswith(".git"):
+        return None
+    return "%s/%s" % (parts[0], repo[:-len(".git")])
+
+
+class AuthdHandler(socketserver.BaseRequestHandler):
+    """单次请求处理（每连接一 JSON 行）：鉴权 → Sentinel 评估 →
+    按次释放 → 回包。任何一步拒绝都回 {ok:false}，不含值。"""
+
+    def handle(self):
+        server = self.server
+        try:
+            creds = self.request.getsockopt(
+                socket.SOL_SOCKET, socket.SO_PEERCRED,
+                struct.calcsize("3i"))
+            _pid, uid, _gid = struct.unpack("3i", creds)
+        except (OSError, struct.error):
+            self._reply({"ok": False, "error": "peercred_failed"})
+            return
+        if uid not in server.allowed_uids:
+            server.audit({"event": "authd_peer_rejected",
+                          "uid": uid})
+            self._reply({"ok": False, "error": "uid_not_allowed"})
+            return
+        try:
+            buf = b""
+            while b"\n" not in buf:
+                chunk = self.request.recv(4096)
+                if not chunk:
+                    break
+                buf += chunk
+                if len(buf) > 65536:
+                    break
+            req = json.loads(buf.decode("utf-8"))
+        except (OSError, ValueError):
+            self._reply({"ok": False, "error": "bad_request"})
+            return
+        self._serve(req)
+
+    def _reply(self, obj):
+        try:
+            self.request.sendall(
+                (json.dumps(obj) + "\n").encode("utf-8"))
+        except OSError:
+            pass
+
+    def _serve(self, req):
+        server = self.server
+        connector = req.get("connector")
+        host = (req.get("host") or "").lower()
+        path = req.get("path") or "/"
+        if connector != "gitea" or not host or not path:
+            server.audit({"event": "authd_credential_refused",
+                          "kind": "bad_request_fields",
+                          "connector": connector})
+            self._reply({"ok": False, "error": "bad_request_fields"})
+            return
+        method, operation = _operation_for_path(path)
+        if method is None:
+            server.audit({"event": "authd_credential_refused",
+                          "kind": "path_not_allowed",
+                          "connector": connector, "host": host,
+                          "path": path})
+            self._reply({"ok": False, "error": "path_not_allowed"})
+            return
+        repo = _repo_from_path(path)
+        if repo is None:
+            server.audit({"event": "authd_credential_refused",
+                          "kind": "repo_unparseable",
+                          "connector": connector, "host": host})
+            self._reply({"ok": False, "error": "repo_unparseable"})
+            return
+        # Sentinel 评估：主机/方法/路径/解析后 IP/SSRF。
+        from cloud.chv.sentinel import EgressRequest
+        scheme = "https"
+        port = 443
+        egress = EgressRequest(connector, method, scheme, host,
+                               port, path)
+        decision = server.policy.evaluate(
+            egress, resolve=getattr(server, "resolve", None))
+        if decision.verdict != "allow":
+            record = decision.audit(egress)
+            server.audit(record)
+            self._reply({"ok": False,
+                         "error": "sentinel_" + decision.verdict,
+                         "reasons": decision.reasons})
+            return
+        server.audit(decision.audit(egress))
+        # 按次释放：绑定须与入库作用域一致（主机/仓库），单次有效。
+        try:
+            release = server.vault.release(
+                connector, "GITEA_TOKEN", host, repo, operation)
+        except Exception as exc:  # AuthdRefusal：kind 在异常上
+            kind = getattr(exc, "kind", "refused")
+            server.audit({"event": "authd_credential_refused",
+                          "kind": kind, "connector": connector,
+                          "host": host, "repo": repo,
+                          "operation": operation})
+            self._reply({"ok": False, "error": kind})
+            return
+        server.audit(server.vault.audit_release(release))
+        try:
+            value = server.vault.consume(release.release_id)
+        except Exception as exc:
+            kind = getattr(exc, "kind", "refused")
+            server.audit({"event": "authd_credential_refused",
+                          "kind": kind, "connector": connector,
+                          "host": host, "repo": repo,
+                          "operation": operation})
+            self._reply({"ok": False, "error": kind})
+            return
+        self._reply({"ok": True, "username": server.gitea_user,
+                     "password": value,
+                     "expires_at": release.expires_at})
+
+
+class AuthdServer(socketserver.ThreadingUnixStreamServer):
+    """Unix socket 服务（线程化，daemon_threads=True 避免僵尸线程）。"""
+
+    daemon_threads = True
+    allow_reuse_address = False
+
+    def __init__(self, socket_path, vault, policy, gitea_user,
+                 allowed_uids, log_path, resolve=None):
+        self.vault = vault
+        self.policy = policy
+        self.gitea_user = gitea_user
+        self.allowed_uids = frozenset(allowed_uids)
+        self.log_path = log_path
+        # 解析器可注入（测试用假解析器；不给时 Sentinel 用真实
+        # getaddrinfo）。与执行器的 sentinel_resolver 同款口径。
+        self.resolve = resolve
+        directory = os.path.dirname(socket_path)
+        if directory:
+            os.makedirs(directory, exist_ok=True)
+            try:
+                os.chmod(directory, 0o700)
+            except OSError:
+                pass
+        if os.path.exists(socket_path):
+            os.unlink(socket_path)
+        super().__init__(socket_path, AuthdHandler)
+        try:
+            os.chmod(socket_path, 0o600)
+        except OSError:
+            pass
+
+    def audit(self, record):
+        _audit(self.log_path, record)
+
+
+def build_server(seed_path=DEFAULT_SEED_PATH,
+                 socket_path=DEFAULT_SOCKET_PATH,
+                 log_path=DEFAULT_LOG_PATH,
+                 lib_dir=DEFAULT_LIB_DIR,
+                 allowed_uids=None,
+                 resolve=None):
+    """建服务（不启动）：读 seed → 入库 → 建策略 → 建 socket 服务。
+
+    seed 缺必填键（GITEA_URL/GITEA_USER/GITEA_TOKEN/SYNC_REPO）
+    即 ValueError（只含缺失键名，不含值）；无 seed 文件即
+    FileNotFoundError——authd 绝不带着“无凭证”启动。"""
+    if lib_dir not in sys.path:
+        sys.path.insert(0, lib_dir)
+    from cloud.chv.authd import CredentialVault
+    from cloud.chv.sentinel import SentinelPolicy
+    seed = _load_seed(seed_path)
+    missing = sorted({"GITEA_URL", "GITEA_USER", "GITEA_TOKEN",
+                      "SYNC_REPO"} - set(seed))
+    if missing:
+        raise ValueError(
+            "authd seed 不完整，缺失：%s" % ",".join(missing))
+    from urllib.parse import urlparse
+    gitea_host = (urlparse(seed["GITEA_URL"]).hostname or "").lower()
+    if not gitea_host:
+        raise ValueError("authd seed 的 GITEA_URL 主机解析不出")
+    vault = CredentialVault()
+    vault.store("gitea", "GITEA_TOKEN", seed["GITEA_TOKEN"],
+                host=gitea_host, repo=seed["SYNC_REPO"])
+    extra = [h.strip().lower()
+             for h in (os.environ.get("CHV_SENTINEL_ALLOW_HOSTS")
+                       or "").split(",") if h.strip()]
+    policy = SentinelPolicy(
+        allow_hosts={gitea_host} | set(extra))
+    uids = allowed_uids
+    if uids is None:
+        raw = os.environ.get("CHV_AUTHD_ALLOWED_UIDS") or ""
+        uids = [int(p) for p in raw.split(",") if p.strip()] \
+            if raw.strip() else list(DEFAULT_ALLOWED_UIDS)
+    server = AuthdServer(socket_path, vault, policy,
+                         seed["GITEA_USER"], uids, log_path,
+                         resolve=resolve)
+    return server
+
+
+def main(argv=None):
+    """前台运行（systemd Type=simple 直接跑本进程，不 daemonize）。"""
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--seed", default=DEFAULT_SEED_PATH)
+    parser.add_argument("--socket", default=DEFAULT_SOCKET_PATH)
+    parser.add_argument("--log", default=DEFAULT_LOG_PATH)
+    parser.add_argument("--lib", default=DEFAULT_LIB_DIR)
+    args = parser.parse_args(argv)
+    server = build_server(seed_path=args.seed,
+                          socket_path=args.socket,
+                          log_path=args.log, lib_dir=args.lib)
+    print("chv-authd: listening on %s" % args.socket,
+          file=sys.stderr)
+    try:
+        server.serve_forever()
+    except KeyboardInterrupt:
+        pass
+    finally:
+        server.server_close()
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

chv-guest/chv-authd.service

@@ -0,0 +1,22 @@
+[Unit]
+Description=CHV authd (credential broker for chv-sync)
+# 注意时序：本服务须在 chv-sync boot 之前就绪。cloud-final 的
+# runcmd（99-chv-sync.cfg）在 chv-sync boot 之前显式
+# `systemctl start chv-authd`；这里不依赖 After=cloud-final
+#（cloud-final 跑完才 After 得到它，太晚）。enable 后开机自启
+# 只是兜底，首启顺序以 runcmd 的显式 start 为准。
+After=network-online.target
+
+[Service]
+Type=simple
+# authd 跑在运行时单元之外（root 身份）：它是凭证的唯一持有者，
+# 单元内的 agent 只能经 Unix socket + SO_PEERCRED 向它要凭证，
+# 读不到它的内存与 seed 文件（P2 单元隔离落地后）。
+ExecStart=/usr/bin/python3 /usr/local/bin/chv-authd
+Restart=on-failure
+RestartSec=2
+StandardOutput=journal
+StandardError=journal
+
+[Install]
+WantedBy=multi-user.target

chv-guest/chv-sync.sh

@@ -7,19 +7,18 @@
 # 的规则：分叉或冲突时失败关闭，不强制覆盖、不自动 rebase、
 # 不静默丢提交。
 #
-# 凭证绝不进镜像。启动时由 seed 盘投递 /etc/chv/guest-env
-# （0600，root 所有），格式（KEY=VALUE，每行一项）：
-#   GITEA_URL=https://gitea.tose.top
-#   GITEA_USER=<Gitea 用户名>
-#   GITEA_TOKEN=<专用 API token，仅此次测试/此次 VM 有效>
-#   SYNC_REPO=<owner/repo>
-#   SYNC_BRANCH=<可选，默认分支>
-#   GITEA_EMAIL=<可选，commit 身份邮箱；不给则留镜像占位>
-# boot（root，cloud-init runcmd 调）读 guest-env，把 token 写进
+# 凭证绝不进镜像。启动时由 seed 盘投递两份文件（0600，root 所有）：
+#   /etc/chv/guest-env（非密钥同步变量：GITEA_URL/GITEA_USER/
+#     SYNC_REPO/SYNC_BRANCH/GITEA_EMAIL）与
+#   /etc/chv/authd-seed（feat-chv-muse-p3：只装 GITEA_TOKEN，
+#     只给客内 authd 守护进程读）。
+# boot（root，cloud-init runcmd 调）读 guest-env，authd 在时配
+# git credential helper 走 authd 按次取凭证（只进 git 子进程
+# 内存，不落盘）；authd 不在时（旧镜像）才回落到把 token 写进
 # /home/chv/.git-credentials（0600，chv 所有，git credential
-# store 标准机制），再以 chv 身份 clone 到 ~/workspace/<仓名>。
-# pull/push/status 以 chv 身份跑，只用已落盘的 credential，不
-# 再读 guest-env；token 不会出现在 git remote URL 或镜像里。
+# store 标准机制）并警告。pull/push/status 以 chv 身份跑，用
+# helper 或已落盘的 credential，不再读 guest-env；token 不会
+# 出现在 git remote URL 或镜像里。
 #
 # 用法：
 #   chv-sync boot              # 启动时 clone（已 clone 则静默退出，不自动 pull）
@@ -64,7 +63,6 @@ cmd_boot() {
     set -a; . "$GUEST_ENV"; set +a
     : "${GITEA_URL:?guest-env 缺 GITEA_URL}"
     : "${GITEA_USER:?guest-env 缺 GITEA_USER}"
-    : "${GITEA_TOKEN:?guest-env 缺 GITEA_TOKEN}"
     : "${SYNC_REPO:?guest-env 缺 SYNC_REPO}"
     [ "$(id -u)" = "0" ] || die "boot 须以 root 运行（cloud-init runcmd）"
     local host="${GITEA_URL#https://}"
@@ -74,12 +72,31 @@ cmd_boot() {
     local dir="$WORKSPACE/$name"
     mkdir -p "$WORKSPACE"
     chown "$SYNC_USER:$SYNC_USER" "$WORKSPACE"
-    # token 只落盘进用户家目录的 credential store（0600），
-    # 不进 git remote URL、不进镜像、不进日志（此处不打印 token）。
-    local creds="$USER_HOME/.git-credentials"
-    printf 'https://%s:%s@%s\n' "$GITEA_USER" "$GITEA_TOKEN" "$host" > "$creds"
-    chmod 0600 "$creds"
-    chown "$SYNC_USER:$SYNC_USER" "$creds"
+    # feat-chv-muse-p3：authd 路径——socket 存在时凭证走 helper，
+    # 不再把真 token 写进 .git-credentials 落盘；helper 按次向
+    # authd 取凭证，只进 git 子进程内存。socket 不存在（旧镜像）
+    # 时走旧路径 + 警告（过渡兼容，旧镜像退役后改为 fail-closed）。
+    # 注意：authd 路径下 GITEA_TOKEN 不在 guest-env 里（它只进
+    # authd-seed 给守护进程），所以这里不要求它；旧路径才要求。
+    local authd_socket="${CHV_AUTHD_SOCKET:-/run/chv-authd/authd.sock}"
+    local helper="/usr/local/bin/git-credential-chv-authd"
+    if [ -S "$authd_socket" ] && [ -x "$helper" ]; then
+        runuser -u "$SYNC_USER" -- env HOME="$USER_HOME" git config --global credential.helper "$helper"
+        # helper 按路径推断仓库与操作（Sentinel L7 绑定），git
+        # 默认不把 path 发给 helper（须 credential.useHttpPath），
+        # 不开它每次都会因“请求不完整”拒绝——新测试抓到的真问题。
+        runuser -u "$SYNC_USER" -- env HOME="$USER_HOME" git config --global credential.useHttpPath true
+        info "凭证走 chv-authd（不落盘）"
+    else
+        : "${GITEA_TOKEN:?guest-env 缺 GITEA_TOKEN（旧路径）}"
+        echo "chv-sync: 警告：chv-authd 未运行，凭证走旧落盘路径（须用含 authd 的新镜像）" >&2
+        # token 只落盘进用户家目录的 credential store（0600），
+        # 不进 git remote URL、不进镜像、不进日志（此处不打印 token）。
+        local creds="$USER_HOME/.git-credentials"
+        printf 'https://%s:%s@%s\n' "$GITEA_USER" "$GITEA_TOKEN" "$host" > "$creds"
+        chmod 0600 "$creds"
+        chown "$SYNC_USER:$SYNC_USER" "$creds"
+    fi
     if [ -d "$dir/.git" ]; then
         info "已 clone：$dir（不自动 pull，改动由用户显式同步）"
         exit 0

chv-guest/git-credential-chv-authd

@@ -0,0 +1,107 @@
+#!/usr/bin/env python3
+"""git credential helper：向 chv-authd 按次取凭证（feat-chv-muse-p3）。
+
+git credential 协议：git 把 `protocol/host/path`（每行 `key=value`，
+空行结束）写进本进程 stdin；本进程向 authd 的 Unix socket 发
+一次 JSON 请求，拿到用户名/口令后按协议回给 git（属性 + 空行）。
+
+口令只进 git 子进程内存：本进程不写文件、不打日志（拒绝时只
+报 kind，不含值）；`password_expiry_utc` 按释放的 expires_at
+回给 git，git 在该时间后重新问 helper（不缓存过期凭证）。
+authd 拒绝 / socket 不通时本进程 exit 1——git 操作随之失败
+（fail-closed，绝不回落成“无凭证也照推”）。
+
+用法（chv-sync boot 配好后 git 自动调，不用手工跑）：
+  git config --global credential.helper /usr/local/bin/git-credential-chv-authd
+  git config --global credential.useHttpPath true
+（后者不可少：git 默认不把 path 发给 helper，没有 path 本进程
+无法推断仓库与操作，会直接拒绝。）
+socket 路径可用 CHV_AUTHD_SOCKET 覆盖（测试用 /tmp 下的假 socket）。
+"""
+
+import json
+import os
+import socket
+import sys
+
+DEFAULT_SOCKET_PATH = "/run/chv-authd/authd.sock"
+SOCKET_TIMEOUT_SECONDS = 5
+
+
+def _read_credential_request():
+    """读 git credential 协议的请求属性（空行结束）→ dict。"""
+    attrs = {}
+    for line in sys.stdin:
+        line = line.rstrip("\n")
+        if not line:
+            break
+        if "=" in line:
+            key, value = line.split("=", 1)
+            attrs[key.strip()] = value.strip()
+    return attrs
+
+
+def _ask_authd(socket_path, host, path):
+    """向 authd 要一次凭证：返回 (username, password, expires_at)。
+    拒绝/通信失败即 RuntimeError（只含 kind，不含值）。"""
+    req = {"connector": "gitea", "host": host, "path": path}
+    sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+    sock.settimeout(SOCKET_TIMEOUT_SECONDS)
+    try:
+        sock.connect(socket_path)
+        sock.sendall((json.dumps(req) + "\n").encode("utf-8"))
+        chunks = []
+        while True:
+            data = sock.recv(65536)
+            if not data:
+                break
+            chunks.append(data)
+            if b"\n" in data:
+                break
+    except OSError as exc:
+        raise RuntimeError("authd_unreachable:%s" % exc)
+    finally:
+        sock.close()
+    try:
+        resp = json.loads(b"".join(chunks).decode("utf-8"))
+    except ValueError:
+        raise RuntimeError("authd_bad_response")
+    if not resp.get("ok"):
+        raise RuntimeError("authd_refused:%s"
+                           % resp.get("error", "refused"))
+    return (resp["username"], resp["password"],
+            resp.get("expires_at"))
+
+
+def main(argv=None):
+    """helper 入口：读协议 → 问 authd → 回属性。拒绝即 exit 1。"""
+    attrs = _read_credential_request()
+    host = attrs.get("host", "")
+    path = attrs.get("path", "")
+    if attrs.get("protocol") != "https" or not host or not path:
+        print("git-credential-chv-authd: 请求不完整（须 protocol=https "
+              "+ host + path），拒绝", file=sys.stderr)
+        return 1
+    if not path.startswith("/"):
+        path = "/" + path
+    # git 的 info/refs 路径带 ?service=... 查询串：Sentinel 的
+    # L7 规则按干净路径匹配，这里先去掉查询串再问 authd。
+    path = path.split("?", 1)[0]
+    socket_path = os.environ.get("CHV_AUTHD_SOCKET") \
+        or DEFAULT_SOCKET_PATH
+    try:
+        username, password, expires_at = _ask_authd(
+            socket_path, host, path)
+    except RuntimeError as exc:
+        print("git-credential-chv-authd: %s" % exc, file=sys.stderr)
+        return 1
+    print("username=%s" % username)
+    print("password=%s" % password)
+    if expires_at:
+        print("password_expiry_utc=%d" % int(expires_at))
+    print()
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

cloud/chv/authd.py

@@ -0,0 +1,311 @@
+"""authd 式替身凭证服务（feat-chv-muse-p3，叶子模块）。
+
+对齐 Muse 官方安全文（security.muse.ai 2026-09-08，见
+docs/chv-muse-p3-gap.md）：凭证由 VM 内独立服务 authd 保管
+（含 OAuth token，存于用户 VM 而非集中基础设施），代理只见
+替身令牌；内置连接器的执行分离——每个 privsep worker 有明确
+凭证白名单（日历 worker 不能仅改参数就索取邮件凭证）；登录
+凭证存入运行时单元之外的安全凭证存储，代理不可见，仅在需要
+时注入。
+
+本模块是纯逻辑（只用标准库，不依赖 cloud.*），宿主侧与客内
+authd 守护进程共用同一份代码（见 chv-guest/authd.py）：
+- `CredentialVault`：只活内存的保险库；`store()` 只接受白名单
+  内的（连接器，凭证键）；`release()` 按次发放，绑定
+  {连接器, 主机, 仓库路径, 操作}，单次有效 + 300 秒 TTL，
+  过期即废；`__repr__` 与审计记录里绝无凭证值。
+- 替身的两档（见差距表 §1）：本期默认是“按次释放”——释放给
+  git 子进程的值仍是真 token，但只进 git 进程内存、不进 agent
+  环境/argv/日志/落盘；真铸 token（Gitea API 铸短期专用 token）
+  由 `mint_gitea_token()` 实现，默认关闭（须真 token 有建 token
+  权限，现有 write:repository 测试 token 铸 token 会失败）。
+- 位置决策：authd 跑在客内、单元外（root 身份的 systemd 服务），
+  不在宿主侧——Muse 的 authd 就在用户 VM 里；宿主侧方案要走
+  vsock（P1 范围、未就绪），且会把全部 VM 的 token 集中到宿主。
+"""
+
+import hashlib
+import hmac
+import json
+import secrets
+import time
+import urllib.error
+import urllib.request
+from collections.abc import Callable, Iterable
+
+# 连接器白名单（privsep 口径）：本期只认 gitea 一个连接器。
+# 未知连接器一律拒绝，不得以“默认放行”扩展。
+ALLOWED_CONNECTORS = frozenset({"gitea"})
+
+# 每连接器的凭证白名单：连接器 → 允许的凭证键。gitea 只允许
+# GITEA_TOKEN；其它键（哪怕调用方传进来）一律拒绝入库。
+CONNECTOR_CREDENTIALS = {
+    "gitea": frozenset({"GITEA_TOKEN"}),
+}
+
+# 按次释放的默认有效期（秒）与次数：单次、300 秒，过期即废。
+DEFAULT_RELEASE_TTL_SECONDS = 300
+DEFAULT_RELEASE_MAX_USES = 1
+
+
+class AuthdRefusal(Exception):
+    """authd 拒绝：kind + 原因，不含任何凭证值。"""
+
+    def __init__(self, kind: str, message: str) -> None:
+        super().__init__("authd_refused:%s: %s" % (kind, message))
+        self.kind = kind
+
+
+class SurrogateRelease:
+    """一次按次释放（替身）：绑定到一次具体的凭证使用。
+
+    release_id 是不透明 id（审计与吊销用，不含值）；value 只在
+    释放对象里出现一次，由调用方（客内守护进程）经 Unix socket
+    交给 git credential helper、再由 helper 交给 git 子进程——
+    除此之外不进任何环境、argv、日志、审计。`consume()` 校验
+    绑定与有效期，单次有效：用过一次即废，过期即废。"""
+
+    def __init__(self, release_id: str, connector: str, key: str,
+                 host: str, repo: str, operation: str, value: str,
+                 expires_at: float, max_uses: int = 1) -> None:
+        self.release_id = release_id
+        self.connector = connector
+        self.key = key
+        self.host = host
+        self.repo = repo
+        self.operation = operation
+        self.value = value
+        self.expires_at = expires_at
+        self.uses_left = max_uses
+
+    def describe(self) -> dict:
+        """审计用描述（只含绑定，不含值）。"""
+        return {"release_id": self.release_id,
+                "connector": self.connector, "key": self.key,
+                "host": self.host, "repo": self.repo,
+                "operation": self.operation,
+                "expires_at": self.expires_at,
+                "uses_left": self.uses_left}
+
+    def __repr__(self) -> str:  # 绝不含 value
+        return ("SurrogateRelease(id=%s connector=%s host=%s "
+                "repo=%s operation=%s)"
+                % (self.release_id, self.connector, self.host,
+                   self.repo, self.operation))
+
+
+class CredentialVault:
+    """只活内存的凭证保险库（一个 VM 一个实例，客内 authd 持有）。
+
+    - 值只存在 `self._store` 的内存 dict 里，不落盘、不进日志；
+    - `store()` 校验白名单：未知连接器 / 该连接器不允许的键 /
+      空值一律拒绝入库（AuthdRefusal），绝不静默收下；
+    - `release()` 按次发放：调用方须给出完整绑定（主机、仓库、
+      操作），绑定与入库时的作用域不一致即拒绝；
+    - `consume()` 单次兑现：release 兑现一次即从库里删除，过期
+      也兑现不了；兑现失败不留可重试的残值。
+    clock 可注入（测试用假时钟测过期），默认 time.time。"""
+
+    def __init__(self, clock: Callable[[], float] | None = None) -> None:
+        self._store: dict[tuple[str, str], dict] = {}
+        self._releases: dict[str, SurrogateRelease] = {}
+        self._clock = clock or time.time
+
+    def __repr__(self) -> str:  # 绝不含任何凭证值
+        return "CredentialVault(connectors=%s)" % sorted(
+            {c for (c, _k) in self._store})
+
+    def store(self, connector: str, key: str, value: str,
+              host: str | None = None, repo: str | None = None) -> None:
+        """入库：白名单校验 + 作用域记录（host/repo 是该凭证
+        允许使用的范围，release 时须一致）。"""
+        if connector not in ALLOWED_CONNECTORS:
+            raise AuthdRefusal(
+                "unknown_connector",
+                "连接器 %s 不在 authd 白名单内" % connector)
+        allowed = CONNECTOR_CREDENTIALS.get(connector, frozenset())
+        if key not in allowed:
+            raise AuthdRefusal(
+                "credential_not_allowlisted",
+                "连接器 %s 不允许凭证键 %s" % (connector, key))
+        if not isinstance(value, str) or not value.strip():
+            raise AuthdRefusal(
+                "empty_credential", "凭证值为空，拒绝入库")
+        self._store[(connector, key)] = {
+            "value": value, "host": host, "repo": repo}
+
+    def has(self, connector: str, key: str) -> bool:
+        """该（连接器，键）是否已入库（不返回值）。"""
+        return (connector, key) in self._store
+
+    def forget(self, connector: str, key: str) -> bool:
+        """删一条入库凭证（VM 停止时调用）；同时作废其全部未兑现
+        释放。未知键返回 False，不抛错。"""
+        gone = self._store.pop((connector, key), None) is not None
+        if gone:
+            doomed = [rid for rid, rel in self._releases.items()
+                      if rel.connector == connector
+                      and rel.key == key]
+            for rid in doomed:
+                self._releases.pop(rid, None)
+        return gone
+
+    def release(self, connector: str, key: str, host: str, repo: str,
+                operation: str,
+                ttl_seconds: int = DEFAULT_RELEASE_TTL_SECONDS,
+                max_uses: int = DEFAULT_RELEASE_MAX_USES
+                ) -> SurrogateRelease:
+        """按次发放：绑定须与入库作用域一致（host/repo 有记录时
+        须一致，无记录时本次绑定即定作用域），否则拒绝。发放的
+        释放单次有效、ttl 后过期；发放本身记入库内待兑现表，
+        须经 consume() 兑现一次才交出值。"""
+        entry = self._store.get((connector, key))
+        if entry is None:
+            raise AuthdRefusal(
+                "credential_not_stored",
+                "连接器 %s 的凭证 %s 未入库" % (connector, key))
+        if entry["host"] is not None and entry["host"] != host:
+            raise AuthdRefusal(
+                "scope_mismatch",
+                "凭证作用域主机不一致，拒绝释放")
+        if entry["repo"] is not None and entry["repo"] != repo:
+            raise AuthdRefusal(
+                "scope_mismatch",
+                "凭证作用域仓库不一致，拒绝释放")
+        if operation not in ("clone", "pull", "push"):
+            raise AuthdRefusal(
+                "operation_not_allowed",
+                "操作 %s 不在允许集内" % operation)
+        release = SurrogateRelease(
+            release_id=secrets.token_hex(8),
+            connector=connector, key=key, host=host, repo=repo,
+            operation=operation, value=entry["value"],
+            expires_at=self._clock() + ttl_seconds,
+            max_uses=max_uses)
+        self._releases[release.release_id] = release
+        return release
+
+    def consume(self, release_id: str) -> str:
+        """兑现一次释放：过期/用完/未知 id 一律拒绝（AuthdRefusal，
+        不含值）。兑现成功返回凭证值，并把该释放从库里删除
+        （单次有效，用过即废）。"""
+        release = self._releases.get(release_id)
+        if release is None:
+            raise AuthdRefusal(
+                "release_unknown", "释放记录不存在或已兑现")
+        if self._clock() >= release.expires_at:
+            self._releases.pop(release_id, None)
+            raise AuthdRefusal("release_expired", "释放已过期")
+        if release.uses_left <= 0:
+            self._releases.pop(release_id, None)
+            raise AuthdRefusal("release_exhausted", "释放已用完")
+        release.uses_left -= 1
+        value = release.value
+        self._releases.pop(release_id, None)
+        return value
+
+    def audit_release(self, release: SurrogateRelease) -> dict:
+        """发放审计记录（只含绑定，不含值）。"""
+        record = release.describe()
+        record["event"] = "authd_credential_released"
+        return record
+
+    def audit_refusal(self, kind: str, connector: str,
+                      key: str | None = None) -> dict:
+        """拒绝审计记录（只含 kind 与名，不含值）。"""
+        return {"event": "authd_credential_refused", "kind": kind,
+                "connector": connector, "key": key}
+
+
+def _default_post(url: str, headers: dict, payload: dict):
+    """默认 HTTP 客户端（urllib）：返回 (status, body_dict)。
+    只用于 mint_gitea_token，测试注入假 post，不联网。"""
+    data = json.dumps(payload).encode("utf-8")
+    req = urllib.request.Request(url, data=data, headers=headers,
+                                 method="POST")
+    try:
+        with urllib.request.urlopen(req, timeout=15) as resp:
+            return resp.status, json.loads(resp.read() or b"{}")
+    except urllib.error.HTTPError as exc:
+        try:
+            body = json.loads(exc.read() or b"{}")
+        except ValueError:
+            body = {}
+        return exc.code, body
+
+
+def mint_gitea_token(api_url: str, user: str, master_token: str,
+                     name: str, scopes: Iterable[str] = ("write:repository",),
+                     post: Callable[[str, dict, dict], tuple[int, dict]]
+                     | None = None) -> str:
+    """经 Gitea API 铸短期专用 token（真替身路径，默认关闭）。
+
+    调 `POST {api_url}/users/{user}/tokens`（Gitea 官方 API），
+    新 token 与 master token 是不同的值、可独立吊销（DELETE 同
+    路径），到期由调用方吊销。返回新 token 值；失败即
+    AuthdRefusal（不含任何 token 值）。
+    前提（见差距表 §3.2）：master token 须有建 token 的权限；
+    现有 write:repository 测试 token 铸 token 会失败（403），
+    开这条路须另配有铸 token 权限的 token 并做真机验证。
+    post 可注入（测试用假 post），默认走 urllib。"""
+    if not api_url or not user or not master_token or not name:
+        raise AuthdRefusal(
+            "mint_args_missing", "铸 token 参数不全")
+    base = api_url.rstrip("/")
+    url = "%s/users/%s/tokens" % (base, user)
+    headers = {"Authorization": "token %s" % master_token,
+               "Content-Type": "application/json"}
+    payload = {"name": name, "scopes": list(scopes)}
+    post_fn = post or _default_post
+    status, body = post_fn(url, headers, payload)
+    if status not in (200, 201):
+        raise AuthdRefusal(
+            "mint_failed",
+            "Gitea 铸 token 失败：status=%s" % status)
+    token = (body or {}).get("sha1") or (body or {}).get("token")
+    if not token:
+        raise AuthdRefusal(
+            "mint_empty", "Gitea 铸 token 未返回 token 值")
+    return token
+
+
+def revoke_gitea_token(api_url: str, user: str, master_token: str,
+                       token_id: object,
+                       delete: Callable[[str, dict], int] | None = None
+                       ) -> bool:
+    """吊销一个铸出的 token（`DELETE {api}/users/{user}/tokens/{id}`）。
+    成功返回 True；404 视为已不在（也返回 True，不报错）；
+    其它失败返回 False（调用方审计，不抛错——吊销失败不能炸任务，
+    但须留审计）。delete 可注入，默认走 urllib。"""
+    if not api_url or not user or not master_token \
+            or token_id is None:
+        return False
+    base = api_url.rstrip("/")
+    url = "%s/users/%s/tokens/%s" % (base, user, token_id)
+    headers = {"Authorization": "token %s" % master_token}
+    if delete is None:
+        def _do_delete(url: str, headers: dict) -> int:
+            req = urllib.request.Request(url, headers=headers,
+                                         method="DELETE")
+            try:
+                with urllib.request.urlopen(req, timeout=15) as resp:
+                    return resp.status
+            except urllib.error.HTTPError as exc:
+                return exc.code
+        delete = _do_delete
+    status = delete(url, headers)
+    return status in (200, 204, 404)
+
+
+def surrogate_fingerprint(value: str) -> str:
+    """凭证值的指纹（审计/日志里唯一允许出现的凭证派生形式）：
+    SHA-256 前 8 位 hex。指纹不可逆推出原值，审计里只记指纹、
+    不记值。"""
+    return hashlib.sha256(value.encode("utf-8")).hexdigest()[:8]
+
+
+def hmac_sign(key: str, message: str) -> str:
+    """HMAC-SHA256 签名（释放记录的防篡改绑定，客内守护进程
+    用它签发 release_id 的校验码；与 Gitea 无关）。"""
+    return hmac.new(key.encode("utf-8"), message.encode("utf-8"),
+                    hashlib.sha256).hexdigest()

cloud/chv/config.py

@@ -153,6 +153,12 @@ def _network_fragment(params: dict, env: dict) -> dict:
         "egress_proxy": env.get("CHV_EGRESS_PROXY"),
         "allow_direct_egress": env.get("CHV_ALLOW_DIRECT_EGRESS")
         == "1",
+        # feat-chv-muse-p3：Sentinel 白名单主机（逗号分隔），
+        # 默认只放行配置的 Gitea 主机（执行器 wake 预审时追加）。
+        "sentinel_allow_hosts": tuple(
+            h.strip().lower()
+            for h in (env.get("CHV_SENTINEL_ALLOW_HOSTS") or "")
+            .split(",") if h.strip()),
     }
 
 

cloud/chv/executor.py

@@ -279,6 +279,7 @@ class CloudHypervisorExecutor(Executor):
     disk_gb: float
     egress_proxy: str | None
     allow_direct_egress: bool
+    sentinel_allow_hosts: tuple
     task_timeout: float | None
     snapshot_url: str | None
     warm_pool_size: int
@@ -303,6 +304,8 @@ class CloudHypervisorExecutor(Executor):
     audit_failure_count: int
     _vms: dict
     _guest_env_delivered: dict
+    _authd_delivered: dict
+    sentinel_resolver: "Callable[..., object] | None"
 
     def __init__(self, binary: str | None = None,
                  kernel: str | None = None,
@@ -329,7 +332,9 @@ class CloudHypervisorExecutor(Executor):
                  api_timeout: float | None = None,
                  ready_timeout: float | None = None,
                  ready_interval: float | None = None,
-                 host_key_mode: str | None = None) -> None:
+                 host_key_mode: str | None = None,
+                 sentinel_resolver: Callable[..., object] | None = None
+                 ) -> None:
         self._env: dict = env if env is not None else os.environ  # type: ignore[assignment]
         # 配置解析已拆到 cloud/chv/config.py（参数优先于环境
         # 变量，默认值与 #216 一致；显式空用户/越界端口的收紧
@@ -342,13 +347,15 @@ class CloudHypervisorExecutor(Executor):
             "ssh_user": ssh_user, "ssh_port": ssh_port,
             "guest_host_py": guest_host_py, "cmdline": cmdline,
             "net": net, "net_json": net_json, "ssh_host": ssh_host,
-            "kvm_path": kvm_path,
-            "cpuinfo_path": cpuinfo_path, "ch_remote": ch_remote,
-            "model": model, "api_timeout": api_timeout,
-            "ready_timeout": ready_timeout,
+            "kvm_path": kvm_path, "cpuinfo_path": cpuinfo_path,
+            "ch_remote": ch_remote, "model": model,
+            "api_timeout": api_timeout, "ready_timeout": ready_timeout,
             "ready_interval": ready_interval,
             "host_key_mode": host_key_mode}, self._env)
         self._copy_config_fields(_cfg)
+        # feat-chv-muse-p3：Sentinel 解析器可注入（测试用假解析器
+        # 测 SSRF 挡截；不给时 evaluate 用 socket.getaddrinfo）。
+        self.sentinel_resolver = sentinel_resolver
         self._init_runtime(runner, launcher, api_client, sleeper)
 
     def _init_runtime(self, runner: Callable[..., object] | None,
@@ -367,6 +374,7 @@ class CloudHypervisorExecutor(Executor):
         # 名 source 客内文件，未投递的密钥名一律拒绝（防止
         # “wake 没投递、执行时却按名 source 一个旧文件”）。
         self._guest_env_delivered = {}
+        self._init_authd_runtime()
         # feat-chv-closeup（P2-20）：_guest_env_delivered 是
         # 跨 wake/build_ssh_command/stop 的共享字典，wake 与
         # 执行可并发在不同线程——此前无锁读写，set/get/pop
@@ -396,6 +404,19 @@ class CloudHypervisorExecutor(Executor):
             self.launcher = _default_popen_launcher
         self._vms = {}
 
+    def _init_authd_runtime(self) -> None:
+        """feat-chv-muse-p3：authd 投递集与 Sentinel 待审批队列。
+        _authd_delivered 是本实例已走 authd 通道投递密钥的 VM
+        名 → 已投递密钥名集合（与 _guest_env_delivered 同锁、
+        同款语义，唯一区别：authd 投递的密钥名在 build_ssh_command
+        里不报错、但绝不 source 进 codex 进程环境——它只给客内
+        authd 守护进程，任务进程里永不出现）。_sentinel_queue
+        是 Sentinel ask 待审批队列（内存态，进程重启即清；
+        ask 本来就意味着这次执行已停止）。"""
+        self._authd_delivered = {}
+        from cloud.chv.sentinel import ApprovalQueue
+        self._sentinel_queue = ApprovalQueue()
+
     def _delivered_secrets(self, name: str) -> set:
         """P2-20 唯一读入口：该 VM 已 seed 投递的密钥名集合。"""
         with self._guest_env_lock:
@@ -415,6 +436,26 @@ class CloudHypervisorExecutor(Executor):
         with self._guest_env_lock:
             self._guest_env_delivered.pop(name, None)
 
+    def _delivered_authd_secrets(self, name: str) -> set:
+        """feat-chv-muse-p3：该 VM 已走 authd 通道投递的密钥名
+        集合（与 _delivered_secrets 同款读入口）。"""
+        with self._guest_env_lock:
+            return set(self._authd_delivered.get(name, set()))
+
+    def _mark_authd_delivered(self, name: str, secrets) -> None:
+        """feat-chv-muse-p3：记下该 VM 已走 authd 通道投递的
+        密钥名集合（secrets 为空即清记录，与旧语义一致）。"""
+        with self._guest_env_lock:
+            if secrets:
+                self._authd_delivered[name] = set(secrets)
+            else:
+                self._authd_delivered.pop(name, None)
+
+    def _clear_authd_delivered(self, name: str) -> None:
+        """feat-chv-muse-p3：stop 时清掉该 VM 的 authd 投递记录。"""
+        with self._guest_env_lock:
+            self._authd_delivered.pop(name, None)
+
     def _copy_config_fields(self, _cfg) -> None:
         """把 build_chv_config 的字段抄到 self（纯移动，字段集不变）。"""
         for _field in (
@@ -431,7 +472,7 @@ class CloudHypervisorExecutor(Executor):
                 "snapshot_max_total_gb", "snapshot_retention_days",
                 "snapshot_dir", "allow_remote_snapshot",
                 "gitea_sync_env", "seed_user_data_extra_file",
-                "seed_network_config_file"):
+                "seed_network_config_file", "sentinel_allow_hosts"):
             setattr(self, _field, getattr(_cfg, _field))
 
     # -- 名字 / 路径 ---------------------------------------------------
@@ -868,7 +909,8 @@ class CloudHypervisorExecutor(Executor):
     def provision_host_key(self, task: dict, secret_env: dict | None = None,
                            guest_env: dict | None = None,
                            user_data_extra: str | None = None,
-                           network_config: str | None = None) -> dict:
+                           network_config: str | None = None,
+                           authd_env: dict | None = None) -> dict:
         """按 VM 预置 host key（补缺口轨强方案，见 hostkeys.py）：
         启动前为该 VM 生成一次性 ed25519 host key 对，公钥
         立即写进该 VM 的 known_hosts（钉死），私钥只写进
@@ -887,17 +929,11 @@ class CloudHypervisorExecutor(Executor):
         # to10：known_hosts 写入已抽到 _write_provisioned_known_hosts，
         # 此处只留生成与 seed 写入的 import（删后不再有未用 import）。
         from cloud.chv.hostkeys import (
-            dirty_marker_path,
             generate_host_keypair,
             write_seed,
         )
         vm = self.vm_name(task)
-        if os.path.exists(dirty_marker_path(self.state_dir, vm)):
-            raise ChvFailure(
-                "vm_dirty_refused",
-                "VM %s 上次清理失败（脏标记存在），拒绝复用旧 "
-                "host key 启动：请人工清理 state_dir 后删脏标记"
-                % vm)
+        self._refuse_dirty_vm(vm)
         paths = generate_host_keypair(
             self.state_dir, vm,
             runner=lambda cmd: self._call_runner(cmd))
@@ -908,13 +944,25 @@ class CloudHypervisorExecutor(Executor):
                               secret_env=secret_env,
                               guest_env=guest_env,
                               user_data_extra=user_data_extra,
-                              network_config=network_config)
+                              network_config=network_config,
+                              authd_env=authd_env)
         self._emit("host_key_generated", task,
                    "vm=%s secret_names=%s"
                    % (vm, ",".join(sorted(secret_env or {}))))
         return {"paths": paths, "seed_dir": seed_dir,
                 "known_hosts": paths["known_hosts"]}
 
+    def _refuse_dirty_vm(self, vm: str) -> None:
+        """上次清理失败留下的脏标记存在时直接拒绝（清理失败
+        即拒起，不再复用旧 key 对）。"""
+        from cloud.chv.hostkeys import dirty_marker_path
+        if os.path.exists(dirty_marker_path(self.state_dir, vm)):
+            raise ChvFailure(
+                "vm_dirty_refused",
+                "VM %s 上次清理失败（脏标记存在），拒绝复用旧 "
+                "host key 启动：请人工清理 state_dir 后删脏标记"
+                % vm)
+
     def capture_host_key(self, task: dict) -> bool:
         """首次捕获客机 host key（TOFU，弱回落，边界见
         ssh_transport.py 与 hostkeys.py）：ssh-keyscan 经本机
@@ -974,25 +1022,42 @@ class CloudHypervisorExecutor(Executor):
             # 构造兼容，wake/wait 路径仍严格解析。
             ssh_host = "127.0.0.1"
         secrets, plain = split_secret_env(extra_env)
-        guest_env_file = None
-        if secrets:
-            from cloud.chv.hostkeys import GUEST_ENV_PATH
-            delivered = self._delivered_secrets(self.vm_name(task))
-            missing: list = [name for name in secrets if name not in delivered]
-            if missing:
-                raise ChvFailure(
-                    "secret_env_not_delivered",
-                    "密钥环境变量 %s 未经 seed 投递，拒绝进 ssh "
-                    "argv：provisioned 模式下 wake 时投递，"
-                    "keyscan 模式不支持密钥投递"
-                    % ",".join(sorted(missing)))
-            guest_env_file = GUEST_ENV_PATH
+        guest_env_file = self._resolve_secret_env_file(task, secrets)
         return _build_ssh_cmd(
             self.ssh_port, self.ssh_key, self.ssh_user,
             self.egress_proxy, self.known_hosts_path(task),
             inner_cmd, ssh_host=ssh_host, extra_env=plain or None,
             guest_env_file=guest_env_file)
 
+    def _resolve_secret_env_file(self, task: dict, secrets) -> str | None:
+        """feat-chv-muse-p3：密钥分两路查投递，返回远程命令
+        该 source 的客内密钥文件（无 guest-env 路密钥即 None）。
+        authd 路的（GITEA_TOKEN）只认“已投递、不报错”，绝不
+        source 进 codex 进程环境；guest-env 路的才 source
+        guest-env 文件。两路任一路未投递都 fail-closed（同款 kind）。"""
+        if not secrets:
+            return None
+        from cloud.chv.gitea_sync import split_authd_env
+        from cloud.chv.hostkeys import GUEST_ENV_PATH
+        authd_secrets, guest_secrets = split_authd_env(secrets)
+        name = self.vm_name(task)
+        missing: list = [
+            n for n in authd_secrets
+            if n not in self._delivered_authd_secrets(name)]
+        missing += [
+            n for n in guest_secrets
+            if n not in self._delivered_secrets(name)]
+        if missing:
+            raise ChvFailure(
+                "secret_env_not_delivered",
+                "密钥环境变量 %s 未经 seed 投递，拒绝进 ssh "
+                "argv：provisioned 模式下 wake 时投递，"
+                "keyscan 模式不支持密钥投递"
+                % ",".join(sorted(missing)))
+        if guest_secrets:
+            return GUEST_ENV_PATH
+        return None
+
     # -- API 模式：VmConfig / API 调用 / 就绪轮询 -----------------------
     def _vm_config_disks(self, task_disk: str | None,
                          seed_disk: str | None) -> list:
@@ -1440,6 +1505,107 @@ class CloudHypervisorExecutor(Executor):
         merged.update(sync)
         return merged
 
+    def _sentinel_gate_gitea(self, task: dict,
+                             extra_env: dict | None) -> None:
+        """feat-chv-muse-p3：wake 时的 Gitea 出口 Sentinel 预审。
+
+        同步已请求（extra_env 里有 GITEA_URL）时才评估；无同步
+        变量即跳过（本次启动不同步，无出口可审）。评估目标：
+        GET https://<Gitea 主机>/api/v1/version。
+        - allow → 审计 sentinel_egress_allow，通过；
+        - deny → 审计 + ChvFailure(sentinel_egress_denied)，
+          起 VMM 前 fail-closed；
+        - ask → 记入待审批队列 + 审计 sentinel_egress_ask，
+          同样 fail-closed：wake 等不了审批 UI，ask 在这里就是
+          deny，须运维把主机加进 CHV_SENTINEL_ALLOW_HOSTS 再起。
+        白名单来源：CHV_SENTINEL_ALLOW_HOSTS + 运维配置的 Gitea
+        主机（CHV_GITEA_URL 的主机，运维可信）；任务级覆盖改了
+        GITEA_URL 主机时不自动放行——改到未知主机就是 ask。
+        SSRF 检查对白名单主机同样生效（DNS 可被投毒，白名单主机
+        解析到私网 IP 仍 deny）。审计 detail 只含 VM 名/主机/
+        verdict，不含任何凭证值。"""
+        sync_url = (extra_env or {}).get("GITEA_URL")
+        if not sync_url:
+            return
+        name = self.vm_name(task)
+        host, scheme, port = self._gate_gitea_target(task, name, sync_url)
+        self._gate_gitea_apply(task, name, host, scheme, port)
+
+    def _gate_gitea_target(self, task: dict, name: str,
+                           sync_url: str) -> tuple:
+        """解析 Gitea 出口预审目标（主机/协议/端口）；主机解析
+        不出即审计 + fail-closed。"""
+        from urllib.parse import urlparse
+        parsed = urlparse(sync_url)
+        host = (parsed.hostname or "").lower()
+        scheme = (parsed.scheme or "").lower()
+        port = parsed.port or (443 if scheme == "https" else 80)
+        if not host:
+            self._emit("sentinel_egress_denied", task,
+                       "vm=%s gitea_url_invalid" % name)
+            raise ChvFailure(
+                "sentinel_egress_denied",
+                "Gitea URL 主机解析不出，拒绝启动")
+        return host, scheme, port
+
+    def _gate_gitea_policy(self, host: str):
+        """预审策略：白名单来源 CHV_SENTINEL_ALLOW_HOSTS +
+        运维配置的 Gitea 主机（CHV_GITEA_URL 的主机，运维可信）；
+        任务级覆盖改了 GITEA_URL 主机时不自动放行——改到未知
+        主机就是 ask。"""
+        from urllib.parse import urlparse
+        from cloud.chv.sentinel import SentinelPolicy
+        cfg_sync = getattr(self, "gitea_sync_env", None) or {}
+        cfg_host = ""
+        cfg_url = cfg_sync.get("GITEA_URL")
+        if cfg_url:
+            cfg_host = (urlparse(cfg_url).hostname or "").lower()
+        extra_allow = {host} if host and host == cfg_host else set()
+        return SentinelPolicy.from_env(
+            self._env, extra_allow_hosts=extra_allow)
+
+    def _gate_gitea_apply(self, task: dict, name: str, host: str,
+                          scheme: str, port: int) -> None:
+        """对预审目标做 Sentinel 评估并按 verdict 处理：
+        allow → 审计通过；ask → 记入待审批队列 + 审计，同样
+        fail-closed（wake 等不了审批 UI，ask 在这里就是 deny，
+        须运维把主机加进 CHV_SENTINEL_ALLOW_HOSTS 再起）；
+        deny → 审计 + ChvFailure，起 VMM 前 fail-closed。
+        SSRF 检查对白名单主机同样生效（DNS 可被投毒，白名单
+        主机解析到私网 IP 仍 deny）。审计 detail 只含 VM 名/
+        主机/verdict，不含任何凭证值。"""
+        from cloud.chv.sentinel import EgressRequest
+        policy = self._gate_gitea_policy(host)
+        request = EgressRequest("gitea", "GET", scheme, host, port,
+                                "/api/v1/version")
+        decision = policy.evaluate(
+            request, resolve=getattr(self, "sentinel_resolver", None))
+        if decision.verdict == "allow":
+            capability = decision.capability
+            assert capability is not None  # allow 恒带严格能力
+            self._emit("sentinel_egress_allow", task,
+                       "vm=%s host=%s capability=%s"
+                       % (name, host, capability.id))
+            return
+        if decision.verdict == "ask":
+            pending_id = self._sentinel_queue.record(
+                request, decision.reasons)
+            self._emit("sentinel_egress_ask", task,
+                       "vm=%s host=%s pending=%s"
+                       % (name, host, pending_id))
+            raise ChvFailure(
+                "sentinel_egress_denied",
+                "Gitea 主机 %s 不在 Sentinel 白名单内（待审批 %s），"
+                "拒绝启动：请运维加进 CHV_SENTINEL_ALLOW_HOSTS"
+                % (host, pending_id))
+        self._emit("sentinel_egress_denied", task,
+                   "vm=%s host=%s reasons=%s"
+                   % (name, host, ",".join(decision.reasons)))
+        raise ChvFailure(
+            "sentinel_egress_denied",
+            "Gitea 出口被 Sentinel 拒绝（%s），拒绝启动"
+            % ",".join(decision.reasons))
+
     def _wake_prechecks(self, task: dict, name: str, extra_env) -> tuple:
         """wake 前置：preflight/容量/配额/net/密钥模式，纯移动。
         失败时发 vm_failed 审计并原样抛 ChvFailure。"""
@@ -1471,6 +1637,9 @@ class CloudHypervisorExecutor(Executor):
                     "provisioned 模式支持（当前 %s）"
                     % (",".join(sorted(secrets)),
                        getattr(self, "host_key_mode", "provisioned")))
+            # feat-chv-muse-p3：Gitea 出口 Sentinel 预审（起 VMM
+            # 前 fail-closed，见 _sentinel_gate_gitea）。
+            self._sentinel_gate_gitea(task, extra_env)
         except ChvFailure as exc:
             self._emit("vm_failed", task,
                        "vm=%s wake 前置失败 kind=%s" % (name, exc.kind),
@@ -1478,6 +1647,20 @@ class CloudHypervisorExecutor(Executor):
             raise
         return secrets, net_entries
 
+    def _split_seed_authd_env(self, secrets,
+                              guest_env: dict | None) -> tuple:
+        """feat-chv-muse-p3：seed 密钥集先拆通道——GITEA_TOKEN
+        走 authd（独立的 authd-seed 文件，只给客内 authd 读），
+        其余密钥仍走 guest-env。guest_env（split_guest_env 拼
+        出来的）里也含 GITEA_TOKEN：一并拆出来（与 secrets 拆
+        出的合并，secrets 的值优先），否则 write_seed 会拒收。
+        返回 (authd_env, guest_secrets, guest_env)。"""
+        from cloud.chv.gitea_sync import split_authd_env
+        authd_env, guest_secrets = split_authd_env(secrets)
+        dup_authd, guest_env = split_authd_env(guest_env)
+        dup_authd.update(authd_env)
+        return dup_authd, guest_secrets, guest_env
+
     def _provision_seed_disk(self, task: dict, name: str, secrets,
                              guest_env: dict | None = None,
                              seed_extras: tuple | None = None) -> str | None:
@@ -1503,16 +1686,24 @@ class CloudHypervisorExecutor(Executor):
                 seed_disk_path,
             )
             extra, netcfg = seed_extras or (None, None)
+            authd_env, guest_secrets, guest_env = \
+                self._split_seed_authd_env(secrets, guest_env)
             provisioned = self.provision_host_key(
-                task, secret_env=secrets or None,
+                task, secret_env=guest_secrets or None,
                 guest_env=guest_env or None,
-                user_data_extra=extra, network_config=netcfg)
+                user_data_extra=extra, network_config=netcfg,
+                authd_env=authd_env or None)
             seed_disk = seed_disk_path(self.state_dir, name)
             build_seed_disk(
                 provisioned["seed_dir"], seed_disk,
                 runner=lambda cmd: self._call_runner(cmd))
             delete_host_private_key(self.state_dir, name)
-            self._mark_guest_env_delivered(name, secrets)
+            # 两路投递记录分开记：guest-env 路的进
+            # _guest_env_delivered（任务期可 source），authd 路的进
+            # _authd_delivered（任务期只认“已投递、不报错”，绝不
+            # source 进 codex 环境）。
+            self._mark_guest_env_delivered(name, guest_secrets)
+            self._mark_authd_delivered(name, authd_env)
         return seed_disk
 
     def _boot_vm(self, task, name, task_disk, seed_disk,
@@ -1869,6 +2060,7 @@ class CloudHypervisorExecutor(Executor):
             self._emit("vm_net_registry_delete_failed", task,
                        "vm=%s error=%s" % (name, _net_exc))
         self._clear_guest_env_delivered(name)
+        self._clear_authd_delivered(name)
         return failed
 
     def _write_dirty_marker(self, name, failed):

cloud/chv/gitea_sync.py

@@ -43,6 +43,26 @@ GITEA_SYNC_KEYS = frozenset({
 # _CHV_SECRET_ENV_NAMES）。
 GITEA_SYNC_SECRET_KEYS = frozenset({"GITEA_TOKEN"})
 
+# 其中 authd 专用密钥（feat-chv-muse-p3）：只进客内 authd 的
+# 独立 seed 文件 /etc/chv/authd-seed（见 hostkeys.AUTHD_SEED_PATH），
+# 不进 guest-env、不进 codex 进程环境。本期只有 GITEA_TOKEN。
+AUTHD_SECRET_KEYS = frozenset({"GITEA_TOKEN"})
+
+
+def split_authd_env(secrets: dict | None) -> tuple[dict, dict]:
+    """把密钥集拆成（走 authd 通道的子集，仍走 guest-env 的子集）。
+
+    authd 子集 = AUTHD_SECRET_KEYS 命中的键（本期只有 GITEA_TOKEN）；
+    其余密钥（如 SUB2API_API_KEY）仍走 guest-env（见差距表 §3.3，
+    网关 key 的 privsep 改造是后续工作）。"""
+    authd, rest = {}, {}
+    for key, value in dict(secrets or {}).items():
+        if key in AUTHD_SECRET_KEYS:
+            authd[key] = value
+        else:
+            rest[key] = value
+    return authd, rest
+
 # 其中非密钥、但 boot 时就要用、也必须进 guest-env 的键
 #（SSH env 前缀到不了 cloud-init 时运行的 chv-sync boot）。
 GITEA_SYNC_PLAIN_KEYS = GITEA_SYNC_KEYS - GITEA_SYNC_SECRET_KEYS

cloud/chv/hostkeys.py

@@ -42,6 +42,17 @@ from cloud.chv.atomic_file import atomic_write_text
 # 除 seed 目录/user-data（随 VM 销毁而删）外不另存该文件。
 GUEST_ENV_PATH = "/etc/chv/guest-env"
 
+# seed 投递的客内 authd 专用密钥文件路径（feat-chv-muse-p3）：
+# 只写 AUTHD_SECRET_KEYS 命中的密钥（本期只有 GITEA_TOKEN），
+# 0600 root，只给客内 authd 守护进程读；codex 进程环境里永不
+# 出现这些键（见执行器 _authd_delivered）。与 guest-env 分离
+# 的理由：guest-env 会被远程命令 source 进任务进程环境，任何
+# 进 guest-env 的密钥都会进 codex 环境；authd 密钥绝不能走
+# 那条路。文件本身仍是 seed 写出的落盘文件（cloud-init 的唯一
+# 投递通道就是写文件），真正让它对 agent 不可见的是 P2 的
+# nspawn 单元隔离，见差距表 §3.1。
+AUTHD_SEED_PATH = "/etc/chv/authd-seed"
+
 
 def host_key_paths(state_dir: str, vm_name: str) -> dict:
     """该 VM 的 host key 文件路径（私钥/公钥/known_hosts）。"""
@@ -180,6 +191,24 @@ def build_seed_user_data(private_key_text: str, public_key_text: str) -> str:
     return "\n".join(lines) + "\n"
 
 
+def _authd_env_entry(authd_env: dict) -> list:
+    """authd-seed 的 write_files 条目行（0600 root，值经
+    shlex.quote 引用）：只装 AUTHD_SECRET_KEYS 命中的密钥，
+    与 guest-env 条目并列写进同一 write_files 列表（绝不另写
+    一个 write_files: 键，见 _guest_env_entry 的 YAML 重复键
+    禁令）。"""
+    lines = [
+        "  - path: %s" % AUTHD_SEED_PATH,
+        "    permissions: '0600'",
+        "    owner: root:root",
+        "    content: |",
+    ]
+    for key in sorted(authd_env):
+        lines.append("      %s=%s"
+                     % (key, shlex.quote(str(authd_env[key]))))
+    return lines
+
+
 def _guest_env_entry(merged_env: dict) -> list:
     """guest-env 的 write_files 条目行（0600 root，值经
     shlex.quote 引用）：只能并进已有的 write_files 列表，
@@ -227,12 +256,68 @@ def _apply_seed_extras(seed_dir: str, user_data: str,
     return user_data
 
 
+def _validate_authd_env(authd_env: dict | None) -> None:
+    """authd_env 的键必须全在 AUTHD_SECRET_KEYS 内（本期只有
+    GITEA_TOKEN），否则 ValueError——authd 通道只装 authd
+    密钥，别的密钥误进这里会绕过 guest-env 的 source 语义。"""
+    if not authd_env:
+        return
+    from cloud.chv.gitea_sync import AUTHD_SECRET_KEYS
+    unknown = sorted(set(authd_env) - AUTHD_SECRET_KEYS)
+    if unknown:
+        raise ValueError(
+            "authd 通道只接受 %s，拒绝：%s"
+            % (",".join(sorted(AUTHD_SECRET_KEYS)),
+               ",".join(unknown)))
+
+
+def _write_seed_meta(seed_dir: str, vm_name: str) -> str:
+    """建 seed 目录并写 meta-data（0600），返回 user-data 路径。
+    feat-chv-cover100：seed 暂存文件走 atomic_file（mkstemp
+    0600 → 写 → fsync → rename）：user-data 含 host 私钥
+    明文，直接写 + 事后 chmod 会在 chmod 前以 umask 权限
+    短暂可读；半截文件到不了 ISO（见 H5），原子写同时堵
+    住权限窗口与半截残留。"""
+    os.makedirs(seed_dir, exist_ok=True)
+    meta_path = os.path.join(seed_dir, "meta-data")
+    user_path = os.path.join(seed_dir, "user-data")
+    atomic_write_text(
+        meta_path,
+        "instance-id: %s\nlocal-hostname: %s\n" % (vm_name, vm_name))
+    return user_path
+
+
+def _seed_env_entries(merged_env: dict,
+                      authd_env: dict | None) -> list:
+    """guest-env 与 authd-seed 两路的 write_files 条目行
+    （皆空即空列表，不写条目）。"""
+    extra_lines = []
+    if merged_env:
+        extra_lines += _guest_env_entry(merged_env)
+    if authd_env:
+        extra_lines += _authd_env_entry(authd_env)
+    return extra_lines
+
+
+def _insert_seed_entries(user_data: str, lines: list) -> str:
+    """条目插在 runcmd 段之前、仍属同一 write_files 列表
+    （seed 自 2026-11-07 起不再带 runcmd，见
+    build_seed_user_data 的根因禁令；marker 分支只作
+    防御，将来模板若恢复 runcmd 仍插在其前）。"""
+    marker = "runcmd:\n"
+    if marker in user_data:
+        head, tail = user_data.split(marker, 1)
+        return head + "\n".join(lines) + "\n" + marker + tail
+    return user_data + "\n".join(lines) + "\n"
+
+
 def write_seed(seed_dir: str, vm_name: str, private_key_text: str,
                public_key_text: str,
                secret_env: dict | None = None,
                guest_env: dict | None = None,
                user_data_extra: str | None = None,
-               network_config: str | None = None) -> str:
+               network_config: str | None = None,
+               authd_env: dict | None = None) -> str:
     """写 seed 目录：meta-data + user-data（均为 0600），
     返回 seed 目录路径。调用方负责随 VM 销毁删掉它。
 
@@ -243,34 +328,28 @@ def write_seed(seed_dir: str, vm_name: str, private_key_text: str,
     装 boot 时要用的非密钥同步变量（GITEA_URL 等，见
     gitea_sync）：只走 SSH 前缀到不了 cloud-init 时运行
     的 chv-sync boot。两者皆空时不写该条目。
+    authd_env（feat-chv-muse-p3）给定时写进另一个独立条目
+    AUTHD_SEED_PATH（0600 root，只给客内 authd 守护进程读）：
+    键必须全在 gitea_sync.AUTHD_SECRET_KEYS 内（本期只有
+    GITEA_TOKEN），否则 ValueError——authd 通道只装 authd
+    密钥，别的密钥误进这里会绕过 guest-env 的 source 语义。
     user_data_extra / network_config 见 _apply_seed_extras
     （客机用户与静态网络的供给，皆空即不写）。"""
-    os.makedirs(seed_dir, exist_ok=True)
-    meta_path = os.path.join(seed_dir, "meta-data")
-    user_path = os.path.join(seed_dir, "user-data")
-    # feat-chv-cover100：seed 暂存文件走 atomic_file（mkstemp
-    # 0600 → 写 → fsync → rename）：user-data 含 host 私钥
-    # 明文，直接写 + 事后 chmod 会在 chmod 前以 umask 权限
-    # 短暂可读；半截文件到不了 ISO（见 H5），原子写同时堵
-    # 住权限窗口与半截残留。
-    atomic_write_text(
-        meta_path,
-        "instance-id: %s\nlocal-hostname: %s\n" % (vm_name, vm_name))
+    _validate_authd_env(authd_env)
+    user_path = _write_seed_meta(seed_dir, vm_name)
     user_data = build_seed_user_data(private_key_text, public_key_text)
     merged_env = dict(guest_env or {})
     merged_env.update(secret_env or {})
-    if merged_env:
-        lines = _guest_env_entry(merged_env)
-        # 条目插在 runcmd 段之前、仍属同一 write_files 列表
-        #（seed 自 2026-11-07 起不再带 runcmd，见
-        # build_seed_user_data 的根因禁令；marker 分支只作
-        # 防御，将来模板若恢复 runcmd 仍插在其前）。
-        marker = "runcmd:\n"
-        if marker in user_data:
-            head, tail = user_data.split(marker, 1)
-            user_data = head + "\n".join(lines) + "\n" + marker + tail
-        else:
-            user_data = user_data + "\n".join(lines) + "\n"
+    # feat-chv-muse-p3：GITEA_TOKEN 绝不进 guest-env——它只能走
+    # authd_env（独立的 authd-seed 条目）。调用方（执行器）须先
+    # 用 gitea_sync.split_authd_env() 拆出来；这里再拦一次，
+    # 免得别的调用点把旧写法带进来、悄悄恢复泄露面。
+    if "GITEA_TOKEN" in merged_env:
+        raise ValueError(
+            "GITEA_TOKEN 须走 authd_env，不进 guest-env")
+    extra_lines = _seed_env_entries(merged_env, authd_env)
+    if extra_lines:
+        user_data = _insert_seed_entries(user_data, extra_lines)
     user_data = _apply_seed_extras(seed_dir, user_data,
                                    user_data_extra, network_config)
     atomic_write_text(user_path, user_data)

cloud/chv/sentinel.py

@@ -0,0 +1,342 @@
+"""Sentinel 式出口审批决策引擎（feat-chv-muse-p3，叶子模块）。
+
+对齐 Muse 官方安全文（security.muse.ai 2026-09-08，见
+docs/chv-muse-p3-gap.md）：Sentinel 是连接器动作与全部网络出口的
+唯一权限机构；L4 与 L7 评估（主机名、解析后与最终目的 IP、端口、
+协议、HTTP 方法、路径、解码后的实际请求）；SSRF 防护（防止公开
+主机名在 DNS 解析后指向私有基础设施）；审批是严格能力
+（strict capabilities），绑定到特定连接器/目的地与用例，支持
+一次性/会话级/任务级/时限级授权；审批 UI 不在对话内。
+
+本模块只是决策层（纯函数 + 内存态队列），不是网络层强制：
+- 它不拦截客内流量。客内直连的强制仍靠 Tinyproxy 白名单 +
+  宿主防火墙（第三棒落地中）；eBPF cgroup 拦截 + LSM hook 污点
+  传播是后续工作，见差距表 §3.4，不得把本模块写成“出口已全部
+  经 Sentinel”。
+- 本期两个接线点：① 执行器 wake 时的 Gitea 出口预审（起 VMM 前
+  fail-closed）；② 客内 authd 放凭证前评估。两者都调本模块的
+  `evaluate()`，决策记录进审计（无值）。
+- `ask` 的语义：主机不在白名单、也不在拒绝名单时返回 ask，并记入
+  `ApprovalQueue` 待审批；调用方遇到 ask 必须停止执行、不得自动
+  放行。审批 UI 不在对话内——本期只留队列，不实现任何对话内审批
+  入口（见差距表 §1）。
+
+本模块只用标准库，不依赖 cloud.*，客内 authd 守护进程可直接
+import 它（见 chv-guest/authd.py）。解析器可注入（`resolve`
+参数），默认 `socket.getaddrinfo`；测试用假解析器，绝不联网。
+审计记录里绝不含凭证值、也绝不含解析出的 IP 以外的内容——
+reason 只含网段名与策略名。
+"""
+
+import ipaddress
+import socket
+import time
+import uuid
+from collections.abc import Callable, Iterable
+
+# SSRF 挡截：解析到这些网段的 IP 一律 deny。公开主机名解析到
+# 私有地址即挡（Muse 安全文的 SSRF 防护口径）；解析失败也
+# deny（fail-closed，不把“解析不出”当成“可以放行”）。
+_BLOCKED_NETWORKS = tuple(
+    ipaddress.ip_network(net) for net in (
+        "0.0.0.0/8",          # 本网络 / 未指定
+        "10.0.0.0/8",         # RFC1918
+        "100.64.0.0/10",      # CGNAT 共享地址空间
+        "127.0.0.0/8",        # 回环
+        "169.254.0.0/16",     # 链路本地
+        "172.16.0.0/12",      # RFC1918
+        "192.0.0.0/24",       # IETF 协议分配
+        "192.0.2.0/24",       # 文档 TEST-NET-1
+        "192.88.99.0/24",     # 6to4 中继（已废弃）
+        "192.168.0.0/16",     # RFC1918
+        # 注意：198.18.0.0/15（基准测试）故意不在表里——本部署的
+        # egress/DNS 基础设施自用该网段，gitea.tose.top 在此环境
+        # 真实解析到 198.18.139.152；挡掉它会把真 Gitea 主机也拒了。
+        "198.51.100.0/24",    # 文档 TEST-NET-2
+        "203.0.113.0/24",     # 文档 TEST-NET-3
+        "224.0.0.0/4",        # 组播
+        "240.0.0.0/4",        # 保留
+        "255.255.255.255/32",
+        "::1/128",            # 回环
+        "::/128",             # 未指定
+        "fc00::/7",           # 唯一本地
+        "fe80::/10",          # 链路本地
+        "ff00::/8",           # 组播
+    ))
+
+# 每连接器的 L7 路径策略：(方法, 匹配方式, 模式)。
+# gitea：API 只放 /api/v1/；git 智能 HTTP 只放三条（clone/pull/push
+# 的实际路径，见 git http-protocol）。其它路径（Web UI、登录页、
+# 任意管理接口）一律不在本连接器的能力内。
+_CONNECTOR_PATH_RULES = {
+    "gitea": (
+        ("GET", "prefix", "/api/v1/"),
+        ("POST", "prefix", "/api/v1/"),
+        ("GET", "suffix", ".git/info/refs"),
+        ("POST", "suffix", ".git/git-upload-pack"),
+        ("POST", "suffix", ".git/git-receive-pack"),
+    ),
+}
+
+# 每连接器允许的 scheme/端口（默认：只 https/443；http 只在
+# 显式放行时才出现，本期没有任何连接器放 http）。
+_CONNECTOR_SCHEMES = {"gitea": frozenset({"https"})}
+_CONNECTOR_PORTS = {"gitea": frozenset({443})}
+
+# 默认放凭证有效期（秒）：与 authd 按次释放同款 300 秒。
+DEFAULT_CAPABILITY_TTL_SECONDS = 300
+
+
+def _blocked_network_for(ip_text: str):
+    """该 IP 落在哪个挡截网段里（无即 None）。非法 IP 文本
+    按“挡不住也说不清”处理：返回 0.0.0.0/8（deny），绝不
+    把解析异常当成放行信号。"""
+    try:
+        addr = ipaddress.ip_address(ip_text)
+    except ValueError:
+        return _BLOCKED_NETWORKS[0]
+    for net in _BLOCKED_NETWORKS:
+        if addr.version == net.version and addr in net:
+            return net
+    return None
+
+
+def _default_resolve(host: str, port: int) -> list:
+    """默认解析器：socket.getaddrinfo 去重后的 IP 列表。"""
+    infos = socket.getaddrinfo(host, port, proto=socket.IPPROTO_TCP)
+    out = []
+    for info in infos:
+        ip = info[4][0]
+        if ip not in out:
+            out.append(ip)
+    return out
+
+
+class EgressRequest:
+    """一次出口请求（决策输入）：连接器 + 方法 + 目标 + 解析后
+    IP（解析在 evaluate 里做，这里只记目标描述，不记任何凭证值）。"""
+
+    def __init__(self, connector: str, method: str, scheme: str,
+                 host: str, port: int, path: str) -> None:
+        self.connector = connector
+        self.method = (method or "").upper()
+        self.scheme = (scheme or "").lower()
+        self.host = (host or "").lower()
+        self.port = int(port)
+        self.path = path or "/"
+
+    def describe(self) -> dict:
+        """审计用描述（只含连接器/目标/方法，不含值）。"""
+        return {"connector": self.connector, "method": self.method,
+                "scheme": self.scheme, "host": self.host,
+                "port": self.port, "path": self.path}
+
+
+class Capability:
+    """严格能力（Muse 口径：绑定到特定连接器/目的地与用例）。
+
+    一次 evaluate().capability 只对：一个连接器、一个主机、声明
+    的方法与路径模式、在有效期内、剩余次数内有效。越界使用
+    （换主机、换方法、换路径、过期、用完）一律 `matches()` 为
+    False，调用方必须重新走 evaluate，不得自行续期。"""
+
+    def __init__(self, connector: str, host: str,
+                 methods: Iterable[str],
+                 path_rules: Iterable[tuple[str, str, str]],
+                 ttl_seconds: int = DEFAULT_CAPABILITY_TTL_SECONDS,
+                 max_uses: int = 1, now: float | None = None) -> None:
+        self.id = uuid.uuid4().hex[:16]
+        self.connector = connector
+        self.host = (host or "").lower()
+        self.methods = frozenset(methods)
+        self.path_rules = tuple(path_rules)
+        self.expires_at = (now if now is not None else time.time()) \
+            + ttl_seconds
+        self.max_uses = max_uses
+        self.uses_left = max_uses
+
+    def matches(self, request: EgressRequest,
+                now: float | None = None) -> bool:
+        """该请求是否在能力绑定内（不消耗次数，只判定）。"""
+        if request.connector != self.connector:
+            return False
+        if request.host != self.host:
+            return False
+        if request.method not in self.methods:
+            return False
+        if not _path_allowed(request, self.path_rules):
+            return False
+        at = now if now is not None else time.time()
+        if at >= self.expires_at:
+            return False
+        if self.uses_left <= 0:
+            return False
+        return True
+
+    def use(self, request: EgressRequest,
+            now: float | None = None) -> bool:
+        """消耗一次能力：绑定内才扣次数并返回 True，否则 False
+        （不抛错，调用方按 deny 处理）。"""
+        if not self.matches(request, now=now):
+            return False
+        self.uses_left -= 1
+        return True
+
+
+class Decision:
+    """一次审批决策：allow / deny / ask + 原因 + 能力（仅 allow
+    时有）+ 审计记录（无值）。"""
+
+    def __init__(self, verdict: str, reasons: Iterable[str],
+                 capability: Capability | None = None) -> None:
+        assert verdict in ("allow", "deny", "ask")
+        self.verdict = verdict
+        self.reasons = list(reasons)
+        self.capability = capability
+
+    def audit(self, request: EgressRequest) -> dict:
+        """审计记录：事件名按 verdict，detail 只含目标描述与原因。"""
+        record = request.describe()
+        record["event"] = "sentinel_egress_" + self.verdict
+        record["verdict"] = self.verdict
+        record["reasons"] = list(self.reasons)
+        if self.capability is not None:
+            record["capability_id"] = self.capability.id
+        return record
+
+
+def _path_allowed(request: EgressRequest, rules) -> bool:
+    """路径是否在该连接器的 L7 规则内。"""
+    for method, kind, pattern in rules:
+        if request.method != method:
+            continue
+        if kind == "prefix" and request.path.startswith(pattern):
+            return True
+        if kind == "suffix" and request.path.endswith(pattern):
+            return True
+    return False
+
+
+class SentinelPolicy:
+    """Sentinel 策略：白名单主机 + 拒绝主机 + 连接器 L7 规则。
+
+    默认拒绝（default-deny）：主机不在白名单、也不在拒绝名单时
+    返回 ask（不是 allow）；拒绝名单命中、SSRF 命中、scheme/
+    端口/路径越界时返回 deny。白名单只放行“已知且解析干净”的
+    目标——白名单不是“解析后是什么都行”，SSRF 检查对白名单
+    主机同样生效（DNS 可被投毒，白名单主机解析到私网 IP 仍 deny）。"""
+
+    def __init__(self, allow_hosts: Iterable[str] = (),
+                 deny_hosts: Iterable[str] = (),
+                 connector_paths: dict[str, tuple] | None = None
+                 ) -> None:
+        self.allow_hosts = frozenset(
+            h.lower() for h in (allow_hosts or ()))
+        self.deny_hosts = frozenset(
+            h.lower() for h in (deny_hosts or ()))
+        self.connector_paths = dict(
+            connector_paths if connector_paths is not None
+            else _CONNECTOR_PATH_RULES)
+
+    @classmethod
+    def from_env(cls: type["SentinelPolicy"], env: dict | None,
+                 extra_allow_hosts: Iterable[str] = ()
+                 ) -> "SentinelPolicy":
+        """从宿主环境变量建策略：CHV_SENTINEL_ALLOW_HOSTS（逗号
+        分隔主机名）+ 调用方追加的白名单（执行器 wake 预审时把
+        配置的 Gitea 主机追加进来——只放行配置的 Gitea 主机，
+        不是放行任意主机）。"""
+        raw = (env or {}).get("CHV_SENTINEL_ALLOW_HOSTS") or ""
+        hosts = [h.strip().lower() for h in raw.split(",") if h.strip()]
+        hosts += [h.lower() for h in (extra_allow_hosts or ())]
+        return cls(allow_hosts=hosts)
+
+    def evaluate(self, request: EgressRequest,
+                 resolve: Callable[[str, int], list] | None = None,
+                 now: float | None = None) -> Decision:
+        """评估一次出口请求（按顺序，首个命中即定 verdict）：
+
+        1. 未知连接器 → deny（没有策略的连接器不放行）；
+        2. scheme 不在该连接器允许集 → deny；
+        3. 端口不在该连接器允许集 → deny；
+        4. 主机在拒绝名单 → deny；
+        5. 解析失败 → deny（fail-closed）；
+        6. 任一解析 IP 落在挡截网段 → deny（SSRF）；
+        7. 主机不在白名单 → ask（记入待审批队列，调用方停止执行）；
+        8. 方法/路径不在连接器 L7 规则内 → deny；
+        9. 全部通过 → allow + 严格能力（单次、300 秒）。
+        """
+        rules = self.connector_paths.get(request.connector)
+        if rules is None:
+            return Decision("deny", ["unknown_connector"])
+        if request.scheme not in _CONNECTOR_SCHEMES.get(
+                request.connector, frozenset()):
+            return Decision("deny", ["scheme_not_allowed"])
+        if request.port not in _CONNECTOR_PORTS.get(
+                request.connector, frozenset()):
+            return Decision("deny", ["port_not_allowed"])
+        if request.host in self.deny_hosts:
+            return Decision("deny", ["host_denied"])
+        resolver = resolve or _default_resolve
+        try:
+            ips = resolver(request.host, request.port)
+        except Exception:
+            return Decision("deny", ["dns_resolution_failed"])
+        if not ips:
+            return Decision("deny", ["dns_resolution_failed"])
+        for ip in ips:
+            net = _blocked_network_for(ip)
+            if net is not None:
+                return Decision(
+                    "deny", ["ssrf_private_ip:%s" % net])
+        if request.host not in self.allow_hosts:
+            return Decision("ask", ["host_not_allowlisted"])
+        if not _path_allowed(request, rules):
+            return Decision("deny", ["path_not_allowed"])
+        capability = Capability(
+            request.connector, request.host,
+            {request.method}, rules,
+            ttl_seconds=DEFAULT_CAPABILITY_TTL_SECONDS,
+            max_uses=1, now=now)
+        return Decision("allow", ["policy_allow"],
+                        capability=capability)
+
+
+class ApprovalQueue:
+    """ask 待审批队列（内存态）：ask 出现即记入，执行停止。
+
+    设计约束（Muse 口径）：审批 UI 不在对话内。本期只留队列与
+    查询/标记接口，不实现任何对话内审批入口，也不自动放行。
+    队列是内存态，进程重启即清——ask 本来就意味着“这次执行
+    已停止”，重启后须重新申请。"""
+
+    def __init__(self, clock: Callable[[], float] | None = None) -> None:
+        self._pending: dict[str, dict] = {}
+        self._clock = clock or time.time
+
+    def record(self, request: EgressRequest,
+               reasons: Iterable[str]) -> str:
+        """记入一条待审批，返回 pending id（只含目标描述与原因）。"""
+        pending_id = uuid.uuid4().hex[:16]
+        record = request.describe()
+        record["pending_id"] = pending_id
+        record["reasons"] = list(reasons)
+        record["ts"] = self._clock()
+        self._pending[pending_id] = record
+        return pending_id
+
+    def list_pending(self) -> list:
+        """全部待审批（副本，按记录时间排序）。"""
+        return [dict(v) for _, v in sorted(
+            self._pending.items(), key=lambda kv: kv[1]["ts"])]
+
+    def resolve(self, pending_id: str, verdict: str) -> bool:
+        """标记一条待审批已处理（verdict 只记 allow/deny，不在
+        这里执行放行——放行须调用方重新走 evaluate）。未知 id
+        返回 False，不抛错。"""
+        if pending_id not in self._pending:
+            return False
+        if verdict not in ("allow", "deny"):
+            return False
+        self._pending.pop(pending_id)
+        return True

cloud/tests/test_chv_arch_maint_fe.py

@@ -76,6 +76,7 @@ EXPECTED_FIELDS = {
     "snapshot_max_total_gb", "snapshot_retention_days",
     "snapshot_dir", "allow_remote_snapshot", "snapshot_url",
     "task_timeout", "api_timeout", "ready_timeout", "ready_interval",
+    "sentinel_allow_hosts",
 }
 
 
@@ -255,7 +256,7 @@ class ConfigFragmentsTest(unittest.TestCase):
             self.assertFalse(seen & set(frag), "片段键重叠")
             seen |= set(frag)
         self.assertEqual(seen, EXPECTED_FIELDS)
-        self.assertEqual(len(EXPECTED_FIELDS), 38)
+        self.assertEqual(len(EXPECTED_FIELDS), 39)
 
     def test_defaults_and_overrides(self):
         cfg = build_chv_config({}, {})

cloud/tests/test_chv_gaps.py

@@ -365,22 +365,34 @@ class HostKeyProvisionTest(ChvGapCase):
     def test_write_seed_with_secret_has_no_runcmd(self):
         # 带 secret_env 时 guest-env 条目仍须并进同一个
         # write_files 列表（单 write_files 键），且 user-data
-        # 里仍不能出现 runcmd。
+        # 里仍不能出现 runcmd。feat-chv-muse-p3：GITEA_TOKEN
+        # 不再走 secret_env，改走 authd_env（独立的 authd-seed
+        # 条目，0600）；guest-env 条目里只剩非 authd 密钥。
         with tempfile.TemporaryDirectory() as tmp:
             seed_dir = os.path.join(tmp, "seed")
             write_seed(seed_dir, "vm1", "PRIV", "ssh-ed25519 AAAAx",
-                       secret_env={"GITEA_TOKEN": "s3cr3t v"})
+                       secret_env={"SUB2API_API_KEY": "gw-key"},
+                       authd_env={"GITEA_TOKEN": "s3cr3t v"})
             with open(os.path.join(seed_dir, "user-data")) as fh:
                 content = fh.read()
         self.assertEqual(content.count("write_files:"), 1)
         parsed = yaml.safe_load(content)
         self.assertNotIn("runcmd", parsed)
-        self.assertEqual(len(parsed["write_files"]), 3)
+        self.assertEqual(len(parsed["write_files"]), 4)
         paths = [e["path"] for e in parsed["write_files"]]
         self.assertIn("/etc/chv/guest-env", paths)
-        # 密钥值经 shlex.quote 引用后才进文件（值本身在 guest-env
-        # 条目内是设计内的，见 hostkeys.write_seed）。
-        self.assertIn("GITEA_TOKEN='s3cr3t v'", content)
+        self.assertIn("/etc/chv/authd-seed", paths)
+        # 密钥值经 shlex.quote 引用后才进文件（值本身在条目内
+        # 是设计内的，见 hostkeys.write_seed）；GITEA_TOKEN
+        # 只在 authd-seed 里，不在 guest-env 里。
+        entries = {e["path"]: e for e in parsed["write_files"]}
+        authd_text = entries["/etc/chv/authd-seed"]["content"]
+        self.assertIn("GITEA_TOKEN=", authd_text)
+        self.assertIn("s3cr3t v", authd_text)
+        self.assertEqual(
+            entries["/etc/chv/authd-seed"]["permissions"], "0600")
+        self.assertNotIn(
+            "GITEA_TOKEN", entries["/etc/chv/guest-env"]["content"])
 
     def test_write_seed_uses_atomic_write(self):
         # feat-chv-cover100：seed 暂存文件（user-data 含 host

cloud/tests/test_chv_gitea_sync.py

@@ -10,7 +10,8 @@ VM、不联网、不含任何真实 token（测试值全是假值）。
   5 键进 guest-env 集；
 - hostkeys.write_seed：guest_env 与 secret_env 合并写同一
   条目、仍无 runcmd、0600；
-- 执行器 wake_vm：完整同步变量进 seed user-data；缺失必填
+- 执行器 wake_vm：完整同步变量进 seed（5 键 guest-env +
+  GITEA_TOKEN 走 authd-seed）；缺失必填
   键时起 VMM 前 ChvFailure 且不启动；无同步变量时正常启动
   且不写 guest-env；seed 投递后 build_ssh_command 才能用
   GITEA_TOKEN 且 argv 里绝无 token 值；
@@ -117,6 +118,13 @@ def _guest_env_lines(user_text):
     assert len(guest) == 1, entries
     return guest[0]["content"], guest[0]["permissions"]
 
+def _authd_seed_lines(user_text):
+    parsed = yaml.safe_load(user_text)
+    entries = parsed["write_files"]
+    authd = [e for e in entries if e["path"] == "/etc/chv/authd-seed"]
+    assert len(authd) == 1, entries
+    return authd[0]["content"], authd[0]["permissions"]
+
 
 # -- (1) gitea_sync 纯函数 ---------------------------------------------
 class GiteaSyncPureTest(unittest.TestCase):
@@ -226,27 +234,39 @@ class SshTransportSyncTest(unittest.TestCase):
 # -- (3) hostkeys.write_seed -------------------------------------------
 class WriteSeedGuestEnvTest(unittest.TestCase):
     def test_guest_env_and_secret_merge_into_one_entry_no_runcmd(self):
+        # feat-chv-muse-p3：GITEA_TOKEN 不再进 guest-env，改走
+        # authd_env（独立的 authd-seed 条目，0600）；guest-env
+        # 条目里只剩非 authd 密钥与同步普通变量。guest_env 里
+        # 直接给 GITEA_TOKEN 即 ValueError（调用方须先
+        # split_authd_env 拆出来，旧的“同名键 secret 优先”语义
+        # 已由“authd 通道只装 authd 密钥”取代）。
         with tempfile.TemporaryDirectory() as tmp:
             seed_dir = os.path.join(tmp, "seed")
             write_seed(seed_dir, "vm1", "PRIV", "ssh-ed25519 AAAAx",
                        secret_env={"SUB2API_API_KEY": "gw-key"},
                        guest_env={"GITEA_URL": "https://gitea.tose.top",
                                   "GITEA_USER": "admin",
-                                  "SYNC_REPO": "admin/chv-sync-test",
-                                  "GITEA_TOKEN": "dup-should-lose"})
+                                  "SYNC_REPO": "admin/chv-sync-test"},
+                       authd_env={"GITEA_TOKEN": "dup-should-lose"})
             with open(os.path.join(seed_dir, "user-data")) as fh:
                 content = fh.read()
         self.assertEqual(content.count("write_files:"), 1)
         parsed = yaml.safe_load(content)
         self.assertNotIn("runcmd", parsed)
-        self.assertEqual(len(parsed["write_files"]), 3)
+        self.assertEqual(len(parsed["write_files"]), 4)
         text, perms = _guest_env_lines(content)
         self.assertEqual(perms, "0600")
         self.assertIn("GITEA_URL=https://gitea.tose.top", text)
         self.assertIn("SUB2API_API_KEY=gw-key", text)
-        # 同名键以 secret_env 为准，不出现两行 GITEA_TOKEN。
-        self.assertEqual(text.count("GITEA_TOKEN="), 1)
-        self.assertIn("GITEA_TOKEN=dup-should-lose", text)
+        self.assertNotIn("GITEA_TOKEN", text)
+        authd_text, authd_perms = _authd_seed_lines(content)
+        self.assertEqual(authd_perms, "0600")
+        self.assertIn("GITEA_TOKEN=dup-should-lose", authd_text)
+        with tempfile.TemporaryDirectory() as tmp:
+            seed_dir = os.path.join(tmp, "seed")
+            with self.assertRaises(ValueError):
+                write_seed(seed_dir, "vm1", "PRIV", "ssh-ed25519 AAAAx",
+                           guest_env={"GITEA_TOKEN": "dup-should-lose"})
 
     def test_guest_env_only_still_no_runcmd(self):
         with tempfile.TemporaryDirectory() as tmp:
@@ -307,7 +327,10 @@ class WakeGuestEnvCase(unittest.TestCase):
     def make_exec(self, env_extra=None):
         env = {"CHV_EGRESS_PROXY": "http://10.0.0.1:8888",
                "CHV_HOST_KEY_MODE": "provisioned",
-               "CHV_SSH_FORWARD": "1"}
+               "CHV_SSH_FORWARD": "1",
+               # feat-chv-muse-p3：wake 前 Sentinel 预审要求
+               # Gitea 主机在白名单内；解析器用假的（测试不联网）。
+               "CHV_SENTINEL_ALLOW_HOSTS": "gitea.tose.top"}
         if env_extra:
             env.update(env_extra)
         ex = CloudHypervisorExecutor(
@@ -317,7 +340,8 @@ class WakeGuestEnvCase(unittest.TestCase):
             cpuinfo_path=self.cpuinfo, state_dir=self.state_dir,
             runner=self._runner, launcher=self._launcher,
             api_client=lambda s, m, p, payload=None: (200, {}),
-            sleeper=lambda s: None, env=env)
+            sleeper=lambda s: None, env=env,
+            sentinel_resolver=lambda host, port: ["93.184.216.34"])
         ex.audit_callback = lambda e, t, d, k=None: self.audit.append(
             (e, t.get("id") if isinstance(t, dict) else t, d, k))
         return ex
@@ -331,8 +355,15 @@ class WakeGuestEnvCase(unittest.TestCase):
         self.assertNotIn("runcmd", yaml.safe_load(text))
         guest_text, perms = _guest_env_lines(text)
         self.assertEqual(perms, "0600")
-        for key, value in FAKE_SYNC.items():
-            self.assertIn("%s=%s" % (key, value), guest_text)
+        # feat-chv-muse-p3：六键仍全进 seed，但分两路——
+        # 5 个普通键进 guest-env，GITEA_TOKEN 只进 authd-seed。
+        for key in ("GITEA_URL", "GITEA_USER", "SYNC_REPO",
+                    "SYNC_BRANCH", "GITEA_EMAIL"):
+            self.assertIn("%s=%s" % (key, FAKE_SYNC[key]), guest_text)
+        self.assertNotIn("GITEA_TOKEN", guest_text)
+        authd_text, authd_perms = _authd_seed_lines(text)
+        self.assertEqual(authd_perms, "0600")
+        self.assertIn("GITEA_TOKEN=fake-token-123", authd_text)
         # seed.iso 已制盘且 0600。
         seed_iso = os.path.join(self.state_dir, "chv-sync1", "seed.iso")
         self.assertTrue(os.path.exists(seed_iso))
@@ -351,8 +382,14 @@ class WakeGuestEnvCase(unittest.TestCase):
         self.assertEqual(out["state"], "running")
         guest_text, _ = _guest_env_lines(
             _seed_user_text(self.state_dir, "chv-sync2"))
-        self.assertIn("GITEA_TOKEN=fake-token-123", guest_text)
+        # feat-chv-muse-p3：GITEA_TOKEN 不再进 guest-env，
+        # 改走 authd-seed；其余同步变量仍在 guest-env 里。
+        self.assertNotIn("GITEA_TOKEN", guest_text)
         self.assertIn("SYNC_BRANCH=main", guest_text)
+        authd_text, authd_perms = _authd_seed_lines(
+            _seed_user_text(self.state_dir, "chv-sync2"))
+        self.assertEqual(authd_perms, "0600")
+        self.assertIn("GITEA_TOKEN=fake-token-123", authd_text)
 
     def test_wake_partial_sync_fails_closed_before_launch(self):
         ex = self.make_exec()
@@ -387,11 +424,24 @@ class WakeGuestEnvCase(unittest.TestCase):
     def test_build_ssh_command_after_wake_sources_guest_env(self):
         ex = self.make_exec()
         ex.wake_vm({"id": "sync5", "gitea_sync": dict(FAKE_SYNC)})
+        # feat-chv-muse-p3：GITEA_TOKEN 走 authd 投递——任务期
+        # 认“已投递、不报错”，但绝不 source 进 codex 进程环境，
+        # 所以命令里既无 token 值、也不 source 任何客内密钥文件。
         cmd = ex.build_ssh_command(
             {"id": "sync5"}, ["true"],
             extra_env={"GITEA_TOKEN": "fake-token-123"})
         joined = " ".join(cmd)
         self.assertNotIn("fake-token-123", joined)
+        self.assertNotIn("/etc/chv/guest-env", joined)
+        self.assertNotIn("/etc/chv/authd-seed", joined)
+        # 非 authd 密钥仍走 guest-env：wake 时投递，任务期 source。
+        ex.wake_vm({"id": "sync5b"},
+                   extra_env={"SUB2API_API_KEY": "gw-key"})
+        cmd = ex.build_ssh_command(
+            {"id": "sync5b"}, ["true"],
+            extra_env={"SUB2API_API_KEY": "gw-key"})
+        joined = " ".join(cmd)
+        self.assertNotIn("gw-key", joined)
         self.assertIn("/etc/chv/guest-env", joined)
         # 未 wake 投递的 VM 用密钥即拒绝。
         with self.assertRaises(ChvFailure) as ctx:

cloud/tests/test_chv_muse_p3.py

@@ -0,0 +1,675 @@
+"""对齐 Muse P3 测试：authd 替身凭证 + Sentinel 式出口审批。
+
+只用标准库与 mock，绝不启动真实 VM、不联网、不含真实 token
+（测试值全是假值）。覆盖：
+- authd 保险库：白名单入库、作用域绑定、单次释放、替身过期、
+  forget 吊销、repr/审计里绝无 token 值；
+- 铸 token / 吊销（假 HTTP 客户端）；
+- Sentinel：SSRF 挡截、deny/ask 路径、L7 路径规则、严格能力
+  （绑定/过期/次数）、待审批队列；
+- seed 拆分：split_authd_env、write_seed 拒收 guest-env 里的
+  GITEA_TOKEN、authd-seed 0600；
+- wake 预审：allow/ask/deny、无同步跳过；
+- 客内 authd 守护进程（真 socket + 真 seed）：放凭证、拒绝
+  路径/主机/仓库、审计无值；
+- git credential helper 端到端：真 git credential fill + 假
+  authd socket（守护进程真跑在测试进程内）。
+"""
+
+import importlib.util
+import json
+import os
+import socket
+import subprocess
+import sys
+import tempfile
+import threading
+import unittest
+from pathlib import Path
+
+try:
+    from cloud.chv.authd import (
+        AuthdRefusal,
+        CredentialVault,
+        mint_gitea_token,
+        revoke_gitea_token,
+        surrogate_fingerprint,
+    )
+    from cloud.chv.gitea_sync import split_authd_env
+    from cloud.chv.hostkeys import AUTHD_SEED_PATH, write_seed
+    from cloud.chv.sentinel import (
+        ApprovalQueue,
+        EgressRequest,
+        SentinelPolicy,
+    )
+    from cloud.sandbox import ChvFailure, CloudHypervisorExecutor
+except ImportError:
+    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+    from cloud.chv.authd import (
+        AuthdRefusal,
+        CredentialVault,
+        mint_gitea_token,
+        revoke_gitea_token,
+        surrogate_fingerprint,
+    )
+    from cloud.chv.gitea_sync import split_authd_env
+    from cloud.chv.hostkeys import AUTHD_SEED_PATH, write_seed
+    from cloud.chv.sentinel import (
+        ApprovalQueue,
+        EgressRequest,
+        SentinelPolicy,
+    )
+    from cloud.sandbox import ChvFailure, CloudHypervisorExecutor
+
+REPO_ROOT = Path(__file__).resolve().parents[2]
+TOKEN = "fake-token-123"
+PUBLIC_IP = "93.184.216.34"
+FAKE_SYNC = {
+    "GITEA_URL": "https://gitea.tose.top",
+    "GITEA_USER": "admin",
+    "GITEA_TOKEN": TOKEN,
+    "SYNC_REPO": "admin/chv-sync-test",
+    "SYNC_BRANCH": "main",
+    "GITEA_EMAIL": "admin@tose.top",
+}
+
+
+def _public_resolver(host, port):
+    return [PUBLIC_IP]
+
+
+def _load_guest_authd():
+    """chv-guest/authd.py（守护进程）按文件加载，不走包名。"""
+    path = REPO_ROOT / "chv-guest" / "authd.py"
+    spec = importlib.util.spec_from_file_location(
+        "chv_guest_authd", path)
+    mod = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(mod)
+    return mod
+
+
+# -- (1) authd 保险库 ----------------------------------------------------
+class VaultTest(unittest.TestCase):
+    def setUp(self):
+        self.now = [1000.0]
+        self.vault = CredentialVault(clock=lambda: self.now[0])
+
+    def _stored(self):
+        self.vault.store("gitea", "GITEA_TOKEN", TOKEN,
+                         host="gitea.tose.top",
+                         repo="admin/chv-sync-test")
+
+    def test_store_rejects_unknown_connector_key_empty(self):
+        with self.assertRaises(AuthdRefusal) as ctx:
+            self.vault.store("github", "GITEA_TOKEN", TOKEN)
+        self.assertEqual(ctx.exception.kind, "unknown_connector")
+        with self.assertRaises(AuthdRefusal) as ctx:
+            self.vault.store("gitea", "OTHER_KEY", TOKEN)
+        self.assertEqual(ctx.exception.kind,
+                         "credential_not_allowlisted")
+        with self.assertRaises(AuthdRefusal) as ctx:
+            self.vault.store("gitea", "GITEA_TOKEN", "   ")
+        self.assertEqual(ctx.exception.kind, "empty_credential")
+        self.assertFalse(self.vault.has("gitea", "GITEA_TOKEN"))
+
+    def test_release_single_use_and_scope_binding(self):
+        self._stored()
+        rel = self.vault.release("gitea", "GITEA_TOKEN",
+                                 "gitea.tose.top",
+                                 "admin/chv-sync-test", "push")
+        self.assertEqual(self.vault.consume(rel.release_id), TOKEN)
+        with self.assertRaises(AuthdRefusal) as ctx:
+            self.vault.consume(rel.release_id)
+        self.assertEqual(ctx.exception.kind, "release_unknown")
+        with self.assertRaises(AuthdRefusal) as ctx:
+            self.vault.release("gitea", "GITEA_TOKEN",
+                               "other.example",
+                               "admin/chv-sync-test", "push")
+        self.assertEqual(ctx.exception.kind, "scope_mismatch")
+        with self.assertRaises(AuthdRefusal) as ctx:
+            self.vault.release("gitea", "GITEA_TOKEN",
+                               "gitea.tose.top",
+                               "admin/other-repo", "push")
+        self.assertEqual(ctx.exception.kind, "scope_mismatch")
+        with self.assertRaises(AuthdRefusal) as ctx:
+            self.vault.release("gitea", "GITEA_TOKEN",
+                               "gitea.tose.top",
+                               "admin/chv-sync-test", "delete")
+        self.assertEqual(ctx.exception.kind, "operation_not_allowed")
+
+    def test_release_expires_after_ttl(self):
+        # 替身过期：300 秒 TTL，301 秒后兑现即废。
+        self._stored()
+        rel = self.vault.release("gitea", "GITEA_TOKEN",
+                                 "gitea.tose.top",
+                                 "admin/chv-sync-test", "pull")
+        self.now[0] += 301
+        with self.assertRaises(AuthdRefusal) as ctx:
+            self.vault.consume(rel.release_id)
+        self.assertEqual(ctx.exception.kind, "release_expired")
+
+    def test_release_unexpired_still_works(self):
+        self._stored()
+        rel = self.vault.release("gitea", "GITEA_TOKEN",
+                                 "gitea.tose.top",
+                                 "admin/chv-sync-test", "clone")
+        self.now[0] += 299
+        self.assertEqual(self.vault.consume(rel.release_id), TOKEN)
+
+    def test_forget_revokes_pending_releases(self):
+        self._stored()
+        rel = self.vault.release("gitea", "GITEA_TOKEN",
+                                 "gitea.tose.top",
+                                 "admin/chv-sync-test", "pull")
+        self.assertTrue(self.vault.forget("gitea", "GITEA_TOKEN"))
+        self.assertFalse(self.vault.has("gitea", "GITEA_TOKEN"))
+        self.assertFalse(self.vault.forget("gitea", "GITEA_TOKEN"))
+        with self.assertRaises(AuthdRefusal) as ctx:
+            self.vault.consume(rel.release_id)
+        self.assertEqual(ctx.exception.kind, "release_unknown")
+        with self.assertRaises(AuthdRefusal) as ctx:
+            self.vault.release("gitea", "GITEA_TOKEN",
+                               "gitea.tose.top",
+                               "admin/chv-sync-test", "pull")
+        self.assertEqual(ctx.exception.kind, "credential_not_stored")
+
+    def test_repr_and_audit_never_contain_value(self):
+        # 真 token 泄露面：repr/审计记录里绝无 token 值。
+        self._stored()
+        rel = self.vault.release("gitea", "GITEA_TOKEN",
+                                 "gitea.tose.top",
+                                 "admin/chv-sync-test", "push")
+        for text in (repr(self.vault), repr(rel),
+                     json.dumps(self.vault.audit_release(rel)),
+                     json.dumps(self.vault.audit_refusal(
+                         "scope_mismatch", "gitea", "GITEA_TOKEN"))):
+            self.assertNotIn(TOKEN, text)
+        self.assertNotEqual(surrogate_fingerprint(TOKEN), TOKEN)
+        self.assertEqual(len(surrogate_fingerprint(TOKEN)), 8)
+        self.assertEqual(surrogate_fingerprint(TOKEN),
+                         surrogate_fingerprint(TOKEN))
+
+
+class MintRevokeTest(unittest.TestCase):
+    def test_mint_returns_new_token_and_failure_is_refusal(self):
+        def fake_post(url, headers, payload):
+            self.assertIn("/users/admin/tokens", url)
+            self.assertEqual(payload["scopes"], ["write:repository"])
+            return 201, {"sha1": "new-short-token"}
+        self.assertEqual(
+            mint_gitea_token("https://gitea.tose.top/api/v1", "admin",
+                             TOKEN, "chv-test", post=fake_post),
+            "new-short-token")
+        with self.assertRaises(AuthdRefusal) as ctx:
+            mint_gitea_token("https://gitea.tose.top/api/v1", "admin",
+                             TOKEN, "chv-test",
+                             post=lambda u, h, p: (403, {}))
+        self.assertEqual(ctx.exception.kind, "mint_failed")
+        with self.assertRaises(AuthdRefusal):
+            mint_gitea_token("", "admin", TOKEN, "x")
+
+    def test_revoke_statuses(self):
+        base = ("https://gitea.tose.top/api/v1", "admin", TOKEN)
+        self.assertTrue(revoke_gitea_token(
+            *base, 49, delete=lambda u, h: 204))
+        self.assertTrue(revoke_gitea_token(
+            *base, 49, delete=lambda u, h: 404))
+        self.assertFalse(revoke_gitea_token(
+            *base, 49, delete=lambda u, h: 500))
+        self.assertFalse(revoke_gitea_token("", "admin", TOKEN, 49))
+
+
+# -- (2) Sentinel 决策 ----------------------------------------------------
+class SentinelTest(unittest.TestCase):
+    def _req(self, host="gitea.tose.top", method="GET",
+             scheme="https", port=443, path="/api/v1/version"):
+        return EgressRequest("gitea", method, scheme, host, port, path)
+
+    def test_allow_public_ip_allowlisted_host(self):
+        policy = SentinelPolicy(allow_hosts={"gitea.tose.top"})
+        dec = policy.evaluate(self._req(), resolve=_public_resolver)
+        self.assertEqual(dec.verdict, "allow")
+        self.assertIsNotNone(dec.capability)
+        self.assertNotIn(TOKEN, json.dumps(dec.audit(self._req())))
+
+    def test_ssrf_private_ip_denied_even_when_allowlisted(self):
+        # SSRF 挡截：白名单主机解析到私网 IP 仍 deny。
+        policy = SentinelPolicy(allow_hosts={"gitea.tose.top"})
+        for ip in ("10.0.0.5", "192.168.1.1", "127.0.0.1",
+                   "169.254.1.2", "100.64.0.9", "::1", "0.0.0.0"):
+            dec = policy.evaluate(
+                self._req(), resolve=lambda h, p, ip=ip: [ip])
+            self.assertEqual(dec.verdict, "deny", ip)
+            self.assertIn("ssrf_private_ip", dec.reasons[0])
+
+    def test_dns_failure_is_deny_not_ask(self):
+        policy = SentinelPolicy(allow_hosts={"gitea.tose.top"})
+        dec = policy.evaluate(
+            self._req(), resolve=lambda h, p: (_ for _ in ()).throw(
+                OSError("boom")))
+        self.assertEqual(dec.verdict, "deny")
+        self.assertEqual(dec.reasons, ["dns_resolution_failed"])
+        dec = policy.evaluate(self._req(), resolve=lambda h, p: [])
+        self.assertEqual(dec.verdict, "deny")
+
+    def test_unknown_host_is_ask_not_allow(self):
+        policy = SentinelPolicy(allow_hosts={"gitea.tose.top"})
+        dec = policy.evaluate(self._req(host="other.example"),
+                              resolve=_public_resolver)
+        self.assertEqual(dec.verdict, "ask")
+        self.assertEqual(dec.reasons, ["host_not_allowlisted"])
+        self.assertIsNone(dec.capability)
+
+    def test_deny_paths(self):
+        policy = SentinelPolicy(allow_hosts={"gitea.tose.top"},
+                                deny_hosts={"bad.example"})
+        cases = [
+            (EgressRequest("github", "GET", "https", "gitea.tose.top",
+                           443, "/api/v1/version"),
+             "unknown_connector"),
+            (self._req(scheme="http"), "scheme_not_allowed"),
+            (self._req(port=80), "port_not_allowed"),
+            (self._req(host="bad.example"), "host_denied"),
+            (self._req(path="/login"), "path_not_allowed"),
+        ]
+        for req, reason in cases:
+            dec = policy.evaluate(req, resolve=_public_resolver)
+            self.assertEqual(dec.verdict, "deny", reason)
+            self.assertEqual(dec.reasons, [reason])
+        # git 智能 HTTP 三条路径放行。
+        for method, path in (
+                ("GET", "/admin/r.git/info/refs"),
+                ("POST", "/admin/r.git/git-upload-pack"),
+                ("POST", "/admin/r.git/git-receive-pack")):
+            dec = policy.evaluate(
+                self._req(method=method, path=path),
+                resolve=_public_resolver)
+            self.assertEqual(dec.verdict, "allow", path)
+
+    def test_capability_binding_expiry_and_uses(self):
+        policy = SentinelPolicy(allow_hosts={"gitea.tose.top"})
+        dec = policy.evaluate(self._req(), resolve=_public_resolver,
+                              now=1000.0)
+        cap = dec.capability
+        req = self._req()
+        self.assertTrue(cap.matches(req, now=1000.0))
+        self.assertFalse(cap.matches(self._req(host="other.example"),
+                                     now=1000.0))
+        self.assertFalse(cap.matches(self._req(method="POST"),
+                                     now=1000.0))
+        self.assertFalse(cap.matches(self._req(path="/login"),
+                                     now=1000.0))
+        self.assertFalse(cap.matches(req, now=1000.0 + 301))
+        self.assertTrue(cap.use(req, now=1000.0))
+        self.assertFalse(cap.use(req, now=1000.0))
+
+    def test_from_env_and_approval_queue(self):
+        policy = SentinelPolicy.from_env(
+            {"CHV_SENTINEL_ALLOW_HOSTS": "a.example, b.example "},
+            extra_allow_hosts={"c.example"})
+        self.assertEqual(policy.allow_hosts,
+                         {"a.example", "b.example", "c.example"})
+        queue = ApprovalQueue(clock=lambda: 1000.0)
+        req = self._req(host="other.example")
+        pid = queue.record(req, ["host_not_allowlisted"])
+        pending = queue.list_pending()
+        self.assertEqual(len(pending), 1)
+        self.assertEqual(pending[0]["pending_id"], pid)
+        self.assertEqual(pending[0]["ts"], 1000.0)
+        self.assertTrue(queue.resolve(pid, "deny"))
+        self.assertFalse(queue.resolve(pid, "deny"))
+        self.assertFalse(queue.resolve("nope", "allow"))
+        self.assertEqual(queue.list_pending(), [])
+
+
+# -- (3) seed 拆分 ---------------------------------------------------------
+class SeedSplitTest(unittest.TestCase):
+    def test_split_authd_env(self):
+        authd, rest = split_authd_env(
+            {"GITEA_TOKEN": TOKEN, "SUB2API_API_KEY": "gw",
+             "OTHER": "1"})
+        self.assertEqual(authd, {"GITEA_TOKEN": TOKEN})
+        self.assertEqual(rest, {"SUB2API_API_KEY": "gw", "OTHER": "1"})
+        self.assertEqual(split_authd_env(None), ({}, {}))
+        self.assertEqual(split_authd_env({}), ({}, {}))
+
+    def test_write_seed_refuses_token_in_guest_env(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            seed_dir = os.path.join(tmp, "seed")
+            with self.assertRaises(ValueError):
+                write_seed(seed_dir, "vm1", "PRIV",
+                           "ssh-ed25519 AAAAx",
+                           secret_env={"GITEA_TOKEN": TOKEN})
+            with self.assertRaises(ValueError):
+                write_seed(seed_dir, "vm1", "PRIV",
+                           "ssh-ed25519 AAAAx",
+                           guest_env={"GITEA_TOKEN": TOKEN})
+            with self.assertRaises(ValueError):
+                write_seed(seed_dir, "vm1", "PRIV",
+                           "ssh-ed25519 AAAAx",
+                           authd_env={"SUB2API_API_KEY": "gw"})
+
+    def test_write_seed_authd_entry_is_separate_0600(self):
+        import yaml
+        with tempfile.TemporaryDirectory() as tmp:
+            seed_dir = os.path.join(tmp, "seed")
+            write_seed(seed_dir, "vm1", "PRIV", "ssh-ed25519 AAAAx",
+                       secret_env={"SUB2API_API_KEY": "gw-key"},
+                       guest_env={"GITEA_URL": "https://gitea.tose.top"},
+                       authd_env={"GITEA_TOKEN": TOKEN})
+            with open(os.path.join(seed_dir, "user-data")) as fh:
+                parsed = yaml.safe_load(fh.read())
+        entries = {e["path"]: e for e in parsed["write_files"]}
+        self.assertEqual(entries[AUTHD_SEED_PATH]["permissions"],
+                         "0600")
+        self.assertIn("GITEA_TOKEN=" + TOKEN,
+                      entries[AUTHD_SEED_PATH]["content"])
+        self.assertNotIn("GITEA_TOKEN",
+                         entries["/etc/chv/guest-env"]["content"])
+
+
+# -- (4) wake 预审 -----------------------------------------------------------
+class _FakeResult:
+    def __init__(self, returncode=0, stdout="", stderr=""):
+        self.returncode = returncode
+        self.stdout = stdout
+        self.stderr = stderr
+
+
+class _FakeProc:
+    def __init__(self, alive=True):
+        self._alive = alive
+
+    def poll(self):
+        return None if self._alive else 1
+
+    def terminate(self):
+        self._alive = False
+
+    def wait(self, timeout=None):
+        return 0
+
+    def kill(self):
+        self._alive = False
+
+
+def _write_file(path, content=b"x"):
+    with open(path, "wb") as fh:
+        fh.write(content)
+    os.chmod(path, 0o755)
+    return path
+
+
+class WakeGateTest(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.dir = self.tmp.name
+        self.binary = _write_file(
+            os.path.join(self.dir, "cloud-hypervisor"))
+        self.ch_remote = _write_file(os.path.join(self.dir, "ch-remote"))
+        self.kernel = _write_file(os.path.join(self.dir, "vmlinux"))
+        self.rootfs = _write_file(os.path.join(self.dir, "disk.raw"),
+                                  b"R" * 100)
+        self.ssh_key = _write_file(os.path.join(self.dir, "id_chv"))
+        self.kvm = _write_file(os.path.join(self.dir, "kvm"))
+        self.cpuinfo = os.path.join(self.dir, "cpuinfo")
+        with open(self.cpuinfo, "w") as fh:
+            fh.write("flags : fpu vmx sse\n")
+        self.state_dir = os.path.join(self.dir, "state")
+        self.audit = []
+        self.launches = []
+
+    def tearDown(self):
+        self.tmp.cleanup()
+
+    def _runner(self, cmd, timeout=None):
+        if cmd and cmd[0] == "ssh-keygen":
+            priv = cmd[cmd.index("-f") + 1]
+            with open(priv, "w") as fh:
+                fh.write("-----BEGIN OPENSSH PRIVATE KEY-----\nFAKE\n")
+            os.chmod(priv, 0o600)
+            with open(priv + ".pub", "w") as fh:
+                fh.write("ssh-ed25519 AAAAfake1 vm-key\n")
+            return _FakeResult(0, "", "")
+        if cmd and cmd[0] == "genisoimage":
+            out = cmd[cmd.index("-output") + 1]
+            with open(out, "wb") as fh:
+                fh.write(b"ISO")
+            return _FakeResult(0, "", "")
+        return _FakeResult(0, "", "")
+
+    def _make_exec(self, env_extra=None, resolver=None):
+        env = {"CHV_EGRESS_PROXY": "http://10.0.0.1:8888",
+               "CHV_HOST_KEY_MODE": "provisioned",
+               "CHV_SSH_FORWARD": "1"}
+        if env_extra:
+            env.update(env_extra)
+        ex = CloudHypervisorExecutor(
+            binary=self.binary, ch_remote=self.ch_remote,
+            kernel=self.kernel, rootfs=self.rootfs,
+            ssh_key=self.ssh_key, kvm_path=self.kvm,
+            cpuinfo_path=self.cpuinfo, state_dir=self.state_dir,
+            runner=self._runner,
+            launcher=lambda cmd: self.launches.append(list(cmd))
+            or _FakeProc(alive=True),
+            api_client=lambda s, m, p, payload=None: (200, {}),
+            sleeper=lambda s: None, env=env,
+            sentinel_resolver=resolver)
+        ex.audit_callback = lambda e, t, d, k=None: self.audit.append(
+            (e, t.get("id") if isinstance(t, dict) else t, d, k))
+        return ex
+
+    def test_wake_allowlisted_gitea_passes_gate(self):
+        ex = self._make_exec(
+            env_extra={"CHV_SENTINEL_ALLOW_HOSTS": "gitea.tose.top"},
+            resolver=_public_resolver)
+        out = ex.wake_vm({"id": "gate1", "gitea_sync": dict(FAKE_SYNC)})
+        self.assertEqual(out["state"], "running")
+        kinds = [a[0] for a in self.audit]
+        self.assertIn("sentinel_egress_allow", kinds)
+
+    def test_wake_unknown_gitea_host_is_ask_and_fail_closed(self):
+        ex = self._make_exec(resolver=_public_resolver)
+        sync = dict(FAKE_SYNC)
+        sync["GITEA_URL"] = "https://other.example"
+        with self.assertRaises(ChvFailure) as ctx:
+            ex.wake_vm({"id": "gate2", "gitea_sync": sync})
+        self.assertEqual(ctx.exception.kind, "sentinel_egress_denied")
+        self.assertEqual(self.launches, [])
+        kinds = [a[0] for a in self.audit]
+        self.assertIn("sentinel_egress_ask", kinds)
+        pending = ex._sentinel_queue.list_pending()
+        self.assertEqual(len(pending), 1)
+        self.assertEqual(pending[0]["host"], "other.example")
+
+    def test_wake_gitea_resolving_to_private_ip_is_deny(self):
+        ex = self._make_exec(
+            env_extra={"CHV_SENTINEL_ALLOW_HOSTS": "gitea.tose.top"},
+            resolver=lambda h, p: ["10.0.0.5"])
+        with self.assertRaises(ChvFailure) as ctx:
+            ex.wake_vm({"id": "gate3", "gitea_sync": dict(FAKE_SYNC)})
+        self.assertEqual(ctx.exception.kind, "sentinel_egress_denied")
+        kinds = [a[0] for a in self.audit]
+        self.assertIn("sentinel_egress_denied", kinds)
+        self.assertEqual(ex._sentinel_queue.list_pending(), [])
+
+    def test_wake_without_sync_skips_gate(self):
+        ex = self._make_exec(resolver=_public_resolver)
+        out = ex.wake_vm({"id": "gate4"})
+        self.assertEqual(out["state"], "running")
+        kinds = [a[0] for a in self.audit]
+        self.assertNotIn("sentinel_egress_allow", kinds)
+        self.assertNotIn("sentinel_egress_denied", kinds)
+        self.assertNotIn("sentinel_egress_ask", kinds)
+
+
+# -- (5) 客内 authd 守护进程 + helper 端到端 -------------------------------
+class GuestAuthdTest(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.dir = self.tmp.name
+        self.seed_path = os.path.join(self.dir, "authd-seed")
+        with open(self.seed_path, "w") as fh:
+            fh.write("GITEA_URL=https://gitea.tose.top\n"
+                     "GITEA_USER=admin\n"
+                     "GITEA_TOKEN=" + TOKEN + "\n"
+                     "SYNC_REPO=admin/chv-sync-test\n")
+        self.socket_path = os.path.join(self.dir, "authd.sock")
+        self.log_path = os.path.join(self.dir, "authd.log")
+        mod = _load_guest_authd()
+        self.server = mod.build_server(
+            seed_path=self.seed_path, socket_path=self.socket_path,
+            log_path=self.log_path, lib_dir=str(REPO_ROOT),
+            allowed_uids=[os.getuid()], resolve=_public_resolver)
+        self.thread = threading.Thread(
+            target=self.server.serve_forever, daemon=True)
+        self.thread.start()
+
+    def tearDown(self):
+        self.server.shutdown()
+        self.server.server_close()
+        self.tmp.cleanup()
+
+    def _ask(self, req):
+        sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+        sock.settimeout(5)
+        try:
+            sock.connect(self.socket_path)
+            sock.sendall((json.dumps(req) + "\n").encode("utf-8"))
+            buf = b""
+            while b"\n" not in buf:
+                chunk = sock.recv(4096)
+                if not chunk:
+                    break
+                buf += chunk
+            return json.loads(buf.decode("utf-8"))
+        finally:
+            sock.close()
+
+    def test_release_for_clone_path(self):
+        resp = self._ask({"connector": "gitea",
+                          "host": "gitea.tose.top",
+                          "path": "/admin/chv-sync-test.git/info/refs"})
+        self.assertTrue(resp["ok"])
+        self.assertEqual(resp["username"], "admin")
+        self.assertEqual(resp["password"], TOKEN)
+        self.assertGreater(resp["expires_at"], 0)
+
+    def test_second_request_gets_fresh_release(self):
+        for _ in range(2):
+            resp = self._ask({"connector": "gitea",
+                              "host": "gitea.tose.top",
+                              "path": "/admin/chv-sync-test.git/"
+                                      "git-upload-pack"})
+            self.assertTrue(resp["ok"])
+            self.assertEqual(resp["password"], TOKEN)
+
+    def test_refused_paths_hosts_and_repos(self):
+        cases = [
+            ({"connector": "github", "host": "gitea.tose.top",
+              "path": "/admin/chv-sync-test.git/info/refs"},
+             "bad_request_fields"),
+            ({"connector": "gitea", "host": "gitea.tose.top",
+              "path": "/admin/chv-sync-test.git/foo"},
+             "path_not_allowed"),
+            ({"connector": "gitea", "host": "other.example",
+              "path": "/admin/chv-sync-test.git/info/refs"},
+             "sentinel_ask"),
+            ({"connector": "gitea", "host": "gitea.tose.top",
+              "path": "/foo.git/info/refs"}, "repo_unparseable"),
+        ]
+        for req, error in cases:
+            resp = self._ask(req)
+            self.assertFalse(resp["ok"], req)
+            self.assertEqual(resp["error"], error, req)
+
+    def test_audit_log_never_contains_value(self):
+        self._ask({"connector": "gitea", "host": "gitea.tose.top",
+                   "path": "/admin/chv-sync-test.git/info/refs"})
+        self._ask({"connector": "gitea", "host": "gitea.tose.top",
+                   "path": "/admin/chv-sync-test.git/foo"})
+        with open(self.log_path) as fh:
+            self.assertNotIn(TOKEN, fh.read())
+
+    def test_incomplete_or_missing_seed_fails_fast(self):
+        mod = _load_guest_authd()
+        bad_seed = os.path.join(self.dir, "bad-seed")
+        with open(bad_seed, "w") as fh:
+            fh.write("GITEA_URL=https://gitea.tose.top\n")
+        with self.assertRaises(ValueError):
+            mod.build_server(seed_path=bad_seed,
+                             socket_path=os.path.join(self.dir, "s2"),
+                             log_path=self.log_path,
+                             lib_dir=str(REPO_ROOT),
+                             allowed_uids=[os.getuid()],
+                             resolve=_public_resolver)
+        with self.assertRaises(FileNotFoundError):
+            mod.build_server(seed_path=os.path.join(self.dir, "nope"),
+                             socket_path=os.path.join(self.dir, "s3"),
+                             log_path=self.log_path,
+                             lib_dir=str(REPO_ROOT),
+                             allowed_uids=[os.getuid()],
+                             resolve=_public_resolver)
+
+
+class HelperE2ETest(unittest.TestCase):
+    """helper + 真 git credential fill（假 authd socket 在后台）。"""
+
+    HELPER = REPO_ROOT / "chv-guest" / "git-credential-chv-authd"
+
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.dir = self.tmp.name
+        self.guest = GuestAuthdTest("test_release_for_clone_path")
+        self.guest.setUp()
+        self.socket_path = self.guest.socket_path
+
+    def tearDown(self):
+        self.guest.tearDown()
+        self.tmp.cleanup()
+
+    def _helper_env(self):
+        env = dict(os.environ)
+        env["CHV_AUTHD_SOCKET"] = self.socket_path
+        return env
+
+    def test_helper_returns_credential(self):
+        proc = subprocess.run(
+            [sys.executable, str(self.HELPER)],
+            input="protocol=https\nhost=gitea.tose.top\n"
+                  "path=/admin/chv-sync-test.git/info/refs"
+                  "?service=git-upload-pack\n\n",
+            capture_output=True, text=True, env=self._helper_env(),
+            check=False)
+        self.assertEqual(proc.returncode, 0, proc.stderr)
+        self.assertIn("username=admin", proc.stdout)
+        self.assertIn("password=" + TOKEN, proc.stdout)
+        self.assertIn("password_expiry_utc=", proc.stdout)
+
+    def test_helper_refusal_is_exit_1(self):
+        proc = subprocess.run(
+            [sys.executable, str(self.HELPER)],
+            input="protocol=https\nhost=gitea.tose.top\n"
+                  "path=/admin/chv-sync-test.git/foo\n\n",
+            capture_output=True, text=True, env=self._helper_env(),
+            check=False)
+        self.assertEqual(proc.returncode, 1)
+        self.assertIn("path_not_allowed", proc.stderr)
+
+    def test_git_credential_fill_end_to_end(self):
+        env = self._helper_env()
+        env["GIT_CONFIG_NOSYSTEM"] = "1"
+        proc = subprocess.run(
+            ["git", "-c", "credential.helper=%s" % self.HELPER,
+             "-c", "credential.useHttpPath=true",
+             "credential", "fill"],
+            input="protocol=https\nhost=gitea.tose.top\n"
+                  "path=/admin/chv-sync-test.git/info/refs\n\n",
+            capture_output=True, text=True, env=env, cwd=self.dir,
+            check=False)
+        self.assertEqual(proc.returncode, 0, proc.stderr)
+        self.assertIn("password=" + TOKEN, proc.stdout)
+
+
+if __name__ == "__main__":
+    unittest.main()

docs/chv-muse-p3-gap.md

@@ -0,0 +1,118 @@
+# 对齐 Muse P3 差距表：authd 替身凭证 + Sentinel 式出口审批
+
+依据：`~/workspace/research_notes/meta-muse-ai-architecture-20261101-0755/report.md`
+（官方以 Meta security.muse.ai 2026-09-08 为准；第三方拆解只作参考，
+逆向细节不得写成官方架构）。Muse 模型一句话：凭证由 VM 内独立服务
+authd 保管、代理只见替身令牌；所有连接器动作与网络出口由 Sentinel
+唯一审批（allow/deny/ask，L4+L7 评估，SSRF 防护，严格能力绑定）；
+审批 UI 不在对话内；连接器执行走 privsep worker，按连接器做凭证白名单。
+
+本轨范围只限 P3：authd 式替身凭证 + Sentinel 式出口审批。
+不碰 P1（常驻 VM、WarmPool、气球、vsock）、不碰 P2（客机 nspawn 单元、
+浏览器子代理独立）。分支 `feat-chv-muse-p3`（从 `feat-chv-cover100`
+分出，基线 HEAD `06c3e17`）。
+
+## 1. 现状 vs Muse
+
+| 项 | 现状（实测/代码为准） | Muse 目标 | 本轨做法 |
+|---|---|---|---|
+| Gitea token 可见面 | 真 token 经 seed 写进 `/etc/chv/guest-env`（0600 root）；`build_ssh_command` 在每个任务的远程命令前 `source` 该文件，codex 进程环境里直接含真值；`chv-sync boot` 再把它写进 `/home/chv/.git-credentials` 落盘 | authd 保管凭证，代理只见替身令牌；凭证存入运行时单元之外的安全凭证存储，代理不可见；仅在需要时注入 | 真 token 改走 seed 里的独立文件 `/etc/chv/authd-seed`（0600 root），只给客内 authd 守护进程读；guest-env 不再含 GITEA_TOKEN；任务期 `build_ssh_command` 不再把 GITEA_TOKEN source 进 codex 环境；`chv-sync` 经 git credential helper 按次向 authd 取凭证，只进 git 子进程内存，不落盘、不进 agent 环境/argv/日志 |
+| authd 位置 | 无 | 存于用户 VM 内（非 Meta 集中基础设施），在运行时单元之外 | 客内 authd 守护进程（`chv-guest/authd.py`，标准库 Unix socket 服务），以 root 身份跑在单元外；宿主侧不存任何 VM 的 token。选客内不选宿主的理由：Muse 的 authd 就在用户 VM 里；宿主侧方案要走 vsock（P1 范围、未就绪），且会把全部 VM 的 token 集中到宿主，违背“存于用户 VM”的口径 |
+| 替身令牌 | 无（真值直给） | authd 签发替身令牌，带作用域与有效期 | 两档：① 本期默认——按次释放：每次 git 操作向 authd 取一次凭证，绑定 {连接器=gitea, 主机, 仓库路径, 操作}，单次有效 + 300 秒 TTL，过期即废，释放全程审计（审计里无值）；② 真铸 token：authd 可经 Gitea API 铸短期专用 token（`POST /api/v1/users/{user}/tokens`），默认关闭，见诚实边界第 2 条 |
+| 凭证白名单（privsep） | 无连接器概念；`_CHV_SECRET_ENV_NAMES` 只有“密钥/非密钥”二分 | 每个 privsep worker 有明确凭证白名单（日历 worker 不能仅改参数就索取邮件凭证） | authd 只认一个连接器 `gitea`，白名单只含 GITEA_TOKEN；未知连接器/越权主机/仓库/操作一律拒绝；SUB2API_API_KEY 仍走 guest-env（见诚实边界第 3 条） |
+| 出口审批 | 只有代码门：preflight 要求代理或显式直连 + `validate_egress_proxy` 格式校验；真正的隔离靠 Tinyproxy 白名单 + 宿主防火墙（第三棒落地中）；客内 root 绕过代理直连在代码层无第二道门 | Sentinel 是连接器动作与全部网络出口的唯一权限机构；L4+L7 评估（主机名、解析后 IP、端口、协议、HTTP 方法、路径、解码后请求）；SSRF 防护；审批是严格能力（绑定连接器/目的地/用例，一次性/会话级/任务级/时限级） | `cloud/chv/sentinel.py` 决策引擎：每动作审批钩子 allow/deny/ask；L4（解析后 IP、端口、协议）+ L7（主机、方法、路径）评估；SSRF 挡截（公开主机名解析到私有/回环/链路本地/CGNAT/保留地址即 deny）；能力绑定 {连接器, 主机, 方法, 路径, 有效期, 次数}；ask 记入待审批队列、执行停止。本期两个接线点：① wake 时 Gitea 出口预审（解析失败/私网 IP/不在白名单 → 起 VMM 前 fail-closed）；② authd 放凭证前先过 Sentinel 评估 |
+| 审批 UI | 无 | Sentinel 直接向客户端发审批请求，弹窗在客户端 UI 内，答复直接回 Sentinel；目的：防对话内容伪造审批 | 本期只做设计约束 + 待审批队列（`sentinel.ApprovalQueue`，内存态 pending 记录）：ask 出现即停止执行、记入队列，不自动放行；UI 不在对话内这一条记为约束，不在本期实现任何对话内审批入口 |
+| 污点出口（eBPF） | 无 | eBPF cgroup 程序做网络拦截与进程归因、LSM hook 做污点传播 | 列为后续，不冒充已做。见诚实边界第 4 条 |
+| 审计 | `_emit` 审计只含 VM 名/路径/URL/kind，不含 key 值（M5） | 审计轨迹展示已做与计划做的事 | 沿用 `_emit` 契约：新增 `sentinel_egress_allow/deny/ask`、`authd_credential_released/refused` 事件，detail 只含连接器/主机/路径/ verdict/原因，不含任何凭证值；authd 客内审计文件 `/var/log/chv-authd.log`（0600）同款口径 |
+
+## 2. 改了什么（本轨交付物）
+
+- `cloud/chv/authd.py`（新，叶子模块，标准库）：凭证保险库（只活内存，
+  `__repr__` 不含值）+ 按次释放（作用域绑定、单次有效、300 秒 TTL、
+  过期即废）+ 连接器白名单（本期只 `gitea`）+ Gitea API 铸 token
+  辅助函数（可注入 HTTP 客户端，默认关闭）+ 无值审计记录。
+- `cloud/chv/sentinel.py`（新，叶子模块，标准库）：决策引擎
+  `evaluate()`（allow/deny/ask）+ L4/L7 评估 + SSRF 挡截
+  （解析器可注入，默认 `socket.getaddrinfo`）+ 严格能力
+  `Capability`（连接器/主机/方法/路径/有效期/次数绑定）+
+  `ApprovalQueue`（ask 待审批队列，内存态）+ 无值审计记录。
+- `cloud/chv/hostkeys.py`：`write_seed` 加 `authd_env` 参数——authd
+  专用密钥写进独立的客内文件 `/etc/chv/authd-seed`（0600 root），
+  不再与 guest-env 混写；`AUTHD_SEED_PATH` 常量。
+- `cloud/chv/gitea_sync.py`：`AUTHD_SECRET_KEYS = {"GITEA_TOKEN"}` +
+  `split_authd_env()`（把 GITEA_TOKEN 从密钥集里拆出来走 authd 通道）。
+- `cloud/chv/ssh_transport.py`：`_CHV_AUTHD_SECRET_NAMES`；`remote_command`
+  语义不变（仍只 source guest-env 文件）。
+- `cloud/chv/executor.py`：`provision_host_key`/`_provision_seed_disk`
+  透传 `authd_env`；新增 `_authd_delivered` 投递集（与
+  `_guest_env_delivered` 同锁）——authd 投递的密钥名在
+  `build_ssh_command` 里不报错、但不 source 进 codex 环境；
+  `_wake_prechecks` 加 Gitea 出口 Sentinel 预审（deny/ask 即
+  `sentinel_egress_denied`，起 VMM 前 fail-closed）。
+- `cloud/chv/config.py`：`CHV_SENTINEL_ALLOW_HOSTS`（逗号分隔主机名白名单，
+  默认只放行配置的 Gitea 主机）。
+- `chv-guest/authd.py`（新，标准库守护进程）：读 `/etc/chv/authd-seed`
+  持 token 于内存、Unix socket `/run/chv-authd/authd.sock`（0700 目录、
+  0600 socket）、SO_PEERCRED 只放行 root 与 chv 用户、每次放凭证前先过
+  Sentinel 评估、审计落盘 0600。
+- `chv-guest/git-credential-chv-authd`（新，标准库 helper）：git
+  credential 协议（stdin 读 `protocol/host/path`，stdout 回用户名/口令），
+  口令只进 git 子进程内存。
+- `chv-guest/chv-authd.service`（新）：systemd 单元（root 运行，单元外）。
+- `chv-guest/chv-sync.sh`：`boot` 在 authd socket 存在时改配
+  `credential.helper` 走 authd，不再把真 token 写进
+  `.git-credentials`；同时配 `credential.useHttpPath=true`
+  （git 默认不把 path 发给 helper，没有 path helper 无法推断
+  仓库与操作、每次都拒绝——新测试抓到的真问题）；socket 不存在
+  时走旧路径 + stderr 警告（过渡兼容，见诚实边界第 5 条）。
+- `cloud/tests/test_chv_muse_p3.py`（新）：替身过期、真 token 泄露面、
+  SSRF 挡截、deny/ask 路径、seed 拆分、wake 预审、helper 端到端
+  （真 git `credential fill` + 假 authd socket）；全量 cloud 套件全过。
+- 镜像集成（只留接入说明，不改 `chv-guest/build.sh`，与 P1 同款规矩，
+  合并时由 P2 一并带进镜像）：
+  `--copy-in chv-guest/authd.py:/usr/local/bin/`、
+  `--copy-in chv-guest/git-credential-chv-authd:/usr/local/bin/`、
+  `--copy-in chv-guest/chv-authd.service:/etc/systemd/system/`、
+  `--copy-in cloud/chv/authd.py:/usr/local/lib/chv-authd/cloud/chv/`、
+  `--copy-in cloud/chv/sentinel.py:/usr/local/lib/chv-authd/cloud/chv/`、
+  `--copy-in cloud/__init__.py:/usr/local/lib/chv-authd/cloud/`、
+  `--copy-in cloud/chv/__init__.py:/usr/local/lib/chv-authd/cloud/chv/`、
+  `--run-command 'chmod +x /usr/local/bin/chv-authd /usr/local/bin/git-credential-chv-authd'`、
+  `--run-command 'systemctl enable chv-authd || true'`、
+  `99-chv-sync.cfg` 的 runcmd 在 `chv-sync boot` 之前加
+  `systemctl start chv-authd || true`。
+
+## 3. 诚实边界（本轮不宣称的东西）
+
+1. **seed 文件仍是 0600 root 落盘**：真 token 在 `/etc/chv/authd-seed`
+   里仍以文件形式存在（cloud-init 的唯一投递通道就是写文件），且 seed
+   ISO 本身挂在客机上。真正让它对 agent 不可见的是 P2 的 nspawn 单元
+   隔离（单元内 root 映射为非特权用户，读不到单元外的 0600 文件）；在
+   P2 落地之前、codex 仍以客内 root 跑的镜像上，root agent 理论上仍能
+   读到它。本轮堵住的是“进程环境/argv/日志/`.git-credentials` 落盘”
+   这条泄露面，不是“客内 root 万能”——后者是 P2 的边界，不是 P3 能修的。
+2. **本期默认的“替身”不是 Gitea 侧可独立校验的新 token**：按次释放给
+   git 子进程的值仍是真 token，只是只进 git 进程内存、不进 agent。
+   真铸 token（Gitea API 铸短期专用 token、到期吊销）已实现辅助函数
+   与假客户端测试，但默认关闭——它要求真 token 有建 token 的权限，
+   现有测试 token 只有 `write:repository`，铸 token 会失败；开它须另配
+   有铸 token 权限的 token，并做真机验证。不得把本期默认写成“Muse 同款
+   替身令牌已落地”。
+3. **SUB2API_API_KEY 不在本轮**：网关 key 仍走 guest-env 进 codex 环境
+   （H2 既有行为）。按 Muse 模型它也该走 privsep worker（由 worker 代
+   codex 调网关、codex 不见 key），那是网关调用链改造，超出本轮范围，
+   列为后续，不静默声称已覆盖。
+4. **eBPF 污点传播未做**：Sentinel 本轮是决策层（wake 预审 + authd 放
+   凭证前评估），不是网络层强制。客内任意进程的直连仍由 Tinyproxy
+   白名单 + 宿主防火墙（第三棒）兜底；eBPF cgroup 拦截 + LSM 污点传播
+   是后续工作，不得写成“出口已全部经 Sentinel”。
+5. **过渡兼容留了旧路径**：`chv-sync boot` 在 authd socket 不存在时仍走
+   旧的 `.git-credentials` 落盘（只多一句 stderr 警告），为的是 lite9/
+   lite10 旧镜像不被一次改断。旧镜像退役后应删掉该回落，改为无 socket
+   即 fail-closed；删除前不得声称“旧路径已关闭”。
+6. **无真机验证**：本轮全部是代码 + 本地单测（含真 git helper 端到端，
+   但 authd 守护进程本身未在真 VM 里跑过）。真机待验证清单：authd 在
+   lite11 镜像里启动时序（须在 `chv-sync boot` 之前就绪）、SO_PEERCRED
+   放行 chv 用户、helper 在 pull/push 下的实际表现、wake 预审对真实
+   Gitea 主机解析的判定、铸 token 的真实 API 行为。测试机（WSL）须等
+   第三棒 V-1～V-9 让出后再做。