scrape-asn.py

#!/usr/bin/python3
################################################################################
#
# Basic scraper to pull IP CIDRs from the RIPE API. Currently hardcoded to grab
# all public Australian AS Numbers, could be quickly modified to scrape based on
# on other criteria. Only currently does IPV4, again quick to modify for v6.
#
# Runs on 8 threads because that's the max allowed concurrent requests for the
# API. There are other requirements if you're scraping them on a routine basis
# read here -> https://stat.ripe.net/docs/data_api
#
# Standard disclaimer: I'm still new and bad at Python. Always interested to
# learn about issues / improvements if you have any.
#
################################################################################
import json
import requests
import threading
import concurrent.futures

def callAPI(command, params):
    URLprefix = 'https://stat.ripe.net/data/'
    URLsuffix = '/data.json'

    r = requests.get(url = URLprefix + command + URLsuffix, params = params, timeout = 15.0)

    if r.status_code == 200:
        return json.loads(r.content.decode('utf-8'))
    else:
        return None

outputLock = threading.Lock()

def grabPrefixes(ASN):
    global outputLock

    print ('Grabbing prefixes for ASN ', ASN)
    ASNdetails = callAPI('ris-prefixes', {'resource': ASN, 'list_prefixes': True})
    if ASNdetails != None:
        for prefix in ASNdetails['data']['prefixes']['v4']['originating']:
            print(ASN, ": ", prefix)
            outputLock.acquire()
            f.write("%s\n" % prefix)
            outputLock.release()

f = open('australia-asn-prefixes.txt', 'w')

print ('Getting Australian AS Numbers')
ASNlist = callAPI('country-asns', { 'resource': 'au', 'lod': 1})

if ASNlist != None:
    with concurrent.futures.ThreadPoolExecutor(max_workers = 8) as executor:
        executor.map(grabPrefixes, ASNlist['data']['countries'][0]['routed'])
        
f.close()