mypy.ini

@@ -12,7 +12,7 @@ plugins = mypy_django_plugin.main, mypy_drf_plugin.main
 django_settings_module = swh.auth.tests.django.app.apptest.settings
 
 # 3rd party libraries without stubs (yet)
-[mypy-jose.*]
+[mypy-jwcrypto.*]
 ignore_missing_imports = True
 
 [mypy-keycloak.*]

requirements.txt

@@ -2,6 +2,5 @@
 # should match https://pypi.python.org/pypi names. For the full spec or
 # dependency lines, see https://pip.readthedocs.org/en/1.1/requirements.html
 click
-python-keycloak >= 0.19.0, <3.9
-python-jose
+python-keycloak >= 4
 pyyaml

swh/auth/cli.py

@@ -275,7 +275,10 @@ def auth_config(ctx: Context, username: str, token: str):
         ctx.fail(keycloak_error_message(ke))
 
     # Save auth configuration file?
+
+    msg = f"Skipping write of authentication configuration file {config_file}"
     if not click.confirm(f"Save authentication settings to {config_file}?"):
+        click.echo(click.style(msg, fg="yellow"))
         ctx.exit(0)
 
     # Save configuration to file

swh/auth/keycloak.py

-# Copyright (C) 2020-2022  The Software Heritage developers
+# Copyright (C) 2020-2024  The Software Heritage developers
 # See the AUTHORS file at the top-level directory of this distribution
 # License: GNU Affero General Public License version 3, or any later version
 # See top-level LICENSE file for more information
 
 import json
-from typing import Any, Dict, Optional
+from typing import Any, Dict
 from urllib.parse import parse_qs, urlencode, urlparse, urlunparse
 
-# add ExpiredSignatureError alias to avoid leaking jose import
+# add ExpiredSignatureError alias to avoid leaking jwcrypto import
 # in swh-auth client code
-from jose.jwt import ExpiredSignatureError  # noqa
+from jwcrypto.jwt import JWTExpired as ExpiredSignatureError  # noqa
 from keycloak import KeycloakOpenID
 
 # add KeycloakError alias to avoid leaking keycloak import
@@ -152,42 +152,35 @@ class KeycloakOpenIDConnect:
         Request a new access token from Keycloak using a refresh token.
 
         Args:
-            refresh_token: A refresh token provided by Keycloak
+            refresh_token: a refresh token provided by Keycloak
 
         Returns:
-            A dictionary filled with tokens info
+            a dictionary filled with tokens info
         """
         return self._keycloak.refresh_token(refresh_token)
 
     def decode_token(
-        self, token: str, options: Optional[Dict[str, Any]] = None
+        self, token: str, validate: bool = True, **kwargs
     ) -> Dict[str, Any]:
         """
         Try to decode a JWT token.
 
         Args:
-            token: A JWT token to decode
-            options: Options for jose.jwt.decode
+            token: a JWT token to decode
+            validate: whether to validate the token
+            kwargs: additional keyword arguments for jwcrypto's JWT object
 
         Returns:
-            A dictionary filled with decoded token content
+            a dictionary filled with decoded token content
         """
-        if not self.realm_public_key:
-            realm_public_key = self._keycloak.public_key()
-            self.realm_public_key = "-----BEGIN PUBLIC KEY-----\n"
-            self.realm_public_key += realm_public_key
-            self.realm_public_key += "\n-----END PUBLIC KEY-----"
-
-        return self._keycloak.decode_token(
-            token, key=self.realm_public_key, options=options
-        )
+        return self._keycloak.decode_token(token, validate=validate, **kwargs)
 
     def logout(self, refresh_token: str) -> None:
         """
         Logout a user by closing its authenticated session.
 
         Args:
-            refresh_token: A refresh token provided by Keycloak
+            refresh_token: a refresh token provided by Keycloak
         """
         self._keycloak.logout(refresh_token)
 
@@ -196,10 +189,10 @@ class KeycloakOpenIDConnect:
         Return user information from its access token.
 
         Args:
-            access_token: An access token provided by Keycloak
+            access_token: an access token provided by Keycloak
 
         Returns:
-            A dictionary fillled with user information
+            a dictionary fillled with user information
         """
         return self._keycloak.userinfo(access_token)
 

swh/auth/pytest_plugin.py

-# Copyright (C) 2020-2021 The Software Heritage developers
+# Copyright (C) 2020-2024 The Software Heritage developers
 # See the AUTHORS file at the top-level directory of this distribution
 # License: GNU Affero General Public License version 3, or any later version
 # See top-level LICENSE file for more information
@@ -104,14 +104,10 @@ class KeycloackOpenIDConnectMock(KeycloakOpenIDConnect):
 
         self.set_auth_success(auth_success, oidc_profile, user_info)
 
-    def decode_token(self, token):
-        options = {}
-        if self.auth_success:
-            # skip signature expiration and audience checks as we use a static
-            # oidc_profile for the tests with expired tokens in it
-            options["verify_exp"] = False
-            options["verify_aud"] = False
-        decoded = super().decode_token(token, options)
+    def decode_token(self, token, **kwargs):
+        # skip signature expiration and audience checks as we use a static
+        # oidc_profile for the tests with expired tokens in it
+        decoded = super().decode_token(token, validate=False, **kwargs)
         # Merge the user info configured to be part of the decode token
         userinfo = self.userinfo()
         if userinfo is not None:

swh/auth/starlette/backends.py

-# Copyright (C) 2023 The Software Heritage developers
+# Copyright (C) 2023-2024 The Software Heritage developers
 # See the AUTHORS file at the top-level directory of this distribution
 # License: GNU General Public License version 3, or any later version
 # See top-level LICENSE file for more information
@@ -8,7 +8,7 @@ import hashlib
 from typing import Any, Dict, Optional, Tuple
 
 from aiocache.base import BaseCache
-from jose.exceptions import JWTError
+from jwcrypto.common import JWException
 from starlette.authentication import (
     AuthCredentials,
     AuthenticationBackend,
@@ -78,7 +78,7 @@ class BearerTokenAuthBackend(AuthenticationBackend):
             return None
         try:
             decoded_token = self.oidc_client.decode_token(access_token)
-        except (KeycloakError, UnicodeEncodeError, ExpiredSignatureError):
+        except (KeycloakError, UnicodeEncodeError, ExpiredSignatureError, ValueError):
             # token is eitehr too old or an invalid one
             decoded_token = None
         return decoded_token
@@ -98,9 +98,8 @@ class BearerTokenAuthBackend(AuthenticationBackend):
             decoded_token = self._decode_token(token)
             if not decoded_token:
                 raise AuthenticationError("Access token failed to be decoded")
-        except JWTError:
+        except JWException:
             # token is a refresh one so backend handles access token renewal
-
             # get the cache key
             cache_key = self._get_token_cache_key(token)
             # read access token from the cache

swh/auth/tests/starlette/test_backend.py

-# Copyright (C) 2023 The Software Heritage developers
+# Copyright (C) 2023-2024 The Software Heritage developers
 # See the AUTHORS file at the top-level directory of this distribution
 # License: GNU General Public License version 3, or any later version
 # See top-level LICENSE file for more information
 
 from aiocache import Cache
+from jwcrypto.jws import InvalidJWSSignature
 import pytest
 from starlette.applications import Starlette
 from starlette.middleware import Middleware
@@ -13,7 +14,7 @@ from starlette.routing import Route
 from starlette.testclient import TestClient
 
 from swh.auth.starlette import backends
-from swh.auth.tests.sample_data import USER_INFO
+from swh.auth.tests.sample_data import DECODED_TOKEN, USER_INFO
 
 
 @pytest.fixture
@@ -71,12 +72,21 @@ def test_invalid_refresh_token(client, keycloak_oidc):
     client.headers = {"Authorization": "Bearer invalid-valid-token"}
     response = client.get("/")
     assert response.status_code == 400
-    assert "Invalid or expired user token" in response.text
+    assert "Access token failed to be decoded" in response.text
 
 
 @pytest.mark.parametrize("token_type", ["access_token", "refresh_token"])
-def test_success_with_tokens(client, keycloak_oidc, token_type):
+def test_success_with_tokens(client, keycloak_oidc, mocker, token_type):
     oidc_profile = keycloak_oidc.login()
+    if token_type == "refresh_token":
+        # simulate invalid decoding of refresh token then valid decoding of
+        # new access token as JWT validation is disabled in keycloak_oidc fixture
+        # due to the use of expired tokens
+        mocker.patch.object(
+            keycloak_oidc,
+            "decode_token",
+            side_effect=[InvalidJWSSignature(), DECODED_TOKEN],
+        )
     client.headers = {"Authorization": f"Bearer {oidc_profile[token_type]}"}
     response = client.get("/")