Unit tests, 100% coverage!

- Refactoring for small methods
- Fix some bugs

Author: cpontvieux-systra

Diff for: .gitignore

@@ -2,5 +2,6 @@ dist/
 project/
 __pycache__/
 reports/
+.coverage
 .venv/
 .cache/

Diff for: .gitlab-ci.yml

@@ -22,8 +22,6 @@ type_checker:
   allow_failure: true
 unit_tests:
   extends: .unit_tests
-  # the code is not ready yet
-  allow_failure: true
 .tag_base:
   extends: .tag
   needs:

Diff for: .gitlab/unit-tests-ci.yml

@@ -9,3 +9,13 @@ include:
   script:
     - 'make install_all_deps'
     - 'make tests'
+  artifacts:
+    when: always
+    paths:
+      - reports/**
+    reports:
+      junit: reports/report.xunit
+      coverage_report:
+        coverage_format: cobertura
+        path: reports/coverage.xml
+  coverage: /^TOTAL .+? ([\d.]+%)$/

Diff for: _CHANGELOGS/Changed/refresh_session_middleware_min_secs.md

@@ -0,0 +1 @@
+- Allow to override `MIN_SECONDS` in `RefreshSessionMiddleware`.

Diff for: _CHANGELOGS/Changed/utc_time.md

@@ -0,0 +1 @@
+- Use UTC time in `RefreshAccessTokenMiddleware`, `RefreshSessionMiddleware`.

Diff for: _CHANGELOGS/Fixed/bearer_authentication_backend_bugfix.md

@@ -0,0 +1 @@
+- Default value for `jwks` in `BearerAuthenticationBackend` should be dict, not a list.

Diff for: _CHANGELOGS/Fixed/blacklist_expiration_bugfix.md

@@ -0,0 +1 @@
+- Fix blacklist expiration for token where seconds where used as hours

Diff for: _CHANGELOGS/Fixed/cache_based_view_clear_cache_fix.md

@@ -0,0 +1 @@
+- Fix `_clear_cache` method in `CacheBaseView`: was not clearing the session correctly.

Diff for: _CHANGELOGS/Fixed/config_reload.md

@@ -0,0 +1 @@
+- Configuration cannot be updated when using unit tests. This is now fixed. No impact on lib usage.

Diff for: _CHANGELOGS/Fixed/login_required_error_field.md

@@ -0,0 +1 @@
+- Respect the optional `fail` parameter of `@login_required` decorator.

Diff for: _CHANGELOGS/Fixed/middleware_inheritance.md

@@ -0,0 +1 @@
+- Middlewares should not inherit depraceted `MiddlewareMixin`.

Diff for: _CHANGELOGS/Fixed/oauth2_middleware_mixin_user_not_exist_on_request.md

@@ -0,0 +1 @@
+- If user does not exist on request, should not crash in `Oauth2MiddlewareMixin.is_oidc_enabled`.

Diff for: leaks.toml

@@ -4,4 +4,8 @@ stopwords = [
     'OIDC_AUTHORIZATION_HEADER_PREFIX',
     'oidc_authentication',
     'oidc_op_authorization_url',
+    'a_client_secret',
+    'client_secret',
+    'BadPrefix',
+    'django_key',
 ]

Diff for: oauth2_authcodeflow/auth.py

@@ -1,13 +1,15 @@
+from datetime import datetime
 from inspect import signature
 from logging import (
     debug,
     warning,
 )
 from re import search
-from time import time
 from typing import (
     Dict,
     Optional,
+    Type,
+    cast,
 )
 
 from django.contrib.auth import get_user_model
@@ -23,6 +25,7 @@
     JWTError,
     jwt,
 )
+from pytz import utc
 from requests import get as request_get
 from requests import post as request_post
 
@@ -35,8 +38,7 @@
 
 
 class AuthenticationMixin:
-    def __init__(self, *args, **kwargs):
-        self.UserModel = get_user_model()
+    UserModel: Type[AbstractUser] = cast(Type[AbstractUser], get_user_model())
 
     def validate_and_decode_id_token(self, id_token: str, nonce: Optional[str], jwks: Dict) -> Dict:
         header = jwt.get_unverified_header(id_token)
@@ -56,14 +58,10 @@ def validate_and_decode_id_token(self, id_token: str, nonce: Optional[str], jwks
             key = settings.OIDC_RP_CLIENT_SECRET
         else:
             raise NotImplementedError(f"Algo {algo} cannot be handled by this authentication backend")
-        if isinstance(key, dict):
-            secret = key
-        else:  # RSA public key (bytes)
-            secret = key
         try:
             claims = jwt.decode(
                 id_token,
-                secret,
+                key,
                 algorithms=settings.OIDC_RP_SIGN_ALGOS_ALLOWED,
                 audience=settings.OIDC_RP_CLIENT_ID,
                 options={
@@ -80,7 +78,7 @@ def validate_and_decode_id_token(self, id_token: str, nonce: Optional[str], jwks
             raise SuspiciousOperation("JWT Nonce verification failed")
         return claims
 
-    def validate_claims(self, claims):
+    def validate_claims(self, claims: Dict) -> None:
         expected_list = [settings.OIDC_OP_EXPECTED_EMAIL_CLAIM] + list(settings.OIDC_OP_EXPECTED_CLAIMS)
         debug(f"Validate claims={claims} against expected {expected_list}")
         for expected in expected_list:
@@ -163,17 +161,10 @@ def authenticate_oauth2(self,
             return None
         try:
             if use_pkce:
-                if any((
-                    not code,
-                    not code_verifier,
-                )):
+                if not code or not code_verifier:
                     raise SuspiciousOperation('code and code_verifier values are required')
             else:
-                if any((
-                    not code,
-                    not state,
-                    not nonce,
-                )):
+                if not code or not state or not nonce:
                     raise SuspiciousOperation('code, state and nonce values are required')
             params = {
                 'grant_type': 'authorization_code',
@@ -190,53 +181,52 @@ def authenticate_oauth2(self,
             result = resp.json()
             id_token = result['id_token']
             access_token = result['access_token']
-            expires_in = result.get('expires_in')  # in secs, could be missing
+            access_expires_in = result.get('expires_in')  # in secs, could be missing
             refresh_token = result.get('refresh_token') if 'offline_access' in settings.OIDC_RP_SCOPES else None
-            id_claims = self.validate_and_decode_id_token(id_token, nonce, request.session[constants.SESSION_OP_JWKS])
+            id_claims = self.validate_and_decode_id_token(id_token, nonce, request.session.get(constants.SESSION_OP_JWKS, {}))
             self.validate_claims(id_claims)
-            now = time()
-            if expires_in:
-                expires_at = now + expires_in
+            now_ts = int(datetime.now(tz=utc).timestamp())
+            session_expires_at = now_ts + settings.OIDC_MIDDLEWARE_SESSION_TIMEOUT_SECONDS
+            if access_expires_in:
+                access_expires_at = now_ts + access_expires_in
             else:
-                expires_at = id_claims.get('exp', now + settings.OIDC_MIDDLEWARE_SESSION_TIMEOUT_SECONDS)
+                access_expires_at = id_claims.get('exp', session_expires_at)
             request.session[constants.SESSION_ID_TOKEN] = id_token
             request.session[constants.SESSION_ACCESS_TOKEN] = access_token
-            request.session[constants.SESSION_ACCESS_EXPIRES_AT] = expires_at
-            request.session[constants.SESSION_EXPIRES_AT] = now + settings.OIDC_MIDDLEWARE_SESSION_TIMEOUT_SECONDS
+            request.session[constants.SESSION_ACCESS_EXPIRES_AT] = access_expires_at
+            request.session[constants.SESSION_EXPIRES_AT] = session_expires_at
             if refresh_token:
                 request.session[constants.SESSION_REFRESH_TOKEN] = refresh_token
             user = self.get_or_create_user(request, id_claims, access_token)
             return user
         except Exception as e:
-            warning(e, str(e))
+            warning(str(e), exc_info=True)
             return None
         finally:
             # be sure the session is in sync
             request.session.save()
 
 
 class BearerAuthenticationBackend(ModelBackend, AuthenticationMixin, OIDCUrlsMixin):
-    def __init__(self, *args, **kwargs):
+    def __init__(self, *args, **kwargs) -> None:
         super().__init__(*args, **kwargs)
         self.authorization_prefix = settings.OIDC_AUTHORIZATION_HEADER_PREFIX
         self.oidc_urls = self.get_oidc_urls({})
 
     def authenticate(self, request: HttpRequest, username: Optional[str] = None, password: Optional[str] = None, **kwargs) -> Optional[AbstractBaseUser]:
         """Authenticates users using the Authorization header and previous OIDC Id Token."""
-        try:
-            prefix, id_token = request.headers.get('Authorization', ' ').split(' ', 1)
-        except ValueError:
-            prefix = id_token = ''
+        auth_header = request.headers.get('Authorization', '')
+        prefix, id_token = auth_header.split(' ', 1) if ' ' in auth_header else ('', '')
         if not prefix or not id_token:
             return None
         try:
             if prefix != self.authorization_prefix:
                 raise SuspiciousOperation(f"Authorization should start with a {self.authorization_prefix} prefix")
             if BlacklistedToken.is_blacklisted(id_token):
                 raise SuspiciousOperation(f"token {id_token} is blacklisted")
-            id_claims = self.validate_and_decode_id_token(id_token, nonce=None, jwks=self.oidc_urls.get(constants.SESSION_OP_JWKS, []))
+            id_claims = self.validate_and_decode_id_token(id_token, nonce=None, jwks=self.oidc_urls.get(constants.SESSION_OP_JWKS, {}))
             user = self.get_or_create_user(request, id_claims, '')
             return user
         except Exception as e:
-            warning(e.args[0])
+            warning(str(e), exc_info=True)
             return None

Diff for: oauth2_authcodeflow/conf.py

@@ -1,20 +1,22 @@
 from base64 import urlsafe_b64encode
 from hashlib import sha1
+from types import FunctionType
 from typing import (
     Any,
     Callable,
     Dict,
+    Optional,
+    Set,
     Tuple,
 )
 
 from django.conf import settings as dj_settings
 from django.core.exceptions import ImproperlyConfigured
+from django.core.signals import setting_changed
 from django.utils.module_loading import import_string
 
 from . import constants
 
-assert constants
-
 
 def import_string_as_func(value: str, attr: str) -> Callable:
     try:
@@ -30,8 +32,6 @@ def get_default_django_username(claims: Dict) -> str:
     ).decode('ascii').rstrip('=')
 
 
-func = type(lambda: 0)
-
 DEFAULTS: Dict[str, Tuple[Any, Any]] = {
     'OIDC_VIEW_AUTHENTICATE': (type, f'{__package__}.views.AuthenticateView'),
     'OIDC_VIEW_CALLBACK': (type, f'{__package__}.views.CallbackView'),
@@ -114,18 +114,18 @@ def get_default_django_username(claims: Dict) -> str:
     # Function or dotted path to a function that compute the django username based on claims.
     # The username should be unique for this app.
     # The default is to use a base64 encode of the email hash (sha1).
-    'OIDC_DJANGO_USERNAME_FUNC': (func, get_default_django_username),
+    'OIDC_DJANGO_USERNAME_FUNC': (FunctionType, get_default_django_username),
     # Default, by value None, is the value of `OIDC_OP_EXPECTED_EMAIL_CLAIM`
     # You can also provide a lambda that takes all the claims as argument and return an email
-    'OIDC_EMAIL_CLAIM': ((str, func), None),
+    'OIDC_EMAIL_CLAIM': ((str, FunctionType), None),
     # You can also provide a lambda that takes all the claims as argument and return a firstname
-    'OIDC_FIRSTNAME_CLAIM': ((str, func), 'given_name'),
+    'OIDC_FIRSTNAME_CLAIM': ((str, FunctionType), 'given_name'),
     # You can also provide a lambda that takes all the claims as argument and return a lastname
-    'OIDC_LASTNAME_CLAIM': ((str, func), 'family_name'),
+    'OIDC_LASTNAME_CLAIM': ((str, FunctionType), 'family_name'),
     # Callable (that takes the user, the claims and optionaly the request and access_token as arguments)
     # to extend user with other potential additional information available in the claims or from another request
     # You can also specify a dotted path to a callable
-    'OIDC_EXTEND_USER': (func, None),
+    'OIDC_EXTEND_USER': (FunctionType, None),
     # Scramble the password on each SSO connection/renewal. If False, it will only scramble it when creating an account.
     'OIDC_UNUSABLE_PASSWORD': (bool, True),
     'OIDC_BLACKLIST_TOKEN_TIMEOUT_SECONDS': (int, 7 * 86400),  # 7 days
@@ -150,28 +150,53 @@ class Settings:
     """
     A settings object, that allows settings to be accessed as properties.
     """
-    def __init__(self, defaults=None):
+    defaults: Dict[str, Tuple[Any, Any]]
+
+    def __init__(self, defaults: Optional[Dict[str, Tuple[Any, Any]]] = None) -> None:
         self.defaults = defaults or DEFAULTS
+        self._cache: Set[str] = set()
+        setting_changed.connect(self.reload)
 
-    def __getattr__(self, attr):
+    def __getattr__(self, attr: str) -> Any:
         atype, def_val = self.defaults.get(attr, (None, None))
         val = getattr(dj_settings, attr, def_val)
         if atype is None:  # other setting
             atype = type(val),
         elif not isinstance(atype, tuple):
             atype = atype,
         val = self._check_type_and_get_value(attr, val, atype)
-        setattr(self, attr, val)  # cache the result
+        # cache the result
+        self._cache.add(attr)
+        setattr(self, attr, val)
         return val
 
     def _check_type_and_get_value(self, attr: str, val: Any, atype: tuple) -> Any:
-        if (func in atype or type in atype) and isinstance(val, str) and '.' in val:
+        if (FunctionType in atype or type in atype) and isinstance(val, str) and '.' in val:
             val = import_string_as_func(val, attr)  # dotted strings to function
         if val is ImproperlyConfigured:
             raise ImproperlyConfigured(f"Setting '{attr}' not found, it should be defined")
         if val is not None and not isinstance(val, atype):
-            raise ImproperlyConfigured(f"Invalid setting: '{attr}' should be of type {atype} and is of type {type(val)}")
+            raise ImproperlyConfigured(f"Invalid setting: '{attr}' should be of type {', '.join(map(str, atype))} and is of type {type(val)}")
         return val
 
+    def reload(self, *args, **kwargs) -> None:
+        attr = str(kwargs.get('setting'))
+        # hasattr cannot be used because it relies on getattr and __getattr__, so use the _cache
+        if attr in self._cache:
+            self._cache.remove(attr)
+        try:
+            delattr(self, attr)
+        except AttributeError:
+            pass
+
 
 settings = Settings()
+
+
+__all__ = [
+    'Settings',
+    'constants',
+    'get_default_django_username',
+    'import_string_as_func',
+    'settings',
+]

Diff for: oauth2_authcodeflow/management/commands/oidc_urls.py

@@ -26,4 +26,4 @@ def handle(self, *args, **options):
             self.stdout.write(f"redirect_url: {redirect_url}")
             self.stdout.write(f"logout_url: {logout_url}")
         except NoReverseMatch as e:
-            raise CommandError(e)
+            raise CommandError(str(e))