weichen
8af945a1d5
Why: cleanup, component synchronization, receipt recording and credential handling could contradict the documented development boundaries. Evidence: SELF 75 isolated governance tests and source checks are captured in .task/feature-governance-reliability.md; combination and CI verification follow. Impact: hardens local task and CI tooling, aligns policy and component documentation, and adds continuous regression tests without changing product behavior. Refs: .task/feature-governance-reliability.md; policy/governance.md; policy/platform-admission.md
214 lines
11 KiB
Python
214 lines
11 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):
|
|
# 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": self.url,
|
|
"GITHUB_REPOSITORY": "weichen/Agent-System-Dev",
|
|
"GITHUB_SHA": sha},
|
|
capture_output=True, text=True, timeout=30)
|
|
|
|
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()
|