VED-26: Use configurable supplier permissions in the API. (#606)

* setup

* replace vaccineTypePermissions

* remove imms-batch-app

* test for new perms and imms batch removel

* test still

* test3

* test_for_search

* test for update

* test for update2

* test suite complete

* remove perm config in test

* poetry

* remove unused files

* env vars

* VED-26: Attach endpoint Lambda functions to the VPC so they can connect to Redis.

* VED-26: Give endpoint Lambdas the required permissions to attach to the VPC.

* VED-26: Try creating a fresh client for every invocation.

* VED-26: Src again :(

* VED-26: Revert Redis client change. Temporarily skip PDS callout.

* VED-26: Revert PDS callout removal.

* integrating the new supplier permissions

* clean up

* clean up2

* test suite

* lint issues

* refactor control flow and duplication error

* e2e fix

* e2e lint

* refactor some logic from review sections

* refactor from review section

* lint issues

* update

* sanity check

* VED-26: Attach Lambda functions and ECS tasks to private subnets only.

* VED-26: Resolve Sonar warning. Add pre-deploy check for private subnets.

* final review

* final review2

---------

Co-authored-by: Matt Jarvis <[email protected]>

Author: Akol125

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

@@ -1,11 +1,24 @@
-"""Initialise s3, kinesis and lambda clients"""
+"""Initialise s3, kinesis, lambda and redis clients"""
 
 from boto3 import client as boto3_client
+import os
+import logging
+import redis
 
-REGION_NAME = "eu-west-2"
+REGION_NAME = os.getenv("AWS_REGION", "eu-west-2")
 
 s3_client = boto3_client("s3", region_name=REGION_NAME)
 kinesis_client = boto3_client("kinesis", region_name=REGION_NAME)
 lambda_client = boto3_client("lambda", region_name=REGION_NAME)
 firehose_client = boto3_client("firehose", region_name=REGION_NAME)
 sqs_client = boto3_client("sqs", region_name=REGION_NAME)
+
+REDIS_HOST = os.getenv("REDIS_HOST", "")
+REDIS_PORT = int(os.getenv("REDIS_PORT", 6379))
+
+
+logging.basicConfig(level="INFO")
+logger = logging.getLogger()
+logger.info(f"Connecting to Redis at {REDIS_HOST}:{REDIS_PORT}")
+
+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, {}))