67 lines
2.5 KiB
Python
67 lines
2.5 KiB
Python
|
"""Enkel rate limiting i minnet.
|
||
|
|
|
||
|
|
Lagt til etter sikkerhetsgjennomgang 2026-08-09 (se ARCHITECTURE_DECISIONS.md
|
||
|
|
-tillegg): ingen av auth-endepunktene hadde noen form for forsøksbegrensning,
|
||
|
|
noe som gjorde passord- og 2FA-kode-gjetting praktisk mulig for en angriper
|
||
|
|
med nok samtidighet.
|
||
|
|
|
||
|
|
Trygt å holde i minnet KUN fordi `teecup_api` kjører som ÉN uvicorn-prosess
|
||
|
|
(ingen `--workers`-flagg, se Dockerfile) -- tilstanden deles derfor ikke
|
||
|
|
mellom flere prosesser/containere, samme kjente begrensning som andre
|
||
|
|
in-memory-cacher i appen (se "Cache over flere prosesser" i
|
||
|
|
ARCHITECTURE_DECISIONS.md sin liste over åpne punkter). Ved fremtidig
|
||
|
|
skalering til flere workers/containere MÅ dette flyttes til Redis eller
|
||
|
|
tilsvarende delt lager -- ikke glem å oppdatere denne kommentaren da.
|
||
|
|
"""
|
||
|
|
|
||
|
|
import time
|
||
|
|
|
||
|
|
from fastapi import Request
|
||
|
|
|
||
|
|
from .errors import app_error
|
||
|
|
|
||
|
|
|
||
|
|
def client_ip(request: Request) -> str:
|
||
|
|
"""Klient-IP via X-Forwarded-For -- trygt KUN fordi appen ikke er nåbar
|
||
|
|
unntatt gjennom Caddy, som terminerer TLS og proxyer videre over vanlig
|
||
|
|
http internt (samme tillitsmodell som should_use_secure_cookies() i
|
||
|
|
app/auth.py). Første adresse i kjeden er den faktiske klienten."""
|
||
|
|
forwarded = request.headers.get("x-forwarded-for")
|
||
|
|
if forwarded:
|
||
|
|
return forwarded.split(",")[0].strip()
|
||
|
|
return request.client.host if request.client else "unknown"
|
||
|
|
|
||
|
|
|
||
|
|
class RateLimiter:
|
||
|
|
"""Fast-vindu-teller per nøkkel.
|
||
|
|
|
||
|
|
Bruk som FastAPI-dependency (`Depends(en_limiter)`) for IP-basert
|
||
|
|
begrensning, eller kall `.check(nøkkel)` direkte inne i en
|
||
|
|
endepunkt-funksjon for en mer presis nøkkel (f.eks. en pending-bruker-ID
|
||
|
|
-- riktig nøkkel når selve RESSURSEN som brute-forces er knyttet til en
|
||
|
|
bestemt konto, ikke avsender-IP-en)."""
|
||
|
|
|
||
|
|
def __init__(
|
||
|
|
self,
|
||
|
|
max_requests: int,
|
||
|
|
window_seconds: float,
|
||
|
|
message: str = "For mange forsøk. Vent litt og prøv igjen.",
|
||
|
|
):
|
||
|
|
self.max_requests = max_requests
|
||
|
|
self.window_seconds = window_seconds
|
||
|
|
self.message = message
|
||
|
|
self._buckets: dict[str, list[float]] = {}
|
||
|
|
|
||
|
|
def check(self, key: str) -> None:
|
||
|
|
now = time.monotonic()
|
||
|
|
window_start = now - self.window_seconds
|
||
|
|
hits = [t for t in self._buckets.get(key, []) if t >= window_start]
|
||
|
|
if len(hits) >= self.max_requests:
|
||
|
|
self._buckets[key] = hits
|
||
|
|
raise app_error(429, "RATE_LIMITED", self.message)
|
||
|
|
hits.append(now)
|
||
|
|
self._buckets[key] = hits
|
||
|
|
|
||
|
|
def __call__(self, request: Request) -> None:
|
||
|
|
self.check(client_ip(request))
|