Port info: bad port.json, ports.json and ports_info.json files can't …

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=<n>` 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.

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:

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}
 

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
 
 

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")