Building IP Intelligence Data Pipelines for Analytics and SIEM

// Python: Batch enrichment with chunked processing
import requests

def enrich_ip_batch(ip_list, api_key, chunk_size=1000):
    """Enrich a list of IPs using batch API calls."""
    results = {}
    for i in range(0, len(ip_list), chunk_size):
        chunk = ip_list[i:i + chunk_size]
        response = requests.post(
            "https://ip-info.app/api/module-app/"
            "v1-bulk-ip-details",
            headers={
                "x-api-key": api_key,
                "Content-Type": "application/json"
            },
            json={"ips": chunk}
        )
        for record in response.json().get("results", []):
            results[record["ip"]] = {
                "country": record.get("country"),
                "city": record.get("city"),
                "isp": record.get("isp"),
                "asn": record.get("asn"),
                "is_vpn": record.get("isVpn", False),
                "risk_score": record.get("riskScore"),
            }
    return results

// Node.js / TypeScript: Stream enrichment middleware
import { moduleAppClient } from "@/sdks/module-app-client";

async function enrichEvent(event: Record<string, unknown>) {
  const ip = event.clientIp as string;
  if (!ip) return event;

  try {
    const { data } = await moduleAppClient
      .v1GetIpDetails
      .v1GetIpDetailsAction({ ip });

    return {
      ...event,
      geo: {
        country: data.country,
        city: data.city,
        timezone: data.timezone,
      },
      network: {
        isp: data.isp,
        asn: data.asn,
        connectionType: data.connectionType,
      },
      threat: {
        isVpn: data.isVpn,
        isProxy: data.isProxy,
        isTor: data.isTor,
        riskScore: data.riskScore,
      },
    };
  } catch {
    // Return original event on API failure
    return { ...event, geo: null, network: null, threat: null };
  }
}

# Elasticsearch: Enrich a query with IP geolocation data
# Assumes a pre-built enrichment index called "ip_intel"

GET /security-events-*/_search
{
  "query": {
    "bool": {
      "must": [
        { "term": { "event_type": "login_failed" } }
      ],
      "should": [
        {
          "term": {
            "ip_intel.threat.is_vpn": true
          }
        }
      ]
    }
  },
  "aggs": {
    "by_country": {
      "terms": { "field": "ip_intel.geo.country" }
    },
    "vpn_traffic_pct": {
      "filters": {
        "filters": {
          "vpn": { "term": { "ip_intel.threat.is_vpn": true } },
          "total": { "match_all": {} }
        }
      }
    }
  }
}