File size: 3,706 Bytes
01a891a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 | #!/usr/bin/env python3
"""Run conservative validation gates against a candidate copy.
Every command must explicitly consume {candidate_file}, {candidate_dir}, or a
RE_AGENT_CANDIDATE_* environment variable. Passing commands remain UNKNOWN
unless --trust-commands attests that they meaningfully validate the candidate.
Exit codes: 0 PASS, 1 FAIL, 2 UNKNOWN.
"""
from __future__ import annotations
import argparse
import os
import re
import subprocess
import sys
from pathlib import Path
MARKERS = (
"{candidate_file}", "{candidate_dir}",
"$RE_AGENT_CANDIDATE_FILE", "${RE_AGENT_CANDIDATE_FILE}",
"$RE_AGENT_CANDIDATE_DIR", "${RE_AGENT_CANDIDATE_DIR}",
"%RE_AGENT_CANDIDATE_FILE%", "%RE_AGENT_CANDIDATE_DIR%",
"$env:RE_AGENT_CANDIDATE_FILE", "$env:RE_AGENT_CANDIDATE_DIR",
)
def safe_component(value: str) -> str:
"""Sanitize a generated host-filesystem path component."""
return re.sub(r"[^A-Za-z0-9_.-]", "_", value) or "candidate"
def consumes_candidate(command: str) -> bool:
return any(marker in command for marker in MARKERS)
def run(candidate: Path, commands: list[str], timeout: int,
trust_commands: bool) -> tuple[str, str, list[str]]:
if not candidate.is_file():
return "FAIL", f"Candidate file does not exist: {candidate}", []
if not commands:
return "UNKNOWN", "No candidate validation commands configured", []
unsafe = [command for command in commands if not consumes_candidate(command)]
if unsafe:
return ("FAIL", "Every command must explicitly consume the candidate",
[f"does not consume candidate: {command}" for command in unsafe])
candidate = candidate.resolve()
env = os.environ.copy()
env["RE_AGENT_CANDIDATE_FILE"] = str(candidate)
env["RE_AGENT_CANDIDATE_DIR"] = str(candidate.parent)
findings: list[str] = []
for command in commands:
rendered = (command.replace("{candidate_file}", str(candidate))
.replace("{candidate_dir}", str(candidate.parent)))
try:
proc = subprocess.run(
rendered, cwd=candidate.parent, env=env, shell=True,
stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
text=True, timeout=timeout, check=False)
except subprocess.TimeoutExpired:
return "FAIL", f"Validation command timed out after {timeout}s", findings
tail = "\n".join(proc.stdout.splitlines()[-20:])
findings.append(f"{command} -> exit {proc.returncode}\n{tail}".rstrip())
if proc.returncode != 0:
return "FAIL", "Candidate validation command failed", findings
if not trust_commands:
return ("UNKNOWN", "Commands passed but are not trusted as meaningful proof; "
"review them before using --trust-commands", findings)
return "PASS", "All trusted candidate validation commands passed", findings
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--candidate", required=True, type=Path)
parser.add_argument("--command", action="append", default=[],
help="Repeat for each ordered validation command")
parser.add_argument("--timeout", type=int, default=900)
parser.add_argument("--trust-commands", action="store_true")
args = parser.parse_args()
verdict, summary, findings = run(
args.candidate, args.command, max(1, args.timeout), args.trust_commands)
print(f"VALIDATION: {verdict}")
print(f"SUMMARY: {summary}")
for finding in findings:
print(f" - {finding}")
return {"PASS": 0, "FAIL": 1, "UNKNOWN": 2}[verdict]
if __name__ == "__main__":
sys.exit(main())
|