VED-901-Auth: authorization for extended attribute (#1017)

Author: Akol125

infrastructure/instance/file_name_processor.tf

@@ -3,8 +3,18 @@ locals {
   filename_lambda_dir     = abspath("${path.root}/../../lambdas/filenameprocessor")
   filename_lambda_files   = fileset(local.filename_lambda_dir, "**")
   filename_lambda_dir_sha = sha1(join("", [for f in local.filename_lambda_files : filesha1("${local.filename_lambda_dir}/${f}")]))
+  dps_bucket_name_for_extended_attribute = (
+    var.environment == "prod"
+    ? "nhsd-dspp-core-prod-extended-attributes-gdp"
+    : "nhsd-dspp-core-ref-extended-attributes-gdp"
+  )
+  dps_bucket_arn_for_extended_attribute = [
+    "arn:aws:s3:::${local.dps_bucket_name_for_extended_attribute}/*"
+  ]
 }
 
+
+
 resource "aws_ecr_repository" "file_name_processor_lambda_repository" {
   image_scanning_configuration {
     scan_on_push = true
@@ -162,6 +172,13 @@ resource "aws_iam_policy" "filenameprocessor_lambda_exec_policy" {
           "firehose:PutRecordBatch"
         ],
         "Resource" : "arn:aws:firehose:*:*:deliverystream/${module.splunk.firehose_stream_name}"
+      },
+      {
+        "Effect" : "Allow",
+        "Action" : [
+          "s3:PutObject"
+        ],
+        "Resource" : local.dps_bucket_arn_for_extended_attribute
       }
     ]
   })
@@ -278,8 +295,10 @@ resource "aws_lambda_function" "file_processor_lambda" {
   environment {
     variables = {
       ACCOUNT_ID           = var.immunisation_account_id
+      DPS_ACCOUNT_ID       = var.dspp_core_account_id
       SOURCE_BUCKET_NAME   = aws_s3_bucket.batch_data_source_bucket.bucket
       ACK_BUCKET_NAME      = aws_s3_bucket.batch_data_destination_bucket.bucket
+      DPS_BUCKET_NAME      = local.dps_bucket_name_for_extended_attribute
       QUEUE_URL            = aws_sqs_queue.batch_file_created.url
       REDIS_HOST           = data.aws_elasticache_cluster.existing_redis.cache_nodes[0].address
       REDIS_PORT           = data.aws_elasticache_cluster.existing_redis.cache_nodes[0].port

infrastructure/instance/s3_config.tf

@@ -84,7 +84,18 @@ resource "aws_s3_bucket_lifecycle_configuration" "datasources_lifecycle" {
     filter {
       prefix = "archive/"
     }
+    expiration {
+      days = 7
+    }
+  }
 
+  rule {
+    id     = "DeleteExtendedAttributesFilesAfter7Days"
+    status = "Enabled"
+
+    filter {
+      prefix = "extended-attributes-archive/"
+    }
     expiration {
       days = 7
     }

lambdas/filenameprocessor/src/constants.py

@@ -12,19 +12,24 @@
 
 SOURCE_BUCKET_NAME = os.getenv("SOURCE_BUCKET_NAME")
 
-# We have used an internal temporary bucket here and an actual dps bucket will replace this
-DPS_DESTINATION_BUCKET_NAME = os.getenv("ACK_BUCKET_NAME")
-EXPECTED_BUCKET_OWNER_ACCOUNT = os.getenv("ACCOUNT_ID")
+
+DPS_DESTINATION_BUCKET_NAME = os.getenv("DPS_BUCKET_NAME")
+EXPECTED_SOURCE_BUCKET_ACCOUNT = os.getenv("ACCOUNT_ID")
+EXPECTED_DPS_DESTINATION_ACCOUNT = os.getenv("DPS_ACCOUNT_ID")
 AUDIT_TABLE_NAME = os.getenv("AUDIT_TABLE_NAME")
 AUDIT_TABLE_TTL_DAYS = os.getenv("AUDIT_TABLE_TTL_DAYS")
 VALID_VERSIONS = ["V5"]
 
 VACCINE_TYPE_TO_DISEASES_HASH_KEY = "vacc_to_diseases"
 ODS_CODE_TO_SUPPLIER_SYSTEM_HASH_KEY = "ods_code_to_supplier"
 EXTENDED_ATTRIBUTES_FILE_PREFIX = "Vaccination_Extended_Attributes"
+
+# Currently only COVID extended attributes files are supported, might be extended in future for other vaccine types
 EXTENDED_ATTRIBUTES_VACC_TYPE = "COVID"
-DPS_DESTINATION_PREFIX = "dps_destination/"
 
+DPS_DESTINATION_PREFIX = "dps_destination"
+EXTENDED_ATTRIBUTES_ARCHIVE_PREFIX = "extended-attributes-archive"
+VALID_EA_VERSIONS = ["V1_5"]
 ERROR_TYPE_TO_STATUS_CODE_MAP = {
     VaccineTypePermissionsError: 403,
     InvalidFileKeyError: 400,  # Includes invalid ODS code, therefore unable to identify supplier
@@ -61,3 +66,9 @@ class AuditTableKeys(StrEnum):
     TIMESTAMP = "timestamp"
     EXPIRES_AT = "expires_at"
     ERROR_DETAILS = "error_details"
+
+
+class Operation(str):
+    CREATE = "C"
+    UPDATE = "U"
+    DELETE = "D"

lambdas/filenameprocessor/src/file_name_processor.py

@@ -12,18 +12,19 @@
 from audit_table import upsert_audit_table
 from common.aws_s3_utils import (
     copy_file_to_external_bucket,
-    delete_file,
     move_file,
 )
 from common.clients import STREAM_NAME, get_s3_client, logger
 from common.log_decorator import logging_decorator
 from common.models.errors import UnhandledAuditTableError
 from constants import (
     DPS_DESTINATION_BUCKET_NAME,
+    DPS_DESTINATION_PREFIX,
     ERROR_TYPE_TO_STATUS_CODE_MAP,
-    EXPECTED_BUCKET_OWNER_ACCOUNT,
+    EXPECTED_DPS_DESTINATION_ACCOUNT,
+    EXPECTED_SOURCE_BUCKET_ACCOUNT,
+    EXTENDED_ATTRIBUTES_ARCHIVE_PREFIX,
     EXTENDED_ATTRIBUTES_FILE_PREFIX,
-    EXTENDED_ATTRIBUTES_VACC_TYPE,
     SOURCE_BUCKET_NAME,
     FileNotProcessedReason,
     FileStatus,
@@ -36,7 +37,7 @@
     VaccineTypePermissionsError,
 )
 from send_sqs_message import make_and_send_sqs_message
-from supplier_permissions import validate_vaccine_type_permissions
+from supplier_permissions import validate_permissions_for_extended_attributes_files, validate_vaccine_type_permissions
 from utils_for_filenameprocessor import get_creation_and_expiry_times
 
 
@@ -107,8 +108,8 @@ def handle_unexpected_bucket_name(bucket_name: str, file_key: str) -> dict:
     config and overarching design"""
     try:
         if file_key.startswith(EXTENDED_ATTRIBUTES_FILE_PREFIX):
-            organization_code = validate_extended_attributes_file_key(file_key)
-            extended_attribute_identifier = f"{organization_code}_{EXTENDED_ATTRIBUTES_VACC_TYPE}"
+            vaccine_type, organisation_code = validate_extended_attributes_file_key(file_key)
+            extended_attribute_identifier = f"{organisation_code}_{vaccine_type}"
             logger.error(
                 "Unable to process file %s due to unexpected bucket name %s",
                 file_key,
@@ -249,8 +250,10 @@ def handle_extended_attributes_file(
 
     extended_attribute_identifier = None
     try:
-        organization_code = validate_extended_attributes_file_key(file_key)
-        extended_attribute_identifier = f"{organization_code}_{EXTENDED_ATTRIBUTES_VACC_TYPE}"
+        vaccine_type, organisation_code = validate_extended_attributes_file_key(file_key)
+        extended_attribute_identifier = validate_permissions_for_extended_attributes_files(
+            vaccine_type, organisation_code
+        )
 
         upsert_audit_table(
             message_id,
@@ -262,16 +265,17 @@ def handle_extended_attributes_file(
         )
 
         # TODO: agree the prefix with DPS
-        dest_file_key = f"dps_destination/{file_key}"
+        dest_file_key = f"{DPS_DESTINATION_PREFIX}/{file_key}"
         copy_file_to_external_bucket(
             bucket_name,
             file_key,
             DPS_DESTINATION_BUCKET_NAME,
             dest_file_key,
-            EXPECTED_BUCKET_OWNER_ACCOUNT,
-            EXPECTED_BUCKET_OWNER_ACCOUNT,
+            EXPECTED_DPS_DESTINATION_ACCOUNT,
+            EXPECTED_SOURCE_BUCKET_ACCOUNT,
         )
-        delete_file(bucket_name, dest_file_key, EXPECTED_BUCKET_OWNER_ACCOUNT)
+
+        move_file(bucket_name, file_key, f"{EXTENDED_ATTRIBUTES_ARCHIVE_PREFIX}/{file_key}")
 
         upsert_audit_table(
             message_id,
@@ -305,7 +309,7 @@ def handle_extended_attributes_file(
             extended_attribute_identifier = "unknown"
 
         # Move file to archive
-        move_file(bucket_name, file_key, f"archive/{file_key}")
+        move_file(bucket_name, file_key, f"{EXTENDED_ATTRIBUTES_ARCHIVE_PREFIX}/{file_key}")
 
         upsert_audit_table(
             message_id,

lambdas/filenameprocessor/src/file_validation.py

@@ -3,7 +3,7 @@
 from datetime import datetime
 from re import match
 
-from constants import VALID_VERSIONS
+from constants import EXTENDED_ATTRIBUTES_FILE_PREFIX, EXTENDED_ATTRIBUTES_VACC_TYPE, VALID_EA_VERSIONS, VALID_VERSIONS
 from elasticache import (
     get_supplier_system_from_cache,
     get_valid_vaccine_types_from_cache,
@@ -37,17 +37,36 @@ def is_valid_datetime(timestamp: str) -> bool:
     return True
 
 
-def validate_extended_attributes_file_key(file_key: str) -> str:
+def validate_extended_attributes_file_key(file_key: str) -> tuple[str, str]:
     """
     Checks that all elements of the file key are valid, raises an exception otherwise.
     Returns a string containing the organization code.
     """
     if not match(r"^[^_.]*_[^_.]*_[^_.]*_[^_.]*_[^_.]*_[^_.]*_[^_.]*", file_key):
         raise InvalidFileKeyError("Initial file validation failed: invalid extended attributes file key format")
 
-    file_key_parts_without_extension, _ = split_file_key(file_key)
+    file_key_parts_without_extension, extension = split_file_key(file_key)
+    file_type = "_".join(file_key_parts_without_extension[:3])
+    version = "_".join(file_key_parts_without_extension[3:5])
     organization_code = file_key_parts_without_extension[5]
-    return organization_code
+    timestamp = file_key_parts_without_extension[6]
+    supplier = get_supplier_system_from_cache(organization_code)
+    valid_vaccine_types = get_valid_vaccine_types_from_cache()
+    vaccine_type = EXTENDED_ATTRIBUTES_VACC_TYPE
+
+    if not (
+        vaccine_type in valid_vaccine_types
+        and file_type == EXTENDED_ATTRIBUTES_FILE_PREFIX.upper()
+        and version in VALID_EA_VERSIONS
+        and supplier  # Note that if supplier could be identified, this also implies that ODS code is valid
+        and is_valid_datetime(timestamp)
+        and (
+            (extension == "CSV") or (extension == "DAT") or (extension == "CTL")
+        )  # The DAT extension has been added for MESH file processing
+    ):
+        raise InvalidFileKeyError("Initial file validation failed: invalid file key")
+
+    return vaccine_type, organization_code
 
 
 def validate_batch_file_key(file_key: str) -> tuple[str, str]:

lambdas/filenameprocessor/src/supplier_permissions.py

@@ -1,7 +1,8 @@
 """Functions for fetching supplier permissions"""
 
 from common.clients import logger
-from elasticache import get_supplier_permissions_from_cache
+from constants import Operation
+from elasticache import get_supplier_permissions_from_cache, get_supplier_system_from_cache
 from models.errors import VaccineTypePermissionsError
 
 
@@ -19,3 +20,26 @@ def validate_vaccine_type_permissions(vaccine_type: str, supplier: str) -> list:
         raise VaccineTypePermissionsError(error_message)
 
     return supplier_permissions
+
+
+def validate_permissions_for_extended_attributes_files(vaccine_type: str, ods_code: str) -> str:
+    """
+    Checks that the supplier has COVID vaccine type and its CUD permissions.
+    Raises an exception if the supplier does not have at least one permission for the vaccine type.
+    """
+    allowed_operations = {
+        Operation.CREATE,
+        Operation.UPDATE,
+        Operation.DELETE,
+    }
+    supplier = get_supplier_system_from_cache(ods_code)
+    supplier_permissions = get_supplier_permissions_from_cache(supplier)
+    cached_operations = [
+        permission.split(".")[1] for permission in supplier_permissions if permission.split(".")[0] == vaccine_type
+    ]
+    if not (cached_operations and allowed_operations.issubset(set(cached_operations[0]))):
+        error_message = f"Initial file validation failed: {supplier} does not have permissions for {vaccine_type}"
+        logger.error(error_message)
+        raise VaccineTypePermissionsError(error_message)
+
+    return f"{ods_code}_{vaccine_type}"

lambdas/filenameprocessor/tests/test_file_key_validation.py

@@ -113,8 +113,8 @@ def test_split_file_key(self, _):
             with self.subTest(f"SubTest for file key: {file_key}"):
                 self.assertEqual(split_file_key(file_key), expected)
 
-    def test_validate_extended_attributes_file_key(self, _):
-        """Tests that validate_extended_attributes_file_key returns organization code if all
+    def test_validate_extended_attributes_file_key(self, mock_get_redis_client):
+        """Tests that validate_extended_attributes_file_key returns organization code and COVID vaccine type if all
         elements pass validation, and raises an exception otherwise"""
         test_cases_for_success_scenarios = [
             # Valid extended attributes file key
@@ -131,8 +131,13 @@ def test_validate_extended_attributes_file_key(self, _):
 
         for file_key, expected_result in test_cases_for_success_scenarios:
             with self.subTest(f"SubTest for file key: {file_key}"):
+                mock_redis = Mock()
+                mock_redis.hget.side_effect = create_mock_hget(MOCK_ODS_CODE_TO_SUPPLIER, {})
+                mock_redis.hkeys.return_value = ["COVID"]
+                mock_get_redis_client.return_value = mock_redis
+                _, supplier = validate_extended_attributes_file_key(file_key)
                 self.assertEqual(
-                    validate_extended_attributes_file_key(file_key),
+                    supplier,
                     expected_result,
                 )