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

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¤

Incoming request can specify tenant context using X-Tenant HTTP header. If this header is not empty, AuthService extracts the tenant ID and verifies if the request is authorized to access that tenant. The request tenant can easily be accessed using asab.contextvars.Tenant.get().

import asab.contextvars
import asab.web.rest

...

async def get_todays_menu(request):
    tenant = asab.contextvars.Tenant.get()
    menu = await get_todays_menu(tenant)
    return asab.web.rest.json_response(request, data=menu)

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.

Source code in asab/web/auth/service.py
 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
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
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)
		self.PublicKeysUrl = Config.get("auth", "public_keys_url") or None

		# 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.IntrospectionUrl = None
		self.MockUserInfo = None

		enabled = Config.get("auth", "enabled", fallback=True)
		if enabled == "mock":
			self.Mode = AuthMode.MOCK
		elif 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._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.Mode == AuthMode.ENABLED:
			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._UserInfo.get("resources", {})
		for tenant in resources.keys():
			if tenant == "*":
				continue
			# Return the first authorized tenant
			return tenant

		return None


	async def initialize(self, app):
		self._check_if_installed()


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

		: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
		for middleware in web_container.WebApp.on_startup:
			if middleware == self._wrap_handlers:
				if len(web_service.Containers) == 1:
					L.warning(
						"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:
					L.warning("WebContainer has authorization middleware installed already.")
				return

		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 not self.TrustedPublicKeys["keys"]:
			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
		"""
		if not self.is_enabled():
			raise ValueError("Cannot build Authorization when AuthService is disabled.")

		# 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.Mode == AuthMode.MOCK:
			authz = Authorization(self, self.MockUserInfo)
		else:
			userinfo = await self._get_userinfo_from_id_token(id_token)
			authz = Authorization(self, 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.Mode == AuthMode.MOCK:
			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]

			if not self.is_enabled():
				return await handler(*args, **kwargs)

			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 _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 unwrapped handler method for signature inspection
		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__

		is_websocket = isinstance(handler_method, WebSocketFactory)

		if hasattr(handler_method, "NoAuth"):
			return
		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)

		# 3) 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)
		if "tenant" in args:
			handler = _pass_tenant(handler)
		if "authz" in args:
			handler = _pass_authz(handler)

		# 2) Authenticate and authorize request, authorize tenant from context, set Authorization context
		handler = self._authorize_request(handler)

		# 1.5) Set tenant context from obsolete locations (no authorization yet)
		# TODO: Deprecated. Ignore tenant in path and query, always use request headers instead.
		if tenant_in_path:
			handler = _set_tenant_context_from_url_path(handler)
		else:
			handler = _set_tenant_context_from_url_query(handler)

		# 1) Set tenant context (no authorization yet)
		# TODO: This should be eventually done by TenantService
		if is_websocket:
			handler = _set_tenant_context_from_sec_websocket_protocol_header(handler)
		else:
			handler = _set_tenant_context_from_x_tenant_header(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 _check_if_installed(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

		for web_container in web_service.Containers.values():
			for middleware in web_container.WebApp.on_startup:
				if middleware == self._wrap_handlers:
					# Container has authorization installed
					break
			else:
				continue

			# Container has authorization installed
			break

		else:
			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 middleware 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
	"""
	if not self.is_enabled():
		raise ValueError("Cannot build Authorization when AuthService is disabled.")

	# 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.Mode == AuthMode.MOCK:
		authz = Authorization(self, self.MockUserInfo)
	else:
		userinfo = await self._get_userinfo_from_id_token(id_token)
		authz = Authorization(self, 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._UserInfo.get("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.Mode == AuthMode.MOCK:
		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
	for middleware in web_container.WebApp.on_startup:
		if middleware == self._wrap_handlers:
			if len(web_service.Containers) == 1:
				L.warning(
					"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:
				L.warning("WebContainer has authorization middleware installed already.")
			return

	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 not self.TrustedPublicKeys["keys"]:
		return False
	return True

asab.web.auth.Authorization ¤

Contains authentication and authorization details, provides methods for checking and enforcing access control.

Requires that AuthService is initialized and enabled.

Source code in asab/web/auth/authorization.py
class Authorization:
	"""
	Contains authentication and authorization details, provides methods for checking and enforcing access control.

	Requires that AuthService is initialized and enabled.
	"""
	def __init__(self, auth_service, userinfo: dict):
		self.AuthService = auth_service
		if not self.AuthService.is_enabled():
			raise ValueError("Cannot create Authorization when AuthService is disabled.")

		# Userinfo should not be accessed directly
		self._UserInfo = userinfo or {}

		self.CredentialsId = self._UserInfo.get("sub")
		self.Username = self._UserInfo.get("preferred_username") or self._UserInfo.get("username")
		self.Email = self._UserInfo.get("email")
		self.Phone = self._UserInfo.get("phone")

		self.Issuer = self._UserInfo.get("iss")  # Who issued the authorization
		self.AuthorizedParty = self._UserInfo.get("azp")  # What party (application) is authorized
		self.IssuedAt = datetime.datetime.fromtimestamp(int(self._UserInfo["iat"]), datetime.timezone.utc)
		self.Expiration = datetime.datetime.fromtimestamp(int(self._UserInfo["exp"]), datetime.timezone.utc)


	def __repr__(self):
		return "<Authorization [{}cid: {!r}, azp: {!r}, iat: {!r}, exp: {!r}]>".format(
			"SUPERUSER, " if self.has_superuser_access() else "",
			self.CredentialsId,
			self.AuthorizedParty,
			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.

		:return: Is the agent a superuser?
		"""
		self.require_valid()
		return is_superuser(self._UserInfo)


	def has_resource_access(self, *resources: typing.Iterable[str]) -> bool:
		"""
		Check whether the agent is authorized to access requested resources.

		:param resources: List of resource IDs whose authorization is requested.
		:return: Is resource access authorized?
		"""
		self.require_valid()
		return has_resource_access(self._UserInfo, resources, tenant=Tenant.get(None))


	def has_tenant_access(self) -> bool:
		"""
		Check whether the agent has access to the tenant in context.

		:return: Is tenant access authorized?
		"""
		self.require_valid()

		try:
			tenant = Tenant.get()
		except LookupError as e:
			raise ValueError("No tenant in context.") from e

		return has_tenant_access(self._UserInfo, 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, "azp": self.AuthorizedParty, "exp": self.Expiration.isoformat()})
			raise NotAuthenticatedError()


	def require_superuser_access(self):
		"""
		Assert that the agent has superuser access.
		"""
		if not self.has_superuser_access():
			L.warning("Superuser authorization required.", struct_data={
				"cid": self.CredentialsId, "azp": self.AuthorizedParty})
			raise AccessDeniedError()


	def require_resource_access(self, *resources: typing.Iterable[str]):
		"""
		Assert that the agent is authorized to access the required resources.

		:param resources: List of resource IDs whose authorization is required.
		"""
		if not self.has_resource_access(*resources):
			L.warning("Resource authorization required.", struct_data={
				"resource": resources, "cid": self.CredentialsId, "azp": self.AuthorizedParty})
			raise AccessDeniedError()


	def require_tenant_access(self):
		"""
		Assert that the agent is authorized to access the tenant in the context.
		"""
		if not self.has_tenant_access():
			L.warning("Tenant authorization required.", struct_data={
				"tenant": Tenant.get(), "cid": self.CredentialsId, "azp": self.AuthorizedParty})
			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.

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


	def user_info(self) -> typing.Dict[str, typing.Any]:
		"""
		Return OpenID Connect UserInfo.

		NOTE: If possible, use Authz attributes (CredentialsId, Username etc.) instead of inspecting the user info
		dictionary directly.

		:return: User info
		"""
		self.require_valid()
		return self._UserInfo

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.

:return: 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.

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

has_resource_access(*resources) ¤

Check whether the agent is authorized to access requested resources.

:param resources: List of resource IDs whose authorization is requested. :return: Is resource access authorized?

Source code in asab/web/auth/authorization.py
def has_resource_access(self, *resources: typing.Iterable[str]) -> bool:
	"""
	Check whether the agent is authorized to access requested resources.

	:param resources: List of resource IDs whose authorization is requested.
	:return: Is resource access authorized?
	"""
	self.require_valid()
	return has_resource_access(self._UserInfo, resources, tenant=Tenant.get(None))

has_superuser_access() ¤

Check whether the agent is a superuser.

:return: Is the agent a superuser?

Source code in asab/web/auth/authorization.py
def has_superuser_access(self) -> bool:
	"""
	Check whether the agent is a superuser.

	:return: Is the agent a superuser?
	"""
	self.require_valid()
	return is_superuser(self._UserInfo)

has_tenant_access() ¤

Check whether the agent has access to the tenant in context.

:return: Is tenant access authorized?

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.

	:return: Is tenant access authorized?
	"""
	self.require_valid()

	try:
		tenant = Tenant.get()
	except LookupError as e:
		raise ValueError("No tenant in context.") from e

	return has_tenant_access(self._UserInfo, 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) ¤

Assert that the agent is authorized to access the required resources.

:param resources: List of resource IDs whose authorization is required.

Source code in asab/web/auth/authorization.py
def require_resource_access(self, *resources: typing.Iterable[str]):
	"""
	Assert that the agent is authorized to access the required resources.

	:param resources: List of resource IDs whose authorization is required.
	"""
	if not self.has_resource_access(*resources):
		L.warning("Resource authorization required.", struct_data={
			"resource": resources, "cid": self.CredentialsId, "azp": self.AuthorizedParty})
		raise AccessDeniedError()

require_superuser_access() ¤

Assert that the agent has superuser access.

Source code in asab/web/auth/authorization.py
def require_superuser_access(self):
	"""
	Assert that the agent has superuser access.
	"""
	if not self.has_superuser_access():
		L.warning("Superuser authorization required.", struct_data={
			"cid": self.CredentialsId, "azp": self.AuthorizedParty})
		raise AccessDeniedError()

require_tenant_access() ¤

Assert that the agent is authorized to access the tenant in the context.

Source code in asab/web/auth/authorization.py
def require_tenant_access(self):
	"""
	Assert that the agent is authorized to access the tenant in the context.
	"""
	if not self.has_tenant_access():
		L.warning("Tenant authorization required.", struct_data={
			"tenant": Tenant.get(), "cid": self.CredentialsId, "azp": self.AuthorizedParty})
		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, "azp": self.AuthorizedParty, "exp": self.Expiration.isoformat()})
		raise NotAuthenticatedError()

user_info() ¤

Return OpenID Connect UserInfo.

NOTE: If possible, use Authz attributes (CredentialsId, Username etc.) instead of inspecting the user info dictionary directly.

:return: User info

Source code in asab/web/auth/authorization.py
def user_info(self) -> typing.Dict[str, typing.Any]:
	"""
	Return OpenID Connect UserInfo.

	NOTE: If possible, use Authz attributes (CredentialsId, Username etc.) instead of inspecting the user info
	dictionary directly.

	:return: User info
	"""
	self.require_valid()
	return self._UserInfo

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 ("tenant", "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