Kyle DammannandClaude Opus 5.5
ef9deaeff7
- Django 5.2 + PostgreSQL 17 via docker compose (app + db, volumes, healthcheck on /healthz, test profile running pytest against Postgres) - Custom e-mail user model with user/admin roles - Invitation-only registration with single-use, hashed, expiring links - Admin area: invitations, role changes, deactivation, password reset via one-time link or temporary password (forced change on login) - First admin created from ADMIN_EMAIL/ADMIN_PASSWORD on startup - Login rate limiting per e-mail and IP (DB-backed, worker-safe) - Proxy middleware honouring X-Forwarded-* only from TRUSTED_PROXIES, optional CLIENT_IP_HEADER (e.g. CF-Connecting-IP) - Security headers (CSP, HSTS, Permissions-Policy), secure __Host- cookies - Mobile-first German UI, CLAUDE.md, README, init-env script - 65 tests for auth, permissions, proxy handling and rate limiting Co-Authored-By: Claude Opus 5.5 <[email protected]>
59 lines
1.9 KiB
Python
59 lines
1.9 KiB
Python
"""Database-backed fixed-window rate limiting.
|
|
|
|
Stored in the database rather than a process-local cache so that the limit holds
|
|
across all gunicorn workers without needing Redis. A limit of 0 means unlimited.
|
|
"""
|
|
|
|
from datetime import timedelta
|
|
|
|
from django.db import IntegrityError, transaction
|
|
from django.utils import timezone
|
|
|
|
from .models import RateLimitBucket
|
|
|
|
|
|
def _expired(bucket, window, now):
|
|
return bucket.window_start + window <= now
|
|
|
|
|
|
def is_limited(key, limit, window: timedelta):
|
|
"""True if the key has already reached the limit in the current window."""
|
|
if limit <= 0:
|
|
return False
|
|
bucket = RateLimitBucket.objects.filter(key=key).first()
|
|
if bucket is None or _expired(bucket, window, timezone.now()):
|
|
return False
|
|
return bucket.count >= limit
|
|
|
|
|
|
def hit(key, limit, window: timedelta):
|
|
"""Count one event. Returns False if this event exceeds the limit."""
|
|
if limit <= 0:
|
|
return True
|
|
now = timezone.now()
|
|
for _ in range(2):
|
|
try:
|
|
with transaction.atomic():
|
|
bucket = RateLimitBucket.objects.select_for_update().filter(key=key).first()
|
|
if bucket is None:
|
|
RateLimitBucket.objects.create(key=key, window_start=now, count=1)
|
|
return True
|
|
if _expired(bucket, window, now):
|
|
bucket.window_start = now
|
|
bucket.count = 0
|
|
bucket.count += 1
|
|
bucket.save(update_fields=["window_start", "count"])
|
|
return bucket.count <= limit
|
|
except IntegrityError:
|
|
# Another worker created the row concurrently; retry with the row present.
|
|
continue
|
|
return False
|
|
|
|
|
|
def reset(key):
|
|
RateLimitBucket.objects.filter(key=key).delete()
|
|
|
|
|
|
def purge_expired(older_than: timedelta = timedelta(days=2)):
|
|
RateLimitBucket.objects.filter(window_start__lt=timezone.now() - older_than).delete()
|