Skip to content

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
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
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


	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_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

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