chore(kinesisfirehose-alpha): refactor encryption property to combine encryptionKey (#31430)

### Reason for this change

The previous `encryption` and `encryptionKey` properties required error handling to enforce when an `encryptionKey` could be specified and when it was invalid (only valid when using `CUSTOMER_MANAGED_KEY`).

The properties should be combined to make this user experience more straightforward and only allow a KMS key to be passed in when using a customer-managed key. 

### Description of changes

BREAKING CHANGE: `encryptionKey` property is removed and `encryption` property type has changed from the `StreamEncryption` enum to the `StreamEncryption` class. 

To pass in a KMS key for the customer managed key case, use `StreamEncryption.customerManagedKey(key)`

#### Details
Replaced `encryption` and `encryptionKey` properties with a single property `encryption` of type `StreamEncryption` and is used by calling one of the 3 methods:
```ts
SreamEncryption.unencrypted()
StreamEncryption.awsOwnedKey()
StreamEncryption.customerManagedKey(key?: IKey)
```

This makes it so it's not longer possible to pass in a key when the encryption type is AWS owned or unencrypted. The `key` is an optional parameter in `StreamEncryption.customerManagedKey(key?: IKey)` so following the previous behaviour, if a key is provided it will be used, otherwise a key will be created for the user. 
### Description of how you validated changes

Generated templates do not change so behaviour remains the same. 

Updated integ/unit tests. 

### 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*

Author: paulhcsun

packages/@aws-cdk/aws-kinesisfirehose-alpha/README.md

@@ -153,18 +153,18 @@ declare const destination: firehose.IDestination;
 
 // SSE with an AWS-owned key
 new firehose.DeliveryStream(this, 'Delivery Stream AWS Owned', {
-  encryption: firehose.StreamEncryption.AWS_OWNED,
+  encryption: firehose.StreamEncryption.awsOwnedKey(),
   destinations: [destination],
 });
 // SSE with an customer-managed key that is created automatically by the CDK
 new firehose.DeliveryStream(this, 'Delivery Stream Implicit Customer Managed', {
-  encryption: firehose.StreamEncryption.CUSTOMER_MANAGED,
+  encryption: firehose.StreamEncryption.customerManagedKey(),
   destinations: [destination],
 });
 // SSE with an customer-managed key that is explicitly specified
 declare const key: kms.Key;
 new firehose.DeliveryStream(this, 'Delivery Stream Explicit Customer Managed', {
-  encryptionKey: key,
+  encryption: firehose.StreamEncryption.customerManagedKey(key),
   destinations: [destination],
 });
 ```

packages/@aws-cdk/aws-kinesisfirehose-alpha/lib/delivery-stream.ts

@@ -9,6 +9,7 @@ import { Construct, Node } from 'constructs';
 import { IDestination } from './destination';
 import { FirehoseMetrics } from 'aws-cdk-lib/aws-kinesisfirehose/lib/kinesisfirehose-canned-metrics.generated';
 import { CfnDeliveryStream } from 'aws-cdk-lib/aws-kinesisfirehose';
+import { StreamEncryption } from './encryption';
 
 const PUT_RECORD_ACTIONS = [
   'firehose:PutRecord',
@@ -162,7 +163,7 @@ abstract class DeliveryStreamBase extends cdk.Resource implements IDeliveryStrea
 /**
  * Options for server-side encryption of a delivery stream.
  */
-export enum StreamEncryption {
+export enum StreamEncryptionType {
   /**
    * Data in the stream is stored unencrypted.
    */
@@ -216,16 +217,9 @@ export interface DeliveryStreamProps {
   /**
    * Indicates the type of customer master key (CMK) to use for server-side encryption, if any.
    *
-   * @default StreamEncryption.UNENCRYPTED - unless `encryptionKey` is provided, in which case this will be implicitly set to `StreamEncryption.CUSTOMER_MANAGED`
+   * @default StreamEncryption.unencrypted()
    */
   readonly encryption?: StreamEncryption;
-
-  /**
-   * Customer managed key to server-side encrypt data in the stream.
-   *
-   * @default - no KMS key will be used; if `encryption` is set to `CUSTOMER_MANAGED`, a KMS key will be created for you
-   */
-  readonly encryptionKey?: kms.IKey;
 }
 
 /**
@@ -334,23 +328,20 @@ export class DeliveryStream extends DeliveryStreamBase {
       throw new Error(`Only one destination is allowed per delivery stream, given ${props.destinations.length}`);
     }
 
-    if (props.encryptionKey || props.sourceStream) {
+    if (props.encryption?.encryptionKey || props.sourceStream) {
       this._role = this._role ?? new iam.Role(this, 'Service Role', {
         assumedBy: new iam.ServicePrincipal('firehose.amazonaws.com'),
       });
     }
 
     if (
       props.sourceStream &&
-        (props.encryption === StreamEncryption.AWS_OWNED || props.encryption === StreamEncryption.CUSTOMER_MANAGED || props.encryptionKey)
+        (props.encryption?.type === StreamEncryptionType.AWS_OWNED || props.encryption?.type === StreamEncryptionType.CUSTOMER_MANAGED)
     ) {
       throw new Error('Requested server-side encryption but delivery stream source is a Kinesis data stream. Specify server-side encryption on the data stream instead.');
     }
-    if ((props.encryption === StreamEncryption.AWS_OWNED || props.encryption === StreamEncryption.UNENCRYPTED) && props.encryptionKey) {
-      throw new Error(`Specified stream encryption as ${StreamEncryption[props.encryption]} but provided a customer-managed key`);
-    }
-    const encryptionKey = props.encryptionKey ?? (props.encryption === StreamEncryption.CUSTOMER_MANAGED ? new kms.Key(this, 'Key') : undefined);
-    const encryptionConfig = (encryptionKey || (props.encryption === StreamEncryption.AWS_OWNED)) ? {
+    const encryptionKey = props.encryption?.encryptionKey ?? (props.encryption?.type === StreamEncryptionType.CUSTOMER_MANAGED ? new kms.Key(this, 'Key') : undefined);
+    const encryptionConfig = (encryptionKey || (props.encryption?.type === StreamEncryptionType.AWS_OWNED)) ? {
       keyArn: encryptionKey?.keyArn,
       keyType: encryptionKey ? 'CUSTOMER_MANAGED_CMK' : 'AWS_OWNED_CMK',
     } : undefined;

packages/@aws-cdk/aws-kinesisfirehose-alpha/lib/encryption.ts

@@ -0,0 +1,44 @@
+import { StreamEncryptionType } from './delivery-stream';
+import { IKey } from 'aws-cdk-lib/aws-kms';
+
+/**
+ * Represents server-side encryption for a Kinesis Firehose Delivery Stream.
+ */
+export abstract class StreamEncryption {
+  /**
+   * No server-side encryption is configured.
+   */
+  public static unencrypted(): StreamEncryption {
+    return new (class extends StreamEncryption {
+    }) (StreamEncryptionType.UNENCRYPTED);
+  }
+
+  /**
+   * Configure server-side encryption using an AWS owned key.
+   */
+  public static awsOwnedKey(): StreamEncryption {
+    return new (class extends StreamEncryption {
+    }) (StreamEncryptionType.AWS_OWNED);
+  }
+
+  /**
+   * Configure server-side encryption using customer managed keys.
+   *
+   * @param encryptionKey the KMS key for the delivery stream.
+   */
+  public static customerManagedKey(encryptionKey?: IKey): StreamEncryption {
+    return new (class extends StreamEncryption {
+
+    }) (StreamEncryptionType.CUSTOMER_MANAGED, encryptionKey);
+  }
+
+  /**
+   * Constructor for StreamEncryption.
+   *
+   * @param type The type of server-side encryption for the Kinesis Firehose delivery stream.
+   * @param encryptionKey Optional KMS key used for customer managed encryption.
+   */
+  private constructor (
+    public readonly type: StreamEncryptionType,
+    public readonly encryptionKey?: IKey) {}
+}

packages/@aws-cdk/aws-kinesisfirehose-alpha/lib/index.ts

@@ -1,5 +1,6 @@
 export * from './delivery-stream';
 export * from './destination';
+export * from './encryption';
 export * from './lambda-function-processor';
 export * from './processor';
 

packages/@aws-cdk/aws-kinesisfirehose-alpha/test/delivery-stream.test.ts

@@ -9,6 +9,7 @@ import * as targets from 'aws-cdk-lib/aws-events-targets';
 import * as cdk from 'aws-cdk-lib';
 import { Construct, Node } from 'constructs';
 import * as firehose from '../lib';
+import { StreamEncryption } from '../lib';
 
 describe('delivery stream', () => {
   let stack: cdk.Stack;
@@ -151,12 +152,12 @@ describe('delivery stream', () => {
     Template.fromStack(stack).resourceCountIs('AWS::IAM::Role', 2);
   });
 
-  test('not providing role but specifying encryptionKey creates two roles', () => {
+  test('not providing role but using customerManagedKey encryption with a key creates two roles', () => {
     const key = new kms.Key(stack, 'Key');
 
     new firehose.DeliveryStream(stack, 'Delivery Stream', {
       destinations: [mockS3Destination],
-      encryptionKey: key,
+      encryption: StreamEncryption.customerManagedKey(key),
     });
 
     Template.fromStack(stack).hasResourceProperties('AWS::IAM::Role', {
@@ -215,7 +216,7 @@ describe('delivery stream', () => {
   test('requesting customer-owned encryption creates key and configuration', () => {
     new firehose.DeliveryStream(stack, 'Delivery Stream', {
       destinations: [mockS3Destination],
-      encryption: firehose.StreamEncryption.CUSTOMER_MANAGED,
+      encryption: firehose.StreamEncryption.customerManagedKey(),
       role: deliveryStreamRole,
     });
 
@@ -246,12 +247,12 @@ describe('delivery stream', () => {
     });
   });
 
-  test('providing encryption key creates configuration', () => {
+  test('using customerManagedKey encryption with provided key creates configuration', () => {
     const key = new kms.Key(stack, 'Key');
 
     new firehose.DeliveryStream(stack, 'Delivery Stream', {
       destinations: [mockS3Destination],
-      encryptionKey: key,
+      encryption: StreamEncryption.customerManagedKey(key),
       role: deliveryStreamRole,
     });
 
@@ -281,7 +282,7 @@ describe('delivery stream', () => {
   test('requesting AWS-owned key does not create key and creates configuration', () => {
     new firehose.DeliveryStream(stack, 'Delivery Stream', {
       destinations: [mockS3Destination],
-      encryption: firehose.StreamEncryption.AWS_OWNED,
+      encryption: firehose.StreamEncryption.awsOwnedKey(),
       role: deliveryStreamRole,
     });
 
@@ -299,7 +300,7 @@ describe('delivery stream', () => {
   test('requesting no encryption creates no configuration', () => {
     new firehose.DeliveryStream(stack, 'Delivery Stream', {
       destinations: [mockS3Destination],
-      encryption: firehose.StreamEncryption.UNENCRYPTED,
+      encryption: firehose.StreamEncryption.unencrypted(),
       role: deliveryStreamRole,
     });
 
@@ -311,42 +312,22 @@ describe('delivery stream', () => {
     });
   });
 
-  test('requesting AWS-owned key and providing a key throws an error', () => {
-    const key = new kms.Key(stack, 'Key');
-
-    expect(() => new firehose.DeliveryStream(stack, 'Delivery Stream', {
-      destinations: [mockS3Destination],
-      encryption: firehose.StreamEncryption.AWS_OWNED,
-      encryptionKey: key,
-    })).toThrowError('Specified stream encryption as AWS_OWNED but provided a customer-managed key');
-  });
-
-  test('requesting no encryption and providing a key throws an error', () => {
-    const key = new kms.Key(stack, 'Key');
-
-    expect(() => new firehose.DeliveryStream(stack, 'Delivery Stream', {
-      destinations: [mockS3Destination],
-      encryption: firehose.StreamEncryption.UNENCRYPTED,
-      encryptionKey: key,
-    })).toThrowError('Specified stream encryption as UNENCRYPTED but provided a customer-managed key');
-  });
-
   test('requesting encryption or providing a key when source is a stream throws an error', () => {
     const sourceStream = new kinesis.Stream(stack, 'Source Stream');
 
     expect(() => new firehose.DeliveryStream(stack, 'Delivery Stream 1', {
       destinations: [mockS3Destination],
-      encryption: firehose.StreamEncryption.AWS_OWNED,
+      encryption: firehose.StreamEncryption.awsOwnedKey(),
       sourceStream,
     })).toThrowError('Requested server-side encryption but delivery stream source is a Kinesis data stream. Specify server-side encryption on the data stream instead.');
     expect(() => new firehose.DeliveryStream(stack, 'Delivery Stream 2', {
       destinations: [mockS3Destination],
-      encryption: firehose.StreamEncryption.CUSTOMER_MANAGED,
+      encryption: firehose.StreamEncryption.customerManagedKey(),
       sourceStream,
     })).toThrowError('Requested server-side encryption but delivery stream source is a Kinesis data stream. Specify server-side encryption on the data stream instead.');
     expect(() => new firehose.DeliveryStream(stack, 'Delivery Stream 3', {
       destinations: [mockS3Destination],
-      encryptionKey: new kms.Key(stack, 'Key'),
+      encryption: StreamEncryption.customerManagedKey(new kms.Key(stack, 'Key')),
       sourceStream,
     })).toThrowError('Requested server-side encryption but delivery stream source is a Kinesis data stream. Specify server-side encryption on the data stream instead.');
   });

packages/@aws-cdk/aws-kinesisfirehose-alpha/test/integ.delivery-stream.ts

@@ -37,7 +37,7 @@ const key = new kms.Key(stack, 'Key', {
 
 new firehose.DeliveryStream(stack, 'Delivery Stream', {
   destinations: [mockS3Destination],
-  encryptionKey: key,
+  encryption: firehose.StreamEncryption.customerManagedKey(key),
 });
 
 new firehose.DeliveryStream(stack, 'Delivery Stream No Source Or Encryption Key', {