diff --git a/AGENTS.md b/AGENTS.md index d3cd45ce..de09a87b 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -134,7 +134,8 @@ policy/delivery.md。 合入与验收须核验 GATE 的运行链接/身份及 SHA,或 REPLAY 的独立审阅者、命令和对象; 本地 evidence 只生成 SELF,不能靠 `SRC` 自行升级;与重验不一致时以重验为准。 - **证据由工具写入**:命令输出进入任务卡/PR 一律经 `CMD="<命令>" make evidence` 写入 - (记录脱敏命令、执行前后对象、退出码、完整脱敏输出与摘要)。凭据只经环境或受保护文件 + (卡内写摘要块:脱敏命令、前后 HEAD、退出码、输出尾部与摘要;含执行前后对象和完整脱敏输出的 + 完整收据一次性存于仓外记录路径,见 policy/governance.md)。凭据只经环境或受保护文件 引用;命令、输出和摘要统一脱敏,摘要只证明记录完整性,不证明执行者身份。手抄、转写、 凭记忆重构的输出视为无效证据;补写输出与伪造测试同级红线。 - **完成声明三段输出**(precheck / task-sync / 测试)必须由 evidence 工具写入任务卡, diff --git a/policy/governance.md b/policy/governance.md index 7e328e43..5b1eeba8 100644 --- a/policy/governance.md +++ b/policy/governance.md @@ -103,6 +103,11 @@ PR 合入由智能体执行:先完成有效 Review,再同步并解决冲突 命令含敏感值时展示脱敏形式,并记录原命令摘要。执行前后对象分别记录;命令改变 HEAD 时不把执行前 SHA 冒充执行后对象。不能读取执行后状态时明确记录缺项。 +写入任务卡(`CMD` 直写或 `IMPORT_RECEIPT` 导入)时,卡内只放摘要块(命令、退出码、前后 HEAD、 +输出尾部及各摘要);完整收据以 0600 一次性写到仓外的记录路径(默认 +`~/.cache/agent-system-dev/evidence/.json`,可用 `AGENT_EVIDENCE_DIR` 覆盖), +不覆盖不同内容。`check-evidence` 在本机可读时用完整收据核对摘要,缺失则警告未验证;两者仍是 SELF。 + ### 候选固定前后 固定前,完成同步、必要检查、任务卡及本地收据提交,核对干净的组件组合,再读取和推送最终 HEAD。 diff --git a/scripts/check-evidence.py b/scripts/check-evidence.py index 70395128..7e56ab96 100644 --- a/scripts/check-evidence.py +++ b/scripts/check-evidence.py @@ -97,6 +97,72 @@ def check_receipt(value): return before_stale or after_stale +SUMMARY_FIELDS = {"summary_schema", "source", "timestamp", "command", "exit_code", "cwd", + "before_head", "after_head", "branch", "target_branch", "target_head", + "before_tracked_diff_sha256", "after_tracked_diff_sha256", "output_tail", + "output_sha256", "receipt_sha256", "receipt_path"} + + +def optional_match(pattern, value): + return value is None or (isinstance(value, str) and pattern.fullmatch(value) is not None) + + +def check_summary_fields(value): + if type(value.get("summary_schema")) is not int or value["summary_schema"] != evidence.SUMMARY_SCHEMA: + raise ValueError("unsupported summary schema") + keys = set(value) + targets = keys & set(evidence.TARGET_FIELDS) + if targets not in (set(), set(evidence.TARGET_FIELDS)) or keys - targets != SUMMARY_FIELDS: + raise ValueError("invalid summary fields") + for key in targets: + if not evidence.valid_branch(value[key]): + raise ValueError("invalid summary target: " + key) + for key in ("source", "timestamp", "command", "cwd", "branch", "target_branch", "receipt_path"): + if not isinstance(value[key], str): + raise ValueError("invalid summary field: " + key) + if value["source"] != "SELF" or type(value["exit_code"]) is not int or not 0 <= value["exit_code"] <= 255: + raise ValueError("invalid summary source or exit code") + try: + datetime.fromisoformat(value["timestamp"].replace("Z", "+00:00")) + except ValueError: + raise ValueError("invalid summary timestamp") from None + if not isinstance(value["before_head"], str) or not evidence.GIT_SHA.fullmatch(value["before_head"]): + raise ValueError("invalid summary before_head") + for key in ("after_head", "target_head"): + if not optional_match(evidence.GIT_SHA, value[key]): + raise ValueError("invalid summary field: " + key) + for key in ("before_tracked_diff_sha256", "output_sha256", "receipt_sha256"): + if not isinstance(value[key], str) or not evidence.SHA256.fullmatch(value[key]): + raise ValueError("invalid summary digest: " + key) + if not optional_match(evidence.SHA256, value["after_tracked_diff_sha256"]) or ( + (value["after_head"] is None) != (value["after_tracked_diff_sha256"] is None)): + raise ValueError("invalid summary digest: after_tracked_diff_sha256") + if not isinstance(value["output_tail"], list) or not all(isinstance(line, str) for line in value["output_tail"]): + raise ValueError("invalid summary output tail") + path = Path(value["receipt_path"]) + if not Path(value["cwd"]).is_absolute() or not path.is_absolute() or ( + path.name != value["receipt_sha256"] + ".json"): + raise ValueError("invalid summary paths") + + +def check_summary(label, value): + """Verify a card summary against its full receipt when it is stored on this machine.""" + check_summary_fields(value) + path = Path(value["receipt_path"]) + if not path.is_file(): + warn(label, f"unverified: full receipt not available locally ({path})") + try: + heads = [value["before_head"]] + ([value["after_head"]] if value["after_head"] else []) + return any([check_head(head) for head in heads]) + except (ValueError, OSError): + return False + full = json.loads(path.read_text(encoding="utf-8")) + stale = check_receipt(full) + if evidence.build_summary(full, value["receipt_path"], len(value["output_tail"])) != value: + raise ValueError("summary does not match full receipt") + return stale + + def records(content): """Yield fenced JSON and timestamped legacy rows without interpreting text.""" block = None @@ -125,7 +191,12 @@ def main(): return for label, kind, record in records(content): try: - stale = check_receipt(json.loads(record)) if kind == "json" else check_head(record) + if kind == "json": + value = json.loads(record) + stale = (check_summary(label, value) if isinstance(value, dict) and "summary_schema" in value + else check_receipt(value)) + else: + stale = check_head(record) except (ValueError, TypeError, KeyError, OSError, evidence.EvidenceError) as exc: warn(label, "unverified: " + str(exc)) continue diff --git a/scripts/evidence.py b/scripts/evidence.py index fc36c290..a40a5e84 100644 --- a/scripts/evidence.py +++ b/scripts/evidence.py @@ -14,9 +14,12 @@ from pathlib import Path import re import subprocess import sys +import tempfile SCHEMA = 1 +SUMMARY_SCHEMA = 1 +DEFAULT_TAIL = 15 SECRET_NAME = re.compile(r"token|password|passwd|secret|api[_-]?key|credential|private[_-]?key", re.I) SHA256 = re.compile(r"[0-9a-f]{64}\Z") GIT_SHA = re.compile(r"(?:[0-9a-f]{40}|[0-9a-f]{64})\Z") @@ -332,8 +335,79 @@ def find_card(): raise EvidenceError("No task card exists for the current branch") +def evidence_dir(): + configured = os.environ.get("AGENT_EVIDENCE_DIR") + if configured: + base = Path(configured) + else: + base = Path(os.environ.get("XDG_CACHE_HOME") or Path.home() / ".cache") / "agent-system-dev" / "evidence" + base.mkdir(mode=0o700, parents=True, exist_ok=True) + return base.resolve() + + +def store_receipt(receipt): + """Write the full receipt once under its digest; identical existing content is accepted.""" + directory = evidence_dir() + path = directory / (receipt["receipt_sha256"] + ".json") + encoded = json.dumps(receipt, ensure_ascii=False, indent=2) + "\n" + if path.exists() or path.is_symlink(): + try: + same = not path.is_symlink() and json.loads(path.read_text(encoding="utf-8")) == receipt + except (OSError, ValueError): + same = False + if not same: + raise EvidenceError("Full receipt path holds different content: " + str(path)) + return path + fd, temporary = tempfile.mkstemp(prefix=".receipt-", dir=directory) + try: + with os.fdopen(fd, "w", encoding="utf-8") as stream: + os.fchmod(stream.fileno(), 0o600) + stream.write(encoded) + stream.flush() + os.fsync(stream.fileno()) + try: + os.link(temporary, path) + except FileExistsError: + return store_receipt(receipt) + finally: + os.unlink(temporary) + return path + + +def tail_lines(): + value = os.environ.get("TAIL", "") + return int(value) if value.isdigit() else DEFAULT_TAIL + + +def build_summary(receipt, receipt_path, tail=None): + """Card-sized view of a full receipt; every field is derived from the sanitized receipt.""" + tail = tail_lines() if tail is None else tail + before, after = receipt["before"], receipt["after"] + summary = { + "summary_schema": SUMMARY_SCHEMA, "source": "SELF", + "timestamp": receipt["timestamp"], "command": receipt["command"], + "exit_code": receipt["exit_code"], "cwd": receipt["cwd"], + "before_head": before["head"], "after_head": after["head"] if after else None, + "branch": before["branch"], "target_branch": before["target_branch"], + "target_head": before["target_head"], + } + for key in TARGET_FIELDS: + if key in receipt: + summary[key] = receipt[key] + summary.update({ + "before_tracked_diff_sha256": before["tracked_diff_sha256"], + "after_tracked_diff_sha256": after["tracked_diff_sha256"] if after else None, + "output_tail": receipt["output"].splitlines()[-tail:] if tail > 0 else [], + "output_sha256": receipt["output_sha256"], + "receipt_sha256": receipt["receipt_sha256"], + "receipt_path": str(receipt_path), + }) + return summary + + def append_card(card, receipt): - encoded = json.dumps(receipt, ensure_ascii=False, indent=2) + path = store_receipt(receipt) + encoded = json.dumps(build_summary(receipt, path), ensure_ascii=False, indent=2) block = "\n## 证据(Evidence)\n\n" + ( "\n" + "```json\n" + encoded + "\n```\n") diff --git a/tests/test_check_evidence.py b/tests/test_check_evidence.py new file mode 100644 index 00000000..0d55a188 --- /dev/null +++ b/tests/test_check_evidence.py @@ -0,0 +1,123 @@ +"""Exercise check-evidence.py on card summaries and legacy full receipt blocks.""" +import json +import os +from pathlib import Path +import subprocess +import tempfile +import unittest + + +SCRIPTS = Path(__file__).resolve().parents[1] / "scripts" + + +class CheckEvidenceTests(unittest.TestCase): + def setUp(self): + self.temp = tempfile.TemporaryDirectory() + self.addCleanup(self.temp.cleanup) + self.base = Path(self.temp.name) + self.repo = self.base / "repo" + self.repo.mkdir() + self.git("init", "-b", "main") + self.git("config", "user.name", "Evidence Test") + self.git("config", "user.email", "evidence@example.invalid") + (self.repo / "tracked").write_text("initial\n") + self.git("add", "tracked") + self.git("commit", "-qm", "Initial") + self.git("update-ref", "refs/remotes/origin/main", self.git("rev-parse", "HEAD")) + self.card = self.repo / ".task/main.md" + self.card.parent.mkdir() + self.card.write_text("# Test task\n") + self.env = os.environ.copy() + for key in ("CMD", "SRC", "RECEIPT", "IMPORT_RECEIPT", "VALIDATE_RECEIPT", "TAIL"): + self.env.pop(key, None) + self.env["AGENT_EVIDENCE_DIR"] = str(self.base / "evidence") + + def git(self, *args): + return subprocess.check_output(["git", *args], cwd=self.repo, text=True, + stderr=subprocess.DEVNULL).strip() + + def record(self, command="printf 'checked\\n'"): + result = subprocess.run(["bash", str(SCRIPTS / "evidence.sh")], cwd=self.repo, + env=self.env | {"CMD": command}, capture_output=True, text=True) + self.assertEqual(result.returncode, 0, result.stderr) + text = self.card.read_text() + block = text.split("```json\n", 1)[1].split("\n```", 1)[0] + return json.loads(block) + + def check(self): + result = subprocess.run(["python3", str(SCRIPTS / "check-evidence.py"), str(self.card)], + cwd=self.repo, capture_output=True, text=True) + return result.stdout + result.stderr + + def replace_block(self, value): + prefix, rest = self.card.read_text().split("```json\n", 1) + suffix = rest.split("\n```", 1)[1] + self.card.write_text(prefix + "```json\n" + json.dumps(value, indent=2) + "\n```" + suffix) + + def advance(self): + (self.repo / "tracked").write_text("changed\n") + self.git("commit", "-qam", "Change code") + + def test_fresh_summary_passes(self): + summary = self.record() + self.assertNotIn("submodules", self.card.read_text()) + self.assertEqual(self.check(), "") + self.assertTrue(Path(summary["receipt_path"]).is_file()) + + def test_summary_detects_stale_head_through_full_receipt(self): + self.record() + self.advance() + output = self.check() + self.assertIn("recorded HEAD predates non-card changes", output) + self.assertNotIn("unverified", output) + + def test_tampered_summary_warns(self): + summary = self.record() + for field, value in (("command", "make test"), ("exit_code", 1), ("output_tail", ["forged"]), + ("before_head", "0" * 40)): + with self.subTest(field=field): + self.replace_block(dict(summary, **{field: value})) + self.assertIn("unverified: summary does not match full receipt", self.check()) + + def test_malformed_summary_fields_warn(self): + summary = self.record() + for field, value in (("receipt_sha256", "zz"), ("exit_code", True), ("output_tail", "text"), + ("receipt_path", "relative.json")): + with self.subTest(field=field): + self.replace_block(dict(summary, **{field: value})) + self.assertIn("unverified", self.check()) + missing = dict(summary) + del missing["command"] + self.replace_block(missing) + self.assertIn("unverified: invalid summary fields", self.check()) + + def test_tampered_full_receipt_warns(self): + summary = self.record() + path = Path(summary["receipt_path"]) + full = json.loads(path.read_text()) + full["output"] = "changed\n" + path.write_text(json.dumps(full)) + self.assertIn("unverified: receipt checksum mismatch", self.check()) + + def test_missing_full_receipt_warns_and_still_checks_summary_heads(self): + summary = self.record() + path = Path(summary["receipt_path"]) + path.unlink() + output = self.check() + self.assertIn(f"unverified: full receipt not available locally ({path})", output) + self.assertNotIn("predates", output) + self.advance() + self.assertIn("recorded HEAD predates non-card changes", self.check()) + + def test_legacy_full_receipt_block_is_still_accepted(self): + summary = self.record() + full = json.loads(Path(summary["receipt_path"]).read_text()) + self.replace_block(full) + self.assertEqual(self.check(), "") + full["output"] = "changed\n" + self.replace_block(full) + self.assertIn("unverified: receipt checksum mismatch", self.check()) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_evidence.py b/tests/test_evidence.py index 520d4a27..01fc018b 100644 --- a/tests/test_evidence.py +++ b/tests/test_evidence.py @@ -57,6 +57,7 @@ class EvidenceTests(unittest.TestCase): self.head = self.git("rev-parse", "HEAD") self.git("update-ref", "refs/remotes/origin/main", self.head) self.receipt = self.base / "receipt.json" + self.evidence_dir = self.base / "evidence" self.card = self.repo / ".task/main.md" self.card.parent.mkdir() self.card.write_text("# Test task\n") @@ -71,6 +72,7 @@ class EvidenceTests(unittest.TestCase): for key in ("CMD", "SRC", "RECEIPT", "IMPORT_RECEIPT", "VALIDATE_RECEIPT", "VALIDATE_RECEIPT_PURPOSE", "TASK_WORK_MODE"): env.pop(key, None) + env["AGENT_EVIDENCE_DIR"] = str(self.evidence_dir) if command is not None: env["CMD"] = command env.update({key: str(value) for key, value in values.items()}) @@ -337,6 +339,58 @@ class EvidenceTests(unittest.TestCase): recorded = self.card.read_text() self.assertEqual(self.import_receipt().returncode, 0) self.assertEqual(self.card.read_text(), recorded) + self.assertEqual(recorded.count("```json"), 1) + + def card_summaries(self): + blocks = self.card.read_text().split("```json\n")[1:] + return [json.loads(block.split("\n```", 1)[0]) for block in blocks] + + def test_card_holds_summary_and_full_receipt_is_stored_outside_repo(self): + result = self.run_evidence("seq 1 30") + self.assertEqual(result.returncode, 0, result.stderr) + text = self.card.read_text() + self.assertIn("