agent-exec transport 49a82f74c1def272b56afa8e0ac78c2e/97ab25ec9da4

This commit is contained in:
agent-exec-transport
2026-10-18 17:20:46 +00:00
parent 804a5749cb
commit 7e0b0e4ff8
3 changed files with 415 additions and 2 deletions
+5 -2
View File
@@ -7,12 +7,15 @@
把 PASS 项的 last_verified 刷新为当天并写回本表(随 PR 合入)。
- 表内验证命令不得含竖线 `|`(保持表格与解析可核);新增事实先实测再登记,不得凭记忆填日期。
- precheck 的 `[facts]` 行报告过期项;AGENTS.md「Gitea 与 Git 访问事实」节为本表的叙述版。
- 凭据类验证命令不得把凭据展开进命令行、输出或证据:经 `scripts/facts-verify.sh` 探针子命令在进程内读取并发送。
- 旧端点 git-panel.u1680832.nyat.app:30319 已退役,一律不得使用——这是政策禁令,不随 F4 核验结果变化;
F4 只登记可观测的 TLS 握手失败现象,DNS 失败、超时、连接失败或握手成功都判 FAIL,不刷新 last_verified。
| ID | 事实 | 验证命令 | last_verified | TTL(天) |
|---|---|---|---|---|
| F1 | Gitea Web/API 可达:https://git-home.u1680832.nyat.app:30319(1.27.3) | `curl -fsS -m 10 https://git-home.u1680832.nyat.app:30319/api/v1/version` | 2026-10-14 | 30 |
| F2 | API token 存在且可用(值在 ~/.secrets/gitea,HOME/IMC 两段各一组 `- ACCESS_TOKEN=…`,本条取 HOME 段;不入仓不回显) | `t=$(awk '/^HOME:/{h=1;next} /^IMC:/{h=0} h&&sub(/^- ACCESS_TOKEN=/,""){print;exit}' ~/.secrets/gitea); test -n "$t" && curl -fsS -m 10 -o /dev/null -H "Authorization: token $t" https://git-home.u1680832.nyat.app:30319/api/v1/user` | 2026-10-14 | 30 |
| F2 | API token 存在且可用(值在 ~/.secrets/gitea,HOME/IMC 两段各一组 `- ACCESS_TOKEN=…`,本条取 HOME 段;不入仓不回显,不进子进程 argv) | `bash scripts/facts-verify.sh api-token https://git-home.u1680832.nyat.app:30319/api/v1/user` | 2026-10-14 | 30 |
| F3 | Git over SSH 可达:ssh://git@git-ssh.u1680832.nyat.app:31411/weichen/<Repo>.git(key ~/.ssh/home_gitea) | `GIT_SSH_COMMAND="ssh -i ~/.ssh/home_gitea -o BatchMode=yes" git ls-remote ssh://git@git-ssh.u1680832.nyat.app:31411/weichen/Development-Guidelines.git HEAD` | 2026-10-14 | 30 |
| F4 | 旧端点 git-panel.u1680832.nyat.app:30319 已失效(TLS 握手失败),不得使用 | `! curl -fsS -m 8 https://git-panel.u1680832.nyat.app:30319/api/v1/version` | 2026-10-14 | 180 |
| F4 | 旧端点 git-panel.u1680832.nyat.app:30319 TCP 可连通但 TLS 握手失败(仅 TLS 层握手错误判 PASS) | `bash scripts/facts-verify.sh tls-handshake-fails git-panel.u1680832.nyat.app 30319` | 2026-10-14 | 180 |
| F5 | 基座 users 表存有姓名(name 列,CreateAccount/ListUsersAdmin 读写;EXT-19 目录自 P3 起向授权管理读面下发) | `cd modules/backend && go test ./internal/app/ -run TestInprocHostSearchAccounts -count=1` | 2026-10-14 | 180 |
| F6 | 基座账号无停用位(users 表无 disabled;EXT-19 Valid 恒为存在性)——停用机制未实现,不得当已具备 | `cd modules/backend && ! grep -rn -e disabled -e Disabled internal/store/auth.go pkg/extension/extension.go` | 2026-10-14 | 180 |
+66
View File
@@ -3,7 +3,73 @@
# 规则依据:AGENTS.md「证据与验证门禁」(外部事实零信任);verification.md「声明与证据来源」。
# 用法:make facts-verify [ARGS=--update] (--update:把 PASS 项 last_verified 刷为当天并写回)
# 表内验证命令约定不含竖线;含 "\|" 转义时按字面还原后执行。
# 登记表引用的探针子命令(单独运行时在 stderr 说明判定,不输出凭据):
# api-token URL 进程内读 ~/.secrets/gitea HOME 段 ACCESS_TOKEN 并 GET URL;
# 0=接受 3=缺失 4=格式不符 5=请求失败/被拒
# tls-handshake-fails HOST PORT 仅当 TCP 已连通且 TLS 层握手报错时 0;
# 3=DNS 失败 4=超时 5=连接失败 6=握手成功(端点仍可用)
set -uo pipefail
SCRIPT_DIR="$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd)"
case "${1:-}" in
api-token)
[ $# -eq 2 ] || { echo "usage: facts-verify.sh api-token URL" >&2; exit 2; }
exec python3 - "$SCRIPT_DIR" "$2" <<'PY'
import sys
sys.path.insert(0, sys.argv[1])
import task_triage
url = sys.argv[2]
token = task_triage.secrets_file_token()
if not token:
print("api-token: ~/.secrets/gitea HOME 段 ACCESS_TOKEN 缺失", file=sys.stderr)
sys.exit(3)
if not task_triage.TOKEN_RE.fullmatch(token):
print("api-token: ~/.secrets/gitea HOME 段 ACCESS_TOKEN 格式不符(应为单个无空白值)", file=sys.stderr)
sys.exit(4)
try:
task_triage.http_get_json(url, token)
except task_triage.Stop as stop:
print("api-token: " + stop.message, file=sys.stderr)
sys.exit(5)
print("api-token: accepted by " + url, file=sys.stderr)
PY
;;
tls-handshake-fails)
[ $# -eq 3 ] || { echo "usage: facts-verify.sh tls-handshake-fails HOST PORT" >&2; exit 2; }
exec python3 - "$2" "$3" <<'PY'
import socket, ssl, sys
host, port = sys.argv[1], int(sys.argv[2])
def verdict(code, text):
print(f"tls-handshake-fails {host}:{port}: {text}", file=sys.stderr)
sys.exit(code)
try:
sock = socket.create_connection((host, port), timeout=8)
except socket.gaierror:
verdict(3, "DNS resolution failed (not a TLS observation)")
except TimeoutError:
verdict(4, "TCP connect timed out (not a TLS observation)")
except OSError as exc:
verdict(5, f"TCP connect failed ({type(exc).__name__}; not a TLS observation)")
try:
with sock, ssl.create_default_context().wrap_socket(sock, server_hostname=host) as tls:
version = tls.version()
except ssl.SSLError as exc:
verdict(0, f"TLS handshake failed ({type(exc).__name__}: {getattr(exc, 'reason', None) or 'unknown'})")
except TimeoutError:
verdict(4, "TLS handshake timed out (not a TLS-layer error)")
except OSError as exc:
verdict(5, f"connection failed during handshake ({type(exc).__name__}; not a TLS-layer error)")
verdict(6, f"TLS handshake succeeded ({version}); endpoint still answers")
PY
;;
esac
F="${FACTS:-policy/facts.md}"
[ -f "$F" ] || { echo "ERROR: 事实登记表 $F 不存在" >&2; exit 1; }
python3 - "$F" "${1:-}" <<'PY'
+344
View File
@@ -0,0 +1,344 @@
"""外部事实重验契约(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 json, os, socket, ssl, subprocess, urllib.error, urllib.request
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
class _Response:
def __init__(self, body):
self._body = body
def read(self):
return self._body
def __enter__(self):
return self
def __exit__(self, *exc):
return False
def _urlopen(request, *args, **kwargs):
url = getattr(request, "full_url", request)
auth = request.get_header("Authorization") if hasattr(request, "get_header") else None
_log("http", url=url, authorization=auth)
mode = os.environ.get("FAKE_HTTP", "200")
if mode == "200":
return _Response(b'{"login": "fixture"}')
if mode == "leak":
raise ValueError("Invalid header value " + repr(auth))
raise urllib.error.HTTPError(url, int(mode), "stub", {}, None)
urllib.request.urlopen = _urlopen
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)
# --- 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()