Agent Skills with Code Execution.

# Wrong: Bash for everything
# tool_use: Bash, command: "cat config.json"
# tool_use: Bash, command: "grep -r 'TODO' src/"

# Right: built-in tools for I/O; Bash only for execution
# tool_use: Read, file_path: "config.json"
# tool_use: Grep, pattern: "TODO", path: "src/"
# tool_use: Glob, pattern: "**/*.py"

# Bash legitimately for execution:
# tool_use: Bash, command: "pytest tests/ --json-report"
# tool_use: Bash, command: "python analyze.py --input data.csv"

import re
FILE_IO_VIA_BASH = re.compile(
    r"^\s*(cat|head|tail|less|more|grep|find|ls|wc|sort|uniq|cut|awk|sed)\s",
)

def warn_on_io_via_bash(tool_name: str, command: str) -> str | None:
    if tool_name != "Bash":
        return None
    if FILE_IO_VIA_BASH.match(command):
        first = command.strip().split()[0]
        return (
            f"Bash command starts with {first!r}. "
            f"For file I/O prefer Read / Grep / Glob; reserve Bash for execution."
        )
    return None

# .claude/hooks/codeexec_blocklist.py
import sys, json, re

BLOCKLIST = re.compile(
    r"\b("
    r"rm\s+-rf"
    r"|sudo\s+"
    r"|drop\s+(database|table)"
    r"|kill\s+-9"
    r"|chmod\s+777"
    r"|>\s*/(etc|usr|var)/"
    r"|curl\s+[^|]+\|\s*sh"
    r")\b",
    re.IGNORECASE,
)

ALLOWLIST_BINS = {
    "python", "python3", "pytest", "node", "npm", "pnpm",
    "tsc", "eslint", "prettier", "ruff", "black", "go", "cargo",
    "kubectl", "docker", "jq",
}

def main():
    payload = json.loads(sys.stdin.read())
    if payload["tool_name"] != "Bash":
        sys.exit(0)
    cmd = (payload["tool_input"].get("command") or "").strip()
    if BLOCKLIST.search(cmd):
        print(f"BLOCKED: command matches destructive pattern. command={cmd!r}", file=sys.stderr)
        sys.exit(2)
    first = cmd.split()[0] if cmd else ""
    if first and first not in ALLOWLIST_BINS:
        print(f"BLOCKED: binary {first!r} not on the code-exec allowlist.", file=sys.stderr)
        sys.exit(2)
    sys.exit(0)

if __name__ == "__main__":
    main()

import subprocess, time

def run_in_sandbox(command: str, timeout_s: int = 30) -> dict:
    """Run command in a Docker sandbox with kernel-level limits."""
    start = time.monotonic()
    try:
        result = subprocess.run(
            [
                "docker", "run", "--rm",
                "--cpus", "2",
                "--memory", "1024m",
                "--memory-swap", "1024m",
                "--network", "none",
                "--read-only",
                "--tmpfs", "/tmp:size=128m",
                "--ipc", "private",
                "--pid", "private",
                "code-exec:latest",
                "bash", "-c", command,
            ],
            capture_output=True, text=True,
            timeout=timeout_s,
        )
        return {
            "status": "ok" if result.returncode == 0 else "exit_nonzero",
            "exit_code": result.returncode,
            "stdout": result.stdout,
            "stderr": result.stderr,
            "duration_ms": int((time.monotonic() - start) * 1000),
        }
    except subprocess.TimeoutExpired:
        return {
            "status": "timeout",
            "exit_code": 124,
            "stdout": "",
            "stderr": f"command exceeded {timeout_s}s timeout",
            "duration_ms": timeout_s * 1000,
        }

# Wrong: Python signal handler
# import signal
# def handler(signum, frame):
#     raise TimeoutError("timed out")
# signal.signal(signal.SIGALRM, handler)
# signal.alarm(30)
# # User script can do try/except and swallow the SIGALRM.

# Right: kernel timeout via systemd-run
import subprocess

def run_with_kernel_timeout(command: str, timeout_s: int = 30) -> int:
    result = subprocess.run(
        [
            "systemd-run", "--user", "--scope",
            f"--property=TimeoutStartSec={timeout_s}s",
            "--property=MemoryMax=1G",
            "--property=CPUQuota=200%",
            "bash", "-c", command,
        ],
    )
    # Exit code 124 means systemd killed it for timeout; user script cannot prevent.
    return result.returncode

import sys, json, re, datetime

ANSI = re.compile(r"\x1b\[[0-9;]*m")
UNIX_TS = re.compile(r"\b1[6-9]\d{8}\b")

def normalize_stdout(text: str, max_bytes: int = 4096) -> str:
    text = ANSI.sub("", text)
    text = UNIX_TS.sub(
        lambda m: datetime.datetime.utcfromtimestamp(int(m.group())).isoformat() + "Z",
        text,
    )
    if len(text) > max_bytes:
        head = text[: max_bytes // 2]
        tail = text[-max_bytes // 2 :]
        omitted_lines = text[max_bytes // 2 : -max_bytes // 2].count("\n")
        text = f"{head}\n... truncated ({omitted_lines} more lines) ...\n{tail}"
    return text

def main():
    payload = json.loads(sys.stdin.read())
    if payload["tool_name"] != "Bash":
        print(json.dumps(payload))
        sys.exit(0)
    raw_result = payload.get("tool_result") or {}
    normalized = {
        "status": raw_result.get("status", "unknown"),
        "exit_code": raw_result.get("exit_code", -1),
        "stdout": normalize_stdout(raw_result.get("stdout", "")),
        "stderr": normalize_stdout(raw_result.get("stderr", "")),
        "duration_ms": raw_result.get("duration_ms"),
        "peak_memory_mb": raw_result.get("peak_memory_mb"),
    }
    payload["tool_result"] = normalized
    print(json.dumps(payload))
    sys.exit(0)

if __name__ == "__main__":
    main()

import json
from typing import TypedDict

class ValidationResult(TypedDict):
    valid: bool
    errors: list[str]

def validate_pytest_output(result: dict) -> ValidationResult:
    errors = []
    try:
        report = json.loads(result["stdout"])
    except json.JSONDecodeError:
        errors.append("pytest stdout is not valid JSON")
        return {"valid": False, "errors": errors}

    summary = report.get("summary", {})
    passed = summary.get("passed", 0)
    failed = summary.get("failed", 0)
    skipped = summary.get("skipped", 0)
    total = summary.get("total", 0)

    if any(v < 0 for v in (passed, failed, skipped, total)):
        errors.append(f"negative test counts in summary: {summary}")
    if passed + failed + skipped != total:
        errors.append(f"counts inconsistent: passed={passed} + failed={failed} + skipped={skipped} != total={total}")
    if total == 0 and result.get("exit_code") == 0:
        errors.append("pytest reported total=0 but exited 0; expected tests to run")
    return {"valid": len(errors) == 0, "errors": errors}

def adversarial_oom_test() -> bool:
    """Allocate 10 GB. Sandbox should kill at 1 GB cap."""
    code = "x = bytearray(10 * 1024 * 1024 * 1024)"
    result = run_in_sandbox(f"python3 -c {code!r}")
    if result["exit_code"] != 137:
        print(f"FAIL: expected exit 137 (OOM kill), got {result['exit_code']}")
        return False
    print("PASS: OOM kill at 1 GB cap")
    return True

def adversarial_timeout_test() -> bool:
    """Busy loop. Sandbox should kill at 30s cap."""
    result = run_in_sandbox("python3 -c 'while True: pass'", timeout_s=30)
    if result["status"] != "timeout":
        print(f"FAIL: expected status=timeout, got {result['status']}")
        return False
    print("PASS: timeout kill at 30s")
    return True

import datetime, json
from pathlib import Path

AUDIT_DIR = Path("audit")
AUDIT_DIR.mkdir(exist_ok=True)

def audit_code_exec(command, pre_decision, sandbox_result, validation, agent_id):
    today = datetime.date.today().isoformat()
    row = {
        "ts": datetime.datetime.utcnow().isoformat() + "Z",
        "agent_id": agent_id,
        "command": command[:200],
        "pre_decision": pre_decision,
        "sandbox": {
            "status": sandbox_result.get("status"),
            "exit_code": sandbox_result.get("exit_code"),
            "duration_ms": sandbox_result.get("duration_ms"),
            "peak_memory_mb": sandbox_result.get("peak_memory_mb"),
        },
        "validation": validation,
    }
    with open(AUDIT_DIR / f"{today}.jsonl", "a") as f:
        f.write(json.dumps(row) + "\n")