diff --git a/backend/app/api/deps.py b/backend/app/api/deps.py index 5f28ec692a..d722d31d78 100644 --- a/backend/app/api/deps.py +++ b/backend/app/api/deps.py @@ -30,7 +30,12 @@ def get_db() -> Generator[Session]: def get_current_user(session: SessionDep, token: TokenDep) -> User: try: payload = jwt.decode( - token, settings.SECRET_KEY, algorithms=[security.ALGORITHM] + token, + settings.SECRET_KEY, + algorithms=[security.ALGORITHM], + # Require an explicit expiry so a token signed with the (now + # fail-closed) key can never be accepted without a lifetime bound. + options={"require": ["exp"]}, ) token_data = TokenPayload(**payload) except InvalidTokenError, ValidationError: diff --git a/backend/app/utils.py b/backend/app/utils.py index d59aefc7bf..36d229dbeb 100644 --- a/backend/app/utils.py +++ b/backend/app/utils.py @@ -118,7 +118,12 @@ def generate_password_reset_token(email: str) -> str: def verify_password_reset_token(token: str) -> str | None: try: decoded_token = jwt.decode( - token, settings.SECRET_KEY, algorithms=[security.ALGORITHM] + token, + settings.SECRET_KEY, + algorithms=[security.ALGORITHM], + # Require the expiry claim so an unexpired reset token can never be + # replayed indefinitely if one is minted without a lifetime bound. + options={"require": ["exp"]}, ) return str(decoded_token["sub"]) except InvalidTokenError: diff --git a/backend/tests/api/routes/test_jwt_claims.py b/backend/tests/api/routes/test_jwt_claims.py new file mode 100644 index 0000000000..412023e5bd --- /dev/null +++ b/backend/tests/api/routes/test_jwt_claims.py @@ -0,0 +1,83 @@ +"""Tests asserting JWTs are rejected when they lack a required `exp` claim. + +Companion hardening for the shipped-default-secret finding: even once the +signing key is fail-closed, a token that carries no expiry should never be +accepted. These tests mint tokens with the application's own key and assert that +access-token and password-reset verification both require `exp`. +""" + +from datetime import UTC, datetime, timedelta + +import jwt +from fastapi.testclient import TestClient + +from app.core import security +from app.core.config import settings +from app.utils import verify_password_reset_token +from tests.utils.utils import random_email + + +def _encode(claims: dict, *, key: str | None = None) -> str: + return jwt.encode( + claims, + key or settings.SECRET_KEY, + algorithm=security.ALGORITHM, + ) + + +def test_access_token_without_exp_is_rejected( + client: TestClient, superuser_token_headers: dict[str, str] +) -> None: + # Locate the superuser id from a valid session first. + me = client.get(f"{settings.API_V1_STR}/users/me", headers=superuser_token_headers) + assert me.status_code == 200 + superuser_id = me.json()["id"] + + # A correctly signed token that omits `exp` must not authenticate. + token_without_exp = _encode({"sub": superuser_id}) + r = client.get( + f"{settings.API_V1_STR}/users/me", + headers={"Authorization": f"Bearer {token_without_exp}"}, + ) + assert r.status_code in (401, 403) + + +def test_access_token_with_exp_is_accepted( + client: TestClient, superuser_token_headers: dict[str, str] +) -> None: + me = client.get(f"{settings.API_V1_STR}/users/me", headers=superuser_token_headers) + assert me.status_code == 200 + superuser_id = me.json()["id"] + + exp = datetime.now(UTC) + timedelta(minutes=30) + token_with_exp = _encode({"sub": superuser_id, "exp": exp}) + r = client.get( + f"{settings.API_V1_STR}/users/me", + headers={"Authorization": f"Bearer {token_with_exp}"}, + ) + assert r.status_code == 200 + + +def test_password_reset_token_requires_exp() -> None: + email = random_email() + assert verify_password_reset_token(_encode({"sub": email})) is None + + exp = datetime.now(UTC) + timedelta(hours=1) + assert verify_password_reset_token(_encode({"sub": email, "exp": exp})) == email + + +def test_token_signed_with_wrong_key_is_rejected( + client: TestClient, superuser_token_headers: dict[str, str] +) -> None: + me = client.get(f"{settings.API_V1_STR}/users/me", headers=superuser_token_headers) + assert me.status_code == 200 + superuser_id = me.json()["id"] + + # Even with an `exp`, a token signed with the wrong key must fail. + exp = datetime.now(UTC) + timedelta(minutes=30) + forged = _encode({"sub": superuser_id, "exp": exp}, key="not-the-real-key") + r = client.get( + f"{settings.API_V1_STR}/users/me", + headers={"Authorization": f"Bearer {forged}"}, + ) + assert r.status_code in (401, 403)