old_ripedb_fiddle.py

"""
An old fiddle when we were investigating the size of the ranges owned by Ziggo…
"""
import aiohttp
import asyncio
import json
import itertools
import netaddr


from typing import List

from contextlib import asynccontextmanager
from typing import Optional

class RIPEDBClient(object):
    session: aiohttp.ClientSession
    base_url: str = 'http://rest.db.ripe.net/search'

    def __init__(self, session: aiohttp.ClientSession):
        self.session = session

    @staticmethod
    @asynccontextmanager
    async def get_instance() -> 'RIPEDBClient':
        headers = {
            'Accept': 'application/json',
        }
        try:
            async with aiohttp.ClientSession(headers=headers) as session:
                yield RIPEDBClient(session)
        finally:
            await asyncio.sleep(0.250)

    async def reverse_lookup_search(
            self,
            query_string: str=None,
            inverse_attribute: str=None,
            type_filter: Optional[str]=None):
        params = {
            'source': 'ripe',
            'query-string': query_string,
            'flags': 'no-referenced',
            'inverse-attribute': inverse_attribute,
        }

        if type_filter:
            params['type-filter'] = type_filter

        async with self.session.get(self.base_url, params=params) as resp:
            assert resp.status == 200
            return await resp.json()


def network_from_ripe(entry) -> List[netaddr.IPRange]:
    for attr in entry['attributes']['attribute']:
        if attr['name'] == 'inetnum':
            begin, end = attr['value'].split(' - ')

            ip_range = netaddr.IPRange(begin, end)
            return list(ip_range.cidrs())

    return []

async def stats_for(rc, maintainer):
    body = await rc.reverse_lookup_search(maintainer, 'mnt-by', 'inetnum')

    networks = itertools.chain.from_iterable([network_from_ripe(entry) for entry in body['objects']['object']])

    merged = netaddr.cidr_merge(networks)
    sized = list(reversed(sorted([(len(netaddr.IPSet(n)), n) for n in merged])))
    biggest = ", ".join(["{} ({})".format(nw[1], nw[0]) for nw in sized[:3]])
    ip_merged = netaddr.IPSet(merged)
    print("{:<20} {:>4} subnets {:>8} ips ({} ...)".format(f"{maintainer}:", len(merged), len(ip_merged), biggest))


async def main():
    async with RIPEDBClient.get_instance() as rc:
        await stats_for(rc, 'ziggo-services-mnt')
        await stats_for(rc, 'ZIGGO-MNT')
        await stats_for(rc, 'MNT-LGI')
        await stats_for(rc, 'AS6830-MNT')
        await stats_for(rc, 'TNF-MNT')
        await stats_for(rc, 'ZIGGO-IPMGMT')


if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())
    loop.close()