PhotosAPI/extensions/users.py

from datetime import datetime, timedelta
from uuid import uuid1

from fastapi import Depends, Form
from fastapi.responses import Response, UJSONResponse
from starlette.status import HTTP_204_NO_CONTENT

from classes.exceptions import (
    UserAlreadyExists,
    UserCredentialsInvalid,
    UserEmailCodeInvalid,
)
from modules.app import app
from modules.database import col_albums, col_emails, col_photos, col_users, col_videos
from modules.mailer import mail_sender
from modules.scheduler import scheduler
from modules.security import (
    User,
    get_current_active_user,
    get_password_hash,
    get_user,
    verify_password,
)
from modules.utils import configGet, logWrite


async def send_confirmation(user: str, email: str):
    confirmation_code = str(uuid1())
    try:
        mail_sender.sendmail(
            from_addr=configGet("sender", "mailer", "smtp"),
            to_addrs=email,
            msg=f'From: {configGet("sender", "mailer", "smtp")}\nSubject: Email confirmation\n\n'
            + configGet(
                "message", "mailer", "messages", "registration_confirmation"
            ).format(
                configGet("external_address")
                + f"/users/{user}/confirm?code={confirmation_code}"
            ),
        )
        col_emails.insert_one(
            {"user": user, "email": email, "used": False, "code": confirmation_code}
        )
        logWrite(f"Sent confirmation email to '{email}' with code {confirmation_code}")
    except Exception as exp:
        logWrite(f"Could not send confirmation email to '{email}' due to: {exp}")


@app.get("/users/me/", response_model=User)
async def user_me(current_user: User = Depends(get_current_active_user)):
    return current_user


user_confirm_responses = {
    200: {
        "description": "Successful Response",
        "content": {
            "application/json": {
                "example": {"detail": configGet("email_confirmed", "messages")}
            }
        },
    },
    400: UserEmailCodeInvalid().openapi,
}
if configGet("registration_requires_confirmation") is True:

    @app.get(
        "/users/{user}/confirm",
        response_class=UJSONResponse,
        responses=user_confirm_responses,
    )
    @app.patch(
        "/users/{user}/confirm",
        response_class=UJSONResponse,
        responses=user_confirm_responses,
    )
    async def user_confirm(user: str, code: str):
        confirm_record = col_emails.find_one(
            {"user": user, "code": code, "used": False}
        )
        if confirm_record is None:
            raise UserEmailCodeInvalid()
        col_emails.find_one_and_update(
            {"_id": confirm_record["_id"]}, {"$set": {"used": True}}
        )
        col_users.find_one_and_update(
            {"user": confirm_record["user"]}, {"$set": {"disabled": False}}
        )
        return UJSONResponse({"detail": configGet("email_confirmed", "messages")})


user_create_responses = {409: UserAlreadyExists().openapi}
if configGet("registration_enabled") is True:

    @app.post(
        "/users", status_code=HTTP_204_NO_CONTENT, responses=user_create_responses
    )
    async def user_create(
        user: str = Form(), email: str = Form(), password: str = Form()
    ):
        if col_users.find_one({"user": user}) is not None:
            raise UserAlreadyExists()
        col_users.insert_one(
            {
                "user": user,
                "email": email,
                "hash": get_password_hash(password),
                "disabled": configGet("registration_requires_confirmation"),
            }
        )
        if configGet("registration_requires_confirmation") is True:
            scheduler.add_job(
                send_confirmation,
                trigger="date",
                run_date=datetime.now() + timedelta(seconds=1),
                kwargs={"user": user, "email": email},
            )
        return Response(status_code=HTTP_204_NO_CONTENT)


user_delete_responses = {401: UserCredentialsInvalid().openapi}


@app.delete(
    "/users/me/", status_code=HTTP_204_NO_CONTENT, responses=user_delete_responses
)
async def user_delete(
    password: str = Form(), current_user: User = Depends(get_current_active_user)
):
    user = get_user(current_user.user)
    if not user:
        return False
    if not verify_password(password, user.hash):
        raise UserCredentialsInvalid()
    col_users.delete_many({"user": current_user.user})
    col_emails.delete_many({"user": current_user.user})
    col_photos.delete_many({"user": current_user.user})
    col_videos.delete_many({"user": current_user.user})
    col_albums.delete_many({"user": current_user.user})
    return Response(status_code=HTTP_204_NO_CONTENT)