Create basic authoriser class

Author: dlzhry2nhs

backend/src/authorisation/ApiOperationCode.py

@@ -0,0 +1,9 @@
+from enum import StrEnum
+
+
+class ApiOperationCode(StrEnum):
+    CREATE = "c"
+    READ = "r"
+    UPDATE = "u"
+    DELETE = "d"
+    SEARCH = "s"

backend/src/authorisation/Authoriser.py

@@ -0,0 +1,51 @@
+import json
+
+from authorisation.ApiOperationCode import ApiOperationCode
+from clients import redis_client, logger
+from constants import SUPPLIER_PERMISSIONS_HASH_KEY
+
+
+class Authoriser:
+    def __init__(self):
+        self._cache_client = redis_client
+
+    @staticmethod
+    def _expand_permissions(permissions: list[str]) -> dict[str, list[ApiOperationCode]]:
+        """Parses and expands permissions data into a dictionary mapping vaccination types to a list of permitted
+        API operations. The raw string from Redis will be in the form VAC.PERMS e.g. COVID19.CRUDS"""
+        expanded_permissions = {}
+
+        for permission in permissions:
+            vaccine_type, operation_codes_str = permission.split(".", maxsplit=1)
+            vaccine_type = vaccine_type.lower()
+            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 _get_supplier_permissions(self, supplier_name: str) -> dict[str, list[ApiOperationCode]]:
+        raw_permissions_data = self._cache_client.hget(SUPPLIER_PERMISSIONS_HASH_KEY, supplier_name)
+        permissions_data = json.loads(raw_permissions_data) if raw_permissions_data else []
+
+        return self._expand_permissions(permissions_data)
+
+    def authorise(
+        self,
+        supplier_name: str,
+        requested_operation: ApiOperationCode,
+        vaccination_types: set[str]
+    ) -> bool:
+        supplier_permissions = self._get_supplier_permissions(supplier_name)
+
+        logger.info(
+            f"operation: {requested_operation}, supplier_permissions: {supplier_permissions}, "
+            f"vaccine_types: {vaccination_types}"
+        )
+        return all(
+            requested_operation in supplier_permissions.get(vaccination_type.lower(), [])
+            for vaccination_type in vaccination_types
+        )

backend/src/authorisation/__init__.py

@@ -1,12 +1,11 @@
 from dataclasses import dataclass
 from enum import Enum
 from functools import wraps
-from typing import Set
 
 from models.errors import UnauthorizedError
 
 
-AUTHENTICATION_HEADER = "AuthenticationType"
+AUTHENTICATION_TYPE_HEADER_NAME = "AuthenticationType"
 
 
 @dataclass
@@ -32,17 +31,17 @@ 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, aws_event: dict):
         auth_type = self._parse_auth_type(aws_event["headers"])
-        
+
         if auth_type not in {AuthType.APP_RESTRICTED, AuthType.CIS2, AuthType.NHS_LOGIN}:
             raise UnauthorizedError()
 
     @staticmethod
-    def _parse_auth_type(headers) -> AuthType:
+    def _parse_auth_type(headers: dict) -> AuthType:
         try:
-            auth_type = headers[AUTHENTICATION_HEADER]
+            auth_type = headers[AUTHENTICATION_TYPE_HEADER_NAME]
             return AuthType(auth_type)
         except ValueError:
             # The value of authentication type comes from apigee regardless of auth type. That's why

backend/src/authorization.py

@@ -23,3 +23,4 @@ class Urls:
 
 
 GENERIC_SERVER_ERROR_DIAGNOSTICS_MESSAGE = "Unable to process request. Issue may be transient."
+SUPPLIER_PERMISSIONS_HASH_KEY = "supplier_permissions"

backend/src/constants.py

@@ -1,12 +1,10 @@
 import base64
-import boto3
 import json
 import os
 import re
 import uuid
 from decimal import Decimal
 from typing import Optional
-import boto3
 from aws_lambda_typing.events import APIGatewayProxyEventV1
 from fhir.resources.R4B.immunization import Immunization
 from boto3 import client as boto3_client
@@ -25,7 +23,6 @@
     ValidationError,
     IdentifierDuplicationError,
     ParameterException,
-    InconsistentIdError,
     UnauthorizedVaxError,
     UnauthorizedVaxOnRecordError,
     UnauthorizedSystemError,
@@ -101,7 +98,7 @@ def get_immunization_by_identifier(self, aws_event) -> dict:
 
         try:
             if resource := self.fhir_service.get_immunization_by_identifier(
-                identifiers, imms_vax_type_perms, identifier, element):
+                identifiers, supplier_system, identifier, element):
                 return FhirController.create_response(200, resource)
         except UnauthorizedVaxError as unauthorized:
             return self.create_response(403, unauthorized.to_operation_outcome())

backend/src/fhir_controller.py

@@ -1,4 +1,3 @@
-from urllib import response
 from responses import logger
 import simplejson as json
 import os
@@ -87,24 +86,21 @@ class ImmunizationRepository:
     def __init__(self, table: Table):
         self.table = table
 
-    def get_immunization_by_identifier(
-        self, identifier_pk: str, imms_vax_type_perms: list[str]
-    ) -> Optional[dict]:
+    def get_immunization_by_identifier(self, identifier_pk: str) -> tuple[Optional[dict], Optional[str]]:
         response = self.table.query(
             IndexName="IdentifierGSI", KeyConditionExpression=Key("IdentifierPK").eq(identifier_pk)
         )
+
         if "Items" in response and len(response["Items"]) > 0:
             item = response["Items"][0]
             resp = dict()
             vaccine_type = self._vaccine_type(item["PatientSK"])
-            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"])
-            return resp
+            return resp, vaccine_type
         else:
-            return None
+            return None, None
 
     def get_immunization_by_id(self, imms_id: str, imms_vax_type_perms: str) -> Optional[dict]:
         response = self.table.get_item(Key={"PK": _make_immunization_pk(imms_id)})
@@ -401,7 +397,7 @@ def find_immunizations(self, patient_identifier: str, vaccine_types: list):
 
         raw_items = self.get_all_items(condition, is_not_deleted)
 
-        if raw_items:    
+        if raw_items:
             # Filter the response to contain only the requested vaccine types
             items = [x for x in raw_items if x["PatientSK"].split("#")[0] in vaccine_types]
 
@@ -430,7 +426,7 @@ def get_all_items(self, condition, is_not_deleted):
             response = self.table.query(**query_args)
             if "Items" not in response:
                 raise UnhandledResponseError(message="No Items in DynamoDB response", response=response)
-  
+
             items = response.get("Items", [])
             all_items.extend(items)
 

backend/src/fhir_repository.py

@@ -15,13 +15,12 @@
 from pydantic import ValidationError
 
 import parameter_parser
+from authorisation.ApiOperationCode import ApiOperationCode
+from authorisation.Authoriser import Authoriser
 from fhir_repository import ImmunizationRepository
-from base_utils.base_utils import obtain_field_value
-from models.field_names import FieldNames
-from models.errors import InvalidPatientId, CustomValidationError, UnhandledResponseError
+from models.errors import InvalidPatientId, CustomValidationError, UnauthorizedVaxError
 from models.fhir_immunization import ImmunizationValidator
 from models.utils.generic_utils import nhs_number_mod11_check, get_occurrence_datetime, create_diagnostics, form_json, get_contained_patient
-from models.constants import Constants
 from models.errors import MandatoryError
 from timer import timed
 from filter import Filter
@@ -54,27 +53,27 @@ def __init__(
         imms_repo: ImmunizationRepository,
         validator: ImmunizationValidator = ImmunizationValidator(),
     ):
+        self.authoriser = Authoriser()
         self.immunization_repo = imms_repo
         self.validator = validator
 
     def get_immunization_by_identifier(
-        self, identifier_pk: str, imms_vax_type_perms: list[str], identifier: str, element: str
+        self, identifier_pk: str, supplier_name: 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,
         return the Immunization without calling PDS or checking S flag.
         """
-        imms_resp = self.immunization_repo.get_immunization_by_identifier(
-            identifier_pk, imms_vax_type_perms
-        )
+        base_url = f"{get_service_url()}/Immunization"
+        imms_resp, vaccination_type = self.immunization_repo.get_immunization_by_identifier(identifier_pk)
+
         if not imms_resp:
-            base_url = f"{get_service_url()}/Immunization"
-            response = form_json(imms_resp, None, None, base_url)
-            return response
-        else:
-            base_url = f"{get_service_url()}/Immunization"
-            response = form_json(imms_resp, element, identifier, base_url)
-            return response
+            return form_json(imms_resp, None, None, base_url)
+
+        if not self.authoriser.authorise(supplier_name, ApiOperationCode.SEARCH, {vaccination_type}):
+            raise UnauthorizedVaxError
+
+        return form_json(imms_resp, element, identifier, base_url)
 
     def get_immunization_by_id(self, imms_id: str, imms_vax_type_perms: list[str]) -> Optional[dict]:
         """

backend/src/fhir_service.py

@@ -2,7 +2,7 @@
 from authorization import (
     Authorization,
     UnknownPermission,
-    AUTHENTICATION_HEADER,
+    AUTHENTICATION_TYPE_HEADER_NAME,
     authorize,
     AuthType
 )
@@ -12,7 +12,7 @@
 def _make_aws_event(auth_type: AuthType):
     return {
         "headers": {
-            AUTHENTICATION_HEADER: auth_type.value
+            AUTHENTICATION_TYPE_HEADER_NAME: auth_type.value
         }
     }
 
@@ -35,7 +35,7 @@ def test_decorator_unauthorized(self):
         controller = TestAuthorizeDecorator.StubController()
         aws_event = {
             "headers": {
-                AUTHENTICATION_HEADER: "InvalidType"
+                AUTHENTICATION_TYPE_HEADER_NAME: "InvalidType"
             }
         }
         with self.assertRaises(UnknownPermission):
@@ -59,7 +59,7 @@ def test_valid_auth_types(self):
     def test_unknown_authorization(self):
         aws_event = {
             "headers": {
-                AUTHENTICATION_HEADER: "unknown auth type"
+                AUTHENTICATION_TYPE_HEADER_NAME: "unknown auth type"
             }
         }
         with self.assertRaises(UnknownPermission):
@@ -80,7 +80,7 @@ def test_valid_app_restricted_auth(self):
     def test_invalid_auth_type_raises_unknown_permission(self):
         aws_event = {
             "headers": {
-                AUTHENTICATION_HEADER: "InvalidAuthType"
+                AUTHENTICATION_TYPE_HEADER_NAME: "InvalidAuthType"
             }
         }
         with self.assertRaises(UnknownPermission):
@@ -101,8 +101,8 @@ def test_valid_cis2_auth(self):
     def test_invalid_auth_type_raises_unknown_permission(self):
         aws_event = {
             "headers": {
-                AUTHENTICATION_HEADER: "InvalidAuthType"
+                AUTHENTICATION_TYPE_HEADER_NAME: "InvalidAuthType"
             }
         }
         with self.assertRaises(UnknownPermission):
-            self.authorization.authorize(aws_event)
+            self.authorization.authorize(aws_event)

backend/tests/test_authorization.py

@@ -9,7 +9,7 @@
     Authorization,
     UnknownPermission,
     AuthType,
-    AUTHENTICATION_HEADER,
+    AUTHENTICATION_TYPE_HEADER_NAME,
 )
 from fhir_controller import FhirController
 from fhir_repository import ImmunizationRepository
@@ -20,7 +20,7 @@
 
 
 def make_aws_event(auth_type: AuthType, permissions=None) -> dict:
-    return {"headers": {AUTHENTICATION_HEADER: str(auth_type)}}
+    return {"headers": {AUTHENTICATION_TYPE_HEADER_NAME: str(auth_type)}}
 
 
 class TestFhirControllerAuthorization(unittest.TestCase):
@@ -36,10 +36,10 @@ def setUp(self):
         self.controller = FhirController(self.authorizer, self.service)
         self.logger_info_patcher = patch("logging.Logger.info")
         self.mock_logger_info = self.logger_info_patcher.start()
-        
+
     def tearDown(self):
         patch.stopall()
-        
+
     def test_get_imms_by_id_authorized(self):
         aws_event = {"pathParameters": {"id": "an-id"}}
 
@@ -111,20 +111,20 @@ def test_update_imms_authorized(self, mock_get_supplier_permissions):
         _ = self.controller.update_immunization(aws_event)
 
         self.authorizer.authorize.assert_called_once_with(aws_event)
-    
+
     @patch("fhir_controller.get_supplier_permissions")
     def test_update_imms_unauthorized_vaxx_in_record(self,mock_get_supplier_permissions):
         mock_get_supplier_permissions.return_value = ["Covid19.CRUDS"]
         imms_id = str(uuid.uuid4())
         aws_event = {"headers": {"E-Tag":1, "SupplierSystem" : "Test"},"pathParameters": {"id": imms_id}, "body": create_covid_19_immunization(imms_id).json()}
         self.service.get_immunization_by_id_all.return_value = {"resource":"new_value","Version":1,"DeletedAt": False, "VaccineType":"Flu"}
-        
+
         response = self.controller.update_immunization(aws_event)
         self.assertEqual(response["statusCode"], 403)
         body = json.loads(response["body"])
         self.assertEqual(body["resourceType"], "OperationOutcome")
-        self.assertEqual(body["issue"][0]["code"], "forbidden")            
-            
+        self.assertEqual(body["issue"][0]["code"], "forbidden")
+
         self.authorizer.authorize.assert_called_once_with(aws_event)
 
     def test_update_imms_unauthorized(self):
@@ -209,4 +209,4 @@ def test_search_imms_unknown_permission(self):
         self.assertEqual(response["statusCode"], 500)
         body = json.loads(response["body"])
         self.assertEqual(body["resourceType"], "OperationOutcome")
-        self.assertEqual(body["issue"][0]["code"], "exception")
+        self.assertEqual(body["issue"][0]["code"], "exception")