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.
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 / 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 nodeis_vpn: known VPN networks, including major commercial VPN providersis_relay: privacy relay egressis_hosting: hosting/datacenter detectionis_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
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:
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.
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
- VPN Detection API: known VPN network, Tor exit, privacy relay and hosting flags
- IP Fraud Detection API: signals for signup, login and payment risk
- Get a visitor’s IP and country in Express
- API Docs