"""Execute the workflow checkout against real disposable HTTP Git repositories.""" import http.server import os from pathlib import Path import subprocess import tempfile import textwrap import threading import unittest from urllib.parse import urlsplit WORKFLOW = Path(__file__).resolve().parents[1] / ".gitea/workflows/ci.yaml" SERVER = "https://git-home.u1680832.nyat.app:30319" SSH = "ssh://git@git-ssh.u1680832.nyat.app:31411/weichen/" ROOT_TOKEN = "synthetic-root-token-not-a-real-secret" SUB_TOKEN = "synthetic-component-token-not-a-real-secret" def checkout_script(): lines = WORKFLOW.read_text().splitlines() start = next(i for i, line in enumerate(lines) if line.strip() == "- name: Checkout") start = next(i for i in range(start, len(lines)) if lines[i].strip() == "run: |") + 1 end = next(i for i in range(start, len(lines)) if lines[i].startswith(" - name:")) return textwrap.dedent("\n".join(lines[start:end])) def rdm_clone_script(): lines = WORKFLOW.read_text().splitlines() start = next(i for i, line in enumerate(lines) if "Clone Agent-System-RDM" in line) start = next(i for i in range(start, len(lines)) if lines[i].strip() == "run: |") + 1 end = next(i for i in range(start, len(lines)) if lines[i].startswith(" - name:")) return textwrap.dedent("\n".join(lines[start:end])) class CheckoutTests(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory() self.addCleanup(self.temp.cleanup) self.base = Path(self.temp.name) self.published = self.base / "published" self.published.mkdir() self.home = self.base / "home" self.home.mkdir() self.checkout = self.base / "checkout" self.checkout.mkdir() self.env = {k: v for k, v in os.environ.items() if not k.startswith(("GIT_", "GITHUB_")) and k not in {"TOKEN", "SUBMODULE_TOKEN"}} self.env.update(HOME=str(self.home), GIT_CONFIG_NOSYSTEM="1", GIT_TERMINAL_PROMPT="0") self.requests = [] self.redirect_repo = None outer = self class Handler(http.server.BaseHTTPRequestHandler): def log_message(self, *args): pass def do_GET(self): self.respond() def do_POST(self): self.respond() def respond(self): parsed = urlsplit(self.path) outer.requests.append((parsed.path, self.headers.get("Authorization"))) if outer.redirect_repo and outer.redirect_repo in parsed.path: self.send_response(302) self.send_header("Location", outer.url + "/untrusted/receiver") self.end_headers() return env = outer.env | {"GIT_PROJECT_ROOT": str(outer.published), "GIT_HTTP_EXPORT_ALL": "1", "PATH_INFO": parsed.path, "QUERY_STRING": parsed.query, "REQUEST_METHOD": self.command, "CONTENT_TYPE": self.headers.get("Content-Type", ""), "CONTENT_LENGTH": self.headers.get("Content-Length", "0")} body = self.rfile.read(int(env["CONTENT_LENGTH"])) result = subprocess.run(["git", "http-backend"], input=body, env=env, capture_output=True) headers, body = result.stdout.split(b"\r\n\r\n", 1) pairs = [line.decode().split(": ", 1) for line in headers.split(b"\r\n")] status = next((int(value.split()[0]) for key, value in pairs if key == "Status"), 200) self.send_response(status) for key, value in pairs: if key != "Status": self.send_header(key, value) self.end_headers() self.wfile.write(body) self.http = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Handler) self.url = f"http://127.0.0.1:{self.http.server_port}" # Any regression that tries a production HTTPS URL must fail locally. self.env.update(https_proxy=self.url, HTTPS_PROXY=self.url, no_proxy="127.0.0.1,localhost", NO_PROXY="127.0.0.1,localhost") thread = threading.Thread(target=self.http.serve_forever, daemon=True) thread.start() self.addCleanup(self.http.server_close) self.addCleanup(self.http.shutdown) def git(self, directory, *args): return subprocess.run(["git", "-C", str(directory), *args], env=self.env, check=True, capture_output=True, text=True).stdout.strip() def publish(self, name, children=(), extra_gitmodules=""): source = self.base / name source.mkdir() self.git(source, "init", "-q", "-b", "main") (source / "README").write_text("Fixture repository\n") # Candidate code tries the channels available after the checkout step. (source / "probe.py").write_text(textwrap.dedent(f"""\ import os, pathlib, subprocess sources = [repr(dict(os.environ)), subprocess.check_output(['git', 'config', '--list'], text=True)] for path in [pathlib.Path.home() / '.gitconfig', *pathlib.Path('.git').rglob('config')]: if path.is_file(): sources.append(path.read_text()) assert not any(secret in source for secret in ({ROOT_TOKEN!r}, {SUB_TOKEN!r}) for source in sources), 'credential readable by candidate' print('candidate cannot read checkout credentials') """)) gitmodules = "" for path, repo, sha, override in children: gitmodules += f'[submodule "{path}"]\n\tpath = {path}\n\turl = {override or SSH + repo + ".git"}\n' if gitmodules or extra_gitmodules: (source / ".gitmodules").write_text(gitmodules + extra_gitmodules) self.git(source, "add", ".") for path, repo, sha, override in children: self.git(source, "update-index", "--add", "--cacheinfo", f"160000,{sha},{path}") self.git(source, "-c", "user.name=Fixture", "-c", "user.email=fixture@example.invalid", "commit", "-q", "-m", "Fixture") sha = self.git(source, "rev-parse", "HEAD") target = self.published / "weichen" / (name + ".git") target.parent.mkdir(exist_ok=True) self.git(self.base, "clone", "-q", "--bare", str(source), str(target)) return sha def run_checkout(self, sha, metadata_url=None): # Only the fixed transport endpoint changes; execute the actual workflow # shell/Python logic, not a test-specific implementation of checkout. script = checkout_script().replace(SERVER, self.url) return subprocess.run(["bash", "-c", script], cwd=self.checkout, env=self.env | {"TOKEN": ROOT_TOKEN, "SUBMODULE_TOKEN": SUB_TOKEN, "GITHUB_SERVER_URL": metadata_url or self.url, "GITHUB_REPOSITORY": "weichen/Agent-System-Dev", "GITHUB_SHA": sha}, capture_output=True, text=True, timeout=30) def test_credential_destination_confined_to_injected_server(self): # 合入语义（PR #213，2026-10-05）：传输基址 = 运行器注入的 GITHUB_SERVER_URL # （run 950 实测为内网 http 形态 http://172.17.0.1:3000），凭据只发往 # 「该基址 + 仓库白名单」路径。锁定三点：任一注入拼写（含/不含尾斜杠）检出 # 成功；每个带凭据请求都落在白名单仓库路径且 root/子仓令牌分离；注入值 # 非 http(s) 时 fail-closed 拒绝——不回落到任何其他端点、不发凭据。 core = self.publish("Agent-System-Core") root = self.publish("Agent-System-Dev", [("modules/core", "Agent-System-Core", core, None)]) for index, injected in enumerate((self.url, self.url + "/")): with self.subTest(injected=injected): self.checkout = self.base / f"injected-checkout-{index}" self.checkout.mkdir() self.requests.clear() result = self.run_checkout(root, metadata_url=injected) self.assertEqual(result.returncode, 0, result.stderr) self.assertEqual(self.git(self.checkout / "modules/core", "rev-parse", "HEAD"), core) self.assertTrue(self.requests) for path, auth in self.requests: self.assertRegex(path, r"^/weichen/Agent-System-(Dev|Core)\.git/(info/refs|git-upload-pack)$") self.assertEqual(auth, "token " + (ROOT_TOKEN if "Agent-System-Dev.git/" in path else SUB_TOKEN)) self.checkout = self.base / "injected-checkout-refused" self.checkout.mkdir() self.requests.clear() result = self.run_checkout(root, metadata_url="git-panel.u1680832.nyat.app:30319") self.assertNotEqual(result.returncode, 0) self.assertFalse(self.requests) def test_candidate_cannot_read_checkout_credentials(self): sha = self.publish("Agent-System-Dev") result = self.run_checkout(sha) self.assertEqual(result.returncode, 0, result.stderr) candidate = subprocess.run(["python3", "probe.py"], cwd=self.checkout, env=self.env, capture_output=True, text=True) self.assertEqual(candidate.returncode, 0, candidate.stderr) self.assertNotIn(ROOT_TOKEN, result.stdout + result.stderr) self.assertNotIn(SUB_TOKEN, result.stdout + result.stderr) def test_recursive_gitlinks_materialize_exact_sha_without_persisting_credentials(self): core = self.publish("Agent-System-Core") backend = self.publish("Agent-System-Backend", [("deps/agent-system-core", "Agent-System-Core", core, None)]) root = self.publish("Agent-System-Dev", [("modules/backend", "Agent-System-Backend", backend, None)]) result = self.run_checkout(root) self.assertEqual(result.returncode, 0, result.stderr) self.assertEqual(self.git(self.checkout / "modules/backend", "rev-parse", "HEAD"), backend) self.assertEqual(self.git(self.checkout / "modules/backend/deps/agent-system-core", "rev-parse", "HEAD"), core) self.assertNotIn("-", self.git(self.checkout, "submodule", "status", "--recursive")[:1]) candidate = subprocess.run(["python3", "probe.py"], cwd=self.checkout, env=self.env, capture_output=True, text=True) self.assertEqual(candidate.returncode, 0, candidate.stderr) for path, auth in self.requests: self.assertEqual(auth, "token " + (ROOT_TOKEN if "Agent-System-Dev.git/" in path else SUB_TOKEN)) for config in (self.checkout / ".git").rglob("config"): self.assertNotIn(SUB_TOKEN, config.read_text()) def test_external_or_unlisted_urls_are_rejected_before_component_fetch(self): core = self.publish("Agent-System-Core") root = self.publish("Agent-System-Dev", [("modules/core", "Agent-System-Core", core, self.url + "/untrusted/receiver")]) result = self.run_checkout(root) self.assertNotEqual(result.returncode, 0) self.assertFalse(any("untrusted" in path for path, auth in self.requests)) self.assertFalse((self.checkout / "modules/core/.git").exists()) def test_candidate_submodule_update_commands_are_rejected(self): core = self.publish("Agent-System-Core") root = self.publish("Agent-System-Dev", [("modules/core", "Agent-System-Core", core, None)], "\tupdate = !touch /tmp/should-never-execute\n") result = self.run_checkout(root) self.assertNotEqual(result.returncode, 0) self.assertFalse(any("Agent-System-Core.git/" in path for path, auth in self.requests)) def test_nested_external_url_is_rejected_before_authenticated_fetch(self): core = self.publish("Agent-System-Core") backend = self.publish("Agent-System-Backend", [("deps/agent-system-core", "Agent-System-Core", core, self.url + "/untrusted/receiver")]) root = self.publish("Agent-System-Dev", [("modules/backend", "Agent-System-Backend", backend, None)]) result = self.run_checkout(root) self.assertNotEqual(result.returncode, 0) self.assertFalse(any("untrusted" in path for path, auth in self.requests)) self.assertFalse(any("Agent-System-Core.git/" in path for path, auth in self.requests)) def test_inherited_template_hook_does_not_execute_with_credentials(self): template = self.base / "template" (template / "hooks").mkdir(parents=True) marker = self.base / "hook-executed" hook = template / "hooks/post-checkout" hook.write_text(f"#!/bin/sh\ntouch '{marker}'\n") hook.chmod(0o755) self.git(self.base, "config", "--global", "init.templateDir", str(template)) root = self.publish("Agent-System-Dev") result = self.run_checkout(root) self.assertEqual(result.returncode, 0, result.stderr) self.assertFalse(marker.exists()) def test_redirects_do_not_forward_credential_or_fetch_unlisted_target(self): core = self.publish("Agent-System-Core") root = self.publish("Agent-System-Dev", [("modules/core", "Agent-System-Core", core, None)]) self.redirect_repo = "Agent-System-Core.git" result = self.run_checkout(root) self.assertNotEqual(result.returncode, 0) self.assertFalse(any("untrusted" in path for path, auth in self.requests)) def test_rdm_pin_fetch_is_three_shas_on_the_allowlist(self): source = self.base / "Agent-System-RDM" source.mkdir() self.git(source, "init", "-q", "-b", "main") shas = [] for index in range(3): (source / "Makefile").write_text( "build:\n\tmkdir -p bin\n\tprintf 'rev%d\\n' > bin/agent-system-rdm\n" % index) self.git(source, "add", ".") self.git(source, "-c", "user.name=Fixture", "-c", "user.email=fixture@example.invalid", "commit", "-q", "-m", "rev %d" % index) shas.append(self.git(source, "rev-parse", "HEAD")) target = self.published / "weichen" / "Agent-System-RDM.git" target.parent.mkdir(parents=True, exist_ok=True) self.git(self.base, "clone", "-q", "--bare", str(source), str(target)) self.git(target, "config", "uploadpack.allowReachableSHA1InWant", "true") self.git(target, "config", "uploadpack.allowAnySHA1InWant", "true") work = self.base / "rdm-job" pin = work / "tests" / "integration" / "lifecycle" pin.mkdir(parents=True) (pin / "rdm.pin").write_text( "sha=%s\nprev_sha=%s\nlegacy_sha=%s\n" % (shas[2], shas[1], shas[0])) script = rdm_clone_script() self.assertNotRegex(script, r"(?m)^\s*source\s") self.requests.clear() result = subprocess.run( ["bash", "-c", script], cwd=work, env=self.env | {"SUBMODULE_TOKEN": SUB_TOKEN, "GITHUB_SERVER_URL": self.url}, capture_output=True, text=True, timeout=60) self.assertEqual(result.returncode, 0, result.stdout + result.stderr) self.assertNotIn(SUB_TOKEN, result.stdout + result.stderr) self.assertTrue(self.requests) for path, auth in self.requests: self.assertRegex(path, r"^/weichen/Agent-System-RDM\.git/") self.assertEqual(auth, "token " + SUB_TOKEN) for sha in shas: self.assertEqual(self.git(work / "rdm", "cat-file", "-t", sha), "commit") self.assertEqual(self.git(work / "rdm", "rev-parse", "HEAD"), shas[2]) digests = (work / "rdm" / "pin-build-digests.txt").read_text() for sha in shas: self.assertIn(sha, digests) gate = WORKFLOW.read_text().split( "Real-combination lifecycle gate", 1)[1].split("\n svc-realcombo:", 1)[0] self.assertIn('export SVC_CI_GIT_TOKEN="$SUBMODULE_TOKEN"', gate) self.assertNotRegex(gate, r"(?m)^\s*source\s") self.assertNotIn(">>", gate) if __name__ == "__main__": unittest.main()