feat(opax-mcp): add safe Gitea file patch tool

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

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

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