Merge remote-tracking branch 'origin/VED-26-permissions-api' into VED-382-backend-configurable-disease-mapping

# Conflicts:
#	backend/src/clients.py
#	backend/src/fhir_controller.py
#	backend/src/fhir_repository.py
#	backend/src/fhir_service.py
#	backend/src/mappings.py
#	backend/src/models/utils/permissions.py
#	backend/tests/test_fhir_controller.py
#	backend/tests/test_fhir_controller_authorization.py
#	backend/tests/test_fhir_repository.py
#	backend/tests/test_fhir_service.py
#	backend/tests/utils/test_permissions.py

Author: nhsdevws

backend/poetry.lock

@@ -12,6 +12,7 @@ python                  = "~3.11"
 boto3                   = "~1.38.42"
 boto3-stubs-lite        = {extras = ["dynamodb"], version = "~1.38.42"}
 aws-lambda-typing       = "~2.20.0"
+redis                   = "^4.6.0"
 moto                    = "^5.1.6"
 requests                = "~2.32.4"
 responses               = "~0.25.7"

backend/pyproject.toml

@@ -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/authorization.py

@@ -19,7 +19,6 @@
 
 logging.basicConfig(level="INFO")
 logger = logging.getLogger()
-logger.setLevel("INFO")
 logger.info(f"Connecting to Redis at {REDIS_HOST}:{REDIS_PORT}")
 
-redis_client = redis.StrictRedis(host=REDIS_HOST, port=REDIS_PORT, decode_responses=True)
+redis_client = redis.StrictRedis(host=REDIS_HOST, port=REDIS_PORT, decode_responses=True)

backend/src/clients.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/create_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/delete_imms_handler.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
@@ -35,6 +35,7 @@
 )
 from models.utils.generic_utils import check_keys_in_sources
 from models.utils.permissions import get_supplier_permissions
+from models.utils.permission_checker import VaccinePermissionChecker
 from pds_service import PdsService
 from parameter_parser import process_params, process_search_params, create_query_string
 import urllib.parse
@@ -79,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:
@@ -120,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"]
@@ -165,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()
@@ -209,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:
@@ -271,9 +272,8 @@ def update_immunization(self, aws_event):
 
         # Check vaccine type permissions on the existing record - start
         try:
-            vax_type_perms = self._parse_vaccine_permissions_controller(imms_vax_type_perms)
-            vax_type_perm = self._vaccine_permission(existing_record["VaccineType"], "update")
-            self._check_permission(vax_type_perm, vax_type_perms)
+            checker = VaccinePermissionChecker(imms_vax_type_perms)
+            checker.validate(existing_record["VaccineType"], "update")
         except UnauthorizedVaxOnRecordError as unauthorized:
             return self.create_response(403, unauthorized.to_operation_outcome())
         # Check vaccine type permissions on the existing record - end
@@ -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:
@@ -428,8 +428,11 @@ def search_immunizations(self, aws_event: APIGatewayProxyEventV1) -> dict:
             return self.create_response(403, unauthorized.to_operation_outcome())
         # Check vaxx type permissions on the existing record - start
         try:
-            vax_type_perms = self._parse_vaccine_permissions_controller(imms_vax_type_perms)
-            vax_type_perm = self._new_vaccine_request(search_params.immunization_targets, "search", vax_type_perms)
+            checker = VaccinePermissionChecker(imms_vax_type_perms)
+            vax_type_perms = checker.expanded_permissions
+            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:
@@ -540,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):
@@ -633,6 +636,7 @@ def check_vaccine_type_permissions(self, aws_event):
             if len(supplier_system) == 0:
                 raise UnauthorizedSystemError()
             imms_vax_type_perms = get_supplier_permissions(supplier_system)
+            print(f" update imms = {imms_vax_type_perms}")
             if len(imms_vax_type_perms) == 0:
                 raise UnauthorizedVaxError()
             # Return the values needed for later use
@@ -661,50 +665,6 @@ def create_response(status_code, body=None, headers=None):
             **({"body": body} if body else {}),
         }
 
-    @staticmethod
-    def _sendack(payload, file_name, message_id, created_at_formatted_string, local_id, operation_requested):
-        payload["file_key"] = file_name
-        payload["row_id"] = message_id
-        payload["created_at_formatted_string"] = created_at_formatted_string
-        payload["local_id"] = local_id
-        payload["operation_requested"] = operation_requested
-        sqs_client.send_message(QueueUrl=queue_url, MessageBody=json.dumps(payload), MessageGroupId=file_name)
-
-    @staticmethod
-    def _vaccine_permission(vaccine_type, operation) -> set:
-        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 _parse_vaccine_permissions_controller(imms_vax_type_perms) -> set:
-        return {str(s).strip().lower() for s in imms_vax_type_perms}
-
-    @staticmethod
-    def _check_permission(requested: set, allowed: set) -> set:
-        if not requested.issubset(allowed):
-            raise UnauthorizedVaxOnRecordError()
-        else:
-            return None
-
-    @staticmethod
-    def _new_vaccine_request(vaccine_type, operation, vaccine_type_permissions: None) -> Optional[list]:
-        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"]