Merge pull request #237 from CenterForOpenScience/modify-get-object

[ENG-7171] Enable HMAC query of configured-storage-addons endpoint using pk in `<guid>:<addon_name>` format

addon_service/configured_addon/storage/views.py

@@ -1,9 +1,11 @@
 from http import HTTPMethod
 
+from django.http import Http404
 from rest_framework.decorators import action
 from rest_framework.response import Response
 
 from addon_service.common.credentials_formats import CredentialsFormats
+from addon_service.common.permissions import IsValidHMACSignedRequest
 from addon_service.common.waterbutler_compat import WaterButlerConfigSerializer
 from addon_service.configured_addon.views import ConfiguredAddonViewSet
 from app.settings import ALLOWED_RESOURCE_URI_PREFIXES
@@ -21,23 +23,35 @@ class ConfiguredStorageAddonViewSet(ConfiguredAddonViewSet):
     )
     serializer_class = ConfiguredStorageAddonSerializer
 
+    def get_object(self):
+        pk = self.kwargs["pk"]
+        if ":" in pk:
+            # Make sure only a valid HMAC request can query this endpoint using
+            # pk in "<guid>:<addon_name>" format
+            IsValidHMACSignedRequest().has_permission(self.request, self)
+            guid, external_service_name = pk.split(":", maxsplit=1)
+            try:
+                addon: ConfiguredStorageAddon = ConfiguredStorageAddon.objects.get(
+                    base_account__external_service__wb_key=external_service_name,
+                    authorized_resource__resource_uri__in=[
+                        f"{prefix}/{guid}" for prefix in ALLOWED_RESOURCE_URI_PREFIXES
+                    ],
+                )
+            except ConfiguredStorageAddon.DoesNotExist:
+                raise Http404
+            self.check_object_permissions(self.request, addon)
+            return addon
+        else:
+            return super().get_object()
+
     @action(
         detail=True,
         methods=[HTTPMethod.GET],
         url_name="waterbutler-credentials",
         url_path="waterbutler-credentials",
     )
     def get_wb_credentials(self, request, pk: str = None):
-        if ":" in pk:
-            guid, external_service_name = pk.split(":", maxsplit=1)
-            addon: ConfiguredStorageAddon = ConfiguredStorageAddon.objects.get(
-                base_account__external_service__wb_key=external_service_name,
-                authorized_resource__resource_uri__in=[
-                    f"{prefix}/{guid}" for prefix in ALLOWED_RESOURCE_URI_PREFIXES
-                ],
-            )
-        else:
-            addon = self.get_object()
+        addon = self.get_object()
         if addon.external_service.credentials_format is CredentialsFormats.OAUTH2:
             addon.base_account.refresh_oauth_access_token__blocking()
         self.resource_name = "waterbutler-credentials"  # for the jsonapi resource type