Building an IP-intelligence enrichment pipeline in Python

What enrichment actually means

An IP address on its own tells you almost nothing. enrichment is the unglamorous work of attaching context to it: what ports the host exposes, whether it has known vulnerabilities, what network it lives in, whether it's a data-center box or a residential line, whether it's flagged as a proxy. None of those facts is a verdict by itself. Together they let you rank a hundred addresses and spend your attention on the five that deserve it.

The input here is the kind of list every homelab and small ops team generates constantly: fail2ban bans, firewall drops, weird entries in an auth log. The output is a CSV sorted by a risk score, which is a thing you can actually act on before your coffee goes cold.

Ground rules before you query anything

Two lines I don't cross, and the pipeline is built around both. First, I only look up addresses I have a legitimate reason to care about, meaning they showed up in my own logs touching my own systems. Second, everything here is passive: we're reading data that services like Shodan have already aggregated, not probing anyone's host ourselves. Nothing in this article sends a single packet to the IPs being investigated.

And when you use someone's free API, behave like a guest. Identify yourself with a real User-Agent, honor rate limits before you hit them, back off when told, and cache so you never ask the same question twice. This isn't just etiquette; polite pipelines are also faster and more reliable, because they don't get banned halfway through a run.

The shape of the pipeline

Five stages, each one a plain function: read and validate the input, check the cache, enrich from two sources, score, write the report. Starting with reading, because garbage in a log file is a certainty, not a possibility:

import ipaddress
import sys

def load_ips(path):
    """Read one IP per line; dedupe, validate, ignore junk."""
    seen = set()
    with open(path) as f:
        for line in f:
            line = line.strip()
            if not line or line.startswith("#"):
                continue
            try:
                ip = ipaddress.ip_address(line)
            except ValueError:
                print(f"skipping junk line: {line!r}", file=sys.stderr)
                continue
            if ip.is_global:          # private/reserved space: nothing to enrich
                seen.add(str(ip))
    return sorted(seen)

The ipaddress module from the standard library does the validation for free, and the is_global check quietly drops RFC 1918 addresses before you embarrass yourself asking Shodan about 192.168.1.50.

Cache first, ask questions later

The cache goes in before any network code gets written, not after. Log files repeat themselves, reruns happen, and scripts crash halfway. A SQLite file keyed on (ip, source) means every question gets asked exactly once a week, no matter how many times the script runs:

import json
import sqlite3
import time

DB = sqlite3.connect("enrich_cache.db")
DB.execute("""
    CREATE TABLE IF NOT EXISTS cache (
        ip         TEXT NOT NULL,
        source     TEXT NOT NULL,
        data       TEXT NOT NULL,
        fetched_at REAL NOT NULL,
        PRIMARY KEY (ip, source)
    )
""")

MAX_AGE = 7 * 24 * 3600   # a week is plenty for triage work

def cache_get(ip, source):
    row = DB.execute(
        "SELECT data, fetched_at FROM cache WHERE ip = ? AND source = ?",
        (ip, source),
    ).fetchone()
    if row and time.time() - row[1] < MAX_AGE:
        return json.loads(row[0])
    return None

def cache_put(ip, source, data):
    DB.execute(
        "REPLACE INTO cache (ip, source, data, fetched_at) VALUES (?, ?, ?, ?)",
        (ip, source, json.dumps(data), time.time()),
    )
    DB.commit()

On my bench, a rerun over a 400-address list goes from six minutes to under a second once the cache is warm. That difference is what makes it practical to iterate on the scoring logic, which is where the actual analysis happens.

Talking to APIs politely

One fetch function handles the manners for every source: a User-Agent that says who you are, a timeout so a dead API can't hang the run, exponential backoff when you're told to slow down, and 404 treated as an answer rather than an error:

import requests

SESSION = requests.Session()
SESSION.headers["User-Agent"] = "ip-triage/0.2 (personal lab; [email protected])"

def fetch(url, tries=4):
    delay = 2.0
    for _ in range(tries):
        resp = SESSION.get(url, timeout=10)
        if resp.status_code == 404:
            return {}          # "no data on this host" is a result; cache it
        if resp.status_code == 429:
            time.sleep(float(resp.headers.get("Retry-After", delay)))
            delay *= 2
            continue
        resp.raise_for_status()
        return resp.json()
    raise RuntimeError(f"gave up on {url}")

The sources

Two free, keyless sources cover a surprising amount of ground. InternetDB is Shodan's fast lookup service: open ports, known CVEs, and tags for any address, no account needed. ip-api adds geolocation, the owning network, and proxy/hosting flags, free for non-commercial use at 45 requests a minute. That limit is why there's a sleep in the second function; staying under it by design beats getting throttled by surprise:

def internetdb(ip):
    cached = cache_get(ip, "internetdb")
    if cached is not None:
        return cached
    data = fetch(f"https://internetdb.shodan.io/{ip}")
    cache_put(ip, "internetdb", data)
    return data

def geo(ip):
    cached = cache_get(ip, "ip-api")
    if cached is not None:
        return cached
    fields = "status,country,as,proxy,hosting"
    data = fetch(f"http://ip-api.com/json/{ip}?fields={fields}")
    cache_put(ip, "ip-api", data)
    time.sleep(1.4)   # free tier allows 45/min; this keeps us at ~42
    return data

Both functions check the cache before touching the network, so the sleep only costs you on cold lookups. When you outgrow the free tier, keyed sources like AbuseIPDB or GreyNoise drop into this same pattern: one function, cache on both ends, manners in the middle.

Scoring for triage

Scoring is where people overthink it. You don't need a machine-learning model to rank a log file; you need a handful of weighted facts and the honesty to call the result an opinion rather than a verdict:

RISKY_PORTS = {23, 445, 1433, 3389, 5900}   # telnet, smb, mssql, rdp, vnc

def score(intel, geo_row):
    pts = 0
    ports = set(intel.get("ports", []))

    pts += 3 * len(intel.get("vulns", []))     # known CVEs are the loudest signal
    pts += 2 * len(ports & RISKY_PORTS)        # classic abuse-friendly services
    pts += min(len(ports), 5)                  # very open hosts are rarely innocent

    if geo_row.get("proxy"):
        pts += 4                               # anonymization infrastructure
    if geo_row.get("hosting"):
        pts += 2                               # data-center box, not a person

    return pts

Every weight in there is an argument you should have with yourself. Is a host behind a proxy flag worse than one with an open RDP port? For my logs, where most noise is rented scanner boxes, yes. For yours, maybe not. The point of keeping the function this small is that changing your mind costs thirty seconds and a cache-warm rerun.

Output a human can use

The report is a CSV sorted by score, highest first, because the entire job of this pipeline is to decide what you look at first:

import csv

def main(path):
    rows = []
    for ip in load_ips(path):
        intel = internetdb(ip)
        g = geo(ip)
        rows.append({
            "ip": ip,
            "score": score(intel, g),
            "ports": " ".join(map(str, intel.get("ports", []))),
            "vulns": " ".join(intel.get("vulns", [])),
            "country": g.get("country", ""),
            "asn": g.get("as", ""),
            "proxy": g.get("proxy", False),
            "hosting": g.get("hosting", False),
        })

    rows.sort(key=lambda r: r["score"], reverse=True)
    with open("triage.csv", "w", newline="") as f:
        w = csv.DictWriter(f, fieldnames=list(rows[0].keys()))
        w.writeheader()
        w.writerows(rows)
    print(f"wrote triage.csv ({len(rows)} hosts)")

if __name__ == "__main__":
    main(sys.argv[1] if len(sys.argv) > 1 else "ips.txt")

The first time I ran this against a week of fail2ban bans from my own VPS, the top of the sheet was exactly what you'd expect: hosting-provider addresses with open Telnet and a stack of CVEs, the standard rented scanner fleet. The interesting find was further down, a residential address with no exposure at all, which turned out to be my own phone on cellular failing Wi-Fi calling authentication. Triage works both directions.

Where to take it

The single-threaded version with the ip-api sleep processes about 40 fresh addresses a minute, which is fine for a daily log review. When the lists get bigger, the upgrade path is asyncio with a semaphore per source, so each API gets its own concurrency budget and the manners survive the speedup. After that: keyed sources for reputation data, a flag that diffs today's report against yesterday's, and piping high scores into whatever alerts you.

But the honest advice is that the cache and the manners scale further than you'd think, and the scoring function is where the real returns live. Tools don't do analysis. They just clear the ground so you can.