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 ID token claims and also contain the resources object. The iss and exp timestamp claims can contain relative value as a string in the format now + 30m, now - 1d etc. They will be converted to absolute timestamps at runtime. Omitted claims will be supplied with default values.

[auth]
enabled=mock
mock_claims_path=./mock-claims.json

Custom claims examples

Custom issue and expiration time

{
    "iat": "now - 1h",
    "exp": "now + 30m"
}

Custom subject ID (sub) and access control (resources)

{
    "sub": "abc123def456-xyz789",
    "resources": {
        "*": [],
        "cool-eshop": ["article:edit", "article:delete"],
    }
}

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

			# Skip OPTIONS routes, they should require no authentication
			if route.method == "OPTIONS":
				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:

Type	Description
Authorization	Authorization object

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

		# Skip OPTIONS routes, they should require no authentication
		if route.method == "OPTIONS":
			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	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	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	Claim name.	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]	UserInfo claims

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