Harden Gitea change proposals

opax-mcp/gitea_handler.py

@@ -113,6 +113,46 @@ def _validate_safe_path(path: str) -> str:
     return path
 
 
+MAX_PUSH_CONTENT_BYTES = 1_048_576  # 1 MB
+
+def validate_repo_for_write(repo_id: str) -> str:
+    """Validates that a repo is in the allowed namespace for write operations."""
+    if not isinstance(repo_id, str) or not _REPO_ID_RE.fullmatch(repo_id):
+        raise ValueError("Invalid repository format. Must be 'owner/repo'.")
+    if not repo_id.startswith(f"{ALLOWED_GITEA_NAMESPACE}/"):
+        raise ValueError(f"Write operations are only allowed in the '{ALLOWED_GITEA_NAMESPACE}' namespace.")
+    return repo_id
+
+def validate_branch_for_write(branch: str) -> str:
+    """Validates that a branch name is safe for write operations."""
+    if not branch or not isinstance(branch, str):
+        raise ValueError("Branch name cannot be empty.")
+    if branch.lower() in ["main", "master"]:
+        raise ValueError(f"Direct writes to protected branch '{branch}' are not allowed.")
+    return _validate_branch_name(branch)
+
+def validate_path_for_write(path: str) -> str:
+    """Validates a file path for write operations with segment-based checks."""
+    if not path or not isinstance(path, str):
+        raise ValueError("Path cannot be empty.")
+    if "\\" in path or "\x00" in path or "//" in path:
+        raise ValueError("Path contains invalid characters.")
+    if path.startswith('/') or '..' in path.split('/'):
+        raise ValueError("Path must be relative and cannot contain traversal elements.")
+
+    path_segments = path.lower().split('/')
+    filename = path_segments[-1]
+
+    if '.git' in path_segments:
+        raise ValueError("Changes within a '.git' directory are not allowed.")
+    if filename == '.env' or filename.startswith('.env.'):
+        raise ValueError("Path targets a '.env' file, which is not allowed.")
+    sensitive_basenames = ["credentials", "service_account", "private_key", "id_rsa", "id_ed25519", "secret", "token"]
+    if filename in sensitive_basenames or any(filename.endswith(ext) for ext in ['.pem', '.key']):
+         raise ValueError(f"Path targets a sensitive basename or extension.")
+    return path
+
+
 async def resolve_branch_to_commit_sha(
     branch_name: str,
     repo_id: str,

opax-mcp/server.py

@@ -66,6 +66,9 @@ from gitea_handler import (
     handle_get_file_content,
     resolve_branch_to_commit_sha,
     download_repo_archive,
+    validate_repo_for_write,
+    validate_branch_for_write,
+    validate_path_for_write,
 )
 from capability_bridge import build_capability_system_context
 from deployment_policy import get_deployment_target
@@ -238,6 +241,9 @@ async def propose_gitea_change(p: dict) -> dict:
         )
     _validate_admin_gitea_path(path)
 
+    validate_path_for_write(path)
+    validate_branch_for_write(branch)
+
     live_sha = await resolve_branch_to_commit_sha(
         branch_name=branch,
         repo_id=repo,

opax-mcp/test_propose_gitea_change.py

@@ -0,0 +1,125 @@
+import unittest
+import server  # Følger eksisterende mønster
+import hashlib
+from unittest.mock import patch, AsyncMock
+
+class TestProposeGiteaChange(unittest.IsolatedAsyncioTestCase):
+
+    def setUp(self):
+        """Set up valid parameters for tests."""
+        self.valid_params = {
+            "repo": server.GITEA_REPO,
+            "branch": "feature/new-idea",
+            "path": "docs/new-file.md",
+            "new_content": "This is new content.",
+            "base_sha": "a" * 40,
+            "commit_message": "A valid commit message.",
+        }
+
+    def tearDown(self):
+        patch.stopall()
+
+    @patch('server.create_gitea_change_plan', new_callable=AsyncMock)
+    @patch('server._get_gitea_file_details', new_callable=AsyncMock)
+    @patch('server.resolve_branch_to_commit_sha', new_callable=AsyncMock)
+    async def test_rejects_main_branch_before_io(self, mock_resolve_sha, mock_get_details, mock_create_plan):
+        params = self.valid_params | {"branch": "main"}
+        with self.assertRaisesRegex(ValueError, "Direct writes to protected branch 'main' are not allowed."):
+            await server.propose_gitea_change(params)
+        
+        mock_resolve_sha.assert_not_awaited()
+        mock_get_details.assert_not_awaited()
+        mock_create_plan.assert_not_awaited()
+
+    @patch('server.create_gitea_change_plan', new_callable=AsyncMock)
+    @patch('server._get_gitea_file_details', new_callable=AsyncMock)
+    @patch('server.resolve_branch_to_commit_sha', new_callable=AsyncMock)
+    async def test_rejects_master_branch_before_io(self, mock_resolve_sha, mock_get_details, mock_create_plan):
+        params = self.valid_params | {"branch": "master"}
+        with self.assertRaisesRegex(ValueError, "Direct writes to protected branch 'master' are not allowed."):
+            await server.propose_gitea_change(params)
+
+        mock_resolve_sha.assert_not_awaited()
+        mock_get_details.assert_not_awaited()
+        mock_create_plan.assert_not_awaited()
+
+    @patch("server.create_gitea_change_plan", new_callable=AsyncMock)
+    @patch("server._get_gitea_file_details", new_callable=AsyncMock)
+    @patch("server.resolve_branch_to_commit_sha", new_callable=AsyncMock)
+    @patch("server._validate_admin_gitea_path")
+    async def test_rejects_dot_git_path_via_new_write_policy(
+        self,
+        mock_admin_path_validator,
+        mock_resolve_sha,
+        mock_get_details,
+        mock_create_plan,
+    ):
+        params = self.valid_params | {"path": "some/dir/.git/config"}
+
+        with self.assertRaisesRegex(
+            ValueError,
+            r"Changes within a '.git' directory are not allowed.",
+        ):
+            await server.propose_gitea_change(params)
+
+        mock_admin_path_validator.assert_called_once_with(params["path"])
+        mock_resolve_sha.assert_not_awaited()
+        mock_get_details.assert_not_awaited()
+        mock_create_plan.assert_not_awaited()
+
+    @patch('server.create_gitea_change_plan', new_callable=AsyncMock)
+    @patch('server._get_gitea_file_details', new_callable=AsyncMock)
+    @patch('server.resolve_branch_to_commit_sha', new_callable=AsyncMock)
+    async def test_rejects_base_sha_mismatch(self, mock_resolve_sha, mock_get_details, mock_create_plan):
+        mock_resolve_sha.return_value = "c" * 40  # Mismatched SHA
+        
+        with self.assertRaisesRegex(ValueError, "Branch head does not match the supplied base_sha"):
+            await server.propose_gitea_change(self.valid_params)
+            
+        mock_resolve_sha.assert_awaited_once()
+        mock_get_details.assert_not_awaited()
+        mock_create_plan.assert_not_awaited()
+
+    @patch('server.create_gitea_change_plan', new_callable=AsyncMock)
+    @patch('server._get_gitea_file_details', new_callable=AsyncMock)
+    @patch('server.resolve_branch_to_commit_sha', new_callable=AsyncMock)
+    async def test_rejects_no_op_diff(self, mock_resolve_sha, mock_get_details, mock_create_plan):
+        mock_resolve_sha.return_value = self.valid_params["base_sha"]
+        mock_get_details.return_value = (self.valid_params["new_content"], "b" * 40)
+        
+        with self.assertRaisesRegex(ValueError, "Proposed content produces no file change."):
+            await server.propose_gitea_change(self.valid_params)
+            
+        mock_resolve_sha.assert_awaited_once()
+        mock_get_details.assert_awaited_once()
+        mock_create_plan.assert_not_awaited()
+
+    @patch('server.create_gitea_change_plan', new_callable=AsyncMock)
+    @patch('server._get_gitea_file_details', new_callable=AsyncMock, return_value=("old content", "b" * 40))
+    @patch('server.resolve_branch_to_commit_sha', new_callable=AsyncMock)
+    async def test_happy_path_creates_pending_plan(self, mock_resolve_sha, mock_get_details, mock_create_plan):
+        mock_resolve_sha.return_value = self.valid_params["base_sha"]
+        
+        result = await server.propose_gitea_change(self.valid_params)
+
+        self.assertEqual(result["status"], "PENDING")
+        self.assertEqual(result["repo"], self.valid_params["repo"])
+        self.assertEqual(result["branch"], self.valid_params["branch"])
+        self.assertEqual(result["path"], self.valid_params["path"])
+        
+        expected_hash = hashlib.sha256(self.valid_params["new_content"].encode("utf-8")).hexdigest()
+        self.assertEqual(result["content_hash"], expected_hash)
+        
+        mock_create_plan.assert_awaited_once()
+        
+        # Verify the object passed to create_gitea_change_plan is the real Pydantic model
+        call_args = mock_create_plan.call_args[0][0]
+        self.assertIsInstance(call_args, server.GiteaChangePlan)
+        self.assertEqual(call_args.repo, self.valid_params["repo"])
+        self.assertEqual(call_args.branch, self.valid_params["branch"])
+        self.assertEqual(call_args.path, self.valid_params["path"])
+        self.assertEqual(call_args.base_sha, self.valid_params["base_sha"])
+        self.assertEqual(call_args.existing_file_sha, "b" * 40)
+
+if __name__ == "__main__":
+    unittest.main()