From e08c4fb2c973021d0a58c65cee24093824e03003 Mon Sep 17 00:00:00 2001 From: badra001 Date: Thu, 12 Mar 2026 11:12:06 -0400 Subject: [PATCH] imitial --- .../ipam/export_ipam_discovery.py | 75 +++++++++++++++++++ 1 file changed, 75 insertions(+) create mode 100755 local-app/python-tools/ipam/export_ipam_discovery.py diff --git a/local-app/python-tools/ipam/export_ipam_discovery.py b/local-app/python-tools/ipam/export_ipam_discovery.py new file mode 100755 index 00000000..6ae18fc2 --- /dev/null +++ b/local-app/python-tools/ipam/export_ipam_discovery.py @@ -0,0 +1,75 @@ +#!/bin/env python3 + +import boto3 +import csv +import argparse +import sys +from botocore.exceptions import ClientError + +def get_args(): + parser = argparse.ArgumentParser(description="Export AWS IPAM discovered resources to CSV.") + parser.add_argument("--profile", help="AWS CLI profile name", default=None) + parser.add_argument("--region", help="AWS region (e.g., us-gov-west-1)", required=True) + parser.add_argument("--output", help="Output CSV filename", default="ipam_discovery.csv") + return parser.parse_args() + +def export_ipam_data(profile, region, output_file): + # Initialize session + session = boto3.Session(profile_name=profile, region_name=region) + ec2 = session.client("ec2") + + try: + # First, we need the IPAM Resource Discovery ID + # Most environments have one, but we'll fetch the first active one found + discoveries = ec2.describe_ipam_resource_discoveries() + if not discoveries["IpamResourceDiscoveries"]: + print(f"Error: No IPAM Resource Discoveries found in {region}.") + return + + discovery_id = discoveries["IpamResourceDiscoveries"][0]["IpamResourceDiscoveryId"] + print(f"Using Discovery ID: {discovery_id}") + + resources = [] + paginator = ec2.get_paginator("get_ipam_discovered_resource_cidrs") + + # Fetching VPCs and Subnets + for resource_type in ["vpc", "subnet"]: + print(f"Fetching discovered {resource_type}s...") + page_iterator = paginator.paginate( + IpamResourceDiscoveryId=discovery_id, + ResourceIdScope=region, + Filters=[{'Name': 'resource-type', 'Values': [resource_type]}] + ) + + for page in page_iterator: + for item in page["IpamDiscoveredResourceCidrs"]: + resources.append({ + "ResourceType": resource_type.upper(), + "ResourceId": item.get("ResourceId"), + "ResourceCidr": item.get("ResourceCidr"), + "ResourceRegion": item.get("ResourceRegion"), + "ResourceOwnerId": item.get("ResourceOwnerId"), + "IpUsage": item.get("IpUsage"), + "VpcId": item.get("VpcId", "N/A"), # Only populates for subnets + "SampleTime": item.get("SampleTime").strftime("%Y-%m-%d %H:%M:%S") + }) + + # Write to CSV + if resources: + keys = resources[0].keys() + with open(output_file, "w", newline="") as f: + dict_writer = csv.DictWriter(f, fieldnames=keys) + dict_writer.writeheader() + dict_writer.writerows(resources) + print(f"Successfully exported {len(resources)} resources to {output_file}") + else: + print("No resources found to export.") + + except ClientError as e: + print(f"AWS Error: {e.response['Error']['Message']}") + except Exception as e: + print(f"An error occurred: {e}") + +if __name__ == "__main__": + args = get_args() + export_ipam_data(args.profile, args.region, args.output)