Multithreaded GitHub Actions IP Exporter

script.py

#!/usr/bin/env python3
"""
Install requests library
    pip install requests

Make the script executable
    chmod +x script.py

Run the script
    ./script.py
"""

import sys
import requests
import ipaddress
import threading
import time
import datetime
from concurrent.futures import ThreadPoolExecutor, as_completed

GITHUB_META_URL = "https://api.github.com/meta"
OUTPUT_FILE = "github_actions_ips.txt"
MAX_WORKERS = 8  # adjust based on your CPU cores

def fetch_github_meta(url: str) -> dict:
    try:
        resp = requests.get(url, headers={"Accept": "application/vnd.github.v3+json"})
        resp.raise_for_status()
        return resp.json()
    except requests.RequestException as e:
        print(f"Error fetching {url}: {e}", file=sys.stderr)
        sys.exit(1)

def format_duration(seconds: float) -> str:
    return str(datetime.timedelta(seconds=int(seconds)))

def expand_and_save_cidr(cidr: str, filename: str, lock: threading.Lock):
    """
    Expand one CIDR block and append all IPs to the file under a lock.
    """
    network = ipaddress.ip_network(cidr, strict=False)
    lines = [f"{ip}\n" for ip in network]
    with lock:
        with open(filename, "a") as out:
            out.writelines(lines)

def main():
    data = fetch_github_meta(GITHUB_META_URL)
    cidrs = data.get("actions", [])
    if not cidrs:
        print("No 'actions' key found in GitHub meta response.", file=sys.stderr)
        sys.exit(1)

    total = len(cidrs)
    print(f"Expanding {total} CIDR blocks from GitHub Actions…\n")

    # clear or create the output file
    open(OUTPUT_FILE, "w").close()

    lock = threading.Lock()
    start_time = time.time()

    with ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor:
        futures = {
            executor.submit(expand_and_save_cidr, cidr, OUTPUT_FILE, lock): idx
            for idx, cidr in enumerate(cidrs, start=1)
        }

        for future in as_completed(futures):
            idx = futures[future]
            cidr = cidrs[idx-1]
            try:
                future.result()
                elapsed = time.time() - start_time
                avg_per = elapsed / idx
                remaining = avg_per * (total - idx)
                print(
                    f"✔️  Finished {cidr} ({idx}/{total})"
                    f" — elapsed {format_duration(elapsed)}, ETA {format_duration(remaining)}"
                )
            except Exception as e:
                print(f"❌ Error processing {cidr}: {e}", file=sys.stderr)

    print(f"\nDone. All IPs written to {OUTPUT_FILE}")

if __name__ == "__main__":
    main()