Add ALB and Target Cluster Proxy

deployment/cdk/opensearch-service-migration/default-values.json

@@ -14,5 +14,7 @@
   "migrationConsoleServiceEnabled": true,
   "trafficReplayerServiceEnabled": true,
   "otelCollectorEnabled": true,
-  "dpPipelineTemplatePath": "./dp_pipeline_template.yaml"
+  "dpPipelineTemplatePath": "./dp_pipeline_template.yaml",
+  "albEnabled": false,
+  "targetClusterProxyServiceEnabled": false
 }

deployment/cdk/opensearch-service-migration/lib/common-utilities.ts

@@ -3,6 +3,8 @@ import {Construct} from "constructs";
 import {StringParameter} from "aws-cdk-lib/aws-ssm";
 import {CpuArchitecture} from "aws-cdk-lib/aws-ecs";
 import {RemovalPolicy} from "aws-cdk-lib";
+import { IApplicationLoadBalancer } from "aws-cdk-lib/aws-elasticloadbalancingv2";
+import { ICertificate } from "aws-cdk-lib/aws-certificatemanager";
 
 export function createOpenSearchIAMAccessPolicy(partition: string, region: string, accountId: string): PolicyStatement {
     return new PolicyStatement({
@@ -149,4 +151,18 @@ export function parseRemovalPolicy(optionName: string, policyNameString?: string
         throw new Error(`Provided '${optionName}' with value '${policyNameString}' does not match a selectable option, for reference https://docs.aws.amazon.com/cdk/api/v2/docs/aws-cdk-lib.RemovalPolicy.html`)
     }
     return policy
+}
+
+
+export type ALBConfig = NewALBListenerConfig;
+
+export interface NewALBListenerConfig {
+    alb: IApplicationLoadBalancer,
+    albListenerCert: ICertificate,
+    albListenerPort?: number,
+}
+
+export function isNewALBListenerConfig(config: ALBConfig): config is NewALBListenerConfig {
+    const parsed = config as NewALBListenerConfig;
+    return parsed.alb !== undefined && parsed.albListenerCert !== undefined;
 }

deployment/cdk/opensearch-service-migration/lib/lambda/acm-cert-importer-handler.ts

@@ -0,0 +1,176 @@
+import {
+  Context,
+  CloudFormationCustomResourceEvent,
+  CloudFormationCustomResourceResponse,
+} from 'aws-lambda';
+import { ACMClient, ImportCertificateCommand, DeleteCertificateCommand } from '@aws-sdk/client-acm';
+import * as https from 'https';
+import * as forge from 'node-forge';
+
+export const handler = async (event: CloudFormationCustomResourceEvent, context: Context): Promise<CloudFormationCustomResourceResponse> => {
+  console.log('Received event type:', event.RequestType);
+  console.log('Received context:', JSON.stringify(context, null, 2));
+  let responseData: { [key: string]: any } = {};
+  let physicalResourceId: string = '';
+
+  try {
+    switch (event.RequestType) {
+      case 'Create':
+        const { certificate, privateKey, certificateChain } = await generateSelfSignedCertificate();
+        const certificateArn = await importCertificate(certificate, privateKey, certificateChain);
+        console.log(`Certificate imported with ARN: ${certificateArn}`);
+        responseData = { CertificateArn: certificateArn };
+        physicalResourceId = certificateArn;
+        break;
+      case 'Update':
+        // No update logic needed, return existing physical resource id
+        physicalResourceId = event.PhysicalResourceId;
+        break;  
+      case 'Delete':
+        const arn = event.PhysicalResourceId;
+        await deleteCertificate(arn);
+        responseData = { CertificateArn: arn };
+        physicalResourceId = arn;
+        break;
+    }
+
+    return await sendResponse(event, context, 'SUCCESS', responseData, physicalResourceId);
+  } catch (error) {
+    console.error(error);
+    return await sendResponse(event, context, 'FAILED', { Error: (error as Error).message }, physicalResourceId);
+  }
+};
+
+async function generateSelfSignedCertificate(): Promise<{ certificate: string, privateKey: string, certificateChain: string }> {
+  return new Promise((resolve, reject) => {
+    const keys = forge.pki.rsa.generateKeyPair(2048);
+    const cert = forge.pki.createCertificate();
+
+    cert.publicKey = keys.publicKey;
+    cert.serialNumber = '01';
+    cert.validity.notBefore = new Date(Date.UTC(1970, 0, 1, 0, 0, 0));
+    cert.validity.notAfter = new Date(Date.UTC(9999, 11, 31, 23, 59, 59, 999));
+    const attrs = [{
+      name: 'commonName',
+      value: 'localhost'
+    }];
+
+    cert.setSubject(attrs);
+    cert.setIssuer(attrs);
+
+    cert.setExtensions([{
+      name: 'basicConstraints',
+      cA: true
+    }, {
+      name: 'subjectAltName',
+      altNames: [{
+        type: 2, // DNS
+        value: 'localhost'
+      }]
+    }, {
+      name: 'keyUsage',
+      keyCertSign: true,
+      digitalSignature: true,
+      keyEncipherment: true
+    }, {
+      name: 'extKeyUsage',
+      serverAuth: true,
+      clientAuth: true
+    },]);
+    cert.sign(keys.privateKey, forge.md.sha384.create());
+
+    const pemCert = forge.pki.certificateToPem(cert);
+    const pemKey = forge.pki.privateKeyToPem(keys.privateKey);
+
+    resolve({
+      certificate: pemCert,
+      privateKey: pemKey,
+      certificateChain: pemCert
+    });
+  });
+}
+
+async function importCertificate(certificate: string, privateKey: string, certificateChain: string): Promise<string> {
+  const client = new ACMClient({ region: process.env.AWS_REGION });
+
+  const command = new ImportCertificateCommand({
+    Certificate: Buffer.from(certificate),
+    PrivateKey: Buffer.from(privateKey),
+    CertificateChain: Buffer.from(certificateChain),
+  });
+
+  const response = await client.send(command);
+  return response.CertificateArn!;
+}
+
+async function deleteCertificate(certificateArn: string): Promise<void> {
+  const client = new ACMClient({ region: process.env.AWS_REGION });
+  const command = new DeleteCertificateCommand({
+    CertificateArn: certificateArn,
+  });
+
+  await client.send(command);
+}
+
+async function sendResponse(event: CloudFormationCustomResourceEvent, context: Context, responseStatus: string, responseData: { [key: string]: any }, physicalResourceId: string): Promise<CloudFormationCustomResourceResponse> {
+  const responseBody = JSON.stringify({
+    Status: responseStatus,
+    Reason: `See the details in CloudWatch Log Stream: ${context.logStreamName}`,
+    PhysicalResourceId: physicalResourceId,
+    StackId: event.StackId,
+    RequestId: event.RequestId,
+    LogicalResourceId: event.LogicalResourceId,
+    Data: responseData
+  });
+
+  console.log('Response body:', responseBody);
+
+  if (event.ResponseURL === "Local") {
+    console.log('Running locally, simulating response success.');
+    return {
+      Status: 'SUCCESS',
+      Reason: 'Running locally, response simulated.',
+      PhysicalResourceId: physicalResourceId,
+      StackId: event.StackId,
+      RequestId: event.RequestId,
+      LogicalResourceId: event.LogicalResourceId,
+      Data: responseData
+    };
+  }
+
+  const parsedUrl = new URL(event.ResponseURL);
+  const options = {
+    hostname: parsedUrl.hostname,
+    port: 443,
+    path: parsedUrl.pathname + parsedUrl.search,
+    method: 'PUT',
+    headers: {
+      'Content-Type': '',
+      'Content-Length': responseBody.length
+    }
+  };
+
+  return new Promise<CloudFormationCustomResourceResponse>((resolve, reject) => {
+    const request = https.request(options, (response) => {
+      console.log(`Status code: ${response.statusCode}`);
+      console.log(`Status message: ${response.statusMessage}`);
+      resolve({
+        Status: responseStatus === 'SUCCESS' || responseStatus === 'FAILED' ? responseStatus : 'FAILED',
+        Reason: `See the details in CloudWatch Log Stream: ${context.logStreamName}`,
+        PhysicalResourceId: physicalResourceId,
+        StackId: event.StackId,
+        RequestId: event.RequestId,
+        LogicalResourceId: event.LogicalResourceId,
+        Data: responseData
+      });
+    });
+
+    request.on('error', (error) => {
+      console.error('sendResponse Error:', error);
+      reject(error);
+    });
+
+    request.write(responseBody);
+    request.end();
+  });
+}

deployment/cdk/opensearch-service-migration/lib/lambda/index.ts

@@ -1,2 +1,3 @@
 export * as msk_ordered_endpoints_handler from './msk-ordered-endpoints-handler'
-export * as msk_public_endpoint_handler from './msk-public-endpoint-handler'
+export * as msk_public_endpoint_handler from './msk-public-endpoint-handler'
+export * as acm_cert_importer_handler from './acm-cert-importer-handler'

deployment/cdk/opensearch-service-migration/lib/network-stack.ts

@@ -7,15 +7,25 @@ import {
 import {Construct} from "constructs";
 import {StackPropsExt} from "./stack-composer";
 import {StringParameter} from "aws-cdk-lib/aws-ssm";
+import { ApplicationLoadBalancer, IApplicationLoadBalancer } from "aws-cdk-lib/aws-elasticloadbalancingv2";
+import { Certificate, ICertificate } from "aws-cdk-lib/aws-certificatemanager";
+import { ARecord, HostedZone, RecordTarget } from "aws-cdk-lib/aws-route53";
+import { LoadBalancerTarget } from "aws-cdk-lib/aws-route53-targets";
+import { AcmCertificateImporter } from "./service-stacks/acm-cert-importer";
 
 export interface NetworkStackProps extends StackPropsExt {
-    readonly vpcId?: string
-    readonly vpcAZCount?: number
-    readonly targetClusterEndpoint?: string
+    readonly vpcId?: string;
+    readonly vpcAZCount?: number;
+    readonly targetClusterEndpoint?: string;
+    readonly albEnabled?: boolean;
+    readonly albAcmCertArn?: string;
+    readonly env?: { [key: string]: any };
 }
 
 export class NetworkStack extends Stack {
     public readonly vpc: IVpc;
+    public readonly alb?: IApplicationLoadBalancer;
+    public readonly albListenerCert?: ICertificate
 
     // Validate a proper url string is provided and return an url string which contains a protocol, host name, and port.
     // If a port is not provided, the default protocol port (e.g. 443, 80) will be explicitly added
@@ -104,6 +114,39 @@ export class NetworkStack extends Stack {
         }
         this.validateVPC(this.vpc)
 
+        if (props.albEnabled) {
+            // Create the ALB with the strongest TLS 1.3 security policy
+            this.alb = new ApplicationLoadBalancer(this, 'ALB', {
+                vpc: this.vpc,
+                internetFacing: false,
+                http2Enabled: false,      
+            });
+            this.exportValue(this.alb.loadBalancerArn);
+            if (props.albAcmCertArn) {
+                const cert = Certificate.fromCertificateArn(this, 'ALBListenerCert', props.albAcmCertArn);
+                this.albListenerCert = cert;
+            } else {
+                const cert = new AcmCertificateImporter(this, 'ALBListenerCertImport').acmCert;
+                this.albListenerCert = cert;
+            }
+
+            const route53 = new HostedZone(this, 'ALBHostedZone', {
+                zoneName: `alb.migration.${props.stage}.local`,
+                vpcs: [this.vpc]
+            });
+
+            const albDnsRecord = new ARecord(this, 'albDnsRecord', {
+                zone: route53,
+                target: RecordTarget.fromAlias(new LoadBalancerTarget(this.alb)),
+            });
+
+            new StringParameter(this, 'SSMParameterAlbUrl', {
+                description: 'OpenSearch migration parameter for ALB to migration services',
+                parameterName: `/migration/${props.stage}/${props.defaultDeployId}/albMigrationUrl`,
+                stringValue: `https://${albDnsRecord.domainName}`
+            });
+        }
+
         if (!props.addOnMigrationDeployId) {
             new StringParameter(this, 'SSMParameterVpcId', {
                 description: 'OpenSearch migration parameter for VPC id',

deployment/cdk/opensearch-service-migration/lib/service-stacks/acm-cert-importer.ts

@@ -0,0 +1,59 @@
+import { Construct } from 'constructs';
+import { Certificate, ICertificate } from "aws-cdk-lib/aws-certificatemanager";
+import { Provider } from 'aws-cdk-lib/custom-resources';
+import { NodejsFunction } from 'aws-cdk-lib/aws-lambda-nodejs';
+import { CustomResource, Duration, RemovalPolicy, Stack } from 'aws-cdk-lib/core';
+import * as path from 'path';
+import { Runtime } from 'aws-cdk-lib/aws-lambda';
+import { PolicyStatement, Role, ServicePrincipal } from 'aws-cdk-lib/aws-iam';
+import { LogGroup, RetentionDays } from 'aws-cdk-lib/aws-logs';
+
+export class AcmCertificateImporter extends Construct {
+    public readonly acmCert: ICertificate;
+
+    constructor(scope: Construct, id: string) {
+        super(scope, id);
+
+        // Create a role for the Lambda function with the necessary permissions
+        const lambdaRole = new Role(this, 'AcmCertificateImporterRole', {
+            assumedBy: new ServicePrincipal('lambda.amazonaws.com'),
+        });
+
+        const partition = Stack.of(this).partition;
+        lambdaRole.addManagedPolicy({
+            managedPolicyArn: `arn:${partition}:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole`
+        });
+        lambdaRole.addManagedPolicy({
+            managedPolicyArn: `arn:${partition}:iam::aws:policy/service-role/AWSLambdaVPCAccessExecutionRole`
+        });
+        lambdaRole.addToPolicy(new PolicyStatement({
+            actions: ['acm:ImportCertificate', 'acm:DeleteCertificate'],
+            resources: ['*'],
+        }));
+
+        const onEvent = new NodejsFunction(this, 'AcmCertificateImporterHandler', {
+            runtime: Runtime.NODEJS_20_X,
+            handler: 'handler',
+            entry: path.join(__dirname, '../lambda/acm-cert-importer-handler.ts'),
+            environment: {
+            },
+            timeout: Duration.seconds(30),
+            role: lambdaRole,
+            logGroup: new LogGroup(this, 'AcmCertificateImporterLogGroup', {
+                logGroupName: `/aws/lambda/AcmCertificateImporterHandler-${id}`,
+                removalPolicy: RemovalPolicy.DESTROY,
+                retention: RetentionDays.ONE_DAY
+            })
+        });
+
+        const myProvider = new Provider(this, 'AcmCertificateImporterProvider'+id, {
+            onEventHandler: onEvent,
+        });
+
+        const resource = new CustomResource(this, 'AcmCertificateImporterResource' + id, {
+            serviceToken: myProvider.serviceToken,
+        });
+        const acmCertArn = resource.ref;
+        this.acmCert = Certificate.fromCertificateArn(this, "ImportedCert"+id, acmCertArn);
+    }
+}