intent/intent/core/throttle.py
Alec Perkins 604c2bbf91
Some checks failed
ci / test (push) Failing after 1m6s
ci / deploy (push) Failing after 0s
Harden for launch: limits, CSP, moderation, admin 2FA (Phase 5)
- Rate limits in the shared cache for comments, issues, uploads and
  invites; a limited HTMX reply keeps the draft.
- Content-Security-Policy (self only, no inline script or style, bucket
  origin for images and frames) and Permissions-Policy on every page.
  htmx configured not to inject styles or eval.
- Moderation: deactivating a user ends their sessions and keeps their
  content; deleted accounts show as "ghost"; reserved usernames; the
  admin won't delete an org's last owner.
- Admin requires a TOTP code (django-otp), with add_admin_totp to enrol.
- Email provider chosen by environment through django-anymail.
- Playwright browser tests: axe WCAG 2.1 AA on every page in light and
  dark, no CSP violations, keyboard walkthrough, theme toggle. Explicit
  Canvas/CanvasText on :root fixes dark-mode contrast detection.
- bin/start creates the data dir if missing; docs/operations.md runbook.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-27 10:51:06 -04:00

31 lines
1.3 KiB
Python

"""Fixed-window rate limits kept in the shared cache (the database cache in production).
Services call `throttle()` before doing work; going over a limit raises a ValidationError
with a user-facing message, which views already show like any other rule violation.
"""
from django.core.cache import cache
from django.core.exceptions import ValidationError
TOO_MANY_MESSAGE = "You're doing that too often. Wait a few minutes and try again."
# (events, seconds). Generous for people; tight enough to stop a script flooding a project.
COMMENTS = (30, 10 * 60) # Comments and new threads, per user.
ISSUES = (20, 60 * 60) # New issues, per user.
UPLOADS = (30, 60 * 60) # File uploads and new versions, per user.
INVITES_PER_SENDER = (30, 60 * 60)
INVITES_PER_RECIPIENT = (3, 60 * 60) # Per org, so one address isn't flooded with email.
def throttle(key: str, limit: int, seconds: int, message: str = TOO_MANY_MESSAGE) -> None:
"""Count an event under `key`; raise once more than `limit` happen within `seconds`."""
key = f"intent:throttle:{key}"
if cache.add(key, 1, timeout=seconds):
return
try:
count = cache.incr(key)
except ValueError: # Expired between add and incr.
cache.add(key, 1, timeout=seconds)
return
if count > limit:
raise ValidationError(message)