Correct permissions, fix auth mocking, update some POST logic

addon_service/authorized_storage_account/models.py

@@ -33,4 +33,4 @@ def account_owner(self):
 
     @property
     def owner_reference(self):
-        return self.external_account.owner.user_uri
+        return self.account_owner.user_uri

addon_service/authorized_storage_account/serializers.py

@@ -1,10 +1,12 @@
+from rest_framework.serializers import ReadOnlyField
 from rest_framework_json_api import serializers
 from rest_framework_json_api.relations import (
     HyperlinkedRelatedField,
     ResourceRelatedField,
 )
 from rest_framework_json_api.utils import get_resource_type_from_model
 
+from addon_service.common.serializer_fields import ReadOnlyResourceRelatedField
 from addon_service.models import (
     AuthorizedStorageAccount,
     ConfiguredStorageAddon,
@@ -18,19 +20,6 @@
 RESOURCE_NAME = get_resource_type_from_model(AuthorizedStorageAccount)
 
 
-class AccountOwnerField(ResourceRelatedField):
-    def to_internal_value(self, data):
-        return UserReference.objects.get(user_uri=data["id"])
-
-
-class ExternalStorageServiceField(ResourceRelatedField):
-    def to_internal_value(self, data):
-        external_storage_service, _ = ExternalStorageService.objects.get_or_create(
-            auth_uri=data["id"],
-        )
-        return external_storage_service
-
-
 class AuthorizedStorageAccountSerializer(serializers.HyperlinkedModelSerializer):
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
@@ -42,12 +31,12 @@ def __init__(self, *args, **kwargs):
     url = serializers.HyperlinkedIdentityField(
         view_name=f"{RESOURCE_NAME}-detail", required=False
     )
-    account_owner = AccountOwnerField(
+    account_owner = ReadOnlyResourceRelatedField(
         many=False,
         queryset=UserReference.objects.all(),
         related_link_view_name=f"{RESOURCE_NAME}-related",
     )
-    external_storage_service = ExternalStorageServiceField(
+    external_storage_service = ResourceRelatedField(
         queryset=ExternalStorageService.objects.all(),
         many=False,
         related_link_view_name=f"{RESOURCE_NAME}-related",
@@ -72,15 +61,16 @@ def __init__(self, *args, **kwargs):
     }
 
     def create(self, validate_data):
-        account_owner = validate_data["account_owner"]
+        session_user_uri = self.context['request'].session.get('user_reference_uri')
+        account_owner, _ =  UserReference.objects.get_or_create(user_uri=session_user_uri)
         external_storage_service = validate_data["external_storage_service"]
         # TODO(ENG-5189): Update this once credentials format is finalized
-        credentials, created = ExternalCredentials.objects.get_or_create(
+        credentials, _ = ExternalCredentials.objects.get_or_create(
             oauth_key=validate_data["username"],
             oauth_secret=validate_data["password"],
         )
 
-        external_account, created = ExternalAccount.objects.get_or_create(
+        external_account, _ = ExternalAccount.objects.get_or_create(
             owner=account_owner,
             credentials=credentials,
             credentials_issuer=external_storage_service.credentials_issuer,

addon_service/authorized_storage_account/views.py

@@ -1,5 +1,6 @@
+from rest_framework.exceptions import MethodNotAllowed
 from addon_service.common.permissions import (
-    CanCreateASA,
+    IsAuthenticated,
     SessionUserIsOwner,
 )
 from addon_service.common.viewsets import RetrieveWriteViewSet
@@ -18,7 +19,4 @@ def get_permissions(self):
 
         if self.action in ["retrieve", "retrieve_related", "update", "destroy"]:
             return [SessionUserIsOwner()]
-        elif self.action == "create":
-            return [CanCreateASA()]
-        else:
-            raise NotImplementedError("view action permission not implemented")
+        return [IsAuthenticated()]

addon_service/common/permissions.py

@@ -3,14 +3,17 @@
     permissions,
 )
 
-from addon_service.models import UserReference
+from addon_service.models import ResourceReference
 from app.authentication import authenticate_resource
 
 
+class IsAuthenticated(permissions.BasePermission):
+
+    def has_permission(self, request, view):
+        return request.session.get("user_reference_uri") is not None
+
+
 class SessionUserIsOwner(permissions.BasePermission):
-    """
-    Decorator to fetch 'user_reference_uri' from the session and pass it to the permission check function.
-    """
 
     def has_object_permission(self, request, view, obj):
         session_user_uri = request.session.get("user_reference_uri")
@@ -19,30 +22,22 @@ def has_object_permission(self, request, view, obj):
         return False
 
 
-class SessionUserIsResourceReferenceOwner(permissions.BasePermission):
+class SessionUserCanViewReferencedResource(permissions.BasePermission):
     def has_object_permission(self, request, view, obj):
-        resource_uri = authenticate_resource(request, obj.resource_uri, "read")
-        return obj.resource_uri == resource_uri
+        return authenticate_resource(request, obj.resource_uri, "read")
 
 
-class CanCreateCSA(permissions.BasePermission):
+class SessionUserIsReferencedResourceAdmin(permissions.BasePermission):
     def has_permission(self, request, view):
-        authorized_resource_id = request.data.get("authorized_resource", {}).get("id")
-        if authenticate_resource(request, authorized_resource_id, "admin"):
-            return True
-        return False
+        resource_uri = None
+        try:
+           resource_uri = ResourceReference.objects.get(
+                id=request.data.get("authoirized_resource", {}).get("id")
+            ).resource_uri
+        except ResourceReference.DoesNotExist:
+            resource_uri = request.data.get("authorized_resource", {}).get("resource_uri")
 
+        if resource_uri is None:
+            raise exceptions.ParseError
 
-class CanCreateASA(permissions.BasePermission):
-    def has_permission(self, request, view):
-        session_user_uri = request.session.get("user_reference_uri")
-        request_user_uri = request.data.get("account_owner", {}).get("id")
-        if not session_user_uri == request_user_uri:
-            raise exceptions.NotAuthenticated(
-                "Account owner ID is missing in the request."
-            )
-        try:
-            UserReference.objects.get(user_uri=request_user_uri)
-            return True
-        except UserReference.DoesNotExist:
-            raise exceptions.NotAuthenticated("User does not exist.")
+        return authenticate_resource(request, resource_uri, "admin")

addon_service/common/serializer_fields.py

@@ -0,0 +1,5 @@
+from rest_framework import serializers as drf_serializers
+from rest_framework_json_api import serializers as json_api_serializers
+
+class ReadOnlyResourceRelatedField(json_api_serializers.ResourceRelatedField, drf_serializers.ReadOnlyField):
+    pass

addon_service/configured_storage_addon/models.py

@@ -32,3 +32,7 @@ def account_owner(self):
     @property
     def owner_reference(self):
         return self.base_account.external_account.owner.user_uri
+
+    @property
+    def resource_uri(self):
+        return self.authorized_resource.resource_uri

addon_service/configured_storage_addon/serializers.py

@@ -15,7 +15,7 @@
 class AuthorizedResourceField(ResourceRelatedField):
     def to_internal_value(self, data):
         resource_reference, _ = ResourceReference.objects.get_or_create(
-            resource_uri=data["id"]
+            resource_uri=data["resource_uri"]
         )
         return resource_reference
 

addon_service/configured_storage_addon/views.py

@@ -1,6 +1,6 @@
 from addon_service.common.permissions import (
-    CanCreateCSA,
-    SessionUserIsOwner,
+    SessionUserCanViewReferencedResource,
+    SessionUserIsReferencedResourceAdmin,
 )
 from addon_service.common.viewsets import RetrieveWriteViewSet
 
@@ -17,8 +17,8 @@ def get_permissions(self):
             return super().get_permissions()
 
         if self.action in ["retrieve", "retrieve_related", "update", "destroy"]:
-            return [SessionUserIsOwner()]
+            return [SessionUserCanViewReferencedResource()]
         elif self.action == "create":
-            return [CanCreateCSA()]
+            return [SessionUserIsReferencedResourceAdmin()]
         else:
             raise NotImplementedError("view action permission not implemented")

addon_service/resource_reference/views.py

@@ -1,4 +1,4 @@
-from addon_service.common.permissions import SessionUserIsResourceReferenceOwner
+from addon_service.common.permissions import SessionUserCanViewReferencedResource
 from addon_service.common.viewsets import RetrieveOnlyViewSet
 from addon_service.serializers import ResourceReferenceSerializer
 
@@ -8,4 +8,4 @@
 class ResourceReferenceViewSet(RetrieveOnlyViewSet):
     queryset = ResourceReference.objects.all()
     serializer_class = ResourceReferenceSerializer
-    permission_classes = [SessionUserIsResourceReferenceOwner]
+    permission_classes = [SessionUserCanViewReferencedResource]

addon_service/tests/_helpers.py

@@ -1,8 +1,12 @@
-from functools import wraps
+import json
+from collections import defaultdict
+from functools import partial, wraps
+from http import HTTPStatus
 from unittest.mock import patch
 
 import httpx
 from django.contrib.sessions.backends.db import SessionStore
+from rest_framework import exceptions as drf_exceptions
 from rest_framework.test import APIRequestFactory
 
 from app import settings
@@ -17,41 +21,112 @@ def get_test_request(user=None, method="get", path="", cookies=None):
             _request.COOKIES[name] = value
     return _request
 
+class MockOSF:
+
+    def __init__(self, permissions=None):
+        '''A lightweight, configurable  mock of OSF for testing remote permissions.
+
+        Accepts a mapping of arbitrary user_uris to their roles on arbitrary resource_uris.
+        i.e.
+        {
+            'osf.io/abcde': {'osf.io/bcdef': 'write', 'osf.io/cdefg': 'admin'},
+            'osf.io/zyxwv': {'osf.io/yxwvut': 'read'}
+        }
+        Intercepts 'get' requests and uses the request url and this mapping to generate a minimal
+        response required for testing GravyValet's behavior.
+
+        Users of the mock can either explicitly tell the Mock which user to assume a call is from,
+        or they can include a cookie with the 'user_uri' in their GET request, and MockOSF will honor
+        that user
+        '''
+        self._permissions = defaultdict(lambda: defaultdict(dict))
+        if permissions:
+            self._permissions.update(permissions)
+        self._configured_caller_uri = None
+        self._mock_auth_request = patch('app.authentication.make_auth_request', side_effect=self._mock_user_check)
+        self._mock_resource_check = patch('addon_service.common.permissions.authenticate_resource', side_effect=self._mock_resource_check)
+        self._mock_auth_request.start()
+        self._mock_resource_check.start()
+
+
+    def stop(self):
+        self._mock_auth_request.stop()
+        self._mock_resource_check.stop()
+
+    def configure_assumed_caller(self, caller_uri):
+        self._configured_caller_uri = caller_uri
+
+    def configure_user_role(self, user_uri, resource_uri, role):
+        self._permissions[user_uri][resource_uri] = role
+
+    def _get_assumed_caller(self, cookies=None):
+        if self._configured_caller_uri:
+            return self._configured_caller_uri
+        if cookies is not None:
+            return cookies.get(settings.USER_REFERENCE_COOKIE)
+        raise ValueError('MockOSF cannot handle requests without configuring a user')
+
+    def _get_user_permissions(self, user_uri, resource_uri):
+        role = self._permissions[user_uri][resource_uri]
+        if not role:
+            return []
+        if role == 'read':
+            return ['read']
+        if role == 'write':
+            return ['read', 'write']
+        if role == 'admin':
+            return ['read', 'write', 'admin']
+
+    def _mock_user_check(self, *args, **kwargs):
+        caller_uri = self._get_assumed_caller(cookies=kwargs.get('cookies'))
+        return {"data": {"links": {"iri": caller_uri}}}
+
+    def _mock_resource_check(self, request, uri, required_permission, *args, **kwargs):
+        caller = self._get_assumed_caller(cookies=request.COOKIES)
+        permissions = self._get_user_permissions(user_uri=caller, resource_uri=uri)
+        if required_permission.lower() not in permissions:
+            raise drf_exceptions.PermissionDenied
+        return uri  # mimicking behavior from the check being mocked
+
+
+
+def with_mocked_httpx_get(response_status=HTTPStatus.OK, user_uri=None):
 
-def with_mocked_httpx_get(response_status=200):
-    """Decorator to mock httpx.Client get requests with a customizable response status."""
 
     def decorator(func):
         @wraps(func)
-        def wrapper(self, *args, **kwargs):
-            with patch(
-                "httpx.Client.get",
-                new=lambda *args, **kwargs: mock_httpx_response(
-                    *args, response_status=response_status
-                ),
-            ):
-                return func(self, *args, **kwargs)
+        def wrapper(testcase, *args, **kwargs):
+            mock_side_effect = partial(
+                _mock_httpx_response,
+                response_status=response_status,
+                user_uri=user_uri or testcase._user.user_uri
+            )
+            patcher = patch('httpx.Client.get', side_effect=mock_side_effect)
+            client = patcher.start()
+            test_result = func(testcase, *args, **kwargs)
+            patcher.stop()
+            return test_result
 
         return wrapper
 
     return decorator
 
 
-def mock_httpx_response(url, current_user, response_status, *args, **kwargs):
+def _mock_httpx_response(response_status, user_uri, url, *args, **kwargs):
     """Generates mock httpx.Response based on the requested URL and response status."""
-    if response_status == 200:
-        if url == settings.USER_REFERENCE_LOOKUP_URL:
-            payload = {"data": {"links": {"iri": current_user.user_uri}}}
-        else:
-            guid = url.rstrip("/").split("/")[-1]
-            payload = {
-                "data": {
-                    "attributes": {
-                        "current_user_permissions": ["read", "write", "admin"]
-                    },
-                    "links": {"iri": f"{settings.URI_ID}{guid}"},
-                }
-            }
-        return httpx.Response(status_code=200, json=payload)
-    else:  # Handles 403 and other statuses explicitly
+    if not response_status.is_success:
         return httpx.Response(status_code=response_status)
+
+    if url == settings.USER_REFERENCE_LOOKUP_URL:
+        payload = {"data": {"links": {"iri": user_uri}}}
+    else:
+        guid = url.rstrip("/").split("/")[-1]
+        payload = {
+            "data": {
+                "attributes": {
+                    "current_user_permissions": ["read", "write", "admin"]
+                },
+                "links": {"iri": f"{settings.URI_ID}{guid}"},
+            }
+        }
+    return httpx.Response(status_code=response_status, json=payload)