Hitonabi
0ea5f10b31
Ampel / ampel (push) Failing after 12m31s
- Metadaten-Preview WIEDERHERGESTELLT (Stub ueberschattete echte prescan-Implementierung), tote Altmodule geloescht - Celery update_state statt Phantom-Task/erfundener API - abcde-Kommando korrigiert (CD-Ripping war nie funktionsfaehig) - JWT: fester Schluessel Pflicht, echtes Logout, Cleanup nur Abgelaufene - main.py: crashende Endpoints (Path/secrets/api_keys), year-Bug, Admin-Login aus .env - Ruff gruen (29 Funde), Tests: auth/cache_keys/ripping_helpers, Placebo-test_health raus - SAVEPOINT: offene MakeMKV-Entscheidung SICHTBAR gemacht (Regel B)
106 lines
3.5 KiB
Python
106 lines
3.5 KiB
Python
"""JWT-Auth-Module für Rippy API."""
|
|
|
|
import time
|
|
from datetime import datetime, timedelta, timezone
|
|
from typing import Dict, Optional
|
|
|
|
import jwt
|
|
from passlib.context import CryptContext
|
|
|
|
from config import settings
|
|
|
|
# Passwort-Hashing-Kontext
|
|
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
|
|
|
|
# Geheimer Schlüssel für JWT — MUSS konfiguriert sein (.env: JWT_SECRET_KEY).
|
|
# Review-Fund 22.07.: der frühere Zufalls-Fallback erzeugte PRO PROZESS einen
|
|
# neuen Schlüssel → jeder Neustart/zweite Worker invalidierte alle Tokens.
|
|
# Lieber laut scheitern als still kaputt sein.
|
|
if not settings.jwt_secret_key:
|
|
raise RuntimeError(
|
|
"JWT_SECRET_KEY ist nicht gesetzt (.env). Ohne festen Schlüssel wären "
|
|
"alle Tokens nach jedem Neustart ungültig — Start verweigert."
|
|
)
|
|
SECRET_KEY = settings.jwt_secret_key
|
|
ALGORITHM = "HS256"
|
|
|
|
# Token-Lifetimes
|
|
ACCESS_TOKEN_EXPIRE_MINUTES = 15
|
|
REFRESH_TOKEN_EXPIRE_DAYS = 7
|
|
|
|
|
|
def verify_password(plain_password: str, hashed_password: str) -> bool:
|
|
"""Verifiziere Passwort."""
|
|
return pwd_context.verify(plain_password, hashed_password)
|
|
|
|
|
|
def get_password_hash(password: str) -> str:
|
|
"""Hash Passwort."""
|
|
return pwd_context.hash(password)
|
|
|
|
|
|
def create_access_token(data: Dict, expires_delta: timedelta = None) -> str:
|
|
"""Erstelle Access Token (Default 15 min)."""
|
|
to_encode = data.copy()
|
|
delta = expires_delta or timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
|
|
expire = datetime.now(timezone.utc) + delta
|
|
to_encode.update({"exp": expire, "type": "access"})
|
|
return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
|
|
|
|
|
|
def create_refresh_token(data: Dict) -> str:
|
|
"""Erstelle Refresh Token (7 Tage)."""
|
|
to_encode = data.copy()
|
|
expire = datetime.now(timezone.utc) + timedelta(days=REFRESH_TOKEN_EXPIRE_DAYS)
|
|
to_encode.update({"exp": expire, "type": "refresh"})
|
|
return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
|
|
|
|
|
|
def decode_token(token: str) -> Optional[Dict]:
|
|
"""Dekodiere Token (None bei abgelaufen/ungültig)."""
|
|
try:
|
|
return jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
|
|
except jwt.ExpiredSignatureError:
|
|
return None
|
|
except jwt.InvalidTokenError:
|
|
return None
|
|
|
|
|
|
def is_access_token(token: str) -> bool:
|
|
"""Prüfe ob Token ein gültiges Access Token ist."""
|
|
payload = decode_token(token)
|
|
return bool(payload and payload.get("type") == "access")
|
|
|
|
|
|
def is_refresh_token(token: str) -> bool:
|
|
"""Prüfe ob Token ein gültiges Refresh Token ist."""
|
|
payload = decode_token(token)
|
|
return bool(payload and payload.get("type") == "refresh")
|
|
|
|
|
|
# Token-Blacklist für Logout: Token → Ablauf-Zeitstempel (exp).
|
|
# Bewusste MVP-Grenze: in-memory = pro Prozess (siehe SAVEPOINT.md).
|
|
# Review-Fund 22.07.: das frühere cleanup löschte die GESAMTE Blacklist —
|
|
# Logout war ein Placebo. Jetzt fliegen nur abgelaufene Tokens raus
|
|
# (die sind eh ungültig, decode_token lehnt sie ab).
|
|
token_blacklist: Dict[str, float] = {}
|
|
|
|
|
|
def add_to_blacklist(token: str) -> None:
|
|
"""Füge gültigen Token zur Blacklist hinzu (bis zu seinem Ablauf)."""
|
|
payload = decode_token(token)
|
|
if payload:
|
|
token_blacklist[token] = float(payload.get("exp", time.time()))
|
|
|
|
|
|
def is_blacklisted(token: str) -> bool:
|
|
"""Prüfe ob Token auf der Blacklist steht."""
|
|
return token in token_blacklist
|
|
|
|
|
|
def cleanup_blacklist() -> None:
|
|
"""Entferne NUR abgelaufene Tokens von der Blacklist."""
|
|
now = time.time()
|
|
for token in [t for t, exp in token_blacklist.items() if exp <= now]:
|
|
del token_blacklist[token]
|