fix: add jwt standard claim

Author: fiqrikm18

.env.example

@@ -10,7 +10,9 @@ SECRET_KEY=your-super-secret-production-key-here
 MAX_REQUEST_SIZE_MB=5242880 #5mb
 
 ALGORITHM=HS256
+JWT_ISSUER=todo-modulith-api
+JWT_AUDIENCE=todo-modulith-client
 ACCESS_TOKEN_EXPIRE_MINUTES=30
 REFRESH_TOKEN_EXPIRE_MINUTES=10080
 
-RATE_LIMIT="100/minute" 
+RATE_LIMIT="100/minute" 

README.md

@@ -233,7 +233,10 @@ APP_NAME=Todo Modulith API
 DATABASE_URL=
 SECRET_KEY=
 ALGORITHM=HS256
+JWT_ISSUER=todo-modulith-api
+JWT_AUDIENCE=todo-modulith-client
 ACCESS_TOKEN_EXPIRE_MINUTES=30
+REFRESH_TOKEN_EXPIRE_MINUTES=10080
 ```
 
 For local development without Docker, point `DATABASE_URL` at your local PostgreSQL host, for example:

src/core/config/setting.py

@@ -19,6 +19,8 @@ class Settings(BaseSettings):
         alias="SECRET_KEY", default="super-secret-key-change-in-production"
     )
     ALGORITHM: str = Field(alias="ALGORITHM", default="HS256")
+    JWT_ISSUER: str = Field(alias="JWT_ISSUER", default="todo-modulith-api")
+    JWT_AUDIENCE: str = Field(alias="JWT_AUDIENCE", default="todo-modulith-client")
     ACCESS_TOKEN_EXPIRE_MINUTES: int = Field(
         alias="ACCESS_TOKEN_EXPIRE_MINUTES", default=30
     )

src/core/middleware/auth.py

@@ -6,6 +6,7 @@
 
 from src.core.security.jwt import JWTService
 from src.core.security.token_revocation import TokenRevocationService
+from src.shared.exceptions.credential_exception import InvalidCredentialsError
 
 PUBLIC_PATHS = frozenset(
     {
@@ -44,6 +45,7 @@ async def dispatch(
 
         try:
             payload = JWTService.decode_token(token)
+            JWTService.require_token_type(payload, JWTService.ACCESS_TOKEN_TYPE)
             if await TokenRevocationService.is_access_token_revoked(token):
                 return JSONResponse(
                     status_code=status.HTTP_401_UNAUTHORIZED,
@@ -61,6 +63,11 @@ async def dispatch(
                 status_code=status.HTTP_401_UNAUTHORIZED,
                 content={"detail": f"Invalid or expired token: {str(e)}"},
             )
+        except InvalidCredentialsError as e:
+            return JSONResponse(
+                status_code=status.HTTP_401_UNAUTHORIZED,
+                content={"detail": str(e)},
+            )
         except Exception:
             return JSONResponse(
                 status_code=status.HTTP_401_UNAUTHORIZED,

src/core/security/jwt.py

@@ -10,29 +10,57 @@
 
 
 class JWTService:
+    ACCESS_TOKEN_TYPE = "access"
+    REFRESH_TOKEN_TYPE = "refresh"
+
     @staticmethod
-    def create_access_token(data: dict) -> str:
+    def _create_token(data: dict, token_type: str, expires_delta: timedelta) -> str:
+        now = datetime.now(timezone.utc)
         to_encode = data.copy()
-        expire = datetime.now(timezone.utc) + timedelta(
-            minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES
+        to_encode.update(
+            {
+                "iss": settings.JWT_ISSUER,
+                "sub": str(to_encode["sub"]),
+                "aud": settings.JWT_AUDIENCE,
+                "exp": now + expires_delta,
+                "nbf": now,
+                "iat": now,
+                "jti": str(uuid4()),
+                "token_type": token_type,
+            }
         )
-        to_encode.update({"exp": expire, "jti": str(uuid4())})
         return jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM)
 
+    @staticmethod
+    def create_access_token(data: dict) -> str:
+        return JWTService._create_token(
+            data=data,
+            token_type=JWTService.ACCESS_TOKEN_TYPE,
+            expires_delta=timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES),
+        )
+
     @staticmethod
     def create_refresh_token(data: dict) -> str:
-        to_encode = data.copy()
-        expire = datetime.now(timezone.utc) + timedelta(
-            minutes=settings.REFRESH_TOKEN_EXPIRE_MINUTES,
+        return JWTService._create_token(
+            data=data,
+            token_type=JWTService.REFRESH_TOKEN_TYPE,
+            expires_delta=timedelta(minutes=settings.REFRESH_TOKEN_EXPIRE_MINUTES),
         )
-        to_encode.update({"exp": expire})
-        return jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM)
 
     @staticmethod
     def decode_token(token: str) -> dict:
         try:
             return jwt.decode(
-                token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM]
+                token,
+                settings.SECRET_KEY,
+                algorithms=[settings.ALGORITHM],
+                issuer=settings.JWT_ISSUER,
+                audience=settings.JWT_AUDIENCE,
             )
         except JWTError:
             raise InvalidCredentialsError("Invalid or expired token")
+
+    @staticmethod
+    def require_token_type(payload: dict, expected_token_type: str) -> None:
+        if payload.get("token_type") != expected_token_type:
+            raise InvalidCredentialsError(f"{expected_token_type.title()} token required")

src/modules/user/application/refresh_token/handler.py

@@ -11,7 +11,10 @@
 from src.modules.user.domain.repositories.refresh_token_repository import (
     RefreshTokenRepository,
 )
-from src.shared.exceptions.credential_exception import InvalidRefreshTokenError
+from src.shared.exceptions.credential_exception import (
+    InvalidCredentialsError,
+    InvalidRefreshTokenError,
+)
 from src.shared.unit_of_work import UnitOfWork
 
 settings = get_settings()
@@ -42,6 +45,15 @@ async def execute(self, command: RefreshTokenCommand) -> dict:
         if stored_token.expires_at < datetime.now(timezone.utc):
             raise InvalidRefreshTokenError("Refresh token has expired")
 
+        try:
+            payload = JWTService.decode_token(command.token)
+            JWTService.require_token_type(payload, JWTService.REFRESH_TOKEN_TYPE)
+        except InvalidCredentialsError:
+            raise InvalidRefreshTokenError("Invalid refresh token")
+
+        if payload.get("sub") != str(stored_token.user_id):
+            raise InvalidRefreshTokenError("Invalid refresh token")
+
         async with self._unit_of_work:
             stored_token.revoke()
             await self._refresh_token_repo.save(stored_token)

src/modules/user/infrastructure/service/auth_service.py

@@ -1,12 +1,7 @@
-from datetime import datetime, timedelta, timezone
-
-from jose import JWTError, jwt
 from passlib.context import CryptContext
 
-from src.core.config.setting import get_settings
-from src.modules.user.domain.exceptions import InvalidCredentialsError
+from src.core.security.jwt import JWTService
 
-settings = get_settings()
 pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
 
 
@@ -15,17 +10,7 @@ def verifyy_password(self, password: str, hashed_password: str) -> bool:
         return pwd_context.verify(password, hashed_password)
 
     def create_access_token(self, data: dict) -> str:
-        to_encode = data.copy()
-        expire = datetime.now(timezone.utc) + timedelta(
-            minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES
-        )
-        to_encode.update({"exp": expire})
-        return jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM)
+        return JWTService.create_access_token(data)
 
     def decode_token(self, token: str) -> dict:
-        try:
-            return jwt.decode(
-                token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM]
-            )
-        except JWTError:
-            raise InvalidCredentialsError("Invalid or expired token")
+        return JWTService.decode_token(token)

tests/core/test_auth_middleware.py

@@ -40,3 +40,44 @@ async def call_next(_request):
         assert response.status_code == 401
 
     asyncio.run(run())
+
+
+def test_authentication_middleware_rejects_refresh_token_on_protected_endpoint(
+    monkeypatch,
+):
+    async def run():
+        async def active_token(_token):
+            return False
+
+        monkeypatch.setattr(
+            TokenRevocationService,
+            "is_access_token_revoked",
+            active_token,
+        )
+
+        token = JWTService.create_refresh_token({"sub": "user-id"})
+        call_next_called = False
+        request = Request(
+            {
+                "type": "http",
+                "method": "GET",
+                "path": "/protected",
+                "headers": [(b"authorization", f"Bearer {token}".encode())],
+                "query_string": b"",
+                "server": ("testserver", 80),
+                "scheme": "http",
+                "client": ("testclient", 50000),
+            }
+        )
+
+        async def call_next(_request):
+            nonlocal call_next_called
+            call_next_called = True
+            return JSONResponse({"ok": True})
+
+        response = await AuthenticationMiddleware(None).dispatch(request, call_next)
+
+        assert response.status_code == 401
+        assert call_next_called is False
+
+    asyncio.run(run())

tests/core/test_jwt_claims.py

@@ -9,10 +9,14 @@ def test_access_token_uses_standard_claim_payload():
     claims = JWTService.decode_token(token)
 
     assert claims["sub"] == "user-id"
+    assert claims["iss"]
+    assert claims["aud"]
     assert claims["token_type"] == "access"
     assert claims["jti"]
     assert claims["iat"]
+    assert claims["nbf"]
     assert claims["exp"]
+    assert claims["nbf"] <= claims["iat"]
     assert claims["iat"] <= claims["exp"]
 
 
@@ -22,10 +26,14 @@ def test_refresh_token_uses_standard_claim_payload():
     claims = JWTService.decode_token(token)
 
     assert claims["sub"] == "user-id"
+    assert claims["iss"]
+    assert claims["aud"]
     assert claims["token_type"] == "refresh"
     assert claims["jti"]
     assert claims["iat"]
+    assert claims["nbf"]
     assert claims["exp"]
+    assert claims["nbf"] <= claims["iat"]
     assert claims["iat"] <= claims["exp"]
 
 

tests/user/test_refresh_token_flow.py

@@ -6,6 +6,7 @@
 
 import pytest
 
+from src.core.security.jwt import JWTService
 from src.modules.user.application.refresh_token.command import RefreshTokenCommand
 from src.modules.user.application.refresh_token import handler as refresh_handler_module
 from src.modules.user.application.refresh_token.handler import RefreshTokenCommandHandler
@@ -90,11 +91,32 @@ async def run():
     asyncio.run(run())
 
 
+def test_refresh_token_rejects_access_token_even_when_hash_exists():
+    async def run():
+        user_id = uuid4()
+        access_token = JWTService.create_access_token({"sub": str(user_id)})
+        token_hash = hashlib.sha256(access_token.encode()).hexdigest()
+        stored_token = RefreshToken.create(
+            user_id=user_id,
+            token_hash=token_hash,
+            expires_at=datetime.now(timezone.utc) + timedelta(days=1),
+        )
+        handler = RefreshTokenCommandHandler(
+            FakeRefreshTokenRepository(stored_token),
+            FakeUnitOfWork(),
+        )
+
+        with pytest.raises(InvalidRefreshTokenError, match="Invalid refresh token"):
+            await handler.execute(RefreshTokenCommand(token=access_token))
+
+    asyncio.run(run())
+
+
 def test_refresh_token_rotates_token_and_revokes_existing_token():
     async def run():
-        raw_token = "raw-refresh-token"
-        token_hash = hashlib.sha256(raw_token.encode()).hexdigest()
         user_id = uuid4()
+        raw_token = JWTService.create_refresh_token({"sub": str(user_id)})
+        token_hash = hashlib.sha256(raw_token.encode()).hexdigest()
         stored_token = RefreshToken.create(
             user_id=user_id,
             token_hash=token_hash,
@@ -126,9 +148,9 @@ async def run():
             "REFRESH_TOKEN_EXPIRE_MINUTES",
             15,
         )
-        raw_token = "raw-refresh-token"
-        token_hash = hashlib.sha256(raw_token.encode()).hexdigest()
         user_id = uuid4()
+        raw_token = JWTService.create_refresh_token({"sub": str(user_id)})
+        token_hash = hashlib.sha256(raw_token.encode()).hexdigest()
         stored_token = RefreshToken.create(
             user_id=user_id,
             token_hash=token_hash,
@@ -152,10 +174,11 @@ async def run():
 
 def test_refresh_token_rotation_rolls_back_when_new_token_save_fails():
     async def run():
-        raw_token = "raw-refresh-token"
+        user_id = uuid4()
+        raw_token = JWTService.create_refresh_token({"sub": str(user_id)})
         token_hash = hashlib.sha256(raw_token.encode()).hexdigest()
         stored_token = RefreshToken.create(
-            user_id=uuid4(),
+            user_id=user_id,
             token_hash=token_hash,
             expires_at=datetime.now(timezone.utc) + timedelta(days=1),
         )