通过公网IP查找AWS组织内EC2实例的自动化方案优化咨询

目标 #

通过公网IP自动化查找AWS组织内任意项目/区域的EC2实例信息（公网IP具有唯一性，最多匹配一个实例）。

前提需求 #

组织管理账号下的IAM用户具备AssumeRole权限，可切换至组织内所有账号的指定同名角色。

现有方案逻辑 #

1. 使用AWS Organizations命令（list-roots、list-children）获取组织内所有账号；
2. 依次通过指定角色名AssumeRole切换至各账号：
   • 切换角色后，调用aws ec2 describe-regions获取该账号下的所有可用区域；
   • 遍历所有区域，执行aws ec2 describe-instances --filter="Name=ipaddress,Value=<ip>" --region <region>命令查找对应公网IP的EC2实例；
   • 找到实例后立即终止流程，未找到则返回“未找到”提示。

疑问 #

现有逻辑可行但API调用量过大，已用Python3/Boto3编写脚本，请问该逻辑是否合理？是否有更优实现方式？

现有逻辑的合理性 #

你的基础逻辑是合理且可行的，核心思路（遍历组织账号→切换角色→遍历区域查询）覆盖了所有可能的实例存在范围，能保证找到目标实例（如果存在的话）。但确实存在API调用量过高的问题——尤其是组织账号数量多、区域多的时候，会产生大量串行的API请求，效率偏低。

优化方案 #

1. 并行化请求减少耗时

用Python的concurrent.futures模块（比如ThreadPoolExecutor）实现多账号、多区域的并行查询，大幅压缩总耗时：

• 先一次性获取所有组织账号列表；
• 对每个账号，并行执行AssumeRole+区域查询的流程；
• 每个账号内部，再并行遍历所有区域执行describe-instances查询；
• 一旦任意线程找到目标实例，立即终止所有线程并返回结果。

示例核心代码：

import boto3
from concurrent.futures import ThreadPoolExecutor, as_completed

def check_account(account_id, role_name, target_ip):
    # 切换角色
    sts_client = boto3.client('sts')
    try:
        assumed_role = sts_client.assume_role(
            RoleArn=f"arn:aws:iam::{account_id}:role/{role_name}",
            RoleSessionName="EC2IPLookupSession"
        )
    except Exception:
        return None  # 跳过无权限或不存在角色的账号

    # 生成带临时凭证的EC2客户端
    def get_regional_ec2(region):
        return boto3.client(
            'ec2',
            region_name=region,
            aws_access_key_id=assumed_role['Credentials']['AccessKeyId'],
            aws_secret_access_key=assumed_role['Credentials']['SecretAccessKey'],
            aws_session_token=assumed_role['Credentials']['SessionToken']
        )

    # 获取账号下所有区域
    ec2_client = get_regional_ec2('us-east-1')
    regions = [r['RegionName'] for r in ec2_client.describe_regions()['Regions']]

    # 并行检查每个区域
    def check_region(region):
        try:
            response = get_regional_ec2(region).describe_instances(
                Filters=[{'Name': 'ip-address', 'Values': [target_ip]}]
            )
            if response['Reservations']:
                return (account_id, region, response['Reservations'][0]['Instances'][0])
        except Exception:
            pass
        return None

    with ThreadPoolExecutor(max_workers=len(regions)) as region_exec:
        futures = [region_exec.submit(check_region, r) for r in regions]
        for future in as_completed(futures):
            res = future.result()
            if res:
                return res
    return None

# 主流程
org_client = boto3.client('organizations')
accounts = [acc['Id'] for acc in org_client.list_accounts()['Accounts']]  # 注意处理分页
target_ip = "x.x.x.x"
role_name = "YourSharedRoleName"

with ThreadPoolExecutor(max_workers=10) as account_exec:
    futures = [account_exec.submit(check_account, acc, role_name, target_ip) for acc in accounts]
    for future in as_completed(futures):
        result = future.result()
        if result:
            print(f"找到实例：账号ID={result[0]}, 区域={result[1]}, 实例ID={result[2]['InstanceId']}")
            # 终止所有未完成任务
            for f in futures:
                f.cancel()
            exit()
print("未找到对应公网IP的EC2实例")

2. 利用AWS Config直接跨账号查询

如果你的组织已启用AWS Config跨账号聚合，可以直接通过Config的查询API一次性检索目标实例，无需遍历所有账号和区域：

• Config会自动跨账号、跨区域同步资源配置信息；
• 用SQL风格语句筛选目标IP：

config_client = boto3.client('config')
query = f"""
SELECT accountId, awsRegion, resourceId, configuration
WHERE resourceType = 'AWS::EC2::Instance'
AND configuration.publicIpAddress = '{target_ip}'
"""
response = config_client.select_resource_config(Expression=query)
if response['Results']:
    print("找到实例：", response['Results'][0])
else:
    print("未找到")

注意：此方式需要确保组织内所有账号都已启用Config，且资源记录是最新的（可配置刷新频率）。

3. 缓存常用信息减少重复调用

如果脚本需要频繁执行，可以缓存：

• 组织账号列表（定期刷新，比如每天一次）；
• 各账号的可用区域列表；
  避免每次执行都重复调用list-children、describe-regions等API。

额外注意事项 #

• 处理权限异常：部分账号可能没有指定角色或权限不足，需捕获异常并跳过；
• 控制并发数：避免因并发过高触发AWS API速率限制，可根据实际情况调整max_workers；
• 优先查询常用区域：如果组织有固定的常用区域，可先查询这些区域，找到结果后直接终止，减少无效查询。

内容的提问来源于stack exchange，提问作者Johnny W