faa9d6a25d
Die Einkaufsliste war ueber /shopping-list und /shopping-list/groups schon abrufbar, aber nur mit einem Login-JWT, das nach 7 Tagen ablaeuft - fuer HA unpraktisch. - Neue Tabelle api_tokens (Name, SHA-256-Hash, Ersteller, zuletzt genutzt). Der Klartext wird nur einmal beim Anlegen zurueckgegeben. - get_current_user akzeptiert jetzt JWT ODER API-Token im Authorization-Header; bei Nutzung wird last_used_at fortgeschrieben. - Endpunkte /api-tokens (Admin): auflisten, erzeugen, widerrufen. - Einstellungsseite: Tokens verwalten inkl. einmaliger Anzeige und einem aufklappbaren Konfigurationsbeispiel fuer Home Assistant. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
55 lines
1.7 KiB
Python
55 lines
1.7 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 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:
|
||
username = payload.get("sub")
|
||
if not username:
|
||
raise _credentials_exc
|
||
user = db.query(User).filter(User.username == username).first()
|
||
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
|