Do you use boto3 or botocore? Want to capture and inspect your AWS API traffic? This script will send all AWS API traffic to a Runscope bucket for analysis and debugging.

capture-boto3.py

# Do you use boto3 or botocore? Want to capture and inspect your AWS API
# traffic? This script will send all AWS API traffic to a Runscope bucket for
# analysis and debugging.
#
# Instructions:
# 1. Sign up for an account at https://www.runscope.com if you don't already
#    have one.
# 2. Your account will have a single bucket. Get the Bucket Key from Runscope's
#    Traffic Inspector tab; paste that into the RUNSCOPE_BUCKET_KEY constant.
# 3. Go to https://www.runscope.com/applications and create a new application.
#    Website URL and Callback URL aren't relevant because you'll never be using
#    OAuth with this. Once you have created the application, copy your Personal
#    Access Token from the bottom of the screen, into RUNSCOPE_ACCESS_TOKEN.
# 4. Set LOG_ALL_TRAFFIC to True if you want to send *all* AWS API traffic to
#    Runscope. Or set it to False to log only errors (HTTP status codes 400-499
#    and 500-599).

import json
import threading
import time

import botocore.hooks
import botocore.session
import botocore.vendored.requests

RUNSCOPE_BUCKET_KEY = "<your bucket key goes here>"
RUNSCOPE_ACCESS_TOKEN = "<your access token goes here>"
LOG_ALL_TRAFFIC = False


def get_boto3_session():
    capture_vars = threading.local()

    def munge_headers(headers):
        # Copies the dict-like `headers` object into a real dict. Applies
        # traditional capitalization rules to headers, e.g. "content-type"
        # becomes "Content-Type".
        result = {}
        for (k, v) in headers.iteritems():
            k = "-".join([word[0].upper() + word[1:] for word in k.split("-")])
            result[k] = v
        return result

    def before_call_handler(*args, **kwargs):
        capture_vars.start_time = time.time()

    def after_call_handler(*args, **kwargs):
        if not isinstance(kwargs.get("http_response"), botocore.vendored.requests.Response):
            return
        response = kwargs['http_response']
        if not LOG_ALL_TRAFFIC and response.status_code < 400:
            return

        end_time = time.time()
        response_time = end_time - capture_vars.start_time

        api_data = {"request": {"method": response.request.method,
                                "url": response.request.url,
                                "headers": munge_headers(response.request.headers),
                                "body": response.request.body,
                                "timestamp": capture_vars.start_time},
                    "response": {"status": response.status_code,
                                 "body": response.text,
                                 "headers": munge_headers(response.headers),
                                 "response_time": end_time - capture_vars.start_time,
                                 "timestamp": end_time}}

        api_headers = {'Authorization': 'Bearer {}'.format(RUNSCOPE_ACCESS_TOKEN),
                       'Content-Type': 'application/json'}

        url = "https://api.runscope.com/buckets/{}/messages".format(RUNSCOPE_BUCKET_KEY)
        botocore.vendored.requests.post(url, headers=api_headers, data=json.dumps(api_data))

    hooks = botocore.hooks.HierarchicalEmitter()
    hooks.register(event_name="before-call.*", handler=before_call_handler)
    hooks.register(event_name="after-call.*", handler=after_call_handler)

    return botocore.session.Session(event_hooks=hooks)