diff --git a/packages/@aws-cdk-testing/framework-integ/test/aws-dynamodb/test/integ.dynamodb.stream-policy.ts b/packages/@aws-cdk-testing/framework-integ/test/aws-dynamodb/test/integ.dynamodb.stream-policy.ts
new file mode 100644
index 0000000000000..5d7507126e28d
--- /dev/null
+++ b/packages/@aws-cdk-testing/framework-integ/test/aws-dynamodb/test/integ.dynamodb.stream-policy.ts
@@ -0,0 +1,42 @@
+import { App, RemovalPolicy, Stack, StackProps } from 'aws-cdk-lib';
+import { Construct } from 'constructs';
+import * as dynamodb from 'aws-cdk-lib/aws-dynamodb';
+import * as iam from 'aws-cdk-lib/aws-iam';
+import { IntegTest } from '@aws-cdk/integ-tests-alpha';
+
+export class TestStack extends Stack {
+
+  readonly table: dynamodb.Table;
+
+  constructor(scope: Construct, id: string, props?: StackProps) {
+    super(scope, id, props);
+
+    const doc = new iam.PolicyDocument({
+      statements: [
+        new iam.PolicyStatement({
+          actions: ['dynamodb:GetRecords', 'dynamodb:DescribeStream'],
+          principals: [new iam.AccountRootPrincipal()],
+          resources: ['*'],
+        }),
+      ],
+    });
+
+    this.table = new dynamodb.Table(this, 'TableTest1', {
+      partitionKey: {
+        name: 'id',
+        type: dynamodb.AttributeType.STRING,
+      },
+      removalPolicy: RemovalPolicy.DESTROY,
+      streamResourcePolicy: doc,
+      stream: dynamodb.StreamViewType.NEW_AND_OLD_IMAGES,
+    });
+
+  }
+}
+
+const app = new App();
+const stack = new TestStack(app, 'resource-policy-stack', {});
+
+new IntegTest(app, 'resource-policy-integ-test', {
+  testCases: [stack],
+});
diff --git a/packages/aws-cdk-lib/aws-dynamodb/TABLE_V1_API.md b/packages/aws-cdk-lib/aws-dynamodb/TABLE_V1_API.md
index 34cf25434d6b2..b2ffe94f9ebd5 100644
--- a/packages/aws-cdk-lib/aws-dynamodb/TABLE_V1_API.md
+++ b/packages/aws-cdk-lib/aws-dynamodb/TABLE_V1_API.md
@@ -273,4 +273,28 @@ new dynamodb.Table(this, 'MyTable', {
 });
 ```
 
-If you have a global table replica, note that it does not support the addition of a resource-based policy.
\ No newline at end of file
+If you have a global table replica, note that it does not support the addition of a resource-based policy.
+
+Using `streamResourcePolicy` you can add a [resource policy](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/access-control-resource-based.html) to a table's stream in the form of a `PolicyDocument`:
+
+```ts
+const policy = new iam.PolicyDocument({
+  statements: [
+    new iam.PolicyStatement({
+      actions: ['dynamodb:GetRecords'],
+      principals: [new iam.AccountRootPrincipal()],
+      resources: ['*'],
+    }),
+  ],
+});
+
+new dynamodb.Table(this, 'MyTable', {
+  partitionKey: {
+    name: 'id',
+    type: dynamodb.AttributeType.STRING,
+  },
+  removalPolicy: RemovalPolicy.DESTROY,
+  streamResourcePolicy: policy,
+  stream: StreamViewType.NEW_AND_OLD_IMAGES,
+});
+```
diff --git a/packages/aws-cdk-lib/aws-dynamodb/lib/table.ts b/packages/aws-cdk-lib/aws-dynamodb/lib/table.ts
index 8a37d443cb97f..979a4885f537a 100644
--- a/packages/aws-cdk-lib/aws-dynamodb/lib/table.ts
+++ b/packages/aws-cdk-lib/aws-dynamodb/lib/table.ts
@@ -397,6 +397,13 @@ export interface TableOptions extends SchemaOptions {
    * @default - No resource policy statement
    */
   readonly resourcePolicy?: iam.PolicyDocument;
+
+  /**
+   * Resource policy to assign to a DynamoDB stream.
+   * @see https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-properties-dynamodb-globaltable-replicaspecification.html#cfn-dynamodb-globaltable-replicaspecification-replicastreamspecification
+   * @default - No resource policy statements are added to the stream
+   */
+  readonly streamResourcePolicy?: iam.PolicyDocument;
 }
 
 /**
@@ -551,6 +558,12 @@ export abstract class TableBase extends Resource implements ITable, iam.IResourc
    */
   public abstract resourcePolicy?: iam.PolicyDocument;
 
+  /**
+   * Resource policy to assign to table.
+   * @attribute
+   */
+  public abstract streamResourcePolicy?: iam.PolicyDocument;
+
   protected readonly regionalArns = new Array<string>();
 
   /**
@@ -1047,6 +1060,7 @@ export class Table extends TableBase {
       public readonly tableStreamArn?: string;
       public readonly encryptionKey?: kms.IKey;
       public resourcePolicy?: iam.PolicyDocument;
+      public streamResourcePolicy?: iam.PolicyDocument;
       protected readonly hasIndex = (attrs.grantIndexPermissions ?? false) ||
         (attrs.globalIndexes ?? []).length > 0 ||
         (attrs.localIndexes ?? []).length > 0;
@@ -1092,6 +1106,13 @@ export class Table extends TableBase {
    */
   public resourcePolicy?: iam.PolicyDocument;
 
+  /**
+   * Resource policy to assign to DynamoDB stream.
+   * @see https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-properties-dynamodb-table-resourcepolicy.html
+   * @default - No resource policy statements are added to the created stream.
+   */
+  public streamResourcePolicy?: iam.PolicyDocument;
+
   /**
    * @attribute
    */
@@ -1139,15 +1160,28 @@ export class Table extends TableBase {
       if (props.stream && props.stream !== StreamViewType.NEW_AND_OLD_IMAGES) {
         throw new Error('`stream` must be set to `NEW_AND_OLD_IMAGES` when specifying `replicationRegions`');
       }
-      streamSpecification = { streamViewType: StreamViewType.NEW_AND_OLD_IMAGES };
+      streamSpecification = {
+        streamViewType: StreamViewType.NEW_AND_OLD_IMAGES,
+        resourcePolicy: props.streamResourcePolicy
+          ? { policyDocument: props.streamResourcePolicy }
+          : undefined,
+      };
 
       this.billingMode = props.billingMode ?? BillingMode.PAY_PER_REQUEST;
     } else {
       this.billingMode = props.billingMode ?? BillingMode.PROVISIONED;
       if (props.stream) {
-        streamSpecification = { streamViewType: props.stream };
+        streamSpecification = {
+          streamViewType: props.stream,
+          resourcePolicy: props.streamResourcePolicy
+            ? { policyDocument: props.streamResourcePolicy }
+            : undefined,
+        };
       }
     }
+    if (props.streamResourcePolicy && !props.stream) {
+      throw new Error('`stream` must be enabled when specifying `streamResourcePolicy`');
+    }
     this.validateProvisioning(props);
 
     this.table = new CfnTable(this, 'Resource', {
diff --git a/packages/aws-cdk-lib/aws-dynamodb/test/dynamodb.test.ts b/packages/aws-cdk-lib/aws-dynamodb/test/dynamodb.test.ts
index 4be16524b2ca3..4b1a762b16e39 100644
--- a/packages/aws-cdk-lib/aws-dynamodb/test/dynamodb.test.ts
+++ b/packages/aws-cdk-lib/aws-dynamodb/test/dynamodb.test.ts
@@ -3601,3 +3601,78 @@ test('Resource policy test', () => {
     },
   });
 });
+
+test('Stream resource policy test to fail as no stream specified', () => {
+  // GIVEN
+  const app = new App();
+  const stack = new Stack(app, 'Stack');
+
+  const doc = new iam.PolicyDocument({
+    statements: [
+      new iam.PolicyStatement({
+        actions: ['dynamodb:DescribeStream', 'dynamodb:GetRecords'],
+        principals: [new iam.ArnPrincipal('arn:aws:iam::111122223333:user/foobar')],
+        resources: ['*'],
+      }),
+    ],
+  });
+
+  expect(() => new Table(stack, 'Table', {
+    partitionKey: TABLE_PARTITION_KEY,
+    streamResourcePolicy: doc,
+  })).toThrow('`stream` must be enabled when specifying `streamResourcePolicy`');
+
+});
+
+test('Stream resource policy test', () => {
+  // GIVEN
+  const app = new App();
+  const stack = new Stack(app, 'Stack');
+
+  const doc = new iam.PolicyDocument({
+    statements: [
+      new iam.PolicyStatement({
+        actions: ['dynamodb:DescribeStream', 'dynamodb:GetRecords'],
+        principals: [new iam.ArnPrincipal('arn:aws:iam::111122223333:user/foobar')],
+        resources: ['*'],
+      }),
+    ],
+  });
+
+  // WHEN
+  const table = new Table(stack, 'Table', {
+    partitionKey: { name: 'id', type: AttributeType.STRING },
+    streamResourcePolicy: doc,
+    stream: StreamViewType.NEW_AND_OLD_IMAGES,
+  });
+
+  // THEN
+  Template.fromStack(stack).hasResourceProperties('AWS::DynamoDB::Table', {
+    KeySchema: [
+      { AttributeName: 'id', KeyType: 'HASH' },
+    ],
+    AttributeDefinitions: [
+      { AttributeName: 'id', AttributeType: 'S' },
+    ],
+  });
+
+  Template.fromStack(stack).hasResourceProperties('AWS::DynamoDB::Table', {
+    'StreamSpecification': {
+      'ResourcePolicy': {
+        'PolicyDocument': {
+          'Version': '2012-10-17',
+          'Statement': [
+            {
+              'Principal': {
+                'AWS': 'arn:aws:iam::111122223333:user/foobar',
+              },
+              'Effect': 'Allow',
+              'Action': ['dynamodb:DescribeStream', 'dynamodb:GetRecords'],
+              'Resource': '*',
+            },
+          ],
+        },
+      },
+    },
+  });
+});
\ No newline at end of file