refactor some logic from review sections

Author: Akol125

backend/src/authorization.py

@@ -5,7 +5,7 @@
 
 from models.errors import UnauthorizedError
 
-PERMISSIONS_HEADER = "Permissions"
+
 AUTHENTICATION_HEADER = "AuthenticationType"
 
 
@@ -14,40 +14,16 @@ class UnknownPermission(RuntimeError):
     """Error when the parsed value can't be converted to Permissions enum."""
 
 
-class EndpointOperation(Enum):
-    """The kind of operation.
-    This maps one-to-one to each endpoint. Authorization class decides whether there are sufficient permissions or not.
-    The caller is responsible for passing the correct operation.
-    """
-    READ = 0,
-    CREATE = 1,
-    UPDATE = 2,
-    DELETE = 3,
-    SEARCH = 4,
-
-
 class AuthType(str, Enum):
     """This backend supports all three types of authentication.
     An Apigee App should specify AuthenticationType in its custom attribute.
     Each Apigee app can only have one type of authentication which is enforced by onboarding process.
     See: https://digital.nhs.uk/developer/guides-and-documentation/security-and-authorisation"""
     APP_RESTRICTED = "ApplicationRestricted",
-    NHS_LOGIN = "NnsLogin",
+    NHS_LOGIN = "NhsLogin",
     CIS2 = "Cis2",
 
 
-class Permission(str, Enum):
-    """Permission name for each operation that can be done to an Immunization Resource
-    An Apigee App should specify a set of these as a comma-separated custom attribute.
-    Permission works the same way as 'scope' but, in this case, they're called permission to distinguish them from
-    OAuth2 scopes"""
-    READ = "immunization:read"
-    CREATE = "immunization:create"
-    UPDATE = "immunization:update"
-    DELETE = "immunization:delete"
-    SEARCH = "immunization:search"
-
-
 class Authorization:
     """ Authorize the call based on the endpoint and the authentication type.
     This class uses the passed headers from Apigee to decide the type of authentication (Application Restricted,
@@ -56,52 +32,13 @@ class Authorization:
     UnknownPermission is due to proxy bad configuration, and should result in 500. Any invalid value, either
     insufficient permissions or bad string, will result in UnauthorizedError if it comes from user.
     """
-
-    def authorize(self, operation: EndpointOperation, aws_event: dict):
+   
+    def authorize(self, aws_event: dict):
         auth_type = self._parse_auth_type(aws_event["headers"])
-        if auth_type == AuthType.APP_RESTRICTED:
-            self._app_restricted(operation, aws_event)
-        if auth_type == AuthType.CIS2:
-            self._cis2(operation, aws_event)
-        # TODO(NhsLogin_AMB-1923) add NHSLogin
-        else:
-            UnauthorizedError()
-
-    _app_restricted_map = {
-        EndpointOperation.READ: {Permission.READ},
-        EndpointOperation.CREATE: {Permission.CREATE},
-        EndpointOperation.UPDATE: {Permission.UPDATE, Permission.CREATE},
-        EndpointOperation.DELETE: {Permission.DELETE},
-        EndpointOperation.SEARCH: {Permission.SEARCH},
-    }
-
-    def _app_restricted(self, operation: EndpointOperation, aws_event: dict) -> None:
-        allowed = self._parse_permissions(aws_event["headers"])
-        requested = self._app_restricted_map[operation]
-        if not requested.issubset(allowed):
+        
+        if auth_type not in {AuthType.APP_RESTRICTED, AuthType.CIS2, AuthType.NHS_LOGIN}:
             raise UnauthorizedError()
 
-    def _cis2(self, operation: EndpointOperation, aws_event: dict) -> None:
-        # Cis2 works exactly the same as ApplicationRestricted
-        self._app_restricted(operation, aws_event)
-
-    @staticmethod
-    def _parse_permissions(headers) -> Set[Permission]:
-        """Given headers return a set of Permissions. Raises UnknownPermission"""
-
-        content = headers.get(PERMISSIONS_HEADER, "")
-        # comma-separate the Permissions header then trim and finally convert to lowercase
-        parsed = [str.strip(str.lower(s)) for s in content.split(",")]
-
-        permissions = set()
-        for s in parsed:
-            try:
-                permissions.add(Permission(s))
-            except ValueError:
-                raise UnknownPermission()
-
-        return permissions
-
     @staticmethod
     def _parse_auth_type(headers) -> AuthType:
         try:
@@ -112,14 +49,13 @@ def _parse_auth_type(headers) -> AuthType:
             #  we raise UnknownPermission in case of an error and not UnauthorizedError
             raise UnknownPermission()
 
-
-def authorize(operation: EndpointOperation):
+def authorize():
     def decorator(func):
         auth = Authorization()
 
         @wraps(func)
         def wrapper(controller_instance, a):
-            auth.authorize(operation, a)
+            auth.authorize(a)
             return func(controller_instance, a)
 
         return wrapper

backend/src/create_imms_handler.py

@@ -3,7 +3,7 @@
 import pprint
 import uuid
 
-from authorization import Permission
+
 from fhir_controller import FhirController, make_controller
 from local_lambda import load_string
 from models.errors import Severity, Code, create_operation_outcome
@@ -42,7 +42,6 @@ def create_immunization(event, controller: FhirController):
         "headers": {
             "Content-Type": "application/x-www-form-urlencoded",
             "AuthenticationType": "ApplicationRestricted",
-            "Permissions": (",".join([Permission.CREATE])),
         },
     }
 

backend/src/delete_imms_handler.py

@@ -3,7 +3,7 @@
 import pprint
 import uuid
 
-from authorization import Permission
+
 from fhir_controller import FhirController, make_controller
 from models.errors import Severity, Code, create_operation_outcome
 from log_structure import function_info
@@ -40,8 +40,7 @@ def delete_immunization(event, controller: FhirController):
         "pathParameters": {"id": args.id},
         "headers": {
             "Content-Type": "application/x-www-form-urlencoded",
-            "AuthenticationType": "ApplicationRestricted",
-            "Permissions": (",".join([Permission.DELETE])),
+            "AuthenticationType": "ApplicationRestricted"
         },
     }
     pprint.pprint(delete_imms_handler(event, {}))

backend/src/fhir_controller.py

@@ -14,7 +14,7 @@
 from fhir.resources.R4B.immunization import Immunization
 from boto3 import client as boto3_client
 
-from authorization import Authorization, EndpointOperation, UnknownPermission
+from authorization import Authorization, UnknownPermission
 from cache import Cache
 from fhir_repository import ImmunizationRepository, create_table
 from fhir_service import FhirService, UpdateOutcome, get_service_url
@@ -80,7 +80,7 @@ def __init__(
     def get_immunization_by_identifier(self, aws_event) -> dict:
         try:
             if aws_event.get("headers"):
-                if response := self.authorize_request(EndpointOperation.SEARCH, aws_event):
+                if response := self.authorize_request(aws_event):
                     return response
                 query_params = aws_event.get("queryStringParameters", {})
             else:
@@ -121,7 +121,7 @@ def get_immunization_by_identifier(self, aws_event) -> dict:
             return self.create_response(403, unauthorized.to_operation_outcome())
 
     def get_immunization_by_id(self, aws_event) -> dict:
-        if response := self.authorize_request(EndpointOperation.READ, aws_event):
+        if response := self.authorize_request(aws_event):
             return response
 
         imms_id = aws_event["pathParameters"]["id"]
@@ -166,7 +166,7 @@ def get_immunization_by_id(self, aws_event) -> dict:
     def create_immunization(self, aws_event):
         try:
             if aws_event.get("headers"):
-                if response := self.authorize_request(EndpointOperation.CREATE, aws_event):
+                if response := self.authorize_request(aws_event):
                         return response
             else:
                 raise UnauthorizedError()
@@ -210,7 +210,7 @@ def create_immunization(self, aws_event):
     def update_immunization(self, aws_event):
         try:
             if aws_event.get("headers"):
-                if response := self.authorize_request(EndpointOperation.UPDATE, aws_event):
+                if response := self.authorize_request(aws_event):
                     return response
                 imms_id = aws_event["pathParameters"]["id"]
             else:
@@ -372,7 +372,7 @@ def update_immunization(self, aws_event):
     def delete_immunization(self, aws_event):
         try:
             if aws_event.get("headers"):
-                if response := self.authorize_request(EndpointOperation.DELETE, aws_event):
+                if response := self.authorize_request(aws_event):
                         return response
                 imms_id = aws_event["pathParameters"]["id"]
             else:
@@ -403,7 +403,7 @@ def delete_immunization(self, aws_event):
             return self.create_response(403, unauthorized.to_operation_outcome())
 
     def search_immunizations(self, aws_event: APIGatewayProxyEventV1) -> dict:
-        if response := self.authorize_request(EndpointOperation.SEARCH, aws_event):
+        if response := self.authorize_request(aws_event):
             return response
 
         try:
@@ -430,7 +430,9 @@ def search_immunizations(self, aws_event: APIGatewayProxyEventV1) -> dict:
         try:
             checker = VaccinePermissionChecker(imms_vax_type_perms)
             vax_type_perms = checker.expanded_permissions
-            vax_type_perm = self._new_vaccine_request(search_params.immunization_targets, "search", vax_type_perms)
+            operation_code = VaccinePermissionChecker.mapped_operations.get("search")
+            vax_type_perm = [ vaccine_type for vaccine_type in search_params.immunization_targets 
+                             if f"{vaccine_type.lower()}.{operation_code}" in vax_type_perms ]
             if not vax_type_perm:
                 raise UnauthorizedVaxError
         except UnauthorizedVaxError as unauthorized:
@@ -541,19 +543,19 @@ def _create_bad_request(self, message):
         )
         return self.create_response(400, error)
 
-    def authorize_request(self, operation: EndpointOperation, aws_event: dict) -> Optional[dict]:
+    
+    def authorize_request(self, aws_event: dict) -> Optional[dict]:
         try:
-            self.authorizer.authorize(operation, aws_event)
+            self.authorizer.authorize(aws_event)
         except UnauthorizedError as e:
             return self.create_response(403, e.to_operation_outcome())
         except UnknownPermission:
-            # TODO: I think when AuthenticationType is not present, then we don't get below message. Double check again
             id_error = create_operation_outcome(
-                resource_id=str(uuid.uuid4()),
-                severity=Severity.error,
-                code=Code.server_error,
-                diagnostics="application includes invalid authorization values",
-            )
+            resource_id=str(uuid.uuid4()),
+            severity=Severity.error,
+            code=Code.server_error,
+            diagnostics="Application includes invalid authorization values",
+        )
             return self.create_response(500, id_error)
 
     def fetch_identifier_system_and_element(self, event: dict):
@@ -663,37 +665,6 @@ def create_response(status_code, body=None, headers=None):
             **({"body": body} if body else {}),
         }
 
-    @staticmethod
-    def _vaccine_permission(vaccine_type, operation) -> set:
-        
-        operation = mappedOperation.mapped_operations.get(operation.lower())
-        if not operation:
-            raise ValueError(f"Unsupported operation: {operation}")
-
-        vaccine_permission = set()
-        if isinstance(vaccine_type, list):
-            for x in vaccine_type:
-                vaccine_permission.add(str.lower(f"{x}.{operation}"))
-            return vaccine_permission
-        else:
-            vaccine_permission.add(str.lower(f"{vaccine_type}.{operation}"))
-            return vaccine_permission
-
-    @staticmethod
-    def _new_vaccine_request(vaccine_type, operation, vaccine_type_permissions: None) -> Optional[list]:
-
-        operation = VaccinePermissionChecker.mapped_operations.get(operation.lower())
-        vaccine_permission = list()
-        if isinstance(vaccine_type, list):
-            for x in vaccine_type:
-                vaccs_prms = set()
-                vaccs_prms.add(str.lower(f"{x}.{operation}"))
-                if vaccs_prms.issubset(vaccine_type_permissions):
-                    vaccine_permission.append(x)
-            return vaccine_permission
-        else:
-            return vaccine_permission
-
     @staticmethod
     def _identify_supplier_system(aws_event):
         supplier_system = aws_event["headers"]["SupplierSystem"]

backend/src/fhir_service.py

@@ -58,7 +58,7 @@ def __init__(
         self.validator = validator
 
     def get_immunization_by_identifier(
-        self, identifier_pk: str, imms_vax_type_perms: str, identifier: str, element: str
+        self, identifier_pk: str, imms_vax_type_perms: list[str], identifier: str, element: str
     ) -> Optional[dict]:
         """
         Get an Immunization by its ID. Return None if not found. If the patient doesn't have an NHS number,
@@ -76,7 +76,7 @@ def get_immunization_by_identifier(
             response = form_json(imms_resp, element, identifier, base_url)
             return response
 
-    def get_immunization_by_id(self, imms_id: str, imms_vax_type_perms: str) -> Optional[dict]:
+    def get_immunization_by_id(self, imms_id: str, imms_vax_type_perms: list[str]) -> Optional[dict]:
         """
         Get an Immunization by its ID. Return None if it is not found. If the patient doesn't have an NHS number,
         return the Immunization without calling PDS or checking S flag.
@@ -155,7 +155,7 @@ def reinstate_immunization(
         imms_id: str,
         immunization: dict,
         existing_resource_version: int,
-        imms_vax_type_perms: str,
+        imms_vax_type_perms: list[str],
         supplier_system: str,
     ) -> tuple[UpdateOutcome, Immunization]:
         immunization["id"] = imms_id
@@ -178,7 +178,7 @@ def update_reinstated_immunization(
         imms_id: str,
         immunization: dict,
         existing_resource_version: int,
-        imms_vax_type_perms: str,
+        imms_vax_type_perms: list[str],
         supplier_system: str,
     ) -> tuple[UpdateOutcome, Immunization]:
         immunization["id"] = imms_id

backend/src/get_imms_handler.py

@@ -3,7 +3,7 @@
 import pprint
 import uuid
 
-from authorization import Permission
+
 from fhir_controller import FhirController, make_controller
 from models.errors import Severity, Code, create_operation_outcome
 from log_structure import function_info
@@ -41,7 +41,6 @@ def get_immunization_by_id(event, controller: FhirController):
         "headers": {
             "Content-Type": "application/x-www-form-urlencoded",
             "AuthenticationType": "ApplicationRestricted",
-            "Permissions": (",".join([Permission.READ])),
         },
     }
     pprint.pprint(get_imms_handler(event, {}))

backend/src/models/errors.py

@@ -233,16 +233,17 @@ def __str__(self):
         return self.message
 
 
-@dataclass
 class UnauthorizedSystemError(RuntimeError):
-    @staticmethod
-    def to_operation_outcome() -> dict:
-        msg = f"Unauthorized system"
+    def __init__(self, message="Unauthorized system"):
+        super().__init__(message)
+        self.message = message
+    
+    def to_operation_outcome(self) -> dict:
         return create_operation_outcome(
             resource_id=str(uuid.uuid4()),
             severity=Severity.error,
             code=Code.forbidden,
-            diagnostics=msg,
+            diagnostics=self.message,
         )