diff --git a/app/classes/controllers/crafty_perms_controller.py b/app/classes/controllers/crafty_perms_controller.py
index 75cf2601..4f50d0bb 100644
--- a/app/classes/controllers/crafty_perms_controller.py
+++ b/app/classes/controllers/crafty_perms_controller.py
@@ -1,12 +1,15 @@
import logging
-from app.classes.models.crafty_permissions import crafty_permissions, Enum_Permissions_Crafty
+from app.classes.models.crafty_permissions import (
+ crafty_permissions,
+ Enum_Permissions_Crafty,
+)
from app.classes.models.users import ApiKeys
logger = logging.getLogger(__name__)
-class Crafty_Perms_Controller:
+class Crafty_Perms_Controller:
@staticmethod
def list_defined_crafty_permissions():
permissions_list = crafty_permissions.get_permissions_list()
@@ -18,23 +21,29 @@ class Crafty_Perms_Controller:
return permissions_mask
@staticmethod
- def set_permission(permission_mask, permission_tested: Enum_Permissions_Crafty, value):
- return crafty_permissions.set_permission(permission_mask, permission_tested, value)
+ def set_permission(
+ permission_mask, permission_tested: Enum_Permissions_Crafty, value
+ ):
+ return crafty_permissions.set_permission(
+ permission_mask, permission_tested, value
+ )
@staticmethod
def can_create_server(user_id):
- return crafty_permissions.can_add_in_crafty(user_id, Enum_Permissions_Crafty.Server_Creation)
+ return crafty_permissions.can_add_in_crafty(
+ user_id, Enum_Permissions_Crafty.Server_Creation
+ )
@staticmethod
- def can_add_user(): # Add back argument 'user_id' when you work on this
- #TODO: Complete if we need a User Addition limit
- #return crafty_permissions.can_add_in_crafty(user_id, Enum_Permissions_Crafty.User_Config)
+ def can_add_user(): # Add back argument 'user_id' when you work on this
+ # TODO: Complete if we need a User Addition limit
+ # return crafty_permissions.can_add_in_crafty(user_id, Enum_Permissions_Crafty.User_Config)
return True
@staticmethod
- def can_add_role(): # Add back argument 'user_id' when you work on this
- #TODO: Complete if we need a Role Addition limit
- #return crafty_permissions.can_add_in_crafty(user_id, Enum_Permissions_Crafty.Roles_Config)
+ def can_add_role(): # Add back argument 'user_id' when you work on this
+ # TODO: Complete if we need a Role Addition limit
+ # return crafty_permissions.can_add_in_crafty(user_id, Enum_Permissions_Crafty.Roles_Config)
return True
@staticmethod
diff --git a/app/classes/controllers/management_controller.py b/app/classes/controllers/management_controller.py
index 59419ca0..f2d62d75 100644
--- a/app/classes/controllers/management_controller.py
+++ b/app/classes/controllers/management_controller.py
@@ -5,18 +5,19 @@ from app.classes.models.servers import servers_helper
logger = logging.getLogger(__name__)
+
class Management_Controller:
- #************************************************************************************************
+ # ************************************************************************************************
# Host_Stats Methods
- #************************************************************************************************
+ # ************************************************************************************************
@staticmethod
def get_latest_hosts_stats():
return management_helper.get_latest_hosts_stats()
- #************************************************************************************************
+ # ************************************************************************************************
# Commands Methods
- #************************************************************************************************
+ # ************************************************************************************************
@staticmethod
def get_unactioned_commands():
return management_helper.get_unactioned_commands()
@@ -26,43 +27,61 @@ class Management_Controller:
server_name = servers_helper.get_server_friendly_name(server_id)
# Example: Admin issued command start_server for server Survival
- management_helper.add_to_audit_log(user_id, f"issued command {command} for server {server_name}", server_id, remote_ip)
+ management_helper.add_to_audit_log(
+ user_id,
+ f"issued command {command} for server {server_name}",
+ server_id,
+ remote_ip,
+ )
management_helper.add_command(server_id, user_id, remote_ip, command)
@staticmethod
def mark_command_complete(command_id=None):
return management_helper.mark_command_complete(command_id)
- #************************************************************************************************
+ # ************************************************************************************************
# Audit_Log Methods
- #************************************************************************************************
+ # ************************************************************************************************
@staticmethod
def get_actity_log():
return management_helper.get_actity_log()
@staticmethod
def add_to_audit_log(user_id, log_msg, server_id=None, source_ip=None):
- return management_helper.add_to_audit_log(user_id, log_msg, server_id, source_ip)
+ return management_helper.add_to_audit_log(
+ user_id, log_msg, server_id, source_ip
+ )
@staticmethod
def add_to_audit_log_raw(user_name, user_id, server_id, log_msg, source_ip):
- return management_helper.add_to_audit_log_raw(user_name, user_id, server_id, log_msg, source_ip)
+ return management_helper.add_to_audit_log_raw(
+ user_name, user_id, server_id, log_msg, source_ip
+ )
- #************************************************************************************************
+ # ************************************************************************************************
# Schedules Methods
- #************************************************************************************************
+ # ************************************************************************************************
@staticmethod
- def create_scheduled_task(server_id, action, interval, interval_type, start_time, command, comment=None, enabled=True):
+ def create_scheduled_task(
+ server_id,
+ action,
+ interval,
+ interval_type,
+ start_time,
+ command,
+ comment=None,
+ enabled=True,
+ ):
return management_helper.create_scheduled_task(
- server_id,
- action,
- interval,
- interval_type,
- start_time,
- command,
- comment,
- enabled
- )
+ server_id,
+ action,
+ interval,
+ interval_type,
+ start_time,
+ command,
+ comment,
+ enabled,
+ )
@staticmethod
def delete_scheduled_task(schedule_id):
@@ -96,16 +115,24 @@ class Management_Controller:
def get_schedules_enabled():
return management_helper.get_schedules_enabled()
- #************************************************************************************************
+ # ************************************************************************************************
# Backups Methods
- #************************************************************************************************
+ # ************************************************************************************************
@staticmethod
def get_backup_config(server_id):
return management_helper.get_backup_config(server_id)
@staticmethod
- def set_backup_config(server_id: int, backup_path: str = None, max_backups: int = None, excluded_dirs: list = None, compress: bool = False,):
- return management_helper.set_backup_config(server_id, backup_path, max_backups, excluded_dirs, compress)
+ def set_backup_config(
+ server_id: int,
+ backup_path: str = None,
+ max_backups: int = None,
+ excluded_dirs: list = None,
+ compress: bool = False,
+ ):
+ return management_helper.set_backup_config(
+ server_id, backup_path, max_backups, excluded_dirs, compress
+ )
@staticmethod
def get_excluded_backup_dirs(server_id: int):
diff --git a/app/classes/controllers/roles_controller.py b/app/classes/controllers/roles_controller.py
index bded6e23..224995df 100644
--- a/app/classes/controllers/roles_controller.py
+++ b/app/classes/controllers/roles_controller.py
@@ -7,11 +7,11 @@ from app.classes.shared.helpers import helper
logger = logging.getLogger(__name__)
-class Roles_Controller:
+class Roles_Controller:
@staticmethod
def get_all_roles():
- return roles_helper.get_all_roles()
+ return roles_helper.get_all_roles()
@staticmethod
def get_roleid_by_name(role_name):
@@ -21,9 +21,8 @@ class Roles_Controller:
def get_role(role_id):
return roles_helper.get_role(role_id)
-
@staticmethod
- def update_role(role_id: str, role_data = None, permissions_mask: str = "00000000"):
+ def update_role(role_id: str, role_data=None, permissions_mask: str = "00000000"):
if role_data is None:
role_data = {}
base_data = Roles_Controller.get_role_with_servers(role_id)
@@ -34,15 +33,17 @@ class Roles_Controller:
if key == "role_id":
continue
elif key == "servers":
- added_servers = role_data['servers'].difference(base_data['servers'])
- removed_servers = base_data['servers'].difference(role_data['servers'])
+ added_servers = role_data["servers"].difference(base_data["servers"])
+ removed_servers = base_data["servers"].difference(role_data["servers"])
elif base_data[key] != role_data[key]:
up_data[key] = role_data[key]
- up_data['last_update'] = helper.get_time_as_string()
- logger.debug(f"role: {role_data} +server:{added_servers} -server{removed_servers}")
+ up_data["last_update"] = helper.get_time_as_string()
+ logger.debug(
+ f"role: {role_data} +server:{added_servers} -server{removed_servers}"
+ )
for server in added_servers:
server_permissions.get_or_create(role_id, server, permissions_mask)
- for server in base_data['servers']:
+ for server in base_data["servers"]:
server_permissions.update_role_permission(role_id, server, permissions_mask)
# TODO: This is horribly inefficient and we should be using bulk queries but im going for functionality at this point
server_permissions.delete_roles_permissions(role_id, removed_servers)
@@ -56,7 +57,7 @@ class Roles_Controller:
@staticmethod
def remove_role(role_id):
role_data = Roles_Controller.get_role_with_servers(role_id)
- server_permissions.delete_roles_permissions(role_id, role_data['servers'])
+ server_permissions.delete_roles_permissions(role_id, role_data["servers"])
users_helper.remove_roles_from_role_id(role_id)
return roles_helper.remove_role(role_id)
@@ -74,9 +75,9 @@ class Roles_Controller:
servers = set()
for s in servers_query:
servers.add(s.server_id.server_id)
- role['servers'] = servers
- #logger.debug("role: ({}) {}".format(role_id, role))
+ role["servers"] = servers
+ # logger.debug("role: ({}) {}".format(role_id, role))
return role
else:
- #logger.debug("role: ({}) {}".format(role_id, {}))
+ # logger.debug("role: ({}) {}".format(role_id, {}))
return {}
diff --git a/app/classes/controllers/server_perms_controller.py b/app/classes/controllers/server_perms_controller.py
index bd0ae36d..931f4f1e 100644
--- a/app/classes/controllers/server_perms_controller.py
+++ b/app/classes/controllers/server_perms_controller.py
@@ -1,6 +1,9 @@
import logging
-from app.classes.models.server_permissions import server_permissions, Enum_Permissions_Server
+from app.classes.models.server_permissions import (
+ server_permissions,
+ Enum_Permissions_Server,
+)
from app.classes.models.users import users_helper, ApiKeys
from app.classes.models.roles import roles_helper
from app.classes.models.servers import servers_helper
@@ -8,8 +11,8 @@ from app.classes.shared.main_models import db_helper
logger = logging.getLogger(__name__)
-class Server_Perms_Controller:
+class Server_Perms_Controller:
@staticmethod
def get_server_user_list(server_id):
return server_permissions.get_server_user_list(server_id)
@@ -42,20 +45,28 @@ class Server_Perms_Controller:
role_list = server_permissions.get_server_roles(old_server_id)
for role in role_list:
server_permissions.add_role_server(
- new_server_id, role.role_id,
- server_permissions.get_permissions_mask(int(role.role_id), int(old_server_id)))
- #server_permissions.add_role_server(new_server_id, role.role_id, '00001000')
+ new_server_id,
+ role.role_id,
+ server_permissions.get_permissions_mask(
+ int(role.role_id), int(old_server_id)
+ ),
+ )
+ # server_permissions.add_role_server(new_server_id, role.role_id, '00001000')
- #************************************************************************************************
+ # ************************************************************************************************
# Servers Permissions Methods
- #************************************************************************************************
+ # ************************************************************************************************
@staticmethod
def get_permissions_mask(role_id, server_id):
return server_permissions.get_permissions_mask(role_id, server_id)
@staticmethod
- def set_permission(permission_mask, permission_tested: Enum_Permissions_Server, value):
- return server_permissions.set_permission(permission_mask, permission_tested, value)
+ def set_permission(
+ permission_mask, permission_tested: Enum_Permissions_Server, value
+ ):
+ return server_permissions.set_permission(
+ permission_mask, permission_tested, value
+ )
@staticmethod
def get_role_permissions_list(role_id):
@@ -86,7 +97,9 @@ class Server_Perms_Controller:
roles_list.append(roles_helper.get_role(u.role_id))
for r in roles_list:
- role_test = server_permissions.get_role_servers_from_role_id(r.get('role_id'))
+ role_test = server_permissions.get_role_servers_from_role_id(
+ r.get("role_id")
+ )
for t in role_test:
role_server.append(t)
@@ -94,6 +107,8 @@ class Server_Perms_Controller:
authorized_servers.append(servers_helper.get_server_data_by_id(s.server_id))
for s in authorized_servers:
- latest = servers_helper.get_latest_server_stats(s.get('server_id'))
- server_data.append({'server_data': s, "stats": db_helper.return_rows(latest)[0]})
+ latest = servers_helper.get_latest_server_stats(s.get("server_id"))
+ server_data.append(
+ {"server_data": s, "stats": db_helper.return_rows(latest)[0]}
+ )
return server_data
diff --git a/app/classes/controllers/servers_controller.py b/app/classes/controllers/servers_controller.py
index 4a596428..cfec8375 100644
--- a/app/classes/controllers/servers_controller.py
+++ b/app/classes/controllers/servers_controller.py
@@ -5,17 +5,21 @@ import json
from app.classes.controllers.roles_controller import Roles_Controller
from app.classes.models.servers import servers_helper
from app.classes.models.users import users_helper, ApiKeys
-from app.classes.models.server_permissions import server_permissions, Enum_Permissions_Server
+from app.classes.models.server_permissions import (
+ server_permissions,
+ Enum_Permissions_Server,
+)
from app.classes.shared.helpers import helper
from app.classes.shared.main_models import db_helper
logger = logging.getLogger(__name__)
+
class Servers_Controller:
- #************************************************************************************************
+ # ************************************************************************************************
# Generic Servers Methods
- #************************************************************************************************
+ # ************************************************************************************************
@staticmethod
def create_server(
name: str,
@@ -27,7 +31,8 @@ class Servers_Controller:
server_log_file: str,
server_stop: str,
server_type: str,
- server_port=25565):
+ server_port=25565,
+ ):
return servers_helper.create_server(
name,
server_uuid,
@@ -38,7 +43,8 @@ class Servers_Controller:
server_log_file,
server_stop,
server_type,
- server_port)
+ server_port,
+ )
@staticmethod
def get_server_obj(server_id):
@@ -66,8 +72,8 @@ class Servers_Controller:
for role in roles_list:
role_id = role.role_id
role_data = Roles_Controller.get_role_with_servers(role_id)
- role_data['servers'] = {server_id}
- server_permissions.delete_roles_permissions(role_id, role_data['servers'])
+ role_data["servers"] = {server_id}
+ server_permissions.delete_roles_permissions(role_id, role_data["servers"])
server_permissions.remove_roles_of_server(server_id)
servers_helper.remove_server(server_id)
@@ -75,9 +81,9 @@ class Servers_Controller:
def get_server_data_by_id(server_id):
return servers_helper.get_server_data_by_id(server_id)
- #************************************************************************************************
+ # ************************************************************************************************
# Servers Methods
- #************************************************************************************************
+ # ************************************************************************************************
@staticmethod
def get_all_defined_servers():
return servers_helper.get_all_defined_servers()
@@ -100,17 +106,26 @@ class Servers_Controller:
@staticmethod
def get_authorized_servers_stats_api_key(api_key: ApiKeys):
server_data = []
- authorized_servers = Servers_Controller.get_authorized_servers(api_key.user.user_id)
+ authorized_servers = Servers_Controller.get_authorized_servers(
+ api_key.user.user_id
+ )
for s in authorized_servers:
- latest = servers_helper.get_latest_server_stats(s.get('server_id'))
- key_permissions = server_permissions.get_api_key_permissions_list(api_key, s.get('server_id'))
+ latest = servers_helper.get_latest_server_stats(s.get("server_id"))
+ key_permissions = server_permissions.get_api_key_permissions_list(
+ api_key, s.get("server_id")
+ )
if Enum_Permissions_Server.Commands in key_permissions:
user_command_permission = True
else:
user_command_permission = False
- server_data.append({'server_data': s, "stats": db_helper.return_rows(latest)[0],
- "user_command_permission": user_command_permission})
+ server_data.append(
+ {
+ "server_data": s,
+ "stats": db_helper.return_rows(latest)[0],
+ "user_command_permission": user_command_permission,
+ }
+ )
return server_data
@staticmethod
@@ -119,18 +134,22 @@ class Servers_Controller:
authorized_servers = Servers_Controller.get_authorized_servers(user_id)
for s in authorized_servers:
- latest = servers_helper.get_latest_server_stats(s.get('server_id'))
+ latest = servers_helper.get_latest_server_stats(s.get("server_id"))
# TODO
- user_permissions = server_permissions.get_user_id_permissions_list(user_id, s.get('server_id'))
+ user_permissions = server_permissions.get_user_id_permissions_list(
+ user_id, s.get("server_id")
+ )
if Enum_Permissions_Server.Commands in user_permissions:
user_command_permission = True
else:
user_command_permission = False
- server_data.append({
- 'server_data': s,
- 'stats': db_helper.return_rows(latest)[0],
- 'user_command_permission': user_command_permission
- })
+ server_data.append(
+ {
+ "server_data": s,
+ "stats": db_helper.return_rows(latest)[0],
+ "user_command_permission": user_command_permission,
+ }
+ )
return server_data
@@ -138,9 +157,9 @@ class Servers_Controller:
def get_server_friendly_name(server_id):
return servers_helper.get_server_friendly_name(server_id)
- #************************************************************************************************
+ # ************************************************************************************************
# Servers_Stats Methods
- #************************************************************************************************
+ # ************************************************************************************************
@staticmethod
def get_server_stats_by_id(server_id):
return servers_helper.get_server_stats_by_id(server_id)
@@ -157,7 +176,9 @@ class Servers_Controller:
def server_id_authorized(server_id_a, user_id):
user_roles = users_helper.user_role_query(user_id)
for role in user_roles:
- for server_id_b in server_permissions.get_role_servers_from_role_id(role.role_id):
+ for server_id_b in server_permissions.get_role_servers_from_role_id(
+ role.role_id
+ ):
if str(server_id_a) == str(server_id_b.server_id):
return True
return False
@@ -197,21 +218,23 @@ class Servers_Controller:
def get_update_status(server_id):
return servers_helper.get_update_status(server_id)
- #************************************************************************************************
+ # ************************************************************************************************
# Servers Helpers Methods
- #************************************************************************************************
+ # ************************************************************************************************
@staticmethod
def get_banned_players(server_id):
stats = servers_helper.get_server_stats_by_id(server_id)
- server_path = stats['server_id']['path']
- path = os.path.join(server_path, 'banned-players.json')
+ server_path = stats["server_id"]["path"]
+ path = os.path.join(server_path, "banned-players.json")
try:
- with open(helper.get_os_understandable_path(path), encoding='utf-8') as file:
+ with open(
+ helper.get_os_understandable_path(path), encoding="utf-8"
+ ) as file:
content = file.read()
file.close()
except Exception as ex:
- print (ex)
+ print(ex)
return None
return json.loads(content)
@@ -219,18 +242,20 @@ class Servers_Controller:
def check_for_old_logs(self):
servers = servers_helper.get_all_defined_servers()
for server in servers:
- logs_path = os.path.split(server['log_path'])[0]
- latest_log_file = os.path.split(server['log_path'])[1]
- logs_delete_after = int(server['logs_delete_after'])
+ logs_path = os.path.split(server["log_path"])[0]
+ latest_log_file = os.path.split(server["log_path"])[1]
+ logs_delete_after = int(server["logs_delete_after"])
if logs_delete_after == 0:
continue
- log_files = list(filter(
- lambda val: val != latest_log_file,
- os.listdir(logs_path)
- ))
+ log_files = list(
+ filter(lambda val: val != latest_log_file, os.listdir(logs_path))
+ )
for log_file in log_files:
log_file_path = os.path.join(logs_path, log_file)
- if helper.check_file_exists(log_file_path) and \
- helper.is_file_older_than_x_days(log_file_path, logs_delete_after):
+ if helper.check_file_exists(
+ log_file_path
+ ) and helper.is_file_older_than_x_days(
+ log_file_path, logs_delete_after
+ ):
os.remove(log_file_path)
diff --git a/app/classes/controllers/users_controller.py b/app/classes/controllers/users_controller.py
index bd51fd8f..af590046 100644
--- a/app/classes/controllers/users_controller.py
+++ b/app/classes/controllers/users_controller.py
@@ -2,17 +2,21 @@ import logging
from typing import Optional
from app.classes.models.users import users_helper
-from app.classes.models.crafty_permissions import crafty_permissions, Enum_Permissions_Crafty
+from app.classes.models.crafty_permissions import (
+ crafty_permissions,
+ Enum_Permissions_Crafty,
+)
from app.classes.shared.helpers import helper
from app.classes.shared.authentication import authentication
logger = logging.getLogger(__name__)
+
class Users_Controller:
- #************************************************************************************************
+ # ************************************************************************************************
# Users Methods
- #************************************************************************************************
+ # ************************************************************************************************
@staticmethod
def get_all_users():
return users_helper.get_all_users()
@@ -59,26 +63,31 @@ class Users_Controller:
if key == "user_id":
continue
elif key == "roles":
- added_roles = user_data['roles'].difference(base_data['roles'])
- removed_roles = base_data['roles'].difference(user_data['roles'])
+ added_roles = user_data["roles"].difference(base_data["roles"])
+ removed_roles = base_data["roles"].difference(user_data["roles"])
elif key == "password":
- if user_data['password'] is not None and user_data['password'] != "":
- up_data['password'] = helper.encode_pass(user_data['password'])
+ if user_data["password"] is not None and user_data["password"] != "":
+ up_data["password"] = helper.encode_pass(user_data["password"])
elif base_data[key] != user_data[key]:
up_data[key] = user_data[key]
- up_data['last_update'] = helper.get_time_as_string()
- up_data['lang'] = user_data['lang']
+ up_data["last_update"] = helper.get_time_as_string()
+ up_data["lang"] = user_data["lang"]
logger.debug(f"user: {user_data} +role:{added_roles} -role:{removed_roles}")
for role in added_roles:
users_helper.get_or_create(user_id=user_id, role_id=role)
- permissions_mask = user_crafty_data.get('permissions_mask', '000')
+ permissions_mask = user_crafty_data.get("permissions_mask", "000")
- if 'server_quantity' in user_crafty_data:
- limit_server_creation = user_crafty_data['server_quantity'][
- Enum_Permissions_Crafty.Server_Creation.name]
+ if "server_quantity" in user_crafty_data:
+ limit_server_creation = user_crafty_data["server_quantity"][
+ Enum_Permissions_Crafty.Server_Creation.name
+ ]
- limit_user_creation = user_crafty_data['server_quantity'][Enum_Permissions_Crafty.User_Config.name]
- limit_role_creation = user_crafty_data['server_quantity'][Enum_Permissions_Crafty.Roles_Config.name]
+ limit_user_creation = user_crafty_data["server_quantity"][
+ Enum_Permissions_Crafty.User_Config.name
+ ]
+ limit_role_creation = user_crafty_data["server_quantity"][
+ Enum_Permissions_Crafty.Roles_Config.name
+ ]
else:
limit_server_creation = 0
limit_user_creation = 0
@@ -89,19 +98,44 @@ class Users_Controller:
permissions_mask,
limit_server_creation,
limit_user_creation,
- limit_role_creation)
+ limit_role_creation,
+ )
users_helper.delete_user_roles(user_id, removed_roles)
users_helper.update_user(user_id, up_data)
@staticmethod
- def add_user(username, password, email="default@example.com", enabled: bool = True, superuser: bool = False):
- return users_helper.add_user(username, password=password, email=email, enabled=enabled, superuser=superuser)
+ def add_user(
+ username,
+ password,
+ email="default@example.com",
+ enabled: bool = True,
+ superuser: bool = False,
+ ):
+ return 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 users_helper.add_rawpass_user(username, password=password, email=email, enabled=enabled, superuser=superuser)
+ def add_rawpass_user(
+ username,
+ password,
+ email="default@example.com",
+ enabled: bool = True,
+ superuser: bool = False,
+ ):
+ return users_helper.add_rawpass_user(
+ username,
+ password=password,
+ email=email,
+ enabled=enabled,
+ superuser=superuser,
+ )
@staticmethod
def remove_user(user_id):
@@ -122,7 +156,7 @@ class Users_Controller:
@staticmethod
def get_user_id_by_api_token(token: str) -> str:
token_data = authentication.check_no_iat(token)
- return token_data['user_id']
+ return token_data["user_id"]
@staticmethod
def get_user_by_api_token(token: str):
@@ -166,10 +200,16 @@ class Users_Controller:
return users_helper.get_user_api_key(key_id)
@staticmethod
- def add_user_api_key(name: str, user_id: str, superuser: bool = False,
- server_permissions_mask: Optional[str] = None,
- crafty_permissions_mask: Optional[str] = None):
- return users_helper.add_user_api_key(name, user_id, superuser, server_permissions_mask, crafty_permissions_mask)
+ def add_user_api_key(
+ name: str,
+ user_id: str,
+ superuser: bool = False,
+ server_permissions_mask: Optional[str] = None,
+ crafty_permissions_mask: Optional[str] = None,
+ ):
+ return users_helper.add_user_api_key(
+ name, user_id, superuser, server_permissions_mask, crafty_permissions_mask
+ )
@staticmethod
def delete_user_api_keys(user_id: str):
diff --git a/app/classes/minecraft/bedrock_ping.py b/app/classes/minecraft/bedrock_ping.py
index 1d2a8c99..c1d44a35 100644
--- a/app/classes/minecraft/bedrock_ping.py
+++ b/app/classes/minecraft/bedrock_ping.py
@@ -3,21 +3,22 @@ import socket
import time
import psutil
+
class BedrockPing:
- magic = b'\x00\xff\xff\x00\xfe\xfe\xfe\xfe\xfd\xfd\xfd\xfd\x12\x34\x56\x78'
- fields = { # (len, signed)
+ magic = b"\x00\xff\xff\x00\xfe\xfe\xfe\xfe\xfd\xfd\xfd\xfd\x12\x34\x56\x78"
+ fields = { # (len, signed)
"byte": (1, False),
"long": (8, True),
"ulong": (8, False),
"magic": (16, False),
"short": (2, True),
- "ushort": (2, False), #unsigned short
- "string": (2, False), #strlen is ushort
+ "ushort": (2, False), # unsigned short
+ "string": (2, False), # strlen is ushort
"bool": (1, False),
"address": (7, False),
- "uint24le": (3, False)
+ "uint24le": (3, False),
}
- byte_order = 'big'
+ byte_order = "big"
def __init__(self, bedrock_addr, bedrock_port, client_guid=0, timeout=5):
self.addr = bedrock_addr
@@ -36,51 +37,61 @@ class BedrockPing:
@staticmethod
def __slice(in_bytes, pattern):
ret = []
- bi = 0 # bytes index
- pi = 0 # pattern index
+ bi = 0 # bytes index
+ pi = 0 # pattern index
while bi < len(in_bytes):
try:
f = BedrockPing.fields[pattern[pi]]
except IndexError as index_error:
- raise IndexError("Ran out of pattern with additional bytes remaining") from index_error
+ raise IndexError(
+ "Ran out of pattern with additional bytes remaining"
+ ) from index_error
if pattern[pi] == "string":
- shl = f[0] # string header length
- sl = int.from_bytes(in_bytes[bi:bi+shl], BedrockPing.byte_order, signed=f[1]) # string length
- l = shl+sl
- ret.append(in_bytes[bi+shl:bi+shl+sl].decode('ascii'))
+ shl = f[0] # string header length
+ sl = int.from_bytes(
+ in_bytes[bi : bi + shl], BedrockPing.byte_order, signed=f[1]
+ ) # string length
+ l = shl + sl
+ ret.append(in_bytes[bi + shl : bi + shl + sl].decode("ascii"))
elif pattern[pi] == "magic":
- l = f[0] # length of field
- ret.append(in_bytes[bi:bi+l])
+ l = f[0] # length of field
+ ret.append(in_bytes[bi : bi + l])
else:
- l = f[0] # length of field
- ret.append(int.from_bytes(in_bytes[bi:bi+l], BedrockPing.byte_order, signed=f[1]))
- bi+=l
- pi+=1
+ l = f[0] # length of field
+ ret.append(
+ int.from_bytes(
+ in_bytes[bi : bi + l], BedrockPing.byte_order, signed=f[1]
+ )
+ )
+ bi += l
+ pi += 1
return ret
@staticmethod
def __get_time():
- #return time.time_ns() // 1000000
+ # return time.time_ns() // 1000000
return time.perf_counter_ns() // 1000000
def __sendping(self):
- pack_id = BedrockPing.__byter(0x01, 'byte')
- now = BedrockPing.__byter(BedrockPing.__get_time(), 'ulong')
+ pack_id = BedrockPing.__byter(0x01, "byte")
+ now = BedrockPing.__byter(BedrockPing.__get_time(), "ulong")
guid = self.guid_bytes
- d2s = pack_id+now+BedrockPing.magic+guid
- #print("S:", d2s)
+ d2s = pack_id + now + BedrockPing.magic + guid
+ # print("S:", d2s)
self.sock.sendto(d2s, (self.addr, self.port))
def __recvpong(self):
data = self.sock.recv(4096)
- if data[0] == 0x1c:
+ if data[0] == 0x1C:
ret = {}
- sliced = BedrockPing.__slice(data,["byte","ulong","ulong","magic","string"])
+ sliced = BedrockPing.__slice(
+ data, ["byte", "ulong", "ulong", "magic", "string"]
+ )
if sliced[3] != BedrockPing.magic:
raise ValueError(f"Incorrect magic received ({sliced[3]})")
ret["server_guid"] = sliced[2]
ret["server_string_raw"] = sliced[4]
- server_info = sliced[4].split(';')
+ server_info = sliced[4].split(";")
ret["server_edition"] = server_info[0]
ret["server_motd"] = (server_info[1], server_info[7])
ret["server_protocol_version"] = server_info[2]
@@ -103,5 +114,7 @@ class BedrockPing:
self.__sendping()
return self.__recvpong()
except ValueError as e:
- print(f"E: {e}, checking next packet. Retries remaining: {rtr}/{retries}")
+ print(
+ f"E: {e}, checking next packet. Retries remaining: {rtr}/{retries}"
+ )
rtr -= 1
diff --git a/app/classes/minecraft/mc_ping.py b/app/classes/minecraft/mc_ping.py
index 1e4e6b7e..b540fac5 100644
--- a/app/classes/minecraft/mc_ping.py
+++ b/app/classes/minecraft/mc_ping.py
@@ -13,24 +13,25 @@ from app.classes.shared.console import console
logger = logging.getLogger(__name__)
+
class Server:
def __init__(self, data):
- self.description = data.get('description')
+ self.description = data.get("description")
# print(self.description)
if isinstance(self.description, dict):
# cat server
if "translate" in self.description:
- self.description = self.description['translate']
+ self.description = self.description["translate"]
# waterfall / bungee
- elif 'extra' in self.description:
+ elif "extra" in self.description:
lines = []
description = self.description
- if 'extra' in description.keys():
- for e in description['extra']:
- #Conversion format code needed only for Java Version
+ if "extra" in description.keys():
+ for e in description["extra"]:
+ # Conversion format code needed only for Java Version
lines.append(get_code_format("reset"))
if "bold" in e.keys():
lines.append(get_code_format("bold"))
@@ -43,36 +44,36 @@ class Server:
if "obfuscated" in e.keys():
lines.append(get_code_format("obfuscated"))
if "color" in e.keys():
- lines.append(get_code_format(e['color']))
- #Then append the text
+ lines.append(get_code_format(e["color"]))
+ # Then append the text
if "text" in e.keys():
- if e['text'] == '\n':
+ if e["text"] == "\n":
lines.append("ยงยง")
else:
- lines.append(e['text'])
+ lines.append(e["text"])
total_text = " ".join(lines)
self.description = total_text
# normal MC
else:
- self.description = self.description['text']
+ self.description = self.description["text"]
- self.icon = base64.b64decode(data.get('favicon', '')[22:])
+ self.icon = base64.b64decode(data.get("favicon", "")[22:])
try:
- self.players = Players(data['players']).report()
+ self.players = Players(data["players"]).report()
except KeyError:
logger.error("Error geting player information key error")
self.players = []
- self.version = data['version']['name']
- self.protocol = data['version']['protocol']
+ self.version = data["version"]["name"]
+ self.protocol = data["version"]["protocol"]
class Players(list):
def __init__(self, data):
- super().__init__(Player(x) for x in data.get('sample', []))
- self.max = data['max']
- self.online = data['online']
+ super().__init__(Player(x) for x in data.get("sample", []))
+ self.max = data["max"]
+ self.online = data["online"]
def report(self):
players = []
@@ -80,35 +81,34 @@ class Players(list):
for x in self:
players.append(str(x))
- r_data = {
- 'online': self.online,
- 'max': self.max,
- 'players': players
- }
+ r_data = {"online": self.online, "max": self.max, "players": players}
return json.dumps(r_data)
class Player:
def __init__(self, data):
- self.id = data['id']
- self.name = data['name']
+ self.id = data["id"]
+ self.name = data["name"]
def __str__(self):
return self.name
+
def get_code_format(format_name):
root_dir = os.path.abspath(os.path.curdir)
- format_file = os.path.join(root_dir, 'app', 'config', 'motd_format.json')
+ format_file = os.path.join(root_dir, "app", "config", "motd_format.json")
try:
- with open(format_file, "r", encoding='utf-8') as f:
+ with open(format_file, "r", encoding="utf-8") as f:
data = json.load(f)
if format_name in data.keys():
return data.get(format_name)
else:
logger.error(f"Format MOTD Error: format name {format_name} does not exist")
- console.error(f"Format MOTD Error: format name {format_name} does not exist")
+ console.error(
+ f"Format MOTD Error: format name {format_name} does not exist"
+ )
return ""
except Exception as e:
@@ -128,10 +128,10 @@ def ping(ip, port):
if not k:
return 0
k = k[0]
- i |= (k & 0x7f) << (j * 7)
+ i |= (k & 0x7F) << (j * 7)
j += 1
if j > 5:
- raise ValueError('var_int too big')
+ raise ValueError("var_int too big")
if not k & 0x80:
return i
@@ -143,15 +143,15 @@ def ping(ip, port):
return False
try:
- host = ip.encode('utf-8')
- data = b'' # wiki.vg/Server_List_Ping
- data += b'\x00' # packet ID
- data += b'\x04' # protocol variant
- data += struct.pack('>b', len(host)) + host
- data += struct.pack('>H', port)
- data += b'\x01' # next state
- data = struct.pack('>b', len(data)) + data
- sock.sendall(data + b'\x01\x00') # handshake + status ping
+ host = ip.encode("utf-8")
+ data = b"" # wiki.vg/Server_List_Ping
+ data += b"\x00" # packet ID
+ data += b"\x04" # protocol variant
+ data += struct.pack(">b", len(host)) + host
+ data += struct.pack(">H", port)
+ data += b"\x01" # next state
+ data = struct.pack(">b", len(data)) + data
+ sock.sendall(data + b"\x01\x00") # handshake + status ping
length = read_var_int() # full packet length
if length < 10:
if length < 0:
@@ -161,7 +161,7 @@ def ping(ip, port):
sock.recv(1) # packet type, 0 for pings
length = read_var_int() # string length
- data = b''
+ data = b""
while len(data) != length:
chunk = sock.recv(length - len(data))
if not chunk:
@@ -176,12 +176,13 @@ def ping(ip, port):
finally:
sock.close()
+
# For the rest of requests see wiki.vg/Protocol
def ping_bedrock(ip, port):
rd = random.Random()
try:
- #pylint: disable=consider-using-f-string
- rd.seed(''.join(re.findall('..', '%012x' % uuid.getnode())))
+ # pylint: disable=consider-using-f-string
+ rd.seed("".join(re.findall("..", "%012x" % uuid.getnode())))
client_guid = uuid.UUID(int=rd.getrandbits(32)).int
except:
client_guid = 0
diff --git a/app/classes/minecraft/server_props.py b/app/classes/minecraft/server_props.py
index 027e11ac..c70652de 100644
--- a/app/classes/minecraft/server_props.py
+++ b/app/classes/minecraft/server_props.py
@@ -1,30 +1,30 @@
import pprint
import os
-class ServerProps:
+class ServerProps:
def __init__(self, filepath):
self.filepath = filepath
self.props = self._parse()
def _parse(self):
"""Loads and parses the file specified in self.filepath"""
- with open(self.filepath, encoding='utf-8') as fp:
+ with open(self.filepath, encoding="utf-8") as fp:
line = fp.readline()
d = {}
if os.path.exists(".header"):
os.remove(".header")
while line:
- if '#' != line[0]:
+ if "#" != line[0]:
s = line
- s1 = s[:s.find('=')]
- if '\n' in s:
- s2 = s[s.find('=')+1:s.find('\n')]
+ s1 = s[: s.find("=")]
+ if "\n" in s:
+ s2 = s[s.find("=") + 1 : s.find("\n")]
else:
- s2 = s[s.find('=')+1:]
+ s2 = s[s.find("=") + 1 :]
d[s1] = s2
else:
- with open(".header", "a+", encoding='utf-8') as h:
+ with open(".header", "a+", encoding="utf-8") as h:
h.write(line)
line = fp.readline()
return d
@@ -47,9 +47,9 @@ class ServerProps:
def save(self):
"""Writes to the new file"""
- with open(self.filepath, "a+", encoding='utf-8') as f:
+ with open(self.filepath, "a+", encoding="utf-8") as f:
f.truncate(0)
- with open(".header", encoding='utf-8') as header:
+ with open(".header", encoding="utf-8") as header:
line = header.readline()
while line:
f.write(line)
diff --git a/app/classes/minecraft/serverjars.py b/app/classes/minecraft/serverjars.py
index 46e040a5..380dbd38 100644
--- a/app/classes/minecraft/serverjars.py
+++ b/app/classes/minecraft/serverjars.py
@@ -18,8 +18,8 @@ try:
except ModuleNotFoundError as err:
helper.auto_installer_fix(err)
-class ServerJars:
+class ServerJars:
def __init__(self):
self.base_url = "https://serverjars.com"
@@ -41,8 +41,8 @@ class ServerJars:
logger.error(f"Unable to parse serverjar.com api result due to error: {e}")
return {}
- api_result = api_data.get('status')
- api_response = api_data.get('response', {})
+ api_result = api_data.get("status")
+ api_response = api_data.get("response", {})
if api_result != "success":
logger.error(f"Api returned a failed status: {api_result}")
@@ -55,7 +55,7 @@ class ServerJars:
cache_file = helper.serverjar_cache
cache = {}
try:
- with open(cache_file, "r", encoding='utf-8') as f:
+ with open(cache_file, "r", encoding="utf-8") as f:
cache = json.load(f)
except Exception as e:
@@ -65,7 +65,7 @@ class ServerJars:
def get_serverjar_data(self):
data = self._read_cache()
- return data.get('servers')
+ return data.get("servers")
def get_serverjar_data_sorted(self):
data = self.get_serverjar_data()
@@ -80,10 +80,10 @@ class ServerJars:
try:
return int(x)
except ValueError:
- temp = x.split('-')
+ temp = x.split("-")
return to_int(temp[0]) + str_to_int(temp[1]) / 100000
- sort_key_fn = lambda x: [to_int(y) for y in x.split('.')]
+ sort_key_fn = lambda x: [to_int(y) for y in x.split(".")]
for key in data.keys():
data[key] = sorted(data[key], key=sort_key_fn)
@@ -125,10 +125,7 @@ class ServerJars:
if cache_old:
logger.info("Cache file is over 1 day old, refreshing")
now = datetime.now()
- data = {
- 'last_refreshed': now.strftime("%m/%d/%Y, %H:%M:%S"),
- 'servers': {}
- }
+ data = {"last_refreshed": now.strftime("%m/%d/%Y, %H:%M:%S"), "servers": {}}
jar_types = self._get_server_type_list()
@@ -141,51 +138,51 @@ class ServerJars:
versions = self._get_jar_details(s)
# add these versions (a list) to the dict with a key of the server type
- data['servers'].update({
- s: versions
- })
+ data["servers"].update({s: versions})
# save our cache
try:
- with open(cache_file, "w", encoding='utf-8') as f:
+ with open(cache_file, "w", encoding="utf-8") as f:
f.write(json.dumps(data, indent=4))
logger.info("Cache file refreshed")
except Exception as e:
logger.error(f"Unable to update serverjars.com cache file: {e}")
- def _get_jar_details(self, jar_type='servers'):
- url = f'/api/fetchAll/{jar_type}'
+ def _get_jar_details(self, jar_type="servers"):
+ url = f"/api/fetchAll/{jar_type}"
response = self._get_api_result(url)
temp = []
for v in response:
- temp.append(v.get('version'))
- time.sleep(.5)
+ temp.append(v.get("version"))
+ time.sleep(0.5)
return temp
def _get_server_type_list(self):
- url = '/api/fetchTypes/'
+ url = "/api/fetchTypes/"
response = self._get_api_result(url)
return response
def download_jar(self, server, version, path, server_id):
- update_thread = threading.Thread(target=self.a_download_jar, daemon=True, args=(server, version, path, server_id))
+ update_thread = threading.Thread(
+ target=self.a_download_jar,
+ daemon=True,
+ args=(server, version, path, server_id),
+ )
update_thread.start()
def a_download_jar(self, server, version, path, server_id):
- #delaying download for server register to finish
+ # delaying download for server register to finish
time.sleep(3)
fetch_url = f"{self.base_url}/api/fetchJar/{server}/{version}"
server_users = server_permissions.get_server_user_list(server_id)
-
- #We need to make sure the server is registered before we submit a db update for it's stats.
+ # We need to make sure the server is registered before we submit a db update for it's stats.
while True:
try:
Servers_Controller.set_download(server_id)
for user in server_users:
- websocket_helper.broadcast_user(user, 'send_start_reload', {
- })
+ websocket_helper.broadcast_user(user, "send_start_reload", {})
break
except:
@@ -194,25 +191,27 @@ class ServerJars:
# open a file stream
with requests.get(fetch_url, timeout=2, stream=True) as r:
try:
- with open(path, 'wb') as output:
+ with open(path, "wb") as output:
shutil.copyfileobj(r.raw, output)
Servers_Controller.finish_download(server_id)
for user in server_users:
- websocket_helper.broadcast_user(user, 'notification', "Executable download finished")
+ websocket_helper.broadcast_user(
+ user, "notification", "Executable download finished"
+ )
time.sleep(3)
- websocket_helper.broadcast_user(user, 'send_start_reload', {
- })
+ websocket_helper.broadcast_user(user, "send_start_reload", {})
return True
except Exception as e:
logger.error(f"Unable to save jar to {path} due to error:{e}")
Servers_Controller.finish_download(server_id)
server_users = server_permissions.get_server_user_list(server_id)
for user in server_users:
- websocket_helper.broadcast_user(user, 'notification', "Executable download finished")
+ websocket_helper.broadcast_user(
+ user, "notification", "Executable download finished"
+ )
time.sleep(3)
- websocket_helper.broadcast_user(user, 'send_start_reload', {
- })
+ websocket_helper.broadcast_user(user, "send_start_reload", {})
return False
diff --git a/app/classes/minecraft/stats.py b/app/classes/minecraft/stats.py
index 456213ce..3fcc780e 100644
--- a/app/classes/minecraft/stats.py
+++ b/app/classes/minecraft/stats.py
@@ -11,8 +11,8 @@ from app.classes.shared.helpers import helper
logger = logging.getLogger(__name__)
-class Stats:
+class Stats:
def __init__(self, controller):
self.controller = controller
@@ -24,30 +24,26 @@ class Stats:
except NotImplementedError:
cpu_freq = psutil._common.scpufreq(current=0, min=0, max=0)
node_stats = {
- 'boot_time': str(boot_time),
- 'cpu_usage': psutil.cpu_percent(interval=0.5) / psutil.cpu_count(),
- 'cpu_count': psutil.cpu_count(),
- 'cpu_cur_freq': round(cpu_freq[0], 2),
- 'cpu_max_freq': cpu_freq[2],
- 'mem_percent': psutil.virtual_memory()[2],
- 'mem_usage': helper.human_readable_file_size(psutil.virtual_memory()[3]),
- 'mem_total': helper.human_readable_file_size(psutil.virtual_memory()[0]),
- 'disk_data': self._all_disk_usage()
+ "boot_time": str(boot_time),
+ "cpu_usage": psutil.cpu_percent(interval=0.5) / psutil.cpu_count(),
+ "cpu_count": psutil.cpu_count(),
+ "cpu_cur_freq": round(cpu_freq[0], 2),
+ "cpu_max_freq": cpu_freq[2],
+ "mem_percent": psutil.virtual_memory()[2],
+ "mem_usage": helper.human_readable_file_size(psutil.virtual_memory()[3]),
+ "mem_total": helper.human_readable_file_size(psutil.virtual_memory()[0]),
+ "disk_data": self._all_disk_usage(),
}
- #server_stats = self.get_servers_stats()
- #data['servers'] = server_stats
- data['node_stats'] = node_stats
+ # server_stats = self.get_servers_stats()
+ # data['servers'] = server_stats
+ data["node_stats"] = node_stats
return data
@staticmethod
def _get_process_stats(process):
if process is None:
- process_stats = {
- 'cpu_usage': 0,
- 'memory_usage': 0,
- 'mem_percentage': 0
- }
+ process_stats = {"cpu_usage": 0, "memory_usage": 0, "mem_percentage": 0}
return process_stats
else:
process_pid = process.pid
@@ -63,19 +59,21 @@ class Stats:
# this is a faster way of getting data for a process
with p.oneshot():
process_stats = {
- 'cpu_usage': real_cpu,
- 'memory_usage': helper.human_readable_file_size(p.memory_info()[0]),
- 'mem_percentage': round(p.memory_percent(), 0)
+ "cpu_usage": real_cpu,
+ "memory_usage": helper.human_readable_file_size(p.memory_info()[0]),
+ "mem_percentage": round(p.memory_percent(), 0),
}
return process_stats
except Exception as e:
- logger.error(f"Unable to get process details for pid: {process_pid} due to error: {e}")
+ logger.error(
+ f"Unable to get process details for pid: {process_pid} due to error: {e}"
+ )
# Dummy Data
process_stats = {
- 'cpu_usage': 0,
- 'memory_usage': 0,
+ "cpu_usage": 0,
+ "memory_usage": 0,
}
return process_stats
@@ -87,7 +85,7 @@ class Stats:
for part in psutil.disk_partitions(all=False):
if helper.is_os_windows():
- if 'cdrom' in part.opts or part.fstype == '':
+ if "cdrom" in part.opts or part.fstype == "":
# skip cd-rom drives with no disk in it; they may raise
# ENOENT, pop-up a Windows GUI error for a non-ready
# partition or just hang.
@@ -95,13 +93,13 @@ class Stats:
usage = psutil.disk_usage(part.mountpoint)
disk_data.append(
{
- 'device': part.device,
- 'total': helper.human_readable_file_size(usage.total),
- 'used': helper.human_readable_file_size(usage.used),
- 'free': helper.human_readable_file_size(usage.free),
- 'percent_used': int(usage.percent),
- 'fs': part.fstype,
- 'mount': part.mountpoint
+ "device": part.device,
+ "total": helper.human_readable_file_size(usage.total),
+ "used": helper.human_readable_file_size(usage.used),
+ "free": helper.human_readable_file_size(usage.free),
+ "percent_used": int(usage.percent),
+ "fs": part.fstype,
+ "mount": part.mountpoint,
}
)
@@ -128,22 +126,20 @@ class Stats:
# server_settings = server.get('server_settings', {})
# server_data = server.get('server_data_obj', {})
-
# TODO: search server properties file for possible override of 127.0.0.1
- internal_ip = server['server_ip']
- server_port = server['server_port']
+ internal_ip = server["server_ip"]
+ server_port = server["server_port"]
logger.debug("Pinging {internal_ip} on port {server_port}")
- if servers_helper.get_server_type_by_id(server_id) != 'minecraft-bedrock':
+ if servers_helper.get_server_type_by_id(server_id) != "minecraft-bedrock":
int_mc_ping = ping(internal_ip, int(server_port))
-
ping_data = {}
# if we got a good ping return, let's parse it
if int_mc_ping:
ping_data = Stats.parse_server_ping(int_mc_ping)
- return ping_data['players']
+ return ping_data["players"]
return []
@staticmethod
@@ -156,21 +152,20 @@ class Stats:
except Exception as e:
logger.info(f"Unable to read json from ping_obj: {e}")
-
try:
server_icon = base64.encodebytes(ping_obj.icon)
- server_icon = server_icon.decode('utf-8')
- except Exception as e:
+ server_icon = server_icon.decode("utf-8")
+ except Exception as e:
server_icon = False
logger.info(f"Unable to read the server icon : {e}")
ping_data = {
- 'online': online_stats.get("online", 0),
- 'max': online_stats.get('max', 0),
- 'players': online_stats.get('players', 0),
- 'server_description': ping_obj.description,
- 'server_version': ping_obj.version,
- 'server_icon': server_icon
+ "online": online_stats.get("online", 0),
+ "max": online_stats.get("max", 0),
+ "players": online_stats.get("players", 0),
+ "server_description": ping_obj.description,
+ "server_version": ping_obj.version,
+ "server_icon": server_icon,
}
return ping_data
@@ -179,59 +174,59 @@ class Stats:
def parse_server_RakNet_ping(ping_obj: object):
try:
- server_icon = base64.encodebytes(ping_obj['icon'])
- except Exception as e:
+ server_icon = base64.encodebytes(ping_obj["icon"])
+ except Exception as e:
server_icon = False
logger.info(f"Unable to read the server icon : {e}")
ping_data = {
- 'online': ping_obj['server_player_count'],
- 'max': ping_obj['server_player_max'],
- 'players': [],
- 'server_description': ping_obj['server_edition'],
- 'server_version': ping_obj['server_version_name'],
- 'server_icon': server_icon
+ "online": ping_obj["server_player_count"],
+ "max": ping_obj["server_player_max"],
+ "players": [],
+ "server_description": ping_obj["server_edition"],
+ "server_version": ping_obj["server_version_name"],
+ "server_icon": server_icon,
}
-
return ping_data
-
def record_stats(self):
stats_to_send = self.get_node_stats()
- node_stats = stats_to_send.get('node_stats')
+ node_stats = stats_to_send.get("node_stats")
- Host_Stats.insert({
- Host_Stats.boot_time: node_stats.get('boot_time', "Unknown"),
- Host_Stats.cpu_usage: round(node_stats.get('cpu_usage', 0), 2),
- Host_Stats.cpu_cores: node_stats.get('cpu_count', 0),
- Host_Stats.cpu_cur_freq: node_stats.get('cpu_cur_freq', 0),
- Host_Stats.cpu_max_freq: node_stats.get('cpu_max_freq', 0),
- Host_Stats.mem_usage: node_stats.get('mem_usage', "0 MB"),
- Host_Stats.mem_percent: node_stats.get('mem_percent', 0),
- Host_Stats.mem_total: node_stats.get('mem_total', "0 MB"),
- Host_Stats.disk_json: node_stats.get('disk_data', '{}')
- }).execute()
+ Host_Stats.insert(
+ {
+ Host_Stats.boot_time: node_stats.get("boot_time", "Unknown"),
+ Host_Stats.cpu_usage: round(node_stats.get("cpu_usage", 0), 2),
+ Host_Stats.cpu_cores: node_stats.get("cpu_count", 0),
+ Host_Stats.cpu_cur_freq: node_stats.get("cpu_cur_freq", 0),
+ Host_Stats.cpu_max_freq: node_stats.get("cpu_max_freq", 0),
+ Host_Stats.mem_usage: node_stats.get("mem_usage", "0 MB"),
+ Host_Stats.mem_percent: node_stats.get("mem_percent", 0),
+ Host_Stats.mem_total: node_stats.get("mem_total", "0 MB"),
+ Host_Stats.disk_json: node_stats.get("disk_data", "{}"),
+ }
+ ).execute()
-# server_stats = stats_to_send.get('servers')#
-#
-# for server in server_stats:
-# Server_Stats.insert({
-# Server_Stats.server_id: server.get('id', 0),
-# Server_Stats.started: server.get('started', ""),
-# Server_Stats.running: server.get('running', False),
-# Server_Stats.cpu: server.get('cpu', 0),
-# Server_Stats.mem: server.get('mem', 0),
-# Server_Stats.mem_percent: server.get('mem_percent', 0),
-# Server_Stats.world_name: server.get('world_name', ""),
-# Server_Stats.world_size: server.get('world_size', ""),
-# Server_Stats.server_port: server.get('server_port', ""),
-# Server_Stats.int_ping_results: server.get('int_ping_results', False),
-# Server_Stats.online: server.get("online", False),
-# Server_Stats.max: server.get("max", False),
-# Server_Stats.players: server.get("players", False),
-# Server_Stats.desc: server.get("desc", False),
-# Server_Stats.version: server.get("version", False)
-# }).execute()
+ # server_stats = stats_to_send.get('servers')#
+ #
+ # for server in server_stats:
+ # Server_Stats.insert({
+ # Server_Stats.server_id: server.get('id', 0),
+ # Server_Stats.started: server.get('started', ""),
+ # Server_Stats.running: server.get('running', False),
+ # Server_Stats.cpu: server.get('cpu', 0),
+ # Server_Stats.mem: server.get('mem', 0),
+ # Server_Stats.mem_percent: server.get('mem_percent', 0),
+ # Server_Stats.world_name: server.get('world_name', ""),
+ # Server_Stats.world_size: server.get('world_size', ""),
+ # Server_Stats.server_port: server.get('server_port', ""),
+ # Server_Stats.int_ping_results: server.get('int_ping_results', False),
+ # Server_Stats.online: server.get("online", False),
+ # Server_Stats.max: server.get("max", False),
+ # Server_Stats.players: server.get("players", False),
+ # Server_Stats.desc: server.get("desc", False),
+ # Server_Stats.version: server.get("version", False)
+ # }).execute()
# delete old data
max_age = helper.get_setting("history_max_age")
@@ -239,4 +234,6 @@ class Stats:
last_week = now.day - max_age
Host_Stats.delete().where(Host_Stats.time < last_week).execute()
+
+
# Server_Stats.delete().where(Server_Stats.created < last_week).execute()
diff --git a/app/classes/models/crafty_permissions.py b/app/classes/models/crafty_permissions.py
index 13a655e5..46b02315 100644
--- a/app/classes/models/crafty_permissions.py
+++ b/app/classes/models/crafty_permissions.py
@@ -5,25 +5,32 @@ from app.classes.shared.permission_helper import permission_helper
from app.classes.models.users import Users, ApiKeys
try:
- from peewee import SqliteDatabase, Model, ForeignKeyField, CharField, IntegerField, DoesNotExist
+ from peewee import (
+ SqliteDatabase,
+ Model,
+ ForeignKeyField,
+ CharField,
+ IntegerField,
+ DoesNotExist,
+ )
from enum import Enum
except ModuleNotFoundError as e:
helper.auto_installer_fix(e)
logger = logging.getLogger(__name__)
-peewee_logger = logging.getLogger('peewee')
+peewee_logger = logging.getLogger("peewee")
peewee_logger.setLevel(logging.INFO)
-database = SqliteDatabase(helper.db_path, pragmas = {
- 'journal_mode': 'wal',
- 'cache_size': -1024 * 10})
+database = SqliteDatabase(
+ helper.db_path, pragmas={"journal_mode": "wal", "cache_size": -1024 * 10}
+)
-#************************************************************************************************
+# ************************************************************************************************
# User_Crafty Class
-#************************************************************************************************
+# ************************************************************************************************
class User_Crafty(Model):
- user_id = ForeignKeyField(Users, backref='users_crafty')
+ user_id = ForeignKeyField(Users, backref="users_crafty")
permissions = CharField(default="00000000")
limit_server_creation = IntegerField(default=-1)
limit_user_creation = IntegerField(default=0)
@@ -33,22 +40,24 @@ class User_Crafty(Model):
created_role = IntegerField(default=0)
class Meta:
- table_name = 'user_crafty'
+ table_name = "user_crafty"
database = database
-#************************************************************************************************
+
+# ************************************************************************************************
# Crafty Permissions Class
-#************************************************************************************************
+# ************************************************************************************************
class Enum_Permissions_Crafty(Enum):
Server_Creation = 0
User_Config = 1
Roles_Config = 2
+
class Permissions_Crafty:
- #************************************************************************************************
+ # ************************************************************************************************
# Crafty Permissions Methods
- #************************************************************************************************
+ # ************************************************************************************************
@staticmethod
def get_permissions_list():
permissions_list = []
@@ -67,15 +76,17 @@ class Permissions_Crafty:
@staticmethod
def has_permission(permission_mask, permission_tested: Enum_Permissions_Crafty):
result = False
- if permission_mask[permission_tested.value] == '1':
+ if permission_mask[permission_tested.value] == "1":
result = True
return result
@staticmethod
- def set_permission(permission_mask, permission_tested: Enum_Permissions_Crafty, value):
+ def set_permission(
+ permission_mask, permission_tested: Enum_Permissions_Crafty, value
+ ):
l = list(permission_mask)
l[permission_tested.value] = str(value)
- permission_mask = ''.join(l)
+ permission_mask = "".join(l)
return permission_mask
@staticmethod
@@ -84,7 +95,7 @@ class Permissions_Crafty:
@staticmethod
def get_crafty_permissions_mask(user_id):
- permissions_mask = ''
+ permissions_mask = ""
user_crafty = crafty_permissions.get_User_Crafty(user_id)
permissions_mask = user_crafty.permissions
return permissions_mask
@@ -108,49 +119,65 @@ class Permissions_Crafty:
}
return quantity_list
- #************************************************************************************************
+ # ************************************************************************************************
# User_Crafty Methods
- #************************************************************************************************
+ # ************************************************************************************************
@staticmethod
def get_User_Crafty(user_id):
try:
- user_crafty = User_Crafty.select().where(User_Crafty.user_id == user_id).get()
+ user_crafty = (
+ User_Crafty.select().where(User_Crafty.user_id == user_id).get()
+ )
except DoesNotExist:
- user_crafty = User_Crafty.insert({
- User_Crafty.user_id: user_id,
- User_Crafty.permissions: "000",
- User_Crafty.limit_server_creation: 0,
- User_Crafty.limit_user_creation: 0,
- User_Crafty.limit_role_creation: 0,
- User_Crafty.created_server: 0,
- User_Crafty.created_user: 0,
- User_Crafty.created_role: 0,
- }).execute()
+ user_crafty = User_Crafty.insert(
+ {
+ User_Crafty.user_id: user_id,
+ User_Crafty.permissions: "000",
+ User_Crafty.limit_server_creation: 0,
+ User_Crafty.limit_user_creation: 0,
+ User_Crafty.limit_role_creation: 0,
+ User_Crafty.created_server: 0,
+ User_Crafty.created_user: 0,
+ User_Crafty.created_role: 0,
+ }
+ ).execute()
user_crafty = crafty_permissions.get_User_Crafty(user_id)
return user_crafty
@staticmethod
def add_user_crafty(user_id, uc_permissions):
- user_crafty = User_Crafty.insert({User_Crafty.user_id: user_id, User_Crafty.permissions: uc_permissions}).execute()
+ user_crafty = User_Crafty.insert(
+ {User_Crafty.user_id: user_id, User_Crafty.permissions: uc_permissions}
+ ).execute()
return user_crafty
@staticmethod
- def add_or_update_user(user_id, permissions_mask, limit_server_creation, limit_user_creation, limit_role_creation):
+ def add_or_update_user(
+ user_id,
+ permissions_mask,
+ limit_server_creation,
+ limit_user_creation,
+ limit_role_creation,
+ ):
try:
- user_crafty = User_Crafty.select().where(User_Crafty.user_id == user_id).get()
+ user_crafty = (
+ User_Crafty.select().where(User_Crafty.user_id == user_id).get()
+ )
user_crafty.permissions = permissions_mask
user_crafty.limit_server_creation = limit_server_creation
user_crafty.limit_user_creation = limit_user_creation
user_crafty.limit_role_creation = limit_role_creation
User_Crafty.save(user_crafty)
except:
- User_Crafty.insert({
- User_Crafty.user_id: user_id,
- User_Crafty.permissions: permissions_mask,
- User_Crafty.limit_server_creation: limit_server_creation,
- User_Crafty.limit_user_creation: limit_user_creation,
- User_Crafty.limit_role_creation: limit_role_creation
- }).execute()
+ User_Crafty.insert(
+ {
+ User_Crafty.user_id: user_id,
+ User_Crafty.permissions: permissions_mask,
+ User_Crafty.limit_server_creation: limit_server_creation,
+ User_Crafty.limit_user_creation: limit_user_creation,
+ User_Crafty.limit_role_creation: limit_role_creation,
+ }
+ ).execute()
@staticmethod
def get_created_quantity_list(user_id):
@@ -173,7 +200,10 @@ class Permissions_Crafty:
can = crafty_permissions.has_permission(user_crafty.permissions, permission)
limit_list = crafty_permissions.get_permission_quantity_list(user_id)
quantity_list = crafty_permissions.get_created_quantity_list(user_id)
- return can and ((quantity_list[permission.name] < limit_list[permission.name]) or limit_list[permission.name] == -1 )
+ return can and (
+ (quantity_list[permission.name] < limit_list[permission.name])
+ or limit_list[permission.name] == -1
+ )
@staticmethod
def add_server_creation(user_id):
@@ -188,12 +218,15 @@ class Permissions_Crafty:
if user.superuser and key.superuser:
return crafty_permissions.get_permissions_list()
else:
- user_permissions_mask = crafty_permissions.get_crafty_permissions_mask(user.user_id)
+ user_permissions_mask = crafty_permissions.get_crafty_permissions_mask(
+ user.user_id
+ )
key_permissions_mask: str = key.crafty_permissions
- permissions_mask = permission_helper.combine_masks(user_permissions_mask, key_permissions_mask)
+ permissions_mask = permission_helper.combine_masks(
+ user_permissions_mask, key_permissions_mask
+ )
permissions_list = crafty_permissions.get_permissions(permissions_mask)
return permissions_list
-
crafty_permissions = Permissions_Crafty()
diff --git a/app/classes/models/management.py b/app/classes/models/management.py
index 927f97b4..2e55bd49 100644
--- a/app/classes/models/management.py
+++ b/app/classes/models/management.py
@@ -9,38 +9,51 @@ from app.classes.shared.main_models import db_helper
from app.classes.web.websocket_helper import websocket_helper
try:
- from peewee import SqliteDatabase, Model, ForeignKeyField, CharField, IntegerField, DateTimeField, FloatField, TextField, AutoField, BooleanField
+ from peewee import (
+ SqliteDatabase,
+ Model,
+ ForeignKeyField,
+ CharField,
+ IntegerField,
+ DateTimeField,
+ FloatField,
+ TextField,
+ AutoField,
+ BooleanField,
+ )
from playhouse.shortcuts import model_to_dict
except ModuleNotFoundError as e:
helper.auto_installer_fix(e)
logger = logging.getLogger(__name__)
-peewee_logger = logging.getLogger('peewee')
+peewee_logger = logging.getLogger("peewee")
peewee_logger.setLevel(logging.INFO)
-database = SqliteDatabase(helper.db_path, pragmas = {
- 'journal_mode': 'wal',
- 'cache_size': -1024 * 10})
+database = SqliteDatabase(
+ helper.db_path, pragmas={"journal_mode": "wal", "cache_size": -1024 * 10}
+)
-#************************************************************************************************
+# ************************************************************************************************
# Audit_Log Class
-#************************************************************************************************
+# ************************************************************************************************
class Audit_Log(Model):
audit_id = AutoField()
created = DateTimeField(default=datetime.datetime.now)
user_name = CharField(default="")
user_id = IntegerField(default=0, index=True)
- source_ip = CharField(default='127.0.0.1')
- server_id = IntegerField(default=None, index=True) # When auditing global events, use server ID 0
- log_msg = TextField(default='')
+ source_ip = CharField(default="127.0.0.1")
+ server_id = IntegerField(
+ default=None, index=True
+ ) # When auditing global events, use server ID 0
+ log_msg = TextField(default="")
class Meta:
database = database
-#************************************************************************************************
+# ************************************************************************************************
# Host_Stats Class
-#************************************************************************************************
+# ************************************************************************************************
class Host_Stats(Model):
time = DateTimeField(default=datetime.datetime.now, index=True)
boot_time = CharField(default="")
@@ -58,16 +71,16 @@ class Host_Stats(Model):
database = database
-#************************************************************************************************
+# ************************************************************************************************
# Commands Class
-#************************************************************************************************
+# ************************************************************************************************
class Commands(Model):
command_id = AutoField()
created = DateTimeField(default=datetime.datetime.now)
- server_id = ForeignKeyField(Servers, backref='server', index=True)
- user = ForeignKeyField(Users, backref='user', index=True)
- source_ip = CharField(default='127.0.0.1')
- command = CharField(default='')
+ server_id = ForeignKeyField(Servers, backref="server", index=True)
+ user = ForeignKeyField(Users, backref="user", index=True)
+ source_ip = CharField(default="127.0.0.1")
+ command = CharField(default="")
executed = BooleanField(default=False)
class Meta:
@@ -75,9 +88,9 @@ class Commands(Model):
database = database
-#************************************************************************************************
+# ************************************************************************************************
# Webhooks Class
-#************************************************************************************************
+# ************************************************************************************************
class Webhooks(Model):
id = AutoField()
name = CharField(max_length=64, unique=True, index=True)
@@ -91,12 +104,12 @@ class Webhooks(Model):
database = database
-#************************************************************************************************
+# ************************************************************************************************
# Schedules Class
-#************************************************************************************************
+# ************************************************************************************************
class Schedules(Model):
schedule_id = IntegerField(unique=True, primary_key=True)
- server_id = ForeignKeyField(Servers, backref='schedule_server')
+ server_id = ForeignKeyField(Servers, backref="schedule_server")
enabled = BooleanField()
action = CharField()
interval = IntegerField()
@@ -110,44 +123,48 @@ class Schedules(Model):
delay = IntegerField(default=0)
class Meta:
- table_name = 'schedules'
+ table_name = "schedules"
database = database
-#************************************************************************************************
+# ************************************************************************************************
# Backups Class
-#************************************************************************************************
+# ************************************************************************************************
class Backups(Model):
excluded_dirs = CharField(null=True)
max_backups = IntegerField()
- server_id = ForeignKeyField(Servers, backref='backups_server')
+ server_id = ForeignKeyField(Servers, backref="backups_server")
compress = BooleanField(default=False)
+
class Meta:
- table_name = 'backups'
+ table_name = "backups"
database = database
+
class helpers_management:
- #************************************************************************************************
+ # ************************************************************************************************
# Host_Stats Methods
- #************************************************************************************************
+ # ************************************************************************************************
@staticmethod
def get_latest_hosts_stats():
- #pylint: disable=no-member
+ # pylint: disable=no-member
query = Host_Stats.select().order_by(Host_Stats.id.desc()).get()
return model_to_dict(query)
- #************************************************************************************************
+ # ************************************************************************************************
# Commands Methods
- #************************************************************************************************
+ # ************************************************************************************************
@staticmethod
def add_command(server_id, user_id, remote_ip, command):
- Commands.insert({
- Commands.server_id: server_id,
- Commands.user: user_id,
- Commands.source_ip: remote_ip,
- Commands.command: command
- }).execute()
+ Commands.insert(
+ {
+ Commands.server_id: server_id,
+ Commands.user: user_id,
+ Commands.source_ip: remote_ip,
+ Commands.command: command,
+ }
+ ).execute()
@staticmethod
def get_unactioned_commands():
@@ -158,13 +175,13 @@ class helpers_management:
def mark_command_complete(command_id=None):
if command_id is not None:
logger.debug(f"Marking Command {command_id} completed")
- Commands.update({
- Commands.executed: True
- }).where(Commands.command_id == command_id).execute()
+ Commands.update({Commands.executed: True}).where(
+ Commands.command_id == command_id
+ ).execute()
- #************************************************************************************************
+ # ************************************************************************************************
# Audit_Log Methods
- #************************************************************************************************
+ # ************************************************************************************************
@staticmethod
def get_actity_log():
q = Audit_Log.select()
@@ -179,22 +196,24 @@ class helpers_management:
server_users = server_permissions.get_server_user_list(server_id)
for user in server_users:
- websocket_helper.broadcast_user(user,'notification', audit_msg)
+ websocket_helper.broadcast_user(user, "notification", audit_msg)
- Audit_Log.insert({
- Audit_Log.user_name: user_data['username'],
- Audit_Log.user_id: user_id,
- Audit_Log.server_id: server_id,
- Audit_Log.log_msg: audit_msg,
- Audit_Log.source_ip: source_ip
- }).execute()
- #deletes records when they're more than 100
+ Audit_Log.insert(
+ {
+ Audit_Log.user_name: user_data["username"],
+ Audit_Log.user_id: user_id,
+ Audit_Log.server_id: server_id,
+ Audit_Log.log_msg: audit_msg,
+ Audit_Log.source_ip: source_ip,
+ }
+ ).execute()
+ # deletes records when they're more than 100
ordered = Audit_Log.select().order_by(+Audit_Log.created)
for item in ordered:
- if not helper.get_setting('max_audit_entries'):
+ if not helper.get_setting("max_audit_entries"):
max_entries = 300
else:
- max_entries = helper.get_setting('max_audit_entries')
+ max_entries = helper.get_setting("max_audit_entries")
if Audit_Log.select().count() > max_entries:
Audit_Log.delete().where(Audit_Log.audit_id == item.audit_id).execute()
else:
@@ -202,28 +221,31 @@ class helpers_management:
@staticmethod
def add_to_audit_log_raw(user_name, user_id, server_id, log_msg, source_ip):
- Audit_Log.insert({
- Audit_Log.user_name: user_name,
- Audit_Log.user_id: user_id,
- Audit_Log.server_id: server_id,
- Audit_Log.log_msg: log_msg,
- Audit_Log.source_ip: source_ip
- }).execute()
- #deletes records when they're more than 100
+ Audit_Log.insert(
+ {
+ Audit_Log.user_name: user_name,
+ Audit_Log.user_id: user_id,
+ Audit_Log.server_id: server_id,
+ Audit_Log.log_msg: log_msg,
+ Audit_Log.source_ip: source_ip,
+ }
+ ).execute()
+ # deletes records when they're more than 100
ordered = Audit_Log.select().order_by(+Audit_Log.created)
for item in ordered:
- #configurable through app/config/config.json
- if not helper.get_setting('max_audit_entries'):
+ # configurable through app/config/config.json
+ if not helper.get_setting("max_audit_entries"):
max_entries = 300
else:
- max_entries = helper.get_setting('max_audit_entries')
+ max_entries = helper.get_setting("max_audit_entries")
if Audit_Log.select().count() > max_entries:
Audit_Log.delete().where(Audit_Log.audit_id == item.audit_id).execute()
else:
return
- #************************************************************************************************
+
+ # ************************************************************************************************
# Schedules Methods
- #************************************************************************************************
+ # ************************************************************************************************
@staticmethod
def create_scheduled_task(
server_id,
@@ -235,24 +257,26 @@ class helpers_management:
comment=None,
enabled=True,
one_time=False,
- cron_string='* * * * *',
+ cron_string="* * * * *",
parent=None,
- delay=0):
- sch_id = Schedules.insert({
- Schedules.server_id: server_id,
- Schedules.action: action,
- Schedules.enabled: enabled,
- Schedules.interval: interval,
- Schedules.interval_type: interval_type,
- Schedules.start_time: start_time,
- Schedules.command: command,
- Schedules.comment: comment,
- Schedules.one_time: one_time,
- Schedules.cron_string: cron_string,
- Schedules.parent: parent,
- Schedules.delay: delay
-
- }).execute()
+ delay=0,
+ ):
+ sch_id = Schedules.insert(
+ {
+ Schedules.server_id: server_id,
+ Schedules.action: action,
+ Schedules.enabled: enabled,
+ Schedules.interval: interval,
+ Schedules.interval_type: interval_type,
+ Schedules.start_time: start_time,
+ Schedules.command: command,
+ Schedules.comment: comment,
+ Schedules.one_time: one_time,
+ Schedules.cron_string: cron_string,
+ Schedules.parent: parent,
+ Schedules.delay: delay,
+ }
+ ).execute()
return sch_id
@staticmethod
@@ -282,7 +306,11 @@ class helpers_management:
@staticmethod
def get_child_schedules_by_server(schedule_id, server_id):
- return Schedules.select().where(Schedules.server_id == server_id, Schedules.parent == schedule_id).execute()
+ return (
+ Schedules.select()
+ .where(Schedules.server_id == server_id, Schedules.parent == schedule_id)
+ .execute()
+ )
@staticmethod
def get_child_schedules(schedule_id):
@@ -294,22 +322,24 @@ class helpers_management:
@staticmethod
def get_schedules_enabled():
- #pylint: disable=singleton-comparison
+ # pylint: disable=singleton-comparison
return Schedules.select().where(Schedules.enabled == True).execute()
- #************************************************************************************************
+ # ************************************************************************************************
# Backups Methods
- #************************************************************************************************
+ # ************************************************************************************************
@staticmethod
def get_backup_config(server_id):
try:
- row = Backups.select().where(Backups.server_id == server_id).join(Servers)[0]
+ row = (
+ Backups.select().where(Backups.server_id == server_id).join(Servers)[0]
+ )
conf = {
"backup_path": row.server_id.backup_path,
"excluded_dirs": row.excluded_dirs,
"max_backups": row.max_backups,
"server_id": row.server_id.server_id,
- "compress": row.compress
+ "compress": row.compress,
}
except IndexError:
conf = {
@@ -322,7 +352,13 @@ class helpers_management:
return conf
@staticmethod
- def set_backup_config(server_id: int, backup_path: str = None, max_backups: int = None, excluded_dirs: list = None, compress: bool = False):
+ def set_backup_config(
+ server_id: int,
+ backup_path: str = None,
+ max_backups: int = None,
+ excluded_dirs: list = None,
+ compress: bool = False,
+ ):
logger.debug(f"Updating server {server_id} backup config with {locals()}")
if Backups.select().where(Backups.server_id == server_id).count() != 0:
new_row = False
@@ -331,34 +367,42 @@ class helpers_management:
conf = {
"excluded_dirs": None,
"max_backups": 0,
- "server_id": server_id,
- "compress": False
+ "server_id": server_id,
+ "compress": False,
}
new_row = True
if max_backups is not None:
- conf['max_backups'] = max_backups
+ conf["max_backups"] = max_backups
if excluded_dirs is not None:
dirs_to_exclude = ",".join(excluded_dirs)
- conf['excluded_dirs'] = dirs_to_exclude
- conf['compress'] = compress
+ conf["excluded_dirs"] = dirs_to_exclude
+ conf["compress"] = compress
if not new_row:
with database.atomic():
if backup_path is not None:
- u1 = Servers.update(backup_path=backup_path).where(Servers.server_id == server_id).execute()
+ u1 = (
+ Servers.update(backup_path=backup_path)
+ .where(Servers.server_id == server_id)
+ .execute()
+ )
else:
u1 = 0
- u2 = Backups.update(conf).where(Backups.server_id == server_id).execute()
+ u2 = (
+ Backups.update(conf).where(Backups.server_id == server_id).execute()
+ )
logger.debug(f"Updating existing backup record. {u1}+{u2} rows affected")
else:
with database.atomic():
conf["server_id"] = server_id
if backup_path is not None:
- Servers.update(backup_path=backup_path).where(Servers.server_id == server_id)
+ Servers.update(backup_path=backup_path).where(
+ Servers.server_id == server_id
+ )
Backups.create(**conf)
logger.debug("Creating new backup record.")
def get_excluded_backup_dirs(self, server_id: int):
- excluded_dirs = self.get_backup_config(server_id)['excluded_dirs']
+ excluded_dirs = self.get_backup_config(server_id)["excluded_dirs"]
if excluded_dirs is not None and excluded_dirs != "":
dir_list = excluded_dirs.split(",")
else:
@@ -372,7 +416,9 @@ class helpers_management:
excluded_dirs = ",".join(dir_list)
self.set_backup_config(server_id=server_id, excluded_dirs=excluded_dirs)
else:
- logger.debug(f"Not adding {dir_to_add} to excluded directories - already in the excluded directory list for server ID {server_id}")
+ logger.debug(
+ f"Not adding {dir_to_add} to excluded directories - already in the excluded directory list for server ID {server_id}"
+ )
def del_excluded_backup_dir(self, server_id: int, dir_to_del: str):
dir_list = self.get_excluded_backup_dirs()
@@ -381,14 +427,18 @@ class helpers_management:
excluded_dirs = ",".join(dir_list)
self.set_backup_config(server_id=server_id, excluded_dirs=excluded_dirs)
else:
- logger.debug(f"Not removing {dir_to_del} from excluded directories - not in the excluded directory list for server ID {server_id}")
+ logger.debug(
+ f"Not removing {dir_to_del} from excluded directories - not in the excluded directory list for server ID {server_id}"
+ )
@staticmethod
def clear_unexecuted_commands():
- Commands.update({
- Commands.executed: True
- #pylint: disable=singleton-comparison
- }).where(Commands.executed == False).execute()
+ Commands.update(
+ {
+ Commands.executed: True
+ # pylint: disable=singleton-comparison
+ }
+ ).where(Commands.executed == False).execute()
management_helper = helpers_management()
diff --git a/app/classes/models/roles.py b/app/classes/models/roles.py
index 782e0af5..6724aafb 100644
--- a/app/classes/models/roles.py
+++ b/app/classes/models/roles.py
@@ -4,22 +4,29 @@ import datetime
from app.classes.shared.helpers import helper
try:
- from peewee import SqliteDatabase, Model, CharField, DoesNotExist, AutoField, DateTimeField
+ from peewee import (
+ SqliteDatabase,
+ Model,
+ CharField,
+ DoesNotExist,
+ AutoField,
+ DateTimeField,
+ )
from playhouse.shortcuts import model_to_dict
except ModuleNotFoundError as e:
helper.auto_installer_fix(e)
logger = logging.getLogger(__name__)
-peewee_logger = logging.getLogger('peewee')
+peewee_logger = logging.getLogger("peewee")
peewee_logger.setLevel(logging.INFO)
-database = SqliteDatabase(helper.db_path, pragmas = {
- 'journal_mode': 'wal',
- 'cache_size': -1024 * 10})
+database = SqliteDatabase(
+ helper.db_path, pragmas={"journal_mode": "wal", "cache_size": -1024 * 10}
+)
-#************************************************************************************************
+# ************************************************************************************************
# Roles Class
-#************************************************************************************************
+# ************************************************************************************************
class Roles(Model):
role_id = AutoField()
created = DateTimeField(default=datetime.datetime.now)
@@ -30,9 +37,10 @@ class Roles(Model):
table_name = "roles"
database = database
-#************************************************************************************************
+
+# ************************************************************************************************
# Roles Helpers
-#************************************************************************************************
+# ************************************************************************************************
class helper_roles:
@staticmethod
def get_all_roles():
@@ -52,10 +60,12 @@ class helper_roles:
@staticmethod
def add_role(role_name):
- role_id = Roles.insert({
- Roles.role_name: role_name.lower(),
- Roles.created: helper.get_time_as_string()
- }).execute()
+ role_id = Roles.insert(
+ {
+ Roles.role_name: role_name.lower(),
+ Roles.created: helper.get_time_as_string(),
+ }
+ ).execute()
return role_id
@staticmethod
@@ -74,4 +84,5 @@ class helper_roles:
return False
return True
+
roles_helper = helper_roles()
diff --git a/app/classes/models/server_permissions.py b/app/classes/models/server_permissions.py
index 9ec9c0cd..a4024873 100644
--- a/app/classes/models/server_permissions.py
+++ b/app/classes/models/server_permissions.py
@@ -7,35 +7,43 @@ from app.classes.shared.helpers import helper
from app.classes.shared.permission_helper import permission_helper
try:
- from peewee import SqliteDatabase, Model, ForeignKeyField, CharField, CompositeKey, JOIN
+ from peewee import (
+ SqliteDatabase,
+ Model,
+ ForeignKeyField,
+ CharField,
+ CompositeKey,
+ JOIN,
+ )
from enum import Enum
except ModuleNotFoundError as e:
helper.auto_installer_fix(e)
logger = logging.getLogger(__name__)
-peewee_logger = logging.getLogger('peewee')
+peewee_logger = logging.getLogger("peewee")
peewee_logger.setLevel(logging.INFO)
-database = SqliteDatabase(helper.db_path, pragmas = {
- 'journal_mode': 'wal',
- 'cache_size': -1024 * 10})
+database = SqliteDatabase(
+ helper.db_path, pragmas={"journal_mode": "wal", "cache_size": -1024 * 10}
+)
-#************************************************************************************************
+# ************************************************************************************************
# Role Servers Class
-#************************************************************************************************
+# ************************************************************************************************
class Role_Servers(Model):
- role_id = ForeignKeyField(Roles, backref='role_server')
- server_id = ForeignKeyField(Servers, backref='role_server')
+ role_id = ForeignKeyField(Roles, backref="role_server")
+ server_id = ForeignKeyField(Servers, backref="role_server")
permissions = CharField(default="00000000")
class Meta:
- table_name = 'role_servers'
- primary_key = CompositeKey('role_id', 'server_id')
+ table_name = "role_servers"
+ primary_key = CompositeKey("role_id", "server_id")
database = database
-#************************************************************************************************
+
+# ************************************************************************************************
# Servers Permissions Class
-#************************************************************************************************
+# ************************************************************************************************
class Enum_Permissions_Server(Enum):
Commands = 0
Terminal = 1
@@ -46,11 +54,13 @@ class Enum_Permissions_Server(Enum):
Config = 6
Players = 7
-class Permissions_Servers:
+class Permissions_Servers:
@staticmethod
def get_or_create(role_id, server, permissions_mask):
- return Role_Servers.get_or_create(role_id=role_id, server_id=server, permissions=permissions_mask)
+ return Role_Servers.get_or_create(
+ role_id=role_id, server_id=server, permissions=permissions_mask
+ )
@staticmethod
def get_permissions_list():
@@ -69,13 +79,15 @@ class Permissions_Servers:
@staticmethod
def has_permission(permission_mask, permission_tested: Enum_Permissions_Server):
- return permission_mask[permission_tested.value] == '1'
+ return permission_mask[permission_tested.value] == "1"
@staticmethod
- def set_permission(permission_mask, permission_tested: Enum_Permissions_Server, value):
+ def set_permission(
+ permission_mask, permission_tested: Enum_Permissions_Server, value
+ ):
list_perms = list(permission_mask)
list_perms[permission_tested.value] = str(value)
- permission_mask = ''.join(list_perms)
+ permission_mask = "".join(list_perms)
return permission_mask
@staticmethod
@@ -86,50 +98,71 @@ class Permissions_Servers:
def get_token_permissions(permissions_mask, api_permissions_mask):
permissions_list = []
for member in Enum_Permissions_Server.__members__.items():
- if permission_helper.both_have_perm(permissions_mask, api_permissions_mask, member[1]):
+ if permission_helper.both_have_perm(
+ permissions_mask, api_permissions_mask, member[1]
+ ):
permissions_list.append(member[1])
return permissions_list
-
- #************************************************************************************************
+ # ************************************************************************************************
# Role_Servers Methods
- #************************************************************************************************
+ # ************************************************************************************************
@staticmethod
def get_role_servers_from_role_id(roleid):
return Role_Servers.select().where(Role_Servers.role_id == roleid)
@staticmethod
def get_servers_from_role(role_id):
- return Role_Servers.select().join(Servers, JOIN.INNER).where(Role_Servers.role_id == role_id)
+ return (
+ Role_Servers.select()
+ .join(Servers, JOIN.INNER)
+ .where(Role_Servers.role_id == role_id)
+ )
@staticmethod
def get_roles_from_server(server_id):
- return Role_Servers.select().join(Roles, JOIN.INNER).where(Role_Servers.server_id == server_id)
+ return (
+ Role_Servers.select()
+ .join(Roles, JOIN.INNER)
+ .where(Role_Servers.server_id == server_id)
+ )
@staticmethod
def add_role_server(server_id, role_id, rs_permissions="00000000"):
- servers = Role_Servers.insert({Role_Servers.server_id: server_id, Role_Servers.role_id: role_id,
- Role_Servers.permissions: rs_permissions}).execute()
+ servers = Role_Servers.insert(
+ {
+ Role_Servers.server_id: server_id,
+ Role_Servers.role_id: role_id,
+ Role_Servers.permissions: rs_permissions,
+ }
+ ).execute()
return servers
@staticmethod
def get_permissions_mask(role_id, server_id):
- permissions_mask = ''
- role_server = Role_Servers.select().where(Role_Servers.role_id == role_id).where(Role_Servers.server_id == server_id).get()
+ permissions_mask = ""
+ role_server = (
+ Role_Servers.select()
+ .where(Role_Servers.role_id == role_id)
+ .where(Role_Servers.server_id == server_id)
+ .get()
+ )
permissions_mask = role_server.permissions
return permissions_mask
@staticmethod
def get_server_roles(server_id):
role_list = []
- roles = Role_Servers.select().where(Role_Servers.server_id == server_id).execute()
+ roles = (
+ Role_Servers.select().where(Role_Servers.server_id == server_id).execute()
+ )
for role in roles:
role_list.append(role.role_id)
return role_list
@staticmethod
def get_role_permissions_list(role_id):
- permissions_mask = '00000000'
+ permissions_mask = "00000000"
role_server = Role_Servers.get_or_none(Role_Servers.role_id == role_id)
if role_server is not None:
permissions_mask = role_server.permissions
@@ -138,7 +171,12 @@ class Permissions_Servers:
@staticmethod
def update_role_permission(role_id, server_id, permissions_mask):
- role_server = Role_Servers.select().where(Role_Servers.role_id == role_id).where(Role_Servers.server_id == server_id).get()
+ role_server = (
+ Role_Servers.select()
+ .where(Role_Servers.role_id == role_id)
+ .where(Role_Servers.server_id == server_id)
+ .get()
+ )
role_server.permissions = permissions_mask
Role_Servers.save(role_server)
@@ -146,12 +184,21 @@ class Permissions_Servers:
def delete_roles_permissions(role_id, removed_servers=None):
if removed_servers is None:
removed_servers = {}
- return Role_Servers.delete().where(Role_Servers.role_id == role_id).where(Role_Servers.server_id.in_(removed_servers)).execute()
+ return (
+ Role_Servers.delete()
+ .where(Role_Servers.role_id == role_id)
+ .where(Role_Servers.server_id.in_(removed_servers))
+ .execute()
+ )
@staticmethod
def remove_roles_of_server(server_id):
with database.atomic():
- return Role_Servers.delete().where(Role_Servers.server_id == server_id).execute()
+ return (
+ Role_Servers.delete()
+ .where(Role_Servers.server_id == server_id)
+ .execute()
+ )
@staticmethod
def get_user_id_permissions_mask(user_id, server_id: str):
@@ -161,14 +208,19 @@ class Permissions_Servers:
@staticmethod
def get_user_permissions_mask(user: Users, server_id: str):
if user.superuser:
- permissions_mask = '1' * len(server_permissions.get_permissions_list())
+ permissions_mask = "1" * len(server_permissions.get_permissions_list())
else:
roles_list = users_helper.get_user_roles_id(user.user_id)
- role_server = Role_Servers.select().where(Role_Servers.role_id.in_(roles_list)).where(Role_Servers.server_id == server_id).execute()
+ role_server = (
+ Role_Servers.select()
+ .where(Role_Servers.role_id.in_(roles_list))
+ .where(Role_Servers.server_id == server_id)
+ .execute()
+ )
try:
permissions_mask = role_server[0].permissions
except IndexError:
- permissions_mask = '0' * len(server_permissions.get_permissions_list())
+ permissions_mask = "0" * len(server_permissions.get_permissions_list())
return permissions_mask
@staticmethod
@@ -197,7 +249,9 @@ class Permissions_Servers:
if user.superuser:
permissions_list = server_permissions.get_permissions_list()
else:
- permissions_mask = server_permissions.get_user_permissions_mask(user, server_id)
+ permissions_mask = server_permissions.get_user_permissions_mask(
+ user, server_id
+ )
permissions_list = server_permissions.get_permissions(permissions_mask)
return permissions_list
@@ -212,11 +266,18 @@ class Permissions_Servers:
if user.superuser and key.superuser:
return server_permissions.get_permissions_list()
else:
- roles_list = users_helper.get_user_roles_id(user['user_id'])
- role_server = Role_Servers.select().where(Role_Servers.role_id.in_(roles_list)).where(Role_Servers.server_id == server_id).execute()
+ roles_list = users_helper.get_user_roles_id(user["user_id"])
+ role_server = (
+ Role_Servers.select()
+ .where(Role_Servers.role_id.in_(roles_list))
+ .where(Role_Servers.server_id == server_id)
+ .execute()
+ )
user_permissions_mask = role_server[0].permissions
key_permissions_mask = key.server_permissions
- permissions_mask = permission_helper.combine_masks(user_permissions_mask, key_permissions_mask)
+ permissions_mask = permission_helper.combine_masks(
+ user_permissions_mask, key_permissions_mask
+ )
permissions_list = server_permissions.get_permissions(permissions_mask)
return permissions_list
diff --git a/app/classes/models/servers.py b/app/classes/models/servers.py
index 07b87d31..8d5526ec 100644
--- a/app/classes/models/servers.py
+++ b/app/classes/models/servers.py
@@ -5,21 +5,31 @@ from app.classes.shared.helpers import helper
from app.classes.shared.main_models import db_helper
try:
- from peewee import SqliteDatabase, Model, ForeignKeyField, CharField, AutoField, DateTimeField, BooleanField, IntegerField, FloatField
+ from peewee import (
+ SqliteDatabase,
+ Model,
+ ForeignKeyField,
+ CharField,
+ AutoField,
+ DateTimeField,
+ BooleanField,
+ IntegerField,
+ FloatField,
+ )
except ModuleNotFoundError as e:
helper.auto_installer_fix(e)
logger = logging.getLogger(__name__)
-peewee_logger = logging.getLogger('peewee')
+peewee_logger = logging.getLogger("peewee")
peewee_logger.setLevel(logging.INFO)
-database = SqliteDatabase(helper.db_path, pragmas = {
- 'journal_mode': 'wal',
- 'cache_size': -1024 * 10})
+database = SqliteDatabase(
+ helper.db_path, pragmas={"journal_mode": "wal", "cache_size": -1024 * 10}
+)
-#************************************************************************************************
+# ************************************************************************************************
# Servers Class
-#************************************************************************************************
+# ************************************************************************************************
class Servers(Model):
server_id = AutoField()
created = DateTimeField(default=datetime.datetime.now)
@@ -45,13 +55,13 @@ class Servers(Model):
database = database
-#************************************************************************************************
+# ************************************************************************************************
# Servers Stats Class
-#************************************************************************************************
+# ************************************************************************************************
class Server_Stats(Model):
stats_id = AutoField()
created = DateTimeField(default=datetime.datetime.now)
- server_id = ForeignKeyField(Servers, backref='server', index=True)
+ server_id = ForeignKeyField(Servers, backref="server", index=True)
started = CharField(default="")
running = BooleanField(default=False)
cpu = FloatField(default=0)
@@ -72,20 +82,19 @@ class Server_Stats(Model):
crashed = BooleanField(default=False)
downloading = BooleanField(default=False)
-
class Meta:
table_name = "server_stats"
database = database
-#************************************************************************************************
+# ************************************************************************************************
# Servers Class
-#************************************************************************************************
+# ************************************************************************************************
class helper_servers:
- #************************************************************************************************
+ # ************************************************************************************************
# Generic Servers Methods
- #************************************************************************************************
+ # ************************************************************************************************
@staticmethod
def create_server(
name: str,
@@ -97,23 +106,25 @@ class helper_servers:
server_log_file: str,
server_stop: str,
server_type: str,
- server_port=25565):
- return Servers.insert({
- Servers.server_name: name,
- Servers.server_uuid: server_uuid,
- Servers.path: server_dir,
- Servers.executable: server_file,
- Servers.execution_command: server_command,
- Servers.auto_start: False,
- Servers.auto_start_delay: 10,
- Servers.crash_detection: False,
- Servers.log_path: server_log_file,
- Servers.server_port: server_port,
- Servers.stop_command: server_stop,
- Servers.backup_path: backup_path,
- Servers.type: server_type
- }).execute()
-
+ server_port=25565,
+ ):
+ return Servers.insert(
+ {
+ Servers.server_name: name,
+ Servers.server_uuid: server_uuid,
+ Servers.path: server_dir,
+ Servers.executable: server_file,
+ Servers.execution_command: server_command,
+ Servers.auto_start: False,
+ Servers.auto_start_delay: 10,
+ Servers.crash_detection: False,
+ Servers.log_path: server_log_file,
+ Servers.server_port: server_port,
+ Servers.stop_command: server_stop,
+ Servers.backup_path: backup_path,
+ Servers.type: server_type,
+ }
+ ).execute()
@staticmethod
def get_server_obj(server_id):
@@ -141,9 +152,9 @@ class helper_servers:
except IndexError:
return {}
- #************************************************************************************************
+ # ************************************************************************************************
# Servers Methods
- #************************************************************************************************
+ # ************************************************************************************************
@staticmethod
def get_all_defined_servers():
query = Servers.select()
@@ -155,10 +166,23 @@ class helper_servers:
server_data = []
try:
for s in servers:
- latest = Server_Stats.select().where(Server_Stats.server_id == s.get('server_id')).order_by(Server_Stats.created.desc()).limit(1)
- server_data.append({'server_data': s, "stats": db_helper.return_rows(latest)[0], "user_command_permission":True})
+ latest = (
+ Server_Stats.select()
+ .where(Server_Stats.server_id == s.get("server_id"))
+ .order_by(Server_Stats.created.desc())
+ .limit(1)
+ )
+ server_data.append(
+ {
+ "server_data": s,
+ "stats": db_helper.return_rows(latest)[0],
+ "user_command_permission": True,
+ }
+ )
except IndexError as ex:
- logger.error(f"Stats collection failed with error: {ex}. Was a server just created?")
+ logger.error(
+ f"Stats collection failed with error: {ex}. Was a server just created?"
+ )
return server_data
@staticmethod
@@ -167,16 +191,26 @@ class helper_servers:
friendly_name = f"{server_data.get('server_name', None)} with ID: {server_data.get('server_id', 0)}"
return friendly_name
- #************************************************************************************************
+ # ************************************************************************************************
# Servers_Stats Methods
- #************************************************************************************************
+ # ************************************************************************************************
@staticmethod
def get_latest_server_stats(server_id):
- return Server_Stats.select().where(Server_Stats.server_id == server_id).order_by(Server_Stats.created.desc()).limit(1)
+ return (
+ Server_Stats.select()
+ .where(Server_Stats.server_id == server_id)
+ .order_by(Server_Stats.created.desc())
+ .limit(1)
+ )
@staticmethod
def get_server_stats_by_id(server_id):
- stats = Server_Stats.select().where(Server_Stats.server_id == server_id).order_by(Server_Stats.created.desc()).limit(1)
+ stats = (
+ Server_Stats.select()
+ .where(Server_Stats.server_id == server_id)
+ .order_by(Server_Stats.created.desc())
+ .limit(1)
+ )
return db_helper.return_rows(stats)[0]
@staticmethod
@@ -188,33 +222,42 @@ class helper_servers:
@staticmethod
def sever_crashed(server_id):
with database.atomic():
- Server_Stats.update(crashed=True).where(Server_Stats.server_id == server_id).execute()
+ Server_Stats.update(crashed=True).where(
+ Server_Stats.server_id == server_id
+ ).execute()
@staticmethod
def set_download(server_id):
with database.atomic():
- Server_Stats.update(downloading=True).where(Server_Stats.server_id == server_id).execute()
+ Server_Stats.update(downloading=True).where(
+ Server_Stats.server_id == server_id
+ ).execute()
@staticmethod
def finish_download(server_id):
with database.atomic():
- Server_Stats.update(downloading=False).where(Server_Stats.server_id == server_id).execute()
+ Server_Stats.update(downloading=False).where(
+ Server_Stats.server_id == server_id
+ ).execute()
@staticmethod
def get_download_status(server_id):
- download_status = Server_Stats.select().where(Server_Stats.server_id == server_id).get()
+ download_status = (
+ Server_Stats.select().where(Server_Stats.server_id == server_id).get()
+ )
return download_status.downloading
-
@staticmethod
def server_crash_reset(server_id):
with database.atomic():
- Server_Stats.update(crashed=False).where(Server_Stats.server_id == server_id).execute()
+ Server_Stats.update(crashed=False).where(
+ Server_Stats.server_id == server_id
+ ).execute()
@staticmethod
def is_crashed(server_id):
svr = Server_Stats.select().where(Server_Stats.server_id == server_id).get()
- #pylint: disable=singleton-comparison
+ # pylint: disable=singleton-comparison
if svr.crashed == True:
return True
else:
@@ -223,44 +266,58 @@ class helper_servers:
@staticmethod
def set_update(server_id, value):
try:
- #Checks if server even exists
+ # Checks if server even exists
Server_Stats.select().where(Server_Stats.server_id == server_id)
except Exception as ex:
logger.error(f"Database entry not found! {ex}")
with database.atomic():
- Server_Stats.update(updating=value).where(Server_Stats.server_id == server_id).execute()
+ Server_Stats.update(updating=value).where(
+ Server_Stats.server_id == server_id
+ ).execute()
@staticmethod
def get_update_status(server_id):
- update_status = Server_Stats.select().where(Server_Stats.server_id == server_id).get()
+ update_status = (
+ Server_Stats.select().where(Server_Stats.server_id == server_id).get()
+ )
return update_status.updating
@staticmethod
def set_first_run(server_id):
- #Sets first run to false
+ # Sets first run to false
try:
- #Checks if server even exists
+ # Checks if server even exists
Server_Stats.select().where(Server_Stats.server_id == server_id)
except Exception as ex:
logger.error(f"Database entry not found! {ex}")
return
with database.atomic():
- Server_Stats.update(first_run=False).where(Server_Stats.server_id == server_id).execute()
+ Server_Stats.update(first_run=False).where(
+ Server_Stats.server_id == server_id
+ ).execute()
@staticmethod
def get_first_run(server_id):
- first_run = Server_Stats.select().where(Server_Stats.server_id == server_id).get()
+ first_run = (
+ Server_Stats.select().where(Server_Stats.server_id == server_id).get()
+ )
return first_run.first_run
@staticmethod
def get_TTL_without_player(server_id):
- last_stat = Server_Stats.select().where(Server_Stats.server_id == server_id).order_by(Server_Stats.created.desc()).first()
- last_stat_with_player = (Server_Stats
- .select()
- .where(Server_Stats.server_id == server_id)
- .where(Server_Stats.online > 0)
- .order_by(Server_Stats.created.desc())
- .first())
+ last_stat = (
+ Server_Stats.select()
+ .where(Server_Stats.server_id == server_id)
+ .order_by(Server_Stats.created.desc())
+ .first()
+ )
+ last_stat_with_player = (
+ Server_Stats.select()
+ .where(Server_Stats.server_id == server_id)
+ .where(Server_Stats.online > 0)
+ .order_by(Server_Stats.created.desc())
+ .first()
+ )
return last_stat.created - last_stat_with_player.created
@staticmethod
@@ -279,11 +336,15 @@ class helper_servers:
except Exception as ex:
logger.error(f"Database entry not found! {ex}")
with database.atomic():
- Server_Stats.update(waiting_start=value).where(Server_Stats.server_id == server_id).execute()
+ Server_Stats.update(waiting_start=value).where(
+ Server_Stats.server_id == server_id
+ ).execute()
@staticmethod
def get_waiting_start(server_id):
- waiting_start = Server_Stats.select().where(Server_Stats.server_id == server_id).get()
+ waiting_start = (
+ Server_Stats.select().where(Server_Stats.server_id == server_id).get()
+ )
return waiting_start.waiting_start
diff --git a/app/classes/models/users.py b/app/classes/models/users.py
index 751fdb1c..24020e61 100644
--- a/app/classes/models/users.py
+++ b/app/classes/models/users.py
@@ -6,22 +6,33 @@ from app.classes.models.roles import Roles, roles_helper
from app.classes.shared.helpers import helper
try:
- from peewee import SqliteDatabase, Model, ForeignKeyField, CharField, AutoField, DateTimeField, BooleanField, CompositeKey, DoesNotExist, JOIN
+ from peewee import (
+ SqliteDatabase,
+ Model,
+ ForeignKeyField,
+ CharField,
+ AutoField,
+ DateTimeField,
+ BooleanField,
+ CompositeKey,
+ DoesNotExist,
+ JOIN,
+ )
from playhouse.shortcuts import model_to_dict
except ModuleNotFoundError as e:
helper.auto_installer_fix(e)
logger = logging.getLogger(__name__)
-peewee_logger = logging.getLogger('peewee')
+peewee_logger = logging.getLogger("peewee")
peewee_logger.setLevel(logging.INFO)
-database = SqliteDatabase(helper.db_path, pragmas = {
- 'journal_mode': 'wal',
- 'cache_size': -1024 * 10})
+database = SqliteDatabase(
+ helper.db_path, pragmas={"journal_mode": "wal", "cache_size": -1024 * 10}
+)
-#************************************************************************************************
+# ************************************************************************************************
# Users Class
-#************************************************************************************************
+# ************************************************************************************************
class Users(Model):
user_id = AutoField()
created = DateTimeField(default=datetime.datetime.now)
@@ -34,7 +45,7 @@ class Users(Model):
enabled = BooleanField(default=True)
superuser = BooleanField(default=False)
lang = CharField(default="en_EN")
- support_logs = CharField(default = '')
+ support_logs = CharField(default="")
valid_tokens_from = DateTimeField(default=datetime.datetime.now)
server_order = CharField(default="")
preparing = BooleanField(default=False)
@@ -49,35 +60,35 @@ class Users(Model):
# ************************************************************************************************
class ApiKeys(Model):
token_id = AutoField()
- name = CharField(default='', unique=True, index=True)
+ name = CharField(default="", unique=True, index=True)
created = DateTimeField(default=datetime.datetime.now)
- user_id = ForeignKeyField(Users, backref='api_token', index=True)
- server_permissions = CharField(default='00000000')
- crafty_permissions = CharField(default='000')
+ user_id = ForeignKeyField(Users, backref="api_token", index=True)
+ server_permissions = CharField(default="00000000")
+ crafty_permissions = CharField(default="000")
superuser = BooleanField(default=False)
class Meta:
- table_name = 'api_keys'
+ table_name = "api_keys"
database = database
-#************************************************************************************************
+# ************************************************************************************************
# User Roles Class
-#************************************************************************************************
+# ************************************************************************************************
class User_Roles(Model):
- user_id = ForeignKeyField(Users, backref='user_role')
- role_id = ForeignKeyField(Roles, backref='user_role')
+ user_id = ForeignKeyField(Users, backref="user_role")
+ role_id = ForeignKeyField(Roles, backref="user_role")
class Meta:
- table_name = 'user_roles'
- primary_key = CompositeKey('user_id', 'role_id')
+ table_name = "user_roles"
+ primary_key = CompositeKey("user_id", "role_id")
database = database
-#************************************************************************************************
-# Users Helpers
-#************************************************************************************************
-class helper_users:
+# ************************************************************************************************
+# Users Helpers
+# ************************************************************************************************
+class helper_users:
@staticmethod
def get_by_id(user_id):
return Users.get_by_id(user_id)
@@ -107,19 +118,19 @@ class helper_users:
def get_user(user_id):
if user_id == 0:
return {
- 'user_id': 0,
- 'created': '10/24/2019, 11:34:00',
- 'last_login': '10/24/2019, 11:34:00',
- 'last_update': '10/24/2019, 11:34:00',
- 'last_ip': "127.27.23.89",
- 'username': "SYSTEM",
- 'password': None,
- 'email': "default@example.com",
- 'enabled': True,
- 'superuser': True,
- 'roles': [],
- 'servers': [],
- 'support_logs': '',
+ "user_id": 0,
+ "created": "10/24/2019, 11:34:00",
+ "last_login": "10/24/2019, 11:34:00",
+ "last_update": "10/24/2019, 11:34:00",
+ "last_ip": "127.27.23.89",
+ "username": "SYSTEM",
+ "password": None,
+ "email": "default@example.com",
+ "enabled": True,
+ "superuser": True,
+ "roles": [],
+ "servers": [],
+ "support_logs": "",
}
user = model_to_dict(Users.get(Users.user_id == user_id))
@@ -128,7 +139,7 @@ class helper_users:
user = users_helper.add_user_roles(user)
return user
else:
- #logger.debug("user: ({}) {}".format(user_id, {}))
+ # logger.debug("user: ({}) {}".format(user_id, {}))
return {}
@staticmethod
@@ -147,31 +158,47 @@ class helper_users:
return user
@staticmethod
- def add_user(username: str, password: str = None, email: Optional[str] = None, enabled: bool = True, superuser: bool = False) -> str:
+ def add_user(
+ username: str,
+ password: str = None,
+ email: Optional[str] = None,
+ enabled: bool = True,
+ superuser: bool = False,
+ ) -> str:
if password is not None:
pw_enc = helper.encode_pass(password)
else:
pw_enc = None
- user_id = Users.insert({
- Users.username: username.lower(),
- Users.password: pw_enc,
- Users.email: email,
- Users.enabled: enabled,
- Users.superuser: superuser,
- Users.created: helper.get_time_as_string()
- }).execute()
+ user_id = Users.insert(
+ {
+ Users.username: username.lower(),
+ Users.password: pw_enc,
+ Users.email: email,
+ Users.enabled: enabled,
+ Users.superuser: superuser,
+ Users.created: helper.get_time_as_string(),
+ }
+ ).execute()
return user_id
@staticmethod
- def add_rawpass_user(username: str, password: str = None, email: Optional[str] = None, enabled: bool = True, superuser: bool = False) -> str:
- user_id = Users.insert({
- Users.username: username.lower(),
- Users.password: password,
- Users.email: email,
- Users.enabled: enabled,
- Users.superuser: superuser,
- Users.created: helper.get_time_as_string()
- }).execute()
+ def add_rawpass_user(
+ username: str,
+ password: str = None,
+ email: Optional[str] = None,
+ enabled: bool = True,
+ superuser: bool = False,
+ ) -> str:
+ user_id = Users.insert(
+ {
+ Users.username: username.lower(),
+ Users.password: password,
+ Users.email: email,
+ Users.enabled: enabled,
+ Users.superuser: superuser,
+ Users.created: helper.get_time_as_string(),
+ }
+ ).execute()
return user_id
@staticmethod
@@ -183,7 +210,9 @@ class helper_users:
@staticmethod
def update_server_order(user_id, user_server_order):
- Users.update(server_order = user_server_order).where(Users.user_id == user_id).execute()
+ Users.update(server_order=user_server_order).where(
+ Users.user_id == user_id
+ ).execute()
@staticmethod
def get_server_order(user_id):
@@ -208,20 +237,22 @@ class helper_users:
@staticmethod
def set_support_path(user_id, support_path):
- Users.update(support_logs = support_path).where(Users.user_id == user_id).execute()
+ Users.update(support_logs=support_path).where(
+ Users.user_id == user_id
+ ).execute()
@staticmethod
def set_prepare(user_id):
- Users.update(preparing = True).where(Users.user_id == user_id).execute()
+ Users.update(preparing=True).where(Users.user_id == user_id).execute()
@staticmethod
def stop_prepare(user_id):
- Users.update(preparing = False).where(Users.user_id == user_id).execute()
+ Users.update(preparing=False).where(Users.user_id == user_id).execute()
@staticmethod
def clear_support_status():
- #pylint: disable=singleton-comparison
- Users.update(preparing = False).where(Users.preparing == True).execute()
+ # pylint: disable=singleton-comparison
+ Users.update(preparing=False).where(Users.preparing == True).execute()
@staticmethod
def user_id_exists(user_id):
@@ -229,9 +260,9 @@ class helper_users:
return False
return True
-#************************************************************************************************
-# User_Roles Methods
-#************************************************************************************************
+ # ************************************************************************************************
+ # User_Roles Methods
+ # ************************************************************************************************
@staticmethod
def get_or_create(user_id, role_id):
@@ -242,7 +273,7 @@ class helper_users:
roles_list = []
roles = User_Roles.select().where(User_Roles.user_id == user_id)
for r in roles:
- roles_list.append(roles_helper.get_role(r.role_id)['role_id'])
+ roles_list.append(roles_helper.get_role(r.role_id)["role_id"])
return roles_list
@staticmethod
@@ -250,37 +281,40 @@ class helper_users:
roles_list = []
roles = User_Roles.select().where(User_Roles.user_id == user_id)
for r in roles:
- roles_list.append(roles_helper.get_role(r.role_id)['role_name'])
+ roles_list.append(roles_helper.get_role(r.role_id)["role_name"])
return roles_list
@staticmethod
def add_role_to_user(user_id, role_id):
- User_Roles.insert({
- User_Roles.user_id: user_id,
- User_Roles.role_id: role_id
- }).execute()
+ User_Roles.insert(
+ {User_Roles.user_id: user_id, User_Roles.role_id: role_id}
+ ).execute()
@staticmethod
def add_user_roles(user: Union[dict, Users]):
if isinstance(user, dict):
- user_id = user['user_id']
+ user_id = user["user_id"]
else:
user_id = user.user_id
# I just copied this code from get_user, it had those TODOs & comments made by mac - Lukas
- roles_query = User_Roles.select().join(Roles, JOIN.INNER).where(User_Roles.user_id == user_id)
+ roles_query = (
+ User_Roles.select()
+ .join(Roles, JOIN.INNER)
+ .where(User_Roles.user_id == user_id)
+ )
# TODO: this query needs to be narrower
roles = set()
for r in roles_query:
roles.add(r.role_id.role_id)
if isinstance(user, dict):
- user['roles'] = roles
+ user["roles"] = roles
else:
user.roles = roles
- #logger.debug("user: ({}) {}".format(user_id, user))
+ # logger.debug("user: ({}) {}".format(user_id, user))
return user
@staticmethod
@@ -293,15 +327,17 @@ class helper_users:
@staticmethod
def delete_user_roles(user_id, removed_roles):
- User_Roles.delete().where(User_Roles.user_id == user_id).where(User_Roles.role_id.in_(removed_roles)).execute()
+ User_Roles.delete().where(User_Roles.user_id == user_id).where(
+ User_Roles.role_id.in_(removed_roles)
+ ).execute()
@staticmethod
def remove_roles_from_role_id(role_id):
User_Roles.delete().where(User_Roles.role_id == role_id).execute()
-# ************************************************************************************************
-# ApiKeys Methods
-# ************************************************************************************************
+ # ************************************************************************************************
+ # ApiKeys Methods
+ # ************************************************************************************************
@staticmethod
def get_user_api_keys(user_id: str):
@@ -313,18 +349,29 @@ class helper_users:
@staticmethod
def add_user_api_key(
- name: str,
- user_id: str,
- superuser: bool = False,
- server_permissions_mask: Optional[str] = None,
- crafty_permissions_mask: Optional[str] = None):
- return ApiKeys.insert({
- ApiKeys.name: name,
- ApiKeys.user_id: user_id,
- **({ApiKeys.server_permissions: server_permissions_mask} if server_permissions_mask is not None else {}),
- **({ApiKeys.crafty_permissions: crafty_permissions_mask} if crafty_permissions_mask is not None else {}),
- ApiKeys.superuser: superuser
- }).execute()
+ name: str,
+ user_id: str,
+ superuser: bool = False,
+ server_permissions_mask: Optional[str] = None,
+ crafty_permissions_mask: Optional[str] = None,
+ ):
+ return ApiKeys.insert(
+ {
+ ApiKeys.name: name,
+ ApiKeys.user_id: user_id,
+ **(
+ {ApiKeys.server_permissions: server_permissions_mask}
+ if server_permissions_mask is not None
+ else {}
+ ),
+ **(
+ {ApiKeys.crafty_permissions: crafty_permissions_mask}
+ if crafty_permissions_mask is not None
+ else {}
+ ),
+ ApiKeys.superuser: superuser,
+ }
+ ).execute()
@staticmethod
def delete_user_api_keys(user_id: str):
@@ -335,5 +382,4 @@ class helper_users:
ApiKeys.delete().where(ApiKeys.token_id == key_id).execute()
-
users_helper = helper_users()
diff --git a/app/classes/shared/authentication.py b/app/classes/shared/authentication.py
index 25053d0b..6592fdd3 100644
--- a/app/classes/shared/authentication.py
+++ b/app/classes/shared/authentication.py
@@ -14,12 +14,13 @@ except ModuleNotFoundError as e:
logger = logging.getLogger(__name__)
+
class Authentication:
def __init__(self):
self.secret = "my secret"
- self.secret = helper.get_setting('apikey_secret', None)
+ self.secret = helper.get_setting("apikey_secret", None)
- if self.secret is None or self.secret == 'random':
+ if self.secret is None or self.secret == "random":
self.secret = helper.random_string_generator(64)
@staticmethod
@@ -27,13 +28,9 @@ class Authentication:
if extra is None:
extra = {}
return jwt.encode(
- {
- 'user_id': user_id,
- 'iat': int(time.time()),
- **extra
- },
+ {"user_id": user_id, "iat": int(time.time()), **extra},
authentication.secret,
- algorithm="HS256"
+ algorithm="HS256",
)
@staticmethod
@@ -49,23 +46,25 @@ class Authentication:
return None
@staticmethod
- def check(token) -> Optional[Tuple[Optional[ApiKeys], Dict[str, Any], Dict[str, Any]]]:
+ def check(
+ token,
+ ) -> Optional[Tuple[Optional[ApiKeys], Dict[str, Any], Dict[str, Any]]]:
try:
data = jwt.decode(token, authentication.secret, algorithms=["HS256"])
except PyJWTError as error:
logger.debug("Error while checking JWT token: ", exc_info=error)
return None
- iat: int = data['iat']
+ iat: int = data["iat"]
key: Optional[ApiKeys] = None
- if 'token_id' in data:
- key_id = data['token_id']
+ if "token_id" in data:
+ key_id = data["token_id"]
key = users_helper.get_user_api_key(key_id)
if key is None:
return None
- user_id: str = data['user_id']
+ user_id: str = data["user_id"]
user = users_helper.get_user(user_id)
# TODO: Have a cache or something so we don't constantly have to query the database
- if int(user.get('valid_tokens_from').timestamp()) < iat:
+ if int(user.get("valid_tokens_from").timestamp()) < iat:
# Success!
return key, data, user
else:
diff --git a/app/classes/shared/command.py b/app/classes/shared/command.py
index 112effbb..8907f4c5 100644
--- a/app/classes/shared/command.py
+++ b/app/classes/shared/command.py
@@ -11,8 +11,8 @@ from app.classes.web.websocket_helper import websocket_helper
logger = logging.getLogger(__name__)
-class MainPrompt(cmd.Cmd):
+class MainPrompt(cmd.Cmd):
def __init__(self, tasks_manager, migration_manager):
super().__init__()
self.tasks_manager = tasks_manager
@@ -25,48 +25,52 @@ class MainPrompt(cmd.Cmd):
def emptyline():
pass
- #pylint: disable=unused-argument
+ # pylint: disable=unused-argument
def do_exit(self, line):
self.tasks_manager._main_graceful_exit()
self.universal_exit()
def do_migrations(self, line):
- if line == 'up':
+ if line == "up":
self.migration_manager.up()
- elif line == 'down':
+ elif line == "down":
self.migration_manager.down()
- elif line == 'done':
+ elif line == "done":
console.info(self.migration_manager.done)
- elif line == 'todo':
+ elif line == "todo":
console.info(self.migration_manager.todo)
- elif line == 'diff':
+ elif line == "diff":
console.info(self.migration_manager.diff)
- elif line == 'info':
- console.info(f'Done: {self.migration_manager.done}')
- console.info(f'FS: {self.migration_manager.todo}')
- console.info(f'Todo: {self.migration_manager.diff}')
- elif line.startswith('add '):
- migration_name = line[len('add '):]
+ elif line == "info":
+ console.info(f"Done: {self.migration_manager.done}")
+ console.info(f"FS: {self.migration_manager.todo}")
+ console.info(f"Todo: {self.migration_manager.diff}")
+ elif line.startswith("add "):
+ migration_name = line[len("add ") :]
self.migration_manager.create(migration_name, False)
else:
- console.info('Unknown migration command')
+ console.info("Unknown migration command")
@staticmethod
def do_threads(_line):
for thread in threading.enumerate():
if sys.version_info >= (3, 8):
- print(f'Name: {thread.name} Identifier: {thread.ident} TID/PID: {thread.native_id}')
+ print(
+ f"Name: {thread.name} Identifier: {thread.ident} TID/PID: {thread.native_id}"
+ )
else:
- print(f'Name: {thread.name} Identifier: {thread.ident}')
+ print(f"Name: {thread.name} Identifier: {thread.ident}")
def do_import3(self, _line):
import3.start_import()
def universal_exit(self):
logger.info("Stopping all server daemons / threads")
- console.info("Stopping all server daemons / threads - This may take a few seconds")
+ console.info(
+ "Stopping all server daemons / threads - This may take a few seconds"
+ )
websocket_helper.disconnect_all()
- console.info('Waiting for main thread to stop')
+ console.info("Waiting for main thread to stop")
while True:
if self.tasks_manager.get_main_thread_run_status():
sys.exit(0)
diff --git a/app/classes/shared/console.py b/app/classes/shared/console.py
index 96b1e324..bcaf17a6 100644
--- a/app/classes/shared/console.py
+++ b/app/classes/shared/console.py
@@ -12,16 +12,18 @@ except ModuleNotFoundError as ex:
logger.critical(f"Import Error: Unable to load {ex.name} module", exc_info=True)
print(f"Import Error: Unable to load {ex.name} module")
from app.classes.shared.installer import installer
- installer.do_install()
-class Console:
+ installer.do_install()
+
+
+class Console:
def __init__(self):
- if 'colorama' in sys.modules:
+ if "colorama" in sys.modules:
init()
@staticmethod
def do_print(message, color):
- if 'termcolor' in sys.modules or 'colorama' in sys.modules:
+ if "termcolor" in sys.modules or "colorama" in sys.modules:
print(colored(message, color))
else:
print(message)
diff --git a/app/classes/shared/exceptions.py b/app/classes/shared/exceptions.py
index 05a46fdb..0b1cac75 100644
--- a/app/classes/shared/exceptions.py
+++ b/app/classes/shared/exceptions.py
@@ -1,8 +1,10 @@
class CraftyException(Exception):
pass
+
class DatabaseException(CraftyException):
pass
+
class SchemaError(DatabaseException):
pass
diff --git a/app/classes/shared/file_helpers.py b/app/classes/shared/file_helpers.py
index 205c9066..a706d115 100644
--- a/app/classes/shared/file_helpers.py
+++ b/app/classes/shared/file_helpers.py
@@ -6,12 +6,9 @@ from zipfile import ZipFile, ZIP_DEFLATED
logger = logging.getLogger(__name__)
+
class FileHelpers:
- allowed_quotes = [
- "\"",
- "'",
- "`"
- ]
+ allowed_quotes = ['"', "'", "`"]
def del_dirs(self, path):
path = pathlib.Path(path)
@@ -32,7 +29,7 @@ class FileHelpers:
path = pathlib.Path(path)
try:
logger.debug(f"Deleting file: {path}")
- #Remove the file
+ # Remove the file
os.remove(path)
return True
except FileNotFoundError:
@@ -59,43 +56,58 @@ class FileHelpers:
@staticmethod
def make_archive(path_to_destination, path_to_zip):
# create a ZipFile object
- path_to_destination += '.zip'
- with ZipFile(path_to_destination, 'w') as z:
+ path_to_destination += ".zip"
+ with ZipFile(path_to_destination, "w") as z:
for root, _dirs, files in os.walk(path_to_zip, topdown=True):
ziproot = path_to_zip
for file in files:
try:
logger.info(f"backing up: {os.path.join(root, file)}")
if os.name == "nt":
- z.write(os.path.join(root, file), os.path.join(root.replace(ziproot, ""), file))
+ z.write(
+ os.path.join(root, file),
+ os.path.join(root.replace(ziproot, ""), file),
+ )
else:
- z.write(os.path.join(root, file), os.path.join(root.replace(ziproot, "/"), file))
+ z.write(
+ os.path.join(root, file),
+ os.path.join(root.replace(ziproot, "/"), file),
+ )
except Exception as e:
- logger.warning(f"Error backing up: {os.path.join(root, file)}! - Error was: {e}")
-
+ logger.warning(
+ f"Error backing up: {os.path.join(root, file)}! - Error was: {e}"
+ )
return True
@staticmethod
def make_compressed_archive(path_to_destination, path_to_zip):
# create a ZipFile object
- path_to_destination += '.zip'
- with ZipFile(path_to_destination, 'w', ZIP_DEFLATED) as z:
+ path_to_destination += ".zip"
+ with ZipFile(path_to_destination, "w", ZIP_DEFLATED) as z:
for root, _dirs, files in os.walk(path_to_zip, topdown=True):
ziproot = path_to_zip
for file in files:
try:
logger.info(f"backing up: {os.path.join(root, file)}")
if os.name == "nt":
- z.write(os.path.join(root, file), os.path.join(root.replace(ziproot, ""), file))
+ z.write(
+ os.path.join(root, file),
+ os.path.join(root.replace(ziproot, ""), file),
+ )
else:
- z.write(os.path.join(root, file), os.path.join(root.replace(ziproot, "/"), file))
+ z.write(
+ os.path.join(root, file),
+ os.path.join(root.replace(ziproot, "/"), file),
+ )
except Exception as e:
- logger.warning(f"Error backing up: {os.path.join(root, file)}! - Error was: {e}")
-
+ logger.warning(
+ f"Error backing up: {os.path.join(root, file)}! - Error was: {e}"
+ )
return True
+
file_helper = FileHelpers()
diff --git a/app/classes/shared/helpers.py b/app/classes/shared/helpers.py
index 4af17478..cbc75b2a 100644
--- a/app/classes/shared/helpers.py
+++ b/app/classes/shared/helpers.py
@@ -37,28 +37,27 @@ except ModuleNotFoundError as err:
print(f"Import Error: Unable to load {err.name} module")
installer.do_install()
+
class Helpers:
- allowed_quotes = [
- "\"",
- "'",
- "`"
- ]
+ allowed_quotes = ['"', "'", "`"]
def __init__(self):
self.root_dir = os.path.abspath(os.path.curdir)
- self.config_dir = os.path.join(self.root_dir, 'app', 'config')
- self.webroot = os.path.join(self.root_dir, 'app', 'frontend')
- self.servers_dir = os.path.join(self.root_dir, 'servers')
- self.backup_path = os.path.join(self.root_dir, 'backups')
- self.migration_dir = os.path.join(self.root_dir, 'app', 'migrations')
+ self.config_dir = os.path.join(self.root_dir, "app", "config")
+ self.webroot = os.path.join(self.root_dir, "app", "frontend")
+ self.servers_dir = os.path.join(self.root_dir, "servers")
+ self.backup_path = os.path.join(self.root_dir, "backups")
+ self.migration_dir = os.path.join(self.root_dir, "app", "migrations")
- self.session_file = os.path.join(self.root_dir, 'app', 'config', 'session.lock')
- self.settings_file = os.path.join(self.root_dir, 'app', 'config', 'config.json')
+ self.session_file = os.path.join(self.root_dir, "app", "config", "session.lock")
+ self.settings_file = os.path.join(self.root_dir, "app", "config", "config.json")
- self.ensure_dir_exists(os.path.join(self.root_dir, 'app', 'config', 'db'))
- self.db_path = os.path.join(self.root_dir, 'app', 'config', 'db', 'crafty.sqlite')
- self.serverjar_cache = os.path.join(self.config_dir, 'serverjars.json')
- self.credits_cache = os.path.join(self.config_dir, 'credits.json')
+ self.ensure_dir_exists(os.path.join(self.root_dir, "app", "config", "db"))
+ self.db_path = os.path.join(
+ self.root_dir, "app", "config", "db", "crafty.sqlite"
+ )
+ self.serverjar_cache = os.path.join(self.config_dir, "serverjars.json")
+ self.credits_cache = os.path.join(self.config_dir, "credits.json")
self.passhasher = PasswordHasher()
self.exiting = False
@@ -74,7 +73,7 @@ class Helpers:
def check_file_perms(self, path):
try:
- open(path, "r", encoding='utf-8').close()
+ open(path, "r", encoding="utf-8").close()
logger.info(f"{path} is readable")
return True
except PermissionError:
@@ -97,7 +96,7 @@ class Helpers:
@staticmethod
def check_internet():
try:
- requests.get('https://google.com', timeout=1)
+ requests.get("https://google.com", timeout=1)
return True
except Exception:
return False
@@ -105,9 +104,9 @@ class Helpers:
@staticmethod
def check_port(server_port):
try:
- ip = get('https://api.ipify.org').content.decode('utf8')
+ ip = get("https://api.ipify.org").content.decode("utf8")
except:
- ip = 'google.com'
+ ip = "google.com"
a_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
a_socket.settimeout(20.0)
@@ -125,7 +124,7 @@ class Helpers:
def check_server_conn(server_port):
a_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
a_socket.settimeout(10.0)
- ip = '127.0.0.1'
+ ip = "127.0.0.1"
location = (ip, server_port)
result_of_check = a_socket.connect_ex(location)
@@ -139,43 +138,55 @@ class Helpers:
@staticmethod
def cmdparse(cmd_in):
# Parse a string into arguments
- cmd_out = [] # "argv" output array
- ci = -1 # command index - pointer to the argument we're building in cmd_out
- np = True # whether we're creating a new argument/parameter
+ cmd_out = [] # "argv" output array
+ ci = -1 # command index - pointer to the argument we're building in cmd_out
+ np = True # whether we're creating a new argument/parameter
esc = False # whether an escape character was encountered
stch = None # if we're dealing with a quote, save the quote type here. Nested quotes to be dealt with by the command
- for c in cmd_in: # for character in string
- if np: # if set, begin a new argument and increment the command index. Continue the loop.
- if c == ' ':
+ for c in cmd_in: # for character in string
+ if (
+ np
+ ): # if set, begin a new argument and increment the command index. Continue the loop.
+ if c == " ":
continue
else:
ci += 1
cmd_out.append("")
np = False
- if esc: # if we encountered an escape character on the last loop, append this char regardless of what it is
+ if (
+ esc
+ ): # if we encountered an escape character on the last loop, append this char regardless of what it is
if c not in Helpers.allowed_quotes:
- cmd_out[ci] += '\\'
+ cmd_out[ci] += "\\"
cmd_out[ci] += c
esc = False
else:
- if c == '\\': # if the current character is an escape character, set the esc flag and continue to next loop
+ if (
+ c == "\\"
+ ): # if the current character is an escape character, set the esc flag and continue to next loop
esc = True
- elif c == ' ' and stch is None: # if we encounter a space and are not dealing with a quote,
- # set the new argument flag and continue to next loop
+ elif (
+ c == " " and stch is None
+ ): # if we encounter a space and are not dealing with a quote,
+ # set the new argument flag and continue to next loop
np = True
- elif c == stch: # if we encounter the character that matches our start quote, end the quote and continue to next loop
+ elif (
+ c == stch
+ ): # if we encounter the character that matches our start quote, end the quote and continue to next loop
stch = None
- elif stch is None and (c in Helpers.allowed_quotes): # if we're not in the middle of a quote and we get a quotable character,
- # start a quote and proceed to the next loop
+ elif stch is None and (
+ c in Helpers.allowed_quotes
+ ): # if we're not in the middle of a quote and we get a quotable character,
+ # start a quote and proceed to the next loop
stch = c
- else: # else, just store the character in the current arg
+ else: # else, just store the character in the current arg
cmd_out[ci] += c
return cmd_out
def get_setting(self, key, default_return=False):
try:
- with open(self.settings_file, "r", encoding='utf-8') as f:
+ with open(self.settings_file, "r", encoding="utf-8") as f:
data = json.load(f)
if key in data.keys():
@@ -187,8 +198,12 @@ class Helpers:
return default_return
except Exception as e:
- logger.critical(f"Config File Error: Unable to read {self.settings_file} due to {e}")
- console.critical(f"Config File Error: Unable to read {self.settings_file} due to {e}")
+ logger.critical(
+ f"Config File Error: Unable to read {self.settings_file} due to {e}"
+ )
+ console.critical(
+ f"Config File Error: Unable to read {self.settings_file} due to {e}"
+ )
return default_return
@@ -196,10 +211,10 @@ class Helpers:
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
try:
# doesn't even have to be reachable
- s.connect(('10.255.255.255', 1))
+ s.connect(("10.255.255.255", 1))
IP = s.getsockname()[0]
except Exception:
- IP = '127.0.0.1'
+ IP = "127.0.0.1"
finally:
s.close()
return IP
@@ -207,7 +222,9 @@ class Helpers:
def get_version(self):
version_data = {}
try:
- with open(os.path.join(self.config_dir, 'version.json'), 'r', encoding='utf-8') as f:
+ with open(
+ os.path.join(self.config_dir, "version.json"), "r", encoding="utf-8"
+ ) as f:
version_data = json.load(f)
except Exception as e:
@@ -217,9 +234,11 @@ class Helpers:
@staticmethod
def get_announcements():
- r = requests.get('https://craftycontrol.com/notify.json', timeout=2)
- data = '[{"id":"1","date":"Unknown","title":"Error getting Announcements","desc":"Error getting ' \
- 'Announcements","link":""}] '
+ r = requests.get("https://craftycontrol.com/notify.json", timeout=2)
+ data = (
+ '[{"id":"1","date":"Unknown","title":"Error getting Announcements","desc":"Error getting '
+ 'Announcements","link":""}] '
+ )
if r.status_code in [200, 201]:
try:
@@ -229,14 +248,13 @@ class Helpers:
return data
-
def get_version_string(self):
version_data = self.get_version()
- major = version_data.get('major', '?')
- minor = version_data.get('minor', '?')
- sub = version_data.get('sub', '?')
- meta = version_data.get('meta', '?')
+ major = version_data.get("major", "?")
+ minor = version_data.get("minor", "?")
+ sub = version_data.get("sub", "?")
+ meta = version_data.get("meta", "?")
# set some defaults if we don't get version_data from our helper
version = f"{major}.{minor}.{sub}-{meta}"
@@ -256,25 +274,31 @@ class Helpers:
# our regex replacements
# note these are in a tuple
- user_keywords = self.get_setting('keywords')
+ user_keywords = self.get_setting("keywords")
replacements = [
- (r'(\[.+?/INFO\])', r'\1'),
- (r'(\[.+?/WARN\])', r'\1'),
- (r'(\[.+?/ERROR\])', r'\1'),
- (r'(\[.+?/FATAL\])', r'\1'),
- (r'(\w+?\[/\d+?\.\d+?\.\d+?\.\d+?\:\d+?\])', r'\1'),
- (r'\[(\d\d:\d\d:\d\d)\]', r'[\1]'),
- (r'(\[.+? INFO\])', r'\1'),
- (r'(\[.+? WARN\])', r'\1'),
- (r'(\[.+? ERROR\])', r'\1'),
- (r'(\[.+? FATAL\])', r'\1')
+ (r"(\[.+?/INFO\])", r'\1'),
+ (r"(\[.+?/WARN\])", r'\1'),
+ (r"(\[.+?/ERROR\])", r'\1'),
+ (r"(\[.+?/FATAL\])", r'\1'),
+ (
+ r"(\w+?\[/\d+?\.\d+?\.\d+?\.\d+?\:\d+?\])",
+ r'\1',
+ ),
+ (r"\[(\d\d:\d\d:\d\d)\]", r'[\1]'),
+ (r"(\[.+? INFO\])", r'\1'),
+ (r"(\[.+? WARN\])", r'\1'),
+ (r"(\[.+? ERROR\])", r'\1'),
+ (r"(\[.+? FATAL\])", r'\1'),
]
# highlight users keywords
for keyword in user_keywords:
# pylint: disable=consider-using-f-string
- search_replace = (r'({})'.format(keyword), r'\1')
+ search_replace = (
+ r"({})".format(keyword),
+ r'\1',
+ )
replacements.append(search_replace)
for old, new in replacements:
@@ -282,9 +306,8 @@ class Helpers:
return line
-
def validate_traversal(self, base_path, filename):
- logger.debug(f"Validating traversal (\"{base_path}\", \"{filename}\")")
+ logger.debug(f'Validating traversal ("{base_path}", "{filename}")')
base = pathlib.Path(base_path).resolve()
file = pathlib.Path(filename)
fileabs = base.joinpath(file).resolve()
@@ -294,7 +317,6 @@ class Helpers:
else:
raise ValueError("Path traversal detected")
-
def tail_file(self, file_name, number_lines=20):
if not self.check_file_exists(file_name):
logger.warning(f"Unable to find file to tail: {file_name}")
@@ -307,7 +329,7 @@ class Helpers:
line_buffer = number_lines * avg_line_length
# open our file
- with open(file_name, 'r', encoding='utf-8') as f:
+ with open(file_name, "r", encoding="utf-8") as f:
# seek
f.seek(0, 2)
@@ -316,14 +338,16 @@ class Helpers:
fsize = f.tell()
# set pos @ last n chars (buffer from above = number of lines * avg_line_length)
- f.seek(max(fsize-line_buffer, 0), 0)
+ f.seek(max(fsize - line_buffer, 0), 0)
# read file til the end
try:
lines = f.readlines()
except Exception as e:
- logger.warning(f'Unable to read a line in the file:{file_name} - due to error: {e}')
+ logger.warning(
+ f"Unable to read a line in the file:{file_name} - due to error: {e}"
+ )
# now we are done getting the lines, let's return it
return lines
@@ -332,7 +356,7 @@ class Helpers:
def check_writeable(path: str):
filename = os.path.join(path, "tempfile.txt")
try:
- open(filename, "w", encoding='utf-8').close()
+ open(filename, "w", encoding="utf-8").close()
os.remove(filename)
logger.info(f"{filename} is writable")
@@ -355,31 +379,36 @@ class Helpers:
return False
def unzipFile(self, zip_path):
- new_dir_list = zip_path.split('/')
- new_dir = ''
- for i in range(len(new_dir_list)-1):
+ new_dir_list = zip_path.split("/")
+ new_dir = ""
+ for i in range(len(new_dir_list) - 1):
if i == 0:
new_dir += new_dir_list[i]
else:
- new_dir += '/'+new_dir_list[i]
+ new_dir += "/" + new_dir_list[i]
if helper.check_file_perms(zip_path) and os.path.isfile(zip_path):
helper.ensure_dir_exists(new_dir)
tempDir = tempfile.mkdtemp()
try:
- with zipfile.ZipFile(zip_path, 'r') as zip_ref:
+ with zipfile.ZipFile(zip_path, "r") as zip_ref:
zip_ref.extractall(tempDir)
for i in enumerate(zip_ref.filelist):
- if len(zip_ref.filelist) > 1 or not zip_ref.filelist[i].filename.endswith('/'):
+ if len(zip_ref.filelist) > 1 or not zip_ref.filelist[
+ i
+ ].filename.endswith("/"):
break
full_root_path = tempDir
for item in os.listdir(full_root_path):
try:
- file_helper.move_dir(os.path.join(full_root_path, item), os.path.join(new_dir, item))
+ file_helper.move_dir(
+ os.path.join(full_root_path, item),
+ os.path.join(new_dir, item),
+ )
except Exception as ex:
- logger.error(f'ERROR IN ZIP IMPORT: {ex}')
+ logger.error(f"ERROR IN ZIP IMPORT: {ex}")
except Exception as ex:
print(ex)
else:
@@ -387,8 +416,8 @@ class Helpers:
return
def ensure_logging_setup(self):
- log_file = os.path.join(os.path.curdir, 'logs', 'commander.log')
- session_log_file = os.path.join(os.path.curdir, 'logs', 'session.log')
+ log_file = os.path.join(os.path.curdir, "logs", "commander.log")
+ session_log_file = os.path.join(os.path.curdir, "logs", "session.log")
logger.info("Checking app directory writable")
@@ -402,13 +431,13 @@ class Helpers:
# ensure the log directory is there
try:
with suppress(FileExistsError):
- os.makedirs(os.path.join(self.root_dir, 'logs'))
+ os.makedirs(os.path.join(self.root_dir, "logs"))
except Exception as e:
console.error(f"Failed to make logs directory with error: {e} ")
# ensure the log file is there
try:
- open(log_file, 'a', encoding='utf-8').close()
+ open(log_file, "a", encoding="utf-8").close()
except Exception as e:
console.critical(f"Unable to open log file! {e}")
sys.exit(1)
@@ -426,7 +455,7 @@ class Helpers:
@staticmethod
def calc_percent(source_path, dest_path):
- #calculates percentable of zip from drive. Not with compression. For backups and support logs
+ # calculates percentable of zip from drive. Not with compression. For backups and support logs
source_size = 0
files_count = 0
for path, _dirs, files in os.walk(source_path):
@@ -435,47 +464,41 @@ class Helpers:
source_size += os.stat(fp).st_size
files_count += 1
dest_size = os.path.getsize(str(dest_path))
- percent = round((dest_size/source_size) * 100, 1)
+ percent = round((dest_size / source_size) * 100, 1)
if percent >= 0:
- results = {
- "percent": percent,
- "total_files": files_count
- }
+ results = {"percent": percent, "total_files": files_count}
else:
- results = {
- "percent": 0,
- "total_files": 0
- }
+ results = {"percent": 0, "total_files": 0}
return results
@staticmethod
def check_file_exists(path: str):
- logger.debug(f'Looking for path: {path}')
+ logger.debug(f"Looking for path: {path}")
if os.path.exists(path) and os.path.isfile(path):
- logger.debug(f'Found path: {path}')
+ logger.debug(f"Found path: {path}")
return True
else:
return False
@staticmethod
- def human_readable_file_size(num: int, suffix='B'):
- for unit in ['', 'K', 'M', 'G', 'T', 'P', 'E', 'Z']:
+ def human_readable_file_size(num: int, suffix="B"):
+ for unit in ["", "K", "M", "G", "T", "P", "E", "Z"]:
if abs(num) < 1024.0:
- # pylint: disable=consider-using-f-string
+ # pylint: disable=consider-using-f-string
return "%3.1f%s%s" % (num, unit, suffix)
num /= 1024.0
# pylint: disable=consider-using-f-string
- return "%.1f%s%s" % (num, 'Y', suffix)
+ return "%.1f%s%s" % (num, "Y", suffix)
@staticmethod
def check_path_exists(path: str):
if not path:
return False
- logger.debug(f'Looking for path: {path}')
+ logger.debug(f"Looking for path: {path}")
if os.path.exists(path):
- logger.debug(f'Found path: {path}')
+ logger.debug(f"Found path: {path}")
return True
else:
return False
@@ -483,12 +506,12 @@ class Helpers:
@staticmethod
def get_file_contents(path: str, lines=100):
- contents = ''
+ contents = ""
if os.path.exists(path) and os.path.isfile(path):
try:
- with open(path, 'r', encoding='utf-8') as f:
- for line in (f.readlines() [-lines:]):
+ with open(path, "r", encoding="utf-8") as f:
+ for line in f.readlines()[-lines:]:
contents = contents + line
return contents
@@ -497,7 +520,9 @@ class Helpers:
logger.error(f"Unable to read file: {path}. \n Error: {e}")
return False
else:
- logger.error(f"Unable to read file: {path}. File not found, or isn't a file.")
+ logger.error(
+ f"Unable to read file: {path}. File not found, or isn't a file."
+ )
return False
def create_session_file(self, ignore=False):
@@ -510,30 +535,33 @@ class Helpers:
file_data = self.get_file_contents(self.session_file)
try:
data = json.loads(file_data)
- pid = data.get('pid')
- started = data.get('started')
+ pid = data.get("pid")
+ started = data.get("started")
if psutil.pid_exists(pid):
- console.critical(f"Another Crafty Controller agent seems to be running...\npid: {pid} \nstarted on: {started}")
+ console.critical(
+ f"Another Crafty Controller agent seems to be running...\npid: {pid} \nstarted on: {started}"
+ )
logger.critical("Found running crafty process. Exiting.")
sys.exit(1)
else:
- logger.info("No process found for pid. Assuming crafty crashed. Deleting stale session.lock")
+ logger.info(
+ "No process found for pid. Assuming crafty crashed. Deleting stale session.lock"
+ )
os.remove(self.session_file)
except Exception as e:
logger.error(f"Failed to locate existing session.lock with error: {e} ")
- console.error(f"Failed to locate existing session.lock with error: {e} ")
+ console.error(
+ f"Failed to locate existing session.lock with error: {e} "
+ )
sys.exit(1)
pid = os.getpid()
now = datetime.now()
- session_data = {
- 'pid': pid,
- 'started': now.strftime("%d-%m-%Y, %H:%M:%S")
- }
- with open(self.session_file, 'w', encoding='utf-8') as f:
+ session_data = {"pid": pid, "started": now.strftime("%d-%m-%Y, %H:%M:%S")}
+ with open(self.session_file, "w", encoding="utf-8") as f:
json.dump(session_data, f, indent=True)
# because this is a recursive function, we will return bytes, and set human readable later
@@ -548,26 +576,30 @@ class Helpers:
@staticmethod
def list_dir_by_date(path: str, reverse=False):
- return [str(p) for p in sorted(pathlib.Path(path).iterdir(), key=os.path.getmtime, reverse=reverse)]
+ return [
+ str(p)
+ for p in sorted(
+ pathlib.Path(path).iterdir(), key=os.path.getmtime, reverse=reverse
+ )
+ ]
def get_human_readable_files_sizes(self, paths: list):
sizes = []
for p in paths:
- sizes.append({
- "path": p,
- "size": self.human_readable_file_size(os.stat(p).st_size)
- })
+ sizes.append(
+ {"path": p, "size": self.human_readable_file_size(os.stat(p).st_size)}
+ )
return sizes
@staticmethod
def base64_encode_string(fun_str: str):
- s_bytes = str(fun_str).encode('utf-8')
+ s_bytes = str(fun_str).encode("utf-8")
b64_bytes = base64.encodebytes(s_bytes)
- return b64_bytes.decode('utf-8')
+ return b64_bytes.decode("utf-8")
@staticmethod
def base64_decode_string(fun_str: str):
- s_bytes = str(fun_str).encode('utf-8')
+ s_bytes = str(fun_str).encode("utf-8")
b64_bytes = base64.decodebytes(s_bytes)
return b64_bytes.decode("utf-8")
@@ -666,23 +698,23 @@ class Helpers:
random_generator() = G8sjO2
random_generator(3, abcdef) = adf
"""
- return ''.join(random.choice(chars) for x in range(size))
+ return "".join(random.choice(chars) for x in range(size))
@staticmethod
def is_os_windows():
- if os.name == 'nt':
+ if os.name == "nt":
return True
else:
return False
@staticmethod
def wtol_path(w_path):
- l_path = w_path.replace('\\', '/')
+ l_path = w_path.replace("\\", "/")
return l_path
@staticmethod
def ltow_path(l_path):
- w_path = l_path.replace('/', '\\')
+ w_path = l_path.replace("/", "\\")
return w_path
@staticmethod
@@ -694,10 +726,10 @@ class Helpers:
data = {}
if self.check_file_exists(default_file):
- with open(default_file, 'r', encoding='utf-8') as f:
+ with open(default_file, "r", encoding="utf-8") as f:
data = json.load(f)
- del_json = helper.get_setting('delete_default_json')
+ del_json = helper.get_setting("delete_default_json")
if del_json:
os.remove(default_file)
@@ -714,14 +746,15 @@ class Helpers:
dir_list.append(item)
else:
unsorted_files.append(item)
- file_list = sorted(dir_list, key=str.casefold) + sorted(unsorted_files, key=str.casefold)
+ file_list = sorted(dir_list, key=str.casefold) + sorted(
+ unsorted_files, key=str.casefold
+ )
for raw_filename in file_list:
filename = html.escape(raw_filename)
rel = os.path.join(folder, raw_filename)
dpath = os.path.join(folder, filename)
if os.path.isdir(rel):
- output += \
- f"""
+ output += f"""
\n
@@ -729,8 +762,7 @@ class Helpers:
{filename}
- \n"""\
-
+ \n"""
else:
if filename != "crafty_managed.txt":
output += f""""""\
-
+ file_list = sorted(dir_list, key=str.casefold) + sorted(
+ unsorted_files, key=str.casefold
+ )
+ output += f""""""
for raw_filename in file_list:
filename = html.escape(raw_filename)
dpath = os.path.join(folder, filename)
rel = os.path.join(folder, raw_filename)
if os.path.isdir(rel):
- output += \
- f"""-
+ output += f"""
-
\n
{filename}
-
- """\
-
+
- """
else:
if filename != "crafty_managed.txt":
output += f"""
- {filename}
"""
- output += '
\n'
+ output += "\n"
return output
@staticmethod
def generate_zip_tree(folder, output=""):
file_list = os.listdir(folder)
file_list = sorted(file_list, key=str.casefold)
- output += \
- f"""