From 7ea52473dcb610a3b66ecddfccfe4df86cb01974 Mon Sep 17 00:00:00 2001 From: Manuel Raynaud Date: Tue, 4 Aug 2026 15:27:15 +0200 Subject: [PATCH] =?UTF-8?q?=E2=9C=A8(backend)=20add=20a=20service=20genera?= =?UTF-8?q?ting=20cached=20RS256=20JWT=20tokens?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit We want to generate jwt token using the RS256 algotrithm. This token will be used for internal call with the yhub service. --- CHANGELOG.md | 1 + documentation/env.md | 2 + src/backend/core/services/jwt_services.py | 117 ++++++++++ .../core/tests/test_services_jwt_services.py | 207 ++++++++++++++++++ src/backend/impress/settings.py | 17 ++ src/backend/pyproject.toml | 2 +- src/backend/uv.lock | 9 +- 7 files changed, 352 insertions(+), 3 deletions(-) create mode 100644 src/backend/core/services/jwt_services.py create mode 100644 src/backend/core/tests/test_services_jwt_services.py diff --git a/CHANGELOG.md b/CHANGELOG.md index e188d43e4..de422ea34 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,7 @@ and this project adheres to ### Added +- ✨(backend) add a service generating cached RS256 JWT tokens - ♿️(frontend) restore skip to content link after header redesign #2510 - 🌐(i18n) rename cn_CN to zh_CN, add eo_PL and zh_TW locales #2486 - ✨(backend) conditional email notification in server to server api #2554 diff --git a/documentation/env.md b/documentation/env.md index 6da1e30e8..65fe38b47 100644 --- a/documentation/env.md +++ b/documentation/env.md @@ -81,6 +81,8 @@ These are the environment variables you can set for the `impress-backend` contai | FRONTEND_JS_URL | To add a external js file to the app | | | FRONTEND_HOMEPAGE_FEATURE_ENABLED | Frontend feature flag to display the homepage | false | | FRONTEND_THEME | Frontend theme to use | | +| JWT_PRIVATE_KEY | PEM encoded RSA private key used to sign the JWT tokens (RS256). Can be read from a file with JWT_PRIVATE_KEY_FILE | | +| JWT_TOKEN_LIFETIME | Lifetime in seconds of the generated JWT tokens. Also used as the cache timeout of these tokens | 3600 | | LANGUAGE_CODE | Default language | en-us | | LANGFUSE_SECRET_KEY | The Langfuse secret key used by the sdk | None | | LANGFUSE_PUBLIC_KEY | The Langfuse public key used by the sdk | None | diff --git a/src/backend/core/services/jwt_services.py b/src/backend/core/services/jwt_services.py new file mode 100644 index 000000000..a57fa6acd --- /dev/null +++ b/src/backend/core/services/jwt_services.py @@ -0,0 +1,117 @@ +"""JWT services.""" + +import hashlib +import json +import logging +from datetime import timedelta + +from django.conf import settings +from django.core.cache import cache +from django.utils import timezone + +import jwt + +logger = logging.getLogger(__name__) + +ALGORITHM = "RS256" +CACHE_KEY_PREFIX = "jwt_token" + + +class JWTError(Exception): + """Base exception for JWT related errors.""" + + +class ConfigurationError(JWTError): + """Raised when the JWT service is not properly configured.""" + + +class TokenGenerationError(JWTError): + """Raised when a token cannot be signed.""" + + +class JWTService: + """ + Service class issuing RS256 signed JSON Web Tokens. + + The claims are injected by the caller at generation time, the service only + owns the signature and the token lifetime. Generated tokens are cached for + their whole lifetime so that repeated calls with the same claims reuse the + same token instead of signing a new one. + """ + + algorithm = ALGORITHM + + @property + def private_key(self): + """Return the RSA private key used to sign the tokens.""" + private_key = settings.JWT_PRIVATE_KEY + if not private_key: + raise ConfigurationError( + "The JWT_PRIVATE_KEY setting is required to sign tokens." + ) + return private_key + + @property + def lifetime(self): + """Return the token lifetime, in seconds.""" + return settings.JWT_TOKEN_LIFETIME + + def get_cache_key(self, claims): + """ + Build the cache key identifying a token for the given claims. + + The signing key and the lifetime are part of the fingerprint so that + rotating the key or changing the lifetime never serves a stale token. + """ + fingerprint = json.dumps( + { + "claims": claims, + "lifetime": self.lifetime, + "key": self.private_key, + }, + sort_keys=True, + default=str, + ) + digest = hashlib.sha256(fingerprint.encode("utf-8")).hexdigest() + return f"{CACHE_KEY_PREFIX}:{digest}" + + def generate_token(self, claims): + """ + Sign a new token embedding the given claims. + + The "iat" and "exp" claims are always set by the service, from the + configured lifetime, and take precedence over the caller's claims. + """ + issued_at = timezone.now() + payload = { + **claims, + "iat": issued_at, + "exp": issued_at + timedelta(seconds=self.lifetime), + } + + try: + return jwt.encode(payload, self.private_key, algorithm=self.algorithm) + except (jwt.PyJWTError, TypeError, ValueError) as err: + logger.exception( + "Unable to sign a JWT token with algorithm %s", self.algorithm + ) + raise TokenGenerationError("Unable to sign the JWT token") from err + + def get_token(self, claims): + """ + Return a token embedding the given claims, generating it if needed. + + The token is cached for its own lifetime, so a cached token can be + returned close to its expiry. Callers needing a guaranteed remaining + validity should account for it in the configured lifetime. + """ + cache_key = self.get_cache_key(claims) + + token = cache.get(cache_key) + if token is not None: + return token + + token = self.generate_token(claims) + cache.set(cache_key, token, self.lifetime) + + return token diff --git a/src/backend/core/tests/test_services_jwt_services.py b/src/backend/core/tests/test_services_jwt_services.py new file mode 100644 index 000000000..423fc3e99 --- /dev/null +++ b/src/backend/core/tests/test_services_jwt_services.py @@ -0,0 +1,207 @@ +""" +This module contains tests for the JWTService class in the +core.services.jwt_services module. +""" + +from datetime import datetime, timezone +from unittest import mock + +from django.core.cache import cache + +import jwt +import pytest +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives.asymmetric import rsa +from freezegun import freeze_time + +from core.services.jwt_services import ( + ConfigurationError, + JWTService, + TokenGenerationError, +) + + +def generate_key_pair(): + """Generate a PEM encoded RSA key pair to sign and verify test tokens.""" + private_key = rsa.generate_private_key(public_exponent=65537, key_size=2048) + private_pem = private_key.private_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PrivateFormat.PKCS8, + encryption_algorithm=serialization.NoEncryption(), + ).decode("utf-8") + public_pem = ( + private_key.public_key() + .public_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PublicFormat.SubjectPublicKeyInfo, + ) + .decode("utf-8") + ) + return private_pem, public_pem + + +# Generating RSA keys is expensive, do it once for the whole module +PRIVATE_KEY, PUBLIC_KEY = generate_key_pair() +OTHER_PRIVATE_KEY, OTHER_PUBLIC_KEY = generate_key_pair() + + +@pytest.fixture(name="jwt_settings") +def jwt_settings_fixture(settings): + """Setup valid settings for the JWT service.""" + settings.JWT_PRIVATE_KEY = PRIVATE_KEY + settings.JWT_TOKEN_LIFETIME = 3600 + return settings + + +@pytest.mark.usefixtures("jwt_settings") +def test_get_token_signs_the_injected_claims_with_rs256(): + """The generated token is signed with RS256 and carries the given claims.""" + token = JWTService().get_token({"sub": "user-id", "abilities": ["read"]}) + + assert jwt.get_unverified_header(token)["alg"] == "RS256" + + payload = jwt.decode(token, PUBLIC_KEY, algorithms=["RS256"]) + assert payload["sub"] == "user-id" + assert payload["abilities"] == ["read"] + + +@pytest.mark.usefixtures("jwt_settings") +def test_get_token_cannot_be_verified_with_another_key(): + """The token is signed with the private key defined in the settings.""" + token = JWTService().get_token({"sub": "user-id"}) + + with pytest.raises(jwt.InvalidSignatureError): + jwt.decode(token, OTHER_PUBLIC_KEY, algorithms=["RS256"]) + + +def test_generate_token_expires_after_the_configured_lifetime(jwt_settings): + """The "iat" and "exp" claims are computed from the configured lifetime.""" + jwt_settings.JWT_TOKEN_LIFETIME = 300 + + now = datetime(2026, 8, 4, 10, 0, 0, tzinfo=timezone.utc) + with freeze_time(now): + token = JWTService().generate_token({"sub": "user-id"}) + payload = jwt.decode(token, PUBLIC_KEY, algorithms=["RS256"]) + + assert payload["iat"] == now.timestamp() + assert payload["exp"] == now.timestamp() + 300 + + +def test_generate_token_ignores_the_expiry_claims_given_by_the_caller(jwt_settings): + """The service owns the token lifetime, the caller cannot extend it.""" + jwt_settings.JWT_TOKEN_LIFETIME = 60 + + now = datetime(2026, 8, 4, 10, 0, 0, tzinfo=timezone.utc) + with freeze_time(now): + token = JWTService().generate_token( + {"sub": "user-id", "iat": 0, "exp": 99999999999} + ) + payload = jwt.decode(token, PUBLIC_KEY, algorithms=["RS256"]) + + assert payload["iat"] == now.timestamp() + assert payload["exp"] == now.timestamp() + 60 + + +@pytest.mark.usefixtures("jwt_settings") +def test_get_token_reuses_the_cached_token(): + """A token already in cache is returned without signing a new one.""" + service = JWTService() + claims = {"sub": "user-id"} + + token = service.get_token(claims) + assert cache.get(service.get_cache_key(claims)) == token + + with mock.patch("core.services.jwt_services.jwt.encode") as mock_encode: + assert service.get_token(claims) == token + + mock_encode.assert_not_called() + + +def test_get_token_caches_the_token_for_its_lifetime(jwt_settings): + """The cache entry expires along with the token it holds.""" + jwt_settings.JWT_TOKEN_LIFETIME = 300 + + service = JWTService() + claims = {"sub": "user-id"} + + with freeze_time("2026-08-04 10:00:00") as frozen_time: + token = service.get_token(claims) + + frozen_time.move_to("2026-08-04 10:04:59") + assert cache.get(service.get_cache_key(claims)) == token + + frozen_time.move_to("2026-08-04 10:05:01") + assert cache.get(service.get_cache_key(claims)) is None + + +@pytest.mark.usefixtures("jwt_settings") +def test_get_token_caches_each_set_of_claims_separately(): + """Two different sets of claims get two different tokens.""" + service = JWTService() + + first_token = service.get_token({"sub": "user-id"}) + second_token = service.get_token({"sub": "other-user-id"}) + + assert first_token != second_token + assert jwt.decode(first_token, PUBLIC_KEY, algorithms=["RS256"])["sub"] == "user-id" + assert ( + jwt.decode(second_token, PUBLIC_KEY, algorithms=["RS256"])["sub"] + == "other-user-id" + ) + + +@pytest.mark.usefixtures("jwt_settings") +def test_get_token_ignores_the_claims_ordering(): + """Claims given in a different order hit the same cache entry.""" + service = JWTService() + + assert service.get_cache_key({"a": 1, "b": 2}) == service.get_cache_key( + {"b": 2, "a": 1} + ) + + +def test_get_token_generates_a_new_token_after_a_key_rotation(jwt_settings): + """A token signed with a rotated out key is never served from the cache.""" + service = JWTService() + claims = {"sub": "user-id"} + + service.get_token(claims) + + jwt_settings.JWT_PRIVATE_KEY = OTHER_PRIVATE_KEY + token = service.get_token(claims) + + assert jwt.decode(token, OTHER_PUBLIC_KEY, algorithms=["RS256"])["sub"] == "user-id" + + +def test_get_token_generates_a_new_token_when_the_lifetime_changes(jwt_settings): + """A token cached with the former lifetime is never served.""" + jwt_settings.JWT_TOKEN_LIFETIME = 300 + + service = JWTService() + claims = {"sub": "user-id"} + + with freeze_time("2026-08-04 10:00:00"): + service.get_token(claims) + + jwt_settings.JWT_TOKEN_LIFETIME = 600 + token = service.get_token(claims) + payload = jwt.decode(token, PUBLIC_KEY, algorithms=["RS256"]) + + assert payload["exp"] - payload["iat"] == 600 + + +@pytest.mark.parametrize("private_key", [None, ""]) +def test_get_token_without_private_key(jwt_settings, private_key): + """The service refuses to issue a token when no private key is configured.""" + jwt_settings.JWT_PRIVATE_KEY = private_key + + with pytest.raises(ConfigurationError, match="JWT_PRIVATE_KEY"): + JWTService().get_token({"sub": "user-id"}) + + +def test_generate_token_with_an_invalid_private_key(jwt_settings): + """An unusable private key is reported as a token generation error.""" + jwt_settings.JWT_PRIVATE_KEY = "not-a-pem-key" + + with pytest.raises(TokenGenerationError, match="Unable to sign the JWT token"): + JWTService().generate_token({"sub": "user-id"}) diff --git a/src/backend/impress/settings.py b/src/backend/impress/settings.py index a6a2f79ff..173bf5c2d 100755 --- a/src/backend/impress/settings.py +++ b/src/backend/impress/settings.py @@ -540,6 +540,23 @@ class Base(Configuration): environ_prefix=None, ) + # JWT + # RSA private key (PEM) used to sign the tokens issued by + # core.services.jwt_services.JWTService. Prefer the JWT_PRIVATE_KEY_FILE + # environment variable, a PEM does not fit well in an environment variable. + JWT_PRIVATE_KEY = SecretFileValue( + None, + environ_name="JWT_PRIVATE_KEY", + environ_prefix=None, + ) + # Lifetime, in seconds, of the tokens issued by the JWT service. It is both + # the "exp" claim horizon and the cache timeout of the generated tokens. + JWT_TOKEN_LIFETIME = values.IntegerValue( + default=3600, + environ_name="JWT_TOKEN_LIFETIME", + environ_prefix=None, + ) + # Frontend FRONTEND_THEME = values.Value( None, environ_name="FRONTEND_THEME", environ_prefix=None diff --git a/src/backend/pyproject.toml b/src/backend/pyproject.toml index cfbf3d367..82cb5ca46 100644 --- a/src/backend/pyproject.toml +++ b/src/backend/pyproject.toml @@ -61,7 +61,7 @@ dependencies = [ "pycrdt==0.14.1", "pydantic==2.13.4", "pydantic-ai-slim[openai,mistral,logfire,web]==1.107.1", - "PyJWT==2.13.0", + "PyJWT[crypto]==2.13.0", "python-magic==0.4.27", "redis<6.0.0", "requests==2.34.2", diff --git a/src/backend/uv.lock b/src/backend/uv.lock index 5db67ffad..e6fd18c29 100644 --- a/src/backend/uv.lock +++ b/src/backend/uv.lock @@ -965,7 +965,7 @@ dependencies = [ { name = "pycrdt" }, { name = "pydantic" }, { name = "pydantic-ai-slim", extra = ["logfire", "mistral", "openai", "web"] }, - { name = "pyjwt" }, + { name = "pyjwt" , extra = ["crypto"] }, { name = "python-magic" }, { name = "redis" }, { name = "requests" }, @@ -1042,7 +1042,7 @@ requires-dist = [ { name = "pydantic", specifier = "==2.13.4" }, { name = "pydantic-ai-slim", extras = ["openai", "mistral", "logfire", "web"], specifier = "==1.107.1" }, { name = "pyfakefs", marker = "extra == 'dev'", specifier = "==6.2.0" }, - { name = "pyjwt", specifier = "==2.13.0" }, + { name = "pyjwt", extras = ["crypto"], specifier = "==2.13.0" }, { name = "pylint", marker = "extra == 'dev'", specifier = "==4.0.6" }, { name = "pylint-django", marker = "extra == 'dev'", specifier = "==2.8.0" }, { name = "pytest", marker = "extra == 'dev'", specifier = "==9.1.1" }, @@ -1978,6 +1978,11 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/a3/5e/ecf12fdb62546d64385c158514e9b2b671f7832108ef2ecd2020ce0af2d1/pyjwt-2.13.0-py3-none-any.whl", hash = "sha256:66adcc2aff09b3f1bbd95fc1e1577df8ac8723c978552fd43304c8a290ac5728", size = 31274, upload-time = "2026-05-21T19:54:35.362Z" }, ] +[package.optional-dependencies] +crypto = [ + { name = "cryptography" }, +] + [[package]] name = "pylint" version = "4.0.6"