feat: update

src/infrastructure/security/jwt.py

@@ -20,7 +20,7 @@ class JwtService(IJwtService):
     def _issuer(self) -> str | None:
         return settings.ADMIN_JWT_ISSUER
 
-    async def create_access_token(self, user_id: str, *, role: str, sid: str | None = None) -> str:
+    async def create_access_token(self, user_id: str, *, role: str) -> str:
         now = datetime.now(timezone.utc)
         exp = now + timedelta(seconds=int(settings.JWT_ACCESS_TTL_SECONDS))
 
@@ -32,8 +32,6 @@ class JwtService(IJwtService):
             'nbf': int(now.timestamp()),
             'exp': int(exp.timestamp()),
         }
-        if sid:
-            payload['sid'] = sid
         if self._issuer:
             payload['iss'] = self._issuer
         if settings.JWT_AUDIENCE:
@@ -41,15 +39,14 @@ class JwtService(IJwtService):
 
         return await self._encode(payload, user_id=user_id, token_kind='access')
 
-    async def create_refresh_token(self, user_id: str, *, sid: str, refresh_jti: str) -> str:
+    async def create_refresh_token(self, user_id: str, *, role: str) -> str:
         now = datetime.now(timezone.utc)
         exp = now + timedelta(seconds=int(settings.JWT_REFRESH_TTL_SECONDS))
 
         payload: dict[str, object] = {
             'sub': user_id,
             'type': 'refresh',
-            'sid': sid,
-            'jti': refresh_jti,
+            'role': role,
             'iat': int(now.timestamp()),
             'nbf': int(now.timestamp()),
             'exp': int(exp.timestamp()),
@@ -82,7 +79,6 @@ class JwtService(IJwtService):
                 sub=str(payload['sub']),
                 type='access',
                 role=str(payload['role']) if payload.get('role') else None,
-                sid=str(payload['sid']) if payload.get('sid') else None,
                 iat=int(payload['iat']),
                 nbf=int(payload['nbf']),
                 exp=int(payload['exp']),
@@ -100,8 +96,7 @@ class JwtService(IJwtService):
             return RefreshTokenPayload(
                 sub=str(payload['sub']),
                 type='refresh',
-                sid=str(payload['sid']),
-                jti=str(payload['jti']),
+                role=str(payload['role']),
                 iat=int(payload['iat']),
                 nbf=int(payload['nbf']),
                 exp=int(payload['exp']),
@@ -150,12 +145,10 @@ class JwtService(IJwtService):
             if 'type' not in payload:
                 raise ApplicationException(status_code=401, message='Missing token claim: type')
             token_type = payload.get('type')
-            if token_type == 'access' and 'role' not in payload:
+            if 'role' not in payload:
                 raise ApplicationException(status_code=401, message='Missing token claim: role')
-            if token_type == 'refresh':
-                for claim in ('sid', 'jti'):
-                    if claim not in payload:
-                        raise ApplicationException(status_code=401, message=f'Missing token claim: {claim}')
+            if token_type not in ('access', 'refresh'):
+                raise ApplicationException(status_code=401, message='Invalid token type')
             return payload
         except ExpiredSignatureError:
             raise ApplicationException(status_code=401, message='Token expired')

src/infrastructure/utils/ulid.py

@@ -0,0 +1,15 @@
+from ulid import ULID
+
+
+def new_ulid() -> str:
+    return str(ULID())
+
+
+def is_valid_ulid(value: str | None) -> bool:
+    if not value:
+        return False
+    try:
+        ULID.parse(value)
+        return True
+    except ValueError:
+        return False