Add permissions to assemble and files

files/migrations/0002_fileblob_organizations.py

@@ -0,0 +1,19 @@
+# Generated by Django 3.2 on 2021-04-16 18:58
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('organizations_ext', '0001_squashed_0009_organization_scrub_ip_addresses'),
+        ('files', '0001_initial'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='fileblob',
+            name='organizations',
+            field=models.ManyToManyField(to='organizations_ext.Organization'),
+        ),
+    ]

files/models.py

@@ -5,6 +5,7 @@ from glitchtip.base_models import CreatedModel
 class FileBlob(CreatedModel):
     upload = models.FileField(upload_to="uploads/")
     checksum = models.CharField(max_length=40, unique=True)
+    organizations = models.ManyToManyField("organizations_ext.Organization")
 
     @classmethod
     def from_files(cls, files, organization=None, logger=None):
@@ -23,3 +24,5 @@ class FileBlob(CreatedModel):
             blob.checksum = file_with_checksum[1]
             blob.upload.save(blob_file.name, blob_file)
             blob.save()
+            if organization:
+                blob.organizations.add(organization)

files/permissions.py

@@ -0,0 +1,6 @@
+from releases.permissions import ReleasePermission
+
+
+class ChunkUploadPermission(ReleasePermission):
+    def get_user_scopes(self, obj, user):
+        return obj.get_user_scopes(user)

files/tests.py

@@ -1,3 +1,45 @@
-from django.test import TestCase
+from io import BytesIO
+from django.core.files.uploadedfile import InMemoryUploadedFile
+from django.shortcuts import reverse
+from glitchtip.test_utils.test_case import GlitchTipTestCase
+from glitchtip.test_utils.test_case import APIPermissionTestCase
 
-# Create your tests here.
+
+class ChunkUploadAPITestCase(GlitchTipTestCase):
+    def setUp(self):
+        self.create_user_and_project()
+        self.url = reverse(
+            "chunk-upload", kwargs={"organization_slug": self.organization}
+        )
+
+    def test_get(self):
+        res = self.client.get(self.url)
+        self.assertEqual(res.status_code, 200)
+
+    def test_post(self):
+        im_io = BytesIO()
+        file = InMemoryUploadedFile(
+            im_io, None, "random-name.jpg", "image/jpeg", len(im_io.getvalue()), None
+        )
+        data = {"file_gzip": file}
+        res = self.client.post(self.url, data)
+        self.assertEqual(res.status_code, 200)
+
+
+class ChunkUploadAPIPermissionTests(APIPermissionTestCase):
+    def setUp(self):
+        self.create_user_org()
+        self.set_client_credentials(self.auth_token.token)
+        self.url = reverse(
+            "chunk-upload", kwargs={"organization_slug": self.organization}
+        )
+
+    def test_get(self):
+        self.assertGetReqStatusCode(self.url, 403)
+        self.auth_token.add_permission("project:read")
+        self.assertGetReqStatusCode(self.url, 200)
+
+    def test_post(self):
+        self.assertGetReqStatusCode(self.url, 403)
+        self.auth_token.add_permission("project:write")
+        self.assertGetReqStatusCode(self.url, 200)

files/views.py

@@ -4,9 +4,12 @@ from io import BytesIO
 from gzip import GzipFile
 from django.conf import settings
 from django.urls import reverse
+from django.shortcuts import get_object_or_404
 from rest_framework import views, status
 from rest_framework.response import Response
+from organizations_ext.models import Organization
 from .models import FileBlob
+from .permissions import ChunkUploadPermission
 
 
 CHUNK_UPLOAD_BLOB_SIZE = 8 * 1024 * 1024  # 8MB
@@ -32,6 +35,8 @@ class GzipChunk(BytesIO):
 
 
 class ChunkUploadAPIView(views.APIView):
+    permission_classes = [ChunkUploadPermission]
+
     def get(self, request, organization_slug):
         url = settings.GLITCHTIP_DOMAIN.geturl() + reverse(
             "chunk-upload", args=[organization_slug]
@@ -54,6 +59,11 @@ class ChunkUploadAPIView(views.APIView):
         logger = logging.getLogger("glitchtip.files")
         logger.info("chunkupload.start")
 
+        organization = get_object_or_404(
+            Organization, slug=organization_slug.lower(), users=self.request.user
+        )
+        self.check_object_permissions(request, organization)
+
         files = request.data.getlist("file")
         files += [GzipChunk(chunk) for chunk in request.data.getlist("file_gzip")]
         if len(files) == 0:
@@ -96,7 +106,7 @@ class ChunkUploadAPIView(views.APIView):
 
         try:
             FileBlob.from_files(
-                zip(files, checksums), organization=organization_slug, logger=logger
+                zip(files, checksums), organization=organization, logger=logger
             )
         except IOError as err:
             logger.info(

releases/tests/test_permissions.py

@@ -48,6 +48,16 @@ class TeamAPIPermissionTests(APIPermissionTestCase):
         self.assertGetReqStatusCode(self.organization_detail_url, 200)
         self.assertGetReqStatusCode(self.project_detail_url, 200)
 
+    def test_assemble(self):
+        url = self.organization_detail_url + "assemble/"
+        data = {
+            "checksum": "94bc085fe32db9b4b1b82236214d65eeeeeeeeee",
+            "chunks": ["94bc085fe32db9b4b1b82236214d65eeeeeeeeee"],
+        }
+        self.assertPostReqStatusCode(url, data, 403)
+        self.auth_token.add_permission("project:write")
+        self.assertPostReqStatusCode(url, data, 200)
+
     def test_create(self):
         self.auth_token.add_permission("project:read")
         data = {"version": "new-version"}