refactor: isolate submission rate limiting

apps/api/app/main.py

@@ -1,7 +1,6 @@
 from __future__ import annotations
 
 from datetime import datetime, timedelta, timezone
-from ipaddress import IPv4Address, IPv6Address, IPv4Network, IPv6Network
 import hashlib
 import hmac
 import json
@@ -16,7 +15,7 @@ import httpx
 from fastapi import Depends, FastAPI, File, Header, HTTPException, Query, Request, Response, UploadFile
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.responses import JSONResponse
-from sqlalchemy import delete, func, select, text
+from sqlalchemy import func, select
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm import Session, joinedload
 
@@ -33,6 +32,8 @@ from .routers.public_data import router as public_data_router
 from .public_cache import public_cache
 from .schemas import ActivityOut, AdminCatchReportOut, CatchReportAccepted, CatchReportCreate, CatchReportCreated, ExternalAliasSuggestionOut, ExternalObservationDecision, ExternalObservationMapping, ExternalObservationOut, ExternalObservationPublished, ImportRunOut, ModerationUpdate
 from .storage import ScreenshotError, client as storage_client, delete_screenshot, signed_screenshot_url, upload_screenshot
+from .submission_security import check_rate_limit
+from .submission_security import is_trusted_proxy as _is_trusted_proxy
 
 
 configure_logging(settings.log_level)
@@ -395,39 +396,5 @@ def delete_report(report_id: UUID, db: Db, moderator: Annotated[str, Depends(_ad
     return Response(status_code=204)
 
 
-def _is_trusted_proxy(address: str, trusted_cidrs: list[str]) -> bool:
-    """Check if address is in trusted proxy CIDRs."""
-    try:
-        addr = IPv4Address(address) if ":" not in address else IPv6Address(address)
-    except ValueError:
-        return False
-    for cidr in trusted_cidrs:
-        try:
-            network = IPv4Network(cidr) if ":" not in cidr else IPv6Network(cidr)
-            if addr in network:
-                return True
-        except ValueError:
-            continue
-    return False
-
-
 def _check_rate_limit(request: Request, db: Session) -> None:
-    now = datetime.now(timezone.utc)
-    cutoff = now - timedelta(minutes=10)
-    # Extract real client IP from forwarded headers
-    client = request.client.host if request.client else "unknown"
-    forwarded = request.headers.get("x-forwarded-for")
-    # Only trust X-Forwarded-For if connection came from a trusted proxy
-    if forwarded and request.client and _is_trusted_proxy(request.client.host, settings.trusted_proxy_cidrs):
-        client = forwarded.split(",")[0].strip()
-    client_hash = hmac.new(settings.rate_limit_secret.encode(), client.encode(), hashlib.sha256).hexdigest()
-    if db.get_bind().dialect.name == "postgresql":
-        lock_key = int(client_hash[:16], 16) & 0x7FFF_FFFF_FFFF_FFFF
-        db.execute(text("SELECT pg_advisory_xact_lock(:lock_key)"), {"lock_key": lock_key})
-    db.execute(delete(SubmissionAttempt).where(SubmissionAttempt.created_at < now - timedelta(days=1)))
-    recent = db.scalar(select(func.count()).select_from(SubmissionAttempt).where(SubmissionAttempt.client_hash == client_hash, SubmissionAttempt.created_at >= cutoff)) or 0
-    if recent >= 5:
-        db.commit()
-        raise HTTPException(status_code=429, detail="too many submissions")
-    db.add(SubmissionAttempt(client_hash=client_hash, created_at=now))
-    db.commit()
+    check_rate_limit(request, db, settings)

apps/api/app/submission_security.py

@@ -0,0 +1,56 @@
+from datetime import datetime, timedelta, timezone
+import hashlib
+import hmac
+from ipaddress import IPv4Address, IPv4Network, IPv6Address, IPv6Network
+from typing import Protocol
+
+from fastapi import HTTPException, Request
+from sqlalchemy import delete, func, select, text
+from sqlalchemy.orm import Session
+
+from .models import SubmissionAttempt
+
+
+class RateLimitConfig(Protocol):
+    rate_limit_secret: str
+    trusted_proxy_cidrs: list[str]
+
+
+def is_trusted_proxy(address: str, trusted_cidrs: list[str]) -> bool:
+    try:
+        addr = IPv4Address(address) if ":" not in address else IPv6Address(address)
+    except ValueError:
+        return False
+    for cidr in trusted_cidrs:
+        try:
+            network = IPv4Network(cidr) if ":" not in cidr else IPv6Network(cidr)
+            if addr in network:
+                return True
+        except ValueError:
+            continue
+    return False
+
+
+def check_rate_limit(request: Request, db: Session, config: RateLimitConfig) -> None:
+    now = datetime.now(timezone.utc)
+    cutoff = now - timedelta(minutes=10)
+    client = request.client.host if request.client else "unknown"
+    forwarded = request.headers.get("x-forwarded-for")
+    if forwarded and request.client and is_trusted_proxy(request.client.host, config.trusted_proxy_cidrs):
+        client = forwarded.split(",")[0].strip()
+    client_hash = hmac.new(config.rate_limit_secret.encode(), client.encode(), hashlib.sha256).hexdigest()
+    if db.get_bind().dialect.name == "postgresql":
+        lock_key = int(client_hash[:16], 16) & 0x7FFF_FFFF_FFFF_FFFF
+        db.execute(text("SELECT pg_advisory_xact_lock(:lock_key)"), {"lock_key": lock_key})
+    db.execute(delete(SubmissionAttempt).where(SubmissionAttempt.created_at < now - timedelta(days=1)))
+    recent = db.scalar(
+        select(func.count()).select_from(SubmissionAttempt).where(
+            SubmissionAttempt.client_hash == client_hash,
+            SubmissionAttempt.created_at >= cutoff,
+        )
+    ) or 0
+    if recent >= 5:
+        db.commit()
+        raise HTTPException(status_code=429, detail="too many submissions")
+    db.add(SubmissionAttempt(client_hash=client_hash, created_at=now))
+    db.commit()