"""外部事实重验契约（scripts/facts-verify.sh + policy/facts.md 的 F2/F4 验证命令）。 全部离线：合成 HOME/IMC 凭据、sitecustomize 桩拦截 urllib 处理器网络边界（保留真实重定向判定） 与 socket/ssl（真实连接一律拒绝）、 PATH 桩记录 curl/wget argv、临时 Git 仓库与临时登记表；不读取真实 ~/.secrets，不连接 Gitea 或旧端点。 """ import datetime import json import os import re import shutil import subprocess import tempfile import unittest from pathlib import Path SOURCE = Path(__file__).resolve().parents[1] REGISTRY = SOURCE / "policy" / "facts.md" HOME_TOKEN = "h0me" + "1" * 36 IMC_TOKEN = "1mc0" + "2" * 36 ENV_DECOY = "dec0" + "3" * 36 OLD_DATE = "2026-01-01" SECRETS = [HOME_TOKEN, IMC_TOKEN] STUB = r''' import http.client, io, json, os, socket, ssl, subprocess, urllib.request, urllib.response def _log(kind, **data): with open(os.environ["FACTS_STUB_LOG"], "a", encoding="utf-8") as fh: fh.write(json.dumps({"kind": kind, **data}) + "\n") def _deny(*args, **kwargs): raise RuntimeError("live network disabled in facts-verify tests") socket.socket.connect = _deny socket.socket.connect_ex = _deny _popen_init = subprocess.Popen.__init__ def _popen(self, args, *rest, **kwargs): argv = [args] if isinstance(args, (str, bytes, os.PathLike)) else list(args) _log("exec", argv=[os.fsdecode(a) if isinstance(a, (bytes, os.PathLike)) else str(a) for a in argv]) _popen_init(self, args, *rest, **kwargs) subprocess.Popen.__init__ = _popen def _response(url, code, body=b"", location=None): headers = http.client.HTTPMessage() if location: headers["Location"] = location resp = urllib.response.addinfourl(io.BytesIO(body), headers, url, code) resp.msg = "stub" return resp REDIRECT_TARGET = "redirect-target.invalid" # Intercept only the handlers' network boundary so the opener's real redirect and # error processing still decides whether a follow-up request is made. def _open(self, request): url = request.full_url auth = request.get_header("Authorization") _log("http", url=url, authorization=auth) if REDIRECT_TARGET in url: return _response(url, 200, b'{"login": "redirect-target"}') mode = os.environ.get("FAKE_HTTP", "200") if mode == "200": return _response(url, 200, b'{"login": "fixture"}') if mode == "leak": raise ValueError("Invalid header value " + repr(auth)) code = int(mode) location = "https://" + REDIRECT_TARGET + "/collect" if 300 <= code < 400 else None return _response(url, code, location=location) urllib.request.HTTPSHandler.https_open = _open urllib.request.HTTPHandler.http_open = _open class _Conn: def close(self): pass def version(self): return "TLSv1.3" def __enter__(self): return self def __exit__(self, *exc): return False def _create_connection(address, *args, **kwargs): _log("connect", host=address[0], port=int(address[1])) mode = os.environ.get("FAKE_NET", "") if mode == "dns": raise socket.gaierror(socket.EAI_NONAME, "Name or service not known") if mode == "timeout": raise TimeoutError("timed out") if mode == "refused": raise ConnectionRefusedError(111, "Connection refused") return _Conn() socket.create_connection = _create_connection def _wrap_socket(self, sock, *args, **kwargs): mode = os.environ.get("FAKE_NET", "") if mode == "tls": raise ssl.SSLError(1, "[SSL: SSLV3_ALERT_HANDSHAKE_FAILURE] sslv3 alert handshake failure") if mode == "reset": raise ConnectionResetError(104, "Connection reset by peer") if mode == "tls-timeout": raise TimeoutError("handshake timed out") return _Conn() ssl.SSLContext.wrap_socket = _wrap_socket ''' SHIM = """#!/bin/sh for a in "$0" "$@"; do printf '%s\\n' "$a"; done >> "$FACTS_ARGV_LOG" exit "${FAKE_CURL_EXIT:-0}" """ # Exit codes the retired-endpoint curl check would see for each simulated condition. CURL_EXIT = {"dns": "6", "timeout": "28", "refused": "7", "reset": "35", "tls-timeout": "28", "tls": "35", "ok": "0"} def registry_row(fid): for line in REGISTRY.read_text(encoding="utf-8").splitlines(): if re.match(r"^\|\s*" + fid + r"\s*\|", line): return line raise AssertionError(fid + " missing from policy/facts.md") def registry_command(fid): cells = [c.strip() for c in registry_row(fid).strip().strip("|").replace("\\|", "\x00").split("|")] return cells[2].strip("`").replace("\x00", "|") def with_date(line, date): return re.sub(r"\|\s*\d{4}-\d{2}-\d{2}\s*\|(\s*\d+\s*\|)\s*$", "| " + date + r" |\1", line) class FactsVerifyTests(unittest.TestCase): def setUp(self): tmp = tempfile.TemporaryDirectory(prefix="facts-verify-") self.addCleanup(tmp.cleanup) self.base = Path(tmp.name) self.repo = self.base / "repo" (self.repo / "scripts").mkdir(parents=True) for name in ("facts-verify.sh", "task_triage.py"): shutil.copy2(SOURCE / "scripts" / name, self.repo / "scripts" / name) self.git("init", "-q", "-b", "main") self.git("add", "scripts") self.git("-c", "user.name=Facts Test", "-c", "user.email=facts@example.invalid", "commit", "-qm", "Initial") self.home = self.base / "home" (self.home / ".secrets").mkdir(parents=True) self.write_secrets(f"HOME:\n- URL_PREFIX=https://home.invalid\n- ACCESS_TOKEN={HOME_TOKEN}\n" f"IMC:\n- URL_PREFIX=https://imc.invalid\n- ACCESS_TOKEN={IMC_TOKEN}\n") self.stub_dir = self.base / "stub" self.stub_dir.mkdir() (self.stub_dir / "sitecustomize.py").write_text(STUB, encoding="utf-8") self.shim_dir = self.base / "bin" self.shim_dir.mkdir() for name in ("curl", "wget"): shim = self.shim_dir / name shim.write_text(SHIM, encoding="utf-8") shim.chmod(0o755) self.stub_log = self.base / "stub.log" self.argv_log = self.base / "argv.log" self.facts = self.base / "facts.md" self.today = datetime.date.today().isoformat() def git(self, *args): subprocess.run(["git", *args], cwd=self.repo, check=True, capture_output=True) def write_secrets(self, text): (self.home / ".secrets" / "gitea").write_text(text, encoding="utf-8") def write_registry(self, *fids): rows = [with_date(registry_row(fid), OLD_DATE) for fid in fids] rows.append(f"| F8 | 对照：总能通过 | `true` | {OLD_DATE} | 30 |") rows.append(f"| F9 | 对照：总是失败 | `false` | {OLD_DATE} | 30 |") self.facts.write_text("| ID | 事实 | 验证命令 | last_verified | TTL(天) |\n|---|---|---|---|---|\n" + "\n".join(rows) + "\n", encoding="utf-8") def env(self, net="", http="200"): env = {k: v for k, v in os.environ.items() if k not in ("CMD", "SRC", "RECEIPT", "IMPORT_RECEIPT", "VALIDATE_RECEIPT", "TAIL")} curl_exit = CURL_EXIT.get(net, "0") if net else ("0" if http == "200" else "22") env.update({ "HOME": str(self.home), "GITEA_TOKEN": ENV_DECOY, "FACTS": str(self.facts), "PATH": str(self.shim_dir) + os.pathsep + env.get("PATH", ""), "PYTHONPATH": str(self.stub_dir), "FACTS_STUB_LOG": str(self.stub_log), "FACTS_ARGV_LOG": str(self.argv_log), "FAKE_NET": net, "FAKE_HTTP": http, "FAKE_CURL_EXIT": curl_exit, "AGENT_EVIDENCE_DIR": str(self.base / "evidence"), "GIT_CONFIG_NOSYSTEM": "1", }) return env def run_script(self, *args, net="", http="200"): return subprocess.run(["bash", "scripts/facts-verify.sh", *args], cwd=self.repo, env=self.env(net, http), capture_output=True, text=True, timeout=120) def run_evidence(self, command, receipt, net="", http="200"): env = self.env(net, http) env.update({"CMD": command, "RECEIPT": str(receipt)}) return subprocess.run(["python3", str(SOURCE / "scripts" / "evidence.py")], cwd=self.repo, env=env, capture_output=True, text=True, timeout=120) def stub_events(self, kind): if not self.stub_log.exists(): return [] events = [json.loads(line) for line in self.stub_log.read_text(encoding="utf-8").splitlines()] return [e for e in events if e["kind"] == kind] def dates(self): found = {} for line in self.facts.read_text(encoding="utf-8").splitlines(): match = re.match(r"^\|\s*(F\d+)\s*\|.*\|\s*(\d{4}-\d{2}-\d{2})\s*\|\s*\d+\s*\|\s*$", line) if match: found[match.group(1)] = match.group(2) return found def assert_no_secrets(self, *texts): exec_argv = "\n".join("\0".join(e["argv"]) for e in self.stub_events("exec")) shim_argv = self.argv_log.read_text(encoding="utf-8") if self.argv_log.exists() else "" for label, text in [("child argv", exec_argv), ("curl/wget argv", shim_argv), *((f"output#{i}", t) for i, t in enumerate(texts))]: for secret in SECRETS: self.assertNotIn(secret, text, f"credential leaked into {label}") def assert_status(self, output, fid, status): self.assertRegex(output, r"(?m)^\[" + fid + r"\] " + status + r"\b") # --- F2：凭据只在进程内使用 ------------------------------------------------- def test_f2_pass_sends_home_token_in_process_and_leaks_nowhere(self): self.write_registry("F2") receipt = self.base / "receipt.json" result = self.run_evidence("bash scripts/facts-verify.sh --update", receipt) self.assertEqual(result.returncode, 1, result.stdout + result.stderr) # F9 control fails self.assert_status(result.stdout, "F2", "PASS") requests = self.stub_events("http") self.assertEqual(len(requests), 1, requests) self.assertEqual(requests[0]["url"], "https://git-home.u1680832.nyat.app:30319/api/v1/user") self.assertEqual(requests[0]["authorization"], "token " + HOME_TOKEN) self.assertEqual(self.dates(), {"F2": self.today, "F8": self.today, "F9": OLD_DATE}) self.assertTrue(receipt.exists(), result.stderr) self.assert_no_secrets(result.stdout, result.stderr, receipt.read_text(encoding="utf-8"), self.facts.read_text(encoding="utf-8")) def test_f2_missing_or_malformed_credentials_fail_without_echo(self): self.write_registry("F2") cases = { "no HOME token": f"HOME:\n- URL_PREFIX=https://home.invalid\nIMC:\n- ACCESS_TOKEN={IMC_TOKEN}\n", "malformed HOME token": f"HOME:\n- ACCESS_TOKEN={HOME_TOKEN} {IMC_TOKEN}\n" f"IMC:\n- ACCESS_TOKEN={IMC_TOKEN}\n", "missing file": None, } for label, text in cases.items(): with self.subTest(label): self.stub_log.unlink(missing_ok=True) self.argv_log.unlink(missing_ok=True) secrets = self.home / ".secrets" / "gitea" secrets.unlink(missing_ok=True) if text is not None: self.write_secrets(text) self.write_registry("F2") receipt = self.base / f"receipt-{len(label)}.json" result = self.run_evidence("bash scripts/facts-verify.sh --update", receipt) self.assert_status(result.stdout, "F2", "FAIL") self.assertEqual(self.stub_events("http"), [], "no request without a valid HOME token") self.assertEqual(self.dates()["F2"], OLD_DATE) direct = self.run_script("api-token", "https://git-home.u1680832.nyat.app:30319/api/v1/user") self.assertNotEqual(direct.returncode, 0) self.assertIn("ACCESS_TOKEN", direct.stderr) self.assert_no_secrets(result.stdout, result.stderr, direct.stdout, direct.stderr, receipt.read_text(encoding="utf-8")) def test_f2_rejected_or_failing_request_does_not_refresh_or_leak(self): for http in ("401", "leak"): with self.subTest(http): self.stub_log.unlink(missing_ok=True) self.write_registry("F2") result = self.run_script("--update", http=http) self.assert_status(result.stdout, "F2", "FAIL") self.assertEqual(self.dates(), {"F2": OLD_DATE, "F8": self.today, "F9": OLD_DATE}) direct = self.run_script("api-token", "https://git-home.u1680832.nyat.app:30319/api/v1/user", http=http) self.assertNotEqual(direct.returncode, 0) self.assertNotIn("Authorization", direct.stderr) self.assert_no_secrets(result.stdout, result.stderr, direct.stdout, direct.stderr) def test_f2_rejects_unregistered_destination_before_reading_credentials(self): bad_urls = [ "http://git-home.u1680832.nyat.app:30319/api/v1/user", "https://evil.invalid/api/v1/user", "https://git-home.u1680832.nyat.app:30320/api/v1/user", "https://git-home.u1680832.nyat.app/api/v1/user", "https://user:pw@git-home.u1680832.nyat.app:30319/api/v1/user", "https://git-home.u1680832.nyat.app:30319@evil.invalid/api/v1/user", "https://git-home.u1680832.nyat.app:30319/api/v1/user#frag", "https://git-home.u1680832.nyat.app:30319/api/v1/user?next=https://evil.invalid", "https://git-home.u1680832.nyat.app:30319/api/v1/users", "https://git-home.u1680832.nyat.app:30319/api/v1/user/../admin", "https://git-home.u1680832.nyat.app:30319/api/v1/user ", ] for secrets_present in (True, False): if not secrets_present: (self.home / ".secrets" / "gitea").unlink() for url in bad_urls: with self.subTest(url=url, secrets_present=secrets_present): self.stub_log.unlink(missing_ok=True) direct = self.run_script("api-token", url) self.assertEqual(direct.returncode, 6, direct.stderr) self.assertEqual(self.stub_events("http"), [], "no request to an unregistered destination") self.assertNotIn(url.strip(), direct.stdout + direct.stderr) self.assertNotIn("evil.invalid", direct.stdout + direct.stderr) self.assertNotIn("ACCESS_TOKEN", direct.stderr, "destination is checked before credentials") self.assert_no_secrets(direct.stdout, direct.stderr) def test_f2_refuses_redirect_without_forwarding_authorization(self): for code in ("301", "302", "303", "307", "308"): with self.subTest(code): self.stub_log.unlink(missing_ok=True) direct = self.run_script("api-token", "https://git-home.u1680832.nyat.app:30319/api/v1/user", http=code) self.assertEqual(direct.returncode, 5, direct.stderr) requests = self.stub_events("http") self.assertEqual([r["url"] for r in requests], ["https://git-home.u1680832.nyat.app:30319/api/v1/user"], "no follow-up request") self.assertNotIn("redirect-target", direct.stdout + direct.stderr) self.assert_no_secrets(direct.stdout, direct.stderr) self.stub_log.unlink(missing_ok=True) self.write_registry("F2") result = self.run_script("--update", http=code) self.assert_status(result.stdout, "F2", "FAIL") self.assertEqual(self.dates()["F2"], OLD_DATE) self.assertEqual(len(self.stub_events("http")), 1) self.assert_no_secrets(result.stdout, result.stderr) # --- F4：只有 TLS 层握手失败才支持该事实 ----------------------------------- def test_f4_only_tls_handshake_failure_passes_and_refreshes(self): expected = {"tls": "PASS", "dns": "FAIL", "timeout": "FAIL", "refused": "FAIL", "reset": "FAIL", "tls-timeout": "FAIL", "ok": "FAIL"} for net, status in expected.items(): with self.subTest(net): self.stub_log.unlink(missing_ok=True) self.write_registry("F4") result = self.run_script("--update", net=net) self.assert_status(result.stdout, "F4", status) self.assertEqual(self.dates(), {"F4": self.today if status == "PASS" else OLD_DATE, "F8": self.today, "F9": OLD_DATE}) connects = self.stub_events("connect") self.assertEqual([(c["host"], c["port"]) for c in connects], [("git-panel.u1680832.nyat.app", 30319)]) def test_f4_helper_reports_each_condition_distinctly(self): codes = {} for net in ("tls", "dns", "timeout", "refused", "reset", "tls-timeout", "ok"): result = self.run_script("tls-handshake-fails", "git-panel.u1680832.nyat.app", "30319", net=net) codes[net] = result.returncode self.assertTrue(result.stderr.strip(), net + " must explain its verdict") self.assertEqual(codes["tls"], 0) self.assertEqual(len({codes["dns"], codes["timeout"], codes["refused"], codes["ok"]}), 4, codes) self.assertNotIn(0, [code for net, code in codes.items() if net != "tls"]) def test_registry_separates_retired_endpoint_policy_from_tls_observation(self): text = REGISTRY.read_text(encoding="utf-8") policy_lines = [line for line in text.splitlines() if "git-panel" in line and not line.startswith("|")] self.assertTrue(any("不得使用" in line for line in policy_lines), "retired-endpoint prohibition must stand outside the F4 verification row") self.assertNotIn("不得使用", registry_row("F4")) for fid in ("F2", "F4"): command = registry_command(fid) self.assertNotIn("curl", command) self.assertNotIn("Authorization", command) self.assertNotIn(".secrets", command) if __name__ == "__main__": unittest.main()