Fix: enforce editor-only access to chat message logs (#25936)

This commit is contained in:
-LAN- 2025-09-18 21:59:51 +08:00 • committed by GitHub
commit b2d8a7eaf1
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 108 additions and 0 deletions

View file

@ -62,6 +62,9 @@ class ChatMessageListApi(Resource):
@account_initialization_required @account_initialization_required
@marshal_with(message_infinite_scroll_pagination_fields) @marshal_with(message_infinite_scroll_pagination_fields)
def get(self, app_model): def get(self, app_model):
if not isinstance(current_user, Account) or not current_user.has_edit_permission:
raise Forbidden()
parser = reqparse.RequestParser() parser = reqparse.RequestParser()
parser.add_argument("conversation_id", required=True, type=uuid_value, location="args") parser.add_argument("conversation_id", required=True, type=uuid_value, location="args")
parser.add_argument("first_id", type=uuid_value, location="args") parser.add_argument("first_id", type=uuid_value, location="args")

View file

@ -1,12 +1,14 @@
"""Integration tests for ChatMessageApi permission verification.""" """Integration tests for ChatMessageApi permission verification."""
import uuid import uuid
from types import SimpleNamespace
from unittest import mock from unittest import mock
import pytest import pytest
from flask.testing import FlaskClient from flask.testing import FlaskClient
from controllers.console.app import completion as completion_api from controllers.console.app import completion as completion_api
from controllers.console.app import message as message_api
from controllers.console.app import wraps from controllers.console.app import wraps
from libs.datetime_utils import naive_utc_now from libs.datetime_utils import naive_utc_now
from models import Account, App, Tenant from models import Account, App, Tenant
@ -99,3 +101,106 @@ class TestChatMessageApiPermissions:
) )
assert response.status_code == status assert response.status_code == status
@pytest.mark.parametrize(
("role", "status"),
[
(TenantAccountRole.OWNER, 200),
(TenantAccountRole.ADMIN, 200),
(TenantAccountRole.EDITOR, 200),
(TenantAccountRole.NORMAL, 403),
(TenantAccountRole.DATASET_OPERATOR, 403),
],
)
def test_get_requires_edit_permission(
self,
test_client: FlaskClient,
auth_header,
monkeypatch,
mock_app_model,
mock_account,
role: TenantAccountRole,
status: int,
):
"""Ensure GET chat-messages endpoint enforces edit permissions."""
mock_load_app_model = mock.Mock(return_value=mock_app_model)
monkeypatch.setattr(wraps, "_load_app_model", mock_load_app_model)
conversation_id = uuid.uuid4()
created_at = naive_utc_now()
mock_conversation = SimpleNamespace(id=str(conversation_id), app_id=str(mock_app_model.id))
mock_message = SimpleNamespace(
id=str(uuid.uuid4()),
conversation_id=str(conversation_id),
inputs=[],
query="hello",
message=[{"text": "hello"}],
message_tokens=0,
re_sign_file_url_answer="",
answer_tokens=0,
provider_response_latency=0.0,
from_source="console",
from_end_user_id=None,
from_account_id=mock_account.id,
feedbacks=[],
workflow_run_id=None,
annotation=None,
annotation_hit_history=None,
created_at=created_at,
agent_thoughts=[],
message_files=[],
message_metadata_dict={},
status="success",
error="",
parent_message_id=None,
)
class MockQuery:
def __init__(self, model):
self.model = model
def where(self, *args, **kwargs):
return self
def first(self):
if getattr(self.model, "__name__", "") == "Conversation":
return mock_conversation
return None
def order_by(self, *args, **kwargs):
return self
def limit(self, *_):
return self
def all(self):
if getattr(self.model, "__name__", "") == "Message":
return [mock_message]
return []
mock_session = mock.Mock()
mock_session.query.side_effect = MockQuery
mock_session.scalar.return_value = False
monkeypatch.setattr(message_api, "db", SimpleNamespace(session=mock_session))
monkeypatch.setattr(message_api, "current_user", mock_account)
class DummyPagination:
def __init__(self, data, limit, has_more):
self.data = data
self.limit = limit
self.has_more = has_more
monkeypatch.setattr(message_api, "InfiniteScrollPagination", DummyPagination)
mock_account.role = role
response = test_client.get(
f"/console/api/apps/{mock_app_model.id}/chat-messages",
headers=auth_header,
query_string={"conversation_id": str(conversation_id)},
)
assert response.status_code == status