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

feat(opax-mcp): add safe Gitea repository catalog

Browse Source
This commit is contained in:
Chris Christiansen 2026-09-20 18:20:56 +00:00
parent 6c4d197051
commit 07626c83f4
3 changed files with 535 additions and 75 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

195
opax-mcp/gitea_handler.py
View File

@ -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")
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
repo_id = p.get("repo", default_repo)
ref = p.get("ref", "main")
full_name = item.get("full_name")
name = item.get("name")
url = f"{gitea_url}/api/v1/repos/{repo_id}/raw/{path}?ref={ref}"
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:
r = await c.get(url, headers=_gitea_headers())
r.raise_for_status()
return {"path": path, "content": r.text, "encoding": "text"}
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."""

17
opax-mcp/server.py
View File

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

352
opax-mcp/test_gitea_helpers.py
View File

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