aws_cloudtrail_logs.py

#! /usr/bin/env nix-shell
#! nix-shell -i python3 -p python3 python3Packages.boto3

import boto3
import json

def fetch_cloudtrail_events():
    # NOTE: this is us-west-2 specific!
    client = boto3.client(
        "cloudtrail",
        aws_access_key_id="...",
        aws_secret_access_key="...",
        region_name="us-west-2"
    )
    lookup_attributes = [{ "AttributeKey": "ResourceType", "AttributeValue": "AWS::EC2::Instance" }]

    all_events = []

    # Initial call to lookup-events
    response = client.lookup_events(LookupAttributes=lookup_attributes, MaxResults=1000)
    all_events.extend(response["Events"])

    # Use pagination to fetch all events
    while "NextToken" in response:
        response = client.lookup_events(LookupAttributes=lookup_attributes, MaxResults=1000, NextToken=response["NextToken"])
        all_events.extend(response["Events"])

    return all_events

events = fetch_cloudtrail_events()
# Sort events by EventTime
for event in sorted(events, key=lambda e: e["EventTime"]):
    print(json.dumps({**event, "EventTime": event["EventTime"].timestamp()}))