✨(oidc) encrypt the refresh token in session
Enforce refresh token encryption for the session storage.
This commit is contained in:
@@ -8,6 +8,10 @@ and this project adheres to
|
||||
|
||||
## [Unreleased]
|
||||
|
||||
## Added
|
||||
|
||||
- ✨(oidc) add refresh token tools #584
|
||||
|
||||
## [2.5.0] - 2025-03-18
|
||||
|
||||
## Added
|
||||
|
||||
@@ -4,3 +4,6 @@ BURST_THROTTLE_RATES="200/minute"
|
||||
DJANGO_SERVER_TO_SERVER_API_TOKENS=test-e2e
|
||||
Y_PROVIDER_API_KEY=yprovider-api-key
|
||||
Y_PROVIDER_API_BASE_URL=http://y-provider:4444/api/
|
||||
|
||||
# - add a key to store the refresh token in tests
|
||||
OIDC_STORE_REFRESH_TOKEN_KEY=qnw7gZrOFLkLuZIixzuxksNORFJyjWyi5ACugNchKJY=
|
||||
|
||||
@@ -1,12 +1,14 @@
|
||||
"""Authentication Backends for the Impress core app."""
|
||||
|
||||
import logging
|
||||
from functools import lru_cache
|
||||
|
||||
from django.conf import settings
|
||||
from django.core.exceptions import SuspiciousOperation
|
||||
from django.utils.translation import gettext_lazy as _
|
||||
|
||||
import requests
|
||||
from cryptography.fernet import Fernet
|
||||
from mozilla_django_oidc.auth import (
|
||||
OIDCAuthenticationBackend as MozillaOIDCAuthenticationBackend,
|
||||
)
|
||||
@@ -17,10 +19,28 @@ from core.models import DuplicateEmailError, User
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
@lru_cache(maxsize=0)
|
||||
def get_cipher_suite():
|
||||
"""Return a Fernet cipher suite."""
|
||||
key = import_from_settings("OIDC_STORE_REFRESH_TOKEN_KEY", None)
|
||||
if not key:
|
||||
raise ValueError("OIDC_STORE_REFRESH_TOKEN_KEY setting is required.")
|
||||
return Fernet(key)
|
||||
|
||||
|
||||
def store_oidc_refresh_token(session, refresh_token):
|
||||
"""Store the OIDC refresh token in the session if enabled in settings."""
|
||||
"""Store the encrypted OIDC refresh token in the session if enabled in settings."""
|
||||
if import_from_settings("OIDC_STORE_REFRESH_TOKEN", False):
|
||||
session["oidc_refresh_token"] = refresh_token
|
||||
encrypted_token = get_cipher_suite().encrypt(refresh_token.encode())
|
||||
session["oidc_refresh_token"] = encrypted_token.decode()
|
||||
|
||||
|
||||
def get_oidc_refresh_token(session):
|
||||
"""Retrieve and decrypt the OIDC refresh token from the session."""
|
||||
encrypted_token = session.get("oidc_refresh_token")
|
||||
if encrypted_token:
|
||||
return get_cipher_suite().decrypt(encrypted_token.encode()).decode()
|
||||
return None
|
||||
|
||||
|
||||
def store_tokens(session, access_token, id_token, refresh_token):
|
||||
|
||||
@@ -38,7 +38,7 @@ from mozilla_django_oidc.utils import (
|
||||
import_from_settings,
|
||||
)
|
||||
|
||||
from core.authentication.backends import store_tokens
|
||||
from core.authentication.backends import get_oidc_refresh_token, store_tokens
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
@@ -138,7 +138,7 @@ class RefreshOIDCAccessToken(SessionRefresh):
|
||||
token_url = self.get_settings("OIDC_OP_TOKEN_ENDPOINT")
|
||||
client_id = self.get_settings("OIDC_RP_CLIENT_ID")
|
||||
client_secret = self.get_settings("OIDC_RP_CLIENT_SECRET")
|
||||
refresh_token = request.session.get("oidc_refresh_token")
|
||||
refresh_token = get_oidc_refresh_token(request.session)
|
||||
|
||||
if not refresh_token:
|
||||
logger.debug("no refresh token stored")
|
||||
|
||||
@@ -10,14 +10,37 @@ from django.test.utils import override_settings
|
||||
|
||||
import pytest
|
||||
import responses
|
||||
from cryptography.fernet import Fernet
|
||||
|
||||
from core import models
|
||||
from core.authentication.backends import OIDCAuthenticationBackend
|
||||
from core.authentication.backends import (
|
||||
OIDCAuthenticationBackend,
|
||||
get_oidc_refresh_token,
|
||||
store_oidc_refresh_token,
|
||||
)
|
||||
from core.factories import UserFactory
|
||||
|
||||
pytestmark = pytest.mark.django_db
|
||||
|
||||
|
||||
def test_oidc_refresh_token_session_store(settings):
|
||||
"""Test that the OIDC refresh token is stored and retrieved from the session."""
|
||||
session = {}
|
||||
|
||||
with pytest.raises(
|
||||
ValueError, match="OIDC_STORE_REFRESH_TOKEN_KEY setting is required."
|
||||
):
|
||||
store_oidc_refresh_token(session, "test-refresh-token")
|
||||
|
||||
settings.OIDC_STORE_REFRESH_TOKEN_KEY = Fernet.generate_key()
|
||||
|
||||
store_oidc_refresh_token(session, "test-refresh-token")
|
||||
assert session["oidc_refresh_token"] is not None
|
||||
assert session["oidc_refresh_token"] != "test-refresh-token"
|
||||
|
||||
assert get_oidc_refresh_token(session) == "test-refresh-token"
|
||||
|
||||
|
||||
def test_authentication_getter_existing_user_no_email(
|
||||
django_assert_num_queries, monkeypatch
|
||||
):
|
||||
@@ -561,6 +584,7 @@ def test_authentication_session_tokens(
|
||||
settings.OIDC_OP_JWKS_ENDPOINT = "http://oidc.endpoint.test/jwks"
|
||||
settings.OIDC_STORE_ACCESS_TOKEN = True
|
||||
settings.OIDC_STORE_REFRESH_TOKEN = True
|
||||
settings.OIDC_STORE_REFRESH_TOKEN_KEY = Fernet.generate_key()
|
||||
|
||||
klass = OIDCAuthenticationBackend()
|
||||
request = rf.get("/some-url", {"state": "test-state", "code": "test-code"})
|
||||
@@ -598,4 +622,4 @@ def test_authentication_session_tokens(
|
||||
|
||||
assert user is not None
|
||||
assert request.session["oidc_access_token"] == "test-access-token"
|
||||
assert request.session["oidc_refresh_token"] == "test-refresh-token"
|
||||
assert get_oidc_refresh_token(request.session) == "test-refresh-token"
|
||||
|
||||
@@ -11,8 +11,14 @@ from django.test import RequestFactory
|
||||
import pytest
|
||||
import requests.exceptions
|
||||
import responses
|
||||
from cryptography.fernet import Fernet
|
||||
|
||||
from core import factories
|
||||
from core.authentication.backends import (
|
||||
get_cipher_suite,
|
||||
get_oidc_refresh_token,
|
||||
store_oidc_refresh_token,
|
||||
)
|
||||
from core.authentication.middleware import RefreshOIDCAccessToken
|
||||
|
||||
pytestmark = pytest.mark.django_db
|
||||
@@ -34,7 +40,13 @@ def fixture_oidc_settings(settings):
|
||||
settings.OIDC_TOKEN_USE_BASIC_AUTH = False
|
||||
settings.OIDC_STORE_ACCESS_TOKEN = True
|
||||
settings.OIDC_STORE_REFRESH_TOKEN = True
|
||||
return settings
|
||||
settings.OIDC_STORE_REFRESH_TOKEN_KEY = Fernet.generate_key()
|
||||
|
||||
get_cipher_suite.cache_clear()
|
||||
|
||||
yield settings
|
||||
|
||||
get_cipher_suite.cache_clear()
|
||||
|
||||
|
||||
def test_anonymous_user(oidc_settings): # pylint: disable=unused-argument
|
||||
@@ -94,7 +106,7 @@ def test_basic_auth_disabled(oidc_settings): # pylint: disable=unused-argument
|
||||
session_middleware.process_request(request)
|
||||
|
||||
request.session["oidc_access_token"] = "old_token"
|
||||
request.session["oidc_refresh_token"] = "refresh_token"
|
||||
store_oidc_refresh_token(request.session, "refresh_token")
|
||||
request.session["oidc_token_expiration"] = time.time() - 100
|
||||
request.session.save()
|
||||
|
||||
@@ -118,7 +130,7 @@ def test_successful_token_refresh(oidc_settings): # pylint: disable=unused-argu
|
||||
session_middleware.process_request(request)
|
||||
|
||||
request.session["oidc_access_token"] = "old_token"
|
||||
request.session["oidc_refresh_token"] = "refresh_token"
|
||||
store_oidc_refresh_token(request.session, "refresh_token")
|
||||
request.session["oidc_token_expiration"] = time.time() - 100
|
||||
request.session.save()
|
||||
|
||||
@@ -135,7 +147,7 @@ def test_successful_token_refresh(oidc_settings): # pylint: disable=unused-argu
|
||||
|
||||
assert response is None
|
||||
assert request.session["oidc_access_token"] == "new_token"
|
||||
assert request.session["oidc_refresh_token"] == "new_refresh_token"
|
||||
assert get_oidc_refresh_token(request.session) == "new_refresh_token"
|
||||
|
||||
|
||||
def test_non_expired_token(oidc_settings): # pylint: disable=unused-argument
|
||||
@@ -169,7 +181,7 @@ def test_refresh_token_request_timeout(oidc_settings): # pylint: disable=unused
|
||||
session_middleware = SessionMiddleware(get_response)
|
||||
session_middleware.process_request(request)
|
||||
request.session["oidc_access_token"] = "old_token"
|
||||
request.session["oidc_refresh_token"] = "refresh_token"
|
||||
store_oidc_refresh_token(request.session, "refresh_token")
|
||||
request.session["oidc_token_expiration"] = time.time() - 100
|
||||
request.session.save()
|
||||
|
||||
@@ -200,7 +212,7 @@ def test_refresh_token_request_error_400(oidc_settings): # pylint: disable=unus
|
||||
session_middleware = SessionMiddleware(get_response)
|
||||
session_middleware.process_request(request)
|
||||
request.session["oidc_access_token"] = "old_token"
|
||||
request.session["oidc_refresh_token"] = "refresh_token"
|
||||
store_oidc_refresh_token(request.session, "refresh_token")
|
||||
request.session["oidc_token_expiration"] = time.time() - 100
|
||||
request.session.save()
|
||||
|
||||
@@ -233,7 +245,7 @@ def test_refresh_token_request_error(oidc_settings): # pylint: disable=unused-a
|
||||
session_middleware = SessionMiddleware(get_response)
|
||||
session_middleware.process_request(request)
|
||||
request.session["oidc_access_token"] = "old_token"
|
||||
request.session["oidc_refresh_token"] = "refresh_token"
|
||||
store_oidc_refresh_token(request.session, "refresh_token")
|
||||
request.session["oidc_token_expiration"] = time.time() - 100
|
||||
request.session.save()
|
||||
|
||||
@@ -265,7 +277,7 @@ def test_refresh_token_request_malformed_json_error(oidc_settings): # pylint: d
|
||||
session_middleware = SessionMiddleware(get_response)
|
||||
session_middleware.process_request(request)
|
||||
request.session["oidc_access_token"] = "old_token"
|
||||
request.session["oidc_refresh_token"] = "refresh_token"
|
||||
store_oidc_refresh_token(request.session, "refresh_token")
|
||||
request.session["oidc_token_expiration"] = time.time() - 100
|
||||
request.session.save()
|
||||
|
||||
@@ -297,7 +309,7 @@ def test_refresh_token_request_exception(oidc_settings): # pylint: disable=unus
|
||||
session_middleware = SessionMiddleware(get_response)
|
||||
session_middleware.process_request(request)
|
||||
request.session["oidc_access_token"] = "old_token"
|
||||
request.session["oidc_refresh_token"] = "refresh_token"
|
||||
store_oidc_refresh_token(request.session, "refresh_token")
|
||||
request.session["oidc_token_expiration"] = time.time() - 100
|
||||
request.session.save()
|
||||
|
||||
|
||||
@@ -493,6 +493,11 @@ class Base(Configuration):
|
||||
OIDC_STORE_REFRESH_TOKEN = values.BooleanValue(
|
||||
default=True, environ_name="OIDC_STORE_REFRESH_TOKEN", environ_prefix=None
|
||||
)
|
||||
OIDC_STORE_REFRESH_TOKEN_KEY = values.Value(
|
||||
default=None,
|
||||
environ_name="OIDC_STORE_REFRESH_TOKEN_KEY",
|
||||
environ_prefix=None,
|
||||
)
|
||||
|
||||
# WARNING: Enabling this setting allows multiple user accounts to share the same email
|
||||
# address. This may cause security issues and is not recommended for production use when
|
||||
|
||||
Reference in New Issue
Block a user