mirror of
https://gitlab.com/crafty-controller/crafty-4.git
synced 2024-08-30 18:23:09 +00:00
312 lines
9.9 KiB
Python
312 lines
9.9 KiB
Python
import logging
|
|
import typing as t
|
|
|
|
from app.classes.models.users import HelperUsers
|
|
from app.classes.models.crafty_permissions import (
|
|
PermissionsCrafty,
|
|
EnumPermissionsCrafty,
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class UsersController:
|
|
class ApiPermissionDict(t.TypedDict):
|
|
name: str
|
|
quantity: int
|
|
enabled: bool
|
|
|
|
def __init__(self, helper, users_helper, authentication):
|
|
self.helper = helper
|
|
self.users_helper = users_helper
|
|
self.authentication = authentication
|
|
|
|
_permissions_props = {
|
|
"name": {
|
|
"type": "string",
|
|
"enum": [
|
|
permission.name
|
|
for permission in PermissionsCrafty.get_permissions_list()
|
|
],
|
|
},
|
|
"quantity": {"type": "number", "minimum": 0},
|
|
"enabled": {"type": "boolean"},
|
|
}
|
|
self.user_jsonschema_props: t.Final = {
|
|
"username": {
|
|
"type": "string",
|
|
"maxLength": 20,
|
|
"minLength": 4,
|
|
"pattern": "^[a-z0-9_]+$",
|
|
"examples": ["admin"],
|
|
"title": "Username",
|
|
},
|
|
"password": {
|
|
"type": "string",
|
|
"maxLength": 20,
|
|
"minLength": 4,
|
|
"examples": ["crafty"],
|
|
"title": "Password",
|
|
},
|
|
"email": {
|
|
"type": "string",
|
|
"format": "email",
|
|
"examples": ["default@example.com"],
|
|
"title": "E-Mail",
|
|
},
|
|
"enabled": {
|
|
"type": "boolean",
|
|
"examples": [True],
|
|
"title": "Enabled",
|
|
},
|
|
"lang": {
|
|
"type": "string",
|
|
"maxLength": 10,
|
|
"minLength": 2,
|
|
"examples": ["en"],
|
|
"title": "Language",
|
|
},
|
|
"superuser": {
|
|
"type": "boolean",
|
|
"examples": [False],
|
|
"title": "Superuser",
|
|
},
|
|
"permissions": {
|
|
"type": "array",
|
|
"items": {
|
|
"type": "object",
|
|
"properties": _permissions_props,
|
|
"required": ["name", "quantity", "enabled"],
|
|
},
|
|
},
|
|
"roles": {
|
|
"type": "array",
|
|
"items": {
|
|
"type": "string",
|
|
"minLength": 1,
|
|
},
|
|
},
|
|
"hints": {"type": "boolean"},
|
|
}
|
|
|
|
# **********************************************************************************
|
|
# Users Methods
|
|
# **********************************************************************************
|
|
@staticmethod
|
|
def get_all_users():
|
|
return HelperUsers.get_all_users()
|
|
|
|
@staticmethod
|
|
def get_all_user_ids() -> t.List[int]:
|
|
return HelperUsers.get_all_user_ids()
|
|
|
|
@staticmethod
|
|
def get_id_by_name(username):
|
|
return HelperUsers.get_user_id_by_name(username)
|
|
|
|
@staticmethod
|
|
def get_user_lang_by_id(user_id):
|
|
return HelperUsers.get_user_lang_by_id(user_id)
|
|
|
|
@staticmethod
|
|
def get_user_by_id(user_id):
|
|
return HelperUsers.get_user(user_id)
|
|
|
|
@staticmethod
|
|
def update_server_order(user_id, user_server_order):
|
|
HelperUsers.update_server_order(user_id, user_server_order)
|
|
|
|
@staticmethod
|
|
def get_server_order(user_id):
|
|
return HelperUsers.get_server_order(user_id)
|
|
|
|
@staticmethod
|
|
def user_query(user_id):
|
|
return HelperUsers.user_query(user_id)
|
|
|
|
@staticmethod
|
|
def set_support_path(user_id, support_path):
|
|
HelperUsers.set_support_path(user_id, support_path)
|
|
|
|
def update_user(self, user_id: str, user_data=None, user_crafty_data=None):
|
|
if user_crafty_data is None:
|
|
user_crafty_data = {}
|
|
if user_data is None:
|
|
user_data = {}
|
|
base_data = HelperUsers.get_user(user_id)
|
|
up_data = {}
|
|
added_roles = set()
|
|
removed_roles = set()
|
|
for key in user_data:
|
|
if key == "user_id":
|
|
continue
|
|
if key == "roles":
|
|
added_roles = set(user_data["roles"]).difference(
|
|
set(base_data["roles"])
|
|
)
|
|
removed_roles = set(base_data["roles"]).difference(
|
|
set(user_data["roles"])
|
|
)
|
|
elif key == "password":
|
|
if user_data["password"] is not None and user_data["password"] != "":
|
|
up_data["password"] = self.helper.encode_pass(user_data["password"])
|
|
elif key == "lang":
|
|
up_data["lang"] = user_data["lang"]
|
|
elif key == "hints":
|
|
up_data["hints"] = user_data["hints"]
|
|
elif base_data[key] != user_data[key]:
|
|
up_data[key] = user_data[key]
|
|
up_data["last_update"] = self.helper.get_time_as_string()
|
|
logger.debug(f"user: {user_data} +role:{added_roles} -role:{removed_roles}")
|
|
for role in added_roles:
|
|
HelperUsers.get_or_create(user_id=user_id, role_id=role)
|
|
permissions_mask = user_crafty_data.get("permissions_mask", "000")
|
|
|
|
if "server_quantity" in user_crafty_data:
|
|
limit_server_creation = user_crafty_data["server_quantity"].get(
|
|
EnumPermissionsCrafty.SERVER_CREATION.name, 0
|
|
)
|
|
|
|
limit_user_creation = user_crafty_data["server_quantity"].get(
|
|
EnumPermissionsCrafty.USER_CONFIG.name, 0
|
|
)
|
|
limit_role_creation = user_crafty_data["server_quantity"].get(
|
|
EnumPermissionsCrafty.ROLES_CONFIG.name, 0
|
|
)
|
|
else:
|
|
limit_server_creation = 0
|
|
limit_user_creation = 0
|
|
limit_role_creation = 0
|
|
|
|
PermissionsCrafty.add_or_update_user(
|
|
user_id,
|
|
permissions_mask,
|
|
limit_server_creation,
|
|
limit_user_creation,
|
|
limit_role_creation,
|
|
)
|
|
|
|
self.users_helper.delete_user_roles(user_id, removed_roles)
|
|
|
|
self.users_helper.update_user(user_id, up_data)
|
|
|
|
def raw_update_user(self, user_id: int, up_data: t.Optional[t.Dict[str, t.Any]]):
|
|
"""Directly passes the data to the model helper.
|
|
|
|
Args:
|
|
user_id (int): The id of the user to update.
|
|
up_data (t.Optional[t.Dict[str, t.Any]]): Update data.
|
|
"""
|
|
self.users_helper.update_user(user_id, up_data)
|
|
|
|
def add_user(
|
|
self,
|
|
username,
|
|
password,
|
|
email="default@example.com",
|
|
enabled: bool = True,
|
|
superuser: bool = False,
|
|
):
|
|
return self.users_helper.add_user(
|
|
username,
|
|
password=password,
|
|
email=email,
|
|
enabled=enabled,
|
|
superuser=superuser,
|
|
)
|
|
|
|
@staticmethod
|
|
def add_rawpass_user(
|
|
username,
|
|
password,
|
|
email="default@example.com",
|
|
enabled: bool = True,
|
|
superuser: bool = False,
|
|
):
|
|
return HelperUsers.add_rawpass_user(
|
|
username,
|
|
password=password,
|
|
email=email,
|
|
enabled=enabled,
|
|
superuser=superuser,
|
|
)
|
|
|
|
def remove_user(self, user_id):
|
|
return self.users_helper.remove_user(user_id)
|
|
|
|
@staticmethod
|
|
def user_id_exists(user_id):
|
|
return HelperUsers.user_id_exists(user_id)
|
|
|
|
@staticmethod
|
|
def set_prepare(user_id):
|
|
return HelperUsers.set_prepare(user_id)
|
|
|
|
@staticmethod
|
|
def stop_prepare(user_id):
|
|
return HelperUsers.stop_prepare(user_id)
|
|
|
|
def get_user_id_by_api_token(self, token: str) -> str:
|
|
token_data = self.authentication.check_no_iat(token)
|
|
return token_data["user_id"]
|
|
|
|
def get_user_by_api_token(self, token: str):
|
|
_, _, user = self.authentication.check_err(token)
|
|
return user
|
|
|
|
def get_api_key_by_token(self, token: str):
|
|
key, _, _ = self.authentication.check(token)
|
|
return key
|
|
|
|
# **********************************************************************************
|
|
# User Roles Methods
|
|
# **********************************************************************************
|
|
|
|
@staticmethod
|
|
def get_user_roles_id(user_id):
|
|
return HelperUsers.get_user_roles_id(user_id)
|
|
|
|
@staticmethod
|
|
def get_user_roles_names(user_id):
|
|
return HelperUsers.get_user_roles_names(user_id)
|
|
|
|
def add_role_to_user(self, user_id, role_id):
|
|
return self.users_helper.add_role_to_user(user_id, role_id)
|
|
|
|
def add_user_roles(self, user):
|
|
return self.users_helper.add_user_roles(user)
|
|
|
|
@staticmethod
|
|
def user_role_query(user_id):
|
|
return HelperUsers.user_role_query(user_id)
|
|
|
|
# **********************************************************************************
|
|
# Api Keys Methods
|
|
# **********************************************************************************
|
|
|
|
@staticmethod
|
|
def get_user_api_keys(user_id: str):
|
|
return HelperUsers.get_user_api_keys(user_id)
|
|
|
|
@staticmethod
|
|
def get_user_api_key(key_id: str):
|
|
return HelperUsers.get_user_api_key(key_id)
|
|
|
|
def add_user_api_key(
|
|
self,
|
|
name: str,
|
|
user_id: str,
|
|
superuser: bool = False,
|
|
server_permissions_mask: t.Optional[str] = None,
|
|
crafty_permissions_mask: t.Optional[str] = None,
|
|
):
|
|
return self.users_helper.add_user_api_key(
|
|
name, user_id, superuser, server_permissions_mask, crafty_permissions_mask
|
|
)
|
|
|
|
def delete_user_api_keys(self, user_id: str):
|
|
return self.users_helper.delete_user_api_keys(user_id)
|
|
|
|
def delete_user_api_key(self, key_id: str):
|
|
return self.users_helper.delete_user_api_key(key_id)
|