Die Anmeldung galt fest 7 Tage - und die Frist lief ab dem *Anmelden*, nicht ab der letzten Nutzung. Wer die App taeglich benutzte, flog trotzdem punktgenau nach einer Woche hinaus. Beides ist jetzt geregelt: - Einstellungen -> Anmeldung: 7 Tage bis 1 Jahr oder unbegrenzt. Gilt fuer Website und App gemeinsam, weil beide dasselbe Token bekommen - auch nach Single Sign-on. - POST /auth/refresh schiebt die Frist beim Start nach vorn. Damit zaehlt die eingestellte Dauer ab dem letzten Besuch, und eine geaenderte Einstellung greift ohne Neuanmeldung. "Unbegrenzt" heisst ein Token ganz ohne exp. Das ist bewusst folgenreich: Es gibt keine Sperrliste, eine erteilte Anmeldung laesst sich nicht einzeln zurueckziehen. Die Oberflaeche sagt das dazu und nennt den einzigen Ausweg (JWT_SECRET aendern). /auth/refresh nimmt bewusst nur JWTs an, keine langlebigen API-Token: aus einem jederzeit widerrufbaren Zugang (Home Assistant) waere sonst ein unwiderruflicher geworden. Der Rueckfallwert aus der Umgebung steigt von 7 auf 30 Tage - er gilt nur noch, solange nichts eingestellt ist. 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
|