diff --git a/src/AptSourceActions.py b/src/AptSourceActions.py index 811c096..aa83047 100755 --- a/src/AptSourceActions.py +++ b/src/AptSourceActions.py @@ -127,26 +127,67 @@ def is_safe_sources(sources_text): if not sources_text or not str(sources_text).strip(): return False - blacklisted_terms = [ - "trusted=yes", "trusted=true", - "allow-insecure", - "signed-by=", - "file://", "copy://", "cdrom://" - ] - - for line in str(sources_text).splitlines(): - - line = line.strip() + allowed_signed_by_prefixes = ( + "/usr/share/keyrings/", + "/etc/apt/keyrings/", + ) + allowed_option_keys = { + "arch", + "lang", + "target", + "pdiffs", + "by-hash", + "signed-by", + "check-valid-until", + "check-date", + } + + for raw_line in str(sources_text).splitlines(): + line = raw_line.strip() if not line or line.startswith("#"): continue - if not re.match(r"^deb(-src)?\s+", line): + m = re.match(r"^deb(-src)?\s+(?:\[(.*?)\]\s+)?([^\s]+)", line) + if not m: + return False + + options = m.group(2) + uri = m.group(3).lower() + + if not (uri.startswith("http://") or uri.startswith("https://")): return False - compact_line = line.lower().replace(" ", "") - for term in blacklisted_terms: - if term.replace(" ", "") in compact_line: - return False + if options: + chunks = re.split(r"[,\s]+(?=[A-Za-z0-9-]+=)", options.strip()) + for chunk in chunks: + chunk = chunk.strip() + if not chunk: + continue + if "=" in chunk: + key, value = chunk.split("=", 1) + else: + key, value = chunk, "" + key = key.lower().strip() + value_l = value.lower().strip() + + if key not in allowed_option_keys: + return False + if key in ("check-valid-until", "check-date"): + if value_l in ("false", "no", "0", "off", "disable", "disabled"): + return False + continue + if key == "signed-by": + for part in value.split(","): + part = part.strip() + if not part: + return False + path = os.path.normpath(part) + if not os.path.isabs(path): + return False + if ".." in path.split(os.sep): + return False + if not any(path.startswith(prefix) for prefix in allowed_signed_by_prefixes): + return False return True diff --git a/src/Group.py b/src/Group.py index e3df5d7..945ba7d 100755 --- a/src/Group.py +++ b/src/Group.py @@ -6,24 +6,60 @@ @author: fatih """ +import os +import pwd +import re +import shutil import subprocess import sys +USERNAME_REGEX = re.compile(r"^[a-zA-Z0-9_.][a-zA-Z0-9_.-]*\$?$") + + +def validate_username(username): + if not username or not USERNAME_REGEX.match(username): + sys.stderr.write(f"Error: Invalid username format: '{username}'.\n") + return False + try: + pwd.getpwnam(username) + return True + except KeyError: + sys.stderr.write(f"Error: User '{username}' does not exist.\n") + return False + def main(): - def addtogroup(user): - subprocess.call(["adduser", user, "pardus-update"]) + if len(sys.argv) < 3: + sys.stderr.write("Usage: Group.py \n") + sys.exit(1) + + action = sys.argv[1] + user = sys.argv[2] + + if not validate_username(user): + sys.exit(1) - def delfromgroup(user): - subprocess.call(["deluser", user, "pardus-update"]) + target_group = "pardus-update" - if len(sys.argv) > 1: - if sys.argv[1] == "add": - addtogroup(sys.argv[2]) - elif sys.argv[1] == "del": - delfromgroup(sys.argv[2]) + if action == "add": + cmd_name = "adduser" + cmd_path = shutil.which(cmd_name) or (f"/usr/sbin/{cmd_name}" if os.path.exists(f"/usr/sbin/{cmd_name}") else None) + if not cmd_path: + sys.stderr.write(f"Error: Command '{cmd_name}' not found on system.\n") + sys.exit(1) + rc = subprocess.call([cmd_path, user, target_group]) + sys.exit(rc) + elif action == "del": + cmd_name = "deluser" + cmd_path = shutil.which(cmd_name) or (f"/usr/sbin/{cmd_name}" if os.path.exists(f"/usr/sbin/{cmd_name}") else None) + if not cmd_path: + sys.stderr.write(f"Error: Command '{cmd_name}' not found on system.\n") + sys.exit(1) + rc = subprocess.call([cmd_path, user, target_group]) + sys.exit(rc) else: - print("no argument passed") + sys.stderr.write(f"Error: Unknown action '{action}'. Use 'add' or 'del'.\n") + sys.exit(1) if __name__ == "__main__": diff --git a/src/SysActions.py b/src/SysActions.py index 940ec6a..53a59dc 100755 --- a/src/SysActions.py +++ b/src/SysActions.py @@ -11,6 +11,7 @@ import re import subprocess import sys +import tempfile from pathlib import Path from shutil import rmtree @@ -132,16 +133,45 @@ def removeauto(): subprocess.call(["apt", "autoremove", "-yq"], env={**os.environ, 'DEBIAN_FRONTEND': 'noninteractive'}) + def write_temp_sources_list(sources_content): + run_dir = "/run/pardus-update" + if os.path.isdir("/run"): + try: + os.makedirs(run_dir, mode=0o700, exist_ok=True) + tmp_path = os.path.join(run_dir, "tmp-sources.list") + if os.path.islink(tmp_path): + os.unlink(tmp_path) + with open(tmp_path, "w") as f: + f.write(sources_content) + f.flush() + try: + os.chmod(tmp_path, 0o600) + except OSError: + pass + return tmp_path + except OSError: + pass + + fd, tmp_path = tempfile.mkstemp(prefix="pardus-update-sources-", suffix=".list") + with os.fdopen(fd, "w") as f: + f.write(sources_content) + f.flush() + return tmp_path + + def cleanup_temp_sources_list(tmp_path): + if tmp_path and os.path.exists(tmp_path): + try: + os.unlink(tmp_path) + except OSError: + pass + def controldistupgrade(sourceslist): if not is_safe_sources(sourceslist): print("Malicious apt source detected. Execution aborted.", file=sys.stderr) sys.exit(1) - sfile = open("/tmp/tmp-sources.list", "w") - sfile.write(sourceslist) - sfile.flush() - sfile.close() + tmp_sources_path = write_temp_sources_list(sourceslist) rc_file = os.path.dirname(os.path.abspath(__file__)) + "/../required_changes_for_upgrade.json" @@ -162,149 +192,151 @@ def controldistupgrade(sourceslist): old_sources_list = apt_pkg.config.find("Dir::Etc::sourcelist") old_sources_list_d = apt_pkg.config.find("Dir::Etc::sourceparts") old_cleanup = apt_pkg.config.find("APT::List-Cleanup") - apt_pkg.init_config() - apt_pkg.config.set("Dir::Etc::sourcelist", os.path.abspath("/tmp/tmp-sources.list")) - apt_pkg.config.set("Dir::Etc::sourceparts", "xxx") - apt_pkg.config.set("APT::List-Cleanup", "0") - apt_pkg.init_system() - cache = apt.Cache() - cache.update() - cache.open() - try: - cache.upgrade(True) - cache_error = False - except Exception as error: - print("cache.upgrade Error: {}".format(error)) - update_cache_error_msg = "{}".format(error) - rcu["cache_error_msg"] = update_cache_error_msg - - for kp in keep_list: - try: - cache[kp].mark_keep() - except Exception as e: - print("{} not found".format(kp)) - print("{}".format(e)) - - changes = cache.get_changes() - print(changes) - if changes: - changes_available = True - for package in changes: - if package.is_installed: - if package.marked_upgrade: - to_upgrade.append(package.name) - elif package.marked_delete: - to_delete.append(package.name) - elif package.marked_install: - to_install.append(package.name) - else: - changes_available = False + apt_pkg.init_config() + apt_pkg.config.set("Dir::Etc::sourcelist", os.path.abspath(tmp_sources_path)) + apt_pkg.config.set("Dir::Etc::sourceparts", "xxx") + apt_pkg.config.set("APT::List-Cleanup", "0") + apt_pkg.init_system() + cache = apt.Cache() + cache.update() + cache.open() - download_size = cache.required_download - space = cache.required_space - if space < 0: - freed_size = space * -1 - install_size = 0 - else: - freed_size = 0 - install_size = space - - if cache.keep_count > 0: - upgradable_cache_packages = [pkg.name for pkg in cache if pkg.is_upgradable] - upgradable_changes_packages = [pkg.name for pkg in changes if pkg.is_upgradable] - to_keep = list(set(upgradable_cache_packages).difference(set(upgradable_changes_packages))) - - to_upgrade = sorted(to_upgrade) - to_install = sorted(to_install) - to_delete = sorted(to_delete) - to_keep = sorted(to_keep) - - rcu["download_size"] = download_size - rcu["freed_size"] = freed_size - rcu["install_size"] = install_size - # rcu["to_upgrade"] = to_upgrade - # rcu["to_install"] = to_install - # rcu["to_delete"] = to_delete - # rcu["to_keep"] = to_keep - rcu["changes_available"] = changes_available - rcu["cache_error"] = cache_error - - def summary(packagename): - package = cache.get(packagename) - if package is None: return "" - try: - return package.candidate.summary - except AttributeError: - sum = package.versions.get(0) - return sum.summary if hasattr(sum, "summary") else "Summary is not found" - - def candidate_version(packagename): - package = cache[packagename] - if package is None: return "" try: - version = package.candidate.version - except: + cache.upgrade(True) + cache_error = False + except Exception as error: + print("cache.upgrade Error: {}".format(error)) + update_cache_error_msg = "{}".format(error) + rcu["cache_error_msg"] = update_cache_error_msg + + for kp in keep_list: + try: + cache[kp].mark_keep() + except Exception as e: + print("{} not found".format(kp)) + print("{}".format(e)) + + changes = cache.get_changes() + print(changes) + if changes: + changes_available = True + for package in changes: + if package.is_installed: + if package.marked_upgrade: + to_upgrade.append(package.name) + elif package.marked_delete: + to_delete.append(package.name) + elif package.marked_install: + to_install.append(package.name) + else: + changes_available = False + + download_size = cache.required_download + space = cache.required_space + if space < 0: + freed_size = space * -1 + install_size = 0 + else: + freed_size = 0 + install_size = space + + if cache.keep_count > 0: + upgradable_cache_packages = [pkg.name for pkg in cache if pkg.is_upgradable] + upgradable_changes_packages = [pkg.name for pkg in changes if pkg.is_upgradable] + to_keep = list(set(upgradable_cache_packages).difference(set(upgradable_changes_packages))) + + to_upgrade = sorted(to_upgrade) + to_install = sorted(to_install) + to_delete = sorted(to_delete) + to_keep = sorted(to_keep) + + rcu["download_size"] = download_size + rcu["freed_size"] = freed_size + rcu["install_size"] = install_size + # rcu["to_upgrade"] = to_upgrade + # rcu["to_install"] = to_install + # rcu["to_delete"] = to_delete + # rcu["to_keep"] = to_keep + rcu["changes_available"] = changes_available + rcu["cache_error"] = cache_error + + def summary(packagename): + package = cache.get(packagename) + if package is None: return "" try: - version = package.versions[0].version + return package.candidate.summary + except AttributeError: + sum = package.versions.get(0) + return sum.summary if hasattr(sum, "summary") else "Summary is not found" + + def candidate_version(packagename): + package = cache[packagename] + if package is None: return "" + try: + version = package.candidate.version + except: + try: + version = package.versions[0].version + except: + version = "" + return version + + def installed_version(packagename): + package = cache[packagename] + if package is None: return "" + try: + version = package.installed.version except: version = "" - return version - - def installed_version(packagename): - package = cache[packagename] - if package is None: return "" - try: - version = package.installed.version - except: - version = "" - return version - - to_install_list = [] - for package in to_install: - to_install_list.append({"name": package, "oldversion": installed_version(package), - "newversion": candidate_version(package), "summary": summary(package)}) - - to_upgrade_list = [] - for package in to_upgrade: - to_upgrade_list.append({"name": package, "oldversion": installed_version(package), - "newversion": candidate_version(package), "summary": summary(package)}) - - to_delete_list = [] - for package in to_delete: - to_delete_list.append({"name": package, "oldversion": installed_version(package), - "newversion": candidate_version(package), "summary": summary(package)}) - - to_keep_list = [] - for package in to_keep: - to_keep_list.append({"name": package, "oldversion": installed_version(package), - "newversion": candidate_version(package), "summary": summary(package)}) - - rcu["to_upgrade"] = to_upgrade_list - rcu["to_install"] = to_install_list - rcu["to_delete"] = to_delete_list - rcu["to_keep"] = to_keep_list - - # print("freed_size {}".format(rcu["freed_size"])) - # print("download_size {}".format(rcu["download_size"])) - # print("install_size {}".format(rcu["install_size"])) - # print("to_upgrade {}".format(rcu["to_upgrade"])) - # print("to_install {}".format(rcu["to_install"])) - # print("to_delete {}".format(rcu["to_delete"])) - # print("to_keep {}".format(rcu["to_keep"])) - # print("changes_available {}".format(rcu["changes_available"])) - # print("cache_error {}".format(rcu["cache_error"])) - - print(rcu) - - changes_file = open(rc_file, "w") - json.dump(rcu, changes_file, indent=2) - changes_file.flush() - changes_file.close() - - apt_pkg.config.set("Dir::Etc::sourcelist", old_sources_list) - apt_pkg.config.set("Dir::Etc::sourceparts", old_sources_list_d) - apt_pkg.config.set("APT::List-Cleanup", old_cleanup) + return version + + to_install_list = [] + for package in to_install: + to_install_list.append({"name": package, "oldversion": installed_version(package), + "newversion": candidate_version(package), "summary": summary(package)}) + + to_upgrade_list = [] + for package in to_upgrade: + to_upgrade_list.append({"name": package, "oldversion": installed_version(package), + "newversion": candidate_version(package), "summary": summary(package)}) + + to_delete_list = [] + for package in to_delete: + to_delete_list.append({"name": package, "oldversion": installed_version(package), + "newversion": candidate_version(package), "summary": summary(package)}) + + to_keep_list = [] + for package in to_keep: + to_keep_list.append({"name": package, "oldversion": installed_version(package), + "newversion": candidate_version(package), "summary": summary(package)}) + + rcu["to_upgrade"] = to_upgrade_list + rcu["to_install"] = to_install_list + rcu["to_delete"] = to_delete_list + rcu["to_keep"] = to_keep_list + + # print("freed_size {}".format(rcu["freed_size"])) + # print("download_size {}".format(rcu["download_size"])) + # print("install_size {}".format(rcu["install_size"])) + # print("to_upgrade {}".format(rcu["to_upgrade"])) + # print("to_install {}".format(rcu["to_install"])) + # print("to_delete {}".format(rcu["to_delete"])) + # print("to_keep {}".format(rcu["to_keep"])) + # print("changes_available {}".format(rcu["changes_available"])) + # print("cache_error {}".format(rcu["cache_error"])) + + print(rcu) + + changes_file = open(rc_file, "w") + json.dump(rcu, changes_file, indent=2) + changes_file.flush() + changes_file.close() + finally: + cleanup_temp_sources_list(tmp_sources_path) + apt_pkg.config.set("Dir::Etc::sourcelist", old_sources_list) + apt_pkg.config.set("Dir::Etc::sourceparts", old_sources_list_d) + apt_pkg.config.set("APT::List-Cleanup", old_cleanup) def downupgrade(sourceslist): @@ -314,33 +346,33 @@ def downupgrade(sourceslist): aptclean() - sfile = open("/tmp/tmp-sources.list", "w") - sfile.write(sourceslist) - sfile.flush() - sfile.close() - - apt_pkg.init_config() - apt_pkg.config.set("Dir::Etc::sourcelist", os.path.abspath("/tmp/tmp-sources.list")) - apt_pkg.config.set("Dir::Etc::sourceparts", "xxx") - apt_pkg.config.set("APT::List-Cleanup", "0") - apt_pkg.init_system() - cache = apt.Cache() - cache.update() - cache.open() + tmp_sources_path = write_temp_sources_list(sourceslist) try: - cache.upgrade(True) - except Exception as error: - print("cache.upgrade Error: {}".format(error)) + apt_pkg.init_config() + apt_pkg.config.set("Dir::Etc::sourcelist", os.path.abspath(tmp_sources_path)) + apt_pkg.config.set("Dir::Etc::sourceparts", "xxx") + apt_pkg.config.set("APT::List-Cleanup", "0") + apt_pkg.init_system() + cache = apt.Cache() + cache.update() + cache.open() - for kp in keep_list: try: - cache[kp].mark_keep() - except Exception as e: - print("{} not found".format(kp)) - print("{}".format(e)) + cache.upgrade(True) + except Exception as error: + print("cache.upgrade Error: {}".format(error)) - cache.fetch_archives() + for kp in keep_list: + try: + cache[kp].mark_keep() + except Exception as e: + print("{} not found".format(kp)) + print("{}".format(e)) + + cache.fetch_archives() + finally: + cleanup_temp_sources_list(tmp_sources_path) # subprocess.call(["apt", "full-upgrade", "-yqd"], # env={**os.environ, 'DEBIAN_FRONTEND': 'noninteractive'}) @@ -494,26 +526,67 @@ def is_safe_sources(sources_text): if not sources_text or not str(sources_text).strip(): return False - blacklisted_terms = [ - "trusted=yes", "trusted=true", - "allow-insecure", - "signed-by=", - "file://", "copy://", "cdrom://" - ] - - for line in str(sources_text).splitlines(): - - line = line.strip() + allowed_signed_by_prefixes = ( + "/usr/share/keyrings/", + "/etc/apt/keyrings/", + ) + allowed_option_keys = { + "arch", + "lang", + "target", + "pdiffs", + "by-hash", + "signed-by", + "check-valid-until", + "check-date", + } + + for raw_line in str(sources_text).splitlines(): + line = raw_line.strip() if not line or line.startswith("#"): continue - if not re.match(r"^deb(-src)?\s+", line): + m = re.match(r"^deb(-src)?\s+(?:\[(.*?)\]\s+)?([^\s]+)", line) + if not m: + return False + + options = m.group(2) + uri = m.group(3).lower() + + if not (uri.startswith("http://") or uri.startswith("https://")): return False - compact_line = line.lower().replace(" ", "") - for term in blacklisted_terms: - if term.replace(" ", "") in compact_line: - return False + if options: + chunks = re.split(r"[,\s]+(?=[A-Za-z0-9-]+=)", options.strip()) + for chunk in chunks: + chunk = chunk.strip() + if not chunk: + continue + if "=" in chunk: + key, value = chunk.split("=", 1) + else: + key, value = chunk, "" + key = key.lower().strip() + value_l = value.lower().strip() + + if key not in allowed_option_keys: + return False + if key in ("check-valid-until", "check-date"): + if value_l in ("false", "no", "0", "off", "disable", "disabled"): + return False + continue + if key == "signed-by": + for part in value.split(","): + part = part.strip() + if not part: + return False + path = os.path.normpath(part) + if not os.path.isabs(path): + return False + if ".." in path.split(os.sep): + return False + if not any(path.startswith(prefix) for prefix in allowed_signed_by_prefixes): + return False return True