marco.morath/einkaufsapp
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Ergänzung Datenschutzerklärung und Impressum, Fristenformular in Einstellungen

Browse Source
This commit is contained in:
Marco Morath
2026-08-10 06:44:11 +02:00
parent 8ffbda875e
commit 777e3e5447
25 changed files with 1310 additions and 139 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
backend/app/config.py
View File

@@ -72,6 +72,13 @@ class Settings(BaseSettings):
cleanup_ops_days: int = 7
# Abstand zwischen zwei Durchlaeufen.
cleanup_interval_hours: int = 24
# Startwerte fuer Fristen, die zur Laufzeit in der Verwaltung
# geaendert werden koennen (siehe app/runtime_settings.py).
invite_days: int = 14
welcome_days: int = 14
verify_hours: int = 24
reset_hours: int = 1
email_change_hours: int = 48
# ---------------- Push-Benachrichtigungen ----------------
# Erzeugen mit: python3 tools/vapid-keys.py

10
backend/app/deps.py
View File

@@ -60,18 +60,24 @@ def self_registration_enabled(db: Session) -> bool:
# --------------------------------------------------------------------------
def create_session(db: Session, user: User, response: Response) -> UserSession:
# Spät importiert: runtime_settings greift auf get_setting in dieser
# Datei zu, ein Import auf Modulebene wäre ein Kreis.
from app.runtime_settings import get_duration
days = get_duration(db, "session_days")
raw = new_token()
csrf = new_token()
sess = UserSession(
token_hash=hash_token(raw),
user_id=user.id,
csrf_token=csrf,
expires_at=utcnow() + timedelta(days=settings.session_days),
expires_at=utcnow() + timedelta(days=days),
)
db.add(sess)
db.flush()
max_age = settings.session_days * 24 * 3600
max_age = days * 24 * 3600
# HttpOnly: fuer JavaScript unsichtbar, damit ein XSS-Fund das Token
# nicht abgreifen kann.
response.set_cookie(

25
backend/app/maintenance.py
View File

@@ -38,6 +38,7 @@ from app.models import (
User,
UserSession,
)
from app.runtime_settings import get_duration
from app.security import utcnow
from app.users import deactivate, delete_user, months_setting
@@ -45,10 +46,19 @@ log = logging.getLogger(__name__)
def run_cleanup(db: Session) -> dict[str, int]:
"""Räumt auf und gibt zurück, was entfernt wurde."""
"""Räumt auf und gibt zurück, was entfernt wurde.
Alle Fristen kommen aus der Laufzeitkonfiguration, damit das, was
hier geschieht, dem entspricht, was in der Verwaltung steht - und
dem, was die Datenschutzerklärung behauptet.
"""
now = utcnow()
counts: dict[str, int] = {}
deleted_days = get_duration(db, "cleanup_deleted_days")
ops_days = get_duration(db, "cleanup_ops_days")
product_days = get_duration(db, "product_cache_days")
def purge(model, condition, label: str) -> None:
result = db.execute(delete(model).where(condition))
if result.rowcount:
@@ -79,34 +89,33 @@ def run_cleanup(db: Session) -> dict[str, int]:
# längst über die Wiederholungsgrenze hinaus versucht.
purge(
AppliedOp,
AppliedOp.created_at < now - timedelta(days=settings.cleanup_ops_days),
AppliedOp.created_at < now - timedelta(days=ops_days),
"Outbox-Quittungen",
)
# --- Abgelaufene Einladungen und öffentliche Links ---
purge(
ListInvite,
ListInvite.expires_at < now - timedelta(days=settings.cleanup_deleted_days),
ListInvite.expires_at < now - timedelta(days=deleted_days),
"alte Einladungen",
)
purge(
PublicShare,
PublicShare.expires_at < now - timedelta(days=settings.cleanup_deleted_days),
PublicShare.expires_at < now - timedelta(days=deleted_days),
"abgelaufene öffentliche Links",
)
# --- Produktzwischenspeicher ---
purge(
ProductCache,
ProductCache.fetched_at
< now - timedelta(days=max(settings.product_cache_days * 2, 365)),
ProductCache.fetched_at < now - timedelta(days=max(product_days * 2, 365)),
"Produktzwischenspeicher",
)
# --- Endgültiges Löschen weicher Löschungen ---
# Erst jetzt verschwinden die Daten wirklich. Bis dahin konnten
# offline gebliebene Geräte erfahren, dass es sie nicht mehr gibt.
cutoff = now - timedelta(days=settings.cleanup_deleted_days)
cutoff = now - timedelta(days=deleted_days)
purge(ListItem, ListItem.deleted_at < cutoff, "gelöschte Einträge")
@@ -135,7 +144,7 @@ def run_cleanup(db: Session) -> dict[str, int]:
# --- Abgeschlossene Adressänderungen ---
purge(
EmailChange,
EmailChange.expires_at < now - timedelta(days=settings.cleanup_deleted_days),
EmailChange.expires_at < now - timedelta(days=deleted_days),
"alte Adressänderungen",
)

4
backend/app/product_lookup.py
View File

@@ -163,9 +163,11 @@ def lookup(db: Session, barcode: str) -> ProductCache | None:
cached = db.get(ProductCache, barcode)
if cached is not None:
from app.runtime_settings import get_duration
age = utcnow() - cached.fetched_at
max_age = timedelta(
days=settings.product_cache_days if cached.found
days=get_duration(db, "product_cache_days") if cached.found
else settings.product_miss_days
)
if age < max_age:

63
backend/app/routers/admin.py
View File

@@ -26,16 +26,18 @@ from app.mail import (
from app.maintenance import describe, run_cleanup
from app.models import EmailChange, User
from app.schemas import MailCheckOut, MailTestIn, MessageOut
from app.runtime_settings import BY_KEY, describe_durations, set_duration
from app.schemas_admin import (
AdminSettingsIn,
AdminSettingsOut,
AdminStatsOut,
AdminUserOut,
DeleteUserIn,
DurationOut,
EmailChangeIn,
UserCreateIn,
)
from app.security import check_rate_limit, normalize_email, utcnow
from app.security import bucket_key, check_rate_limit, normalize_email, utcnow
from app.users import (
EMAIL_CHANGE_HOURS,
WELCOME_DAYS,
@@ -66,18 +68,17 @@ def _settings_out(db) -> AdminSettingsOut:
return AdminSettingsOut(
allow_self_registration=self_registration_enabled(db),
locked_by_env=registration_locked_by_env(),
auto_deactivate_months=months_setting(
db, "auto_deactivate_months", DEFAULT_DEACTIVATE_MONTHS
),
auto_delete_months=months_setting(
db, "auto_delete_months", DEFAULT_DELETE_MONTHS
),
auto_archive_months=months_setting(
db, "auto_archive_months", DEFAULT_ARCHIVE_MONTHS
),
archive_delete_months=months_setting(
db, "archive_delete_months", DEFAULT_ARCHIVE_DELETE_MONTHS
),
durations=[DurationOut(**entry) for entry in describe_durations(db)],
fixed={
"Zähler der Missbrauchsabwehr": "24 Stunden",
"Fehlgeschlagene Produktabfragen":
f"{settings.product_miss_days} Tage (PRODUCT_MISS_DAYS)",
"Aufräumlauf":
f"alle {settings.cleanup_interval_hours} Stunden "
"(CLEANUP_INTERVAL_HOURS)",
"Zugriffsprotokolle des Webservers":
"Sache der Protokollrotation auf dem Server, nicht der Anwendung",
},
)
@@ -102,14 +103,12 @@ def write_settings(payload: AdminSettingsIn, db: DbSession, admin: AdminUser):
"true" if payload.allow_self_registration else "false",
)
if payload.auto_deactivate_months is not None:
set_setting(db, "auto_deactivate_months", str(payload.auto_deactivate_months))
if payload.auto_delete_months is not None:
set_setting(db, "auto_delete_months", str(payload.auto_delete_months))
if payload.auto_archive_months is not None:
set_setting(db, "auto_archive_months", str(payload.auto_archive_months))
if payload.archive_delete_months is not None:
set_setting(db, "archive_delete_months", str(payload.archive_delete_months))
for key, value in (payload.durations or {}).items():
if key not in BY_KEY:
raise HTTPException(
status.HTTP_400_BAD_REQUEST, f"Unbekannte Frist: {key}"
)
set_duration(db, key, value)
db.commit()
return _settings_out(db)
@@ -222,7 +221,8 @@ def add_user(
background: BackgroundTasks,
):
if not check_rate_limit(
db, f"admin-create:{client_ip(request)}", limit=30, window_minutes=60
db, bucket_key(db, "admin-create", client_ip(request)),
limit=30, window_minutes=60,
):
db.commit()
raise HTTPException(status.HTTP_429_TOO_MANY_REQUESTS, "Zu viele Anlagen.")
@@ -241,7 +241,8 @@ def add_user(
background.add_task(
send_welcome, email, token,
admin.display_name or admin.email.split("@")[0], WELCOME_DAYS,
admin.display_name or admin.email.split("@")[0],
get_duration(db, "welcome_days"),
)
return _user_out(user, {}, {}, {})
@@ -268,16 +269,19 @@ def resend_welcome(
"nutzt die Person „Passwort vergessen“ auf der Anmeldeseite.",
)
if not check_rate_limit(db, f"admin-welcome:{user.id}", limit=5, window_minutes=60):
if not check_rate_limit(
db, bucket_key(db, "admin-welcome", user.id), limit=5, window_minutes=60
):
db.commit()
raise HTTPException(status.HTTP_429_TOO_MANY_REQUESTS, "Zu viele Versuche.")
token = issue_token(db, user, "welcome", hours=WELCOME_DAYS * 24)
days = get_duration(db, "welcome_days")
token = issue_token(db, user, "welcome", hours=days * 24)
db.commit()
background.add_task(
send_welcome, user.email, token,
admin.display_name or admin.email.split("@")[0], WELCOME_DAYS,
admin.display_name or admin.email.split("@")[0], days,
)
return MessageOut(detail=f"Willkommensnachricht erneut an {user.email} versendet.")
@@ -310,12 +314,13 @@ def change_email(
change, raw_new, raw_old = start_email_change(db, user, target, admin.id)
db.commit()
change_hours = get_duration(db, "email_change_hours")
background.add_task(
send_email_change_verify, target, raw_new, old_email, EMAIL_CHANGE_HOURS
send_email_change_verify, target, raw_new, old_email, change_hours
)
if change.requires_old and raw_old:
background.add_task(
send_email_change_verify_old, old_email, raw_old, target, EMAIL_CHANGE_HOURS
send_email_change_verify_old, old_email, raw_old, target, change_hours
)
detail = (
f"Bestätigungslinks an {target} und {old_email} versendet. "
@@ -324,7 +329,7 @@ def change_email(
)
else:
background.add_task(
send_email_change_notice, old_email, target, EMAIL_CHANGE_HOURS
send_email_change_notice, old_email, target, change_hours
)
detail = (
f"Bestätigungslink an {target} versendet, Hinweis an {old_email}. "

24
backend/app/routers/auth.py
View File

@@ -28,6 +28,7 @@ from app.schemas import (
RegisterIn,
UserOut,
)
from app.runtime_settings import get_duration
from app.users import apply_if_complete
from app.security import (
check_rate_limit,
@@ -96,7 +97,8 @@ def register(
)
if not check_rate_limit(
db, f"register:{client_ip(request)}", limit=5, window_minutes=60
db, bucket_key(db, "register", client_ip(request)),
limit=5, window_minutes=60,
):
db.commit()
raise HTTPException(status.HTTP_429_TOO_MANY_REQUESTS, "Zu viele Versuche.")
@@ -113,12 +115,14 @@ def register(
)
db.add(user)
db.flush()
raw = _issue_token(db, user, "verify", hours=24)
raw = _issue_token(db, user, "verify", hours=get_duration(db, "verify_hours"))
background.add_task(send_verification, email, raw)
elif existing.verified_at is None:
# Unbestaetigtes Konto: neuen Link schicken, statt zu verraten,
# dass die Adresse schon vergeben ist.
raw = _issue_token(db, existing, "verify", hours=24)
raw = _issue_token(
db, existing, "verify", hours=get_duration(db, "verify_hours")
)
background.add_task(send_verification, email, raw)
# Bestaetigtes Konto: nichts tun, aber gleiche Antwort geben.
@@ -148,8 +152,13 @@ def verify(token: str, db: DbSession):
def login(payload: LoginIn, request: Request, response: Response, db: DbSession):
email = normalize_email(payload.email)
ip_ok = check_rate_limit(db, f"login-ip:{client_ip(request)}", limit=20, window_minutes=15)
acct_ok = check_rate_limit(db, f"login-acct:{email}", limit=8, window_minutes=15)
ip_ok = check_rate_limit(
db, bucket_key(db, "login-ip", client_ip(request)),
limit=20, window_minutes=15,
)
acct_ok = check_rate_limit(
db, bucket_key(db, "login-acct", email), limit=8, window_minutes=15
)
if not (ip_ok and acct_ok):
db.commit()
raise HTTPException(
@@ -266,7 +275,8 @@ def reset_request(
background: BackgroundTasks,
):
if not check_rate_limit(
db, f"reset:{client_ip(request)}", limit=5, window_minutes=60
db, bucket_key(db, "reset", client_ip(request)),
limit=5, window_minutes=60,
):
db.commit()
raise HTTPException(status.HTTP_429_TOO_MANY_REQUESTS, "Zu viele Versuche.")
@@ -274,7 +284,7 @@ def reset_request(
email = normalize_email(payload.email)
user = db.scalar(select(User).where(User.email == email))
if user is not None and user.is_active and user.verified_at is not None:
raw = _issue_token(db, user, "reset", hours=1)
raw = _issue_token(db, user, "reset", hours=get_duration(db, "reset_hours"))
background.add_task(send_password_reset, email, raw)
db.commit()

5
backend/app/routers/public.py
View File

@@ -31,7 +31,7 @@ from app.schemas_shopping import (
PublicToggleIn,
PublicViewOut,
)
from app.security import check_rate_limit, hash_token, new_token, utcnow
from app.security import bucket_key, check_rate_limit, hash_token, new_token, utcnow
router = APIRouter(tags=["public"])
@@ -189,7 +189,8 @@ def _load_share(db: DbSession, token: str, request: Request) -> PublicShare:
# bremsen, bevor jemand Token durchprobiert. Bei 256 Bit Zufall ist
# Raten aussichtslos, aber die Last soll trotzdem begrenzt bleiben.
if not check_rate_limit(
db, f"public:{client_ip(request)}", limit=120, window_minutes=15
db, bucket_key(db, "public", client_ip(request)),
limit=120, window_minutes=15,
):
db.commit()
raise HTTPException(

26
backend/app/routers/sharing.py
View File

@@ -40,10 +40,17 @@ from app.security import (
router = APIRouter(tags=["sharing"])
# Vorgabewert; maßgeblich ist die Einstellung in der Verwaltung.
INVITE_DAYS = 14
MAX_OPEN_INVITES = 50
def _invite_days(db) -> int:
from app.runtime_settings import get_duration
return get_duration(db, "invite_days")
def _status(invite: ListInvite) -> str:
if invite.revoked_at:
return "revoked"
@@ -78,8 +85,13 @@ def _guard_invite_rate(db: DbSession, request: Request, list_id: str) -> None:
unerwünschte Mails missbrauchen - jede Einladung erzeugt schließlich
eine Nachricht an eine frei wählbare Adresse.
"""
ok_list = check_rate_limit(db, f"invite-list:{list_id}", limit=20, window_minutes=60)
ok_ip = check_rate_limit(db, f"invite-ip:{client_ip(request)}", limit=30, window_minutes=60)
ok_list = check_rate_limit(
db, bucket_key(db, "invite-list", list_id), limit=20, window_minutes=60
)
ok_ip = check_rate_limit(
db, bucket_key(db, "invite-ip", client_ip(request)),
limit=30, window_minutes=60,
)
if not (ok_list and ok_ip):
db.commit()
raise HTTPException(
@@ -159,6 +171,7 @@ def create_invite(
).all():
old.revoked_at = utcnow()
valid_days = _invite_days(db)
raw = new_token()
invite = ListInvite(
list_id=lst.id,
@@ -166,7 +179,7 @@ def create_invite(
role=payload.role,
token_hash=hash_token(raw),
invited_by=user.id,
expires_at=utcnow() + timedelta(days=INVITE_DAYS),
expires_at=utcnow() + timedelta(days=valid_days),
)
db.add(invite)
db.commit()
@@ -174,7 +187,7 @@ def create_invite(
background.add_task(
send_invitation, email, raw,
lst.name, user.display_name or user.email.split("@")[0], INVITE_DAYS,
lst.name, user.display_name or user.email.split("@")[0], valid_days,
)
return _invite_out(invite)
@@ -204,10 +217,11 @@ def resend_invite(
_guard_invite_rate(db, request, lst.id)
valid_days = _invite_days(db)
raw = new_token()
invite.token_hash = hash_token(raw)
invite.revoked_at = None
invite.expires_at = utcnow() + timedelta(days=INVITE_DAYS)
invite.expires_at = utcnow() + timedelta(days=valid_days)
invite.last_sent_at = utcnow()
invite.send_count += 1
db.commit()
@@ -215,7 +229,7 @@ def resend_invite(
background.add_task(
send_invitation, invite.email, raw,
lst.name, user.display_name or user.email.split("@")[0], INVITE_DAYS,
lst.name, user.display_name or user.email.split("@")[0], valid_days,
)
return _invite_out(invite)

183
backend/app/runtime_settings.py Normal file
View File

@@ -0,0 +1,183 @@
"""Fristen, die sich zur Laufzeit einstellen lassen.
Bis hierher standen sie verstreut: teils als Konstante im Quelltext
(`INVITE_DAYS = 14`), teils in der Umgebung (`CLEANUP_DELETED_DAYS`),
teils in der Einstellungstabelle. Wer nachsehen wollte, was tatsächlich
gilt, musste an drei Stellen suchen - und die Datenschutzerklärung
behauptete Werte, die niemand überprüfen konnte.
Jetzt an einer Stelle, mit klarer Rangfolge:
Einstellungstabelle → Umgebungsvariable → Vorgabewert hier
Die Umgebungsvariable wirkt also als Startwert; sobald jemand in der
Verwaltung etwas ändert, gilt der gespeicherte Wert. Das entspricht dem
Verhalten der Selbstregistrierung.
Fristen, die NICHT hierhergehören, weil sie außerhalb der Anwendung
liegen: die Aufbewahrung der Webserver-Protokolle (Sache der
Protokollrotation auf dem Server) und alles, was fremde Dienste mit
ihren eigenen Daten tun.
"""
import logging
from dataclasses import dataclass
from sqlalchemy.orm import Session
from app.config import settings
from app.deps import get_setting, set_setting
log = logging.getLogger(__name__)
@dataclass(frozen=True)
class Duration:
key: str
label: str
unit: str # "Tage" | "Monate" | "Stunden"
default: int
minimum: int
maximum: int
# Was passiert, wenn 0 eingetragen wird? None = nicht erlaubt.
zero_means: str | None = None
note: str = ""
# Reihenfolge bestimmt die Darstellung in der Verwaltung.
DURATIONS: tuple[Duration, ...] = (
# ---- Konten ----
Duration(
"auto_deactivate_months", "Konto deaktivieren nach", "Monaten",
default=12, minimum=0, maximum=600, zero_means="abgeschaltet",
note="ohne Anmeldung; Administratorkonten sind ausgenommen",
),
Duration(
"auto_delete_months", "Konto löschen nach", "Monaten",
default=12, minimum=0, maximum=600, zero_means="abgeschaltet",
note="Deaktivierung",
),
# ---- Listen ----
Duration(
"auto_archive_months", "Liste archivieren nach", "Monaten",
default=12, minimum=0, maximum=600, zero_means="abgeschaltet",
note="ohne Öffnen; gilt je Person getrennt",
),
Duration(
"archive_delete_months", "Liste löschen nach", "Monaten",
default=12, minimum=0, maximum=600, zero_means="abgeschaltet",
note="im Archiv des Eigentümers; Vorwarnung 7 Tage und 1 Tag vorher",
),
# ---- Aufbewahrung ----
Duration(
"cleanup_deleted_days", "Gelöschtes endgültig entfernen nach", "Tagen",
default=30, minimum=1, maximum=365,
note="so lange kann ein Gerät offline bleiben und vom Löschen "
"noch erfahren",
),
Duration(
"cleanup_ops_days", "Quittungen der Warteschlange aufbewahren", "Tage",
default=7, minimum=1, maximum=90,
note="verhindern, dass eine wiederholte Änderung doppelt ausgeführt wird",
),
Duration(
"product_cache_days", "Produktabfragen gültig", "Tage",
default=180, minimum=1, maximum=3650,
note="danach wird derselbe Strichcode erneut abgefragt",
),
# ---- Sitzungen und Links ----
Duration(
"session_days", "Angemeldet bleiben", "Tage",
default=30, minimum=1, maximum=365,
note="danach ist eine erneute Anmeldung nötig",
),
Duration(
"invite_days", "Einladung gültig", "Tage",
default=14, minimum=1, maximum=90,
note="danach muss sie erneut versendet werden",
),
Duration(
"welcome_days", "Willkommenslink gültig", "Tage",
default=14, minimum=1, maximum=90,
note="Link zum Festlegen des ersten Passworts",
),
Duration(
"verify_hours", "Bestätigungslink gültig", "Stunden",
default=24, minimum=1, maximum=720,
note="Bestätigung der E-Mail-Adresse nach der Registrierung",
),
Duration(
"reset_hours", "Passwortlink gültig", "Stunden",
default=1, minimum=1, maximum=72,
note="kurz gehalten: Der Link erlaubt die Übernahme des Kontos",
),
Duration(
"email_change_hours", "Adresswechsel bestätigen binnen", "Stunden",
default=48, minimum=1, maximum=336,
note="bei Administratorkonten müssen beide Adressen zustimmen",
),
)
BY_KEY = {d.key: d for d in DURATIONS}
def _env_default(entry: Duration) -> int:
"""Startwert aus der Umgebung, falls es dort ein passendes Feld gibt."""
value = getattr(settings, entry.key, None)
if isinstance(value, int) and not isinstance(value, bool):
return value
return entry.default
def get_duration(db: Session, key: str) -> int:
"""Aktuell geltender Wert."""
entry = BY_KEY[key]
raw = get_setting(db, key, "")
if not raw:
return _env_default(entry)
try:
value = int(raw)
except ValueError:
log.warning(
"Einstellung %s ist keine Zahl (%r) - verwende %d",
key, raw, _env_default(entry),
)
return _env_default(entry)
if entry.zero_means and value == 0:
return 0
return max(entry.minimum, min(value, entry.maximum))
def set_duration(db: Session, key: str, value: int) -> int:
entry = BY_KEY[key]
if entry.zero_means and value == 0:
clamped = 0
else:
clamped = max(entry.minimum, min(int(value), entry.maximum))
set_setting(db, key, str(clamped))
return clamped
def all_durations(db: Session) -> dict[str, int]:
return {entry.key: get_duration(db, entry.key) for entry in DURATIONS}
def describe_durations(db: Session) -> list[dict]:
"""Beschreibung für die Verwaltungsoberfläche - Beschriftung,
Einheit, Grenzen und Bedeutung der Null kommen aus dieser Datei,
damit die Oberfläche sie nicht doppelt vorhalten muss."""
return [
{
"key": entry.key,
"label": entry.label,
"unit": entry.unit,
"value": get_duration(db, entry.key),
"default": _env_default(entry),
"minimum": entry.minimum,
"maximum": entry.maximum,
"zero_means": entry.zero_means,
"note": entry.note,
}
for entry in DURATIONS
]

36
backend/app/schemas_admin.py
View File

@@ -58,23 +58,39 @@ class DeleteUserIn(BaseModel):
confirm_email: EmailStr
class DurationOut(BaseModel):
"""Eine einstellbare Frist samt Beschreibung.
Beschriftung, Einheit und Grenzen kommen vom Server, damit die
Oberflaeche sie nicht doppelt vorhalten muss - und damit nicht das
eine steht und das andere gilt.
"""
key: str
label: str
unit: str
value: int
default: int
minimum: int
maximum: int
# Was bedeutet 0? None = nicht erlaubt.
zero_means: str | None = None
note: str = ""
class AdminSettingsOut(BaseModel):
allow_self_registration: bool
locked_by_env: bool
# 0 = abgeschaltet
auto_deactivate_months: int
auto_delete_months: int
# Listen: archivieren nach Nichtbenutzung, loeschen nach Archivdauer
auto_archive_months: int
archive_delete_months: int
durations: list[DurationOut]
# Fristen, die ausserhalb der Anwendung liegen - nur zur Anzeige,
# damit die Uebersicht nicht unvollstaendig wirkt.
fixed: dict[str, str]
class AdminSettingsIn(BaseModel):
allow_self_registration: bool | None = None
auto_deactivate_months: int | None = Field(default=None, ge=0, le=600)
auto_delete_months: int | None = Field(default=None, ge=0, le=600)
auto_archive_months: int | None = Field(default=None, ge=0, le=600)
archive_delete_months: int | None = Field(default=None, ge=0, le=600)
# Schluessel aus DURATIONS -> neuer Wert
durations: dict[str, int] | None = None
class WelcomeCompleteIn(BaseModel):

44
backend/app/security.py
View File

@@ -59,6 +59,50 @@ 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:

39
backend/app/users.py
View File

@@ -24,6 +24,7 @@ from app.security import hash_token, new_token, normalize_email, utcnow
log = logging.getLogger(__name__)
# Vorgabewerte; maßgeblich ist, was in der Verwaltung eingestellt ist.
WELCOME_DAYS = 14
EMAIL_CHANGE_HOURS = 48
@@ -32,22 +33,18 @@ EMAIL_CHANGE_HOURS = 48
# Einstellungen zur automatischen Bereinigung
# ==========================================================================
def months_setting(db: Session, key: str, fallback: int) -> int:
"""Liest einen Monatswert aus der Laufzeitkonfiguration.
def months_setting(db: Session, key: str, fallback: int = 0) -> int:
"""Monatswert aus der Laufzeitkonfiguration.
0 bedeutet ausdrücklich "abgeschaltet" - nicht "sofort". Ein
Tippfehler soll nicht dazu führen, dass beim nächsten Durchlauf alle
Konten verschwinden.
Nur noch eine Weiterleitung - die Fristen liegen gesammelt in
app/runtime_settings.py. Der Name bleibt, damit die vorhandenen
Aufrufstellen unverändert weiterlaufen.
0 bedeutet ausdrücklich "abgeschaltet", nicht "sofort".
"""
raw = get_setting(db, key, "")
if not raw:
return fallback
try:
value = int(raw)
except ValueError:
log.warning("Einstellung %s ist keine Zahl (%r) - verwende %d", key, raw, fallback)
return fallback
return max(0, min(value, 600))
from app.runtime_settings import get_duration
return get_duration(db, key)
# ==========================================================================
@@ -77,6 +74,12 @@ def issue_token(db: Session, user: User, purpose: str, hours: int) -> str:
return raw
def welcome_days(db: Session) -> int:
from app.runtime_settings import get_duration
return get_duration(db, "welcome_days")
def create_user(
db: Session, *, email: str, display_name: str | None, is_admin: bool = False
) -> tuple[User, str]:
@@ -99,7 +102,7 @@ def create_user(
)
db.add(user)
db.flush()
token = issue_token(db, user, "welcome", hours=WELCOME_DAYS * 24)
token = issue_token(db, user, "welcome", hours=welcome_days(db) * 24)
return user, token
@@ -133,6 +136,10 @@ def start_email_change(
).all():
pending.cancelled_at = utcnow()
from app.runtime_settings import get_duration
hours = get_duration(db, "email_change_hours")
raw_new = new_token()
raw_old = new_token() if user.is_admin else None
@@ -144,7 +151,7 @@ def start_email_change(
token_old_hash=hash_token(raw_old) if raw_old else None,
requires_old=user.is_admin,
requested_by=requested_by,
expires_at=utcnow() + timedelta(hours=EMAIL_CHANGE_HOURS),
expires_at=utcnow() + timedelta(hours=hours),
)
db.add(change)
db.flush()