Skip to content

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
 79
 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
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
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 Exception(
					"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 Exception("WebContainer has authorization middleware installed already.")

		try:
			tenant_wrapper_idx = web_container.WebApp.on_startup.index(set_up_tenant_web_wrapper)
			# Tenant wrapper is installed - Auth wrapper must be applied before it
			web_container.WebApp.on_startup.insert(tenant_wrapper_idx, self.set_up_auth_web_wrapper)
		except ValueError:
			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 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 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

		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

			# Ensure the wrappers are applied in the correct order
			try:
				tenant_wrapper_idx = web_container.WebApp.on_startup.index(set_up_tenant_web_wrapper)
			except ValueError:
				# Tenant wrapper not installed here
				continue

			if auth_wrapper_idx > tenant_wrapper_idx:
				raise Exception("TenantService.install() must be called before AuthService.install()")

		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.info("WebContainer authorization 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 Exception(
				"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 Exception("WebContainer has authorization middleware installed already.")

	try:
		tenant_wrapper_idx = web_container.WebApp.on_startup.index(set_up_tenant_web_wrapper)
		# Tenant wrapper is installed - Auth wrapper must be applied before it
		web_container.WebApp.on_startup.insert(tenant_wrapper_idx, self.set_up_auth_web_wrapper)
	except ValueError:
		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.

Source code in asab/web/auth/authorization.py
class Authorization:
	"""
	Contains authentication and authorization claims, provides methods for checking and enforcing access control.
	"""
	def __init__(self, claims: dict):
		# Userinfo should not be accessed directly
		self._Claims = claims or {}

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


	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._Claims, 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._Claims, 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 authorized_resources(self) -> typing.Optional[typing.Set[str]]:
		"""
		Return the set of authorized resources.

		NOTE: If possible, use methods has_resource_access(resource_id) and has_superuser_access() instead of inspecting
		the set of resources directly.

		Returns:
			Set of authorized resources.
		"""
		self.require_valid()
		return get_authorized_resources(self._Claims, Tenant.get(None))


	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)

authorized_resources() ¤

Return the set of authorized resources.

NOTE: If possible, use methods has_resource_access(resource_id) and has_superuser_access() instead of inspecting the set of resources directly.

Returns:

Type Description
Optional[Set[str]]

Set of authorized resources.

Source code in asab/web/auth/authorization.py
def authorized_resources(self) -> typing.Optional[typing.Set[str]]:
	"""
	Return the set of authorized resources.

	NOTE: If possible, use methods has_resource_access(resource_id) and has_superuser_access() instead of inspecting
	the set of resources directly.

	Returns:
		Set of authorized resources.
	"""
	self.require_valid()
	return get_authorized_resources(self._Claims, Tenant.get(None))

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

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._Claims, 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 decorator_require(handler):

		@functools.wraps(handler)
		async def wrapper(*args, **kwargs):
			authz = Authz.get()
			if authz is None:
				raise AccessDeniedError()

			authz.require_resource_access(*resources)

			return await handler(*args, **kwargs)

		return wrapper

	return decorator_require

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 wrapper(*args, **kwargs):
		return await handler(*args, **kwargs)

	return wrapper