new: use local patricia tree to lookup prefixes
parent
9979c9abe1
commit
3544dcb7e4
|
@ -42,10 +42,11 @@ class PrefixDatabase():
|
||||||
with gzip.open(BytesIO(r.content), 'r') as f:
|
with gzip.open(BytesIO(r.content), 'r') as f:
|
||||||
for line in f:
|
for line in f:
|
||||||
prefix, length, asns = line.decode().strip().split('\t')
|
prefix, length, asns = line.decode().strip().split('\t')
|
||||||
for asn in re.split('[,_]', asns):
|
# The meaning of AS set and multi-origin AS in unclear. Tacking the first ASN in the list only.
|
||||||
network = ip_network('{}/{}'.format(prefix, length))
|
asn = re.split('[,_]', asns)[0]
|
||||||
to_import[asn][address_family].add(str(network))
|
network = ip_network('{}/{}'.format(prefix, length))
|
||||||
to_import[asn]['ipcount'] += network.num_addresses
|
to_import[asn][address_family].add(str(network))
|
||||||
|
to_import[asn]['ipcount'] += network.num_addresses
|
||||||
|
|
||||||
p = self.redis_cache.pipeline()
|
p = self.redis_cache.pipeline()
|
||||||
p.sadd('asns', *to_import.keys())
|
p.sadd('asns', *to_import.keys())
|
||||||
|
|
|
@ -2,59 +2,59 @@
|
||||||
# -*- coding: utf-8 -*-
|
# -*- coding: utf-8 -*-
|
||||||
|
|
||||||
import logging
|
import logging
|
||||||
import json
|
|
||||||
from redis import Redis
|
from redis import Redis
|
||||||
|
|
||||||
from .libs.StatsRipeText import RIPECaching
|
import time
|
||||||
import asyncio
|
import pytricia
|
||||||
|
import ipaddress
|
||||||
|
|
||||||
|
|
||||||
class RISPrefixLookup(RIPECaching):
|
class RISPrefixLookup():
|
||||||
|
|
||||||
def __init__(self, sourceapp: str='bgpranking-ng', loglevel: int=logging.DEBUG):
|
def __init__(self, loglevel: int=logging.DEBUG):
|
||||||
super().__init__(sourceapp, loglevel)
|
self.__init_logger(loglevel)
|
||||||
self.logger.debug('Starting RIS Prefix fetcher')
|
self.logger.debug('Starting RIS Prefix fetcher')
|
||||||
|
self.prefix_db = Redis(host='localhost', port=6582, db=0, decode_responses=True)
|
||||||
|
self.longest_prefix_matching = Redis(host='localhost', port=6581, db=0, decode_responses=True)
|
||||||
|
self.tree_v4 = pytricia.PyTricia()
|
||||||
|
self.tree_v6 = pytricia.PyTricia(128)
|
||||||
|
self.init_tree()
|
||||||
|
|
||||||
def cache_prefix(self, redis_cache, ip, network_info, prefix_overview):
|
def __init_logger(self, loglevel):
|
||||||
prefix = network_info['prefix']
|
self.logger = logging.getLogger('{}'.format(self.__class__.__name__))
|
||||||
asns = network_info['asns']
|
self.logger.setLevel(loglevel)
|
||||||
description = prefix_overview['block']['desc']
|
|
||||||
if not description:
|
def cache_prefix(self, ip, prefix, asns):
|
||||||
description = prefix_overview['block']['name']
|
p = self.longest_prefix_matching.pipeline()
|
||||||
p = redis_cache.pipeline()
|
p.hmset(ip, {'asn': asns, 'prefix': prefix})
|
||||||
for asn in asns:
|
p.expire(ip, 43200) # 12H
|
||||||
p.hmset(ip, {'asn': asn, 'prefix': prefix, 'description': description})
|
|
||||||
p.expire(ip, 43200) # 12H
|
|
||||||
p.execute()
|
p.execute()
|
||||||
|
|
||||||
async def run(self):
|
def init_tree(self):
|
||||||
redis_cache = Redis(host='localhost', port=6581, db=0, decode_responses=True)
|
for asn in self.prefix_db.smembers('asns'):
|
||||||
reader, writer = await asyncio.open_connection(self.hostname, self.port)
|
for prefix in self.prefix_db.smembers('{}|{}'.format(asn, 'v4')):
|
||||||
|
self.tree_v4[prefix] = asn
|
||||||
|
for prefix in self.prefix_db.smembers('{}|{}'.format(asn, 'v6')):
|
||||||
|
self.tree_v6[prefix] = asn
|
||||||
|
|
||||||
writer.write(b'-k\n')
|
def run(self):
|
||||||
while True:
|
while True:
|
||||||
ip = redis_cache.spop('for_ris_lookup')
|
ip = self.longest_prefix_matching.spop('for_ris_lookup')
|
||||||
if not ip: # TODO: add a check against something to stop the loop
|
if not ip: # TODO: add a check against something to stop the loop
|
||||||
self.logger.debug('Nothing to lookup')
|
self.logger.debug('Nothing to lookup')
|
||||||
await asyncio.sleep(10)
|
time.sleep(10)
|
||||||
continue
|
continue
|
||||||
if redis_cache.exists(ip):
|
if self.longest_prefix_matching.exists(ip):
|
||||||
self.logger.debug('Already cached: {}'.format(ip))
|
self.logger.debug('Already cached: {}'.format(ip))
|
||||||
continue
|
continue
|
||||||
self.logger.debug('RIS lookup: {}'.format(ip))
|
ip = ipaddress.ip_address(ip)
|
||||||
to_send = '-d network-info {} sourceapp={}\n'.format(ip, self.sourceapp)
|
if ip.version == 4:
|
||||||
writer.write(to_send.encode())
|
prefix = self.tree_v4.get_key(ip)
|
||||||
data = await reader.readuntil(b'\n}\n')
|
asns = self.tree_v4.get(ip)
|
||||||
network_info = json.loads(data)
|
else:
|
||||||
if not network_info.get('prefix'):
|
prefix = self.tree_v6.get_key(ip)
|
||||||
|
asns = self.tree_v6.get(ip)
|
||||||
|
if not prefix:
|
||||||
self.logger.warning('The IP {} does not seem to be announced'.format(ip))
|
self.logger.warning('The IP {} does not seem to be announced'.format(ip))
|
||||||
continue
|
continue
|
||||||
self.logger.debug('Prefix lookup: {}'.format(ip))
|
self.cache_prefix(ip, prefix, asns)
|
||||||
to_send = '-d prefix-overview {} sourceapp={}\n'.format(network_info['prefix'], self.sourceapp)
|
|
||||||
writer.write(to_send.encode())
|
|
||||||
data = await reader.readuntil(b'\n}\n')
|
|
||||||
prefix_overview = json.loads(data)
|
|
||||||
self.logger.debug('RIS cache prefix info: {}'.format(ip))
|
|
||||||
self.cache_prefix(redis_cache, ip, network_info, prefix_overview)
|
|
||||||
writer.write(b'-k\n')
|
|
||||||
writer.close()
|
|
||||||
|
|
16
ris.py
16
ris.py
|
@ -2,7 +2,6 @@
|
||||||
# -*- coding: utf-8 -*-
|
# -*- coding: utf-8 -*-
|
||||||
|
|
||||||
import logging
|
import logging
|
||||||
import asyncio
|
|
||||||
from listimport.risfetcher import RISPrefixLookup
|
from listimport.risfetcher import RISPrefixLookup
|
||||||
|
|
||||||
logging.basicConfig(format='%(asctime)s %(name)s %(levelname)s:%(message)s',
|
logging.basicConfig(format='%(asctime)s %(name)s %(levelname)s:%(message)s',
|
||||||
|
@ -14,19 +13,10 @@ class RISManager():
|
||||||
def __init__(self, loglevel: int=logging.DEBUG):
|
def __init__(self, loglevel: int=logging.DEBUG):
|
||||||
self.ris_fetcher = RISPrefixLookup(loglevel=loglevel)
|
self.ris_fetcher = RISPrefixLookup(loglevel=loglevel)
|
||||||
|
|
||||||
async def run_fetcher(self):
|
def run_fetcher(self):
|
||||||
await asyncio.gather(
|
self.ris_fetcher.run()
|
||||||
self.ris_fetcher.run(),
|
|
||||||
self.ris_fetcher.run(),
|
|
||||||
self.ris_fetcher.run(),
|
|
||||||
self.ris_fetcher.run(),
|
|
||||||
self.ris_fetcher.run(),
|
|
||||||
self.ris_fetcher.run(),
|
|
||||||
# self.ris_fetcher.run(2)
|
|
||||||
)
|
|
||||||
|
|
||||||
|
|
||||||
if __name__ == '__main__':
|
if __name__ == '__main__':
|
||||||
modules_manager = RISManager()
|
modules_manager = RISManager()
|
||||||
loop = asyncio.get_event_loop()
|
modules_manager.run_fetcher()
|
||||||
loop.run_until_complete(modules_manager.run_fetcher())
|
|
||||||
|
|
Loading…
Reference in New Issue