Harden Gitea repository file listing

opax-mcp/gitea_handler.py

@@ -1,7 +1,7 @@
 import os
 import httpx
 import logging
-from urllib.parse import quote
+from urllib.parse import quote, unquote
 import re
 import base64
 import json
@@ -75,40 +75,75 @@ def _validate_commit_sha(commit_sha: str) -> str:
     return commit_sha
 
 def _validate_safe_path(path: str) -> str:
-    """Validates a file path against a strict allowlist and safety rules."""
+    """Validates a repository-relative path using strict POSIX semantics."""
-    if not path or not isinstance(path, str):
+    if not isinstance(path, str):
-        raise ValueError("Repository file request is not allowed.")
+        raise ValueError("PATH_NOT_ALLOWED")
 
-    # 1. Reject malformed path
+    decoded_path = unquote(path)
-    if '\\' in path or '\0' in path or '//' in path:
-        raise ValueError("Repository file request is not allowed.")
 
-    # 2. Reject traversal/absolute
+    for candidate in (path, decoded_path):
-    if path.startswith('/') or '..' in path.split('/'):
+        if not candidate.strip() or "\0" in candidate or "\\" in candidate:
-        raise ValueError("Repository file request is not allowed.")
+            raise ValueError("PATH_NOT_ALLOWED")
 
-    # 3. Apply exact root/prefix allowlist
+        if candidate.startswith("/"):
-    is_allowed = (
+            raise ValueError("PATH_NOT_ALLOWED")
-        path == "README.md" or
-        path == "GEMINI.md" or
-        path.startswith("docs/") or
-        path.startswith(".gemini/") or
-        path.startswith("README/")
-    )
-    if not is_allowed:
-        raise ValueError("Repository file request is not allowed.")
 
-    # 4. Apply deny rules
+        components = candidate.split("/")
-    denied_patterns = [
+        if any(component in ("", ".", "..") for component in components):
-        ".env", ".pem", ".key", ".p12", ".pfx", "id_rsa", "id_ed25519",
+            raise ValueError("PATH_NOT_ALLOWED")
-        "credentials", "secret"
+
-    ]
+    decoded_components = decoded_path.split("/")
-    path_segments = path.lower().split('/')
+    filename = decoded_components[-1]
-    for segment in path_segments:
+
-        if any(pattern in segment for pattern in denied_patterns):
+    if ".git" in decoded_components:
-            raise ValueError("Repository file request is not allowed.")
+        raise ValueError("SECRET_PATH_DENIED")
-    if not path.startswith("docs/") and path.endswith(".json"):
+
-        raise ValueError("Repository file request is not allowed.")
+    if filename == ".env" or (
+        filename.startswith(".env.") and filename != ".env.example"
+    ):
+        raise ValueError("SECRET_PATH_DENIED")
+
+    denied_basenames = {
+        ".netrc",
+        ".npmrc",
+        ".pypirc",
+        "id_rsa",
+        "id_ed25519",
+        "credentials",
+        "credentials.json",
+        "service_account.json",
+        "service-account.json",
+        "private_key",
+        "private_key.json",
+        "token",
+        "token.json",
+    }
+    if filename in denied_basenames:
+        raise ValueError("SECRET_PATH_DENIED")
+
+    denied_components = {
+        "secrets",
+        "credentials",
+        "service_accounts",
+        "service-accounts",
+    }
+    if any(component in denied_components for component in decoded_components):
+        raise ValueError("SECRET_PATH_DENIED")
+
+    denied_suffixes = {
+        ".pem",
+        ".key",
+        ".p12",
+        ".pfx",
+        ".jks",
+        ".keystore",
+        ".kubeconfig",
+        ".crt",
+        ".cer",
+        ".der",
+    }
+    if any(filename.endswith(suffix) for suffix in denied_suffixes):
+        raise ValueError("SECRET_PATH_DENIED")
 
     return path
 
@@ -378,25 +413,12 @@ async def handle_list_repo_files(p: dict, default_repo: str) -> dict:
     if set(p.keys()) - {"path"}:
         raise ValueError("Unsupported list_repo_files input field.")
 
-    path = p.get("path")
+    requested_path = p.get("path")
 
-    if path is None or path == "" or path == ".":
+    if requested_path in (None, ""):
         normalized_path = ""
     else:
-        if not isinstance(path, str):
+        normalized_path = _validate_safe_path(requested_path)
-            raise ValueError("Invalid repository path.")
-        normalized_path = path
-
-    if normalized_path.startswith('/') or \
-       '\\' in normalized_path or \
-       any(c in normalized_path for c in (':', '?', '#')):
-        raise ValueError("Invalid repository path.")
-
-    parts = normalized_path.split('/')
-    if any(part in ('.', '..') for part in parts):
-        raise ValueError("Invalid repository path.")
-    if '' in parts and normalized_path != '': # check for empty segments
-        raise ValueError("Invalid repository path.")
 
     repo_id = default_repo
     ref = "main"
@@ -422,8 +444,15 @@ async def handle_list_repo_files(p: dict, default_repo: str) -> dict:
             if not isinstance(entry, dict):
                 raise ValueError("Invalid upstream response.")
 
-            name = entry.get("name")
             entry_path = entry.get("path")
+            try:
+                _validate_safe_path(entry_path)
+            except ValueError as e:
+                if str(e) in ("PATH_NOT_ALLOWED", "SECRET_PATH_DENIED"):
+                    continue
+                raise
+
+            name = entry.get("name")
             entry_type = entry.get("type")
 
             if not all([name, entry_path, entry_type]):

opax-mcp/test_gitea_handler.py

@@ -79,17 +79,6 @@ class TestGiteaHandler(unittest.TestCase):
         import asyncio
         asyncio.run(run_test())
 
-    @patch("httpx.AsyncClient")
-    def test_list_repo_files_dot_path(self, mock_async_client_constructor):
-        payload = []
-        mock_async_client, _ = self._prepare_mock_client(payload)
-        mock_async_client_constructor.return_value = mock_async_client
-        async def run_test():
-            result = await handle_list_repo_files({"path": "."}, "test/repo")
-            self.assertEqual(result["path"], "")
-        import asyncio
-        asyncio.run(run_test())
-
     @patch("httpx.AsyncClient")
     def test_nested_path(self, mock_async_client_constructor):
         payload = []
@@ -114,6 +103,164 @@ class TestGiteaHandler(unittest.TestCase):
         import asyncio
         asyncio.run(run_test())
 
+    @patch("httpx.AsyncClient")
+    def test_hardened_root_none(self, mock_async_client_constructor):
+        """Verify path=None retains root-listing behavior and params."""
+        mock_async_client, mock_client = self._prepare_mock_client([])
+        mock_async_client_constructor.return_value = mock_async_client
+
+        async def run_test():
+            await handle_list_repo_files({"path": None}, "test/repo")
+            mock_client.get.assert_called_once()
+            called_url = mock_client.get.call_args[0][0]
+            self.assertTrue(called_url.endswith("/contents/"))
+            self.assertEqual(mock_client.get.call_args[1].get("params"), {"ref": "main"})
+
+        import asyncio
+        asyncio.run(run_test())
+
+    @patch("httpx.AsyncClient")
+    def test_hardened_root_empty(self, mock_async_client_constructor):
+        """Verify path="" retains root-listing behavior and params."""
+        mock_async_client, mock_client = self._prepare_mock_client([])
+        mock_async_client_constructor.return_value = mock_async_client
+
+        async def run_test():
+            await handle_list_repo_files({"path": ""}, "test/repo")
+            mock_client.get.assert_called_once()
+            called_url = mock_client.get.call_args[0][0]
+            self.assertTrue(called_url.endswith("/contents/"))
+            self.assertEqual(mock_client.get.call_args[1].get("params"), {"ref": "main"})
+
+        import asyncio
+        asyncio.run(run_test())
+
+    @patch("httpx.AsyncClient")
+    def test_hardened_dot_path_denied(self, mock_async_client_constructor):
+        """Verify path='.' is denied before making an HTTP request."""
+        mock_async_client, mock_client = self._prepare_mock_client([])
+        mock_async_client_constructor.return_value = mock_async_client
+
+        async def run_test():
+            with self.assertRaises(ValueError) as cm:
+                await handle_list_repo_files({"path": "."}, "test/repo")
+            self.assertEqual(str(cm.exception), "PATH_NOT_ALLOWED")
+            mock_client.get.assert_not_called()
+
+        import asyncio
+        asyncio.run(run_test())
+
+    @patch("httpx.AsyncClient")
+    def test_hardened_safe_nested_path(self, mock_async_client_constructor):
+        """Verify a safe nested path reaches the client and preserves params."""
+        mock_async_client, mock_client = self._prepare_mock_client([])
+        mock_async_client_constructor.return_value = mock_async_client
+
+        async def run_test():
+            await handle_list_repo_files({"path": "docs/sub"}, "test/repo")
+            mock_client.get.assert_called_once()
+            called_url = mock_client.get.call_args[0][0]
+            self.assertTrue(called_url.endswith("/contents/docs/sub"))
+            self.assertEqual(mock_client.get.call_args[1].get("params"), {"ref": "main"})
+
+        import asyncio
+        asyncio.run(run_test())
+
+    @patch("httpx.AsyncClient")
+    def test_hardened_traversal_path_denied(self, mock_async_client_constructor):
+        """Verify a traversal path is denied before any HTTP request."""
+        mock_async_client, mock_client = self._prepare_mock_client([])
+        mock_async_client_constructor.return_value = mock_async_client
+        async def run_test():
+            with self.assertRaises(ValueError) as cm:
+                await handle_list_repo_files({"path": "a/../b"}, "test/repo")
+            self.assertEqual(str(cm.exception), "PATH_NOT_ALLOWED")
+            mock_client.get.assert_not_called()
+        import asyncio
+        asyncio.run(run_test())
+
+    @patch("httpx.AsyncClient")
+    def test_hardened_secret_request_path_denied(self, mock_async_client_constructor):
+        """Verify a secret requested path is denied before any HTTP request."""
+        mock_async_client, mock_client = self._prepare_mock_client([])
+        mock_async_client_constructor.return_value = mock_async_client
+        async def run_test():
+            with self.assertRaises(ValueError) as cm:
+                await handle_list_repo_files({"path": ".env"}, "test/repo")
+            self.assertEqual(str(cm.exception), "SECRET_PATH_DENIED")
+            mock_client.get.assert_not_called()
+        import asyncio
+        asyncio.run(run_test())
+
+    @patch("httpx.AsyncClient")
+    def test_hardened_output_filtering(self, mock_async_client_constructor):
+        """Verify secret paths are omitted from the returned list."""
+        gitea_response_payload = [
+            {"name": "README.md", "path": "README.md", "type": "file", "sha": "safe_sha", "size": 100},
+            {"name": ".env", "path": ".env", "type": "file"},
+            {"name": "config", "path": ".git/config", "type": "file"},
+            {"name": "app.txt", "path": "secrets/app.txt", "type": "file"},
+            {"name": "secret.pem", "path": "keys/secret.pem", "type": "file"},
+        ]
+        mock_async_client, _ = self._prepare_mock_client(gitea_response_payload)
+        mock_async_client_constructor.return_value = mock_async_client
+
+        async def run_test():
+            result = await handle_list_repo_files({"path": ""}, "test/repo")
+            self.assertEqual(result["files"], [{"name": "README.md", "path": "README.md", "type": "file", "sha": "safe_sha", "size": 100}])
+            returned_paths = {f["path"] for f in result["files"]}
+            self.assertNotIn(".env", returned_paths)
+            self.assertNotIn(".git/config", returned_paths)
+            self.assertNotIn("secrets/app.txt", returned_paths)
+            self.assertNotIn("keys/secret.pem", returned_paths)
+
+        import asyncio
+        asyncio.run(run_test())
+
+    def test_hardened_unexpected_error_reraised(self):
+        """Verify other ValueErrors are not suppressed during filtering."""
+        import gitea_handler
+        real_validator = gitea_handler._validate_safe_path
+
+        def validator_side_effect(path, *args, **kwargs):
+            if path == "trigger.txt":
+                raise ValueError("UNEXPECTED_ERROR")
+            return real_validator(path, *args, **kwargs)
+
+        async def run_test():
+            with patch("gitea_handler._validate_safe_path") as mock_validator:
+                mock_validator.side_effect = validator_side_effect
+
+                with patch("httpx.AsyncClient") as mock_async_client_constructor:
+                    mock_async_client, mock_client = self._prepare_mock_client(
+                        [
+                            {
+                                "name": "safe.txt",
+                                "path": "safe.txt",
+                                "type": "file",
+                                "sha": "safe_sha",
+                                "size": 1,
+                            },
+                            {
+                                "name": "trigger.txt",
+                                "path": "trigger.txt",
+                                "type": "file",
+                                "sha": "trigger_sha",
+                                "size": 1,
+                            },
+                        ]
+                    )
+                    mock_async_client_constructor.return_value = mock_async_client
+
+                    with self.assertRaises(ValueError) as cm:
+                        await handle_list_repo_files({"path": ""}, "test/repo")
+
+                    self.assertEqual(str(cm.exception), "UNEXPECTED_ERROR")
+                    mock_client.get.assert_called_once()
+
+        import asyncio
+        asyncio.run(run_test())
+
     def test_invalid_paths(self):
         invalid_paths = [
             "../secrets",
@@ -121,23 +268,30 @@ class TestGiteaHandler(unittest.TestCase):
             "docs\\secrets",
             "/etc/passwd",
             "https://example.com",
-            "docs?ref=other",
-            "docs#fragment",
             "docs//security",
             "docs/./security",
         ]
         async def run_test():
             for path in invalid_paths:
                 with self.subTest(path=path):
-                    with self.assertRaisesRegex(ValueError, "Invalid repository path."):
+                    # Use patch here to ensure no real network call is made for paths
+                    # that might otherwise be valid URLs.
+                    with patch("httpx.AsyncClient") as mock_client:
+                        mock_async_client, mocked_http_client = self._prepare_mock_client([])
+                        mock_client.return_value = mock_async_client
+                        with self.assertRaises(ValueError) as cm:
                             await handle_list_repo_files({"path": path}, "test/repo")
+                        self.assertEqual(str(cm.exception), "PATH_NOT_ALLOWED")
+                        # Ensure no network call was attempted for invalid paths
+                        mocked_http_client.get.assert_not_called()
         import asyncio
         asyncio.run(run_test())
 
     def test_invalid_input_types(self):
         async def run_test():
-            with self.assertRaisesRegex(ValueError, "Invalid repository path."):
+            with self.assertRaises(ValueError) as cm:
                 await handle_list_repo_files({"path": 123}, "test/repo")
+            self.assertEqual(str(cm.exception), "PATH_NOT_ALLOWED")
         import asyncio
         asyncio.run(run_test())