Merge branch '523-improve-validation-checks' into 'v3'

Improve Validation Checks

Closes #523

See merge request openflexure/openflexure-microscope-server!524
This commit is contained in:
Joe Knapper 2026-03-09 19:19:18 +00:00
commit 2e99e4c4c2
6 changed files with 430 additions and 21 deletions

View file

@ -353,6 +353,9 @@ class ScanDirectoryManager:
For more explanation on the scan naming see `new_scan_dir` For more explanation on the scan naming see `new_scan_dir`
""" """
scan_name = make_name_safe(scan_name) 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 # A regex with the scan name and a group for the numbers
scan_regex = re.compile( scan_regex = re.compile(
"^" + scan_name + "_([0-9]{" + str(SCAN_ZERO_PAD_DIGITS) + "})$" "^" + scan_name + "_([0-9]{" + str(SCAN_ZERO_PAD_DIGITS) + "})$"

View file

@ -18,7 +18,7 @@ from pydantic import BaseModel
import labthings_fastapi as lt 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" IS_WINDOWS = os.name == "nt"
@ -29,6 +29,22 @@ STITCH_TILE_SIZE = 8192
DEFAULT_OVERLAP = 0.1 DEFAULT_OVERLAP = 0.1
DEFAULT_RESIZE = 0.5 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): class StitchingSettings(BaseModel):
"""The data needed to stitch a scan.""" """The data needed to stitch a scan."""
@ -49,17 +65,26 @@ class StitcherValidationError(RuntimeError):
def validate_command(cmd: list[str]) -> None: 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 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. :raises StitcherValidationError: if any element in the command is not safe.
""" """
for element in cmd: for element in cmd:
if element != make_path_safe(element): # Check against forbidden commands (case-insensitive)
if element.lower() in FORBIDDEN_COMMANDS:
raise StitcherValidationError( 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. :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( raise StitcherValidationError(
"Invalid directory path: Contains unsafe characters." "Invalid directory path: Contains unsafe characters."
) )

View file

@ -76,34 +76,131 @@ def requires_lock(
# Compiled regular expressions for unsafe characters # Compiled regular expressions for unsafe characters
# Matches anything that isn't a-z, A-Z, 0-9, _, ., -, :, /, \ # Matches anything that isn't a-z, A-Z, 0-9, _, ., -, :, /, \, space
_WINDOWS_UNSAFE_PATTERN = re.compile(r"[^a-zA-Z0-9_.\-:/\\]") _WINDOWS_UNSAFE_PATTERN = re.compile(r"[^a-zA-Z0-9_.\-:/\ \\]")
# Matches anything that isn't a-z, A-Z, 0-9, _, ., -, \ # Matches anything that isn't a-z, A-Z, 0-9, _, ., -, \, space
_POSIX_UNSAFE_PATTERN = re.compile(r"[^a-zA-Z0-9_.\-/]") _POSIX_UNSAFE_PATTERN = re.compile(r"[^a-zA-Z0-9_.\-/ ]")
# Matches anything that isn't a-z, A-Z, 0-9, _, ., - # Matches anything that isn't a-z, A-Z, 0-9, _, ., -
_NAME_UNSAFE_PATTERN = re.compile(r"[^a-zA-Z0-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 def is_path_safe(unsafe_path_string: str) -> bool:
unsafe filepath. """Check if a file path has any unsafe elements in it.
This ensures compatibility across platforms by preserving only valid characters, The path is not coerced into a safe form because if we ask
including path separators (forward slash on POSIX, and forward for a file to be written to a location we shouldn't be changing
slash/backslash/colon on Windows). that location as we have no "consent" to write to this other location.
:param unsafe_path_string: The original path string to sanitise. :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 = ( unsafe_character_pattern = (
_WINDOWS_UNSAFE_PATTERN _WINDOWS_UNSAFE_PATTERN
if sys.platform.startswith("win") if sys.platform.startswith("win")
else _POSIX_UNSAFE_PATTERN 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: 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 This excludes all path separators, ensuring the result is safe to use as a
standalone filename or identifier component. 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. :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. :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): class VersionData(BaseModel):

View file

@ -167,7 +167,7 @@ def test_bad_scan_names():
scan_dir = scan_dir_manager.new_scan_dir("fake scan") scan_dir = scan_dir_manager.new_scan_dir("fake scan")
assert scan_dir.name == "fake_scan_0001" assert scan_dir.name == "fake_scan_0001"
scan_dir = scan_dir_manager.new_scan_dir("fake scan;rm -rf /;") 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): def test_scan_sequence_and_listing(caplog):
@ -247,6 +247,63 @@ def test_scan_sequence_and_listing(caplog):
assert len(caplog.records) == 0 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(): def test_scan_name_non_sequential():
"""Check new scan has the correct name if the directories are not sequential.""" """Check new scan has the correct name if the directories are not sequential."""
_clear_scan_dir() _clear_scan_dir()

View file

@ -16,11 +16,13 @@ from pydantic import BaseModel
import labthings_fastapi as lt import labthings_fastapi as lt
from openflexure_microscope_server.stitching import ( from openflexure_microscope_server.stitching import (
FORBIDDEN_COMMANDS,
BaseStitcher, BaseStitcher,
FinalStitcher, FinalStitcher,
PreviewStitcher, PreviewStitcher,
StitcherValidationError, StitcherValidationError,
StitchingSettings, StitchingSettings,
validate_command,
) )
from ..shared_utils.lt_test_utils import LabThingsTestEnv 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") 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(): def test_base_stitcher():
"""Test the logic in BaseStitcher. """Test the logic in BaseStitcher.

View file

@ -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<error>.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
)