Add validation checks to prevent arbitrary code execution via stitching.

src/openflexure_microscope_server/scan_directories.py

@@ -12,6 +12,7 @@ from datetime import datetime, timedelta
 from pydantic import BaseModel, field_validator, field_serializer
 
 from openflexure_microscope_server.utilities import requires_lock
+from openflexure_microscope_server.utilities import make_name_safe
 
 IMG_DIR_NAME = "images"
 SCAN_ZERO_PAD_DIGITS = 4
@@ -257,6 +258,7 @@ class ScanDirectoryManager:
 
         For more explanation on the scan naming see `new_scan_dir`
         """
+        scan_name = make_name_safe(scan_name)
         # A regex with the scan name and a group for the numbers
         scan_regex = re.compile(
             "^" + scan_name + "_([0-9]{" + str(SCAN_ZERO_PAD_DIGITS) + "})$"

src/openflexure_microscope_server/stitching.py

@@ -14,6 +14,8 @@ import time
 
 import labthings_fastapi as lt
 
+from openflexure_microscope_server.utilities import make_path_safe
+
 STITCHING_CMD = "openflexure-stitch"
 STITCHING_RESOLUTION = (820, 616)
 
@@ -21,6 +23,10 @@ DEFAULT_OVERLAP = 0.1
 DEFAULT_RESIZE = 0.5
 
 
+class StitcherValidationError(RuntimeError):
+    """The stitcher received values that it deems unsafe to create a command from."""
+
+
 class BaseStitcher:
     """A base stitching class for all stitchers. Don't initialise this directly."""
 
@@ -36,24 +42,49 @@ class BaseStitcher:
         # Set minimum overlap to 90% of the scan overlap to catch only images
         # directly adjacent, not images with overlapping corners.
         self.images_dir = images_dir
+        try:
+            overlap = float(overlap)
+            correlation_resize = float(correlation_resize)
+        except ValueError as e:
+            raise StitcherValidationError(
+                "Stitching inputs overlap or correlation_resize were not floats as "
+                "expected."
+            ) from e
+        self.validate_path()
+
         self.min_overlap = round(overlap * 0.9, 2)
-        self.correlation_resize = correlation_resize
+        self.correlation_resize = float(correlation_resize)
         self._mode = "all"
         self._extra_args = []
 
     @property
     def command(self) -> list[str]:
         """The command to run with subprocess.Popen."""
+        # Revalidate for good measure,
+        self.validate_path()
         # The command, and the mode
         initial_args = [STITCHING_CMD, "--stitching_mode", self._mode]
+        # Use float() just to really ensure that anything input is a float
         setting_args = [
             "--minimum_overlap",
-            f"{self.min_overlap}",
+            f"{float(self.min_overlap)}",
             "--resize",
-            f"{self.correlation_resize}",
+            f"{float(self.correlation_resize)}",
         ]
         return initial_args + self._extra_args + setting_args + [self.images_dir]
 
+    def validate_path(self):
+        """Check path is safe before making a command to run with subprocess.
+
+        This is essential for stopping arbitrary code execution.
+
+        :raises RuntimeError: if inputs are unsafe.
+        """
+        if self.images_dir != make_path_safe(self.images_dir):
+            raise StitcherValidationError(
+                "Invalid directory path contains unsafe characters."
+            )
+
     def start(self) -> None:
         """Start stitching a stitching process.
 
@@ -91,20 +122,6 @@ class PreviewStitcher(BaseStitcher):
 
         self._mode = "preview_stitch"
 
-    @property
-    def command(self) -> list[str]:
-        """The command to run with subprocess.Popen."""
-        return [
-            STITCHING_CMD,
-            "--stitching_mode",
-            "preview_stitch",
-            "--minimum_overlap",
-            f"{self.min_overlap}",
-            "--resize",
-            f"{self.correlation_resize}",
-            self.images_dir,
-        ]
-
     def start(self) -> None:
         """Start stitching a preview of the scan in a background subprocess.
 

src/openflexure_microscope_server/utilities.py

@@ -3,6 +3,7 @@
 from typing import TypeVar, Callable, ParamSpec
 import os
 import re
+import sys
 from threading import Thread
 import logging
 from importlib.metadata import version
@@ -126,6 +127,46 @@ def _wrap_and_catch_errors(target, error_buffer, *args, **kwargs):
         error_buffer.append(e)
 
 
+# Compiled regular expressions for unsafe characters
+_WINDOWS_UNSAFE_PATTERN = re.compile(r"[^a-zA-Z0-9_.\-:/\\]")
+_POSIX_UNSAFE_PATTERN = re.compile(r"[^a-zA-Z0-9_.\-/]")
+_NAME_UNSAFE_PATTERN = re.compile(r"[^a-zA-Z0-9_.\-]")
+
+
+def make_path_safe(unsafe_path_string: str) -> str:
+    """Replace unsafe characters in a file path with underscores, preserving separators.
+
+    This can be used to check that user inputs have not been concatenated into an
+    unsafe filepath.
+
+    This ensures compatibility across platforms by preserving only valid characters,
+    including path separators (forward slash on POSIX, and forward
+    slash/backslash/colon on Windows).
+
+    :param unsafe_path_string: The original path string to sanitise.
+    :returns: A version of the input string safe to use as a file path.
+    """
+    unsafe_character_pattern = (
+        _WINDOWS_UNSAFE_PATTERN
+        if sys.platform.startswith("win")
+        else _POSIX_UNSAFE_PATTERN
+    )
+    return unsafe_character_pattern.sub("_", unsafe_path_string)
+
+
+def make_name_safe(unsafe_name_string: str) -> str:
+    """Replace unsafe characters in a filename or identifier with underscores.
+
+    This excludes all path separators, ensuring the result is safe to use as a
+    standalone filename or identifier component.
+
+    :param unsafe_name_string: The original name string to sanitise.
+
+    :returns: A version of the input string safe to use as a file name or identifier.
+    """
+    return _NAME_UNSAFE_PATTERN.sub("_", unsafe_name_string)
+
+
 class VersionData(BaseModel):
     """A BaseModel containing the information about the server version.
 

tests/test_scan_directories.py

@@ -160,6 +160,16 @@ def test_basic_directory_operations():
     assert not os.path.isdir(scan_path)
 
 
+def test_bad_scan_names():
+    """Check scan names with spaces, or worse BASH commands are not allowed."""
+    _clear_scan_dir()
+    scan_dir_manager = ScanDirectoryManager(BASE_SCAN_DIR)
+    scan_dir = scan_dir_manager.new_scan_dir("fake scan")
+    assert scan_dir.name == "fake_scan_0001"
+    scan_dir = scan_dir_manager.new_scan_dir("fake scan;rm -rf /;")
+    assert scan_dir.name == "fake_scan_rm_-rf____0001"
+
+
 def test_scan_sequence_and_listing():
     """Check created scans are added in order and listed correctly."""
     _clear_scan_dir()

tests/test_stitching.py

@@ -5,6 +5,7 @@ generated, and the subprocess calling works as expected.
 """
 
 import logging
+import os
 from copy import copy
 
 import pytest
@@ -14,11 +15,12 @@ from openflexure_microscope_server.stitching import (
     PreviewStitcher,
     FinalStitcher,
     STITCHING_RESOLUTION,
+    StitcherValidationError,
 )
 
 # A global logger pretending to be an Invocation Logger
 LOGGER = logging.getLogger("mock-invocation_logger")
-FAKE_DIR = "a/dir/that/is/fake"
+FAKE_DIR = os.path.join("a", "dir", "that", "is", "fake")
 
 
 def test_base_stitcher():
@@ -45,7 +47,7 @@ def test_base_stitcher():
         ]
         stitcher = BaseStitcher(FAKE_DIR, overlap=overlap, correlation_resize=0.5)
         assert stitcher.command == expected_command
-        # check that a BaseSticher can't start
+        # check that a BaseStitcher can't start
         with pytest.raises(NotImplementedError):
             stitcher.start()
 
@@ -148,3 +150,26 @@ def test_final_stitcher_command_set_with_dict():
             scan_data_dict={"overlap": 0.4, resolution_key: resolution},
         )
         assert stitcher.command == expected_command
+
+
+def _validation_error_tester(scan_path, **kwargs):
+    """Check each type of stitcher throws a validation error for the given init args."""
+    # If scan_data_dict is in the kwargs only test the scan_data_dict
+    if "scan_data_dict" not in kwargs:
+        with pytest.raises(StitcherValidationError):
+            BaseStitcher(scan_path, **kwargs).command
+        with pytest.raises(StitcherValidationError):
+            PreviewStitcher(scan_path, **kwargs).command
+    with pytest.raises(StitcherValidationError):
+        FinalStitcher(scan_path, logger=LOGGER, **kwargs).command
+
+
+def test_validation_error():
+    """Test a number of way to try to inject mallicious arguments into the stitcher.
+
+    The stitcher should throw a validation error each attempt.
+    """
+    _validation_error_tester("/dir;rm -rf /;", overlap=".2", correlation_resize=".25")
+    _validation_error_tester(FAKE_DIR, overlap=".2", correlation_resize=".25;rm -rf /;")
+    _validation_error_tester(FAKE_DIR, overlap=".2;rm -rf /;", correlation_resize=".25")
+    _validation_error_tester(FAKE_DIR, scan_data_dict={"overlap": ".2;rm -rf /;"})