109 lines
2.9 KiB
Python
109 lines
2.9 KiB
Python
from http import HTTPStatus
|
|
from typing import Annotated
|
|
from uuid import uuid4
|
|
|
|
import jwt
|
|
from aws_lambda_powertools.event_handler import (
|
|
Response,
|
|
)
|
|
from aws_lambda_powertools.event_handler.api_gateway import Router
|
|
from aws_lambda_powertools.event_handler.exceptions import ForbiddenError, NotFoundError
|
|
from aws_lambda_powertools.event_handler.openapi.params import Body
|
|
from aws_lambda_powertools.shared.cookies import Cookie
|
|
from layercake.dateutils import now, ttl
|
|
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
|
|
|
|
router = Router()
|
|
oauth2_layer = DynamoDBPersistenceLayer(OAUTH2_TABLE, dynamodb_client)
|
|
|
|
|
|
@router.post('/session')
|
|
def session(
|
|
username: Annotated[str, Body()],
|
|
password: Annotated[str, Body()],
|
|
):
|
|
user_id, password_hash = _get_user(username)
|
|
|
|
if not pbkdf2_sha256.verify(password, password_hash):
|
|
raise ForbiddenError('Invalid credentials')
|
|
|
|
return Response(
|
|
status_code=HTTPStatus.OK,
|
|
cookies=[
|
|
Cookie(
|
|
name='session_id',
|
|
value=new_session(user_id),
|
|
http_only=True,
|
|
secure=True,
|
|
same_site=None,
|
|
max_age=JWT_EXP_SECONDS,
|
|
)
|
|
],
|
|
)
|
|
|
|
|
|
def _get_user(username: str) -> tuple[str, str]:
|
|
sk = SortKey(username, path_spec='user_id')
|
|
user = oauth2_layer.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()
|
|
|
|
password = oauth2_layer.collection.get_item(
|
|
KeyPair(user['user_id'], 'PASSWORD'),
|
|
exc_cls=UserNotFoundError,
|
|
)
|
|
|
|
return user['user_id'], password['hash']
|
|
|
|
|
|
def new_session(sub: str) -> str:
|
|
now_ = now()
|
|
sid = str(uuid4())
|
|
exp = ttl(start_dt=now_, seconds=JWT_EXP_SECONDS)
|
|
token = jwt.encode(
|
|
{
|
|
'sid': sid,
|
|
'sub': sub,
|
|
'iss': ISSUER,
|
|
'iat': int(now_.timestamp()),
|
|
'exp': exp,
|
|
},
|
|
JWT_SECRET,
|
|
algorithm=JWT_ALGORITHM,
|
|
)
|
|
|
|
with oauth2_layer.transact_writer() as transact:
|
|
transact.put(
|
|
item={
|
|
'id': 'SESSION',
|
|
'sk': sid,
|
|
'user_id': sub,
|
|
'ttl': exp,
|
|
'created_at': now_,
|
|
}
|
|
)
|
|
transact.put(
|
|
item={
|
|
'id': sub,
|
|
'sk': f'SESSION#{sid}',
|
|
'ttl': exp,
|
|
'created_at': now_,
|
|
}
|
|
)
|
|
|
|
return token
|
|
|
|
|
|
class UserNotFoundError(NotFoundError):
|
|
def __init__(self, *_):
|
|
super().__init__('User not found')
|