69 lines
2.4 KiB
Python
69 lines
2.4 KiB
Python
from typing import Optional, List
|
|
|
|
from fastapi import Depends, APIRouter, Body, Request
|
|
from fastapi.security import SecurityScopes
|
|
from pydantic import BaseModel
|
|
|
|
from auth.models import UserModel, OAuth2PasswordRequestFormJson, CreateTokenModel, OAuth2OptRequest, \
|
|
PasswordForgottenInput
|
|
from auth.utils import get_current_active_user, get_current_user, generate_token, reset_password
|
|
from config import conf
|
|
from core.app import app
|
|
from core.utils import to_http_res
|
|
from core.utils.router import BaseRoute, safe_route
|
|
|
|
_pre = conf.server.prefix_url
|
|
|
|
router = APIRouter(
|
|
prefix=f"{_pre}/auth",
|
|
route_class=BaseRoute,
|
|
)
|
|
|
|
_tag_base = ['token']
|
|
|
|
|
|
class ValidateTokenInput(BaseModel):
|
|
token: str
|
|
scopes: Optional[List[str]] = ['account']
|
|
|
|
|
|
@router.post('/login/otp', response_model=CreateTokenModel, tags=_tag_base)
|
|
async def login_with_otp(request: Request, form_data: OAuth2OptRequest = Depends()):
|
|
client_host = request.client.host
|
|
return await safe_route('login', generate_token, form_data, client_host, request.headers.get('user-agent'))
|
|
|
|
|
|
@router.post("/token", response_model=CreateTokenModel, tags=_tag_base)
|
|
async def login_for_access_token(request: Request, form_data: OAuth2PasswordRequestFormJson = Depends()):
|
|
client_host = request.client.host
|
|
return await safe_route('login', generate_token, form_data, client_host, request.headers.get('user-agent'))
|
|
|
|
|
|
@router.post('/login', response_model=CreateTokenModel, tags=_tag_base)
|
|
async def login(request: Request, data: OAuth2PasswordRequestFormJson = Depends()):
|
|
client_host = request.client.host
|
|
return await safe_route('login', generate_token, data, client_host, request.headers.get('user-agent'))
|
|
|
|
|
|
@router.post('/reset-password', response_model=bool, tags=_tag_base)
|
|
async def reset_password_p(_input: PasswordForgottenInput):
|
|
return await safe_route('update', reset_password, _input)
|
|
|
|
|
|
@router.get('/logout', response_model=bool, tags=_tag_base)
|
|
async def logout():
|
|
return to_http_res(True, 'token')
|
|
|
|
|
|
@router.get("/status", tags=_tag_base)
|
|
async def token_status(current_user: UserModel = Depends(get_current_active_user)):
|
|
return to_http_res('get', {"status": "token is valid", "scope": current_user.scopes})
|
|
|
|
|
|
@router.post('/validate-token', tags=_tag_base)
|
|
async def validate_token(body: ValidateTokenInput = Body(...)):
|
|
return await safe_route('get', get_current_user, SecurityScopes(scopes=body.scopes), body.token)
|
|
|
|
|
|
app.include_router(router)
|