# Copyright 2025 Softwell S.r.l.
# Licensed under the Apache License, Version 2.0
"""Authentication mixin for AsgiServer.
Provides server.authenticate(scope) as the single extension point for auth.
The whole backend logic (bearer, basic, JWT) lives in authentication.AuthCore;
this mixin binds it to the server class and adds the local identity source.
When the server carries a ``user_store`` (``server(users=True)``), both
credential paths consult the store FIRST, then fall back to the config ``basic``
dict (bootstrap/legacy compat). The store's tags enter the auth result exactly
where the config tags do, so ``/_server/login`` and Basic-auth requests both work
with store users unchanged. AuthCore itself stays config-only: it is also used
standalone by AuthMiddleware, which has no store.
Usage:
class MyServer(BasicAuthMixin, RoutingClass):
def __init__(self):
BasicAuthMixin.__init__(self, bearer={...}, basic={...})
...
# In middleware or dispatcher:
auth = server.authenticate(scope)
"""
from __future__ import annotations
import base64
import binascii
from typing import Any
from ..authentication import AuthCore
__all__ = ["BasicAuthMixin"]
[docs]
class BasicAuthMixin(AuthCore):
"""Server-side auth mixin: bearer, basic, JWT, plus the local user store.
Backend configuration and verification are inherited from AuthCore. The two
``basic`` paths (the login helper and the Authorization-header dispatch) are
extended here so a ``user_store`` user authenticates as well, checked before
the config dict.
"""
__slots__ = ()
@property
def _store(self) -> Any:
"""The server's user store, or None when it has none (standalone mixin)."""
return getattr(self, "user_store", None)
[docs]
def verify_credentials(self, username: str, password: str) -> dict[str, Any] | None:
"""Verify username/password against the user store first, then config.
Args:
username: The username to verify.
password: The password to verify.
Returns:
Auth dict ``{"identity": ..., "tags": ...}`` if valid, None otherwise.
"""
store = self._store
if store is not None:
record = store.verify(username, password)
if record is not None:
return {"identity": record["username"], "tags": record.get("tags", [])}
return super().verify_credentials(username, password)
def _auth_basic(self, *, credentials: str, **kw: Any) -> dict[str, Any] | None:
"""Authenticate HTTP Basic credentials, store users before config users.
Args:
credentials: Base64-encoded "username:password" from the header.
**kw: Additional args from dynamic dispatch (unused).
Returns:
Auth dict if valid, None if the credentials match neither source.
"""
store = self._store
if store is not None:
decoded = self._decode_basic(credentials)
if decoded is not None:
username, password = decoded
record = store.verify(username, password)
if record is not None:
return {
"identity": record["username"],
"tags": record.get("tags", []),
"backend": "basic",
}
return super()._auth_basic(credentials=credentials, **kw)
def _decode_basic(self, credentials: str) -> tuple[str, str] | None:
"""Decode a Basic header blob to ``(username, password)`` or None if malformed."""
try:
raw = base64.b64decode(credentials).decode()
except (binascii.Error, UnicodeDecodeError):
return None
if ":" not in raw:
return None
username, password = raw.split(":", 1)
return username, password
if __name__ == "__main__":
pass