VPN, proxy, and Tor detection in Python

Quick start (requests)

The minimum useful integration is a five-second-timeout POST to /v1/check:

import requests

def check_vpn(ip: str) -> dict:
    r = requests.post(
        "https://iplogs.com/v1/check",
        json={"ip": ip},
        timeout=5.0,
    )
    r.raise_for_status()
    return r.json()

verdict = check_vpn("1.1.1.1")
print(verdict["verdict"], verdict["score"])
print(verdict["ip_info"]["org"])

The response contains verdict (one of clean, suspicious, vpn_likely, vpn_detected), score (0 to 1), is_vpn, an ip_info object with ASN, geo, and provider data, and a signals[] array showing every feed or probe that matched. See the API docs for the full schema.

Async variant (httpx)

import httpx

async def check_vpn(ip: str) -> dict:
    async with httpx.AsyncClient(timeout=5.0) as client:
        r = await client.post(
            "https://iplogs.com/v1/check",
            json={"ip": ip},
        )
        r.raise_for_status()
        return r.json()

Use httpx in any async stack (FastAPI, Starlette, aiohttp handlers, Sanic). Async lets a single worker overlap the 5-second IPLogs call with other I/O instead of blocking the event loop.

Reading the visitor IP from a trusted proxy

Behind a reverse proxy or CDN the client IP arrives in a header. Read the right one for your edge — Caddy, Nginx, and Traefik populate X-Forwarded-For; Cloudflare uses CF-Connecting-IP; many proxies also set X-Real-IP to the immediate peer. Never trust the raw header from an untrusted source.

from flask import request

def client_ip() -> str:
    # Trust only because this app is behind a single-hop reverse proxy.
    # If you have multiple hops, parse the chain explicitly.
    cf = request.headers.get("CF-Connecting-IP")
    if cf:
        return cf.strip()
    xff = request.headers.get("X-Forwarded-For", "")
    if xff:
        # Caddy / Nginx append the real peer last; read the LAST entry.
        return xff.split(",")[-1].strip()
    return request.remote_addr or ""

Caching verdicts per IP

A 5-minute per-IP cache drops latency to sub-millisecond on the hot path and reduces upstream load. The shape of the cache key is just the IP string — verdicts are deterministic per IP at a given point in time.

from cachetools import TTLCache
import requests

_cache: TTLCache = TTLCache(maxsize=10_000, ttl=300)

def check_vpn_cached(ip: str) -> dict:
    if ip in _cache:
        return _cache[ip]
    r = requests.post(
        "https://iplogs.com/v1/check",
        json={"ip": ip},
        timeout=5.0,
    )
    r.raise_for_status()
    data = r.json()
    _cache[ip] = data
    return data

For multi-instance services use Redis with the same 300-second TTL so workers share the cache. Keep the cache key plain (the IP string); avoid prefixes that change across deploys, which would cold-start the cache every release.

Retries and failure handling

/v1/check has a server-side cap of around 8 seconds for the heaviest IPs and a 30-second handler timeout. A single 5-second client timeout is the right default. On timeout or 5xx, fail open (allow the request) rather than hard-fail your own service:

from typing import Optional
import requests

def check_vpn_safe(ip: str) -> Optional[dict]:
    try:
        r = requests.post(
            "https://iplogs.com/v1/check",
            json={"ip": ip},
            timeout=5.0,
        )
        if r.status_code >= 500:
            return None
        r.raise_for_status()
        return r.json()
    except (requests.Timeout, requests.ConnectionError):
        return None

Callers should treat Noneas “unknown” and degrade the policy gracefully — for example, skip the VPN check but keep the rest of the risk score. Aggressive automatic retries are the wrong default; one in-line retry with backoff is usually enough.

Batch with /v1/bulk-check

For offline fraud-queue review or analytical jobs, post up to 100 IPs at once. The bulk endpoint returns a compact per-row response (verdict, score, country, ASN, provider) instead of the full signal set:

import requests

def bulk_check(ips: list[str]) -> list[dict]:
    r = requests.post(
        "https://iplogs.com/v1/bulk-check",
        json={"ips": ips},
        timeout=95.0,  # endpoint has a 90s server-side budget
    )
    r.raise_for_status()
    return r.json()["results"]

# Chunk a larger list into batches of 100
def review_queue(all_ips: list[str]) -> dict:
    results = []
    for i in range(0, len(all_ips), 100):
        results.extend(bulk_check(all_ips[i : i + 100]))
    return {row["ip"]: row for row in results}

Type definitions

Static types catch shape drift at PR time rather than in production logs. A minimal TypedDict / Pydantic model:

from typing import Literal, TypedDict

Verdict = Literal["clean", "suspicious", "vpn_likely", "vpn_detected"]

class IPInfo(TypedDict, total=False):
    ip: str
    asn: str
    org: str
    country: str
    country_code: str
    is_vpn: bool
    vpn_provider: str
    vpn_provider_sources: list[str]

class Signal(TypedDict):
    type: str
    weight: float
    matched: bool
    detail: str

class CheckResponse(TypedDict):
    verdict: Verdict
    score: float
    is_vpn: bool
    confidence: float
    signals: list[Signal]
    ip_info: IPInfo
    request_id: str

Applying policy by verdict

The verdict bucket maps cleanly to a policy decision per route. See the longer compliance write-up for the why; here is the code:

def policy(verdict: dict, action: str) -> str:
    v = verdict.get("verdict", "clean")
    is_write = action in {"signup", "payment", "kyc", "withdrawal", "admin"}

    if v == "vpn_detected" and is_write:
        return "block"
    if v in {"vpn_detected", "vpn_likely"} and is_write:
        return "step_up_auth"
    if v == "suspicious" and is_write:
        return "elevate_risk"
    return "allow"

Avoid hard-blocking read traffic on any verdict — false positives there are visible to users and tend to be ordinary customers on CGNAT, Apple Private Relay, or Cloudflare WARP. See the glossary for why those are distinct categories.

Testing without hitting the network

# pytest + responses
import responses
from myapp.ip import check_vpn

@responses.activate
def test_check_vpn_clean():
    responses.add(
        responses.POST,
        "https://iplogs.com/v1/check",
        json={"verdict": "clean", "score": 0.0, "is_vpn": False,
              "ip_info": {"ip": "1.1.1.1"}, "signals": []},
        status=200,
    )
    out = check_vpn("1.1.1.1")
    assert out["verdict"] == "clean"

FAQ