#!/usr/bin/env python3 import argparse import json import time from collections import defaultdict from contextlib import redirect_stdout from io import StringIO from ipaddress import IPv4Network, IPv6Network, ip_network from itertools import combinations from pathlib import Path import toml NEO_NETWORK_POOL = [ip_network("10.127.0.0/16"), ip_network("fd10:127::/32")] def pick(entity: dict, fields: [str], **kwargs: dict): new_entity = {} for field in fields: new_entity[field] = entity.get(field) for old_field, new_field in kwargs.items(): new_entity[new_field] = entity.get(old_field) return new_entity def is_neo_network(address): return any( address.version == neo.version and address.subnet_of(neo) for neo in NEO_NETWORK_POOL ) def is_neo_network_asn(asn: int): return 4201270000 <= asn <= 4201279999 def is_dn42_asn(asn: int): return 4242420000 <= asn <= 4242429999 def iter_toml_file(path: str): for item in Path(path).iterdir(): if not item.is_file() or item.suffix != ".toml": continue yield item, toml.loads(item.read_text()) def load_entities(): return {item.stem: entity for item, entity in iter_toml_file("entity")} def load_asn(entities: dict): def assert_entity(entity, asn): owner = entity.get("owner") source = entity.get("source") if is_neo_network_asn(asn): source = "NeoNetwork" elif is_dn42_asn(asn): source = "DN42" entity["source"] = source assert owner in entities assert source in ["NeoNetwork", "DN42", "Internet"] return entity mapping = { int(item.stem.lstrip("AS")): entity for item, entity in iter_toml_file("asn") } return {asn: assert_entity(entity, asn) for asn, entity in mapping.items()} def node_to_asn(orignal_asn_set: set): node_table = dict() for _, entities in iter_toml_file("node"): mapping = {name: entity["asn"] for (name, entity) in entities.items()} asn_set = set(mapping.values()) assert orignal_asn_set & asn_set == asn_set node_table.update(mapping) return node_table def assert_peer(nodes: set): for item, entities in iter_toml_file("peer"): peers = set(entities["to-peer"]) assert item.stem in nodes assert nodes & peers == peers def route_to_roa(asn_table: dict): def make_route(): for item, entity in iter_toml_file("route"): asn = int(item.stem.lstrip("AS")) for prefix, fields in entity.items(): if fields["type"] not in ("loopback", "subnet"): continue fields["asn"] = asn fields["prefix"] = ip_network(prefix, strict=True) supernet = fields.get("supernet") fields["supernet"] = ( ip_network(supernet, strict=True) if supernet else None ) assert fields["name"] assert is_neo_network(fields["prefix"]) assert not fields["supernet"] or is_neo_network(fields["supernet"]) yield fields entities = ( pick(route, ["asn", "name", "prefix", "supernet"]) for route in make_route() ) entities = sorted(entities, key=lambda item: item["asn"]) prefixes = [item["prefix"] for item in entities] for net1, net2 in combinations( sorted(entities, key=lambda net: net["prefix"].prefixlen), 2 ): if not net1["prefix"].overlaps(net2["prefix"]): continue assert net1["prefix"] != net2["prefix"] assert net1["prefix"].supernet_of(net2["prefix"]) s1net, s2net = (net1["supernet"], net2["supernet"]) assert s2net # please include supernet = in your route # if net1(the bigger net) has a supernet s1net, then s1net and net1 # will be checked or must have been checked, same for net2 assert not s1net or s1net in prefixes # net1.supernet is garbage assert s2net == net1["prefix"] or s2net in prefixes # net2.supernet is garbage return entities def prehandle_roa(asn_table: dict, args): roa = route_to_roa(asn_table) max_prefixlen = IPv4Network(0).max_prefixlen roa4 = filter(lambda item: isinstance(item["prefix"], IPv4Network), roa) roa6 = filter(lambda item: isinstance(item["prefix"], IPv6Network), roa) if args.ipv4: roa6 = [] elif args.ipv6: roa4 = [] roa4 = [ r for r in roa4 if r["prefix"].prefixlen <= args.max or r["prefix"].prefixlen == max_prefixlen ] roa6 = [r for r in roa6 if r["prefix"].prefixlen <= args.max6] for r in roa4: r["maxLength"] = args.max if r["prefix"].prefixlen == max_prefixlen: r["maxLength"] = max_prefixlen for r in roa6: r["maxLength"] = args.max6 for r in (*roa4, *roa6): r["prefix"] = r["prefix"].with_prefixlen return roa4, roa6 def make_export(roa4, roa6): entities = load_entities() asn_table = load_asn(entities) current = int(time.time()) output = { "metadata": {"generated": current, "valid": current + 14 * 86400}, "people": { owner: {"info": entity, "asns": []} for owner, entity in entities.items() }, } for asn, asn_info in asn_table.items(): owner = asn_info["owner"] asn_item = { "asn": asn, "name": asn_info["name"], "source": asn_info["source"], "routes": { "ipv4": [ pick(roa, ["prefix", "maxLength"]) for roa in roa4 if roa["asn"] == asn ], "ipv6": [ pick(roa, ["prefix", "maxLength"]) for roa in roa6 if roa["asn"] == asn ], }, } output["people"][owner]["asns"].append(asn_item) return json.dumps(output, indent=2) def make_json(roa4, roa6): current = int(time.time()) output = { "metadata": { "counts": len(roa4) + len(roa6), "generated": current, "valid": current + 14 * 86400, }, "roas": [ {"asn": "AS%d" % roa["asn"], **pick(roa, ["prefix", "maxLength"])} for roa in (*roa4, *roa6) ], } return json.dumps(output, indent=2) def make_rfc8416(roa4, roa6): output = { "slurmVersion": 1, "validationOutputFilters": {"prefixFilters": [], "bgpsecFilters": []}, "locallyAddedAssertions": { "bgpsecAssertions": [], "prefixAssertions": [ pick( roa, ["asn", "prefix"], maxLength="maxPrefixLength", name="comment", ) for roa in (*roa4, *roa6) ], }, } return json.dumps(output, indent=2) def make_roa_records(roa4, roa6): records = [ "route {asn} max {prefix} as {maxLength};".format_map(roa) for roa in (*roa4, *roa6) ] return "\n".join(["# NeoNetwork ROA tool", "", *records]) def make_summary(): from tabulate import tabulate entities = load_entities() asn_table = load_asn(entities) node_table = node_to_asn(set(asn_table.keys())) stream = StringIO() with redirect_stdout(stream): print("Entity table:") entity_table = tabulate( ( ( entity["name"], entity.get("contact", {}).get("email"), entity.get("contact", {}).get("telegram"), ) for entity in entities.values() ), headers=["Name", "Email", "Telegram"], tablefmt="presto", ) print(entity_table) print() print("AS table:") as_table = tabulate( ( (entity["source"], "AS{}".format(asn), entity["owner"], entity["name"]) for asn, entity in sorted(asn_table.items(), key=lambda item: item[0]) ), headers=["Source", "ASN", "Owner", "Name"], tablefmt="presto", ) print(as_table) print() print("Node table:") node_table = tabulate( ( ("AS{}".format(asn), name) for name, asn in sorted(node_table.items(), key=lambda item: item[1]) ), headers=["ASN", "Name"], tablefmt="presto", ) print(node_table) print() print("Peer table:") peer_table = tabulate( ( (item.stem, downstream) for item, entity in iter_toml_file("peer") for downstream in entity["to-peer"] ), headers=["Upstream", "Downstream"], tablefmt="presto", colalign=("right",), ) print(peer_table) print() print("Route table:") route_table = tabulate( ( ( "AS{asn}".format_map(entity), entity["name"], entity["prefix"] or "", entity["supernet"] or "", ) for entity in route_to_roa(asn_table) ), headers=["ASN", "Name", "Prefix", "Supernet"], tablefmt="presto", ) print(route_table) return stream.getvalue() def main(args): entities = load_entities() asn_table = load_asn(entities) node_table = node_to_asn(set(asn_table.keys())) assert_peer(set(node_table.keys())) roa4, roa6 = prehandle_roa(asn_table, args) if args.export: return make_export(roa4, roa6) elif args.json: return make_json(roa4, roa6) elif args.rfc8416: return make_rfc8416(roa4, roa6) elif args.summary: return make_summary() else: return make_roa_records(roa4, roa6) if __name__ == "__main__": parser = argparse.ArgumentParser(description="NeoNetwork ROA tool") parser.add_argument( "-m", "--max", type=int, default=29, help="set ipv4 max prefix length" ) parser.add_argument( "-M", "--max6", type=int, default=64, help="set ipv6 max prefix length" ) parser.add_argument("-j", "--json", action="store_true", help="output json") parser.add_argument("-r", "--rfc8416", action="store_true", help="output rfc8416") parser.add_argument("-s", "--summary", action="store_true", help="output summary") parser.add_argument("-o", "--output", default="", help="write output to file") parser.add_argument("-4", "--ipv4", action="store_true", help="print ipv4 only") parser.add_argument("-6", "--ipv6", action="store_true", help="print ipv6 only") parser.add_argument( "-e", "--export", action="store_true", help="export registry to json" ) args = parser.parse_args() if ( args.max < 0 or args.max6 < 0 or args.max > IPv4Network(0).max_prefixlen or args.max6 > IPv6Network(0).max_prefixlen ): parser.error("check your max prefix length") output = main(args) if not args.output or args.output == "-": print(output) elif output: Path(args.output).write_text(output) print("written to", args.output)