Merge pull request #2482 from onaio/append-azure-token-to-attachment-blob-url

Ensure sas token is appended to azure blob attachment url

Author: KipSigei

onadata/apps/viewer/tests/test_attachment_url.py

@@ -1,9 +1,14 @@
 import os
 
 from django.conf import settings
+from django.contrib.auth import authenticate
+from django.http import HttpResponseRedirect
 from django.urls import reverse
+from mock import patch
+from rest_framework.test import APIRequestFactory
 
 from onadata.apps.logger.models import Attachment
+from onadata.apps.logger.views import submission
 from onadata.apps.main.tests.test_base import TestBase
 from onadata.apps.viewer.views import attachment_url
 
@@ -18,6 +23,8 @@ def setUp(self):
         self._submit_transport_instance_w_attachment()
         self.url = reverse(
             attachment_url, kwargs={'size': 'original'})
+        self._submission_url = reverse(
+            'submissions', kwargs={'username': self.user.username})
 
     def test_attachment_url(self):
         self.assertEqual(
@@ -61,6 +68,54 @@ def test_attachment_url_w_media_id_no_redirect(self):
                        'no_redirect': 'true'})
         self.assertEqual(response.status_code, 200)  # no redirects to amazon
 
+    @patch("onadata.apps.viewer.views.generate_media_download_url")
+    def test_attachment_url_has_azure_sas_token(self, mock_media_url):
+        """Test attachment url has azure sas token"""
+        self._publish_xls_file(
+            os.path.join(
+                settings.PROJECT_ROOT,
+                "apps",
+                "main",
+                "tests",
+                "fixtures",
+                "transportation",
+                "transportation_encrypted.xlsx",
+            )
+        )
+        files = {}
+        for filename in ["submission.xml", "submission.xml.enc"]:
+            files[filename] = os.path.join(
+                settings.PROJECT_ROOT,
+                "apps",
+                "main",
+                "tests",
+                "fixtures",
+                "transportation",
+                "instances_encrypted",
+                filename,
+            )
+        with open(files["submission.xml.enc"], "rb") as encryped_file:
+            with open(files["submission.xml"], "rb") as f:
+                post_data = {
+                    "xml_submission_file": f,
+                    "submission.xml.enc": encryped_file,
+                }
+                self.factory = APIRequestFactory()
+                request = self.factory.post(self._submission_url, post_data)
+                request.user = authenticate(username="bob", password="bob")
+                response = submission(request, username=self.user.username)
+                self.assertEqual(response.status_code, 201)
+
+        # get submission enc attachment
+        attachment = Attachment.objects.all()[1]
+        sas_token = "se=ab736fba7261" # nosec
+        expected_url = f"http://testserver/{attachment.media_file.name}?{sas_token}"
+        mock_media_url.return_value = HttpResponseRedirect(redirect_to=expected_url)
+        response = self.client.get(self.url, {"media_file": attachment.media_file.name})
+        self.assertEqual(response.status_code, 302)
+        self.assertEqual(response.url, expected_url)
+        self.assertIn(f"?{sas_token}", str(response.url))
+
     def tearDown(self):
         path = os.path.join(settings.MEDIA_ROOT, self.user.username)
         for root, dirs, files in os.walk(path, topdown=False):

onadata/apps/viewer/views.py

@@ -56,7 +56,7 @@
     str_to_bool,
 )
 from onadata.libs.utils.google import create_flow
-from onadata.libs.utils.image_tools import image_url
+from onadata.libs.utils.image_tools import image_url, generate_media_download_url
 from onadata.libs.utils.log import Actions, audit_log
 from onadata.libs.utils.logger_tools import (
     generate_content_disposition_header,
@@ -894,7 +894,7 @@ def attachment_url(request, size="medium"):
 
         return response
     if not attachment.mimetype.startswith("image"):
-        return redirect(attachment.media_file.url)
+        return generate_media_download_url(attachment)
     media_url = image_url(attachment, size)
     if media_url:
         return redirect(media_url)