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

# Conflicts:
#	backend/tests/test_fhir_controller.py

Author: nhsdevws

backend/src/fhir_controller.py

@@ -35,7 +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 models.utils.permission_checker import ApiOperationCode, validate_permissions, _expand_permissions
 from pds_service import PdsService
 from parameter_parser import process_params, process_search_params, create_query_string
 import urllib.parse
@@ -271,11 +271,8 @@ def update_immunization(self, aws_event):
         # Validate if the imms resource does not exist - end
 
         # Check vaccine type permissions on the existing record - start
-        try:
-            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())
+        if not validate_permissions(imms_vax_type_perms, ApiOperationCode.UPDATE, [existing_record["VaccineType"]]):
+            return self.create_response(403, UnauthorizedVaxOnRecordError().to_operation_outcome())
         # Check vaccine type permissions on the existing record - end
 
         existing_resource_version = int(existing_record["Version"])
@@ -428,11 +425,15 @@ 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:
-            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 ]
+            expanded_permissions = _expand_permissions(imms_vax_type_perms)
+            vax_type_perm = [
+                vaccine_type
+                for vaccine_type in search_params.immunization_targets
+                if ApiOperationCode.SEARCH in expanded_permissions.get(vaccine_type.lower(), [])
+            ]
+            # vax_type_perms = _expand_permissions(imms_vax_type_perms, ApiOperationCode.SEARCH)
+            # vax_type_perm = [ vaccine_type for vaccine_type in search_params.immunization_targets 
+            #                  if f"{vaccine_type.lower()}.{ApiOperationCode.SEARCH}" in vax_type_perms ]
             if not vax_type_perm:
                 raise UnauthorizedVaxError
         except UnauthorizedVaxError as unauthorized:

backend/src/fhir_repository.py

@@ -8,7 +8,7 @@
 import boto3
 import botocore.exceptions
 from boto3.dynamodb.conditions import Attr, Key
-from models.utils.permission_checker import VaccinePermissionChecker
+from models.utils.permission_checker import ApiOperationCode, validate_permissions
 from botocore.config import Config
 from models.errors import (
     ResourceNotFoundError,
@@ -95,8 +95,8 @@ def get_immunization_by_identifier(
             item = response["Items"][0]
             resp = dict()
             vaccine_type = self._vaccine_type(item["PatientSK"])
-            checker = VaccinePermissionChecker(imms_vax_type_perms)
-            checker.validate(vaccine_type, "search")
+            if not validate_permissions(imms_vax_type_perms,ApiOperationCode.SEARCH, [vaccine_type]):
+                raise UnauthorizedVaxError()
             resource = json.loads(item["Resource"])
             resp["id"] = resource.get("id")
             resp["version"] = int(response["Items"][0]["Version"])
@@ -112,17 +112,17 @@ def get_immunization_by_id(self, imms_id: str, imms_vax_type_perms: str) -> Opti
             if "DeletedAt" in response["Item"]:
                 if response["Item"]["DeletedAt"] == "reinstated":
                     vaccine_type = self._vaccine_type(response["Item"]["PatientSK"])
-                    checker = VaccinePermissionChecker(imms_vax_type_perms)
-                    checker.validate(vaccine_type, "read")
+                    if not validate_permissions(imms_vax_type_perms,ApiOperationCode.READ, [vaccine_type]):
+                        raise UnauthorizedVaxError()
                     resp["Resource"] = json.loads(response["Item"]["Resource"])
                     resp["Version"] = response["Item"]["Version"]
                     return resp
                 else:
                     return None
             else:
                 vaccine_type = self._vaccine_type(response["Item"]["PatientSK"])
-                checker = VaccinePermissionChecker(imms_vax_type_perms)
-                checker.validate(vaccine_type, "read")
+                if not validate_permissions(imms_vax_type_perms,ApiOperationCode.READ, [vaccine_type]):
+                    raise UnauthorizedVaxError()
                 resp["Resource"] = json.loads(response["Item"]["Resource"])
                 resp["Version"] = response["Item"]["Version"]
                 return resp
@@ -168,8 +168,8 @@ def create_immunization(
         new_id = str(uuid.uuid4())
         immunization["id"] = new_id
         attr = RecordAttributes(immunization, patient)
-        checker = VaccinePermissionChecker(imms_vax_type_perms)
-        checker.validate(attr.vaccine_type, "create")
+        if not validate_permissions(imms_vax_type_perms,ApiOperationCode.CREATE, [attr.vaccine_type]):
+            raise UnauthorizedVaxError()
         query_response = _query_identifier(self.table, "IdentifierGSI", "IdentifierPK", attr.identifier)
 
         if query_response is not None:
@@ -270,8 +270,7 @@ def update_reinstated_immunization(
         )
 
     def _handle_permissions(self, imms_vax_type_perms: list[str], attr: RecordAttributes):
-        checker = VaccinePermissionChecker(imms_vax_type_perms)
-        checker.validate(attr.vaccine_type, "update")
+        validate_permissions(imms_vax_type_perms, ApiOperationCode.UPDATE, [attr.vaccine_type])
 
     def _build_update_expression(self, is_reinstate: bool) -> str:
         if is_reinstate:
@@ -367,8 +366,8 @@ def delete_immunization(
                     if resp["Item"]["DeletedAt"] == "reinstated":
                         pass
                 vaccine_type = self._vaccine_type(resp["Item"]["PatientSK"])
-                checker = VaccinePermissionChecker(imms_vax_type_perms)
-                checker.validate(vaccine_type, "delete")
+                if not validate_permissions(imms_vax_type_perms, ApiOperationCode.DELETE, [vaccine_type]):
+                    raise UnauthorizedVaxError()
 
             response = self.table.update_item(
                 Key={"PK": _make_immunization_pk(imms_id)},

backend/src/models/utils/permission_checker.py

@@ -1,61 +1,29 @@
-import json
-from clients import redis_client
-from models.errors import UnauthorizedVaxOnRecordError
-
-
-class VaccinePermissionChecker:
-    """Centralized vaccine permission checker."""
-
-    mapped_operations = {
-        "create": "c",
-        "read": "r",
-        "update": "u",
-        "delete": "d",
-        "search": "s"
-    }
-
-    def __init__(self, supplier_system: str):
-        self.supplier_permissions = supplier_system
-        self.expanded_permissions = self._expand_permissions(self.supplier_permissions)
-    
-    # Expand permissions from the supplier's permission list
-    @staticmethod    
-    def _expand_permissions(supplier_permissions: list[str]) -> set[str]:
-        expanded = set()
-        for permissions in supplier_permissions:
-            if '.' not in permissions:
-                continue  # skip invalid format
-        vaccine_type, allowed_operations = permissions.split('.', 1)
+from enum import StrEnum
+
+class ApiOperationCode(StrEnum):
+    CREATE = "c"
+    READ = "r"
+    UPDATE = "u"
+    DELETE = "d"
+    SEARCH = "s"
+
+def _expand_permissions(permissions: list[str]) -> dict[str, list[ApiOperationCode]]:
+    expanded_permissions = {}
+    for permission in permissions:
+        vaccine_type, operation_codes_str = permission.split(".", maxsplit=1)
         vaccine_type = vaccine_type.lower()
-        for operation in allowed_operations.lower():
-            if operation not in {'c', 'r', 'u', 'd', 's'}:
-                raise ValueError(f"Unknown operation code: {operation} in a permission {permissions}")
-            expanded.add(f"{vaccine_type}.{operation}")
-        return expanded
-
-    # Check if the requested permission is a subset of the expanded permissions
-    def _vaccine_permission(self, vaccine_type, operation) -> set:
-        
-        operation = self.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
-
-    # Check if the requested permission is allowed
-    def _check_permission(self, requested: set[str]) -> None:
-        if not requested.issubset(self.expanded_permissions):
-            raise UnauthorizedVaxOnRecordError()
-        return None
-
-    # Validate the requested vaccine type and operation against the supplier permissions
-    def validate(self, vaccine_type, operation) -> None:
-        requested_perm = self._vaccine_permission(vaccine_type, operation)
-        self._check_permission(requested_perm)
+        operation_codes = [
+            operation_code
+            for operation_code in operation_codes_str.lower()
+            if operation_code in list(ApiOperationCode)
+        ]
+        expanded_permissions[vaccine_type] = operation_codes
+    return expanded_permissions
+
+def validate_permissions(permissions: list[str], operation: ApiOperationCode, vaccine_types: list[str]):
+    expanded_permissions = _expand_permissions(permissions)
+    print(f"operation: {operation}, expanded_permissions: {expanded_permissions}, vaccine_types: {vaccine_types}")
+    return all([
+        operation in expanded_permissions.get(vaccine_type.lower(), [])
+        for vaccine_type in vaccine_types
+    ])

backend/tests/sample_data/permissions_config.py

@@ -151,7 +151,7 @@ def test_get_imms_by_identifer_header_missing(self):
         response = self.controller.get_immunization_by_identifier(lambda_event)
 
         self.assertEqual(response["statusCode"], 403)
-    
+
     @patch("fhir_controller.get_supplier_permissions")
     def test_not_found_for_identifier(self, mock_get_permissions):
         """it should return not-found OperationOutcome if it doesn't exist"""
@@ -183,7 +183,7 @@ def test_not_found_for_identifier(self, mock_get_permissions):
 
         imms = identifier.replace("|", "#")
         # When
-        
+
         response = self.controller.get_immunization_by_identifier(lambda_event)
 
         # Then
@@ -218,7 +218,7 @@ def test_get_imms_by_identifer_patient_identifier_and_element_present(self, mock
         self.assertEqual(response["statusCode"], 400)
         body = json.loads(response["body"])
         self.assertEqual(body["resourceType"], "OperationOutcome")
-    
+
     @patch("fhir_controller.get_supplier_permissions")
     def test_get_imms_by_identifer_both_body_and_query_params_present(self, mock_get_supplier_permissions):
         """it should return Immunization Id if it exists"""
@@ -440,7 +440,7 @@ def test_validate_immunization_identifier_having_whitespace(self,mock_get_suppli
         self.assertEqual(response["statusCode"], 400)
         outcome = json.loads(response["body"])
         self.assertEqual(outcome["resourceType"], "OperationOutcome")
-    
+
     @patch("fhir_controller.get_supplier_permissions")
     def test_validate_imms_id_invalid_vaccinetype(self, mock_get_permissions):
         """it should validate lambda's Immunization id"""
@@ -756,7 +756,7 @@ def test_validate_immunization_identifier_having_whitespace(self,mock_get_suppli
         self.assertEqual(response["statusCode"], 400)
         outcome = json.loads(response["body"])
         self.assertEqual(outcome["resourceType"], "OperationOutcome")
-   
+
     @patch("fhir_controller.get_supplier_permissions")
     def test_validate_imms_id_invalid_vaccinetype(self, mock_get_permissions):
         """it should validate lambda's Immunization id"""
@@ -838,7 +838,7 @@ def test_get_imms_by_id_unauthorised_vax_error(self,mock_permissions):
         # Then
         mock_permissions.assert_called_once_with("test")
         self.assertEqual(response["statusCode"], 403)
-    
+
     @patch("fhir_controller.get_supplier_permissions")
     def test_get_imms_by_id_no_vax_permission(self, mock_permissions):
         """it should return Immunization Id if it exists"""
@@ -901,7 +901,7 @@ def setUp(self):
     @patch("fhir_controller.get_supplier_permissions")
     def test_create_immunization(self,mock_get_permissions):
         """it should create Immunization and return resource's location"""
-        mock_get_permissions.return_value = ["COVID19.CRUDS"]
+        mock_get_permissions.return_value = ["COVID19.CRUDS", "FLU.CRUDS"]
         imms_id = str(uuid.uuid4())
         imms = create_covid_19_immunization(imms_id)
         aws_event = {
@@ -914,7 +914,7 @@ def test_create_immunization(self,mock_get_permissions):
 
         imms_obj = json.loads(aws_event["body"])
         mock_get_permissions.assert_called_once_with("Test")
-        self.service.create_immunization.assert_called_once_with(imms_obj, ["COVID19.CRUDS"], "Test")
+        self.service.create_immunization.assert_called_once_with(imms_obj, ["COVID19.CRUDS", "FLU.CRUDS"], "Test")
         self.assertEqual(response["statusCode"], 201)
         self.assertTrue("body" not in response)
         self.assertTrue(response["headers"]["Location"].endswith(f"Immunization/{imms_id}"))
@@ -1110,7 +1110,7 @@ def test_update_immunization_etag_missing(self, mock_get_supplier_permissions):
     @patch("fhir_controller.get_supplier_permissions")
     def test_update_immunization_duplicate(self, mock_get_supplier_permissions):
         """it should not update the Immunization record"""
-        mock_get_supplier_permissions.return_value = ["Covid19.U"]
+        mock_get_supplier_permissions.return_value = ["COVID19.U"]
         # Given
         imms_id = "valid-id"
         imms = {"id": "valid-id"}
@@ -1160,11 +1160,11 @@ def test_update_immunization_UnauthorizedVaxError(self, mock_get_supplier_permis
         response = self.controller.update_immunization(aws_event)
         mock_get_supplier_permissions.assert_called_once_with("Test")
         self.assertEqual(response["statusCode"], 403)
-    
+
     @patch("fhir_controller.get_supplier_permissions")
     def test_update_immunization_UnauthorizedVaxError_check_for_non_batch(self, mock_get_supplier_permissions):
         """it should not update the Immunization record"""
-        mock_get_supplier_permissions.return_value = ["COVID19.U"]
+        mock_get_supplier_permissions.return_value = ["COVID19.CRDS"]
         imms_id = "valid-id"
         imms = {"id": "valid-id"}
         aws_event = {
@@ -1591,7 +1591,7 @@ def test_immunization_exception_not_found(self, mock_get_permissions):
         body = json.loads(response["body"])
         self.assertEqual(body["resourceType"], "OperationOutcome")
         self.assertEqual(body["issue"][0]["code"], "not-found")
-    
+
     @patch("fhir_controller.get_supplier_permissions")
     def test_immunization_unhandled_error(self, mock_get_supplier_permissions):
         """it should return server-error OperationOutcome if service throws UnhandledResponseError"""
@@ -1635,7 +1635,7 @@ def tearDown(self):
     def test_get_search_immunizations(self, mock_get_supplier_permissions):
         """it should search based on patient_identifier and immunization_target"""
 
-        mock_get_supplier_permissions.return_value = ["covid19.s"]
+        mock_get_supplier_permissions.return_value = ["COVID19.S"]
         search_result = Bundle.construct()
         self.service.search_immunizations.return_value = search_result
 
@@ -1818,7 +1818,7 @@ def test_post_search_immunizations(self,mock_get_supplier_permissions):
             "headers": {"Content-Type": "application/x-www-form-urlencoded", "SupplierSystem": "Test"},
             "body": base64_encoded_body,
         }
-        
+
         # When
         response = self.controller.search_immunizations(lambda_event)
         # Then
@@ -1999,7 +1999,7 @@ def test_search_immunizations_returns_200_remove_vaccine_not_done(self, mock_get
         "This method should return 200 but remove the data which has status as not done."
         search_result = load_json_data("sample_immunization_response _for _not_done_event.json")
         bundle = Bundle.parse_obj(search_result)
-        mock_get_supplier_permissions.return_value = ["covid19.cruds"]
+        mock_get_supplier_permissions.return_value = ["COVID19.CRUDS"]
         self.service.search_immunizations.return_value = bundle
         vaccine_type = "COVID19"
         lambda_event = {