7a0da7932b
OSV-Scanner (Scheduled) / scan-scheduled (push) Failing after 0s
Create Release / test-gate (push) Has been cancelled
Create Release / release-gate (push) Has been cancelled
Create Release / ci-gate (push) Has been cancelled
Create Release / version-check (push) Has been cancelled
Create Release / e2e-test-gate (push) Has been cancelled
Create Release / responsive-test-gate (push) Has been cancelled
Create Release / compat-test-gate (push) Has been cancelled
Create Release / compose-integration-gate (push) Has been cancelled
Create Release / vulture-gate (push) Has been cancelled
Create Release / build (push) Has been cancelled
Create Release / provenance (push) Has been cancelled
Create Release / prerelease-docker (push) Has been cancelled
Create Release / publish-docker (push) Has been cancelled
Create Release / create-release (push) Has been cancelled
Create Release / cleanup-changelog (push) Has been cancelled
Create Release / trigger-pypi (push) Has been cancelled
Create Release / monitor-pypi (push) Has been cancelled
Create Release / Clean up orphan prerelease tags and signatures (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [research-form] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [research-metrics] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [research-workflow] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [settings-core] (push) Has been cancelled
CodeQL Advanced / Analyze (javascript-typescript) (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [history-news] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [library] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [link-analytics] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [chat-core] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [chat-lifecycle] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [error-benchmark] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [settings-pages] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) (push) Has been cancelled
Docker Tests (Consolidated) / Accessibility Tests (push) Has been cancelled
Docker Tests (Consolidated) / LLM Unit Tests (push) Has been cancelled
Docker Tests (Consolidated) / LLM Example Tests (push) Has been cancelled
Docker Tests (Consolidated) / Production Image Smoke Test (push) Has been cancelled
Docker Tests (Consolidated) / Infrastructure Tests (push) Has been cancelled
OSSF Scorecard / OSSF Security Scorecard Analysis (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [mobile] (push) Has been cancelled
Backwards Compatibility / Verify Encryption Constants (push) Has been cancelled
Backwards Compatibility / PyPI Version Compatibility (push) Has been cancelled
Backwards Compatibility / Database Migration Tests (push) Has been cancelled
CodeQL Advanced / Analyze (python) (push) Has been cancelled
Docker Tests (Consolidated) / detect-changes (push) Has been cancelled
Docker Tests (Consolidated) / Build Test Image (push) Has been cancelled
Docker Tests (Consolidated) / All Pytest Tests + Coverage (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [accessibility] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [api-crud] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [auth-login] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [auth-pages] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [auth-register] (push) Has been cancelled
1041 lines
37 KiB
Python
1041 lines
37 KiB
Python
"""
|
|
Tests for research_library.utils module.
|
|
|
|
Tests path traversal protection and utility functions.
|
|
"""
|
|
|
|
from pathlib import Path
|
|
from unittest.mock import patch, Mock
|
|
|
|
from local_deep_research.constants import (
|
|
FILE_PATH_METADATA_ONLY,
|
|
FILE_PATH_TEXT_ONLY,
|
|
FILE_PATH_BLOB_DELETED,
|
|
)
|
|
|
|
|
|
class TestGetAbsoluteLibraryPath:
|
|
"""Tests for get_absolute_library_path function."""
|
|
|
|
def test_valid_relative_path_returns_absolute_path(self, tmp_path):
|
|
"""Valid relative paths should return absolute paths."""
|
|
from local_deep_research.research_library.utils import (
|
|
get_absolute_library_path,
|
|
)
|
|
|
|
with patch(
|
|
"local_deep_research.research_library.utils.get_library_storage_path"
|
|
) as mock_storage:
|
|
mock_storage.return_value = tmp_path
|
|
|
|
result = get_absolute_library_path("pdfs/test.pdf", "testuser")
|
|
|
|
assert result is not None
|
|
assert result == tmp_path / "pdfs/test.pdf"
|
|
|
|
def test_path_traversal_with_dotdot_returns_none(self, tmp_path):
|
|
"""Path traversal attempts with .. should return None."""
|
|
from local_deep_research.research_library.utils import (
|
|
get_absolute_library_path,
|
|
)
|
|
|
|
with patch(
|
|
"local_deep_research.research_library.utils.get_library_storage_path"
|
|
) as mock_storage:
|
|
mock_storage.return_value = tmp_path
|
|
|
|
result = get_absolute_library_path(
|
|
"../../../etc/passwd", "testuser"
|
|
)
|
|
|
|
assert result is None
|
|
|
|
def test_path_traversal_with_nested_dotdot_returns_none(self, tmp_path):
|
|
"""Path traversal with nested .. should return None."""
|
|
from local_deep_research.research_library.utils import (
|
|
get_absolute_library_path,
|
|
)
|
|
|
|
with patch(
|
|
"local_deep_research.research_library.utils.get_library_storage_path"
|
|
) as mock_storage:
|
|
mock_storage.return_value = tmp_path
|
|
|
|
# Various traversal attempts using actual .. patterns
|
|
traversal_paths = [
|
|
"pdfs/../../secret.txt",
|
|
"./../../etc/passwd",
|
|
"subdir/../../../etc/passwd",
|
|
]
|
|
|
|
for path in traversal_paths:
|
|
result = get_absolute_library_path(path, "testuser")
|
|
assert result is None, f"Path '{path}' should have been blocked"
|
|
|
|
def test_absolute_path_outside_root_returns_none(self, tmp_path):
|
|
"""Absolute paths outside library root should return None."""
|
|
from local_deep_research.research_library.utils import (
|
|
get_absolute_library_path,
|
|
)
|
|
|
|
with patch(
|
|
"local_deep_research.research_library.utils.get_library_storage_path"
|
|
) as mock_storage:
|
|
mock_storage.return_value = tmp_path
|
|
|
|
result = get_absolute_library_path("/etc/passwd", "testuser")
|
|
|
|
assert result is None
|
|
|
|
def test_empty_path_returns_none(self, tmp_path):
|
|
"""Empty path should return None."""
|
|
from local_deep_research.research_library.utils import (
|
|
get_absolute_library_path,
|
|
)
|
|
|
|
with patch(
|
|
"local_deep_research.research_library.utils.get_library_storage_path"
|
|
) as mock_storage:
|
|
mock_storage.return_value = tmp_path
|
|
|
|
result = get_absolute_library_path("", "testuser")
|
|
|
|
assert result is None
|
|
|
|
def test_nested_valid_path_returns_correct_path(self, tmp_path):
|
|
"""Deeply nested valid paths should work correctly."""
|
|
from local_deep_research.research_library.utils import (
|
|
get_absolute_library_path,
|
|
)
|
|
|
|
with patch(
|
|
"local_deep_research.research_library.utils.get_library_storage_path"
|
|
) as mock_storage:
|
|
mock_storage.return_value = tmp_path
|
|
|
|
result = get_absolute_library_path(
|
|
"pdfs/2024/01/document.pdf", "testuser"
|
|
)
|
|
|
|
assert result is not None
|
|
assert result == tmp_path / "pdfs/2024/01/document.pdf"
|
|
|
|
def test_symlink_blocked(self, tmp_path):
|
|
"""Symlinks should be blocked in get_absolute_library_path."""
|
|
import pytest
|
|
|
|
from local_deep_research.research_library.utils import (
|
|
get_absolute_library_path,
|
|
)
|
|
|
|
# Create a symlink inside library root pointing outside
|
|
link_path = tmp_path / "evil_link"
|
|
try:
|
|
link_path.symlink_to("/etc/passwd")
|
|
except OSError:
|
|
pytest.skip("Cannot create symlinks in this environment")
|
|
|
|
with patch(
|
|
"local_deep_research.research_library.utils.get_library_storage_path"
|
|
) as mock_storage:
|
|
mock_storage.return_value = tmp_path
|
|
|
|
result = get_absolute_library_path("evil_link", "testuser")
|
|
assert result is None
|
|
|
|
|
|
class TestGetAbsolutePathFromSettings:
|
|
"""Tests for get_absolute_path_from_settings function."""
|
|
|
|
def test_valid_relative_path_returns_absolute_path(self, tmp_path):
|
|
"""Valid relative paths should return absolute paths."""
|
|
from local_deep_research.research_library.utils import (
|
|
get_absolute_path_from_settings,
|
|
)
|
|
|
|
mock_settings = Mock()
|
|
mock_settings.get_setting.return_value = str(tmp_path)
|
|
|
|
with patch(
|
|
"local_deep_research.utilities.db_utils.get_settings_manager",
|
|
return_value=mock_settings,
|
|
):
|
|
result = get_absolute_path_from_settings("pdfs/test.pdf")
|
|
|
|
assert result is not None
|
|
assert result == tmp_path / "pdfs/test.pdf"
|
|
|
|
def test_empty_path_returns_library_root(self, tmp_path):
|
|
"""Empty path should return the library root."""
|
|
from local_deep_research.research_library.utils import (
|
|
get_absolute_path_from_settings,
|
|
)
|
|
|
|
mock_settings = Mock()
|
|
mock_settings.get_setting.return_value = str(tmp_path)
|
|
|
|
with patch(
|
|
"local_deep_research.utilities.db_utils.get_settings_manager",
|
|
return_value=mock_settings,
|
|
):
|
|
result = get_absolute_path_from_settings("")
|
|
|
|
assert result is not None
|
|
assert result == tmp_path
|
|
|
|
def test_path_traversal_returns_none(self, tmp_path):
|
|
"""Path traversal attempts should return None."""
|
|
from local_deep_research.research_library.utils import (
|
|
get_absolute_path_from_settings,
|
|
)
|
|
|
|
mock_settings = Mock()
|
|
mock_settings.get_setting.return_value = str(tmp_path)
|
|
|
|
with patch(
|
|
"local_deep_research.utilities.db_utils.get_settings_manager",
|
|
return_value=mock_settings,
|
|
):
|
|
traversal_paths = [
|
|
"../../../etc/passwd",
|
|
"pdfs/../../secret.txt",
|
|
]
|
|
|
|
for path in traversal_paths:
|
|
result = get_absolute_path_from_settings(path)
|
|
assert result is None, f"Path '{path}' should have been blocked"
|
|
|
|
def test_symlink_blocked(self, tmp_path):
|
|
"""Symlinks should be blocked in get_absolute_path_from_settings."""
|
|
import pytest
|
|
|
|
from local_deep_research.research_library.utils import (
|
|
get_absolute_path_from_settings,
|
|
)
|
|
|
|
# Create a symlink inside library root pointing outside
|
|
link_path = tmp_path / "evil_link"
|
|
try:
|
|
link_path.symlink_to("/etc/passwd")
|
|
except OSError:
|
|
pytest.skip("Cannot create symlinks in this environment")
|
|
|
|
mock_settings = Mock()
|
|
mock_settings.get_setting.return_value = str(tmp_path)
|
|
|
|
with patch(
|
|
"local_deep_research.utilities.db_utils.get_settings_manager",
|
|
return_value=mock_settings,
|
|
):
|
|
result = get_absolute_path_from_settings("evil_link")
|
|
assert result is None
|
|
|
|
|
|
class TestLibraryServiceSafeAbsolutePath:
|
|
"""Tests for LibraryService._get_safe_absolute_path method."""
|
|
|
|
def test_metadata_only_returns_none(self):
|
|
"""metadata_only marker should return None."""
|
|
from local_deep_research.research_library.services.library_service import (
|
|
LibraryService,
|
|
)
|
|
|
|
service = LibraryService("testuser")
|
|
|
|
result = service._get_safe_absolute_path(FILE_PATH_METADATA_ONLY)
|
|
|
|
assert result is None
|
|
|
|
def test_text_only_not_stored_returns_none(self):
|
|
"""text_only_not_stored marker should return None."""
|
|
from local_deep_research.research_library.services.library_service import (
|
|
LibraryService,
|
|
)
|
|
|
|
service = LibraryService("testuser")
|
|
|
|
result = service._get_safe_absolute_path(FILE_PATH_TEXT_ONLY)
|
|
|
|
assert result is None
|
|
|
|
def test_empty_path_returns_none(self):
|
|
"""Empty path should return None."""
|
|
from local_deep_research.research_library.services.library_service import (
|
|
LibraryService,
|
|
)
|
|
|
|
service = LibraryService("testuser")
|
|
|
|
result = service._get_safe_absolute_path("")
|
|
|
|
assert result is None
|
|
|
|
def test_none_path_returns_none(self):
|
|
"""None path should return None."""
|
|
from local_deep_research.research_library.services.library_service import (
|
|
LibraryService,
|
|
)
|
|
|
|
service = LibraryService("testuser")
|
|
|
|
result = service._get_safe_absolute_path(None)
|
|
|
|
assert result is None
|
|
|
|
def test_valid_path_returns_string(self, tmp_path):
|
|
"""Valid paths should return a string."""
|
|
from local_deep_research.research_library.services.library_service import (
|
|
LibraryService,
|
|
)
|
|
|
|
mock_settings = Mock()
|
|
mock_settings.get_setting.return_value = str(tmp_path)
|
|
|
|
with patch(
|
|
"local_deep_research.utilities.db_utils.get_settings_manager",
|
|
return_value=mock_settings,
|
|
):
|
|
service = LibraryService("testuser")
|
|
|
|
result = service._get_safe_absolute_path("pdfs/test.pdf")
|
|
|
|
assert result is not None
|
|
assert isinstance(result, str)
|
|
assert result == str(tmp_path / "pdfs/test.pdf")
|
|
|
|
def test_path_traversal_returns_none(self, tmp_path):
|
|
"""Path traversal attempts should return None."""
|
|
from local_deep_research.research_library.services.library_service import (
|
|
LibraryService,
|
|
)
|
|
|
|
mock_settings = Mock()
|
|
mock_settings.get_setting.return_value = str(tmp_path)
|
|
|
|
with patch(
|
|
"local_deep_research.utilities.db_utils.get_settings_manager",
|
|
return_value=mock_settings,
|
|
):
|
|
service = LibraryService("testuser")
|
|
|
|
result = service._get_safe_absolute_path("../../../etc/passwd")
|
|
|
|
assert result is None
|
|
|
|
|
|
class TestPDFStorageManagerSafeFilePath:
|
|
"""Tests for PDFStorageManager._get_safe_file_path method."""
|
|
|
|
def test_valid_path_returns_path_object(self, tmp_path):
|
|
"""Valid relative paths should return Path objects."""
|
|
from local_deep_research.research_library.services.pdf_storage_manager import (
|
|
PDFStorageManager,
|
|
)
|
|
|
|
manager = PDFStorageManager(
|
|
library_root=tmp_path,
|
|
storage_mode="filesystem",
|
|
)
|
|
|
|
result = manager._get_safe_file_path("pdfs/test.pdf")
|
|
|
|
assert result is not None
|
|
assert isinstance(result, Path)
|
|
assert result == tmp_path / "pdfs/test.pdf"
|
|
|
|
def test_metadata_only_returns_none(self, tmp_path):
|
|
"""metadata_only marker should return None."""
|
|
from local_deep_research.research_library.services.pdf_storage_manager import (
|
|
PDFStorageManager,
|
|
)
|
|
|
|
manager = PDFStorageManager(
|
|
library_root=tmp_path,
|
|
storage_mode="filesystem",
|
|
)
|
|
|
|
result = manager._get_safe_file_path(FILE_PATH_METADATA_ONLY)
|
|
|
|
assert result is None
|
|
|
|
def test_text_only_not_stored_returns_none(self, tmp_path):
|
|
"""text_only_not_stored marker should return None."""
|
|
from local_deep_research.research_library.services.pdf_storage_manager import (
|
|
PDFStorageManager,
|
|
)
|
|
|
|
manager = PDFStorageManager(
|
|
library_root=tmp_path,
|
|
storage_mode="filesystem",
|
|
)
|
|
|
|
result = manager._get_safe_file_path(FILE_PATH_TEXT_ONLY)
|
|
|
|
assert result is None
|
|
|
|
def test_empty_path_returns_none(self, tmp_path):
|
|
"""Empty path should return None."""
|
|
from local_deep_research.research_library.services.pdf_storage_manager import (
|
|
PDFStorageManager,
|
|
)
|
|
|
|
manager = PDFStorageManager(
|
|
library_root=tmp_path,
|
|
storage_mode="filesystem",
|
|
)
|
|
|
|
result = manager._get_safe_file_path("")
|
|
|
|
assert result is None
|
|
|
|
def test_path_traversal_returns_none(self, tmp_path):
|
|
"""Path traversal attempts should return None."""
|
|
from local_deep_research.research_library.services.pdf_storage_manager import (
|
|
PDFStorageManager,
|
|
)
|
|
|
|
manager = PDFStorageManager(
|
|
library_root=tmp_path,
|
|
storage_mode="filesystem",
|
|
)
|
|
|
|
traversal_paths = [
|
|
"../../../etc/passwd",
|
|
"pdfs/../../secret.txt",
|
|
]
|
|
|
|
for path in traversal_paths:
|
|
result = manager._get_safe_file_path(path)
|
|
assert result is None, f"Path '{path}' should have been blocked"
|
|
|
|
def test_none_path_returns_none(self, tmp_path):
|
|
"""None path should return None."""
|
|
from local_deep_research.research_library.services.pdf_storage_manager import (
|
|
PDFStorageManager,
|
|
)
|
|
|
|
manager = PDFStorageManager(
|
|
library_root=tmp_path,
|
|
storage_mode="filesystem",
|
|
)
|
|
|
|
result = manager._get_safe_file_path(None)
|
|
|
|
assert result is None
|
|
|
|
|
|
class TestPathTraversalIntegration:
|
|
"""Integration tests for path traversal protection across components."""
|
|
|
|
def test_malicious_document_file_path_blocked_in_has_pdf(self, tmp_path):
|
|
"""PDFStorageManager.has_pdf should handle malicious paths safely."""
|
|
from unittest.mock import patch
|
|
|
|
from local_deep_research.research_library.services.pdf_storage_manager import (
|
|
PDFStorageManager,
|
|
)
|
|
|
|
manager = PDFStorageManager(
|
|
library_root=tmp_path,
|
|
storage_mode="filesystem",
|
|
)
|
|
|
|
# Create a mock document with malicious file_path
|
|
mock_doc = Mock()
|
|
mock_doc.file_type = "pdf"
|
|
mock_doc.file_path = "../../../etc/passwd"
|
|
|
|
mock_session = Mock()
|
|
mock_session.query.return_value.filter_by.return_value.first.return_value = None
|
|
|
|
# Patch DocumentBlob so its .id attribute is accessible outside SQLAlchemy
|
|
mock_blob_class = Mock()
|
|
with patch(
|
|
"local_deep_research.database.models.library.DocumentBlob",
|
|
mock_blob_class,
|
|
):
|
|
# Should not raise an error, just return False
|
|
result = manager.has_pdf(mock_doc, mock_session)
|
|
|
|
assert result is False
|
|
|
|
def test_malicious_document_file_path_blocked_in_load(self, tmp_path):
|
|
"""PDFStorageManager._load_from_filesystem should handle malicious paths."""
|
|
from local_deep_research.research_library.services.pdf_storage_manager import (
|
|
PDFStorageManager,
|
|
)
|
|
|
|
manager = PDFStorageManager(
|
|
library_root=tmp_path,
|
|
storage_mode="filesystem",
|
|
)
|
|
|
|
# Create a mock document with malicious file_path
|
|
mock_doc = Mock()
|
|
mock_doc.file_path = "../../../etc/passwd"
|
|
|
|
# Should return None, not raise or attempt to read /etc/passwd
|
|
result = manager._load_from_filesystem(mock_doc)
|
|
|
|
assert result is None
|
|
|
|
def test_malicious_document_file_path_blocked_in_delete(self, tmp_path):
|
|
"""PDFStorageManager.delete_pdf should handle malicious paths safely."""
|
|
from local_deep_research.research_library.services.pdf_storage_manager import (
|
|
PDFStorageManager,
|
|
)
|
|
|
|
manager = PDFStorageManager(
|
|
library_root=tmp_path,
|
|
storage_mode="filesystem",
|
|
)
|
|
|
|
# Create a mock document with malicious file_path
|
|
mock_doc = Mock()
|
|
mock_doc.file_path = "../../../etc/passwd"
|
|
mock_doc.storage_mode = "filesystem"
|
|
|
|
mock_session = Mock()
|
|
mock_session.query.return_value.filter_by.return_value.first.return_value = None
|
|
|
|
# Should not raise an error or delete /etc/passwd
|
|
result = manager.delete_pdf(mock_doc, mock_session)
|
|
|
|
# Should return True (successful "deletion" since there was nothing to delete)
|
|
# The important thing is it doesn't try to delete outside the library
|
|
assert result is True
|
|
|
|
def test_is_already_downloaded_handles_malicious_path(self, tmp_path):
|
|
"""DownloadService.is_already_downloaded should not crash on malicious paths."""
|
|
from local_deep_research.research_library.services.download_service import (
|
|
DownloadService,
|
|
)
|
|
from local_deep_research.database.models.download_tracker import (
|
|
DownloadTracker,
|
|
)
|
|
|
|
# Create a mock tracker with malicious file_path
|
|
mock_tracker = Mock(spec=DownloadTracker)
|
|
mock_tracker.file_path = "../../../etc/passwd"
|
|
mock_tracker.is_downloaded = True
|
|
|
|
mock_session = Mock()
|
|
mock_session.query.return_value.filter_by.return_value.first.return_value = mock_tracker
|
|
|
|
mock_settings = Mock()
|
|
mock_settings.get_setting.return_value = str(tmp_path)
|
|
|
|
with (
|
|
patch(
|
|
"local_deep_research.utilities.db_utils.get_settings_manager",
|
|
return_value=mock_settings,
|
|
),
|
|
patch(
|
|
"local_deep_research.research_library.services.download_service.get_user_db_session"
|
|
) as mock_db_session,
|
|
):
|
|
# Make the context manager return our mock session
|
|
mock_db_session.return_value.__enter__ = Mock(
|
|
return_value=mock_session
|
|
)
|
|
mock_db_session.return_value.__exit__ = Mock(return_value=False)
|
|
|
|
service = DownloadService.__new__(DownloadService)
|
|
service.username = "testuser"
|
|
service.password = None
|
|
|
|
# Should not raise AttributeError on None.exists()
|
|
# Should return (False, None) because path traversal is blocked
|
|
result = service.is_already_downloaded(
|
|
"https://example.com/doc.pdf"
|
|
)
|
|
|
|
assert result == (False, None)
|
|
|
|
|
|
class TestPathTraversalAttackVectors:
|
|
"""Test various path traversal attack vectors are blocked."""
|
|
|
|
def test_simple_dotdot_traversal(self, tmp_path):
|
|
"""Basic ../ traversal should be blocked."""
|
|
from local_deep_research.research_library.utils import (
|
|
get_absolute_path_from_settings,
|
|
)
|
|
|
|
mock_settings = Mock()
|
|
mock_settings.get_setting.return_value = str(tmp_path)
|
|
|
|
with patch(
|
|
"local_deep_research.utilities.db_utils.get_settings_manager",
|
|
return_value=mock_settings,
|
|
):
|
|
attacks = [
|
|
"../secret.txt",
|
|
"../../secret.txt",
|
|
"../../../etc/passwd",
|
|
"../../../../etc/shadow",
|
|
]
|
|
for attack in attacks:
|
|
result = get_absolute_path_from_settings(attack)
|
|
assert result is None, f"Attack '{attack}' should be blocked"
|
|
|
|
def test_nested_traversal_escaping_root(self, tmp_path):
|
|
"""Traversal that escapes library root should be blocked."""
|
|
from local_deep_research.research_library.utils import (
|
|
get_absolute_path_from_settings,
|
|
)
|
|
|
|
mock_settings = Mock()
|
|
mock_settings.get_setting.return_value = str(tmp_path)
|
|
|
|
with patch(
|
|
"local_deep_research.utilities.db_utils.get_settings_manager",
|
|
return_value=mock_settings,
|
|
):
|
|
# These paths escape the library root
|
|
attacks = [
|
|
"pdfs/../../../etc/passwd",
|
|
"documents/subfolder/../../../../../../tmp/evil",
|
|
"a/b/c/d/../../../../../../../../etc/passwd",
|
|
]
|
|
for attack in attacks:
|
|
result = get_absolute_path_from_settings(attack)
|
|
assert result is None, f"Attack '{attack}' should be blocked"
|
|
|
|
def test_traversal_staying_in_root_is_allowed(self, tmp_path):
|
|
"""Traversal that stays within library root should be allowed."""
|
|
from local_deep_research.research_library.utils import (
|
|
get_absolute_path_from_settings,
|
|
)
|
|
|
|
mock_settings = Mock()
|
|
mock_settings.get_setting.return_value = str(tmp_path)
|
|
|
|
with patch(
|
|
"local_deep_research.utilities.db_utils.get_settings_manager",
|
|
return_value=mock_settings,
|
|
):
|
|
# This path uses .. but stays within library root
|
|
# valid/path/to/../../../outside resolves to "outside" which is in root
|
|
result = get_absolute_path_from_settings(
|
|
"valid/path/to/../../../outside"
|
|
)
|
|
# This is actually safe - resolves to tmp_path/outside
|
|
assert result is not None
|
|
assert result == tmp_path / "outside"
|
|
|
|
def test_absolute_path_injection(self, tmp_path):
|
|
"""Absolute paths should be blocked."""
|
|
from local_deep_research.research_library.utils import (
|
|
get_absolute_path_from_settings,
|
|
)
|
|
|
|
mock_settings = Mock()
|
|
mock_settings.get_setting.return_value = str(tmp_path)
|
|
|
|
with patch(
|
|
"local_deep_research.utilities.db_utils.get_settings_manager",
|
|
return_value=mock_settings,
|
|
):
|
|
attacks = [
|
|
"/etc/passwd",
|
|
"/etc/shadow",
|
|
"/root/.ssh/id_rsa",
|
|
"/var/log/auth.log",
|
|
]
|
|
for attack in attacks:
|
|
result = get_absolute_path_from_settings(attack)
|
|
assert result is None, f"Attack '{attack}' should be blocked"
|
|
|
|
def test_dot_current_dir_traversal(self, tmp_path):
|
|
"""Traversal using ./ patterns should be blocked when escaping."""
|
|
from local_deep_research.research_library.utils import (
|
|
get_absolute_path_from_settings,
|
|
)
|
|
|
|
mock_settings = Mock()
|
|
mock_settings.get_setting.return_value = str(tmp_path)
|
|
|
|
with patch(
|
|
"local_deep_research.utilities.db_utils.get_settings_manager",
|
|
return_value=mock_settings,
|
|
):
|
|
attacks = [
|
|
"./../../../etc/passwd",
|
|
"./././../../../etc/passwd",
|
|
"././../secret.txt",
|
|
]
|
|
for attack in attacks:
|
|
result = get_absolute_path_from_settings(attack)
|
|
assert result is None, f"Attack '{attack}' should be blocked"
|
|
|
|
def test_double_slash_traversal(self, tmp_path):
|
|
"""Double slashes with traversal should be handled."""
|
|
from local_deep_research.research_library.utils import (
|
|
get_absolute_path_from_settings,
|
|
)
|
|
|
|
mock_settings = Mock()
|
|
mock_settings.get_setting.return_value = str(tmp_path)
|
|
|
|
with patch(
|
|
"local_deep_research.utilities.db_utils.get_settings_manager",
|
|
return_value=mock_settings,
|
|
):
|
|
# These should be blocked due to .. patterns
|
|
attacks = [
|
|
"..//..//etc/passwd",
|
|
"pdfs//..//..//secret.txt",
|
|
]
|
|
for attack in attacks:
|
|
result = get_absolute_path_from_settings(attack)
|
|
assert result is None, f"Attack '{attack}' should be blocked"
|
|
|
|
def test_backslash_traversal_windows_style(self, tmp_path):
|
|
"""Windows-style backslash traversal - safe on Linux, blocked on Windows.
|
|
|
|
On Linux, backslashes are valid filename characters, not path separators.
|
|
So '..\\..\\etc\\passwd' is treated as a literal filename, which is safe.
|
|
On Windows, werkzeug's safe_join would block these.
|
|
"""
|
|
import sys
|
|
|
|
from local_deep_research.research_library.utils import (
|
|
get_absolute_path_from_settings,
|
|
)
|
|
|
|
mock_settings = Mock()
|
|
mock_settings.get_setting.return_value = str(tmp_path)
|
|
|
|
with patch(
|
|
"local_deep_research.utilities.db_utils.get_settings_manager",
|
|
return_value=mock_settings,
|
|
):
|
|
attacks = [
|
|
"..\\..\\..\\etc\\passwd",
|
|
"..\\..\\windows\\win.ini",
|
|
]
|
|
for attack in attacks:
|
|
result = get_absolute_path_from_settings(attack)
|
|
if sys.platform == "win32":
|
|
# On Windows, these should be blocked
|
|
assert result is None, (
|
|
f"Attack '{attack}' should be blocked"
|
|
)
|
|
else:
|
|
# On Linux, backslashes are literal - path stays in library
|
|
if result is not None:
|
|
assert str(tmp_path) in str(result)
|
|
|
|
def test_mixed_slash_with_real_traversal(self, tmp_path):
|
|
"""Mixed slash patterns with real .. traversal should be blocked."""
|
|
from local_deep_research.research_library.utils import (
|
|
get_absolute_path_from_settings,
|
|
)
|
|
|
|
mock_settings = Mock()
|
|
mock_settings.get_setting.return_value = str(tmp_path)
|
|
|
|
with patch(
|
|
"local_deep_research.utilities.db_utils.get_settings_manager",
|
|
return_value=mock_settings,
|
|
):
|
|
# These contain real forward-slash traversal that escapes root
|
|
attacks = [
|
|
"pdfs/../../../etc/passwd",
|
|
"subdir/../../../../../../etc/passwd",
|
|
]
|
|
for attack in attacks:
|
|
result = get_absolute_path_from_settings(attack)
|
|
assert result is None, f"Attack '{attack}' should be blocked"
|
|
|
|
def test_mangled_patterns_stay_in_library(self, tmp_path):
|
|
"""Mangled patterns like '..../' don't escape library.
|
|
|
|
Pattern '....' is treated as a literal directory name, not traversal.
|
|
These patterns are safe because they stay within the library root.
|
|
"""
|
|
from local_deep_research.research_library.utils import (
|
|
get_absolute_path_from_settings,
|
|
)
|
|
|
|
mock_settings = Mock()
|
|
mock_settings.get_setting.return_value = str(tmp_path)
|
|
|
|
with patch(
|
|
"local_deep_research.utilities.db_utils.get_settings_manager",
|
|
return_value=mock_settings,
|
|
):
|
|
# These don't actually traverse - '....' is a literal dirname
|
|
patterns = [
|
|
"....//....//file.pdf",
|
|
"..../file.pdf",
|
|
]
|
|
for pattern in patterns:
|
|
result = get_absolute_path_from_settings(pattern)
|
|
# These are allowed because they stay in library
|
|
if result is not None:
|
|
assert str(tmp_path) in str(result)
|
|
|
|
def test_null_byte_injection(self, tmp_path):
|
|
"""Null byte injection should be blocked or handled safely.
|
|
|
|
Null bytes (%00) can truncate paths in some languages (PHP < 5.3.4).
|
|
While Python handles this better, we should still test it.
|
|
"""
|
|
from local_deep_research.research_library.utils import (
|
|
get_absolute_path_from_settings,
|
|
)
|
|
|
|
mock_settings = Mock()
|
|
mock_settings.get_setting.return_value = str(tmp_path)
|
|
|
|
with patch(
|
|
"local_deep_research.utilities.db_utils.get_settings_manager",
|
|
return_value=mock_settings,
|
|
):
|
|
# Null byte injection patterns
|
|
attacks = [
|
|
"../../../etc/passwd\x00.pdf",
|
|
"pdfs/test.pdf\x00/../../../etc/passwd",
|
|
"../../../etc/passwd%00.pdf", # URL-encoded null (if decoded)
|
|
]
|
|
for attack in attacks:
|
|
result = get_absolute_path_from_settings(attack)
|
|
assert result is None, (
|
|
f"Null byte attack '{attack!r}' should be blocked"
|
|
)
|
|
|
|
def test_unc_path_injection_windows(self, tmp_path):
|
|
"""UNC paths (Windows network shares) should be blocked.
|
|
|
|
UNC paths like \\\\server\\share can be used for SSRF or credential theft.
|
|
"""
|
|
from local_deep_research.research_library.utils import (
|
|
get_absolute_path_from_settings,
|
|
)
|
|
|
|
mock_settings = Mock()
|
|
mock_settings.get_setting.return_value = str(tmp_path)
|
|
|
|
with patch(
|
|
"local_deep_research.utilities.db_utils.get_settings_manager",
|
|
return_value=mock_settings,
|
|
):
|
|
# UNC path patterns
|
|
attacks = [
|
|
"\\\\attacker.com\\share\\malicious.pdf",
|
|
"\\\\localhost\\c$\\windows\\win.ini",
|
|
"//attacker.com/share/file.pdf",
|
|
]
|
|
for attack in attacks:
|
|
result = get_absolute_path_from_settings(attack)
|
|
# Should be blocked or stay within library
|
|
if result is not None:
|
|
assert str(tmp_path) in str(result), (
|
|
f"Attack '{attack}' resulted in path outside library"
|
|
)
|
|
|
|
def test_dot_segment_variations(self, tmp_path):
|
|
"""Various dot segment patterns should be handled.
|
|
|
|
Tests patterns like '...' and '....' which some systems mishandle.
|
|
"""
|
|
from local_deep_research.research_library.utils import (
|
|
get_absolute_path_from_settings,
|
|
)
|
|
|
|
mock_settings = Mock()
|
|
mock_settings.get_setting.return_value = str(tmp_path)
|
|
|
|
with patch(
|
|
"local_deep_research.utilities.db_utils.get_settings_manager",
|
|
return_value=mock_settings,
|
|
):
|
|
# Various dot patterns
|
|
attacks = [
|
|
".../.../.../etc/passwd",
|
|
"..../..../etc/passwd",
|
|
"..;/..;/..;/etc/passwd", # Tomcat bypass
|
|
]
|
|
for attack in attacks:
|
|
result = get_absolute_path_from_settings(attack)
|
|
# These are unusual patterns - should either be blocked or safe
|
|
if result is not None:
|
|
assert str(tmp_path) in str(result), (
|
|
f"Attack '{attack}' resulted in path outside library"
|
|
)
|
|
|
|
|
|
class TestPathTraversalEdgeCases:
|
|
"""Test edge cases and boundary conditions."""
|
|
|
|
def test_whitespace_paths(self, tmp_path):
|
|
"""Paths with only whitespace should be handled."""
|
|
from local_deep_research.research_library.utils import (
|
|
get_absolute_path_from_settings,
|
|
)
|
|
|
|
mock_settings = Mock()
|
|
mock_settings.get_setting.return_value = str(tmp_path)
|
|
|
|
with patch(
|
|
"local_deep_research.utilities.db_utils.get_settings_manager",
|
|
return_value=mock_settings,
|
|
):
|
|
# Whitespace-only paths should return None or be sanitized
|
|
whitespace_paths = [" ", "\t", "\n", " \t\n "]
|
|
for path in whitespace_paths:
|
|
result = get_absolute_path_from_settings(path)
|
|
# After strip(), these become empty, which should return library_root
|
|
# or None depending on implementation
|
|
assert result is None or result == tmp_path
|
|
|
|
def test_path_with_spaces(self, tmp_path):
|
|
"""Valid paths with spaces should work."""
|
|
from local_deep_research.research_library.utils import (
|
|
get_absolute_path_from_settings,
|
|
)
|
|
|
|
mock_settings = Mock()
|
|
mock_settings.get_setting.return_value = str(tmp_path)
|
|
|
|
with patch(
|
|
"local_deep_research.utilities.db_utils.get_settings_manager",
|
|
return_value=mock_settings,
|
|
):
|
|
result = get_absolute_path_from_settings("pdfs/my document.pdf")
|
|
assert result is not None
|
|
assert result == tmp_path / "pdfs/my document.pdf"
|
|
|
|
def test_path_with_special_characters(self, tmp_path):
|
|
"""Paths with allowed special characters should work."""
|
|
from local_deep_research.research_library.utils import (
|
|
get_absolute_path_from_settings,
|
|
)
|
|
|
|
mock_settings = Mock()
|
|
mock_settings.get_setting.return_value = str(tmp_path)
|
|
|
|
with patch(
|
|
"local_deep_research.utilities.db_utils.get_settings_manager",
|
|
return_value=mock_settings,
|
|
):
|
|
# Valid special characters
|
|
valid_paths = [
|
|
"pdfs/test-file.pdf",
|
|
"pdfs/test_file.pdf",
|
|
"pdfs/test.file.pdf",
|
|
]
|
|
for path in valid_paths:
|
|
result = get_absolute_path_from_settings(path)
|
|
assert result is not None, f"Path '{path}' should be allowed"
|
|
assert result == tmp_path / path
|
|
|
|
def test_very_long_path(self, tmp_path):
|
|
"""Very long paths should be handled without crashing."""
|
|
from local_deep_research.research_library.utils import (
|
|
get_absolute_path_from_settings,
|
|
)
|
|
|
|
mock_settings = Mock()
|
|
mock_settings.get_setting.return_value = str(tmp_path)
|
|
|
|
with patch(
|
|
"local_deep_research.utilities.db_utils.get_settings_manager",
|
|
return_value=mock_settings,
|
|
):
|
|
# Create a very long but valid path
|
|
long_path = "/".join(["subdir"] * 50) + "/file.pdf"
|
|
result = get_absolute_path_from_settings(long_path)
|
|
# Should either work or fail gracefully (no crash)
|
|
assert result is None or isinstance(result, Path)
|
|
|
|
def test_unicode_path(self, tmp_path):
|
|
"""Unicode paths should be handled."""
|
|
from local_deep_research.research_library.utils import (
|
|
get_absolute_path_from_settings,
|
|
)
|
|
|
|
mock_settings = Mock()
|
|
mock_settings.get_setting.return_value = str(tmp_path)
|
|
|
|
with patch(
|
|
"local_deep_research.utilities.db_utils.get_settings_manager",
|
|
return_value=mock_settings,
|
|
):
|
|
# Unicode filenames are common
|
|
unicode_paths = [
|
|
"pdfs/документ.pdf", # Russian
|
|
"pdfs/文档.pdf", # Chinese
|
|
"pdfs/ドキュメント.pdf", # Japanese
|
|
]
|
|
for path in unicode_paths:
|
|
# Should not crash - just verify no exception is raised
|
|
_ = get_absolute_path_from_settings(path)
|
|
# Result depends on filesystem support
|
|
|
|
|
|
class TestCascadeHelperPathSafety:
|
|
"""Test CascadeHelper path handling."""
|
|
|
|
def test_delete_filesystem_file_rejects_special_markers(self):
|
|
"""CascadeHelper should reject special path markers."""
|
|
from local_deep_research.research_library.deletion.utils.cascade_helper import (
|
|
CascadeHelper,
|
|
)
|
|
|
|
markers = [
|
|
FILE_PATH_METADATA_ONLY,
|
|
FILE_PATH_TEXT_ONLY,
|
|
FILE_PATH_BLOB_DELETED,
|
|
]
|
|
for marker in markers:
|
|
result = CascadeHelper.delete_filesystem_file(marker)
|
|
assert result is False, f"Marker '{marker}' should be rejected"
|
|
|
|
def test_delete_filesystem_file_handles_none(self):
|
|
"""CascadeHelper should handle None path."""
|
|
from local_deep_research.research_library.deletion.utils.cascade_helper import (
|
|
CascadeHelper,
|
|
)
|
|
|
|
result = CascadeHelper.delete_filesystem_file(None)
|
|
assert result is False
|
|
|
|
def test_delete_filesystem_file_handles_empty_string(self):
|
|
"""CascadeHelper should handle empty string path."""
|
|
from local_deep_research.research_library.deletion.utils.cascade_helper import (
|
|
CascadeHelper,
|
|
)
|
|
|
|
result = CascadeHelper.delete_filesystem_file("")
|
|
assert result is False
|
|
|
|
def test_delete_filesystem_file_nonexistent_returns_false(self, tmp_path):
|
|
"""CascadeHelper should return False for nonexistent files."""
|
|
from local_deep_research.research_library.deletion.utils.cascade_helper import (
|
|
CascadeHelper,
|
|
)
|
|
|
|
result = CascadeHelper.delete_filesystem_file(
|
|
str(tmp_path / "nonexistent.pdf")
|
|
)
|
|
assert result is False
|
|
|
|
def test_delete_filesystem_file_deletes_valid_file(self, tmp_path):
|
|
"""CascadeHelper should delete valid files."""
|
|
from local_deep_research.research_library.deletion.utils.cascade_helper import (
|
|
CascadeHelper,
|
|
)
|
|
|
|
# Create a test file
|
|
test_file = tmp_path / "test.pdf"
|
|
test_file.write_text("test content")
|
|
assert test_file.exists()
|
|
|
|
result = CascadeHelper.delete_filesystem_file(str(test_file))
|
|
|
|
assert result is True
|
|
assert not test_file.exists()
|