1

AWS ECS Fargate 批量禁止公网访问

 1 month ago
source link: https://blog.51cto.com/jiemei/10514233
Go to the source link to view the article. You can view the picture content, updated content and better typesetting reading experience. If the link is broken, please click the button below to view the snapshot at that time.

AWS ECS Fargate 批量禁止公网访问

精选 原创

IvwDcwSo 2024-04-18 20:50:53 博主文章分类：运维 ©著作权

文章标签 AWS Fargate ecs 安全 运维 文章分类 运维 yyds干货盘点 阅读数264

在 AWS 上部署容器化应用时,使用 ECS Fargate 是一个很好的选择。Fargate 是一种无需管理底层基础设施的容器化计算引擎,可以轻松运行容器,而无需预置或管理 EC2 实例。

然而,默认情况下,Fargate 任务使用公共子网,这意味着容器可以直接访问互联网。虽然这对某些应用程序来说是必需的,但从安全角度来看,最好将容器与互联网隔离开来,只允许出站流量。

在本文中,我们将介绍如何使用 Python 脚本批量禁止 ECS Fargate 服务的公网访问,并使用 NAT 网关解决镜像拉取问题。

  • AWS 账户和适当的 IAM 权限
  • Python 3.x 及其 boto3 库
  • 已创建的 ECS 集群和服务

我们将编写一个 Python 脚本,该脚本可以:

  1. 列出指定 ECS 集群中的所有服务
  2. 检查每个服务的网络配置
  3. 如果服务使用公共子网,则更新其配置以使用私有子网和 NAT 网关
  4. 在更新配置后,等待一段时间再处理下一个服务,以避免频繁更新导致的问题
import boto3
import time

ecs_client = boto3.client('ecs', region_name='us-east-1')
ec2_client = boto3.client('ec2', region_name='us-east-1')

# 定义睡眠时间(以秒为单位)
sleep_time = 60

# 定义 VPC 对应的私有子网和安全组配置
vpc_config = {
    'vpc-97f8a9f1': {
        'subnets': ['subnet-03271df83933bc34c', 'subnet-0ba1ee5da9fa8a5fe'],
        'security_groups': ['sg-023218fbc132324f0']
    },
    'vpc-01bc326c162920649': {
        'subnets': ['subnet-0406036d337ad20aa', 'subnet-01f8076d8bb161937', 'subnet-0b7927873dfe98061'],
        'security_groups': ['sg-011d6a642cf4adddb']
    }
}

def get_cluster_services(cluster):
    services = []
    next_token = None

    while True:
        if next_token:
            response = ecs_client.list_services(cluster=cluster, nextToken=next_token)
        else:
            response = ecs_client.list_services(cluster=cluster)

        services.extend(response['serviceArns'])

        if 'nextToken' in response:
            next_token = response['nextToken']
        else:
            break

    return services

def get_vpc_id_from_subnet(subnet_id):
    response = ec2_client.describe_subnets(SubnetIds=[subnet_id])
    vpc_id = response['Subnets'][0]['VpcId']
    return vpc_id

def update_service_subnet(cluster, service):
    # 获取服务信息
    response = ecs_client.describe_services(
        cluster=cluster,
        services=[service]
    )
    service_info = response['services'][0]

    # 获取服务所在的子网
    subnet_ids = service_info.get('networkConfiguration', {}).get('awsvpcConfiguration', {}).get('subnets', [])

    # 如果服务关联了子网
    if subnet_ids:
        # 获取 VPC ID
        vpc_id = get_vpc_id_from_subnet(subnet_ids[0])

        # 如果服务所在的 VPC ID 在配置中存在
        if vpc_id in vpc_config:
            vpc_config_data = vpc_config[vpc_id]
            subnets = vpc_config_data['subnets']
            security_groups = vpc_config_data['security_groups']

            # 更新服务的子网和安全组配置
            response = ecs_client.update_service(
                cluster=cluster,
                service=service,
                networkConfiguration={
                    'awsvpcConfiguration': {
                        'subnets': subnets,
                        'securityGroups': security_groups,
                        'assignPublicIp': 'DISABLED'
                    }
                }
            )
            print(f"Updated service {service} subnet and security groups for cluster {cluster}")
            time.sleep(sleep_time)
        else:
            print(f"VPC ID {vpc_id} not found in the configuration")
    else:
        print(f"Service {service} is not associated with any subnet")

# 示例用法
clusters = ['cluster1','cluster2','cluster3']

for cluster in clusters:
    services = get_cluster_services(cluster)
    for service in services:
        update_service_subnet(cluster, service)
  1. 首先,我们导入必要的 AWS SDK 库并初始化 ECS 和 EC2 客户端。
  2. 定义一个睡眠时间变量,用于在更新每个服务后等待一段时间,避免频繁更新导致的问题。
  3. 定义一个 vpc_config 字典,用于存储每个 VPC 对应的私有子网和安全组配置。你需要根据实际情况填写正确的子网 ID 和安全组 ID。
  4. get_cluster_services 函数用于列出指定 ECS 集群中的所有服务 ARN。
  5. get_vpc_id_from_subnet 函数用于从子网 ID 获取 VPC ID。
  6. update_service_subnet 函数是主要逻辑所在。它会获取服务的网络配置,如果服务使用公共子网,则根据 vpc_config 中的配置更新服务的子网和安全组,并禁用公网 IP。更新后,它会等待指定的睡眠时间。
  7. 最后,我们定义了一个示例集群列表,并遍历每个集群和服务,调用 update_service_subnet 函数进行更新。
  1. 将上述脚本保存为 Python 文件,例如 update_ecs_service_subnets.py。
  2. 根据实际情况,在 vpc_config 字典中填写正确的私有子网 ID 和安全组 ID。
  3. 根据需要,修改 clusters 列表中的集群名称。
  4. 确保您的 AWS 凭证已正确配置,或者使用适当的角色或实例配置文件。
  5. 运行脚本:
python update_ecs_service_subnets.py
  1. 脚本将逐个处理每个集群中的服务,如果服务使用公共子网,则更新其配置以使用私有子网和 NAT 网关。
  2. 更新完成后,您的 ECS Fargate 服务将无法直接访问互联网,但可以通过 NAT 网关拉取 Docker 镜像和访问其他出站资源。

通过执行此脚本,您可以批量禁止 ECS Fargate 服务的公网访问,提高应用程序的安全性,同时使用 NAT 网关解决镜像拉取问题。请注意,根据您的服务数量和集群规模,此过程可能需要一些时间。

  • 赞
  • 收藏
  • 评论
  • 分享
  • 举报

report

Recommend

About Joyk


Aggregate valuable and interesting links.
Joyk means Joy of geeK