host
a90cf78d84
- 新测试 cloud/tests/test_chv_muse_p3.py 30 项：替身过期、真 token 泄露面、SSRF 挡截、deny/ask 路径、authd-seed 拆分、wake 预审、 客内 authd 守护进程、git credential helper 端到端（真 git fill） - 8 个旧测试按新 authd 行为改写（未删测试、未放宽断言）： GITEA_TOKEN 改走 authd-seed，build_ssh_command 不再 source 进 codex 环境；wake 测试补 Sentinel 白名单 + 假解析器 - 7 处超 50 行函数拆到 50 行内（不改行为）；补 authd/sentinel 公共 API 注解，mypy 在改动文件上干净 - 真问题修复：git 默认不把 path 发给 helper，chv-sync boot 加 credential.useHttpPath=true，否则 authd 路径每次都拒绝 - 全量 cloud 套件 1897 项 OK（skipped=8） 真机待验证（未做，等第三棒让出 WSL 测试机）：见 docs/chv-muse-p3-gap.md §3.6 清单
676 lines
28 KiB
Python
676 lines
28 KiB
Python
"""对齐 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()
|