From 7569c7d6b249c47d4f056c93e557014d449b0d55 Mon Sep 17 00:00:00 2001 From: Julian Stirling Date: Sun, 3 Aug 2025 18:54:11 +0100 Subject: [PATCH] Add validation checks to prevent arbitrary code execution via stitching. --- .../scan_directories.py | 2 + .../stitching.py | 51 ++++++++++++------- .../utilities.py | 41 +++++++++++++++ tests/test_scan_directories.py | 10 ++++ tests/test_stitching.py | 29 ++++++++++- 5 files changed, 114 insertions(+), 19 deletions(-) diff --git a/src/openflexure_microscope_server/scan_directories.py b/src/openflexure_microscope_server/scan_directories.py index 122ff2ce..da93085e 100644 --- a/src/openflexure_microscope_server/scan_directories.py +++ b/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) + "})$" diff --git a/src/openflexure_microscope_server/stitching.py b/src/openflexure_microscope_server/stitching.py index 2bc91357..ec4fe5b6 100644 --- a/src/openflexure_microscope_server/stitching.py +++ b/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. diff --git a/src/openflexure_microscope_server/utilities.py b/src/openflexure_microscope_server/utilities.py index 21099216..1031aa23 100644 --- a/src/openflexure_microscope_server/utilities.py +++ b/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. diff --git a/tests/test_scan_directories.py b/tests/test_scan_directories.py index 3fb317cc..04c4b2d1 100644 --- a/tests/test_scan_directories.py +++ b/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() diff --git a/tests/test_stitching.py b/tests/test_stitching.py index 3e8d1d98..92189bd7 100644 --- a/tests/test_stitching.py +++ b/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 /;"})