Add WSO2OAuthClient

CHANGES.rst

@@ -5,7 +5,17 @@ Changelog of nens-auth-client
 1.4.5 (unreleased)
 ------------------
 
-- Nothing changed yet.
+- Moved provider-specific logic (``extract_provider_name``, ``extract_username``) to
+  to ``CognitoOAuthClient``.
+
+- Made the OAuth2 provider backend configurable through ``NENS_AUTH_OAUTH_BACKEND``.
+
+- Added ``WSO2OAuthClient`` for login using WSO2 Identity Server.
+
+- Added the possibility of wildcards in ``NENS_AUTH_TRUSTED_PROVIDERS[_NEW_USERS]``.
+
+- Usage of Bearer tokens with WSO2 is not (yet) supported; currently a NotImplementedError
+  is raised with the token claims as argument.
 
 
 1.4.4 (2024-01-26)

README.rst

@@ -237,6 +237,11 @@ from a trusted identity provider. For such sites, enable the
 The setting contains the list of provider names (as configured in cognito) that we trust to
 have correct email addresses.
 
+If you want to auto-accept all users that authenticate through OAuth2, use a wildecard as follows::
+
+    NENS_AUTH_TRUSTED_PROVIDERS = ["*"]
+    NENS_AUTH_TRUSTED_PROVIDERS_NEW_USERS = ["*"]
+
 
 Auto-accepting N&S users
 ------------------------

nens_auth_client/backends.py

@@ -1,6 +1,7 @@
-from .users import _extract_provider_name
+from .oauth import get_oauth_client
 from .users import create_remote_user
 from .users import create_user
+from .users import found_or_wildcard
 from django.conf import settings
 from django.contrib.auth import get_user_model
 from django.contrib.auth.backends import ModelBackend
@@ -49,7 +50,7 @@ def _nens_user_extract_username(claims):
     email domain @nelen-schuurmans.nl.
     """
     # Get the provider name, return False if not present
-    provider_name = _extract_provider_name(claims)
+    provider_name = get_oauth_client().extract_provider_name(claims)
     if not provider_name:
         return
 
@@ -149,21 +150,22 @@ def authenticate(self, request, claims):
         Returns:
           user or None
         """
-        provider_name = _extract_provider_name(claims)
+        provider_name = get_oauth_client().extract_provider_name(claims)
         email = claims.get("email")
-        # We need proper claims with provider_name and email, otherwise we
-        # don't need to bother to look.
-        if not provider_name or not email:
+        # We need email
+        if not email:
             return
 
-        if provider_name not in settings.NENS_AUTH_TRUSTED_PROVIDERS:
+        if not found_or_wildcard(provider_name, settings.NENS_AUTH_TRUSTED_PROVIDERS):
             logger.debug("%s not in special list of trusted providers", provider_name)
             return
 
         try:
             user = UserModel.objects.get(email__iexact=email)
         except ObjectDoesNotExist:
-            if provider_name in settings.NENS_AUTH_TRUSTED_PROVIDERS_NEW_USERS:
+            if found_or_wildcard(
+                provider_name, settings.NENS_AUTH_TRUSTED_PROVIDERS_NEW_USERS
+            ):
                 return create_user(claims)
             else:
                 return

nens_auth_client/cognito.py

@@ -148,3 +148,24 @@ def load_key(header, payload):
 
         claims.validate(leeway=leeway)
         return claims
+
+    @staticmethod
+    def extract_provider_name(claims):
+        """Return provider name from claim and `None` if not found"""
+        # Also used by backends.py
+        try:
+            return claims["identities"][0]["providerName"]
+        except (KeyError, IndexError):
+            return
+
+    @staticmethod
+    def extract_username(claims) -> str:
+        """Return username from claims"""
+        username = ""
+        if claims.get("identities"):
+            # External identity providers result in usernames that are not
+            # recognizable by the end user. Use the email instead.
+            username = claims.get("email")
+        if not username:
+            username = claims["cognito:username"]
+        return username

nens_auth_client/conf.py

@@ -15,6 +15,7 @@ class NensAuthClientAppConf(AppConf):
     RESOURCE_SERVER_ID = None  # For Access Tokens ("aud" should equal this)
 
     PERMISSION_BACKEND = "nens_auth_client.permissions.DjangoPermissionBackend"
+    OAUTH_BACKEND = "nens_auth_client.cognito.CognitoOAuthClient"
 
     INVITATION_EMAIL_SUBJECT = "Invitation"
     INVITATION_EXPIRY_DAYS = 14  # change this to change the default expiry

nens_auth_client/oauth.py

@@ -1,24 +1,24 @@
 from authlib.integrations.django_client import OAuth
 from authlib.oidc.discovery import get_well_known_url
 from django.conf import settings
-from nens_auth_client.cognito import CognitoOAuthClient
+from django.utils.module_loading import import_string
 
 # Create the global OAuth registry
 oauth_registry = OAuth()
 
 
 def get_oauth_client():
-    client = oauth_registry.create_client("cognito")
+    client = oauth_registry.create_client("oauth")
     if client is not None:
         return client
 
     url = get_well_known_url(settings.NENS_AUTH_ISSUER, external=True)
     oauth_registry.register(
-        name="cognito",
+        name="oauth",
         client_id=settings.NENS_AUTH_CLIENT_ID,
         client_secret=settings.NENS_AUTH_CLIENT_SECRET,
         server_metadata_url=url,
         client_kwargs={"scope": " ".join(settings.NENS_AUTH_SCOPE)},
-        client_cls=CognitoOAuthClient,
+        client_cls=import_string(settings.NENS_AUTH_OAUTH_BACKEND),
     )
-    return oauth_registry.create_client("cognito")
+    return oauth_registry.create_client("oauth")

nens_auth_client/tests/conftest.py

@@ -172,7 +172,7 @@ def func(id_token, user=None, code="code", state="state", nonce="nonce"):
             "http://testserver/authorize/?code={}&state={}".format(code, state)
         )
         request.session = {
-            f"_state_cognito_{state}": {"data": {"nonce": nonce}},
+            f"_state_oauth_{state}": {"data": {"nonce": nonce}},
             LOGIN_REDIRECT_SESSION_KEY: "http://testserver/success",
         }
         return request

nens_auth_client/tests/test_authorize_view.py

@@ -306,7 +306,7 @@ def test_authorize_wrong_state(id_token_generator, auth_req_generator):
     # we solve this by restarting the login process (keeping the success_url).
     id_token, claims = id_token_generator()
     request = auth_req_generator(id_token, state="a")
-    request.session.pop("_state_cognito_a")
+    request.session.pop("_state_oauth_a")
     request.session[LOGIN_REDIRECT_SESSION_KEY] = "/success"
     response = views.authorize(request)
 
@@ -400,7 +400,7 @@ def test_token_error(rq_mocker, rf, openid_configuration):
     )
     # Create the request
     request = rf.get("http://testserver/authorize/?code=abc&state=my_state")
-    request.session = {"_state_cognito_my_state": {"data": {"nonce": "x"}}}
+    request.session = {"_state_oauth_my_state": {"data": {"nonce": "x"}}}
     with pytest.raises(OAuthError, match="some_error: bla"):
         views.authorize(request)
 
@@ -416,7 +416,7 @@ def test_token_error_code_already_used(rq_mocker, rf, openid_configuration):
     )
     # Create the request
     request = rf.get("http://testserver/authorize/?code=abc&state=my_state")
-    request.session = {"_state_cognito_my_state": {"data": {"nonce": "bla"}}}
+    request.session = {"_state_oauth_my_state": {"data": {"nonce": "bla"}}}
     response = views.authorize(request)
 
     assert response.status_code == 302

nens_auth_client/tests/test_cognito.py

@@ -1,3 +1,4 @@
+from nens_auth_client.cognito import CognitoOAuthClient
 from nens_auth_client.cognito import preprocess_access_token
 
 import pytest
@@ -20,3 +21,34 @@ def test_preprocess_access_token(claims, expected, settings):
     settings.NENS_AUTH_RESOURCE_SERVER_ID = "api/"
     preprocess_access_token(claims)
     assert claims == expected
+
+
+def test_extract_provider_name_present():
+    # Extract provider name when it is present.
+    claims = {"identities": [{"providerName": "Google"}]}
+    assert CognitoOAuthClient.extract_provider_name(claims) == "Google"
+
+
+def test_extract_provider_name_absent():
+    # Return None when a provider name cannot be found.
+    claims = {"some": "claim"}
+    assert CognitoOAuthClient.extract_provider_name(claims) is None
+
+
+@pytest.mark.parametrize(
+    "claims,expected",
+    [
+        ({"cognito:username": "foo", "email": "[email protected]"}, "foo"),
+        ({"cognito:username": "foo", "email": "[email protected]", "identities": []}, "foo"),
+        (
+            {
+                "cognito:username": "abc123",
+                "email": "[email protected]",
+                "identities": ["something"],
+            },
+            "[email protected]",
+        ),
+    ],
+)
+def test_extract_username(claims, expected):
+    assert CognitoOAuthClient.extract_username(claims) == expected

nens_auth_client/tests/test_login_view.py

@@ -30,7 +30,7 @@ def test_login(rf, openid_configuration):
     assert qs["redirect_uri"] == ["http://testserver/authorize/"]
     assert qs["scope"] == [" ".join(settings.NENS_AUTH_SCOPE)]
 
-    state = request.session[f'_state_cognito_{qs["state"][0]}']
+    state = request.session[f'_state_oauth_{qs["state"][0]}']
     assert qs["nonce"] == [state["data"]["nonce"]]
     assert request.session[views.LOGIN_REDIRECT_SESSION_KEY] == "/a"
 
@@ -117,7 +117,7 @@ def test_login_with_forced_logout(rf, openid_configuration, mocker, logged_in):
     assert qs["redirect_uri"] == ["http://testserver/authorize/"]
     assert qs["scope"] == [" ".join(settings.NENS_AUTH_SCOPE)]
 
-    state = request.session[f'_state_cognito_{qs["state"][0]}']
+    state = request.session[f'_state_oauth_{qs["state"][0]}']
     assert qs["nonce"] == [state["data"]["nonce"]]
     assert request.session[views.LOGIN_REDIRECT_SESSION_KEY] == "/a"
 

nens_auth_client/tests/test_users.py

@@ -1,6 +1,5 @@
 from django.contrib.auth.models import User
 from django.db import IntegrityError
-from nens_auth_client.users import _extract_provider_name
 from nens_auth_client.users import create_remote_user
 from nens_auth_client.users import create_user
 from nens_auth_client.users import update_remote_user
@@ -182,15 +181,3 @@ def test_create_user_username_exists(user_mgr, create_user_m, mocker):
     first_call, second_call = create_user_m.call_args_list
     assert first_call[0] == ("testuser", "abc")
     assert second_call[0] == ("testuserx23f", "abc")
-
-
-def test_extract_provider_name_present():
-    # Extract provider name when it is present.
-    claims = {"identities": [{"providerName": "Google"}]}
-    assert _extract_provider_name(claims) == "Google"
-
-
-def test_extract_provider_name_absent():
-    # Return None when a provider name cannot be found.
-    claims = {"some": "claim"}
-    assert not _extract_provider_name(claims)

nens_auth_client/tests/test_wso2.py

@@ -0,0 +1,27 @@
+from nens_auth_client.wso2 import WSO2AuthClient
+
+import pytest
+
+
+def test_extract_provider_name():
+    # Extract provider name when it is present.
+    claims = {"sub": "abc123"}
+    assert WSO2AuthClient.extract_provider_name(claims) is None
+
+
+@pytest.mark.parametrize(
+    "claims,expected",
+    [
+        ({"email": "[email protected]"}, "[email protected]"),
+    ],
+)
+def test_extract_username(claims, expected):
+    assert WSO2AuthClient.extract_username(claims) == expected
+
+
+def test_parse_access_token_includes_claims(access_token_generator):
+    with pytest.raises(NotImplementedError) as e:
+        WSO2AuthClient.parse_access_token(None, access_token_generator())
+
+    # error is raised with claims as arg
+    assert e.value.args[0]["client_id"] == "1234"

nens_auth_client/testsettings.py

@@ -118,11 +118,10 @@
 )
 
 # Add your production name here
-ALLOWED_HOSTS = ["localhost"]
+ALLOWED_HOSTS = ["localhost", "0.0.0.0"]
 
 AUTHENTICATION_BACKENDS = [
     "nens_auth_client.backends.RemoteUserBackend",
-    "nens_auth_client.backends.SSOMigrationBackend",
     "django.contrib.auth.backends.ModelBackend",
 ]
 

nens_auth_client/users.py

@@ -1,10 +1,12 @@
 from .models import RemoteUser
+from .oauth import get_oauth_client
 from django.conf import settings
 from django.contrib.auth import get_user_model
 from django.db import IntegrityError
 from django.db import transaction
 from django.utils import timezone
 from django.utils.crypto import get_random_string
+from typing import List
 
 import logging
 
@@ -14,13 +16,8 @@
 User = get_user_model()
 
 
-def _extract_provider_name(claims):
-    """Return provider name from claim and `None` if not found"""
-    # Also used by backends.py
-    try:
-        return claims["identities"][0]["providerName"]
-    except (KeyError, IndexError):
-        return
+def found_or_wildcard(elem: str, allowed_elements: List[str]):
+    return "*" in allowed_elements or elem in allowed_elements
 
 
 def create_remote_user(user, claims):
@@ -69,8 +66,6 @@ def _create_user(username, external_id):
 def create_user(claims):
     """Create User and associate it with an external one through RemoteUser.
 
-    The username is taken from the "cognito:username" field.
-
     Raises an IntegrityError if this username already exists. This is expected
     to happen very rarely, in which case we do want to see this in our bug
     tracker.
@@ -82,15 +77,9 @@ def create_user(claims):
       django User (created or, in case of a race condition, retrieved)
       RemoteUser (created or, in case of a race condition, retrieved)
     """
-    # Format a username from the claims.
-    username = ""
-    if claims.get("identities"):
-        # External identity providers result in usernames that are not
-        # recognizable by the end user. Use the email instead.
-        username = claims.get("email")
-    if not username:
-        username = claims["cognito:username"]
-    username = username[: settings.NENS_AUTH_USERNAME_MAX_LENGTH]
+    username = get_oauth_client().extract_username(claims)[
+        : settings.NENS_AUTH_USERNAME_MAX_LENGTH
+    ]
 
     external_id = claims["sub"]
     try:
@@ -118,10 +107,9 @@ def update_user(user, claims):
     """
     user.first_name = claims.get("given_name", "")
     user.last_name = claims.get("family_name", "")
-    provider_name = _extract_provider_name(claims)
-    if (
-        claims.get("email_verified")
-        or provider_name in settings.NENS_AUTH_TRUSTED_PROVIDERS
+    provider_name = get_oauth_client().extract_provider_name(claims)
+    if claims.get("email_verified") or found_or_wildcard(
+        provider_name, settings.NENS_AUTH_TRUSTED_PROVIDERS
     ):
         user.email = claims.get("email", "")
     else: