Hitonabi
ba2429612a
- Doppelte Arcane-Einträge behoben (rippy statt Rippy) - Import-Fixes: relative → absolute imports in main.py, cache/__init__.py, prescan/__init__.py, clients/__init__.py - Neue Dateien: cache.py, prescan.py, Settings.tsx - API-Port 8000 in docker-compose.yml gemappt - UI mit Sidebar, Dark Mode, Einstellungen-Tabs Fixes: #2978 (doppelte Einträge), #2888 (Import-Fehler)
99 lines
2.7 KiB
Python
99 lines
2.7 KiB
Python
"""JWT-Auth-Module für Rippy API."""
|
|
|
|
import secrets
|
|
import time
|
|
from datetime import datetime, timedelta
|
|
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
|
|
SECRET_KEY = settings.jwt_secret_key or secrets.token_urlsafe(32)
|
|
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 (15 min)."""
|
|
to_encode = data.copy()
|
|
if expires_delta:
|
|
expire = datetime.utcnow() + expires_delta
|
|
else:
|
|
expire = datetime.utcnow() + timedelta(minutes=15)
|
|
|
|
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.utcnow() + timedelta(days=7)
|
|
|
|
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."""
|
|
try:
|
|
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
|
|
return payload
|
|
except jwt.ExpiredSignatureError:
|
|
return None
|
|
except jwt.InvalidTokenError:
|
|
return None
|
|
|
|
|
|
def is_access_token(token: str) -> bool:
|
|
"""Prüfe ob Token ein Access Token ist."""
|
|
payload = decode_token(token)
|
|
return payload and payload.get("type") == "access"
|
|
|
|
|
|
def is_refresh_token(token: str) -> bool:
|
|
"""Prüfe ob Token ein Refresh Token ist."""
|
|
payload = decode_token(token)
|
|
return payload and payload.get("type") == "refresh"
|
|
|
|
|
|
# In-Memory Token Blacklist für Logout
|
|
token_blacklist: set = set()
|
|
|
|
|
|
def add_to_blacklist(token: str) -> None:
|
|
"""Füge Token zur Blacklist hinzu."""
|
|
payload = decode_token(token)
|
|
if payload:
|
|
token_blacklist.add(token)
|
|
|
|
|
|
def is_blacklisted(token: str) -> bool:
|
|
"""Prüfe ob Token auf Blacklist steht."""
|
|
return token in token_blacklist
|
|
|
|
|
|
def cleanup_blacklist() -> None:
|
|
"""Räume alte Token von Blacklist."""
|
|
current_time = time.time()
|
|
token_blacklist.clear() # In Produktion mit Redis implementieren
|