update

Author: Akol125

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

@@ -883,7 +883,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 = {
@@ -896,7 +896,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}"))
@@ -1092,7 +1092,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"}
@@ -1146,7 +1146,7 @@ def test_update_immunization_UnauthorizedVaxError(self, mock_get_supplier_permis
     @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 = {
@@ -1161,7 +1161,7 @@ def test_update_immunization_UnauthorizedVaxError_check_for_non_batch(self, mock
     @patch("fhir_controller.get_supplier_permissions")
     def test_update_immunization_Unauthorizedsystem_check_for_non_batch(self, mock_get_supplier_permissions):
         """it should not update the Immunization record"""
-        mock_get_supplier_permissions.return_value = ["COVID19.CRUD"]
+        mock_get_supplier_permissions.return_value = ["COVID19.CRD"]
         imms_id = "valid-id"
         imms = {"id": "valid-id"}
         aws_event = {
@@ -1612,7 +1612,7 @@ def setUp(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
 
@@ -1975,7 +1975,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 = {

backend/tests/test_fhir_controller.py

@@ -79,7 +79,7 @@ def test_unauthorized_get_immunization_by_identifier(self):
                 ]
             }
         )
-        with self.assertRaises(UnauthorizedVaxOnRecordError) as e:
+        with self.assertRaises(UnauthorizedVaxError) as e:
             # When
             self.repository.get_immunization_by_identifier(imms_id, ["FLU.CRUD"])
 
@@ -112,7 +112,7 @@ def test_get_immunization_by_id(self):
                 }
             }
         )
-        imms = self.repository.get_immunization_by_id(imms_id, ["COVID19.CRUD"])
+        imms = self.repository.get_immunization_by_id(imms_id, ["COVID19.CRUDS"])
 
         # Validate the results
         self.assertDictEqual(resource, imms)
@@ -133,7 +133,7 @@ def test_unauthorized_get_immunization_by_id(self):
                 }
             }
         )
-        with self.assertRaises(UnauthorizedVaxOnRecordError) as e:
+        with self.assertRaises(UnauthorizedVaxError) as e:
             # When
             self.repository.get_immunization_by_id(imms_id, ["FLU.CRUD"])
 
@@ -328,10 +328,22 @@ def test_create_patient_with_unauthorised_vaccine_type_permissions(self):
         """Patient record should not be created"""
         imms = create_covid_19_immunization_dict("an-id")
 
+        self.repository.table.query.return_value = {
+            "Count": 0,
+            "Items": []
+        }
+
+        self.repository.table.put_item.return_value = {
+            "ResponseMetadata": {
+                "HTTPStatusCode": 200
+            }
+        }
+
         update_target_disease_code(imms, DiseaseCodes.flu)
-        with self.assertRaises(UnauthorizedVaxOnRecordError) as e:
+        with self.assertRaises(UnauthorizedVaxError) as e:
             # When
-            self.repository.create_immunization(imms, self.patient, ["COVID.CRUD"], "Test")
+            self.repository.create_immunization(imms, self.patient, ["COVID19.CRUD"], "Test")
+            
 
 
 class TestUpdateImmunization(unittest.TestCase):
@@ -515,7 +527,16 @@ def test_unauthorised_vax_delete(self):
             }
         )
 
-        with self.assertRaises(UnauthorizedVaxOnRecordError) as e:
+        self.repository.table.update_item.return_value = {
+        "ResponseMetadata": {
+            "HTTPStatusCode": 200
+        },
+        "Attributes": {
+            "Resource": json.dumps({"id": "valid-id", "status": "deleted"})
+        }
+    }
+
+        with self.assertRaises(UnauthorizedVaxError) as e:
             self.repository.delete_immunization(imms_id, ["COVID19.CRUD"], "Test")
 
     def test_multiple_delete_should_not_update_timestamp(self):