| |
| """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()) |
|
|
