add-vpc and remove-vpc use Events to create/destroy per-ENI mirroring (#42)

* add-vpc uses events to create Traffic Mirroring Sessions

Signed-off-by: Chris Helma <chelma+github@amazon.com>

* remove-vpc uses events to destroy Traffic Mirroring Sessions

* The CreateEniMirroring lambda now emits outcome metrics

* Given the move towards an event-based architecture, we needed a
  way to more easily keep track of what is happening in our system.
  I added CloudWatch Metrics to the lambda to indicate when each
  possible outcome occurred.

Signed-off-by: Chris Helma <chelma+github@amazon.com>

* DestroyEniMirror Lambda now emits outcome metrics

* Like the CreateEniMirror Lambda, the Destroy Lambda emits
  metrics to CloudWatch on each possible outcome to help evaluate
  what's happening in the system.

Signed-off-by: Chris Helma <chelma+github@amazon.com>

---------

Signed-off-by: Chris Helma <chelma+github@amazon.com>

Author: chelma

cdk-lib/capture-stacks/capture-nodes-stack.ts

@@ -2,16 +2,20 @@ import * as cdk from 'aws-cdk-lib';
 import * as autoscaling from 'aws-cdk-lib/aws-autoscaling';
 import * as ec2 from 'aws-cdk-lib/aws-ec2';
 import * as ecs from 'aws-cdk-lib/aws-ecs';
+import * as events from 'aws-cdk-lib/aws-events';
+import * as targets from 'aws-cdk-lib/aws-events-targets';
 import * as elbv2 from 'aws-cdk-lib/aws-elasticloadbalancingv2';
 import * as iam from 'aws-cdk-lib/aws-iam';
 import * as kms from 'aws-cdk-lib/aws-kms';
+import * as logs from 'aws-cdk-lib/aws-logs';
 import * as opensearch from 'aws-cdk-lib/aws-opensearchservice';
 import * as s3 from 'aws-cdk-lib/aws-s3';
 import * as secretsmanager from 'aws-cdk-lib/aws-secretsmanager';
 import * as ssm from 'aws-cdk-lib/aws-ssm';
 import * as path from 'path'
 import { Construct } from 'constructs';
 
+import * as constants from '../core/constants'
 import {ClusterSsmValue} from '../core/ssm-wrangling'
 
 export interface CaptureNodesStackProps extends cdk.StackProps {
@@ -204,15 +208,51 @@ export class CaptureNodesStack extends cdk.Stack {
             allowedPrincipals: [`arn:aws:iam::${this.account}:root`],
         });
 
-        // This SSM parameter will enable us share the details of our Capture setup.
-        const clusterParamValue: ClusterSsmValue = {clusterName: props.clusterName, vpceServiceId: gwlbEndpointService.ref}
+        /**
+         * Set up shared resources for event-based management of mirroring.
+         */
+        const clusterBus = new events.EventBus(this, 'ClusterBus', {})
+
+        // Store a copy of the Arkime events that occur for later replay
+        clusterBus.archive('Archive', {
+            archiveName: `Arkime-${props.clusterName}`,
+            description: `Archive of Arkime events for Cluster ${props.clusterName}`,
+            eventPattern: {
+                source: [constants.EVENT_SOURCE],
+            },
+            retention: cdk.Duration.days(365), // Arbitrarily chosen
+        });
+
+        // Make a human-readable log of the Arkime events that occur on the bus
+        const clusterLogGroup = new logs.LogGroup(this, 'LogGroup', {
+            logGroupName: `Arkime-${props.clusterName}`,
+            removalPolicy: cdk.RemovalPolicy.DESTROY // The archive contains the real events
+        });
+        const logClusterEventsRule = new events.Rule(this, 'RuleLogClusterEvents', {
+            eventBus: clusterBus,
+            eventPattern: {
+                source: [constants.EVENT_SOURCE],
+            },
+            targets: [new targets.CloudWatchLogGroup(clusterLogGroup)]
+        });
+        
+        /**
+         * This SSM parameter will enable us share the details of our Capture setup.
+         */
+        const clusterParamValue: ClusterSsmValue = {
+            busArn: clusterBus.eventBusArn,
+            busName: clusterBus.eventBusName,
+            clusterName: props.clusterName, 
+            vpceServiceId: gwlbEndpointService.ref
+        }
         const clusterParam = new ssm.StringParameter(this, 'ClusterParam', {
             allowedPattern: '.*',
             description: 'The Cluster\'s details',
             parameterName: props.ssmParamNameCluster,
             stringValue: JSON.stringify(clusterParamValue),
             tier: ssm.ParameterTier.STANDARD,
         });
-        clusterParam.node.addDependency(gwlbEndpointService);       
+        clusterParam.node.addDependency(gwlbEndpointService);
+        clusterParam.node.addDependency(clusterBus);
     }
 }

cdk-lib/cloud-demo.ts

@@ -72,6 +72,7 @@ switch(params.type) {
         break;
     case 'MirrorMgmtParams':
         new VpcMirrorStack(app, params.nameVpcMirrorStack, {
+            eventBusArn: params.arnEventBus,
             subnetIds: params.listSubnetIds,
             subnetSsmParamNames: params.listSubnetSsmParams,
             vpcId: params.idVpc,

cdk-lib/core/command-params.ts

@@ -29,6 +29,7 @@ export interface ClusterMgmtParamsRaw extends CommandParamsRaw {
  */
 export interface MirrorMgmtParamsRaw extends CommandParamsRaw {
     type: 'MirrorMgmtParamsRaw';
+    arnEventBus: string;
     nameVpcMirrorStack: string;
     nameVpcSsmParam: string;
     idVni: string;
@@ -88,6 +89,7 @@ export interface ClusterMgmtParams extends CommandParams {
  */
 export interface MirrorMgmtParams extends CommandParams {
     type: 'MirrorMgmtParams';
+    arnEventBus: string;
     nameVpcMirrorStack: string;
     nameVpcSsmParam: string;
     idVni: string;

cdk-lib/core/constants.ts

@@ -4,6 +4,9 @@
 export const CDK_CONTEXT_CMD_VAR: string = 'ARKIME_CMD'
 export const CDK_CONTEXT_REGION_VAR: string = 'ARKIME_REGION'
 export const CDK_CONTEXT_PARAMS_VAR: string = 'ARKIME_PARAMS'
+export const EVENT_SOURCE: string = "arkime";
+export const EVENT_DETAIL_TYPE_CREATE_ENI_MIRROR: string = "CreateEniMirror";
+export const EVENT_DETAIL_TYPE_DESTROY_ENI_MIRROR: string = "DestroyEniMirror";
 
 /**
  * These map directly to the specific commands executed by the user via the management CLI.  Since the strings are

cdk-lib/core/context-wrangling.ts

@@ -110,6 +110,7 @@ function validateArgs(args: ValidateArgs) : (prms.ClusterMgmtParams | prms.Deplo
             const rawMirrorMgmtParamsObj: prms.MirrorMgmtParamsRaw = JSON.parse(args.cmdParamsRaw)
             const mirrorMgmtParams: prms.MirrorMgmtParams = {
                 type: 'MirrorMgmtParams',
+                arnEventBus: rawMirrorMgmtParamsObj.arnEventBus,
                 awsAccount: args.awsAccount,
                 awsRegion: args.awsRegion,
                 nameVpcMirrorStack: rawMirrorMgmtParamsObj.nameVpcMirrorStack,

cdk-lib/core/ssm-wrangling.ts

@@ -5,6 +5,8 @@
  */
 
 export interface ClusterSsmValue {
+    readonly busArn: string;
+    readonly busName: string;
     readonly clusterName: string;
     readonly vpceServiceId: string;
 }

cdk-lib/mirror-stacks/vpc-mirror-stack.ts

@@ -1,13 +1,20 @@
 import assert = require('assert');
 
 import { Construct } from 'constructs';
-import { Stack, StackProps } from 'aws-cdk-lib';
+import { Duration, Stack, StackProps } from 'aws-cdk-lib';
 import * as ec2 from 'aws-cdk-lib/aws-ec2';
+import * as events from 'aws-cdk-lib/aws-events';
+import * as targets from 'aws-cdk-lib/aws-events-targets';
+import * as iam from 'aws-cdk-lib/aws-iam';
+import * as lambda from 'aws-cdk-lib/aws-lambda';
 import * as ssm from 'aws-cdk-lib/aws-ssm';
+import * as path from 'path'
 
 import {SubnetSsmValue, VpcSsmValue} from '../core/ssm-wrangling'
+import * as constants from '../core/constants'
 
 export interface VpcMirrorStackProps extends StackProps {
+    readonly eventBusArn: string;
     readonly subnetIds: string[];
     readonly subnetSsmParamNames: string[];
     readonly vpcId: string;
@@ -68,7 +75,7 @@ export class VpcMirrorStack extends Stack {
         // See: https://docs.aws.amazon.com/vpc/latest/mirroring/tm-example-non-vpc.html
         const filter = new ec2.CfnTrafficMirrorFilter(this, `Filter`, {
             description: 'Mirror non-local VPC traffic',
-            tags: [{key: "Name", value: props.vpcId}]
+            tags: [{key: 'Name', value: props.vpcId}]
         });
         new ec2.CfnTrafficMirrorFilterRule(this, `FRule-RejectLocalOutbound`, {
             destinationCidrBlock: '10.0.0.0/16', // TODO: Need to figure this out instead of hardcode
@@ -118,5 +125,147 @@ export class VpcMirrorStack extends Stack {
             tier: ssm.ParameterTier.STANDARD,
         });
         vpcParam.node.addDependency(filter);
+
+        /**
+         * Configure the resources required for event-based mirroring configuration
+         */
+        // Get a handle to the cluster event bus
+        const clusterBus = events.EventBus.fromEventBusArn(this, 'ClusterBus', props.eventBusArn);
+
+        // Archive Arkime events related to this User VPC to enable replay, with a focus on shorter-term debugging
+        clusterBus.archive('Archive', {
+            archiveName: `Arkime-${props.vpcId}`,
+            description: `Archive of Arkime events for VPC ${props.vpcId}`,
+            eventPattern: {
+                source: [constants.EVENT_SOURCE],
+                detail: {
+                    'vpc_id': events.Match.exactString(props.vpcId)
+                }
+            },
+            retention: Duration.days(30),
+        });
+
+        // Create the Lambda that will set up the traffic mirroring for ENIs in our VPC
+        const createLambda = new lambda.Function(this, 'CreateEniMirrorLambda', {
+            functionName: `CreateEniMirror-${props.vpcId}`,
+            runtime: lambda.Runtime.PYTHON_3_9,
+            code: lambda.Code.fromAsset(path.resolve(__dirname, '..', '..', 'manage_arkime')),            
+            handler: 'lambda_handlers.create_eni_mirror_handler',
+            timeout:  Duration.seconds(30), // Something has gone very wrong if this is exceeded            
+        });
+        createLambda.addToRolePolicy(
+            new iam.PolicyStatement({
+                effect: iam.Effect.ALLOW,
+                actions: [
+                    // TODO: Should scope this down.
+                    // We need ec2:CreateTrafficMirrorSession in order to set up our session, but whenever I add *just*
+                    // that, I get an UnauthorizedOperation.  The docs say that's all that should be required, but the
+                    // documentation appears wrong or there's something extra mysterious going on here.  Even CloudTrail
+                    // indicates the only call being made is CreateTrafficMirrorSession, but it's still failing.  The
+                    // exception also doesn't indicate otherwise.
+                    // See: https://docs.aws.amazon.com/vpc/latest/mirroring/traffic-mirroring-security.html
+                    'ec2:*',
+                ],
+                resources: [
+                    `arn:aws:ec2:${this.region}:${this.account}:*`
+                ]
+            })
+        );
+        createLambda.addToRolePolicy(
+            new iam.PolicyStatement({
+                effect: iam.Effect.ALLOW,
+                actions: [
+                    'ssm:GetParameter',
+                    'ssm:PutParameter',
+                ],
+                resources: [
+                    `arn:aws:ssm:${this.region}:${this.account}:*`
+                ]
+            })
+        );
+        createLambda.addToRolePolicy(
+            new iam.PolicyStatement({
+                effect: iam.Effect.ALLOW,
+                actions: [
+                    'cloudwatch:PutMetricData',
+                ],
+                resources: [
+                    "*"
+                ]
+            })
+        );
+
+        // Create a rule to funnel appropriate events to our setup lambda
+        const createRule = new events.Rule(this, 'RuleCreateEniMirror', {
+            eventBus: clusterBus,
+            eventPattern: {
+                source: [constants.EVENT_SOURCE],
+                detailType: [constants.EVENT_DETAIL_TYPE_CREATE_ENI_MIRROR],
+                detail: {
+                    'vpc_id': events.Match.exactString(props.vpcId)
+                }
+            },
+            targets: [new targets.LambdaFunction(createLambda)]
+        });
+        createRule.node.addDependency(clusterBus);
+
+        // Create the Lambda that will tear down the traffic mirroring for ENIs in our VPC
+        const destroyLambda = new lambda.Function(this, 'DestroyEniMirrorLambda', {
+            functionName: `DestroyEniMirror-${props.vpcId}`,
+            runtime: lambda.Runtime.PYTHON_3_9,
+            code: lambda.Code.fromAsset(path.resolve(__dirname, '..', '..', 'manage_arkime')),            
+            handler: 'lambda_handlers.destroy_eni_mirror_handler',
+            timeout:  Duration.seconds(30), // Something has gone very wrong if this is exceeded            
+        });
+        destroyLambda.addToRolePolicy(
+            new iam.PolicyStatement({
+                effect: iam.Effect.ALLOW,
+                actions: [
+                    // TODO: Should scope this down.
+                    // Just need ec2:DeleteTrafficMirroringSession, but failing similar to the Create Lambda
+                    'ec2:*',
+                ],
+                resources: [
+                    `arn:aws:ec2:${this.region}:${this.account}:*`
+                ]
+            })
+        );
+        destroyLambda.addToRolePolicy(
+            new iam.PolicyStatement({
+                effect: iam.Effect.ALLOW,
+                actions: [
+                    'ssm:GetParameter',
+                    'ssm:DeleteParameter',
+                ],
+                resources: [
+                    `arn:aws:ssm:${this.region}:${this.account}:*`
+                ]
+            })
+        );
+        destroyLambda.addToRolePolicy(
+            new iam.PolicyStatement({
+                effect: iam.Effect.ALLOW,
+                actions: [
+                    'cloudwatch:PutMetricData',
+                ],
+                resources: [
+                    "*"
+                ]
+            })
+        );
+
+        // Create a rule to funnel appropriate events to our teardwon lambda
+        const destroyRule = new events.Rule(this, 'RuleDestroyEniMirror', {
+            eventBus: clusterBus,
+            eventPattern: {
+                source: [constants.EVENT_SOURCE],
+                detailType: [constants.EVENT_DETAIL_TYPE_DESTROY_ENI_MIRROR],
+                detail: {
+                    'vpc_id': events.Match.exactString(props.vpcId)
+                }
+            },
+            targets: [new targets.LambdaFunction(destroyLambda)]
+        });
+        destroyRule.node.addDependency(clusterBus);
     }
 }

manage_arkime.py

@@ -3,16 +3,16 @@
 
 import click
 
-from manage_arkime.commands.add_vpc import cmd_add_vpc
-from manage_arkime.commands.create_cluster import cmd_create_cluster
-from manage_arkime.commands.destroy_cluster import cmd_destroy_cluster
-from manage_arkime.commands.deploy_demo_traffic import cmd_deploy_demo_traffic
-from manage_arkime.commands.destroy_demo_traffic import cmd_destroy_demo_traffic
-from manage_arkime.commands.get_login_details import cmd_get_login_details
-from manage_arkime.commands.list_clusters import cmd_list_clusters
-from manage_arkime.commands.remove_vpc import cmd_remove_vpc
-import manage_arkime.constants as constants
-from manage_arkime.logging_wrangler import LoggingWrangler
+from commands.add_vpc import cmd_add_vpc
+from commands.create_cluster import cmd_create_cluster
+from commands.destroy_cluster import cmd_destroy_cluster
+from commands.deploy_demo_traffic import cmd_deploy_demo_traffic
+from commands.destroy_demo_traffic import cmd_destroy_demo_traffic
+from commands.get_login_details import cmd_get_login_details
+from commands.list_clusters import cmd_list_clusters
+from commands.remove_vpc import cmd_remove_vpc
+import constants as constants
+from logging_wrangler import LoggingWrangler, set_boto_log_level
 
 logger = logging.getLogger(__name__)
 
@@ -127,4 +127,5 @@ def main():
 
 
 if __name__ == "__main__":
-    main()
+    with set_boto_log_level("WARNING"): # Prevent overwhelming boto spam in our debug log
+        main()