37 lines
1.1 KiB
Python
37 lines
1.1 KiB
Python
|
"""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")
|
||
|
|
|
||
|
|
token = authorization.replace("Bearer ", "").strip()
|
||
|
|
|
||
|
|
# 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
|