136 lines
4.0 KiB
Python
136 lines
4.0 KiB
Python
import hashlib
|
|
import hmac
|
|
import secrets
|
|
from datetime import UTC, datetime, timedelta
|
|
|
|
from argon2 import PasswordHasher
|
|
from argon2.exceptions import InvalidHashError, VerifyMismatchError
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.models import RateLimit
|
|
|
|
_hasher = PasswordHasher()
|
|
|
|
# Dummy-Hash gegen Timing-Angriffe: bei unbekanntem Konto wird trotzdem
|
|
# eine Verifikation durchgefuehrt, damit die Antwortzeit gleich bleibt.
|
|
_DUMMY_HASH = _hasher.hash("dummy-password-for-constant-time-comparison")
|
|
|
|
|
|
def hash_password(password: str) -> str:
|
|
return _hasher.hash(password)
|
|
|
|
|
|
def verify_password(password: str, password_hash: str | None) -> bool:
|
|
try:
|
|
_hasher.verify(password_hash or _DUMMY_HASH, password)
|
|
except (VerifyMismatchError, InvalidHashError):
|
|
return False
|
|
return password_hash is not None
|
|
|
|
|
|
def needs_rehash(password_hash: str) -> bool:
|
|
try:
|
|
return _hasher.check_needs_rehash(password_hash)
|
|
except InvalidHashError:
|
|
return True
|
|
|
|
|
|
def new_token() -> str:
|
|
"""256 Bit Zufall, URL-tauglich."""
|
|
return secrets.token_urlsafe(32)
|
|
|
|
|
|
def hash_token(token: str) -> str:
|
|
"""Token landen nur als Hash in der Datenbank. Kein Salt noetig -
|
|
der Eingabewert ist bereits hochentropisch."""
|
|
return hashlib.sha256(token.encode()).hexdigest()
|
|
|
|
|
|
def tokens_equal(a: str, b: str) -> bool:
|
|
return hmac.compare_digest(a, b)
|
|
|
|
|
|
def utcnow() -> datetime:
|
|
return datetime.now(UTC).replace(tzinfo=None)
|
|
|
|
|
|
def normalize_email(email: str) -> str:
|
|
return email.strip().lower()
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# Schluessel fuer die Missbrauchsbremse
|
|
# --------------------------------------------------------------------------
|
|
#
|
|
# Die Zaehler brauchen einen Schluessel je Absender - IP-Adresse oder
|
|
# E-Mail-Adresse. Beides im Klartext abzulegen waere unnoetig: Fuer das
|
|
# Hochzaehlen genuegt ein Wert, der gleich bleibt; lesbar muss er nicht
|
|
# sein.
|
|
#
|
|
# Deshalb ein Streuwert mit einem zufaelligen, installationsweiten
|
|
# Zusatz. Das ist Pseudonymisierung, keine Anonymisierung - wer die
|
|
# Datenbank samt Zusatz hat, koennte eine vermutete Adresse
|
|
# nachrechnen. Es verhindert aber, dass beim Blick in die Tabelle
|
|
# Adressen mitgelesen werden, und die Werte verschwinden ohnehin nach
|
|
# 24 Stunden.
|
|
|
|
_bucket_salt: str | None = None
|
|
|
|
|
|
def _salt(db: Session) -> bytes:
|
|
"""Zufallszusatz, beim ersten Bedarf erzeugt und dann bestaendig."""
|
|
global _bucket_salt
|
|
if _bucket_salt is None:
|
|
from app.models import Setting
|
|
|
|
row = db.get(Setting, "ratelimit_salt")
|
|
if row is None:
|
|
value = secrets.token_hex(16)
|
|
db.add(Setting(key="ratelimit_salt", value=value))
|
|
db.flush()
|
|
else:
|
|
value = row.value
|
|
_bucket_salt = value
|
|
return _bucket_salt.encode()[:32]
|
|
|
|
|
|
def bucket_key(db: Session, prefix: str, value: str) -> str:
|
|
"""Erzeugt einen Zaehlerschluessel, ohne den Wert selbst abzulegen."""
|
|
digest = hashlib.blake2s(
|
|
value.encode("utf-8"), key=_salt(db), digest_size=16
|
|
).hexdigest()
|
|
return f"{prefix}:{digest}"
|
|
|
|
|
|
def check_rate_limit(
|
|
db: Session, bucket: str, *, limit: int, window_minutes: int
|
|
) -> bool:
|
|
"""True = Anfrage erlaubt. Zaehlt hoch und gibt False zurueck,
|
|
sobald das Limit im aktuellen Fenster erreicht ist."""
|
|
now = utcnow()
|
|
window = now.replace(second=0, microsecond=0)
|
|
window = window - timedelta(minutes=window.minute % window_minutes)
|
|
|
|
row = db.get(RateLimit, (bucket[:160], window))
|
|
if row is None:
|
|
row = RateLimit(bucket=bucket[:160], window_start=window, count=1)
|
|
db.add(row)
|
|
db.flush()
|
|
return True
|
|
|
|
if row.count >= limit:
|
|
return False
|
|
|
|
row.count += 1
|
|
db.flush()
|
|
return True
|
|
|
|
|
|
def purge_rate_limits(db: Session, older_than_hours: int = 24) -> None:
|
|
cutoff = utcnow() - timedelta(hours=older_than_hours)
|
|
for row in db.scalars(
|
|
select(RateLimit).where(RateLimit.window_start < cutoff)
|
|
).all():
|
|
db.delete(row)
|