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
public_keys_url	URL	The URL of the authorization server's public keys (also known as jwks_uri in OAuth 2.0
enabled	boolean or "mock"	Enables or disables authentication and authorization or switches to mock authorization. In mock mode, all incoming requests are authorized with mock user info. There is no communication with the authorization server (so it is not necessary to configure public_keys_url in dev mode).
mock_user_info_path	path	Path to JSON file that contains user info claims used in mock mode. The structure of user info should follow the OpenID Connect userinfo definition and also contain the resources object.

Default options:

public_keys_url=http://localhost:3081/.well-known/jwks.json
enabled=yes
mock_user_info_path=/conf/mock-userinfo.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.
	"""

	_PUBLIC_KEYS_URL_DEFAULT = "http://localhost:3081/.well-known/jwks.json"

	def __init__(self, app, service_name="asab.AuthService"):
		super().__init__(app, service_name)

		# To enable Service Discovery, initialize Api Service and call its initialize_zookeeper() method before AuthService initialization
		self.DiscoveryService = self.App.get_service("asab.DiscoveryService")

		self.PublicKeysUrl = Config.get("auth", "public_keys_url") or None
		self.IntrospectionUrl = None
		self.MockUserInfo = None
		self.MockMode = False

		# Configure mock mode
		enabled = Config.get("auth", "enabled", fallback=True)
		if enabled == "mock":
			self.MockMode = True
		elif string_to_boolean(enabled) is True:
			self.MockMode = False
		else:
			raise ValueError(
				"Disabling AuthService is deprecated. "
				"For development pupropses use mock mode instead ([auth] enabled=mock)."
			)

		# Set up auth server keys URL
		if self.MockMode is True:
			self._prepare_mock_mode()
		elif 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.")
		elif not self.PublicKeysUrl and self.DiscoveryService is None:
			self.PublicKeysUrl = self._PUBLIC_KEYS_URL_DEFAULT
			L.warning(
				"No 'public_keys_url' provided in [auth] config section. "
				"Defaulting to {!r}.".format(self._PUBLIC_KEYS_URL_DEFAULT)
			)

		self.TrustedPublicKeys: jwcrypto.jwk.JWKSet = jwcrypto.jwk.JWKSet()
		# Limit the frequency of auth server requests to save network traffic
		self.AuthServerCheckCooldown = datetime.timedelta(minutes=5)
		self.AuthServerLastSuccessfulCheck = None

		if self.PublicKeysUrl:
			self.App.TaskService.schedule(self._fetch_public_keys_if_needed())

		self.Authorizations: typing.Dict[typing.Tuple[str, str], Authorization] = {}
		self.App.PubSub.subscribe("Application.housekeeping!", self._delete_invalid_authorizations)

		# Try to auto-install authorization middleware
		self._try_auto_install()


	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 _prepare_mock_mode(self):
		"""
		Try to set up direct introspection. In not available, set up static mock userinfo.
		"""
		introspection_url = Config.get("auth", "introspection_url", fallback=None)
		if introspection_url is not None:
			self.IntrospectionUrl = introspection_url
			L.warning(
				"AuthService is running in MOCK MODE. Web requests will be authorized with direct introspection "
				"call to {!r}.".format(self.IntrospectionUrl)
			)
			return

		# Load custom user info
		mock_user_info_path = Config.get("auth", "mock_user_info_path")
		if os.path.isfile(mock_user_info_path):
			with open(mock_user_info_path, "rb") as fp:
				user_info = json.load(fp)
		else:
			user_info = MOCK_USERINFO_DEFAULT
		# Validate user info
		resources = user_info.get("resources", {})
		if not isinstance(resources, dict) or not all(
			map(lambda kv: isinstance(kv[0], str) and isinstance(kv[1], list), resources.items())
		):
			raise ValueError("User info 'resources' must be an object with string keys and array values.")
		L.warning(
			"AuthService is running in MOCK MODE. Web requests will be authorized with mock user info, which "
			"currently grants access to the following tenants: {}. To customize mock mode authorization (add or "
			"remove tenants and resources, change username etc.), provide your own user info in {!r}.".format(
				list(t for t in user_info.get("resources", {}).keys() if t != "*"),
				mock_user_info_path
			)
		)
		self.MockUserInfo = user_info


	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)



	def is_ready(self):
		"""
		Check if the service is ready to authorize requests.
		"""
		if self.PublicKeysUrl and not self.TrustedPublicKeys["keys"]:
			# Auth server keys have not been loaded yet
			return False
		return True


	async def build_authorization(self, id_token: str) -> Authorization:
		"""
		Build authorization from ID token string.

		:param id_token: Base64-encoded JWToken from Authorization header
		:return: Valid asab.web.auth.Authorization object
		"""
		# Try if the object already exists
		authz = self.Authorizations.get(id_token)
		if authz is not None:
			try:
				authz.require_valid()
			except NotAuthenticatedError as e:
				del self.Authorizations[id_token]
				raise e
			return authz

		# Create a new Authorization object and store it
		if id_token == "MOCK" and self.MockMode is True:
			authz = Authorization(self.MockUserInfo)
		else:
			userinfo = await self._get_userinfo_from_id_token(id_token)
			authz = Authorization(userinfo)

		self.Authorizations[id_token] = authz
		return authz


	async def _delete_invalid_authorizations(self, message_name):
		"""
		Check for expired Authorization objects and delete them
		"""
		# Find expired
		expired = []
		for key, authz in self.Authorizations.items():
			if not authz.is_valid():
				expired.append(key)

		# Delete expired
		for key in expired:
			del self.Authorizations[key]


	async def get_bearer_token_from_authorization_header(self, request):
		"""
		Validate the Authorizetion header and extract the Bearer token value
		"""
		if self.MockMode is True:
			if not self.IntrospectionUrl:
				return "MOCK"

			# Send the request headers for introspection
			async with aiohttp.ClientSession() as session:
				async with session.post(self.IntrospectionUrl, headers=request.headers) as response:
					if response.status != 200:
						L.warning("Access token introspection failed.")
						raise aiohttp.web.HTTPUnauthorized()
					authorization_header = response.headers.get(aiohttp.hdrs.AUTHORIZATION)

		else:
			authorization_header = request.headers.get(aiohttp.hdrs.AUTHORIZATION)
			if authorization_header is None:
				L.warning("No Authorization header.")
				raise aiohttp.web.HTTPUnauthorized()
		try:
			auth_type, token_value = authorization_header.split(" ", 1)
		except ValueError:
			L.warning("Cannot parse Authorization header.")
			raise aiohttp.web.HTTPBadRequest()
		if auth_type != "Bearer":
			L.warning("Unsupported Authorization header type: {!r}".format(auth_type))
			raise aiohttp.web.HTTPUnauthorized()
		return token_value


	async def _fetch_public_keys_if_needed(self, *args, **kwargs):
		"""
		Check if public keys have been fetched from the authorization server and fetch them if not yet.
		"""
		# TODO: Refactor into Key Providers
		# Add internal shared auth key
		if self.DiscoveryService is not None:
			if self.DiscoveryService.InternalAuthKey is not None:
				self.TrustedPublicKeys.add(self.DiscoveryService.InternalAuthKey.public())
			else:
				L.debug("Internal auth key is not ready yet.")
				self.App.TaskService.schedule(self._fetch_public_keys_if_needed())

			if not self.PublicKeysUrl:
				# Only internal authorization is supported
				return

		# Either DiscoveryService or PublicKeysUrl must be defined
		assert self.PublicKeysUrl is not None

		now = datetime.datetime.now(datetime.timezone.utc)
		if self.AuthServerLastSuccessfulCheck is not None \
			and now < self.AuthServerLastSuccessfulCheck + self.AuthServerCheckCooldown:
			# Public keys have been fetched recently
			return


		async def fetch_keys(session):
			try:
				async with session.get(self.PublicKeysUrl) as response:
					if response.status != 200:
						L.error("HTTP error while loading public keys.", struct_data={
							"status": response.status,
							"url": self.PublicKeysUrl,
							"text": await response.text(),
						})
						return
					try:
						data = await response.json()
					except json.JSONDecodeError:
						L.error("JSON decoding error while loading public keys.", struct_data={
							"url": self.PublicKeysUrl,
							"data": data,
						})
						return
					try:
						key_data = data["keys"].pop()
					except (IndexError, KeyError):
						L.error("Error while loading public keys: No public keys in server response.", struct_data={
							"url": self.PublicKeysUrl,
							"data": data,
						})
						return
					try:
						public_key = jwcrypto.jwk.JWK(**key_data)
					except Exception as e:
						L.error("JWK decoding error while loading public keys: {}.".format(e), struct_data={
							"url": self.PublicKeysUrl,
							"data": data,
						})
						return
			except aiohttp.client_exceptions.ClientConnectorError as e:
				L.error("Connection error while loading public keys: {}".format(e), struct_data={
					"url": self.PublicKeysUrl,
				})
				return
			return public_key

		if self.DiscoveryService is None:
			async with aiohttp.ClientSession() as session:
				public_key = await fetch_keys(session)

		else:
			async with self.DiscoveryService.session() as session:
				try:
					public_key = await fetch_keys(session)
				except NotDiscoveredError as e:
					L.error("Service Discovery error while loading public keys: {}".format(e), struct_data={
						"url": self.PublicKeysUrl,
					})
					return

		if public_key is None:
			return

		self.TrustedPublicKeys.add(public_key)
		self.AuthServerLastSuccessfulCheck = datetime.datetime.now(datetime.timezone.utc)
		L.debug("Public key loaded.", struct_data={"url": self.PublicKeysUrl})


	def _authorize_request(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_request_wrapper(*args, **kwargs):
			request = args[-1]

			bearer_token = await self.get_bearer_token_from_authorization_header(request)
			authz = await self.build_authorization(bearer_token)

			# 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_request_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_request(handler)

		route._handler = handler


	async def _get_userinfo_from_id_token(self, bearer_token):
		"""
		Parse the bearer ID token and extract user info.
		"""
		if not self.is_ready():
			# Try to load the public keys again
			if not self.TrustedPublicKeys["keys"]:
				await self._fetch_public_keys_if_needed()
			if not self.is_ready():
				L.error("Cannot authenticate request: Failed to load authorization server's public keys.")
				raise aiohttp.web.HTTPUnauthorized()

		try:
			return _get_id_token_claims(bearer_token, self.TrustedPublicKeys)
		except (jwcrypto.jws.InvalidJWSSignature, jwcrypto.jwt.JWTMissingKey):
			# Authz server keys may have changed. Try to reload them.
			await self._fetch_public_keys_if_needed()

		try:
			return _get_id_token_claims(bearer_token, self.TrustedPublicKeys)
		except (jwcrypto.jws.InvalidJWSSignature, jwcrypto.jwt.JWTMissingKey) as e:
			L.error("Cannot authenticate request: {}".format(str(e)))
			raise NotAuthenticatedError()


	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.")

build_authorization(id_token) async

Build authorization from ID token string.

:param id_token: Base64-encoded JWToken from Authorization header :return: Valid asab.web.auth.Authorization object

Source code in asab/web/auth/service.py

async def build_authorization(self, id_token: str) -> Authorization:
	"""
	Build authorization from ID token string.

	:param id_token: Base64-encoded JWToken from Authorization header
	:return: Valid asab.web.auth.Authorization object
	"""
	# Try if the object already exists
	authz = self.Authorizations.get(id_token)
	if authz is not None:
		try:
			authz.require_valid()
		except NotAuthenticatedError as e:
			del self.Authorizations[id_token]
			raise e
		return authz

	# Create a new Authorization object and store it
	if id_token == "MOCK" and self.MockMode is True:
		authz = Authorization(self.MockUserInfo)
	else:
		userinfo = await self._get_userinfo_from_id_token(id_token)
		authz = Authorization(userinfo)

	self.Authorizations[id_token] = authz
	return authz

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

get_bearer_token_from_authorization_header(request) async

Validate the Authorizetion header and extract the Bearer token value

Source code in asab/web/auth/service.py

async def get_bearer_token_from_authorization_header(self, request):
	"""
	Validate the Authorizetion header and extract the Bearer token value
	"""
	if self.MockMode is True:
		if not self.IntrospectionUrl:
			return "MOCK"

		# Send the request headers for introspection
		async with aiohttp.ClientSession() as session:
			async with session.post(self.IntrospectionUrl, headers=request.headers) as response:
				if response.status != 200:
					L.warning("Access token introspection failed.")
					raise aiohttp.web.HTTPUnauthorized()
				authorization_header = response.headers.get(aiohttp.hdrs.AUTHORIZATION)

	else:
		authorization_header = request.headers.get(aiohttp.hdrs.AUTHORIZATION)
		if authorization_header is None:
			L.warning("No Authorization header.")
			raise aiohttp.web.HTTPUnauthorized()
	try:
		auth_type, token_value = authorization_header.split(" ", 1)
	except ValueError:
		L.warning("Cannot parse Authorization header.")
		raise aiohttp.web.HTTPBadRequest()
	if auth_type != "Bearer":
		L.warning("Unsupported Authorization header type: {!r}".format(auth_type))
		raise aiohttp.web.HTTPUnauthorized()
	return token_value

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

is_ready()

Check if the service is ready to authorize requests.

Source code in asab/web/auth/service.py

def is_ready(self):
	"""
	Check if the service is ready to authorize requests.
	"""
	if self.PublicKeysUrl and not self.TrustedPublicKeys["keys"]:
		# Auth server keys have not been loaded yet
		return False
	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, 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	End-user's preferred username.
Email	str	End-user email address.
Phone	str	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, 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): End-user's preferred username.
		Email (str): End-user email address.
		Phone (str): 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.).
		"""
		# Userinfo 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
		self.IssuedAt = datetime.datetime.fromtimestamp(int(self._Claims["iat"]), datetime.timezone.utc)
		self.Expiration = datetime.datetime.fromtimestamp(int(self._Claims["exp"]), datetime.timezone.utc)


	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: Do I have superuser access?

		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: Am I authorized to access requested resources?

		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: Am I authorized to access requested tenant?

		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.
		"""
		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:
			AccessDeniedError: If I do 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.

		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:
			AccessDeniedError: If the agent does not have access to the 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
		"""
		self.require_valid()
		return self._Claims


	def get_claim(self, key: str) -> 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).
		"""
		self.require_valid()
		return self._Claims.get(key)


	def _resources(self) -> typing.Optional[typing.Set[str]]:
		"""
		Return the set of authorized resources.

		Returns:
			set: Authorized resources.
		"""
		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.).
	"""
	# Userinfo 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
	self.IssuedAt = datetime.datetime.fromtimestamp(int(self._Claims["iat"]), datetime.timezone.utc)
	self.Expiration = datetime.datetime.fromtimestamp(int(self._Claims["exp"]), datetime.timezone.utc)

get_claim(key)

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).

Source code in asab/web/auth/authorization.py

def get_claim(self, key: str) -> 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).
	"""
	self.require_valid()
	return self._Claims.get(key)

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	Am I authorized to access requested resources?

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: Am I authorized to access requested resources?

	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	Do I have superuser access?

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: Do I have superuser access?

	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	Am I authorized to access requested tenant?

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: Am I authorized to access requested tenant?

	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.	()

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.

	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
AccessDeniedError	If I do 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:
		AccessDeniedError: If I do 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
AccessDeniedError	If the agent does not have access to the 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:
		AccessDeniedError: If the agent does not have access to the 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.

Source code in asab/web/auth/authorization.py

def require_valid(self):
	"""
	Ensure that the authorization is not expired.
	"""
	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

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
	"""
	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("my-app:token:generate")
async def generate_token(self, request):
        data = await self.service.generate_token()
        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("my-app:token:generate")
	async def generate_token(self, request):
		data = await self.service.generate_token()
		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.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