diff --git a/api/packs.py b/api/packs.py index 578a16a..5870faa 100644 --- a/api/packs.py +++ b/api/packs.py @@ -32,6 +32,7 @@ except ImportError: web = MockWeb() import os +import re import shutil import tempfile @@ -45,6 +46,16 @@ else: from services.packs.pack_archive import PackArchive, PackError from services.packs.pack_registry import PackRegistry +# Strict pattern for pack name/version URL route parameters. +_SAFE_SEGMENT_RE = re.compile(r"^[a-zA-Z0-9._-]+$") + + +def _is_safe_pack_segment(value: str) -> bool: + """Check if a pack name or version segment is safe for filesystem use.""" + if not value or value in (".", ".."): + return False + return bool(_SAFE_SEGMENT_RE.match(value)) + if web: @@ -155,6 +166,11 @@ class PacksHandlers: {"ok": False, "error": "Missing name/version"}, status=400 ) + if not _is_safe_pack_segment(name) or not _is_safe_pack_segment(version): + return web.json_response( + {"ok": False, "error": "Invalid name or version format"}, status=400 + ) + try: success = self.registry.uninstall_pack(name, version) if success: @@ -182,6 +198,11 @@ class PacksHandlers: {"ok": False, "error": "Missing name/version"}, status=400 ) + if not _is_safe_pack_segment(name) or not _is_safe_pack_segment(version): + return web.json_response( + {"ok": False, "error": "Invalid name or version format"}, status=400 + ) + pack_path = self.registry.get_pack_path(name, version) if not pack_path: return web.json_response( @@ -206,7 +227,7 @@ class PacksHandlers: return CleanupFileResponse( temp_zip, headers={ - "Content-Disposition": f'attachment; filename="{name}-{version}.zip"', + "Content-Disposition": f'attachment; filename="{name.replace(chr(34), "")}-{version.replace(chr(34), "")}.zip"', "Content-Type": "application/zip", }, ) diff --git a/tests/test_packs_integrity.py b/tests/test_packs_integrity.py index 1adae20..90a22ad 100644 --- a/tests/test_packs_integrity.py +++ b/tests/test_packs_integrity.py @@ -7,7 +7,7 @@ import unittest import zipfile from unittest.mock import ANY, AsyncMock, MagicMock, patch -from api.packs import CleanupFileResponse, PacksHandlers +from api.packs import CleanupFileResponse, PacksHandlers, _is_safe_pack_segment from services.packs.pack_archive import PackArchive, PackError from services.packs.pack_manifest import create_manifest @@ -193,5 +193,60 @@ class TestPacksApiAsync(unittest.IsolatedAsyncioTestCase): self.assertFalse(os.path.exists(path), "File should be deleted") +class TestPacksApiInputValidation(unittest.IsolatedAsyncioTestCase): + """Test that API handlers reject path traversal in URL route parameters.""" + + def setUp(self): + self.test_dir = tempfile.mkdtemp() + self.handlers = PacksHandlers(self.test_dir) + self.handlers._check_auth = AsyncMock(return_value=True) + + def tearDown(self): + shutil.rmtree(self.test_dir) + + def test_safe_segment_rejects_traversal(self): + self.assertFalse(_is_safe_pack_segment("../../etc")) + self.assertFalse(_is_safe_pack_segment("..")) + self.assertFalse(_is_safe_pack_segment(".")) + self.assertFalse(_is_safe_pack_segment("")) + self.assertFalse(_is_safe_pack_segment("foo/bar")) + self.assertFalse(_is_safe_pack_segment("foo\\bar")) + + def test_safe_segment_accepts_valid(self): + self.assertTrue(_is_safe_pack_segment("my-pack")) + self.assertTrue(_is_safe_pack_segment("1.0.0")) + self.assertTrue(_is_safe_pack_segment("pack_v2.1")) + + @patch("api.packs.web") + async def test_delete_rejects_traversal_name(self, mock_web): + mock_web.json_response = MagicMock() + request = MockRequest() + request.match_info = {"name": "../../etc", "version": "1.0.0"} + await self.handlers.delete_pack_handler(request) + mock_web.json_response.assert_called_with( + {"ok": False, "error": "Invalid name or version format"}, status=400 + ) + + @patch("api.packs.web") + async def test_delete_rejects_traversal_version(self, mock_web): + mock_web.json_response = MagicMock() + request = MockRequest() + request.match_info = {"name": "my-pack", "version": "../.."} + await self.handlers.delete_pack_handler(request) + mock_web.json_response.assert_called_with( + {"ok": False, "error": "Invalid name or version format"}, status=400 + ) + + @patch("api.packs.web") + async def test_export_rejects_traversal(self, mock_web): + mock_web.json_response = MagicMock() + request = MockRequest() + request.match_info = {"name": "../../etc", "version": "passwd"} + await self.handlers.export_pack_handler(request) + mock_web.json_response.assert_called_with( + {"ok": False, "error": "Invalid name or version format"}, status=400 + ) + + if __name__ == "__main__": unittest.main()