chris/OSVauco
1
0
Fork 0
Code Issues 1 Pull Requests Actions 2 Packages Projects Releases Wiki Activity

feat(opax-mcp): add safe read-only Gitea repository listing

Browse Source
This commit is contained in:
Chris Christiansen 2026-09-19 23:28:01 +00:00
parent 25b3deb19b
commit 8e8952ce20
7 changed files with 295 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
opax-mcp/capability_registry.py
View File

@ -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,

80
opax-mcp/gitea_handler.py
View File

@ -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()
return {
"path": path,
"files": [{"name": f.get("name"), "type": f.get("type"), "path": f.get("path")} for f in files]
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": normalized_path,
"files": result_files
}

1
opax-mcp/policy/tool_policy.py
View File

@ -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",

8
opax-mcp/server.py
View File

@ -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"]}),

6
opax-mcp/test_capability_bridge.py
View File

@ -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)

17
opax-mcp/test_capability_registry.py
View File

@ -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()

190
opax-mcp/test_gitea_handler.py Normal file
View File

@ -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()