add revoke

id.saladeaula.digital/app/routes/session.py

@@ -15,10 +15,16 @@ from layercake.dynamodb import DynamoDBPersistenceLayer, KeyPair, SortKey
 from passlib.hash import pbkdf2_sha256
 
 from boto3clients import dynamodb_client
-from config import ISSUER, JWT_ALGORITHM, JWT_EXP_SECONDS, JWT_SECRET, OAUTH2_TABLE
+from config import (
+    ISSUER,
+    JWT_ALGORITHM,
+    JWT_SECRET,
+    OAUTH2_REFRESH_TOKEN_EXPIRES_IN,
+    OAUTH2_TABLE,
+)
 
 router = Router()
-oauth2_layer = DynamoDBPersistenceLayer(OAUTH2_TABLE, dynamodb_client)
+dyn = DynamoDBPersistenceLayer(OAUTH2_TABLE, dynamodb_client)
 
 
 @router.post('/session')
@@ -26,11 +32,7 @@ def session(
     username: Annotated[str, Body()],
     password: Annotated[str, Body()],
 ):
-    (
-        user_id,
-        password_hash,
-        scope,
-    ) = _get_user(username)
+    user_id, password_hash = _get_user(username)
 
     if not pbkdf2_sha256.verify(password, password_hash):
         raise ForbiddenError('Invalid credentials')
@@ -40,28 +42,27 @@ def session(
         cookies=[
             Cookie(
                 name='session_id',
-                value=new_session(user_id, scope),
+                value=new_session(user_id),
                 http_only=True,
                 secure=True,
                 same_site=None,
-                max_age=JWT_EXP_SECONDS,
+                max_age=OAUTH2_REFRESH_TOKEN_EXPIRES_IN,
             )
         ],
     )
 
 
-def _get_user(username: str) -> tuple[str, str, str | None]:
+def _get_user(username: str) -> tuple[str, str]:
     sk = SortKey(username, path_spec='user_id')
-    user = oauth2_layer.collection.get_items(
+    user = dyn.collection.get_items(
         KeyPair(pk='email', sk=sk, rename_key=sk.path_spec)
         + KeyPair(pk='cpf', sk=sk, rename_key=sk.path_spec),
-        flatten_top=False,
     )
 
     if not user:
         raise UserNotFoundError()
 
-    userdata = oauth2_layer.collection.get_items(
+    password = dyn.collection.get_item(
         KeyPair(
             pk=user['user_id'],
             sk=SortKey(
@@ -69,46 +70,34 @@ def _get_user(username: str) -> tuple[str, str, str | None]:
                 path_spec='hash',
                 rename_key='password',
             ),
-        )
-        + KeyPair(
-            pk=user['user_id'],
-            sk=SortKey(
-                sk='SCOPE',
-                path_spec='scope',
-                rename_key='scope',
-            ),
         ),
-        flatten_top=False,
+        exc_cls=UserNotFoundError,
     )
 
-    if not userdata:
-        raise UserNotFoundError()
-
-    return user['user_id'], userdata['password'], userdata.get('scope')
+    return user['user_id'], password
 
 
-def new_session(sub: str, scope: str | None) -> str:
+def new_session(sub: str) -> str:
+    session_id = str(uuid4())
     now_ = now()
-    sid = str(uuid4())
-    exp = ttl(start_dt=now_, seconds=JWT_EXP_SECONDS)
+    exp = ttl(start_dt=now_, seconds=OAUTH2_REFRESH_TOKEN_EXPIRES_IN)
     token = jwt.encode(
         {
-            'sid': sid,
+            'sid': session_id,
             'sub': sub,
             'iss': ISSUER,
             'iat': int(now_.timestamp()),
             'exp': exp,
-            'scope': scope,
         },
         JWT_SECRET,
         algorithm=JWT_ALGORITHM,
     )
 
-    with oauth2_layer.transact_writer() as transact:
+    with dyn.transact_writer() as transact:
         transact.put(
             item={
                 'id': 'SESSION',
-                'sk': sid,
+                'sk': session_id,
                 'user_id': sub,
                 'ttl': exp,
                 'created_at': now_,
@@ -117,7 +106,7 @@ def new_session(sub: str, scope: str | None) -> str:
         transact.put(
             item={
                 'id': sub,
-                'sk': f'SESSION#{sid}',
+                'sk': f'SESSION#{session_id}',
                 'ttl': exp,
                 'created_at': now_,
             }