我们希望仅使用 Egress-only internet gateway 和 IPv6 来实现此配置

Source: Dev.to

使用仅限出口的 IPv6 Internet 网关的 VPC 配置

import * as cdk from 'aws-cdk-lib';
import { Construct } from 'constructs';
import { SubnetType, CfnEIP, CfnSubnet } from 'aws-cdk-lib/aws-ec2';
import * as aws_ec2 from '@aws-cdk/aws-ec2-alpha';
import { IpAddressType } from 'aws-cdk-lib/aws-elasticloadbalancingv2';
import { AvailabilityZoneRebalancing } from 'aws-cdk-lib/aws-ecs';

export class VPCSubnet extends Construct {
  constructor(scope: Construct, id: string) {
    super(scope, id);

    const myVpc = new aws_ec2.VpcV2(this, 'Vpc', {
      primaryAddressBlock: aws_ec2.IpAddresses.ipv4('10.1.0.0/16'),
      vpcName: 'vpc-Develop-Network',
      secondaryAddressBlocks: [
        aws_ec2.IpAddresses.amazonProvidedIpv6({
          cidrBlockName: 'AmazonProvided'
        })
      ]
    });

    // Internet Gateway for Public Subnet
    const igw = new aws_ec2.InternetGateway(this, 'IGW', {
      vpc: myVpc
    });

    // Egress‑Only Internet Gateway for IPv6
    const eigw = new aws_ec2.EgressOnlyInternetGateway(this, 'EIGW', {
      vpc: myVpc
    });

    // Public Subnets (Multi‑AZ)
    const publicRouteTable = new aws_ec2.RouteTable(this, 'PublicRouteTable', {
      vpc: myVpc
    });
    publicRouteTable.addRoute('PublicIGWRoute', '0.0.0.0/0', { gateway: igw });

    const publicSubnetAZ1Name = cdk.Fn.join('', ['sub-public-', cdk.Fn.select(0, cdk.Fn.getAzs())]);
    const publicSubnetAZ1 = new aws_ec2.SubnetV2(this, 'PublicSubnetAZ1', {
      vpc: myVpc,
      subnetName: publicSubnetAZ1Name,
      availabilityZone: cdk.Fn.select(0, cdk.Fn.getAzs()),
      ipv4CidrBlock: new aws_ec2.IpCidr('10.1.0.0/24'),
      subnetType: SubnetType.PUBLIC,
      routeTable: publicRouteTable
    });
    const cfnPublicSubnetAZ1 = publicSubnetAZ1.node.defaultChild as CfnSubnet;
    cdk.Tags.of(cfnPublicSubnetAZ1).add('Name', publicSubnetAZ1Name);

    const publicSubnetAZ2Name = cdk.Fn.join('', ['sub-public-', cdk.Fn.select(1, cdk.Fn.getAzs())]);
    const publicSubnetAZ2 = new aws_ec2.SubnetV2(this, 'PublicSubnetAZ2', {
      vpc: myVpc,
      subnetName: publicSubnetAZ2Name,
      availabilityZone: cdk.Fn.select(1, cdk.Fn.getAzs()),
      ipv4CidrBlock: new aws_ec2.IpCidr('10.1.1.0/24'),
      subnetType: SubnetType.PUBLIC,
      routeTable: publicRouteTable
    });
    const cfnPublicSubnetAZ2 = publicSubnetAZ2.node.defaultChild as CfnSubnet;
    cdk.Tags.of(cfnPublicSubnetAZ2).add('Name', publicSubnetAZ2Name);

    // Private Subnets (Multi‑AZ) with IPv6 egress
    const privateRouteTable = new aws_ec2.RouteTable(this, 'PrivateRouteTable', {
      vpc: myVpc
    });
    privateRouteTable.addRoute('PrivateEIGWRoute', '::/0', { gateway: eigw });

    const privateSubnetAZ1Name = cdk.Fn.join('', ['sub-private-', cdk.Fn.select(0, cdk.Fn.getAzs())]);
    const privateSubnetAZ1 = new aws_ec2.SubnetV2(this, 'PrivateSubnetAZ1', {
      vpc: myVpc,
      subnetName: privateSubnetAZ1Name,
      availabilityZone: cdk.Fn.select(0, cdk.Fn.getAzs()),
      ipv4CidrBlock: new aws_ec2.IpCidr('10.1.2.0/24'),
      ipv6CidrBlock: new aws_ec2.IpCidr(
        cdk.Fn.select(2, cdk.Fn.cidr(cdk.Fn.select(0, myVpc.ipv6CidrBlocks), 4, '64'))
      ),
      subnetType: SubnetType.PRIVATE_WITH_EGRESS,
      routeTable: privateRouteTable
    });
    const cfnPrivateSubnetAZ1 = privateSubnetAZ1.node.defaultChild as CfnSubnet;
    cdk.Tags.of(cfnPrivateSubnetAZ1).add('Name', privateSubnetAZ1Name);

    const privateSubnetAZ2Name = cdk.Fn.join('', ['sub-private-', cdk.Fn.select(1, cdk.Fn.getAzs())]);
    const privateSubnetAZ2 = new aws_ec2.SubnetV2(this, 'PrivateSubnetAZ2', {
      vpc: myVpc,
      subnetName: privateSubnetAZ2Name,
      availabilityZone: cdk.Fn.select(1, cdk.Fn.getAzs()),
      ipv4CidrBlock: new aws_ec2.IpCidr('10.1.3.0/24'),
      ipv6CidrBlock: new aws_ec2.IpCidr(
        cdk.Fn.select(3, cdk.Fn.cidr(cdk.Fn.select(0, myVpc.ipv6CidrBlocks), 4, '64'))
      ),
      subnetType: SubnetType.PRIVATE_WITH_EGRESS,
      routeTable: privateRouteTable
    });
    const cfnPrivateSubnetAZ2 = privateSubnetAZ2.node.defaultChild as CfnSubnet;
    cdk.Tags.of(cfnPrivateSubnetAZ2).add('Name', privateSubnetAZ2Name);
  }
}

使用 Lambda 验证 IPv6 连通性

下面的 Lambda 函数通过调用 ifconfig.me 来获取源 IPv6 地址。请确保子网已启用 “Allow IPv6 traffic for dual‑stack subnets”。

import json
import urllib.request

def lambda_handler(event, context):
    """
    Calls http://ifconfig.me/ to obtain the caller's IPv6 address.
    """
    try:
        with urllib.request.urlopen('http://ifconfig.me/', timeout=10) as response:
            ip_address = response.read().decode('utf-8').strip()

        return {
            'statusCode': 200,
            'body': json.dumps({
                'message': 'Successfully retrieved IP address',
                'ip_address': ip_address
            })
        }

    except urllib.error.URLError as e:
        return {
            'statusCode': 500,
            'body': json.dumps({
                'message': 'Error retrieving IP address',
                'error': str(e)
            })
        }

    except Exception as e:
        return {
            'statusCode': 500,
            'body': json.dumps({
                'message': 'Unexpected error occurred',
                'error': str(e)
            })
        }

# Local test
if __name__ == '__main__':
    result = lambda_handler({}, {})
    print(json.dumps(result, indent=2, ensure_ascii=False))

示例响应

{
  "statusCode": 200,
  "body": "{\"message\": \"Successfully retrieved IP address\", \"ip_address\": \"2600:1f13:94:1a02:ef65:ac4:50ab:5e1\"}"
}

AWS 服务配置

在使用 IPv6 的 AWS SDK 时，请启用双栈端点：

my_config = Config(
    use_dualstack_endpoint=True,
    retries={'max_attempts': 3, 'mode': 'standard'}
)

更多细节请参阅 AWS 文档。

S3 示例（支持 IPv6）

my_config = Config(
    use_dualstack_endpoint=True,
    retries={'max_attempts': 3, 'mode': 'standard'}
)

bucket_name = ''
try:
    s3_resource = boto3.resource("s3", config=my_config)
    buckets = list(s3_resource.buckets.all())
    for bucket in buckets:
        bucket_name = bucket.name
        logger.info(bucket_name)
except Exception as e:
    logger.error(f"Error accessing S3: {e}")