[ELI-688] - Adding Kinesis Data Stream as a buffer for Firehose

infrastructure/modules/kinesis_firehose/kinesis_firehose_delivery_stream.tf

@@ -2,6 +2,11 @@ resource "aws_kinesis_firehose_delivery_stream" "eligibility_audit_firehose_deli
   name        = "${terraform.workspace == "default" ? "" : "${terraform.workspace}-"}${var.project_name}-${var.environment}-${var.audit_firehose_delivery_stream_name}"
   destination = "extended_s3"
 
+  kinesis_source_configuration {
+    kinesis_stream_arn = var.kinesis_source_stream_arn
+    role_arn           = var.audit_firehose_role.arn
+  }
+
   extended_s3_configuration {
     role_arn   = var.audit_firehose_role.arn
     bucket_arn = var.s3_audit_bucket_arn
@@ -14,19 +19,14 @@ resource "aws_kinesis_firehose_delivery_stream" "eligibility_audit_firehose_deli
 
     cloudwatch_logging_options {
       enabled         = true
-      log_group_name  = var.kinesis_cloud_watch_log_group_name
-      log_stream_name = var.kinesis_cloud_watch_log_stream
+      log_group_name  = var.firehose_cloud_watch_log_group_name
+      log_stream_name = var.firehose_cloud_watch_log_stream
     }
   }
 
-  server_side_encryption {
-    enabled  = true
-    key_arn  = aws_kms_key.firehose_cmk.arn
-    key_type = "CUSTOMER_MANAGED_CMK"
-  }
-
   depends_on = [
     aws_kms_key.firehose_cmk,
+    var.kinesis_source_stream_arn,
     var.audit_firehose_role
   ]
 

infrastructure/modules/kinesis_firehose/variables.tf

@@ -13,12 +13,12 @@ variable "s3_audit_bucket_arn" {
   type        = string
 }
 
-variable "kinesis_cloud_watch_log_group_name" {
+variable "firehose_cloud_watch_log_group_name" {
   description = "kinesis cloud watch log group name"
   type        = string
 }
 
-variable "kinesis_cloud_watch_log_stream" {
+variable "firehose_cloud_watch_log_stream" {
   description = "kinesis cloud watch log stream"
   type        = string
 }
@@ -28,6 +28,11 @@ variable "eligibility_lambda_role_arn" {
   type        = any
 }
 
+variable "kinesis_source_stream_arn" {
+  description = "the arn of the kinesis data stream the lambda puts records to"
+  type        = any
+}
+
 
 
 

infrastructure/modules/lambda/lambda.tf

@@ -20,7 +20,7 @@ resource "aws_lambda_function" "eligibility_signposting_lambda" {
       PERSON_TABLE_NAME            = var.eligibility_status_table_name,
       RULES_BUCKET_NAME            = var.eligibility_rules_bucket_name,
       CONSUMER_MAPPING_BUCKET_NAME = var.eligibility_consumer_mappings_bucket_name,
-      KINESIS_AUDIT_STREAM_TO_S3   = var.kinesis_audit_stream_to_s3_name
+      KINESIS_AUDIT_STREAM         = var.kinesis_audit_stream_name
       ENV                          = var.environment
       LOG_LEVEL                    = var.log_level
       ENABLE_XRAY_PATCHING         = var.enable_xray_patching

infrastructure/modules/lambda/variables.tf

@@ -54,7 +54,7 @@ variable "eligibility_status_table_name" {
   type        = string
 }
 
-variable "kinesis_audit_stream_to_s3_name" {
+variable "kinesis_audit_stream_name" {
   description = "kinesis audit stream to s3 name"
   type        = string
 }

infrastructure/stacks/api-layer/iam_policies.tf

@@ -256,6 +256,109 @@ resource "aws_iam_role_policy_attachment" "lambda_insights_policy" {
   policy_arn = "arn:aws:iam::aws:policy/CloudWatchLambdaInsightsExecutionRolePolicy"
 }
 
+# Policy document to read from Kinesis Source stream
+data "aws_iam_policy_document" "kinesis_source_access" {
+  statement {
+    sid    = "AllowReadFromKinesisSourceStream"
+    effect = "Allow"
+
+    actions = [
+      "kinesis:DescribeStream",
+      "kinesis:DescribeStreamSummary",
+      "kinesis:GetRecords",
+      "kinesis:GetShardIterator",
+      "kinesis:ListShards",
+      "kinesis:SubscribeToShard",
+    ]
+
+    resources = [
+      aws_kinesis_stream.kinesis_source_stream.arn,
+    ]
+  }
+}
+
+# Policy document to use KMS key for reading from Kinesis Source stream
+data "aws_iam_policy_document" "kinesis_source_kms_read_access" {
+  statement {
+    sid    = "AllowUseOfKinesisSourceKeyForReads"
+    effect = "Allow"
+
+    actions = [
+      "kms:Decrypt",
+      "kms:GenerateDataKey",
+      "kms:DescribeKey"
+    ]
+
+    resources = [
+      aws_kms_key.kinesis_data_stream_kms_key.arn
+    ]
+  }
+}
+
+# Attach kinesis read policy to firehose role
+resource "aws_iam_role_policy" "kinesis_firehose_read_policy" {
+  name   = "KinesisSourceReadAccess"
+  role   = aws_iam_role.eligibility_audit_firehose_role.id
+  policy = data.aws_iam_policy_document.kinesis_source_access.json
+}
+
+# Attach kinesis source stream KMS read policy to firehose role
+resource "aws_iam_role_policy" "firehose_kinesis_source_kms_policy" {
+  name   = "KinesisSourceKmsReadAccess"
+  role   = aws_iam_role.eligibility_audit_firehose_role.id
+  policy = data.aws_iam_policy_document.kinesis_source_kms_read_access.json
+}
+
+# Policy document for Lambda to write to Kinesis stream
+data "aws_iam_policy_document" "kinesis_write_access" {
+  statement {
+    sid    = "AllowWriteToKinesisStream"
+    effect = "Allow"
+
+    actions = [
+      "kinesis:PutRecord",
+      "kinesis:PutRecords"
+    ]
+
+    resources = [
+      aws_kinesis_stream.kinesis_source_stream.arn
+    ]
+  }
+}
+
+# Policy document to use the KMS key
+data "aws_iam_policy_document" "kinesis_kms_write_access" {
+  statement {
+    sid    = "AllowUseOfKinesisStreamKeyForWrites"
+    effect = "Allow"
+
+    actions = [
+      "kms:Encrypt",
+      "kms:GenerateDataKey",
+      "kms:DescribeKey"
+    ]
+
+    resources = [
+      aws_kms_key.kinesis_data_stream_kms_key.arn
+    ]
+  }
+}
+
+
+# Attach kinesis write policy to Lambda role
+resource "aws_iam_role_policy" "lambda_kinesis_write_policy" {
+  name   = "KinesisWriteAccess"
+  role   = aws_iam_role.eligibility_lambda_role.id
+  policy = data.aws_iam_policy_document.kinesis_write_access.json
+}
+
+# Attach kinesis KMS access policy to Lambda role
+resource "aws_iam_role_policy" "lambda_kinesis_kms_policy" {
+  name   = "KinesisStreamKmsWriteAccess"
+  role   = aws_iam_role.eligibility_lambda_role.id
+  policy = data.aws_iam_policy_document.kinesis_kms_write_access.json
+}
+
 # Policy doc for S3 Audit bucket
 data "aws_iam_policy_document" "s3_audit_bucket_policy" {
   statement {

infrastructure/stacks/api-layer/kinesis_data_stream.tf

@@ -0,0 +1,82 @@
+resource "aws_kms_key" "kinesis_data_stream_kms_key" {
+  description             = "${terraform.workspace == "default" ? "" : "${terraform.workspace}-"} kinesis_data_stream_kms Master Key"
+  deletion_window_in_days = 14
+  is_enabled              = true
+  enable_key_rotation     = true
+}
+
+resource "aws_kms_alias" "kinesis_data_stream_kms_key" {
+  name          = "alias/${var.project_name}-${var.environment}-kinesis-audit-stream"
+  target_key_id = aws_kms_key.kinesis_data_stream_kms_key.key_id
+}
+
+
+data "aws_iam_policy_document" "kinesis_stream_kms_key_policy" {
+  statement {
+    sid    = "EnableRootPermissions"
+    effect = "Allow"
+
+    principals {
+      type        = "AWS"
+      identifiers = ["arn:aws:iam::${data.aws_caller_identity.current.account_id}:root"]
+    }
+
+    actions   = ["kms:*"]
+    resources = ["*"]
+  }
+
+  statement {
+    sid    = "AllowLambdaUseOfKey"
+    effect = "Allow"
+
+    principals {
+      type        = "AWS"
+      identifiers = [aws_iam_role.eligibility_lambda_role.arn]
+    }
+
+    actions = [
+      "kms:Encrypt",
+      "kms:Decrypt",
+      "kms:ReEncrypt*",
+      "kms:GenerateDataKey*",
+      "kms:DescribeKey"
+    ]
+
+    resources = ["*"]
+  }
+
+  statement {
+    sid    = "AllowFirehoseRoleUseOfKey"
+    effect = "Allow"
+
+    principals {
+      type        = "AWS"
+      identifiers = [aws_iam_role.eligibility_audit_firehose_role.arn]
+    }
+
+    actions = [
+      "kms:Decrypt",
+      "kms:GenerateDataKey*",
+      "kms:DescribeKey"
+    ]
+
+    resources = ["*"]
+  }
+}
+
+resource "aws_kms_key_policy" "kinesis_stream_kms_key_policy" {
+  key_id = aws_kms_key.kinesis_data_stream_kms_key.id
+  policy = data.aws_iam_policy_document.kinesis_stream_kms_key_policy.json
+}
+
+resource "aws_kinesis_stream" "kinesis_source_stream" {
+  name             = "${var.project_name}-${var.environment}-kinesis-audit-stream"
+  retention_period = 24
+
+  stream_mode_details {
+    stream_mode = "ON_DEMAND" # can discuss later
+  }
+
+  encryption_type = "KMS"
+  kms_key_id      = aws_kms_key.kinesis_data_stream_kms_key.arn
+}

infrastructure/stacks/api-layer/kinesis_firehose.tf

@@ -7,7 +7,8 @@ module "eligibility_audit_firehose_delivery_stream" {
   stack_name                          = local.stack_name
   workspace                           = local.workspace
   tags                                = local.tags
-  kinesis_cloud_watch_log_group_name  = aws_cloudwatch_log_group.firehose_audit.name
-  kinesis_cloud_watch_log_stream      = aws_cloudwatch_log_stream.firehose_audit_stream.name
+  firehose_cloud_watch_log_group_name  = aws_cloudwatch_log_group.firehose_audit.name
+  firehose_cloud_watch_log_stream      = aws_cloudwatch_log_stream.firehose_audit_stream.name
   eligibility_lambda_role_arn            = aws_iam_role.eligibility_lambda_role.arn
+  kinesis_source_stream_arn = aws_kinesis_stream.kinesis_source_stream.arn
 }

infrastructure/stacks/api-layer/lambda.tf

@@ -25,7 +25,7 @@ module "eligibility_signposting_lambda_function" {
   eligibility_rules_bucket_name             = module.s3_rules_bucket.storage_bucket_name
   eligibility_consumer_mappings_bucket_name = module.s3_consumer_mappings_bucket.storage_bucket_name
   eligibility_status_table_name             = module.eligibility_status_table.table_name
-  kinesis_audit_stream_to_s3_name           = module.eligibility_audit_firehose_delivery_stream.firehose_stream_name
+  kinesis_audit_stream_name                 = aws_kinesis_stream.kinesis_source_stream.name
   hashing_secret_name                       = module.secrets_manager.aws_hashing_secret_name
   lambda_insights_extension_version         = 38
   log_level                                 = "INFO"

infrastructure/stacks/iams-developer-roles/github_actions_policies.tf

@@ -674,6 +674,41 @@ resource "aws_iam_policy" "firehose_readonly" {
   tags = merge(local.tags, { Name = "firehose-describe-access" })
 }
 
+resource "aws_iam_policy" "kinesis_management" {
+  name        = "kinesis-management"
+  description = "Allow GitHub Actions to manage project Kinesis streams"
+  path        = "/service-policies/"
+
+  policy = jsonencode({
+    Version = "2012-10-17",
+    Statement = [
+      {
+        Effect = "Allow",
+        Action = [
+          "kinesis:CreateStream",
+          "kinesis:DeleteStream",
+          "kinesis:DescribeStream",
+          "kinesis:DescribeStreamSummary",
+          "kinesis:ListStreams",
+          "kinesis:ListTagsForStream",
+          "kinesis:AddTagsToStream",
+          "kinesis:RemoveTagsFromStream",
+          "kinesis:ListShards",
+          "kinesis:IncreaseStreamRetentionPeriod",
+          "kinesis:DecreaseStreamRetentionPeriod",
+          "kinesis:StartStreamEncryption",
+          "kinesis:StopStreamEncryption"
+        ],
+        Resource = [
+          "arn:aws:kinesis:${var.default_aws_region}:${data.aws_caller_identity.current.account_id}:stream/eligibility-signposting-api-*"
+        ]
+      }
+    ]
+  })
+
+  tags = merge(local.tags, { Name = "kinesis-management" })
+}
+
 resource "aws_iam_policy" "cloudwatch_management" {
   #checkov:skip=CKV_AWS_355: GetMetricWidgetImage requires wildcard resource
   #checkov:skip=CKV_AWS_290: GetMetricWidgetImage requires wildcard resource
@@ -788,3 +823,8 @@ resource "aws_iam_role_policy_attachment" "cloudwatch_management" {
   role       = aws_iam_role.github_actions.name
   policy_arn = aws_iam_policy.cloudwatch_management.arn
 }
+
+resource "aws_iam_role_policy_attachment" "kinesis_management_attach" {
+  role       = aws_iam_role.github_actions.name
+  policy_arn = aws_iam_policy.kinesis_management.arn
+}

infrastructure/stacks/iams-developer-roles/iams_permissions_boundary.tf

@@ -86,6 +86,22 @@ data "aws_iam_policy_document" "permissions_boundary" {
       "firehose:StartDeliveryStreamEncryption",
       "firehose:StopDeliveryStreamEncryption",
 
+      # Kinesis Stream - audit log streaming
+      "kinesis:CreateStream",
+      "kinesis:DeleteStream",
+      "kinesis:DescribeStream",
+      "kinesis:ListStreams",
+      "kinesis:PutRecord",
+      "kinesis:PutRecords",
+      "kinesis:TagStream",
+      "kinesis:ListTagsForStream",
+      "kinesis:UntagStream",
+      "kinesis:GetShardIterator",
+      "kinesis:GetRecords",
+      "kinesis:ListShards",
+      "kinesis:SubscribeToShard",
+      "kinesis:DescribeStreamSummary",
+
       # IAM - specific role and policy management
       "iam:GetRole*",
       "iam:GetPolicy*",

src/eligibility_signposting_api/audit/audit_context.py

@@ -140,7 +140,7 @@ def add_response_details(response_id: UUID, last_updated: datetime) -> None:
         g.audit_log.response.last_updated = last_updated
 
     @staticmethod
-    def write_to_firehose(service: AuditService) -> None:
+    def write_audit_record(service: AuditService) -> None:
         service.audit(g.audit_log.model_dump(by_alias=True))
 
     @staticmethod