diff --git a/opax-mcp/gitea_handler.py b/opax-mcp/gitea_handler.py index b13f646..1885c51 100644 --- a/opax-mcp/gitea_handler.py +++ b/opax-mcp/gitea_handler.py @@ -3,6 +3,9 @@ import httpx import logging from urllib.parse import quote import re +import base64 +import json +import binascii logger = logging.getLogger(__name__) @@ -19,6 +22,13 @@ MAX_SOURCE_ARCHIVE_BYTES = int( os.environ.get("OPAX_MAX_SOURCE_ARCHIVE_BYTES", "104857600") ) +ALLOWED_GITEA_NAMESPACE = "chris" +CATALOG_PAGE_SIZE = 50 +CATALOG_MAX_PAGES = 2 +CATALOG_MAX_RESULTS = 100 +MAX_READ_FILE_BYTES = 1_048_576 +MAX_GET_FILE_RESPONSE_BYTES = 1_500_000 + def _gitea_headers() -> dict: """Constructs headers for Gitea API requests.""" return {"Authorization": f"token {os.environ.get('GITEA_TOKEN')}", "Accept": "application/json"} @@ -28,7 +38,6 @@ def _validate_repo_id(repo_id: str) -> str: raise ValueError("Invalid configured Gitea repository ID") return repo_id - def _validate_gitea_url(gitea_url: str) -> str: if not isinstance(gitea_url, str): raise ValueError("GITEA_URL environment variable is not set") @@ -40,7 +49,6 @@ def _validate_gitea_url(gitea_url: str) -> str: return normalized - def _validate_branch_name(branch_name: str) -> str: if not isinstance(branch_name, str): raise ValueError("Branch name must be a string") @@ -60,13 +68,50 @@ def _validate_branch_name(branch_name: str) -> str: return branch_name - def _validate_commit_sha(commit_sha: str) -> str: if not isinstance(commit_sha, str) or not _SHA_RE.fullmatch(commit_sha): raise ValueError("Invalid commit SHA") 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.") + + # 1. Reject malformed path + if '\\' in path or '\0' in path or '//' in path: + raise ValueError("Repository file request is not allowed.") + + # 2. Reject traversal/absolute + if path.startswith('/') or '..' in path.split('/'): + raise ValueError("Repository file request is 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.") + + # 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.") + + return path + async def resolve_branch_to_commit_sha( branch_name: str, @@ -145,25 +190,141 @@ async def download_repo_archive( return b"".join(chunks) -async def handle_get_file_content(p: dict, default_repo: str) -> dict: - """Gets the raw content of a file from the Gitea repository.""" - gitea_url = os.environ.get("GITEA_URL") - if not gitea_url: - raise ValueError("GITEA_URL environment variable is not set.") - - path = p.get("path") - if not path: - raise ValueError("Missing required parameter: 'path' for get_file_content") - - repo_id = p.get("repo", default_repo) - ref = p.get("ref", "main") - - url = f"{gitea_url}/api/v1/repos/{repo_id}/raw/{path}?ref={ref}" - - async with httpx.AsyncClient(timeout=15) as c: - r = await c.get(url, headers=_gitea_headers()) - r.raise_for_status() - return {"path": path, "content": r.text, "encoding": "text"} + +def _normalize_repository_item(item: dict) -> dict | None: + """Safely extracts and transforms a single repository item from the Gitea API response.""" + if not isinstance(item, dict): + return None + + full_name = item.get("full_name") + name = item.get("name") + + if not all(isinstance(val, str) and val for val in [full_name, name]): + return None + + if not full_name.startswith(f"{ALLOWED_GITEA_NAMESPACE}/"): + return None + + return { + "name": name, + "full_name": full_name, + "default_branch": item.get("default_branch"), + "updated_at": item.get("updated_at"), + "archived": item.get("archived", False), + } + + +async def list_allowed_namespace_repositories() -> dict: + """ + Fetches a read-only, metadata-only catalog of Gitea repositories + from a fixed, approved namespace. + """ + gitea_url = _validate_gitea_url(os.environ.get("GITEA_URL")) + + all_repos = [] + seen_repos = set() + + try: + async with httpx.AsyncClient(timeout=15.0, follow_redirects=False) as client: + for page in range(1, CATALOG_MAX_PAGES + 1): + if len(all_repos) >= CATALOG_MAX_RESULTS: + break + + url = ( + f"{gitea_url}/api/v1/users/{ALLOWED_GITEA_NAMESPACE}/repos" + f"?limit={CATALOG_PAGE_SIZE}&page={page}" + ) + + response = await client.get(url, headers=_gitea_headers()) + response.raise_for_status() + + page_items = response.json() + if not isinstance(page_items, list) or not page_items: + break + + for item in page_items: + normalized = _normalize_repository_item(item) + if normalized and normalized["full_name"] not in seen_repos: + seen_repos.add(normalized["full_name"]) + all_repos.append(normalized) + + except (httpx.HTTPError, json.JSONDecodeError) as e: + logger.error(f"Gitea repository catalog failed: {e}") + raise ValueError("Repository catalog unavailable.") + + return {"repositories": all_repos[:CATALOG_MAX_RESULTS]} + + +async def handle_get_file_content(p: dict, server_repo_id: str) -> dict: + """ + Gets the raw content of a file from the Gitea repository after strict validation. + Uses the JSON/base64 Contents API with bounded reads. + """ + gitea_url = _validate_gitea_url(os.environ.get("GITEA_URL")) + validated_server_repo = _validate_repo_id(server_repo_id) + + caller_repo = p.get("repo") + if caller_repo is not None and caller_repo != validated_server_repo: + raise ValueError("Repository file request is not allowed.") + + ref = _validate_commit_sha(p.get("ref")) + path = _validate_safe_path(p.get("path")) + + url = f"{gitea_url}/api/v1/repos/{validated_server_repo}/contents/{quote(path, safe='')}?ref={ref}" + + try: + async with httpx.AsyncClient(timeout=15) as c: + async with c.stream("GET", url, headers=_gitea_headers()) as response: + response.raise_for_status() + + content_length = -1 + content_length_str = response.headers.get("content-length") + if content_length_str: + try: + content_length = int(content_length_str) + except (ValueError, TypeError): + content_length = -1 + + if content_length >= 0 and content_length > MAX_GET_FILE_RESPONSE_BYTES: + raise ValueError("Repository file exceeds the allowed size.") + + body_bytes = bytearray() + async for chunk in response.aiter_bytes(): + if len(body_bytes) + len(chunk) > MAX_GET_FILE_RESPONSE_BYTES: + raise ValueError("Repository file exceeds the allowed size.") + body_bytes.extend(chunk) + + data = json.loads(body_bytes) + + except (httpx.HTTPError, json.JSONDecodeError): + raise ValueError("Repository file is unavailable.") + + if not isinstance(data, dict) or "content" not in data: + raise ValueError("Repository file is unavailable.") + + size = data.get("size") + if isinstance(size, int) and not isinstance(size, bool) and size >= 0: + if size > MAX_READ_FILE_BYTES: + raise ValueError("Repository file exceeds the allowed size.") + + try: + decoded_content = base64.b64decode(data["content"], validate=True) + except (TypeError, ValueError, binascii.Error): + raise ValueError("Repository file content is not readable text.") + + if len(decoded_content) > MAX_READ_FILE_BYTES: + raise ValueError("Repository file exceeds the allowed size.") + + try: + text_content = decoded_content.decode('utf-8') + except UnicodeDecodeError: + raise ValueError("Repository file content is not readable text.") + + if '\0' in text_content: + raise ValueError("Repository file content is not readable text.") + + return {"path": path, "content": text_content, "encoding": "utf-8"} + async def handle_list_repo_files(p: dict, default_repo: str) -> dict: """Lists files and directories in a given path in the Gitea repository.""" diff --git a/opax-mcp/server.py b/opax-mcp/server.py index ae8172f..a90d0df 100644 --- a/opax-mcp/server.py +++ b/opax-mcp/server.py @@ -58,7 +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 gitea_handler import handle_list_repo_files, list_allowed_namespace_repositories, handle_get_file_content from capability_bridge import build_capability_system_context from email.mime.text import MIMEText from datetime import datetime, timezone, timedelta @@ -327,7 +327,7 @@ async def create_a2h2a_ticket(ticket: A2H2ATicket, request: Request, test_mode: response_payload = {"status": "success", "ticket_id": ticket.ticket_id, "parameter_hash": ticket.proposed_action.parameter_hash} if test_mode: response_payload['raw_token'] = raw_token - + logger.info(f"A2H2A ticket created successfully: {ticket.ticket_id}", extra={"ticket_id": ticket.ticket_id, "event_type": "TICKET_CREATED"}) A2H2A_METRICS["tickets_created"] += 1 return response_payload @@ -364,7 +364,7 @@ async def review_a2h2a_ticket(ticket_id: str, token: str, request: Request): logger.warning(f"A2H2A review attempted for expired ticket {ticket_id}.", extra={"ticket_id": ticket_id, "event_type": "REVIEW_TICKET_EXPIRED"}) A2H2A_METRICS["tickets_expired"] += 1 return HTMLResponse(content="

400: Ticket Expired

", status_code=400) - + provided_token_hash = hashlib.sha256(token.encode()).hexdigest() if not secrets.compare_digest(provided_token_hash, ticket.governance.approval_token_hash): logger.error(f"Invalid approval token for ticket {ticket_id}.", extra={"ticket_id": ticket_id, "event_type": "REVIEW_INVALID_TOKEN"}) @@ -436,7 +436,7 @@ async def _process_approval_action(ticket_id: str, token: str, user_email: str, if ticket.is_expired(): A2H2A_METRICS["tickets_expired"] += 1 raise HTTPException(status_code=400, detail="Ticket has expired") - + provided_token_hash = hashlib.sha256(token.encode()).hexdigest() if not secrets.compare_digest(provided_token_hash, ticket.governance.approval_token_hash): raise HTTPException(status_code=403, detail="Invalid token") @@ -446,7 +446,7 @@ async def _process_approval_action(ticket_id: str, token: str, user_email: str, if not stored_csrf_hash: logger.error(f"CSRF token missing or already used for ticket {ticket_id}.", extra={"ticket_id": ticket_id, "user_email": user_email, "event_type": "APPROVAL_CSRF_TOKEN_MISSING"}) raise HTTPException(status_code=403, detail="CSRF token missing or already used.") - + provided_csrf_hash = hashlib.sha256(csrf_token.encode()).hexdigest() if not secrets.compare_digest(provided_csrf_hash, stored_csrf_hash): logger.error(f"Invalid CSRF token for ticket {ticket_id}.", extra={"ticket_id": ticket_id, "user_email": user_email, "event_type": "APPROVAL_INVALID_CSRF_TOKEN"}) @@ -563,15 +563,15 @@ def _get_identity_token(audience: str) -> str: async def _agent_headers() -> dict: """Headers for machine-to-machine calls to osvauco-agent.""" token = await asyncio.to_thread(_get_identity_token, OSVAUCO_AGENT_URL) - + headers = { "X-Internal-Key": INTERNAL_API_KEY, "Content-Type": "application/json", } - + if token: headers["Authorization"] = f"Bearer {token}" - + return headers async def _agent_get(path: str) -> Any: @@ -735,13 +735,13 @@ async def trigger_build(p: dict) -> dict: try: client = cloudbuild_v1.CloudBuildClient() source = cloudbuild_v1.RepoSource(branch_name=branch, substitutions=substitutions) - + response = client.run_build_trigger( project_id=project_id, trigger_id=trigger_id, source=source, ) - + build_id = response.metadata.build.id logger.info(f"Successfully triggered Cloud Build. Build ID: {build_id}") return {"status": "success", "build_id": build_id} @@ -943,8 +943,18 @@ async def run_emma(p: dict) -> dict: async def run_emma_fast(p): return await _ollama_chat(EMMA_FAST_MODEL, p.get("prompt", p.get("message", "")), p.get("system", "Du er en rask og konsis AI-assistent...")) async def run_qwen(p): return await _ollama_chat(EMMA_LIGHT_MODEL, p.get("prompt", p.get("message", ""))) async def list_emma_models(p): return await _ollama_models() +async def list_gitea_repositories(p: dict) -> dict: + """Wrapper for the read-only, metadata-only repository catalog.""" + if p: + raise ValueError("list_gitea_repositories does not accept any arguments.") + return await list_allowed_namespace_repositories() + 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 get_file(p: dict) -> dict: + """Wrapper for the hardened get_file_content handler.""" + # Note: server_repo_id (GITEA_REPO) is validated inside the handler now. + return await handle_get_file_content(p, GITEA_REPO) 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: @@ -953,7 +963,7 @@ async def list_repo_files(p: dict) -> dict: 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): +async def push_file(p): # ... (beholdt uendret) repo, path = p.get("repo", GITEA_REPO), p.get("path") content = base64.b64encode(p.get("content", "").encode()).decode() @@ -977,12 +987,12 @@ def _resolve_memory_path(file_name: str) -> str: """Resolves a file path and ensures it is within the project/ directory.""" if not file_name or '..' in file_name or file_name.startswith('/'): raise ValueError("Invalid file name.") - + abs_path = os.path.abspath(os.path.join(PROJECT_DIR, file_name)) - + if not abs_path.startswith(PROJECT_DIR): raise ValueError("Security violation: Path is outside the allowed memory bank directory.") - + return abs_path async def read_memory_bank(p: dict) -> dict: @@ -990,9 +1000,9 @@ async def read_memory_bank(p: dict) -> dict: file_name = p.get("file_name") if not file_name: raise ValueError("Missing required parameter: 'file_name'") - + file_path = _resolve_memory_path(file_name) - + try: with open(file_path, 'r', encoding='utf-8') as f: content = f.read() @@ -1030,13 +1040,13 @@ async def build_and_deploy_service(p: dict) -> dict: """ branch_to_deploy = p.get("branch", "main") logger.info(f"Triggering production deployment for branch: {branch_to_deploy}") - + result = await trigger_build({ "trigger_id": PROD_BUILD_TRIGGER_ID, "branch": branch_to_deploy, "substitutions": p.get("substitutions", {}) }) - + return result @@ -1246,8 +1256,9 @@ TOOLS = { "list_customers": (list_customers, "List alle kunder (alias for get_state)", {}), "run_terminal": (run_terminal, "Kjør terminalkommando på VM", {"type":"object","properties":{"command":{"type":"string"}}}), # Gitea / VCS + "list_gitea_repositories": (list_gitea_repositories, "Lists repositories in the approved 'chris' namespace. Listing does not grant read, build, or deploy authority.", {}), "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"]}), + "get_file": (get_file, "Hent fil fra Gitea-repo", {"type":"object","properties":{"path":{"type":"string"}, "ref":{"type":"string"}, "repo":{"type":"string"}},"required":["path", "ref"]}), "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"]}), diff --git a/opax-mcp/test_gitea_helpers.py b/opax-mcp/test_gitea_helpers.py index 890e2e7..ef41559 100644 --- a/opax-mcp/test_gitea_helpers.py +++ b/opax-mcp/test_gitea_helpers.py @@ -3,18 +3,63 @@ from unittest.mock import AsyncMock, MagicMock, patch import os import sys from pathlib import Path +import base64 +import httpx +import json +import binascii + +# Correction 4: Schema test isolation +# Before importing server, install only a minimal test stub in sys.modules for +# the unrelated runtime dependency `emma_adapter`. +emma_adapter_stub = MagicMock() +emma_adapter_stub.CanonicalEmma = MagicMock() +sys.modules['emma_adapter'] = emma_adapter_stub REPO_ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(REPO_ROOT / "opax-mcp")) +# Now that the path is set, we can import the modules import gitea_handler +import server + +def make_stream_context(response_content, headers=None): + async def aiter_bytes(): + if isinstance(response_content, list): + for item in response_content: + yield item + else: + yield response_content + + response = MagicMock() + response.aiter_bytes = aiter_bytes + response.raise_for_status = MagicMock() + response.headers = headers if headers is not None else {} -def make_stream_context(response): context = MagicMock() context.__aenter__ = AsyncMock(return_value=response) context.__aexit__ = AsyncMock(return_value=False) return context +def configure_stream(client, response_content, headers=None): + client.stream = MagicMock( + return_value=make_stream_context(response_content, headers=headers) + ) + +def _get_mock_file_response(content=b"test content", size=None, content_b64=None): + """Helper to create a default valid mock response for get_file tests.""" + if content_b64 is None: + encoded_content = base64.b64encode(content).decode('ascii') + else: + encoded_content = content_b64 + + if size is None: + size = len(content) + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.json.return_value = {"size": size, "content": encoded_content} + mock_response.raise_for_status = MagicMock() + return mock_response + class TestGiteaHelpers(unittest.IsolatedAsyncioTestCase): def test_validate_branch_name_valid(self): self.assertEqual(gitea_handler._validate_branch_name("main"), "main") @@ -29,13 +74,13 @@ class TestGiteaHelpers(unittest.IsolatedAsyncioTestCase): gitea_handler._validate_branch_name("feat//bad") def test_validate_commit_sha_valid(self): - self.assertEqual(gitea_handler._validate_commit_sha("395aad5d70d1d30ccd4c3f84e3dfb56e38a2c2c9"), "395aad5d70d1d30ccd4c3f84e3dfb56e38a2c2c9") + self.assertEqual(gitea_handler._validate_commit_sha("f" * 40), "f" * 40) def test_validate_commit_sha_invalid(self): with self.assertRaises(ValueError): gitea_handler._validate_commit_sha("main") with self.assertRaises(ValueError): - gitea_handler._validate_commit_sha("395aad5d70d1d30ccd4c3f84e3dfb56e38a2c2c") # 39 chars + gitea_handler._validate_commit_sha("f" * 39) def test_validate_gitea_url_https_required(self): with self.assertRaises(ValueError): @@ -45,29 +90,20 @@ class TestGiteaHelpers(unittest.IsolatedAsyncioTestCase): async def test_resolve_branch_to_commit_sha(self, mock_client): mock_response = unittest.mock.Mock() mock_response.status_code = 200 - mock_response.json.return_value = {"commit": {"id": "395aad5d70d1d30ccd4c3f84e3dfb56e38a2c2c9"}} + mock_response.json.return_value = {"commit": {"id": "f" * 40}} mock_client.return_value.__aenter__.return_value.get.return_value = mock_response sha = await gitea_handler.resolve_branch_to_commit_sha("main", "chris/OSVauco", "https://gitea.example.com") - self.assertEqual(sha, "395aad5d70d1d30ccd4c3f84e3dfb56e38a2c2c9") + self.assertEqual(sha, "f" * 40) @patch("gitea_handler.httpx.AsyncClient") async def test_download_repo_archive_uses_sha(self, mock_client): - response = MagicMock() - response.headers = {} - response.raise_for_status = MagicMock() - - async def aiter_bytes(): - yield b"archive-data" - - response.aiter_bytes = aiter_bytes - - stream_context = make_stream_context(response) + stream_context = make_stream_context(b"archive-data") client = mock_client.return_value.__aenter__.return_value client.stream = MagicMock(return_value=stream_context) archive = await gitea_handler.download_repo_archive( - "395aad5d70d1d30ccd4c3f84e3dfb56e38a2c2c9", + "f" * 40, "chris/OSVauco", "https://gitea.example.com", ) @@ -77,9 +113,8 @@ class TestGiteaHelpers(unittest.IsolatedAsyncioTestCase): client.stream.assert_called_once_with( "GET", ( - "https://gitea.example.com/api/v1/repos/" - "chris/OSVauco/archive/" - "395aad5d70d1d30ccd4c3f84e3dfb56e38a2c2c9.tar.gz" + f"https://gitea.example.com/api/v1/repos/" + f"chris/OSVauco/archive/{'f' * 40}.tar.gz" ), headers=unittest.mock.ANY, ) @@ -94,37 +129,30 @@ class TestGiteaHelpers(unittest.IsolatedAsyncioTestCase): } response.raise_for_status = MagicMock() - stream_context = make_stream_context(response) + stream_context = make_stream_context([]) # Empty stream client = mock_client.return_value.__aenter__.return_value client.stream = MagicMock(return_value=stream_context) + # The response is used to check headers, not the stream content + stream_context.__aenter__.return_value = response with self.assertRaisesRegex( ValueError, "Source archive exceeds allowed size", ): await gitea_handler.download_repo_archive( - "395aad5d70d1d30ccd4c3f84e3dfb56e38a2c2c9", + "f" * 40, "chris/OSVauco", "https://gitea.example.com", ) @patch("gitea_handler.httpx.AsyncClient") async def test_download_repo_archive_no_redirects(self, mock_client): - response = MagicMock() - response.headers = {} - response.raise_for_status = MagicMock() - - async def aiter_bytes(): - yield b"" - - response.aiter_bytes = aiter_bytes - - stream_context = make_stream_context(response) + stream_context = make_stream_context(b"") client = mock_client.return_value.__aenter__.return_value client.stream = MagicMock(return_value=stream_context) await gitea_handler.download_repo_archive( - "395aad5d70d1d30ccd4c3f84e3dfb56e38a2c2c9", + "f" * 40, "chris/OSVauco", "https://gitea.example.com", ) @@ -133,5 +161,265 @@ class TestGiteaHelpers(unittest.IsolatedAsyncioTestCase): mock_client.call_args.kwargs["follow_redirects"] ) +class TestGiteaCatalog(unittest.IsolatedAsyncioTestCase): + def setUp(self): + self.mock_env = patch.dict(os.environ, { + "GITEA_URL": "https://gitea.example.com", + "GITEA_TOKEN": "fake-token", + }) + self.mock_env.start() + + def tearDown(self): + self.mock_env.stop() + + @patch("gitea_handler.httpx.AsyncClient") + async def test_catalog_uses_fixed_namespace(self, mock_client): + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.json.return_value = [] + mock_client.return_value.__aenter__.return_value.get.return_value = mock_response + + await gitea_handler.list_allowed_namespace_repositories() + + mock_client.return_value.__aenter__.return_value.get.assert_called_once() + call_url = mock_client.return_value.__aenter__.return_value.get.call_args[0][0] + self.assertIn("/api/v1/users/chris/repos", call_url) + + @patch("gitea_handler.httpx.AsyncClient") + async def test_catalog_pagination_is_bounded(self, mock_client): + async def get_response(url, headers): + mock_response = MagicMock() + mock_response.status_code = 200 + if "page=1" in url: + mock_response.json.return_value = [{"full_name": f"chris/repo{i}", "name": f"repo{i}"} for i in range(50)] + elif "page=2" in url: + mock_response.json.return_value = [{"full_name": f"chris/repo{i}", "name": f"repo{i}"} for i in range(50, 100)] + else: # Should not be called for page 3 + mock_response.json.return_value = [] + return mock_response + + mock_client.return_value.__aenter__.return_value.get.side_effect = get_response + + result = await gitea_handler.list_allowed_namespace_repositories() + self.assertEqual(mock_client.return_value.__aenter__.return_value.get.call_count, 2) + self.assertEqual(len(result["repositories"]), 100) + + @patch("gitea_handler.httpx.AsyncClient") + async def test_catalog_normalization_and_security(self, mock_client): + gitea_response = [ + {"full_name": "chris/repo1", "name": "repo1", "clone_url": "sensitive", "permissions": {}}, + {"full_name": "other/repo2", "name": "repo2"}, + {"full_name": "chris/repo3"}, + {"full_name": "chris/repo1", "name": "repo1"}, + ] + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.json.return_value = gitea_response + mock_client.return_value.__aenter__.return_value.get.return_value = mock_response + + result = await gitea_handler.list_allowed_namespace_repositories() + self.assertEqual(len(result["repositories"]), 1) + repo = result["repositories"][0] + self.assertEqual(repo["full_name"], "chris/repo1") + self.assertNotIn("clone_url", repo) + self.assertNotIn("permissions", repo) + + @patch("gitea_handler.httpx.AsyncClient") + async def test_catalog_upstream_failure_500(self, mock_client): + mock_client.return_value.__aenter__.return_value.get.side_effect = httpx.HTTPStatusError( + "Internal Server Error", request=MagicMock(), response=MagicMock(status_code=500) + ) + with self.assertRaisesRegex(ValueError, "Repository catalog unavailable."): + await gitea_handler.list_allowed_namespace_repositories() + + @patch("gitea_handler.httpx.AsyncClient") + async def test_catalog_upstream_failure_timeout(self, mock_client): + mock_client.return_value.__aenter__.return_value.get.side_effect = httpx.TimeoutException("Timeout") + with self.assertRaisesRegex(ValueError, "Repository catalog unavailable."): + await gitea_handler.list_allowed_namespace_repositories() + + @patch("gitea_handler.httpx.AsyncClient") + async def test_catalog_malformed_json(self, mock_client): + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.json.side_effect = json.JSONDecodeError("err", "doc", 0) + mock_client.return_value.__aenter__.return_value.get.return_value = mock_response + with self.assertRaisesRegex(ValueError, "Repository catalog unavailable."): + await gitea_handler.list_allowed_namespace_repositories() + + +class TestGetFileHardening(unittest.IsolatedAsyncioTestCase): + def setUp(self): + self.mock_env_patcher = patch.dict(os.environ, { + "GITEA_URL": "https://gitea.example.com", + "GITEA_TOKEN": "fake-token", + "GITEA_REPO": "chris/OSVauco", + }) + self.mock_env = self.mock_env_patcher.start() + self.valid_sha = "f" * 40 + self.params = {"path": "README.md", "ref": self.valid_sha} + self.repo = "chris/OSVauco" + + + def tearDown(self): + self.mock_env_patcher.stop() + + @patch("gitea_handler.httpx.AsyncClient") + async def test_get_file_repo_compatibility(self, mock_client): + client = mock_client.return_value.__aenter__.return_value + configure_stream(client, json.dumps({'content': 'YQ==', 'size': 1}).encode("utf-8")) + + # Allowed cases + await gitea_handler.handle_get_file_content( + {"path": "README.md", "ref": self.valid_sha}, "chris/OSVauco" + ) + await gitea_handler.handle_get_file_content( + {"path": "README.md", "ref": self.valid_sha, "repo": None}, "chris/OSVauco" + ) + await gitea_handler.handle_get_file_content( + {"path": "README.md", "ref": self.valid_sha, "repo": "chris/OSVauco"}, "chris/OSVauco" + ) + self.assertEqual(client.stream.call_count, 3) + + # Rejected cases + rejected_repos = { + "empty_string": "", + "wrong_type_false": False, + "wrong_type_zero": 0, + "mismatch": "other/repo", + } + for name, repo_val in rejected_repos.items(): + with self.subTest(rejected_case=name): + client.stream.reset_mock() + with self.assertRaisesRegex(ValueError, "Repository file request is not allowed."): + await gitea_handler.handle_get_file_content( + {"path": "README.md", "ref": self.valid_sha, "repo": repo_val}, "chris/OSVauco" + ) + client.stream.assert_not_called() + + async def test_get_file_ref_validation(self): + with self.assertRaisesRegex(ValueError, "Invalid commit SHA"): + await gitea_handler.handle_get_file_content({"path": "README.md", "ref": "main"}, "chris/OSVauco") + with self.assertRaisesRegex(ValueError, "Invalid commit SHA"): + await gitea_handler.handle_get_file_content({"path": "README.md", "ref": None}, "chris/OSVauco") + + async def test_gitea_repo_validation(self): + with self.assertRaisesRegex(ValueError, "Invalid configured Gitea repository ID"): + await gitea_handler.handle_get_file_content({"path": "README.md", "ref": self.valid_sha}, "invalid-repo-id") + with self.assertRaisesRegex(ValueError, "Invalid configured Gitea repository ID"): + await gitea_handler.handle_get_file_content({"path": "README.md", "ref": self.valid_sha}, "") + + @patch("gitea_handler.httpx.AsyncClient") + async def test_get_file_path_validation(self, mock_client): + client = mock_client.return_value.__aenter__.return_value + configure_stream(client, json.dumps({'content': 'YQ==', 'size': 1}).encode("utf-8")) + for path in ["../secrets.txt", "/etc/passwd", "src/main.py", "docs/.env", "file.json"]: + with self.subTest(path=path): + with self.assertRaisesRegex(ValueError, "Repository file request is not allowed."): + await gitea_handler.handle_get_file_content({"path": path, "ref": self.valid_sha}, "chris/OSVauco") + for path in ["README.md", "docs/ARCHITECTURE.md", ".gemini/GEMINI.md"]: + with self.subTest(path=path): + await gitea_handler.handle_get_file_content({"path": path, "ref": self.valid_sha}, "chris/OSVauco") + + @patch("gitea_handler.httpx.AsyncClient") + async def test_get_file_size_limit(self, mock_client): + client = mock_client.return_value.__aenter__.return_value + + # Declared size too large + response_body = json.dumps({"size": gitea_handler.MAX_READ_FILE_BYTES + 1, "content": ""}).encode("utf-8") + configure_stream(client, response_body) + with self.assertRaisesRegex(ValueError, "Repository file exceeds the allowed size."): + await gitea_handler.handle_get_file_content({"path": "README.md", "ref": self.valid_sha}, "chris/OSVauco") + + # Transport size too large + large_body = b'{' * (gitea_handler.MAX_GET_FILE_RESPONSE_BYTES + 1) + configure_stream(client, large_body) + with self.assertRaisesRegex(ValueError, "Repository file exceeds the allowed size."): + await gitea_handler.handle_get_file_content({"path": "README.md", "ref": self.valid_sha}, "chris/OSVauco") + + @patch("gitea_handler.httpx.AsyncClient") + async def test_get_file_content_validation(self, mock_client): + client = mock_client.return_value.__aenter__.return_value + + # Malformed Base64 + configure_stream(client, json.dumps({'content': 'invalid-b64!', 'size': 12}).encode("utf-8")) + with self.assertRaisesRegex(ValueError, "Repository file content is not readable text."): + await gitea_handler.handle_get_file_content({"path": "README.md", "ref": self.valid_sha}, "chris/OSVauco") + + # Invalid UTF-8 + invalid_utf8_b64 = base64.b64encode(b"\xff\xff\xff").decode('ascii') + configure_stream(client, json.dumps({'content': invalid_utf8_b64, 'size': 3}).encode("utf-8")) + with self.assertRaisesRegex(ValueError, "Repository file content is not readable text."): + await gitea_handler.handle_get_file_content({"path": "README.md", "ref": self.valid_sha}, "chris/OSVauco") + + # NUL byte + nul_byte_b64 = base64.b64encode(b"hello\0world").decode('ascii') + configure_stream(client, json.dumps({'content': nul_byte_b64, 'size': 11}).encode("utf-8")) + with self.assertRaisesRegex(ValueError, "Repository file content is not readable text."): + await gitea_handler.handle_get_file_content({"path": "README.md", "ref": self.valid_sha}, "chris/OSVauco") + + @patch("gitea_handler.httpx.AsyncClient") + async def test_content_length_validation(self, mock_client): + client = mock_client.return_value.__aenter__.return_value + test_cases = { + "absent": None, + "empty": "", + "non-numeric": "not-a-number", + "negative": "-100", + } + for name, length_val in test_cases.items(): + with self.subTest(name=name): + headers = {"content-length": length_val} if length_val is not None else {} + configure_stream(client, + json.dumps({'content': 'YQ==', 'size': 1}).encode("utf-8"), headers=headers + ) + # No ValueError should be raised, should proceed and succeed + result = await gitea_handler.handle_get_file_content(self.params, self.repo) + self.assertEqual(result["content"], "a") + + # Test valid but too large + with self.subTest(name="valid_too_large"): + headers = {"content-length": str(gitea_handler.MAX_GET_FILE_RESPONSE_BYTES + 1)} + configure_stream(client, b"", headers=headers) + with self.assertRaisesRegex(ValueError, "Repository file exceeds the allowed size."): + await gitea_handler.handle_get_file_content(self.params, self.repo) + + @patch("gitea_handler.httpx.AsyncClient") + async def test_json_size_validation(self, mock_client): + client = mock_client.return_value.__aenter__.return_value + test_cases = { + "True": True, + "False": False, + "negative": -1, + "string": "123", + } + for name, size_val in test_cases.items(): + with self.subTest(name=name): + body = json.dumps({"content": "YQ==", "size": size_val}).encode("utf-8") + configure_stream(client, body) + # No error should be raised, should proceed and succeed + result = await gitea_handler.handle_get_file_content(self.params, self.repo) + self.assertEqual(result["content"], "a") + + # Test valid int over the limit + with self.subTest(name="valid_too_large"): + body = json.dumps({ + "content": "", "size": gitea_handler.MAX_READ_FILE_BYTES + 1 + }).encode("utf-8") + configure_stream(client, body) + with self.assertRaisesRegex(ValueError, "Repository file exceeds the allowed size."): + await gitea_handler.handle_get_file_content(self.params, self.repo) + + +class TestMCPSchema(unittest.TestCase): + def test_get_file_schema(self): + _handler, _desc, schema = server.TOOLS["get_file"] + self.assertEqual(schema["properties"].keys(), {"path", "ref", "repo"}) + self.assertEqual(schema["required"], ["path", "ref"]) + + def test_list_gitea_repositories_schema(self): + _handler, _desc, schema = server.TOOLS["list_gitea_repositories"] + self.assertEqual(schema, {}) + if __name__ == "__main__": - unittest.main() \ No newline at end of file + unittest.main()