from __future__ import annotations
from typing import Any, Dict, List
import boto3
from aws_mcp_audit.normalize.redaction import redact_tags
def collect_ec2_region(session: boto3.Session, region: str) -> Dict[str, Any]:
ec2 = session.client("ec2", region_name=region)
instances: List[Dict[str, Any]] = []
paginator = ec2.get_paginator("describe_instances")
for page in paginator.paginate():
for res in page.get("Reservations", []):
for inst in res.get("Instances", []):
tags = {t["Key"]: t["Value"] for t in inst.get("Tags", [])}
tags = redact_tags(tags)
instances.append(
{
"instance_id": inst.get("InstanceId"),
"state": (inst.get("State") or {}).get("Name"),
"instance_type": inst.get("InstanceType"),
"az": inst.get("Placement", {}).get("AvailabilityZone"),
"public_ip": inst.get("PublicIpAddress"),
"private_ip": inst.get("PrivateIpAddress"),
"vpc_id": inst.get("VpcId"),
"subnet_id": inst.get("SubnetId"),
"security_group_ids": [sg.get("GroupId") for sg in inst.get("SecurityGroups", [])],
"name": tags.get("Name"),
"tags": tags,
}
)
security_groups: List[Dict[str, Any]] = []
sg_p = ec2.get_paginator("describe_security_groups")
for page in sg_p.paginate():
for sg in page.get("SecurityGroups", []):
security_groups.append(
{
"group_id": sg.get("GroupId"),
"group_name": sg.get("GroupName"),
"vpc_id": sg.get("VpcId"),
"ip_permissions": sg.get("IpPermissions", []),
}
)
volumes: List[Dict[str, Any]] = []
vol_p = ec2.get_paginator("describe_volumes")
for page in vol_p.paginate():
for vol in page.get("Volumes", []):
attachments = vol.get("Attachments", [])
volumes.append(
{
"volume_id": vol.get("VolumeId"),
"size_gb": vol.get("Size"),
"encrypted": vol.get("Encrypted"),
"state": vol.get("State"),
"az": vol.get("AvailabilityZone"),
"attached_instance_id": attachments[0].get("InstanceId") if attachments else None,
}
)
eips: List[Dict[str, Any]] = []
addr_p = ec2.get_paginator("describe_addresses")
for page in addr_p.paginate():
for a in page.get("Addresses", []):
eips.append(
{
"public_ip": a.get("PublicIp"),
"allocation_id": a.get("AllocationId"),
"association_id": a.get("AssociationId"),
"instance_id": a.get("InstanceId"),
"network_interface_id": a.get("NetworkInterfaceId"),
}
)
return {
"region": region,
"instances": instances,
"security_groups": security_groups,
"volumes": volumes,
"eips": eips,
}
