From ea741698e78d7b74187b92f09056eee4d1816fc7 Mon Sep 17 00:00:00 2001
From: Valentin Rigal <rigal@teklia.com>
Date: Fri, 30 Jun 2023 10:58:21 +0000
Subject: [PATCH] Restrict thumbnail S3 PUT URL
---
arkindex/documents/serializers/elements.py | 16 ++++++++--
.../documents/tests/test_retrieve_elements.py | 29 +++++++++++++++++--
2 files changed, 41 insertions(+), 4 deletions(-)
diff --git a/arkindex/documents/serializers/elements.py b/arkindex/documents/serializers/elements.py
index fdcaff8144..e125a4f395 100644
--- a/arkindex/documents/serializers/elements.py
+++ b/arkindex/documents/serializers/elements.py
@@ -23,7 +23,7 @@ from arkindex.documents.serializers.light import (
from arkindex.documents.serializers.ml import ClassificationSerializer, WorkerRunSummarySerializer
from arkindex.images.models import Image
from arkindex.images.serializers import ZoneSerializer
-from arkindex.ponos.utils import is_admin_or_ponos_task
+from arkindex.ponos.models import Task
from arkindex.process.models import WorkerVersion
from arkindex.project.fields import Array
from arkindex.project.mixins import SelectionMixin
@@ -437,7 +437,19 @@ class ElementSlimSerializer(ElementTinySerializer):
@extend_schema_field(serializers.CharField(allow_null=True))
def get_thumbnail_put_url(self, element):
- if is_admin_or_ponos_task(self.context['request']) and element.type.folder:
+ """
+ Only set the Thumbnail PUT URL for Ponos tasks that
+ are running the thumbnails generation on a folder.
+ """
+ # TODO: This check would be simplified to process.thumbnails once that attribute
+ # is available, allowing to use the get_process_from_ponos_auth helper directly.
+ task = self.context.get('request') and self.context['request'].auth
+ if (
+ isinstance(task, Task)
+ and element.type.folder
+ and task.image == settings.ARKINDEX_TASKS_IMAGE
+ and "generate_thumbnails" in task.command
+ ):
return element.thumbnail.s3_put_url
class Meta(ElementTinySerializer.Meta):
diff --git a/arkindex/documents/tests/test_retrieve_elements.py b/arkindex/documents/tests/test_retrieve_elements.py
index 1433be4a4a..aacf31f7cf 100644
--- a/arkindex/documents/tests/test_retrieve_elements.py
+++ b/arkindex/documents/tests/test_retrieve_elements.py
@@ -1,3 +1,4 @@
+from django.test import override_settings
from django.urls import reverse
from rest_framework import status
@@ -110,9 +111,11 @@ class TestRetrieveElements(FixtureAPITestCase):
self.assertIsNone(response.json()['thumbnail_url'])
self.assertIsNone(response.json()['thumbnail_put_url'])
+ @override_settings(ARKINDEX_TASKS_IMAGE='task_image')
def test_get_element_thumbnail_put_ponos_task(self):
"""
- Check getting an element only gives a thumbnail PUT URL to bots for folders
+ Check getting an element only gives a thumbnail PUT URL Ponos tasks
+ running the thumbnails generation command on a folder element
"""
process = Process.objects.create(
mode=ProcessMode.Repository,
@@ -121,8 +124,10 @@ class TestRetrieveElements(FixtureAPITestCase):
)
process.start()
task = process.tasks.get()
+ task.image = 'task_image'
+ task.command = 'python generate_thumbnails'
- self.assertTrue(self.vol.type.folder)
+ task.save()
response = self.client.get(
reverse('api:element-retrieve', kwargs={'pk': str(self.vol.id)}),
HTTP_AUTHORIZATION=f'Ponos {task.token}',
@@ -138,6 +143,26 @@ class TestRetrieveElements(FixtureAPITestCase):
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertIsNone(response.json()['thumbnail_put_url'])
+ def test_get_element_thumbnail_put_requires_thumbnails_task(self):
+ """
+ Only tasks that are intended to generate thumbnails (ARKINDEX_TASKS_IMAGE + thumbnails_generation command)
+ can retrieve the thumbnails PUT URL.
+ """
+ process = Process.objects.create(
+ mode=ProcessMode.Repository,
+ revision=self.worker_version.revision,
+ creator=self.user,
+ )
+ process.start()
+ task = process.tasks.get()
+ self.assertTrue(self.vol.type.folder)
+ response = self.client.get(
+ reverse('api:element-retrieve', kwargs={'pk': str(self.vol.id)}),
+ HTTP_AUTHORIZATION=f'Ponos {task.token}',
+ )
+ self.assertEqual(response.status_code, status.HTTP_200_OK)
+ self.assertIsNone(response.json()['thumbnail_put_url'])
+
def test_get_element_creator(self):
self.vol.creator = self.user
self.vol.save()
--
GitLab