from abc import abstractmethod
from base64 import b64decode, b64encode
from typing import TYPE_CHECKING, Final, Self
from django.views.decorators.debug import sensitive_variables
from typing_extensions import override
from dmr.exceptions import NotAuthenticatedError
from dmr.metadata import EndpointMetadata
from dmr.openapi.objects import Reference, SecurityRequirement, SecurityScheme
from dmr.security.base import AsyncAuth, SyncAuth
if TYPE_CHECKING:
from dmr.controller import Controller
from dmr.endpoint import Endpoint
from dmr.serializer import BaseSerializer
# Default protection space that we advertise in `WWW-Authenticate`:
_DEFAULT_BASIC_REALM: Final = 'api'
class _HttpBasicAuth: # noqa: WPS214
"""
Shared parts of the sync and async http basic auth.
.. versionchanged:: 0.15.0
Added the ``auth_scheme`` prefix, which is required by default.
See :class:`HttpBasicSyncAuth` and :class:`HttpBasicAsyncAuth`.
"""
__slots__ = (
'auth_scheme',
'header',
'realm',
'security_scheme_name',
'www_authenticate',
)
def __init__( # noqa: WPS211
self,
*,
security_scheme_name: str = 'http_basic',
header: str = 'Authorization',
auth_scheme: str = 'Basic',
www_authenticate: bool = True,
realm: str = _DEFAULT_BASIC_REALM,
) -> None:
"""
Apply possible customizations.
- *security_scheme_name* is the name
used in the OpenAPI security scheme map
- *header* selects the header to read the credentials from
- *auth_scheme* is the prefix that the header value must start with,
it is matched exactly, so ``Basic`` won't accept ``basic``.
Pass an empty string to read the credentials without any prefix
- *www_authenticate* controls whether ``401`` responses
advertise this auth in the ``WWW-Authenticate`` header.
Turn it off to stop browsers from showing
their native login prompt for this API.
- *realm* names the protection space in that challenge.
:rfc:`7617#section-2` requires it for the ``Basic`` scheme.
"""
self.security_scheme_name = security_scheme_name
self.header = header
self.auth_scheme = auth_scheme
self.www_authenticate = www_authenticate
self.realm = realm
@property
def www_authenticate_challenge(self) -> str | None:
"""
Challenge for the ``Basic`` scheme.
Returns ``None`` for a custom *header* or *auth_scheme*,
because a challenge can only ask the client
for the ``Basic`` prefix in the ``Authorization`` header.
"""
if (
not self.www_authenticate
or not self._uses_standard_http_basic_auth()
):
return None
# `charset` is defined by RFC 7617 and tells the client to encode
# credentials as UTF-8, which is what we decode them as
# in `_get_username_and_password`.
return f'Basic realm={_quote_auth_param(self.realm)}, charset="UTF-8"'
def security_schemes(
self,
metadata: EndpointMetadata,
controller_cls: type['Controller[BaseSerializer]'],
) -> dict[str, 'SecurityScheme | Reference']:
"""Provides a security schema definition."""
if self._uses_standard_http_basic_auth():
return {
self.security_scheme_name: SecurityScheme(
type='http',
scheme='basic',
description='Http Basic auth',
),
}
return {
self.security_scheme_name: SecurityScheme(
type='apiKey',
name=self.header,
security_scheme_in='header',
description=self._get_custom_security_scheme_description(),
),
}
def security_requirements(
self,
metadata: EndpointMetadata,
controller_cls: type['Controller[BaseSerializer]'],
) -> list[SecurityRequirement]:
"""Provides a security schema usage requirement."""
return [{self.security_scheme_name: []}]
@sensitive_variables()
def _get_username_and_password(
self,
controller: 'Controller[BaseSerializer]',
) -> tuple[str, str] | None:
# We return `None` here, because it might be some other auth.
# We don't want to falsely trigger any errors just yet.
header = controller.request.headers.get(self.header)
if not header:
return None
encoded = self._split_encoded_credentials(header)
if encoded is None:
return None
# After this point we are sure that these are basic auth credentials.
# So, broken ones are an error and not a reason to try other authes.
try:
username, password = b64decode(encoded).decode().split(':', 1)
except Exception:
raise NotAuthenticatedError from None
return username, password
def _split_encoded_credentials(self, header: str) -> str | None:
"""Splits string like 'Basic credentials' and returns 'credentials'."""
if not self.auth_scheme: # Empty scheme means "no prefix at all".
return header
parts = header.split(' ')
if len(parts) != 2 or parts[0] != self.auth_scheme:
# This header does not belong to us: it can be a header
# of any other auth from the chain. Raising here would fail
# the whole request and would not let the others even try.
# So, we return `None` and the next auth gets its chance.
return None
return parts[1]
def _uses_standard_http_basic_auth(self) -> bool:
"""Whether the auth contract matches OpenAPI HTTP basic auth."""
return self.header == 'Authorization' and self.auth_scheme == 'Basic'
def _get_custom_security_scheme_description(self) -> str:
"""Describe non-standard basic auth header contracts."""
# Empty `auth_scheme` means that the header carries
# the credentials alone, without a prefix and a space after it.
scheme_prefix = f'{self.auth_scheme} ' if self.auth_scheme else ''
return (
'HTTP Basic auth via '
f'`{self.header}` header using '
f'`{scheme_prefix}<base64(username:password)>` format'
)
[docs]
class HttpBasicSyncAuth(_HttpBasicAuth, SyncAuth):
"""
Uses HTTP Basic Auth.
Subclass this type to provide actual username/password
check according to your needs.
This class is used for sync endpoints.
Note that this class does not set ``request.user`` by design.
Because many users might use the same auth parameters.
.. warning::
HTTP Basic Auth is not really secure and should
not be used for anything serious.
Consider using JWT instead.
See also:
https://developer.mozilla.org/en-US/docs/Web/HTTP/Guides/Authentication#basic_authentication_scheme
.. versionchanged:: 0.15.0
The ``auth_scheme`` prefix is now required and configurable,
it is ``Basic`` by default. Header values without a prefix
are only accepted with ``auth_scheme=''``.
"""
__slots__ = ()
[docs]
@override
@sensitive_variables()
def __call__(
self,
endpoint: 'Endpoint',
controller: 'Controller[BaseSerializer]',
) -> Self | None:
"""Does the login routine."""
login_data = self._get_username_and_password(controller)
if login_data is None:
return None
return self.authenticate(endpoint, controller, *login_data)
[docs]
@abstractmethod
def authenticate(
self,
endpoint: 'Endpoint',
controller: 'Controller[BaseSerializer]',
username: str,
password: str,
) -> Self | None:
"""Override this method to provide an actual user/password check."""
raise NotImplementedError
[docs]
class HttpBasicAsyncAuth(_HttpBasicAuth, AsyncAuth):
"""
Uses HTTP Basic Auth.
Subclass this type to provide actual username/password
check according to your needs.
This class is used for async endpoints.
Note that this class does not set ``request.user`` by design.
Because many users might use the same auth parameters.
.. warning::
HTTP Basic Auth is not really secure and should
not be used for anything serious.
Consider using JWT instead.
See also:
https://developer.mozilla.org/en-US/docs/Web/HTTP/Guides/Authentication#basic_authentication_scheme
.. versionchanged:: 0.15.0
The ``auth_scheme`` prefix is now required and configurable,
it is ``Basic`` by default. Header values without a prefix
are only accepted with ``auth_scheme=''``.
"""
__slots__ = ()
[docs]
@override
@sensitive_variables()
async def __call__(
self,
endpoint: 'Endpoint',
controller: 'Controller[BaseSerializer]',
) -> Self | None:
"""Does the login routine."""
login_data = self._get_username_and_password(controller)
if login_data is None:
return None
return await self.authenticate(endpoint, controller, *login_data)
[docs]
@abstractmethod
async def authenticate(
self,
endpoint: 'Endpoint',
controller: 'Controller[BaseSerializer]',
username: str,
password: str,
) -> Self | None:
"""Override this method to provide an actual user/password check."""
raise NotImplementedError
[docs]
def basic_auth(username: str, password: str, *, prefix: str = 'Basic ') -> str:
"""
Return a header value for basic auth for a given *username* and *password*.
The *prefix* must match the ``auth_scheme`` of the auth class
that will read this header, including the trailing space.
.. code:: python
>>> basic_auth('admin', 'pass')
'Basic YWRtaW46cGFzcw=='
>>> basic_auth('admin', 'pass', prefix='Custom ')
'Custom YWRtaW46cGFzcw=='
"""
token = b64encode(f'{username}:{password}'.encode()).decode('utf8')
return f'{prefix}{token}'
def _quote_auth_param(auth_param: str) -> str:
r"""
Return *value* as a ``quoted-string`` auth param, as :rfc:`9110` wants it.
.. code:: python
>>> _quote_auth_param('api')
'"api"'
>>> _quote_auth_param('say "hi"')
'"say \\"hi\\""'
"""
escaped = auth_param.replace('\\', r'\\').replace('"', r'\"')
return f'"{escaped}"'