"""对齐 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"]) def test_write_seed_authd_entry_carries_sync_keys(self): # round-11 真机修复（层 2）：authd-seed 须同时有 # GITEA_URL/GITEA_USER/SYNC_REPO，build_server 才会放行 #（只 GITEA_TOKEN 时 ValueError，探针 fixprobe3 即从 # guest-env 补这三键）；三键只在 guest-env 里有才抄。 import yaml with tempfile.TemporaryDirectory() as tmp: seed_dir = os.path.join(tmp, "seed") write_seed(seed_dir, "vm1", "PRIV", "ssh-ed25519 AAAAx", guest_env={ "GITEA_URL": "https://gitea.tose.top", "GITEA_USER": "admin", "SYNC_REPO": "admin/chv-sync-test", "SYNC_BRANCH": "main"}, 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"]} content = entries[AUTHD_SEED_PATH]["content"] for key in ("GITEA_URL=https://gitea.tose.top", "GITEA_USER=admin", "SYNC_REPO=admin/chv-sync-test", "GITEA_TOKEN="): self.assertIn(key, content) # 非同步键不进 authd-seed。 self.assertNotIn("SYNC_BRANCH", content) def test_write_seed_authd_entry_without_sync_keys(self): # guest-env 里没有三键时 authd-seed 只装密钥，不编默认值。 import yaml with tempfile.TemporaryDirectory() as tmp: seed_dir = os.path.join(tmp, "seed") write_seed(seed_dir, "vm1", "PRIV", "ssh-ed25519 AAAAx", 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"]} content = entries[AUTHD_SEED_PATH]["content"] self.assertIn("GITEA_TOKEN=", content) self.assertNotIn("GITEA_URL", content) self.assertNotIn("GITEA_USER", content) self.assertNotIn("SYNC_REPO", 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()