Differential D5366 Diff 19219 swh/auth/django/backends.py

Changeset View

Standalone View

swh/auth/django/backends.py

# Copyright (C) 2020-2021 The Software Heritage developers		# Copyright (C) 2020-2021 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution		# See the AUTHORS file at the top-level directory of this distribution
# License: GNU Affero General Public License version 3, or any later version		# License: GNU Affero General Public License version 3, or any later version
# See top-level LICENSE file for more information		# See top-level LICENSE file for more information

		from datetime import datetime
		import hashlib
from typing import Any, Dict, Optional		from typing import Any, Dict, Optional

from django.core.cache import cache		from django.core.cache import cache
from django.http import HttpRequest		from django.http import HttpRequest
from django.utils import timezone		from django.utils import timezone
		from rest_framework.authentication import BaseAuthentication
		from rest_framework.exceptions import AuthenticationFailed, ValidationError
import sentry_sdk		import sentry_sdk

from swh.auth.django.models import OIDCUser		from swh.auth.django.models import OIDCUser
from swh.auth.django.utils import (		from swh.auth.django.utils import (
keycloak_oidc_client,		keycloak_oidc_client,
oidc_profile_cache_key,		oidc_profile_cache_key,
		oidc_user_from_decoded_token,
oidc_user_from_profile,		oidc_user_from_profile,
)		)
from swh.auth.keycloak import KeycloakOpenIDConnect		from swh.auth.keycloak import KeycloakOpenIDConnect


def _update_cached_oidc_profile(		def _update_cached_oidc_profile(
oidc_client: KeycloakOpenIDConnect, oidc_profile: Dict[str, Any], user: OIDCUser		oidc_client: KeycloakOpenIDConnect, oidc_profile: Dict[str, Any], user: OIDCUser
) -> None:		) -> None:
▲ Show 20 Lines • Show All 80 Lines • ▼ Show 20 Lines	def get_user(self, user_id: int) -> Optional[OIDCUser]:
# restore auth backend		# restore auth backend
setattr(user, "backend", f"{__name__}.{self.__class__.__name__}")		setattr(user, "backend", f"{__name__}.{self.__class__.__name__}")
return user		return user
except Exception as e:		except Exception as e:
sentry_sdk.capture_exception(e)		sentry_sdk.capture_exception(e)
return None		return None
else:		else:
return None		return None


		class OIDCBearerTokenAuthentication(BaseAuthentication):
		"""
		Django REST Framework authentication backend using bearer tokens for
		Keycloak OpenID Connect.

		It enables to authenticate a Web API user by sending a long-lived
		OpenID Connect refresh token in HTTP Authorization headers.
		Long lived refresh tokens can be generated by opening an OpenID Connect
		session with the following scope: ``openid offline_access``.

		To use that backend globally in your DRF application, proceed as follow:

		* add ``"swh.auth.django.backends.OIDCBearerTokenAuthentication"``
		to the ``REST_FRAMEWORK["DEFAULT_AUTHENTICATION_CLASSES"]``
		django setting.

		* configure Keycloak URL, realm and client by adding
		``SWH_AUTH_SERVER_URL``, ``SWH_AUTH_REALM_NAME`` and ``SWH_AUTH_CLIENT_ID``
		in django settings

		Users will then be able to perform authenticated Web API calls by sending
		their refresh token in HTTP Authorization headers, for instance:
		``curl -H "Authorization: Bearer ${TOKEN}" https://...``.

		"""

		def authenticate(self, request):
		auth_header = request.META.get("HTTP_AUTHORIZATION")
		if auth_header is None:
		return None

		try:
		auth_type, refresh_token = auth_header.split(" ", 1)
		except ValueError:
		raise AuthenticationFailed("Invalid HTTP authorization header format")

		if auth_type != "Bearer":
		raise AuthenticationFailed(
		(f"Invalid or unsupported HTTP authorization" f" type ({auth_type}).")
		)
		try:

		oidc_client = keycloak_oidc_client()

		# compute a cache key from the token that does not exceed
		# memcached key size limit
		hasher = hashlib.sha1()
		hasher.update(refresh_token.encode("ascii"))
		cache_key = f"api_token_{hasher.hexdigest()}"

		# check if an access token is cached
		access_token = cache.get(cache_key)

		# attempt to decode access token
		try:
		decoded_token = oidc_client.decode_token(access_token)
		except Exception:
		# access token is None or it has expired
		decoded_token = None

		if access_token is None or decoded_token is None:
		# get a new access token from authentication provider
		access_token = oidc_client.refresh_token(refresh_token)["access_token"]
		# decode access token
		decoded_token = oidc_client.decode_token(access_token)
		# compute access token expiration
		exp = datetime.fromtimestamp(decoded_token["exp"])
		ttl = int(exp.timestamp() - timezone.now().timestamp())
		# save access token in cache while it is valid
		cache.set(cache_key, access_token, timeout=max(0, ttl))

		# create Django user
		user = oidc_user_from_decoded_token(decoded_token, oidc_client.client_id)
		except UnicodeEncodeError as e:
		sentry_sdk.capture_exception(e)
		raise ValidationError("Invalid bearer token")
		except Exception as e:
		sentry_sdk.capture_exception(e)
		raise AuthenticationFailed(str(e))

		return user, None