weichen
b84667a697
Why: independent review found shared submodule metadata risk, quoted header leakage, stale JSON receipt blindness and missing abandon coverage. Evidence: SELF 92 governance regressions pass; the original review receipt and replay output are recorded in the task card. Impact: preserves external component worktrees, handles both evidence formats and adds targeted failure-path coverage; no product behavior changes. Refs: .task/feature-governance-reliability.md; .task/reviews/governance-reliability.json
319 lines
15 KiB
Python
319 lines
15 KiB
Python
#!/usr/bin/env python3
|
||
"""Record local command evidence without upgrading its trust beyond SELF.
|
||
|
||
Receipts preserve sanitized full output and the exact command when it contains no
|
||
credentials. A hash of the original command remains when redaction is necessary.
|
||
Checksums detect accidental modification, not forgery by a local file's author.
|
||
"""
|
||
import datetime
|
||
import fcntl
|
||
import hashlib
|
||
import json
|
||
import os
|
||
from pathlib import Path
|
||
import re
|
||
import subprocess
|
||
import sys
|
||
|
||
|
||
SCHEMA = 1
|
||
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")
|
||
|
||
|
||
class EvidenceError(Exception):
|
||
pass
|
||
|
||
|
||
def digest(value):
|
||
return hashlib.sha256(value.encode() if isinstance(value, str) else value).hexdigest()
|
||
|
||
|
||
def canonical(value):
|
||
return json.dumps(value, sort_keys=True, ensure_ascii=False, separators=(",", ":"))
|
||
|
||
|
||
def sanitize(value):
|
||
"""Redact known secret values and conventional credential syntax everywhere."""
|
||
for name, secret in sorted(os.environ.items(), key=lambda item: -len(item[1])):
|
||
if SECRET_NAME.search(name) and len(secret) >= 4:
|
||
value = value.replace(secret, "<redacted>")
|
||
value = re.sub(r"-----BEGIN [^-\r\n]*PRIVATE KEY-----.*?-----END [^-\r\n]*PRIVATE KEY-----",
|
||
"<redacted-private-key>", value, flags=re.S)
|
||
value = re.sub(r"(?i)(\bAuthorization['\"]?[ \t]*[=:][ \t]*['\"]?"
|
||
r"(?:Bearer|token|Basic)[ \t]+)[^\s'\";]+", r"\1<redacted>", value)
|
||
value = re.sub(r"(?i)\b(https?|ssh)://[^\s/@:]+:[^\s/@]+@", r"\1://<redacted>@", value)
|
||
credential = r"[\w-]*(?:token|password|passwd|secret|api[_-]?key|credential)[\w-]*"
|
||
prefix = r"(?:\b" + credential + r"['\"]?[ \t]*[=:][ \t]*|--" + credential + r"[ \t]+)"
|
||
value = re.sub(r"(?i)(" + prefix + r")(['\"])(.*?)\2",
|
||
r"\1\2<redacted>\2", value, flags=re.S)
|
||
value = re.sub(r"(?i)(--" + credential + r"[ \t]+)(?!<redacted>)[^\s'\";]+",
|
||
r"\1<redacted>", value)
|
||
value = re.sub(
|
||
r"(?i)(\b[\w-]*(?:token|password|passwd|secret|api[_-]?key|credential)[\w-]*"
|
||
r"['\"]?[ \t]*[=:][ \t]*['\"]?)(?!<redacted>)[^\s'\"&;,}]+",
|
||
r"\1<redacted>", value)
|
||
return value
|
||
|
||
|
||
def sanitize_tree(value):
|
||
if isinstance(value, str):
|
||
return sanitize(value)
|
||
if isinstance(value, list):
|
||
return [sanitize_tree(item) for item in value]
|
||
if isinstance(value, dict):
|
||
return {sanitize(key): sanitize_tree(item) for key, item in value.items()}
|
||
return value
|
||
|
||
|
||
def git(*args, required=True):
|
||
result = subprocess.run(["git", *args], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
|
||
if result.returncode and required:
|
||
raise EvidenceError("Git object could not be recorded: " + " ".join(args[:2]))
|
||
return result.stdout if result.returncode == 0 else b""
|
||
|
||
|
||
def git_text(*args, **kwargs):
|
||
return git(*args, **kwargs).decode("utf-8", "replace").strip()
|
||
|
||
|
||
def common_dir():
|
||
return str(Path(git_text("rev-parse", "--git-common-dir")).resolve())
|
||
|
||
|
||
def snapshot():
|
||
untracked = {}
|
||
root = Path(git_text("rev-parse", "--show-toplevel"))
|
||
for raw in git("-C", str(root), "ls-files", "--full-name", "--others", "--exclude-standard", "-z").split(b"\0"):
|
||
if not raw:
|
||
continue
|
||
name = raw.decode("utf-8", "replace")
|
||
path = root / name
|
||
if path.is_symlink():
|
||
untracked[name] = digest(os.fsencode(os.readlink(path)))
|
||
elif path.is_file():
|
||
with path.open("rb") as stream:
|
||
untracked[name] = hashlib.file_digest(stream, "sha256").hexdigest()
|
||
else:
|
||
untracked[name] = None
|
||
target = os.environ.get("TASK_TARGET_BRANCH", "main")
|
||
return {
|
||
"head": git_text("rev-parse", "HEAD"),
|
||
"branch": git_text("rev-parse", "--abbrev-ref", "HEAD"),
|
||
"target_branch": target,
|
||
"target_head": git_text("rev-parse", "--verify", "refs/remotes/origin/" + target,
|
||
required=False) or None,
|
||
"status": git("status", "--porcelain=v1", "--untracked-files=all").decode("utf-8", "replace"),
|
||
"tracked_diff_sha256": digest(git("diff", "--binary", "HEAD", "--")),
|
||
"untracked": untracked,
|
||
"submodules": git("submodule", "status", "--recursive").decode("utf-8", "replace"),
|
||
}
|
||
|
||
|
||
def create_receipt(command, output, exit_code, before, after, *, timestamp=None,
|
||
cwd=None, git_common_dir=None, after_error=None, capture_note=None):
|
||
"""Package an actual captured run; callers must not reconstruct missing output."""
|
||
receipt = sanitize_tree({
|
||
"schema": SCHEMA, "source": "SELF",
|
||
"timestamp": timestamp or datetime.datetime.now(datetime.timezone.utc).isoformat(),
|
||
"command": command, "command_sha256": digest(command),
|
||
"cwd": cwd or str(Path.cwd().resolve()),
|
||
"common_dir": git_common_dir or common_dir(),
|
||
"before": before, "after": after, "after_error": after_error,
|
||
"exit_code": exit_code, "output": output,
|
||
})
|
||
if capture_note is not None:
|
||
receipt["capture_note"] = sanitize(capture_note)
|
||
receipt["output_sha256"] = digest(receipt["output"])
|
||
receipt["receipt_sha256"] = digest(canonical(receipt))
|
||
return receipt
|
||
|
||
|
||
def validate_snapshot(value):
|
||
if not isinstance(value, dict):
|
||
raise EvidenceError("Invalid receipt object snapshot")
|
||
expected = {"head", "branch", "target_branch", "target_head", "status",
|
||
"tracked_diff_sha256", "submodules", "untracked"}
|
||
if set(value) != expected:
|
||
raise EvidenceError("Invalid receipt object fields")
|
||
for key in ("head", "branch", "target_branch", "status", "tracked_diff_sha256", "submodules"):
|
||
if not isinstance(value.get(key), str):
|
||
raise EvidenceError("Invalid receipt object field: " + key)
|
||
if not GIT_SHA.fullmatch(value["head"]) or not SHA256.fullmatch(value["tracked_diff_sha256"]):
|
||
raise EvidenceError("Invalid receipt object digest")
|
||
if value.get("target_head") is not None and (
|
||
not isinstance(value["target_head"], str) or not GIT_SHA.fullmatch(value["target_head"])):
|
||
raise EvidenceError("Invalid receipt target identity")
|
||
if not isinstance(value.get("untracked"), dict):
|
||
raise EvidenceError("Invalid receipt untracked object")
|
||
for name, checksum in value["untracked"].items():
|
||
if not isinstance(name, str) or (checksum is not None and (
|
||
not isinstance(checksum, str) or not SHA256.fullmatch(checksum))):
|
||
raise EvidenceError("Invalid receipt untracked digest")
|
||
|
||
|
||
def read_receipt(filename):
|
||
if not Path(filename).is_absolute():
|
||
raise EvidenceError("Receipt path must be absolute")
|
||
value = json.loads(Path(filename).read_text())
|
||
if not isinstance(value, dict) or type(value.get("schema")) is not int or value["schema"] != SCHEMA:
|
||
raise EvidenceError("Unsupported receipt schema")
|
||
expected = {"schema", "source", "timestamp", "command", "command_sha256", "cwd", "common_dir",
|
||
"before", "after", "after_error", "exit_code", "output", "output_sha256", "receipt_sha256"}
|
||
if set(value) not in (expected, expected | {"capture_note"}):
|
||
raise EvidenceError("Invalid receipt fields")
|
||
for key in ("source", "timestamp", "command", "command_sha256", "cwd", "common_dir",
|
||
"output", "output_sha256", "receipt_sha256"):
|
||
if not isinstance(value.get(key), str):
|
||
raise EvidenceError("Invalid receipt field: " + key)
|
||
if value["source"] != "SELF" or type(value.get("exit_code")) is not int or not 0 <= value["exit_code"] <= 255:
|
||
raise EvidenceError("Invalid receipt source or exit code")
|
||
datetime.datetime.fromisoformat(value["timestamp"].replace("Z", "+00:00"))
|
||
for key in ("cwd", "common_dir"):
|
||
if not Path(value[key]).is_absolute():
|
||
raise EvidenceError("Receipt identity paths must be absolute")
|
||
for key in ("command_sha256", "output_sha256", "receipt_sha256"):
|
||
if not SHA256.fullmatch(value[key]):
|
||
raise EvidenceError("Invalid receipt digest: " + key)
|
||
validate_snapshot(value.get("before"))
|
||
if value.get("after") is not None:
|
||
validate_snapshot(value["after"])
|
||
elif not isinstance(value.get("after_error"), str) or not value["after_error"]:
|
||
raise EvidenceError("Missing after object or capture error")
|
||
if value.get("after_error") is not None and not isinstance(value["after_error"], str):
|
||
raise EvidenceError("Invalid after capture error")
|
||
if "capture_note" in value and not isinstance(value["capture_note"], str):
|
||
raise EvidenceError("Invalid capture note")
|
||
if digest(value["output"]) != value["output_sha256"]:
|
||
raise EvidenceError("Receipt output checksum mismatch")
|
||
payload = {key: item for key, item in value.items() if key != "receipt_sha256"}
|
||
if digest(canonical(payload)) != value["receipt_sha256"]:
|
||
raise EvidenceError("Receipt checksum mismatch")
|
||
if value != sanitize_tree(value):
|
||
raise EvidenceError("Receipt contains unredacted credential syntax")
|
||
if Path(value["common_dir"]).resolve() != Path(common_dir()):
|
||
raise EvidenceError("Receipt belongs to a different Git common directory")
|
||
return value
|
||
|
||
|
||
def validate_precheck(receipt):
|
||
after = receipt["after"]
|
||
current = snapshot()
|
||
if receipt["command"] not in ("make precheck", "bash scripts/task.sh precheck"):
|
||
raise EvidenceError("Task start requires a precheck receipt")
|
||
if receipt["exit_code"] != 0 or not re.search(r"^\[verdict\] READY(?:\s|$)", receipt["output"], re.M):
|
||
raise EvidenceError("Precheck receipt is not READY")
|
||
if after is None or receipt["after_error"] is not None:
|
||
raise EvidenceError("Precheck object capture is incomplete")
|
||
if receipt["before"]["head"] != current["head"] or after["head"] != current["head"]:
|
||
raise EvidenceError("Precheck source HEAD has changed")
|
||
if after["target_branch"] != current["target_branch"] or not after["target_head"] or after["target_head"] != current["target_head"]:
|
||
raise EvidenceError("Precheck authoritative target has changed")
|
||
|
||
|
||
def find_card():
|
||
root = Path(git_text("rev-parse", "--show-toplevel"))
|
||
branch = git_text("rev-parse", "--abbrev-ref", "HEAD")
|
||
card = root / ".task" / (branch.replace("/", "-") + ".md")
|
||
if card.is_file():
|
||
return card
|
||
for candidate in (root / ".task").glob("*.md"):
|
||
if re.search(r"^- 工作区：.*分支：" + re.escape(branch) + r"(?:\s|$)", candidate.read_text(), re.M):
|
||
return candidate
|
||
raise EvidenceError("No task card exists for the current branch")
|
||
|
||
|
||
def append_card(card, receipt):
|
||
encoded = json.dumps(receipt, ensure_ascii=False, indent=2)
|
||
block = "\n## 证据（Evidence）\n\n" + (
|
||
"<!-- Local receipt: SELF. Digests detect corruption; they do not authenticate a reviewer or CI. -->\n"
|
||
+ "```json\n" + encoded + "\n```\n")
|
||
with card.open("r+", encoding="utf-8") as stream:
|
||
fcntl.flock(stream, fcntl.LOCK_EX)
|
||
existing = stream.read()
|
||
if receipt["receipt_sha256"] in existing:
|
||
return
|
||
if "## 证据（Evidence）" in existing:
|
||
block = block.replace("\n## 证据（Evidence）\n\n", "\n", 1)
|
||
stream.write(block)
|
||
stream.flush()
|
||
os.fsync(stream.fileno())
|
||
|
||
|
||
def write_receipt(path, receipt):
|
||
with path.open("x", encoding="utf-8") as stream:
|
||
os.chmod(path, 0o600)
|
||
json.dump(receipt, stream, ensure_ascii=False, indent=2)
|
||
stream.write("\n")
|
||
stream.flush()
|
||
os.fsync(stream.fileno())
|
||
|
||
|
||
def report(receipt, destination):
|
||
print(sanitize("[evidence] " + destination + " exit=" + str(receipt["exit_code"])
|
||
+ " src=SELF before=" + receipt["before"]["head"]
|
||
+ " receipt_sha256=" + receipt["receipt_sha256"]))
|
||
|
||
|
||
def main():
|
||
child_exit = None
|
||
try:
|
||
if os.environ.get("SRC", "SELF") not in ("", "SELF"):
|
||
raise EvidenceError("Local evidence is SELF; GATE/REPLAY require external execution identity")
|
||
capture, imported, validated = (os.environ.get(key) for key in
|
||
("RECEIPT", "IMPORT_RECEIPT", "VALIDATE_RECEIPT"))
|
||
command = os.environ.get("CMD", "")
|
||
if sum(bool(item) for item in (capture, imported, validated)) > 1 or ((imported or validated) and command):
|
||
raise EvidenceError("Choose one capture/import/validate operation")
|
||
if imported or validated:
|
||
receipt = read_receipt(imported or validated)
|
||
if validated:
|
||
validate_precheck(receipt)
|
||
report(receipt, "precheck validated")
|
||
else:
|
||
card = find_card()
|
||
append_card(card, receipt)
|
||
report(receipt, "card=" + str(card))
|
||
return receipt["exit_code"]
|
||
if not command or "\n" in command or "\r" in command or "\0" in command:
|
||
raise EvidenceError("CMD must be a nonempty single-line command")
|
||
path = Path(capture) if capture else None
|
||
if path is not None and (not path.is_absolute() or path.exists() or path.is_symlink()):
|
||
raise EvidenceError("Capture requires an absolute, unused receipt path")
|
||
card = find_card() if path is None else None
|
||
before = snapshot()
|
||
source_cwd, source_common = str(Path.cwd().resolve()), common_dir()
|
||
timestamp = datetime.datetime.now(datetime.timezone.utc).isoformat()
|
||
child_env = os.environ.copy()
|
||
for key in ("CMD", "SRC", "RECEIPT", "IMPORT_RECEIPT", "VALIDATE_RECEIPT", "TAIL"):
|
||
child_env.pop(key, None)
|
||
result = subprocess.run(["bash", "-o", "pipefail", "-c", command],
|
||
stdout=subprocess.PIPE, stderr=subprocess.STDOUT, env=child_env)
|
||
child_exit = result.returncode if result.returncode >= 0 else 128 - result.returncode
|
||
after, after_error = None, None
|
||
try:
|
||
after = snapshot()
|
||
except (EvidenceError, OSError) as error:
|
||
after_error = str(error)
|
||
receipt = create_receipt(command, result.stdout.decode("utf-8", "replace"), child_exit,
|
||
before, after, timestamp=timestamp, cwd=source_cwd,
|
||
git_common_dir=source_common, after_error=after_error)
|
||
if path is not None:
|
||
write_receipt(path, receipt)
|
||
report(receipt, "receipt=" + str(path))
|
||
else:
|
||
append_card(card, receipt)
|
||
report(receipt, "card=" + str(card))
|
||
print(receipt["output"], end="" if receipt["output"].endswith("\n") else "\n")
|
||
if after_error:
|
||
print("[evidence] after object unavailable: " + sanitize(after_error), file=sys.stderr)
|
||
return child_exit
|
||
except (EvidenceError, OSError, ValueError, TypeError) as error:
|
||
print("[evidence] ERROR: " + sanitize(str(error)), file=sys.stderr)
|
||
return child_exit if child_exit else 2
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|