Authentication and authorization
The asab.web.auth
module provides authentication and
authorization of incoming requests.
This enables your application to differentiate between users,
grant or deny them API access depending on their permissions, and work
with other relevant user data.
The auth
module requires an authorization server to function.
It works best with TeskaLabs Seacat Auth
authorization server
(See its documentation for setup instructions).
Getting started
To get started, add asab.web
module to your application and initialize asab.web.auth.AuthService
:
import asab
import asab.web
import asab.web.auth
...
class MyApplication(asab.Application):
def __init__(self):
super().__init__()
# Initialize web module
asab.web.create_web_server(self)
# Initialize authorization service
self.AuthService = asab.web.auth.AuthService(self)
Note
If your app has more than one web container, you will need to call AuthService.install(web_container)
to apply
the authorization.
Note
You may also need to specify your authorization server's public_keys_url
(also known as jwks_uri
in OAuth 2.0).
In case you don't have any authorization server at hand,
you can run the auth module in "mock mode". See the Configuration section for details.
Every handler in your web server now accepts only requests with a valid authentication.
Unauthenticated requests are automatically answered with
HTTP 401: Unauthorized.
For every authenticated request, an asab.web.auth.Authorization
object is created and stored
in asab.contextvars.Authz
for easy access.
It contains authorization and authentication details, such as CredentialsId
, Username
or Email
, and
access-checking methods has_resource_access
, require_superuser_access
and more (see reference).
import asab.contextvars
import asab.web.rest
...
async def order_breakfast(request):
authz = asab.contextvars.Authz.get()
username = authz.Username
# This will raise asab.exceptions.AccessDeniedError when the user is not authorized for resource `breakfast:access`
authz.require_resource_access("breakfast:access")
print("{} is ordering breakfast.".format(username))
if authz.has_resource_access("breakfast:pancakes"):
print("{} can get pancakes for breakfast!".format(username))
if authz.has_superuser_access():
print("{} can get anything they want!".format(username))
return asab.web.rest.json_response(request, {
"result": "Good morning {}, your breakfast will be ready in a minute!".format(username)
})
See examples/web-auth.py for a full demo ASAB application with auth module.
Multitenancy
If your web container runs in multi-tenant mode, AuthService will ensure the authorization of tenant context.
Read more in the Multitenancy chapter.
Configuration
The asab.web.auth
module is configured in the [auth]
section with the following options:
Option |
Type |
Meaning |
enabled |
production (default), mock or development |
Switches authentication and authorization mode. In mock mode all incoming requests are authorized with mock user info without any communication with the auth server. In development mode the app is expected to run without Nginx reverse proxy and does auth introspection by itself (it is necessary to configure introspection_url ). |
public_keys_url |
URL |
The URL of the authorization server's public keys (also known as jwks_uri in OAuth 2.0 |
introspection_url |
URL |
The URL of Seacat Auth introspection endpoint used in development mode. |
mock_claims_path |
path |
Path to JSON file that contains auth claims (aka User Info) used in mock mode. The structure of the JSON object should follow the OpenID Connect userinfo definition and also contain the resources object. |
Default values:
enabled=production
public_keys_url=
introspection_url=
mock_claims_path=/conf/mock-claims.json
Without any configuration, all incoming requests will be denied with HTTP 401,
since the auth service is in production mode and no public keys are provided.
Production mode
In production mode, the AuthService
uses the authorization server to authenticate and authorize incoming requests.
The application is expected to run behind Nginx reverse proxy with auth introspection enabled.
You need to configure public_keys_url
to point to the auth server's public keys endpoint to be able to verify incoming requests.
[auth]
public_keys_url=https://example.teskalabs.com/.well-known/jwks.json
Internal auth
To use ASAB's cluster-internal authentication (together with the web authentication above),
initialize asab.api.ApiService
and asab.zookeeper.Module
in your application.
This will automatically set up shared keys necessary for internal auth, no extra configuration necessary.
You can also use internal authentication without web authentication by leaving public_keys_url
empty.
Development mode
In development mode, the AuthService
expects the application to run without Nginx reverse proxy and
does auth token introspection by itself.
To use the development mode, set the enabled
option to development
and specify
the introspection_url
to point to the auth server's access token introspection endpoint.
You also need to configure public_keys_url
to the auth server's public keys endpoint.
[auth]
enabled=development
public_keys_url=http://localhost:8900/.well-known/jwks.json
introspection_url=http://localhost:8900/nginx/introspect/openidconnect
Mock mode
In mock mode, the AuthService
authorizes all incoming requests with mock authorization claims (aka User Info) without
any communication with the auth server.
You can also customize the claims by providing a path to a JSON file with mock claims.
The file content should comply with the OpenID Connect userinfo response definition
and also contain the resources
object.
[auth]
enabled=mock
mock_claims_path=./mock-claims.json
Reference
asab.web.auth.AuthService
Bases: Service
Provides authentication and authorization of incoming requests.
Source code in asab/web/auth/service.py
| class AuthService(Service):
"""
Provides authentication and authorization of incoming requests.
"""
def __init__(self, app, service_name="asab.AuthService"):
super().__init__(app, service_name)
if jwcrypto is None:
raise ModuleNotFoundError(
"You are trying to use asab.web.auth module without 'jwcrypto' installed. "
"Please run 'pip install jwcrypto' "
"or install asab with 'authz' optional dependency."
)
self.DiscoveryService = self.App.get_service("asab.DiscoveryService")
self.Providers: list = []
self._set_up_providers()
# Try to auto-install authorization middleware
self._try_auto_install()
def get_provider(self, provider_type: str):
for provider in self.Providers:
if provider.Type == provider_type:
return provider
return None
def get_authorized_tenant(self, request=None) -> typing.Optional[str]:
"""
DEPRECATED. Get the request's authorized tenant.
"""
authz = Authz.get()
resources = authz.get_claim("resources", {})
for tenant in resources.keys():
if tenant == "*":
continue
# Return the first authorized tenant
return tenant
return None
async def initialize(self, app):
self._validate_wrapper_installation()
def is_enabled(self) -> bool:
"""
OBSOLETE. Check if the AuthService is enabled. Mock mode counts as enabled too.
"""
LogObsolete.warning(
"AuthService.is_enabled() is obsolete since it is not possible to disable AuthService anymore.",
struct_data={"eol": "2025-03-31"}
)
return True
def install(self, web_container):
"""
Apply authorization to all web handlers in a web container.
:param web_container: Web container to be protected by authorization.
:type web_container: asab.web.WebContainer
"""
web_service = self.App.get_service("asab.WebService")
# Check that the middleware has not been installed yet
if self.set_up_auth_web_wrapper in web_container.WebApp.on_startup:
if len(web_service.Containers) == 1:
raise RuntimeError(
"WebContainer has authorization middleware installed already. "
"You don't need to call `AuthService.install()` in applications with a single WebContainer; "
"it is called automatically at init time."
)
else:
raise RuntimeError("WebContainer has authorization middleware installed already.")
tenant_service = self.App.get_service("asab.TenantService")
if tenant_service is None:
web_container.WebApp.on_startup.append(self.set_up_auth_web_wrapper)
return
tenant_wrapper_idx = tenant_service.get_web_wrapper_position(web_container)
if tenant_wrapper_idx is not None:
# Tenant wrapper is present - Auth wrapper must be applied before it
web_container.WebApp.on_startup.insert(tenant_wrapper_idx, self.set_up_auth_web_wrapper)
else:
web_container.WebApp.on_startup.append(self.set_up_auth_web_wrapper)
async def authorize_request(self, request: aiohttp.web.Request) -> Authorization:
"""
Authenticate and authorize request with first valid provider
Args:
request (aiohttp.web.Request): Request to authenticate and authorize
Returns:
Authorization object
Raises:
NotAuthenticatedError: When no provider is able to authorize the request
"""
for provider in self.Providers:
try:
return await provider.authorize(request)
except NotAuthenticatedError:
# Provider was unable to authenticate request
# L.debug("Authorization failed.", struct_data={"auth_provider": provider.Type})
pass
L.warning("Cannot authenticate request: No valid authorization provider found.")
raise NotAuthenticatedError()
def _set_up_providers(self):
# Always set up an ID token provider without public keys
# The public keys are set up based on app configuration or added later
from .providers import IdTokenAuthProvider
id_token_provider = IdTokenAuthProvider(self.App)
self.Providers.append(id_token_provider)
public_keys_url = Config.get("auth", "public_keys_url") or None
if public_keys_url:
from .providers.key_providers import UrlPublicKeyProvider
public_key_provider = UrlPublicKeyProvider(self.App, public_keys_url)
id_token_provider.register_key_provider(public_key_provider)
else:
public_key_provider = None
enabled = Config.get("auth", "enabled", fallback="production")
try:
enabled = string_to_boolean(enabled)
except ValueError:
pass
if enabled == "production" or enabled is True:
# Production mode is enabled by default
# The ID token provider is already set up
return
elif enabled == "development":
introspection_url = Config.get("auth", "introspection_url", fallback=None)
if introspection_url is None or public_key_provider is None:
raise ValueError(
"AuthService is in development mode, but introspection_url or public_keys_url is not set."
)
L.warning(
"AuthService is in development mode. "
"Web requests will be authorized with introspection call to {}.".format(introspection_url)
)
from .providers import AccessTokenAuthProvider
provider = AccessTokenAuthProvider(self.App, introspection_url=introspection_url)
provider.register_key_provider(public_key_provider)
self.Providers.append(provider)
elif enabled == "mock":
from .providers import MockAuthProvider
auth_claims_path = (
Config.get("auth", "mock_claims_path", fallback=None)
or Config.get("auth", "mock_user_info_path", fallback=None)
)
provider = MockAuthProvider(self.App, auth_claims_path=auth_claims_path)
# Make sure the mock provider is the first one and catches every request
self.Providers.insert(0, provider)
return
elif enabled is False:
raise ValueError("Disabling AuthService is deprecated. Use mock mode instead.")
else:
raise ValueError("Unsupported auth mode: {!r}".format(enabled))
def _authorize_handler(self, handler):
"""
Authenticate the request by the JWT ID token in the Authorization header.
Extract the token claims into Authorization context so that they can be used for authorization checks.
"""
@functools.wraps(handler)
async def _authorize_handler_wrapper(*args, **kwargs):
request = args[-1]
# Authenticate and authorize request with first valid provider
authz = await self.authorize_request(request)
# Authorize tenant context
tenant = Tenant.get(None)
if tenant is not None:
authz.require_tenant_access()
authz_ctx = Authz.set(authz)
try:
return await handler(*args, **kwargs)
finally:
Authz.reset(authz_ctx)
return _authorize_handler_wrapper
async def set_up_auth_web_wrapper(self, aiohttp_app: aiohttp.web.Application):
"""
Inspect all registered handlers and wrap them in decorators according to their parameters.
"""
for route in aiohttp_app.router.routes():
# Skip non-coroutines
if not inspect.iscoroutinefunction(route.handler):
continue
try:
self._set_handler_auth(route)
except Exception as e:
raise Exception("Failed to initialize auth for handler {!r}.".format(route.handler.__qualname__)) from e
def _set_handler_auth(self, route: aiohttp.web.AbstractRoute):
"""
Inspect handler and apply suitable auth wrappers.
"""
# Extract the actual unwrapped handler method for signature inspection
handler_method = route.handler
# Exclude endpoints with @noauth decorator
if hasattr(handler_method, "NoAuth") and handler_method.NoAuth is True:
return
while hasattr(handler_method, "__wrapped__"):
# While loop unwraps handlers wrapped in multiple decorators.
# NOTE: This requires all the decorators to use @functools.wraps().
handler_method = handler_method.__wrapped__
if hasattr(handler_method, "__func__"):
handler_method = handler_method.__func__
argspec = inspect.getfullargspec(handler_method)
args = set(argspec.kwonlyargs).union(argspec.args)
# Extract the whole handler including its existing decorators and wrappers
handler = route.handler
# Apply the decorators IN REVERSE ORDER (the last applied wrapper affects the request first)
# 2) Pass authorization attributes to handler method
if "resources" in args:
LogObsolete.warning(
"The 'resources' argument is deprecated. "
"Use the access-checking methods of asab.contextvars.Authz instead.",
struct_data={"handler": handler.__qualname__, "eol": "2025-03-01"},
)
handler = _pass_resources(handler)
if "user_info" in args:
LogObsolete.warning(
"The 'user_info' argument is deprecated. "
"Use the Authorization object in asab.contextvars.Authz instead.",
struct_data={"handler": handler.__qualname__, "eol": "2025-03-01"},
)
handler = _pass_user_info(handler)
# 1) Authenticate and authorize request, authorize tenant from context, set Authorization context
handler = self._authorize_handler(handler)
route._handler = handler
def _validate_wrapper_installation(self):
"""
Check if there is at least one web container with authorization installed
"""
web_service = self.App.get_service("asab.WebService")
if web_service is None or len(web_service.Containers) == 0:
L.warning("Authorization is not installed: There are no web containers.")
return
tenant_service = self.App.get_service("asab.TenantService")
auth_wrapper_installed = False
for web_container in web_service.Containers.values():
try:
auth_wrapper_idx = web_container.WebApp.on_startup.index(self.set_up_auth_web_wrapper)
auth_wrapper_installed = True
except ValueError:
# Authorization wrapper not installed here
continue
if tenant_service is None:
# Without tenant service there are no tenant web wrappers
continue
# Ensure the wrappers are applied in the correct order
tenant_wrapper_idx = tenant_service.get_web_wrapper_position(web_container)
if tenant_wrapper_idx is not None and auth_wrapper_idx > tenant_wrapper_idx:
L.error(
"TenantService.install(web_container) must be called before AuthService.install(web_container). "
"Otherwise authorization will not work properly."
)
if not auth_wrapper_installed:
L.warning(
"Authorization is not installed in any web container. "
"In applications with more than one WebContainer there is no automatic installation; "
"you have to call `AuthService.install(web_container)` explicitly."
)
return
def _try_auto_install(self):
"""
If there is exactly one web container, install authorization wrapper on it.
"""
web_service = self.App.get_service("asab.WebService")
if web_service is None:
return
if len(web_service.Containers) != 1:
return
web_container = web_service.WebContainer
self.install(web_container)
L.debug("WebContainer authorization wrapper will be installed automatically.")
|
authorize_request(request)
async
Authenticate and authorize request with first valid provider
Parameters:
Name |
Type |
Description |
Default |
request
|
Request
|
Request to authenticate and authorize
|
required
|
Returns:
Raises:
Type |
Description |
NotAuthenticatedError
|
When no provider is able to authorize the request
|
Source code in asab/web/auth/service.py
| async def authorize_request(self, request: aiohttp.web.Request) -> Authorization:
"""
Authenticate and authorize request with first valid provider
Args:
request (aiohttp.web.Request): Request to authenticate and authorize
Returns:
Authorization object
Raises:
NotAuthenticatedError: When no provider is able to authorize the request
"""
for provider in self.Providers:
try:
return await provider.authorize(request)
except NotAuthenticatedError:
# Provider was unable to authenticate request
# L.debug("Authorization failed.", struct_data={"auth_provider": provider.Type})
pass
L.warning("Cannot authenticate request: No valid authorization provider found.")
raise NotAuthenticatedError()
|
get_authorized_tenant(request=None)
DEPRECATED. Get the request's authorized tenant.
Source code in asab/web/auth/service.py
| def get_authorized_tenant(self, request=None) -> typing.Optional[str]:
"""
DEPRECATED. Get the request's authorized tenant.
"""
authz = Authz.get()
resources = authz.get_claim("resources", {})
for tenant in resources.keys():
if tenant == "*":
continue
# Return the first authorized tenant
return tenant
return None
|
install(web_container)
Apply authorization to all web handlers in a web container.
:param web_container: Web container to be protected by authorization.
:type web_container: asab.web.WebContainer
Source code in asab/web/auth/service.py
| def install(self, web_container):
"""
Apply authorization to all web handlers in a web container.
:param web_container: Web container to be protected by authorization.
:type web_container: asab.web.WebContainer
"""
web_service = self.App.get_service("asab.WebService")
# Check that the middleware has not been installed yet
if self.set_up_auth_web_wrapper in web_container.WebApp.on_startup:
if len(web_service.Containers) == 1:
raise RuntimeError(
"WebContainer has authorization middleware installed already. "
"You don't need to call `AuthService.install()` in applications with a single WebContainer; "
"it is called automatically at init time."
)
else:
raise RuntimeError("WebContainer has authorization middleware installed already.")
tenant_service = self.App.get_service("asab.TenantService")
if tenant_service is None:
web_container.WebApp.on_startup.append(self.set_up_auth_web_wrapper)
return
tenant_wrapper_idx = tenant_service.get_web_wrapper_position(web_container)
if tenant_wrapper_idx is not None:
# Tenant wrapper is present - Auth wrapper must be applied before it
web_container.WebApp.on_startup.insert(tenant_wrapper_idx, self.set_up_auth_web_wrapper)
else:
web_container.WebApp.on_startup.append(self.set_up_auth_web_wrapper)
|
is_enabled()
OBSOLETE. Check if the AuthService is enabled. Mock mode counts as enabled too.
Source code in asab/web/auth/service.py
| def is_enabled(self) -> bool:
"""
OBSOLETE. Check if the AuthService is enabled. Mock mode counts as enabled too.
"""
LogObsolete.warning(
"AuthService.is_enabled() is obsolete since it is not possible to disable AuthService anymore.",
struct_data={"eol": "2025-03-31"}
)
return True
|
set_up_auth_web_wrapper(aiohttp_app)
async
Inspect all registered handlers and wrap them in decorators according to their parameters.
Source code in asab/web/auth/service.py
| async def set_up_auth_web_wrapper(self, aiohttp_app: aiohttp.web.Application):
"""
Inspect all registered handlers and wrap them in decorators according to their parameters.
"""
for route in aiohttp_app.router.routes():
# Skip non-coroutines
if not inspect.iscoroutinefunction(route.handler):
continue
try:
self._set_handler_auth(route)
except Exception as e:
raise Exception("Failed to initialize auth for handler {!r}.".format(route.handler.__qualname__)) from e
|
asab.web.auth.Authorization
Contains authentication and authorization claims (aka UserInfo), provides methods for checking and enforcing access control.
Attributes:
Name |
Type |
Description |
CredentialsId |
str
|
Unique identifier of the authorized entity in the ASAB ecosystem.
Usually corresponds to JWT attribute "sub".
|
Username |
str | None
|
End-user's preferred username.
|
Email |
str | None
|
|
Phone |
str | None
|
|
SessionId |
str
|
Sign-on session identifier.
|
Issuer |
str
|
Unique identifier of the server that issued the authorization.
|
IssuedAt |
datetime
|
Timestamp when the authorization was issued.
|
Expiration |
datetime
|
Timestamp when the authorization expires.
|
Source code in asab/web/auth/authorization.py
| class Authorization:
"""
Contains authentication and authorization claims (aka UserInfo), provides methods for checking and enforcing access control.
Attributes:
CredentialsId (str):
Unique identifier of the authorized entity in the ASAB ecosystem.
Usually corresponds to JWT attribute "sub".
Username (str | None): End-user's preferred username.
Email (str | None): End-user email address.
Phone (str | None): End-user phone number.
SessionId (str): Sign-on session identifier.
Issuer (str): Unique identifier of the server that issued the authorization.
IssuedAt (datetime.datetime): Timestamp when the authorization was issued.
Expiration (datetime.datetime): Timestamp when the authorization expires.
"""
def __init__(self, claims: dict):
"""
Initialize Authorization object from authorization server claims.
Args:
claims (dict): Authorization server claims (from ID token, UserInfo etc.).
"""
# Claims dict should not be accessed directly
self._Claims = claims or {}
self._Resources = self._Claims.get("resources", {})
self.CredentialsId = self._Claims.get("sub")
self.Username = self._Claims.get("preferred_username") or self._Claims.get("username")
self.Email = self._Claims.get("email")
self.Phone = self._Claims.get("phone")
self.SessionId = self._Claims.get("sid")
self.Issuer = self._Claims.get("iss") # Who issued the authorization
try:
self.IssuedAt = datetime.datetime.fromtimestamp(int(self._Claims["iat"]), datetime.timezone.utc)
except KeyError:
L.error("ID token is missing the required 'iat' field.")
raise NotAuthenticatedError()
try:
self.Expiration = datetime.datetime.fromtimestamp(int(self._Claims["exp"]), datetime.timezone.utc)
except KeyError:
L.error("ID token is missing the required 'exp' field.")
raise NotAuthenticatedError()
def __repr__(self):
return "<Authorization [{}cid: {!r}, iat: {!r}, exp: {!r}]>".format(
"SUPERUSER, " if self.has_superuser_access() else "",
self.CredentialsId,
self.IssuedAt.isoformat(),
self.Expiration.isoformat(),
)
def is_valid(self):
"""
Check that the authorization is not expired.
"""
return datetime.datetime.now(datetime.timezone.utc) < self.Expiration
def has_superuser_access(self) -> bool:
"""
Check whether the agent is a superuser.
Returns:
bool: Does the agent have superuser access?
Raises:
NotAuthenticatedError: When the authorization is expired or otherwise invalid.
Examples:
>>> import asab.contextvars
>>> authz = asab.contextvars.Authz.get()
>>> if authz.has_superuser_access():
>>> print("I am a superuser and can do anything!")
>>> else:
>>> print("I am but a mere mortal.")
"""
self.require_valid()
return is_superuser(self._Resources)
def has_resource_access(self, *resources: str) -> bool:
"""
Check whether the agent is authorized to access requested resources.
Args:
*resources (str): A variable number of resource IDs whose authorization is requested.
Returns:
bool: Is the agent authorized to access requested resources?
Raises:
NotAuthenticatedError: When the authorization is expired or otherwise invalid.
Examples:
>>> import asab.contextvars
>>> authz = asab.contextvars.Authz.get()
>>> if authz.has_resource_access("article:read", "article:write"):
>>> print("I can read and write articles!")
>>> else:
>>> print("Not much to do here.")
"""
self.require_valid()
return has_resource_access(self._Resources, resources, tenant=Tenant.get(None))
def has_tenant_access(self) -> bool:
"""
Check whether the agent has access to the tenant in context.
Returns:
bool: Is the agent authorized to access requested tenant?
Raises:
NotAuthenticatedError: When the authorization is expired or otherwise invalid.
Examples:
>>> import asab.contextvars
>>> authz = asab.contextvars.Authz.get()
>>> tenant_ctx = asab.contextvars.Tenant.set("big-corporation")
>>> try:
>>> if authz.has_tenant_access():
>>> print("I have access to Big Corporation!")
>>> else:
>>> print("Not much to do here.")
>>> finally:
>>> asab.contextvars.Tenant.reset(tenant_ctx)
"""
self.require_valid()
try:
tenant = Tenant.get()
except LookupError as e:
raise ValueError("No tenant in context.") from e
return has_tenant_access(self._Resources, tenant)
def require_valid(self):
"""
Ensure that the authorization is not expired.
Raises:
NotAuthenticatedError: When the authorization is expired or otherwise invalid.
"""
if not self.is_valid():
L.warning("Authorization expired.", struct_data={
"cid": self.CredentialsId, "exp": self.Expiration.isoformat()})
raise NotAuthenticatedError()
def require_superuser_access(self):
"""
Ensure that the agent has superuser access.
Raises:
NotAuthenticatedError: When the authorization is expired or otherwise invalid.
AccessDeniedError: When the agent does not have superuser access.
Examples:
>>> import asab.contextvars
>>> authz = asab.contextvars.Authz.get()
>>> authz.require_superuser_access()
>>> print("I am a superuser and can do anything!")
"""
if not self.has_superuser_access():
L.warning("Superuser authorization required.", struct_data={
"cid": self.CredentialsId})
raise AccessDeniedError()
def require_resource_access(self, *resources: str):
"""
Ensure that the agent is authorized to access the specified resources.
Args:
*resources (str): A variable number of resource IDs whose authorization is requested.
Raises:
NotAuthenticatedError: When the authorization is expired or otherwise invalid.
AccessDeniedError: When the agent does not have access to the requested resources.
Examples:
>>> import asab.contextvars
>>> authz = asab.contextvars.Authz.get()
>>> authz.require_resource_access("article:read", "article:write")
>>> print("I can read and write articles!")
"""
if not self.has_resource_access(*resources):
L.warning("Resource authorization required.", struct_data={
"resource": resources, "cid": self.CredentialsId})
raise AccessDeniedError()
def require_tenant_access(self):
"""
Ensures that the agent is authorized to access the tenant in the current context.
Raises:
NotAuthenticatedError: When the authorization is expired or otherwise invalid.
AccessDeniedError: When the agent does not have access to the requested tenant.
Examples:
>>> import asab.contextvars
>>> authz = asab.contextvars.Authz.get()
>>> tenant_ctx = asab.contextvars.Tenant.set("big-corporation")
>>> try:
>>> authz.require_tenant_access()
>>> print("I have access to Big Corporation!")
>>> finally:
>>> asab.contextvars.Tenant.reset(tenant_ctx)
"""
if not self.has_tenant_access():
L.warning("Tenant authorization required.", struct_data={
"tenant": Tenant.get(), "cid": self.CredentialsId})
raise AccessDeniedError()
def user_info(self) -> typing.Dict[str, typing.Any]:
"""
Return OpenID Connect UserInfo claims (or JWToken claims).
NOTE: If possible, use Authz attributes (CredentialsId, Username etc.) instead of inspecting
the claims directly.
Returns:
dict: UserInfo claims
Raises:
NotAuthenticatedError: When the authorization is expired or otherwise invalid.
"""
self.require_valid()
return self._Claims
def get_claim(self, key: str, default=None) -> typing.Any:
"""
Get the value of a token claim.
Args:
key: Claim name.
Returns:
Value of the requested claim (or `None` if the claim is not present).
Raises:
NotAuthenticatedError: When the authorization is expired or otherwise invalid.
"""
self.require_valid()
return self._Claims.get(key, default)
def _resources(self) -> typing.Optional[typing.Set[str]]:
"""
Return the set of authorized resources.
Returns:
set: Authorized resources.
Raises:
NotAuthenticatedError: When the authorization is expired or otherwise invalid.
"""
self.require_valid()
resources = _authorized_resources(self._Resources, Tenant.get(None))
if self.has_superuser_access():
# Ensure superuser resource is present no matter the tenant
resources.add(SUPERUSER_RESOURCE_ID)
return resources
|
__init__(claims)
Initialize Authorization object from authorization server claims.
Parameters:
Name |
Type |
Description |
Default |
claims
|
dict
|
Authorization server claims (from ID token, UserInfo etc.).
|
required
|
Source code in asab/web/auth/authorization.py
| def __init__(self, claims: dict):
"""
Initialize Authorization object from authorization server claims.
Args:
claims (dict): Authorization server claims (from ID token, UserInfo etc.).
"""
# Claims dict should not be accessed directly
self._Claims = claims or {}
self._Resources = self._Claims.get("resources", {})
self.CredentialsId = self._Claims.get("sub")
self.Username = self._Claims.get("preferred_username") or self._Claims.get("username")
self.Email = self._Claims.get("email")
self.Phone = self._Claims.get("phone")
self.SessionId = self._Claims.get("sid")
self.Issuer = self._Claims.get("iss") # Who issued the authorization
try:
self.IssuedAt = datetime.datetime.fromtimestamp(int(self._Claims["iat"]), datetime.timezone.utc)
except KeyError:
L.error("ID token is missing the required 'iat' field.")
raise NotAuthenticatedError()
try:
self.Expiration = datetime.datetime.fromtimestamp(int(self._Claims["exp"]), datetime.timezone.utc)
except KeyError:
L.error("ID token is missing the required 'exp' field.")
raise NotAuthenticatedError()
|
get_claim(key, default=None)
Get the value of a token claim.
Parameters:
Name |
Type |
Description |
Default |
key
|
str
|
|
required
|
Returns:
Type |
Description |
Any
|
Value of the requested claim (or None if the claim is not present).
|
Raises:
Type |
Description |
NotAuthenticatedError
|
When the authorization is expired or otherwise invalid.
|
Source code in asab/web/auth/authorization.py
| def get_claim(self, key: str, default=None) -> typing.Any:
"""
Get the value of a token claim.
Args:
key: Claim name.
Returns:
Value of the requested claim (or `None` if the claim is not present).
Raises:
NotAuthenticatedError: When the authorization is expired or otherwise invalid.
"""
self.require_valid()
return self._Claims.get(key, default)
|
has_resource_access(*resources)
Check whether the agent is authorized to access requested resources.
Parameters:
Name |
Type |
Description |
Default |
*resources
|
str
|
A variable number of resource IDs whose authorization is requested.
|
()
|
Returns:
Name | Type |
Description |
bool |
bool
|
Is the agent authorized to access requested resources?
|
Raises:
Type |
Description |
NotAuthenticatedError
|
When the authorization is expired or otherwise invalid.
|
Examples:
>>> import asab.contextvars
>>> authz = asab.contextvars.Authz.get()
>>> if authz.has_resource_access("article:read", "article:write"):
>>> print("I can read and write articles!")
>>> else:
>>> print("Not much to do here.")
Source code in asab/web/auth/authorization.py
| def has_resource_access(self, *resources: str) -> bool:
"""
Check whether the agent is authorized to access requested resources.
Args:
*resources (str): A variable number of resource IDs whose authorization is requested.
Returns:
bool: Is the agent authorized to access requested resources?
Raises:
NotAuthenticatedError: When the authorization is expired or otherwise invalid.
Examples:
>>> import asab.contextvars
>>> authz = asab.contextvars.Authz.get()
>>> if authz.has_resource_access("article:read", "article:write"):
>>> print("I can read and write articles!")
>>> else:
>>> print("Not much to do here.")
"""
self.require_valid()
return has_resource_access(self._Resources, resources, tenant=Tenant.get(None))
|
has_superuser_access()
Check whether the agent is a superuser.
Returns:
Name | Type |
Description |
bool |
bool
|
Does the agent have superuser access?
|
Raises:
Type |
Description |
NotAuthenticatedError
|
When the authorization is expired or otherwise invalid.
|
Examples:
>>> import asab.contextvars
>>> authz = asab.contextvars.Authz.get()
>>> if authz.has_superuser_access():
>>> print("I am a superuser and can do anything!")
>>> else:
>>> print("I am but a mere mortal.")
Source code in asab/web/auth/authorization.py
| def has_superuser_access(self) -> bool:
"""
Check whether the agent is a superuser.
Returns:
bool: Does the agent have superuser access?
Raises:
NotAuthenticatedError: When the authorization is expired or otherwise invalid.
Examples:
>>> import asab.contextvars
>>> authz = asab.contextvars.Authz.get()
>>> if authz.has_superuser_access():
>>> print("I am a superuser and can do anything!")
>>> else:
>>> print("I am but a mere mortal.")
"""
self.require_valid()
return is_superuser(self._Resources)
|
has_tenant_access()
Check whether the agent has access to the tenant in context.
Returns:
Name | Type |
Description |
bool |
bool
|
Is the agent authorized to access requested tenant?
|
Raises:
Type |
Description |
NotAuthenticatedError
|
When the authorization is expired or otherwise invalid.
|
Examples:
>>> import asab.contextvars
>>> authz = asab.contextvars.Authz.get()
>>> tenant_ctx = asab.contextvars.Tenant.set("big-corporation")
>>> try:
>>> if authz.has_tenant_access():
>>> print("I have access to Big Corporation!")
>>> else:
>>> print("Not much to do here.")
>>> finally:
>>> asab.contextvars.Tenant.reset(tenant_ctx)
Source code in asab/web/auth/authorization.py
| def has_tenant_access(self) -> bool:
"""
Check whether the agent has access to the tenant in context.
Returns:
bool: Is the agent authorized to access requested tenant?
Raises:
NotAuthenticatedError: When the authorization is expired or otherwise invalid.
Examples:
>>> import asab.contextvars
>>> authz = asab.contextvars.Authz.get()
>>> tenant_ctx = asab.contextvars.Tenant.set("big-corporation")
>>> try:
>>> if authz.has_tenant_access():
>>> print("I have access to Big Corporation!")
>>> else:
>>> print("Not much to do here.")
>>> finally:
>>> asab.contextvars.Tenant.reset(tenant_ctx)
"""
self.require_valid()
try:
tenant = Tenant.get()
except LookupError as e:
raise ValueError("No tenant in context.") from e
return has_tenant_access(self._Resources, tenant)
|
is_valid()
Check that the authorization is not expired.
Source code in asab/web/auth/authorization.py
| def is_valid(self):
"""
Check that the authorization is not expired.
"""
return datetime.datetime.now(datetime.timezone.utc) < self.Expiration
|
require_resource_access(*resources)
Ensure that the agent is authorized to access the specified resources.
Parameters:
Name |
Type |
Description |
Default |
*resources
|
str
|
A variable number of resource IDs whose authorization is requested.
|
()
|
Raises:
Type |
Description |
NotAuthenticatedError
|
When the authorization is expired or otherwise invalid.
|
AccessDeniedError
|
When the agent does not have access to the requested resources.
|
Examples:
>>> import asab.contextvars
>>> authz = asab.contextvars.Authz.get()
>>> authz.require_resource_access("article:read", "article:write")
>>> print("I can read and write articles!")
Source code in asab/web/auth/authorization.py
| def require_resource_access(self, *resources: str):
"""
Ensure that the agent is authorized to access the specified resources.
Args:
*resources (str): A variable number of resource IDs whose authorization is requested.
Raises:
NotAuthenticatedError: When the authorization is expired or otherwise invalid.
AccessDeniedError: When the agent does not have access to the requested resources.
Examples:
>>> import asab.contextvars
>>> authz = asab.contextvars.Authz.get()
>>> authz.require_resource_access("article:read", "article:write")
>>> print("I can read and write articles!")
"""
if not self.has_resource_access(*resources):
L.warning("Resource authorization required.", struct_data={
"resource": resources, "cid": self.CredentialsId})
raise AccessDeniedError()
|
require_superuser_access()
Ensure that the agent has superuser access.
Raises:
Type |
Description |
NotAuthenticatedError
|
When the authorization is expired or otherwise invalid.
|
AccessDeniedError
|
When the agent does not have superuser access.
|
Examples:
>>> import asab.contextvars
>>> authz = asab.contextvars.Authz.get()
>>> authz.require_superuser_access()
>>> print("I am a superuser and can do anything!")
Source code in asab/web/auth/authorization.py
| def require_superuser_access(self):
"""
Ensure that the agent has superuser access.
Raises:
NotAuthenticatedError: When the authorization is expired or otherwise invalid.
AccessDeniedError: When the agent does not have superuser access.
Examples:
>>> import asab.contextvars
>>> authz = asab.contextvars.Authz.get()
>>> authz.require_superuser_access()
>>> print("I am a superuser and can do anything!")
"""
if not self.has_superuser_access():
L.warning("Superuser authorization required.", struct_data={
"cid": self.CredentialsId})
raise AccessDeniedError()
|
require_tenant_access()
Ensures that the agent is authorized to access the tenant in the current context.
Raises:
Type |
Description |
NotAuthenticatedError
|
When the authorization is expired or otherwise invalid.
|
AccessDeniedError
|
When the agent does not have access to the requested tenant.
|
Examples:
>>> import asab.contextvars
>>> authz = asab.contextvars.Authz.get()
>>> tenant_ctx = asab.contextvars.Tenant.set("big-corporation")
>>> try:
>>> authz.require_tenant_access()
>>> print("I have access to Big Corporation!")
>>> finally:
>>> asab.contextvars.Tenant.reset(tenant_ctx)
Source code in asab/web/auth/authorization.py
| def require_tenant_access(self):
"""
Ensures that the agent is authorized to access the tenant in the current context.
Raises:
NotAuthenticatedError: When the authorization is expired or otherwise invalid.
AccessDeniedError: When the agent does not have access to the requested tenant.
Examples:
>>> import asab.contextvars
>>> authz = asab.contextvars.Authz.get()
>>> tenant_ctx = asab.contextvars.Tenant.set("big-corporation")
>>> try:
>>> authz.require_tenant_access()
>>> print("I have access to Big Corporation!")
>>> finally:
>>> asab.contextvars.Tenant.reset(tenant_ctx)
"""
if not self.has_tenant_access():
L.warning("Tenant authorization required.", struct_data={
"tenant": Tenant.get(), "cid": self.CredentialsId})
raise AccessDeniedError()
|
require_valid()
Ensure that the authorization is not expired.
Raises:
Type |
Description |
NotAuthenticatedError
|
When the authorization is expired or otherwise invalid.
|
Source code in asab/web/auth/authorization.py
| def require_valid(self):
"""
Ensure that the authorization is not expired.
Raises:
NotAuthenticatedError: When the authorization is expired or otherwise invalid.
"""
if not self.is_valid():
L.warning("Authorization expired.", struct_data={
"cid": self.CredentialsId, "exp": self.Expiration.isoformat()})
raise NotAuthenticatedError()
|
user_info()
Return OpenID Connect UserInfo claims (or JWToken claims).
NOTE: If possible, use Authz attributes (CredentialsId, Username etc.) instead of inspecting
the claims directly.
Returns:
Name | Type |
Description |
dict |
Dict[str, Any]
|
|
Raises:
Type |
Description |
NotAuthenticatedError
|
When the authorization is expired or otherwise invalid.
|
Source code in asab/web/auth/authorization.py
| def user_info(self) -> typing.Dict[str, typing.Any]:
"""
Return OpenID Connect UserInfo claims (or JWToken claims).
NOTE: If possible, use Authz attributes (CredentialsId, Username etc.) instead of inspecting
the claims directly.
Returns:
dict: UserInfo claims
Raises:
NotAuthenticatedError: When the authorization is expired or otherwise invalid.
"""
self.require_valid()
return self._Claims
|
asab.web.auth.require(*resources)
Require that the request have authorized access to one or more resources.
Requests without these resources result in AccessDeniedError and consequently in an HTTP 403 response.
Parameters:
Name |
Type |
Description |
Default |
resources
|
Iterable
|
Resources whose authorization is required.
|
()
|
Examples:
@asab.web.auth.require("insider-info:access")
async def get_insider_info(self, request):
data = await self.service.get_insider_info()
return asab.web.rest.json_response(request, data)
Source code in asab/web/auth/decorator.py
| def require(*resources):
"""
Require that the request have authorized access to one or more resources.
Requests without these resources result in AccessDeniedError and consequently in an HTTP 403 response.
Args:
resources (Iterable): Resources whose authorization is required.
Examples:
```python
@asab.web.auth.require("insider-info:access")
async def get_insider_info(self, request):
data = await self.service.get_insider_info()
return asab.web.rest.json_response(request, data)
```
"""
def _require_resource_access_decorator(handler):
@functools.wraps(handler)
async def _require_resource_access_wrapper(*args, **kwargs):
authz = Authz.get()
if authz is None:
raise AccessDeniedError()
authz.require_resource_access(*resources)
return await handler(*args, **kwargs)
return _require_resource_access_wrapper
return _require_resource_access_decorator
|
asab.web.auth.require_superuser(handler)
Require that the request have authorized access to the superuser resource.
Requests without superuser access result in AccessDeniedError and consequently in an HTTP 403 response.
Examples:
@asab.web.auth.require_superuser
async def get_confidential_info(self, request):
data = await self.service.get_confidential_info()
return asab.web.rest.json_response(request, data)
Source code in asab/web/auth/decorator.py
| def require_superuser(handler):
"""
Require that the request have authorized access to the superuser resource.
Requests without superuser access result in AccessDeniedError and consequently in an HTTP 403 response.
Examples:
```python
@asab.web.auth.require_superuser
async def get_confidential_info(self, request):
data = await self.service.get_confidential_info()
return asab.web.rest.json_response(request, data)
```
"""
@functools.wraps(handler)
async def _require_superuser_access_wrapper(*args, **kwargs):
authz = Authz.get()
if authz is None:
raise AccessDeniedError()
authz.require_superuser_access()
return await handler(*args, **kwargs)
return _require_superuser_access_wrapper
|
asab.web.auth.noauth(handler)
Exempt the decorated handler from authentication and authorization.
Examples:
@asab.web.auth.noauth
async def get_public_info(self, request):
data = await self.service.get_public_info()
return asab.web.rest.json_response(request, data)
Source code in asab/web/auth/decorator.py
| def noauth(handler):
"""
Exempt the decorated handler from authentication and authorization.
Examples:
```python
@asab.web.auth.noauth
async def get_public_info(self, request):
data = await self.service.get_public_info()
return asab.web.rest.json_response(request, data)
```
"""
argspec = inspect.getfullargspec(handler)
args = set(argspec.kwonlyargs).union(argspec.args)
for arg in ("user_info", "resources", "authz"):
if arg in args:
raise Exception(
"{}(): Handler with @noauth cannot have {!r} in its arguments.".format(handler.__qualname__, arg))
handler.NoAuth = True
@functools.wraps(handler)
async def _noauth_wrapper(*args, **kwargs):
return await handler(*args, **kwargs)
return _noauth_wrapper
|