weichen
9374986b63
ci / test (pull_request) Successful in 2m10s
Why: 复核轮遗留的安全测试草稿按「常量=传输端点」设计，与 PR #213 已合入的 「注入基址=传输端点」语义相反，原样提交会红。适配为锁定真实不变量。 Evidence: python3 -m unittest tests.test_ci_credentials 8/8 OK（含新用例： 任一注入拼写检出成功、凭据仅落白名单路径且令牌分离、非 http(s) 注入 fail-closed）。 Impact: 仅测试文件；不改 ci.yaml 行为。 Refs: PR #213 run 950 诊断（GITHUB_SERVER_URL=http://172.17.0.1:3000）。
242 lines
13 KiB
Python
242 lines
13 KiB
Python
"""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]))
|
||
|
||
|
||
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))
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|