feat: 1) Add default s3 bucket name, s3 object and account id of the bucket properties in S3 provder

      2) Add s3 source pipeline trigger via s3 notification events function
      3) Add cross account s3 event rule fowarder function

Author: danxie1999

docs/providers-guide.md

@@ -125,15 +125,41 @@ Please add the required S3 read permissions to the `adf-codecomit-role` via the
 the `adf-codecommit-role` S3 read permissions in the bucket policy of the
 source bucket.
 
+If the `poll_for_changes` proerty set to `False`, ADF will monitor the S3 events 
+`Object Created` or `Object Copy` for the defined `object_key` of the defined 
+`bucket_name` and trigger the related pipeline.
+
+The source S3 bucket should enable `Bucket Versioning` and `Amazon EventBridge`, 
+otherwise, the auto pipeline trigger will not work.
+
+ADF supports source S3 bucket in a target account other than default Deployment account. 
+To make it work, an event bridge policy should be manually added to the default event bus 
+in the Deployment map. For example:
+
+```
+{
+  "Version": "2012-10-17",
+  "Statement": [{
+    "Sid": "allow_account_to_put_events",
+    "Effect": "Allow",
+    "Principal": {
+      "AWS": "arn:aws:iam::<target-account-id>:root"
+    },
+    "Action": "events:PutEvents",
+    "Resource": "arn:aws:events:eu-central-1:<deployment-account-id>:event-bus/default"
+  }]
+}
+```
+
 Provider type: `s3`.
 
 #### Properties
 
-- *account_id* - *(String)* **(required)**
-  - The AWS Account ID where the source S3 Bucket is located.
-- *bucket_name* - *(String)* **(required)**
+- *account_id* - *(String)* **(optional)**
+  - The AWS Account ID where the source S3 Bucket is located. By default, it is Deployment account ID.
+- *bucket_name* - *(String)* 
   - The Name of the S3 Bucket that will be the source of the pipeline.
-- *object_key* - *(String)* **(required)**
+- *object_key* - *(String)* 
   - The Specific Object within the bucket that will trigger the pipeline
     execution.
 - *trigger_on_changes* - *(Boolean)* default: `True`.
@@ -144,7 +170,15 @@ Provider type: `s3`.
   - **By default**, it will trigger on changes using the polling mechanism of
     CodePipeline. Monitoring the S3 object so it can trigger a release when an
     update took place.
-
+- *poll_for_changes* - *(Boolean)* default: `True`.
+  - If CodePipeline should poll the repository for changes, defaults to `False`
+    in favor of Amazon EventBridge events. As the name implies, when polling
+    for changes it will check the repository for updates every minute or so.
+    This will show up as actions in CloudTrail.
+  - **By default**, it will poll for changes, however, if set to `False`, it 
+    will use the event triggered by S3 notification when an update to the 
+    s3 object took place.
+  
 ### CodeConnections
 
 Use CodeConnections as a source to trigger your pipeline. The source action retrieves

src/lambda_codebase/initial_commit/adfconfig.yml.j2

@@ -33,6 +33,7 @@ config:
     default-scm-branch: main
     # Optional:
     # default-scm-codecommit-account-id: "123456789012"
+    # default-s3-source-bucket-name: "mys3deploymentbucket"
   deployment-maps:
     allow-empty-target: disabled
     # ^ Needs to be set to "enabled" to activate. Defaults to "disabled" when

src/lambda_codebase/initial_commit/bootstrap_repository/adf-bootstrap/deployment/lambda_codebase/pipeline_management/create_or_update_rule.py

@@ -23,7 +23,8 @@
 CLOUDWATCH = boto3.client("cloudwatch")
 METRICS = ADFMetrics(CLOUDWATCH, "PIPELINE_MANAGEMENT/RULE")
 
-_CACHE = None
+_CACHE_S3 = None
+_CACHE_CODECOMMIT = None
 
 
 def lambda_handler(event, _):
@@ -38,31 +39,52 @@ def lambda_handler(event, _):
         event (dict): The ADF Pipeline Management State Machine execution
             input object.
     """
-
     # pylint: disable=W0603
     # Global variable here to cache across lambda execution runtimes.
-    global _CACHE
-    if not _CACHE:
-        _CACHE = Cache()
+    global _CACHE_S3, _CACHE_CODECOMMIT
+
+    if not _CACHE_S3:
+        _CACHE_S3 = Cache()
+        METRICS.put_metric_data(
+            {"MetricName": "S3CacheInitialized", "Value": 1, "Unit": "Count"}
+        )
+
+    if not _CACHE_CODECOMMIT:
+        _CACHE_CODECOMMIT = Cache()
         METRICS.put_metric_data(
-            {"MetricName": "CacheInitialized", "Value": 1, "Unit": "Count"}
+            {"MetricName": "CodeCommitCacheInitialized", "Value": 1, "Unit": "Count"}
         )
 
     LOGGER.info(event)
 
     pipeline = event['pipeline_definition']
 
-    source_provider = (
-        pipeline.get("default_providers", {})
-        .get("source", {})
-        .get("provider", "codecommit")
-    )
-    source_account_id = (
-        pipeline.get("default_providers", {})
-        .get("source", {})
-        .get("properties", {})
-        .get("account_id")
-    )
+    default_source_provider = pipeline.get("default_providers", {}).get("source", {})
+    source_provider = default_source_provider.get("provider", "codecommit")
+    source_provider_properties = default_source_provider.get("properties", {})
+    source_account_id = source_provider_properties.get("account_id")
+    source_bucket_name = source_provider_properties.get("bucket_name")
+    if source_provider == "s3":
+        if not source_account_id:
+            source_account_id = DEPLOYMENT_ACCOUNT_ID
+            pipeline["default_providers"]["source"].setdefault("properties", {})["account_id"] = source_account_id
+        if not source_bucket_name:
+            try:
+                parameter_store = ParameterStore(DEPLOYMENT_ACCOUNT_REGION, boto3)
+                default_s3_source_bucket_name = parameter_store.fetch_parameter(
+                    "/adf/scm/default-s3-source-bucket-name"
+                )
+            except ParameterNotFoundError:
+                default_s3_source_bucket_name = os.environ["S3_BUCKET_NAME"]
+                LOGGER.debug("default_s3_source_bucket_name not found in SSM - Fall back to s3_bucket_name.")
+            pipeline["default_providers"]["source"].setdefault("properties", {})["bucket_name"] = default_s3_source_bucket_name
+            source_bucket_name = default_s3_source_bucket_name
+        event_params = {
+                "SourceS3BucketName": source_bucket_name
+        }
+    else:
+        event_params = {}
+        
 
     # Resolve codecommit source_account_id in case it is not set
     if source_provider == "codecommit" and not source_account_id:
@@ -98,25 +120,36 @@ def lambda_handler(event, _):
         )
 
     if (
-        source_provider == "codecommit"
-        and source_account_id
+        source_account_id
         and int(source_account_id) != int(DEPLOYMENT_ACCOUNT_ID)
-        and not _CACHE.exists(source_account_id)
+        and (
+            (source_provider == "codecommit" and not _CACHE_CODECOMMIT.exists(source_account_id))
+            or (source_provider == "s3" and not _CACHE_S3.exists(source_account_id))
+        )
     ):
         LOGGER.info(
-            "Source is CodeCommit and the repository is hosted in the %s "
+            "Source is %s and the repository/bucket is hosted in the %s "
             "account instead of the deployment account (%s). Creating or "
             "updating EventBridge forward rule to forward change events "
             "from the source account to the deployment account in "
             "EventBridge.",
+            source_provider,
             source_account_id,
             DEPLOYMENT_ACCOUNT_ID,
         )
-        rule = Rule(source_account_id)
+
+        rule = Rule(source_account_id, source_provider, event_params)
         rule.create_update()
-        _CACHE.add(source_account_id, True)
-        METRICS.put_metric_data(
-            {"MetricName": "CreateOrUpdate", "Value": 1, "Unit": "Count"}
-        )
+
+        if source_provider == "codecommit":
+            _CACHE_CODECOMMIT.add(source_account_id, True)
+            METRICS.put_metric_data(
+                {"MetricName": "CodeCommitCreateOrUpdate", "Value": 1, "Unit": "Count"}
+            )
+        elif source_provider == "s3":
+            _CACHE_S3.add(source_account_id, True)
+            METRICS.put_metric_data(
+                {"MetricName": "S3CreateOrUpdate", "Value": 1, "Unit": "Count"}
+            )
 
     return event

src/lambda_codebase/initial_commit/bootstrap_repository/adf-bootstrap/global.yml

@@ -70,6 +70,7 @@ Resources:
               - "codecommit:UploadArchive"
               - "codepipeline:StartPipelineExecution"
               - "events:PutEvents"
+              - "s3:Get*"
             Resource: "*"
           - Effect: Allow
             Action:
@@ -372,7 +373,7 @@ Resources:
               - "iam:TagRole"
               - "iam:UntagRole"
             Resource:
-              - !Sub "arn:${AWS::Partition}:iam::${AWS::AccountId}:role/adf/cross-account-events/adf-cc-event-from-${AWS::AccountId}-to-${DeploymentAccountId}"
+              - !Sub "arn:${AWS::Partition}:iam::${AWS::AccountId}:role/adf/cross-account-events/adf-*-event-from-${AWS::AccountId}-to-${DeploymentAccountId}"
           - Effect: Allow
             Sid: "IAMFullPathAndNameOnly"
             Action:
@@ -381,21 +382,21 @@ Resources:
               - "iam:GetRolePolicy"
               - "iam:PutRolePolicy"
             Resource:
-              - !Sub "arn:${AWS::Partition}:iam::${AWS::AccountId}:role/adf/cross-account-events/adf-cc-event-from-${AWS::AccountId}-to-${DeploymentAccountId}"
-              - !Sub "arn:${AWS::Partition}:iam::${AWS::AccountId}:role/adf-cc-event-from-${AWS::AccountId}-to-${DeploymentAccountId}"
+              - !Sub "arn:${AWS::Partition}:iam::${AWS::AccountId}:role/adf/cross-account-events/adf-*-event-from-${AWS::AccountId}-to-${DeploymentAccountId}"
+              - !Sub "arn:${AWS::Partition}:iam::${AWS::AccountId}:role/adf-*-event-from-${AWS::AccountId}-to-${DeploymentAccountId}"
           - Effect: Allow
             Sid: "IAMPassRole"
             Action:
               - "iam:PassRole"
             Resource:
-              - !Sub "arn:${AWS::Partition}:iam::${AWS::AccountId}:role/adf/cross-account-events/adf-cc-event-from-${AWS::AccountId}-to-${DeploymentAccountId}"
+              - !Sub "arn:${AWS::Partition}:iam::${AWS::AccountId}:role/adf/cross-account-events/adf-*-event-from-${AWS::AccountId}-to-${DeploymentAccountId}"
             Condition:
               StringEquals:
                 'iam:PassedToService':
                   - "events.amazonaws.com"
               ArnEquals:
                 'iam:AssociatedResourceArn':
-                  - !Sub "arn:${AWS::Partition}:events:${AWS::Region}:${AWS::AccountId}:rule/adf-cc-event-from-${AWS::AccountId}-to-${DeploymentAccountId}"
+                  - !Sub "arn:${AWS::Partition}:events:${AWS::Region}:${AWS::AccountId}:rule/adf-*-event-from-${AWS::AccountId}-to-${DeploymentAccountId}"
           - Effect: Allow
             Sid: "KMS"
             Action:

src/lambda_codebase/initial_commit/bootstrap_repository/adf-build/main.py

@@ -188,6 +188,15 @@ def prepare_deployment_account(sts, deployment_account_id, config):
             .get('default-scm-codecommit-account-id', deployment_account_id)
         )
     )
+    # TODO merge
+    deployment_account_parameter_store.put_parameter(
+        'scm/default-s3-source-bucket-name',
+        (
+            config.config
+            .get('scm', {})
+            .get('default-s3-source-bucket-name', S3_BUCKET_NAME)
+        )
+    )
     deployment_account_parameter_store.put_parameter(
         'deployment_maps/allow_empty_target',
         config.config.get('deployment-maps', {}).get(

src/lambda_codebase/initial_commit/bootstrap_repository/adf-build/shared/cdk/cdk_constructs/adf_codepipeline.py

@@ -158,14 +158,18 @@ def _generate_configuration(self):
                     .get('default_providers', {})
                     .get('source', {})
                     .get('properties', {})
-                    .get('object_key')
+                    .get('object_key', f"{self.map_params['name']}.zip")
                 ),
                 "PollForSourceChanges": (
-                    self.map_params
-                    .get('default_providers', {})
-                    .get('source', {})
-                    .get('properties', {})
-                    .get('trigger_on_changes', True)
+                    (
+                        self.map_params['default_providers']['source']
+                        .get('properties', {})
+                        .get('trigger_on_changes', True)
+                    ) and (
+                        self.map_params['default_providers']['source']
+                        .get('properties', {})
+                        .get('poll_for_changes', True)
+                    )
                 ),
             }
         if self.provider == "S3" and self.category == "Deploy":
@@ -202,7 +206,7 @@ def _generate_configuration(self):
                         .get('default_providers', {})
                         .get('deploy', {})
                         .get('properties', {})
-                        .get('object_key')
+                        .get('object_key', f"{self.map_params['name']}.zip")
                     ))
                 ),
                 "KMSEncryptionKeyARN": (
@@ -706,7 +710,12 @@ def __init__(
             'pipeline',
             **pipeline_args
         )
-        adf_events.Events(self, 'events', {
+        _provider = (map_params
+                     .get('default_providers', {})
+                     .get('source', {})
+                     .get('provider')
+        )
+        _event_params = {
             "pipeline": (
                 f'arn:{ADF_DEPLOYMENT_PARTITION}:codepipeline:'
                 f'{ADF_DEPLOYMENT_REGION}:{ADF_DEPLOYMENT_ACCOUNT_ID}:'
@@ -753,7 +762,7 @@ def __init__(
                     .get('default_providers', {})
                     .get('source', {})
                     .get('properties', {})
-                    .get('poll_for_changes', False)
+                    .get('poll_for_changes', True if _provider == "s3" else False)
                 ),
                 "trigger_on_changes": (
                     map_params
@@ -763,7 +772,21 @@ def __init__(
                     .get('trigger_on_changes', True)
                 ),
             }
-        })
+        }
+        if _provider == "s3":
+            _event_params["s3_bucket_name"] = (map_params
+                    .get('default_providers', {})
+                    .get('source', {})
+                    .get('properties', {})
+                    .get('bucket_name')
+                )
+            _event_params["s3_object_key"] = (map_params
+                    .get('default_providers', {})
+                    .get('source', {})
+                    .get('properties', {})
+                    .get('object_key', f"{map_params['name']}.zip")
+                )
+        adf_events.Events(self, 'events', _event_params)
 
     @staticmethod
     def restructure_tags(current_tags):