evgeny/modoboa-installer
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Default gid & uid for folder creation

Browse Source
This commit is contained in:
Spitap
2025-11-01 17:41:02 +01:00
parent a81cd4196c
commit 39e78649aa
1 changed files with 79 additions and 70 deletions
Download Patch File Download Diff File Expand all files Collapse all files

149
modoboa_installer/utils.py
View File

@@ -24,7 +24,6 @@ BLACK, RED, GREEN, YELLOW, BLUE, MAGENTA, CYAN, WHITE = range(8)
class FatalError(Exception): class FatalError(Exception):
"""A simple exception.""" """A simple exception."""
pass pass
@@ -77,16 +76,14 @@ def dist_info():
info = {} info = {}
with open(path) as fp: with open(path) as fp:
for line in fp.readlines(): for line in fp.readlines():
if line == '\n': if line == "\n":
continue continue
key, value = line.split("=") key, value = line.split("=")
value = value.rstrip('"\n') value = value.rstrip('"\n')
value = value.strip('"') value = value.strip('"')
info[key] = value info[key] = value
return info["NAME"], info["VERSION_ID"] return info["NAME"], info["VERSION_ID"]
printcolor( printcolor("Failed to retrieve information about your system, aborting.", RED)
"Failed to retrieve information about your system, aborting.",
RED)
sys.exit(1) sys.exit(1)
@@ -99,14 +96,27 @@ def is_dist_debian_based() -> (bool, str):
"""Check if current OS is Debian based or not.""" """Check if current OS is Debian based or not."""
status, codename = exec_cmd("lsb_release -c -s") status, codename = exec_cmd("lsb_release -c -s")
codename = codename.decode().strip().lower() codename = codename.decode().strip().lower()
return codename in [ return (
"bionic", "bookworm", "bullseye", "buster", codename
"focal", "jammy", "jessie", "sid", "stretch", in [
"trusty", "wheezy", "xenial" "bionic",
], codename "bookworm",
"bullseye",
"buster",
"focal",
"jammy",
"jessie",
"sid",
"stretch",
"trusty",
"wheezy",
"xenial",
],
codename,
)
def mkdir(path, mode, uid, gid): def mkdir(path, mode=0o777, uid=os.getuid(), gid=os.getgid()):
"""Create a directory.""" """Create a directory."""
if not os.path.exists(path): if not os.path.exists(path):
os.mkdir(path, mode) os.mkdir(path, mode)
@@ -115,7 +125,7 @@ def mkdir(path, mode, uid, gid):
os.chown(path, uid, gid) os.chown(path, uid, gid)
def mkdir_safe(path, mode, uid, gid): def mkdir_safe(path, mode=0o777, uid=os.getuid(), gid=os.getgid()):
"""Create a directory. Safe way (-p)""" """Create a directory. Safe way (-p)"""
if not os.path.exists(path): if not os.path.exists(path):
os.makedirs(os.path.abspath(path), mode) os.makedirs(os.path.abspath(path), mode)
@@ -125,8 +135,9 @@ def mkdir_safe(path, mode, uid, gid):
def make_password(length=16): def make_password(length=16):
"""Create a random password.""" """Create a random password."""
return "".join( return "".join(
random.SystemRandom().choice( random.SystemRandom().choice(string.ascii_letters + string.digits)
string.ascii_letters + string.digits) for _ in range(length)) for _ in range(length)
)
@contextlib.contextmanager @contextlib.contextmanager
@@ -149,8 +160,7 @@ def backup_file(fname):
"""Create a backup of a given file.""" """Create a backup of a given file."""
for f in glob.glob("{}.old.*".format(fname)): for f in glob.glob("{}.old.*".format(fname)):
os.unlink(f) os.unlink(f)
bak_name = "{}.old.{}".format( bak_name = "{}.old.{}".format(fname, datetime.datetime.now().isoformat())
fname, datetime.datetime.now().isoformat())
shutil.copy(fname, bak_name) shutil.copy(fname, bak_name)
@@ -171,17 +181,13 @@ def copy_from_template(template, dest, context):
if os.path.isfile(dest): if os.path.isfile(dest):
backup_file(dest) backup_file(dest)
with open(dest, "w") as fp: with open(dest, "w") as fp:
fp.write( fp.write("# This file was automatically installed on {}\n".format(now))
"# This file was automatically installed on {}\n"
.format(now))
fp.write(ConfigFileTemplate(buf).substitute(context)) fp.write(ConfigFileTemplate(buf).substitute(context))
def check_config_file(dest, def check_config_file(
interactive=False, dest, interactive=False, upgrade=False, backup=False, restore=False
upgrade=False, ):
backup=False,
restore=False):
"""Create a new installer config file if needed.""" """Create a new installer config file if needed."""
is_present = True is_present = True
if os.path.exists(dest): if os.path.exists(dest):
@@ -189,22 +195,25 @@ def check_config_file(dest,
if upgrade: if upgrade:
error( error(
"You cannot upgrade an existing installation without a " "You cannot upgrade an existing installation without a "
"configuration file.") "configuration file."
)
sys.exit(1) sys.exit(1)
elif backup: elif backup:
is_present = False is_present = False
error( error(
"Your configuration file hasn't been found. A new one will be generated. " "Your configuration file hasn't been found. A new one will be generated. "
"Please edit it with correct password for the databases !") "Please edit it with correct password for the databases !"
)
elif restore: elif restore:
error( error(
"You cannot restore an existing installation without a " "You cannot restore an existing installation without a "
f"configuration file. (file : {dest} has not been found...") f"configuration file. (file : {dest} has not been found..."
)
sys.exit(1) sys.exit(1)
printcolor( printcolor(
"Configuration file {} not found, creating new one." "Configuration file {} not found, creating new one.".format(dest), YELLOW
.format(dest), YELLOW) )
gen_config(dest, interactive) gen_config(dest, interactive)
return is_present, None return is_present, None
@@ -217,6 +226,7 @@ def has_colours(stream):
return False # auto color only on TTYs return False # auto color only on TTYs
try: try:
import curses import curses
curses.setupterm() curses.setupterm()
return curses.tigetnum("colors") > 2 return curses.tigetnum("colors") > 2
except: except:
@@ -251,8 +261,9 @@ def convert_version_to_int(version):
numbers = [int(number_string) for number_string in version.split(".")] numbers = [int(number_string) for number_string in version.split(".")]
if len(numbers) > len(number_bits): if len(numbers) > len(number_bits):
raise NotImplementedError( raise NotImplementedError(
"Versions with more than {0} decimal places are not supported" "Versions with more than {0} decimal places are not supported".format(
.format(len(number_bits) - 1) len(number_bits) - 1
)
) )
# add 0s for missing numbers # add 0s for missing numbers
numbers.extend([0] * (len(number_bits) - len(numbers))) numbers.extend([0] * (len(number_bits) - len(numbers)))
@@ -263,8 +274,9 @@ def convert_version_to_int(version):
max_num = (bits + 1) - 1 max_num = (bits + 1) - 1
if num >= 1 << max_num: if num >= 1 << max_num:
raise ValueError( raise ValueError(
"Number {0} cannot be stored with only {1} bits. Max is {2}" "Number {0} cannot be stored with only {1} bits. Max is {2}".format(
.format(num, bits, max_num) num, bits, max_num
)
) )
number += num << total_bits number += num << total_bits
total_bits += bits total_bits += bits
@@ -315,7 +327,9 @@ def validate(value, config_entry):
return True return True
def get_entry_value(entry: dict, interactive: bool, config: configparser.ConfigParser) -> string: def get_entry_value(
entry: dict, interactive: bool, config: configparser.ConfigParser
) -> string:
default_entry = entry["default"] default_entry = entry["default"]
if type(default_entry) is type(list()): if type(default_entry) is type(list()):
default_value = str(check_if_condition(config, default_entry)).lower() default_value = str(check_if_condition(config, default_entry)).lower()
@@ -325,7 +339,7 @@ def get_entry_value(entry: dict, interactive: bool, config: configparser.ConfigP
default_value = default_entry default_value = default_entry
user_value = None user_value = None
if entry.get("customizable") and interactive: if entry.get("customizable") and interactive:
while (user_value != '' and not validate(user_value, entry)): while user_value != "" and not validate(user_value, entry):
question = entry.get("question") question = entry.get("question")
if entry.get("values"): if entry.get("values"):
question += " from the list" question += " from the list"
@@ -366,14 +380,10 @@ def load_config_template(interactive):
config.add_section(section["name"]) config.add_section(section["name"])
for config_entry in section["values"]: for config_entry in section["values"]:
if config_entry.get("if") is not None: if config_entry.get("if") is not None:
interactive_section = (interactive_section and interactive_section = interactive_section and check_if_condition(
check_if_condition( config, config_entry["if"]
config, config_entry["if"] )
) value = get_entry_value(config_entry, interactive_section, config)
)
value = get_entry_value(config_entry,
interactive_section,
config)
config.set(section["name"], config_entry["option"], value) config.set(section["name"], config_entry["option"], value)
return config return config
@@ -393,10 +403,11 @@ def update_config(path, apply_update=True):
dropped_sections = list(set(old_sections) - set(new_sections)) dropped_sections = list(set(old_sections) - set(new_sections))
added_sections = list(set(new_sections) - set(old_sections)) added_sections = list(set(new_sections) - set(old_sections))
if len(dropped_sections) > 0 and apply_update: if len(dropped_sections) > 0 and apply_update:
printcolor("Following section(s) will not be ported " printcolor(
"due to being deleted or renamed: " + "Following section(s) will not be ported "
', '.join(dropped_sections), "due to being deleted or renamed: " + ", ".join(dropped_sections),
RED) RED,
)
if len(dropped_sections) + len(added_sections) > 0: if len(dropped_sections) + len(added_sections) > 0:
update = True update = True
@@ -409,11 +420,12 @@ def update_config(path, apply_update=True):
dropped_options = list(set(old_options) - set(new_options)) dropped_options = list(set(old_options) - set(new_options))
added_options = list(set(new_options) - set(old_options)) added_options = list(set(new_options) - set(old_options))
if len(dropped_options) > 0 and apply_update: if len(dropped_options) > 0 and apply_update:
printcolor(f"Following option(s) from section: {section}, " printcolor(
"will not be ported due to being " f"Following option(s) from section: {section}, "
"deleted or renamed: " + "will not be ported due to being "
', '.join(dropped_options), "deleted or renamed: " + ", ".join(dropped_options),
RED) RED,
)
if len(dropped_options) + len(added_options) > 0: if len(dropped_options) + len(added_options) > 0:
update = True update = True
@@ -466,29 +478,27 @@ def validate_backup_path(path: str, silent_mode: bool):
"""Check if provided backup path is valid or not.""" """Check if provided backup path is valid or not."""
path_exists = os.path.exists(path) path_exists = os.path.exists(path)
if path_exists and os.path.isfile(path): if path_exists and os.path.isfile(path):
printcolor( printcolor("Error, you provided a file instead of a directory!", RED)
"Error, you provided a file instead of a directory!", RED)
return None return None
if not path_exists: if not path_exists:
if not silent_mode: if not silent_mode:
create_dir = input( create_dir = input(
f"\"{path}\" doesn't exist, would you like to create it? [y/N]\n" f'"{path}" doesn\'t exist, would you like to create it? [y/N]\n'
).lower() ).lower()
if silent_mode or (not silent_mode and create_dir.startswith("y")): if silent_mode or (not silent_mode and create_dir.startswith("y")):
pw = pwd.getpwnam("root") pw = pwd.getpwnam("root")
mkdir_safe(path, stat.S_IRWXU | stat.S_IRWXG, pw[2], pw[3]) mkdir_safe(path, stat.S_IRWXU | stat.S_IRWXG, pw[2], pw[3])
else: else:
printcolor( printcolor("Error, backup directory not present.", RED)
"Error, backup directory not present.", RED
)
return None return None
if len(os.listdir(path)) != 0: if len(os.listdir(path)) != 0:
if not silent_mode: if not silent_mode:
delete_dir = input( delete_dir = input(
"Warning: backup directory is not empty, it will be purged if you continue... [y/N]\n").lower() "Warning: backup directory is not empty, it will be purged if you continue... [y/N]\n"
).lower()
if silent_mode or (not silent_mode and delete_dir.startswith("y")): if silent_mode or (not silent_mode and delete_dir.startswith("y")):
try: try:
@@ -496,22 +506,19 @@ def validate_backup_path(path: str, silent_mode: bool):
except FileNotFoundError: except FileNotFoundError:
pass pass
shutil.rmtree(os.path.join(path, "custom"), shutil.rmtree(os.path.join(path, "custom"), ignore_errors=False)
ignore_errors=False)
shutil.rmtree(os.path.join(path, "mails"), ignore_errors=False) shutil.rmtree(os.path.join(path, "mails"), ignore_errors=False)
shutil.rmtree(os.path.join(path, "databases"), shutil.rmtree(os.path.join(path, "databases"), ignore_errors=False)
ignore_errors=False)
else: else:
printcolor( printcolor("Error: backup directory not clean.", RED)
"Error: backup directory not clean.", RED
)
return None return None
backup_path = path backup_path = path
pw = pwd.getpwnam("root") pw = pwd.getpwnam("root")
for dir in ["custom/", "databases/"]: for dir in ["custom/", "databases/"]:
mkdir_safe(os.path.join(backup_path, dir), mkdir_safe(
stat.S_IRWXU | stat.S_IRWXG, pw[2], pw[3]) os.path.join(backup_path, dir), stat.S_IRWXU | stat.S_IRWXG, pw[2], pw[3]
)
return backup_path return backup_path
@@ -539,7 +546,9 @@ def check_app_compatibility(section, config):
if section in APP_INCOMPATIBILITY.keys(): if section in APP_INCOMPATIBILITY.keys():
for app in APP_INCOMPATIBILITY[section]: for app in APP_INCOMPATIBILITY[section]:
if config.getboolean(app, "enabled"): if config.getboolean(app, "enabled"):
error(f"{section} cannot be installed if {app} is enabled. " error(
"Please disable one of them.") f"{section} cannot be installed if {app} is enabled. "
"Please disable one of them."
)
incompatible_app.append(app) incompatible_app.append(app)
return len(incompatible_app) == 0 return len(incompatible_app) == 0