diff --git a/apps/oss/tests.py b/apps/oss/tests.py index 4703c42b873..a7b7c4283af 100644 --- a/apps/oss/tests.py +++ b/apps/oss/tests.py @@ -1,7 +1,12 @@ +from unittest import mock + from django.test import SimpleTestCase from django.urls import resolve +from common.exception.app_exception import AppUnauthorizedFailed +from knowledge.models import FileSourceType from maxkb.const import CONFIG +from oss.serializers.file import auth from oss.views import FileRetrievalView, FileView, GetUrlView @@ -52,3 +57,102 @@ def test_get_url_retrieval_routes_pass_application_id(self): 'chat_oss_retrieval', {'application_id': 'application-id'}, ) + + +class FileAuthTestCase(SimpleTestCase): + """Regression tests for anonymous retrieval of APPLICATION-source files.""" + + @staticmethod + def _make_file(source_type, source_id, meta=None): + file = mock.Mock() + file.id = 'f0000000-0000-0000-0000-000000000001' + file.source_type = source_type + file.source_id = source_id + file.meta = meta or {} + return file + + def test_application_file_without_token_is_denied(self): + """A visitor-created APPLICATION file with no chat_id must not be publicly retrievable.""" + file = self._make_file(FileSourceType.APPLICATION, 'application-1') + with mock.patch('oss.serializers.file.CONFIG', {'FILE_AUTH': '1'}), \ + mock.patch('oss.serializers.file.QuerySet') as qs: + qs.return_value.filter.return_value.exists.return_value = False + with self.assertRaises(AppUnauthorizedFailed): + auth(file, None) + + def test_public_temporary_file_does_not_require_token(self): + """Temporary/public source types remain anonymously retrievable.""" + file = self._make_file(FileSourceType.TEMPORARY_120_MINUTE, FileSourceType.TEMPORARY_120_MINUTE.value) + with mock.patch('oss.serializers.file.CONFIG', {'FILE_AUTH': '1'}): + # It should return without raising. + self.assertIsNone(auth(file, None)) + + +class FileUploadPermissionTestCase(SimpleTestCase): + """The /oss/file upload endpoint must not let chat/anon clients bind files to protected sources.""" + + @staticmethod + def _chat_request(source_type, source_id): + request = mock.Mock() + request.user = None + request.auth = mock.Mock() + request.auth.chat_user_id = 'chat-user-1' + request.data = {'source_type': source_type, 'source_id': source_id} + request.FILES = {'file': mock.Mock()} + request.META = {} + request.path = '/chat/api/oss/file' + request.query_params = {} + return request + + def test_chat_upload_binding_to_application_is_denied(self): + request = self._chat_request(FileSourceType.APPLICATION.value, 'application-1') + with mock.patch('common.log.log.Log'): + with self.assertRaises(AppUnauthorizedFailed): + FileView().post(request) + + def test_chat_upload_binding_to_chat_is_allowed(self): + request = self._chat_request(FileSourceType.CHAT.value, 'chat-1') + with mock.patch('common.log.log.Log'), \ + mock.patch('oss.views.file.QuerySet') as qs, \ + mock.patch('oss.views.file.FileSerializer') as serializer_cls: + qs.return_value.filter.return_value.first.return_value = mock.Mock() + serializer_cls.return_value.upload.return_value = './oss/file/1' + response = FileView().post(request) + self.assertEqual(response.status_code, 200) + + def test_chat_upload_binding_to_foreign_chat_is_denied(self): + request = self._chat_request(FileSourceType.CHAT.value, 'foreign-chat') + with mock.patch('common.log.log.Log'), \ + mock.patch('oss.views.file.QuerySet') as qs: + qs.return_value.filter.return_value.first.return_value = None + with self.assertRaises(AppUnauthorizedFailed): + FileView().post(request) + + def test_system_user_upload_binding_to_application_is_allowed(self): + request = mock.Mock() + request.user = mock.Mock() + request.user.id = 'user-1' + request.data = {'source_type': FileSourceType.APPLICATION.value, 'source_id': 'application-1'} + request.FILES = {'file': mock.Mock()} + request.META = {} + request.path = '/admin/api/oss/file' + request.query_params = {} + with mock.patch('common.log.log.Log'), \ + mock.patch('oss.views.file.FileSerializer') as serializer_cls: + serializer_cls.return_value.upload.return_value = './oss/file/1' + response = FileView().post(request) + self.assertEqual(response.status_code, 200) + + def test_logged_in_user_on_chat_path_cannot_bind_to_application(self): + """Even a logged-in user must not bind files to protected sources via the /chat path.""" + request = mock.Mock() + request.user = mock.Mock() + request.user.id = 'user-1' + request.data = {'source_type': FileSourceType.APPLICATION.value, 'source_id': 'application-1'} + request.FILES = {'file': mock.Mock()} + request.META = {} + request.path = '/chat/api/oss/file' + request.query_params = {} + with mock.patch('common.log.log.Log'): + with self.assertRaises(AppUnauthorizedFailed): + FileView().post(request)