codex-app-server/cloud/tests/test_chv_muse_p3.py

"""对齐 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()