2026-07-05 22:18:00 +02:00
|
|
|
"""Token-basierte Auth."""
|
|
|
|
|
from fastapi import Header, HTTPException, Depends
|
|
|
|
|
from typing import Optional
|
|
|
|
|
import aiosqlite
|
|
|
|
|
|
|
|
|
|
from config import settings
|
|
|
|
|
from db import validate_session, get_db
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
async def get_current_user(
|
|
|
|
|
authorization: Optional[str] = Header(None),
|
|
|
|
|
db: aiosqlite.Connection = Depends(get_db),
|
|
|
|
|
) -> str:
|
|
|
|
|
"""Validiert Token und gibt User-ID zurück."""
|
|
|
|
|
if not authorization:
|
|
|
|
|
raise HTTPException(status_code=401, detail="Missing Authorization header")
|
|
|
|
|
|
2026-07-06 07:10:35 +02:00
|
|
|
# Case-insensitive Bearer prefix handling (C3)
|
|
|
|
|
auth_value = authorization.strip()
|
|
|
|
|
if auth_value.lower().startswith("bearer "):
|
|
|
|
|
token = auth_value[7:].strip()
|
|
|
|
|
else:
|
|
|
|
|
token = auth_value
|
2026-07-05 22:18:00 +02:00
|
|
|
|
|
|
|
|
# Static-Auth-Token als Master-Key (für Setup/Bootstrap)
|
|
|
|
|
if token == settings.AUTH_TOKEN:
|
|
|
|
|
return "admin"
|
|
|
|
|
|
|
|
|
|
# Session-Token aus DB
|
|
|
|
|
user_id = await validate_session(db, token)
|
|
|
|
|
if not user_id:
|
|
|
|
|
raise HTTPException(status_code=401, detail="Invalid or expired token")
|
|
|
|
|
|
|
|
|
|
return user_id
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
async def require_admin(user_id: str = Depends(get_current_user)) -> str:
|
|
|
|
|
"""Erfordert Admin-User."""
|
|
|
|
|
if user_id != "admin":
|
|
|
|
|
raise HTTPException(status_code=403, detail="Admin required")
|
|
|
|
|
return user_id
|
