From cac7dd6e8a3501e4365ce39cb7ef87cbd6ecc053 Mon Sep 17 00:00:00 2001 From: Chris Christiansen Date: Mon, 21 Sep 2026 05:30:06 +0000 Subject: [PATCH] Harden Gitea change proposals --- opax-mcp/gitea_handler.py | 40 +++++++++ opax-mcp/server.py | 6 ++ opax-mcp/test_propose_gitea_change.py | 125 ++++++++++++++++++++++++++ 3 files changed, 171 insertions(+) create mode 100644 opax-mcp/test_propose_gitea_change.py diff --git a/opax-mcp/gitea_handler.py b/opax-mcp/gitea_handler.py index 1885c51..f16b4a9 100644 --- a/opax-mcp/gitea_handler.py +++ b/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, diff --git a/opax-mcp/server.py b/opax-mcp/server.py index 2e43285..ffc9d1e 100644 --- a/opax-mcp/server.py +++ b/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, diff --git a/opax-mcp/test_propose_gitea_change.py b/opax-mcp/test_propose_gitea_change.py new file mode 100644 index 0000000..29bae46 --- /dev/null +++ b/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()