115 lines
5.2 KiB
Python
115 lines
5.2 KiB
Python
from fastapi import HTTPException
|
|
from role.models import Roles
|
|
from users.models import Users
|
|
from users.permissions import PermissionTag,has_permission
|
|
from users.serializers import serialize_user
|
|
from users.plugins import clean_user_payload,verify_password,decode_token
|
|
from dotenv import load_dotenv
|
|
load_dotenv()
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
import jwt
|
|
|
|
|
|
class User:
|
|
def __init__(self,session:AsyncSession):
|
|
self.session=session
|
|
|
|
async def _check_role_assignment(self,current_user,role_id,existing_role_id=None):
|
|
if role_id==existing_role_id:
|
|
return
|
|
if not has_permission(current_user.get("permissions") or [],PermissionTag.RBAC_USERS_MANAGE):
|
|
raise HTTPException(status_code=403,detail="Assigning a role requires rbac_users.manage")
|
|
if role_id is None:
|
|
return
|
|
role=await Roles.get_role_by_id(self.session,role_id)
|
|
if role is None or role.is_deleted:
|
|
raise HTTPException(status_code=404,detail="Role not found")
|
|
if not role.is_active:
|
|
raise HTTPException(status_code=400,detail="Role is not active")
|
|
target=set(await Roles.resolve_tags(self.session,role))
|
|
missing=sorted(target-set(current_user.get("permissions") or []))
|
|
if missing:
|
|
raise HTTPException(status_code=403,detail=f"Cannot assign a role with permissions you do not hold: {', '.join(missing)}")
|
|
|
|
async def create_user(self,payload,current_user):
|
|
existing=await Users.get_user_by_email(self.session,payload.get("email"))
|
|
if existing:
|
|
raise HTTPException(status_code=409,detail="Email already registered")
|
|
await self._check_role_assignment(current_user,payload.get("role_id"),None)
|
|
# this is for password hasshing
|
|
fields=clean_user_payload(payload)
|
|
if not fields.get("password"):
|
|
raise HTTPException(status_code=400,detail="Password is required")
|
|
return await Users.insert_user(self.session,fields)
|
|
|
|
async def get_users(self,top,skip,search=None):
|
|
users=await Users.get_users(self.session,top,skip,search)
|
|
return [serialize_user(u) for u in users]
|
|
|
|
async def get_user_by_id(self,record_id):
|
|
user=await Users.get_user_by_id(self.session,record_id)
|
|
if not user:
|
|
raise HTTPException(status_code=404,detail="User not found")
|
|
return serialize_user(user)
|
|
|
|
async def update_user(self,record_id,payload):
|
|
user=await Users.get_user_by_id(self.session,record_id)
|
|
if not user:
|
|
raise HTTPException(status_code=404,detail="User not found")
|
|
# this is for password hashing and dehashing
|
|
fields=clean_user_payload(payload,partial=True)
|
|
email=fields.get("email")
|
|
if email and email!=user.email:
|
|
clash=await Users.get_user_by_email(self.session,email)
|
|
if clash:
|
|
raise HTTPException(status_code=409,detail="Email already registered")
|
|
updated=await Users.update_user(self.session,record_id,fields)
|
|
return serialize_user(updated)
|
|
|
|
async def assign_role(self,record_id,role_id,current_user):
|
|
user=await Users.get_user_by_id(self.session,record_id)
|
|
if not user:
|
|
raise HTTPException(status_code=404,detail="User not found")
|
|
await self._check_role_assignment(current_user,role_id,user.role_id)
|
|
updated=await Users.update_user(self.session,record_id,{"role_id":role_id})
|
|
return serialize_user(updated)
|
|
|
|
async def remove_role(self,record_id,current_user):
|
|
user=await Users.get_user_by_id(self.session,record_id)
|
|
if not user:
|
|
raise HTTPException(status_code=404,detail="User not found")
|
|
await self._check_role_assignment(current_user,None,user.role_id)
|
|
updated=await Users.update_user(self.session,record_id,{"role_id":None})
|
|
return serialize_user(updated)
|
|
|
|
async def delete_user(self,record_id):
|
|
user=await Users.soft_delete_user(self.session,record_id)
|
|
if not user:
|
|
raise HTTPException(status_code=404,detail="User not found")
|
|
return serialize_user(user)
|
|
|
|
async def count_users(self,search=None):
|
|
return await Users.count_users(self.session,search)
|
|
|
|
async def authenticate_user(self,email,password):
|
|
user=await Users.get_user_by_email(self.session,email)
|
|
if not user or not verify_password(password,user.password):
|
|
raise HTTPException(
|
|
status_code=401,
|
|
detail="Incorrect email or password",
|
|
headers={"WWW-Authenticate":"Bearer"},
|
|
)
|
|
if user.is_deleted or not user.is_active:
|
|
raise HTTPException(status_code=401,detail="User is inactive")
|
|
return await Users.get_user_by_id(self.session,user.id)
|
|
|
|
async def refresh_access_token(self,refresh_token):
|
|
try:
|
|
payload=decode_token(refresh_token,expected_type="refresh")
|
|
except jwt.PyJWTError:
|
|
raise HTTPException(status_code=401,detail="Invalid or expired refresh token")
|
|
user=await Users.get_user_by_id(self.session,payload.get("sub"))
|
|
if not user or user.is_deleted or not user.is_active:
|
|
raise HTTPException(status_code=401,detail="User is inactive or does not exist")
|
|
return user
|