kodi/webmanager-mvp
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: download - download safeguard

Browse Source
This commit is contained in:
kodi
2026-03-14 13:24:17 +01:00
parent 7e7c2f3958
commit ea337338e3
4 changed files with 248 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files
webui/backend/tests/golden/__pycache__/test_api_download_golden.cpython-313.pyc
BIN
View File
Binary file not shown.
webui/backend/tests/golden/test_api_download_golden.py
+104 -10
View File
@@ -16,7 +16,7 @@ from backend.app.dependencies import get_file_ops_service
from backend.app.fs.filesystem_adapter import FilesystemAdapter
from backend.app.main import app
from backend.app.security.path_guard import PathGuard
from backend.app.services.file_ops_service import FileOpsService
from backend.app.services.file_ops_service import FileOpsService, ZipDownloadPreflightLimits
class DownloadApiGoldenTest(unittest.TestCase):
@@ -24,13 +24,9 @@ class DownloadApiGoldenTest(unittest.TestCase):
self.temp_dir = tempfile.TemporaryDirectory()
self.root = Path(self.temp_dir.name) / "root"
self.root.mkdir(parents=True, exist_ok=True)
path_guard = PathGuard({"storage1": str(self.root), "storage2": str(self.root)})
service = FileOpsService(path_guard=path_guard, filesystem=FilesystemAdapter())
async def _override_file_ops_service() -> FileOpsService:
return service
app.dependency_overrides[get_file_ops_service] = _override_file_ops_service
self.path_guard = PathGuard({"storage1": str(self.root), "storage2": str(self.root)})
self.filesystem = FilesystemAdapter()
self._override_service()
def tearDown(self) -> None:
app.dependency_overrides.clear()
@@ -44,6 +40,24 @@ class DownloadApiGoldenTest(unittest.TestCase):
return asyncio.run(_run())
def _override_service(
self,
*,
limits: ZipDownloadPreflightLimits | None = None,
monotonic=None,
) -> None:
service = FileOpsService(
path_guard=self.path_guard,
filesystem=self.filesystem,
zip_download_preflight_limits=limits or ZipDownloadPreflightLimits(),
monotonic=monotonic,
)
async def _override_file_ops_service() -> FileOpsService:
return service
app.dependency_overrides[get_file_ops_service] = _override_file_ops_service
def test_download_success_for_allowed_file(self) -> None:
src = self.root / "report.txt"
src.write_text("hello download", encoding="utf-8")
@@ -55,7 +69,7 @@ class DownloadApiGoldenTest(unittest.TestCase):
self.assertIn('attachment; filename="report.txt"', response.headers.get("content-disposition", ""))
self.assertEqual(response.headers.get("content-type"), "text/plain; charset=utf-8")
def test_download_directory_type_conflict(self) -> None:
def test_download_single_directory_as_zip(self) -> None:
(self.root / "docs").mkdir()
(self.root / "docs" / "a.txt").write_text("a", encoding="utf-8")
@@ -121,6 +135,64 @@ class DownloadApiGoldenTest(unittest.TestCase):
self.assertIn("photos/", archive.namelist())
self.assertIn("photos/nested/img.txt", archive.namelist())
def test_download_zip_rejected_when_max_items_exceeded(self) -> None:
(self.root / "docs").mkdir()
(self.root / "docs" / "a.txt").write_text("A", encoding="utf-8")
(self.root / "docs" / "b.txt").write_text("B", encoding="utf-8")
(self.root / "docs" / "c.txt").write_text("C", encoding="utf-8")
self._override_service(
limits=ZipDownloadPreflightLimits(
max_items=3,
max_total_input_bytes=1024,
max_individual_file_bytes=1024,
scan_timeout_seconds=10.0,
)
)
response = self._get("/api/files/download?path=storage1/docs")
self.assertEqual(response.status_code, 409)
self.assertEqual(response.json()["error"]["code"], "download_preflight_failed")
self.assertEqual(response.json()["error"]["message"], "Zip download preflight failed")
self.assertEqual(response.json()["error"]["details"]["reason"], "max_items_exceeded")
def test_download_zip_rejected_when_max_total_input_bytes_exceeded(self) -> None:
(self.root / "a.txt").write_text("AAAA", encoding="utf-8")
(self.root / "b.txt").write_text("BBBB", encoding="utf-8")
self._override_service(
limits=ZipDownloadPreflightLimits(
max_items=10,
max_total_input_bytes=7,
max_individual_file_bytes=1024,
scan_timeout_seconds=10.0,
)
)
response = self._get("/api/files/download?path=storage1/a.txt&path=storage1/b.txt")
self.assertEqual(response.status_code, 409)
self.assertEqual(response.json()["error"]["code"], "download_preflight_failed")
self.assertEqual(response.json()["error"]["details"]["reason"], "max_total_input_bytes_exceeded")
def test_download_zip_rejected_when_individual_file_too_large(self) -> None:
(self.root / "docs").mkdir()
(self.root / "docs" / "large.bin").write_bytes(b"123456")
self._override_service(
limits=ZipDownloadPreflightLimits(
max_items=10,
max_total_input_bytes=1024,
max_individual_file_bytes=5,
scan_timeout_seconds=10.0,
)
)
response = self._get("/api/files/download?path=storage1/docs")
self.assertEqual(response.status_code, 409)
self.assertEqual(response.json()["error"]["code"], "download_preflight_failed")
self.assertEqual(response.json()["error"]["details"]["reason"], "max_individual_file_size_exceeded")
self.assertEqual(response.json()["error"]["details"]["path"], "storage1/docs/large.bin")
def test_download_directory_with_symlink_rejected(self) -> None:
target = self.root / "real.txt"
target.write_text("x", encoding="utf-8")
@@ -130,7 +202,29 @@ class DownloadApiGoldenTest(unittest.TestCase):
response = self._get("/api/files/download?path=storage1/docs")
self.assertEqual(response.status_code, 409)
self.assertEqual(response.json()["error"]["code"], "type_conflict")
self.assertEqual(response.json()["error"]["code"], "download_preflight_failed")
self.assertEqual(response.json()["error"]["details"]["reason"], "symlink_detected")
self.assertEqual(response.json()["error"]["details"]["path"], "storage1/docs/link.txt")
def test_download_zip_preflight_timeout_rejected_cleanly(self) -> None:
(self.root / "a.txt").write_text("A", encoding="utf-8")
(self.root / "b.txt").write_text("B", encoding="utf-8")
ticks = iter([0.0, 11.0, 11.0, 11.0])
self._override_service(
limits=ZipDownloadPreflightLimits(
max_items=10,
max_total_input_bytes=1024,
max_individual_file_bytes=1024,
scan_timeout_seconds=10.0,
),
monotonic=lambda: next(ticks),
)
response = self._get("/api/files/download?path=storage1/a.txt&path=storage1/b.txt")
self.assertEqual(response.status_code, 409)
self.assertEqual(response.json()["error"]["code"], "download_preflight_failed")
self.assertEqual(response.json()["error"]["details"]["reason"], "preflight_timeout")
def test_download_path_not_found(self) -> None:
response = self._get("/api/files/download?path=storage1/missing.txt")