adding some invariants

Author: CamDavidsonPilon

CHANGELOG.md

@@ -19,6 +19,7 @@
    ```
 
    This gives UI, plugin, and monitoring consumers a lightweight MQTT-native way to observe assignment changes while the leader database remains authoritative.
+ - Bumps the base RPi OS image to 18 Jun 2026 which includes Linux kernel 6.18 (_new_ images only).
 
 #### Bug fixes
 

core/pioreactor/calibrations/__init__.py

@@ -12,6 +12,7 @@
 from pioreactor import types as pt
 from pioreactor.calibrations.registry import CalibrationProtocol  # re-export
 from pioreactor.calibrations.registry import get_calibration_protocols  # re-export
+from pioreactor.structs import artifact_path_component
 from pioreactor.utils import local_persistent_storage
 from pioreactor.whoami import is_testing_env
 
@@ -62,11 +63,13 @@ def load_active_calibration(device: Device) -> structs.AnyCalibration | None:
 
 
 def load_calibration(device: Device, calibration_name: str) -> structs.AnyCalibration:
-    target_file = CALIBRATION_PATH / device / f"{calibration_name}.yaml"
+    valid_device = artifact_path_component(device, "device")
+    valid_calibration_name = artifact_path_component(calibration_name, "calibration_name")
+    target_file = CALIBRATION_PATH / valid_device / f"{valid_calibration_name}.yaml"
 
     if not target_file.is_file():
         raise FileNotFoundError(
-            f"Calibration {calibration_name} was not found in {CALIBRATION_PATH / device}"
+            f"Calibration {calibration_name} was not found in {CALIBRATION_PATH / valid_device}"
         )
     elif target_file.stat().st_size == 0:
         raise FileNotFoundError(f"Calibration {calibration_name} is empty")
@@ -79,7 +82,8 @@ def load_calibration(device: Device, calibration_name: str) -> structs.AnyCalibr
 
 
 def list_of_calibrations_by_device(device: Device) -> list[str]:
-    device_dir = CALIBRATION_PATH / device
+    valid_device = artifact_path_component(device, "device")
+    device_dir = CALIBRATION_PATH / valid_device
     if not device_dir.is_dir():
         return []
     return [file.stem for file in device_dir.glob("*.yaml")]

core/pioreactor/estimators/__init__.py

@@ -11,6 +11,7 @@
 from msgspec.yaml import decode as yaml_decode
 from pioreactor import structs
 from pioreactor import types as pt
+from pioreactor.structs import artifact_path_component
 from pioreactor.utils import local_persistent_storage
 from pioreactor.whoami import is_testing_env
 
@@ -21,6 +22,8 @@
 
 
 def _estimator_path_for(device: str, name: str) -> Path:
+    device = artifact_path_component(device, "device")
+    name = artifact_path_component(name, "estimator_name")
     return ESTIMATOR_PATH / device / f"{name}.yaml"
 
 
@@ -67,7 +70,8 @@ def load_estimator(device: Device, estimator_name: str) -> structs.AnyEstimator:
 
 
 def list_of_estimators_by_device(device: Device) -> list[str]:
-    device_dir = ESTIMATOR_PATH / device
+    valid_device = artifact_path_component(device, "device")
+    device_dir = ESTIMATOR_PATH / valid_device
     if not device_dir.is_dir():
         return []
     return [file.stem for file in device_dir.glob("*.yaml")]

core/pioreactor/structs.py

@@ -16,6 +16,7 @@
 from pioreactor import exc
 from pioreactor import types as pt
 from pioreactor.logging import create_logger
+from pioreactor.utils.files import is_valid_unix_filename
 
 
 T = t.TypeVar("T")
@@ -73,6 +74,18 @@ def type(self) -> str:
 type CalibrationCurveData = PolyFitCoefficients | SplineFitData | AkimaFitData
 
 
+def artifact_path_component(value: str, field: str) -> str:
+    """
+    Calibration and estimator identities are filesystem path components.
+
+    Keep this invariant at the artifact layer so CLI, API, and Huey writers
+    cannot construct paths from malformed names.
+    """
+    if not is_valid_unix_filename(value):
+        raise ValueError(f"{field} must be a valid filename component.")
+    return value
+
+
 class AutomationSettings(JSONPrintedStruct):
     """
     Metadata produced when settings in an automation job change
@@ -226,8 +239,10 @@ def calibration_type(self) -> str:
     def path_on_disk_for_device(self, device: str) -> Path:
         from pioreactor.calibrations import CALIBRATION_PATH
 
-        calibration_dir = CALIBRATION_PATH / device
-        out_file = calibration_dir / f"{self.calibration_name}.yaml"
+        calibration_dir = CALIBRATION_PATH / artifact_path_component(device, "device")
+        out_file = (
+            calibration_dir / f"{artifact_path_component(self.calibration_name, 'calibration_name')}.yaml"
+        )
         return out_file
 
     def save_to_disk_for_device(self, device: str) -> str:
@@ -248,29 +263,37 @@ def set_as_active_calibration_for_device(self, device: str) -> None:
         from pioreactor.utils import local_persistent_storage
 
         logger = create_logger("calibrations", experiment="$experiment")
+        device = artifact_path_component(device, "device")
+        calibration_name = artifact_path_component(self.calibration_name, "calibration_name")
 
         if not self.exists_on_disk_for_device(device):
             self.save_to_disk_for_device(device)
 
         with local_persistent_storage("active_calibrations") as c:
-            c[device] = self.calibration_name
+            c[device] = calibration_name
 
         logger.info(f"Set {self.calibration_name} as active calibration for {device}")
 
     def remove_as_active_calibration_for_device(self, device: str) -> None:
         from pioreactor.utils import local_persistent_storage
 
         logger = create_logger("calibrations", experiment="$experiment")
+        device = artifact_path_component(device, "device")
+        calibration_name = artifact_path_component(self.calibration_name, "calibration_name")
 
         with local_persistent_storage("active_calibrations") as c:
-            if c.get(device) == self.calibration_name:
+            if c.get(device) == calibration_name:
                 del c[device]
                 logger.info(f"Removed {self.calibration_name} as active calibration for {device}")
 
     def exists_on_disk_for_device(self, device: str) -> bool:
         from pioreactor.calibrations import CALIBRATION_PATH
 
-        target_file = CALIBRATION_PATH / device / f"{self.calibration_name}.yaml"
+        target_file = (
+            CALIBRATION_PATH
+            / artifact_path_component(device, "device")
+            / f"{artifact_path_component(self.calibration_name, 'calibration_name')}.yaml"
+        )
 
         return target_file.is_file()
 
@@ -377,7 +400,9 @@ def is_active(self, device: str) -> bool:
         from pioreactor.utils import local_persistent_storage
 
         with local_persistent_storage("active_calibrations") as c:
-            return c.get(device) == self.calibration_name
+            return c.get(artifact_path_component(device, "device")) == artifact_path_component(
+                self.calibration_name, "calibration_name"
+            )
 
 
 class EstimatorBase(Struct, tag_field="estimator_type", kw_only=True):
@@ -392,8 +417,8 @@ def estimator_type(self) -> str:
     def path_on_disk_for_device(self, device: str) -> Path:
         from pioreactor.estimators import ESTIMATOR_PATH
 
-        estimator_dir = ESTIMATOR_PATH / device
-        out_file = estimator_dir / f"{self.estimator_name}.yaml"
+        estimator_dir = ESTIMATOR_PATH / artifact_path_component(device, "device")
+        out_file = estimator_dir / f"{artifact_path_component(self.estimator_name, 'estimator_name')}.yaml"
         return out_file
 
     def save_to_disk_for_device(self, device: str) -> str:
@@ -414,37 +439,47 @@ def set_as_active_calibration_for_device(self, device: str) -> None:
         from pioreactor.utils import local_persistent_storage
 
         logger = create_logger("estimators", experiment="$experiment")
+        device = artifact_path_component(device, "device")
+        estimator_name = artifact_path_component(self.estimator_name, "estimator_name")
 
         if not self.exists_on_disk_for_device(device):
             self.save_to_disk_for_device(device)
 
         with local_persistent_storage("active_estimators") as c:
-            c[device] = self.estimator_name
+            c[device] = estimator_name
 
         logger.info(f"Set {self.estimator_name} as active estimator for {device}")
 
     def remove_as_active_calibration_for_device(self, device: str) -> None:
         from pioreactor.utils import local_persistent_storage
 
         logger = create_logger("estimators", experiment="$experiment")
+        device = artifact_path_component(device, "device")
+        estimator_name = artifact_path_component(self.estimator_name, "estimator_name")
 
         with local_persistent_storage("active_estimators") as c:
-            if c.get(device) == self.estimator_name:
+            if c.get(device) == estimator_name:
                 del c[device]
                 logger.info(f"Removed {self.estimator_name} as active estimator for {device}")
 
     def exists_on_disk_for_device(self, device: str) -> bool:
         from pioreactor.estimators import ESTIMATOR_PATH
 
-        target_file = ESTIMATOR_PATH / device / f"{self.estimator_name}.yaml"
+        target_file = (
+            ESTIMATOR_PATH
+            / artifact_path_component(device, "device")
+            / f"{artifact_path_component(self.estimator_name, 'estimator_name')}.yaml"
+        )
 
         return target_file.is_file()
 
     def is_active(self, device: str) -> bool:
         from pioreactor.utils import local_persistent_storage
 
         with local_persistent_storage("active_estimators") as c:
-            return c.get(device) == self.estimator_name
+            return c.get(artifact_path_component(device, "device")) == artifact_path_component(
+                self.estimator_name, "estimator_name"
+            )
 
 
 class ODCalibration(CalibrationBase, kw_only=True, tag="od"):

core/pioreactor/utils/files.py

@@ -0,0 +1,27 @@
+# -*- coding: utf-8 -*-
+import re
+
+
+_ALLOWED_UNIX_FILENAME = re.compile(r"^[A-Za-z0-9._-]+( [A-Za-z0-9._-]+)*$")
+
+
+def is_valid_unix_filename(name: str, *, max_bytes: int = 255) -> bool:
+    """
+    Return True iff *name* is a single portable filename component.
+
+    Artifact names are path components, not arbitrary strings. Validate them
+    before constructing calibration, estimator, profile, or descriptor paths.
+    """
+    if not name:
+        return False
+    if name in {".", ".."}:
+        return False
+    if name[0] in ".-":
+        return False
+    if "/" in name or "\\" in name:
+        return False
+    if any(ord(c) < 0x20 for c in name):
+        return False
+    if len(name.encode()) > max_bytes:
+        return False
+    return bool(_ALLOWED_UNIX_FILENAME.fullmatch(name))

core/pioreactor/web/tasks.py

@@ -1437,6 +1437,12 @@ def pio_update_app(*args: str, env: dict[str, str] | None = None) -> bool:
 
 @huey.task()
 def rm(path: str) -> bool:
+    """
+    Delete a validated path.
+
+    This is a low-level sink. Callers must first constrain the path to the
+    intended root; this task deliberately does not infer containment.
+    """
     logger.debug(f"Deleting {path}.")
     if whoami.is_testing_env():
         return True
@@ -1543,6 +1549,12 @@ def pios(*args: str, env: dict[str, str] | None = None) -> bool:
 
 @huey.task()
 def save_file(path: str, content: str) -> bool:
+    """
+    Write to a validated path.
+
+    This is a low-level sink. Callers must first constrain the path to the
+    intended root and validate filename components where user input is involved.
+    """
     try:
         with open(path, "w") as f:
             f.write(content)