threatintel.py

"""Offline threat-intel matching.

Loads blocklists from a local directory (default ./threatintel) and matches
IPs, domains, and JA3 hashes against them. Fully offline — nothing is fetched
at query time, so checking a suspicious indicator never leaks it.

Drop any number of ``*.txt`` feed files into the directory; one indicator per
line (``#`` comments and blank lines ignored). Each line is auto-classified:

    * dotted IPv4            -> IP blocklist
    * 32 hex chars           -> JA3 blocklist
    * anything else with a . -> domain blocklist

Get free feeds with ``./download_feeds.sh`` (abuse.ch Feodo/SSLBL, Tor exits, …).
"""

from __future__ import annotations

import os
import re
import threading

TI_DIR = os.environ.get("THREATINTEL_DIR",
                        os.path.join(os.path.dirname(__file__), "threatintel"))

_IPV4 = re.compile(r"^\d{1,3}(\.\d{1,3}){3}$")
_JA3 = re.compile(r"^[0-9a-f]{32}$", re.I)

_lock = threading.Lock()
_loaded = False
_ips: set[str] = set()
_domains: set[str] = set()
_ja3: set[str] = set()
_sources: list[str] = []


def _load():
    global _loaded
    if _loaded:
        return
    with _lock:
        if _loaded:
            return
        if os.path.isdir(TI_DIR):
            for fn in sorted(os.listdir(TI_DIR)):
                if not fn.lower().endswith((".txt", ".csv", ".ipset")):
                    continue
                path = os.path.join(TI_DIR, fn)
                n = 0
                try:
                    with open(path, "r", errors="ignore") as fh:
                        for line in fh:
                            line = line.strip()
                            if not line or line.startswith(("#", ";")):
                                continue
                            tok = line.split(",")[0].split()[0].strip().lower()
                            if _IPV4.match(tok):
                                _ips.add(tok); n += 1
                            elif _JA3.match(tok):
                                _ja3.add(tok); n += 1
                            elif "." in tok:
                                _domains.add(tok.lstrip(".")); n += 1
                except OSError:
                    continue
                if n:
                    _sources.append(f"{fn} ({n})")
        _loaded = True


def available() -> bool:
    _load()
    return bool(_ips or _domains or _ja3)


def stats() -> dict:
    _load()
    return {"ips": len(_ips), "domains": len(_domains), "ja3": len(_ja3),
            "sources": _sources}


def match_ip(ip):
    _load()
    return ip in _ips


def match_ja3(h):
    _load()
    return bool(h) and h.lower() in _ja3


def match_domain(domain):
    """Match a domain or any of its parent domains against the blocklist."""
    _load()
    if not domain or not _domains:
        return None
    d = domain.lower().rstrip(".")
    if d in _domains:
        return d
    parts = d.split(".")
    for i in range(1, len(parts) - 1):
        parent = ".".join(parts[i:])
        if parent in _domains:
            return parent
    return None