Merge pull request #43 from maykinmedia/feature/nested-claims

✨ Support extraction of nested configurable claims

README.rst

@@ -217,6 +217,28 @@ The name of the claim that is used for the ``User.username`` property
 can be configured via the admin. By default, the username is derived from the ``sub`` claim that
 is returned by the OIDC provider.
 
+If the desired claim is nested in one or more objects, its path can be specified with dots, e.g.:
+
+.. code-block:: json
+
+    {
+        "some": {
+            "nested": {
+                "claim": "foo"
+            }
+        }
+    }
+
+Can be retrieved by setting the username claim to ``some.nested.claim``
+
+**NOTE**: the username claim does not support claims that have dots in their name, it cannot be configured to retrieve the following claim for instance:
+
+.. code-block:: json
+
+    {
+        "some.dotted.claim": "foo"
+    }
+
 .. |build-status| image:: https://github.com/maykinmedia/mozilla-django-oidc-db/workflows/Run%20CI/badge.svg?branch=master
     :target: https://github.com/maykinmedia/mozilla-django-oidc-db/actions?query=workflow%3A%22Run+CI%22+branch%3Amaster
 

mozilla_django_oidc_db/backends.py

@@ -6,6 +6,7 @@
 from django.contrib.auth.models import Group
 from django.core.exceptions import ObjectDoesNotExist
 
+from glom import glom
 from mozilla_django_oidc.auth import (
     OIDCAuthenticationBackend as _OIDCAuthenticationBackend,
 )
@@ -21,6 +22,8 @@ class OIDCAuthenticationBackend(SoloConfigMixin, _OIDCAuthenticationBackend):
     as unique identifier (default `sub`).
     """
 
+    config_identifier_field = "username_claim"
+
     def __getattribute__(self, attr):
         if attr.startswith("OIDC"):
             return self.get_settings(attr, None)
@@ -37,6 +40,13 @@ def __init__(self, *args, **kwargs):
         # to avoid a large number of `OpenIDConnectConfig.get_solo` calls when
         # `OIDCAuthenticationBackend.__init__` is called for permission checks
 
+    def retrieve_identifier_claim(self, claims: dict) -> str:
+        # NOTE: this does not support the extraction of claims that contain dots "." in
+        # their name (e.g. {"foo.bar": "baz"})
+        identifier_claim_name = getattr(self.config, self.config_identifier_field)
+        unique_id = glom(claims, identifier_claim_name, default="")
+        return unique_id
+
     def authenticate(self, *args, **kwargs):
         if not self.config.enabled:
             return None
@@ -48,14 +58,13 @@ def get_user_instance_values(self, claims) -> Dict[str, Any]:
         Map the names and values of the claims to the fields of the User model
         """
         return {
-            model_field: claims.get(claims_field, "")
+            model_field: glom(claims, claims_field, default="")
             for model_field, claims_field in self.config.claim_mapping.items()
         }
 
     def create_user(self, claims):
         """Return object for a newly created user account."""
-        username_claim = self.config.username_claim
-        unique_id = claims.get(username_claim)
+        unique_id = self.retrieve_identifier_claim(claims)
 
         logger.debug("Creating OIDC user: %s", unique_id)
 
@@ -68,8 +77,7 @@ def create_user(self, claims):
 
     def filter_users_by_claims(self, claims):
         """Return all users matching the specified subject."""
-        username_claim = self.config.username_claim
-        unique_id = claims.get(username_claim)
+        unique_id = self.retrieve_identifier_claim(claims)
 
         if not unique_id:
             return self.UserModel.objects.none()
@@ -79,16 +87,14 @@ def filter_users_by_claims(self, claims):
 
     def verify_claims(self, claims) -> bool:
         """Verify the provided claims to decide if authentication should be allowed."""
-        scopes = self.get_settings("OIDC_RP_SCOPES", "openid email")
-
         logger.debug("OIDC claims received: %s", claims)
 
-        username_claim = self.config.username_claim
+        identifier_claim_name = getattr(self.config, self.config_identifier_field)
 
-        if username_claim not in claims:
+        if not glom(claims, identifier_claim_name, default=""):
             logger.error(
                 "%s not in OIDC claims, cannot proceed with authentication",
-                username_claim,
+                identifier_claim_name,
             )
             return False
         return True
@@ -122,9 +128,8 @@ def update_user_groups(self, user, claims):
         if groups_claim:
             # Update the user's group memberships
             django_groups = [group.name for group in user.groups.all()]
-
-            if groups_claim in claims:
-                claim_groups = claims[groups_claim]
+            claim_groups = glom(claims, groups_claim, default=[])
+            if claim_groups:
                 if not isinstance(claim_groups, list):
                     claim_groups = [
                         claim_groups,

setup.cfg

@@ -33,6 +33,7 @@ install_requires =
     django-better-admin-arrayfield
     django-solo
     mozilla-django-oidc >=1.0.0, <2.0.0
+    glom
 tests_require =
     psycopg2
     pytest

tests/test_backend.py

@@ -47,6 +47,36 @@ def test_backend_get_user_instance_values(mock_get_solo):
     }
 
 
+@patch("mozilla_django_oidc_db.models.OpenIDConnectConfig.get_solo")
+def test_backend_get_user_instance_values_nested_claims(mock_get_solo):
+    mock_get_solo.return_value = OpenIDConnectConfig(
+        claim_mapping={
+            "email": "user_info.email",
+            "first_name": "user_info.given_name",
+            "last_name": "user_info.family_name",
+        }
+    )
+
+    claims = {
+        "sub": "123456",
+        "user_info": {
+            "email": "admin@localhost",
+            "given_name": "John",
+            "family_name": "Doe",
+        },
+    }
+
+    backend = OIDCAuthenticationBackend()
+
+    user_values = backend.get_user_instance_values(claims)
+
+    assert user_values == {
+        "email": "admin@localhost",
+        "first_name": "John",
+        "last_name": "Doe",
+    }
+
+
 @pytest.mark.django_db
 @patch("mozilla_django_oidc_db.models.OpenIDConnectConfig.get_solo")
 def test_backend_create_user(mock_get_solo):
@@ -331,6 +361,44 @@ def test_backend_create_user_sync_groups_according_to_pattern(mock_get_solo):
     assert list(user.groups.values_list("name", flat=True)) == ["groupadmin"]
 
 
+@pytest.mark.django_db
+@patch("mozilla_django_oidc_db.models.OpenIDConnectConfig.get_solo")
+def test_backend_create_user_sync_all_groups_nested_groups_claim(mock_get_solo):
+    mock_get_solo.return_value = OpenIDConnectConfig(
+        enabled=True,
+        oidc_rp_client_id="testid",
+        oidc_rp_client_secret="secret",
+        oidc_rp_sign_algo="HS256",
+        oidc_rp_scopes_list=["openid", "email"],
+        oidc_op_jwks_endpoint="http://some.endpoint/v1/jwks",
+        oidc_op_authorization_endpoint="http://some.endpoint/v1/auth",
+        oidc_op_token_endpoint="http://some.endpoint/v1/token",
+        oidc_op_user_endpoint="http://some.endpoint/v1/user",
+        groups_claim="nested_object.roles",
+        sync_groups=True,
+        sync_groups_glob_pattern="*",
+    )
+
+    claims = {
+        "sub": "123456",
+        "nested_object": {"roles": ["useradmin", "groupadmin"]},
+    }
+
+    backend = OIDCAuthenticationBackend()
+
+    user = backend.create_user(claims)
+
+    # Verify that the groups were created
+    assert Group.objects.count() == 2
+
+    # Verify that a user is created with the correct values
+    assert user.username == "123456"
+    assert list(user.groups.values_list("name", flat=True)) == [
+        "useradmin",
+        "groupadmin",
+    ]
+
+
 @pytest.mark.django_db
 @patch("mozilla_django_oidc_db.models.OpenIDConnectConfig.get_solo")
 def test_backend_create_user_with_profile_settings(mock_get_solo):