400 lines
18 KiB
Python
400 lines
18 KiB
Python
"""外部事实重验契约（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()
|