Merge pull request #774 from uw-it-aca/hotfix/missing-f-string-spec

Hotfix/missing f string spec

Author: mikeseibel

endorsement/exceptions.py

@@ -87,3 +87,7 @@ class EmptyDelegateRightsException(AccessRecordException):
 
 class DelegateRightMismatchException(AccessRecordException):
     pass
+
+
+class DelegateParameterException(Exception):
+    pass

endorsement/resources/msca/file/mbx/v1/jstaff/SetDelegatePerms/u_javerage_admin/FullAccess

@@ -0,0 +1,7 @@
+{
+    "TargetNetid": "jstaff",
+    "Delegates": [{
+        "User": "u_javerage_admin",
+        "AccessRights": "FullAccess"
+     }]
+}

endorsement/test/views/api/test_access.py

@@ -0,0 +1,73 @@
+# Copyright 2025 UW-IT, University of Washington
+# SPDX-License-Identifier: Apache-2.0
+
+
+from django.urls import reverse
+from django.test import Client
+from userservice.user import get_original_user
+from endorsement.models import (
+    Accessee, Accessor, AccessRight, AccessRecord, AccessRecordConflict)
+from endorsement.test.views import require_url
+from endorsement.test.api import EndorsementApiTest
+
+
+@require_url('access_api', 'access url not configured')
+class TestAccess(EndorsementApiTest):
+    fixtures = ['test_data/accessright.json',
+                'test_data/accessee.json',
+                'test_data/accessor.json']
+
+    def test_access_api(self):
+        test_data = {
+            'access_type': "FullAccess",
+            'delegate': "u_javerage_admin",
+            'mailbox': "jstaff"
+        }
+
+        self.assertEqual(0, AccessRecord.objects.all().count())
+
+        self.set_user('jstaff')
+
+        url = reverse('access_api')
+        response = self.client.post(
+            url, test_data, content_type='application/json')
+
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(1, AccessRecord.objects.all().count())
+
+    def test_access_api_missing_param(self):
+        test_data = {
+            'access_type': "FullAccess",
+            'delegate': "u_javerage_admin",
+            'mailbox': None
+        }
+
+        self.assertEqual(0, AccessRecord.objects.all().count())
+
+        self.set_user('jstaff')
+
+        url = reverse('access_api')
+        response = self.client.post(
+            url, test_data, content_type='application/json')
+
+        self.assertEqual(response.status_code, 400)
+        self.assertEqual(0, AccessRecord.objects.all().count())
+
+    def test_access_api_renew(self):
+        test_data = {
+            'action': 'renew',
+            'access_type': "FullAccess",
+            'delegate': "u_javerage_admin",
+            'mailbox': "jstaff"
+        }
+
+        self.assertEqual(0, AccessRecord.objects.all().count())
+
+        self.set_user('jstaff')
+
+        url = reverse('access_api')
+        response = self.client.post(
+            url, test_data, content_type='application/json')
+
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(1, AccessRecord.objects.all().count())

endorsement/views/api/office/access.py

@@ -11,7 +11,8 @@
 from endorsement.dao.office import is_office_permitted, get_office_accessor
 from endorsement.views.rest_dispatch import (
     RESTDispatch, invalid_session, invalid_endorser, data_error)
-from endorsement.exceptions import UnrecognizedUWNetid, InvalidNetID
+from endorsement.exceptions import (
+    UnrecognizedUWNetid, InvalidNetID, DelegateParameterException)
 from endorsement.util.auth import is_support_user
 from uw_msca.access_rights import get_access_rights
 from restclients_core.exceptions import DataFailureException
@@ -68,10 +69,16 @@ def post(self, request, *args, **kwargs):
         except InvalidNetID:
             return invalid_endorser(logger)
 
-        mailbox = request.data.get('mailbox', None)
-        delegate = request.data.get('delegate', None)
-        access_type = request.data.get('access_type', None)
-        previous_access_type = request.data.get('previous_access_type', None)
+        try:
+            mailbox = self._validate_param(request.data, 'mailbox')
+            delegate = self._validate_param(request.data, 'delegate')
+            access_type = self._validate_param(request.data, 'access_type')
+            previous_access_type = self._validate_param(
+                request.data, 'previous_access_type', True)
+        except DelegateParameterException as ex:
+            message = f"Access.post parameter: {ex}"
+            logger.error(message)
+            return self.error_response(400, message=message)
 
         if not is_office_permitted(mailbox):
             return invalid_endorser(logger)
@@ -89,7 +96,8 @@ def post(self, request, *args, **kwargs):
                 accessee, accessor, access_type, acted_as)
         except DataFailureException as ex:
             logger.error(
-                f"ERROR: set/revoke access {ex.url} ({ex.status}): {ex.msg}")
+                f"Access.post {ex.url} FAILURE data: {request.data} "
+                f"status: {ex.msg} ({ex.status})")
             return self.error_response(ex.status, message=ex.msg)
 
         return self.json_response(access.json_data())
@@ -102,11 +110,16 @@ def patch(self, request, *args, **kwargs):
         except InvalidNetID:
             return invalid_endorser(logger)
 
-        action = request.data.get('action', None)
-        mailbox = request.data.get('mailbox', None)
-        delegate = request.data.get('delegate', None)
-        access_type = request.data.get('access_type', None)
-        previous_access_type = request.data.get('previous_access_type', None)
+        try:
+            action = self._validate_param(request.data, 'action', True)
+            mailbox = self._validate_param(request.data, 'mailbox')
+            delegate = self._validate_param(request.data, 'delegate')
+            access_type = self._validate_param(request.data, 'access_type')
+            previous_access_type = self._validate_param(
+                request.data, 'previous_access_type', True)
+        except DelegateParameterException as ex:
+            logger.error(f"Access.patch parameter: {ex}")
+            return self.error_response(400, message=str(ex))
 
         if not is_office_permitted(mailbox):
             return invalid_endorser(logger)
@@ -123,10 +136,16 @@ def patch(self, request, *args, **kwargs):
                     accessee, accessor, previous_access_type,
                     access_type, acted_as)
             else:
+                logger.error(
+                    f"Access.patch missing parameter: {request.data}")
                 return self.error_response(404, message="Insufficient Data")
         except AccessRecord.DoesNotExist:
+            logger.error(
+                f"Access.patch missing access record: {request.data}")
             return self.error_response(404, message="Unknown Access Record")
         except DataFailureException as ex:
+            logger.error(
+                f"Access.patch request: '{ex.url}' ({ex.status}): {ex.msg}")
             return self.error_response(ex.status, message=ex.msg)
 
         return self.json_response(access.json_data())
@@ -149,10 +168,30 @@ def delete(self, request, *args, **kwargs):
         try:
             access = revoke_access(accessee, accessor, access_type, acted_as)
         except DataFailureException as ex:
+            logger.error(
+                f"Access.delete request: {ex.url} ({ex.status}): {ex.msg}")
             return self.error_response(ex.status, message=ex.msg)
 
         return self.json_response(access.json_data())
 
+    def _validate_param(self, data, field, nullable=False):
+        try:
+            value = data[field].strip()
+            if not value and not nullable:
+                raise DelegateParameterException(f"Empty {field} field")
+
+            return value
+        except AttributeError as ex:
+            if nullable:
+                return None
+
+            raise DelegateParameterException(f"Null {field} field")
+        except KeyError as ex:
+            if nullable:
+                return None
+
+            raise DelegateParameterException(f"Missing {field} field")
+
     def _load_access_for_accessee(self, accessee):
         access = AccessRecord.objects.get_access_for_accessee(accessee)
         return [ar.json_data() for ar in access]