FastAPI (Python)

FastAPI создан для высокопроизводительных асинхронных веб-приложений. Чтобы сохранить эту скорость, мы интегрируем ProxyTracer, используя пользовательский BaseHTTPMiddleware и асинхронный HTTP-клиент (httpx), чтобы никогда не блокировать event loop.

Middleware

Эта реализация безопасно извлекает реальный IP клиента, использует асинхронный пул соединений для обращения к ProxyTracer и возвращает строгий ответ 403 Forbidden, если флаг proxy возвращает значение true.


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")
        # Инициализировать постоянный асинхронный клиент для пула соединений и максимальной скорости
        self.client = httpx.AsyncClient(timeout=0.5) # Строгий таймаут 500 мс
 
    async def dispatch(self, request: Request, call_next):
        # 1. Безопасно извлечь реальный IP за балансировщиками нагрузки/прокси
        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
 
        # Пропускать IP-адреса локальной разработки
        if ip and ip not in ["127.0.0.1", "::1", "testclient"]:
            try:
                # 2. Выполнить асинхронный запрос к API ProxyTracer
                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. Немедленно сбросить соединение, если обнаружен прокси
                    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: записать сетевую ошибку в лог и разрешить выполнение запроса
                print(f"ProxyTracer API timeout or failure: {e}")
 
        # 4. Трафик чист, перейти к обработчику маршрута
        return await call_next(request)
 
# Применить middleware глобально
app.add_middleware(ProxyTracerMiddleware)

Заметка по зависимостям: Для этой реализации требуется библиотека httpx для выполнения асинхронных HTTP-запросов. Установите ее с помощью pip install httpx.