Flag Risky Signups in Python by IP: Tor, Hosting and Known VPN Networks

A FastAPI/Flask recipe that turns IP traits into an allow / verify / review decision, without locking out real users.

Sign up Read the docs

Free: 1,000 req/day · HTTPS · commercial OK

Why hard-blocking VPN users backfires

Plenty of legitimate people sign up from a VPN, a privacy relay or a phone on a mobile carrier network. Block them outright and you lose real customers while determined abusers just switch IPs. A better pattern: treat the IP as a risk signal, add friction (email verification, CAPTCHA, manual review) when signals stack up, and keep the reasons on the user record so you can tune later.

Get the real client IP in FastAPI or Flask

Behind Nginx or a load balancer, the socket address is your proxy. Tell the framework which proxy to trust, and never parse X-Forwarded-For yourself: a client can put any IP in it.

FastAPI + Uvicorn
# FastAPI / Starlette behind one trusted proxy (Uvicorn)
# uvicorn app:app --proxy-headers --forwarded-allow-ips="10.0.0.5"

from fastapi import Request

def client_ip(request: Request) -> str:
    # With --proxy-headers, request.client.host is the real visitor IP
    return request.client.host
Flask + ProxyFix
# Flask behind one proxy
from flask import Flask
from werkzeug.middleware.proxy_fix import ProxyFix

app = Flask(__name__)
app.wsgi_app = ProxyFix(app.wsgi_app, x_for=1)   # trust exactly one hop
# request.remote_addr is now the real visitor IP

Look up the IP

One call to GET /v0/ip/{ip} with your key in Authorization: Bearer returns location, network and traits. This async client uses httpx, a one-second timeout and a small per-IP cache, and returns None on failure so signups never break:

ndl.py
# ndl.py
import os, time
import httpx

API = "https://api.networkdatalabs.com/v0/ip/"
HEADERS = {"Authorization": f"Bearer {os.environ['NDL_API_KEY']}"}
_client = httpx.AsyncClient(timeout=httpx.Timeout(1.0), headers=HEADERS)
_cache: dict[str, tuple[float, dict]] = {}
TTL = 6 * 3600

async def lookup(ip: str) -> dict | None:
    hit = _cache.get(ip)
    if hit and hit[0] > time.time():
        return hit[1]
    try:
        r = await _client.get(API + ip)
        r.raise_for_status()
    except httpx.HTTPError:
        return None                      # fail open
    data = r.json()
    _cache[ip] = (time.time() + TTL, data)
    return data

The traits that matter for signups:

  • is_tor: the IP is a current Tor exit node
  • is_vpn: known VPN networks, including major commercial VPN providers
  • is_relay: privacy relay egress
  • is_hosting: hosting/datacenter detection
  • is_mobile: mobile carrier networks (an estimate)

A policy table: allow, verify or review

SignalWhat it meansSuggested action
is_torCurrent Tor exit nodeManual review
is_vpnKnown VPN networks, including major commercial VPN providersVerify email
is_hostingHosting/datacenter IP: often scripts and botsVerify email (or CAPTCHA)
is_relayPrivacy relay: usually a normal, privacy-minded userAllow, log it
is_mobileMobile carrier networks: many users share IPsAllow; don’t rate-limit by IP alone
Country mismatch + high/medium confidenceIP country differs from the declared oneVerify; review if combined with another signal
Two or more signalsStacked riskManual review

The same table in code, combined with a country check that only counts when confidence is high or medium:

risk.py
# risk.py
def assess(data: dict | None, declared_country: str | None = None) -> tuple[str, list[str]]:
    """Return ("allow" | "verify" | "review", reasons)."""
    if data is None:
        return "allow", ["lookup_unavailable"]

    t = data.get("traits", {})
    loc = data.get("location", {})
    reasons = []

    if t.get("is_tor"):
        reasons.append("tor_exit")
    if t.get("is_vpn"):
        reasons.append("known_vpn_network")
    if t.get("is_relay"):
        reasons.append("privacy_relay")
    if t.get("is_hosting"):
        reasons.append("hosting_datacenter")

    country = loc.get("country_code")
    confident = loc.get("confidence") in ("high", "medium")
    if declared_country and country and confident and country != declared_country:
        reasons.append("country_mismatch")

    if "tor_exit" in reasons or len(reasons) >= 2:
        return "review", reasons
    if reasons and reasons != ["privacy_relay"]:
        return "verify", reasons
    return "allow", reasons

Wire it into the signup endpoint

app.py (FastAPI)
# app.py
from fastapi import FastAPI, Request
from pydantic import BaseModel
from ndl import lookup
from risk import assess

app = FastAPI()

class Signup(BaseModel):
    email: str
    country: str | None = None   # from the billing/profile form, e.g. "US"

@app.post("/signup")
async def signup(body: Signup, request: Request):
    ip = request.client.host
    decision, reasons = assess(await lookup(ip), body.country)

    user = create_user(body.email, signup_ip=ip, risk=reasons)
    if decision == "verify":
        send_verification_email(user)        # extra step, not a wall
    elif decision == "review":
        queue_for_review(user)
    return {"status": "ok"}

In Flask it’s the same idea: call assess() with request.remote_addr inside the view (use a sync httpx.Client).

Try it with a US sample IP

Quick check
import asyncio
from ndl import lookup
from risk import assess

async def main():
    data = await lookup("8.8.8.8")
    print(data["location"]["country_code"], data["location"]["city"])  # US Mountain View
    print(data["traits"]["is_hosting"])                                # True
    print(assess(data, "US"))  # ('verify', ['hosting_datacenter'])

asyncio.run(main())

8.8.8.8 is a public DNS resolver in a datacenter, so is_hosting is true and the signup lands in verify. Its country (US, medium confidence) matches the declared one, so there’s no mismatch. For the rest of the logic, unit-test assess() with fixture dicts instead of live IPs:

test_risk.py
def test_tor_goes_to_review():
    data = {"location": {"country_code": "US", "confidence": "high"},
            "traits": {"is_tor": True}}
    assert assess(data, "US")[0] == "review"

def test_residential_us_is_allowed():
    data = {"location": {"country_code": "US", "confidence": "high"},
            "traits": {}}
    assert assess(data, "US") == ("allow", [])

Score existing users in bulk

To backfill risk for users you already have, send their signup IPs to POST /v0/bulk, up to 100 per request. Each IP counts toward the daily quota, and a malformed IP shows up under errors without failing the batch.

Backfill with /v0/bulk
import httpx, os

def enrich(ips: list[str]) -> dict:
    out = {}
    with httpx.Client(headers={"Authorization": f"Bearer {os.environ['NDL_API_KEY']}"}) as c:
        for i in range(0, len(ips), 100):          # max 100 IPs per request
            r = c.post("https://api.networkdatalabs.com/v0/bulk",
                       json={"ips": ips[i:i + 100]}, timeout=10)
            r.raise_for_status()
            out.update(r.json()["results"])        # per-IP failures are in "errors"
    return out

Production checklist

  • Keep the key server-side in NDL_API_KEY.
  • Fail open on timeouts and errors; log lookup_unavailable.
  • Store the reasons with the user, not just the decision.
  • Recheck at sensitive moments (payment, payout, password reset), not only at signup.

Related

Sign up Read the docs