♻️(backend) audience is an enum to be used by the JWTService

To ease the use of the audience with the JWTService, we choose to create
an enum holding all the possible values and then use them in the Yhub
and Y-converter services.
This commit is contained in:
Manuel Raynaud
2026-08-13 14:32:06 +02:00
parent 9d9c706559
commit 3b9318de70
6 changed files with 76 additions and 33 deletions
@@ -9,15 +9,10 @@ from django.conf import settings
import requests
from core.services import mime_types
from core.services.jwt_services import JWTService
from core.services.jwt_services import Audiences, JWTService
logger = logging.getLogger(__name__)
# Audience of the admin token y-provider expects. Scoping the token to it
# prevents an admin JWT issued for another backend service from being
# replayed against y-provider.
Y_CONVERTER_AUDIENCE = "y-converter"
class ConversionError(Exception):
"""Base exception for conversion-related errors."""
@@ -115,7 +110,7 @@ class YdocConverter:
@property
def auth_header(self):
"""Build microservice authentication header."""
token = JWTService().get_admin_token({"aud": Y_CONVERTER_AUDIENCE})
token = JWTService().get_admin_token(audience=Audiences.Y_CONVERTER)
return f"Bearer {token}"
def _request(self, url, data, content_type, accept):
+10 -3
View File
@@ -5,6 +5,7 @@ import hashlib
import json
import logging
from datetime import timedelta
from enum import StrEnum
from django.conf import settings
from django.core.cache import cache
@@ -19,6 +20,13 @@ ALGORITHM = "RS256"
CACHE_KEY_PREFIX = "jwt_token"
class Audiences(StrEnum):
"""Enum of the audiences we can use."""
Y_CONVERTER = "y-converter"
YHUB = "yhub"
class JWTError(Exception):
"""Base exception for JWT related errors."""
@@ -172,12 +180,11 @@ class JWTService:
return token
def get_admin_token(self, claims=None):
def get_admin_token(self, audience: Audiences, claims=None):
"""
Return a token with the `admin: true` claim.
Extra claims can be injected alongside it. They cannot turn the "admin"
claim off: a token issued by this method always grants admin.
"""
return self.get_token({**(claims or {}), "admin": True})
return self.get_token({**(claims or {}), "admin": True, "aud": audience})
+5 -2
View File
@@ -23,7 +23,7 @@ from django.conf import settings
import requests
from core.services.jwt_services import JWTService
from core.services.jwt_services import Audiences, JWTService
logger = logging.getLogger(__name__)
@@ -116,7 +116,10 @@ class YHubService:
document without going through the abilities of a user. The subject it
may carry is who the call is for, it never restricts what it can do.
"""
return f"Bearer {JWTService().get_admin_token(self.claims)}"
token = JWTService().get_admin_token(
audience=Audiences.YHUB, claims=self.claims
)
return f"Bearer {token}"
def build_url(self, endpoint, document_id):
"""Build the url of a document scoped endpoint of the yhub API."""
@@ -13,6 +13,7 @@ from core.services.converter_services import (
ValidationError,
YdocConverter,
)
from core.services.jwt_services import Audiences
from core.tests.utils.jwt_helper import generate_key_pair
# Generating an RSA key is expensive, do it once for the whole module
@@ -34,10 +35,10 @@ def test_auth_header():
assert scheme == "Bearer"
payload = jwt.decode(
token, PUBLIC_KEY, algorithms=["RS256"], audience="y-converter"
token, PUBLIC_KEY, algorithms=["RS256"], audience=Audiences.Y_CONVERTER
)
assert payload["admin"] is True
assert payload["aud"] == "y-converter"
assert payload["aud"] == Audiences.Y_CONVERTER
def test_convert_empty_text():
@@ -13,6 +13,7 @@ import pytest
from freezegun import freeze_time
from core.services.jwt_services import (
Audiences,
ConfigurationError,
JWTService,
TokenGenerationError,
@@ -132,30 +133,43 @@ def test_get_token_caches_each_set_of_claims_separately():
@pytest.mark.usefixtures("jwt_settings")
def test_get_admin_token_carries_the_admin_claim():
"""The admin token is a regular token carrying the "admin" claim."""
token = JWTService().get_admin_token()
token = JWTService().get_admin_token(audience=Audiences.YHUB)
payload = jwt.decode(token, PUBLIC_KEY, algorithms=["RS256"])
payload = jwt.decode(
token, PUBLIC_KEY, algorithms=["RS256"], audience=Audiences.YHUB
)
assert payload["admin"] is True
assert payload["aud"] == Audiences.YHUB
@pytest.mark.usefixtures("jwt_settings")
def test_get_admin_token_embeds_the_extra_claims():
"""Extra claims are carried alongside the "admin" one."""
token = JWTService().get_admin_token({"sub": "user-id", "scope": "read"})
token = JWTService().get_admin_token(
audience=Audiences.YHUB, claims={"sub": "user-id", "scope": "read"}
)
payload = jwt.decode(token, PUBLIC_KEY, algorithms=["RS256"])
payload = jwt.decode(
token, PUBLIC_KEY, algorithms=["RS256"], audience=Audiences.YHUB
)
assert payload["admin"] is True
assert payload["sub"] == "user-id"
assert payload["scope"] == "read"
assert payload["aud"] == Audiences.YHUB
@pytest.mark.usefixtures("jwt_settings")
def test_get_admin_token_extra_claims_cannot_turn_admin_off():
"""🔒 A token issued by get_admin_token always grants admin."""
token = JWTService().get_admin_token({"admin": False})
token = JWTService().get_admin_token(
audience=Audiences.YHUB, claims={"admin": False}
)
payload = jwt.decode(token, PUBLIC_KEY, algorithms=["RS256"])
payload = jwt.decode(
token, PUBLIC_KEY, algorithms=["RS256"], audience=Audiences.YHUB
)
assert payload["admin"] is True
assert payload["aud"] == Audiences.YHUB
@pytest.mark.usefixtures("jwt_settings")
@@ -163,13 +177,24 @@ def test_get_admin_token_caches_each_set_of_extra_claims_separately():
"""Two callers passing different extra claims get their own token."""
service = JWTService()
first_token = service.get_admin_token({"sub": "user-id"})
second_token = service.get_admin_token({"sub": "other-user-id"})
first_token = service.get_admin_token(
audience=Audiences.YHUB, claims={"sub": "user-id"}
)
second_token = service.get_admin_token(
audience=Audiences.YHUB, claims={"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"]
jwt.decode(
first_token, PUBLIC_KEY, algorithms=["RS256"], audience=Audiences.YHUB
)["sub"]
== "user-id"
)
assert (
jwt.decode(
second_token, PUBLIC_KEY, algorithms=["RS256"], audience=Audiences.YHUB
)["sub"]
== "other-user-id"
)
@@ -179,7 +204,7 @@ def test_get_admin_token_does_not_mutate_the_given_claims():
"""The caller's dictionary is left untouched."""
claims = {"sub": "user-id"}
JWTService().get_admin_token(claims)
JWTService().get_admin_token(audience=Audiences.YHUB, claims=claims)
assert claims == {"sub": "user-id"}
@@ -189,10 +214,10 @@ def test_get_admin_token_reuses_the_cached_token():
"""The admin token is cached, like any other token."""
service = JWTService()
token = service.get_admin_token()
token = service.get_admin_token(audience=Audiences.YHUB)
with mock.patch("core.services.jwt_services.jwt.encode") as mock_encode:
assert service.get_admin_token() == token
assert service.get_admin_token(audience=Audiences.YHUB) == token
mock_encode.assert_not_called()
@@ -205,7 +230,7 @@ def test_get_admin_token_is_not_served_to_a_non_admin_caller():
"""
service = JWTService()
admin_token = service.get_admin_token()
admin_token = service.get_admin_token(audience=Audiences.YHUB)
tokens = [
service.get_token({"admin": False}),
service.get_token({"sub": "user-id"}),
@@ -225,8 +250,10 @@ def test_get_admin_token_expires_like_any_other_token(jwt_settings):
now = datetime(2026, 8, 4, 10, 0, 0, tzinfo=timezone.utc)
with freeze_time(now):
token = JWTService().get_admin_token()
payload = jwt.decode(token, PUBLIC_KEY, algorithms=["RS256"])
token = JWTService().get_admin_token(audience=Audiences.YHUB)
payload = jwt.decode(
token, PUBLIC_KEY, algorithms=["RS256"], audience=Audiences.YHUB
)
assert payload["exp"] == now.timestamp() + 120
@@ -9,6 +9,7 @@ import pytest
import requests
from core.factories import UserFactory
from core.services.jwt_services import Audiences
from core.services.yhub_services import (
APIError,
ConfigurationError,
@@ -62,8 +63,11 @@ def test_auth_header():
scheme, token = YHubService().auth_header.split(" ")
assert scheme == "Bearer"
payload = jwt.decode(token, PUBLIC_KEY, algorithms=["RS256"])
payload = jwt.decode(
token, PUBLIC_KEY, algorithms=["RS256"], audience=Audiences.YHUB
)
assert payload["admin"] is True
assert payload["aud"] == Audiences.YHUB
assert "sub" not in payload
@@ -73,19 +77,25 @@ def test_auth_header_with_user():
_scheme, token = YHubService(user=user).auth_header.split(" ")
payload = jwt.decode(token, PUBLIC_KEY, algorithms=["RS256"])
payload = jwt.decode(
token, PUBLIC_KEY, algorithms=["RS256"], audience=Audiences.YHUB
)
assert payload["sub"] == str(user.pk)
# naming a subject should not restrict what the call can do
assert payload["admin"] is True
assert payload["aud"] == Audiences.YHUB
def test_auth_header_with_anonymous_user():
"""An anonymous user is no subject, the token should not name one."""
_scheme, token = YHubService(user=AnonymousUser()).auth_header.split(" ")
payload = jwt.decode(token, PUBLIC_KEY, algorithms=["RS256"])
payload = jwt.decode(
token, PUBLIC_KEY, algorithms=["RS256"], audience=Audiences.YHUB
)
assert "sub" not in payload
assert payload["admin"] is True
assert payload["aud"] == Audiences.YHUB
@patch("requests.request")