FastAPI（Python）

FastAPI は ASGI ミドルウェアを使用してリクエストとレスポンスを非同期に処理します。BaseHTTPMiddleware 経由で ProxyTracer を統合することで、1ms 未満の極小オーバーヘッドで IP レピュテーションを評価し、非同期 API エンドポイントを自動化攻撃やプロキシの不正利用から保護できます。

ミドルウェアの実装

このミドルウェアはクライアント IP を抽出し、httpx を使用して ProxyTracer への非同期 HTTP リクエストを実行して、プロキシが検出された場合にステータスコード 403 の JSONResponse を即座に返します。


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) # 厳格な500msのタイムアウト
 
    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. ProxyTracer APIに非同期でクエリを実行
                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:
                # フェイルオープン: ネットワークエラーをログに記録し、リクエストの進行を許可
                print(f"ProxyTracer API timeout or failure: {e}")
 
        # 4. トラフィックは正常です。ルートハンドラーへ処理を渡します
        return await call_next(request)
 
# ミドルウェアをグローバルに適用
app.add_middleware(ProxyTracerMiddleware)

インストール: 高性能な非同期 HTTP リクエストをサポートするために、Python 環境に httpx がインストールされていることを確認してください（pip install httpx）。