admin/codex-app-server
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(chv): 对齐 Muse P3 收尾——authd 替身凭证 + Sentinel 出口审批

Browse Source 
- 新测试 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 清单
This commit is contained in:
host committed 2026-11-06 21:21:55 +00:00
1 parent 06c3e172fe
commit a90cf78d84
15 files changed
+2370 -93
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

cloud/tests/test_chv_arch_maint_fe.py
+2 -1
View File
@@ -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
+18 -6
View File
@@ -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
+62 -12
View File
@@ -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
+675
View File
@@ -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()