From 6f18765b62075e02104736d8bb519ea007888d1f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Sevban=20D=C3=B6nmez?= <82449360+byjanke@users.noreply.github.com> Date: Fri, 4 Sep 2026 23:19:39 +0300 Subject: [PATCH 1/4] fix: enforce strict apt sources validation, secure temporary files and harden group management (CWE-184, CWE-377, CWE-250) --- src/AptSourceActions.py | 44 ++++++++---- src/Group.py | 56 ++++++++++++--- src/SysActions.py | 152 ++++++++++++++++++++++++++-------------- 3 files changed, 178 insertions(+), 74 deletions(-) diff --git a/src/AptSourceActions.py b/src/AptSourceActions.py index 811c096..32202be 100755 --- a/src/AptSourceActions.py +++ b/src/AptSourceActions.py @@ -127,25 +127,43 @@ 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() + 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): + # Must start with deb or deb-src followed by optional [options] and URI + m = re.match(r"^deb(-src)?\s+(?:\[(.*?)\]\s+)?([^\s]+)", line) + if not m: + return False + + options = m.group(2) + uri = m.group(3).lower() + + # 1. URI must strictly use http:// or https:// (reject file:, copy:, cdrom:, etc.) + if not (uri.startswith("http://") or uri.startswith("https://")): return False + # 2. If options are present [options], verify that no security-weakening options exist + if options: + opt_compact = options.lower().replace(" ", "") + disallowed_keywords = [ + "trusted", + "allow-insecure", + "allow-downgrade", + "signed-by", + "check-date", + "check-valid-until" + ] + for kw in disallowed_keywords: + if kw in opt_compact: + return False + + # 3. Double-check entire line for forbidden protocols or bypass attempts compact_line = line.lower().replace(" ", "") - for term in blacklisted_terms: - if term.replace(" ", "") in compact_line: + forbidden_schemes = ["file:", "copy:", "cdrom:"] + for scheme in forbidden_schemes: + if scheme in compact_line: 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..deec71f 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,14 +192,15 @@ 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: + 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() try: cache.upgrade(True) @@ -301,10 +332,11 @@ def installed_version(packagename): 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) + 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,25 +526,43 @@ 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() + 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): + # Must start with deb or deb-src followed by optional [options] and URI + m = re.match(r"^deb(-src)?\s+(?:\[(.*?)\]\s+)?([^\s]+)", line) + if not m: + return False + + options = m.group(2) + uri = m.group(3).lower() + + # 1. URI must strictly use http:// or https:// (reject file:, copy:, cdrom:, etc.) + if not (uri.startswith("http://") or uri.startswith("https://")): return False + # 2. If options are present [options], verify that no security-weakening options exist + if options: + opt_compact = options.lower().replace(" ", "") + disallowed_keywords = [ + "trusted", + "allow-insecure", + "allow-downgrade", + "signed-by", + "check-date", + "check-valid-until" + ] + for kw in disallowed_keywords: + if kw in opt_compact: + return False + + # 3. Double-check entire line for forbidden protocols or bypass attempts compact_line = line.lower().replace(" ", "") - for term in blacklisted_terms: - if term.replace(" ", "") in compact_line: + forbidden_schemes = ["file:", "copy:", "cdrom:"] + for scheme in forbidden_schemes: + if scheme in compact_line: return False return True From 9b9fa6f5479a89d610edcf0f663e2788b5769c58 Mon Sep 17 00:00:00 2001 From: jankesec Date: Sat, 5 Sep 2026 00:31:10 +0300 Subject: [PATCH 2/4] fix: indent controldistupgrade try/finally so SysActions.py compiles The outer try around APT cache init was closed before the existing cache.upgrade try, which is a SyntaxError. Nest the upgrade/report body inside the same try so the finally always cleans the temp sources list. --- src/SysActions.py | 248 +++++++++++++++++++++++----------------------- 1 file changed, 124 insertions(+), 124 deletions(-) diff --git a/src/SysActions.py b/src/SysActions.py index deec71f..49b1095 100755 --- a/src/SysActions.py +++ b/src/SysActions.py @@ -202,136 +202,136 @@ def controldistupgrade(sourceslist): 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)) + 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 - 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 + for kp in keep_list: + try: + cache[kp].mark_keep() + except Exception as e: + print("{} not found".format(kp)) + print("{}".format(e)) - 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: + 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: + 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.versions[0].version + 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() + 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) From 2f27296303a23103b455caeae6bafa88345762dd Mon Sep 17 00:00:00 2001 From: jankesec Date: Sat, 5 Sep 2026 00:37:50 +0300 Subject: [PATCH 3/4] fix: allow official signed-by and arch options in sources.list checks Reject only security-weakening options (trusted=*, allow-insecure, check-valid-until=false, attacker-controlled signed-by paths). Keep http/https URIs, [arch=...], and signed-by under /usr/share/keyrings or /etc/apt/keyrings so dist-upgrade of stock Pardus/Debian lines is unchanged. --- src/AptSourceActions.py | 44 +++++++++++++++++++++++++---------------- src/SysActions.py | 44 +++++++++++++++++++++++++---------------- 2 files changed, 54 insertions(+), 34 deletions(-) diff --git a/src/AptSourceActions.py b/src/AptSourceActions.py index 32202be..6c2649d 100755 --- a/src/AptSourceActions.py +++ b/src/AptSourceActions.py @@ -127,12 +127,16 @@ def is_safe_sources(sources_text): if not sources_text or not str(sources_text).strip(): return False + allowed_signed_by_prefixes = ( + "/usr/share/keyrings/", + "/etc/apt/keyrings/", + ) + for raw_line in str(sources_text).splitlines(): line = raw_line.strip() if not line or line.startswith("#"): continue - # Must start with deb or deb-src followed by optional [options] and URI m = re.match(r"^deb(-src)?\s+(?:\[(.*?)\]\s+)?([^\s]+)", line) if not m: return False @@ -140,29 +144,35 @@ def is_safe_sources(sources_text): options = m.group(2) uri = m.group(3).lower() - # 1. URI must strictly use http:// or https:// (reject file:, copy:, cdrom:, etc.) if not (uri.startswith("http://") or uri.startswith("https://")): return False - # 2. If options are present [options], verify that no security-weakening options exist if options: - opt_compact = options.lower().replace(" ", "") - disallowed_keywords = [ - "trusted", - "allow-insecure", - "allow-downgrade", - "signed-by", - "check-date", - "check-valid-until" - ] - for kw in disallowed_keywords: - if kw in opt_compact: + for token in options.replace(",", " ").split(): + if "=" in token: + key, value = token.split("=", 1) + else: + key, value = token, "" + key = key.lower() + value_l = value.lower() + + if key == "trusted": + return False + if key.startswith("allow-insecure") or key.startswith("allow-weak") or key.startswith("allow-downgrade"): 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": + path = os.path.normpath(value) + if ".." in path.split(os.sep): + return False + if not any(path.startswith(prefix) for prefix in allowed_signed_by_prefixes): + return False - # 3. Double-check entire line for forbidden protocols or bypass attempts compact_line = line.lower().replace(" ", "") - forbidden_schemes = ["file:", "copy:", "cdrom:"] - for scheme in forbidden_schemes: + for scheme in ("file:", "copy:", "cdrom:"): if scheme in compact_line: return False diff --git a/src/SysActions.py b/src/SysActions.py index 49b1095..5b459fd 100755 --- a/src/SysActions.py +++ b/src/SysActions.py @@ -526,12 +526,16 @@ def is_safe_sources(sources_text): if not sources_text or not str(sources_text).strip(): return False + allowed_signed_by_prefixes = ( + "/usr/share/keyrings/", + "/etc/apt/keyrings/", + ) + for raw_line in str(sources_text).splitlines(): line = raw_line.strip() if not line or line.startswith("#"): continue - # Must start with deb or deb-src followed by optional [options] and URI m = re.match(r"^deb(-src)?\s+(?:\[(.*?)\]\s+)?([^\s]+)", line) if not m: return False @@ -539,29 +543,35 @@ def is_safe_sources(sources_text): options = m.group(2) uri = m.group(3).lower() - # 1. URI must strictly use http:// or https:// (reject file:, copy:, cdrom:, etc.) if not (uri.startswith("http://") or uri.startswith("https://")): return False - # 2. If options are present [options], verify that no security-weakening options exist if options: - opt_compact = options.lower().replace(" ", "") - disallowed_keywords = [ - "trusted", - "allow-insecure", - "allow-downgrade", - "signed-by", - "check-date", - "check-valid-until" - ] - for kw in disallowed_keywords: - if kw in opt_compact: + for token in options.replace(",", " ").split(): + if "=" in token: + key, value = token.split("=", 1) + else: + key, value = token, "" + key = key.lower() + value_l = value.lower() + + if key == "trusted": + return False + if key.startswith("allow-insecure") or key.startswith("allow-weak") or key.startswith("allow-downgrade"): 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": + path = os.path.normpath(value) + if ".." in path.split(os.sep): + return False + if not any(path.startswith(prefix) for prefix in allowed_signed_by_prefixes): + return False - # 3. Double-check entire line for forbidden protocols or bypass attempts compact_line = line.lower().replace(" ", "") - forbidden_schemes = ["file:", "copy:", "cdrom:"] - for scheme in forbidden_schemes: + for scheme in ("file:", "copy:", "cdrom:"): if scheme in compact_line: return False From 02dbb56a0228acd1258a3673d29eea7e80650b6e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Sevban=20D=C3=B6nmez?= <82449360+byjanke@users.noreply.github.com> Date: Sat, 5 Sep 2026 01:49:01 +0300 Subject: [PATCH 4/4] fix: reject signed-by path lists and unknown apt source options Comma-splitting options treated extra signed-by paths as unknown tokens and allowed them. Parse option boundaries so each signed-by path must stay under /usr/share/keyrings or /etc/apt/keyrings. Unknown keys such as trusted and inrelease-path are denied. --- src/AptSourceActions.py | 51 ++++++++++++++++++++++++++--------------- src/SysActions.py | 51 ++++++++++++++++++++++++++--------------- 2 files changed, 64 insertions(+), 38 deletions(-) diff --git a/src/AptSourceActions.py b/src/AptSourceActions.py index 6c2649d..aa83047 100755 --- a/src/AptSourceActions.py +++ b/src/AptSourceActions.py @@ -131,6 +131,16 @@ def is_safe_sources(sources_text): "/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() @@ -148,33 +158,36 @@ def is_safe_sources(sources_text): return False if options: - for token in options.replace(",", " ").split(): - if "=" in token: - key, value = token.split("=", 1) + 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 = token, "" - key = key.lower() - value_l = value.lower() + key, value = chunk, "" + key = key.lower().strip() + value_l = value.lower().strip() - if key == "trusted": - return False - if key.startswith("allow-insecure") or key.startswith("allow-weak") or key.startswith("allow-downgrade"): + 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": - path = os.path.normpath(value) - if ".." in path.split(os.sep): - return False - if not any(path.startswith(prefix) for prefix in allowed_signed_by_prefixes): - return False - - compact_line = line.lower().replace(" ", "") - for scheme in ("file:", "copy:", "cdrom:"): - if scheme in compact_line: - return False + 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/SysActions.py b/src/SysActions.py index 5b459fd..53a59dc 100755 --- a/src/SysActions.py +++ b/src/SysActions.py @@ -530,6 +530,16 @@ def is_safe_sources(sources_text): "/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() @@ -547,33 +557,36 @@ def is_safe_sources(sources_text): return False if options: - for token in options.replace(",", " ").split(): - if "=" in token: - key, value = token.split("=", 1) + 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 = token, "" - key = key.lower() - value_l = value.lower() + key, value = chunk, "" + key = key.lower().strip() + value_l = value.lower().strip() - if key == "trusted": - return False - if key.startswith("allow-insecure") or key.startswith("allow-weak") or key.startswith("allow-downgrade"): + 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": - path = os.path.normpath(value) - if ".." in path.split(os.sep): - return False - if not any(path.startswith(prefix) for prefix in allowed_signed_by_prefixes): - return False - - compact_line = line.lower().replace(" ", "") - for scheme in ("file:", "copy:", "cdrom:"): - if scheme in compact_line: - return False + 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