背景介绍
随着云计算和容器化技术的普及，AWS ECS Fargate 成为了许多企业在云上运行应用程序的首选服务。然而，在容器化环境中确保应用程序的安全性是至关重要的。本文将介绍如何利用 AWS SDK、Prisma Cloud 和 Python 编写的脚本，以自动化的方式向 AWS ECS Fargate 中的服务添加 Prisma Cloud 安全防护。
Prisma Cloud 的重要性
Prisma Cloud 是一款全面的云原生安全平台，提供实时的安全威胁检测和防护功能。它不仅支持多云环境，还能够提供对容器和云基础设施的全面可见性，帮助用户更好地保护其云上工作负载。
开发过程
1. 导出当前任务定义
使用 AWS SDK for Python (Boto3) 获取指定服务的当前任务定义，并从中删除不必要的字段。这确保了导出的任务定义是干净且仅包含必要信息的。
def export_ecs_task(cluster_name, service_name):
ecs = boto3.client('ecs')
# Get the current task definition ARN
response = ecs.describe_services(cluster=cluster_name, services=[service_name])
task_definition_arn = response['services'][0]['taskDefinition']
# Export the current task definition
response = ecs.describe_task_definition(taskDefinition=task_definition_arn)
task_definition = response['taskDefinition']
# Delete invalid fields from dictionary
for field in ['taskDefinitionArn','revision','volumes','status','requiresAttributes','placementConstraints','compatibilities','registeredAt','registeredBy']:
if field in task_definition.keys():
del task_definition[f'{field}']
return task_definition2. 调用 Prisma Cloud API 添加防护
通过向 Prisma Cloud API 发送 HTTP POST 请求，将导出的任务定义添加到 Prisma Cloud 的防护中。这需要提供正确的身份验证和请求参数。
def ecs_defenders(data,token):
headers = {
'Accept': 'application/json, text/plain, */*',
'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8,zh-TW;q=0.7',
'Authorization': f'Bearer {token}',
'Connection': 'keep-alive',
'Content-Type': 'application/x-www-form-urlencoded',
'Origin': 'https://us-east1.cloud.twistlock.com', #注意更新为自己的
'Sec-Fetch-Dest': 'empty',
'Sec-Fetch-Mode': 'cors',
'Sec-Fetch-Site': 'same-origin',
'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/117.0.0.0 Safari/537.36',
'sec-ch-ua': '"Google Chrome";v="117", "Not;A=Brand";v="8", "Chromium";v="117"',
'sec-ch-ua-mobile': '?0',
'sec-ch-ua-platform': '"macOS"',
}
params = {
'project': 'Central Console',
'consoleaddr': 'us-east1.cloud.twistlock.com', #注意更新为自己的
'interpreter': '',
'cloudFormation': 'false',
'filesystemMonitoring': 'false',
'extractEntrypoint': 'true',
'registryType': 'aws',
'registryCredentialID': '821700710668', #注意更新为自己的云账号ID
}
response = requests.post(
'https://us-east1.cloud.twistlock.com/us-1-111579032/api/v1/defenders/fargate.json', #注意更新为自己的
params=params,
headers=headers,
data=json.dumps(data),
)
return response.text3. 更新 ECS 服务
使用 Boto3 更新 ECS 服务，将新的任务定义应用于指定的服务。
def update_ecs(cluster_name,service_name,task_definition):
ecs = boto3.client('ecs')
response = ecs.register_task_definition(**task_definition)
task_definition_arn = response['taskDefinition']['taskDefinitionArn']
#
# # Update the service to use the new task definition
response = ecs.update_service(cluster=cluster_name, service=service_name, taskDefinition=task_definition_arn)
print(f"{service_name} add twistlock_defender success")4. 判断 Twistlock Defender 是否已存在
在添加之前，通过检查任务定义中是否已经存在 Twistlock Defender 容器来确保不会重复添加。
def judge_twistlock_defender(cluster_name, service_name, region_name=None):
# 创建ECS客户端
ecs_client = boto3.client('ecs', region_name=region_name) # 替换为您的AWS区域
# 获取服务的任务定义
response = ecs_client.describe_services(
cluster=cluster_name,
services=[service_name]
)
task_definition_arn = response['services'][0]['taskDefinition']
# 获取任务定义的详细信息
response = ecs_client.describe_task_definition(
taskDefinition=task_definition_arn
)
task_definition = response['taskDefinition']
# 检查任务定义中是否存在TwistlockDefender容器
containers = task_definition['containerDefinitions']
twistlock_defender_found = any(container['name'] == 'TwistlockDefender' for container in containers)
return twistlock_defender_found5. 获取 ECS 服务列表
通过 Boto3 获取指定 ECS 集群中的所有服务的列表，以便进行批量处理。
def get_service_names(cluster_name,region_name = None):
# 创建ECS客户端
ecs_client = boto3.client('ecs', region_name=region_name) # 替换为您的AWS区域
# 分页获取服务名称
next_token = None
service_names = []
while True:
# 获取一页的服务信息
if next_token:
response = ecs_client.list_services(cluster=cluster_name, nextToken=next_token)
else:
response = ecs_client.list_services(cluster=cluster_name)
services = response['serviceArns']
# 过滤服务名称
for service in services:
service_name = service.split('/')[-1]
service_names.append(service_name)
# 检查是否有下一页
if 'nextToken' in response:
next_token = response['nextToken']
else:
break
return service_names6. 获取 Prisma Cloud 认证 Token
使用用户名和密码通过 Prisma Cloud API 获取认证 Token，确保能够正确向 Prisma Cloud 发送请求。
def get_token(username,password):
url = "https://us-east1.cloud.twistlock.com/us-1-111579032/api/v1/authenticate" #注意更新为自己的
payload = json.dumps({
"password": password,
# "token": "string",
"username": username
})
headers = {
'Content-Type': 'application/json',
'Accept': 'application/json'
}
response = requests.request("POST", url, headers=headers, data=payload)
return json.loads(response.text)["token"]7. 主函数
通过调用以上函数，实现对 ECS Fargate 中所有服务的批量添加 Prisma Cloud 防护。
def main():
cluster_name = 'govee-test' #注意更新为自己的AWS ECS Fargate的集群名称
token = get_token('test@qq.com', 'passwd') #注意更新为自己的prisma cloud账号信息
service_names = get_service_names(cluster_name)
for service_name in service_names:
value = judge_twistlock_defender(cluster_name, service_name, region_name=None)
if not value:
task = export_ecs_task(cluster_name, service_name)
try:
defender_task = ecs_defenders(task,token)
update_ecs(cluster_name, service_name, json.loads(defender_task))
except:
pass
main()结果和效益
通过批量添加 Prisma Cloud 防护，我们实现了对 AWS ECS Fargate 服务的自动化安全防护。这种自动化操作不仅提高了安全性，还提升了运维效率，确保所有服务都能够受益于 Prisma Cloud 的强大安全功能。
总的来说，这种自动化容器安全的实践是在现代云环境中确保应用程序安全性的关键一步。通过结合 AWS 服务、Prisma Cloud 和 Python 脚本，我们成功地实现了批量添加防护，为云原生应用提供了更加坚实的安全基石。