mirror of
https://github.com/d3vyce/fastapi-toolsets.git
synced 2026-08-05 08:04:08 +00:00
d3vyceandGitHub
99eabe3b23
* feat: add security module * feat(security): add oauth helpers * docs: add authentication example * fix: cleanup + simplify * docs: update module and reference * fix: multiple security bugs + remove example for now * feat: use async_lru for caching * fix: rename nonce by state_token
56 lines
1.7 KiB
Python
56 lines
1.7 KiB
Python
"""Abstract base class for authentication sources."""
|
|
|
|
import functools
|
|
import inspect
|
|
from abc import ABC, abstractmethod
|
|
from typing import Any, Callable
|
|
|
|
from fastapi import Request
|
|
from fastapi.security import SecurityScopes
|
|
|
|
from fastapi_toolsets.exceptions import UnauthorizedError
|
|
|
|
|
|
def _ensure_async(fn: Callable[..., Any]) -> Callable[..., Any]:
|
|
"""Wrap *fn* so it can always be awaited, caching the coroutine check at init time."""
|
|
if inspect.iscoroutinefunction(fn):
|
|
return fn
|
|
|
|
@functools.wraps(fn)
|
|
async def wrapper(*args: Any, **kwargs: Any) -> Any:
|
|
return fn(*args, **kwargs)
|
|
|
|
return wrapper
|
|
|
|
|
|
class AuthSource(ABC):
|
|
"""Abstract base class for authentication sources."""
|
|
|
|
def __init__(self) -> None:
|
|
"""Set up the default FastAPI dependency signature."""
|
|
source = self
|
|
|
|
async def _call(
|
|
request: Request,
|
|
security_scopes: SecurityScopes, # noqa: ARG001
|
|
) -> Any:
|
|
credential = await source.extract(request)
|
|
if credential is None:
|
|
raise UnauthorizedError()
|
|
return await source.authenticate(credential)
|
|
|
|
self._call_fn: Callable[..., Any] = _call
|
|
self.__signature__ = inspect.signature(_call)
|
|
|
|
@abstractmethod
|
|
async def extract(self, request: Request) -> str | None:
|
|
"""Extract the raw credential from the request without validating."""
|
|
|
|
@abstractmethod
|
|
async def authenticate(self, credential: str) -> Any:
|
|
"""Validate a credential and return the authenticated identity."""
|
|
|
|
async def __call__(self, **kwargs: Any) -> Any:
|
|
"""FastAPI dependency dispatch."""
|
|
return await self._call_fn(**kwargs)
|