Access-check page in chat manager.

guerler flagged /api/chat letting any user read any page_id. Add
ChatManager.get_accessible_page using the standard
base.security_check pattern; route create_page_chat,
get_page_chat_history, and the query() page-context lookup through
it. API 403 tests for both POST /api/chat and
GET /api/chat/page/{id}/history.

Author: jmchilton

lib/galaxy/managers/chat.py

@@ -20,15 +20,18 @@
     NoResultFound,
 )
 
+from galaxy import exceptions
 from galaxy.exceptions import (
     InconsistentDatabase,
     InternalServerError,
     RequestParameterInvalidException,
 )
+from galaxy.managers import base
 from galaxy.managers.context import ProvidesUserContext
 from galaxy.model import (
     ChatExchange,
     ChatExchangeMessage,
+    Page,
 )
 from galaxy.util import unicodify
 
@@ -60,6 +63,13 @@ def create(self, trans: ProvidesUserContext, job_id: Optional[int], message: str
         trans.sa_session.commit()
         return chat_exchange
 
+    def get_accessible_page(self, trans: ProvidesUserContext, page_id: int) -> Page:
+        """Return a Page the current user is allowed to read, or raise."""
+        page = trans.sa_session.get(Page, page_id)
+        if not page:
+            raise exceptions.ObjectNotFound("Page not found")
+        return base.security_check(trans, page, check_ownership=False, check_accessible=True)
+
     def create_page_chat(
         self,
         trans: ProvidesUserContext,
@@ -71,6 +81,7 @@ def create_page_chat(
         """Create a chat exchange scoped to a page."""
         import json
 
+        self.get_accessible_page(trans, page_id)
         chat_exchange = ChatExchange(user=trans.user, page_id=page_id)
 
         conversation_data: dict[str, Any]
@@ -96,6 +107,7 @@ def create_page_chat(
 
     def get_page_chat_history(self, trans: ProvidesUserContext, page_id: int, limit: int = 50) -> list[ChatExchange]:
         """Get chat exchanges scoped to a page, ordered most-recent first."""
+        self.get_accessible_page(trans, page_id)
         try:
             stmt = (
                 select(ChatExchange)

lib/galaxy/webapps/galaxy/api/chat.py

@@ -172,8 +172,12 @@ async def query(
 
         # Check for page scope and populate agent context
         page_id = None
+        page_obj = None
         if payload is not None and hasattr(payload, "page_id") and payload.page_id:
             page_id = payload.page_id
+            # Access-check outside the try below so 403s propagate instead of
+            # being masked as 500.
+            page_obj = self.chat_manager.get_accessible_page(trans, page_id)
 
         # Use new agent system if available, otherwise fallback to legacy
         try:
@@ -185,9 +189,6 @@ async def query(
                 # Content is exported (IDs encoded) so the agent sees the same
                 # text the editor has — hashes and proposals match the client.
                 if page_id:
-                    from galaxy.model import Page
-
-                    page_obj = trans.sa_session.get(Page, page_id)
                     if page_obj:
                         full_context["history_id"] = page_obj.history_id
                         # Fallback to session history for standalone pages

lib/galaxy_test/api/test_pages_history_attached.py

@@ -334,3 +334,17 @@ def test_page_chat_delete_exchange(self):
         self._assert_status_code_is(delete_response, 200)
         history = self.dataset_populator.get_page_chat_history(page["id"])
         assert len(history) == 0
+
+    def test_page_chat_403_on_unowned_page(self):
+        history_id = self.dataset_populator.new_history()
+        page = self.dataset_populator.new_history_page(history_id, content="# Private")
+        with self._different_user():
+            response = self.dataset_populator.send_page_chat_raw(page["id"], "Leak this")
+            self._assert_status_code_is(response, 403)
+
+    def test_page_chat_history_403_on_unowned_page(self):
+        history_id = self.dataset_populator.new_history()
+        page = self.dataset_populator.new_history_page(history_id, content="# Private")
+        with self._different_user():
+            response = self.dataset_populator.get_page_chat_history_raw(page["id"])
+            self._assert_status_code_is(response, 403)

lib/galaxy_test/base/populators.py

@@ -2083,22 +2083,34 @@ def revert_page_revision(self, page_id: str, revision_id: str) -> dict[str, Any]
         api_asserts.assert_status_code_is(response, 200)
         return response.json()
 
-    def send_page_chat(
+    def send_page_chat_raw(
         self,
         page_id: str,
         query: str,
         agent_type: str = "page_assistant",
         exchange_id: Optional[str] = None,
-    ) -> dict[str, Any]:
+    ):
         payload: dict[str, Any] = {"query": query, "page_id": page_id}
         if exchange_id is not None:
             payload["exchange_id"] = exchange_id
-        response = self._post(f"chat?agent_type={agent_type}", payload, json=True)
+        return self._post(f"chat?agent_type={agent_type}", payload, json=True)
+
+    def send_page_chat(
+        self,
+        page_id: str,
+        query: str,
+        agent_type: str = "page_assistant",
+        exchange_id: Optional[str] = None,
+    ) -> dict[str, Any]:
+        response = self.send_page_chat_raw(page_id, query, agent_type=agent_type, exchange_id=exchange_id)
         api_asserts.assert_status_code_is(response, 200)
         return response.json()
 
+    def get_page_chat_history_raw(self, page_id: str):
+        return self._get(f"chat/page/{page_id}/history")
+
     def get_page_chat_history(self, page_id: str) -> list[dict[str, Any]]:
-        response = self._get(f"chat/page/{page_id}/history")
+        response = self.get_page_chat_history_raw(page_id)
         api_asserts.assert_status_code_is(response, 200)
         return response.json()