From e9b51245a4684f36f0c3bfcea48ce8f910a61650 Mon Sep 17 00:00:00 2001 From: Chris Christiansen Date: Wed, 23 Sep 2026 20:26:10 +0000 Subject: [PATCH] feat(opax-mcp): add safe Gitea file patch tool --- opax-mcp/gitea_handler.py | 157 +++++++++++++++++++++++++++++++++ opax-mcp/server.py | 45 ++++++++++ opax-mcp/test_gitea_handler.py | 119 ++++++++++++++++++++++++- 3 files changed, 320 insertions(+), 1 deletion(-) diff --git a/opax-mcp/gitea_handler.py b/opax-mcp/gitea_handler.py index c266b8f..a2e3f52 100644 --- a/opax-mcp/gitea_handler.py +++ b/opax-mcp/gitea_handler.py @@ -6,6 +6,158 @@ import re import base64 import json import binascii +from typing import List, Literal, Callable, Awaitable, Any, Dict +from pydantic import BaseModel, Field, ValidationError + + +# --- Gitea Change Operation Models & Logic --- + +MAX_GITEA_PATCH_OPERATIONS = 50 +MAX_GITEA_PATCH_INPUT_BYTES = 131_072 + + +class ExactReplacementOperation(BaseModel, extra="forbid"): + """A single, deterministic replacement operation.""" + type: Literal["replace_once"] + find: str = Field(min_length=1) + replace: str + + +def apply_gitea_patch_operations( + original_content: str, + operations: List[ExactReplacementOperation], +) -> str: + """ + Applies a sequence of deterministic patch operations to a string. + Ensures that each `find` pattern matches exactly once in the content + as it evolves through the sequence of operations. + """ + if not isinstance(original_content, str): + raise TypeError("original_content must be a string.") + + if not operations: + raise ValueError("Change operations cannot be empty.") + + if len(operations) > MAX_GITEA_PATCH_OPERATIONS: + raise ValueError( + f"Too many patch operations: {len(operations)} > " + f"{MAX_GITEA_PATCH_OPERATIONS}." + ) + + total_patch_bytes = sum( + len(operation.find.encode("utf-8")) + + len(operation.replace.encode("utf-8")) + for operation in operations + ) + if total_patch_bytes > MAX_GITEA_PATCH_INPUT_BYTES: + raise ValueError( + f"Patch input exceeds {MAX_GITEA_PATCH_INPUT_BYTES} bytes." + ) + + content = original_content + for i, op in enumerate(operations): + if op.type != "replace_once": + raise ValueError(f"Unsupported operation type at index {i}: {op.type}") + + if not op.find: + raise ValueError(f"Operation at index {i} has an empty 'find' value.") + + match_count = content.count(op.find) + if match_count == 0: + raise ValueError( + f"Operation at index {i} failed: The 'find' string was not found." + ) + if match_count > 1: + raise ValueError( + f"Operation at index {i} failed: The 'find' string matched {match_count} times (expected 1)." + ) + + content = content.replace(op.find, op.replace, 1) + + if content == original_content: + raise ValueError("The applied operations resulted in no net change to the file content.") + + return content + +async def orchestrate_gitea_patch( + params: Dict[str, Any], + read_file_func: Callable[ + [Dict[str, Any], str], + Awaitable[Dict[str, Any]], + ], + write_file_func: Callable[ + [str, Dict[str, Any]], + Awaitable[Dict[str, Any]], + ], +) -> Dict[str, Any]: + """ + Orchestrates a safe, server-side patch workflow against Gitea. + This pure helper contains the core logic and is dependency-injected for testability. + """ + if not isinstance(params, dict): + raise ValueError("patch_gitea_file input must be an object.") + + allowed_keys = {"repo", "branch", "path", "commit_message", "operations"} + unsupported_keys = set(params) - allowed_keys + if unsupported_keys: + raise ValueError( + f"Unsupported patch_gitea_file input fields: {sorted(unsupported_keys)}" + ) + + # 1. Extract and validate all inputs before any network access. + repo = params.get("repo") + branch = params.get("branch") + path = params.get("path") + commit_message = params.get("commit_message") + operations_data = params.get("operations") + + validate_repo_for_write(repo) + validate_branch_for_write(branch) + validate_path_for_write(path) + + if not isinstance(commit_message, str) or not commit_message.strip(): + raise ValueError("commit_message must be a non-empty string.") + + if not isinstance(operations_data, list) or not operations_data: + raise ValueError("operations must be a non-empty list.") + + try: + parsed_operations = [ExactReplacementOperation(**op) for op in operations_data] + except (TypeError, ValidationError) as e: + raise ValueError(f"Invalid operation object provided: {e}") from e + + # 2. Read current file state from Gitea. + file_state = await read_file_func( + {"repo": repo, "path": path, "ref": branch}, + repo, + ) + original_content = file_state["content"] + file_sha = file_state["file_sha"] + + # 3. Apply the patch operations to the fetched content. + try: + patched_content = apply_gitea_patch_operations( + original_content, + parsed_operations, + ) + except ValueError as e: + raise ValueError(f"Failed to apply patch: {e}") from e + + # 4. Write the update to Gitea. + encoded_patched_content = base64.b64encode(patched_content.encode("utf-8")).decode("ascii") + put_body = { + "branch": branch, + "message": commit_message, + "content": encoded_patched_content, + "sha": file_sha, + } + + return await write_file_func( + f"/repos/{repo}/contents/{path}", + put_body, + ) + + logger = logging.getLogger(__name__) @@ -459,11 +611,16 @@ async def handle_get_file_content(p: dict, server_repo_id: str) -> dict: if '\0' in text_content: raise ValueError("Repository file content is not readable text.") + file_sha = data.get("sha") + if not isinstance(file_sha, str) or not file_sha: + raise ValueError("Repository file is unavailable: upstream response missing file SHA.") + return { "repo": validated_server_repo, "path": path, "requested_ref": requested_ref, "resolved_commit_sha": resolved_commit_sha, + "file_sha": _validate_commit_sha(file_sha), "content": text_content, "encoding": "utf-8", } diff --git a/opax-mcp/server.py b/opax-mcp/server.py index 7178c17..bd9f36e 100644 --- a/opax-mcp/server.py +++ b/opax-mcp/server.py @@ -70,6 +70,7 @@ from gitea_handler import ( validate_repo_for_write, validate_branch_for_write, validate_path_for_write, + orchestrate_gitea_patch, ) from capability_bridge import build_capability_system_context from deployment_policy import get_deployment_target @@ -1354,6 +1355,15 @@ 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 patch_gitea_file(p: dict) -> dict: + """Applies a patch to a file in Gitea via the injected orchestration helper.""" + return await orchestrate_gitea_patch( + p, + handle_get_file_content, + _gitea_put, + ) + + async def push_file(p): # ... (beholdt uendret) repo, path = p.get("repo", GITEA_REPO), p.get("path") @@ -1730,6 +1740,41 @@ TOOLS = { "required": ["repo", "branch", "path", "new_content", "base_sha", "commit_message"] }, ), + "patch_gitea_file": ( + patch_gitea_file, + "Applies ordered exact-once replacements to an existing text file and commits the result. Each find value must match exactly once.", + { + "type": "object", + "additionalProperties": False, + "properties": { + "repo": {"type": "string"}, + "branch": {"type": "string"}, + "path": {"type": "string"}, + "commit_message": {"type": "string"}, + "operations": { + "type": "array", + "minItems": 1, + "items": { + "type": "object", + "additionalProperties": False, + "properties": { + "type": { + "type": "string", + "enum": ["replace_once"], + }, + "find": { + "type": "string", + "minLength": 1, + }, + "replace": {"type": "string"}, + }, + "required": ["type", "find", "replace"], + }, + }, + }, + "required": ["repo", "branch", "path", "commit_message", "operations"], + }, + ), # 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_gitea_handler.py b/opax-mcp/test_gitea_handler.py index b609cc5..16cba90 100644 --- a/opax-mcp/test_gitea_handler.py +++ b/opax-mcp/test_gitea_handler.py @@ -10,7 +10,12 @@ 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, _validate_read_branch_ref, handle_get_file_content +from gitea_handler import ( + handle_list_repo_files, + _validate_read_branch_ref, + handle_get_file_content, + orchestrate_gitea_patch, +) class TestValidateReadBranchRef(unittest.TestCase): @@ -723,5 +728,117 @@ class TestHandleGetFileContent(unittest.TestCase): client.stream.assert_called_once() +class TestOrchestrateGiteaPatch(unittest.IsolatedAsyncioTestCase): + def setUp(self): + self.base_params = { + "repo": "chris/OSVauco", + "branch": "agent/test-patch", + "path": "path/to/file.txt", + "commit_message": "Patch file", + "operations": [{"type": "replace_once", "find": "before", "replace": "after"}], + } + self.read_func = AsyncMock() + self.write_func = AsyncMock() + + async def test_success_flow(self): + """Tests the ideal success path, verifying exact collaborator calls.""" + self.read_func.return_value = {"content": "before", "file_sha": "f" * 40} + self.write_func.return_value = {"commit": {"sha": "c" * 40}} + + result = await orchestrate_gitea_patch( + self.base_params, self.read_func, self.write_func + ) + self.assertEqual(result, {"commit": {"sha": "c" * 40}}) + + self.read_func.assert_awaited_once_with( + {"repo": "chris/OSVauco", "path": "path/to/file.txt", "ref": "agent/test-patch"}, + "chris/OSVauco", + ) + + expected_body = { + "branch": "agent/test-patch", + "message": "Patch file", + "content": base64.b64encode(b"after").decode("ascii"), + "sha": "f" * 40, + } + self.write_func.assert_awaited_once_with( + "/repos/chris/OSVauco/contents/path/to/file.txt", + expected_body, + ) + + async def test_non_dict_params_rejected_before_io(self): + """Tests that non-dict input is rejected with a precise error.""" + with self.assertRaisesRegex( + ValueError, r"^patch_gitea_file input must be an object\.$" + ): + await orchestrate_gitea_patch("not a dict", self.read_func, self.write_func) + self.read_func.assert_not_awaited() + self.write_func.assert_not_awaited() + + async def test_unsupported_top_level_field_rejected_before_io(self): + """Tests that unsupported top-level fields are rejected with a precise error.""" + params = {**self.base_params, "extra": "field"} + with self.assertRaisesRegex( + ValueError, r"^Unsupported patch_gitea_file input fields: \['extra'\]$" + ): + await orchestrate_gitea_patch(params, self.read_func, self.write_func) + self.read_func.assert_not_awaited() + self.write_func.assert_not_awaited() + + async def test_field_and_operation_validation_failures(self): + """Tests that invalid required fields and operations fail before any I/O.""" + test_cases = { + "bad_repo": ("repo", "other/repo"), + "protected_branch": ("branch", "main"), + "invalid_path": ("path", "../secrets"), + "empty_commit": ("commit_message", " "), + "empty_ops": ("operations", []), + "non_list_ops": ("operations", "not-a-list"), + "malformed_ops_obj": ("operations", [{"find": "a"}]), + "unknown_operation_field": ( + "operations", + [{"type": "replace_once", "find": "a", "replace": "b", "extra": True}], + ), + } + for name, (key, value) in test_cases.items(): + with self.subTest(name=name): + bad_params = self.base_params.copy() + bad_params[key] = value + with self.assertRaises(ValueError): + await orchestrate_gitea_patch(bad_params, self.read_func, self.write_func) + self.read_func.assert_not_awaited() + self.write_func.assert_not_awaited() + self.read_func.reset_mock(return_value=True, side_effect=True) + self.write_func.reset_mock(return_value=True, side_effect=True) + + async def test_patch_logic_failures_prevent_write(self): + """Tests that patch application failures (no-match, multi-match) prevent writes.""" + test_cases = { + "no_match": "content does not match", + "multiple_matches": "before before", + } + for name, content in test_cases.items(): + with self.subTest(name=name): + self.read_func.return_value = {"content": content, "file_sha": "f" * 40} + with self.assertRaisesRegex(ValueError, "Failed to apply patch"): + await orchestrate_gitea_patch( + self.base_params, self.read_func, self.write_func + ) + self.read_func.assert_awaited_once() + self.write_func.assert_not_awaited() + self.read_func.reset_mock(return_value=True, side_effect=True) + self.write_func.reset_mock(return_value=True, side_effect=True) + + async def test_read_failure_prevents_write(self): + """Tests that a failure in the injected read function prevents writes.""" + self.read_func.side_effect = ValueError("Upstream read failed") + with self.assertRaisesRegex(ValueError, "Upstream read failed"): + await orchestrate_gitea_patch( + self.base_params, self.read_func, self.write_func + ) + self.read_func.assert_awaited_once() + self.write_func.assert_not_awaited() + + if __name__ == '__main__': unittest.main()