138 lines
6.2 KiB
Python
138 lines
6.2 KiB
Python
#!/usr/bin/env python3
|
|
"""Warn when a task card's recorded objects no longer support current checks.
|
|
|
|
Receipt commands and output are data only. This is a freshness check, not an
|
|
upgrade of local SELF evidence to reviewer or CI evidence.
|
|
"""
|
|
import json
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
import re
|
|
import shlex
|
|
import subprocess
|
|
import sys
|
|
|
|
# Reading a receipt must not create ignored files that prevent task-finish.
|
|
sys.dont_write_bytecode = True
|
|
import evidence
|
|
|
|
|
|
def warn(label, reason):
|
|
print(f"[evidence] WARN {label}: {reason} ——关键检查须重跑并以 make evidence 重新写入")
|
|
|
|
|
|
def check_head(head):
|
|
if not isinstance(head, str) or not re.fullmatch(r"[0-9a-f]{4,64}", head):
|
|
raise ValueError("invalid recorded HEAD")
|
|
resolved = subprocess.run(["git", "rev-parse", "--verify", head + "^{commit}"],
|
|
stdout=subprocess.PIPE, stderr=subprocess.PIPE)
|
|
if resolved.returncode:
|
|
raise ValueError("recorded HEAD is unavailable or ambiguous")
|
|
result = subprocess.run(["git", "diff", "--quiet", "--no-ext-diff",
|
|
resolved.stdout.decode().strip(), "HEAD", "--", ".", ":(exclude).task"],
|
|
stdout=subprocess.PIPE, stderr=subprocess.PIPE)
|
|
if result.returncode not in (0, 1):
|
|
raise ValueError("recorded HEAD could not be compared")
|
|
return result.returncode == 1
|
|
|
|
|
|
def card_path(path):
|
|
return path == ".task" or path.startswith(".task/")
|
|
|
|
|
|
def check_snapshot(snapshot):
|
|
evidence.validate_snapshot(snapshot)
|
|
stale = check_head(snapshot["head"])
|
|
for line in snapshot["status"].splitlines():
|
|
if len(line) < 4 or line[2] != " ":
|
|
raise ValueError("invalid recorded worktree status")
|
|
# Parse quoting as text, never as a shell command. Renames must have
|
|
# both paths inside .task to qualify as metadata-only changes.
|
|
paths = shlex.split(line[3:]) if line[3:].startswith('"') else [line[3:]]
|
|
paths = [part for path in paths for part in path.split(" -> ") if part != "->"]
|
|
if not paths or not all(card_path(path) for path in paths):
|
|
raise ValueError("recorded object includes uncommitted non-card changes")
|
|
if any(not card_path(path) for path in snapshot["untracked"]):
|
|
raise ValueError("recorded object includes untracked non-card files")
|
|
if any(not line.startswith(" ") for line in snapshot["submodules"].splitlines()):
|
|
raise ValueError("recorded component state was not at its pinned SHA")
|
|
if not snapshot["status"] and snapshot["tracked_diff_sha256"] != evidence.digest(b""):
|
|
raise ValueError("recorded diff cannot be reconstructed from HEAD")
|
|
return stale
|
|
|
|
|
|
def check_receipt(value):
|
|
if not isinstance(value, dict) or type(value.get("schema")) is not int or value["schema"] != evidence.SCHEMA:
|
|
raise ValueError("unsupported or malformed receipt")
|
|
if not evidence.receipt_keys_ok(value):
|
|
raise ValueError("invalid receipt fields")
|
|
for key in evidence.TARGET_FIELDS:
|
|
if key in value and not evidence.valid_branch(value[key]):
|
|
raise ValueError("invalid receipt target: " + key)
|
|
for key in ("source", "timestamp", "command", "cwd", "common_dir", "output"):
|
|
if not isinstance(value[key], str):
|
|
raise ValueError("invalid receipt field: " + key)
|
|
for key in ("command_sha256", "output_sha256", "receipt_sha256"):
|
|
if not isinstance(value[key], str) or not evidence.SHA256.fullmatch(value[key]):
|
|
raise ValueError("invalid receipt digest: " + key)
|
|
if value["source"] != "SELF" or type(value["exit_code"]) is not int or not 0 <= value["exit_code"] <= 255:
|
|
raise ValueError("invalid receipt source or exit code")
|
|
if any(not Path(value[key]).is_absolute() for key in ("cwd", "common_dir")):
|
|
raise ValueError("invalid receipt identity paths")
|
|
if "capture_note" in value and not isinstance(value["capture_note"], str):
|
|
raise ValueError("invalid capture note")
|
|
try:
|
|
datetime.fromisoformat(value["timestamp"].replace("Z", "+00:00"))
|
|
except ValueError:
|
|
raise ValueError("invalid receipt timestamp") from None
|
|
payload = {key: item for key, item in value.items() if key != "receipt_sha256"}
|
|
if evidence.digest(evidence.canonical(payload)) != value.get("receipt_sha256"):
|
|
raise ValueError("receipt checksum mismatch")
|
|
if not isinstance(value.get("output"), str) or evidence.digest(value["output"]) != value.get("output_sha256"):
|
|
raise ValueError("receipt output checksum mismatch")
|
|
if value.get("after") is None or value.get("after_error") is not None:
|
|
raise ValueError("receipt has no complete after object")
|
|
before_stale = check_snapshot(value.get("before"))
|
|
after_stale = check_snapshot(value["after"])
|
|
return before_stale or after_stale
|
|
|
|
|
|
def records(content):
|
|
"""Yield fenced JSON and timestamped legacy rows without interpreting text."""
|
|
block = None
|
|
start = None
|
|
for number, line in enumerate(content.splitlines(), 1):
|
|
if block is not None:
|
|
if line.strip() == "```":
|
|
yield f"JSON receipt line {start}", "json", "\n".join(block)
|
|
block = None
|
|
else:
|
|
block.append(line)
|
|
elif line.strip() == "```json":
|
|
block, start = [], number
|
|
elif re.match(r"^\|\s*\d{4}-\d{2}-\d{2}T[^|]*\|", line):
|
|
fields = re.split(r"(?<!\\)\|", line)
|
|
yield f"legacy receipt line {number}", "legacy", fields[3].strip(" `") if len(fields) > 3 else ""
|
|
if block is not None:
|
|
yield f"JSON receipt line {start}", "json", ""
|
|
|
|
|
|
def main():
|
|
try:
|
|
content = Path(sys.argv[1]).read_text(encoding="utf-8")
|
|
except (IndexError, OSError, UnicodeError):
|
|
warn("task card", "unverified: evidence could not be read")
|
|
return
|
|
for label, kind, record in records(content):
|
|
try:
|
|
stale = check_receipt(json.loads(record)) if kind == "json" else check_head(record)
|
|
except (ValueError, TypeError, KeyError, OSError, evidence.EvidenceError) as exc:
|
|
warn(label, "unverified: " + str(exc))
|
|
continue
|
|
if stale:
|
|
warn(label, "recorded HEAD predates non-card changes")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|