feat(opax-mcp): harden branch-resolved Gitea reads and plan transitions

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:

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

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

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):
+    @patch("gitea_handler.resolve_branch_to_commit_sha", new_callable=AsyncMock)
-        with self.assertRaisesRegex(ValueError, "Invalid commit SHA"):
+    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")
+        self.assertEqual(client.stream.call_count, 3)
 
     @patch("gitea_handler.httpx.AsyncClient")
     async def test_get_file_size_limit(self, mock_client):

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 = {
@@ -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()