90 lines
2.8 KiB
Python
90 lines
2.8 KiB
Python
"""WESP-shaped auth endpoints for copied static UI (/api/auth/*)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Request, Response, status
|
|
from pydantic import BaseModel, Field
|
|
|
|
from app.core.dependencies import get_current_user
|
|
from app.modules.auth.router import _set_refresh_cookie
|
|
from app.modules.auth.service import login as auth_login
|
|
from app.modules.users.models import User
|
|
|
|
router = APIRouter(prefix="/auth", tags=["zootech-wesp-auth"])
|
|
|
|
_USER_RULES = {
|
|
"password_min_len": 8,
|
|
"login_min_len": 3,
|
|
}
|
|
|
|
|
|
class WespLoginIn(BaseModel):
|
|
login: str = Field(min_length=1)
|
|
password: str = Field(min_length=1)
|
|
remember: bool = False
|
|
|
|
|
|
@router.post("/login")
|
|
def wesp_auth_login(payload: WespLoginIn, response: Response):
|
|
email = payload.login.strip()
|
|
if "@" not in email:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail={"status": "error", "message": "Используйте email (admin@compton.example)"},
|
|
)
|
|
try:
|
|
access_token, refresh_token, user = auth_login(email, payload.password)
|
|
except ValueError:
|
|
return {"status": "error", "message": "Неверный логин или пароль"}
|
|
except PermissionError as exc:
|
|
detail = str(exc)
|
|
if detail == "EMAIL_NOT_VERIFIED":
|
|
return {"status": "error", "message": "Email не подтверждён"}
|
|
if detail == "ACCOUNT_BLOCKED":
|
|
return {"status": "error", "message": "Учётная запись заблокирована"}
|
|
return {"status": "error", "message": "Вход временно недоступен"}
|
|
|
|
_set_refresh_cookie(response, refresh_token)
|
|
return {
|
|
"status": "success",
|
|
"message": "Авторизация успешна",
|
|
"authenticated": True,
|
|
"user_login": user.email,
|
|
"access_token": access_token,
|
|
"user": {
|
|
"id": user.id,
|
|
"email": user.email,
|
|
"role": user.role,
|
|
"is_superuser": user.is_superuser,
|
|
"status": user.status,
|
|
},
|
|
}
|
|
|
|
|
|
@router.get("/check")
|
|
def wesp_auth_check(user: User = Depends(get_current_user)):
|
|
return {
|
|
"status": "success",
|
|
"authenticated": True,
|
|
"user_login": user.email,
|
|
"remember_login": False,
|
|
"is_superuser": bool(user.is_superuser),
|
|
"can_lab": bool(user.is_superuser),
|
|
"user_rules": _USER_RULES,
|
|
}
|
|
|
|
|
|
@router.get("/get_current_credentials")
|
|
def wesp_get_current_credentials(user: User = Depends(get_current_user)):
|
|
return {
|
|
"status": "success",
|
|
"login": user.email,
|
|
"password": "",
|
|
"user_rules": _USER_RULES,
|
|
}
|
|
|
|
|
|
@router.post("/logout")
|
|
def wesp_logout():
|
|
return {"status": "success", "message": "Выход выполнен успешно"}
|