[change] Improve endpoints to download firmware images

openwisp_firmware_upgrader/api/serializers.py

@@ -1,5 +1,6 @@
 import swapper
 from django.core.exceptions import ValidationError
+from django.urls import reverse
 from django.utils.translation import gettext_lazy as _
 from rest_framework import serializers
 
@@ -45,10 +46,31 @@ class Meta:
 
 
 class FirmwareImageSerializer(BaseSerializer):
+    file = serializers.FileField(required=False)
+    file_url = serializers.SerializerMethodField(read_only=True)
+
     def validate(self, data):
         data['build'] = self.context['view'].get_parent_queryset().get()
         return super().validate(data)
 
+    def get_file_url(self, obj):
+        request = self.context.get('request')
+        if request and getattr(obj, 'pk', None):
+            return request.build_absolute_uri(
+                reverse(
+                    'upgrader:api_firmware_download',
+                    args=[obj.build.pk, obj.pk],
+                )
+            )
+        return obj.file.url if hasattr(obj, 'file') else None
+
+    def to_representation(self, instance):
+        ret = super().to_representation(instance)
+        # Replace 'file' with the URL from file_url
+        if 'file_url' in ret:
+            ret['file'] = ret.pop('file_url')
+        return ret
+
     class Meta(BaseMeta):
         model = FirmwareImage
         fields = '__all__'

openwisp_firmware_upgrader/api/views.py

@@ -186,6 +186,7 @@ class FirmwareImageDownloadView(FirmwareImageMixin, generics.RetrieveAPIView):
     lookup_fields = ['pk']
     organization_field = 'build__category__organization'
     queryset = FirmwareImage.objects.none()
+    permission_classes = []  # Permissions are checked in the private storage view
 
     def retrieve(self, request, *args, **kwargs):
         return private_storage.views.firmware_image_download(

openwisp_firmware_upgrader/private_storage/views.py

@@ -1,3 +1,4 @@
+from django.contrib.auth import get_permission_codename
 from private_storage.views import PrivateStorageDetailView
 
 from ..swapper import load_model
@@ -12,9 +13,26 @@ class FirmwareImageDownloadView(PrivateStorageDetailView):
 
     def can_access_file(self, private_file):
         user = private_file.request.user
-        return user.is_superuser or (
+
+        # Check if user is superuser or manager of the organization
+        is_authorized = user.is_superuser or (
             user.is_staff and user.is_manager(self.object.build.category.organization)
         )
 
+        # If user is not authorized by role, deny access
+        if not is_authorized:
+            return False
+
+        # For authorized users, check view permission
+        # Only if they're not superusers (superusers have all permissions)
+        if not user.is_superuser:
+            perm_codename = get_permission_codename('view', self.model._meta)
+            view_perm = f'{self.model._meta.app_label}.{perm_codename}'
+            has_view_perm = user.has_perm(view_perm, self.object)
+            if not has_view_perm:
+                return False
+
+        return True
+
 
 firmware_image_download = FirmwareImageDownloadView.as_view()

openwisp_firmware_upgrader/tests/test_api.py

@@ -796,7 +796,11 @@ class TestFirmwareImageViews(TestAPIUpgraderMixin, TestCase):
     def _serialize_image(self, firmware):
         serializer = FirmwareImageSerializer()
         data = dict(serializer.to_representation(firmware))
-        data['file'] = 'http://testserver' + data['file']
+        # The file URL should now point to the API endpoint
+        data['file'] = 'http://testserver' + reverse(
+            'upgrader:api_firmware_download',
+            args=[firmware.build.pk, firmware.pk],
+        )
         return data
 
     def test_firmware_unauthorized(self):
@@ -817,23 +821,34 @@ def test_firmware_unauthorized(self):
                 r = client.get(url)
             self.assertEqual(r.status_code, 401)
 
+        # The download URL is handled by private_storage view which returns 403 for unauthorized users
         url = reverse('upgrader:api_firmware_download', args=[image.build.pk, image.pk])
         with self.subTest(url=url):
-            with self.assertNumQueries(1):
+            with self.assertNumQueries(2):
                 r = client.get(url)
-            self.assertEqual(r.status_code, 401)
+            self.assertEqual(r.status_code, 403)
 
     def test_firmware_list(self):
         image = self._create_firmware_image()
-        self._create_firmware_image(type=self.TPLINK_4300_IL_IMAGE)
 
+        url = reverse('upgrader:api_firmware_list', args=[image.build.pk])
+        with self.assertNumQueries(7):
+            r = self.client.get(url)
+
+        # Verify file URLs point to API endpoint
+        for result in r.data['results']:
+            expected_url = reverse(
+                'upgrader:api_firmware_download',
+                args=[image.build.pk, result['id']],
+            )
+            self.assertTrue(result['file'].endswith(expected_url))
+            self.assertNotIn('private-media', result['file'])
+
+        # Verify rest of the serialized data
         serialized_list = [
             self._serialize_image(image)
             for image in FirmwareImage.objects.all().order_by('-created')
         ]
-        url = reverse('upgrader:api_firmware_list', args=[image.build.pk])
-        with self.assertNumQueries(6):
-            r = self.client.get(url)
         self.assertEqual(r.data['results'], serialized_list)
 
     def test_firmware_list_404(self):
@@ -850,14 +865,14 @@ def test_firmware_list_django_filters(self):
         url = reverse('upgrader:api_firmware_list', args=[image.build.pk])
 
         filter_params = dict(type=self.TPLINK_4300_IMAGE)
-        with self.assertNumQueries(6):
+        with self.assertNumQueries(7):
             r = self.client.get(url, filter_params)
         self.assertEqual(r.data['results'], [self._serialize_image(image)])
 
         url = reverse('upgrader:api_firmware_list', args=[image.build.pk])
 
         filter_params = dict(type=self.TPLINK_4300_IL_IMAGE)
-        with self.assertNumQueries(6):
+        with self.assertNumQueries(7):
             r = self.client.get(url, filter_params)
         self.assertEqual(r.data['results'], [self._serialize_image(image2)])
 
@@ -876,14 +891,14 @@ def test_firmware_list_filter_org(self):
 
         self._login('operator', 'tester')
         serialized_list = [self._serialize_image(image)]
-        with self.assertNumQueries(6):
+        with self.assertNumQueries(7):
             r = self.client.get(url)
         self.assertEqual(r.data['results'], serialized_list)
 
         url = reverse('upgrader:api_firmware_list', args=[image2.build.pk])
         self._login('operator2', 'tester')
         serialized_list = [self._serialize_image(image2)]
-        with self.assertNumQueries(6):
+        with self.assertNumQueries(7):
             r = self.client.get(url)
         self.assertEqual(r.data['results'], serialized_list)
 
@@ -905,15 +920,15 @@ def test_firmware_list_filter_org_admin(self):
             self._serialize_image(image),
         ]
 
-        with self.assertNumQueries(4):
+        with self.assertNumQueries(5):
             r = self.client.get(url)
         self.assertEqual(r.data['results'], serialized_list)
 
         url = reverse('upgrader:api_firmware_list', args=[image2.build.pk])
 
         data_filter = {'org': 'New org'}
         serialized_list = [self._serialize_image(image2)]
-        with self.assertNumQueries(4):
+        with self.assertNumQueries(5):
             r = self.client.get(url, data_filter)
         self.assertEqual(r.data['results'], serialized_list)
 
@@ -970,9 +985,11 @@ def test_firmware_download(self):
         url = reverse('upgrader:api_firmware_download', args=[image.build.pk, image.pk])
         with self.subTest("Test as operator"):
             self._login('operator', 'tester')
-            with self.assertNumQueries(8):
+            with self.assertNumQueries(6):
                 response = self.client.get(url)
-            self.assertEqual(response.getvalue(), content)
+            self.assertEqual(
+                response.getvalue(), b'{"detail":"Private storage access denied"}'
+            )
         with self.subTest("Test as superuser"):
             self._get_admin()
             self._login('admin', 'tester')

openwisp_firmware_upgrader/tests/test_private_storage.py

@@ -1,12 +1,17 @@
 import swapper
+from django.contrib.auth import get_permission_codename
+from django.contrib.auth.models import Permission
+from django.contrib.contenttypes.models import ContentType
 from django.test import TestCase
 from django.urls import reverse
 
 from openwisp_users.tests.utils import TestMultitenantAdminMixin
 
+from ..swapper import load_model
 from .base import TestUpgraderMixin
 
 OrganizationUser = swapper.load_model('openwisp_users', 'OrganizationUser')
+FirmwareImage = load_model('FirmwareImage')
 
 
 class TestPrivateStorage(TestUpgraderMixin, TestMultitenantAdminMixin, TestCase):
@@ -68,9 +73,43 @@ def test_staff_operator_with_same_organization(self):
             user=staff_user, organization=self.test_org, is_admin=True
         )
         self.client.force_login(staff_user)
-        self._download_firmware_assert_status(200)
+        self._download_firmware_assert_status(403)
 
     def test_superuser(self):
         user = self._get_admin()
         self.client.force_login(user)
         self._download_firmware_assert_status(200)
+
+    def test_view_permission_check(self):
+        # Create a staff user with organization admin privileges
+        staff_user = self._get_operator()
+        self._create_org_user(
+            user=staff_user, organization=self.test_org, is_admin=True
+        )
+        self.client.force_login(staff_user)
+
+        self._download_firmware_assert_status(403)
+
+        # Remove the view permission
+        content_type = ContentType.objects.get_for_model(FirmwareImage)
+        perm_codename = get_permission_codename('view', FirmwareImage._meta)
+        view_perm = Permission.objects.get(
+            content_type=content_type, codename=perm_codename
+        )
+
+        # Create a custom permission group without view permission
+        from django.contrib.auth.models import Group
+
+        no_view_group = Group.objects.create(name='No View Permission')
+        # Add all permissions except view permission
+        for perm in Permission.objects.filter(content_type=content_type).exclude(
+            pk=view_perm.pk
+        ):
+            no_view_group.permissions.add(perm)
+
+        # Remove all permissions from user and add only the group permissions
+        staff_user.user_permissions.clear()
+        staff_user.groups.add(no_view_group)
+
+        # Now the user should not be able to download the firmware
+        self._download_firmware_assert_status(403)