imitial

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)