You need to enable JavaScript to run this app.
最新活动
大模型
产品
解决方案
定价
生态与合作
支持与服务
开发者
了解我们

FastAPI中IP白名单过滤的最佳实践问询

FastAPI中IP白名单过滤的最佳实践问询

我完全懂你的感受!从Django转FastAPI,一开始确实会怀念Django那些开箱即用的安全特性,比如ALLOWED_HOSTS。你说得没错,FastAPI自带的CORSMiddleware只管浏览器的CORS头,TrustedHostMiddleware也只是校验请求的Host头,和来源IP完全不沾边,确实没法直接满足IP白名单的需求。

很遗憾的是,FastAPI官方目前并没有提供专门做IP白名单过滤的现成中间件,不过自定义一个其实非常简单,而且完全能适配你用.env配置ALLOWED_IPS的需求,下面给你一步步说怎么做:

1. 基础版自定义IP白名单中间件

首先我们可以用Pydantic的BaseSettings来加载.env里的配置,这样比直接读环境变量更规范:

from fastapi import FastAPI, Request
from fastapi.middleware.base import BaseHTTPMiddleware
from pydantic import BaseSettings
from starlette.responses import JSONResponse

class Settings(BaseSettings):
    allowed_ips: list = ["127.0.0.1", "192.168.1.100"]  # 会自动从.env的ALLOWED_IPS读取,格式用逗号分隔即可,比如ALLOWED_IPS=127.0.0.1,192.168.1.100

    class Config:
        env_file = ".env"

settings = Settings()

class IPWhitelistMiddleware(BaseHTTPMiddleware):
    def __init__(self, app: FastAPI, allowed_ips: list):
        super().__init__(app)
        self.allowed_ips = set(allowed_ips)  # 转成集合提升查询效率

    async def dispatch(self, request: Request, call_next):
        # 获取客户端IP
        client_ip = request.client.host
        # 如果是Docker+反向代理(比如Nginx)场景,需要从X-Forwarded-For头取真实客户端IP
        # client_ip = request.headers.get("X-Forwarded-For", client_ip).split(",")[0].strip()
        
        if client_ip not in self.allowed_ips:
            return JSONResponse(
                status_code=403,
                content={"detail": "Forbidden: 你的IP不在允许访问的列表中"}
            )
        response = await call_next(request)
        return response

app = FastAPI()

# 注册中间件
app.add_middleware(IPWhitelistMiddleware, allowed_ips=settings.allowed_ips)

# 测试接口
@app.get("/")
async def root():
    return {"message": "来自白名单IP的访问成功!"}

2. 关键注意事项(尤其是Docker环境)

  • 真实IP获取:如果你的FastAPI跑在Docker容器里,前面还有Nginx这类反向代理,直接用request.client.host拿到的会是代理容器的IP,而非真实客户端IP。这时候需要从X-Forwarded-For请求头提取,上面代码里我注释了对应的处理逻辑,你可以根据自己的代理配置启用它。
  • 网段支持:上面的基础版只支持单个IP,如果你需要支持CIDR网段(比如192.168.1.0/24),可以用Python内置的ipaddress模块改造校验逻辑,下面给你补充进阶版:

3. 进阶:支持CIDR网段的白名单版本

import ipaddress
from fastapi import FastAPI, Request
from fastapi.middleware.base import BaseHTTPMiddleware
from pydantic import BaseSettings
from starlette.responses import JSONResponse

class Settings(BaseSettings):
    allowed_ips: list = ["127.0.0.1", "192.168.1.0/24"]

    class Config:
        env_file = ".env"

settings = Settings()

class IPWhitelistMiddleware(BaseHTTPMiddleware):
    def __init__(self, app: FastAPI, allowed_ips: list):
        super().__init__(app)
        # 提前解析所有允许的IP/网段,提升请求校验性能
        self.allowed_networks = set()
        for ip_str in allowed_ips:
            try:
                self.allowed_networks.add(ipaddress.ip_network(ip_str, strict=False))
            except ValueError:
                # 处理单个IP的情况,自动转成/32网段
                self.allowed_networks.add(ipaddress.ip_network(f"{ip_str}/32", strict=False))

    async def dispatch(self, request: Request, call_next):
        client_ip = request.client.host
        # 反向代理场景下启用下面的逻辑
        # client_ip = request.headers.get("X-Forwarded-For", client_ip).split(",")[0].strip()
        
        try:
            client_ip_obj = ipaddress.ip_address(client_ip)
        except ValueError:
            return JSONResponse(status_code=400, content={"detail": "无效的IP地址格式"})
        
        # 检查客户端IP是否在任何一个允许的网段内
        is_allowed = any(client_ip_obj in network for network in self.allowed_networks)
        if not is_allowed:
            return JSONResponse(
                status_code=403,
                content={"detail": "Forbidden: 你的IP不在允许访问的列表中"}
            )
        response = await call_next(request)
        return response

app = FastAPI()
app.add_middleware(IPWhitelistMiddleware, allowed_ips=settings.allowed_ips)

总的来说,虽然没有现成的官方中间件,但自定义的方式非常灵活,代码量也不大,完全能适配你的需求。如果后续需要动态更新白名单,还可以在中间件里加逻辑从数据库或者配置中心拉取最新的白名单列表~

火山引擎 最新活动