FastAPI (Python)

FastAPI est conçu pour des applications web asynchrones très performantes. Pour maintenir cette vitesse, nous intégrons ProxyTracer à l'aide d'un BaseHTTPMiddleware personnalisé et d'un client HTTP asynchrone (httpx) pour s'assurer que la boucle d'événements n'est jamais bloquée.

L'Intergiciel

Cette implémentation extrait en toute sécurité la véritable adresse IP du client, utilise un pool de connexions asynchrones pour interroger ProxyTracer, et renvoie un strict 403 Forbidden si le drapeau proxy est vrai.


import os
import httpx
from fastapi import FastAPI, Request, status
from fastapi.responses import JSONResponse
from starlette.middleware.base import BaseHTTPMiddleware
 
app = FastAPI()
 
class ProxyTracerMiddleware(BaseHTTPMiddleware):
    def __init__(self, app):
        super().__init__(app)
        self.api_key = os.getenv("PROXYTRACER_API_KEY")
        # Initialize a persistent async client for connection pooling and raw speed
        self.client = httpx.AsyncClient(timeout=0.5) # Strict 500ms timeout
 
    async def dispatch(self, request: Request, call_next):
        # 1. Safely extract the real IP behind load balancers/proxies
        forwarded_for = request.headers.get("x-forwarded-for")
        if forwarded_for:
            ip = forwarded_for.split(",")[0].strip()
        else:
            ip = request.client.host if request.client else None
 
        # Skip local development IPs
        if ip and ip not in ["127.0.0.1", "::1", "testclient"]:
            try:
                # 2. Query ProxyTracer API asynchronously
                response = await self.client.get(
                    f"https://api.proxytracer.com/v1/check/{ip}",
                    headers={"Authorization": f"Bearer {self.api_key}"}
                )
                
                if response.status_code == 200:
                    data = response.json()
                    
                    # 3. Drop the connection immediately if a proxy is detected
                    if data.get("proxy") is True:
                        return JSONResponse(
                            status_code=status.HTTP_403_FORBIDDEN,
                            content={"error": "Access Denied: VPN or Proxy detected."}
                        )
            except httpx.RequestError as e:
                # Fail open: Log the network error and allow the request to proceed
                print(f"ProxyTracer API timeout or failure: {e}")
 
        # 4. Traffic is clean, proceed to the route handler
        return await call_next(request)
 
# Apply the middleware globally
app.add_middleware(ProxyTracerMiddleware)

Note de dépendance: Cette implémentation nécessite la bibliothèque httpx pour les requêtes HTTP asynchrones. Installez-la via pip install httpx.