FastAPI (Python)

FastAPI został stworzony z myślą o wysokowydajnych, asynchronicznych aplikacjach webowych. Aby utrzymać tę szybkość, integrujemy ProxyTracer przy użyciu niestandardowego BaseHTTPMiddleware i asynchronicznego klienta HTTP (httpx), co gwarantuje, że pętla zdarzeń (event loop) nigdy nie zostanie zablokowana.

Middleware

Ta implementacja bezpiecznie pobiera prawdziwy adres IP klienta, korzysta z asynchronicznej puli połączeń do odpytywania ProxyTracer i zwraca restrykcyjną odpowiedź 403 Forbidden, jeśli flaga proxy ma wartość 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")
        # Zainicjalizuj trwały klient asynchroniczny z pulą połączeń dla maksymalnej wydajności
        self.client = httpx.AsyncClient(timeout=0.5) # Ścisły limit czasu 500 ms
 
    async def dispatch(self, request: Request, call_next):
        # 1. Bezpiecznie pobierz prawdziwy adres IP zza load balancerów / proxy
        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
 
        # Pomiń lokalne adresy IP środowiska deweloperskiego
        if ip and ip not in ["127.0.0.1", "::1", "testclient"]:
            try:
                # 2. Odpytaj asynchronicznie 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. Natychmiast odrzuć połączenie, jeśli wykryto proxy
                    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: Zaloguj błąd sieci i zezwól na kontynuowanie żądania
                print(f"ProxyTracer API timeout or failure: {e}")
 
        # 4. Ruch jest czysty, przekaż żądanie do handlera trasy
        return await call_next(request)
 
# Zastosuj middleware globalnie
app.add_middleware(ProxyTracerMiddleware)

Informacja o zależnościach: Ta implementacja wymaga biblioteki httpx do asynchronicznych żądań HTTP. Zainstaluj ją za pomocą polecenia pip install httpx.