Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
75 changes: 71 additions & 4 deletions codeframe/core/adapters/claude_code.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,10 +2,61 @@

from __future__ import annotations

import json
import shlex
import sys
from pathlib import Path

from codeframe.core.adapters.subprocess_adapter import SubprocessAdapter

_GUARD_MODULE = "codeframe.core.claude_code_guard"

# Tools known to be incapable of changing the workspace. Everything *not* listed
# here — Edit/Write/Bash, an MCP write tool, Task (whose subagent can write), a
# tool that does not exist yet — is assumed write-capable.
#
# The polarity matters. Reading a write tool as read-only would silently switch
# off the zero-file guard and re-open the #739 false completion; reading a
# read-only tool as a write tool merely fails an analysis run loudly. So the
# unknown case must land on "write". Keeping an allowlist of the safe names, not
# a blocklist of the dangerous ones, is what makes that the default. (#819 review)
_READ_ONLY_TOOLS = frozenset(
{"read", "grep", "glob", "ls", "notebookread", "webfetch", "websearch"}
)


def _grants_write_access(allowlist: list[str]) -> bool:
"""True unless every allowlist entry is a known read-only tool.

Entries may be bare tool names ("Edit") or rule-shaped ("Bash(git *)"), so
match on the tool name preceding any rule parentheses.
"""
return not all(
entry.split("(", 1)[0].strip().lower() in _READ_ONLY_TOOLS
for entry in allowlist
)


def _guard_settings() -> str:
"""Build the --settings payload registering the dangerous-command guard.

The hook runs under the interpreter already running CodeFrame, so the guard
module is importable regardless of what is on the delegated CLI's PATH.
"""
hook_command = f"{shlex.quote(sys.executable)} -m {_GUARD_MODULE}"
return json.dumps(
{
"hooks": {
"PreToolUse": [
{
"matcher": "Bash",
"hooks": [{"type": "command", "command": hook_command}],
}
]
}
}
)


class ClaudeCodeAdapter(SubprocessAdapter):
"""Adapter that delegates code execution to Claude Code CLI.
Expand All @@ -20,7 +71,7 @@ class ClaudeCodeAdapter(SubprocessAdapter):
def __init__(
self,
allowlist: list[str] | None = None,
require_file_changes: bool = True,
require_file_changes: bool | None = None,
) -> None:
"""Initialize the Claude Code adapter.

Expand All @@ -32,9 +83,16 @@ def __init__(
are auto-approved in non-interactive ``--print`` mode.
Without this, ``--print`` silently denies those tools and
the delegated agent can analyze but never modify files. (#739)
require_file_changes: If True (default), a run that exits 0 but touches
no files is downgraded to ``failed`` — a coding task that
writes nothing is a false completion.
Either way a PreToolUse hook vets Bash commands against
CodeFrame's dangerous-command patterns — the permission
mode skips approval prompts, not hooks. (#819)
require_file_changes: If True, a run that exits 0 but touches no files
is downgraded to ``failed`` — a coding task that writes
nothing is a false completion. Defaults to whether the
adapter actually grants write access: on for the
bypassPermissions default, and for an allowlist carrying a
write tool; off for a read-only allowlist, whose whole
point is a run that writes nothing. (#819)
"""
cli_args = ["--print"]
if allowlist:
Expand All @@ -43,6 +101,15 @@ def __init__(
else:
cli_args.extend(["--permission-mode", "bypassPermissions"])

# Attached on both paths: an allowlist granting Bash has the same blast
# radius as bypass mode, and a read-only one loses nothing by carrying it.
cli_args.extend(["--settings", _guard_settings()])

if require_file_changes is None:
require_file_changes = (
_grants_write_access(allowlist) if allowlist else True
)

super().__init__(
binary="claude",
cli_args=cli_args,
Expand Down
26 changes: 25 additions & 1 deletion codeframe/core/adapters/git_utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,15 +2,23 @@

from __future__ import annotations

import logging
import subprocess
from pathlib import Path

logger = logging.getLogger(__name__)


def detect_modified_files(workspace_path: Path) -> list[str]:
"""Detect files modified in a workspace via git diff.

Combines modified, staged, and untracked files. Returns an empty list
if git is unavailable or the workspace is not a git repo.

Errors are logged rather than swallowed silently: an empty list means
"failed" for ``require_file_changes`` adapters, so a transient git error
(lock contention, missing binary, timeout) would otherwise be
indistinguishable from "the agent changed nothing". (#819)
"""
try:
result = subprocess.run(
Expand All @@ -21,6 +29,13 @@ def detect_modified_files(workspace_path: Path) -> list[str]:
timeout=10,
)
if result.returncode != 0:
logger.warning(
"git diff failed in %s (exit %d): %s — reporting no modified "
"files, which callers may read as 'agent did nothing'.",
workspace_path,
result.returncode,
(result.stderr or "").strip() or "<no stderr>",
)
return []

files = [f for f in result.stdout.strip().splitlines() if f]
Expand All @@ -34,7 +49,16 @@ def detect_modified_files(workspace_path: Path) -> list[str]:
)
if untracked.returncode == 0:
files.extend(f for f in untracked.stdout.strip().splitlines() if f)
else:
logger.warning(
"git ls-files failed in %s (exit %d): %s — untracked files are "
"missing from the result.",
workspace_path,
untracked.returncode,
(untracked.stderr or "").strip() or "<no stderr>",
)

return list(dict.fromkeys(files))
except (FileNotFoundError, OSError, subprocess.TimeoutExpired):
except (FileNotFoundError, OSError, subprocess.TimeoutExpired) as e:
logger.warning("git could not run in %s: %s", workspace_path, e)
return []
41 changes: 34 additions & 7 deletions codeframe/core/adapters/subprocess_adapter.py
Original file line number Diff line number Diff line change
Expand Up @@ -242,12 +242,27 @@ def _write_stdin() -> None:
and head_after != head_before
)
if in_git_repo and not committed:
result.status = "failed"
result.error = (
f"'{self._binary}' exited successfully but modified no files. "
"A coding task must change at least one file; the agent likely "
"lacked write permission or produced no edits."
)
# A zero-file exit-0 run may be the agent *asking* rather than
# failing: `--print` has no way to prompt, so a genuine ambiguity
# gets printed and the process exits clean. Route those to the
# blocker flow a human can answer instead of hard-failing with a
# misleading "lacked write permission". Tactical questions
# classify as None by design — those are the agent's own job, so
# they keep failing. (#819)
category = classify_error_for_blocker(result.output or "")
if category is not None:
result.status = "blocked"
result.error = None
result.blocker_question = self._extract_blocker_question(
result.output or ""
)
else:
result.status = "failed"
result.error = (
f"'{self._binary}' exited successfully but modified no "
"files. A coding task must change at least one file; the "
"agent likely lacked write permission or produced no edits."
)

return result

Expand Down Expand Up @@ -296,6 +311,9 @@ def _git_head(self, workspace_path: Path) -> str | None:

None means "not a git repo, git unavailable, or an unborn HEAD" — i.e.
a state where modified-file detection can't judge whether work happened.
Errors are logged: since empty now means failed for require_file_changes
adapters, a git hiccup would otherwise be indistinguishable from "the
agent changed nothing" in the logs. (#819)
"""
try:
result = subprocess.run(
Expand All @@ -306,9 +324,18 @@ def _git_head(self, workspace_path: Path) -> str | None:
timeout=10,
)
if result.returncode != 0:
logger.warning(
"git rev-parse HEAD failed in %s (exit %d): %s",
workspace_path,
result.returncode,
(result.stderr or "").strip() or "<no stderr>",
)
return None
return result.stdout.strip() or None
except (FileNotFoundError, OSError, subprocess.TimeoutExpired):
except (FileNotFoundError, OSError, subprocess.TimeoutExpired) as e:
logger.warning(
"git rev-parse HEAD could not run in %s: %s", workspace_path, e
)
return None

def _extract_blocker_question(self, output: str) -> str:
Expand Down
87 changes: 87 additions & 0 deletions codeframe/core/claude_code_guard.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,87 @@
"""PreToolUse hook: block dangerous Bash commands from a delegated claude CLI.

``ClaudeCodeAdapter`` runs ``claude`` with ``--permission-mode bypassPermissions``
so ``--print`` mode does not silently deny Edit/Write/Bash (#739). That hands the
delegated agent unrestricted Bash with none of the ``DANGEROUS_PATTERNS``
filtering the built-in ReAct engine applies to every command it runs
(``core/tools.py`` -> ``is_dangerous_command``). With GitHub Issues import (#565)
turning externally-authored issue bodies into task prompts, that gap widens the
prompt-injection blast radius.

``claude`` runs this module as a PreToolUse hook, which fires *even under*
``bypassPermissions`` — the permission mode skips approval prompts, not hooks.
Re-using ``is_dangerous_command`` keeps one source of truth for the patterns, so
both engines block the same set. (#819)

Import cost is paid *per Bash call* (a fresh subprocess each time), so this module
is kept deliberately cheap to import:

- The patterns live in the stdlib-only ``core.dangerous_commands`` leaf rather
than in ``core.executor``, which drags in ``codeframe.adapters.llm`` and the
openai SDK.
- It sits in ``core`` rather than beside its adapter in ``core.adapters``, whose
``__init__`` eagerly imports every adapter (and so the LLM SDKs again).

Together those took the guard from ~930 imported modules to ~300. (#819 review)

This is the same grade of protection ReAct has, no more: a regex matcher over the
command string, evadable by an agent that means to evade it (``bash -c``, string
splicing). It raises the floor for accidental and injected destructive commands;
it is not a sandbox.

Contract (https://code.claude.com/docs/en/hooks): the hook JSON arrives on stdin;
exit 0 with no output allows the call; exit 0 with a ``permissionDecision: deny``
payload blocks it and feeds the reason back to the agent.
"""

from __future__ import annotations

import json
import sys

from codeframe.core.dangerous_commands import is_dangerous_command


def main() -> int:
"""Vet one PreToolUse payload. Returns the process exit code."""
try:
payload = json.load(sys.stdin)
except (json.JSONDecodeError, ValueError, UnicodeDecodeError):
# Not a payload we understand. Allow: a guard that wedges every run on a
# parse hiccup is worse than one that misses a hook it can't read.
return 0

if not isinstance(payload, dict) or payload.get("tool_name") != "Bash":
return 0

tool_input = payload.get("tool_input")
if not isinstance(tool_input, dict):
return 0

command = tool_input.get("command")
if not isinstance(command, str) or not command.strip():
return 0

dangerous, reason = is_dangerous_command(command)
if not dangerous:
return 0

json.dump(
{
"hookSpecificOutput": {
"hookEventName": "PreToolUse",
"permissionDecision": "deny",
"permissionDecisionReason": (
f"CodeFrame blocked this command: {reason}. "
"Destructive commands are refused for delegated runs; "
"achieve the task another way."
),
}
},
sys.stdout,
)
return 0


if __name__ == "__main__":
sys.exit(main())
75 changes: 75 additions & 0 deletions codeframe/core/dangerous_commands.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,75 @@
"""Dangerous shell-command patterns — the shared filter for every engine.

Deliberately a **leaf module**: it imports nothing but the stdlib. Both consumers
need it cheaply.

- The built-in ReAct engine imports it once per process (``core/tools.py``).
- The claude-code guard (``core/claude_code_guard.py``) is spawned as a fresh
subprocess for *every* Bash call the delegated CLI makes, so it pays this
module's import cost per command. Living under ``core/executor.py`` — which
pulls in ``codeframe.adapters.llm`` and the openai SDK, ~930 modules — cost
~420ms per Bash call for what is a regex match. (#819 review)

``core/executor.py`` re-exports both names, so the historical import path keeps
working and the patterns stay a single source of truth.
"""

from __future__ import annotations

import re
import shlex

# Module-level dangerous command patterns (importable by other modules like tools.py)
DANGEROUS_PATTERNS: list[tuple[str, str]] = [
# Recursive delete of root or home
(r"\brm\s+(-[rf]+\s+)*[/~]", "recursive deletion of root or home"),
(r"\brm\s+--no-preserve-root", "rm with --no-preserve-root"),
# Writing to /dev/ devices
(r">\s*/dev/", "redirect to /dev device"),
(r"\bdd\s+.*of=/dev/", "dd writing to device"),
# Filesystem destruction
(r"\bmkfs\b", "filesystem format command"),
(r"\bfdisk\b", "disk partition command"),
# Fork bombs
(r":\s*\(\s*\)\s*\{", "potential fork bomb"),
(r"\bfork\s*while\s*true", "potential fork bomb"),
# Dangerous dd operations
(r"\bdd\s+if=/dev/", "dd reading from device"),
# Dangerous chmod
(r"\bchmod\s+(-[Rr]\s+)?777\s+/", "chmod 777 on root"),
# Wget/curl piped to shell (potential malware download)
(r"\b(wget|curl)\s+.*\|\s*(ba)?sh", "download piped to shell"),
# Overwriting important system files
(r">\s*/(etc|bin|usr|lib|sbin)/", "overwriting system directory"),
]


def is_dangerous_command(command: str) -> tuple[bool, str]:
"""Check if a command matches dangerous patterns.

Uses regex-based patterns that are harder to bypass than substring matching.
Normalizes whitespace and handles common shell escapes.

Args:
command: The shell command to check

Returns:
Tuple of (is_dangerous, description) where description explains the match
"""
# Normalize the command for comparison
try:
# Use shlex to handle escapes, then rejoin
tokens = shlex.split(command)
normalized = " ".join(tokens)
except ValueError:
# If shlex fails, use the original with normalized whitespace
normalized = " ".join(command.split())

for pattern, description in DANGEROUS_PATTERNS:
if re.search(pattern, normalized, re.IGNORECASE):
return (True, description)
# Also check original command in case normalization removed something
if re.search(pattern, command, re.IGNORECASE):
return (True, description)

return (False, "")
Loading