GuidoandClaude Opus 4.8
2c795a57c8
Grundgerüst für Certain Finance: - Backend: FastAPI + SQLAlchemy + SQLite, Alembic-Migrationen (Batch-Mode), argon2-Passwort-Hashing, server-seitige Sitzungen im Cookie, Erst-Admin, Benutzerverwaltung mit Temporär-Passwort, Einstellungen je Benutzer. - Ein Prozess, ein Port: uvicorn liefert API (/api) und gebautes SPA aus, inkl. SPA-Deeplink-Fallback. - Frontend: React 18 + Vite + JSX, Design-Tokens (Light/Dark), i18n DE/EN, Inline-SVG-Icons (keine Emojis), Login, Passwort-ändern, Dashboard-Platzhalter, Benutzer- und Einstellungsseiten. - Deploy: nativer LXC-Installer + Update-Skript (systemd), env-Vorlage, Doku. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
79 lines
2.3 KiB
Python
79 lines
2.3 KiB
Python
"""Server-seitige Sitzungen: anlegen, prüfen, widerrufen.
|
||
|
||
Im Cookie liegt nur das Klartext-Token; in der Datenbank nur dessen SHA-256-Hash.
|
||
So lässt sich eine Sitzung serverseitig beenden (Logout, „überall abmelden"),
|
||
und ein Blick in die DB verrät kein gültiges Cookie.
|
||
"""
|
||
|
||
from datetime import datetime, timedelta, timezone
|
||
|
||
from sqlalchemy.orm import Session as DBSession
|
||
|
||
from ..config import get_settings
|
||
from ..models import Session as UserSession, User
|
||
from ..security import generate_session_token, hash_session_token
|
||
|
||
|
||
def _now() -> datetime:
|
||
return datetime.now(timezone.utc)
|
||
|
||
|
||
def _aware(dt: datetime) -> datetime:
|
||
"""SQLite gibt datetimes teils ohne tzinfo zurück – als UTC interpretieren."""
|
||
return dt if dt.tzinfo is not None else dt.replace(tzinfo=timezone.utc)
|
||
|
||
|
||
def create_session(db: DBSession, user: User, user_agent: str | None = None) -> str:
|
||
"""Legt eine Sitzung an und gibt das Klartext-Token zurück (nur hier sichtbar)."""
|
||
settings = get_settings()
|
||
raw = generate_session_token()
|
||
entry = UserSession(
|
||
user_id=user.id,
|
||
token_hash=hash_session_token(raw),
|
||
expires_at=_now() + timedelta(days=settings.session_ttl_days),
|
||
last_seen_at=_now(),
|
||
user_agent=(user_agent or "")[:255] or None,
|
||
)
|
||
db.add(entry)
|
||
db.commit()
|
||
return raw
|
||
|
||
|
||
def validate(db: DBSession, raw_token: str) -> User | None:
|
||
"""Gibt den Benutzer zurück, wenn das Token gültig und nicht abgelaufen ist."""
|
||
if not raw_token:
|
||
return None
|
||
entry = (
|
||
db.query(UserSession)
|
||
.filter(UserSession.token_hash == hash_session_token(raw_token))
|
||
.first()
|
||
)
|
||
if entry is None:
|
||
return None
|
||
if _aware(entry.expires_at) < _now():
|
||
db.delete(entry)
|
||
db.commit()
|
||
return None
|
||
user = db.get(User, entry.user_id)
|
||
if user is None or not user.is_active:
|
||
return None
|
||
entry.last_seen_at = _now()
|
||
db.commit()
|
||
return user
|
||
|
||
|
||
def revoke(db: DBSession, raw_token: str) -> None:
|
||
entry = (
|
||
db.query(UserSession)
|
||
.filter(UserSession.token_hash == hash_session_token(raw_token))
|
||
.first()
|
||
)
|
||
if entry is not None:
|
||
db.delete(entry)
|
||
db.commit()
|
||
|
||
|
||
def revoke_all(db: DBSession, user: User) -> None:
|
||
db.query(UserSession).filter(UserSession.user_id == user.id).delete()
|
||
db.commit()
|