mirror of
https://github.com/suitenumerique/meet.git
synced 2026-09-03 14:17:59 +00:00
lebaudantoine
3a93826166
The user `sub` field was rejecting some ASCII characters that are actually valid according to the OIDC spec. Loosen the validation to accept the full ASCII range except control characters, so the field is compliant with the RFC and works with any spec-compliant identity provider. Based on the Stack Overflow discussion in question 279832. Closes #1609.
132 lines
4.3 KiB
Python
132 lines
4.3 KiB
Python
"""Authentication Backends for the Meet core app."""
|
|
|
|
import contextlib
|
|
|
|
from django.conf import settings
|
|
from django.core.exceptions import (
|
|
ImproperlyConfigured,
|
|
SuspiciousOperation,
|
|
ValidationError,
|
|
)
|
|
from django.utils.translation import gettext_lazy as _
|
|
|
|
from lasuite.oidc_login.backends import (
|
|
OIDCAuthenticationBackend as LaSuiteOIDCAuthenticationBackend,
|
|
)
|
|
from rest_framework.authentication import SessionAuthentication
|
|
|
|
from core.models import User
|
|
from core.services.marketing import (
|
|
ContactCreationError,
|
|
ContactData,
|
|
get_marketing_service,
|
|
)
|
|
from core.validators import sub_validator
|
|
|
|
|
|
class OIDCAuthenticationBackend(LaSuiteOIDCAuthenticationBackend):
|
|
"""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_extra_claims(self, user_info):
|
|
"""
|
|
Return extra claims from user_info.
|
|
|
|
Args:
|
|
user_info (dict): The user information dictionary.
|
|
|
|
Returns:
|
|
dict: A dictionary of extra claims.
|
|
|
|
"""
|
|
return {
|
|
# Get user's full name from OIDC fields defined in settings
|
|
"full_name": self.compute_full_name(user_info),
|
|
"short_name": user_info.get(settings.OIDC_USERINFO_SHORTNAME_FIELD),
|
|
}
|
|
|
|
def post_get_or_create_user(self, user, claims, is_new_user):
|
|
"""
|
|
Post-processing after user creation or retrieval.
|
|
|
|
Args:
|
|
user (User): The user instance.
|
|
claims (dict): The claims dictionary.
|
|
is_new_user (bool): Indicates if the user was newly created.
|
|
|
|
Returns:
|
|
- None
|
|
|
|
"""
|
|
email = claims["email"]
|
|
if is_new_user and email and settings.SIGNUP_NEW_USER_TO_MARKETING_EMAIL:
|
|
self.signup_to_marketing_email(email)
|
|
|
|
@staticmethod
|
|
def signup_to_marketing_email(email):
|
|
"""Pragmatic approach to newsletter signup during authentication flow.
|
|
|
|
Details:
|
|
1. Uses a very short timeout (1s) to prevent blocking the auth process
|
|
2. Silently fails if the marketing service is down/slow to prioritize user experience
|
|
3. Trade-off: May miss some signups but ensures auth flow remains fast
|
|
|
|
Note: For a more robust solution, consider using Async task processing (Celery/Django-Q)
|
|
"""
|
|
with contextlib.suppress(
|
|
ContactCreationError, ImproperlyConfigured, ImportError
|
|
):
|
|
marketing_service = get_marketing_service()
|
|
contact_data = ContactData(
|
|
email=email, attributes={"VISIO_SOURCE": ["SIGNIN"]}
|
|
)
|
|
marketing_service.create_contact(
|
|
contact_data, timeout=settings.BREVO_API_TIMEOUT
|
|
)
|
|
|
|
def get_existing_user(self, sub, email):
|
|
"""Fetch existing user by sub or email."""
|
|
|
|
sub = str(sub)
|
|
|
|
try:
|
|
sub_validator(sub)
|
|
except ValidationError as err:
|
|
raise SuspiciousOperation(
|
|
"User info contained an invalid sub claim"
|
|
) from err
|
|
|
|
if len(sub) > 255:
|
|
raise SuspiciousOperation("User info contained an invalid sub claim")
|
|
|
|
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__iexact=email)
|
|
except User.DoesNotExist:
|
|
pass
|
|
except User.MultipleObjectsReturned as e:
|
|
raise SuspiciousOperation(
|
|
"Multiple user accounts share a common email."
|
|
) from e
|
|
return None
|
|
|
|
|
|
class SessionAuthenticationWith401(SessionAuthentication):
|
|
"""
|
|
Identical to DRF's SessionAuthentication, but returns a WWW-Authenticate
|
|
header so unauthenticated requests get a 401 instead of a 403.
|
|
|
|
The scheme is deliberately NOT 'Basic' — that would trigger the browser's
|
|
native login popup. 'Session' is ignored by the browser's auth UI but is
|
|
still truthy, so DRF keeps the status at 401.
|
|
"""
|
|
|
|
def authenticate_header(self, request):
|
|
return "Session"
|