diff --git a/src/backend/core/services/converter_services.py b/src/backend/core/services/converter_services.py index ce527dcb7..a5edd4195 100644 --- a/src/backend/core/services/converter_services.py +++ b/src/backend/core/services/converter_services.py @@ -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): diff --git a/src/backend/core/services/jwt_services.py b/src/backend/core/services/jwt_services.py index 92fa15288..7ac666fba 100644 --- a/src/backend/core/services/jwt_services.py +++ b/src/backend/core/services/jwt_services.py @@ -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}) diff --git a/src/backend/core/services/yhub_services.py b/src/backend/core/services/yhub_services.py index 9cf6a66a4..84b103278 100644 --- a/src/backend/core/services/yhub_services.py +++ b/src/backend/core/services/yhub_services.py @@ -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.""" diff --git a/src/backend/core/tests/test_services_converter_services.py b/src/backend/core/tests/test_services_converter_services.py index 949088599..f19a120f6 100644 --- a/src/backend/core/tests/test_services_converter_services.py +++ b/src/backend/core/tests/test_services_converter_services.py @@ -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(): diff --git a/src/backend/core/tests/test_services_jwt_services.py b/src/backend/core/tests/test_services_jwt_services.py index eaf0f2989..c8ffdfb67 100644 --- a/src/backend/core/tests/test_services_jwt_services.py +++ b/src/backend/core/tests/test_services_jwt_services.py @@ -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 diff --git a/src/backend/core/tests/test_services_yhub_services.py b/src/backend/core/tests/test_services_yhub_services.py index 1b92569ce..f57da0424 100644 --- a/src/backend/core/tests/test_services_yhub_services.py +++ b/src/backend/core/tests/test_services_yhub_services.py @@ -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")