forked from cartsnitch/cartsnitch
Coupon Carl
27fe957074
Consolidate API gateway service into monorepo. Squashed from https://github.com/cartsnitch/api main (89bacb1). Co-Authored-By: Paperclip <noreply@paperclip.ing>
40 lines
1.1 KiB
Python
40 lines
1.1 KiB
Python
"""FastAPI dependency injection for authentication."""
|
|
|
|
from uuid import UUID
|
|
|
|
from fastapi import Depends, Header, HTTPException, status
|
|
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
|
|
|
|
from cartsnitch_api.auth.jwt import decode_token
|
|
from cartsnitch_api.config import settings
|
|
|
|
bearer_scheme = HTTPBearer()
|
|
|
|
|
|
async def get_current_user(
|
|
credentials: HTTPAuthorizationCredentials = Depends(bearer_scheme),
|
|
) -> UUID:
|
|
try:
|
|
payload = decode_token(credentials.credentials)
|
|
except ValueError:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Invalid or expired token",
|
|
) from None
|
|
|
|
if payload.get("type") != "access":
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Invalid token type",
|
|
) from None
|
|
|
|
return UUID(payload["sub"])
|
|
|
|
|
|
async def verify_service_key(x_service_key: str = Header()) -> None:
|
|
if x_service_key != settings.service_key:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Invalid service key",
|
|
)
|