5ef6359b7c
When OIDC providers return random values in the "sub" field instead of stable identifiers, implement email-based user matching as fallback. Note: Current implementation needs improvement. Tests forthcoming. Original: @sampaccoud (ff7914f) on Impress
103 lines
3.5 KiB
Python
103 lines
3.5 KiB
Python
"""Authentication Backends for the Meet core app."""
|
|
|
|
from django.conf import settings
|
|
from django.core.exceptions import SuspiciousOperation
|
|
from django.utils.translation import gettext_lazy as _
|
|
|
|
import requests
|
|
from mozilla_django_oidc.auth import (
|
|
OIDCAuthenticationBackend as MozillaOIDCAuthenticationBackend,
|
|
)
|
|
|
|
from core.models import User
|
|
|
|
|
|
class OIDCAuthenticationBackend(MozillaOIDCAuthenticationBackend):
|
|
"""Custom OpenID Connect (OIDC) Authentication Backend.
|
|
|
|
This class overrides the default OIDC Authentication Backend to accommodate differences
|
|
in the User and Identity models, and handles signed and/or encrypted UserInfo response.
|
|
"""
|
|
|
|
def get_userinfo(self, access_token, id_token, payload):
|
|
"""Return user details dictionary.
|
|
|
|
Parameters:
|
|
- access_token (str): The access token.
|
|
- id_token (str): The id token (unused).
|
|
- payload (dict): The token payload (unused).
|
|
|
|
Note: The id_token and payload parameters are unused in this implementation,
|
|
but were kept to preserve base method signature.
|
|
|
|
Note: It handles signed and/or encrypted UserInfo Response. It is required by
|
|
Agent Connect, which follows the OIDC standard. It forces us to override the
|
|
base method, which deal with 'application/json' response.
|
|
|
|
Returns:
|
|
- dict: User details dictionary obtained from the OpenID Connect user endpoint.
|
|
"""
|
|
|
|
user_response = requests.get(
|
|
self.OIDC_OP_USER_ENDPOINT,
|
|
headers={"Authorization": f"Bearer {access_token}"},
|
|
verify=self.get_settings("OIDC_VERIFY_SSL", True),
|
|
timeout=self.get_settings("OIDC_TIMEOUT", None),
|
|
proxies=self.get_settings("OIDC_PROXY", None),
|
|
)
|
|
user_response.raise_for_status()
|
|
userinfo = self.verify_token(user_response.text)
|
|
return userinfo
|
|
|
|
def get_or_create_user(self, access_token, id_token, payload):
|
|
"""Return a User based on userinfo. Get or create a new user if no user matches the Sub.
|
|
|
|
Parameters:
|
|
- access_token (str): The access token.
|
|
- id_token (str): The ID token.
|
|
- payload (dict): The user payload.
|
|
|
|
Returns:
|
|
- User: An existing or newly created User instance.
|
|
|
|
Raises:
|
|
- Exception: Raised when user creation is not allowed and no existing user is found.
|
|
"""
|
|
|
|
user_info = self.get_userinfo(access_token, id_token, payload)
|
|
sub = user_info.get("sub")
|
|
|
|
if not sub:
|
|
raise SuspiciousOperation(
|
|
_("User info contained no recognizable user identification")
|
|
)
|
|
|
|
email = user_info.get("email")
|
|
user = self.get_existing_user(sub, email)
|
|
|
|
if not user and self.get_settings("OIDC_CREATE_USER", True):
|
|
user = User.objects.create(
|
|
sub=sub,
|
|
email=email,
|
|
password="!", # noqa: S106
|
|
)
|
|
elif not user:
|
|
return None
|
|
|
|
if not user.is_active:
|
|
raise SuspiciousOperation(_("User account is disabled"))
|
|
|
|
return user
|
|
|
|
def get_existing_user(self, sub, email):
|
|
"""Fetch existing user by sub or email."""
|
|
try:
|
|
return User.objects.get(sub=sub)
|
|
except User.DoesNotExist:
|
|
if email and settings.OIDC_FALLBACK_TO_EMAIL_FOR_IDENTIFICATION:
|
|
try:
|
|
return User.objects.get(email=email)
|
|
except User.DoesNotExist:
|
|
pass
|
|
return None
|