How to Detect a Leaked API Key From Its Traffic

The fastest way to detect a leaked API key isn't a traffic spike. It's the network. A key that has called your API from AWS in Frankfurt for six months doesn't start calling from a residential proxy in Omaha unless somebody else is holding it.

Volume alerts won't tell you. Google's threat intelligence team published an LLMjacking report on November 2 describing a threat actor who harvested LLM API keys from public GitHub repos and tested them with single-token requests (max_tokens=1) that stayed under the victims' SIEM volume alerts. By count, that sweep is invisible. By source network, it can show on the first call.

This post builds that check in Python: a per-key network baseline, one enrichment lookup per new key-and-network pair, a drift score, and an alert that tells on-call exactly what network just showed up. Every response below comes from a live lookup.

TL;DR

• Volume alerts miss the quiet testing phase of a stolen API key. Google's report documents max_tokens=1 validation sweeps that stayed under SIEM thresholds.
• A server-side key's identity is its set of networks (ASNs), not its IP addresses. Autoscaling changes IPs all day. It rarely changes the ASN.
• For machine keys, the usual IP risk model flips. A hosting ASN is normal, and a residential proxy is the strongest single sign of a leaked API key.
• Enrich only the first time a key shows up from a new /24 (or IPv6 /64), off the hot path. You pay per new network, not per request.
• Score the drift, alert with evidence, and never auto-revoke on one signal. The hot path costs one atomic Redis script call per request. Everything else runs in a worker that looks up each new network once, compares it with what that key has used before, and alerts only when the new network looks like someone else's infrastructure rather than the customer's next deploy.

Why volume alerts can't detect a leaked API key early

Volume-based API key abuse detection loses to the way stolen keys get used: validate quietly, then spend fast. Across the incidents in Google's LLMjacking report, fraudulent charges passed $100,000 in a single day, and some organizations reached $1 million before access was revoked. One actor ran more than 500,000 LLM API calls in under eight hours. A volume alert catches that second phase. By then the bill already exists.

The API key leak itself often never touches a repository. In October, OpenCode said a breach of its hosted Console may have exposed the provider keys saved by 1.4% of Console users. Secret scanning can't find a key copied out of somebody else's database. OpenCode couldn't cap what happened next either, because Console only sees requests that pass through its own gateway.

That's the general rule: the only party that sees every request made with a key is the provider that issued it. If you run an API with customer keys, detection has to live where those requests land, at your gateway, keyed by credential.

For API keys, the IP risk model is upside down

Most IP-risk advice is written for users. At a signup form, a datacenter IP is suspicious because nobody fills in a registration form from an EC2 instance. Switch the subject to a server-side API key and the logic inverts.

Attackers pick residential exits because they look like customers. In the 2024 credential-stuffing wave Okta tracked, "millions of the requests" went through residential proxies, and most of the traffic appeared to come from everyday users' devices rather than VPS ranges. That camouflage works against a login form. Against a key that has only ever called from AWS, it's a flare.

Stripe's guidance already leans on this. For a compromised API key, it says to restrict keys to known IP addresses, and for keys without that restriction, to look for events from IPs you don't recognize. The rest of this post is that second instruction, automated.

Scope: this is for server-side keys. Browser and mobile keys come from residential networks by design, so the inversion doesn't hold for them.

What a key's network baseline holds

A baseline is three small sets per key: the ASNs it has called from, their types, and the countries.

{
  "key_id": "key_acme_prod",
  "asns": ["AS16509"],
  "types": ["HOSTING"],
  "countries": ["DE"]
}

Three rules keep it honest. Key it by an internal key ID, never the raw key, so the detector's Redis doesn't become a second place your secrets leak from. Give each new key a learning window (seven days here) where clean networks get added freely, since everything is new in week one. And never learn a network that scored as suspicious, or a patient attacker teaches your baseline that their proxy pool is normal.

Group addresses before you look them up: IPv4 by /24, IPv6 by /64. An autoscaling group can rotate through dozens of addresses an hour inside the same few subnets, and IPv6 clients change addresses inside one /64 all the time. You want to notice a new network, not a new address. If you want background on the asn.type values themselves, I went through them in Classify an IP by ASN Type.

The hot path: one Redis call per request

Everything below goes in one file, keywatch.py. You need Python 3.10+, a Redis instance, and pip install redis requests. Call observe() right after your auth layer resolves the API key to a key ID.

import ipaddress
import json
import logging
import os
import time

import redis
import requests

log = logging.getLogger("keywatch")
r = redis.Redis.from_url(os.environ.get("REDIS_URL", "redis://localhost:6379/0"),
                         decode_responses=True)
API_KEY = os.environ.get("IPGEO_API_KEY")  # server-side only, never in client code
QUEUE = "kw:queue"

# Atomically mark a bucket as seen and enqueue its first observation.
# This prevents a crash between SADD and RPUSH from creating a blind spot.
ENQUEUE_IF_NEW = r.register_script("""
local added = redis.call('SADD', KEYS[1], ARGV[1])
if added == 1 then
    redis.call('RPUSH', KEYS[2], ARGV[2])
end
return added
""")


def network_bucket(ip: str) -> tuple[str, str] | None:
    """Return the canonical public IP and its network bucket."""
    try:
        addr = ipaddress.ip_address(ip)
    except ValueError:
        return None

    # Some servers expose IPv4 clients as ::ffff:8.8.8.8.
    # Normalize them before choosing the prefix length.
    if isinstance(addr, ipaddress.IPv6Address) and addr.ipv4_mapped:
        addr = addr.ipv4_mapped

    if not addr.is_global:
        return None  # private, loopback, CGNAT, link-local, or reserved

    prefix = 24 if addr.version == 4 else 64
    bucket = ipaddress.ip_network(f"{addr}/{prefix}", strict=False)
    return str(addr), str(bucket)


def observe(key_id: str, client_ip: str) -> None:
    """Call once per authenticated request. One Redis round trip."""
    normalized = network_bucket(client_ip)
    if normalized is None:
        return

    client_ip, bucket = normalized
    job = {
        "key_id": key_id,
        "ip": client_ip,
        "bucket": bucket,
        "ts": time.time(),
    }

    try:
        ENQUEUE_IF_NEW(
            keys=[f"kw:seen:{key_id}", QUEUE],
            args=[bucket, json.dumps(job)],
        )
    except redis.RedisError as exc:
        # Fail open: a detector outage must never fail a customer's API call.
        log.warning("keywatch observe skipped: %s", type(exc).__name__)

A key calling from a network it already knows costs one script call that returns 0. For a new network, the same atomic script marks the bucket as seen and queues one job before returning. If Redis is down, the request goes through and you lose a sighting, which is the right trade for something that sits in front of every call.

Pitfall: client_ip has to be the address your edge saw, not whatever is in X-Forwarded-For. Set your framework's trusted-proxy option (uvicorn's --forwarded-allow-ips, Express's trust proxy) to the exact proxies you run. Otherwise an attacker writes a baseline-friendly AWS address into a header and the detector believes them.

Enrichment: one lookup per new network

Any IP data source that returns ASN type and a residential-proxy flag works here. MaxMind's GeoIP Anonymous IP database has an is_residential_proxy field, and ipinfo sells a residential proxy API; pair either with ASN data and the scoring carries over. I'm using IPGeolocation's IP security data because one /v3/ipgeo call with include=security returns location, the ASN with its type, and the proxy flags with provider names together.

That call costs 3 credits: 1 for the base lookup, 2 for security. The $19 plan's 150,000 credits cover 50,000 lookups a month, and you only look a network up the first time a key appears on it. The cache is shared across keys, so a shared egress IP that ten of your customers call from (a PaaS, a CI provider) is looked up once. Security data needs a paid plan; a free key gets a 401 for include=security.

Here's the call for the IP that shows up in the worked example later:

curl -s "https://api.ipgeolocation.io/v3/ipgeo?apiKey=$IPGEO_API_KEY&ip=145.223.7.7&include=security&fields=location.country_code2,location.city,asn,security"

{
  "ip": "145.223.7.7",
  "location": {
    "country_code2": "US",
    "city": "Omaha"
  },
  "asn": {
    "as_number": "AS209854",
    "organization": "Cyberzone S.A.",
    "country": "PA",
    "type": "HOSTING",
    "domain": "",
    "date_allocated": "2018-11-21",
    "rir": "RIPE"
  },
  "security": {
    "threat_score": 90,
    "is_tor": false,
    "is_proxy": true,
    "proxy_provider_names": [
      "DataImpulse", "ProxyScrape", "Oxy Labs", "ProxyEmpire", "Proxy-Store",
      "SpyderProxy", "FleetProxy", "Proxy4u", "FlashProxy", "Proxy001", "Databay"
    ],
    "proxy_confidence_score": 99,
    "proxy_last_seen": "2026-11-10",
    "proxy_days_seen_30d": 27,
    "proxy_days_seen_7d": 7,
    "proxy_provider_count_30d": 10,
    "is_residential_proxy": true,
    "is_vpn": true,
    "vpn_provider_names": ["SurfShark VPN"],
    "vpn_confidence_score": 99,
    "vpn_last_seen": "",
    "is_relay": false,
    "relay_provider_name": "",
    "is_anonymous": true,
    "is_known_attacker": true,
    "is_bot": false,
    "bot_confidence_score": 0,
    "bot_operator_name": "",
    "bot_type": "",
    "is_known_good_bot": false,
    "bot_last_seen": "",
    "is_spam": true,
    "is_cloud_provider": true,
    "cloud_provider_name": "Brander Group Inc.",
    "is_corporate_gateway": false,
    "corporate_gateway_type": "",
    "corporate_gateway_provider_name": ""
  }
}

The fields parameter trims the location object and drops time zone and currency, which the detector doesn't need. The asn and security objects are complete. One gotcha if you pull the company object too: baseline on asn.type, not company.type. For 54.239.28.85, an amazon.com address, the ASN says HOSTING and the company record says ISP.

IPGEO_URL = "https://api.ipgeolocation.io/v3/ipgeo"
FIELDS = "location.country_code2,location.city,asn,security"


class EnrichmentError(Exception):
    pass


def enrich(ip: str) -> dict | None:
    """Location, ASN and security in one call (3 credits). Cached 24h for every key."""
    cached = r.get(f"kw:intel:{ip}")
    if cached:
        return json.loads(cached)
    try:
        resp = requests.get(IPGEO_URL, params={
            "apiKey": API_KEY, "ip": ip, "include": "security", "fields": FIELDS,
        }, timeout=(2.0, 5.0))  # off the hot path, so a slower timeout is fine
    except requests.RequestException as exc:
        # Never include the exception object: it may contain the URL and API key.
        raise EnrichmentError(
            f"network error: {type(exc).__name__}"
        ) from None
    if resp.status_code == 423:
        return None  # bogon or reserved address; rare after network_bucket()
    if resp.status_code != 200:
        raise EnrichmentError(f"HTTP {resp.status_code}: {resp.text[:200]}")
    try:
        intel = resp.json()
    except ValueError as exc:
        raise EnrichmentError("response was not JSON") from exc
    r.set(f"kw:intel:{ip}", json.dumps(intel), ex=86400)
    return intel

The timeout is looser than you'd use inline, (2.0, 5.0) seconds, because nothing is waiting on this. Errors raise instead of returning a default so the worker can decide to retry rather than score empty data.

Scoring the drift

The weights below are starting points. Tune them against a week of your own traffic.

Alert at 3, contain at 5. The 45 cutoff matches the start of the "elevated risk" band on the same security API page.

ALERT_AT = 3
STEP_UP_AT = 5


def score(ip: str, baseline: dict, intel: dict, declared: set[str]) -> tuple[int, list[str]]:
    asn = intel.get("asn") or {}
    sec = intel.get("security") or {}
    country = (intel.get("location") or {}).get("country_code2")
    as_number = asn.get("as_number")
    pts, why = 0, []

    def add(n: int, reason: str) -> None:
        nonlocal pts
        pts += n
        why.append(f"{n:+d} {reason}")

    # Novelty only means something once the key has finished learning.
    if baseline["learned"]:
        if as_number and as_number not in baseline["asns"]:
            add(1, f"new ASN {as_number} ({asn.get('organization', 'unknown')})")
        if asn.get("type") == "ISP" and "ISP" not in baseline["types"]:
            add(2, "first call from a consumer ISP network")
        if country and country not in baseline["countries"]:
            add(1, f"new country {country}")

    # Intrinsic signals count from the very first request.
    if sec.get("is_residential_proxy"):
        add(3, "residential proxy exit")
    elif sec.get("is_proxy") or sec.get("is_vpn") or sec.get("is_tor"):
        add(2, "proxy, VPN or Tor exit")
    if sec.get("proxy_days_seen_7d", 0) >= 3 and sec.get("proxy_provider_count_30d", 0) >= 2:
        add(1, "busy proxy exit (several providers, most days this week)")
    if sec.get("threat_score", 0) >= 45:
        add(1, f"threat_score {sec['threat_score']}")

    # Known-good egress the customer will legitimately use.
    if sec.get("is_corporate_gateway"):
        add(-2, f"corporate gateway ({sec.get('corporate_gateway_provider_name') or 'unnamed'})")
    addr = ipaddress.ip_address(ip)
    if any(addr in ipaddress.ip_network(cidr, strict=False) for cidr in declared):
        add(-2, "inside a network the customer declared")
    return pts, why

Novelty signals only count after the learning window, but proxy and threat signals count from request one. A key that leaks in its first week still alerts: on a brand-new key, the Omaha exit scores 5 on intrinsic signals alone.

Three live lookups, scored against a key whose baseline is AWS in Frankfurt (AS16509, HOSTING, DE):

Why the proxy flag outweighs ASN type

Look at the third row again. That residential proxy exit sits on a HOSTING network, so a rule that only fires when a cloud key moves to a consumer ISP scores it zero on that line. The +3 came from is_residential_proxy. Proxy pools mix exits on home connections with exits on hosting space, which is why ASN type and the proxy flags are separate signals here instead of one.

Signals that should lower the score

Corporate gateways are the common case. A developer on your customer's team testing from the office comes through Zscaler or Netskope, from the vendor's address space, often in another country. The Dubai row above is exactly this, and the -2 keeps it from paging anyone.

CI runners are the other one. GitHub's docs now say eligible Linux runners use a multi-cloud compute pool, so a customer's test suite can call you from a cloud the key has never touched. GitHub publishes those ranges in the actions key of its /meta API. Let customers declare ranges like that per key in kw:declared:{key_id}, and the -2 covers them.

The worker: learn clean networks, alert on the rest

The baseline lives in three Redis sets per key, plus a timestamp for the first sighting:

LEARNING_SECONDS = 7 * 86400


def load_baseline(key_id: str) -> dict:
    r.set(f"kw:born:{key_id}", time.time(), nx=True)  # first sighting starts the clock
    born = float(r.get(f"kw:born:{key_id}"))
    return {
        "learned": time.time() - born > LEARNING_SECONDS,
        "asns": r.smembers(f"kw:asns:{key_id}"),
        "types": r.smembers(f"kw:types:{key_id}"),
        "countries": r.smembers(f"kw:countries:{key_id}"),
    }


def learn(key_id: str, intel: dict) -> None:
    asn = intel.get("asn") or {}
    country = (intel.get("location") or {}).get("country_code2")
    pipe = r.pipeline()
    if asn.get("as_number"):
        pipe.sadd(f"kw:asns:{key_id}", asn["as_number"])
    if asn.get("type"):
        pipe.sadd(f"kw:types:{key_id}", asn["type"])
    if country:
        pipe.sadd(f"kw:countries:{key_id}", country)
    pipe.execute()

And the loop that ties it together:

def handle(job: dict) -> None:
    key_id, ip = job["key_id"], job["ip"]
    try:
        intel = enrich(ip)
    except EnrichmentError as exc:
        log.warning("enrichment failed for %s: %s", ip, exc)
        r.srem(f"kw:seen:{key_id}", job["bucket"])  # retry on the next request
        return
    if intel is None:
        return
    pts, why = score(ip, load_baseline(key_id), intel, r.smembers(f"kw:declared:{key_id}"))
    if pts >= ALERT_AT:
        raise_alert(key_id, ip, intel, pts, why)  # never learn a suspicious network
    else:
        learn(key_id, intel)


def run_worker() -> None:
    if not API_KEY:
        raise RuntimeError("IPGEO_API_KEY is not set")

    while True:
        item = r.blpop(QUEUE, timeout=5)
        if item is None:
            continue

        job = None
        try:
            job = json.loads(item[1])
            handle(job)
        except Exception:
            # One bad job must not kill the worker.
            log.exception("keywatch job failed")

            # Let the next request enqueue this network again.
            if isinstance(job, dict) and job.get("key_id") and job.get("bucket"):
                try:
                    r.srem(f"kw:seen:{job['key_id']}", job["bucket"])
                except redis.RedisError as exc:
                    log.warning(
                        "keywatch retry-marker cleanup failed: %s",
                        type(exc).__name__,
                    )

Run it as its own process with python -c "import keywatch; keywatch.run_worker()". A failed lookup removes the seen marker, so the next request from that network queues it again instead of leaving a permanent blind spot. BLPOP drops a job if the worker dies mid-handle. If you need stronger delivery, move the queue to a Redis Stream with a consumer group.

What the alert should say

An alert that says "key_acme_prod: weird IP 145.223.7.7" gets ignored. One that names the network gets acted on. The alert step makes one more call, to the dedicated /v3/asn endpoint, for 1 credit, only when something has already scored high:

ASN_URL = "https://api.ipgeolocation.io/v3/asn"


def asn_context(as_number: str) -> dict:
    """1 credit, alerts only: who runs this network and how big is it."""
    try:
        resp = requests.get(ASN_URL, params={
            "apiKey": API_KEY, "asn": as_number.removeprefix("AS"),
        }, timeout=(2.0, 5.0))
        resp.raise_for_status()
        return resp.json().get("asn") or {}
    except requests.HTTPError as exc:
        # Do not log the exception object because it may include the API key.
        status = exc.response.status_code if exc.response is not None else "unknown"
        log.warning("no ASN context for %s: HTTP %s", as_number, status)
        return {}
    except requests.RequestException as exc:
        log.warning(
            "no ASN context for %s: %s",
            as_number,
            type(exc).__name__,
        )
        return {}
    except ValueError:
        log.warning("no ASN context for %s: invalid JSON", as_number)
        return {}  # the alert still goes out without it


def raise_alert(key_id: str, ip: str, intel: dict, pts: int, why: list[str]) -> None:
    as_number = (intel.get("asn") or {}).get("as_number") or "unknown"
    # One alert per key and network per hour, not one per rotating proxy IP.
    if not r.set(f"kw:alerted:{key_id}:{as_number}", 1, nx=True, ex=3600):
        return
    ctx = asn_context(as_number) if as_number != "unknown" else {}
    sec = intel.get("security") or {}
    alert = {
        "key_id": key_id, "ip": ip, "score": pts, "reasons": why,
        "action": "contain_and_notify" if pts >= STEP_UP_AT else "notify",
        "network": {"asn": as_number, "name": ctx.get("asn_name"),
                    "org": ctx.get("organization"), "type": ctx.get("type"),
                    "ipv4_routes": ctx.get("num_of_ipv4_routes")},
        "proxy_providers": sec.get("proxy_provider_names", []),
    }
    log.warning("possible leaked key: %s", json.dumps(alert))
    # Send `alert` to your pager and the customer's security contact here.

For the Omaha request, that produces:

{
  "key_id": "key_acme_prod",
  "ip": "145.223.7.7",
  "score": 7,
  "reasons": [
    "+1 new ASN AS209854 (Cyberzone S.A.)",
    "+1 new country US",
    "+3 residential proxy exit",
    "+1 busy proxy exit (several providers, most days this week)",
    "+1 threat_score 90"
  ],
  "action": "contain_and_notify",
  "network": {
    "asn": "AS209854",
    "name": "Cyberzonehub",
    "org": "Cyberzone S.A.",
    "type": "HOSTING",
    "ipv4_routes": "365"
  },
  "proxy_providers": [
    "DataImpulse", "ProxyScrape", "Oxy Labs", "ProxyEmpire", "Proxy-Store",
    "SpyderProxy", "FleetProxy", "Proxy4u", "FlashProxy", "Proxy001", "Databay"
  ]
}

What to do with it depends on the score:

• 3 to 4: notify the customer with that evidence. At this level it's often a new deploy region or a contractor's laptop, so lead with a question, not an accusation.
• 5 and up: if the customer configured an exact CIDR allowlist, enforce it. Otherwise temporarily cap spend, restrict sensitive endpoints, or require customer confirmation while asking them to rotate the key. Do not turn the learned ASN baseline into an allowlist: a shared cloud ASN contains many unrelated tenants.
• Revoke on confirmation, or when the customer goes quiet and the spend keeps climbing. I wouldn't auto-revoke on a score alone. A wrong revoke takes down a customer's production. A wrong containment action can block legitimate traffic.

Where this breaks

An attacker in the same cloud and region. Someone who rents a VM in AWS Frankfurt keeps the ASN and the country, and this detector scores them zero. Catching that takes request-shape signals on top of network ones, the rest of API key anomaly detection: endpoints the key never called, a new user agent, or the max_tokens=1 micro-requests Google's report recommends watching for.

Slow drift gets learned. After the learning window, a new cloud ASN in a new country scores 2 and gets added to the baseline without an alert. That's deliberate, because customers do move clouds, but it's also the door a careful attacker on clean cloud space walks through. For high-value keys, drop ALERT_AT to 2 and review every new network.

Mixers. The same report describes "LLM Mixers" that pool stolen keys behind their own gateways. That traffic comes from hosting space, so the residential signal won't fire. The new-ASN point still will.

You hold the key instead of issuing it. Then you can't run observe(), but you can still tell if your API key has been leaked when your provider shows request IPs. Stripe's request logs do, and the same scoring works on an export of them.

Pin keys first, detect the rest

If you issue keys, ship an allowlist too. Let customers pin a key to exact CIDRs they control, the way Stripe supports IP restrictions. Treat ASNs as anomaly-detection context unless the customer owns a dedicated ASN and explicitly chooses to trust the full network. Read a week of alerts before you let anything contain a key automatically, and expect the first few to be a contractor's laptop rather than an attacker.