fix(security): per-IP rate limiting on login and ingest

.env.example

@@ -34,3 +34,8 @@ TRENDS_RETENTION_DAYS=365
 # The rollup re-aggregates the last N complete hours each run (catches late data
 # and missed runs). Set to 0 to disable trend rollups.
 TRENDS_ROLLUP_LOOKBACK_HOURS=6
+
+# Per-IP rate limits (requests/minute) for abuse-prone endpoints; 0 disables.
+# In-memory per process — behind a load balancer, rate-limit at the proxy too.
+RATE_LIMIT_LOGIN_PER_MINUTE=10
+RATE_LIMIT_INGEST_PER_MINUTE=600

app/api/main.py

@@ -8,6 +8,7 @@
 from starlette.middleware.sessions import SessionMiddleware
 
 from app.api.middleware import add_security_headers
+from app.api.ratelimit import add_rate_limit
 from app.api.routes import api_router, health_router
 from app.api.routes.web import router as web_router
 from app.core.config import get_settings
@@ -54,6 +55,7 @@ def create_app(*, lifespan: Lifespan = scheduler_lifespan) -> FastAPI:
         https_only=settings.cookie_secure,
     )
     add_security_headers(app)
+    add_rate_limit(app)
 
     Instrumentator(
         should_group_status_codes=True,

app/api/ratelimit.py

@@ -0,0 +1,71 @@
+"""In-memory sliding-window rate limiting for abuse-prone endpoints (login,
+ingest). Per-process — fine for a single self-hosted instance; a multi-instance
+deployment would need a shared store (Redis), out of scope here. 0 disables a limit.
+"""
+
+from __future__ import annotations
+
+import time
+from collections import defaultdict, deque
+from collections.abc import Awaitable, Callable
+
+from fastapi import FastAPI, Request, Response
+from fastapi.responses import JSONResponse
+
+from app.core.config import get_settings
+
+_LOGIN_PATHS = frozenset({"/api/auth/login", "/login"})
+_INGEST_PATH = "/api/ingest"
+_WINDOW_SECONDS = 60.0
+
+
+class RateLimiter:
+    def __init__(self) -> None:
+        self._hits: dict[str, deque[float]] = defaultdict(deque)
+        self._calls = 0
+
+    def allow(self, key: str, limit: int, window: float) -> bool:
+        now = time.monotonic()
+        cutoff = now - window
+        bucket = self._hits[key]
+        while bucket and bucket[0] < cutoff:
+            bucket.popleft()
+        # Opportunistically drop drained buckets so the map can't grow unbounded.
+        self._calls += 1
+        if self._calls % 1024 == 0:
+            for k in [k for k, b in self._hits.items() if not b]:
+                del self._hits[k]
+        if len(bucket) >= limit:
+            return False
+        bucket.append(now)
+        return True
+
+
+def _limit_for(request: Request) -> int:
+    if request.method != "POST":
+        return 0
+    settings = get_settings()
+    if request.url.path in _LOGIN_PATHS:
+        return settings.rate_limit_login_per_minute
+    if request.url.path == _INGEST_PATH:
+        return settings.rate_limit_ingest_per_minute
+    return 0
+
+
+def add_rate_limit(app: FastAPI) -> None:
+    limiter = RateLimiter()
+
+    @app.middleware("http")
+    async def _rate_limit(
+        request: Request, call_next: Callable[[Request], Awaitable[Response]]
+    ) -> Response:
+        limit = _limit_for(request)
+        if limit > 0:
+            client = request.client.host if request.client else "unknown"
+            if not limiter.allow(f"{request.url.path}:{client}", limit, _WINDOW_SECONDS):
+                return JSONResponse(
+                    {"detail": "rate limit exceeded"},
+                    status_code=429,
+                    headers={"Retry-After": str(int(_WINDOW_SECONDS))},
+                )
+        return await call_next(request)

app/core/config.py

@@ -54,6 +54,10 @@ class Settings(BaseSettings):
     trends_retention_days: int = 365
     trends_rollup_lookback_hours: int = 6
 
+    # Per-IP rate limits (requests/minute) for abuse-prone endpoints; 0 disables.
+    rate_limit_login_per_minute: int = 10
+    rate_limit_ingest_per_minute: int = 600
+
     @property
     def cookie_secure(self) -> bool:
         """Mark session cookies `Secure` everywhere except local development, so a

tests/test_rate_limit.py

@@ -0,0 +1,27 @@
+"""Per-IP rate limiting on abuse-prone endpoints."""
+
+from __future__ import annotations
+
+import httpx
+
+from app.api.ratelimit import RateLimiter
+
+
+def test_rate_limiter_sliding_window() -> None:
+    limiter = RateLimiter()
+    assert all(limiter.allow("k", limit=3, window=60) for _ in range(3))
+    assert limiter.allow("k", limit=3, window=60) is False  # 4th over the limit
+    assert limiter.allow("other", limit=3, window=60) is True  # independent key
+
+
+async def test_login_is_rate_limited(client: httpx.AsyncClient) -> None:
+    # Default login limit is 10/min; the 11th attempt from the same IP is blocked.
+    last = None
+    for _ in range(11):
+        last = await client.post(
+            "/api/auth/login",
+            data={"username": "[email protected]", "password": "wrong-pass-12"},
+        )
+    assert last is not None
+    assert last.status_code == 429
+    assert last.headers.get("retry-after") == "60"