Category: Uncategorized

When you build a Python web application that serves thousands—or even millions—of requests, protecting your API from abuse is essential. Rate limiting is the go‑to strategy for throttling traffic, preventing denial‑of‑service attacks, and ensuring a fair usage policy for all users. Among the many back‑ends you can choose, Redis stands out for its speed, atomic operations, and built‑in data structures that make implementing robust rate limits both simple and scalable. In this post we’ll explore why Redis is ideal for rate limiting, walk through several popular algorithms, and provide ready‑to‑use code snippets for Flask, Django, and FastAPI.

Why Use Redis for Rate Limiting?

Redis offers a unique blend of features that make it perfect for throttling requests in real time:

• In‑memory speed: Sub‑millisecond latency ensures your rate‑limit checks don’t become a bottleneck.
• Atomic commands: Operations like INCR, EXPIRE, and Lua scripts run atomically, eliminating race conditions in high‑concurrency environments.
• TTL support: Expiration times let you automatically reset counters without extra cleanup logic.
• Distributed safety: Multiple web workers can share the same Redis instance, guaranteeing consistent limits across a cluster.
• Rich data structures: Sorted sets, hashes, and strings enable sophisticated algorithms such as sliding windows or token buckets.

Choosing the Right Rate‑Limiting Algorithm

Not every algorithm fits every use case. Below are the three most common patterns you can implement with Redis, along with their trade‑offs.

1. Fixed Window Counter

The simplest approach: count requests in a fixed time bucket (e.g., per minute). If the count exceeds the allowed threshold, reject the request.

• Pros: Easy to understand, minimal Redis commands.
• Cons: Bursts at bucket boundaries can exceed the limit (known as the “thundering herd” problem).

2. Sliding Window Log

Store timestamps of each request in a sorted set and prune entries older than the window. This yields a true sliding window behavior.

• Pros: Precise enforcement; no burst spikes.
• Cons: Higher memory usage and more Redis operations per request.

3. Token Bucket (or Leaky Bucket)

Maintain a bucket that refills at a steady rate. Each request consumes a token; if the bucket is empty, the request is throttled.

• Pros: Allows short bursts while preserving an average rate.
• Cons: Slightly more complex logic, but still a perfect fit for Redis Lua scripts.

Implementing Fixed Window Counter in Flask

Below is a minimal Flask middleware that limits each IP address to 100 requests per minute using Redis.

from flask import Flask, request, jsonify
import redis
import time

app = Flask(__name__)
r = redis.StrictRedis(host='localhost', port=6379, db=0)

LIMIT = 100          # requests
WINDOW = 60          # seconds

def is_allowed(ip):
    key = f"rl:{ip}"
    # Increment the counter and set expiry if this is the first hit
    current = r.incr(key)
    if current == 1:
        r.expire(key, WINDOW)
    return current <= LIMIT

@app.before_request
def rate_limit():
    ip = request.remote_addr
    if not is_allowed(ip):
        return jsonify({"error": "Rate limit exceeded"}), 429

@app.route('/')
def hello():
    return "Hello, world!"

if __name__ == '__main__':
    app.run()

This example demonstrates the core Redis commands (INCR and EXPIRE) that keep the implementation atomic and fast.

Sliding Window Log with Redis Sorted Sets (Django)

Django developers can leverage a custom middleware that uses a sorted set (ZSET) to store request timestamps. The following snippet shows how to limit a user (identified by user.id) to 20 actions per 10 seconds.

import time
import redis
from django.utils.deprecation import MiddlewareMixin
from django.http import JsonResponse

r = redis.StrictRedis(host='localhost', port=6379, db=0)

class SlidingWindowRateLimitMiddleware(MiddlewareMixin):
    LIMIT = 20
    WINDOW = 10  # seconds

    def process_request(self, request):
        if not request.user.is_authenticated:
            return  # skip unauthenticated users or handle differently

        user_key = f"rl:user:{request.user.id}"
        now = int(time.time() * 1000)  # milliseconds

        # Add current timestamp to the sorted set
        r.zadd(user_key, {now: now})
        # Remove entries older than the window
        r.zremrangebyscore(user_key, 0, now - self.WINDOW * 1000)

        # Get the count of timestamps in the window
        count = r.zcard(user_key)
        # Set a TTL so the key eventually disappears
        r.expire(user_key, self.WINDOW + 1)

        if count > self.LIMIT:
            return JsonResponse(
                {"error": "Too many requests"},
                status=429
            )

Because ZADD, ZREMRangeByScore, and ZCARD are executed sequentially, you should wrap them in a Lua script for absolute atomicity in a production environment. The next section shows exactly that.

Atomic Token Bucket with a Redis Lua Script (FastAPI)

FastAPI’s async nature pairs nicely with Redis’ aioredis client. The Lua script below implements a token bucket that refills at rate tokens per second and allows up to capacity tokens.

import asyncio
from fastapi import FastAPI, Request, HTTPException
import aioredis

app = FastAPI()
redis = None

TOKEN_BUCKET_SCRIPT = """
local key = KEYS[1]
local capacity = tonumber(ARGV[1])
local refill_rate = tonumber(ARGV[2])
local now = tonumber(ARGV[3])

-- Get the current bucket state
local data = redis.call('HMGET', key, 'tokens', 'timestamp')
local tokens = tonumber(data[1]) or capacity
local timestamp = tonumber(data[2]) or now

-- Refill tokens based on elapsed time
local elapsed = now - timestamp
local refill = elapsed * refill_rate
tokens = math.min(capacity, tokens + refill)

if tokens < 1 then
    -- Not enough tokens, reject request
    redis.call('HMSET', key, 'tokens', tokens, 'timestamp', now)
    redis.call('EXPIRE', key, math.ceil(capacity / refill_rate))
    return 0
else
    -- Consume a token and allow request
    tokens = tokens - 1
    redis.call('HMSET', key, 'tokens', tokens, 'timestamp', now)
    redis.call('EXPIRE', key, math.ceil(capacity / refill_rate))
    return 1
end
"""

@app.on_event("startup")
async def startup():
    global redis
    redis = await aioredis.create_redis_pool('redis://localhost')

@app.on_event("shutdown")
async def shutdown():
    redis.close()
    await redis.wait_closed()

async def is_allowed(client_id: str, capacity: int = 10, rate: float = 1.0) -> bool:
    now = int(asyncio.get_event_loop().time())
    result = await redis.eval(
        TOKEN_BUCKET_SCRIPT,
        keys=[f"rl:bucket:{client_id}"],
        args=[capacity, rate, now]
    )
    return bool(result)

@app.middleware("http")
async def rate_limit(request: Request, call_next):
    client_id = request.headers.get("X-API-Key", request.client.host)
    allowed = await is_allowed(client_id, capacity=10, rate=1.0)  # 10 req/min
    if not allowed:
        raise HTTPException(status_code=429, detail="Rate limit exceeded")
    response = await call_next(request)
    return response

@app.get("/")
async def root():
    return {"message": "Welcome to a fast, rate‑limited API!"}

Because the entire logic lives inside a Lua script, the check is atomic—even under heavy concurrency. Adjust capacity and rate to suit your API’s SLA.

Best Practices for Production‑Ready Rate Limiting

1. Centralize configuration: Keep limits (requests per window, burst size, etc.) in a single config file or environment variables so you can change them without redeploying code.
2. Use consistent keys: Prefix keys (e.g., rl:user:, rl:ip:) to avoid collisions with other Redis data.
3. Monitor Redis health: Rate limiting adds write load; set up alerts for latency, memory usage, and eviction events.
4. Graceful degradation: When Redis is unavailable, decide whether to fail open (allow traffic) or fail closed (reject traffic). A common pattern is to let a limited “fallback” bucket operate locally for a short window.
5. Cache static limits: If you have per‑plan limits (free vs. premium), cache the plan data in Redis or in‑process memory to avoid extra DB lookups on every request.
6. Log throttled attempts: Recording rejected requests helps you spot abusive patterns and fine‑tune limits.

Testing Your Rate Limiter

Before shipping, verify that the limiter

Cross‑Origin Resource Sharing (CORS) can feel like a hidden gatekeeper for modern web applications, especially when you’re building APIs with Python. If you’ve ever seen a “No ‘Access‑Control‑Allow‑Origin’ header” error in your browser console, you know how frustrating it can be. This guide walks you through everything you need to know to configure CORS correctly in Python—whether you’re using Flask, Django, FastAPI, or a custom WSGI setup. By the end, you’ll have a solid, SEO‑friendly foundation to keep your APIs secure, performant, and ready for any front‑end client.

What is CORS and Why It Matters in Python Web Apps

Cross‑Origin Resource Sharing (CORS) is a browser security feature that restricts web pages from making requests to a different domain than the one that served the page. While this protects users from malicious sites, it also blocks legitimate API calls unless the server explicitly permits them.

In Python web development, CORS configuration is essential for:

• Enabling single‑page applications (SPAs) built with React, Vue, or Angular to communicate with a Python back‑end.
• Allowing mobile apps, IoT devices, or third‑party services to consume your API.
• Preventing accidental data leaks by restricting which origins can access sensitive endpoints.

Getting CORS right improves both user experience and security posture, making it a non‑negotiable part of any production‑grade Python API.

Core Principles of CORS Configuration

Before diving into code, understand the four main CORS response headers that control access:

• Access-Control-Allow-Origin – Specifies which origins are allowed.
• Access-Control-Allow-Methods – Lists HTTP methods (GET, POST, etc.) the client may use.
• Access-Control-Allow-Headers – Declares which custom headers the client can send.
• Access-Control-Allow-Credentials – Indicates whether cookies or HTTP authentication are permitted.

Origins

The Origin header sent by the browser tells the server where the request originated. You can respond with a single origin (e.g., https://example.com) or a wildcard (*) for open access. Use wildcards sparingly; they bypass the security benefits of CORS.

Methods

Only expose the HTTP methods your API truly needs. Over‑exposing methods can invite unintended side effects. A typical safe list includes:

GET, POST, PUT, PATCH, DELETE, OPTIONS

Headers

If your client sends custom headers like Authorization or X-Requested-With, you must list them in Access-Control-Allow-Headers. Otherwise the browser will block the request before it reaches your Python code.

Credentials

When you need to send cookies or HTTP authentication tokens across origins, set Access-Control-Allow-Credentials: true. Remember: you cannot use the wildcard (*) for the origin when credentials are allowed.

Implementing CORS in Popular Python Frameworks

Flask

Flask’s lightweight nature makes it a perfect candidate for adding CORS via the flask-cors extension. Install it first:

pip install flask-cors

Then apply it globally or to specific blueprints:

from flask import Flask
from flask_cors import CORS

app = Flask(__name__)

# Global CORS with a whitelist
CORS(app, resources={
    r"/api/*": {
        "origins": ["https://frontend.example.com", "https://admin.example.com"],
        "methods": ["GET", "POST", "PUT", "DELETE"],
        "allow_headers": ["Content-Type", "Authorization"],
        "supports_credentials": True
    }
})

@app.route('/api/data')
def data():
    return {"message": "CORS is configured!"}

Django

Django users typically rely on django-cors-headers. Install and add it to your project:

pip install django-cors-headers

Update settings.py:

# settings.py
INSTALLED_APPS = [
    # …
    'corsheaders',
    # …
]

MIDDLEWARE = [
    'corsheaders.middleware.CorsMiddleware',
    'django.middleware.common.CommonMiddleware',
    # …
]

# Allow specific origins only
CORS_ALLOWED_ORIGINS = [
    "https://frontend.example.com",
    "https://admin.example.com",
]

CORS_ALLOW_METHODS = [
    "GET",
    "POST",
    "PUT",
    "PATCH",
    "DELETE",
]

CORS_ALLOW_HEADERS = [
    "content-type",
    "authorization",
]

CORS_ALLOW_CREDENTIALS = True

FastAPI

FastAPI, built on Starlette, includes built‑in CORS middleware. Add it during app creation:

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

origins = [
    "https://frontend.example.com",
    "https://admin.example.com",
]

app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,
    allow_methods=["GET", "POST", "PUT", "DELETE", "OPTIONS"],
    allow_headers=["*"],  # Adjust as needed
    allow_credentials=True,
)

@app.get("/api/items")
async def read_items():
    return {"message": "FastAPI CORS setup complete"}

Advanced CORS Strategies

For larger projects, static whitelists may not be enough. Consider these advanced techniques to keep your Python CORS configuration flexible and secure.

Dynamic Origin Validation

Instead of a hard‑coded list, you can check the incoming Origin against a database or environment variable at runtime. Example for Flask:

from flask import request
from flask_cors import cross_origin

def is_allowed_origin(origin):
    # Example: fetch from DB or config service
    allowed = {"https://frontend.example.com", "https://admin.example.com"}
    return origin in allowed

@app.route("/api/secure")
@cross_origin(origin= lambda: request.headers.get("Origin") if is_allowed_origin(request.headers.get("Origin")) else "")
def secure_endpoint():
    return {"status": "dynamic CORS applied"}

Environment‑Specific Settings

Use separate CORS configurations for development, staging, and production. Store each set in environment variables (e.g., CORS_ORIGINS_DEV) and load them at startup. This prevents accidental exposure of production data during testing.

Fine‑Grained Path Control

Apply different CORS rules per endpoint. For instance, a public read‑only endpoint can use a wildcard origin, while a private /admin route requires strict origin and credential checks. Both Flask and Django support per‑route decorators or view‑level settings.

Cache CORS Preflight Responses

Browsers send an OPTIONS preflight request before the actual request. To reduce latency, set Access-Control-Max-Age (e.g., 86400 seconds) so the browser caches the permission for a day.

Testing and Debugging CORS

Even with perfect code, misconfigurations happen. Follow this checklist to verify your CORS setup:

1. Use browser dev tools. Look for the Access-Control-Allow-* headers in the Network tab.
2. Test with curl. Example: curl -i -H "Origin: https://frontend.example.com" -X GET http://api.example.com/api/data should return the appropriate CORS headers.
3. Check preflight handling. Ensure the OPTIONS route returns 200 and includes all required headers.
4. Validate credentials. When Access-Control-Allow-Credentials: true is set, confirm the Access-Control-Allow-Origin is not a wildcard.
5. Automate with integration tests. Tools like pytest combined with requests can assert header presence programmatically.

If you encounter “CORS policy: No ‘Access‑Control‑Allow‑Origin’ header is present” errors, double‑check that the origin string exactly matches the one in your whitelist (including protocol and port).

Cross‑Site Request Forgery (CSRF) remains one of the most common yet misunderstood security threats facing modern web applications. Even though the attack vector is simple—tricking a logged‑in user into sending an unwanted request—its impact can be devastating, from unauthorized data changes to full account takeover. Fortunately, Python’s rich ecosystem provides robust, battle‑tested tools to defend against CSRF. In this guide we’ll explore the theory behind CSRF, walk through best‑practice implementation in popular Python frameworks, and share practical tips to keep your web apps safe without sacrificing developer productivity.

Understanding CSRF: The Attack in Plain English

Before you can protect your application, you need to understand how CSRF works. The attacker exploits the trust a browser has in a user’s session cookies. By embedding a malicious link, image tag, or form on a third‑party site, the attacker forces the victim’s browser to send an authenticated request to your server—often without the user realizing it.

• Victim is logged in: The user’s session cookie is automatically attached to every request to your domain.
• Malicious site triggers a request: A hidden form or JavaScript call submits data to your endpoint.
• Server processes the request: Because the cookie is valid, the server assumes the request is legitimate.

Since the attacker cannot read the response (due to the Same‑Origin Policy), they rely on side‑effects—such as changing a password, posting a comment, or initiating a financial transaction—to achieve their goal.

Why CSRF Tokens Are the Gold Standard

CSRF tokens are unique, unpredictable values generated by the server and tied to a user’s session. When a form is rendered, the token is embedded as a hidden field (or as a request header for AJAX calls). The server then validates the token on every state‑changing request. If the token is missing, expired, or mismatched, the request is rejected.

Key properties of a good CSRF token:

• Unpredictable: Must be generated using a cryptographically secure random source.
• Per‑session or per‑request: Either tied to the user session or regenerated for each form.
• Bound to the user: The token should be stored server‑side (e.g., in the session) so it cannot be forged.

Implementing CSRF Protection in Django

Django ships with built‑in CSRF protection that works out of the box for most use cases. The framework automatically adds a hidden csrfmiddlewaretoken field to every <form> rendered with the {% csrf_token %} template tag.

Step‑by‑step setup

1. Enable the middleware: Ensure 'django.middleware.csrf.CsrfViewMiddleware' is listed in MIDDLEWARE in settings.py.
2. Add the token to templates:

   {% raw %}
   
       {% csrf_token %}
       
   
   {% endraw %}

3. Validate in views: For function‑based views, use @csrf_exempt only when you have a good reason; otherwise, Django validates automatically.
4. Handle AJAX requests: Include the token in the X‑CSRFToken header.

   $.ajax({
       type: "POST",
       url: "/api/update/",
       data: JSON.stringify(payload),
       contentType: "application/json",
       beforeSend: function(xhr) {
           xhr.setRequestHeader("X-CSRFToken", "{{ csrf_token }}");
       }
   });

When a request fails validation, Django returns a 403 Forbidden response with a helpful error page. You can customize this behavior by overriding CsrfViewMiddleware.process_view or providing a custom 403.html template.

Common pitfalls

• Disabling the middleware globally (@csrf_exempt) defeats the protection.
• Using csrf_exempt on API endpoints that accept cookies; prefer token‑based authentication (e.g., JWT) for stateless APIs.
• Forcing the token into GET requests—CSRF tokens should only protect state‑changing (POST, PUT, DELETE, PATCH) methods.

CSRF Defense in Flask

Flask does not include CSRF protection by default, but the Flask‑WTF extension makes it straightforward. Flask‑WTF integrates wtforms with CSRF handling and stores the token in the session.

Quick start guide

1. Install the extension: pip install Flask-WTF
2. Configure the secret key:

   app = Flask(__name__)
   app.config['SECRET_KEY'] = 'a-very-secret-key'

3. Create a form class:

   from flask_wtf import FlaskForm
   from wtforms import StringField, SubmitField
   
   class CommentForm(FlaskForm):
       comment = StringField('Comment')
       submit = SubmitField('Post')

4. Render the form in a template:

   {% raw %}
   
       {{ form.hidden_tag() }}
       {{ form.comment.label }} {{ form.comment() }}
       {{ form.submit() }}
   
   {% endraw %}

5. Validate in the view:

   @app.route('/comment', methods=['GET', 'POST'])
   def comment():
       form = CommentForm()
       if form.validate_on_submit():
           # process comment
           flash('Comment posted!')
           return redirect(url_for('index'))
       return render_template('comment.html', form=form)

Flask‑WTF automatically checks the token on every POST request. If the token is missing or invalid, a 400 Bad Request error is raised.

CSRF for API‑only endpoints

When building a JSON‑based API that does not use server‑side sessions, you should consider one of the following strategies:

• Stateless tokens: Use a double‑submit cookie pattern where the client reads a cookie named csrf_token and sends it back in a custom header.
• Header‑only protection: Require a custom header (e.g., X-CSRF-Token) that the client obtains via a prior authenticated request.
• Switch to token‑based auth: JWT or OAuth2 eliminates the need for CSRF protection because the token is not automatically sent by the browser.

CSRF Mitigation Techniques Beyond Tokens

While tokens are the most widely adopted solution, a layered defense strategy can further reduce risk.

SameSite Cookies

Setting the SameSite attribute on session cookies tells browsers not to include the cookie on cross‑origin requests. Modern browsers support three values:

• Strict: Cookie is sent only for same‑site navigation.
• Lax: Cookie is sent for top‑level GET navigation but not for POST, PUT, etc. (good balance for most sites).
• None: Cookie is sent on all requests; must be paired with Secure.

In Django, you can enable it with:

SESSION_COOKIE_SAMESITE = 'Lax'

In Flask (using Flask-Session or directly on Response.set_cookie):

response.set_cookie('session', value, samesite='Lax', secure=True)

Content Security Policy (CSP)

A strict CSP can prevent malicious scripts from loading on your pages, reducing the chance that an attacker can craft a forged request. Example header:

Content-Security-Policy: default-src 'self'; script-src 'self' https://cdnjs.cloudflare.com;

Enforce HTTP Referrer Checks (with caution)

Some legacy systems also verify the Referer header to ensure the request originates from an allowed domain. This method is not recommended as a primary defense because the header can be stripped or spoofed, but it can serve as a supplemental check.

Testing Your CSRF Implementation

Automated tests should verify that legitimate requests succeed while forged requests are blocked. Below is a simple pytest example for a Flask app using Flask‑WTF.

import pytest

from app import app

@pytest.fixture

def client():

    app.config['WTF_CSRF_ENABLED'] = True

    with app.test_client() as client:

        yield client
def test_valid_csrf(client):

    # Get the form to retrieve the token

    resp = client.get('/comment')

    token = re.search(r'name="csrf_token" value="([^"]+)"', resp.get_data(as_text=True)).group(1)
    # Submit with the token

    resp = client.post('/comment', data={'comment': 'test', 'csrf_token': token})

    assert resp.status_code == 302  # redirect on success
def test_missing_csrf(client):

    resp = client.post('/comment', data={'comment': 'test'})