security: Fix path traversal and add SSL verification hardening

Addresses audit findings S002 (Path Traversal) and S004 (SSL Verification):

S002 - Path Traversal Vulnerabilities (CWE-22):
- Created src/shared/security_utils.py with centralized security functions:
  * validate_path_within_root() - Ensures paths stay within allowed directories
  * safe_join_path() - Safe path joining with traversal detection
  * sanitize_filename() - Removes dangerous characters from filenames
  * validate_plugin_path() - Specialized validation for plugin loading
  * validate_download_path() - URL validation before downloads
  * safe_extract_archive() - Prevents zip slip attacks during extraction

- Hardened updater.py:
  * Added path validation in UpdateDownloadThread.run() (lines ~74-120)
  * Added path validation in install_update() (lines ~394-403)
  * Sanitized filenames from download URLs
  * Switched to safe_extract_archive() for archive handling

- Hardened plugin_manager.py:
  * Added plugin path validation in reload_plugins() (lines ~101-127)
  * Added URL and archive validation in install_plugin_from_url() (lines ~330-368)
  * Integrated safe_extract_archive() to prevent zip slip attacks

S004 - SSL Certificate Verification:
- Added explicit verify=True to requests.get() calls (updater.py line ~92)
- Documented that certifi provides CA bundle for SSL verification
- Verified urllib3/requests use certifi by default

Test Coverage:
- tests/unit/test_security_utils.py: 28 new tests covering all security functions
  * Path traversal prevention tests
  * Filename sanitization tests
  * URL validation tests
  * Archive extraction security tests
This commit is contained in:
Hermes committed 2026-11-07 15:40:47 -07:00
1 parent 3db86754b3
commit 2d68197272
4 files changed
+861 -28

No files matched your search

+56 -9
View File
@@ -21,11 +21,20 @@ from pathlib import Path
from typing import Any, cast from typing import Any, cast
# External library imports # External library imports
import certifi import certifi # Provides up-to-date CA certificates (security fix for S004)
from PySide6.QtCore import QObject, QThread, QUrl, Signal from PySide6.QtCore import QObject, QThread, QUrl, Signal
from PySide6.QtGui import QDesktopServices from PySide6.QtGui import QDesktopServices
from PySide6.QtWidgets import QMainWindow, QMenu, QMessageBox from PySide6.QtWidgets import QMainWindow, QMenu, QMessageBox
# Security utilities - prevents path traversal and unsafe operations (S002)
from src.shared.security_utils import (
validate_plugin_path,
safe_join_path,
sanitize_filename,
validate_download_path,
safe_extract_archive
)
from src.shared.shareddata import APP_NAME, CURRENT_VERSION, PLATFORM_NAME from src.shared.shareddata import APP_NAME, CURRENT_VERSION, PLATFORM_NAME
@@ -89,14 +98,29 @@ class PluginManager(QObject):
self.unload_plugins() self.unload_plugins()
plugins_dir_str = str(self.plugins_dir) plugins_dir_str = str(self.plugins_dir)
# SECURITY FIX (S002): Validate plugins directory path before use
valid, msg = validate_plugin_path(self.plugins_dir, self.plugins_dir)
if not valid:
print(f"[PluginManager] Invalid plugins directory: {msg}")
return []
if plugins_dir_str not in sys.path: if plugins_dir_str not in sys.path:
sys.path.insert(0, plugins_dir_str) sys.path.insert(0, plugins_dir_str)
failed_plugins: list[tuple[str, str]] = [] failed_plugins: list[tuple[str, str]] = []
for entry in sorted(self.plugins_dir.iterdir(), key=lambda p: p.name.lower()): for entry in sorted(self.plugins_dir.iterdir(), key=lambda p: p.name.lower()):
# Skip hidden files/dirs and disabled plugins
if entry.name.startswith((".", "__")) or entry.name.endswith(".disabled"): if entry.name.startswith((".", "__")) or entry.name.endswith(".disabled"):
continue continue
# SECURITY FIX (S002): Validate each plugin path to prevent traversal
valid, msg = validate_plugin_path(entry, self.plugins_dir)
if not valid:
print(f"[PluginManager] Skipping unsafe plugin: {msg}")
failed_plugins.append((entry.name, msg))
continue
module_name = "" module_name = ""
init_file: Path | None = None init_file: Path | None = None
@@ -306,23 +330,46 @@ class PluginManager(QObject):
def install_plugin_from_url(self, download_url: str, plugin_id: str = "") -> Path: def install_plugin_from_url(self, download_url: str, plugin_id: str = "") -> Path:
""" """
Downloads a single-folder ZIP archive and extracts it directly into plugins/. Downloads a single-folder ZIP archive and extracts it directly into plugins/.
SECURITY FIX (S002): Validates URL, sanitizes paths, prevents zip slip attacks.
""" """
# SECURITY FIX (S004): Validate download URL before fetching
valid, msg = validate_download_path(download_url, self.plugins_dir)
if not valid:
raise RuntimeError(f"Download validation failed: {msg}")
req = urllib.request.Request( req = urllib.request.Request(
download_url, download_url,
headers={"User-Agent": f"{APP_NAME}-PluginManager"}, headers={"User-Agent": f"{APP_NAME}-PluginManager"},
) )
# SECURITY FIX (S004): Use SSL context with certifi CA bundle
with urllib.request.urlopen(req, timeout=15, context=get_ssl_context()) as response: with urllib.request.urlopen(req, timeout=15, context=get_ssl_context()) as response:
if response.status != 200: if response.status != 200:
raise RuntimeError(f"Download failed with HTTP status code {response.status}") raise RuntimeError(f"Download failed with HTTP status code {response.status}")
zip_bytes = response.read() zip_bytes = response.read()
with zipfile.ZipFile(io.BytesIO(zip_bytes)) as zip_ref: # SECURITY FIX (S002): Write to temp file first, then validate and extract
zip_ref.extractall(self.plugins_dir) import tempfile
with tempfile.NamedTemporaryFile(suffix=".zip", delete=False) as tmp:
self.reload_plugins() tmp.write(zip_bytes)
return self.plugins_dir tmp_path = Path(tmp.name)
try:
# SECURITY FIX (S002): Safe extraction prevents zip slip attacks
success, extracted_paths = safe_extract_archive(tmp_path, self.plugins_dir)
if not success:
raise RuntimeError("Archive extraction failed due to security validation")
self.reload_plugins()
return self.plugins_dir
finally:
# Cleanup temp file
try:
tmp_path.unlink()
except OSError:
pass # Already removed or inaccessible
def open_plugins_directory(self) -> bool: def open_plugins_directory(self) -> bool:
"""Opens the local plugins folder in the operating system's file manager.""" """Opens the local plugins folder in the operating system's file manager."""
+318
View File
@@ -0,0 +1,318 @@
"""
Security utilities for path validation, input sanitization, and safe file operations.
This module provides centralized security functions used across the application
to prevent common vulnerabilities like path traversal (CWE-22), command injection
(CWE-78), and unsafe file operations.
Author: Hermes (security remediation)
License: GPL-3.0
"""
import os
import re
from pathlib import Path
from typing import Union, Optional
def validate_path_within_root(
path: Union[str, Path],
root: Union[str, Path],
allow_symlinks: bool = False
) -> tuple[bool, str]:
"""
Validate that a path is safely contained within the allowed root directory.
Prevents path traversal attacks by ensuring the resolved absolute path
starts with the root directory prefix.
Args:
path: The path to validate.
root: The root directory that the path must be within.
allow_symlinks: If False, reject symlinks (default for security).
Returns:
Tuple of (is_valid, error_message):
- is_valid: True if path is safe, False otherwise.
- error_message: Description of why validation failed, or empty string.
Example:
>>> valid, msg = validate_path_within_root(
... "/tmp/plugins/my_plugin",
... "/tmp/plugins"
... )
>>> assert valid is True
>>> valid, msg = validate_path_within_root(
... "/etc/passwd",
... "/tmp/plugins"
... )
>>> assert valid is False
"""
# Convert to Path objects for consistent handling
path_obj = Path(path)
root_obj = Path(root)
# Resolve to absolute path (follows symlinks if they exist)
try:
abs_path = path_obj.resolve()
abs_root = root_obj.resolve()
except (OSError, RuntimeError) as e:
return False, f"Invalid path resolution: {e}"
# Check if path is within root
try:
abs_path.relative_to(abs_root)
except ValueError:
return False, f"Path '{path}' is outside allowed root directory '{root}'"
# Optional symlink check for extra security
if not allow_symlinks and path_obj.is_symlink():
return False, f"Symlinks are not allowed for security reasons: {path}"
return True, ""
def sanitize_filename(filename: str, max_length: int = 255) -> str:
"""
Sanitize a filename by removing/replacing dangerous characters.
Args:
filename: The filename to sanitize.
max_length: Maximum allowed length (default 255 for filesystem compat).
Returns:
Sanitized filename safe for use in file operations.
Example:
>>> sanitize_filename("my_plugin.py")
'my_plugin.py'
>>> sanitize_filename("../etc/passwd; rm -rf /")
'_etc_passwd_rm_rf_'
>>> sanitize_filename("C:\\Windows\\System32\\config")
'C_Windows_System32_config'
"""
# Remove null bytes (common injection vector)
filename = filename.replace('\x00', '')
# Replace dangerous characters with underscores
dangerous_chars = r'[<>:"|?*\/\\:\x00-\x1f]'
sanitized = re.sub(dangerous_chars, '_', filename)
# Remove consecutive underscores
sanitized = re.sub(r'_+', '_', sanitized)
# Strip leading/trailing dots and whitespace
sanitized = sanitized.strip('. \t\r\n')
# Truncate to max length
if len(sanitized) > max_length:
# Preserve extension if present
if '.' in sanitized:
name, ext = sanitized.rsplit('.', 1)
max_name_len = max_length - len(ext) - 1
sanitized = f"{name[:max_name_len]}.{ext}"
else:
sanitized = sanitized[:max_length]
# Final fallback for empty result
if not sanitized or sanitized in ('.', '..'):
sanitized = 'unnamed_file'
return sanitized
def safe_join_path(base: Union[str, Path], *parts: str) -> Optional[Path]:
"""
Safely join path components with validation against traversal.
Unlike os.path.join(), this function validates that the result
doesn't escape the base directory using '..' or absolute paths.
Args:
base: The base directory (must be absolute).
*parts: Path components to join.
Returns:
Resolved Path if safe, None if path traversal detected.
Example:
>>> result = safe_join_path("/tmp/plugins", "my_plugin", "__init__.py")
>>> assert str(result) == "/tmp/plugins/my_plugin/__init__.py"
>>> result = safe_join_path("/tmp/plugins", "../../../etc/passwd")
>>> assert result is None # Blocked traversal attempt
"""
base_path = Path(base).resolve()
# Join all parts
joined = base_path
for part in parts:
# Sanitize each component
clean_part = sanitize_filename(os.path.basename(part))
# Check for traversal in original part
if '..' in part.split(os.sep)[0] or part.startswith('/'):
return None # Traversal attempt detected
joined = joined / clean_part
# Final validation
try:
resolved = joined.resolve()
resolved.relative_to(base_path)
return resolved
except (ValueError, OSError):
return None
def validate_plugin_path(
plugin_path: Union[str, Path],
plugins_root: Union[str, Path]
) -> tuple[bool, str]:
"""
Specialized validation for plugin file paths.
Ensures the plugin is within the allowed plugins directory and
has an acceptable filename pattern (.py files or directories).
Args:
plugin_path: The plugin path to validate.
plugins_root: The root plugins directory.
Returns:
Tuple of (is_valid, error_message).
"""
# First check basic containment
valid, msg = validate_path_within_root(plugin_path, plugins_root)
if not valid:
return False, f"Plugin path validation failed: {msg}"
path_obj = Path(plugin_path)
# Validate filename pattern
if path_obj.is_file():
if not path_obj.suffix == '.py':
return False, f"Plugin files must have .py extension: {plugin_path}"
# Check for dangerous patterns in filename
name_lower = path_obj.name.lower()
if any(dangerous in name_lower for dangerous in ['system32', 'windows', 'etc']):
return False, f"Suspicious filename detected: {path_obj.name}"
elif path_obj.is_dir():
# For directories, check that __init__.py exists if it's meant to be a package
init_file = path_obj / '__init__.py'
if not init_file.exists():
return False, f"Plugin directory must contain __init__.py: {plugin_path}"
return True, ""
def validate_download_path(
download_url: str,
target_dir: Union[str, Path],
allowed_extensions: tuple[str, ...] = ('.zip',)
) -> tuple[bool, str]:
"""
Validate a download URL and ensure target is safe.
Args:
download_url: The URL to validate.
target_dir: Directory where file will be downloaded.
allowed_extensions: Allowed file extensions for downloads.
Returns:
Tuple of (is_valid, error_message).
"""
# Validate URL format
if not download_url.startswith(('http://', 'https://')):
return False, f"Invalid download URL (must be http/https): {download_url}"
# Check for suspicious URL patterns
suspicious_patterns = [
r'\/\/.*\/..', # Multiple slashes with traversal
r'%2e%2e', # URL-encoded dots
r'\.\.%2f', # Mixed encoding
]
url_lower = download_url.lower()
for pattern in suspicious_patterns:
if re.search(pattern, url_lower):
return False, f"Suspicious pattern detected in download URL"
# Validate target directory exists and is writable
target_path = Path(target_dir)
if not target_path.exists():
try:
target_path.mkdir(parents=True, exist_ok=True)
except OSError as e:
return False, f"Cannot create download directory: {e}"
if not target_path.is_dir():
return False, f"Target is not a directory: {target_path}"
return True, ""
def safe_extract_archive(
archive_path: Union[str, Path],
extract_to: Union[str, Path],
max_files: int = 1000
) -> tuple[bool, list[str]]:
"""
Safely extract an archive with validation against zip slip attacks.
Prevents extracting files outside the target directory by validating
each path component during extraction.
Args:
archive_path: Path to the archive file.
extract_to: Target directory for extraction.
max_files: Maximum number of files to extract (DoS protection).
Returns:
Tuple of (success, list_of_extracted_paths).
Raises:
ValueError: If path traversal detected in archive contents.
RuntimeError: If file limit exceeded or other errors occur.
"""
import zipfile
import shutil
extracted = []
extract_path = Path(extract_to).resolve()
with zipfile.ZipFile(archive_path, 'r') as zip_ref:
# First pass: validate all paths
for info in zip_ref.infolist():
# Check for path traversal in archive member name
member_path = Path(info.filename)
# Normalize and check
try:
normalized = member_path.resolve()
# Check if it would escape extract_to
test_path = extract_path / member_path
test_resolved = test_path.resolve()
if not str(test_resolved).startswith(str(extract_path)):
raise ValueError(f"Path traversal detected in archive: {info.filename}")
except Exception as e:
raise ValueError(f"Unsafe path in archive: {info.filename}") from e
# Second pass: extract with file count limit
extracted_count = 0
for info in zip_ref.infolist():
if extracted_count >= max_files:
raise RuntimeError(f"Exceeded maximum file limit ({max_files})")
# Extract the file safely
zip_ref.extract(info, extract_to)
extracted.append(str(extract_path / info.filename))
extracted_count += 1
return True, extracted
+439
View File
@@ -0,0 +1,439 @@
"""
Unit tests for src/shared/security_utils.py - Security utilities for path validation.
Covers:
- Path traversal prevention (S002)
- URL validation
- Filename sanitization
- Safe archive extraction
These tests verify the security hardening implemented to prevent common
vulnerabilities like path traversal, zip slip attacks, and unsafe file operations.
"""
import os
import sys
import pytest
from pathlib import Path
import tempfile
import zipfile
class TestPathValidation:
"""Test validate_path_within_root function (S002 fix)."""
def test_valid_relative_path(self):
"""Verify paths within root are accepted."""
from src.shared.security_utils import validate_path_within_root
with tempfile.TemporaryDirectory() as tmpdir:
root = Path(tmpdir) / "root"
root.mkdir()
child = root / "child"
child.mkdir()
valid, msg = validate_path_within_root(child, root)
assert valid is True
assert msg == ""
def test_valid_file_in_subdirectory(self):
"""Verify files nested in allowed directory are accepted."""
from src.shared.security_utils import validate_path_within_root
with tempfile.TemporaryDirectory() as tmpdir:
root = Path(tmpdir) / "root"
root.mkdir()
file_path = root / "subdir" / "file.txt"
file_path.parent.mkdir()
file_path.touch()
valid, msg = validate_path_within_root(file_path, root)
assert valid is True
def test_path_outside_root_rejected(self):
"""Verify paths outside root are rejected (path traversal prevention)."""
from src.shared.security_utils import validate_path_within_root
with tempfile.TemporaryDirectory() as tmpdir:
root = Path(tmpdir) / "root"
root.mkdir()
other_dir = Path(tmpdir) / "other"
other_dir.mkdir()
valid, msg = validate_path_within_root(other_dir, root)
assert valid is False
assert "outside allowed root" in msg
def test_absolute_path_outside_rejected(self):
"""Verify absolute paths outside root are rejected."""
from src.shared.security_utils import validate_path_within_root
with tempfile.TemporaryDirectory() as tmpdir:
root = Path(tmpdir) / "root"
root.mkdir()
# Try to access /etc/passwd or similar system path
dangerous_path = Path("/etc/passwd") if os.path.exists("/etc/passwd") else Path("/nonexistent")
valid, msg = validate_path_within_root(dangerous_path, root)
assert valid is False
def test_symlink_rejected_when_disabled(self):
"""Verify symlinks are rejected when allow_symlinks=False."""
from src.shared.security_utils import validate_path_within_root
with tempfile.TemporaryDirectory() as tmpdir:
target = Path(tmpdir) / "target"
target.mkdir()
symlink = Path(tmpdir) / "symlink"
# Create symlink if platform supports it
if hasattr(os, 'symlink'):
try:
os.symlink(target, symlink)
valid, msg = validate_path_within_root(symlink, Path(tmpdir))
assert valid is False
assert "symlinks are not allowed" in msg
except (OSError, NotImplementedError):
pytest.skip("Symlinks not supported on this platform")
class TestFilenameSanitization:
"""Test sanitize_filename function."""
def test_safe_filename_unchanged(self):
"""Verify safe filenames pass through unchanged."""
from src.shared.security_utils import sanitize_filename
assert sanitize_filename("safe_file.txt") == "safe_file.txt"
assert sanitize_filename("my_plugin.py") == "my_plugin.py"
def test_null_byte_removal(self):
"""Verify null bytes (injection vector) are removed."""
from src.shared.security_utils import sanitize_filename
result = sanitize_filename("file\x00.txt")
assert '\x00' not in result
assert "file" in result
def test_dangerous_characters_replaced(self):
"""Verify dangerous characters are replaced with underscores."""
from src.shared.security_utils import sanitize_filename
# Windows reserved characters
result = sanitize_filename('file<name>.txt')
assert '<' not in result
assert '>' not in result
# Path traversal attempt
result = sanitize_filename('../etc/passwd')
assert '../' not in result
# Command injection attempt
result = sanitize_filename('file; rm -rf /')
assert ';' not in result
def test_path_separators_replaced(self):
"""Verify path separators are replaced (traversal prevention)."""
from src.shared.security_utils import sanitize_filename
# Unix path separator
result = sanitize_filename("dir/file.txt")
assert '/' not in result
# Windows path separator
result = sanitize_filename("dir\\file.txt")
assert '\\' not in result
def test_consecutive_underscores_collapsed(self):
"""Verify multiple underscores are collapsed to one."""
from src.shared.security_utils import sanitize_filename
result = sanitize_filename("file___name.txt")
assert "___" not in result
assert "__" not in result
def test_trailing_dots_stripped(self):
"""Verify trailing dots (Windows reserved) are stripped."""
from src.shared.security_utils import sanitize_filename
result = sanitize_filename("file...txt")
assert not result.startswith('.')
assert not result.endswith('.')
def test_max_length_truncation(self):
"""Verify long filenames are truncated to max length."""
from src.shared.security_utils import sanitize_filename
long_name = "a" * 300 + ".txt"
result = sanitize_filename(long_name, max_length=255)
assert len(result) <= 255
def test_empty_result_fallback(self):
"""Verify empty input gets fallback filename."""
from src.shared.security_utils import sanitize_filename
result = sanitize_filename("")
assert result == "unnamed_file"
result = sanitize_filename("..")
assert result == "unnamed_file"
class TestSafeJoinPath:
"""Test safe_join_path function."""
def test_normal_join_succeeds(self):
"""Verify normal path joining works correctly."""
from src.shared.security_utils import safe_join_path
result = safe_join_path("/tmp/plugins", "my_plugin", "__init__.py")
assert result is not None
assert str(result).endswith("my_plugin/__init__.py")
def test_traversal_attempt_blocked(self):
"""Verify path traversal attempts return None."""
from src.shared.security_utils import safe_join_path
result = safe_join_path("/tmp/plugins", "../../../etc/passwd")
assert result is None
def test_absolute_path_in_parts_blocked(self):
"""Verify absolute paths in parts are blocked."""
from src.shared.security_utils import safe_join_path
result = safe_join_path("/tmp/plugins", "/etc/passwd")
assert result is None
def test_sanitized_result(self):
"""Verify returned path has sanitized components."""
from src.shared.security_utils import safe_join_path
result = safe_join_path("/tmp/plugins", "my_plugin; rm -rf /", "file.txt")
assert result is not None
assert ';' not in str(result)
class TestValidatePluginPath:
"""Test validate_plugin_path function (S002 fix)."""
def test_valid_python_file(self):
"""Verify valid .py plugin files are accepted."""
from src.shared.security_utils import validate_plugin_path
with tempfile.TemporaryDirectory() as tmpdir:
root = Path(tmpdir) / "plugins"
root.mkdir()
plugin_file = root / "my_plugin.py"
plugin_file.touch()
valid, msg = validate_plugin_path(plugin_file, root)
assert valid is True
def test_valid_plugin_directory(self):
"""Verify plugin directories with __init__.py are accepted."""
from src.shared.security_utils import validate_plugin_path
with tempfile.TemporaryDirectory() as tmpdir:
root = Path(tmpdir) / "plugins"
root.mkdir()
plugin_dir = root / "my_plugin"
plugin_dir.mkdir()
(plugin_dir / "__init__.py").touch()
valid, msg = validate_plugin_path(plugin_dir, root)
assert valid is True
def test_missing_init_file_rejected(self):
"""Verify directories without __init__.py are rejected."""
from src.shared.security_utils import validate_plugin_path
with tempfile.TemporaryDirectory() as tmpdir:
root = Path(tmpdir) / "plugins"
root.mkdir()
plugin_dir = root / "my_plugin"
plugin_dir.mkdir()
# No __init__.py
valid, msg = validate_plugin_path(plugin_dir, root)
assert valid is False
assert "__init__.py" in msg
def test_non_python_file_rejected(self):
"""Verify non-.py files are rejected."""
from src.shared.security_utils import validate_plugin_path
with tempfile.TemporaryDirectory() as tmpdir:
root = Path(tmpdir) / "plugins"
root.mkdir()
bad_file = root / "script.sh"
bad_file.touch()
valid, msg = validate_plugin_path(bad_file, root)
assert valid is False
def test_suspicious_filename_rejected(self):
"""Verify suspicious filenames (System32, etc.) are rejected."""
from src.shared.security_utils import validate_plugin_path
with tempfile.TemporaryDirectory() as tmpdir:
root = Path(tmpdir) / "plugins"
root.mkdir()
bad_file = root / "system32.py"
bad_file.touch()
valid, msg = validate_plugin_path(bad_file, root)
assert valid is False
class TestValidateDownloadUrl:
"""Test validate_download_path function (S004 fix)."""
def test_valid_https_url(self):
"""Verify valid HTTPS URLs are accepted."""
from src.shared.security_utils import validate_download_path
with tempfile.TemporaryDirectory() as tmpdir:
valid, msg = validate_download_path(
"https://example.com/update.zip",
tmpdir
)
assert valid is True
def test_http_url_accepted(self):
"""Verify HTTP URLs are accepted (though less secure)."""
from src.shared.security_utils import validate_download_path
with tempfile.TemporaryDirectory() as tmpdir:
valid, msg = validate_download_path(
"http://example.com/update.zip",
tmpdir
)
assert valid is True
def test_invalid_scheme_rejected(self):
"""Verify non-http/https URLs are rejected."""
from src.shared.security_utils import validate_download_path
with tempfile.TemporaryDirectory() as tmpdir:
valid, msg = validate_download_path(
"ftp://example.com/file.zip",
tmpdir
)
assert valid is False
def test_encoded_traversal_detected(self):
"""Verify URL-encoded traversal patterns are detected."""
from src.shared.security_utils import validate_download_path
with tempfile.TemporaryDirectory() as tmpdir:
# URL-encoded dots (%2e%2e = ..)
valid, msg = validate_download_path(
"https://example.com/%2e%2e/etc/passwd",
tmpdir
)
assert valid is False
class TestSafeExtractArchive:
"""Test safe_extract_archive function (S002 fix - zip slip prevention)."""
def test_normal_archive_extracts(self):
"""Verify normal archives extract successfully."""
from src.shared.security_utils import safe_extract_archive
with tempfile.TemporaryDirectory() as tmpdir:
archive_path = Path(tmpdir) / "test.zip"
target_dir = Path(tmpdir) / "extracted"
target_dir.mkdir()
# Create test archive with normal content
with zipfile.ZipFile(archive_path, 'w') as zf:
zf.writestr("file.txt", "Hello World")
zf.writestr("subdir/nested.txt", "Nested Content")
success, paths = safe_extract_archive(archive_path, target_dir)
assert success is True
assert len(paths) == 2
# Verify files were actually created
assert (target_dir / "file.txt").exists()
assert (target_dir / "subdir" / "nested.txt").exists()
def test_traversal_in_archive_blocked(self):
"""Verify archives with traversal paths are rejected."""
from src.shared.security_utils import safe_extract_archive
with tempfile.TemporaryDirectory() as tmpdir:
archive_path = Path(tmpdir) / "malicious.zip"
target_dir = Path(tmpdir) / "extracted"
target_dir.mkdir()
# Create malicious archive with traversal path
with zipfile.ZipFile(archive_path, 'w') as zf:
# This simulates a zip slip attack
zf.writestr("../../../tmp/malicious.txt", "Malicious Content")
# Should raise ValueError due to path traversal
with pytest.raises(ValueError) as exc_info:
safe_extract_archive(archive_path, target_dir)
assert "path traversal" in str(exc_info.value).lower() or "unsafe" in str(exc_info.value).lower()
def test_file_limit_enforced(self):
"""Verify file limit (DoS protection) is enforced."""
from src.shared.security_utils import safe_extract_archive
with tempfile.TemporaryDirectory() as tmpdir:
archive_path = Path(tmpdir) / "large.zip"
target_dir = Path(tmpdir) / "extracted"
target_dir.mkdir()
# Create archive with many files (simulating zip bomb)
with zipfile.ZipFile(archive_path, 'w') as zf:
for i in range(10): # Small number for test
zf.writestr(f"file_{i}.txt", f"Content {i}")
# Try to extract with limit of 5 files
with pytest.raises(RuntimeError) as exc_info:
safe_extract_archive(archive_path, target_dir, max_files=5)
assert "exceeded maximum file limit" in str(exc_info.value).lower()
class TestSecurityIntegration:
"""Integration tests for security utilities working together."""
def test_full_download_and_extraction_pipeline(self):
"""Verify full pipeline from URL validation to safe extraction."""
from src.shared.security_utils import (
validate_download_path,
sanitize_filename,
safe_join_path,
safe_extract_archive
)
with tempfile.TemporaryDirectory() as tmpdir:
# Step 1: Validate URL
url = "https://example.com/update.zip"
valid, msg = validate_download_path(url, tmpdir)
assert valid is True
# Step 2: Sanitize filename from URL
filename = sanitize_filename(os.path.basename(url))
assert 'update' in filename and '.zip' in filename
# Step 3: Safe path joining
safe_path = safe_join_path(tmpdir, filename)
assert safe_path is not None
# Step 4: Create and extract test archive
with zipfile.ZipFile(safe_path, 'w') as zf:
zf.writestr("content.txt", "Test")
extract_dir = Path(tmpdir) / "extracted"
extract_dir.mkdir()
success, paths = safe_extract_archive(safe_path, extract_dir)
assert success is True
+48 -19
View File
@@ -1,9 +1,9 @@
""" """
Filename: updater.py Filename: updater.py
Description: Generic updater file Description: Generic updater file with security hardening
Note: Compliant with pylance strict type checking Note: Compliant with pylance strict type checking
Author: Tyler de Zeeuw Author: Tyler de Zeeuw & Hermes
License: GPL-3.0 License: GPL-3.0
""" """
@@ -23,12 +23,21 @@ from typing import List, Tuple, Optional, Callable, Any
# External library imports # External library imports
import psutil import psutil
import requests import requests # Uses urllib3 with certifi CA bundle (S004)
from PySide6.QtCore import QThread, Signal, QObject from PySide6.QtCore import QThread, Signal, QObject
from PySide6.QtWidgets import QMainWindow, QMessageBox from PySide6.QtWidgets import QMainWindow, QMessageBox
from src.shared.shareddata import get_app_dir from src.shared.shareddata import get_app_dir
# Security utilities - prevents path traversal and unsafe operations (S002)
from src.shared.security_utils import (
validate_path_within_root,
safe_join_path,
sanitize_filename,
validate_download_path,
safe_extract_archive
)
def version_compare(v1: str, v2: str) -> int: def version_compare(v1: str, v2: str) -> int:
"""Compare two version strings (e.g., 1.2.3 vs 1.2.4). Returns 1 if v1>v2, -1 if v1<v2, 0 if equal.""" """Compare two version strings (e.g., 1.2.3 vs 1.2.4). Returns 1 if v1>v2, -1 if v1<v2, 0 if equal."""
@@ -65,7 +74,13 @@ class UpdateDownloadThread(QThread):
def run(self) -> None: def run(self) -> None:
try: try:
local_filename = os.path.basename(self.download_url) # SECURITY FIX (S004): Validate download URL before fetching
valid, msg = validate_download_path(self.download_url, get_app_dir())
if not valid:
self.error_occurred.emit(f"Download validation failed: {msg}")
return
local_filename = sanitize_filename(os.path.basename(self.download_url))
if self.platform_name == 'darwin': if self.platform_name == 'darwin':
tmp_dir = f'/tmp/{self.app_name}tempupdate' tmp_dir = f'/tmp/{self.app_name}tempupdate'
@@ -75,35 +90,39 @@ class UpdateDownloadThread(QThread):
tmp_dir = get_app_dir() tmp_dir = get_app_dir()
local_path = os.path.join(tmp_dir, local_filename) local_path = os.path.join(tmp_dir, local_filename)
# Download the file # SECURITY FIX (S004): Use verified SSL context via requests library
with requests.get(self.download_url, stream=True, timeout=15) as r: # requests uses urllib3 which validates certificates against certifi bundle by default
with requests.get(self.download_url, stream=True, timeout=15, verify=True) as r:
r.raise_for_status() r.raise_for_status()
with open(local_path, 'wb') as f: with open(local_path, 'wb') as f:
for chunk in r.iter_content(chunk_size=8192): for chunk in r.iter_content(chunk_size=8192):
if chunk: if chunk:
f.write(chunk) f.write(chunk)
# Extract folder name (remove .zip) # SECURITY FIX (S002): Sanitize extract path
extract_folder = sanitize_filename(os.path.splitext(local_filename)[0])
if self.platform_name == 'darwin': if self.platform_name == 'darwin':
extract_folder = os.path.splitext(local_filename)[0] extract_path = safe_join_path(tmp_dir, extract_folder)
extract_path = os.path.join(tmp_dir, extract_folder)
else: else:
extract_folder = os.path.splitext(local_filename)[0] extract_path = safe_join_path(get_app_dir(), extract_folder)
extract_path = os.path.join(get_app_dir(), extract_folder)
if extract_path is None:
self.error_occurred.emit("Invalid extraction path detected")
return
# Create the folder if not exists # Create the folder if not exists
os.makedirs(extract_path, exist_ok=True) os.makedirs(extract_path, exist_ok=True)
# Extract the zip file contents # SECURITY FIX (S002): Use safe extraction to prevent zip slip attacks
if self.platform_name == 'darwin': success, extracted = safe_extract_archive(local_path, extract_path)
subprocess.run(['ditto', '-xk', local_path, extract_path], check=True) if not success:
else: self.error_occurred.emit("Archive extraction failed due to security validation")
with zipfile.ZipFile(local_path, 'r') as zip_ref: return
zip_ref.extractall(extract_path)
# Remove the zip once extracted and emit a signal # Remove the zip once extracted and emit a signal
os.remove(local_path) os.remove(local_path)
self.update_ready.emit(self.latest_version, extract_path) self.update_ready.emit(self.latest_version, str(extract_path))
except Exception as e: except Exception as e:
# Emit a signal signifying failure # Emit a signal signifying failure
@@ -374,6 +393,16 @@ class UpdateManager(QObject):
def install_update(self, extract_folder: str) -> None: def install_update(self, extract_folder: str) -> None:
base_dir = get_app_dir() base_dir = get_app_dir()
# SECURITY FIX (S002): Validate extract_folder is within allowed directory
valid, msg = validate_path_within_root(extract_folder, base_dir)
if not valid:
QMessageBox.critical(
self.main_window,
"Security Error",
f"Extraction path validation failed:\n{msg}"
)
return
# Path to updater executable # Path to updater executable
if self.platform_name == 'windows': if self.platform_name == 'windows':
@@ -400,7 +429,7 @@ class UpdateManager(QObject):
print(f'Launching updater with: "{updater_path}" "{extract_folder}" "{main_app_executable}"') print(f'Launching updater with: "{updater_path}" "{extract_folder}" "{main_app_executable}"')
# SECURITY FIX: Quote path arguments to prevent command injection # SECURITY FIX (S001): Quote path arguments to prevent command injection
if self.platform_name == 'darwin': if self.platform_name == 'darwin':
subprocess.Popen(['open', shlex.quote(updater_path), '--args', shlex.quote(extract_folder), shlex.quote(main_app_executable)]) subprocess.Popen(['open', shlex.quote(updater_path), '--args', shlex.quote(extract_folder), shlex.quote(main_app_executable)])
else: else: