PhotosAPI/extensions/users.py

95 lines
4.3 KiB
Python
Raw Normal View History

2023-01-09 16:01:30 +02:00
from datetime import datetime, timedelta
2023-02-16 15:55:03 +02:00
from classes.exceptions import UserAlreadyExists, UserCredentialsInvalid, UserEmailCodeInvalid
2023-02-16 16:04:28 +02:00
from classes.models import EmailConfirmed
2023-01-09 16:01:30 +02:00
from modules.database import col_users, col_albums, col_photos, col_emails, col_videos, col_emails
from modules.app import app
from modules.utils import configGet, logWrite
from modules.scheduler import scheduler
from modules.mailer import mail_sender
from uuid import uuid1
2023-02-16 15:55:03 +02:00
from fastapi import Depends, Form
2023-01-09 16:01:30 +02:00
from fastapi.responses import Response, UJSONResponse
2023-02-16 15:55:03 +02:00
from starlette.status import HTTP_204_NO_CONTENT
2023-01-09 16:01:30 +02:00
from modules.security import (
User,
get_current_active_user,
get_password_hash,
get_user,
verify_password
)
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} )
2023-01-12 17:21:41 +02:00
logWrite(f"Sent confirmation email to '{email}' with code {confirmation_code}")
2023-01-09 16:01:30 +02:00
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
2023-02-16 15:55:03 +02:00
user_confirm_responses = {
2023-02-16 16:04:28 +02:00
200: {
"description": "Successful Response",
"content": {
"application/json": {
"example": {
"detail": configGet("email_confirmed", "messages")
}
}
}
},
2023-02-16 15:55:03 +02:00
400: UserEmailCodeInvalid().openapi
}
2023-01-09 16:01:30 +02:00
if configGet("registration_requires_confirmation") is True:
2023-02-16 16:04:28 +02:00
@app.get("/users/{user}/confirm", response_class=UJSONResponse, response_model=EmailConfirmed, responses=user_confirm_responses)
@app.patch("/users/{user}/confirm", response_class=UJSONResponse, response_model=EmailConfirmed, responses=user_confirm_responses)
2023-01-09 16:01:30 +02:00
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:
2023-02-16 15:55:03 +02:00
raise UserEmailCodeInvalid()
2023-01-09 16:01:30 +02:00
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")} )
2023-02-16 15:55:03 +02:00
user_create_responses = {
409: UserAlreadyExists().openapi
}
2023-01-09 16:01:30 +02:00
if configGet("registration_enabled") is True:
2023-02-16 15:55:03 +02:00
@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()):
2023-01-09 16:01:30 +02:00
if col_users.find_one( {"user": user} ) is not None:
2023-02-16 15:55:03 +02:00
raise UserAlreadyExists()
2023-01-09 16:01:30 +02:00
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)
2023-02-16 15:55:03 +02:00
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)):
2023-01-09 16:01:30 +02:00
user = get_user(current_user.user)
if not user:
return False
if not verify_password(password, user.hash):
2023-02-16 15:55:03 +02:00
raise UserCredentialsInvalid()
2023-01-09 16:01:30 +02:00
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)