ScarriffleandClaude Opus 5
9be772557e
Das Sitzungs-Token benannte den Benutzer ueber seinen Namen (sub = username). Damit war der Name die Kennung der Sitzung - ein Umbenennen haette jede laufende Anmeldung dieses Benutzers ungueltig gemacht, und deshalb war es gar nicht erst vorgesehen. Beim Anmeldedienst geaenderte Namen liessen sich in Vorrania folglich nirgends nachziehen. Neue Token tragen die ID (sub = "<id>", zusaetzlich uid), und deps.py loest die Sitzung darueber auf. Alles andere haengt ohnehin schon an der ID: Bewegungen, Dashboards, API-Tokens. Der Name ist damit reine Anzeige. Benutzer lassen sich jetzt unter "Benutzer" umbenennen - auch sich selbst, ohne aus der Sitzung zu fliegen. Der Anmeldeweg selbst sucht weiterhin ueber den Namen; das ist das Feld im Formular und soll so bleiben. Bestandsschutz: Token ohne uid werden weiter ueber den Namen aufgeloest, damit beim Update niemand abgemeldet wird. Der Zweig darf nach Ablauf der Token-Lebensdauer weg. Das zusaetzliche uid unterscheidet alt von neu eindeutig - ein Benutzer koennte schlicht "42" heissen. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
73 lines
2.4 KiB
Python
73 lines
2.4 KiB
Python
from fastapi import Depends, HTTPException, status
|
||
from fastapi.security import OAuth2PasswordBearer
|
||
from sqlalchemy.orm import Session
|
||
|
||
from datetime import datetime, timezone
|
||
|
||
from .database import get_db
|
||
from .models import ApiToken, Role, User
|
||
from .security import decode_access_token, hash_api_token
|
||
|
||
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="auth/login")
|
||
|
||
_credentials_exc = HTTPException(
|
||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||
detail="Ungültige oder abgelaufene Anmeldung",
|
||
headers={"WWW-Authenticate": "Bearer"},
|
||
)
|
||
|
||
|
||
def _nutzer_aus_payload(db: Session, payload: dict) -> User | None:
|
||
"""Loest die Sitzung auf – bevorzugt ueber die Benutzer-ID.
|
||
|
||
Der Benutzername ist Anzeige und darf sich aendern; die ID nicht. Deshalb
|
||
benennt ein neues Token den Benutzer ueber ``uid``.
|
||
"""
|
||
uid = payload.get("uid")
|
||
if uid is not None:
|
||
try:
|
||
return db.get(User, int(uid))
|
||
except (TypeError, ValueError):
|
||
return None
|
||
|
||
# Token aus der Zeit vor der Umstellung (``sub`` = Benutzername). Laeuft
|
||
# spaetestens nach jwt_expire_minutes aus; dieser Zweig darf danach weg.
|
||
username = payload.get("sub")
|
||
if not username:
|
||
return None
|
||
return db.query(User).filter(User.username == username).first()
|
||
|
||
|
||
def get_current_user(
|
||
token: str = Depends(oauth2_scheme), db: Session = Depends(get_db)
|
||
) -> User:
|
||
"""Akzeptiert ein JWT aus dem Login oder ein langlebiges API-Token."""
|
||
payload = decode_access_token(token)
|
||
if payload is not None:
|
||
user = _nutzer_aus_payload(db, payload)
|
||
if user is None:
|
||
raise _credentials_exc
|
||
return user
|
||
|
||
# Kein gültiges JWT – als API-Token prüfen (z.B. Home Assistant).
|
||
entry = (
|
||
db.query(ApiToken).filter(ApiToken.token_hash == hash_api_token(token)).first()
|
||
)
|
||
if entry is None:
|
||
raise _credentials_exc
|
||
user = db.get(User, entry.user_id) if entry.user_id else None
|
||
if user is None:
|
||
raise _credentials_exc
|
||
entry.last_used_at = datetime.now(timezone.utc)
|
||
db.commit()
|
||
return user
|
||
|
||
|
||
def require_admin(current_user: User = Depends(get_current_user)) -> User:
|
||
if current_user.role != Role.admin:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_403_FORBIDDEN,
|
||
detail="Nur Administratoren dürfen diese Aktion ausführen",
|
||
)
|
||
return current_user
|