65 lines
1.8 KiB
Python
65 lines
1.8 KiB
Python
# scopes.api.auth user
|
|
|
|
from datetime import datetime
|
|
from fastapi import APIRouter, Body
|
|
import jwt
|
|
from pwdlib import PasswordHash
|
|
from pydantic import BaseModel, Field
|
|
from typing import Annotated, List
|
|
|
|
from scopes.api import deps
|
|
from scopes.config import settings
|
|
from scopes.storage import user as st_user
|
|
from scopes.storage.common import Storage
|
|
|
|
class User(BaseModel):
|
|
uid: str | None = None
|
|
timeStamp: datetime | None = None
|
|
name: str
|
|
login: str
|
|
display_name: str
|
|
email: str | None = None
|
|
pw_hash: str | None = Field(exclude=True, default=None)
|
|
groups: List[str] = Field(default_factory=list)
|
|
|
|
class UserIn(User):
|
|
password: str | None = None
|
|
|
|
# password handling, authentication
|
|
|
|
pw_hash = PasswordHash.recommended()
|
|
|
|
def hashPassword(pw):
|
|
return pw_hash.hash(pw)
|
|
|
|
def verifyPassword(plain, hashed):
|
|
return pw_hash.verify(plain, hashed)
|
|
|
|
def getUser(storage: Storage, userName: str) -> User:
|
|
users = storage.create(st_user.Users)
|
|
user = users.queryLast(name=userName)
|
|
return user.toObject(User)
|
|
|
|
async def authenticate(storage: deps.Storage, token: deps.Auth):
|
|
payload = jwt.decode(token, settings.jwt_key, algorithms=[settings.jwt_algorithm])
|
|
userName = payload.get('sub')
|
|
user = getUser(storage, userName)
|
|
return user
|
|
|
|
# routes
|
|
|
|
router = APIRouter()
|
|
|
|
@router.get('/users', response_model=list[User])
|
|
async def get_users(storage: deps.Storage):
|
|
users = storage.create(st_user.Users)
|
|
result = [tr.toObject(User) for tr in users.query()]
|
|
return result
|
|
|
|
@router.post('/users', response_model=User)
|
|
async def post_user(user: Annotated[UserIn, Body()], storage: deps.Storage):
|
|
users = storage.create(st_user.Users)
|
|
tr = st_user.User.fromObject(user)
|
|
users.save(tr)
|
|
return tr.toObject(User)
|
|
|