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

155 lines
4.7 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
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)
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 not request.app.state.ENABLE_SIGNUP:
raise HTTPException(400, detail=ERROR_MESSAGES.ACCESS_PROHIBITED)
if not validate_email_format(form_data.email.lower()):
raise HTTPException(400, detail=ERROR_MESSAGES.INVALID_EMAIL_FORMAT)
if Users.get_user_by_email(form_data.email.lower()):
raise HTTPException(400, detail=ERROR_MESSAGES.EMAIL_TAKEN)
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))
############################
# 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,
)