add ft to migrate from cognito

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

@@ -14,7 +14,7 @@ from aws_lambda_powertools.event_handler.exceptions import (
 from layercake.dynamodb import DynamoDBPersistenceLayer, KeyPair, SortKey
 
 from boto3clients import dynamodb_client
-from config import ISSUER, JWT_ALGORITHM, JWT_SECRET, OAUTH2_TABLE
+from config import ISSUER, OAUTH2_TABLE, SESSION_SECRET
 from oauth2 import server
 
 router = Router()
@@ -63,12 +63,11 @@ def authorize():
 def verify_session(session_id: str) -> tuple[str, str | None]:
     payload = jwt.decode(
         session_id,
-        JWT_SECRET,
-        algorithms=[JWT_ALGORITHM],
+        SESSION_SECRET,
+        algorithms=['HS256'],
         issuer=ISSUER,
         options={
             'require': ['exp', 'sub', 'iss', 'sid'],
-            'leeway': 60,
         },
     )
 

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

@@ -2,6 +2,7 @@ from http import HTTPStatus
 from typing import Annotated
 from uuid import uuid4
 
+import boto3
 import jwt
 from aws_lambda_powertools.event_handler import (
     Response,
@@ -17,14 +18,14 @@ from passlib.hash import pbkdf2_sha256
 from boto3clients import dynamodb_client
 from config import (
     ISSUER,
-    JWT_ALGORITHM,
-    JWT_SECRET,
     OAUTH2_TABLE,
     SESSION_EXPIRES_IN,
+    SESSION_SECRET,
 )
 
 router = Router()
 dyn = DynamoDBPersistenceLayer(OAUTH2_TABLE, dynamodb_client)
+idp = boto3.client('cognito-idp')
 
 
 @router.post('/session')
@@ -34,8 +35,11 @@ def session(
 ):
     user_id, password_hash = _get_user(username)
 
-    if not pbkdf2_sha256.verify(password, password_hash):
-        raise ForbiddenError('Invalid credentials')
+    if not password_hash:
+        _get_idp_user(user_id, username, password)
+    else:
+        if not pbkdf2_sha256.verify(password, password_hash):
+            raise ForbiddenError('Invalid credentials')
 
     return Response(
         status_code=HTTPStatus.OK,
@@ -52,7 +56,7 @@ def session(
     )
 
 
-def _get_user(username: str) -> tuple[str, str]:
+def _get_user(username: str) -> tuple[str, str | None]:
     sk = SortKey(username, path_spec='user_id')
     user = dyn.collection.get_items(
         KeyPair(pk='email', sk=sk, rename_key=sk.path_spec)
@@ -71,12 +75,57 @@ def _get_user(username: str) -> tuple[str, str]:
                 rename_key='password',
             ),
         ),
-        exc_cls=UserNotFoundError,
+        raise_on_error=False,
+        default=None,
+        # Uncomment the following line when removing support for Cognito
+        # exc_cls=UserNotFoundError,
     )
 
     return user['user_id'], password
 
 
+def _get_idp_user(
+    user_id: str,
+    username: str,
+    password: str,
+) -> bool:
+    import base64
+    import hashlib
+    import hmac
+
+    client_id = '3ijacqc7r2jc9l4oli2b41f7te'
+    client_secret = 'amktf9l40g1mlqdo9fjlcfvpn2cp3mvh4pt97hu55sfelccos58'
+
+    dig = hmac.new(
+        client_secret.encode('utf-8'),
+        msg=(username + client_id).encode('utf-8'),
+        digestmod=hashlib.sha256,
+    ).digest()
+
+    try:
+        idp.initiate_auth(
+            AuthFlow='USER_PASSWORD_AUTH',
+            AuthParameters={
+                'USERNAME': username,
+                'PASSWORD': password,
+                'SECRET_HASH': base64.b64encode(dig).decode(),
+            },
+            ClientId=client_id,
+        )
+
+        dyn.put_item(
+            item={
+                'id': user_id,
+                'sk': 'PASSWORD',
+                'hash': pbkdf2_sha256.hash(password),
+            }
+        )
+    except Exception:
+        raise ForbiddenError('Invalid credentials')
+
+    return True
+
+
 def new_session(sub: str) -> str:
     session_id = str(uuid4())
     now_ = now()
@@ -89,8 +138,8 @@ def new_session(sub: str) -> str:
             'iat': int(now_.timestamp()),
             'exp': exp,
         },
-        JWT_SECRET,
-        algorithm=JWT_ALGORITHM,
+        SESSION_SECRET,
+        algorithm='HS256',
     )
 
     with dyn.transact_writer() as transact: