From fa0d749dfbc7762bb9abd5f7b5179b42363f91b0 Mon Sep 17 00:00:00 2001 From: mohamed zwanski Date: Thu, 22 Oct 2026 21:01:30 +0100 Subject: [PATCH] Record denied Bash commands for later inspection (#73) The security hook now appends every blocked Bash command to .autoforge/denied_commands.jsonl (timestamp, offending command, full command line, reason) so users can see which tools the agent wanted and extend their allowlists. The log is gitignored and rotates at ~1 MB. Adds GET/DELETE /api/projects/{name}/denied-commands to view the log aggregated by command and to clear it, plus tests and docs. Co-Authored-By: Claude Opus 5.5 --- CLAUDE.md | 6 ++ autoforge_paths.py | 12 +++ examples/README.md | 26 ++++++- security.py | 151 +++++++++++++++++++++++++++++++++---- server/routers/projects.py | 62 +++++++++++++++ server/schemas.py | 23 ++++++ test_security.py | 79 +++++++++++++++++++ 7 files changed, 343 insertions(+), 16 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index 6f59910be..8ed0a1206 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -287,6 +287,7 @@ Projects can be stored in any directory (registered in `~/.autoforge/registry.db - `.autoforge/features.db` - SQLite database with feature test cases - `.autoforge/.agent.lock` - Lock file to prevent multiple agent instances - `.autoforge/allowed_commands.yaml` - Project-specific bash command allowlist (optional) +- `.autoforge/denied_commands.jsonl` - Log of bash commands blocked by the security hook (gitignored) - `.autoforge/.gitignore` - Ignores runtime files - `.claude/skills/playwright-cli/` - Playwright CLI skill for browser automation - `.playwright/cli.config.json` - Browser configuration (headless, viewport, etc.) @@ -393,6 +394,11 @@ blocked_commands: - Blocklisted commands (sudo, dd, shutdown, etc.) can NEVER be allowed - Org-level blocked commands cannot be overridden by project configs +**Denied Command Log:** +- Every blocked Bash command is appended to `.autoforge/denied_commands.jsonl` (timestamp, command, full command, reason) so users can review it and extend their allowlists +- Rotates to `denied_commands.jsonl.1` above ~1 MB; gitignored +- API: `GET /api/projects/{name}/denied-commands` (aggregated by command + recent entries), `DELETE` to clear + **Files:** - `security.py` - Command validation logic and hardcoded blocklist - `test_security.py` - Unit tests for security system diff --git a/autoforge_paths.py b/autoforge_paths.py index e720dfb47..fe0c75be0 100644 --- a/autoforge_paths.py +++ b/autoforge_paths.py @@ -45,6 +45,8 @@ .claude_settings.expand.*.json .progress_cache .migration_version +denied_commands.jsonl +denied_commands.jsonl.1 """ @@ -178,6 +180,16 @@ def get_expand_settings_path(project_dir: Path, uuid_hex: str) -> Path: return project_dir / ".autoforge" / f".claude_settings.expand.{uuid_hex}.json" +def get_denied_commands_log_path(project_dir: Path) -> Path: + """Return the path to ``denied_commands.jsonl``. + + The security hook appends one JSON line per blocked Bash command so users + can review what the agent tried to run and extend their allowlists. + Always stored in ``.autoforge/``. + """ + return project_dir / ".autoforge" / "denied_commands.jsonl" + + # --------------------------------------------------------------------------- # Lock-file safety check # --------------------------------------------------------------------------- diff --git a/examples/README.md b/examples/README.md index 21ba3c80b..07eec413f 100644 --- a/examples/README.md +++ b/examples/README.md @@ -11,6 +11,7 @@ This directory contains example configuration files for controlling which bash c - [Pattern Matching](#pattern-matching) - [Common Use Cases](#common-use-cases) - [Security Best Practices](#security-best-practices) +- [Reviewing Denied Commands](#reviewing-denied-commands) --- @@ -388,11 +389,34 @@ These commands are **NEVER allowed**, even with user approval: --- +## Reviewing Denied Commands + +Every Bash command the security hook blocks is recorded in +`{project}/.autoforge/denied_commands.jsonl` (one JSON object per line), so you can +review afterwards which tools the agent wanted (e.g. `pytest`, `openssl`) and add the +safe ones to `allowed_commands.yaml`: + +```json +{"timestamp": "2026-01-01T12:00:00+00:00", "command": "pytest", "full_command": "pytest -q tests/", "reason": "Command 'pytest' is not allowed."} +``` + +- `command` is the specific command that caused the block; `full_command` is the whole command line (truncated to 2000 chars) +- The log is gitignored and rotates to `denied_commands.jsonl.1` once it exceeds ~1 MB +- With the server running, `GET /api/projects/{name}/denied-commands` returns the commands aggregated by name (count, last seen, example) plus the most recent entries; `DELETE` on the same URL clears the log + +Quick summary from the shell: + +```bash +jq -r .command .autoforge/denied_commands.jsonl | sort | uniq -c | sort -rn +``` + +--- + ## Troubleshooting ### Error: "Command 'X' is not allowed" -**Solution:** Add the command to your project config: +**Solution:** Add the command to your project config (see [Reviewing Denied Commands](#reviewing-denied-commands) to find every command that was blocked): ```yaml # In .autoforge/allowed_commands.yaml commands: diff --git a/security.py b/security.py index 9d928b5a9..c71e05a49 100644 --- a/security.py +++ b/security.py @@ -6,10 +6,12 @@ Uses an allowlist approach - only explicitly permitted commands can run. """ +import json import logging import os import re import shlex +from datetime import datetime, timezone from pathlib import Path from typing import Optional @@ -893,6 +895,112 @@ def is_command_allowed(command: str, allowed_commands: set[str]) -> bool: return False +# Rotate the denied-commands log once it grows beyond this size +DENIED_COMMANDS_LOG_MAX_BYTES = 1_000_000 +# Truncate stored command strings to keep log entries small +DENIED_COMMANDS_MAX_COMMAND_CHARS = 2000 + + +def record_denied_command( + project_dir: Optional[Path], + command: str, + denied_command: Optional[str], + reason: str, +) -> None: + """ + Append a blocked Bash command to the project's denied-commands log. + + The log (``.autoforge/denied_commands.jsonl``) lets users review which + commands the agent tried to run so they can extend their allowlists. + Failures are logged and swallowed: recording must never affect the + security decision itself. + + Args: + project_dir: Project directory, or None (nothing is recorded) + command: The full command string the agent attempted + denied_command: The specific command that caused the block, if known + reason: The block reason returned to the agent + """ + if project_dir is None: + return + + from autoforge_paths import get_denied_commands_log_path + + entry = { + "timestamp": datetime.now(timezone.utc).isoformat(), + "command": denied_command, + "full_command": command[:DENIED_COMMANDS_MAX_COMMAND_CHARS], + "reason": reason.splitlines()[0] if reason else "", + } + + try: + log_path = get_denied_commands_log_path(project_dir) + log_path.parent.mkdir(parents=True, exist_ok=True) + if log_path.exists() and log_path.stat().st_size > DENIED_COMMANDS_LOG_MAX_BYTES: + log_path.replace(log_path.with_name(log_path.name + ".1")) + with open(log_path, "a", encoding="utf-8") as f: + f.write(json.dumps(entry) + "\n") + except OSError as e: + logger.warning(f"Failed to record denied command for {project_dir}: {e}") + + +def load_denied_commands(project_dir: Path) -> list[dict]: + """ + Load all recorded denied-command entries for a project (oldest first). + + Includes the rotated ``.1`` log if present. Malformed lines are skipped. + """ + from autoforge_paths import get_denied_commands_log_path + + log_path = get_denied_commands_log_path(project_dir) + entries: list[dict] = [] + for path in (log_path.with_name(log_path.name + ".1"), log_path): + if not path.exists(): + continue + try: + lines = path.read_text(encoding="utf-8").splitlines() + except OSError as e: + logger.warning(f"Failed to read denied commands log {path}: {e}") + continue + for line in lines: + try: + entry = json.loads(line) + except json.JSONDecodeError: + continue + if isinstance(entry, dict): + entries.append(entry) + return entries + + +def summarize_denied_commands(entries: list[dict]) -> list[dict]: + """ + Aggregate denied-command entries by command name. + + Returns: + List of dicts with ``command``, ``count``, ``last_seen`` and + ``example`` (most recent full command), sorted by count descending. + """ + summary: dict[str, dict] = {} + for entry in entries: + name = entry.get("command") or "(unparseable)" + item = summary.setdefault(name, {"command": name, "count": 0, "last_seen": "", "example": ""}) + item["count"] += 1 + timestamp = entry.get("timestamp") or "" + if timestamp >= item["last_seen"]: + item["last_seen"] = timestamp + item["example"] = entry.get("full_command") or "" + return sorted(summary.values(), key=lambda i: (-i["count"], i["command"])) + + +def clear_denied_commands(project_dir: Path) -> None: + """Delete the project's denied-commands log (including rotated log).""" + from autoforge_paths import get_denied_commands_log_path + + log_path = get_denied_commands_log_path(project_dir) + for path in (log_path, log_path.with_name(log_path.name + ".1")): + path.unlink(missing_ok=True) + + async def bash_security_hook(input_data, tool_use_id=None, context=None): """ Pre-tool-use hook that validates bash commands using an allowlist. @@ -914,6 +1022,26 @@ async def bash_security_hook(input_data, tool_use_id=None, context=None): if not command: return {} + # Get project directory from context + project_dir = None + if context and isinstance(context, dict): + project_dir_str = context.get("project_dir") + if project_dir_str: + project_dir = Path(project_dir_str) + + result, denied_command = _validate_bash_command(command, project_dir) + if result.get("decision") == "block": + record_denied_command(project_dir, command, denied_command, result.get("reason", "")) + return result + + +def _validate_bash_command(command: str, project_dir: Optional[Path]) -> tuple[dict, Optional[str]]: + """ + Validate a Bash command string against the effective allow/block lists. + + Returns: + Tuple of (hook result, name of the command that caused a block or None) + """ # Extract all commands from the command string commands = extract_commands(command) @@ -922,14 +1050,7 @@ async def bash_security_hook(input_data, tool_use_id=None, context=None): return { "decision": "block", "reason": f"Could not parse command for security validation: {command}", - } - - # Get project directory from context - project_dir = None - if context and isinstance(context, dict): - project_dir_str = context.get("project_dir") - if project_dir_str: - project_dir = Path(project_dir_str) + }, None # Get effective commands using hierarchy resolution allowed_commands, blocked_commands = get_effective_commands(project_dir) @@ -947,7 +1068,7 @@ async def bash_security_hook(input_data, tool_use_id=None, context=None): return { "decision": "block", "reason": f"Command '{cmd}' is blocked at organization level and cannot be approved.", - } + }, cmd # Check allowlist (with pattern matching) if not is_command_allowed(cmd, allowed_commands): @@ -960,7 +1081,7 @@ async def bash_security_hook(input_data, tool_use_id=None, context=None): return { "decision": "block", "reason": error_msg, - } + }, cmd # Additional validation for sensitive commands if cmd in COMMANDS_NEEDING_EXTRA_VALIDATION: @@ -979,18 +1100,18 @@ async def bash_security_hook(input_data, tool_use_id=None, context=None): extra_procs = pkill_processes - DEFAULT_PKILL_PROCESSES allowed, reason = validate_pkill_command(cmd_segment, extra_procs if extra_procs else None) if not allowed: - return {"decision": "block", "reason": reason} + return {"decision": "block", "reason": reason}, cmd elif cmd == "chmod": allowed, reason = validate_chmod_command(cmd_segment) if not allowed: - return {"decision": "block", "reason": reason} + return {"decision": "block", "reason": reason}, cmd elif cmd == "init.sh": allowed, reason = validate_init_script(cmd_segment) if not allowed: - return {"decision": "block", "reason": reason} + return {"decision": "block", "reason": reason}, cmd elif cmd == "playwright-cli": allowed, reason = validate_playwright_command(cmd_segment) if not allowed: - return {"decision": "block", "reason": reason} + return {"decision": "block", "reason": reason}, cmd - return {} + return {}, None diff --git a/server/routers/projects.py b/server/routers/projects.py index ff0e38b8a..edf8ce07e 100644 --- a/server/routers/projects.py +++ b/server/routers/projects.py @@ -15,6 +15,9 @@ from fastapi import APIRouter, HTTPException from ..schemas import ( + DeniedCommandEntry, + DeniedCommandsResponse, + DeniedCommandSummary, ProjectCreate, ProjectDetail, ProjectPrompts, @@ -394,6 +397,65 @@ async def get_project_stats_endpoint(name: str): return get_project_stats(project_dir) +# Maximum number of individual denied-command entries returned by the API +DENIED_COMMANDS_RECENT_LIMIT = 100 + + +def _get_existing_project_dir(name: str) -> Path: + """Resolve a registered project's directory or raise 404.""" + (_, _, get_project_path, _, _, _, _, _, _) = _get_registry_functions() + + name = validate_project_name(name) + project_dir = get_project_path(name) + + if not project_dir: + raise HTTPException(status_code=404, detail=f"Project '{name}' not found") + + if not project_dir.exists(): + raise HTTPException(status_code=404, detail="Project directory not found") + + return Path(project_dir) + + +@router.get("/{name}/denied-commands", response_model=DeniedCommandsResponse) +async def get_denied_commands(name: str): + """List Bash commands the agent tried to run but were blocked by the security hook.""" + _init_imports() + from security import load_denied_commands, summarize_denied_commands + + project_dir = _get_existing_project_dir(name) + entries = load_denied_commands(project_dir) + + return DeniedCommandsResponse( + total=len(entries), + commands=[DeniedCommandSummary(**item) for item in summarize_denied_commands(entries)], + recent=[ + DeniedCommandEntry( + timestamp=str(e.get("timestamp") or ""), + command=e.get("command"), + full_command=str(e.get("full_command") or ""), + reason=str(e.get("reason") or ""), + ) + for e in reversed(entries[-DENIED_COMMANDS_RECENT_LIMIT:]) + ], + ) + + +@router.delete("/{name}/denied-commands") +async def clear_denied_commands_endpoint(name: str): + """Clear the denied-commands log for a project.""" + _init_imports() + from security import clear_denied_commands + + project_dir = _get_existing_project_dir(name) + try: + clear_denied_commands(project_dir) + except OSError as e: + raise HTTPException(status_code=500, detail=f"Failed to clear denied commands: {e}") + + return {"success": True, "message": "Denied commands cleared"} + + @router.post("/{name}/reset") async def reset_project(name: str, full_reset: bool = False): """ diff --git a/server/schemas.py b/server/schemas.py index d54f00aa9..67a1f5ecd 100644 --- a/server/schemas.py +++ b/server/schemas.py @@ -69,6 +69,29 @@ class ProjectPrompts(BaseModel): coding_prompt: str = "" +class DeniedCommandEntry(BaseModel): + """A single Bash command blocked by the security hook.""" + timestamp: str = "" + command: str | None = None + full_command: str = "" + reason: str = "" + + +class DeniedCommandSummary(BaseModel): + """Denied commands aggregated by command name.""" + command: str + count: int + last_seen: str = "" + example: str = "" + + +class DeniedCommandsResponse(BaseModel): + """Commands the agent tried to run but was not allowed to.""" + total: int = 0 + commands: list[DeniedCommandSummary] = [] + recent: list[DeniedCommandEntry] = [] + + class ProjectPromptsUpdate(BaseModel): """Request schema for updating project prompts.""" app_spec: str | None = None diff --git a/test_security.py b/test_security.py index ccd234645..6fde425ab 100644 --- a/test_security.py +++ b/test_security.py @@ -15,13 +15,17 @@ from pathlib import Path from security import ( + DENIED_COMMANDS_LOG_MAX_BYTES, bash_security_hook, + clear_denied_commands, extract_commands, get_effective_commands, get_effective_pkill_processes, + load_denied_commands, load_org_config, load_project_commands, matches_pattern, + summarize_denied_commands, validate_chmod_command, validate_init_script, validate_pkill_command, @@ -988,6 +992,76 @@ def test_playwright_cli_validation(): return passed, failed +def test_denied_commands_logging(): + """Test that blocked commands are recorded for later inspection.""" + print("\nTesting denied commands logging:\n") + passed = 0 + failed = 0 + + def check(condition, description): + nonlocal passed, failed + if condition: + print(f" PASS: {description}") + passed += 1 + else: + print(f" FAIL: {description}") + failed += 1 + + with tempfile.TemporaryDirectory() as tmphome: + with tempfile.TemporaryDirectory() as tmpproject: + with temporary_home(tmphome): + project_dir = Path(tmpproject) + context = {"project_dir": str(project_dir)} + log_path = project_dir / ".autoforge" / "denied_commands.jsonl" + + def run(command): + input_data = {"tool_name": "Bash", "tool_input": {"command": command}} + return asyncio.run(bash_security_hook(input_data, context=context)) + + run("ls -la") + check(not log_path.exists(), "allowed command is not recorded") + + run("pytest -q tests/") + run("npm test && pytest") + run("openssl rand -hex 16") + run("sudo rm -rf /") + run("pkill chrome") + + entries = load_denied_commands(project_dir) + check(len(entries) == 5, f"5 denied commands recorded (got {len(entries)})") + names = [e.get("command") for e in entries] + check(names == ["pytest", "pytest", "openssl", "sudo", "pkill"], f"denied command names recorded ({names})") + check(entries[1].get("full_command") == "npm test && pytest", "full command string recorded") + check(all(e.get("timestamp") and e.get("reason") for e in entries), "timestamp and reason recorded") + + summary = summarize_denied_commands(entries) + check(summary[0]["command"] == "pytest" and summary[0]["count"] == 2, "summary aggregates by command") + check(summary[0]["example"] == "npm test && pytest", "summary example is most recent occurrence") + + # Without a project dir nothing is recorded (and nothing crashes) + asyncio.run(bash_security_hook({"tool_name": "Bash", "tool_input": {"command": "pytest"}})) + check(len(load_denied_commands(project_dir)) == 5, "no recording without project_dir") + + # Malformed lines are skipped + with open(log_path, "a", encoding="utf-8") as f: + f.write("not json\n") + check(len(load_denied_commands(project_dir)) == 5, "malformed log lines are skipped") + + # Log rotates once it exceeds the size limit; rotated entries are still loaded + with open(log_path, "a", encoding="utf-8") as f: + f.write(" " * (DENIED_COMMANDS_LOG_MAX_BYTES + 1) + "\n") + run("openssl version") + rotated = log_path.with_name(log_path.name + ".1") + check(rotated.exists() and log_path.stat().st_size < 1000, "log rotates when too large") + check(len(load_denied_commands(project_dir)) == 6, "rotated entries still loaded") + + clear_denied_commands(project_dir) + check(not log_path.exists() and not rotated.exists(), "clear removes logs") + check(load_denied_commands(project_dir) == [], "no entries after clear") + + return passed, failed + + def main(): print("=" * 70) print(" SECURITY HOOK TESTS") @@ -1061,6 +1135,11 @@ def main(): passed += pw_passed failed += pw_failed + # Test denied commands logging + denied_passed, denied_failed = test_denied_commands_logging() + passed += denied_passed + failed += denied_failed + # Commands that SHOULD be blocked # Note: blocklisted commands (sudo, shutdown, dd, aws) are tested in # test_blocklist_enforcement(). chmod validation is tested in