builder updates

Author: superstes

src/build.py

@@ -6,7 +6,7 @@
 
 from config import *
 from util import log
-from enrich_data import ip_asn_info
+from enrich_data import ip_asn_info, net_asn_info
 from write import write_ip_asn, write_nets
 
 
@@ -36,7 +36,9 @@ def build_dbs_ip_asn(reports: dict, ptrs: dict, lookup_lists: dict, networks: di
                 lookup_lists=lookup_lists,
             )
             net = {'network': networks[key][networks['ip_to_net'][ip]]}
-            net['network'].pop('ipv')
+            if 'ipv' in net['network']:
+                net['network'].pop('ipv')
+
             net_sm = {'network': {
                 'reported_ips': net['network']['reported_ips'],
                 'reputation': net['network']['reputation'],
@@ -140,7 +142,9 @@ def build_dbs_net(networks: dict):
 
         for n, nv in net_list.items():
             ipv = nv.pop('ipv')
+            nv = {**nv, **net_asn_info(n)}
             n = f"{n}/{BGP_NET_SIZE[ipv]}"
+
             if ipv =='4':
                 json4[n] = nv
 

src/config.py

@@ -3,10 +3,11 @@
 INFO_CATEGORIES = ['hosting', 'vpn', 'proxy']
 CATEGORIES = ['bot', 'probe', 'rate', 'attack', 'crawler']
 
-BASE_PATH = '/tmp/risk-db'
+BASE_PATH = Path('/tmp/risk-db')
 MMDB_DESCRIPTION = 'OXL RISK-Database - risk.oxl.app (CC BY-SA 4.0)'
 REPORT_COOLDOWN = 60
 ASN_JSON_FILE = Path('/tmp/asn_full.json')  # source: https://github.com/O-X-L/geoip-asn
+ASN_MMDB_FILE = BASE_PATH / 'oxl_geoip_asn.mmdb'  # source: https://github.com/O-X-L/geoip-asn
 SRC_PATH = Path(__file__).resolve().parent
 ASN_FILE_HOSTING = SRC_PATH / 'kind' / 'hosting.txt'
 ASN_FILE_PROXY = SRC_PATH / 'kind' / 'proxy.txt'

src/enrich_data.py

@@ -86,7 +86,7 @@ def load_lookup_lists() -> dict:
 
 
 def ip_asn_info(ip: str, reports: dict, lookup_lists: dict, ptrs: dict) -> dict:
-    with mmdb_database(f'{BASE_PATH}/oxl_geoip_asn.mmdb') as m:
+    with mmdb_database(ASN_MMDB_FILE) as m:
         ip_md = m.get(ip)
 
     try:
@@ -106,6 +106,7 @@ def ip_asn_info(ip: str, reports: dict, lookup_lists: dict, ptrs: dict) -> dict:
         },
         'url': {
             'asn': f'https://risk.oxl.app/api/asn/{asn}',
+            'net': f'https://risk.oxl.app/api/net/{ip}',
             'ipinfo': f'https://ipinfo.io/{ip}',
             'shodan': f'https://www.shodan.io/host/{ip}',
         },
@@ -130,3 +131,21 @@ def ip_asn_info(ip: str, reports: dict, lookup_lists: dict, ptrs: dict) -> dict:
         'full': d,
         'small': d_small,
     }
+
+
+def net_asn_info(ip: str) -> dict:
+    with mmdb_database(ASN_MMDB_FILE) as m:
+        ip_md = m.get(ip)
+
+    try:
+        asn = int(ip_md['asn'][2:])
+
+    except ValueError:
+        return {}
+
+    return {
+        'asn': asn,
+        'url': {
+            'asn': f'https://risk.oxl.app/api/asn/{asn}',
+        }
+    }

src/main.py

@@ -27,8 +27,8 @@ def main():
     lookup_lists = load_lookup_lists()
 
     log('Building and writing DBs')
-    build_dbs_ip_asn(reports=reports, ptrs=ptrs, lookup_lists=lookup_lists, networks=networks)
     build_dbs_net(networks=networks)
+    build_dbs_ip_asn(reports=reports, ptrs=ptrs, lookup_lists=lookup_lists, networks=networks)
 
     log('Done')
 

src/reputation.py

@@ -1,6 +1,6 @@
-from ipaddress import IPv4Address, AddressValueError, IPv4Interface, IPv6Interface
-
 from config import *
+from util import get_ip_version, get_network_address
+
 
 # pylint: disable=W0613
 def _reporter_reputation(r: dict) -> int:
@@ -72,15 +72,8 @@ def reports_by_network_reputation(reports: list[dict]) -> dict:
     ip_to_net = {}
 
     for r in reports:
-        try:
-            IPv4Address(r['ip'])
-            ipv = '4'
-            n = IPv4Interface(f"{r['ip']}/{BGP_NET_SIZE[ipv]}").network.network_address.compressed
-
-        except AddressValueError:
-            ipv = '6'
-            n = IPv6Interface(f"{r['ip']}/{BGP_NET_SIZE[ipv]}").network.network_address.compressed
-
+        ipv = get_ip_version(r['ip'])
+        n = get_network_address(r['ip'])
         reputation = _reporter_reputation(r)
 
         _save_net_report(dst=reported_nets_all, r=r, n=n)

src/util.py

@@ -1,7 +1,32 @@
 from time import time
+from ipaddress import IPv4Address, AddressValueError, IPv4Interface, IPv6Interface
+
+from config import BGP_NET_SIZE
 
 start_time = time()
 
 
 def log(msg: str):
     print(f'{msg} ({int(time() - start_time)}s)')
+
+
+def get_ip_version(ip: str) -> str:
+    try:
+        IPv4Address(ip)
+        return '4'
+
+    except AddressValueError:
+        return '6'
+
+
+def get_network_address(ip: str) -> str:
+    try:
+        IPv4Address(ip)
+        return IPv4Interface(f"{ip}/{BGP_NET_SIZE['4']}").network.network_address.compressed
+
+    except AddressValueError:
+        return IPv6Interface(f"{ip}/{BGP_NET_SIZE['6']}").network.network_address.compressed
+
+
+# def get_network_cidr(ip: str) -> str:
+#     return f"{get_network_address(ip)}/{BGP_NET_SIZE[get_ip_version(ip)]}"