Implement forgot/reset password endpoints

docker-compose.yml

@@ -20,7 +20,6 @@ services:
 
   api-pgd:
     image: ghcr.io/gestaogovbr/api-pgd:latest
-    pull_policy: always
     container_name: api-pgd
     depends_on:
       db:
@@ -40,8 +39,24 @@ services:
       # to new `SECRET` run openssl rand -hex 32
       SECRET: b8a3054ba3457614e95a88cc0807384430c1b338a54e95e4245f41e060da68bc
       ACCESS_TOKEN_EXPIRE_MINUTES: 30
+      MAIL_USERNAME: ''
+      MAIL_FROM: [email protected]
+      MAIL_PORT: 25
+      MAIL_SERVER: smtp4dev
+      MAIL_FROM_NAME: [email protected]
+      MAIL_PASSWORD: ''  
     healthcheck:
       test: ["CMD", "curl", "-f", "http://0.0.0.0:5057/docs"]
       interval: 5s
       timeout: 5s
       retries: 20
+
+  smtp4dev:
+    image: rnwood/smtp4dev:v3
+    restart: always
+    ports:
+      - '5000:80'
+      - '25:25' # Change the number before : to the port the SMTP server should be accessible on
+      - '143:143' # Change the number before : to the port the IMAP server should be accessible on
+    environment:
+      - ServerOptions__HostName=smtp4dev

requirements.txt

@@ -8,3 +8,4 @@ httpx==0.24.1
 python-jose[cryptography]==3.3.0
 passlib[bcrypt]==1.7.4
 python-multipart==0.0.6
+fastapi-mail==1.4.1

src/api.py

@@ -12,11 +12,13 @@
 from fastapi.responses import RedirectResponse
 from sqlalchemy.exc import IntegrityError
 
+
 import schemas
 import crud
 from db_config import DbContextManager, create_db_and_tables
 import crud_auth
 from create_admin_user import init_user_admin
+import email_config
 
 ACCESS_TOKEN_EXPIRE_MINUTES = int(os.environ.get("ACCESS_TOKEN_EXPIRE_MINUTES"))
 
@@ -104,13 +106,12 @@ async def login_for_access_token(
     tags=["Auth"],
 )
 async def get_users(
-    user_logged: Annotated[
+    user_logged: Annotated[ # pylint: disable=unused-argument
         schemas.UsersSchema,
         Depends(crud_auth.get_current_admin_user),
     ],
     db: DbContextManager = Depends(DbContextManager),
 ) -> list[schemas.UsersGetSchema]:
-
     return await crud_auth.get_all_users(db)
 
 
@@ -120,9 +121,8 @@ async def get_users(
     tags=["Auth"],
 )
 async def create_or_update_user(
-    user_logged: Annotated[
-        schemas.UsersSchema,
-        Depends(crud_auth.get_current_admin_user)
+    user_logged: Annotated[  # pylint: disable=unused-argument
+        schemas.UsersSchema, Depends(crud_auth.get_current_admin_user)
     ],
     user: schemas.UsersSchema,
     email: str,
@@ -171,7 +171,7 @@ async def create_or_update_user(
     tags=["Auth"],
 )
 async def get_user(
-    user_logged: Annotated[
+    user_logged: Annotated[  # pylint: disable=unused-argument
         schemas.UsersSchema,
         Depends(crud_auth.get_current_admin_user),
     ],
@@ -218,6 +218,50 @@ async def delete_user(
         ) from exception
 
 
+@app.post(
+    "/user/forgot_password/{email}",
+    summary="Recuperação de Acesso",
+    tags=["Auth"],
+)
+async def forgot_password(
+    email: str,
+    db: DbContextManager = Depends(DbContextManager),
+) -> schemas.UsersInputSchema:
+    user = await crud_auth.get_user(db, email)
+
+    if user:
+        access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
+        access_token = crud_auth.create_access_token(
+            data={"sub": user.email}, expires_delta=access_token_expires
+        )
+
+        return await email_config.send_reset_password_mail(email, access_token)
+
+    raise HTTPException(
+        status.HTTP_404_NOT_FOUND, detail=f"Usuário `{email}` não existe"
+    )
+
+
+@app.get(
+    "/user/reset_password/",
+    summary="Criar nova senha a partir do token de acesso",
+    tags=["Auth"],
+)
+async def reset_password(
+    access_token: str,
+    password: str,
+    db: DbContextManager = Depends(DbContextManager),
+):
+    """
+    Gera uma nova senha através do token fornecido por email.
+    """
+    try:
+        return await crud_auth.user_reset_password(db, access_token, password)
+
+    except ValueError as e:
+        raise HTTPException(status_code=400, detail=f"{e}") from e
+
+
 # ## DATA --------------------------------------------------
 
 
@@ -260,11 +304,6 @@ async def create_or_update_plano_entregas(
     plano_entregas: schemas.PlanoEntregasSchema,
     response: Response,
     db: DbContextManager = Depends(DbContextManager),
-    # TODO: Obter meios de verificar permissão opcional. O código abaixo
-    #       bloqueia o acesso, mesmo informando que é opcional.
-    # access_token_info: Optional[FiefAccessTokenInfo] = Depends(
-    #     auth_backend.authenticated(permissions=["all:read"], optional=True)
-    # ),
 ):
     """Cria um novo plano de entregas ou, se existente, substitui um
     plano de entregas por um novo com os dados informados."""
@@ -388,11 +427,6 @@ async def create_or_update_plano_trabalho(
     plano_trabalho: schemas.PlanoTrabalhoSchema,
     response: Response,
     db: DbContextManager = Depends(DbContextManager),
-    # TODO: Obter meios de verificar permissão opcional. O código abaixo
-    #       bloqueia o acesso, mesmo informando que é opcional.
-    # access_token_info: Optional[FiefAccessTokenInfo] = Depends(
-    #     auth_backend.authenticated(permissions=["all:read"], optional=True)
-    # ),
 ):
     """Cria um novo plano de trabalho ou, se existente, substitui um
     plano de trabalho por um novo com os dados informados."""

src/crud_auth.py

@@ -93,11 +93,7 @@ def create_access_token(data: dict, expires_delta: timedelta | None = None):
 
     return encoded_jwt
 
-
-async def get_current_user(
-    token: Annotated[str, Depends(oauth2_scheme)],
-    db: DbContextManager = Depends(DbContextManager),
-):
+async def verify_token(token: str, db: DbContextManager):
     credentials_exception = HTTPException(
         status_code=status.HTTP_401_UNAUTHORIZED,
         detail="Credenciais não podem ser validadas",
@@ -120,6 +116,18 @@ async def get_current_user(
 
     return user
 
+async def get_current_user(
+    token: Annotated[str, Depends(oauth2_scheme)],
+    db: DbContextManager = Depends(DbContextManager),
+):
+    return await verify_token(token, db)
+
+async def get_user_by_token(
+    token: str,
+    db: DbContextManager = Depends(DbContextManager),
+):
+    return await verify_token(token, db)
+
 
 async def get_current_active_user(
     current_user: Annotated[schemas.UsersSchema, Depends(get_current_user)]
@@ -234,3 +242,30 @@ async def delete_user(
         await session.commit()
 
     return f"Usuário `{email}` deletado"
+
+async def user_reset_password(db_session: DbContextManager, 
+                              token: str,
+                              new_password: str) -> str:
+    """Reset password of a user by passing a access token.
+
+    Args:
+        db_session (DbContextManager): Session with api database
+        token (str): access token sended by email
+        new_password (str): the new password for encryption
+
+    Returns:
+        str: Message about updated password
+    """    
+
+
+    user = await get_user_by_token(token, db_session)
+
+    user.password = get_password_hash(new_password)
+
+    async with db_session as session:
+        await session.execute(
+            update(models.Users).filter_by(email=user.email).values(**user.model_dump())
+        )
+        await session.commit()
+
+    return f"Senha do Usuário {user.email} atualizada"

src/email_config.py

@@ -0,0 +1,69 @@
+import os
+import logging
+from fastapi_mail import FastMail, MessageSchema, ConnectionConfig, MessageType
+from fastapi_mail.errors import DBProvaiderError, ConnectionErrors, ApiError
+from starlette.responses import JSONResponse
+
+conf = ConnectionConfig(
+    MAIL_USERNAME=os.environ["MAIL_USERNAME"],
+    MAIL_FROM=os.environ["MAIL_FROM"],
+    MAIL_PORT=os.environ["MAIL_PORT"],
+    MAIL_SERVER=os.environ["MAIL_SERVER"],
+    MAIL_FROM_NAME=os.environ["MAIL_FROM_NAME"],
+    MAIL_STARTTLS=False,
+    MAIL_SSL_TLS=False,
+    MAIL_PASSWORD=os.environ["MAIL_PASSWORD"],
+    USE_CREDENTIALS=False,
+    VALIDATE_CERTS=False
+)
+
+async def send_reset_password_mail(email: str, 
+                                   token: str, 
+                                   ) -> JSONResponse:
+    """Envia o e-mail contendo token para redefinir a senha.
+
+    Args:
+        email (str): o email do usuário.
+        token (str): o token para redefinir a senha.
+
+    Raises:
+        e: exceção gerada pelo FastAPI Mail.
+
+    Returns:
+        JSONResponse: resposta retornada para o endpoint.
+    """
+    token_expiration_minutes = os.environ["ACCESS_TOKEN_EXPIRE_MINUTES"]
+    body = f"""
+        <html>
+          <body>
+            <h3>Recuperação de acesso</h3>
+            <p>Olá, {email}.</p>
+            <p>Foi solicitada a recuperação de sua senha da API PGD.
+            Caso essa solicitação não tenha sido feita por você, por
+            favor ignore esta mensagem.</p>
+            <p>Foi gerado o seguinte token para geração de uma nova
+            senha.</p>
+            <dl>
+              <dt>Token</dt>
+              <dd><code>{token}</code></dd>
+              <dt>Prazo de validade</dt>
+              <dd>{token_expiration_minutes} minutos</dd>
+            </dl>
+            <p>Utilize o endpoint <code>/user/reset_password</code> com
+            este token para redefinir a senha.</p>
+          </body>
+        </html>
+    """
+    try:
+        message = MessageSchema(
+            subject="Recuperação de acesso",
+            recipients=[email],
+            body=body,
+            subtype=MessageType.html
+        )
+        fm = FastMail(conf)
+        await fm.send_message(message)
+        return JSONResponse(status_code=200, content={"message": "Email enviado!"})
+    except (DBProvaiderError, ConnectionErrors, ApiError) as e:
+        logging.error("Erro ao enviar o email %e", e)
+        raise e

tests/conftest.py

@@ -121,6 +121,7 @@ def delete_user(username: str, password: str, del_user_email: str) -> httpx.Resp
 
     return response
 
+
 def create_user(username: str, password: str, new_user: dict) -> httpx.Response:
     """_summary_
 
@@ -135,7 +136,9 @@ def create_user(username: str, password: str, new_user: dict) -> httpx.Response:
 
     headers = prepare_header(username, password)
     new_user_pop = {key: value for key, value in new_user.items() if key != "username"}
-    response = httpx.put(f"{API_BASE_URL}/user/{new_user['email']}", headers=headers, json=new_user_pop)
+    response = httpx.put(
+        f"{API_BASE_URL}/user/{new_user['email']}", headers=headers, json=new_user_pop
+    )
     response.raise_for_status()
 
     return response
@@ -269,7 +272,9 @@ def fixture_truncate_users(admin_credentials: dict):
     ):
         if del_user_email != admin_credentials["username"]:
             response = delete_user(
-                admin_credentials["username"], admin_credentials["password"], del_user_email
+                admin_credentials["username"],
+                admin_credentials["password"],
+                del_user_email,
             )
             response.raise_for_status()
 
@@ -280,7 +285,9 @@ def fixture_register_user_1(
     user1_credentials: dict,
     admin_credentials: dict,
 ) -> httpx.Response:
-    response = create_user(admin_credentials["username"], admin_credentials["password"], user1_credentials)
+    response = create_user(
+        admin_credentials["username"], admin_credentials["password"], user1_credentials
+    )
     response.raise_for_status()
 
     return response
@@ -292,11 +299,14 @@ def fixture_register_user_2(
     user2_credentials: dict,
     admin_credentials: dict,
 ) -> httpx.Response:
-    response = create_user(admin_credentials["username"], admin_credentials["password"], user2_credentials)
+    response = create_user(
+        admin_credentials["username"], admin_credentials["password"], user2_credentials
+    )
     response.raise_for_status()
 
     return response
 
+
 @pytest.fixture(scope="module")
 def header_not_logged_in() -> dict:
     return prepare_header(username=None, password=None)