|
1 | 1 | """Authentication Backends for the Impress core app."""
|
2 | 2 |
|
3 | 3 | import logging
|
| 4 | +import os |
4 | 5 |
|
5 | 6 | from django.conf import settings
|
6 | 7 | from django.core.exceptions import SuspiciousOperation
|
7 |
| -from django.utils.translation import gettext_lazy as _ |
8 | 8 |
|
9 |
| -import requests |
10 |
| -from mozilla_django_oidc.auth import ( |
11 |
| - OIDCAuthenticationBackend as MozillaOIDCAuthenticationBackend, |
| 9 | +from lasuite.oidc_login.backends import ( |
| 10 | + OIDCAuthenticationBackend as LaSuiteOIDCAuthenticationBackend, |
12 | 11 | )
|
13 | 12 |
|
14 |
| -from core.models import DuplicateEmailError, User |
| 13 | +from core.models import DuplicateEmailError |
15 | 14 |
|
16 | 15 | logger = logging.getLogger(__name__)
|
17 | 16 |
|
| 17 | +# Settings renamed warnings |
| 18 | +if os.environ.get("USER_OIDC_FIELDS_TO_FULLNAME"): |
| 19 | + logger.warning( |
| 20 | + "USER_OIDC_FIELDS_TO_FULLNAME has been renamed to " |
| 21 | + "OIDC_USERINFO_FULLNAME_FIELDS please update your settings." |
| 22 | + ) |
18 | 23 |
|
19 |
| -class OIDCAuthenticationBackend(MozillaOIDCAuthenticationBackend): |
| 24 | +if os.environ.get("USER_OIDC_FIELD_TO_SHORTNAME"): |
| 25 | + logger.warning( |
| 26 | + "USER_OIDC_FIELD_TO_SHORTNAME has been renamed to " |
| 27 | + "OIDC_USERINFO_SHORTNAME_FIELD please update your settings." |
| 28 | + ) |
| 29 | + |
| 30 | + |
| 31 | +class OIDCAuthenticationBackend(LaSuiteOIDCAuthenticationBackend): |
20 | 32 | """Custom OpenID Connect (OIDC) Authentication Backend.
|
21 | 33 |
|
22 | 34 | This class overrides the default OIDC Authentication Backend to accommodate differences
|
23 | 35 | in the User and Identity models, and handles signed and/or encrypted UserInfo response.
|
24 | 36 | """
|
25 | 37 |
|
26 |
| - def get_userinfo(self, access_token, id_token, payload): |
27 |
| - """Return user details dictionary. |
28 |
| -
|
29 |
| - Parameters: |
30 |
| - - access_token (str): The access token. |
31 |
| - - id_token (str): The id token (unused). |
32 |
| - - payload (dict): The token payload (unused). |
33 |
| -
|
34 |
| - Note: The id_token and payload parameters are unused in this implementation, |
35 |
| - but were kept to preserve base method signature. |
36 |
| -
|
37 |
| - Note: It handles signed and/or encrypted UserInfo Response. It is required by |
38 |
| - Agent Connect, which follows the OIDC standard. It forces us to override the |
39 |
| - base method, which deal with 'application/json' response. |
40 |
| -
|
41 |
| - Returns: |
42 |
| - - dict: User details dictionary obtained from the OpenID Connect user endpoint. |
| 38 | + def get_extra_claims(self, user_info): |
43 | 39 | """
|
| 40 | + Return extra claims from user_info. |
44 | 41 |
|
45 |
| - user_response = requests.get( |
46 |
| - self.OIDC_OP_USER_ENDPOINT, |
47 |
| - headers={"Authorization": f"Bearer {access_token}"}, |
48 |
| - verify=self.get_settings("OIDC_VERIFY_SSL", True), |
49 |
| - timeout=self.get_settings("OIDC_TIMEOUT", None), |
50 |
| - proxies=self.get_settings("OIDC_PROXY", None), |
51 |
| - ) |
52 |
| - user_response.raise_for_status() |
| 42 | + Args: |
| 43 | + user_info (dict): The user information dictionary. |
53 | 44 |
|
54 |
| - try: |
55 |
| - userinfo = user_response.json() |
56 |
| - except ValueError: |
57 |
| - try: |
58 |
| - userinfo = self.verify_token(user_response.text) |
59 |
| - except Exception as e: |
60 |
| - raise SuspiciousOperation( |
61 |
| - _("Invalid response format or token verification failed") |
62 |
| - ) from e |
63 |
| - |
64 |
| - return userinfo |
65 |
| - |
66 |
| - def verify_claims(self, claims): |
67 |
| - """ |
68 |
| - Verify the presence of essential claims and the "sub" (which is mandatory as defined |
69 |
| - by the OIDC specification) to decide if authentication should be allowed. |
| 45 | + Returns: |
| 46 | + dict: A dictionary of extra claims. |
70 | 47 | """
|
71 |
| - essential_claims = settings.USER_OIDC_ESSENTIAL_CLAIMS |
72 |
| - missing_claims = [claim for claim in essential_claims if claim not in claims] |
73 |
| - |
74 |
| - if missing_claims: |
75 |
| - logger.error("Missing essential claims: %s", missing_claims) |
76 |
| - return False |
77 |
| - |
78 |
| - return True |
79 |
| - |
80 |
| - def get_or_create_user(self, access_token, id_token, payload): |
81 |
| - """Return a User based on userinfo. Create a new user if no match is found.""" |
82 |
| - |
83 |
| - user_info = self.get_userinfo(access_token, id_token, payload) |
84 |
| - |
85 |
| - if not self.verify_claims(user_info): |
86 |
| - raise SuspiciousOperation("Claims verification failed.") |
87 |
| - |
88 |
| - sub = user_info["sub"] |
89 |
| - email = user_info.get("email") |
90 |
| - |
91 |
| - # Get user's full name from OIDC fields defined in settings |
92 |
| - full_name = self.compute_full_name(user_info) |
93 |
| - short_name = user_info.get(settings.USER_OIDC_FIELD_TO_SHORTNAME) |
94 |
| - |
95 |
| - claims = { |
96 |
| - "email": email, |
97 |
| - "full_name": full_name, |
98 |
| - "short_name": short_name, |
| 48 | + return { |
| 49 | + "full_name": self.compute_full_name(user_info), |
| 50 | + "short_name": user_info.get(settings.OIDC_USERINFO_SHORTNAME_FIELD), |
99 | 51 | }
|
100 | 52 |
|
| 53 | + def get_existing_user(self, sub, email): |
| 54 | + """Fetch existing user by sub or email.""" |
| 55 | + |
101 | 56 | try:
|
102 |
| - user = User.objects.get_user_by_sub_or_email(sub, email) |
| 57 | + return self.UserModel.objects.get_user_by_sub_or_email(sub, email) |
103 | 58 | except DuplicateEmailError as err:
|
104 | 59 | raise SuspiciousOperation(err.message) from err
|
105 |
| - |
106 |
| - if user: |
107 |
| - if not user.is_active: |
108 |
| - raise SuspiciousOperation(_("User account is disabled")) |
109 |
| - self.update_user_if_needed(user, claims) |
110 |
| - elif self.get_settings("OIDC_CREATE_USER", True): |
111 |
| - user = User.objects.create(sub=sub, password="!", **claims) # noqa: S106 |
112 |
| - |
113 |
| - return user |
114 |
| - |
115 |
| - def compute_full_name(self, user_info): |
116 |
| - """Compute user's full name based on OIDC fields in settings.""" |
117 |
| - name_fields = settings.USER_OIDC_FIELDS_TO_FULLNAME |
118 |
| - full_name = " ".join( |
119 |
| - user_info[field] for field in name_fields if user_info.get(field) |
120 |
| - ) |
121 |
| - return full_name or None |
122 |
| - |
123 |
| - def update_user_if_needed(self, user, claims): |
124 |
| - """Update user claims if they have changed.""" |
125 |
| - has_changed = any( |
126 |
| - value and value != getattr(user, key) for key, value in claims.items() |
127 |
| - ) |
128 |
| - if has_changed: |
129 |
| - updated_claims = {key: value for key, value in claims.items() if value} |
130 |
| - self.UserModel.objects.filter(id=user.id).update(**updated_claims) |
0 commit comments