Keeping your API secure is essential, especially in today’s digital age where threats are omnipresent. The key to ensuring the integrity and performance of your web applications lies in effective strategies such as rate limiting and IP blocking. This helps to prevent misuse and protects your resources from excessive traffic. Let’s dive into how you can implement these defenses using FastAPI, a modern, high-performance framework for building APIs.

First off, why should you bother with rate limiting? It’s all about keeping your system safe and sound. Rate limiting puts a cap on the number of requests an API can handle within a set timeframe. This is super important for defending against Distributed Denial of Service (DDoS) attacks, where attackers try to flood your server with a load of requests to knock it offline. It also makes life harder for those trying brute force attacks by limiting their number of attempts. Plus, it curbs those pesky automated scripts that scrape your data and drain your resources.

Now, FastAPI doesn’t come with built-in rate limiting, but don’t sweat it, you can easily integrate it using external libraries. One of the popular ones is slowapi. This library is straightforward to use and gels well with FastAPI.

First, you’ll need to install slowapi. You can do this using pip:

pip install slowapi

Here’s a quick example of how to apply rate limiting to an endpoint using slowapi:

from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.middleware import SlowAPIMiddleware
from fastapi import FastAPI, Request

app = FastAPI()

limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter
app.add_middleware(SlowAPIMiddleware)

@app.get("/")
@limiter.limit("15/minute")
async def root(request: Request):
    return {"message": "Hello from IP-limited endpoint!"}

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

In this example, your / endpoint is capped at 15 requests per minute per IP address. Simple, effective, and neat!

You can mix and match rate limiting techniques to fine-tune your control over API usage. For example, use user-based limits for logged-in users and IP-based limits for public endpoints:

@app.get("/public")
@limiter.limit("20/minute")
async def public_endpoint(request: Request):
    return {"message": "Public endpoint with IP-based limiting"}

@app.get("/private")
@limiter.limit("5/minute", key_func=lambda request: get_current_user().username)
async def private_endpoint(request: Request, user: User = Depends(get_current_user)):
    return {"message": f"Private endpoint for {user.username} with user-based limiting"}

This way, both public and private endpoints get proper rate limits, bolstering overall security and performance.

Another cool way to implement rate limiting is using the token bucket algorithm. Essentially, this generates tokens at a fixed rate, which are then used up with each API call. Once tokens run out, any more requests are denied until new tokens are available.

Here’s a basic example using the token bucket algorithm in FastAPI:

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

class TokenBucket:
    def __init__(self, rate, capacity):
        self.rate = rate
        self.capacity = capacity
        self.tokens = capacity
        self.last_update = 0

    def consume(self, amount=1):
        now = time.time()
        elapsed = now - self.last_update
        self.last_update = now
        self.tokens = min(self.capacity, self.tokens + elapsed * self.rate)
        if self.tokens < amount:
            return False
        self.tokens -= amount
        return True

app = FastAPI()

bucket = TokenBucket(rate=1, capacity=10)  # 1 token per second, max 10 tokens

@app.get("/")
async def root(request: Request):
    if not bucket.consume():
        return JSONResponse(content={"error": "Rate limit exceeded"}, status_code=429)
    return {"message": "Hello from rate-limited endpoint!"}

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

In this example, the / endpoint is rate-limited to 1 request per second with a maximum of 10 tokens in the bucket.

Apart from rate limiting, blocking specific IPs identified as malicious is another layer of defense you might want to add. This involves maintaining a list of blocked IPs and checking each incoming request against this list.

Here’s an example of how you can do IP blocking in FastAPI:

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

app = FastAPI()

blocked_ips = set(["192.168.1.100", "192.168.1.101"])

@app.middleware("http")
async def block_ips(request: Request, call_next):
    if request.client.host in blocked_ips:
        return JSONResponse(content={"error": "IP blocked"}, status_code=403)
    return await call_next(request)

@app.get("/")
async def root(request: Request):
    return {"message": "Hello from IP-checked endpoint!"}

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

Any requests from the IPs listed will be blocked with a 403 Forbidden response. Easy peasy!

For more advanced rate limiting, especially in scenarios involving multiple workers or distributed systems, using an external storage system like Redis is advisable. This ensures that the rate-limiting state is consistent across all workers.

Here’s a quick example using fastapi-limiter which integrates with Redis:

import redis.asyncio as redis
from contextlib import asynccontextmanager
from fastapi import Depends, FastAPI
from fastapi_limiter import FastAPILimiter
from fastapi_limiter.depends import RateLimiter

app = FastAPI()

@asynccontextmanager
async def lifespan(_: FastAPI):
    redis_connection = redis.from_url("redis://localhost:6379", encoding="utf8")
    await FastAPILimiter.init(redis_connection)
    yield
    await FastAPILimiter.close()

app.lifespan_context = lifespan

@app.get("/", dependencies=[Depends(RateLimiter(times=2, seconds=5))])
async def index():
    return {"msg": "Hello World"}

if __name__ == "__main__":
    import uvicorn
    uvicorn.run("main:app", debug=True)

In this example, the / endpoint is rate-limited to 2 requests per 5 seconds, managed by Redis to ensure consistency across different workers.

To wrap it up, rate limiting and IP blocking are crucial for keeping your FastAPI application secure and high-performing. Whether you prefer using libraries like slowapi or fastapi-limiter, or fancy a custom solution like the token bucket algorithm, it’s essential to pick the right approach that fits your application’s needs. And remember, scalability and consistency of your method are just as important, especially in a distributed setup.