117 lines
3.1 KiB
Python
117 lines
3.1 KiB
Python
from dataclasses import dataclass
|
|
from http import HTTPStatus
|
|
from typing import Annotated
|
|
from uuid import uuid4
|
|
|
|
from aws_lambda_powertools.event_handler.api_gateway import Response, Router
|
|
from aws_lambda_powertools.event_handler.exceptions import NotFoundError
|
|
from aws_lambda_powertools.event_handler.openapi.params import Body
|
|
from aws_lambda_powertools.utilities.data_masking import DataMasking
|
|
from layercake.dateutils import now, ttl
|
|
from layercake.dynamodb import DynamoDBPersistenceLayer, KeyPair, SortKey
|
|
from layercake.extra_types import CpfStr
|
|
from layercake.funcs import pick
|
|
from pydantic import EmailStr
|
|
|
|
from boto3clients import dynamodb_client
|
|
from config import USER_TABLE
|
|
|
|
router = Router()
|
|
dyn = DynamoDBPersistenceLayer(USER_TABLE, dynamodb_client)
|
|
data_masker = DataMasking()
|
|
|
|
masking_rules = {
|
|
'email': {'regex_pattern': '(.)(.*)(..)(@.*)', 'mask_format': r'\1****\3\4'},
|
|
}
|
|
|
|
|
|
class UserNotFoundError(NotFoundError): ...
|
|
|
|
|
|
@router.post('/forgot')
|
|
def forgot(username: Annotated[EmailStr | CpfStr, Body(embed=True)]):
|
|
now_ = now()
|
|
user = _get_user(username)
|
|
reset_ttl = ttl(start_dt=now_, hours=3)
|
|
code = uuid4()
|
|
|
|
with dyn.transact_writer() as transact:
|
|
transact.update(
|
|
key=KeyPair(
|
|
pk='PASSWORD_RESET',
|
|
sk=f'USER#{user.id}',
|
|
),
|
|
update_expr='SET #name = :name, \
|
|
email = :email, \
|
|
#ttl = :ttl, \
|
|
updated_at = :now \
|
|
ADD code_attempts :code',
|
|
expr_attr_names={
|
|
'#name': 'name',
|
|
'#ttl': 'ttl',
|
|
},
|
|
expr_attr_values={
|
|
':name': user.name,
|
|
':email': user.email,
|
|
':ttl': reset_ttl,
|
|
':code': {code},
|
|
':now': now_,
|
|
},
|
|
)
|
|
dyn.put_item(
|
|
item={
|
|
'id': 'PASSWORD_RESET',
|
|
'sk': f'CODE#{code}',
|
|
'name': user.name,
|
|
'user_id': user.id,
|
|
'ttl': reset_ttl,
|
|
'created_at': now_,
|
|
}
|
|
)
|
|
|
|
return Response(
|
|
status_code=HTTPStatus.CREATED,
|
|
body=data_masker.erase(
|
|
{
|
|
'email': user.email,
|
|
},
|
|
masking_rules=masking_rules,
|
|
),
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class User:
|
|
id: str
|
|
name: str
|
|
email: str
|
|
|
|
|
|
def _get_user(username: str) -> User:
|
|
user_id = dyn.collection.get_items(
|
|
KeyPair(
|
|
pk='email',
|
|
sk=SortKey(username, path_spec='user_id'), # type: ignore
|
|
rename_key='user_id',
|
|
)
|
|
+ KeyPair(
|
|
pk='cpf',
|
|
sk=SortKey(username, path_spec='user_id'), # type: ignore
|
|
rename_key='user_id',
|
|
),
|
|
flatten_top=False,
|
|
).get('user_id')
|
|
|
|
if not user_id:
|
|
raise UserNotFoundError('User not found')
|
|
|
|
user = dyn.get_item(
|
|
KeyPair(
|
|
pk=user_id,
|
|
sk='0',
|
|
)
|
|
)
|
|
return User(
|
|
**pick(('id', 'name', 'email'), user),
|
|
)
|