SQS: support bucket notifications

packages/@aws-cdk/aws-kms/lib/key.ts

@@ -1,4 +1,4 @@
-import { Construct, DeletionPolicy, Output, PolicyDocument, PolicyStatement } from '@aws-cdk/cdk';
+import { Construct, DeletionPolicy, Output, PolicyDocument, PolicyStatement, resolve } from '@aws-cdk/cdk';
 import { EncryptionKeyAlias } from './alias';
 import { cloudformation, KeyArn } from './kms.generated';
 
@@ -54,10 +54,15 @@ export abstract class EncryptionKeyRef extends Construct {
 
     /**
      * Adds a statement to the KMS key resource policy.
+     * @param statement The policy statement to add
+     * @param allowNoOp If this is set to `false` and there is no policy
+     * defined (i.e. external key), the operation will fail. Otherwise, it will
+     * no-op.
      */
-    public addToResourcePolicy(statement: PolicyStatement) {
+    public addToResourcePolicy(statement: PolicyStatement, allowNoOp = true) {
         if (!this.policy) {
-            return;
+            if (allowNoOp) { return; }
+            throw new Error(`Unable to add statement to IAM resource policy for KMS key: ${JSON.stringify(resolve(this.keyArn))}`);
         }
 
         this.policy.addStatement(statement);

packages/@aws-cdk/aws-kms/test/test.key.ts

@@ -1,7 +1,7 @@
 import { exactlyMatchTemplate, expect } from '@aws-cdk/assert';
 import { App, PolicyDocument, PolicyStatement, Stack } from '@aws-cdk/cdk';
 import { Test } from 'nodeunit';
-import { EncryptionKey } from '../lib';
+import { EncryptionKey, KeyArn } from '../lib';
 
 export = {
     'default key'(test: Test) {
@@ -349,5 +349,31 @@ export = {
         });
 
         test.done();
+    },
+
+    'addToResourcePolicy allowNoOp and there is no policy': {
+        'succeed if set to true (default)'(test: Test) {
+            const stack = new Stack();
+
+            const key = EncryptionKey.import(stack, 'Imported', { keyArn: new KeyArn('foo/bar') });
+
+            key.addToResourcePolicy(new PolicyStatement().addResource('*').addAction('*'));
+
+            test.done();
+        },
+
+        'fails if set to false'(test: Test) {
+
+            const stack = new Stack();
+
+            const key = EncryptionKey.import(stack, 'Imported', { keyArn: new KeyArn('foo/bar') });
+
+            test.throws(() =>
+              key.addToResourcePolicy(new PolicyStatement().addResource('*').addAction('*'), /* allowNoOp */ false),
+              'Unable to add statement to IAM resource policy for KMS key: "foo/bar"');
+
+            test.done();
+
+        }
     }
 };

packages/@aws-cdk/aws-sqs/lib/queue-ref.ts

@@ -1,15 +1,17 @@
-import { Construct, Output, PolicyStatement, Token } from '@aws-cdk/cdk';
+import kms = require('@aws-cdk/aws-kms');
+import s3n = require('@aws-cdk/aws-s3-notifications');
+import cdk = require('@aws-cdk/cdk');
 import { QueuePolicy } from './policy';
 import { QueueArn } from './sqs.generated';
 
 /**
  * Reference to a new or existing Amazon SQS queue
  */
-export abstract class QueueRef extends Construct {
+export abstract class QueueRef extends cdk.Construct implements s3n.IBucketNotificationDestination {
     /**
      * Import an existing queue
      */
-    public static import(parent: Construct, name: string, props: QueueRefProps) {
+    public static import(parent: cdk.Construct, name: string, props: QueueRefProps) {
         new ImportedQueue(parent, name, props);
     }
 
@@ -23,6 +25,11 @@ export abstract class QueueRef extends Construct {
      */
     public abstract readonly queueUrl: QueueUrl;
 
+    /**
+     * If this queue is server-side encrypted, this is the KMS encryption key.
+     */
+    public abstract readonly encryptionMasterKey?: kms.EncryptionKeyRef;
+
     /**
      * Controls automatic creation of policy objects.
      *
@@ -32,13 +39,21 @@ export abstract class QueueRef extends Construct {
 
     private policy?: QueuePolicy;
 
+    /**
+     * The set of S3 bucket IDs that are allowed to send notifications to this queue.
+     */
+    private readonly notifyingBuckets = new Set<string>();
+
     /**
      * Export a queue
      */
     public export(): QueueRefProps {
         return {
-            queueArn: new Output(this, 'QueueArn', { value: this.queueArn }).makeImportValue(),
-            queueUrl: new Output(this, 'QueueUrl', { value: this.queueUrl }).makeImportValue(),
+            queueArn: new cdk.Output(this, 'QueueArn', { value: this.queueArn }).makeImportValue(),
+            queueUrl: new cdk.Output(this, 'QueueUrl', { value: this.queueUrl }).makeImportValue(),
+            keyArn: this.encryptionMasterKey
+                ? new cdk.Output(this, 'KeyArn', { value: this.encryptionMasterKey.keyArn }).makeImportValue()
+                : undefined
         };
     }
 
@@ -49,7 +64,7 @@ export abstract class QueueRef extends Construct {
      * will be automatically created upon the first call to `addToPolicy`. If
      * the queue is improted (`Queue.import`), then this is a no-op.
      */
-    public addToResourcePolicy(statement: PolicyStatement) {
+    public addToResourcePolicy(statement: cdk.PolicyStatement) {
         if (!this.policy && this.autoCreatePolicy) {
             this.policy = new QueuePolicy(this, 'Policy', { queues: [ this ] });
         }
@@ -59,14 +74,61 @@ export abstract class QueueRef extends Construct {
         }
     }
 
+    /**
+     * Allows using SQS queues as destinations for bucket notifications.
+     * Use `bucket.onEvent(event, queue)` to subscribe.
+     * @param bucketArn The ARN of the notifying bucket.
+     * @param bucketId A unique ID for the notifying bucket.
+     */
+    public asBucketNotificationDestination(bucketArn: cdk.Arn, bucketId: string): s3n.BucketNotificationDestinationProps {
+        if (!this.notifyingBuckets.has(bucketId)) {
+            this.addToResourcePolicy(new cdk.PolicyStatement()
+                .addServicePrincipal('s3.amazonaws.com')
+                .addAction('sqs:SendMessage')
+                .addResource(this.queueArn)
+                .addCondition('ArnLike', { 'aws:SourceArn': bucketArn }));
+
+            // if this queue is encrypted, we need to allow S3 to read messages since that's how
+            // it verifies that the notification destination configuration is valid.
+            // by setting allowNoOp to false, we ensure that only custom keys that we can actually
+            // control access to can be used here as described in:
+            // https://docs.aws.amazon.com/AmazonS3/latest/dev/ways-to-add-notification-config-to-bucket.html
+            if (this.encryptionMasterKey) {
+                this.encryptionMasterKey.addToResourcePolicy(new cdk.PolicyStatement()
+                    .addServicePrincipal('s3.amazonaws.com')
+                    .addAction('kms:GenerateDataKey')
+                    .addAction('kms:Decrypt')
+                    .addResource('*'), /* allowNoOp */ false);
+            }
+
+            this.notifyingBuckets.add(bucketId);
+        }
+
+        return {
+            arn: this.queueArn,
+            type: s3n.BucketNotificationDestinationType.Queue
+        };
+    }
 }
 
 /**
  * Reference to a queue
  */
 export interface QueueRefProps {
+    /**
+     * The ARN of the queue.
+     */
     queueArn: QueueArn;
+
+    /**
+     * The URL of the queue.
+     */
     queueUrl: QueueUrl;
+
+    /**
+     * KMS encryption key, if this queue is server-side encrypted by a KMS key.
+     */
+    keyArn?: kms.KeyArn;
 }
 
 /**
@@ -75,18 +137,25 @@ export interface QueueRefProps {
 class ImportedQueue extends QueueRef {
     public readonly queueArn: QueueArn;
     public readonly queueUrl: QueueUrl;
+    public readonly encryptionMasterKey?: kms.EncryptionKeyRef;
 
     protected readonly autoCreatePolicy = false;
 
-    constructor(parent: Construct, name: string, props: QueueRefProps) {
+    constructor(parent: cdk.Construct, name: string, props: QueueRefProps) {
         super(parent, name);
         this.queueArn = props.queueArn;
         this.queueUrl = props.queueUrl;
+
+        if (props.keyArn) {
+            this.encryptionMasterKey = kms.EncryptionKey.import(this, 'Key', {
+                keyArn: props.keyArn
+            });
+        }
     }
 }
 
 /**
  * URL of a queue
  */
-export class QueueUrl extends Token {
+export class QueueUrl extends cdk.Token {
 }

packages/@aws-cdk/aws-sqs/lib/queue.ts

@@ -195,6 +195,11 @@ export class Queue extends QueueRef {
      */
     public readonly queueUrl: QueueUrl;
 
+    /**
+     * If this queue is encrypted, this is the KMS key.
+     */
+    public encryptionMasterKey?: kms.EncryptionKeyRef;
+
     protected readonly autoCreatePolicy = true;
 
     constructor(parent: cdk.Construct, name: string, props: QueueProps = {}) {
@@ -255,30 +260,34 @@ export class Queue extends QueueRef {
     }
 
     private determineEncryptionProps(props: QueueProps): EncryptionProps {
-        const encryption = props.encryption || QueueEncryption.Unencrypted;
+        let encryption = props.encryption || QueueEncryption.Unencrypted;
 
         if (encryption !== QueueEncryption.Kms && props.encryptionMasterKey) {
-            throw new Error("Encryption key is specified, so 'encryption' must be set to 'Kms'.");
+            encryption = QueueEncryption.Kms; // KMS is implied by specifying an encryption key
         }
 
         if (encryption === QueueEncryption.Unencrypted) {
             return {};
         }
 
         if (encryption === QueueEncryption.KmsManaged) {
+            this.encryptionMasterKey = kms.EncryptionKey.import(this, 'Key', {
+                keyArn: new kms.KeyArn('alias/aws/sqs')
+            });
+
             return {
                 kmsMasterKeyId: 'alias/aws/sqs',
                 kmsDataKeyReusePeriodSeconds: props.dataKeyReuseSec
             };
         }
 
-        if (props.encryption === QueueEncryption.Kms) {
-            const encryptionKey = props.encryptionMasterKey || new kms.EncryptionKey(this, 'Key', {
+        if (encryption === QueueEncryption.Kms) {
+            this.encryptionMasterKey = props.encryptionMasterKey || new kms.EncryptionKey(this, 'Key', {
                 description: `Created by ${this.path}`
             });
 
             return {
-                kmsMasterKeyId: encryptionKey.keyArn,
+                kmsMasterKeyId: this.encryptionMasterKey.keyArn,
                 kmsDataKeyReusePeriodSeconds: props.dataKeyReuseSec
             };
         }

packages/@aws-cdk/aws-sqs/package.json

@@ -51,10 +51,12 @@
     "cdk-build-tools": "^0.8.1",
     "cdk-integ-tools": "^0.8.1",
     "cfn2ts": "^0.8.1",
-    "pkglint": "^0.8.1"
+    "pkglint": "^0.8.1",
+    "@aws-cdk/aws-s3": "^0.8.1"
   },
   "dependencies": {
     "@aws-cdk/aws-kms": "^0.8.1",
+    "@aws-cdk/aws-s3-notifications": "^0.8.1",
     "@aws-cdk/cdk": "^0.8.1"
   },
   "homepage": "https://github.com/awslabs/aws-cdk"