feat(forwarder): add source_object_keys parameter

examples/forwarder-s3/README.md

@@ -0,0 +1,74 @@
+# Subscribing objects via EventBridge
+
+This module demonstrates how to forward objects from a source bucket using
+EventBridge to trigger the Forwarder module.
+
+To run this example you need to execute:
+
+```
+$ terraform init
+$ terraform plan
+$ terraform apply
+```
+
+The module will output three source buckets, one for each subscription type.
+It will also output a destination bucket, where objects will be copied to by the forwarder.
+
+```
+→ aws s3 cp ./main.tf s3://`terraform output -raw direct_source`
+upload: ./main.tf to s3://forwarder-s3-bucket-notification-src-20240617155010087500000002/main.tf
+→ aws s3 ls s3://`terraform output -raw destination`
+2024-06-17 08:52:55        869 main.tf
+```
+
+Note that this example may create resources which can cost money. Run terraform destroy when you don't need these resources.
+
+
+<!-- BEGINNING OF PRE-COMMIT-TERRAFORM DOCS HOOK -->
+## Requirements
+
+| Name | Version |
+|------|---------|
+| <a name="requirement_terraform"></a> [terraform](#requirement\_terraform) | >= 1.3 |
+| <a name="requirement_aws"></a> [aws](#requirement\_aws) | >= 5.0 |
+
+## Providers
+
+| Name | Version |
+|------|---------|
+| <a name="provider_aws"></a> [aws](#provider\_aws) | >= 5.0 |
+
+## Modules
+
+| Name | Source | Version |
+|------|--------|---------|
+| <a name="module_forwarder"></a> [forwarder](#module\_forwarder) | ../..//modules/forwarder | n/a |
+
+## Resources
+
+| Name | Type |
+|------|------|
+| [aws_s3_bucket.destination](https://registry.terraform.io/providers/hashicorp/aws/latest/docs/resources/s3_bucket) | resource |
+| [aws_s3_bucket.direct](https://registry.terraform.io/providers/hashicorp/aws/latest/docs/resources/s3_bucket) | resource |
+| [aws_s3_bucket.eventbridge](https://registry.terraform.io/providers/hashicorp/aws/latest/docs/resources/s3_bucket) | resource |
+| [aws_s3_bucket.sns](https://registry.terraform.io/providers/hashicorp/aws/latest/docs/resources/s3_bucket) | resource |
+| [aws_s3_bucket_notification.direct](https://registry.terraform.io/providers/hashicorp/aws/latest/docs/resources/s3_bucket_notification) | resource |
+| [aws_s3_bucket_notification.eventbridge](https://registry.terraform.io/providers/hashicorp/aws/latest/docs/resources/s3_bucket_notification) | resource |
+| [aws_s3_bucket_notification.sns](https://registry.terraform.io/providers/hashicorp/aws/latest/docs/resources/s3_bucket_notification) | resource |
+| [aws_sns_topic.this](https://registry.terraform.io/providers/hashicorp/aws/latest/docs/resources/sns_topic) | resource |
+| [aws_sns_topic_policy.s3_to_sns](https://registry.terraform.io/providers/hashicorp/aws/latest/docs/resources/sns_topic_policy) | resource |
+| [aws_iam_policy_document.s3_to_sns](https://registry.terraform.io/providers/hashicorp/aws/latest/docs/data-sources/iam_policy_document) | data source |
+
+## Inputs
+
+No inputs.
+
+## Outputs
+
+| Name | Description |
+|------|-------------|
+| <a name="output_destination"></a> [destination](#output\_destination) | Destination bucket objects are copied to. |
+| <a name="output_direct_source"></a> [direct\_source](#output\_direct\_source) | Source bucket subscribed directly |
+| <a name="output_eventbridge_source"></a> [eventbridge\_source](#output\_eventbridge\_source) | Source bucket subscribed via eventbridge |
+| <a name="output_sns_source"></a> [sns\_source](#output\_sns\_source) | Source bucket subscribed via SNS |
+<!-- END OF PRE-COMMIT-TERRAFORM DOCS HOOK -->

examples/forwarder-s3/direct.tf

@@ -0,0 +1,12 @@
+resource "aws_s3_bucket" "direct" {
+  bucket_prefix = "${local.name_prefix}direct-"
+  force_destroy = true
+}
+
+resource "aws_s3_bucket_notification" "direct" {
+  bucket = aws_s3_bucket.direct.id
+  queue {
+    queue_arn = module.forwarder.queue_arn
+    events    = ["s3:ObjectCreated:*"]
+  }
+}

examples/forwarder-s3/eventbridge.tf

@@ -0,0 +1,9 @@
+resource "aws_s3_bucket" "eventbridge" {
+  bucket_prefix = "${local.name_prefix}eventbridge-"
+  force_destroy = true
+}
+
+resource "aws_s3_bucket_notification" "eventbridge" {
+  bucket      = aws_s3_bucket.eventbridge.id
+  eventbridge = true
+}

examples/forwarder-s3/main.tf

@@ -0,0 +1,26 @@
+locals {
+  name        = basename(abspath(path.root))
+  name_prefix = "${local.name}-"
+}
+
+# we'll write data to this bucket
+resource "aws_s3_bucket" "destination" {
+  bucket_prefix = "${local.name_prefix}dst-"
+  force_destroy = true
+}
+
+module "forwarder" {
+  # Prefer using the hashicorp registry:
+  # source = "observeinc/collection/aws//modules/forwarder"
+  # For validation purposes we will instead refer to a local version of the
+  # module:
+  source = "../..//modules/forwarder"
+
+  name = local.name
+  destination = {
+    bucket = aws_s3_bucket.destination.id
+    prefix = ""
+  }
+  source_bucket_names = ["${local.name_prefix}*"]
+  source_object_keys  = ["*"]
+}

examples/forwarder-s3/outputs.tf

@@ -0,0 +1,20 @@
+output "direct_source" {
+  description = "Source bucket subscribed directly"
+  value       = aws_s3_bucket.direct.id
+}
+
+output "sns_source" {
+  description = "Source bucket subscribed via SNS"
+  value       = aws_s3_bucket.sns.id
+}
+
+output "eventbridge_source" {
+  description = "Source bucket subscribed via eventbridge"
+  value       = aws_s3_bucket.eventbridge.id
+}
+
+
+output "destination" {
+  description = "Destination bucket objects are copied to."
+  value       = aws_s3_bucket.destination.id
+}

examples/forwarder-s3/sns.tf

@@ -0,0 +1,34 @@
+resource "aws_s3_bucket" "sns" {
+  bucket_prefix = "${local.name_prefix}sns-"
+  force_destroy = true
+}
+
+resource "aws_sns_topic" "this" {
+  name_prefix = local.name_prefix
+}
+
+data "aws_iam_policy_document" "s3_to_sns" {
+  statement {
+    actions   = ["SNS:Publish"]
+    resources = [aws_sns_topic.this.arn]
+    principals {
+      type        = "Service"
+      identifiers = ["s3.amazonaws.com"]
+    }
+  }
+}
+
+resource "aws_sns_topic_policy" "s3_to_sns" {
+  arn    = aws_sns_topic.this.arn
+  policy = data.aws_iam_policy_document.s3_to_sns.json
+}
+
+resource "aws_s3_bucket_notification" "sns" {
+  bucket = aws_s3_bucket.sns.id
+  topic {
+    topic_arn = aws_sns_topic.this.arn
+    events    = ["s3:ObjectCreated:*"]
+  }
+
+  depends_on = [aws_sns_topic_policy.s3_to_sns]
+}

examples/forwarder-s3/tests/example.tftest.hcl

@@ -0,0 +1,2 @@
+# only verifies module can be installed and removed correctly
+run "install" {}

examples/forwarder-s3/versions.tf

@@ -0,0 +1,10 @@
+terraform {
+  required_version = ">= 1.3"
+
+  required_providers {
+    aws = {
+      source  = "hashicorp/aws"
+      version = ">= 5.0"
+    }
+  }
+}

modules/forwarder/README.md

@@ -118,6 +118,7 @@ module "forwarder" {
 | <a name="input_sam_release_version"></a> [sam\_release\_version](#input\_sam\_release\_version) | Release version for SAM apps as defined on github.com/observeinc/aws-sam-apps. | `string` | `""` | no |
 | <a name="input_source_bucket_names"></a> [source\_bucket\_names](#input\_source\_bucket\_names) | A list of bucket names which the forwarder is allowed to read from.  This<br>list only affects permissions, and supports wildcards. In order to have<br>files copied to Filedrop, you must also subscribe S3 Bucket Notifications<br>to the forwarder. | `list(string)` | `[]` | no |
 | <a name="input_source_kms_key_arns"></a> [source\_kms\_key\_arns](#input\_source\_kms\_key\_arns) | A list of KMS Key ARNs the forwarder is allowed to use to decrypt objects in S3. | `list(string)` | `[]` | no |
+| <a name="input_source_object_keys"></a> [source\_object\_keys](#input\_source\_object\_keys) | A list of object key patterns the forwarder is allowed to read from for<br>provided source buckets. | `list(string)` | <pre>[<br>  "*"<br>]</pre> | no |
 | <a name="input_source_topic_arns"></a> [source\_topic\_arns](#input\_source\_topic\_arns) | A list of SNS topics the forwarder is allowed to be subscribed to. | `list(string)` | `[]` | no |
 | <a name="input_verbosity"></a> [verbosity](#input\_verbosity) | Logging verbosity for Lambda. Highest log verbosity is 9. | `number` | `1` | no |
 

modules/forwarder/eventbridge.tf

@@ -17,6 +17,9 @@ resource "aws_cloudwatch_event_rule" "this" {
       # list must have elements, so we introduce an empty match
       for name in concat([""], var.source_bucket_names) : { "wildcard" : name }
     ],
+    "detail.object.key" = [
+      for key in concat([""], var.source_object_keys) : { "wildcard" : key }
+    ],
   })
 }
 

modules/forwarder/iam.tf

@@ -80,7 +80,11 @@ data "aws_iam_policy_document" "reader" {
       "s3:GetObject",
       "s3:GetObjectTagging",
     ]
-    resources = [for name in var.source_bucket_names : "arn:aws:s3:::${name}/*"]
+    # NOTE: this is a deviation from our CloudFormation template. In terraform
+    # we can compute the product of two lists, allowing us to be strict as to what buckets and keys we grant the function read access for. 
+    resources = [
+      for pair in setproduct(var.source_bucket_names, var.source_object_keys) : "arn:aws:s3:::${pair[0]}/${pair[1]}"
+    ]
   }
 }
 

modules/forwarder/lambda.tf

@@ -15,6 +15,7 @@ resource "aws_lambda_function" "this" {
       MAX_FILE_SIZE               = var.max_file_size != null ? var.max_file_size : local.default_limits.max_file_size
       CONTENT_TYPE_OVERRIDES      = join(",", [for o in var.content_type_overrides : "${o["pattern"]}=${o["content_type"]}"])
       SOURCE_BUCKET_NAMES         = join(",", var.source_bucket_names)
+      SOURCE_OBJECT_KEYS          = join(",", var.source_object_keys)
       OTEL_EXPORTER_OTLP_ENDPOINT = var.debug_endpoint
       OTEL_TRACES_EXPORTER        = var.debug_endpoint == "" ? "none" : "otlp"
       VERBOSITY                   = var.verbosity

modules/forwarder/variables.tf

@@ -64,6 +64,21 @@ variable "source_bucket_names" {
   }
 }
 
+variable "source_object_keys" {
+  description = <<-EOF
+    A list of object key patterns the forwarder is allowed to read from for
+    provided source buckets.
+  EOF
+  type        = list(string)
+  nullable    = false
+  default     = ["*"]
+
+  validation {
+    condition     = length(var.source_object_keys) > 0
+    error_message = "At least one S3 object key match must be provided."
+  }
+}
+
 variable "source_topic_arns" {
   description = <<-EOF
     A list of SNS topics the forwarder is allowed to be subscribed to.
@@ -220,3 +235,4 @@ variable "code_uri" {
   default     = ""
   nullable    = false
 }
+

modules/stack/README.md

@@ -170,7 +170,7 @@ You can additionally configure other submodules in this manner:
 | <a name="input_configsubscription"></a> [configsubscription](#input\_configsubscription) | Variables for AWS Config subscription. | <pre>object({<br>    delivery_bucket_name = string<br>  })</pre> | `null` | no |
 | <a name="input_debug_endpoint"></a> [debug\_endpoint](#input\_debug\_endpoint) | Endpoint to send debugging telemetry to. Sets OTEL\_EXPORTER\_OTLP\_ENDPOINT environment variable for supported lambda functions. | `string` | `null` | no |
 | <a name="input_destination"></a> [destination](#input\_destination) | Destination filedrop | <pre>object({<br>    arn    = optional(string, "")<br>    bucket = optional(string, "")<br>    prefix = optional(string, "")<br>    # exclusively for backward compatible HTTP endpoint<br>    uri = optional(string, "")<br>  })</pre> | n/a | yes |
-| <a name="input_forwarder"></a> [forwarder](#input\_forwarder) | Variables for forwarder module. | <pre>object({<br>    source_bucket_names                      = optional(list(string), [])<br>    source_topic_arns                        = optional(list(string), [])<br>    content_type_overrides                   = optional(list(object({ pattern = string, content_type = string })), [])<br>    max_file_size                            = optional(number)<br>    lambda_memory_size                       = optional(number)<br>    lambda_timeout                           = optional(number)<br>    lambda_env_vars                          = optional(map(string))<br>    retention_in_days                        = optional(number)<br>    queue_max_receive_count                  = optional(number)<br>    queue_delay_seconds                      = optional(number)<br>    queue_message_retention_seconds          = optional(number)<br>    queue_batch_size                         = optional(number)<br>    queue_maximum_batching_window_in_seconds = optional(number)<br>    code_uri                                 = optional(string)<br>    sam_release_version                      = optional(string)<br>  })</pre> | `{}` | no |
+| <a name="input_forwarder"></a> [forwarder](#input\_forwarder) | Variables for forwarder module. | <pre>object({<br>    source_bucket_names                      = optional(list(string), [])<br>    source_object_keys                       = optional(list(string))<br>    source_topic_arns                        = optional(list(string), [])<br>    content_type_overrides                   = optional(list(object({ pattern = string, content_type = string })), [])<br>    max_file_size                            = optional(number)<br>    lambda_memory_size                       = optional(number)<br>    lambda_timeout                           = optional(number)<br>    lambda_env_vars                          = optional(map(string))<br>    retention_in_days                        = optional(number)<br>    queue_max_receive_count                  = optional(number)<br>    queue_delay_seconds                      = optional(number)<br>    queue_message_retention_seconds          = optional(number)<br>    queue_batch_size                         = optional(number)<br>    queue_maximum_batching_window_in_seconds = optional(number)<br>    code_uri                                 = optional(string)<br>    sam_release_version                      = optional(string)<br>  })</pre> | `{}` | no |
 | <a name="input_logwriter"></a> [logwriter](#input\_logwriter) | Variables for AWS CloudWatch Logs collection. | <pre>object({<br>    log_group_name_patterns         = optional(list(string))<br>    log_group_name_prefixes         = optional(list(string))<br>    exclude_log_group_name_prefixes = optional(list(string))<br>    buffering_interval              = optional(number)<br>    buffering_size                  = optional(number)<br>    filter_name                     = optional(string)<br>    filter_pattern                  = optional(string)<br>    num_workers                     = optional(number)<br>    discovery_rate                  = optional(string, "24 hours")<br>    lambda_memory_size              = optional(number)<br>    lambda_timeout                  = optional(number)<br>    code_uri                        = optional(string)<br>    sam_release_version             = optional(string)<br>  })</pre> | `null` | no |
 | <a name="input_metricstream"></a> [metricstream](#input\_metricstream) | Variables for AWS CloudWatch Metrics Stream collection. | <pre>object({<br>    include_filters     = optional(list(object({ namespace = string, metric_names = optional(list(string)) })))<br>    exclude_filters     = optional(list(object({ namespace = string, metric_names = optional(list(string)) })))<br>    buffering_interval  = optional(number)<br>    buffering_size      = optional(number)<br>    sam_release_version = optional(string)<br>  })</pre> | `null` | no |
 | <a name="input_name"></a> [name](#input\_name) | Name of role. Since this name must be unique within the<br>account, it will be reused for most of the resources created by this<br>module. | `string` | n/a | yes |

modules/stack/forwarder.tf

@@ -8,6 +8,7 @@ module "forwarder" {
     aws_s3_bucket.this.id,
     var.configsubscription != null ? var.configsubscription.delivery_bucket_name : "",
   ], var.forwarder.source_bucket_names))
+  source_object_keys                       = var.forwarder.source_object_keys
   source_topic_arns                        = concat([aws_sns_topic.this.arn], var.forwarder.source_topic_arns)
   content_type_overrides                   = local.content_type_overrides
   max_file_size                            = var.forwarder.max_file_size

modules/stack/variables.tf

@@ -30,6 +30,7 @@ variable "forwarder" {
   EOF
   type = object({
     source_bucket_names                      = optional(list(string), [])
+    source_object_keys                       = optional(list(string))
     source_topic_arns                        = optional(list(string), [])
     content_type_overrides                   = optional(list(object({ pattern = string, content_type = string })), [])
     max_file_size                            = optional(number)