feat(stepfunctions): expectedBucketOwner parameter for cross-account s3 access

packages/@aws-cdk-testing/framework-integ/test/aws-stepfunctions/test/integ.distributed-map.js.snapshot/aws-stepfunctions-map-integ.template.json

@@ -210,7 +210,11 @@
        {
         "Ref": "Bucket83908E77"
        },
-       "\",\"Key\":\"my-key.csv\"}}}}}"
+       "\",\"Key\":\"my-key.csv\",\"ExpectedBucketOwner\":\"",
+       {
+        "Ref": "AWS::AccountId"
+       },
+       "\"}}}}}"
       ]
      ]
     },

packages/@aws-cdk-testing/framework-integ/test/aws-stepfunctions/test/integ.distributed-map.ts

@@ -22,6 +22,7 @@ class DistributedMapStack extends cdk.Stack {
         bucket: this.bucket,
         key: CSV_KEY,
         csvHeaders: sfn.CsvHeaders.useFirstRow(),
+        expectedBucketOwner: this.account,
       }),
       resultWriter: new sfn.ResultWriter({
         bucket: this.bucket,

packages/aws-cdk-lib/aws-stepfunctions/README.md

@@ -1033,6 +1033,20 @@ distributedMap.itemProcessor(new sfn.Pass(this, 'Pass State'));
   ```
 * CSV file stored in S3
 * S3 inventory manifest stored in S3
+* When your Step Functions state machine needs to read S3 objects from a bucket in a different AWS account, specify the bucket owner's [account ID](https://docs.aws.amazon.com/accounts/latest/reference/manage-acct-identifiers.html#FindAccountId) using the `expectedBucketOwner` parameter:
+
+  ```ts
+  import * as s3 from 'aws-cdk-lib/aws-s3';
+
+  const distributedMap = new sfn.DistributedMap(this, 'DistributedMap', {
+    itemReader: new sfn.S3JsonItemReader({
+      bucket: s3.Bucket.fromBucketName(this, 'CrossAccountBucket', 'bucket-in-account-a'),
+      key: 'input.json',
+      expectedBucketOwner: '123456789012', , // Account ID that owns the bucket
+    }),
+  });
+  distributedMap.itemProcessor(new sfn.Pass(this, 'Pass'));
+  ```
 
 Map states in Distributed mode also support writing results of the iterator to an S3 bucket and optional prefix.  Use a `ResultWriterV2` object provided via the optional `resultWriter` property to configure which S3 location iterator results will be written. The default behavior id `resultWriter` is omitted is to use the state output payload. However, if the iterator results are larger than the 256 kb limit for Step Functions payloads then the State Machine will fail.
 

packages/aws-cdk-lib/aws-stepfunctions/lib/states/distributed-map/item-reader.ts

@@ -70,6 +70,13 @@ export interface ItemReaderProps {
    * @default - Distributed Map state will iterate over all items provided by the ItemReader
    */
   readonly maxItems?: number;
+
+  /**
+   * The account ID of the expected bucket owner for cross-account access
+   *
+   * @default - No expected bucket owner validation
+   */
+  readonly expectedBucketOwner?: string;
 }
 
 /**
@@ -125,11 +132,19 @@ export class S3ObjectsItemReader implements IItemReader {
    */
   readonly maxItems?: number;
 
+  /**
+   * The account ID of the expected bucket owner for cross-account access
+   *
+   * @default - No expected bucket owner validation
+   */
+  readonly expectedBucketOwner?: string;
+
   constructor(props: S3ObjectsItemReaderProps) {
     this._bucket = props.bucket;
     this.bucketNamePath = props.bucketNamePath;
     this.prefix = props.prefix;
     this.maxItems = props.maxItems;
+    this.expectedBucketOwner = props.expectedBucketOwner;
     this.resource = Arn.format({
       region: '',
       account: '',
@@ -150,6 +165,7 @@ export class S3ObjectsItemReader implements IItemReader {
       ...(this._bucket && { Bucket: this._bucket.bucketName }),
       ...(this.bucketNamePath && { Bucket: this.bucketNamePath }),
       ...(this.prefix && { Prefix: this.prefix }),
+      ...(this.expectedBucketOwner && { ExpectedBucketOwner: this.expectedBucketOwner }),
     };
     return FieldUtils.renderObject({
       Resource: this.resource,
@@ -239,13 +255,21 @@ abstract class S3FileItemReader implements IItemReader {
    */
   readonly maxItems?: number;
 
+  /**
+   * The account ID of the expected bucket owner for cross-account access
+   *
+   * @default - No expected bucket owner validation
+   */
+  readonly expectedBucketOwner?: string;
+
   protected abstract readonly inputType: string;
 
   constructor(props: S3FileItemReaderProps) {
     this._bucket = props.bucket;
     this.bucketNamePath = props.bucketNamePath;
     this.key = props.key;
     this.maxItems = props.maxItems;
+    this.expectedBucketOwner = props.expectedBucketOwner;
     this.resource = Arn.format({
       region: '',
       account: '',
@@ -266,6 +290,7 @@ abstract class S3FileItemReader implements IItemReader {
       ...(this._bucket && { Bucket: this._bucket.bucketName }),
       ...(this.bucketNamePath && { Bucket: this.bucketNamePath }),
       Key: this.key,
+      ...(this.expectedBucketOwner && { ExpectedBucketOwner: this.expectedBucketOwner }),
     };
 
     return FieldUtils.renderObject({