mirror of
https://github.com/suitenumerique/docs.git
synced 2026-08-17 21:25:43 +02:00
✨(backend) add a service generating cached RS256 JWT tokens
We want to generate jwt token using the RS256 algotrithm. This token will be used for internal call with the yhub service.
This commit is contained in:
Manuel Raynaud
@@ -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
|
||||
|
||||
@@ -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 |
|
||||
|
||||
@@ -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
|
||||
@@ -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"})
|
||||
@@ -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
|
||||
|
||||
@@ -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",
|
||||
|
||||
Generated
+7
-2
@@ -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"
|
||||
|
||||
Reference in New Issue
Block a user