From 93daa17678fc892f8aa1c269eee42b3c164adc45 Mon Sep 17 00:00:00 2001 From: neo <397763@gmail.com> Date: Mon, 29 Jun 2026 21:46:26 +0500 Subject: [PATCH] Add scripts/preflight.py --- scripts/preflight.py | 203 +++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 203 insertions(+) create mode 100644 scripts/preflight.py diff --git a/scripts/preflight.py b/scripts/preflight.py new file mode 100644 index 0000000..a7597ef --- /dev/null +++ b/scripts/preflight.py @@ -0,0 +1,203 @@ +#!/usr/bin/env python3 +"""Agent Poka-Yoke: Pre-flight check utility for Hermes Agent operations.""" + +import json +import re +import sys +from dataclasses import dataclass, field +from typing import Optional + + +# ─── GPU patterns (absolute blocklist) ────────────────────────────────────── +GPU_PATTERNS = [ + r'(?i)nvidia', r'(?i)cuda', r'(?i)nvcc', r'(?i)vram', r'(?i)gpu', + r'(?i)nvidia-smi', r'(?i)nccl', r'(?i)cudnn', r'(?i)torch\.cuda', + r'(?i)tf\.device', r'(?i)tensorflow.*gpu', r'(?i)mxnet.*gpu', + r'(?i)onnx.*cuda', r'(?i)opencl', r'(?i)rocm', +] +GPU_COMPILED = [re.compile(p) for p in GPU_PATTERNS] + +# ─── Dangerous command patterns ───────────────────────────────────────────── +DANGEROUS_PATTERNS = [ + (r'rm\s+-rf\s+/', 'rm -rf on root filesystem — IMMEDIATELY DANGEROUS'), + (r'rm\s+-rf\s+\.', 'rm -rf on current directory — DESTRUCTIVE'), + (r'>\s*/dev/sd[a-z]', 'Disk overwrite via dd/redirect — DATA LOSS'), + (r'dd\s+if=.*of=/dev/', 'Raw disk write via dd — DATA LOSS'), + (r'sudo\s+.*\brm\s+-rf\b', 'sudo rm -rf — DESTRUCTIVE WITH PRIVILEGES'), + (r'chmod\s+777\b', 'chmod 777 — security risk (too permissive)'), + (r'chown\s+root', 'chown root — privilege escalation'), +] +DANGEROUS_COMPILED = [(re.compile(p), msg) for p, msg in DANGEROUS_PATTERNS] + + +@dataclass +class CheckResult: + passed: bool + layer: str + check: str + message: str + severity: str = "info" # info, warning, block + + def to_dict(self): + return { + "passed": self.passed, + "layer": self.layer, + "check": self.check, + "message": self.message, + "severity": self.severity, + } + + +@dataclass +class PreflightReport: + results: list = field(default_factory=list) + + @property + def all_passed(self) -> bool: + return all(r.passed for r in self.results) + + @property + def blocked(self) -> bool: + return any(r.severity == "block" and not r.passed for r in self.results) + + def add(self, result: CheckResult): + self.results.append(result) + + def to_dict(self) -> dict: + return { + "all_passed": self.all_passed, + "blocked": self.blocked, + "total_checks": len(self.results), + "checks": [r.to_dict() for r in self.results], + } + + def summary(self) -> str: + lines = [f"### Poka-Yoke Pre-flight Report\n"] + for r in self.results: + icon = "✅" if r.passed else "❌" + if r.severity == "block" and not r.passed: + icon = "🛑" + elif r.severity == "warning": + icon = "⚠️" + lines.append(f"- {icon} **[{r.layer}]** {r.check}: {r.message}") + lines.append("") + status = "PASSED" if self.all_passed else "BLOCKED" + lines.append(f"**Status: {status}**") + return "\n".join(lines) + + +def check_gpu(command: str) -> CheckResult: + """Layer 2b: GPU guardrail — absolute block.""" + for pattern in GPU_COMPILED: + if pattern.search(command): + return CheckResult( + passed=False, + layer="GPU", + check=f"GPU pattern detected: {pattern.pattern}", + message=f"GPU operations are prohibited by iron rule. Command matches '{pattern.pattern}'", + severity="block", + ) + return CheckResult( + passed=True, + layer="GPU", + check="No GPU patterns detected", + message="Safe — no GPU access patterns found", + ) + + +def check_dangerous(command: str) -> CheckResult: + """Layer 2d: Dangerous command detection.""" + for pattern, msg in DANGEROUS_COMPILED: + if pattern.search(command): + return CheckResult( + passed=False, + layer="Dangerous Ops", + check=f"Dangerous pattern: {pattern.pattern}", + message=msg, + severity="block", + ) + return CheckResult( + passed=True, + layer="Dangerous Ops", + check="No dangerous patterns", + message="Safe — no destructive patterns detected", + severity="warning", + ) + + +def check_path_traversal(command: str) -> CheckResult: + """Check for path traversal attempts.""" + if '../' in command or '..\\' in command: + return CheckResult( + passed=False, + layer="Schema", + check="Path traversal detected", + message="Command contains ../ or ..\\ — potential path traversal", + severity="block", + ) + return CheckResult( + passed=True, + layer="Schema", + check="No path traversal", + message="Safe — no path traversal detected", + ) + + +def check_api_access(fqdn: str) -> CheckResult: + """Layer 2a: API access verification (known hosts).""" + try: + host = fqdn.replace('http://', '').replace('https://', '').rstrip('/') + known_hosts = ['192.168.1.230:8000'] + for known in known_hosts: + if known in host: + return CheckResult( + passed=True, + layer="Policy", + check=f"Known infrastructure host: {host}", + message=f"Verified as trusted host: {host}", + ) + return CheckResult( + passed=True, + layer="Policy", + check=f"External host: {host}", + message=f"Warning: {host} is not in known infrastructure. Verify before use.", + severity="warning", + ) + except Exception as e: + return CheckResult( + passed=False, + layer="Policy", + check="Host verification failed", + message=f"Could not verify host: {str(e)}", + severity="block", + ) + + +def run_preflight(command: str, fqdn: Optional[str] = None) -> PreflightReport: + """Run all pre-flight checks on a command.""" + report = PreflightReport() + report.add(check_gpu(command)) + report.add(check_dangerous(command)) + report.add(check_path_traversal(command)) + if fqdn: + report.add(check_api_access(fqdn)) + return report + + +if __name__ == "__main__": + import argparse + + parser = argparse.ArgumentParser(description="Agent Poka-Yoke Pre-flight Check") + parser.add_argument("command", help="Command to check") + parser.add_argument("--fqdn", help="API endpoint to verify") + parser.add_argument("--json", action="store_true", help="Output JSON") + args = parser.parse_args() + + report = run_preflight(args.command, args.fqdn) + + if args.json: + print(json.dumps(report.to_dict(), indent=2)) + else: + print(report.summary()) + + sys.exit(0 if report.all_passed else 1)