243 lines
7.0 KiB
Python
243 lines
7.0 KiB
Python
import datetime
|
|
import uuid
|
|
from typing import Optional, Union, List, Literal
|
|
|
|
import pydantic
|
|
from fastapi import Body
|
|
from pydantic import Field, validator
|
|
|
|
from core.models.base import BaseModel, BaseCUAt
|
|
from core.utils.utils import is_mobile_number, is_email
|
|
|
|
|
|
class AuthModel(BaseModel):
|
|
username: str
|
|
password: str
|
|
|
|
|
|
class SessionModel(BaseModel):
|
|
id: Optional[uuid.UUID] = Field(alias='_id')
|
|
user_id: Optional[uuid.UUID]
|
|
access_token: Optional[str]
|
|
token: Optional[str]
|
|
refresh_token: Optional[str]
|
|
token_type: Optional[str]
|
|
agent: Optional[List[str]]
|
|
ip_info: Optional[dict]
|
|
|
|
|
|
class NotifyConfigModel(pydantic.BaseModel):
|
|
sms_info: Optional[bool]
|
|
sms_emergency: Optional[bool]
|
|
push: Optional[bool]
|
|
email: Optional[bool]
|
|
|
|
|
|
class UserOptions(pydantic.BaseModel):
|
|
active_password_login: Optional[bool]
|
|
last_password_changes: Optional[datetime.datetime]
|
|
active_otp: Optional[bool]
|
|
active_token: Optional[bool]
|
|
dark_mode: Optional[Literal['LIGHT', 'DARK', 'SYSTEM', 'DYNAMIC']]
|
|
time_format: Optional[str]
|
|
date_format: Optional[str]
|
|
date_type: Optional[str]
|
|
lang: Optional[str]
|
|
notify: Optional[NotifyConfigModel]
|
|
|
|
|
|
class VisualUserUpdate(pydantic.BaseModel):
|
|
storage: Optional[str]
|
|
|
|
|
|
class UserUpdateModel(BaseModel):
|
|
username: Optional[str]
|
|
email: Optional[str]
|
|
name: Optional[str]
|
|
first_name: Optional[str]
|
|
last_name: Optional[str]
|
|
unit: Optional[str]
|
|
personnel_code: Optional[str]
|
|
avatar: Optional[str]
|
|
desc: Optional[str]
|
|
gender: Optional[Literal[1, 2, 3]]
|
|
home_path: Optional[str]
|
|
force_password_change: Optional[bool]
|
|
disabled: Optional[bool]
|
|
|
|
|
|
class PasswordChangeModel(pydantic.BaseModel):
|
|
old_password: str
|
|
new_password: str = Field(min_length=6, max_length=28)
|
|
|
|
|
|
class UserModel(BaseModel, BaseCUAt):
|
|
id: Optional[Union[uuid.UUID, str]] = Field(alias='_id')
|
|
username: Optional[str]
|
|
mobile: Optional[str]
|
|
email: Optional[str]
|
|
name: Optional[Union[str,List[str]]]
|
|
first_name: Optional[str]
|
|
last_name: Optional[str]
|
|
is_su: Optional[bool]
|
|
unit: Optional[str]
|
|
units: Optional[List[str]]
|
|
personnel_code: Optional[str]
|
|
avatar: Optional[str]
|
|
desc: Optional[str]
|
|
gender: Optional[Literal[1, 2, 3]]
|
|
scopes: Optional[List[str]]
|
|
options: Optional[UserOptions]
|
|
home_path: Optional[str]
|
|
force_password_change: Optional[bool]
|
|
roles: Optional[List[uuid.UUID]]
|
|
_options: Optional[dict]
|
|
disabled: Optional[bool]
|
|
|
|
async def get_roles(self, exclude: [] = None, include: [] = None):
|
|
from auth.utils.mongo import use_auth_collection
|
|
projection = {}
|
|
if exclude:
|
|
projection = {
|
|
f: 0 for f in exclude
|
|
}
|
|
|
|
if include:
|
|
projection = {
|
|
f: 1 for f in include
|
|
}
|
|
|
|
cursor = use_auth_collection('roles').find(
|
|
{'_id': {'$in': [uuid.UUID(str(role)) for role in self.roles]}}, projection)
|
|
return [role async for role in cursor]
|
|
|
|
async def get_units(self):
|
|
roles = await self.get_roles(include=['units'])
|
|
units = []
|
|
for role in roles:
|
|
units.extend(role.get('units', []))
|
|
return units
|
|
|
|
def is_superuser(self) -> bool:
|
|
return self.is_su
|
|
|
|
@validator("mobile")
|
|
def mobile_validation(cls, v):
|
|
if not is_mobile_number(v):
|
|
raise ValueError('Mobile not valid')
|
|
return v
|
|
|
|
@validator("email")
|
|
def email_validation(cls, v):
|
|
if not is_email(v):
|
|
raise ValueError('Email not valid')
|
|
return v
|
|
|
|
|
|
class OAuth2PasswordRequestFormJson:
|
|
"""
|
|
This is a dependency class, use it like:
|
|
|
|
@app.post("/login")
|
|
def login(form_data: OAuth2PasswordRequestForm = Depends()):
|
|
data = form_data.parse()
|
|
print(data.username)
|
|
print(data.password)
|
|
for scope in data.scopes:
|
|
print(scope)
|
|
if data.client_id:
|
|
print(data.client_id)
|
|
if data.client_secret:
|
|
print(data.client_secret)
|
|
return data
|
|
|
|
|
|
It creates the following Form request parameters in your endpoint:
|
|
|
|
grant_type: the OAuth2 spec says it is required and MUST be the fixed string "password".
|
|
Nevertheless, this dependency class is permissive and allows not passing it. If you want to enforce it,
|
|
use instead the OAuth2PasswordRequestFormStrict dependency.
|
|
username: username string. The OAuth2 spec requires the exact field name "username".
|
|
password: password string. The OAuth2 spec requires the exact field name "password".
|
|
scope: Optional string. Several scopes (each one a string) separated by spaces. E.g.
|
|
"items:read items:write users:read profile openid"
|
|
client_id: optional string. OAuth2 recommends sending the client_id and client_secret (if any)
|
|
using HTTP Basic auth, as: client_id:client_secret
|
|
client_secret: optional string. OAuth2 recommends sending the client_id and client_secret (if any)
|
|
using HTTP Basic auth, as: client_id:client_secret
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
grant_type: str = Body(None, regex="password"),
|
|
username: str = Body(...),
|
|
password: str = Body(...),
|
|
scopes: Optional[List[str]] = Body(['default']),
|
|
client_id: Optional[str] = Body(None),
|
|
client_secret: Optional[str] = Body(None),
|
|
):
|
|
self.grant_type = grant_type
|
|
self.username = username
|
|
self.password = password
|
|
self.scopes = scopes
|
|
self.client_id = client_id
|
|
self.client_secret = client_secret
|
|
|
|
|
|
class OAuth2OptRequest:
|
|
|
|
def __init__(
|
|
self,
|
|
grant_type: str = Body(None, regex="password"),
|
|
allow_register: bool = Body(None),
|
|
mobile: str = Body(...),
|
|
token: Optional[int] = Body(None),
|
|
scopes: Optional[List[str]] = Body(['customer']),
|
|
step: Literal[1, 2] = Body(1),
|
|
client_id: Optional[str] = Body(None),
|
|
client_secret: Optional[str] = Body(None),
|
|
):
|
|
self.grant_type = grant_type
|
|
self.allow_register = allow_register
|
|
self.mobile = mobile
|
|
self.token = token
|
|
self.scopes = scopes
|
|
self.step = step
|
|
self.client_id = client_id
|
|
self.client_secret = client_secret
|
|
|
|
|
|
class Token(BaseModel):
|
|
access_token: str
|
|
token_type: str
|
|
|
|
|
|
class TokenData(BaseModel):
|
|
username: Optional[str] = None
|
|
scopes: List[str] = []
|
|
|
|
|
|
class UserInDB(UserModel):
|
|
password: Optional[str]
|
|
|
|
|
|
class CreateTokenModel(BaseModel):
|
|
id: Optional[Union[uuid.UUID, str]] = Field(alias='_id')
|
|
user_id: Optional[uuid.UUID]
|
|
mode: Optional[Literal['user_admin', 'customer']]
|
|
access_token: str
|
|
token: str
|
|
refresh_token: str
|
|
token_type: str = 'bearer'
|
|
|
|
|
|
class PasswordForgottenInput(pydantic.BaseModel):
|
|
password: str
|
|
mode: Optional[Literal['user_admin', 'customer']]
|
|
mobile: str
|
|
token: int
|
|
|
|
|
|
class UIdsInput(BaseModel):
|
|
u_ids: List[uuid.UUID]
|