# Copyright 2025 Softwell S.r.l.
# Licensed under the Apache License, Version 2.0
"""Authentication middleware for ASGI applications.
Supports Bearer tokens, Basic auth, and JWT with O(1) lookup at request time.
Sets scope["auth"] with authentication result for downstream handlers.
The backend logic (config + verification) lives in authentication.AuthCore.
Backends:
bearer: Static token lookup. O(1) via dict.
basic: Username/password. O(1) via base64-encoded key.
jwt: Token verification via pyjwt. Falls back from bearer if not found.
Config:
bearer: Dict of {name: {token: "...", tags: "..."}}
basic: Dict of {username: {password: "...", tags: "..."}}
jwt: Dict of {name: {secret: "...", algorithm: "...", tags: "..."}}
scope["auth"] format:
{"tags": [...], "identity": "...", "backend": "bearer|basic|jwt:name"}
None if no Authorization header present.
Raises:
HTTPException(401): If credentials present but invalid/expired.
Example:
Configure authentication in the config.py recipe::
root.authMiddleware(
bearer={"api_key": {"token": "sk_live_abc123", "tags": "api,read"}},
basic={"admin": {"password": "secret", "tags": "admin"}},
jwt={"internal": {"secret": "my-jwt-secret", "algorithm": "HS256"}},
)
"""
from __future__ import annotations
from typing import TYPE_CHECKING, Any
from . import BaseMiddleware, headers_dict
from ..authentication import AuthCore
if TYPE_CHECKING:
from ..types import ASGIApp, Receive, Scope, Send
__all__ = ["AuthMiddleware"]
[docs]
class AuthMiddleware(AuthCore, BaseMiddleware):
"""Authentication middleware with O(1) credential lookup.
Extracts Authorization header, validates credentials against the
AuthCore backends, and sets scope["auth"] with result.
Attributes:
_auth_config: Dict mapping auth type to credentials dict.
Class Attributes:
middleware_name: "auth" - identifier for config.
middleware_order: 400 - runs after CORS.
middleware_default: False - disabled by default.
"""
middleware_name = "auth"
middleware_order = 400
middleware_default = False
__slots__ = ("_auth_config",)
[docs]
def __init__(self, app: ASGIApp, **entries: Any) -> None:
"""Initialize authentication middleware.
Args:
app: Next ASGI application in the middleware chain.
**entries: Auth configuration by type (bearer, basic, jwt).
Note:
Configuration is processed by AuthCore._configure_{type} methods.
Unknown auth types are silently ignored.
"""
BaseMiddleware.__init__(self, app)
AuthCore.__init__(self, **entries)
@headers_dict
async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
"""Process request with authentication.
For HTTP and WebSocket requests, extracts and validates Authorization
header, setting scope["auth"] and scope["_filters"]["auth_tags"]
with the result.
Delegates to server.authenticate(scope) when server is available
in the middleware chain. Falls back to the inherited authenticate()
when used standalone (no server).
Args:
scope: ASGI scope dictionary.
receive: ASGI receive callable.
send: ASGI send callable.
Note:
Uses @headers_dict decorator to populate scope["_headers"].
Non-HTTP/non-WebSocket requests pass through without auth processing.
"""
if scope["type"] in ("http", "websocket"):
server = self.server
if server is not None:
auth = server.authenticate(scope)
else:
auth = self.authenticate(scope)
scope["auth"] = auth
filters = scope.setdefault("_filters", {})
filters["auth_tags"] = auth["tags"] if auth is not None else []
await self.app(scope, receive, send)
if __name__ == "__main__":
pass