"""Session auth. Identity always comes from the session, never from a client profile header.""" from __future__ import annotations import secrets from datetime import datetime, timedelta from typing import Optional import bcrypt from fastapi import Header, HTTPException from db import get_db, row_to_dict def hash_password(password: str) -> str: return bcrypt.hashpw(password.encode(), bcrypt.gensalt()).decode() def verify_password(password: str, stored_hash: str) -> bool: if not stored_hash: return False try: return bcrypt.checkpw(password.encode(), stored_hash.encode()) except Exception: return False def make_token() -> str: return secrets.token_urlsafe(32) def get_session(token: str) -> dict | None: if not token: return None with get_db() as conn: row = conn.execute( """ SELECT s.token, s.profile_id, s.expires_at, p.role, p.name, p.email FROM sessions s JOIN profiles p ON p.id = s.profile_id WHERE s.token = ? AND s.expires_at > datetime('now') """, (token,), ).fetchone() return row_to_dict(row) def require_auth(x_auth_token: Optional[str] = Header(default=None, alias="X-Auth-Token")): session = get_session(x_auth_token) if not session: raise HTTPException(401, "Nicht eingeloggt") return session def require_admin_dep(x_auth_token: Optional[str] = Header(default=None, alias="X-Auth-Token")): session = get_session(x_auth_token) if not session: raise HTTPException(401, "Nicht eingeloggt") if session.get("role") != "admin": raise HTTPException(403, "Keine Berechtigung") return session def create_session(profile_id: str, session_days: int = 30) -> tuple[str, datetime]: token = make_token() expires = datetime.utcnow() + timedelta(days=session_days) with get_db() as conn: conn.execute( "INSERT INTO sessions (token, profile_id, expires_at) VALUES (?, ?, ?)", (token, profile_id, expires.isoformat(sep=" ")), ) return token, expires