from __future__ import annotations
from abc import ABCMeta
from abc import abstractmethod
from collections.abc import Iterable
from typing import ClassVar
from typing import TYPE_CHECKING
from falcon.asgi import Request as AsyncRequest
from .exc import BackendNotApplicable
from .utils import await_
from .utils import greenlet_spawn
if TYPE_CHECKING:
from falcon import Request
[docs]
class Getter(metaclass=ABCMeta):
"""Represents a class that extracts authentication information from a request.
Note:
Subclasses that wish to only support the :meth:`.load_async` method are also
required to override the :meth:`.load` method since it is defined as abstract.
In these cases the sync version may just raise an exception.
"""
async_calls_sync_load: ClassVar[bool | None] = None
"""Indicates if this Getter has an async load implementation that is not just a fallback to
sync :meth:`.load` method, like the default :meth:`.load_async` method.
This property is automatically set by the :class:`Getter` when a subclass is defined
(using ``__init_subclass__``) if not specified directly by a subclass
(by setting it to a valued different than ``None``).
"""
def __init_subclass__(cls) -> None:
# Use dict instead of accessing it directly to properly control nested subclasses
if cls.__dict__.get("async_calls_sync_load") is None:
cls.async_calls_sync_load = cls.load_async == Getter.load_async
[docs]
@abstractmethod
def load(self, req: Request, *, challenges: Iterable[str] | None = None) -> str:
"""Loads the specified attribute from the provided request.
If a getter cannot be used with the current request, a :class:`~.BackendNotApplicable`
is raised. The ``challenges``, when provided, will be added to ``WWW-Authenticate`` header
in case of error.
Args:
req (Request): The current request. This may be a wsgi or an asgi falcon request.
Keyword Args:
challenges (Optional[Iterable[str]], optional): One or more authentication challenges
to use as the value of the ``WWW-Authenticate`` header in case of errors.
Returns:
str: The loaded data, in case of success.
"""
[docs]
async def load_async(
self, req: AsyncRequest, *, challenges: Iterable[str] | None = None
) -> str:
"""Async version of :meth:`.load`.
The default implementation simply calls :meth:`.load`, but subclasses may override
this implementation to provide an async version.
"""
return self.load(req, challenges=challenges)
[docs]
class ParamGetter(Getter):
"""Returns the specified parameter from the request.
If the parameter appears multiple times an error will be raised.
Note:
When the falcon ``Request`` option ``RequestOptions.auto_parse_form_urlencoded`` is set
to ``True``, this getter can also retrieve parameter in the body of a
``form-urlencoded`` request.
Args:
param_name (str): the name of the param to load.
"""
def __init__(self, param_name: str):
self.param_name = param_name
[docs]
def load(self, req: Request, *, challenges: Iterable[str] | None = None) -> str:
"""Loads the parameter from the provided request"""
param_value = req.get_param_as_list(self.param_name)
if not param_value:
raise BackendNotApplicable(
description=f"Missing {self.param_name} parameter", challenges=challenges
)
if len(param_value) > 1:
raise BackendNotApplicable(
description=f"Invalid {self.param_name} parameter: Multiple value passed",
challenges=challenges,
)
return param_value[0]
[docs]
class CookieGetter(Getter):
"""Returns the specified cookie from the request.
If the cookie appears multiple times an error will be raised.
Args:
cookie_name (str): the name of the cookie to load.
"""
def __init__(self, cookie_name: str):
self.cookie_name = cookie_name
[docs]
def load(self, req: Request, *, challenges: Iterable[str] | None = None) -> str:
"""Loads the cookie from the provided request"""
cookie_value = req.get_cookie_values(self.cookie_name)
if not cookie_value:
raise BackendNotApplicable(
description=f"Missing {self.cookie_name} cookie", challenges=challenges
)
if len(cookie_value) > 1:
raise BackendNotApplicable(
description=f"Invalid {self.cookie_name} cookie: Multiple value passed",
challenges=challenges,
)
return cookie_value[0]
[docs]
class MultiGetter(Getter):
"""Combines multiple getters. This is useful if a value can be passed in multiple ways
to the server, like using an header or a query parameter.
Will use the first value successfully returned, ignoring all :class:`~.BackendNotApplicable`
exceptions raised by the previously tried getters. If no getter can return a valid value
an exception will only be raised.
Args:
getters (Iterable[Getter]): The getters to use. They will be tried in order and the first
value successfully returned is used.
"""
async_calls_sync_load = True
def __init__(self, getters: Iterable[Getter]):
self.getters = tuple(getters)
if len(self.getters) < 2:
raise ValueError("Must pass more than one getter")
if any(not isinstance(g, Getter) for g in self.getters):
raise TypeError("All getter must inherit from Getter")
[docs]
def load(self, req: Request, *, challenges: Iterable[str] | None = None) -> str:
"""Loads the value from the provided request using the provided getters"""
is_async = isinstance(req, AsyncRequest)
for g in self.getters:
try:
if is_async and not g.async_calls_sync_load:
return await_(g.load_async(req, challenges=challenges)) # type: ignore[arg-type]
else:
return g.load(req, challenges=challenges)
except BackendNotApplicable:
pass
raise BackendNotApplicable(
description="No authentication information found", challenges=challenges
)
[docs]
async def load_async(
self, req: AsyncRequest, *, challenges: Iterable[str] | None = None
) -> str:
"""Async version of :meth:`.load`.
Makes sure ``load`` is called inside a greenlet spawn context
"""
return await greenlet_spawn(self.load, req, challenges=challenges)