87 lines
2.5 KiB
Python
87 lines
2.5 KiB
Python
# scopes.api.auth user
|
|
|
|
from datetime import datetime
|
|
from fastapi import APIRouter, Body, Depends, HTTPException
|
|
from fastapi.security import HTTPBasic, HTTPBasicCredentials
|
|
import jwt
|
|
from pwdlib import PasswordHash
|
|
from pydantic import BaseModel, Field
|
|
from typing import Annotated, List
|
|
|
|
from scopes.api import deps
|
|
from scopes.api.auth import config
|
|
from scopes.storage import user as st_user
|
|
from scopes.storage.common import Storage
|
|
|
|
class UserBase(BaseModel):
|
|
name: str
|
|
login: str
|
|
display_name: str
|
|
email: str | None = None
|
|
groups: List[str] = Field(default_factory=list)
|
|
|
|
class User(UserBase):
|
|
uid: str | None = None
|
|
timeStamp: datetime | None = None
|
|
pw_hash: str | None = Field(exclude=True, default=None)
|
|
|
|
class UserEdit(UserBase):
|
|
password: str | None = None
|
|
|
|
def hashPassword(self) -> str | None:
|
|
pw = self.password
|
|
return pw and hashPassword(pw) or None
|
|
|
|
# password handling, authentication
|
|
|
|
def dummy_auth() -> User:
|
|
return User(name='tester', login='tester', display_name='Tester')
|
|
config.auth_methods['testing'] = dummy_auth
|
|
|
|
pw_hash = PasswordHash.recommended()
|
|
|
|
def hashPassword(pw):
|
|
return pw_hash.hash(pw)
|
|
|
|
def verifyPassword(plain, hashed):
|
|
return pw_hash.verify(plain, hashed)
|
|
|
|
security = HTTPBasic()
|
|
|
|
type Credentials = Annotated[HTTPBasicCredentials, Depends(security)]
|
|
|
|
def basicAuth(users: deps.Users, cred: Credentials) -> User:
|
|
tr = users.queryLast(login=cred.username)
|
|
users.storage.commit()
|
|
user = tr.toObject(User)
|
|
if verifyPassword(cred.password, user.pw_hash):
|
|
return user
|
|
raise HTTPException(status_code=401, detail='credentials',
|
|
headers={"WWW-Authenticate": "Basic"})
|
|
config.auth_methods['basic'] = basicAuth
|
|
|
|
def userFromToken(users: deps.Users, token: deps.AuthToken) -> User:
|
|
payload = jwt.decode(token, config.settings.jwt_key,
|
|
algorithms=[config.settings.jwt_algorithm])
|
|
userName = payload.get('sub')
|
|
tr = users.queryLast(name=userName)
|
|
users.storage.commit()
|
|
return tr.toObject(User)
|
|
|
|
# routes
|
|
|
|
router = APIRouter()
|
|
|
|
@router.get('/users', response_model=list[User])
|
|
async def get_users(users: deps.Users):
|
|
result = [tr.toObject(User) for tr in users.query()]
|
|
return result
|
|
|
|
@router.post('/users', response_model=User)
|
|
async def post_user(user: Annotated[UserEdit, Body()], users: deps.Users,
|
|
auth_user: deps.AuthUser):
|
|
tr = st_user.User.fromObject(user)
|
|
users.save(tr)
|
|
#print('*** post_user', auth_user, tr)
|
|
return tr.toObject(User)
|
|
|