open-webui/backend/apps/web/routers/auths.py

161 lines
4.9 KiB
Python
Raw Normal View History

from fastapi import Response, Request
2023-11-19 01:47:12 +01:00
from fastapi import Depends, FastAPI, HTTPException, status
from datetime import datetime, timedelta
from typing import List, Union
from fastapi import APIRouter
from pydantic import BaseModel
import time
import uuid
2024-01-03 01:22:48 +01:00
2023-11-19 01:47:12 +01:00
from apps.web.models.auths import (
SigninForm,
SignupForm,
2023-12-29 09:12:30 +01:00
UpdatePasswordForm,
2023-11-19 01:47:12 +01:00
UserResponse,
SigninResponse,
Auths,
)
from apps.web.models.users import Users
from utils.utils import get_password_hash, get_current_user, create_token
2024-01-03 01:22:48 +01:00
from utils.misc import get_gravatar_url, validate_email_format
2023-11-19 09:13:59 +01:00
from constants import ERROR_MESSAGES
2023-11-19 01:47:12 +01:00
2023-11-19 09:13:59 +01:00
router = APIRouter()
2023-11-19 01:47:12 +01:00
############################
# GetSessionUser
############################
@router.get("/", response_model=UserResponse)
async def get_session_user(user=Depends(get_current_user)):
return {
"id": user.id,
"email": user.email,
"name": user.name,
"role": user.role,
"profile_image_url": user.profile_image_url,
}
2023-11-19 01:47:12 +01:00
2023-12-29 09:12:30 +01:00
############################
# Update Password
############################
@router.post("/update/password", response_model=bool)
2023-12-30 12:00:21 +01:00
async def update_password(
form_data: UpdatePasswordForm, session_user=Depends(get_current_user)
):
2023-12-29 09:26:47 +01:00
if session_user:
user = Auths.authenticate_user(session_user.email, form_data.password)
2023-12-29 09:12:30 +01:00
2023-12-29 09:26:47 +01:00
if user:
hashed = get_password_hash(form_data.new_password)
2023-12-29 09:29:18 +01:00
return Auths.update_user_password_by_id(user.id, hashed)
2023-12-29 09:26:47 +01:00
else:
raise HTTPException(400, detail=ERROR_MESSAGES.INVALID_PASSWORD)
2023-12-29 09:12:30 +01:00
else:
raise HTTPException(400, detail=ERROR_MESSAGES.INVALID_CRED)
2023-11-19 01:47:12 +01:00
############################
# SignIn
############################
@router.post("/signin", response_model=SigninResponse)
async def signin(form_data: SigninForm):
user = Auths.authenticate_user(form_data.email.lower(), form_data.password)
if user:
token = create_token(data={"email": user.email})
return {
"token": token,
"token_type": "Bearer",
"id": user.id,
"email": user.email,
"name": user.name,
"role": user.role,
2023-11-19 06:41:43 +01:00
"profile_image_url": user.profile_image_url,
2023-11-19 01:47:12 +01:00
}
else:
2023-11-19 06:41:43 +01:00
raise HTTPException(400, detail=ERROR_MESSAGES.INVALID_CRED)
2023-11-19 01:47:12 +01:00
############################
# SignUp
############################
@router.post("/signup", response_model=SigninResponse)
async def signup(request: Request, form_data: SignupForm):
if request.app.state.ENABLE_SIGNUP:
2024-01-03 01:22:48 +01:00
if validate_email_format(form_data.email.lower()):
if not Users.get_user_by_email(form_data.email.lower()):
try:
role = "admin" if Users.get_num_users() == 0 else "pending"
hashed = get_password_hash(form_data.password)
user = Auths.insert_new_auth(
form_data.email.lower(), hashed, form_data.name, role
)
if user:
token = create_token(data={"email": user.email})
# response.set_cookie(key='token', value=token, httponly=True)
return {
"token": token,
"token_type": "Bearer",
"id": user.id,
"email": user.email,
"name": user.name,
"role": user.role,
"profile_image_url": user.profile_image_url,
}
else:
raise HTTPException(
500, detail=ERROR_MESSAGES.CREATE_USER_ERROR
)
except Exception as err:
raise HTTPException(500, detail=ERROR_MESSAGES.DEFAULT(err))
else:
raise HTTPException(400, detail=ERROR_MESSAGES.EMAIL_TAKEN)
else:
2024-01-03 01:22:48 +01:00
raise HTTPException(400, detail=ERROR_MESSAGES.INVALID_EMAIL_FORMAT)
else:
raise HTTPException(400, detail=ERROR_MESSAGES.ACCESS_PROHIBITED)
############################
# ToggleSignUp
############################
@router.get("/signup/enabled", response_model=bool)
async def get_sign_up_status(request: Request, user=Depends(get_current_user)):
if user.role == "admin":
return request.app.state.ENABLE_SIGNUP
else:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail=ERROR_MESSAGES.ACCESS_PROHIBITED,
)
@router.get("/signup/enabled/toggle", response_model=bool)
async def toggle_sign_up(request: Request, user=Depends(get_current_user)):
if user.role == "admin":
request.app.state.ENABLE_SIGNUP = not request.app.state.ENABLE_SIGNUP
return request.app.state.ENABLE_SIGNUP
2023-11-19 01:47:12 +01:00
else:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail=ERROR_MESSAGES.ACCESS_PROHIBITED,
)