fix: sanitize shell input

- Replaced unsafe string concatenation with `shlex.join()` for POSIX systems, `mslex.join()` for Windows systems.
- Add utility function `safe_join`.
- Modified `pyproject.toml` to include `mslex` dependency for Windows environments.
This commit is contained in:
Aditya Bavadekar
2025-10-27 13:33:24 +05:30
parent 38355bc1ea
commit c2a695d466
4 changed files with 48 additions and 9 deletions
@@ -10,6 +10,8 @@ import subprocess
import urllib.parse
from typing import Any, Dict, List, Optional
from computer.utils import safe_join
# Setup logging
logger = logging.getLogger(__name__)
@@ -59,7 +61,7 @@ def lume_api_get(
# --max-time: Maximum time for the whole operation (20 seconds)
# -f: Fail silently (no output at all) on server errors
# Add single quotes around URL to ensure special characters are handled correctly
cmd = ["curl", "--connect-timeout", "15", "--max-time", "20", "-s", "-f", f"'{api_url}'"]
cmd = ["curl", "--connect-timeout", "15", "--max-time", "20", "-s", "-f", api_url]
# For logging and display, show the properly escaped URL
display_cmd = ["curl", "--connect-timeout", "15", "--max-time", "20", "-s", "-f", api_url]
@@ -71,7 +73,7 @@ def lume_api_get(
# Execute the command - for execution we need to use shell=True to handle URLs with special characters
try:
# Use a single string with shell=True for proper URL handling
shell_cmd = " ".join(cmd)
shell_cmd = safe_join(cmd)
result = subprocess.run(shell_cmd, shell=True, capture_output=True, text=True)
# Handle curl exit codes
@@ -514,7 +516,7 @@ def lume_api_delete(
"-s",
"-X",
"DELETE",
f"'{api_url}'",
api_url,
]
# For logging and display, show the properly escaped URL
@@ -537,7 +539,7 @@ def lume_api_delete(
# Execute the command - for execution we need to use shell=True to handle URLs with special characters
try:
# Use a single string with shell=True for proper URL handling
shell_cmd = " ".join(cmd)
shell_cmd = safe_join(cmd)
result = subprocess.run(shell_cmd, shell=True, capture_output=True, text=True)
# Handle curl exit codes
+25
View File
@@ -1,7 +1,10 @@
import base64
import io
import os
import shlex
from typing import Any, Dict, Optional, Tuple
import mslex
from PIL import Image, ImageDraw
@@ -104,3 +107,25 @@ def parse_vm_info(vm_info: Dict[str, Any]) -> Optional[Dict[str, Any]]:
"""Parse VM info from pylume response."""
if not vm_info:
return None
def safe_join(argv: list[str]) -> str:
"""
Return a platform-correct string that safely quotes `argv` for shell execution.
- On POSIX: uses `shlex.join`.
- On Windows: uses `shlex.join`.
Args:
argv: iterable of argument strings (will be coerced to str).
Returns:
A safely quoted command-line string appropriate for the current platform that protects against
shell injection vulnerabilities.
"""
if os.name == "nt":
# On Windows, use mslex for proper quoting
return mslex.join(argv)
else:
# On POSIX systems, use shlex
return shlex.join(argv)
+3 -2
View File
@@ -16,7 +16,8 @@ dependencies = [
"websockets>=12.0",
"aiohttp>=3.9.0",
"cua-core>=0.1.0,<0.2.0",
"pydantic>=2.11.1"
"pydantic>=2.11.1",
"mslex>=1.3.0",
]
requires-python = ">=3.12"
@@ -47,4 +48,4 @@ source-includes = ["tests/", "README.md", "LICENSE"]
[tool.pytest.ini_options]
asyncio_mode = "auto"
testpaths = ["tests"]
python_files = "test_*.py"
python_files = "test_*.py"