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

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 / 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 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 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

Signal What it means Suggested action
is_tor Current Tor exit node Manual review
is_vpn Known VPN networks, including major commercial VPN providers Verify email
is_hosting Hosting/datacenter IP: often scripts and bots Verify email (or CAPTCHA)
is_relay Privacy relay: usually a normal, privacy-minded user Allow, log it
is_mobile Mobile carrier networks: many users share IPs Allow; don’t rate-limit by IP alone
Country mismatch + high/medium confidence IP country differs from the declared one Verify; review if combined with another signal
Two or more signals Stacked risk Manual review

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

# 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 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

Want to try it? Network Data Labs has a free plan with 1,000 requests/day and the full response schema on every plan.