Merge branch 'drf-24' into 'main'

Drf 24

See merge request pfl/django-pyoidc!25

DEVELOPMENT.md

@@ -36,6 +36,11 @@ Then run :
 
 ```
 pip install pip-tools
-pip-compile requirements.in # freeze package versions
-pip-compile requirements-test.in
+pip-compile --output-file=requirements.txt pyproject.toml # freeze package versions
+pip-compile --output-file=requirements-test.txt requirements-test.in
+```
+
+FIXME: possible alternative for tests requirements would be:
+```
+python -m piptools compile --extra test -o requirements-test.txt pyproject.toml
 ```

django_pyoidc/VERSION

@@ -1 +1 @@
-0.0.6
+0.0.7

django_pyoidc/__init__.py

@@ -1,23 +1,75 @@
 from typing import Dict
 
 from django.contrib.auth import get_user_model
+from django.core.exceptions import SuspiciousOperation
 
+from django_pyoidc.exceptions import ClaimNotFoundError
+from django_pyoidc.utils import extract_claim_from_tokens
 
-def get_user_by_email(userinfo_token: Dict[str, str], id_token_claims: Dict):
+
+def get_user_by_email(tokens: Dict):
     User = get_user_model()
 
-    username = ""
+    username = None
+    preferred_username = None
+    email = None
+    client_host = None
+    client_address = None
+    client_id = None
+    django_username = None
+
+    try:
+        email = extract_claim_from_tokens("email", tokens)
+    except ClaimNotFoundError:
+        pass
+
+    try:
+        preferred_username = extract_claim_from_tokens("preferred_username", tokens)
+    except ClaimNotFoundError:
+        pass
 
-    if "preferred_username" in id_token_claims:
-        username = id_token_claims["preferred_username"]
-    elif "preferred_username" in userinfo_token:
-        username = userinfo_token["preferred_username"]
+    try:
+        username = extract_claim_from_tokens("username", tokens)
+    except ClaimNotFoundError:
+        pass
+
+    if preferred_username is None:
+        if username is None:
+            if email:
+                django_username = email
+            else:
+                raise SuspiciousOperation(
+                    "Cannot extract username or email from available OIDC tokens."
+                )
+        else:
+            django_username = username
     else:
-        username = userinfo_token["email"]
+        django_username = preferred_username
+
+    # Currently client Credential logins in Keycloak adds these mappers in access tokens.
+    # we can use that to detect M2M accounts.
+    # TODO: check if any other mapper can/should be used for other providers
+    try:
+        client_host = extract_claim_from_tokens("clientHost", tokens)
+    except ClaimNotFoundError:
+        pass
+    try:
+        client_address = extract_claim_from_tokens("clientAddress", tokens)
+    except ClaimNotFoundError:
+        pass
+
+    if email is None and client_host is not None and client_address is not None:
+        # that's a M2M connexion in grant=client_credential mode
+        try:
+            client_id = extract_claim_from_tokens("client_id", tokens)
+        except ClaimNotFoundError:
+            client_id = django_username
+        # Build a fake email for the service accounts
+        email = f"{client_id}@localhost.lan"
 
     user, created = User.objects.get_or_create(
-        email=userinfo_token["email"],
-        username=username,
+        email=email,
+        username=django_username,
     )
     user.backend = "django.contrib.auth.backends.ModelBackend"
     return user

django_pyoidc/client.py

@@ -0,0 +1,59 @@
+import logging
+
+# import oic
+from oic.extension.client import Client as ClientExtension
+from oic.oic.consumer import Consumer
+from oic.utils.authn.client import CLIENT_AUTHN_METHOD
+
+from django_pyoidc.session import OIDCCacheSessionBackendForDjango
+from django_pyoidc.utils import OIDCCacheBackendForDjango, get_setting_for_sso_op
+
+logger = logging.getLogger(__name__)
+
+
+class OIDCClient:
+    def __init__(self, op_name, session_id=None):
+        self._op_name = op_name
+
+        self.session_cache_backend = OIDCCacheSessionBackendForDjango(self._op_name)
+        self.general_cache_backend = OIDCCacheBackendForDjango(self._op_name)
+
+        consumer_config = {
+            # "debug": True,
+            "response_type": "code"
+        }
+
+        client_config = {
+            "client_id": get_setting_for_sso_op(op_name, "OIDC_CLIENT_ID"),
+            "client_authn_method": CLIENT_AUTHN_METHOD,
+        }
+
+        self.consumer = Consumer(
+            session_db=self.session_cache_backend,
+            consumer_config=consumer_config,
+            client_config=client_config,
+        )
+        # used in token introspection
+        self.client_extension = ClientExtension(**client_config)
+
+        provider_info_uri = get_setting_for_sso_op(
+            op_name, "OIDC_PROVIDER_DISCOVERY_URI"
+        )
+        client_secret = get_setting_for_sso_op(op_name, "OIDC_CLIENT_SECRET")
+        self.client_extension.client_secret = client_secret
+
+        if session_id:
+            self.consumer.restore(session_id)
+        else:
+
+            cache_key = self.general_cache_backend.generate_hashed_cache_key(
+                provider_info_uri
+            )
+            try:
+                config = self.general_cache_backend[cache_key]
+            except KeyError:
+                config = self.consumer.provider_config(provider_info_uri)
+                # shared microcache for provider config
+                # FIXME: Setting for duration
+                self.general_cache_backend.set(cache_key, config, 60)
+            self.consumer.client_secret = client_secret

django_pyoidc/drf/authentication.py

@@ -0,0 +1,126 @@
+import logging
+
+from django.conf import settings
+from django.core.exceptions import PermissionDenied
+from rest_framework import exceptions
+from rest_framework.authentication import BaseAuthentication
+
+from django_pyoidc.client import OIDCClient
+from django_pyoidc.engine import OIDCEngine
+from django_pyoidc.utils import (
+    OIDCCacheBackendForDjango,
+    check_audience,
+    get_setting_for_sso_op,
+)
+
+logger = logging.getLogger(__name__)
+
+
+class OIDCBearerAuthentication(BaseAuthentication):
+    def __init__(self, *args, **kwargs):
+        super(OIDCBearerAuthentication, self).__init__(*args, **kwargs)
+        self.op_name = self.extract_drf_opname()
+        self.general_cache_backend = OIDCCacheBackendForDjango(self.op_name)
+        self.client = OIDCClient(self.op_name)
+        self.engine = OIDCEngine(self.op_name)
+
+    def extract_drf_opname(self):
+        """
+        Given a list of opnames and setting in DJANGO_PYOIDC conf, extract the one having USED_BY_REST_FRAMEWORK=True.
+        """
+        op = None
+        found = False
+        for op_name, configs in settings.DJANGO_PYOIDC.items():
+            if (
+                "USED_BY_REST_FRAMEWORK" in configs
+                and configs["USED_BY_REST_FRAMEWORK"]
+            ):
+                if found:
+                    raise RuntimeError(
+                        "Several DJANGO_PYOIDC sections are declared as USED_BY_REST_FRAMEWORK, only one should be used."
+                    )
+                found = True
+                op = op_name
+        if found:
+            return op
+        else:
+            raise RuntimeError(
+                "No DJANGO_PYOIDC sections are declared with USED_BY_REST_FRAMEWORK configuration option."
+            )
+
+    def extract_access_token(self, request) -> str:
+        val = request.headers.get("Authorization")
+        if not val:
+            msg = "Request missing the authorization header."
+            raise RuntimeError(msg)
+        val = val.strip()
+        bearer_name, access_token_jwt = val.split(maxsplit=1)
+        requested_bearer_name = get_setting_for_sso_op(
+            self.op_name, "OIDC_API_BEARER_NAME", "Bearer"
+        )
+        if not bearer_name.lower() == requested_bearer_name.lower():
+            msg = f"Bad authorization header, invalid Keyword for the bearer, expecting {requested_bearer_name}."
+            raise RuntimeError(msg)
+        return access_token_jwt
+
+    def authenticate(self, request):
+        """
+        Returns two-tuple of (user, token) if authentication succeeds,
+        or None otherwise.
+        """
+        try:
+            user = None
+            access_token_claims = None
+
+            # Extract the access token from an HTTP Authorization Bearer header
+            try:
+                access_token_jwt = self.extract_access_token(request)
+            except RuntimeError as e:
+                logger.error(e)
+                # we return None, and not an Error.
+                # API auth failed, but maybe anon access is allowed
+                return None
+
+            # This introspection of the token is made by the SSO server
+            # so it is quite slow, but there's a cache added based on the token expiration
+            access_token_claims = self.engine.introspect_access_token(
+                access_token_jwt, client=self.client
+            )
+            logger.debug(access_token_claims)
+            if not access_token_claims.get("active"):
+                msg = "Inactive access token."
+                raise exceptions.AuthenticationFailed(msg)
+
+            # FIXME: add an option to request userinfo here, but that may be quite slow
+
+            if access_token_claims:
+                logger.debug("Request has valid access token.")
+
+                # FIXME: Add a setting to disable
+                client_id = get_setting_for_sso_op(self.op_name, "OIDC_CLIENT_ID")
+                if not check_audience(client_id, access_token_claims):
+                    raise PermissionDenied(
+                        f"Invalid result for acces token audiences check for {client_id}."
+                    )
+
+                logger.debug("Let application load user via user hook.")
+                user = self.engine.call_get_user_function(
+                    tokens={
+                        "access_token_jwt": access_token_jwt,
+                        "access_token_claims": access_token_claims,
+                    }
+                )
+
+            if not user:
+                logger.error(
+                    "OIDC Bearer Authentication process failure. Cannot set active authenticated user."
+                )
+                return None
+
+        except PermissionDenied as exp:
+            raise exp
+        except Exception as exp:
+            logger.exception(exp)
+            return None
+
+        return (user, access_token_claims)

django_pyoidc/engine.py

@@ -0,0 +1,78 @@
+import datetime
+import logging
+
+from django_pyoidc import get_user_by_email
+from django_pyoidc.client import OIDCClient
+from django_pyoidc.utils import (
+    OIDCCacheBackendForDjango,
+    get_setting_for_sso_op,
+    import_object,
+)
+
+logger = logging.getLogger(__name__)
+
+
+class OIDCEngine:
+    def __init__(self, op_name: str):
+        self.op_name = op_name
+        self.general_cache_backend = OIDCCacheBackendForDjango(self.op_name)
+
+    def call_function(self, setting_name, *args, **kwargs):
+        function_path = get_setting_for_sso_op(self.op_name, setting_name)
+        if function_path:
+            func = import_object(function_path, "")
+            return func(*args, **kwargs)
+
+    def call_get_user_function(self, tokens={}):
+        if get_setting_for_sso_op(self.op_name, "HOOK_GET_USER"):
+            logger.debug("OIDC, Calling user hook on get_user")
+            return self.call_function("HOOK_GET_USER", tokens)
+        else:
+            return get_user_by_email(tokens)
+
+    def introspect_access_token(self, access_token_jwt: str, client: OIDCClient):
+        """
+        Perform a cached intropesction call to extract claims from encoded jwt of the access_token
+        """
+        # FIXME: allow a non-cached mode by global settings
+        access_token_claims = None
+
+        # FIXME: in what case could we not have an access token available?
+        # should we raise an error then?
+        if access_token_jwt is not None:
+            cache_key = self.general_cache_backend.generate_hashed_cache_key(
+                access_token_jwt
+            )
+            try:
+                access_token_claims = self.general_cache_backend["cache_key"]
+            except KeyError:
+                # CACHE MISS
+
+                # RFC 7662: token introspection: ask SSO to validate and render the jwt as json
+                # this means a slow web call
+                request_args = {
+                    "token": access_token_jwt,
+                    "token_type_hint": "access_token",
+                }
+                client_auth_method = client.consumer.registration_response.get(
+                    "introspection_endpoint_auth_method", "client_secret_basic"
+                )
+                introspection = client.client_extension.do_token_introspection(
+                    request_args=request_args,
+                    authn_method=client_auth_method,
+                    endpoint=client.consumer.introspection_endpoint,
+                )
+                access_token_claims = introspection.to_dict()
+
+                # store it in cache
+                current = datetime.datetime.now().strftime("%s")
+                if "exp" not in access_token_claims:
+                    raise RuntimeError("No expiry set on the access token.")
+                access_token_expiry = access_token_claims["exp"]
+                exp = int(access_token_expiry) - int(current)
+                logger.debug(
+                    f"Token expiry: {exp}  - current is {current} "
+                    f"and expiry is set to {access_token_expiry} in the token"
+                )
+                self.general_cache_backend.set(cache_key, access_token_claims, exp)
+        return access_token_claims

django_pyoidc/exceptions.py

@@ -0,0 +1,6 @@
+class InvalidSIDException(Exception):
+    pass
+
+
+class ClaimNotFoundError(Exception):
+    pass