Post

Secure Your FastAPI App with TrustedHostMiddleware - Allow Only Specific Hosts or IPs

Learn how to use FastAPI's TrustedHostMiddleware to allow only specific hosts or IPs to access your API securely and prevent host header attacks.

Secure Your FastAPI App with TrustedHostMiddleware - Allow Only Specific Hosts or IPs

FastAPI’s TrustedHostMiddleware is a security feature used to protect your application from Host header attacks by allowing only specified hostnames or IP addresses to access your API. When a client makes a request to your FastAPI app, it includes a Host header. If your app is running behind a proxy or on a public-facing server, malicious users could send requests with fake or unexpected Host headers to trick your app or exploit vulnerabilities. The TrustedHostMiddleware checks the Host header in incoming requests and only allows requests from hosts you trust. If the Host doesn’t match the allowed list, the middleware blocks the request and returns a 400 Bad Request. Let’s see how to implement both examples.

Limit API access only to specific hosts

The Host header is not the client’s IP address, but the hostname, such as localhost, example.com, *.example.com.

1
2
3
4
5
6
7
8
9
10
11
12
from fastapi import FastAPI, Request
from fastapi.middleware.trustedhost import TrustedHostMiddleware

app = FastAPI()
app.add_middleware(
    TrustedHostMiddleware, allowed_hosts=["localhost", "example.de"]
)   

@app.get("/")
async def main(request: Request):
    print(request.headers['host'])
    return {"message": "Hello from Engineering Minds"}

Limit API access only to specific IP addresses

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
from fastapi import FastAPI, Request, status
from fastapi.responses import JSONResponse


app = FastAPI()
IPS_ALLOW_LIST = ["127.0.0.1", "3.23.10.22"]


@app.middleware('http')
async def trusted_ip_middleware(request: Request, call_next):
    ip = str(request.client.host)
    print(ip)
    
    if ip not in IPS_ALLOW_LIST:
        return JSONResponse(content="403 - Forbidden: Access is denied", status_code=403)

    return await call_next(request)
    

@app.get("/")
async def main():
    return {"message": "Hello from Engineering Minds"}

In some scenarios, we need to send a specific header value to the API endpoint (e.g. from API Marketplace platforms). In that case, we can implement Specific Header Middleware that will check if a request has the proper header and value set.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
class SpecificHeaderMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        required_header = "api_required_header"
        expected_value = "api_header_expected_value"

        if required_header not in request.headers or \
           request.headers[required_header] != expected_value:
            return JSONResponse(
                status_code=status.HTTP_401_UNAUTHORIZED,
                content={"detail": "Authentication required or invalid token"},
            )
        response = await call_next(request)
        return response

app.add_middleware(SpecificHeaderMiddleware)
This post is licensed under CC BY 4.0 by the author.