如何使用boto3验证两台EC2实例指定端口/协议的连通性

要确保实例A和实例B（同账号/跨账号）能通过指定端口/协议通信，核心要检查**安全组（SG）和网络访问控制列表（NACL）**的规则，以下是具体实操步骤和代码实现：

一、前置准备 #

• 确保已配置好Boto3的认证（同账号用默认凭证，跨账号需切换对应凭证或使用IAM角色）
• 明确通信方向：单向（A→B）或双向，以及目标端口（如80、443）和协议（tcp/udp/icmp）

二、核心检查步骤 #

1. 获取实例基础信息 #

先拿到两台实例的子网ID、安全组ID、私有IP/公有IP（跨账号通信通常用公有IP，同账号内网优先用私有IP）：

import boto3

def get_instance_details(ec2_client, instance_id):
    response = ec2_client.describe_instances(InstanceIds=[instance_id])
    instance = response['Reservations'][0]['Instances'][0]
    return {
        'subnet_id': instance['SubnetId'],
        'sg_ids': [sg['GroupId'] for sg in instance['SecurityGroups']],
        'private_ip': instance['PrivateIpAddress'],
        'public_ip': instance.get('PublicIpAddress')
    }

# 示例：初始化EC2客户端（跨账号需指定对应region或凭证）
ec2_a = boto3.client('ec2', region_name='us-east-1')
ec2_b = boto3.client('ec2', region_name='us-east-1')  # 跨账号场景替换为对应账号的客户端配置

instance_a_info = get_instance_details(ec2_a, 'i-xxxxxx')
instance_b_info = get_instance_details(ec2_b, 'i-yyyyyy')

2. 检查安全组规则 #

安全组是状态化的，单向通信只需检查：

• A的出站规则允许访问B的目标IP+端口+协议
• B的入站规则允许来自A的源IP+端口+协议

双向通信需额外检查B的出站和A的入站，以下是通用检查函数：

def check_sg_rule(ec2_client, sg_ids, direction, target_ip, port, protocol):
    # direction: 'ingress' 或 'egress'
    for sg_id in sg_ids:
        sg = ec2_client.describe_security_groups(GroupIds=[sg_id])['SecurityGroups'][0]
        for rule in sg[f'{direction}IpPermissions']:
            # 匹配协议（tcp=6, udp=17, icmp=1，支持数字或字符串）
            if rule['IpProtocol'] != '-1' and rule['IpProtocol'] != str(protocol):
                continue
            # 匹配端口范围（仅tcp/udp）
            if protocol in [6, 17, 'tcp', 'udp']:
                from_port = rule.get('FromPort', -1)
                to_port = rule.get('ToPort', -1)
                if not (from_port <= port <= to_port):
                    continue
            # 匹配IP范围（CIDR）
            for ip_range in rule.get('IpRanges', []):
                if cidr_contains(ip_range['CidrIp'], target_ip):
                    return True
            # 匹配安全组引用（同账号同VPC场景）
            for sg_range in rule.get('UserIdGroupPairs', []):
                target_sgs = instance_b_info['sg_ids'] if direction == 'egress' else instance_a_info['sg_ids']
                if sg_range['GroupId'] in target_sgs:
                    return True
    return False

# 辅助函数：判断IP是否在CIDR范围内
def cidr_contains(cidr, ip):
    import ipaddress
    return ipaddress.ip_address(ip) in ipaddress.ip_network(cidr, strict=False)

3. 检查NACL规则 #

NACL是非状态化的，需同时检查入站（允许源IP→目标IP）和出站（允许目标IP→源IP），规则按优先级执行（数字越小优先级越高，拒绝规则优先于允许）：

def check_nacl_rule(ec2_client, subnet_id, direction, source_ip, dest_ip, port, protocol):
    # direction: 'ingress' 或 'egress'
    subnet = ec2_client.describe_subnets(SubnetIds=[subnet_id])['Subnets'][0]
    nacl_id = subnet['NetworkAclId']
    nacl = ec2_client.describe_network_acls(NetworkAclIds=[nacl_id])['NetworkAcls'][0]
    
    # 按优先级排序规则
    rules = sorted(nacl['Entries'], key=lambda x: x['RuleNumber'])
    allow_found = False
    for rule in rules:
        if rule['Egress'] != (direction == 'egress'):
            continue
        # 跳过默认全允许规则（规则编号32767）
        if rule['RuleNumber'] == 32767:
            allow_found = True
            break
        # 匹配协议
        if rule['Protocol'] != '-1' and rule['Protocol'] != str(protocol):
            continue
        # 匹配端口范围（仅tcp/udp）
        if protocol in [6, 17, 'tcp', 'udp']:
            from_port = int(rule.get('PortRange', {}).get('From', -1))
            to_port = int(rule.get('PortRange', {}).get('To', -1))
            if not (from_port <= port <= to_port):
                continue
        # 匹配源/目标CIDR
        if direction == 'ingress':
            if not cidr_contains(rule['CidrBlock'], source_ip):
                continue
        else:
            if not cidr_contains(rule['CidrBlock'], dest_ip):
                continue
        # 检查规则动作
        if rule['RuleAction'] == 'deny':
            return False
        elif rule['RuleAction'] == 'allow':
            allow_found = True
    return allow_found

4. 组合检查逻辑 #

以A→B的TCP 80端口通信为例，组合上述函数完成验证：

# 确定通信IP（同账号用私有IP，跨账号替换为公有IP）
source_ip = instance_a_info['private_ip']
dest_ip = instance_b_info['private_ip']
port = 80
protocol = 'tcp'

# 检查安全组
sg_ok = check_sg_rule(ec2_a, instance_a_info['sg_ids'], 'egress', dest_ip, port, protocol) and \
        check_sg_rule(ec2_b, instance_b_info['sg_ids'], 'ingress', source_ip, port, protocol)

# 检查NACL
nacl_ok = check_nacl_rule(ec2_a, instance_a_info['subnet_id'], 'egress', source_ip, dest_ip, port, protocol) and \
          check_nacl_rule(ec2_b, instance_b_info['subnet_id'], 'ingress', source_ip, dest_ip, port, protocol)

if sg_ok and nacl_ok:
    print(f"实例A与B可通过TCP {port}端口通信")
else:
    print(f"实例A与B无法通过TCP {port}端口通信，请检查安全组或NACL规则")

三、注意事项 #

• 跨账号场景：若使用VPC对等连接，需用私有IP，但要额外检查对等连接的路由表配置
• ICMP协议：需匹配对应类型和代码（如ping用类型8、代码0），需在检查函数中补充对应逻辑
• 安全组引用：仅支持同账号同VPC的安全组，跨账号需用CIDR指定对方IP
• NACL默认规则：默认入站和出站均拒绝所有，需手动添加允许规则

内容的提问来源于stack exchange，提问作者Ishimwe etienne