146 lines
4.7 KiB
Python
146 lines
4.7 KiB
Python
import logging
|
|
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
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
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}
|
|
)
|
|
logger.info(
|
|
"Sent confirmation email to '%s' with code %s", email, confirmation_code
|
|
)
|
|
except Exception as exp:
|
|
logger.error("Could not send confirmation email to '%s' due to: %s", email, 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)
|