feat(kinesis): streams are encrypted by default (#7102)

Currently, the default configuration for a stream is to be unencrypted. This change uses
the KMS master key and enables encryption on Streams by default.

Stream encryption is not supported in `cn-north-1` and `cn-northwest-1` regions
so there's a condition that sets the default in those regions to unencrypted.

**Why is this not a breaking change**

Modifying the encryption settings on a Stream does not require interruption.
Adding/removing the `StreamEncryption` properties will enable/disable the encryption
on the stream accordingly without interruption

packages/@aws-cdk/aws-kinesis/README.md

@@ -41,7 +41,7 @@ new Stream(this, "MyFirstStream", {
 ```
 
 You can also specify properties such as `shardCount` to indicate how many shards the stream should choose and a `retentionPeriod`
-to specify how many logn the data in the shards should remain accessible.
+to specify how long the data in the shards should remain accessible.
 Read more at [Creating and Managing Streams](https://docs.aws.amazon.com/streams/latest/dev/working-with-streams.html)
 
 ```ts
@@ -52,19 +52,15 @@ new Stream(this, "MyFirstStream", {
 });
 ```
 
-Streams are not encrypted by default.
-
 ### Encryption
 
 [Stream encryption](https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-properties-kinesis-stream-streamencryption.html) enables
 server-side encryption using an AWS KMS key for a specified stream.
 
-You can enable encryption on your stream with the master key owned by Kinesis Data Streams by specifying the `encryption` property.
+Encryption is enabled by default on your stream with the master key owned by Kinesis Data Streams in regions where it is supported.
 
 ```ts
-new Stream(this, 'MyEncryptedStream', {
-  encryption: StreamEncryption.MANAGED
-});
+new Stream(this, 'MyEncryptedStream');
 ```
 
 You can enable encryption on your stream with a user-managed key by specifying the `encryption` property.

packages/@aws-cdk/aws-kinesis/lib/stream.ts

@@ -1,6 +1,7 @@
 import * as iam from '@aws-cdk/aws-iam';
 import * as kms from '@aws-cdk/aws-kms';
-import { Construct, Duration, IResource, Resource, Stack } from '@aws-cdk/core';
+import { Aws, CfnCondition, Construct, Duration, Fn, IResource, Resource, Stack } from '@aws-cdk/core';
+import { IResolvable } from 'constructs';
 import { CfnStream } from './kinesis.generated';
 
 /**
@@ -201,7 +202,9 @@ export interface StreamProps {
    * If you choose KMS, you can specify a KMS key via `encryptionKey`. If
    * encryption key is not specified, a key will automatically be created.
    *
-   * @default - StreamEncryption.KMS if encryptionKey is specified, or StreamEncryption.UNENCRYPTED otherwise
+   * @default - StreamEncryption.KMS if encrypted Streams are supported in the region
+   *   or StreamEncryption.UNENCRYPTED otherwise.
+   *   StreamEncryption.KMS if an encryption key is supplied through the encryptionKey property
    */
   readonly encryption?: StreamEncryption;
 
@@ -210,8 +213,9 @@ export interface StreamProps {
    *
    * The 'encryption' property must be set to "Kms".
    *
-   * @default - If encryption is set to "KMS" and this property is undefined, a
-   * new KMS key will be created and associated with this stream.
+   * @default - Kinesis Data Streams master key ('/alias/aws/kinesis').
+   *   If encryption is set to StreamEncryption.KMS and this property is undefined, a new KMS key
+   *   will be created and associated with this stream.
    */
   readonly encryptionKey?: kms.IKey;
 }
@@ -290,13 +294,36 @@ export class Stream extends StreamBase {
    * user's configuration.
    */
   private parseEncryption(props: StreamProps): {
-    streamEncryption?: CfnStream.StreamEncryptionProperty,
+    streamEncryption?: CfnStream.StreamEncryptionProperty | IResolvable
     encryptionKey?: kms.IKey
   } {
 
+    // if encryption properties are not set, default to KMS in regions where KMS is available
+    if (!props.encryption && !props.encryptionKey) {
+
+      const conditionName = 'AwsCdkKinesisEncryptedStreamsUnsupportedRegions';
+      const existing = Stack.of(this).node.tryFindChild(conditionName);
+
+      // create a single condition for the Stack
+      if (!existing) {
+        new CfnCondition(Stack.of(this), conditionName, {
+          expression: Fn.conditionOr(
+            Fn.conditionEquals(Aws.REGION, 'cn-north-1'),
+            Fn.conditionEquals(Aws.REGION, 'cn-northwest-1')
+          )
+        });
+      }
+
+      return {
+        streamEncryption: Fn.conditionIf(conditionName,
+          Aws.NO_VALUE,
+          { EncryptionType: 'KMS', KeyId: 'alias/aws/kinesis'})
+      };
+    }
+
     // default based on whether encryption key is specified
     const encryptionType = props.encryption ??
-    (props.encryptionKey ? StreamEncryption.KMS : StreamEncryption.UNENCRYPTED);
+      (props.encryptionKey ? StreamEncryption.KMS : StreamEncryption.UNENCRYPTED);
 
     // if encryption key is set, encryption must be set to KMS.
     if (encryptionType !== StreamEncryption.KMS && props.encryptionKey) {