bgp

- research & experiments with border gateway protocol
git clone git://git.acid.vegas/bgp.git
bgp.py (802B)
      1 import pybgpstream
      2 from collections import defaultdict
      3 
      4 def fetch_asn_ip_mappings():
      5     asn_ip_mappings = defaultdict(list)
      6     stream = pybgpstream.BGPStream(
      7         project="ris-live",
      8         filter="collector rrc00",
      9     )
     10     for rec in stream:
     11         for elem in rec:
     12             if elem.type == "A":
     13                 origin_asn = elem.fields["as-path"].split()[-1]
     14                 prefix = elem.fields["prefix"]
     15                 if prefix not in asn_ip_mappings[origin_asn]:
     16                     asn_ip_mappings[origin_asn].append(prefix)
     17                     print(f'{origin_asn} - {prefix}')
     18     return asn_ip_mappings
     19 
     20 if __name__ == "__main__":
     21     mappings = fetch_asn_ip_mappings()
     22     for asn, prefixes in mappings.items():
     23         print(f"ASN: {asn}, Prefixes: {', '.join(prefixes)}")
     24 
     25 
     26