diff --git a/src/openflexure_microscope_server/scan_directories.py b/src/openflexure_microscope_server/scan_directories.py index 295c36dd..4f2e5215 100644 --- a/src/openflexure_microscope_server/scan_directories.py +++ b/src/openflexure_microscope_server/scan_directories.py @@ -353,6 +353,9 @@ class ScanDirectoryManager: For more explanation on the scan naming see `new_scan_dir` """ scan_name = make_name_safe(scan_name) + + # Strip all trailing underscores from the base name + scan_name = scan_name.strip("_") # 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 1f077ec6..49916e10 100644 --- a/src/openflexure_microscope_server/stitching.py +++ b/src/openflexure_microscope_server/stitching.py @@ -18,7 +18,7 @@ from pydantic import BaseModel import labthings_fastapi as lt -from openflexure_microscope_server.utilities import make_path_safe +from openflexure_microscope_server.utilities import is_path_safe IS_WINDOWS = os.name == "nt" @@ -29,6 +29,22 @@ STITCH_TILE_SIZE = 8192 DEFAULT_OVERLAP = 0.1 DEFAULT_RESIZE = 0.5 +# A list of commands that are forbidden in any part of a generated CLI command. +# This provides defense-in-depth against trying to execute arbitrary shells +# or elevation tools. +FORBIDDEN_COMMANDS = { + "sudo", + "sh", + "bash", + "perl", + "ruby", + "php", + "nc", + "netcat", + "curl", + "wget", +} + class StitchingSettings(BaseModel): """The data needed to stitch a scan.""" @@ -49,17 +65,26 @@ class StitcherValidationError(RuntimeError): def validate_command(cmd: list[str]) -> None: - """Validate that the command only characters that are allowed in a path. + """Validate that the command only contains characters that are allowed in a path. The values in the commands should be numbers, commandline flags, paths, and - executables. All of these should be allowed by ``make_path_safe``. + executables. All of these should be allowed by ``is_path_safe``. + + This also checks against a blacklist of forbidden commands for defense-in-depth. :raises StitcherValidationError: if any element in the command is not safe. """ for element in cmd: - if element != make_path_safe(element): + # Check against forbidden commands (case-insensitive) + if element.lower() in FORBIDDEN_COMMANDS: raise StitcherValidationError( - "Invalid stiching command: Contains unsafe characters." + f"Invalid stitching command: Forbidden element '{element}' detected." + ) + + # Ensure characters are safe for a path component + if not is_path_safe(element): + raise StitcherValidationError( + f"Invalid stitching command: Element '{element}' contains unsafe characters." ) @@ -129,7 +154,7 @@ class BaseStitcher: :raises RuntimeError: if inputs are unsafe. """ - if self.images_dir != make_path_safe(self.images_dir): + if not is_path_safe(self.images_dir): raise StitcherValidationError( "Invalid directory path: Contains unsafe characters." ) diff --git a/src/openflexure_microscope_server/utilities.py b/src/openflexure_microscope_server/utilities.py index d38703eb..7da53cc6 100644 --- a/src/openflexure_microscope_server/utilities.py +++ b/src/openflexure_microscope_server/utilities.py @@ -76,34 +76,131 @@ def requires_lock( # Compiled regular expressions for unsafe characters -# Matches anything that isn't a-z, A-Z, 0-9, _, ., -, :, /, \ -_WINDOWS_UNSAFE_PATTERN = re.compile(r"[^a-zA-Z0-9_.\-:/\\]") -# Matches anything that isn't a-z, A-Z, 0-9, _, ., -, \ -_POSIX_UNSAFE_PATTERN = re.compile(r"[^a-zA-Z0-9_.\-/]") +# Matches anything that isn't a-z, A-Z, 0-9, _, ., -, :, /, \, space +_WINDOWS_UNSAFE_PATTERN = re.compile(r"[^a-zA-Z0-9_.\-:/\ \\]") +# Matches anything that isn't a-z, A-Z, 0-9, _, ., -, \, space +_POSIX_UNSAFE_PATTERN = re.compile(r"[^a-zA-Z0-9_.\-/ ]") # Matches anything that isn't a-z, A-Z, 0-9, _, ., - _NAME_UNSAFE_PATTERN = re.compile(r"[^a-zA-Z0-9_.\-]") +# Windows reserved names (case-insensitive) +_WINDOWS_RESERVED_NAMES = { + "CON", + "PRN", + "AUX", + "NUL", + "COM1", + "COM2", + "COM3", + "COM4", + "COM5", + "COM6", + "COM7", + "COM8", + "COM9", + "LPT1", + "LPT2", + "LPT3", + "LPT4", + "LPT5", + "LPT6", + "LPT7", + "LPT8", + "LPT9", +} -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. +def is_path_safe(unsafe_path_string: str) -> bool: + """Check if a file path has any unsafe elements in it. - This ensures compatibility across platforms by preserving only valid characters, - including path separators (forward slash on POSIX, and forward - slash/backslash/colon on Windows). + The path is not coerced into a safe form because if we ask + for a file to be written to a location we shouldn't be changing + that location as we have no "consent" to write to this other location. :param unsafe_path_string: The original path string to sanitise. - :returns: A version of the input string safe to use as a file path. + :returns: A boolean indicating if any unsafe features were found. """ + # Split by separators first to sanitise components independently + + components = re.split(r"([/\\])", unsafe_path_string) + unsafe_character_pattern = ( _WINDOWS_UNSAFE_PATTERN if sys.platform.startswith("win") else _POSIX_UNSAFE_PATTERN ) - return unsafe_character_pattern.sub("_", unsafe_path_string) + + # Ensure each warning only gets logged once per path + unsafe_relative_path = False + trailing_dots = False + trailing_whitespace = False + unsafe_characters = False + reserved_word = False + + for i, component in enumerate(components): + # Skip separators and empty strings + if component in ("/", "\\") or not component: + continue + + # 1. Check for relative paths in the wrong place + if component in (".", "..") and not unsafe_relative_path: + is_first = i == 0 + is_second = ( + i == 2 + and components[i - 1] in ("/", "\\") + and components[i - 2] == ".." + ) + if not (is_first or is_second): + LOGGER.warning( + f"File path {unsafe_path_string} may be unsafe due to unexpected relative navigation." + ) + unsafe_relative_path = True + continue + + # 2. Check for trailing dots in path + if "." in component and i != len(components) - 1 and not trailing_dots: + # Check for trailing dots in the file path before the file name + # e.g.: foo/bar./file is bad, but foo/bar.file.py is fine + LOGGER.warning( + f"File path {unsafe_path_string} may be unsafe due to trailing dots." + ) + trailing_dots = True + + # Check for trailing spaces - Windows specific + if ( + sys.platform.startswith("win") + and component.endswith(" ") + and not trailing_whitespace + ): + LOGGER.warning(f"{unsafe_path_string} contains unsafe trailing spaces.") + trailing_whitespace = True + + # 3. Check for unsafe characters (Regex) + if unsafe_character_pattern.search(component) and not unsafe_characters: + LOGGER.warning( + f"{unsafe_path_string} contains characters that may be unsafe on this platform." + ) + unsafe_characters = True + + # 4. Check for Reserved Names (e.g., CON, PRN, LPT1) + # Assuming _sanitise_reserved returns a different string if it's a reserved name + if ( + _sanitise_reserved(component, is_filename=False) != component + and not reserved_word + ): + LOGGER.warning( + f"{unsafe_path_string} contains a reserved system name and may cause issues." + ) + reserved_word = True + + return not ( + unsafe_relative_path + or trailing_dots + or trailing_whitespace + or unsafe_characters + or reserved_word + ) def make_name_safe(unsafe_name_string: str) -> str: @@ -112,11 +209,44 @@ def make_name_safe(unsafe_name_string: str) -> str: This excludes all path separators, ensuring the result is safe to use as a standalone filename or identifier component. + This also handles Windows reserved names and trailing dots/spaces. + :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) + # 1. Strip trailing dots and spaces + # 2. Apply unsafe character regex + # 3. Check for reserved names + name = unsafe_name_string.rstrip(". ") + if not name: + return "_" + + name = _NAME_UNSAFE_PATTERN.sub("_", name) + return _sanitise_reserved(name) + + +def _sanitise_reserved(name: str, is_filename: bool = True) -> str: + """Check a name against Windows reserved names. + + :param name: The component to check. + :param is_filename: Whether the component is a filename. + If True, names will be forced into lowercase. + + :returns: The sanitised component, in lower case. + """ + # Check for Windows reserved names. + # These are reserved even with an extension (e.g. NUL.txt). + # We check the part before the first dot. + base_name = name.split(".", maxsplit=1)[0].upper() + if base_name in _WINDOWS_RESERVED_NAMES: + return f"{name.lower()}_" if is_filename else f"{name}_" + + # We force names to be lowercase to avoid issues on + # Windows where files with the same name + # but different case can cause problems when checking if + # a file already exists. + return name.lower() if is_filename else name class VersionData(BaseModel): diff --git a/tests/unit_tests/test_scan_directories.py b/tests/unit_tests/test_scan_directories.py index 603587a3..0fc157fc 100644 --- a/tests/unit_tests/test_scan_directories.py +++ b/tests/unit_tests/test_scan_directories.py @@ -167,7 +167,7 @@ def test_bad_scan_names(): 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" + assert scan_dir.name == "fake_scan_rm_-rf_0001" def test_scan_sequence_and_listing(caplog): @@ -247,6 +247,63 @@ def test_scan_sequence_and_listing(caplog): assert len(caplog.records) == 0 +def test_scan_sequence_case_sensitive(): + """Check created scans are added in order and listed correctly, even when cases are different.""" + _clear_scan_dir() + scan_dir_manager = ScanDirectoryManager(BASE_SCAN_DIR) + + # Create some scan data and mark it as successful to get an end date. + scan_data = fake_active_scan_data() + scan_data.set_final_data(result="Success") + # Make 4 scans + scan_dir = scan_dir_manager.new_scan_dir("fake_scan") + scan_dir.save_scan_data(scan_data) + scan_dir = scan_dir_manager.new_scan_dir("FAKE_SCAN") + scan_dir.save_scan_data(scan_data) + scan_dir = scan_dir_manager.new_scan_dir("Fake_Scan") + scan_dir.save_scan_data(scan_data) + scan_dir = scan_dir_manager.new_scan_dir("FAKE_scan") + scan_dir.save_scan_data(scan_data) + scan_dir = scan_dir_manager.new_scan_dir("fAkE_sCaN") + scan_dir.save_scan_data(scan_data) + + # Check they exist and are numbered sequentially + all_scans = scan_dir_manager.all_scans + assert_unique_of_length(all_scans, 5) + assert "fake_scan_0001" in all_scans + assert "fake_scan_0002" in all_scans + assert "fake_scan_0003" in all_scans + assert "fake_scan_0004" in all_scans + assert "fake_scan_0005" in all_scans + + +def test_scan_names_have_single_underscores(): + """Check created scans have only one underscore in them, not two.""" + _clear_scan_dir() + scan_dir_manager = ScanDirectoryManager(BASE_SCAN_DIR) + + # Create some scan data and mark it as successful to get an end date. + scan_data = fake_active_scan_data() + scan_data.set_final_data(result="Success") + # Make 4 scans + scan_dir = scan_dir_manager.new_scan_dir("fake_scan_") + scan_dir.save_scan_data(scan_data) + scan_dir = scan_dir_manager.new_scan_dir("fake_scan__") + scan_dir.save_scan_data(scan_data) + scan_dir = scan_dir_manager.new_scan_dir("fake_scan_______") + scan_dir.save_scan_data(scan_data) + scan_dir = scan_dir_manager.new_scan_dir("fake_scan") + scan_dir.save_scan_data(scan_data) + + # Check they exist and are numbered sequentially + all_scans = scan_dir_manager.all_scans + assert_unique_of_length(all_scans, 4) + assert "fake_scan_0001" in all_scans + assert "fake_scan_0002" in all_scans + assert "fake_scan_0003" in all_scans + assert "fake_scan_0004" in all_scans + + def test_scan_name_non_sequential(): """Check new scan has the correct name if the directories are not sequential.""" _clear_scan_dir() diff --git a/tests/unit_tests/test_stitching.py b/tests/unit_tests/test_stitching.py index 349d975c..ead2276b 100644 --- a/tests/unit_tests/test_stitching.py +++ b/tests/unit_tests/test_stitching.py @@ -16,11 +16,13 @@ from pydantic import BaseModel import labthings_fastapi as lt from openflexure_microscope_server.stitching import ( + FORBIDDEN_COMMANDS, BaseStitcher, FinalStitcher, PreviewStitcher, StitcherValidationError, StitchingSettings, + validate_command, ) from ..shared_utils.lt_test_utils import LabThingsTestEnv @@ -32,6 +34,29 @@ THIS_DIR: str = os.path.dirname(os.path.realpath(__file__)) MOCK_STITCHER: str = os.path.join(THIS_DIR, "mock_stitching", "mock-stitch.py") +def test_validate_command_success(): + """Test valid commands pass validation.""" + validate_command(["openflexure-stitch", "--stitching_mode", "all", "path/to/scan"]) + validate_command(["--resize", "0.5", "8192"]) + + +@pytest.mark.parametrize("cmd_name", FORBIDDEN_COMMANDS) +def test_validate_command_forbidden(cmd_name): + """Test forbidden commands raise error.""" + # As the main command + with pytest.raises( + StitcherValidationError, match=f"Forbidden element '{cmd_name}' detected" + ): + validate_command([cmd_name, "some_arg"]) + + # As an argument (case-insensitive) + with pytest.raises( + StitcherValidationError, + match=f"Forbidden element '{cmd_name.upper()}' detected", + ): + validate_command(["safe-command", cmd_name.upper()]) + + def test_base_stitcher(): """Test the logic in BaseStitcher. diff --git a/tests/unit_tests/test_utilities.py b/tests/unit_tests/test_utilities.py new file mode 100644 index 00000000..bc9076cd --- /dev/null +++ b/tests/unit_tests/test_utilities.py @@ -0,0 +1,169 @@ +"""Unit tests for utility functions.""" + +import logging +import sys + +import pytest + +from openflexure_microscope_server.utilities import ( + _WINDOWS_RESERVED_NAMES, + is_path_safe, + make_name_safe, +) + +PATHS_WITH_DOTS = {"a./b/c.", "path./to/file.", "./path./to/file."} + +PATHS_WITH_SPACES = {"path /to /file ", "path/to /file", "path/a long path name /file"} + +RESERVED_NAME_PATHS = {"path/CON/file", "path/NUL.txt/file", "C:/AUX/test"} + +if sys.platform.startswith("win"): + UNSAFE_CHARACTER_PATHS = { + "C:/path/asterisk/data*.csv", + "data/files/read_me_first!.txt", + "logs/system.log", + "projects/alpha#1/notes.txt", + } +else: + UNSAFE_CHARACTER_PATHS = { + "/var/log/app:backup.log", + "/etc/configs/network-settings\v1", + "/tmp/file_with_@_symbol.txt", + "/home/user/script(1).py", + } + + +def test_make_name_safe_basic(): + """Test basic functionality of make_name_safe.""" + assert make_name_safe("normal_name") == "normal_name" + assert make_name_safe("name with spaces") == "name_with_spaces" + assert make_name_safe("name/with/slashes") == "name_with_slashes" + assert make_name_safe("name\\with\\backslashes") == "name_with_backslashes" + assert make_name_safe("sPoNgEmoCk") == "spongemock" + + +def test_make_name_safe_trailing_chars(): + """Test that trailing dots and spaces are removed.""" + assert make_name_safe("name.") == "name" + assert make_name_safe("name ") == "name" + assert make_name_safe("name. ") == "name" + assert make_name_safe("name .") == "name" + assert make_name_safe(".") == "_" + assert make_name_safe(" ") == "_" + assert make_name_safe(". ") == "_" + + +@pytest.mark.parametrize("name", _WINDOWS_RESERVED_NAMES) +def test_make_name_safe_reserved_names(name): + """Test Windows reserved names.""" + # Base name should be sanitized + assert make_name_safe(name) == f"{name.lower()}_" + # Case-insensitive + assert make_name_safe(name.lower()) == f"{name.lower()}_" + # With extension + assert make_name_safe(f"{name}.txt") == f"{name.lower()}.txt_" + # Multiple extensions + assert make_name_safe(f"{name}.tar.gz") == f"{name.lower()}.tar.gz_" + + +def test_make_name_safe_reserved_names_false_positives(): + """Test that names containing but not equal to reserved names are safe.""" + assert make_name_safe("CONSTANT") == "constant" + assert make_name_safe("CON2") == "con2" + assert make_name_safe("ICON") == "icon" + assert make_name_safe("icon") == "icon" + assert make_name_safe("iCon") == "icon" + assert make_name_safe("iCOn") == "icon" + assert make_name_safe("icOn") == "icon" + assert make_name_safe("CHILLI CON CARNE") == "chilli_con_carne" + + +def test_is_path_safe_basic(caplog): + """Test basic functionality of is_path_safe.""" + # Note: behavior depends on platform for separators + with caplog.at_level(logging.WARNING): + if sys.platform.startswith("win"): + assert is_path_safe("C:\\path\\to/file") + assert is_path_safe("C:\\a very long path\\to/file") + else: + assert is_path_safe("path/to/file") + assert is_path_safe("a very long path/to/file") + + # No errors should be raised. These paths are safe. + assert len(caplog.records) == 0 + + +@pytest.mark.parametrize("path", UNSAFE_CHARACTER_PATHS) +def test_is_path_safe_unsafe_characters_in_components(caplog, path): + """Test unsafe characters within path components.""" + with caplog.at_level(logging.WARNING): + assert not is_path_safe(path) + assert ( + f"{path} contains characters that may be unsafe on this platform." + in caplog.messages + ) + + +@pytest.mark.parametrize("path", RESERVED_NAME_PATHS) +def test_is_path_safe_reserved_in_components(caplog, path): + """Test reserved names within path components.""" + with caplog.at_level(logging.WARNING): + assert not is_path_safe(path) + assert ( + f"{path} contains a reserved system name and may cause issues." + in caplog.messages + ) + + +def test_is_path_safe_reserved_name_components(caplog): + """Test reserved names are okay as part of a larger dir name.""" + with caplog.at_level(logging.WARNING): + assert is_path_safe("chilli con carne/recipe") + assert len(caplog.records) == 0 + + +@pytest.mark.parametrize("path", PATHS_WITH_SPACES) +def test_is_path_safe_trailing_space_in_components(caplog, path): + """Test trailing spaces in path components.""" + with caplog.at_level(logging.WARNING): + # Test Windows - cannot have trailing spaces + if sys.platform.startswith("win"): + assert not is_path_safe(path) + assert len(caplog.records) == 1 + assert f"{path} contains unsafe trailing spaces." in caplog.messages + else: + # Trailing spaces allows in POSIX systems + assert is_path_safe(path) + # No warnings should be raised + assert len(caplog.records) == 0 + + +def test_is_path_safe_relative(caplog): + """Test that relative path components are preserved.""" + # Relative imports ./foo, ../foo or ../../foo are allowed. + with caplog.at_level(logging.WARNING): + assert is_path_safe("./openflexure/data/") + assert is_path_safe("../openflexure/data/") + assert is_path_safe("../../openflexure/data/") + assert is_path_safe(".") + assert is_path_safe("..") + + assert len(caplog.records) == 0 + + assert not is_path_safe("a_bad/../relative_path") + assert len(caplog.records) == 1 + assert ( + "File path a_bad/../relative_path may be unsafe due to unexpected relative navigation." + in caplog.messages + ) + + +@pytest.mark.parametrize("filepath", PATHS_WITH_DOTS) +def test_is_path_safe_dots_warning(caplog, filepath): + """Test that a user is warned about trailing dots in their file paths. The paths should remain unchanged.""" + with caplog.at_level(logging.WARNING): + assert not is_path_safe(filepath) + assert ( + f"File path {filepath} may be unsafe due to trailing dots." + in caplog.messages + )