update archiver to process reports day-by-day

Author: superstes

src/riskdb/archiver/config.py

@@ -1,6 +1,8 @@
 from os import environ
+from datetime import datetime
 
 REPO_ARCHIVE = 'github.com/O-X-L/risk-db-archive'
-ARCHIVE_DEDUPE_FIELDS = ['fp', 'cmt', 'user', 'by', 'cat']
-HEADERS_ARCHIVE_CSV = ['time', 'ip', 'an', 'cat', 'cmt', 'by', 'user', 'fp']
+ARCHIVE_DEDUPE_FIELDS = ['cmt', 'user', 'by', 'cat']
+HEADERS_ARCHIVE_CSV = ['time', 'ip', 'an', 'cat', 'cmt', 'by', 'user']
 GIT_TOKEN = environ.get('GIT_TOKEN')
+ARCHIVE_START_DATE = datetime(year=2024, month=11, day=1)

src/riskdb/archiver/main.py

@@ -5,29 +5,28 @@
 from time import time
 from hashlib import md5
 from pathlib import Path
-from datetime import datetime
 from os import system as shell
 from operator import itemgetter
 from sys import path as sys_path
 from ipaddress import ip_network
+from datetime import datetime, timedelta
 
 sys_path.append(str(Path(__file__).parent.parent.parent))
 
 from riskdb.config import NET_SIZE
-from riskdb.archiver.config import REPO_ARCHIVE, HEADERS_ARCHIVE_CSV, ARCHIVE_DEDUPE_FIELDS
 from riskdb.builder.util import log
 from riskdb.builder.load_reports import FileLoader
 from riskdb.archiver.util import git_commit_and_push, git_clone, git_check_token
+from riskdb.archiver.config import REPO_ARCHIVE, HEADERS_ARCHIVE_CSV, ARCHIVE_DEDUPE_FIELDS, ARCHIVE_START_DATE
 
 
-# NOTE: de-duplicating raw-report values to make the archive more compact
-def _reports_by_day(tmp_dir: str) -> dict[list[dict]]:
-    reports = {}
-    tmp_dir_dedupe = f'{tmp_dir}/dedupe'
-    dedupe_map = {k: [] for k in ARCHIVE_DEDUPE_FIELDS}
-    shell(f'mkdir -p {tmp_dir_dedupe}')
+def _generate_archive_for_day(date: datetime, dedupe_map: dict, tmp_dir: Path) -> dict:
+    reports = []
+    for r in FileLoader(sliding_window=False, match_date=date):
+        rdate = datetime.fromtimestamp(r['time'])
+        if rdate.year != date.year or rdate.month != date.month or rdate.day != date.day:
+            continue
 
-    for r in FileLoader():
         for k, v in r.items():
             if v is None:
                 r[k] = ''
@@ -46,25 +45,19 @@ def _reports_by_day(tmp_dir: str) -> dict[list[dict]]:
         if 'v' in r:
             r.pop('v')
 
-        day = datetime.fromtimestamp(r['time']).strftime('%Y_%m_%d')
-        if day not in reports:
-            reports[day] = []
-
         if 'an' not in r:
             r['an'] = ''
 
-        if 'fp' not in r:
-            r['fp'] = ''
+        if r['by'] != '':
+            if r['by'].find(':') != -1:
+                cidr = NET_SIZE['6']
 
-        if r['by'].find(':') != -1:
-            cidr = NET_SIZE['6']
+            else:
+                cidr = NET_SIZE['4']
 
-        else:
-            cidr = NET_SIZE['4']
-
-        r['by'] = str(ip_network(f"{r['by']}/{cidr}", strict=False)).split('/', 1)[0]
-        if r['by'] in ['::', '::1', '127.0.0.0']:
-            r['by'] = ''
+            r['by'] = str(ip_network(f"{r['by']}/{cidr}", strict=False)).split('/', 1)[0]
+            if r['by'] in ['::', '::1', '127.0.0.0']:
+                r['by'] = ''
 
         for k in ARCHIVE_DEDUPE_FIELDS:
             if r[k] == '':
@@ -77,47 +70,60 @@ def _reports_by_day(tmp_dir: str) -> dict[list[dict]]:
 
             r[k] = dedupe_map[k].index(r[k])
 
-        reports[day].append(r)
+        reports.append(r)
 
-    for k in ARCHIVE_DEDUPE_FIELDS:
-        with open(f'{tmp_dir_dedupe}/field_{k}.csv', 'w', encoding='utf-8') as f:
-            f.write('Key,Value\n')
-            f.write('\n'.join([f'{i},{v}' for i, v in enumerate(dedupe_map[k])]))
+    reports = sorted(reports, key=itemgetter('time'))
+
+    if len(reports) == 0:
+        return dedupe_map
 
-    for day in reports:
-        reports[day] = sorted(reports[day], key=itemgetter('time'))
+    y = str(date.year).zfill(2)
+    m = str(date.month).zfill(2)
+    d = str(date.day).zfill(2)
+    tmp_dir_mon = tmp_dir / y / m
+    shell(f'mkdir -p {tmp_dir_mon}')
+    with open(f'{tmp_dir_mon}/{y}_{m}_{d}.csv', 'w', encoding='utf-8') as f:
+        f.write(f"{','.join(HEADERS_ARCHIVE_CSV)}\n")
+        for r in reports:
+            f.write(
+                f"{r['time']},"
+                f"{r['ip']},{r['an']},{r['cat']},{r['cmt']},"
+                f"{r['by']},{r['user']}\n"
+            )
 
-    return reports
+    return dedupe_map
 
 
-def _write_reports(reports: dict[list[dict]], tmp_dir: str):
-    for y_m_d in reports:
-        y, m, _ = y_m_d.split('_')
-        tmp_dir_mon = f'{tmp_dir}/{y}/{m}'
-        shell(f'mkdir -p {tmp_dir_mon}')
-        with open(f'{tmp_dir_mon}/{y_m_d}.csv', 'w', encoding='utf-8') as f:
-            f.write(f"{','.join(HEADERS_ARCHIVE_CSV)}\n")
-            for r in reports[y_m_d]:
-                f.write(
-                    f"{r['time']},"
-                    f"{r['ip']},{r['an']},{r['cat']},{r['cmt']},"
-                    f"{r['by']},{r['user']},{r['fp']}\n"
-                )
+# todo: multi-threading
+def _generate_archive(tmp_dir: Path):
+    today = datetime.now()
+    date = ARCHIVE_START_DATE
+    dedupe_map = {k: [] for k in ARCHIVE_DEDUPE_FIELDS}
+
+    while date.year < today.year or date.month < today.month or date.day <= today.day:
+        log(f'Generating archive for day: '
+            f'{str(date.year).zfill(2)}-{str(date.month).zfill(2)}-{str(date.day).zfill(2)}')
+        dedupe_map = _generate_archive_for_day(date=date, dedupe_map=dedupe_map, tmp_dir=tmp_dir)
+        date += timedelta(days=1)
+
+    log('Writing dedupe-maps')
+    tmp_dir_dedupe = tmp_dir / 'dedupe'
+    shell(f'mkdir -p {tmp_dir_dedupe}')
+    for k in ARCHIVE_DEDUPE_FIELDS:
+        with open(f'{tmp_dir_dedupe}/field_{k}.csv', 'w', encoding='utf-8') as f:
+            f.write('Key,Value\n')
+            f.write('\n'.join([f'{i},{v}' for i, v in enumerate(dedupe_map[k])]))
 
     git_commit_and_push(user='Report Updater', cmt='Report updates', repo=REPO_ARCHIVE, tmp_dir=tmp_dir)
 
 
 def main():
     log('Prepare Repository')
     git_check_token()
-    tmp_dir = f'/tmp/risk_db_archive_{int(time())}'
+    tmp_dir = Path(f'/tmp/risk_db_archive_{int(time())}')
     git_clone(repo=REPO_ARCHIVE, tmp_dir=tmp_dir)
 
-    log('Loading & Sorting Reports by Day')
-    reports_by_day = _reports_by_day(tmp_dir)
-
-    log('Write Reports')
-    _write_reports(reports_by_day, tmp_dir)
+    _generate_archive(tmp_dir)
 
 
 if __name__ == '__main__':

src/riskdb/archiver/util.py

@@ -1,14 +1,19 @@
-from os import system as shell
+from pathlib import Path
 from datetime import datetime
+from os import system as shell
 
+from riskdb.config import MODE_TEST
 from riskdb.archiver.config import GIT_TOKEN
 
 
-def git_clone(repo: str, tmp_dir: str):
+def git_clone(repo: str, tmp_dir: Path):
     shell(f'git clone https://{repo} {tmp_dir} >/dev/null')
 
 
-def git_commit_and_push(user: str, cmt: str, repo: str, tmp_dir: str):
+def git_commit_and_push(user: str, cmt: str, repo: str, tmp_dir: Path):
+    if MODE_TEST == '1':
+        return
+
     today = datetime.now().strftime('%Y-%m-%d')
     shell(
         f"cd {tmp_dir} && "
@@ -22,5 +27,8 @@ def git_commit_and_push(user: str, cmt: str, repo: str, tmp_dir: str):
 
 
 def git_check_token():
+    if MODE_TEST == '1':
+        return
+
     if GIT_TOKEN is None or not GIT_TOKEN.startswith('ghp_'):
         raise PermissionError('Required GIT-Token was not supplied!')

src/riskdb/builder/config.py

@@ -1,5 +1,5 @@
 from os import environ
-# from datetime import timedelta
+from datetime import timedelta
 
 from riskdb.config import DL_DIR
 
@@ -15,7 +15,7 @@
 
 MMDB_DESCRIPTION = 'OXL RISK-Database - risk.oxl.app (BSD-3-Clause)'
 REPORT_COOLDOWN = 10  # sec
-# REPORT_DAYS = timedelta(days=30)  # sliding window
+REPORT_DAYS = timedelta(days=30)  # sliding window
 TOR_EXIT_NODE_LIST = 'https://check.torproject.org/torbulkexitlist'
 
 PTR_LOOKUP_THREADS = 50

src/riskdb/builder/examples/reports.txt …/builder/examples/reports_2025-05-15.txtsrc/riskdb/builder/examples/reports.txt renamed to src/riskdb/builder/examples/reports_2025-05-15.txt src/riskdb/builder/examples/reports.txt renamed to src/riskdb/builder/examples/reports_2025-05-15.txt

@@ -1,14 +1,15 @@
 # pylint: disable=R0915
 
 from os import listdir
+from datetime import datetime
 from json import JSONDecodeError
 from json import loads as json_loads
 from ipaddress import ip_address, AddressValueError
 
 from maxminddb import open_database as mmdb_database
 
 from riskdb.config import EXCLUDE_NETS_IP4, EXCLUDE_NETS_IP6, REPORT_DIR, USER_TOKENS
-from riskdb.builder.config import REPORT_COOLDOWN, ASN_MMDB_FILE_IP4, ASN_MMDB_FILE_IP6
+from riskdb.builder.config import REPORT_COOLDOWN, ASN_MMDB_FILE_IP4, ASN_MMDB_FILE_IP6, REPORT_DAYS
 from riskdb.builder.obj.ip import IP
 from riskdb.builder.obj.asn import ASN
 from riskdb.builder.obj.report import Report
@@ -17,6 +18,7 @@
 from riskdb.builder.util import log
 
 SKIP_REASONS_DEFAULT = {'no_cat': 0, 'bad_ip': 0, 'cooldown': 0, 'ignored': 0, 'bad_json': 0}
+SLIDING_WINDOW_START = datetime.now() - REPORT_DAYS
 
 
 class ReportLoader:
@@ -85,12 +87,31 @@ def __iter__(self):
 
 
 class FileLoader:
-    def __init__(self, path: str = REPORT_DIR):
+    def __init__(self, path: str = REPORT_DIR, sliding_window: bool = False, match_date: datetime = None):
         self.path = path
+        self.match_date = match_date
+        self.sliding_window = sliding_window
         self.skip_reasons = SKIP_REASONS_DEFAULT.copy()
 
     def load(self):
         for file in listdir(REPORT_DIR):
+            file_path = REPORT_DIR / file
+            if self.sliding_window or self.match_date is not None:
+                ct = datetime.fromtimestamp(file_path.stat().st_ctime)
+
+                if self.sliding_window and ct < SLIDING_WINDOW_START:
+                    continue
+
+                # only get reports of that day (even if we created the file later on)
+                if self.match_date is not None:
+                    y = str(self.match_date.year).zfill(2)
+                    m = str(self.match_date.month).zfill(2)
+                    d = str(self.match_date.day).zfill(2)
+                    ys1, ys2 = f'{y}-{m}-{d}', f'{y}_{m}_{d}'
+                    if (ct.year != y or ct.month != m or ct.day != d) and \
+                            file.find(ys1) == -1 and file.find(ys2) == -1:
+                        continue
+
             loader = ReportLoader(f'{REPORT_DIR}/{file}')
             yield from loader
 

src/riskdb/builder/load_reports.py

@@ -2,6 +2,8 @@
 from pathlib import Path
 from ipaddress import ip_network
 
+MODE_TEST = environ.get('RISKDB_TEST', '0')
+
 USER_TOKENS = [
     'ceaf6e70-71c7-4415-92c0-2be6ea5f743b',  # dummy test-token
 ]