From 8e8952ce2055c21ad11680226c9d4c20380063a6 Mon Sep 17 00:00:00 2001 From: Chris Christiansen Date: Sat, 19 Sep 2026 23:28:01 +0000 Subject: [PATCH] feat(opax-mcp): add safe read-only Gitea repository listing --- opax-mcp/capability_registry.py | 7 + opax-mcp/gitea_handler.py | 78 +++++++++-- opax-mcp/policy/tool_policy.py | 1 + opax-mcp/server.py | 8 ++ opax-mcp/test_capability_bridge.py | 6 +- opax-mcp/test_capability_registry.py | 17 ++- opax-mcp/test_gitea_handler.py | 190 +++++++++++++++++++++++++++ 7 files changed, 295 insertions(+), 12 deletions(-) create mode 100644 opax-mcp/test_gitea_handler.py diff --git a/opax-mcp/capability_registry.py b/opax-mcp/capability_registry.py index a3d4599..d304daa 100644 --- a/opax-mcp/capability_registry.py +++ b/opax-mcp/capability_registry.py @@ -133,6 +133,13 @@ _CAPABILITIES_TUPLE: Tuple[Capability, ...] = ( ("authenticated_user", "gitea_read"), False, None, True, False, False, ("git.vauco.no",), "internal_authoritative_only", ), + _capability( + "gitea.list_repo_files", "List Gitea repo files", + "List files in a directory from the authoritative Gitea service.", "gitea", RiskLevel.READ, + Availability.PLANNED, "opax-mcp", "list_repo_files", + ("authenticated_user", "gitea_read"), False, None, True, False, False, + ("git.vauco.no",), "internal_authoritative_only", + ), _capability( "cloudbuild.read_status", "Read Cloud Build status", "Read Cloud Build status in the approved VAUCO project.", "cloudbuild", RiskLevel.READ, diff --git a/opax-mcp/gitea_handler.py b/opax-mcp/gitea_handler.py index 7ce6453..db7060f 100644 --- a/opax-mcp/gitea_handler.py +++ b/opax-mcp/gitea_handler.py @@ -1,6 +1,7 @@ import os import httpx import logging +from urllib.parse import quote logger = logging.getLogger(__name__) @@ -34,17 +35,78 @@ async def handle_list_repo_files(p: dict, default_repo: str) -> dict: if not gitea_url: raise ValueError("GITEA_URL environment variable is not set.") - path = p.get("path", "") - repo_id = p.get("repo", default_repo) - ref = p.get("ref", "main") + if not isinstance(p, dict): + raise ValueError("Invalid list_repo_files input.") - url = f"{gitea_url}/api/v1/repos/{repo_id}/contents/{path}?ref={ref}" + if set(p.keys()) - {"path"}: + raise ValueError("Unsupported list_repo_files input field.") + + path = p.get("path") + + if path is None or path == "" or path == ".": + 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.") + + repo_id = default_repo + ref = "main" + + safe_path = quote(normalized_path, safe='/') + + url = f"{gitea_url.rstrip('/')}/api/v1/repos/{repo_id}/contents/{safe_path}" async with httpx.AsyncClient(timeout=15) as c: - r = await c.get(url, headers=_gitea_headers()) + r = await c.get(url, headers=_gitea_headers(), params={"ref": ref}) r.raise_for_status() - files = r.json() + + response_data = r.json() + + if isinstance(response_data, dict): + raise ValueError("Path is a file, not a directory. Use get_file_content instead.") + + if not isinstance(response_data, list): + raise ValueError("Invalid upstream response.") + + result_files = [] + for entry in response_data: + if not isinstance(entry, dict): + raise ValueError("Invalid upstream response.") + + name = entry.get("name") + entry_path = entry.get("path") + entry_type = entry.get("type") + + if not all([name, entry_path, entry_type]): + raise ValueError("Invalid upstream response.") + + result_entry = { + "name": name, + "path": entry_path, + "type": entry_type + } + + if "sha" in entry and entry["sha"] is not None: + result_entry["sha"] = entry["sha"] + + if "size" in entry and entry["size"] is not None: + result_entry["size"] = entry["size"] + + result_files.append(result_entry) + return { - "path": path, - "files": [{"name": f.get("name"), "type": f.get("type"), "path": f.get("path")} for f in files] + "path": normalized_path, + "files": result_files } diff --git a/opax-mcp/policy/tool_policy.py b/opax-mcp/policy/tool_policy.py index 47f17b1..36cba3c 100644 --- a/opax-mcp/policy/tool_policy.py +++ b/opax-mcp/policy/tool_policy.py @@ -14,6 +14,7 @@ _TOOL_POLICY_MATRIX: Dict[str, ToolRiskLevel] = { "get_telemetry": "read_only", "list_commits": "read_only", "get_file": "read_only", + "list_repo_files": "read_only", "list_open_issues": "read_only", "list_emma_models": "read_only", "run_emma": "read_only", diff --git a/opax-mcp/server.py b/opax-mcp/server.py index 0c2c5a3..0a46536 100644 --- a/opax-mcp/server.py +++ b/opax-mcp/server.py @@ -58,6 +58,7 @@ except Exception as e: print(f"Failed to load provision_new_mcp_module: {e}") provision_new_mcp_module = None from emma_adapter import CanonicalEmma +from gitea_handler import handle_list_repo_files from capability_bridge import build_capability_system_context from email.mime.text import MIMEText from datetime import datetime, timezone, timedelta @@ -919,6 +920,12 @@ async def list_emma_models(p): return await _ollama_models() async def list_commits(p): return await _gitea_get(f"/repos/{p.get('repo', GITEA_REPO)}/commits?limit={p.get('limit', 10)}") async def get_file(p): return await _gitea_get(f"/repos/{p.get('repo', GITEA_REPO)}/contents/{p.get('path', '')}?ref={p.get('ref', 'main')}") async def list_open_issues(p): return await _gitea_get(f"/repos/{p.get('repo', GITEA_REPO)}/issues?state=open&limit=20") + +async def list_repo_files(p: dict) -> dict: + """Wrapper for handle_list_repo_files that enforces server-side constraints.""" + if set(p.keys()) - {"path"}: + raise ValueError("Unsupported list_repo_files input field.") + return await handle_list_repo_files({"path": p.get("path")}, GITEA_REPO) async def create_issue(p): return await _gitea_post(f"/repos/{p.get('repo', GITEA_REPO)}/issues", {"title": p.get("title"), "body": p.get("body", "")}) async def push_file(p): # ... (beholdt uendret) @@ -1215,6 +1222,7 @@ TOOLS = { # Gitea / VCS "list_commits": (list_commits, "List siste commits i Gitea-repo", {}), "get_file": (get_file, "Hent fil fra Gitea-repo", {"type":"object","properties":{"path":{"type":"string"}},"required":["path"]}), + "list_repo_files": (list_repo_files, "Lists files and directories in a Gitea repository path.", {"type": "object", "properties": {"path": {"type": "string"}}}), # Google Workspace "create_email_alias": (create_email_alias, "Opprett et nytt e-postalias", {"type":"object","properties":{"user_key":{"type":"string"},"alias":{"type":"string"}},"required":["user_key","alias"]}), "list_user_aliases": (list_user_aliases, "List en brukers e-postaliaser", {"type":"object","properties":{"user_key":{"type":"string"}},"required":["user_key"]}), diff --git a/opax-mcp/test_capability_bridge.py b/opax-mcp/test_capability_bridge.py index 071e674..3605eb2 100644 --- a/opax-mcp/test_capability_bridge.py +++ b/opax-mcp/test_capability_bridge.py @@ -46,7 +46,11 @@ class TestCapabilityBridge(unittest.TestCase): planned_capabilities = [ c for c in list_capabilities() if c.availability == Availability.PLANNED ] - self.assertEqual(len(planned_capabilities), 16) + self.assertEqual(len(planned_capabilities), 17) + + planned_capability_ids = {c.id for c in planned_capabilities} + self.assertIn("gitea.list_repo_files", planned_capability_ids) + for capability in planned_capabilities: self.assertIn(f"- {capability.display_name}: {capability.description}", self.context_output) diff --git a/opax-mcp/test_capability_registry.py b/opax-mcp/test_capability_registry.py index 7a28841..5a8c0df 100644 --- a/opax-mcp/test_capability_registry.py +++ b/opax-mcp/test_capability_registry.py @@ -20,9 +20,9 @@ class TestCapabilityRegistry(unittest.TestCase): def setUp(self): self.capabilities = list_capabilities() - def test_registry_contains_exactly_nineteen_capabilities(self): - self.assertEqual(len(self.capabilities), 19) - self.assertEqual(len(CAPABILITIES), 19) + def test_registry_contains_exactly_twenty_capabilities(self): + self.assertEqual(len(self.capabilities), 20) + self.assertEqual(len(CAPABILITIES), 20) def test_identifiers_are_unique(self): identifiers = [capability.id for capability in self.capabilities] @@ -135,6 +135,17 @@ class TestCapabilityRegistry(unittest.TestCase): capability.id in rollback_required_ids, ) + def test_gitea_list_repo_files_capability(self): + capability = get_capability("gitea.list_repo_files") + self.assertIsNotNone(capability) + self.assertEqual(capability.risk_level, RiskLevel.READ) + self.assertEqual(capability.availability, Availability.PLANNED) + self.assertEqual(capability.backend_tool, "list_repo_files") + self.assertIn("authenticated_user", capability.required_actor_scope) + self.assertIn("gitea_read", capability.required_actor_scope) + self.assertIn("git.vauco.no", capability.allowed_targets) + self.assertEqual(capability.external_source_policy, "internal_authoritative_only") + if __name__ == "__main__": unittest.main() diff --git a/opax-mcp/test_gitea_handler.py b/opax-mcp/test_gitea_handler.py new file mode 100644 index 0000000..e4d39c5 --- /dev/null +++ b/opax-mcp/test_gitea_handler.py @@ -0,0 +1,190 @@ +import unittest +from pathlib import Path +import sys +from unittest.mock import AsyncMock, MagicMock, patch +import os +import httpx + +REPO_ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(REPO_ROOT / "opax-mcp")) + +from gitea_handler import handle_list_repo_files + +class TestGiteaHandler(unittest.TestCase): + + def setUp(self): + self.original_gitea_url = os.environ.get("GITEA_URL") + self.original_gitea_token = os.environ.get("GITEA_TOKEN") + os.environ["GITEA_URL"] = "https://gitea.example.com" + os.environ["GITEA_TOKEN"] = "test-token" + + def tearDown(self): + if self.original_gitea_url is not None: + os.environ["GITEA_URL"] = self.original_gitea_url + else: + del os.environ["GITEA_URL"] + if self.original_gitea_token is not None: + os.environ["GITEA_TOKEN"] = self.original_gitea_token + else: + del os.environ["GITEA_TOKEN"] + + def _prepare_mock_client(self, payload, is_file=False): + response = MagicMock() + response.raise_for_status = MagicMock() + if is_file: + response.json.return_value = {"name": "file.txt", "path": "file.txt", "type": "file"} + else: + response.json.return_value = payload + + client = MagicMock() + client.get = AsyncMock(return_value=response) + + async_client = MagicMock() + async_client.__aenter__ = AsyncMock(return_value=client) + async_client.__aexit__ = AsyncMock(return_value=None) + + return async_client, client + + @patch("httpx.AsyncClient") + def test_list_repo_files_root(self, mock_async_client_constructor): + payload = [ + {"name": "README.md", "path": "README.md", "type": "file", "sha": "sha123", "size": 1024}, + {"name": "docs", "path": "docs", "type": "dir"} + ] + 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": None}, "test/repo") + self.assertEqual(result["path"], "") + self.assertEqual(len(result["files"]), 2) + self.assertEqual(result["files"][0]["name"], "README.md") + self.assertIn("sha", result["files"][0]) + self.assertIn("size", result["files"][0]) + self.assertEqual(result["files"][1]["name"], "docs") + self.assertNotIn("sha", result["files"][1]) + self.assertNotIn("size", result["files"][1]) + + import asyncio + asyncio.run(run_test()) + + @patch("httpx.AsyncClient") + def test_list_repo_files_empty_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_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 = [] + mock_async_client, mock_client = self._prepare_mock_client(payload) + mock_async_client_constructor.return_value = mock_async_client + async def run_test(): + await handle_list_repo_files({"path": "docs/security"}, "test/repo") + mock_client.get.assert_called_once() + called_url = mock_client.get.call_args[0][0] + self.assertIn("docs/security", called_url) + + import asyncio + asyncio.run(run_test()) + + @patch("httpx.AsyncClient") + def test_path_is_file(self, mock_async_client_constructor): + mock_async_client, _ = self._prepare_mock_client(None, is_file=True) + mock_async_client_constructor.return_value = mock_async_client + async def run_test(): + with self.assertRaisesRegex(ValueError, "Path is a file, not a directory. Use get_file_content instead."): + await handle_list_repo_files({"path": "file.txt"}, "test/repo") + import asyncio + asyncio.run(run_test()) + + def test_invalid_paths(self): + invalid_paths = [ + "../secrets", + "docs/../secrets", + "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") + import asyncio + asyncio.run(run_test()) + + def test_invalid_input_types(self): + async def run_test(): + with self.assertRaisesRegex(ValueError, "Invalid repository path."): + await handle_list_repo_files({"path": 123}, "test/repo") + import asyncio + asyncio.run(run_test()) + + def test_unsupported_fields(self): + unsupported = [ + {"repo": "x"}, + {"ref": "y"}, + {"branch": "z"}, + {"extra": "key"}, + ] + async def run_test(): + for p in unsupported: + with self.subTest(p=p): + with self.assertRaisesRegex(ValueError, "Unsupported list_repo_files input field."): + await handle_list_repo_files(p, "test/repo") + import asyncio + asyncio.run(run_test()) + + @patch("httpx.AsyncClient") + def test_uses_trusted_repo(self, mock_async_client_constructor): + payload = [] + mock_async_client, mock_client = self._prepare_mock_client(payload) + mock_async_client_constructor.return_value = mock_async_client + async def run_test(): + await handle_list_repo_files({"path": "docs", "repo": "user/bad-repo"}, "test/repo") + # The handler should raise an error due to the unsupported 'repo' field. + import asyncio + with self.assertRaisesRegex(ValueError, "Unsupported list_repo_files input field."): + asyncio.run(run_test()) + + @patch("httpx.AsyncClient") + def test_http_error_propagation(self, mock_async_client_constructor): + response = MagicMock() + response.raise_for_status.side_effect = httpx.HTTPStatusError("Error", request=MagicMock(), response=MagicMock()) + client = MagicMock() + client.get = AsyncMock(return_value=response) + async_client = MagicMock() + async_client.__aenter__ = AsyncMock(return_value=client) + async_client.__aexit__ = AsyncMock(return_value=None) + mock_async_client_constructor.return_value = async_client + + async def run_test(): + with self.assertRaises(httpx.HTTPStatusError): + await handle_list_repo_files({"path": ""}, "test/repo") + + import asyncio + asyncio.run(run_test()) + +if __name__ == '__main__': + unittest.main()