fix(s3): incorrect account used for S3 event source mapping (aws#29365)

### Issue # (if applicable)
aws#21628

initial PR was closed for naming issues: aws#29023

Closes #<issue number here>.
1

### Reason for this change
A Customers has a stack intending to deploy with alongside a Lambda to account "A"

A Bucket (in account "B") was referenced using fromBucketAttributes in the same stack and specified account "B" as the account attribute. When hooked to Lambda using addEventSource, it was expected that the IAM configuration generated will specify account "B" as part of the conditional grant.

However, Account "A" is defined as the "source account" which is incorrect. The Bucket lives in account "B" and was only referenced in the stack whose resources get deployed to "A".

Today S3 bucket is added as an event source to lambda, the account for the bucket is sourced from the stack not from the bucket configuration.


CDK fails to reference customer's bucket account rather is results to using the stack account which might not necessary be the bucket account.

### Description of changes




### Description of how you validated changes

1. Extensive testing was conducted by creating an application and validating the generated templates
2. Unit test was also added to test the new change

```
aws-cdk-lib % yarn test aws-lambda-event-sources 
yarn run v1.22.19
$ jest aws-lambda-event-sources
 PASS  aws-lambda-event-sources/test/sns.test.ts (56.025 s)
 PASS  aws-lambda-event-sources/test/s3.test.ts (56.097 s)
 PASS  aws-lambda-event-sources/test/api.test.ts (56.436 s)
 PASS  aws-lambda-event-sources/test/kinesis.test.ts (56.558 s)
 PASS  aws-lambda-event-sources/test/dynamo.test.ts (57.016 s)
 PASS  aws-lambda-event-sources/test/sqs.test.ts (56.816 s)
 PASS  aws-lambda-event-sources/test/kafka.test.ts (57.452 s)
A worker process has failed to exit gracefully and has been force exited. This is likely caused by tests leaking due to improper teardown. Try running with --detectOpenHandles to find leaks. Active timers can also cause this, ensure that .unref() was called on them.

```



### Checklist
- [x] My code adheres to the [CONTRIBUTING GUIDE](https://github.com/aws/aws-cdk/blob/main/CONTRIBUTING.md) and [DESIGN GUIDELINES](https://github.com/aws/aws-cdk/blob/main/docs/DESIGN_GUIDELINES.md)


*By submitting this pull request, I confirm that my contribution is made under the terms of the Apache-2.0 license*

packages/aws-cdk-lib/aws-lambda-event-sources/test/s3.test.ts

@@ -228,4 +228,86 @@ describe('S3EventSource', () => {
       },
     });
   });
+  test('Cross account buckect access', () => {
+    // GIVEN
+    const app = new cdk.App();
+    const stack = new cdk.Stack(app, 'stack');
+    const fn = new TestFunction(stack, 'Fn');
+
+    let accountB = '1234567';
+    //WHEN
+    const foreignBucket =
+      s3.Bucket.fromBucketAttributes(stack, 'ImportedBucket', {
+        bucketArn: 'arn:aws:s3:::some-bucket-not-in-this-account',
+        // The account the bucket really lives in
+        account: accountB,
+      });
+
+    // This will generate the IAM bindings
+    fn.addEventSource(new sources.S3EventSource(foreignBucket as s3.Bucket,
+      { events: [s3.EventType.OBJECT_CREATED] }));
+
+    // THEN
+    Template.fromStack(stack).hasResourceProperties('AWS::Lambda::Permission', {
+      'Principal': 's3.amazonaws.com',
+      'SourceAccount': '1234567',
+      'SourceArn': 'arn:aws:s3:::some-bucket-not-in-this-account',
+    });
+  });
+
+  test('Test bucket account is referenced intrinsicly', () => {
+    // GIVEN
+    const stack = new cdk.Stack();
+    const fn = new TestFunction(stack, 'Fn');
+    const bucket = new s3.Bucket(stack, 'B');
+
+    // WHEN
+    fn.addEventSource(new sources.S3EventSource(bucket, {
+      events: [s3.EventType.OBJECT_CREATED, s3.EventType.OBJECT_REMOVED],
+      filters: [
+        { prefix: 'prefix/' },
+        { suffix: '.png' },
+      ],
+    }));
+
+    // THEN
+    Template.fromStack(stack).hasResourceProperties('AWS::Lambda::Permission', {
+      'Principal': 's3.amazonaws.com',
+      'SourceAccount': {
+        'Ref': 'AWS::AccountId',
+      },
+      'SourceArn': {
+        'Fn::GetAtt': ['B08E7C7AF', 'Arn'],
+      },
+    });
+  });
+
+  test('Default to stack account if bucket account doesnt exist', () => {
+    // GIVEN
+    const app = new cdk.App();
+    const stack = new cdk.Stack(app, 'stack');
+    const fn = new TestFunction(stack, 'Fn');
+
+    let accountB = '';
+    //WHEN
+    const foreignBucket =
+      s3.Bucket.fromBucketAttributes(stack, 'ImportedBucket', {
+        bucketArn: 'arn:aws:s3:::some-bucket-not-in-this-account',
+        // The account the bucket really lives in
+        account: accountB,
+      });
+
+    // This will generate the IAM bindings
+    fn.addEventSource(new sources.S3EventSource(foreignBucket as s3.Bucket,
+      { events: [s3.EventType.OBJECT_CREATED] }));
+
+    // THEN
+    Template.fromStack(stack).hasResourceProperties('AWS::Lambda::Permission', {
+      'Principal': 's3.amazonaws.com',
+      'SourceAccount': {
+        'Ref': 'AWS::AccountId',
+      },
+      'SourceArn': 'arn:aws:s3:::some-bucket-not-in-this-account',
+    });
+  });
 });

packages/aws-cdk-lib/aws-s3-notifications/lib/lambda.ts

@@ -21,7 +21,7 @@ export class LambdaDestination implements s3.IBucketNotificationDestination {
 
     if (bucket.node.tryFindChild(permissionId) === undefined) {
       this.fn.addPermission(permissionId, {
-        sourceAccount: Stack.of(bucket).account,
+        sourceAccount: bucket.account || Stack.of(bucket).account,
         principal: new iam.ServicePrincipal('s3.amazonaws.com'),
         sourceArn: bucket.bucketArn,
         // Placing the permissions node in the same scope as the s3 bucket.

packages/aws-cdk-lib/aws-s3/lib/bucket.ts

@@ -95,6 +95,11 @@ export interface IBucket extends IResource {
    */
   policy?: BucketPolicy;
 
+  /**
+   * The account bucket belongs to.
+   */
+  readonly account?: string;
+
   /**
    * Adds a statement to the resource policy for a principal (i.e.
    * account/role/service) to perform actions on this bucket and/or its
@@ -1754,6 +1759,7 @@ export class Bucket extends BucketBase {
       public readonly bucketWebsiteNewUrlFormat = attrs.bucketWebsiteNewUrlFormat ?? false;
       public readonly encryptionKey = attrs.encryptionKey;
       public readonly isWebsite = attrs.isWebsite ?? false;
+      public readonly account = attrs.account;
       public policy?: BucketPolicy = undefined;
       protected autoCreatePolicy = false;
       protected disallowPublicAccess = false;
@@ -1967,11 +1973,11 @@ export class Bucket extends BucketBase {
 
     if (props.serverAccessLogsBucket instanceof Bucket) {
       props.serverAccessLogsBucket.allowLogDelivery(this, props.serverAccessLogsPrefix);
-    // It is possible that `serverAccessLogsBucket` was specified but is some other `IBucket`
-    // that cannot have the ACLs or bucket policy applied. In that scenario, we should only
-    // setup log delivery permissions to `this` if a bucket was not specified at all, as documented.
-    // For example, we should not allow log delivery to `this` if given an imported bucket or
-    // another situation that causes `instanceof` to fail
+      // It is possible that `serverAccessLogsBucket` was specified but is some other `IBucket`
+      // that cannot have the ACLs or bucket policy applied. In that scenario, we should only
+      // setup log delivery permissions to `this` if a bucket was not specified at all, as documented.
+      // For example, we should not allow log delivery to `this` if given an imported bucket or
+      // another situation that causes `instanceof` to fail
     } else if (!props.serverAccessLogsBucket && props.serverAccessLogsPrefix) {
       this.allowLogDelivery(this, props.serverAccessLogsPrefix);
     } else if (props.serverAccessLogsBucket) {
@@ -2225,7 +2231,7 @@ export class Bucket extends BucketBase {
     function parseLifecycleRule(rule: LifecycleRule): CfnBucket.RuleProperty {
       const enabled = rule.enabled ?? true;
       if ((rule.expiredObjectDeleteMarker)
-      && (rule.expiration || rule.expirationDate || self.parseTagFilters(rule.tagFilters))) {
+        && (rule.expiration || rule.expirationDate || self.parseTagFilters(rule.tagFilters))) {
         // ExpiredObjectDeleteMarker cannot be specified with ExpirationInDays, ExpirationDate, or TagFilters.
         throw new Error('ExpiredObjectDeleteMarker cannot be specified with expiration, ExpirationDate, or TagFilters.');
       }
@@ -2350,7 +2356,7 @@ export class Bucket extends BucketBase {
     }
 
     if (accessControlRequiresObjectOwnership && this.objectOwnership === ObjectOwnership.BUCKET_OWNER_ENFORCED) {
-      throw new Error (`objectOwnership must be set to "${ObjectOwnership.OBJECT_WRITER}" when accessControl is "${this.accessControl}"`);
+      throw new Error(`objectOwnership must be set to "${ObjectOwnership.OBJECT_WRITER}" when accessControl is "${this.accessControl}"`);
     }
 
     return {