38 lines
1.1 KiB
Python
38 lines
1.1 KiB
Python
from fastapi import Depends, HTTPException, Request, status
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.auth.tokens import hash_token
|
|
from app.db.base import get_db
|
|
from app.db.models import User
|
|
|
|
|
|
def _extract_token(request: Request) -> str | None:
|
|
auth = request.headers.get("Authorization", "")
|
|
if auth.lower().startswith("bearer "):
|
|
token = auth[7:].strip()
|
|
if token:
|
|
return token
|
|
header = request.headers.get("X-API-Token", "").strip()
|
|
if header:
|
|
return header
|
|
query = request.query_params.get("token", "").strip()
|
|
return query or None
|
|
|
|
|
|
def get_current_user(
|
|
request: Request,
|
|
db: Session = Depends(get_db),
|
|
) -> User:
|
|
token = _extract_token(request)
|
|
if not token:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Missing API token")
|
|
|
|
token_hash = hash_token(token)
|
|
user = db.scalar(
|
|
select(User).where(User.api_token_hash == token_hash, User.is_active.is_(True))
|
|
)
|
|
if not user:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid API token")
|
|
return user
|