eulaurarien/feed_asn.py

69 строки
1.9 KiB
Python
Executable File

#!/usr/bin/env python3
import database
import argparse
import requests
import typing
import ipaddress
import logging
import time
IPNetwork = typing.Union[ipaddress.IPv4Network, ipaddress.IPv6Network]
def get_ranges(asn: str) -> typing.Iterable[str]:
req = requests.get(
"https://stat.ripe.net/data/as-routing-consistency/data.json",
params={"resource": asn},
)
data = req.json()
for pref in data["data"]["prefixes"]:
yield pref["prefix"]
def get_name(asn: str) -> str:
req = requests.get(
"https://stat.ripe.net/data/as-overview/data.json", params={"resource": asn}
)
data = req.json()
return data["data"]["holder"]
if __name__ == "__main__":
log = logging.getLogger("feed_asn")
# Parsing arguments
parser = argparse.ArgumentParser(
description="Add the IP ranges associated to the AS in the database"
)
args = parser.parse_args()
DB = database.Database()
def add_ranges(
path: database.Path,
match: database.Match,
) -> None:
assert isinstance(path, database.AsnPath)
assert isinstance(match, database.AsnNode)
asn_str = database.Database.unpack_asn(path)
DB.enter_step("asn_get_name")
name = get_name(asn_str)
match.name = name
DB.enter_step("asn_get_ranges")
for prefix in get_ranges(asn_str):
parsed_prefix: IPNetwork = ipaddress.ip_network(prefix)
if parsed_prefix.version == 4:
DB.set_ip4network(prefix, source=path, updated=int(time.time()))
log.info("Added %s from %s (%s)", prefix, path, name)
elif parsed_prefix.version == 6:
log.warning("Unimplemented prefix version: %s", prefix)
else:
log.error("Unknown prefix version: %s", prefix)
for _ in DB.exec_each_asn(add_ranges):
pass
DB.save()