6c47d3c57d
- Add admin auth dependency and role checks - Expand channel and restaurant API routes - Improve YouTube transcript fetching - Enhance daemon worker with better error handling and scheduling Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
41 lines
1.3 KiB
Python
41 lines
1.3 KiB
Python
"""FastAPI dependencies for authentication."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from fastapi import Header, HTTPException
|
|
|
|
from core.auth import verify_jwt
|
|
|
|
|
|
def get_current_user(authorization: str = Header(None)) -> dict:
|
|
"""Extract and verify Bearer token, return user payload.
|
|
|
|
Raises 401 if token is missing or invalid.
|
|
"""
|
|
if not authorization or not authorization.startswith("Bearer "):
|
|
raise HTTPException(401, "Missing or invalid Authorization header")
|
|
token = authorization.removeprefix("Bearer ").strip()
|
|
try:
|
|
return verify_jwt(token)
|
|
except Exception:
|
|
raise HTTPException(401, "Invalid or expired token")
|
|
|
|
|
|
def get_optional_user(authorization: str = Header(None)) -> dict | None:
|
|
"""Same as get_current_user but returns None if no token."""
|
|
if not authorization or not authorization.startswith("Bearer "):
|
|
return None
|
|
token = authorization.removeprefix("Bearer ").strip()
|
|
try:
|
|
return verify_jwt(token)
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def get_admin_user(authorization: str = Header(None)) -> dict:
|
|
"""Require authenticated admin user. Raises 401/403."""
|
|
user = get_current_user(authorization)
|
|
if not user.get("is_admin"):
|
|
raise HTTPException(403, "관리자 권한이 필요합니다")
|
|
return user
|