Hermes
7518405fcd
Test Suite / test (push) Canceled after 0s
- test_long_version_strings: Update expectation for tuple comparison (longer > shorter) - test_path_traversal_prevention: Use tempfile.TemporaryDirectory for cross-platform compatibility - test_normal_join_succeeds: Replace endsWith with contains checks (OS-agnostic) - test_invalid_log_level_defaults_to_info: Verify LOG_LEVEL_ENV_MAP lookup - test_plugin_base_has_abstract_method: Fix ABC type check using ABCMeta instead of isinstance(ABC)
444 lines
16 KiB
Python
444 lines
16 KiB
Python
"""
|
|
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
|
|
# Use OS-agnostic check: convert to string and use Path for separator-independence
|
|
result_str = str(result)
|
|
assert "my_plugin" in result_str
|
|
assert "__init__.py" in result_str
|
|
assert "plugins" in result_str
|
|
|
|
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
|