diff --git a/src/backend/core/services/jwt_services.py b/src/backend/core/services/jwt_services.py index 5a4ac4eb7..92fa15288 100644 --- a/src/backend/core/services/jwt_services.py +++ b/src/backend/core/services/jwt_services.py @@ -171,3 +171,13 @@ class JWTService: cache.set(cache_key, token, self.lifetime) return token + + def get_admin_token(self, 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}) diff --git a/src/backend/core/tests/test_services_jwt_services.py b/src/backend/core/tests/test_services_jwt_services.py index 6e77a2dd5..be8b58418 100644 --- a/src/backend/core/tests/test_services_jwt_services.py +++ b/src/backend/core/tests/test_services_jwt_services.py @@ -150,6 +150,108 @@ 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() + + payload = jwt.decode(token, PUBLIC_KEY, algorithms=["RS256"]) + assert payload["admin"] is True + + +@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"}) + + payload = jwt.decode(token, PUBLIC_KEY, algorithms=["RS256"]) + assert payload["admin"] is True + assert payload["sub"] == "user-id" + assert payload["scope"] == "read" + + +@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}) + + payload = jwt.decode(token, PUBLIC_KEY, algorithms=["RS256"]) + assert payload["admin"] is True + + +@pytest.mark.usefixtures("jwt_settings") +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"}) + + 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_admin_token_does_not_mutate_the_given_claims(): + """The caller's dictionary is left untouched.""" + claims = {"sub": "user-id"} + + JWTService().get_admin_token(claims) + + assert claims == {"sub": "user-id"} + + +@pytest.mark.usefixtures("jwt_settings") +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() + + with mock.patch("core.services.jwt_services.jwt.encode") as mock_encode: + assert service.get_admin_token() == token + + mock_encode.assert_not_called() + + +@pytest.mark.usefixtures("jwt_settings") +def test_get_admin_token_is_not_served_to_a_non_admin_caller(): + """ + 🔒 The admin token has its own cache entry. Asking for any other set of + claims must never hand out a token granting admin. + """ + service = JWTService() + + admin_token = service.get_admin_token() + tokens = [ + service.get_token({"admin": False}), + service.get_token({"sub": "user-id"}), + service.get_token({}), + ] + + assert admin_token not in tokens + for token in tokens: + assert ( + jwt.decode(token, PUBLIC_KEY, algorithms=["RS256"]).get("admin") is not True + ) + + +def test_get_admin_token_expires_like_any_other_token(jwt_settings): + """The admin token does not outlive the configured lifetime.""" + jwt_settings.JWT_TOKEN_LIFETIME = 120 + + 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"]) + + assert payload["exp"] == now.timestamp() + 120 + + @pytest.mark.usefixtures("jwt_settings") def test_get_jwks_exposes_only_the_public_key(): """🔒 The JWKS must never carry the private components of the key."""