LockWatch

Drop-in API security middleware for FastAPI and Flask: sliding-window rate limiting (Redis sorted set), JWT rotation with refresh-token blacklist, IP burst anomaly detection with webhook alerts, and Postgres audit logging — all wired into a single add_middleware call.

---

Architecture

                    ┌─────────────────────────────────────────────┐
  Incoming          │            LockWatchMiddleware               │
  Request  ────────►│                                             │
                    │  ┌──────────────┐   ┌───────────────────┐  │
                    │  │  RateLimiter │   │  AnomalyDetector  │  │
                    │  │              │   │                   │  │
                    │  │ Redis sorted │   │ burst window +    │  │
                    │  │ set sliding  │   │ webhook alert     │  │
                    │  │ window       │   │ (fire-and-forget) │  │
                    │  └──────┬───────┘   └────────┬──────────┘  │
                    │         │ 429 if over limit   │             │
                    │         ▼                     ▼             │
                    │  ┌─────────────────────────────────────┐    │
                    │  │          Application Handler        │    │
                    │  └─────────────────┬───────────────────┘    │
                    │                    │                         │
                    │                    ▼                         │
                    │  ┌─────────────────────────────────────┐    │
                    │  │  AuditLogger (background task)      │    │
                    │  │  → Postgres lockwatch_audit_log     │    │
                    │  └─────────────────────────────────────┘    │
                    └─────────────────────────────────────────────┘
                                         │
                               Response ◄┘

Every request flows through the rate limiter first (hard block at 429), then the anomaly detector (non-blocking webhook alert), then your application, then an async audit log write that is never on the response critical path.

---

Why I Built This

Bolting rate limiters, audit logs, and JWT rotation into every FastAPI project is tedious boilerplate — the same 50 lines of Redis plumbing, every time. LockWatch makes it a single add_middleware call so the security layer doesn't crowd the application logic.

---

Install

# FastAPI / Starlette
pip install lockwatch[fastapi]

# Flask / WSGI
pip install lockwatch[flask]

# Postgres audit log
pip install lockwatch[audit]

# Everything
pip install lockwatch[all]

---

Quickstart — FastAPI

from fastapi import FastAPI
from lockwatch import LockWatchMiddleware

app = FastAPI()

app.add_middleware(
    LockWatchMiddleware,
    redis_url="redis://localhost:6379",
    rate_limit_requests=100,          # allow 100 requests …
    rate_limit_window_seconds=60,     # … per 60-second sliding window
    burst_threshold=50,               # anomaly alert if >50 req in 10s
    burst_window_seconds=10,
    webhook_urls=["https://hooks.example.com/lockwatch-alert"],
    audit_database_url="postgresql+asyncpg://user:pass@host/db",  # optional
)

@app.get("/items")
async def list_items():
    return {"items": []}

When a client exceeds the limit, LockWatch automatically returns:

HTTP 429 Too Many Requests
Retry-After: 42
X-RateLimit-Limit: 100
X-RateLimit-Remaining: 0
X-RateLimit-Reset: 1717516800

{"error": "rate_limit_exceeded", "retry_after": 42}

---

Quickstart — Flask

from flask import Flask
from lockwatch.middleware import LockWatchFlaskMiddleware

app = Flask(__name__)
app.wsgi_app = LockWatchFlaskMiddleware(
    app.wsgi_app,
    redis_url="redis://localhost:6379",
    rate_limit_requests=200,
    rate_limit_window_seconds=60,
)

---

Flask — Anomaly Detection + JWT Verification

from flask import Flask, g
from lockwatch import JWTRotationManager, JWTConfig
from lockwatch.middleware import flask_jwt_required, LockWatchFlaskMiddleware

app = Flask(__name__)

# Rate limiting + anomaly detection wired at the WSGI layer
app.wsgi_app = LockWatchFlaskMiddleware(
    app.wsgi_app,
    redis_url="redis://localhost:6379",
    rate_limit_requests=200,
    rate_limit_window_seconds=60,
    burst_threshold=50,           # enables anomaly detection; 0 = disabled
    burst_window_seconds=10,
    webhook_urls=["https://hooks.example.com/alert"],
)

# JWT verification per route via decorator
jwt_mgr = JWTRotationManager(redis_client, JWTConfig(rsa_private_key_pem=pem))

@app.route("/protected")
@flask_jwt_required(jwt_mgr)
def protected():
    return {"user": g.jwt_claims["sub"]}

---

JWT Rotation

Manage short-lived access tokens and refresh-token blacklisting without a database round-trip on every request:

from cryptography.hazmat.primitives.asymmetric import rsa
from cryptography.hazmat.primitives.serialization import (
    Encoding, PrivateFormat, NoEncryption,
)
from lockwatch import JWTRotationManager, JWTConfig

# Generate once; persist the PEM in your secrets store
private_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
pem = private_key.private_bytes(Encoding.PEM, PrivateFormat.TraditionalOpenSSL, NoEncryption()).decode()

manager = JWTRotationManager(
    redis_client=redis,
    config=JWTConfig(rsa_private_key_pem=pem),
)

pair = await manager.issue_token_pair(subject="user:42", claims={"role": "admin"})
# pair.access_token  — RS256 JWT, 15-minute TTL
# pair.refresh_token — RS256 JWT, 7-day TTL, hash stored in Redis

new_pair = await manager.rotate(pair.refresh_token)  # atomic swap via WATCH/MULTI/EXEC

---

Anomaly Detection

Standalone burst detector — usable without the full middleware:

from lockwatch import AnomalyDetector, AnomalyConfig

detector = AnomalyDetector(
    redis_client=redis,
    config=AnomalyConfig(
        burst_threshold=50,
        burst_window_seconds=10,
        webhook_urls=["https://hooks.example.com/alert"],
    ),
)

is_burst = await detector.check(ip="1.2.3.4", endpoint="/api/login", method="POST")

---

Audit Log

Query who did what and when:

from lockwatch import AuditLogger, AuditQuery

logger = AuditLogger(database_url="postgresql+asyncpg://user:pass@host/db")
await logger.init()  # creates table if not exists (idempotent)

entries = await logger.query(AuditQuery(
    user_id="user:42",
    anomalies_only=True,
    limit=50,
))

Run Alembic migrations against your Postgres instance:

export LOCKWATCH_DATABASE_URL="postgresql+asyncpg://user:pass@host/db"
alembic upgrade head

---

Configuration Reference

Parameter	Env var override	Default	Description
redis_url	REDIS_URL	—	Redis connection URL (required)
rate_limit_requests	—	100	Max requests per window
rate_limit_window_seconds	—	60	Sliding window size in seconds
burst_threshold	—	50	Request count that triggers anomaly alert
burst_window_seconds	—	10	Burst detection window in seconds
webhook_urls	—	[]	List of HTTPS endpoints for anomaly alerts
audit_database_url	LOCKWATCH_DATABASE_URL	None	Postgres URL for audit log
on_redis_failure	—	"allow"	"allow" (fail open) or "deny" (fail closed)
jwt_secret	JWT_SECRET	—	HMAC secret for JWT signing
access_token_ttl	—	900	Access token lifetime in seconds
anomaly_window	—	10	Alias for burst_window_seconds

---

What it Does

• Sliding-window rate limiter — Redis sorted set records each request timestamp; window is computed by pruning entries older than rate_limit_window_seconds. O(log N) per operation, true sliding semantics (no thundering-herd at window boundary), atomic pipeline.
• JWT rotation with refresh blacklist — issues short-lived access tokens backed by opaque refresh tokens stored in Redis. rotate() atomically swaps old for new and blacklists the old refresh token with a TTL — no stale token reuse.
• IP burst anomaly detection — separate Redis sorted set per IP in a short burst window. When the threshold is crossed, fires async webhook alerts with a 60-second dedup window (one alert per IP per minute). Non-blocking — never delays the response.
• Postgres audit log — every request gets one row in lockwatch_audit_log including timestamp, IP, user_id, endpoint, method, status code, rate_limit_hit flag, anomaly_detected flag, and response latency. Writes are background tasks (asyncio.create_task) — a failed write logs a warning but never affects the response.

---

Running Tests

pip install lockwatch[all]
pip install pytest pytest-asyncio fakeredis[aioredis] aiosqlite
pytest --cov=lockwatch --cov-report=term-missing

---

License

MIT © Jordan Ho