Add scripts/preflight.py
This commit is contained in:
neo committed
1 parent
99dde1cc1d
commit
93daa17678
1 file changed
+203
@@ -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)
|
||||
Reference in new issue
Block a user