前言

近期无意中查看日志发现一直有个ip各种注入和大批量访问刷接口，故加上ip限制

代码

中间件Middleware代码：

from fastapi import FastAPI
from starlette.middleware.base import BaseHTTPMiddleware, RequestResponseEndpoint
from starlette.requests import Request
from starlette.responses import JSONResponse


class IPBlockerMiddleware(BaseHTTPMiddleware):
    def __init__(self, app: FastAPI, blocked_ips):
        super().__init__(app)
        self.blocked_ips = blocked_ips

    async def dispatch(self, request: Request, call_next: RequestResponseEndpoint):
        if str(request.client.host) in self.blocked_ips:
            return JSONResponse(status_code=403, content={"detail": "IP blocked"})
        response = await call_next(request)
        return response

app启动文件中注入即可：

1
2
3

# 定义被禁止访问的IP列表
blocked_ips = ["192.168.1.5"]  
application.add_middleware(IPBlockerMiddleware, blocked_ips=blocked_ips)

另一种：

blocked_ips = ["192.168.1.5"]  
@application.middleware("http")
async def ip_blocker(request: Request, call_next):
    ips = request.headers.get("X-Forwarded-For").replace(" ", "")
    if "," in ips:
        for ip_item in ips.split(","):
            if ip_item in blocked_ips:
                raise HTTPException(status_code=403, detail="IP blocked")
    else:
        if ips in blocked_ips:
            raise HTTPException(status_code=403, detail="IP blocked")
    response = await call_next(request)
    return response

异常处理

刚开始通过raise HTTPException抛出异常但一直报fastapi.exceptions.HTTPException
,ERROR:uvicorn.error:Exception in ASGI application这个错误，改成JSONResponse或者Starlette的Response即可。

GPT4的解释如下：
FastAPI的HTTPException是特别设计的，无法在非路由中的函数（像中间件或依赖项）中正确工作，因为它需要依赖捕获并处理这个异常。

这是因为HTTPException实际上并不是一个真正的异常，而是用于生成合适的HTTP响应的。当在路由中抛出HTTPException时，FastAPI会捕获它，并根据内容生成合适的响应。

在中间件中，您可能需要返回一个Response对象，而不是使用HTTPException。—
title: fastapi拉黑恶意访问ip
date: 2023-07-21 00:09:00
updated: 2023-07-21 00:18:27
categories: fastapi
tags:

id: fastapi-ip
cover:

前言

近期无意中查看日志发现一直有个ip各种注入和大批量访问刷接口，故加上ip限制

代码

中间件Middleware代码：

from fastapi import FastAPI
from starlette.middleware.base import BaseHTTPMiddleware, RequestResponseEndpoint
from starlette.requests import Request
from starlette.responses import JSONResponse


class IPBlockerMiddleware(BaseHTTPMiddleware):
    def __init__(self, app: FastAPI, blocked_ips):
        super().__init__(app)
        self.blocked_ips = blocked_ips

    async def dispatch(self, request: Request, call_next: RequestResponseEndpoint):
        if str(request.client.host) in self.blocked_ips:
            return JSONResponse(status_code=403, content={"detail": "IP blocked"})
        response = await call_next(request)
        return response

app启动文件中注入即可：

1
2
3

# 定义被禁止访问的IP列表
blocked_ips = ["192.168.1.5"]  
application.add_middleware(IPBlockerMiddleware, blocked_ips=blocked_ips)

另一种：

blocked_ips = ["192.168.1.5"]  
@application.middleware("http")
async def ip_blocker(request: Request, call_next):
    ips = request.headers.get("X-Forwarded-For").replace(" ", "")
    if "," in ips:
        for ip_item in ips.split(","):
            if ip_item in blocked_ips:
                raise HTTPException(status_code=403, detail="IP blocked")
    else:
        if ips in blocked_ips:
            raise HTTPException(status_code=403, detail="IP blocked")
    response = await call_next(request)
    return response