The problem with manual enrichment

You have a list of 80 indicators from an incident report, a phishing wave, or last night's alerts. Pasting them into a web UI one at a time is fine for five. For eighty it is an afternoon, and for a daily feed it is a job nobody wants.

The web interface of mlab.sh is the fast path for one-off lookups. Everything it shows you (reputation, passive DNS, WHOIS, geolocation, related indicators) is also available as JSON through the REST API, which means it can be scripted. This tutorial walks through the patterns that matter: authentication, batch loops, rate limit handling, caching, and getting the results somewhere useful.

One note before the code: endpoint paths and response fields below follow the general shape of the API but are simplified for readability. Check the developer documentation for the exact endpoints, parameters, and current response schemas before you build against them.


Setup and authentication

Generate an API key from your account settings, then keep it out of your code. Environment variables are the minimum bar:

import os
import requests

API_BASE = "https://api.mlab.sh/v1"
API_KEY = os.environ["MLAB_API_KEY"]

session = requests.Session()
session.headers.update({
    "Authorization": f"Bearer {API_KEY}",
    "Accept": "application/json",
    "User-Agent": "soc-enrichment-script/1.0",
})

Using a Session matters more than it looks: it reuses TCP connections across requests, which is noticeably faster once you are doing hundreds of lookups.


Your first lookup

Each indicator type has its own lookup path. A minimal IP lookup:

resp = session.get(f"{API_BASE}/lookup/ip/192.0.2.15", timeout=15)
resp.raise_for_status()
data = resp.json()

print(data["reputation"]["score"])
print(data["geo"]["country"], data["asn"]["name"])
for record in data.get("passive_dns", [])[:5]:
    print(record["domain"], record["last_seen"])

Domains, hashes, and URLs follow the same pattern (/lookup/domain/..., /lookup/hash/..., /lookup/url with the URL passed as a parameter). The response bundles what you would otherwise collect from four different tools: reputation verdicts, WHOIS, resolution history, and geolocation in one call.


Batch enrichment done properly

The naive loop works until it hits a rate limit or a transient error at indicator 62 of 80 and loses everything. A production-worthy loop needs three things: type detection, rate limit handling, and partial-failure tolerance.


Detecting indicator types

import re

def classify(ioc: str) -> str:
    ioc = ioc.strip().lower().replace("[.]", ".").replace("hxxp", "http")
    if re.fullmatch(r"(\d{1,3}\.){3}\d{1,3}", ioc):
        return "ip"
    if re.fullmatch(r"[a-f0-9]{32}|[a-f0-9]{40}|[a-f0-9]{64}", ioc):
        return "hash"
    if ioc.startswith(("http://", "https://")):
        return "url"
    if re.fullmatch(r"([a-z0-9-]+\.)+[a-z]{2,}", ioc):
        return "domain"
    return "unknown"

Note the defanging cleanup on the first line: real-world IOC lists arrive with [.] and hxxp in them, and an enrichment script that chokes on defanged input will be abandoned within a week.


Respecting rate limits

API plans come with request quotas. When you exceed the per-minute rate, the API answers 429 Too Many Requests, typically with a Retry-After header. Handle it explicitly instead of crashing:

import time

def lookup(ioc: str, ioc_type: str, max_retries: int = 3) -> dict | None:
    url = f"{API_BASE}/lookup/{ioc_type}/{ioc}"
    for attempt in range(max_retries):
        resp = session.get(url, timeout=15)
        if resp.status_code == 429:
            wait = int(resp.headers.get("Retry-After", 2 ** attempt))
            time.sleep(wait)
            continue
        if resp.status_code == 404:
            return {"ioc": ioc, "found": False}
        if resp.status_code >= 500:
            time.sleep(2 ** attempt)  # exponential backoff
            continue
        resp.raise_for_status()
        return resp.json()
    return None  # gave up; log it, do not lose the batch

Two details worth copying: 404 is treated as a valid answer ("we have nothing on this indicator" is information, not an error), and server errors get exponential backoff rather than immediate retry.

If the API offers a bulk endpoint that accepts a list of indicators in one POST, prefer it for large jobs; it is kinder to your quota and much faster. Again, the developer docs are the source of truth for what is available on your plan.


Caching: stop paying for the same answer

Incident indicator lists repeat themselves constantly. The same C2 IP shows up in twelve alerts; enriching it twelve times wastes quota and time. A small SQLite cache with a TTL fixes this:

import json
import sqlite3
import time

DB = sqlite3.connect("ioc_cache.db")
DB.execute("""CREATE TABLE IF NOT EXISTS cache
              (ioc TEXT PRIMARY KEY, data TEXT, fetched_at REAL)""")
TTL = 6 * 3600  # 6 hours; reputation data ages fast

def cached_lookup(ioc: str, ioc_type: str) -> dict | None:
    row = DB.execute("SELECT data, fetched_at FROM cache WHERE ioc = ?",
                     (ioc,)).fetchone()
    if row and time.time() - row[1] < TTL:
        return json.loads(row[0])
    data = lookup(ioc, ioc_type)
    if data is not None:
        DB.execute("REPLACE INTO cache VALUES (?, ?, ?)",
                   (ioc, json.dumps(data), time.time()))
        DB.commit()
    return data

Keep the TTL short for reputation verdicts (hours, not days). WHOIS and passive DNS history tolerate longer caching if you want to split TTLs by field.


From JSON to something your team uses

Enriched JSON sitting in a script variable helps nobody. The two most common destinations are a CSV for humans and a SIEM for machines.


CSV for the investigation channel

import csv

FIELDS = ["ioc", "type", "verdict", "score", "country", "asn",
          "first_seen", "pdns_count"]

def to_row(ioc, ioc_type, d):
    return {
        "ioc": ioc,
        "type": ioc_type,
        "verdict": d.get("reputation", {}).get("verdict", "unknown"),
        "score": d.get("reputation", {}).get("score", ""),
        "country": d.get("geo", {}).get("country", ""),
        "asn": d.get("asn", {}).get("name", ""),
        "first_seen": d.get("first_seen", ""),
        "pdns_count": len(d.get("passive_dns", [])),
    }

with open("iocs.txt") as f, open("enriched.csv", "w", newline="") as out:
    writer = csv.DictWriter(out, fieldnames=FIELDS)
    writer.writeheader()
    for line in f:
        ioc = line.strip()
        t = classify(ioc)
        if t == "unknown" or not ioc:
            continue
        data = cached_lookup(ioc, t)
        if data:
            writer.writerow(to_row(ioc, t, data))

That is the whole tool: an indicator file in, a sortable spreadsheet out, with malicious verdicts floating to the top when you sort by score.


Pushing to a SIEM

For continuous use, feed enrichment into the SIEM so alerts arrive pre-decorated. The pattern for Splunk HEC (Elastic and Sentinel equivalents look nearly identical):

def push_to_splunk(event: dict):
    requests.post(
        "https://splunk.example.com:8088/services/collector/event",
        headers={"Authorization": f"Splunk {os.environ['HEC_TOKEN']}"},
        json={"sourcetype": "mlab:enrichment", "event": event},
        timeout=10,
    )

Two integration styles work well:

  • Lookup table refresh: a scheduled job enriches all indicators seen in the last 24 hours and writes a lookup table the SIEM joins against at search time. Cheap, simple, slightly stale.
  • Enrich-on-alert: your SOAR or a small webhook service calls the API when an alert fires and attaches the result to the ticket. Fresh data, and quota is only spent on indicators someone will actually look at.

A blocklist export is the same loop with a filter: keep entries where the verdict is malicious and the score clears your threshold, write one indicator per line, and publish it where your proxy or firewall picks it up. Add an expiry so stale entries age out; yesterday's C2 IP is next month's reassigned cloud address.


Scheduling it

The final step is removing yourself from the loop. A cron entry running the feed job every morning:

# m h dom mon dow command
15 6 * * * /usr/bin/python3 /opt/soc/enrich_feed.py >> /var/log/enrich.log 2>&1

For event-driven flows, run the enrichment behind a small HTTP endpoint and point your alerting pipeline's webhook at it. Either way, log every run: indicators processed, cache hits, API errors, and quota consumed. When enrichment silently breaks, alerts quietly get worse, and nobody notices until triage feels slow again.


Where this fits

Automated enrichment is not an intel program by itself. It is plumbing, but it is the plumbing that turns a raw indicator list into something an analyst can act on in seconds, and it removes the most repetitive fifteen minutes from every triage. Start with the CSV script, graduate to the SIEM integration, and keep the developer documentation open while you build; that is where the exact endpoints, quotas, and response fields live.

Enrichment is a machine's job. Write the loop once, cache aggressively, respect the rate limits, and let your analysts spend their attention on the verdicts instead of the lookups.