Authorization and multitenancy

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 module also implements multitenancy, meaning that your application can be used by a number of independent subjects (tenants, for example companies) without interfering with each other.

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, initialize AuthService and install it in your asab.web.WebContainer:

class MyApplication(asab.Application):
    def __init__(self):
        super().__init__()

        # Initialize web container
        self.add_module(asab.web.Module)
        self.WebService = self.get_service("asab.WebService")
        self.WebContainer = asab.web.WebContainer(self.WebService, "web")

        # Initialize authorization service and install the decorators
        self.AuthService = asab.web.auth.AuthService(self)
        self.AuthService.install(self.WebContainer)

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 WebContainer now accepts only requests with a valid authentication. Unauthenticated requests are automatically answered with HTTP 401: Unauthorized. Authenticated requests will be inspected for user info, authorized user tenant and resources. These attributes are passed to the handler method, if the method in question has the corresponding keyword arguments (user_info, tenant and resources). In the following example, the method will receive user_info and resources from the request:

async def order_breakfast(self, request, *, tenant, user_info, resources):
    user_id = user_info["sub"]
    user_name = user_info["preferred_username"]
    if "pancakes:eat" in resources:
        ...
    return asab.web.rest.json_response(request, {
        "result": "Good morning {}, your breakfast will be ready in a minute!".format(user_name)
    })

See examples/web-auth.py for a full demo ASAB application with auth module.

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

Multitenancy

Strictly multitenant endpoints

Strictly multitenant endpoints always operate within a tenant, hence they need the tenant parameter to be always provided. Such endpoints must define the tenant parameter in their URL path and include tenant argument in the handler method. Auth service extracts the tenant from the URL path, validates the tenant existence, checks if the request is authorized for the tenant, and finally passes the tenant name to the handler method.

Example handler:

class MenuHandler:
    def __init__(self, app):
        self.MenuService = app.get_service("MenuService")
        router = app.WebContainer.WebApp.router
        # Add a path with `tenant` parameter
        router.add_get("/{tenant}/todays-menu", self.get_todays_menu)

    # Define handler method with `tenant` argument in the signature
    async def get_todays_menu(self, request, *, tenant):
        menu = await self.MenuService.get_todays_menu(tenant)
        return asab.web.rest.json_response(request, data=menu)

Example request:

GET http://localhost:8080/lazy-raccoon-bistro/todays-menu

Configurable multitenant endpoints

Configurable multitenant endpoints usually operate within a tenant, but they can also operate in tenantless mode if the application is designed for that.

When you create an endpoint without tenant parameter in the URL path and with tenant argument in the handler method, the Auth service will either expect the tenant parameter to be provided in the URL query. If it is not in the query, the tenant variable is set to None.

Example handler:

class MenuHandler:
    def __init__(self, app):
        self.MenuService = app.get_service("MenuService")
        router = app.WebContainer.WebApp.router
        # Add a path without `tenant` parameter
        router.add_get("/todays-menu", self.get_todays_menu)

    # Define handler method with `tenant` argument in the signature
    async def get_todays_menu(self, request, *, tenant):
        menu = await self.MenuService.get_todays_menu(tenant)
        return asab.web.rest.json_response(request, data=menu)

Example requests:

GET http://localhost:8080/todays-menu?tenant=lazy-raccoon-bistro

Mock mode

In mock mode, actual authorization is disabled and replaced with mock authorization, which is useful when you want to develop your ASAB application, but don't have any authorization server at hand. Activate mock mode by setting enabled=mock in the [auth] config section. You can also specify a path to a JSON file with your own mock userinfo payload [auth] mock_user_info_path:

[auth]
enabled=mock
mock_user_info_path=${THIS_DIR}/mock_userinfo.json

When dev mode is enabled, you don't have to provide [public_keys_url] since this option is ignored.

Reference

asab.web.auth.AuthService

Bases: Service

Provides authentication and authorization of incoming requests.

Configuration

Configuration section: auth Configuration options: public_keys_url: - default: "" - URL containing the authorization server's public JWKey set (usually found at "/.well-known/jwks.json") enabled: - default: "yes" - options: "yes", "no", "mocked" - Switch authentication and authorization on, off or activate mock mode. - In MOCK MODE - no authorization server is needed, - all incoming requests are mock-authorized with pre-defined user info, - custom mock user info can supplied in a JSON file. mock_user_info_path: - default: "/conf/mock-userinfo.json"

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

class AuthService(asab.Service):
	"""
	Provides authentication and authorization of incoming requests.

	Configuration:
		Configuration section: auth
		Configuration options:
			public_keys_url:
				- default: ""
				- URL containing the authorization server's public JWKey set (usually found at "/.well-known/jwks.json")
			enabled:
				- default: "yes"
				- options: "yes", "no", "mocked"
				- Switch authentication and authorization on, off or activate mock mode.
				- In MOCK MODE
					- no authorization server is needed,
					- all incoming requests are mock-authorized with pre-defined user info,
					- custom mock user info can supplied in a JSON file.
			mock_user_info_path:
				- default: "/conf/mock-userinfo.json"
	"""

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

	def __init__(self, app, service_name="asab.AuthService"):
		super().__init__(app, service_name)
		self.PublicKeysUrl = asab.Config.get("auth", "public_keys_url")

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

		enabled = asab.Config.get("auth", "enabled", fallback=True)
		if enabled == "mock":
			self.Mode = AuthMode.MOCK
		elif asab.utils.string_to_boolean(enabled):
			self.Mode = AuthMode.ENABLED
		else:
			self.Mode = AuthMode.DISABLED

		if self.Mode == AuthMode.DISABLED:
			pass
		elif self.Mode == AuthMode.MOCK:
			self.MockUserInfo = self._prepare_mock_user_info()
		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 len(self.PublicKeysUrl) == 0:
			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.AuthServerPublicKey = None  # TODO: Support multiple public keys
		# Limit the frequency of auth server requests to save network traffic
		self.AuthServerCheckCooldown = datetime.timedelta(minutes=5)
		self.AuthServerLastSuccessfulCheck = None

	def _prepare_mock_user_info(self):
		# Load custom user info
		mock_user_info_path = asab.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. All 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))
		return user_info


	async def initialize(self, app):
		if self.Mode == AuthMode.ENABLED:
			await self._fetch_public_keys_if_needed()


	def is_enabled(self) -> bool:
		"""
		Check if the AuthService is enabled. Mock mode counts as enabled too.
		"""
		return self.Mode in {AuthMode.ENABLED, AuthMode.MOCK}


	def install(self, web_container):
		"""
		Apply authorization to all web handlers in a web container, according to their arguments and path parameters.

		:param web_container: Web container to be protected by authorization.
		:type web_container: asab.web.WebContainer
		"""
		# TODO: Call this automatically if there is only one container
		web_container.WebApp.on_startup.append(self._wrap_handlers)


	def is_ready(self):
		"""
		Check if the service is ready to authorize requests.
		"""
		if self.Mode == AuthMode.DISABLED:
			return True
		if self.Mode == AuthMode.MOCK:
			return True
		if self.AuthServerPublicKey is None:
			return False
		return True


	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 self.AuthServerPublicKey is None:
				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.AuthServerPublicKey)
		except jwcrypto.jws.InvalidJWSSignature:
			# Authz server keys may have changed. Try to reload them.
			L.warning("Invalid ID token signature.")
			await self._fetch_public_keys_if_needed()

		try:
			return _get_id_token_claims(bearer_token, self.AuthServerPublicKey)
		except jwcrypto.jws.InvalidJWSSignature:
			L.error("Cannot authenticate request: Invalid ID token signature.")
			raise asab.exceptions.NotAuthenticatedError()


	def get_authorized_tenant(self, request) -> typing.Optional[str]:
		"""
		Get the request's authorized tenant.
		"""
		if hasattr(request, "_AuthorizedTenants"):
			for tenant in request._AuthorizedTenants:
				# Return the first authorized tenant
				return tenant
		return None


	def has_superuser_access(self, authorized_resources: typing.Iterable) -> bool:
		"""
		Check if the superuser resource is present in the authorized resource list.
		"""
		if self.Mode == AuthMode.DISABLED:
			return True
		return SUPERUSER_RESOURCE in authorized_resources


	def has_resource_access(self, authorized_resources: typing.Iterable, required_resources: typing.Iterable) -> bool:
		"""
		Check if the requested resources or the superuser resource are present in the authorized resource list.
		"""
		if self.Mode == AuthMode.DISABLED:
			return True
		if self.has_superuser_access(authorized_resources):
			return True
		for resource in required_resources:
			if resource not in authorized_resources:
				return False
		return True


	def has_tenant_access(
		self, authorized_resources: typing.Iterable, authorized_tenants: typing.Iterable, tenant: str
	) -> bool:
		"""
		Check if the request is authorized to access a tenant.
		If the request has superuser access, tenant access is always implicitly granted.
		"""
		if self.Mode == AuthMode.DISABLED:
			return True
		if self.has_superuser_access(authorized_resources):
			return True
		if tenant in authorized_tenants:
			return True
		return False

	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.
		"""
		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.AuthServerPublicKey = public_key
		self.AuthServerLastSuccessfulCheck = datetime.datetime.now(datetime.timezone.utc)
		L.debug("Public key loaded.", struct_data={"url": self.PublicKeysUrl})


	def _authenticate_request(self, handler):
		"""
		Authenticate the request by the JWT ID token in the Authorization header.
		Extract the token claims into request attributes so that they can be used for authorization checks.
		"""
		@functools.wraps(handler)
		async def wrapper(*args, **kwargs):
			request = args[-1]
			if self.Mode == AuthMode.DISABLED:
				user_info = None
			elif self.Mode == AuthMode.MOCK:
				user_info = self.MockUserInfo
			else:
				# Extract user info from the request Authorization header
				bearer_token = _get_bearer_token(request)
				user_info = await self.get_userinfo_from_id_token(bearer_token)

			# Add userinfo, tenants and global resources to the request
			if self.Mode != AuthMode.DISABLED:
				assert user_info is not None
				request._UserInfo = user_info
				resource_dict = request._UserInfo["resources"]
				request._AuthorizedResources = frozenset(resource_dict.get("*", []))
				request._AuthorizedTenants = frozenset(t for t in resource_dict.keys() if t != "*")
			else:
				request._UserInfo = None
				request._AuthorizedResources = None
				request._AuthorizedTenants = None

			# Add access control methods to the request
			def has_resource_access(*required_resources: list) -> bool:
				return self.has_resource_access(request._AuthorizedResources, required_resources)
			request.has_resource_access = has_resource_access

			def has_tenant_access(tenant: str) -> bool:
				return self.has_tenant_access(request._AuthorizedResources, request._AuthorizedTenants, tenant)
			request.has_tenant_access = has_tenant_access

			def has_superuser_access() -> bool:
				return self.has_superuser_access(request._AuthorizedResources)
			request.has_superuser_access = has_superuser_access

			return await handler(*args, **kwargs)
		return wrapper


	async def _wrap_handlers(self, aiohttp_app):
		"""
		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 auth for HEAD requests
			if route.method == "HEAD":
				continue

			try:
				self._wrap_handler(route)
			except Exception as e:
				raise Exception("Failed to initialize auth for handler {!r}.".format(route.handler.__qualname__)) from e


	def _wrap_handler(self, route):
		"""
		Inspect handler and apply suitable auth wrappers.
		"""
		# Check if tenant is in route path
		route_info = route.get_info()
		tenant_in_path = "formatter" in route_info and "{tenant}" in route_info["formatter"]

		# Extract the actual handler method for signature checks
		handler_method = route.handler
		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__

		if hasattr(handler_method, "NoAuth"):
			return
		argspec = inspect.getfullargspec(handler_method)
		args = set(argspec.kwonlyargs).union(argspec.args)

		# Extract the whole handler for wrapping
		handler = route.handler

		# Apply the decorators in reverse order (the last applied wrapper affects the request first)
		if "resources" in args:
			handler = _add_resources(handler)
		if "user_info" in args:
			handler = _add_user_info(handler)
		if "tenant" in args:
			if tenant_in_path:
				handler = self._add_tenant_from_path(handler)
			else:
				handler = self._add_tenant_from_query(handler)

		handler = self._authenticate_request(handler)
		route._handler = handler


	def _authorize_tenant_request(self, request, tenant):
		"""
		Check access to requested tenant and add tenant resources to the request
		"""
		# Check if tenant access is authorized
		if tenant not in request._AuthorizedTenants:
			L.warning("Tenant not authorized.", struct_data={"tenant": tenant, "sub": request._UserInfo.get("sub")})
			raise asab.exceptions.AccessDeniedError()

		# Extend globally granted resources with tenant-granted resources
		request._AuthorizedResources = set(request._AuthorizedResources.union(
			request._UserInfo["resources"].get(tenant, [])))


	def _add_tenant_from_path(self, handler):
		"""
		Extract tenant from request path and authorize it
		"""

		@functools.wraps(handler)
		async def wrapper(*args, **kwargs):
			request = args[-1]
			tenant = request.match_info["tenant"]
			if self.Mode != AuthMode.DISABLED:
				self._authorize_tenant_request(request, tenant)
			return await handler(*args, tenant=tenant, **kwargs)

		return wrapper


	def _add_tenant_from_query(self, handler):
		"""
		Extract tenant from request query and authorize it
		"""

		@functools.wraps(handler)
		async def wrapper(*args, **kwargs):
			request = args[-1]
			if "tenant" not in request.query:
				return await handler(*args, tenant=None, **kwargs)
			else:
				tenant = request.query["tenant"]
				if self.Mode != AuthMode.DISABLED:
					self._authorize_tenant_request(request, tenant)
			return await handler(*args, tenant=tenant, **kwargs)

		return wrapper

get_authorized_tenant(request)

Get the request's authorized tenant.

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

def get_authorized_tenant(self, request) -> typing.Optional[str]:
	"""
	Get the request's authorized tenant.
	"""
	if hasattr(request, "_AuthorizedTenants"):
		for tenant in request._AuthorizedTenants:
			# Return the first authorized tenant
			return tenant
	return None

get_userinfo_from_id_token(bearer_token) async

Parse the bearer ID token and extract user info.

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

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 self.AuthServerPublicKey is None:
			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.AuthServerPublicKey)
	except jwcrypto.jws.InvalidJWSSignature:
		# Authz server keys may have changed. Try to reload them.
		L.warning("Invalid ID token signature.")
		await self._fetch_public_keys_if_needed()

	try:
		return _get_id_token_claims(bearer_token, self.AuthServerPublicKey)
	except jwcrypto.jws.InvalidJWSSignature:
		L.error("Cannot authenticate request: Invalid ID token signature.")
		raise asab.exceptions.NotAuthenticatedError()

has_resource_access(authorized_resources, required_resources)

Check if the requested resources or the superuser resource are present in the authorized resource list.

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

def has_resource_access(self, authorized_resources: typing.Iterable, required_resources: typing.Iterable) -> bool:
	"""
	Check if the requested resources or the superuser resource are present in the authorized resource list.
	"""
	if self.Mode == AuthMode.DISABLED:
		return True
	if self.has_superuser_access(authorized_resources):
		return True
	for resource in required_resources:
		if resource not in authorized_resources:
			return False
	return True

has_superuser_access(authorized_resources)

Check if the superuser resource is present in the authorized resource list.

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

def has_superuser_access(self, authorized_resources: typing.Iterable) -> bool:
	"""
	Check if the superuser resource is present in the authorized resource list.
	"""
	if self.Mode == AuthMode.DISABLED:
		return True
	return SUPERUSER_RESOURCE in authorized_resources

has_tenant_access(authorized_resources, authorized_tenants, tenant)

Check if the request is authorized to access a tenant. If the request has superuser access, tenant access is always implicitly granted.

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

def has_tenant_access(
	self, authorized_resources: typing.Iterable, authorized_tenants: typing.Iterable, tenant: str
) -> bool:
	"""
	Check if the request is authorized to access a tenant.
	If the request has superuser access, tenant access is always implicitly granted.
	"""
	if self.Mode == AuthMode.DISABLED:
		return True
	if self.has_superuser_access(authorized_resources):
		return True
	if tenant in authorized_tenants:
		return True
	return False

install(web_container)

Apply authorization to all web handlers in a web container, according to their arguments and path parameters.

: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, according to their arguments and path parameters.

	:param web_container: Web container to be protected by authorization.
	:type web_container: asab.web.WebContainer
	"""
	# TODO: Call this automatically if there is only one container
	web_container.WebApp.on_startup.append(self._wrap_handlers)

is_enabled()

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:
	"""
	Check if the AuthService is enabled. Mock mode counts as enabled too.
	"""
	return self.Mode in {AuthMode.ENABLED, AuthMode.MOCK}

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.Mode == AuthMode.DISABLED:
		return True
	if self.Mode == AuthMode.MOCK:
		return True
	if self.AuthServerPublicKey is None:
		return False
	return True

asab.web.auth.require(*resources)

Specify resources required for endpoint access. Requests without these resources result in HTTP 403 response.

Parameters:

Name	Type	Description	Default
resources	Iterable	Resources required to access the decorated method.	()

Examples:

@asab.web.authz.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):
	"""
	Specify resources required for endpoint access.
	Requests without these resources result in HTTP 403 response.

	Args:
		resources (Iterable): Resources required to access the decorated method.

	Examples:

	```python
	@asab.web.authz.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 decorator_require(handler):

		@functools.wraps(handler)
		async def wrapper(*args, **kwargs):
			request = args[-1]

			if not hasattr(request, "has_resource_access"):
				raise Exception(
					"Cannot check resource access. Make sure that AuthService is installed and that "
					"the handler method does not use both the '@noauth' and the '@require' decorators at once.")

			if not request.has_resource_access(*resources):
				raise asab.exceptions.AccessDeniedError()

			return await handler(*args, **kwargs)

		return wrapper

	return decorator_require

asab.web.auth.noauth(handler)

Exempt the decorated handler from authentication and authorization. The tenant, user_info and resources arguments are not available in the handler.

Examples:

@asab.web.authz.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.
	The `tenant`, `user_info` and `resources` arguments are not available in the handler.

	Examples:

	```python
	@asab.web.authz.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 ("tenant", "user_info", "resources"):
		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 wrapper(*args, **kwargs):
		return await handler(*args, **kwargs)

	return wrapper