From 56d27684738871ee26985975369a5e4dc24bbda7 Mon Sep 17 00:00:00 2001 From: Chris Christiansen Date: Wed, 23 Sep 2026 01:58:05 +0000 Subject: [PATCH] feat(opax-mcp): harden branch-resolved Gitea reads and plan transitions --- opax-mcp/gitea_handler.py | 74 ++++- opax-mcp/server.py | 78 +++++- opax-mcp/test_gitea_handler.py | 385 +++++++++++++++++++++++++- opax-mcp/test_gitea_helpers.py | 33 ++- opax-mcp/test_propose_gitea_change.py | 285 ++++++++++++++++++- 5 files changed, 832 insertions(+), 23 deletions(-) diff --git a/opax-mcp/gitea_handler.py b/opax-mcp/gitea_handler.py index d83c192..c266b8f 100644 --- a/opax-mcp/gitea_handler.py +++ b/opax-mcp/gitea_handler.py @@ -18,6 +18,10 @@ _BRANCH_RE = re.compile( r"^[A-Za-z0-9][A-Za-z0-9._/-]{0,127}$" ) +_READ_BRANCH_REF_RE = re.compile( + r"^[A-Za-z0-9][A-Za-z0-9._-]*(?:/[A-Za-z0-9][A-Za-z0-9._-]*)*$" +) + MAX_SOURCE_ARCHIVE_BYTES = int( os.environ.get("OPAX_MAX_SOURCE_ARCHIVE_BYTES", "104857600") ) @@ -188,6 +192,34 @@ def validate_path_for_write(path: str) -> str: return path +def _validate_read_branch_ref(ref: str) -> str: + """Validate a safe branch name used only for read resolution.""" + if not isinstance(ref, str) or not ref: + raise ValueError("Invalid git reference.") + + if any(char.isspace() or ord(char) < 32 or ord(char) == 127 for char in ref): + raise ValueError("Invalid git reference.") + + if ( + ref.upper() == "HEAD" + or ref.startswith("refs/") + or ref.startswith("/") + or ref.endswith("/") + or "//" in ref + or "\\" in ref + ): + raise ValueError("Invalid git reference.") + + components = ref.split("/") + if any(component in (".", "..") for component in components): + raise ValueError("Invalid git reference.") + + if not _READ_BRANCH_REF_RE.fullmatch(ref): + raise ValueError("Invalid git reference.") + + return ref + + async def resolve_branch_to_commit_sha( branch_name: str, repo_id: str, @@ -342,10 +374,39 @@ async def handle_get_file_content(p: dict, server_repo_id: str) -> dict: 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}" + requested_ref = p.get("ref") + if not isinstance(requested_ref, str) or not requested_ref: + raise ValueError("Invalid git reference.") + + if _SHA_RE.fullmatch(requested_ref.lower()): + resolved_commit_sha = _validate_commit_sha(requested_ref.lower()) + else: + validated_branch_ref = _validate_read_branch_ref(requested_ref) + try: + resolved_commit_sha = await resolve_branch_to_commit_sha( + branch_name=validated_branch_ref, + repo_id=validated_server_repo, + gitea_url=gitea_url, + ) + except httpx.HTTPStatusError as exc: + if exc.response.status_code == 404: + raise ValueError( + "Unknown or inaccessible branch reference." + ) from exc + logger.warning( + "Gitea branch resolution failed", + extra={"status_code": exc.response.status_code}, + ) + raise ValueError("Repository file is unavailable.") from exc + except Exception: + logger.error("Gitea branch resolution failed unexpectedly") + raise ValueError("Repository file is unavailable.") + + resolved_commit_sha = _validate_commit_sha(resolved_commit_sha) + + url = f"{gitea_url}/api/v1/repos/{validated_server_repo}/contents/{quote(path, safe='')}?ref={resolved_commit_sha}" try: async with httpx.AsyncClient(timeout=15) as c: @@ -398,7 +459,14 @@ 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.") - return {"path": path, "content": text_content, "encoding": "utf-8"} + return { + "repo": validated_server_repo, + "path": path, + "requested_ref": requested_ref, + "resolved_commit_sha": resolved_commit_sha, + "content": text_content, + "encoding": "utf-8", + } async def handle_list_repo_files(p: dict, default_repo: str) -> dict: diff --git a/opax-mcp/server.py b/opax-mcp/server.py index 2fed182..41fffc3 100644 --- a/opax-mcp/server.py +++ b/opax-mcp/server.py @@ -100,13 +100,21 @@ class EncryptedPayloadV1(BaseModel): ciphertext: str +GITEA_CHANGE_PLAN_STATUS_PENDING = "PENDING" +GITEA_CHANGE_PLAN_STATUS_APPROVED = "APPROVED" +GITEA_CHANGE_PLAN_STATUS_APPLYING = "APPLYING" +GITEA_CHANGE_PLAN_STATUS_APPLIED = "APPLIED" +GITEA_CHANGE_PLAN_STATUS_REJECTED = "REJECTED" +GITEA_CHANGE_PLAN_STATUS_EXPIRED = "EXPIRED" + + class GiteaChangePlan(BaseModel): """Immutable admin-approved one-file Gitea change plan.""" plan_id: str = Field( default_factory=lambda: f"gitea-change-{uuid.uuid4().hex}" ) - status: str = "PENDING" + status: str = GITEA_CHANGE_PLAN_STATUS_PENDING created_at: datetime = Field(default_factory=datetime.utcnow) expires_at: datetime = Field( @@ -134,6 +142,38 @@ class GiteaChangePlan(BaseModel): apply_error_message: Optional[str] = None +def is_gitea_change_plan_expired(plan: GiteaChangePlan) -> bool: + """Checks if a Gitea change plan has expired.""" + now_utc = datetime.now(timezone.utc) + expires_at = plan.expires_at + + if expires_at.tzinfo is None: + expires_at = expires_at.replace(tzinfo=timezone.utc) + + return now_utc >= expires_at + + +_ALLOWED_GITEA_PLAN_TRANSITIONS = { + GITEA_CHANGE_PLAN_STATUS_PENDING: { + GITEA_CHANGE_PLAN_STATUS_APPROVED, + GITEA_CHANGE_PLAN_STATUS_REJECTED, + GITEA_CHANGE_PLAN_STATUS_EXPIRED, + }, + GITEA_CHANGE_PLAN_STATUS_APPROVED: { + GITEA_CHANGE_PLAN_STATUS_APPLYING, + GITEA_CHANGE_PLAN_STATUS_EXPIRED, + }, + GITEA_CHANGE_PLAN_STATUS_APPLYING: {GITEA_CHANGE_PLAN_STATUS_APPLIED}, +} + +def is_valid_gitea_change_plan_status_transition( + current_status: str, + next_status: str, +) -> bool: + """Checks if a Gitea change plan status transition is allowed.""" + return next_status in _ALLOWED_GITEA_PLAN_TRANSITIONS.get(current_status, set()) + + async def create_gitea_change_plan(plan: GiteaChangePlan) -> dict: """Persist a pending one-file Gitea change plan in Firestore.""" from google.cloud import firestore @@ -230,6 +270,42 @@ async def get_gitea_change_plan( return GiteaChangePlan(**document.to_dict()) +async def transition_gitea_change_plan_status( + plan_id: str, + expected_status: str, + next_status: str, + updates: Optional[dict] = None, +) -> GiteaChangePlan: + if not is_valid_gitea_change_plan_status_transition(expected_status, next_status): + raise ValueError("Invalid Gitea change plan status transition") + if updates and "status" in updates: + raise ValueError("Gitea change plan updates cannot include status") + + from google.cloud import firestore + + db = firestore.AsyncClient(project=GOOGLE_CLOUD_PROJECT) + plan_ref = db.collection("gitea_change_plans").document(plan_id) + transaction = db.transaction() + + @firestore.async_transactional + async def transactional_update(transaction): + snapshot = await plan_ref.get(transaction=transaction) + if not snapshot.exists: + raise ValueError("Gitea change plan not found") + if snapshot.get("status") != expected_status: + raise ValueError("Gitea change plan status changed") + + update_data = dict(updates or {}) + update_data["status"] = next_status + transaction.update(plan_ref, update_data) + + await transactional_update(transaction) + updated_plan = await get_gitea_change_plan(plan_id) + if updated_plan is None: + raise ValueError("Gitea change plan not found") + return updated_plan + + def _validate_admin_gitea_path(path: str) -> None: """Validate a repository-relative admin file path.""" if not isinstance(path, str) or not path or path.isspace(): diff --git a/opax-mcp/test_gitea_handler.py b/opax-mcp/test_gitea_handler.py index bb4d496..b609cc5 100644 --- a/opax-mcp/test_gitea_handler.py +++ b/opax-mcp/test_gitea_handler.py @@ -3,12 +3,68 @@ from pathlib import Path import sys from unittest.mock import AsyncMock, MagicMock, patch import os +import base64 +import json 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 +from gitea_handler import handle_list_repo_files, _validate_read_branch_ref, handle_get_file_content + + +class TestValidateReadBranchRef(unittest.TestCase): + def test_accepted_refs(self): + accepted = [ + "main", + "feat/opax-domain-decouple", + "docs/mcp-diagnostics", + "fix/example", + "chore/example", + "release/1.2.3", + ] + for ref in accepted: + with self.subTest(ref=ref): + self.assertEqual(_validate_read_branch_ref(ref), ref) + + def test_rejected_refs(self): + rejected = [ + "", + " main", + "main ", + "feature branch", + "main\tbranch", + "main\nbranch", + "main\x00branch", + "main\x7fbranch", + "HEAD", + "head", + "refs/heads/main", + "/main", + "main/", + "main//next", + "main/./next", + "main/../next", + "main\\next", + "main?x=1", + "main#fragment", + "https://example.invalid", + "user@host", + "feature*", + "feature~1", + "feature^", + "feature{a}", + "feature;cmd", + "feature|cmd", + "feature%2Fbranch", + "feature'quote", + 'feature"quote', + "feature[abc]", + ] + for ref in rejected: + with self.subTest(ref=ref): + with self.assertRaisesRegex(ValueError, r"^Invalid git reference\.$"): + _validate_read_branch_ref(ref) class TestGiteaHandler(unittest.TestCase): @@ -340,5 +396,332 @@ class TestGiteaHandler(unittest.TestCase): import asyncio asyncio.run(run_test()) + +class TestHandleGetFileContent(unittest.TestCase): + def setUp(self): + self.original_gitea_url = os.environ.get("GITEA_URL") + self.original_gitea_token = os.environ.get("GITEA_TOKEN") + os.environ["GITEA_URL"] = "https://gitea.example.com" + os.environ["GITEA_TOKEN"] = "test-token" + self.server_repo_id = "chris/OSVauco" + + def tearDown(self): + if self.original_gitea_url is None: + os.environ.pop("GITEA_URL", None) + else: + os.environ["GITEA_URL"] = self.original_gitea_url + if self.original_gitea_token is None: + os.environ.pop("GITEA_TOKEN", None) + else: + os.environ["GITEA_TOKEN"] = self.original_gitea_token + + def _mock_contents_client(self, text): + body = json.dumps({ + "content": base64.b64encode(text.encode("utf-8")).decode("ascii"), + "encoding": "base64", + }).encode("utf-8") + + async def chunks(): + yield body + + response = MagicMock() + response.headers = {"content-length": str(len(body))} + response.aiter_bytes = chunks + stream = MagicMock() + stream.__aenter__.return_value = response + client = MagicMock() + client.stream.return_value = stream + async_client = MagicMock() + async_client.__aenter__.return_value = client + return async_client, client + + @patch("gitea_handler.resolve_branch_to_commit_sha", new_callable=AsyncMock) + @patch("httpx.AsyncClient") + def test_direct_lowercase_sha_success( + self, mock_async_client_constructor, mock_resolver + ): + import asyncio + + sha = "a" * 40 + mock_async_client, client = self._mock_contents_client("file content") + mock_async_client_constructor.return_value = mock_async_client + + async def run_test(): + return await handle_get_file_content( + { + "repo": self.server_repo_id, + "path": "README.md", + "ref": sha, + }, + self.server_repo_id, + ) + + result = asyncio.run(run_test()) + + mock_resolver.assert_not_called() + client.stream.assert_called_once() + request_url = client.stream.call_args.args[1] + self.assertIn(f"ref={sha}", request_url) + self.assertEqual(result["requested_ref"], sha) + self.assertEqual(result["resolved_commit_sha"], sha) + self.assertEqual(result["content"], "file content") + self.assertEqual(result["encoding"], "utf-8") + + @patch("gitea_handler.resolve_branch_to_commit_sha", new_callable=AsyncMock) + @patch("httpx.AsyncClient") + def test_direct_uppercase_sha_normalizes_for_request( + self, mock_async_client_constructor, mock_resolver + ): + import asyncio + + requested_sha = "A" * 40 + resolved_sha = requested_sha.lower() + mock_async_client, client = self._mock_contents_client("file content") + mock_async_client_constructor.return_value = mock_async_client + + async def run_test(): + return await handle_get_file_content( + { + "repo": self.server_repo_id, + "path": "README.md", + "ref": requested_sha, + }, + self.server_repo_id, + ) + + result = asyncio.run(run_test()) + + mock_resolver.assert_not_called() + request_url = client.stream.call_args.args[1] + self.assertIn(f"ref={resolved_sha}", request_url) + self.assertNotIn(f"ref={requested_sha}", request_url) + self.assertEqual(result["requested_ref"], requested_sha) + self.assertEqual(result["resolved_commit_sha"], resolved_sha) + + @patch("gitea_handler.resolve_branch_to_commit_sha", new_callable=AsyncMock) + @patch("httpx.AsyncClient") + def test_branch_resolves_to_sha_before_contents_request( + self, mock_async_client_constructor, mock_resolver + ): + import asyncio + + resolved_sha = "b" * 40 + mock_resolver.return_value = resolved_sha + mock_async_client, client = self._mock_contents_client("file content") + mock_async_client_constructor.return_value = mock_async_client + + async def run_test(): + return await handle_get_file_content( + { + "repo": self.server_repo_id, + "path": "README.md", + "ref": "main", + }, + self.server_repo_id, + ) + + result = asyncio.run(run_test()) + + mock_resolver.assert_awaited_once_with( + branch_name="main", + repo_id=self.server_repo_id, + gitea_url="https://gitea.example.com", + ) + request_url = client.stream.call_args.args[1] + self.assertIn(f"ref={resolved_sha}", request_url) + self.assertNotIn("ref=main", request_url) + self.assertEqual(result["repo"], self.server_repo_id) + self.assertEqual(result["path"], "README.md") + self.assertEqual(result["requested_ref"], "main") + self.assertEqual(result["resolved_commit_sha"], resolved_sha) + self.assertEqual(result["content"], "file content") + self.assertEqual(result["encoding"], "utf-8") + + @patch("gitea_handler.resolve_branch_to_commit_sha", new_callable=AsyncMock) + @patch("httpx.AsyncClient") + def test_unresolved_branch_raises_sanitized_error( + self, mock_async_client_constructor, mock_resolver + ): + import asyncio + + mock_resolver.side_effect = httpx.HTTPStatusError( + "Not Found", + request=MagicMock(), + response=MagicMock(status_code=404), + ) + mock_async_client, client = self._mock_contents_client("file content") + mock_async_client_constructor.return_value = mock_async_client + + async def run_test(): + with self.assertRaisesRegex( + ValueError, r"^Unknown or inaccessible branch reference\.$" + ): + await handle_get_file_content( + { + "repo": self.server_repo_id, + "path": "README.md", + "ref": "missing-branch", + }, + self.server_repo_id, + ) + + asyncio.run(run_test()) + + client.stream.assert_not_called() + + @patch("gitea_handler.resolve_branch_to_commit_sha", new_callable=AsyncMock) + @patch("httpx.AsyncClient") + def test_resolver_non_404_http_error_is_sanitized( + self, mock_async_client_constructor, mock_resolver + ): + import asyncio + + mock_resolver.side_effect = httpx.HTTPStatusError( + "Server Error", + request=MagicMock(), + response=MagicMock(status_code=500), + ) + mock_async_client, client = self._mock_contents_client("file content") + mock_async_client_constructor.return_value = mock_async_client + + async def run_test(): + with self.assertRaisesRegex( + ValueError, r"^Repository file is unavailable\.$" + ): + await handle_get_file_content( + { + "repo": self.server_repo_id, + "path": "README.md", + "ref": "main", + }, + self.server_repo_id, + ) + + asyncio.run(run_test()) + + client.stream.assert_not_called() + + @patch("gitea_handler.resolve_branch_to_commit_sha", new_callable=AsyncMock) + @patch("httpx.AsyncClient") + def test_resolver_unexpected_exception_is_sanitized( + self, mock_async_client_constructor, mock_resolver + ): + import asyncio + + mock_resolver.side_effect = RuntimeError( + "upstream internal details must not reach the caller" + ) + mock_async_client, client = self._mock_contents_client("file content") + mock_async_client_constructor.return_value = mock_async_client + + async def run_test(): + with self.assertRaisesRegex( + ValueError, r"^Repository file is unavailable\.$" + ): + await handle_get_file_content( + { + "repo": self.server_repo_id, + "path": "README.md", + "ref": "main", + }, + self.server_repo_id, + ) + + asyncio.run(run_test()) + + client.stream.assert_not_called() + + @patch("gitea_handler.resolve_branch_to_commit_sha", new_callable=AsyncMock) + @patch("httpx.AsyncClient") + def test_resolver_returns_malformed_sha_is_rejected( + self, mock_async_client_constructor, mock_resolver + ): + import asyncio + + mock_resolver.return_value = "not-a-40-character-commit-sha" + mock_async_client, client = self._mock_contents_client("file content") + mock_async_client_constructor.return_value = mock_async_client + + async def run_test(): + with self.assertRaisesRegex(ValueError, r"^Invalid commit SHA$"): + await handle_get_file_content( + { + "repo": self.server_repo_id, + "path": "README.md", + "ref": "main", + }, + self.server_repo_id, + ) + + asyncio.run(run_test()) + + client.stream.assert_not_called() + + @patch("gitea_handler.resolve_branch_to_commit_sha", new_callable=AsyncMock) + @patch("httpx.AsyncClient") + def test_invalid_branch_ref_is_rejected( + self, mock_async_client_constructor, mock_resolver + ): + import asyncio + + mock_async_client, client = self._mock_contents_client("file content") + mock_async_client_constructor.return_value = mock_async_client + + async def run_test(): + with self.assertRaisesRegex(ValueError, r"^Invalid git reference\.$"): + await handle_get_file_content( + { + "repo": self.server_repo_id, + "path": "README.md", + "ref": "main/../secret", + }, + self.server_repo_id, + ) + + asyncio.run(run_test()) + + mock_resolver.assert_not_called() + client.stream.assert_not_called() + + @patch("gitea_handler.resolve_branch_to_commit_sha", new_callable=AsyncMock) + @patch("httpx.AsyncClient") + def test_contents_api_404_is_sanitized( + self, mock_async_client_constructor, mock_resolver + ): + import asyncio + + resolved_sha = "c" * 40 + mock_resolver.return_value = resolved_sha + mock_async_client, client = self._mock_contents_client("file content") + + response_mock = client.stream.return_value.__aenter__.return_value + response_mock.raise_for_status.side_effect = httpx.HTTPStatusError( + "Not Found", request=MagicMock(), response=MagicMock(status_code=404) + ) + mock_async_client_constructor.return_value = mock_async_client + + async def run_test(): + with self.assertRaisesRegex( + ValueError, r"^Repository file is unavailable\.$" + ): + await handle_get_file_content( + { + "repo": self.server_repo_id, + "path": "README.md", + "ref": "main", + }, + self.server_repo_id, + ) + + asyncio.run(run_test()) + + mock_resolver.assert_awaited_once_with( + branch_name="main", + repo_id=self.server_repo_id, + gitea_url="https://gitea.example.com", + ) + client.stream.assert_called_once() + + if __name__ == '__main__': unittest.main() diff --git a/opax-mcp/test_gitea_helpers.py b/opax-mcp/test_gitea_helpers.py index ef41559..e984609 100644 --- a/opax-mcp/test_gitea_helpers.py +++ b/opax-mcp/test_gitea_helpers.py @@ -297,11 +297,19 @@ class TestGetFileHardening(unittest.IsolatedAsyncioTestCase): ) client.stream.assert_not_called() - async def test_get_file_ref_validation(self): - with self.assertRaisesRegex(ValueError, "Invalid commit SHA"): + @patch("gitea_handler.resolve_branch_to_commit_sha", new_callable=AsyncMock) + async def test_get_file_ref_validation(self, mock_resolver): + mock_resolver.side_effect = httpx.ConnectError("Network error") + with self.assertRaisesRegex(ValueError, r"^Repository file is unavailable\.$"): await gitea_handler.handle_get_file_content({"path": "README.md", "ref": "main"}, "chris/OSVauco") - with self.assertRaisesRegex(ValueError, "Invalid commit SHA"): + mock_resolver.assert_awaited_once() + self.assertEqual(mock_resolver.call_args.kwargs['branch_name'], "main") + self.assertEqual(mock_resolver.call_args.kwargs['repo_id'], "chris/OSVauco") + + with self.assertRaisesRegex(ValueError, r"^Invalid git reference\.$"): await gitea_handler.handle_get_file_content({"path": "README.md", "ref": None}, "chris/OSVauco") + # Verify the resolver was not called a second time for the invalid ref + self.assertEqual(mock_resolver.await_count, 1) async def test_gitea_repo_validation(self): with self.assertRaisesRegex(ValueError, "Invalid configured Gitea repository ID"): @@ -309,17 +317,24 @@ class TestGetFileHardening(unittest.IsolatedAsyncioTestCase): with self.assertRaisesRegex(ValueError, "Invalid configured Gitea repository ID"): await gitea_handler.handle_get_file_content({"path": "README.md", "ref": self.valid_sha}, "") + async def test_get_file_path_not_allowed(self): + for path in ["../secrets.txt", "/etc/passwd"]: + with self.subTest(path=path): + with self.assertRaisesRegex(ValueError, r"^PATH_NOT_ALLOWED$"): + await gitea_handler.handle_get_file_content({"path": path, "ref": self.valid_sha}, "chris/OSVauco") + + async def test_get_file_secret_path_denied(self): + with self.assertRaisesRegex(ValueError, r"^SECRET_PATH_DENIED$"): + await gitea_handler.handle_get_file_content({"path": "docs/.env", "ref": self.valid_sha}, "chris/OSVauco") + @patch("gitea_handler.httpx.AsyncClient") - async def test_get_file_path_validation(self, mock_client): + async def test_get_file_valid_paths(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") + await gitea_handler.handle_get_file_content({"path": path, "ref": self.valid_sha}, "chris/OSVauco") + self.assertEqual(client.stream.call_count, 3) @patch("gitea_handler.httpx.AsyncClient") async def test_get_file_size_limit(self, mock_client): diff --git a/opax-mcp/test_propose_gitea_change.py b/opax-mcp/test_propose_gitea_change.py index 0aaa301..e371fbb 100644 --- a/opax-mcp/test_propose_gitea_change.py +++ b/opax-mcp/test_propose_gitea_change.py @@ -18,6 +18,26 @@ KMS_KEY_NAME = "projects/p/locations/l/keyRings/k/cryptoKeys/k" class TestProposeGiteaChange(unittest.IsolatedAsyncioTestCase): + @patch('server.handle_get_file_content', new_callable=AsyncMock) + async def test_get_file_wrapper_enforces_server_repo(self, mock_handler): + sentinel_response = { + "requested_ref": "main", + "resolved_commit_sha": "a" * 40, + "content": "file content" + } + mock_handler.return_value = sentinel_response + + params = { + "path": "README.md", + "ref": "main", + "repo": "untrusted/repo", + } + result = await server.get_file(params) + + mock_handler.assert_awaited_once_with(params, server.GITEA_REPO) + self.assertIs(result, sentinel_response) + + def setUp(self): """Set up valid parameters for tests.""" self.valid_params = { @@ -39,7 +59,7 @@ class TestProposeGiteaChange(unittest.IsolatedAsyncioTestCase): 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() @@ -85,10 +105,10 @@ class TestProposeGiteaChange(unittest.IsolatedAsyncioTestCase): @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() @@ -99,10 +119,10 @@ class TestProposeGiteaChange(unittest.IsolatedAsyncioTestCase): 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() @@ -118,7 +138,7 @@ class TestProposeGiteaChange(unittest.IsolatedAsyncioTestCase): mock_kms_constructor.return_value.__aexit__ = AsyncMock(return_value=False) mock_resolve_sha.return_value = self.valid_params["base_sha"] - + with patch.dict(os.environ, {"GITEA_PLAN_KMS_KEY_NAME": KMS_KEY_NAME}): result = await server.propose_gitea_change(self.valid_params) @@ -127,13 +147,13 @@ class TestProposeGiteaChange(unittest.IsolatedAsyncioTestCase): self.assertEqual(result["branch"], self.valid_params["branch"]) self.assertEqual(result["path"], self.valid_params["path"]) self.assertIn("approval_subject_hash", result) - + expected_hash = hashlib.sha256(self.valid_params["new_content"].encode("utf-8")).hexdigest() self.assertEqual(result["content_hash"], expected_hash) - + mock_kms_client.encrypt.assert_awaited_once() 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) @@ -344,5 +364,252 @@ class TestProposeGiteaChange(unittest.IsolatedAsyncioTestCase): self.assertNotEqual(base_hash, server._calculate_approval_subject_hash(modified_plan_cipher)) +class TestIsGiteaChangePlanExpired(unittest.TestCase): + def _create_test_plan(self, expires_at): + return server.GiteaChangePlan( + repo="o/r", + branch="b", + path="p", + base_sha="a" * 40, + content_hash="c" * 64, + commit_message="m", + unified_diff="d", + expires_at=expires_at, + ) + + def test_future_aware_is_not_expired(self): + from datetime import datetime, timedelta, timezone + future_time = datetime.now(timezone.utc) + timedelta(days=1) + plan = self._create_test_plan(future_time) + self.assertFalse(server.is_gitea_change_plan_expired(plan)) + + def test_past_aware_is_expired(self): + from datetime import datetime, timedelta, timezone + past_time = datetime.now(timezone.utc) - timedelta(days=1) + plan = self._create_test_plan(past_time) + self.assertTrue(server.is_gitea_change_plan_expired(plan)) + + def test_past_naive_is_expired(self): + from datetime import datetime, timedelta + past_time_naive = datetime.utcnow() - timedelta(days=1) + plan = self._create_test_plan(past_time_naive) + self.assertTrue(server.is_gitea_change_plan_expired(plan)) + + def test_default_status_is_pending_constant(self): + from datetime import datetime, timedelta, timezone + future_time = datetime.now(timezone.utc) + timedelta(days=1) + plan = self._create_test_plan(future_time) + self.assertEqual( + plan.status, + server.GITEA_CHANGE_PLAN_STATUS_PENDING, + ) + self.assertEqual(server.GITEA_CHANGE_PLAN_STATUS_PENDING, "PENDING") + self.assertEqual(server.GITEA_CHANGE_PLAN_STATUS_APPROVED, "APPROVED") + self.assertEqual(server.GITEA_CHANGE_PLAN_STATUS_APPLYING, "APPLYING") + self.assertEqual(server.GITEA_CHANGE_PLAN_STATUS_APPLIED, "APPLIED") + self.assertEqual(server.GITEA_CHANGE_PLAN_STATUS_REJECTED, "REJECTED") + self.assertEqual(server.GITEA_CHANGE_PLAN_STATUS_EXPIRED, "EXPIRED") + + +class TestGiteaChangePlanStatusTransitions(unittest.TestCase): + def test_allowed_transitions(self): + self.assertTrue( + server.is_valid_gitea_change_plan_status_transition( + server.GITEA_CHANGE_PLAN_STATUS_PENDING, + server.GITEA_CHANGE_PLAN_STATUS_APPROVED, + ) + ) + self.assertTrue( + server.is_valid_gitea_change_plan_status_transition( + server.GITEA_CHANGE_PLAN_STATUS_PENDING, + server.GITEA_CHANGE_PLAN_STATUS_REJECTED, + ) + ) + self.assertTrue( + server.is_valid_gitea_change_plan_status_transition( + server.GITEA_CHANGE_PLAN_STATUS_PENDING, + server.GITEA_CHANGE_PLAN_STATUS_EXPIRED, + ) + ) + self.assertTrue( + server.is_valid_gitea_change_plan_status_transition( + server.GITEA_CHANGE_PLAN_STATUS_APPROVED, + server.GITEA_CHANGE_PLAN_STATUS_APPLYING, + ) + ) + self.assertTrue( + server.is_valid_gitea_change_plan_status_transition( + server.GITEA_CHANGE_PLAN_STATUS_APPROVED, + server.GITEA_CHANGE_PLAN_STATUS_EXPIRED, + ) + ) + self.assertTrue( + server.is_valid_gitea_change_plan_status_transition( + server.GITEA_CHANGE_PLAN_STATUS_APPLYING, + server.GITEA_CHANGE_PLAN_STATUS_APPLIED, + ) + ) + + def test_disallowed_transitions(self): + self.assertFalse( + server.is_valid_gitea_change_plan_status_transition( + server.GITEA_CHANGE_PLAN_STATUS_PENDING, + server.GITEA_CHANGE_PLAN_STATUS_PENDING, + ) + ) + self.assertFalse( + server.is_valid_gitea_change_plan_status_transition( + server.GITEA_CHANGE_PLAN_STATUS_APPROVED, + server.GITEA_CHANGE_PLAN_STATUS_APPLIED, + ) + ) + self.assertFalse( + server.is_valid_gitea_change_plan_status_transition( + server.GITEA_CHANGE_PLAN_STATUS_APPLYING, + server.GITEA_CHANGE_PLAN_STATUS_APPROVED, + ) + ) + self.assertFalse( + server.is_valid_gitea_change_plan_status_transition( + server.GITEA_CHANGE_PLAN_STATUS_APPLIED, + server.GITEA_CHANGE_PLAN_STATUS_APPLYING, + ) + ) + self.assertFalse( + server.is_valid_gitea_change_plan_status_transition( + server.GITEA_CHANGE_PLAN_STATUS_REJECTED, + server.GITEA_CHANGE_PLAN_STATUS_APPROVED, + ) + ) + self.assertFalse( + server.is_valid_gitea_change_plan_status_transition( + server.GITEA_CHANGE_PLAN_STATUS_EXPIRED, + server.GITEA_CHANGE_PLAN_STATUS_APPROVED, + ) + ) + self.assertFalse( + server.is_valid_gitea_change_plan_status_transition( + "UNKNOWN", server.GITEA_CHANGE_PLAN_STATUS_PENDING + ) + ) + self.assertFalse( + server.is_valid_gitea_change_plan_status_transition( + server.GITEA_CHANGE_PLAN_STATUS_PENDING, "UNKNOWN" + ) + ) + + +class TestTransitionGiteaChangePlanStatus(unittest.IsolatedAsyncioTestCase): + @patch("google.cloud.firestore.AsyncClient") + async def test_rejects_invalid_transition_before_firestore(self, mock_db_client): + with self.assertRaisesRegex( + ValueError, r"^Invalid Gitea change plan status transition$" + ): + await server.transition_gitea_change_plan_status( + plan_id="plan-123", + expected_status=server.GITEA_CHANGE_PLAN_STATUS_PENDING, + next_status=server.GITEA_CHANGE_PLAN_STATUS_APPLIED, + ) + mock_db_client.assert_not_called() + + @patch("google.cloud.firestore.AsyncClient") + async def test_rejects_status_in_updates_before_firestore(self, mock_db_client): + with self.assertRaisesRegex( + ValueError, r"^Gitea change plan updates cannot include status$" + ): + await server.transition_gitea_change_plan_status( + plan_id="plan-123", + expected_status=server.GITEA_CHANGE_PLAN_STATUS_PENDING, + next_status=server.GITEA_CHANGE_PLAN_STATUS_APPROVED, + updates={"status": server.GITEA_CHANGE_PLAN_STATUS_REJECTED}, + ) + mock_db_client.assert_not_called() + + @patch("server.get_gitea_change_plan", new_callable=AsyncMock) + @patch("google.cloud.firestore.async_transactional") + @patch("google.cloud.firestore.AsyncClient") + async def test_transitions_matching_status_and_returns_refetched_plan( + self, mock_db_client, mock_transactional, mock_get_plan + ): + def transactional_side_effect(callback): + async def wrapped(transaction): + return await callback(transaction) + return wrapped + mock_transactional.side_effect = transactional_side_effect + + mock_transaction = MagicMock() + mock_transaction.update = MagicMock() + mock_db_client.return_value.transaction.return_value = mock_transaction + + snapshot = MagicMock() + snapshot.exists = True + snapshot.get.return_value = server.GITEA_CHANGE_PLAN_STATUS_PENDING + + mock_plan_ref = MagicMock() + mock_plan_ref.get = AsyncMock(return_value=snapshot) + mock_db_client.return_value.collection.return_value.document.return_value = mock_plan_ref + + final_plan = server.GiteaChangePlan( + status=server.GITEA_CHANGE_PLAN_STATUS_APPROVED, + repo="o/r", branch="b", path="p", base_sha="a"*40, + content_hash="c"*64, commit_message="m", unified_diff="d", + approved_by="admin@example.com" + ) + mock_get_plan.return_value = final_plan + + updates = {"approved_by": "admin@example.com"} + result = await server.transition_gitea_change_plan_status( + plan_id="plan-123", + expected_status=server.GITEA_CHANGE_PLAN_STATUS_PENDING, + next_status=server.GITEA_CHANGE_PLAN_STATUS_APPROVED, + updates=updates, + ) + + mock_transaction.update.assert_called_once_with( + mock_plan_ref, + { + "approved_by": "admin@example.com", + "status": server.GITEA_CHANGE_PLAN_STATUS_APPROVED, + }, + ) + self.assertEqual(updates, {"approved_by": "admin@example.com"}) + mock_get_plan.assert_awaited_once_with("plan-123") + self.assertIs(result, final_plan) + + @patch("server.get_gitea_change_plan", new_callable=AsyncMock) + @patch("google.cloud.firestore.async_transactional") + @patch("google.cloud.firestore.AsyncClient") + async def test_rejects_stale_status_inside_transaction( + self, mock_db_client, mock_transactional, mock_get_plan + ): + def transactional_side_effect(callback): + async def wrapped(transaction): + return await callback(transaction) + return wrapped + mock_transactional.side_effect = transactional_side_effect + + mock_transaction = MagicMock() + mock_transaction.update = MagicMock() + mock_db_client.return_value.transaction.return_value = mock_transaction + + snapshot = MagicMock() + snapshot.exists = True + snapshot.get.return_value = server.GITEA_CHANGE_PLAN_STATUS_APPROVED + + mock_plan_ref = MagicMock() + mock_plan_ref.get = AsyncMock(return_value=snapshot) + mock_db_client.return_value.collection.return_value.document.return_value = mock_plan_ref + + with self.assertRaisesRegex(ValueError, r"^Gitea change plan status changed$"): + await server.transition_gitea_change_plan_status( + plan_id="plan-123", + expected_status=server.GITEA_CHANGE_PLAN_STATUS_PENDING, + next_status=server.GITEA_CHANGE_PLAN_STATUS_APPROVED, + ) + + mock_transaction.update.assert_not_called() + mock_get_plan.assert_not_awaited() + + if __name__ == "__main__": unittest.main()