From 3de7e302c440e03539d153b053cacc1956f3aa22 Mon Sep 17 00:00:00 2001 From: kloptops Date: Thu, 22 Oct 2026 18:36:52 +0800 Subject: [PATCH] Port info: bad port.json, ports.json and ports_info.json files can't stop PortMaster A hand-written version 3 port.json without "attr.runtime" stopped PortMaster starting (KeyError: 'runtime'). A fuzzer found the rest: - port_info_load hands back the types everything else expects: name, status, source and files are checked and cleaned, min_glibc must be a version, runtimes must be plain file names. - A port.json name with nothing left once cleaned ("/") is found from its items, like a missing one. - ports.json: entries that aren't objects, have no download url, or have bad sections are skipped, not fatal. A string entry is no longer read as a port.json path. Util and runtime names must be file names. - ports_info.json is cleaned when loaded: a port.json is only written to a "file" inside the ports folder, and owners need a usable name. port_stats.json download counts must be numbers. - JSON nested too deeply loads as None instead of raising. - is_safe_port_item also refuses NUL bytes and names too long for the file system; installed screenshots use it. - muOS: the application folder comes from the script's own name, so a "../" in a files key can't reach outside the applications folder. - Text the font can't draw ("\n", zero-width characters, a width narrower than a character) no longer crashes the GUI. tests/test_fuzz_port_json.py fuzzes all of this through startup, sources, zip installs, ports_info.json and script signatures, the CLI and the GUI screens, checking every delete and write stays in the test's tmp dir. A short fixed-seed run is part of the suite; 100,000 cases per path on Python 3.7 and 3.11 found nothing more. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01NxABmR1rUUgi4iicVLZchY --- CLAUDE.md | 2 + PortMaster/pylibs/harbourmaster/hm_info.py | 89 ++- PortMaster/pylibs/harbourmaster/hm_ports.py | 25 +- PortMaster/pylibs/harbourmaster/info.py | 105 ++- .../pylibs/harbourmaster/platform/muos.py | 8 +- .../pylibs/harbourmaster/source/portmaster.py | 49 +- PortMaster/pylibs/harbourmaster/util/data.py | 7 + PortMaster/pylibs/pugwash/sdl/text.py | 27 +- tests/test_fuzz_port_json.py | 713 ++++++++++++++++++ tests/test_gui_screens.py | 19 + tests/test_hm_info.py | 52 ++ tests/test_hm_ports.py | 91 +++ tests/test_info.py | 113 +++ tests/test_platform_hooks.py | 15 + tests/test_source.py | 78 ++ tests/test_util_edges.py | 11 + 16 files changed, 1371 insertions(+), 33 deletions(-) create mode 100644 tests/test_fuzz_port_json.py diff --git a/CLAUDE.md b/CLAUDE.md index cfd2f97b..a2e86ce2 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -58,6 +58,8 @@ Device info comes from `harbourmaster.hardware.HardwareDetector`, which reads th Installing is covered in `test_install.py` (bad data, layouts, rollback, runtimes, URLs, themes, artwork), `test_fifo_install.py` (fifo_control installs must always answer `OKAY`/`FAIL`, never stay on `WAIT`) and `test_gui_commands.py`. +Port info comes from hand-written files, so `info.port_info_load` (port.json and ports.json entries) and `hm_info.clean_ports_info` (ports_info.json) hand back the types everything else expects, dropping what they can't use. `tests/test_fuzz_port_json.py` fuzzes them through startup, sources, zip installs, ports_info.json/script signatures, the CLI and the GUI screens: nothing may raise, and no delete or write may leave the test's tmp dir. A short fixed-seed run is part of the suite; `PM_FUZZ_N=20000 PM_FUZZ_SEED=` runs a long one. + Guard tests for restructuring: - `test_release_layout.py` runs both scripts from a `pylibs.zip` layout with no `.git`, which is the on-device bootstrap path. - `test_api_snapshot.py` checks that every public name in `tests/data/api_snapshot.json` is still reachable. Update its `AREAS` table when code moves. diff --git a/PortMaster/pylibs/harbourmaster/hm_info.py b/PortMaster/pylibs/harbourmaster/hm_info.py index 489e253c..a912cf69 100644 --- a/PortMaster/pylibs/harbourmaster/hm_info.py +++ b/PortMaster/pylibs/harbourmaster/hm_info.py @@ -5,7 +5,80 @@ import zipfile from gettext import gettext as _ from loguru import logger -from .util import calculate_git_blob_sha_data, calculate_git_blob_sha_file, datetime_compare, download, fetch_data, fetch_text, json_safe_load, json_safe_loads +from .info import is_safe_port_item +from .util import calculate_git_blob_sha_data, calculate_git_blob_sha_file, datetime_compare, download, fetch_data, fetch_text, json_safe_load, json_safe_loads, name_cleaner + + +def _is_port_name(name): + ## Ports are looked up by their cleaned name, so there has to be one. + return isinstance(name, str) and name_cleaner(name).strip('.') != "" + + +def _str_or_strs(value, is_valid=lambda item: isinstance(item, str)): + ## "x" or ["x", "y"], as get_dict_list() reads them; None when there's nothing usable. + if isinstance(value, str): + return value if is_valid(value) else None + + if isinstance(value, list): + value = [item for item in value if is_valid(item)] + if len(value) > 0: + return value + + return None + + +def clean_ports_info(data): + """ + ports_info.json as load_ports() reads it. It is downloaded and cached, so a bad copy + (or a bad release of it) loses the entries it can't use instead of stopping PortMaster. + """ + result = {"items": {}, "md5": {}, "ports": {}, "portsmd_fix": {}} + if not isinstance(data, dict): + return result + + ## items: {"Script.sh": "port.zip" or ["port.zip", ...]} + if isinstance(data.get("items"), dict): + for name, value in data["items"].items(): + value = _str_or_strs(value, _is_port_name) + if value is not None: + result["items"][name] = value + + if isinstance(data.get("portsmd_fix"), dict): + for name, value in data["portsmd_fix"].items(): + value = _str_or_strs(value) + if value is not None: + result["portsmd_fix"][name] = value + + if isinstance(data.get("md5"), dict): + result["md5"] = { + md5: name + for md5, name in data["md5"].items() + if isinstance(name, str)} + + if isinstance(data.get("ports"), dict): + for name, port in data["ports"].items(): + ## A port.json is written to "file", and uninstalling deletes the "items". + if (not isinstance(port, dict) or + not isinstance(port.get("items"), list) or + not all(is_safe_port_item(item) for item in port["items"]) or + not is_safe_port_item(port.get("file")) or + not port["file"].endswith("port.json")): + logger.warning(f"ports_info.json: skipping bad port {name!r}") + continue + + port = dict(port) + for port_name in (port.get("name"), name): + if _is_port_name(port_name): + port["name"] = port_name + break + + else: + logger.warning(f"ports_info.json: skipping port with no name {name!r}") + continue + + result["ports"][name] = port + + return result class InfoMixin: @@ -16,10 +89,7 @@ class InfoMixin: def ports_info(self): if self._PORTS_INFO is None: with open(self.cfg_dir / "ports_info.json", 'r') as fh: - self._PORTS_INFO = json_safe_load(fh) - - if self._PORTS_INFO is None: - self._PORTS_INFO = {"items": {}, "md5": {}, "ports": {}, "portsmd_fix": {}} + self._PORTS_INFO = clean_ports_info(json_safe_load(fh)) return self._PORTS_INFO @@ -32,12 +102,17 @@ def port_downloads(self, port_name): with port_stats.open('r') as fh: port_stats_data = json_safe_load(fh) - if isinstance(port_stats_data, dict) and "ports" in port_stats_data: + if isinstance(port_stats_data, dict) and isinstance(port_stats_data.get("ports"), dict): self._PORTS_DOWNLOAD = port_stats_data["ports"] else: self._PORTS_DOWNLOAD = {} - return self._PORTS_DOWNLOAD.get(port_name, 0) + downloads = self._PORTS_DOWNLOAD.get(port_name, 0) + ## Sorted by, so it has to be a number. + if isinstance(downloads, bool) or not isinstance(downloads, int): + return 0 + + return downloads def porters(self): if self._PORTERS is None: diff --git a/PortMaster/pylibs/harbourmaster/hm_ports.py b/PortMaster/pylibs/harbourmaster/hm_ports.py index f799a314..1ce389e7 100644 --- a/PortMaster/pylibs/harbourmaster/hm_ports.py +++ b/PortMaster/pylibs/harbourmaster/hm_ports.py @@ -7,7 +7,7 @@ from . import config as hm_config from .config import HM_SORT_ORDER from .util import PORT_SORT_FUNCS, add_dict_list_unique, add_list_unique, add_pm_signature, get_dict_list, hash_file, load_pm_signature, match_requirements, name_cleaner, oc_join, remove_dict_list, timeit, version_parse -from .info import port_info_load, port_info_merge +from .info import is_safe_port_item, port_info_load, port_info_merge class PortsMixin: @@ -67,6 +67,10 @@ def _load_port_info(self, port_file): changed = False # Its possible for the port_info to be in a bad way, lets try and fix it. + if port_info.get('name', None) is not None and name_cleaner(port_info['name']).strip('.') == "": + ## Nothing left of it once cleaned (eg: "/"). + port_info['name'] = None + if port_info.get('name', None) is None: # No name, check the items and see if it matches our internal database, we can get the port name from a script. logger.error(f"No 'name' in {port_file!r}") @@ -85,6 +89,10 @@ def _load_port_info(self, port_file): logger.error(f"Unable to load {port_file}, unknown items.") return None + if name_cleaner(port_owners[0]).strip('.') == "": + logger.error(f"Unable to load {port_file}, bad port name {port_owners[0]!r} for its items.") + return None + changed = True port_info['name'] = name_cleaner(port_owners[0]) @@ -434,6 +442,11 @@ def load_ports(self): ## Create new port.json files for any new ports, these only contain the most basic of information. for new_port in new_ports: + if new_port not in ports_info['ports']: + ## ports_info.json names a port for this file, but doesn't have the port. + logger.warning(f"Unknown port {new_port!r} in ports_info.json") + continue + port_info_raw = ports_info['ports'][new_port] port_info = port_info_load(port_info_raw) @@ -465,7 +478,8 @@ def load_ports(self): if port_info.get('status', None) is None: port_info['status'] = {} - port_info['name'] = port_info['name'].casefold() + ## As _load_port_info() does, so it is found by the cleaned name. + port_info['name'] = name_cleaner(port_info['name']) port_info['status']['source'] = "Unknown" port_info['status']['md5'] = None @@ -953,12 +967,11 @@ def _installed_port_images(self, port_name): candidates += ['screenshot.png', 'screenshot.jpg'] for candidate in candidates: - candidate_path = Path(candidate) - - ## Stay inside the port's own directory. - if candidate_path.is_absolute() or '..' in candidate_path.parts: + ## Stay inside the port's own directory (and a name the file system takes). + if not is_safe_port_item(candidate): continue + candidate_path = Path(candidate) if (port_dir / candidate_path).is_file(): return {'screenshot': port_dir / candidate_path} diff --git a/PortMaster/pylibs/harbourmaster/info.py b/PortMaster/pylibs/harbourmaster/info.py index ab39b4d0..1ac2839b 100644 --- a/PortMaster/pylibs/harbourmaster/info.py +++ b/PortMaster/pylibs/harbourmaster/info.py @@ -3,6 +3,7 @@ # System imports import pathlib +import re from pathlib import Path @@ -51,6 +52,14 @@ PORT_INFO_LIST_ATTRS = ('genres', 'porter', 'runtime', 'reqs', 'arch') PORT_INFO_TEXT_ATTRS = ('title', 'desc', 'inst', 'min_glibc') +## ports.json's "source" block: dropped when they aren't these types, so the defaults are used. +PORT_INFO_SOURCE_TEXT = ('md5', 'url', 'date_added', 'date_updated', 'release_id') +PORT_INFO_SOURCE_NUMBERS = ('size', 'downloads') +PORT_INFO_STATUS_TEXT = ('source', 'status') + +## The longest file name most file systems allow, in bytes. +NAME_MAX = 255 + def _clean_attr_types(attr, source_name): """ @@ -94,6 +103,85 @@ def _clean_attr_types(attr, source_name): for key in ('rtr', 'exp'): attr[key] = bool(attr.get(key, False)) + ## Compared with the device's glibc version, so it has to be one: "2.31". + if not re.fullmatch(r'\d+(\.\d+)*', attr['min_glibc'].strip()): + if attr['min_glibc'].strip() != "": + logger.warning(f"port_info['attr']['min_glibc'] is not a version in {source_name!r}: {attr['min_glibc']!r}") + + attr['min_glibc'] = "" + + ## Runtimes are file names in the libs folder. + attr['runtime'] = [runtime for runtime in attr['runtime'] if is_safe_file_name(runtime)] + + +def is_safe_file_name(name): + """A plain file name, with no folders in it.""" + return ( + isinstance(name, str) and name.strip() not in ("", ".", "..") and + not any(c in name for c in '/\\\0') and + len(name.encode('utf-8', 'surrogatepass')) <= NAME_MAX) + + +def _clean_root_types(port_info, source_name): + """ + The same for the root keys: everything that reads port info expects these types, + and a port.json that isn't right stopped PortMaster starting. + """ + if not isinstance(port_info['name'], str) or port_info['name'].strip() == "": + if port_info['name'] is not None: + logger.warning(f"port_info['name'] has a bad value in {source_name!r}: {port_info['name']!r}") + + port_info['name'] = None + + status = port_info.get('status') + if 'status' in port_info and not isinstance(status, dict): + logger.warning(f"port_info['status'] has a bad value in {source_name!r}: {status!r}") + del port_info['status'] + + elif isinstance(status, dict): + status = port_info['status'] = dict(status) + for key in PORT_INFO_STATUS_TEXT: + if key in status and not isinstance(status[key], str): + del status[key] + + if not isinstance(status.get('md5'), str): + status['md5'] = None + + source = port_info.get('source') + if 'source' in port_info and not isinstance(source, dict): + logger.warning(f"port_info['source'] has a bad value in {source_name!r}: {source!r}") + del port_info['source'] + + elif isinstance(source, dict): + source = port_info['source'] = dict(source) + for key in PORT_INFO_SOURCE_TEXT: + if key in source and not isinstance(source[key], str): + del source[key] + + for key in PORT_INFO_SOURCE_NUMBERS: + if key in source and (isinstance(source[key], bool) or not isinstance(source[key], int) or source[key] < 0): + del source[key] + + ## {"port.json": "port/port.json", item: item or [items]}: uninstalling deletes the values. + files = port_info.get('files') + if 'files' in port_info and files is not None: + if not isinstance(files, dict): + logger.warning(f"port_info['files'] has a bad value in {source_name!r}: {files!r}") + files = {} + + clean_files = {} + for key, value in files.items(): + values = value if isinstance(value, list) else [value] + values = [item for item in values if is_safe_port_item(item)] + + if not is_safe_port_item(key) or len(values) == 0: + logger.warning(f"port_info['files'] has a bad entry in {source_name!r}: {key!r}: {value!r}") + continue + + clean_files[key] = values[0] if len(values) == 1 else values + + port_info['files'] = clean_files + def is_safe_port_item(item): """ @@ -103,10 +191,11 @@ def is_safe_port_item(item): if not isinstance(item, str) or item.strip() in ("", ".", "./"): return False - if item.startswith(('/', '\\')): + if item.startswith(('/', '\\')) or '\0' in item: return False - return not any(part == '..' for part in item.replace('\\', '/').split('/')) + parts = item.replace('\\', '/').split('/') + return not any(part == '..' or len(part.encode('utf-8', 'surrogatepass')) > NAME_MAX for part in parts) def port_info_load(raw_info, source_name=None, do_default=False): @@ -227,11 +316,13 @@ def port_info_load(raw_info, source_name=None, do_default=False): if info.get('version', None) == 3: info['version'] = 4 - if info['attr']['runtime'] is None: - info['attr']['runtime'] = [] + ## Hand-written port.json files leave out "attr" or its "runtime"; filled in below. + if 'runtime' in info.get('attr', {}): + if info['attr']['runtime'] is None: + info['attr']['runtime'] = [] - elif isinstance(info['attr']['runtime'], str): - info['attr']['runtime'] = [ info['attr']['runtime'] ] + elif isinstance(info['attr']['runtime'], str): + info['attr']['runtime'] = [ info['attr']['runtime'] ] # Catch errors because who fucking uses version numbers. if info.get('attr', {}).get('runtime', []) == None: @@ -290,6 +381,8 @@ def port_info_load(raw_info, source_name=None, do_default=False): if genre.casefold() in HM_GENRES: port_info['attr']['genres'].append(genre.casefold()) + _clean_root_types(port_info, source_name) + return port_info diff --git a/PortMaster/pylibs/harbourmaster/platform/muos.py b/PortMaster/pylibs/harbourmaster/platform/muos.py index e3765bfb..fc59f06c 100644 --- a/PortMaster/pylibs/harbourmaster/platform/muos.py +++ b/PortMaster/pylibs/harbourmaster/platform/muos.py @@ -127,8 +127,8 @@ def port_install(self, port_name, port_info, port_files, fix_perm_files): APP_NAME = port_info['name'].rsplit('.', 1)[0] for port_file in port_info['files']: if port_file.lower().endswith('.sh'): - # Use the first bash script we find. - APP_NAME = port_file.rsplit('.', 1)[0] + # Use the first bash script we find. (Only its name: it is a folder in MUOSAPP_DIR.) + APP_NAME = port_file.replace('\\', '/').rsplit('/', 1)[-1].rsplit('.', 1)[0] break MUOSAPP_DIR = Path("/run/muos/storage/application") @@ -175,8 +175,8 @@ def port_uninstall(self, port_name, port_info, port_files): APP_NAME = port_info['name'].rsplit('.', 1)[0] for port_file in port_info['files']: if port_file.lower().endswith('.sh'): - # Use the first bash script we find. - APP_NAME = port_file.rsplit('.', 1)[0] + # Use the first bash script we find. (Only its name: it is a folder in MUOSAPP_DIR.) + APP_NAME = port_file.replace('\\', '/').rsplit('/', 1)[-1].rsplit('.', 1)[0] break MUOSAPP_DIR = Path("/run/muos/storage/application") diff --git a/PortMaster/pylibs/harbourmaster/source/portmaster.py b/PortMaster/pylibs/harbourmaster/source/portmaster.py index 32f023df..8a45e0bf 100644 --- a/PortMaster/pylibs/harbourmaster/source/portmaster.py +++ b/PortMaster/pylibs/harbourmaster/source/portmaster.py @@ -8,7 +8,7 @@ from gettext import gettext as _ from loguru import logger from ..config import HM_MAX_TEMP_SIZE -from ..info import port_info_load, port_info_merge +from ..info import is_safe_file_name, port_info_load, port_info_merge from ..util import json_safe_load from ..util import net from .base import BaseSource @@ -59,17 +59,36 @@ def update(self): self.utils = [] self.images = {} + ## A ports.json from any source, so nothing in it is trusted to be the right type. + ports = data.get('ports') if isinstance(data, dict) else None + utils = data.get('utils') if isinstance(data, dict) else None + if not isinstance(ports, dict) or not isinstance(utils, dict): + logger.error(f"Bad ports.json from {self._config['url']}") + ports = ports if isinstance(ports, dict) else {} + utils = utils if isinstance(utils, dict) else {} + ## Load data from the assets. - for key, asset in data['ports'].items(): - asset = port_info_load(asset) + for key, asset in ports.items(): + if not isinstance(asset, dict) or not is_safe_file_name(key): + logger.warning(f"Skipping bad port {key!r} in ports.json") + continue + + asset = port_info_load(asset, source_name=f"ports.json: {key}") if asset is None: ## Skip bad items. continue + if 'url' not in asset.get('source', {}): + logger.warning(f"Skipping port {key!r} in ports.json: no download url") + continue + + if asset['name'] is None: + asset['name'] = key + result = { 'name': asset['name'], - 'size': asset['source']['size'], - 'md5': asset['source']['md5'], + 'size': asset['source'].get('size', 0), + 'md5': asset['source'].get('md5'), 'url': asset['source']['url'], } @@ -77,11 +96,19 @@ def update(self): self._info[self.clean_name(key)] = asset self._data[self.clean_name(key)] = result - for key, asset in data['utils'].items(): + for key, asset in utils.items(): + ## Runtimes are saved in the libs folder by these names. + if (not isinstance(asset, dict) or not is_safe_file_name(key) or + not isinstance(asset.get('url'), str) or not isinstance(asset.get('name', ''), str)): + logger.warning(f"Skipping bad util {key!r} in ports.json") + continue + + size = asset.get('size', 0) + md5 = asset.get('md5') result = { - 'name': asset['name'], - 'size': asset['size'], - 'md5': asset['md5'], + 'name': asset.get('name', key), + 'size': size if isinstance(size, int) and not isinstance(size, bool) and size >= 0 else 0, + 'md5': md5 if isinstance(md5, str) else None, 'url': asset['url'], } @@ -90,6 +117,10 @@ def update(self): if key.endswith('.squashfs'): if 'runtime_name' in asset: + if not is_safe_file_name(asset['runtime_name']) or not isinstance(asset.get('runtime_arch'), str): + logger.warning(f"Skipping bad runtime {key!r} in ports.json") + continue + key=asset['runtime_name'] arch=asset['runtime_arch'] self.hm.runtimes_info.setdefault(key, {}).setdefault('remote', {})[arch] = result.copy() diff --git a/PortMaster/pylibs/harbourmaster/util/data.py b/PortMaster/pylibs/harbourmaster/util/data.py index 36f1c179..229e6e0c 100644 --- a/PortMaster/pylibs/harbourmaster/util/data.py +++ b/PortMaster/pylibs/harbourmaster/util/data.py @@ -15,6 +15,10 @@ def json_safe_loads(*args): except json.JSONDecodeError as err: logger.error(f"Unable to load json_data {err.doc}:{err.pos}:{err.lineno}:{err.colno}") return None + except RecursionError: + ## Nested too deep for the json module: [[[[[... + logger.error("Unable to load json_data: nested too deeply") + return None def json_safe_load(*args): @@ -27,6 +31,9 @@ def json_safe_load(*args): ## Not even text (UnicodeDecodeError): a file cut off or scrambled on the SD card. logger.error(f"Unable to load json_data: {err}") return None + except RecursionError: + logger.error("Unable to load json_data: nested too deeply") + return None def json_load_dict(file_name, default=None): diff --git a/PortMaster/pylibs/pugwash/sdl/text.py b/PortMaster/pylibs/pugwash/sdl/text.py index becea948..0cbd5367 100644 --- a/PortMaster/pylibs/pugwash/sdl/text.py +++ b/PortMaster/pylibs/pugwash/sdl/text.py @@ -4,14 +4,20 @@ # Fonts and rendered text. import collections +import unicodedata from ctypes import c_int, byref import sdl2 import sdl2.ext import sdl2.sdlmixer +from loguru import logger from .errors import GUIValueError from .images import Texture +## Control, format (zero width), surrogate and unassigned characters: fonts draw nothing for them. +UNDRAWABLE = ('Cc', 'Cf', 'Cs', 'Co', 'Cn') + + def get_text_size(font, text=''): ''' Calculate the size of given text using the given font, or if @@ -55,7 +61,26 @@ def quick_render(self, text, size, width=None, align='left', line_h=None): if text == "": text = " " - return self.render_text(text, style_key, width=width, align=align, line_h=line_h) + try: + return self.render_text(text, style_key, width=width, align=align, line_h=line_h) + + except Exception as err: + ## sdl2.ext fails on text with nothing it can draw ("\n", "\u200b"), or a width narrower + ## than a character. Port info comes from anyone's port.json, so draw what we can. + logger.warning(f"Unable to render {text[:40]!r}: {err}") + + text = ''.join(c for c in text if c == '\n' or unicodedata.category(c) not in UNDRAWABLE).strip('\n') + if text.strip() == "": + text = " " + + for try_width in (width, None): + try: + return self.render_text(text, style_key, width=try_width, align=align, line_h=line_h) + + except Exception: + pass + + return self.render_text(" ", style_key, align=align, line_h=line_h) class TextManager: diff --git a/tests/test_fuzz_port_json.py b/tests/test_fuzz_port_json.py new file mode 100644 index 00000000..0c457e59 --- /dev/null +++ b/tests/test_fuzz_port_json.py @@ -0,0 +1,713 @@ +# SPDX-License-Identifier: MIT +# +# A fuzzer for the files that describe ports. Each case mutates real port info (wrong types, +# missing keys, unsafe names, deep nesting, mangled bytes) and pushes it through the ways it +# reaches PortMaster: +# 1. port.json files of installed ports, loaded at startup +# 2. ports.json entries from a source +# 3. port.json inside a port zip being installed +# 4. ports_info.json, port_stats.json and the scripts' PORTMASTER: signatures +# 5. the GUI's ports list and port info screens +# then lists, sorts, filters, prints (the CLI) and uninstalls what loaded. Nothing may raise, +# and every delete, write, rename and mkdir must stay inside the test's tmp dir. +# +# A short run with a fixed seed is part of the suite. For a long one: +# PM_FUZZ_N=20000 PM_FUZZ_SEED=7 .venv/bin/pytest -q tests/test_fuzz_port_json.py + +import json +import os +import random +import shutil +import traceback +from collections import defaultdict +from pathlib import Path + +import pytest + +from harbourmaster.config import HM_SORT_ORDER +from harbourmaster.util import net as hm_net + +from conftest import DATA_DIR, basic_port_files, write_port_zip + + +## Cases per test (the GUI test does a tenth, at least 20). +N = int(os.environ.get('PM_FUZZ_N', '40')) +SEED = int(os.environ.get('PM_FUZZ_SEED', '1')) + +CRASHES = defaultdict(list) +ESCAPES = [] + + +class SandboxEscape(BaseException): + ## Not an Exception, so harbourmaster's own `except Exception:` can't hide it. + pass + + +################################################################################ +## Values +def weird_strings(outside): + return [ + "", " ", "x", "blank", "None", "null", ".", "..", "../", "../..", "a/../..", "./", "/", + str(outside), str(outside) + "/", "../outside", "../outside/canary.txt", "ports/../../outside", + "\x00", "a\x00b", "\n", "\t\r", "é", "日本語", "🎮", "\udcff"[:0] + "‮", "%s%n", "{0}", "{port_info.title}", + "x" * 5000, "Test Port.sh", "testport/", "testport", "TESTPORT.ZIP", "testport.zip", "game.zip", + "mono.squashfs", "godot", ".squashfs", "2.31", "99.99", "1.2.3.4.5", "abc", "-1", "nan", + "action", "ACTION", "rtr", "!rtr", "installed", "broken", "aarch64", "x86_64|armhf", + "", "", "x", "", "<", ">", "<<>>", "", " 2: + return rng.choice(pool) + + kind = rng.random() + if kind < 0.55: + return rng.choice(pool) + + if kind < 0.8: + return [weird_values(rng, outside, depth + 1) for _ in range(rng.randint(0, 4))] + + if kind < 0.97: + keys = ['md5', 'source', 'status', 'name', 'screenshot', 'covers', 'url', 'size', 'date_added', + 'date_updated', 'port.json', 'x', '', 'runtime', 'title'] + return {rng.choice(keys): weird_values(rng, outside, depth + 1) for _ in range(rng.randint(0, 4))} + + ## Deep nesting. + value = rng.choice(pool) + for _ in range(rng.randint(10, 200)): + value = [value] if rng.random() < 0.5 else {'x': value} + return value + + +ROOT_KEYS = ['version', 'name', 'items', 'items_opt', 'attr', 'status', 'files', 'source', 'md5', 'changed'] +ATTR_KEYS = ['title', 'desc', 'desc_md', 'inst', 'inst_md', 'genres', 'porter', 'image', 'rtr', 'exp', + 'runtime', 'reqs', 'arch', 'min_glibc', 'store', 'availability'] + + +def mutate(rng, info, outside): + info = json.loads(json.dumps(info)) + + for _ in range(rng.randint(1, 4)): + where = rng.random() + if where < 0.3: + target, keys = info, ROOT_KEYS + else: + if not isinstance(info.get('attr'), dict): + info['attr'] = {} + target, keys = info['attr'], ATTR_KEYS + + key = rng.choice(keys) + action = rng.random() + if action < 0.15: + target.pop(key, None) + + elif action < 0.3 and isinstance(target.get(key), list) and target[key]: + ## Change one element of a list. + target[key][rng.randrange(len(target[key]))] = weird_values(rng, outside, 2) + + elif action < 0.4 and isinstance(target.get(key), dict) and target[key]: + sub = rng.choice(list(target[key])) + target[key][sub] = weird_values(rng, outside, 2) + + else: + target[key] = weird_values(rng, outside) + + return info + + +def to_bytes(rng, info): + """The port.json file's bytes, sometimes mangled the way an SD card or an editor would.""" + try: + text = json.dumps(info, indent=rng.choice([None, 2])) + except (ValueError, RecursionError): + text = "{}" + + data = text.encode('utf-8', 'surrogatepass') + roll = rng.random() + if roll < 0.80: + return data + if roll < 0.84: + return b'\xef\xbb\xbf' + data + if roll < 0.87: + return data[:rng.randrange(len(data) + 1)] + if roll < 0.90: + data = bytearray(data) + for _ in range(rng.randint(1, 5)): + if data: + data[rng.randrange(len(data))] = rng.randrange(256) + return bytes(data) + if roll < 0.92: + return text.encode('utf-16') + if roll < 0.94: + return text.encode('latin-1', 'replace').replace(b'"', b"'") + return rng.choice([b'', b'null', b'[]', b'"x"', b'1', b'NaN', b'{"attr": NaN}', b'\x00' * 64, + b'[' * 100000, b'{"a":' * 50000, b'{}{}', b'{"name": "a", "name": 1}']) + + +################################################################################ +## Harness +@pytest.fixture +def guard(tmp_path, monkeypatch): + """Every delete must stay inside tmp_path.""" + root = tmp_path.resolve() + + def check(path): + path = Path(os.path.abspath(os.fspath(path))) + if root != path and root not in path.parents: + ESCAPES.append((str(path), traceback.format_stack(limit=8))) + raise SandboxEscape(f"OUTSIDE SANDBOX: {path}") + + real_rmtree, real_remove, real_unlink, real_rmdir = shutil.rmtree, os.remove, os.unlink, os.rmdir + real_path_unlink, real_path_rmdir = Path.unlink, Path.rmdir + + def rmtree(path, *a, **kw): + check(path) + return real_rmtree(path, *a, **kw) + + def remove(path, *a, **kw): + if kw.get('dir_fd') is None: + ## (With dir_fd it is rmtree inside a tree already checked.) + check(path) + return real_remove(path, *a, **kw) + + def unlink(path, *a, **kw): + if kw.get('dir_fd') is None: + ## (With dir_fd it is rmtree inside a tree already checked.) + check(path) + return real_unlink(path, *a, **kw) + + def rmdir(path, *a, **kw): + if kw.get('dir_fd') is None: + ## (With dir_fd it is rmtree inside a tree already checked.) + check(path) + return real_rmdir(path, *a, **kw) + + def path_unlink(self, *a, **kw): + check(self) + return real_path_unlink(self, *a, **kw) + + def path_rmdir(self, *a, **kw): + check(self) + return real_path_rmdir(self, *a, **kw) + + import builtins + import io + real_open, real_rename, real_replace, real_mkdir = builtins.open, os.rename, os.replace, os.mkdir + + def guarded_open(file, mode='r', *a, **kw): + if isinstance(file, (str, bytes, os.PathLike)) and any(c in mode for c in 'wax+'): + check(file) + return real_open(file, mode, *a, **kw) + + def rename(src, dst, *a, **kw): + if kw.get('src_dir_fd') is None: + check(src) + check(dst) + return real_rename(src, dst, *a, **kw) + + def replace(src, dst, *a, **kw): + if kw.get('src_dir_fd') is None: + check(src) + check(dst) + return real_replace(src, dst, *a, **kw) + + def mkdir(path, *a, **kw): + if kw.get('dir_fd') is None: + check(path) + return real_mkdir(path, *a, **kw) + + monkeypatch.setattr(builtins, 'open', guarded_open) + monkeypatch.setattr(io, 'open', guarded_open) + monkeypatch.setattr(os, 'rename', rename) + monkeypatch.setattr(os, 'replace', replace) + monkeypatch.setattr(os, 'mkdir', mkdir) + monkeypatch.setattr(shutil, 'rmtree', rmtree) + monkeypatch.setattr(os, 'remove', remove) + monkeypatch.setattr(os, 'unlink', unlink) + monkeypatch.setattr(os, 'rmdir', rmdir) + monkeypatch.setattr(Path, 'unlink', path_unlink) + monkeypatch.setattr(Path, 'rmdir', path_rmdir) + + +@pytest.fixture(autouse=True) +def no_crashes(): + """Fails the test with a summary of every distinct crash and sandbox escape.""" + CRASHES.clear() + ESCAPES.clear() + yield + + lines = [f"{len(ESCAPES)} file operations outside the sandbox, {len(CRASHES)} distinct crashes (N={N}, seed={SEED})"] + for path, stack in ESCAPES[:5]: + lines.append(f"OUTSIDE SANDBOX: {path}\n" + "".join(stack)) + + for key in sorted(CRASHES): + lines.append(f"{len(CRASHES[key]):5d} {key}\n e.g. {CRASHES[key][0]!r}"[:600]) + + assert not ESCAPES and not CRASHES, "\n".join(lines) + + +def where(err): + frames = traceback.extract_tb(err.__traceback__) + ours = [f for f in frames if '/pylibs/' in f.filename] + frame = (ours or frames)[-1] + return f"{type(err).__name__} @ {frame.filename.split('/pylibs/')[-1]}:{frame.lineno} ({frame.name}): {str(err)[:80]}" + + +def record(stage, err, case): + key = f"[{stage}] {where(err)}" + CRASHES[key].append(case) + + +def run_cli(hm, command, argv): + import contextlib + import io + from harbourmaster import cli + + with contextlib.redirect_stdout(io.StringIO()): + getattr(cli, command)(hm, argv) + + +def exercise(hm, names, stage, case): + """Everything the GUI and CLI do with loaded ports.""" + portsjson = str(hm.temp_dir / "ports-out.json") + steps = [ + ('cli_list', lambda: run_cli(hm, 'do_list', [])), + ('cli_ports', lambda: run_cli(hm, 'do_ports', [])), + ('cli_portsmd', lambda: run_cli(hm, 'do_portsmd', [])), + ('cli_portsjson', lambda: run_cli(hm, 'do_portsjson', [portsjson])), + ('cli_runtime_list', lambda: run_cli(hm, 'do_runtime_list', [])), + ('list_ports', lambda: [hm.list_ports_new(filters=f, sort_by=s) for f in ([], ['installed'], ['rtr'], ['!rtr'], ['action'], ['update available'], ['broken']) for s in HM_SORT_ORDER]), + ('list_ports_names', lambda: hm.list_ports_names_new(['installed'])), + ('list_ports_old', lambda: [hm.list_ports(filters=f, sort_by=s) for f in ([], ['installed']) for s in HM_SORT_ORDER]), + ('list_utils', lambda: hm.list_utils()), + ] + for name in names: + steps += [ + ('port_info', lambda name=name: hm.port_info(name)), + ('port_info_installed', lambda name=name: hm.port_info(name, installed=True)), + ('portmd', lambda name=name: hm.port_info(name) and hm.portmd(hm.port_info(name))), + ('match_requirements', lambda name=name: hm.port_info(name) and hm.match_requirements(hm.port_info(name))), + ('port_images', lambda name=name: hm.port_images(name)), + ('port_download_size', lambda name=name: hm.port_download_size(name)), + ('port_download_url', lambda name=name: hm.port_download_url(name)), + ] + + for step, fn in steps: + try: + fn() + except (AssertionError, SandboxEscape): + raise + except Exception as err: + record(f"{stage}:{step}", err, case) + + +def outside_intact(outside): + assert (outside / "canary.txt").is_file(), "canary outside the ports dir was deleted" + assert (outside / "dir" / "keep.txt").is_file(), "canary dir outside the ports dir was deleted" + + +def make_outside(tmp_path): + outside = tmp_path / "outside" + (outside / "dir").mkdir(parents=True, exist_ok=True) + (outside / "canary.txt").write_text("keep") + (outside / "dir" / "keep.txt").write_text("keep") + return outside + + +def base_infos(): + with open(DATA_DIR / "ports.json") as fh: + data = json.load(fh) + + infos = [] + for info in data['ports'].values(): + info = dict(info) + source = info.pop('source', None) + info['status'] = {'source': "PortMaster", 'md5': (source or {}).get('md5'), 'status': "Installed"} + infos.append(info) + + files = basic_port_files() + infos.append(files["testport/port.json"]) + return data, infos + + +################################################################################ +## 1. Installed port.json files, loaded at startup +def test_fuzz_installed(tmp_path, make_hm, hm_dirs, guard): + rng = random.Random(SEED) + _, infos = base_infos() + ports_dir = hm_dirs['ports_dir'] + outside = make_outside(tmp_path) + + for case_no in range(N): + info = mutate(rng, rng.choice(infos), outside) + raw = to_bytes(rng, info) + case = raw[:300] + + shutil.rmtree(ports_dir) + ports_dir.mkdir() + files = basic_port_files() + files.pop("testport/port.json") + for name, data in files.items(): + path = ports_dir / name + if name.endswith('/'): + path.mkdir(parents=True, exist_ok=True) + else: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(data) + + (ports_dir / "testport" / "port.json").write_bytes(raw) + if rng.random() < 0.2: + ## Port folders as symlinks to the outside. + try: + (ports_dir / "linked").symlink_to(outside, target_is_directory=True) + except OSError: + ## No symlinks on this file system (exFAT on a device). + pass + + try: + hm = make_hm() + except (AssertionError, SandboxEscape): + raise + except Exception as err: + record("installed:startup", err, case) + continue + + names = list(hm.installed_ports) + list(hm.broken_ports) + ["testport.zip"] + exercise(hm, names, "installed", case) + + for name in names: + try: + hm.uninstall_port(name) + except AssertionError: + raise + except Exception as err: + record("installed:uninstall", err, case) + + outside_intact(outside) + + try: + hm.load_ports() + except Exception as err: + record("installed:reload", err, case) + + +################################################################################ +## 2. ports.json entries from a source +def test_fuzz_source(tmp_path, make_hm, monkeypatch, guard): + rng = random.Random(SEED + 1) + data, _ = base_infos() + outside = make_outside(tmp_path) + + for case_no in range(N): + fuzzed = json.loads(json.dumps(data)) + names = list(fuzzed['ports']) + for _ in range(rng.randint(1, 3)): + name = rng.choice(names) + entry = mutate(rng, fuzzed['ports'][name], outside) + if rng.random() < 0.3: + entry['source'] = weird_values(rng, outside) + elif isinstance(entry.get('source'), dict) and rng.random() < 0.5: + entry['source'][rng.choice(['md5', 'url', 'size', 'date_added', 'date_updated', 'release_id'])] = weird_values(rng, outside) + fuzzed['ports'][name] = entry + + if rng.random() < 0.1: + fuzzed['ports'][rng.choice(['', '../x.zip', 'UPPER.ZIP', 'noext'])] = rng.choice(list(fuzzed['ports'].values())) + if rng.random() < 0.05: + fuzzed['ports'] = weird_values(rng, outside) + if rng.random() < 0.05: + fuzzed['utils'] = weird_values(rng, outside) + + case = json.dumps(fuzzed, default=str)[:300] + monkeypatch.setattr(hm_net, "fetch_json", lambda url, fuzzed=fuzzed: json.loads(json.dumps(fuzzed))) + + def fake_download(file_name, file_url, md5_source=None, md5_result=None, callback=None, no_check=False): + files = basic_port_files(name=file_name.name.rsplit('.', 1)[0]) + return write_port_zip(file_name, files) + + monkeypatch.setattr(hm_net, "download", fake_download) + + try: + hm = make_hm(offline=False) + hm.sources['pm'].update() + except (AssertionError, SandboxEscape): + raise + except Exception as err: + record("source:load", err, case) + continue + + names = list(fuzzed['ports']) if isinstance(fuzzed['ports'], dict) else [] + exercise(hm, names, "source", case) + + for name in names[:2]: + try: + hm.install_port(name) + except AssertionError: + raise + except Exception as err: + record("source:install", err, case) + + try: + hm.uninstall_port(name) + except AssertionError: + raise + except Exception as err: + record("source:uninstall", err, case) + + outside_intact(outside) + + ## And again from the cache. + try: + make_hm(offline=True) + except Exception as err: + record("source:cached", err, case) + + +################################################################################ +## 3. port.json inside a port zip +def test_fuzz_zip(tmp_path, make_hm, hm_dirs, guard): + rng = random.Random(SEED + 2) + _, infos = base_infos() + outside = make_outside(tmp_path) + hm = make_hm() + + for case_no in range(N): + files = basic_port_files() + info = mutate(rng, files["testport/port.json"] if rng.random() < 0.5 else rng.choice(infos), outside) + files["testport/port.json"] = to_bytes(rng, info) + case = files["testport/port.json"][:300] + zip_file = write_port_zip(hm_dirs['temp_dir'] / "testport.zip", files) + + try: + hm.install_port(str(zip_file)) + except (AssertionError, SandboxEscape): + raise + except Exception as err: + record("zip:install", err, case) + + try: + hm.load_ports() + except Exception as err: + record("zip:load_ports", err, case) + hm = make_hm() + continue + + names = list(hm.installed_ports) + list(hm.broken_ports) + exercise(hm, names, "zip", case) + for name in names: + try: + hm.uninstall_port(name) + except AssertionError: + raise + except Exception as err: + record("zip:uninstall", err, case) + + outside_intact(outside) + + +################################################################################ +## 4. The rest of what load_ports reads: ports_info.json, port_stats.json, script signatures +PORTS_INFO = { + "items": { + "Test Port.sh": "testport.zip", "testport/": "testport.zip", + "Abes Adventure.sh": "abes.zip", "abes/": "abes.zip", + "Shared.sh": ["one.zip", "two.zip"], + "SPROUT.sh": "sprout.zip", "sprout/": "sprout.zip", + }, + "md5": {}, + "ports": { + "testport.zip": {"date": ["2021-01-01", "2022-01-01"], "file": "testport/port.json", "items": ["Test Port.sh", "testport/"], "name": "testport.zip"}, + "abes.zip": {"date": ["2020-01-01", "2020-01-01"], "file": "abes/abes.port.json", "items": ["Abes Adventure.sh", "abes/"], "name": "abes.zip"}, + "sprout.zip": {"date": ["2025-07-16", "2025-07-16"], "file": "sprout/port.json", "items": ["SPROUT.sh", "sprout/"], "name": "sprout.zip"}, + }, + "portsmd_fix": {"cebion": ["Cebion"]}, + } + + +def mutate_any(rng, value, outside, depth=0): + """Mutate somewhere in a nested json value.""" + value = json.loads(json.dumps(value)) + if rng.random() < 0.15 or depth > 4 or not isinstance(value, (dict, list)) or not value: + return weird_values(rng, outside) + + if isinstance(value, dict): + key = rng.choice(list(value) + ['new-key']) + roll = rng.random() + if roll < 0.15: + value.pop(key, None) + elif roll < 0.25: + value[rng.choice(weird_strings(outside))] = value.get(key, weird_values(rng, outside)) + else: + value[key] = mutate_any(rng, value.get(key), outside, depth + 1) + else: + index = rng.randrange(len(value)) + value[index] = mutate_any(rng, value[index], outside, depth + 1) + + return value + + +def test_fuzz_ports_dir(tmp_path, make_hm, hm_dirs, guard): + rng = random.Random(SEED + 3) + ports_dir = hm_dirs['ports_dir'] + outside = make_outside(tmp_path) + hm = make_hm() + + for case_no in range(N): + ports_info = PORTS_INFO + for _ in range(rng.randint(1, 4)): + ports_info = mutate_any(rng, ports_info, outside) + + shutil.rmtree(ports_dir) + ports_dir.mkdir() + + layout = {} + for script, folder in (("Test Port.sh", "testport"), ("Abes Adventure.sh", "abes"), ("SPROUT.sh", "sprout"), ("Shared.sh", None), ("Unknown.sh", "unknown")): + if rng.random() < 0.2: + continue + + header = "#!/bin/bash\n" + if rng.random() < 0.5: + sig = [rng.choice(weird_strings(outside) + ["testport.zip", "abes.zip", "sprout.zip"]) for _ in range(rng.randint(0, 3))] + header += "# PORTMASTER: " + ", ".join(sig).replace("\n", " ") + "\n" + layout[script] = header + "echo\n" + + if folder: + layout[f"{folder}/"] = "" + if rng.random() < 0.5: + files = basic_port_files(name=folder, script=script) + info = files[f"{folder}/port.json"] + if rng.random() < 0.6: + info = mutate(rng, info, outside) + json_name = rng.choice(["port.json", f"{folder}.port.json"]) + layout[f"{folder}/{json_name}"] = to_bytes(rng, info) + if rng.random() < 0.1: + layout[f"{folder}/other.port.json"] = to_bytes(rng, mutate(rng, info, outside)) + + if rng.random() < 0.3: + ## A script renamed by the user, signed with its old name. + layout["Renamed.sh"] = "#!/bin/bash\n# PORTMASTER: testport.zip, Test Port.sh\necho\n" + + for name, data in layout.items(): + path = ports_dir / name + if name.endswith('/'): + path.mkdir(parents=True, exist_ok=True) + continue + path.parent.mkdir(parents=True, exist_ok=True) + path.write_bytes(data if isinstance(data, bytes) else data.encode('utf-8', 'surrogatepass')) + + (hm.cfg_dir / "ports_info.json").write_bytes(to_bytes(rng, ports_info)) + stats = {"ports": {"testport.zip": 10, "sprout.zip": 3}} + if rng.random() < 0.5: + stats = mutate_any(rng, stats, outside) + (hm.cfg_dir / "port_stats.json").write_bytes(to_bytes(rng, stats)) + + hm._PORTS_INFO = None + hm._PORTS_DOWNLOAD = None + case = (json.dumps(ports_info, default=str)[:300], sorted(layout)) + + try: + hm.load_ports() + except (AssertionError, SandboxEscape): + raise + except Exception as err: + record("ports_dir:load_ports", err, case) + continue + + names = list(hm.installed_ports) + list(hm.broken_ports) + exercise(hm, names, "ports_dir", case) + for name in names: + try: + hm.uninstall_port(name) + except AssertionError: + raise + except Exception as err: + record("ports_dir:uninstall", err, case) + + outside_intact(outside) + + +################################################################################ +## 5. The GUI's ports list and port info screens, drawn with fuzzed ports +@pytest.mark.sdl +def test_fuzz_gui(gui, tmp_path, make_hm, hm_dirs, monkeypatch): + from pugwash import scenes + + rng = random.Random(SEED + 4) + data, infos = base_infos() + outside = make_outside(tmp_path) + ports_dir = hm_dirs['ports_dir'] + saved_scenes = list(gui.scenes) + monkeypatch.setattr(gui, "message_box", lambda *a, **kw: True) + + def fake_download(file_name, file_url, md5_source=None, md5_result=None, callback=None, no_check=False): + return write_port_zip(file_name, basic_port_files(name=file_name.name.rsplit('.', 1)[0])) + + monkeypatch.setattr(hm_net, "download", fake_download) + + try: + for case_no in range(max(N // 10, 20)): + fuzzed = json.loads(json.dumps(data)) + for name in rng.sample(list(fuzzed['ports']), rng.randint(1, len(fuzzed['ports']))): + fuzzed['ports'][name] = mutate(rng, fuzzed['ports'][name], outside) + if rng.random() < 0.3: + ## Text that looks like theme template tags. + attr = fuzzed['ports'][name].setdefault('attr', {}) + if isinstance(attr, dict): + attr[rng.choice(['title', 'desc', 'inst'])] = rng.choice([ + "{port_info.title}", "{", "}", "{{}}", "{0}", "{if:port_info.title}", "{endif}", + "{port_info.description}" * 50, "\n" * 200, "x" * 20000, "\t\x00\u202e", + "\n", " \n ", "\r", "\t", "\u200b", "\ufeff", "\x00", "a\x00b", "\U0001F3AE", + "x" * 500 + " " + "y" * 500, "\n\n\nx", "\u65e5\u672c\u8a9e" * 300]) + + monkeypatch.setattr(hm_net, "fetch_json", lambda url, fuzzed=fuzzed: json.loads(json.dumps(fuzzed))) + + shutil.rmtree(ports_dir) + ports_dir.mkdir() + files = basic_port_files() + files["testport/port.json"] = to_bytes(rng, mutate(rng, files["testport/port.json"], outside)) + for name, value in files.items(): + path = ports_dir / name + if name.endswith('/'): + path.mkdir(parents=True, exist_ok=True) + continue + path.parent.mkdir(parents=True, exist_ok=True) + path.write_bytes(value if isinstance(value, bytes) else value.encode('utf-8')) + + case = json.dumps(fuzzed, default=str)[:300] + try: + hm = make_hm(offline=False) + hm.sources['pm'].update() + except Exception as err: + record("gui:hm", err, case) + continue + + gui.hm = hm + gui.clear_port_sizes() + try: + for mode, options in (('install', {'mode': 'install', 'name': None}), + ('uninstall', {'mode': 'uninstall', 'base_filters': ['installed']})): + gui.scenes = list(saved_scenes) + list_scene = scenes.PortsListScene(gui, options) + gui.push_scene('ports', list_scene) + gui.do_loop(no_delay=True) + + for index, port_name in enumerate(list(list_scene.port_list)): + list_scene.tags['ports_list'].selected = index + gui.do_loop(no_delay=True) + + gui.push_scene('port_info', scenes.PortInfoScene(gui, port_name, mode, list_scene)) + gui.do_loop(no_delay=True) + gui.pop_scene() + gui.do_loop(no_delay=True) + + except (AssertionError, SandboxEscape): + raise + except Exception as err: + record("gui:screens", err, case) + + finally: + gui.scenes = saved_scenes + gui.hm = None + gui.clear_port_sizes() diff --git a/tests/test_gui_screens.py b/tests/test_gui_screens.py index a0ed5c91..069eb994 100644 --- a/tests/test_gui_screens.py +++ b/tests/test_gui_screens.py @@ -7,6 +7,7 @@ from pathlib import Path import pytest +import sdl2 import harbourmaster @@ -575,3 +576,21 @@ def test_schemes_back_and_blank(screen_gui, schemes, monkeypatch): def test_schemes_b(screen_gui, schemes, monkeypatch): press(screen_gui, monkeypatch, 'B') assert top(screen_gui) is not schemes + + +################################################################################ +## Text from anyone's port.json +TEXTS = ["\n", "\n" * 50, " \n ", "\r", "\t", "\x00", "a\x00b", "\u200b", "\ufeff", "\u202e", "x" * 2000] + + +@pytest.mark.parametrize('text', TEXTS, ids=[repr(text[:8]) for text in TEXTS]) +@pytest.mark.parametrize('width', [None, 300, 1]) +def test_text_the_font_cant_draw(gui, text, width): + ## sdl2.ext fails on these (nothing to draw, or narrower than a character). + font = next(iter(gui.text.fonts.values())) + + surface = font.quick_render(text, 20, width=width) + + assert surface.w > 0 + sdl2.SDL_FreeSurface(surface) + diff --git a/tests/test_hm_info.py b/tests/test_hm_info.py index 196ed365..5b75c260 100644 --- a/tests/test_hm_info.py +++ b/tests/test_hm_info.py @@ -353,3 +353,55 @@ def test_git_blob_sha_file(tmp_path): path.write_bytes(b"hello\n") assert calculate_git_blob_sha_file(path) == "ce013625030ba8dba906f756967f9e9ca394464a" assert calculate_git_blob_sha_file(tmp_path / "missing") is None + + +def test_ports_info_with_bad_values_is_cleaned(hm): + ## A bad copy of ports_info.json stopped PortMaster starting. + (hm.cfg_dir / "ports_info.json").write_text(json.dumps({ + "items": {"Good.sh": "good.zip", "Shared.sh": ["a.zip", 1, "b.zip"], "Bad.sh": 5, "Nameless.sh": "\u202e", "Empty.sh": []}, + "md5": {"abc": "Good.sh", "def": 1}, + "ports": { + "good.zip": {"name": "good.zip", "items": ["Good.sh", "good/"], "file": "good/port.json"}, + "noname.zip": {"name": 1, "items": ["N.sh"], "file": "n/port.json"}, + "outside.zip": {"name": "outside.zip", "items": ["O.sh"], "file": "../outside/port.json"}, + "notjson.zip": {"name": "notjson.zip", "items": ["J.sh"], "file": "j/script.sh"}, + "unsafe.zip": {"name": "unsafe.zip", "items": ["../../home"], "file": "u/port.json"}, + "noitems.zip": {"name": "noitems.zip", "file": "x/port.json"}, + "list.zip": ["x"], + }, + "portsmd_fix": {"cebion": ["Cebion"], "x": 1}, + })) + hm._PORTS_INFO = None + + info = hm.ports_info() + + assert info["items"] == {"Good.sh": "good.zip", "Shared.sh": ["a.zip", "b.zip"]} + assert info["md5"] == {"abc": "Good.sh"} + assert sorted(info["ports"]) == ["good.zip", "noname.zip"] + assert info["ports"]["noname.zip"]["name"] == "noname.zip" + assert info["portsmd_fix"] == {"cebion": ["Cebion"]} + + +@pytest.mark.parametrize("contents", ["[]", '"x"', '{"items": [], "ports": 1}', "[" * 100000]) +def test_ports_info_wrong_shape(hm, contents): + (hm.cfg_dir / "ports_info.json").write_text(contents) + hm._PORTS_INFO = None + + assert hm.ports_info() == {"items": {}, "md5": {}, "ports": {}, "portsmd_fix": {}} + hm.load_ports() + + +@pytest.mark.parametrize("count, expected", [(5, 5), ("5", 0), (None, 0), (True, 0), ([1], 0)]) +def test_port_downloads_must_be_numbers(hm, count, expected): + ## They are sorted by, with each other. + (hm.cfg_dir / "port_stats.json").write_text(json.dumps({"ports": {"game.zip": count}})) + hm._PORTS_DOWNLOAD = None + + assert hm.port_downloads("game.zip") == expected + + +def test_port_stats_wrong_shape(hm): + (hm.cfg_dir / "port_stats.json").write_text('{"ports": ["game.zip"]}') + hm._PORTS_DOWNLOAD = None + + assert hm.port_downloads("game.zip") == 0 diff --git a/tests/test_hm_ports.py b/tests/test_hm_ports.py index df161b8a..87a1cfd0 100644 --- a/tests/test_hm_ports.py +++ b/tests/test_hm_ports.py @@ -82,6 +82,18 @@ def test_installed_port_found(installed): assert installed.unknown_ports == [] +def test_port_json_without_runtime(hm, ports_dir): + ## A hand-written version 3 port.json with no runtime stopped PortMaster starting. + files = basic_port_files() + files["testport/port.json"] = {"version": 3, "name": "testport.zip", "items": ["Test Port.sh", "testport/"], "attr": {"title": "Test Port"}} + install_files(ports_dir, files) + + hm.load_ports() + + assert list(hm.installed_ports) == ["testport.zip"] + assert hm.installed_ports["testport.zip"]['attr']['runtime'] == [] + + def test_old_port_json_name_is_renamed(hm, ports_dir): files = basic_port_files() files["testport/testport.port.json"] = files.pop("testport/port.json") @@ -772,3 +784,82 @@ def test_installed_port_images(installed, ports_dir, image, files, expected): def test_installed_port_images_bad_port_json_entry(installed): installed.installed_ports["testport.zip"]['files']['port.json'] = ["a", "b"] assert installed.port_images("testport.zip") is None + + +################################################################################ +## load_ports: port.json files and ports_info.json it can't fully use +@pytest.mark.parametrize('changes', [ + {'status': "Installed"}, + {'status': ["x"]}, + {'files': "port.json"}, + {'files': ["x"]}, + {'files': {'port.json': ["a", "b"], 'Test Port.sh': {"x": 1}}}, + {'source': "x"}, + ], ids=['status-str', 'status-list', 'files-str', 'files-list', 'files-values', 'source-str']) +def test_port_json_with_bad_root_values(hm, ports_dir, changes): + ## Each of these stopped PortMaster starting. + files = basic_port_files() + files["testport/port.json"].update(changes) + install_files(ports_dir, files) + + hm.load_ports() + + assert list(hm.installed_ports) == ["testport.zip"] + assert hm.installed_ports["testport.zip"]['status']['status'] == "Installed" + hm.list_ports_new() + + +@pytest.mark.parametrize('name', ["/", "...", "<<>>", "\u202e"]) +def test_port_json_name_with_nothing_left_once_cleaned(hm, ports_dir, name): + ## It is found from its items instead, as with no name. + set_ports_info(hm) + files = basic_port_files() + files["testport/port.json"]['name'] = name + install_files(ports_dir, files) + + hm.load_ports() + + assert list(hm.installed_ports) == ["testport.zip"] + hm.list_ports_new() + + +def test_port_json_named_by_a_bad_ports_info_owner(hm, ports_dir): + set_ports_info(hm, items={"Test Port.sh": "\u202e", "testport/": "\u202e"}) + files = basic_port_files() + del files["testport/port.json"]['name'] + install_files(ports_dir, files) + + hm.load_ports() + + assert hm.installed_ports == {} + hm.list_ports_new() + + +def test_new_port_missing_from_ports_info(hm, ports_dir): + ## ports_info.json says whose script it is, but doesn't have that port. + set_ports_info(hm, items={"Test Port.sh": "gone.zip"}) + install_files(ports_dir, {"Test Port.sh": "#!/bin/bash\n"}) + + hm.load_ports() + + assert hm.installed_ports == {} + + +def test_new_port_json_is_never_written_outside_the_ports_folder(hm, ports_dir, tmp_path): + data = json.loads(json.dumps(PORTS_INFO)) + data['ports']['testport.zip']['file'] = "../outside/port.json" + set_ports_info(hm, data) + (tmp_path / "outside").mkdir() + install_files(ports_dir, {"Test Port.sh": "#!/bin/bash\n", "testport/": ""}) + + hm.load_ports() + + assert not (tmp_path / "outside" / "port.json").exists() + assert hm.installed_ports == {} + + +@pytest.mark.parametrize('screenshot', ["x" * 300, "a\x00b", "../../etc/passwd", "/etc/passwd"]) +def test_installed_port_screenshot_with_a_bad_name(installed, screenshot): + installed.installed_ports["testport.zip"]['attr']['image'] = {'screenshot': screenshot} + + assert installed.port_images("testport.zip") is None diff --git a/tests/test_info.py b/tests/test_info.py index 83ad04c2..4a4a4050 100644 --- a/tests/test_info.py +++ b/tests/test_info.py @@ -54,6 +54,119 @@ def test_load_v3_blank_runtime(): assert info['attr']['runtime'] == [] +@pytest.mark.parametrize('version', [1, 2, 3, "3"]) +@pytest.mark.parametrize('info', [{}, {'attr': {}}, {'attr': {'title': "Game"}}], ids=['no-attr', 'empty-attr', 'no-runtime']) +def test_load_old_version_without_runtime(version, info): + ## Hand-written port.json files leave these out; this stopped PortMaster starting. + info = port_info_load(dict(info, version=version)) + + assert info['version'] == 4 + assert info['attr']['runtime'] == [] + assert set(info['attr']) == set(PORT_INFO_ATTR_ATTRS) + + +@pytest.mark.parametrize('runtime, expected', [ + (None, []), + ("mono.squashfs", ["mono.squashfs"]), + (["a.squashfs", "b.squashfs"], ["a.squashfs", "b.squashfs"]), + ]) +def test_load_v3_runtime_upgrade(runtime, expected): + assert port_info_load({'version': 3, 'attr': {'runtime': runtime}})['attr']['runtime'] == expected + + +VALUES = [None, True, 0, 1.5, "", "x", "blank", [], [None], [1, "a"], ["a"], {}, {"a": 1}] + + +@pytest.mark.parametrize('version', [None, 1, 2, 3, 4, 5, "3", "x", True, -1, 2.0]) +def test_load_never_raises(version): + ## Every attr and root key with every type of value, at every version. + infos = [{}, *({'attr': value} for value in VALUES)] + infos += [{key: value, 'attr': {}} for key in ('name', 'items', 'items_opt', 'status', 'files', 'source', 'md5') for value in VALUES] + infos += [{'attr': {key: value}} for key in PORT_INFO_ATTR_ATTRS for value in VALUES] + + for info in infos: + if version is not None: + info['version'] = version + + for do_default in (True, False): + loaded = port_info_load(json.loads(json.dumps(info)), do_default=do_default) + if loaded is not None: + port_info_merge(port_info_load({}), loaded) + + +@pytest.mark.parametrize('status', ["Installed", ["Installed"], 1, True]) +def test_load_bad_status_is_dropped(status): + assert 'status' not in port_info_load({'name': 'game.zip', 'status': status}) + + +def test_load_status_values_are_cleaned(): + info = port_info_load({'name': 'game.zip', 'status': {'source': ["x"], 'status': "Installed", 'md5': 5}}) + assert info['status'] == {'status': "Installed", 'md5': None} + + +@pytest.mark.parametrize('source', ["github", ["x"], 1, None]) +def test_load_bad_source_is_dropped(source): + assert 'source' not in port_info_load({'name': 'game.zip', 'source': source}) + + +def test_load_source_values_are_cleaned(): + ## Sorted by and compared with numbers, so the wrong types would crash the ports list. + info = port_info_load({'name': 'game.zip', 'source': { + 'md5': None, 'url': "https://x/game.zip", 'size': "100", 'downloads': -1, + 'date_added': None, 'date_updated': "2024-01-01", 'release_id': 7}}) + + assert info['source'] == {'url': "https://x/game.zip", 'date_updated': "2024-01-01"} + + +def test_load_size_must_be_a_number(): + assert port_info_load({'source': {'size': True}})['source'] == {} + assert port_info_load({'source': {'size': 10}})['source'] == {'size': 10} + + +@pytest.mark.parametrize('files', ["port.json", ["port.json"], 1]) +def test_load_bad_files_are_emptied(files): + assert port_info_load({'name': 'game.zip', 'files': files})['files'] == {} + + +def test_load_unsafe_files_are_dropped(): + ## Uninstalling deletes the values, and muOS makes folders from the keys. + info = port_info_load({'name': 'game.zip', 'files': { + 'port.json': "game/port.json", + 'Game.sh': ["Game.sh", "../Other.sh", None], + '../../escape.sh': "escape.sh", + 'game/': "/etc", + 'data/': {"a": 1}, + 'more/': ["more/", "also/"], + }}) + + assert info['files'] == {'port.json': "game/port.json", 'Game.sh': "Game.sh", 'more/': ["more/", "also/"]} + + +@pytest.mark.parametrize('name', [1, ["game.zip"], "", " "]) +def test_load_bad_name_is_none(name): + assert port_info_load({'name': name})['name'] is None + + +@pytest.mark.parametrize('min_glibc, expected', [ + ("2.31", "2.31"), (" 2.35 ", " 2.35 "), ("2", "2"), ("abc", ""), ("2.a", ""), ("2.31-r1", ""), (2.31, "2.31"), (None, "")]) +def test_load_min_glibc_must_be_a_version(min_glibc, expected): + ## It is compared with the device's glibc: "abc" crashed the ports list. + assert port_info_load({'attr': {'min_glibc': min_glibc}})['attr']['min_glibc'] == expected + + +def test_load_runtimes_are_file_names(): + info = port_info_load({'attr': {'runtime': ["mono.squashfs", "../../x.squashfs", "a/b.squashfs", "", "x" * 300, "a\x00b"]}}) + assert info['attr']['runtime'] == ["mono.squashfs"] + + +@pytest.mark.parametrize('item, safe', [ + ("Game.sh", True), ("game/", True), ("game/data", True), + ("a\x00b", False), ("x" * 256, False), ("x" * 255, True), ("game/" + "x" * 256, False), ("日" * 86, False)]) +def test_safe_port_item_names(item, safe): + from harbourmaster.info import is_safe_port_item + assert is_safe_port_item(item) == safe + + def test_load_string_version_and_reqs_dict(): info = port_info_load({'version': '4', 'attr': {'reqs': {'opengl': True, 'power': True}}}) diff --git a/tests/test_platform_hooks.py b/tests/test_platform_hooks.py index 5e1aa4b9..33c0c6c1 100644 --- a/tests/test_platform_hooks.py +++ b/tests/test_platform_hooks.py @@ -842,3 +842,18 @@ def test_muos_install_fills_the_catalogue(make_hm, hm_dirs, device, monkeypatch, assert hm.install_port(str(zip_file)) == 0 assert device.path(catalogue + "/box/Test Port-pre.png").read_text() == "cover" assert device.path(catalogue + "/text/Test Port.txt").read_text().startswith("First line") + + +def test_muos_port_uninstall_app_folder_from_a_bad_files_key(hm, device): + ## Only the script's own name makes the folder, so `..` in a files key can't reach outside. + app_dir = device.mkdir("/run/muos/storage/application") + escape = device.mkdir("/run/muos/storage/escape/Test Port") + platform = muos.PlatformmuOS(hm) + (escape / platform.MUOSAPP_PM_MARKER).write_text("Test Port") + (app_dir / "Test Port").mkdir() + (app_dir / "Test Port" / platform.MUOSAPP_PM_MARKER).write_text("Test Port") + + platform.port_uninstall("testport.zip", muos_port_info(script="../escape/Test Port.sh"), []) + + assert escape.is_dir() + assert not (app_dir / "Test Port").exists() diff --git a/tests/test_source.py b/tests/test_source.py index daf74bd7..9d4c4836 100644 --- a/tests/test_source.py +++ b/tests/test_source.py @@ -184,3 +184,81 @@ def test_unknown_source_api_is_skipped(make_hm): assert "custom" not in hm.sources assert sorted(hm.sources) == ['pm', 'pmmv'] + + +################################################################################ +## A ports.json from any source: entries it can't use are skipped, not fatal. +def update_with(hm, monkeypatch, data): + monkeypatch.setattr(hm_net, "fetch_json", lambda url: json.loads(json.dumps(data))) + source = hm.sources['pm'] + source.update() + return source + + +@pytest.mark.parametrize('source_block', [None, "x", [], {}, {'md5': "abc", 'size': 1}]) +def test_port_without_a_download_is_skipped(hm, monkeypatch, ports_json_data, source_block): + if source_block is None: + del ports_json_data['ports']['sprout.zip']['source'] + else: + ports_json_data['ports']['sprout.zip']['source'] = source_block + + source = update_with(hm, monkeypatch, ports_json_data) + + assert 'sprout.zip' not in source.ports + assert sorted(source.ports) == sorted(set(ports_json_data['ports']) - {'sprout.zip'}) + + +def test_port_with_bad_source_values(hm, monkeypatch, ports_json_data): + ports_json_data['ports']['sprout.zip']['source'].update(size="big", md5=None, date_added=None) + source = update_with(hm, monkeypatch, ports_json_data) + + assert source.port_download_size('sprout.zip', check_runtime=False) == 0 + hm.list_ports_new(sort_by='recently_added') + + +@pytest.mark.parametrize('entry', ["/etc/hostname", 1, None, ["x"]]) +def test_port_entry_that_isnt_an_object(hm, monkeypatch, ports_json_data, entry): + ## A string isn't read as a port.json path either. + ports_json_data['ports']['sprout.zip'] = entry + source = update_with(hm, monkeypatch, ports_json_data) + + assert 'sprout.zip' not in source.ports + assert len(source.ports) == len(ports_json_data['ports']) - 1 + + +def test_port_with_no_name_uses_its_key(hm, monkeypatch, ports_json_data): + ports_json_data['ports']['sprout.zip']['name'] = 5 + source = update_with(hm, monkeypatch, ports_json_data) + + assert source.port_info('sprout.zip')['name'] == 'sprout.zip' + hm.list_ports_new() + + +@pytest.mark.parametrize('key', ["ports", "utils"]) +@pytest.mark.parametrize('value', [None, [], "x"]) +def test_ports_json_with_bad_sections(hm, monkeypatch, ports_json_data, key, value, log_errors): + ports_json_data[key] = value + update_with(hm, monkeypatch, ports_json_data) + + assert any("Bad ports.json" in message for message in log_errors) + + +@pytest.mark.parametrize('data', [None, [], "x", 1]) +def test_ports_json_that_isnt_an_object(hm, monkeypatch, data): + source = update_with(hm, monkeypatch, {'x': data} if data is None else data) + assert source.ports == [] + + +def test_util_names_are_file_names(hm, monkeypatch, ports_json_data): + ## Runtimes are saved in the libs folder under these names. + good = next(iter(ports_json_data['utils'].values())) + ports_json_data['utils']['../../escape.squashfs'] = dict(good) + ports_json_data['utils']['named.squashfs'] = dict(good, runtime_name="../escape.squashfs", runtime_arch="aarch64") + ports_json_data['utils']['no-url.squashfs'] = {'name': "x", 'md5': "y"} + ports_json_data['utils']['sized.squashfs'] = dict(good, size="big", md5=7) + + source = update_with(hm, monkeypatch, ports_json_data) + + assert not any('escape' in runtime for runtime in hm.runtimes_info) + assert 'no-url.squashfs' not in source.utils + assert source.port_download_size('sized.squashfs') == 0 diff --git a/tests/test_util_edges.py b/tests/test_util_edges.py index 45afd938..aaf0db12 100644 --- a/tests/test_util_edges.py +++ b/tests/test_util_edges.py @@ -567,3 +567,14 @@ def test_uninstall_ignores_files_outside_the_ports_folder(sandbox, value, log_er assert sandbox.uninstall_port("testport.zip") == 0 assert (sandbox.root / "precious.txt").read_text() == "keep me" assert not (sandbox.ports_dir / "Test Port.sh").exists() + + +@pytest.mark.parametrize('text', ["[" * 100000, '{"a":' * 100000]) +def test_json_nested_too_deep(text, tmp_path): + ## A RecursionError in the json module, from a scrambled file or a bad download. + from harbourmaster.util import json_safe_load, json_safe_loads + + assert json_safe_loads(text) is None + (tmp_path / "deep.json").write_text(text) + with (tmp_path / "deep.json").open() as fh: + assert json_safe_load(fh) is None