diff --git a/opax-mcp/gitea_handler.py b/opax-mcp/gitea_handler.py index f16b4a9..d83c192 100644 --- a/opax-mcp/gitea_handler.py +++ b/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.""" - if not path or not isinstance(path, str): - raise ValueError("Repository file request is not allowed.") + """Validates a repository-relative path using strict POSIX semantics.""" + if not isinstance(path, str): + raise ValueError("PATH_NOT_ALLOWED") - # 1. Reject malformed path - if '\\' in path or '\0' in path or '//' in path: - raise ValueError("Repository file request is not allowed.") + decoded_path = unquote(path) - # 2. Reject traversal/absolute - if path.startswith('/') or '..' in path.split('/'): - raise ValueError("Repository file request is not allowed.") + for candidate in (path, decoded_path): + if not candidate.strip() or "\0" in candidate or "\\" in candidate: + raise ValueError("PATH_NOT_ALLOWED") - # 3. Apply exact root/prefix allowlist - is_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.") + if candidate.startswith("/"): + raise ValueError("PATH_NOT_ALLOWED") - # 4. Apply deny rules - denied_patterns = [ - ".env", ".pem", ".key", ".p12", ".pfx", "id_rsa", "id_ed25519", - "credentials", "secret" - ] - path_segments = path.lower().split('/') - for segment in path_segments: - if any(pattern in segment for pattern in denied_patterns): - raise ValueError("Repository file request is not allowed.") - if not path.startswith("docs/") and path.endswith(".json"): - raise ValueError("Repository file request is not allowed.") + components = candidate.split("/") + if any(component in ("", ".", "..") for component in components): + raise ValueError("PATH_NOT_ALLOWED") + + decoded_components = decoded_path.split("/") + filename = decoded_components[-1] + + if ".git" in decoded_components: + raise ValueError("SECRET_PATH_DENIED") + + 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): - 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.") + normalized_path = _validate_safe_path(requested_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]): diff --git a/opax-mcp/test_gitea_handler.py b/opax-mcp/test_gitea_handler.py index e4d39c5..bb4d496 100644 --- a/opax-mcp/test_gitea_handler.py +++ b/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."): - await handle_list_repo_files({"path": path}, "test/repo") + # 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())