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
2 changes: 1 addition & 1 deletion README.md
Original file line number Diff line number Diff line change
Expand Up @@ -37,7 +37,7 @@ DevMind: [searches the project] [runs tests] [reads code]
- **Reliable sessions:** named saves, autosave, atomic persistence, and compacted conversation history.
- **Visible operation:** streaming output, structured logs, task state, retry metrics, and per-tool latency.
- **Cost awareness:** token usage and estimated USD cost are recorded for each API call.
- **Offline confidence:** 156 tests cover tools, security guards, persistence, metrics, plugins, and control flow without paid API calls.
- **Offline confidence:** 170 tests cover tools, security guards, persistence, metrics, plugins, and control flow without paid API calls.

## Architecture

Expand Down
191 changes: 178 additions & 13 deletions security.py
Original file line number Diff line number Diff line change
Expand Up @@ -96,18 +96,166 @@ def validate_path(
# Command validation
# ============================================================
_SUSPICIOUS_CHAR_RUNS = re.compile(r"`.+`|\$\([^)]+\)")
_SHELL_OPERATORS = re.compile(r"&&|\|\||[;|\n]")
_ENV_ASSIGNMENT = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*=")
_WRAPPERS = {
"builtin",
"command",
"env",
"exec",
"nice",
"nohup",
"setsid",
"stdbuf",
"time",
"timeout",
"xargs",
}
_WRAPPER_VALUE_OPTIONS = {
"env": {"-u", "--unset", "-C", "--chdir"},
"exec": {"-a"},
"nice": {"-n", "--adjustment"},
"stdbuf": {"-i", "-o", "-e", "--input", "--output", "--error"},
"time": {"-f", "--format", "-o", "--output"},
"timeout": {"-s", "--signal", "-k", "--kill-after"},
"xargs": {
"-a",
"--arg-file",
"-d",
"--delimiter",
"-E",
"--eof",
"-I",
"--replace",
"-L",
"--max-lines",
"-n",
"--max-args",
"-P",
"--max-procs",
"-s",
"--max-chars",
},
}
_PRIVILEGE_TOOLS = {"sudo", "su", "doas", "pkexec"}
_SHELLS = {"bash", "sh", "zsh", "ksh", "fish", "dash", "powershell", "pwsh", "cmd"}
_INLINE_CODE = {
"py": {"-c"},
"python": {"-c"},
"python3": {"-c"},
"node": {"-e", "--eval"},
"perl": {"-e"},
"ruby": {"-e"},
"php": {"-r"},
}


def _segments(command: str) -> list[str]:
return [
segment.strip()
for segment in _SHELL_OPERATORS.split(command)
if segment.strip()
]


def _split_env_string(value: str) -> list[str]:
try:
return shlex.split(value, posix=True)
except ValueError:
return value.split()


def _executable(tokens: list[str]) -> tuple[str, list[str]]:
while tokens:
if _ENV_ASSIGNMENT.match(tokens[0]):
tokens = tokens[1:]
continue
head = Path(tokens[0]).name.lower()
if head not in _WRAPPERS:
break
tokens = tokens[1:]

while tokens and tokens[0].startswith("-"):
option = tokens[0]
tokens = tokens[1:]
if option == "--":
break
if head == "env" and (
option in {"-S", "--split-string"}
or option.startswith("--split-string=")
or (option.startswith("-S") and len(option) > 2)
):
if option.startswith("--split-string="):
value = option.split("=", 1)[1]
elif option.startswith("-S") and len(option) > 2:
value = option[2:]
elif tokens:
value = tokens[0]
tokens = tokens[1:]
else:
break
tokens = _split_env_string(value) + tokens
continue
if option in _WRAPPER_VALUE_OPTIONS.get(head, set()) and tokens:
tokens = tokens[1:]

if head == "timeout" and tokens and not tokens[0].startswith("-"):
tokens = tokens[1:]

if not tokens:
return "", []
return Path(tokens[0]).name.lower(), tokens[1:]


def _shell_has_command_flag(arguments: list[str]) -> bool:
for argument in arguments:
lowered = argument.lower()
if lowered in {"-c", "/c", "/k"}:
return True
if (
lowered.startswith("-")
and not lowered.startswith("--")
and "c" in lowered[1:]
):
return True
return False


def _has_inline_flag(arguments: list[str], flags: set[str]) -> bool:
short_flags = {
flag[1:] for flag in flags if flag.startswith("-") and not flag.startswith("--")
}
for argument in arguments:
lowered = argument.lower()
if not lowered.startswith("-"):
break
if lowered in flags:
return True
if lowered.startswith("-") and not lowered.startswith("--"):
if any(flag in lowered[1:] for flag in short_flags):
return True
return False


def validate_command(command: str) -> tuple[bool, str]:
"""
Validate a shell command before running it.
Best-effort shell guardrail for the bash tool.

This is a denylist over parsed shell segments, not an OS sandbox: a
determined caller can still run arbitrary code through an allowed
interpreter. The agent-level approval flow is the real boundary.

Rules:
- Must be a non-empty string
- Null bytes rejected
- Must not contain any blocked substring from config.tool.blocked_commands
- Tokens starting with `sudo` are blocked
- Length is capped (10 KB) to avoid abuse
- Command substitution (backticks, `$()`) rejected
- Blocked substrings from config.tool.blocked_commands rejected
- Privilege escalation (sudo/su/doas/pkexec) rejected anywhere,
including via env wrappers and absolute paths
- Nested shell `-c` and inline interpreter code (`python -c`, `node -e`)
rejected
- base64 decode piped into a shell rejected
- Length capped (10 KB)

Returns:
(is_valid, error_message)
Expand All @@ -121,6 +269,9 @@ def validate_command(command: str) -> tuple[bool, str]:
return False, "Invalid command (null byte)."
if len(command) > 10_000:
return False, "Command too long (max 10 KB)."
if _SUSPICIOUS_CHAR_RUNS.search(stripped):
log.warning(f"Command substitution blocked: {stripped[:80]}")
return False, "Command substitution (backticks or $()) is blocked."

lower = stripped.lower()

Expand All @@ -135,16 +286,30 @@ def validate_command(command: str) -> tuple[bool, str]:
),
)

# Block sudo and su outright
try:
tokens = shlex.split(stripped, posix=True)
except ValueError:
# Unbalanced quotes etc. — allow bash itself to error out
tokens = stripped.split()
if re.search(r"\bbase64\b[^|]*(?:-d|--decode)[^|]*\|\s*(?:ba|z|k)?sh\b", lower):
return False, "Decoding base64 into a shell is blocked."

if tokens and tokens[0] in ("sudo", "su", "doas"):
log.warning(f"Privilege escalation blocked: {stripped[:80]}")
return False, "Privilege escalation commands (sudo/su/doas) are blocked."
for segment in _segments(stripped):
try:
tokens = shlex.split(segment, posix=True)
except ValueError:
tokens = segment.split()
executable, arguments = _executable(tokens)
if not executable:
continue
if executable in _PRIVILEGE_TOOLS:
log.warning(f"Privilege escalation blocked: {stripped[:80]}")
return False, "Privilege escalation commands (sudo/su/doas) are blocked."
if executable == "eval":
return False, "Shell eval is blocked."
if executable in _SHELLS and _shell_has_command_flag(arguments):
return False, "Nested shell execution (-c) is blocked."
inline_flags = _INLINE_CODE.get(executable, set())
if inline_flags and _has_inline_flag(arguments, inline_flags):
return False, (
f"Inline code execution via {executable} is blocked; "
"write a script file instead."
)

return True, ""

Expand Down
49 changes: 49 additions & 0 deletions tests/test_security.py
Original file line number Diff line number Diff line change
Expand Up @@ -67,6 +67,55 @@ def test_blocks_shutdown(self):
ok, err = validate_command("shutdown -h now")
assert ok is False

def test_blocks_absolute_sudo(self):
ok, err = validate_command("/usr/bin/sudo id")
assert ok is False

def test_blocks_sudo_through_env(self):
ok, err = validate_command("env FOO=bar sudo id")
assert ok is False

def test_blocks_sudo_through_wrapper_options(self):
for command in (
"env -u FOO sudo id",
"env -S 'sudo id'",
"time -f '%E' sudo id",
"timeout 5 sudo id",
"xargs -I {} sudo id",
"stdbuf -o 0 sudo id",
):
ok, err = validate_command(command)
assert ok is False, (command, err)

def test_blocks_nested_shell_command(self):
ok, err = validate_command("bash -lc 'sudo id'")
assert ok is False
ok, err = validate_command('cmd /k "sudo id"')
assert ok is False

def test_blocks_inline_python(self):
ok, err = validate_command("python -c \"import shutil; shutil.rmtree('.')\"")
assert ok is False
ok, err = validate_command("py -c \"import os; os.system('calc')\"")
assert ok is False

def test_blocks_base64_pipe_to_shell(self):
ok, err = validate_command("echo cm0gLXJmIC8= | base64 -d | bash")
assert ok is False

def test_blocks_command_substitution(self):
ok, err = validate_command("echo $(sudo id)")
assert ok is False

def test_allows_python_module_command(self):
ok, err = validate_command("python -m pytest -c pytest.ini -q")
assert ok is True

def test_allows_regular_dev_commands(self):
for command in ("git status", "python --version", "pytest -q", "ls -la"):
ok, err = validate_command(command)
assert ok is True, (command, err)


class TestValidatePath:
def test_accepts_normal_relative_path(self, tmp_path, monkeypatch):
Expand Down
58 changes: 53 additions & 5 deletions tests/test_tools.py
Original file line number Diff line number Diff line change
Expand Up @@ -298,8 +298,9 @@ def test_skips_binary_files(self, tmp_path, monkeypatch):
# ListFilesTool Tests
# ============================================================
class TestListFilesTool:
def test_list_directory(self, tmp_path):
def test_list_directory(self, tmp_path, monkeypatch):
"""Should list all items in a directory"""
monkeypatch.chdir(tmp_path)
(tmp_path / "file1.py").write_text("x=1")
(tmp_path / "file2.txt").write_text("hello")
(tmp_path / "subdir").mkdir()
Expand All @@ -309,18 +310,21 @@ def test_list_directory(self, tmp_path):
assert "file2.txt" in result
assert "subdir" in result

def test_nonexistent_directory(self):
def test_nonexistent_directory(self, tmp_path, monkeypatch):
"""Should return an error for a missing directory"""
result = list_files_tool.invoke({"directory": "/fake/nonexistent/path"})
monkeypatch.chdir(tmp_path)
result = list_files_tool.invoke({"directory": "missing-dir"})
assert "not found" in result.lower()

def test_empty_directory(self, tmp_path):
def test_empty_directory(self, tmp_path, monkeypatch):
"""Should report that an empty directory is empty"""
monkeypatch.chdir(tmp_path)
result = list_files_tool.invoke({"directory": str(tmp_path)})
assert "empty" in result.lower()

def test_file_sizes_shown(self, tmp_path):
def test_file_sizes_shown(self, tmp_path, monkeypatch):
"""File sizes must be displayed"""
monkeypatch.chdir(tmp_path)
(tmp_path / "small.txt").write_text("hi")
result = list_files_tool.invoke({"directory": str(tmp_path)})
assert "B" in result or "KB" in result
Expand Down Expand Up @@ -350,6 +354,50 @@ def test_tools_have_descriptions(self):
assert t.description, f"{t.name} is missing a description!"


class TestReadToolTraversal:
def _setup(self, tmp_path, monkeypatch):
workspace = tmp_path / "workspace"
outside = tmp_path / "outside"
workspace.mkdir()
outside.mkdir()
(outside / "secret.txt").write_text("SECRET_TOKEN=abcd1234", encoding="utf-8")
(workspace / "notes.txt").write_text("a normal needle", encoding="utf-8")
monkeypatch.chdir(workspace)
return workspace, outside

def test_grep_blocks_parent_directory(self, tmp_path, monkeypatch):
_, _ = self._setup(tmp_path, monkeypatch)
result = grep_tool.invoke(
{"pattern": "SECRET_TOKEN", "directory": "../outside"}
)
assert "Security" in result
assert "abcd1234" not in result

def test_grep_blocks_absolute_outside_path(self, tmp_path, monkeypatch):
_, outside = self._setup(tmp_path, monkeypatch)
result = grep_tool.invoke(
{"pattern": "SECRET_TOKEN", "directory": str(outside)}
)
assert "Security" in result
assert "abcd1234" not in result

def test_grep_allows_workspace(self, tmp_path, monkeypatch):
self._setup(tmp_path, monkeypatch)
result = grep_tool.invoke({"pattern": "needle", "directory": "."})
assert "needle" in result

def test_list_files_blocks_parent_directory(self, tmp_path, monkeypatch):
_, _ = self._setup(tmp_path, monkeypatch)
result = list_files_tool.invoke({"directory": "../outside"})
assert "Security" in result
assert "secret.txt" not in result

def test_list_files_allows_workspace(self, tmp_path, monkeypatch):
self._setup(tmp_path, monkeypatch)
result = list_files_tool.invoke({"directory": "."})
assert "notes.txt" in result


if __name__ == "__main__":
import subprocess

Expand Down
Loading
Loading