Detectar registros sospechosos por IP en Python: Tor, hosting y VPN conocidas

Una receta para FastAPI y Flask que convierte los traits de la IP en permitir, verificar o revisar, sin dejar afuera a usuarios reales.

Registrarse Leer la documentación

Gratis: 1.000 req/día · HTTPS · uso comercial OK

Por qué bloquear de una a quien usa VPN te sale caro

Mucha gente legítima se registra desde una VPN, un relay de privacidad o el celular con datos móviles. Si los bloqueás de entrada perdés clientes reales, y el que quiere abusar simplemente cambia de IP. Funciona mejor tomar la IP como una señal de riesgo: sumar fricción (verificar el email, un CAPTCHA, revisión manual) cuando se juntan señales, y guardar los motivos en el usuario para ajustar después.

Obtené la IP real en FastAPI o Flask

Detrás de Nginx o un balanceador, la dirección del socket es la de tu proxy. Decile al framework en qué proxy confiar y nunca parsees X-Forwarded-For a mano: el cliente puede poner cualquier IP ahí.

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

Consultá la IP

Una llamada a GET /v0/ip/{ip} con tu key en Authorization: Bearer te devuelve ubicación, red y traits. Este cliente async usa httpx, timeout de un segundo y una caché chica por IP, y devuelve None si algo falla, para que el registro nunca se rompa:

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

Los traits que importan en un registro:

  • is_tor: la IP es un exit node de Tor actual
  • is_vpn: redes VPN conocidas, incluidos los principales proveedores comerciales de VPN
  • is_relay: salida de un relay de privacidad
  • is_hosting: detección de hosting/datacenter
  • is_mobile: redes de operadores móviles (estimación)

Una tabla de política: permitir, verificar o revisar

SeñalQué significaAcción sugerida
is_torExit node de Tor actualRevisión manual
is_vpnRedes VPN conocidas, incluidos los principales proveedores comerciales de VPNVerificar email
is_hostingIP de hosting/datacenter: muchas veces scripts y botsVerificar email (o CAPTCHA)
is_relayRelay de privacidad: suele ser un usuario común que cuida su privacidadPermitir y registrar
is_mobileRedes de operadores móviles: muchos usuarios comparten IPPermitir; no limitar solo por IP
País distinto + confianza high/mediumEl país de la IP no coincide con el declaradoVerificar; revisión si se suma otra señal
Dos o más señalesRiesgo acumuladoRevisión manual

La misma tabla en código, con un chequeo de país que solo cuenta cuando confidence es high o 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

Conectalo al endpoint de registro

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"}

En Flask es la misma idea: llamás a assess() con request.remote_addr dentro de la vista (con un httpx.Client sincrónico).

Probalo con una IP de ejemplo de EE. UU.

Prueba rápida 
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 es un resolver DNS público en un datacenter, así que is_hosting da true y el registro cae en verify. El país (US, confianza medium) coincide con el declarado, así que no suma mismatch. Para el resto de la lógica, testeá assess() con diccionarios de ejemplo en vez de IPs reales:

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", [])

Puntuá a los usuarios que ya tenés, en bulk

Para calcular el riesgo de tu base actual, mandá las IPs de registro a POST /v0/bulk, hasta 100 por pedido. Cada IP cuenta para la cuota diaria, y si una IP está mal formada aparece en errors sin hacer fallar el lote.

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

Checklist para producción

  • La key queda en el servidor, en NDL_API_KEY.
  • Fail-open ante timeouts y errores; registrá lookup_unavailable.
  • Guardá los motivos en el usuario, no solo la decisión.
  • Volvé a chequear en momentos sensibles (pago, retiro, cambio de contraseña), no solo al registrarse.

Relacionado

Registrarse Leer la documentación