This mitmproxy Python script captures and logs HTTP(S) API calls made to any subdomain of onepeloton.com. It extracts request and response details (including GraphQL operation names), ensures uniqueness, assigns a sequence number to each entry, and saves the results to a CSV file.

proxy.py

import csv
import signal
import re
import time
from mitmproxy import http
from multiprocessing import Manager
import json

# mitmdump -s proxy.py -p 8080
# set browser proxy (use firefox with foxy proxy extenstion) to localhost:8080
# Visit http://mitm.it/ and install the mitmproxy certificate for full HTTPS logging.
# to exit : press cntrl+c. This will write the results to a csv file. 

# Regex pattern for subdomains of onepeloton.com
ONEPELOTON_PATTERN = re.compile(r"https?://([a-zA-Z0-9.-]+)?\.onepeloton\.com")

# Shared variables to control logging state
manager = Manager()
logging_enabled = manager.Value('i', 1)  # 0 = Logging Off, 1 = Logging On
results = manager.list()
referer_host_method_url_set = manager.dict()
log_sequence = manager.Value('i', 1)

def response(flow: http.HTTPFlow):
    """
    Intercepts HTTP requests and logs API calls along with their referer headers only when logging is enabled.
    """
    print(f"Logging enabled Value : {logging_enabled.value}")
    if not logging_enabled.value:
        return  # Skip logging if disabled

    url = flow.request.url
    method = flow.request.method
    referer = flow.request.headers.get("Referer", "No Referer")
    host = flow.request.host
    content_type = flow.response.headers.get("Content-Type", "").split(";")[0]
    path = flow.request.path
    graphQL_operation_name = None
    is_graphql = False

    # Log only requests to subdomains of onepeloton.com
    if ONEPELOTON_PATTERN.match(url):
        print(f"************Matched: host: {host}")
        seq = log_sequence.value
        log_sequence.value += 1
       
        #check if the request is a graphql request
        if "graphql" in path and flow.request.method == "POST":
            is_graphql = True
            #extract the graphql operation name
            try:
                data = json.loads(flow.request.get_text())
                graphQL_operation_name = data.get("operationName", None)
            except Exception as e:
                graphQL_operation_name = f"<error: {str(e)}>"
       
        # Log the request only if it's unique
        request_tuple = (referer, host, method, url, graphQL_operation_name)
        if request_tuple not in referer_host_method_url_set:
            url_info = {
                "Sequence": seq,
                "Referer": referer,
                "Host": host,
                "Method": method,
                "Content-Type": content_type,
                "URL": url, 
                "GraphQL_Operation_Name": graphQL_operation_name,
                "Is_GraphQL": is_graphql
                
            }
            results.append(url_info)
            referer_host_method_url_set[request_tuple] = True  # Mark request as logged

            # Console output (optional)
            print(f"[API Call] {method} {url}\nReferer: {referer}\nHost: {host}\nContent-Type: {content_type}\n")

# Register the request handler
addons = [response]

def save_logs_to_csv():
    """Saves logged API calls to a CSV file when logging is stopped."""
   # print(f"results: {results}")
    if results:
        filename = "onepeloton_logs.csv"
        with open(filename, mode="w", newline="") as file:
            writer = csv.DictWriter(file, fieldnames=["Sequence","Referer","Host", "Method", "Content-Type", "URL", "GraphQL_Operation_Name", 'Is_GraphQL'])
            writer.writeheader()
            writer.writerows(results)
        print(f"\n✅ Logs saved to {filename}")
        results[:] = []  # Clear logs after saving
        referer_host_method_url_set.clear()  # Reset unique tracking

def start_logging():
    """Starts logging API calls."""
    logging_enabled.value = 1
    print("🟢 Logging started.")

def stop_logging():
    """Stops logging API calls and saves logs."""
    logging_enabled.value = 0
    print("\n🛑 Logging stopped.")
    save_logs_to_csv()
    print("Saved file to CSV. Exiting...")
    time.sleep(1)  # Wait for CSV write to complete before exiting

def signal_handler(sig, frame):
    """Handles CTRL+C to stop logging gracefully before exiting."""
    stop_logging()
    exit(0)

# Attach signal handler for CTRL+C
signal.signal(signal.SIGINT, signal_handler)