Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 7 additions & 0 deletions lelab/calibrate.py
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,7 @@
)
from lerobot.utils.utils import init_logging

from .utils.config import is_valid_robot_name
from .utils.devices import safe_disconnect_device

logger = logging.getLogger(__name__)
Expand Down Expand Up @@ -157,6 +158,12 @@ def _cache_recorded_ranges(self, positions: dict[str, float]) -> None:
def start_calibration(self, request: CalibrationRequest) -> dict[str, Any]:
"""Start calibration process"""
try:
# config_file becomes the LeRobot device id, and LeRobot writes the calibration to
# calibration_dir / f"{id}.json", so reject path-traversal characters (/, \, ..) first.
# Same guard as the calibration-config delete endpoint and robot records.
if not is_valid_robot_name(request.config_file):
return {"success": False, "message": "Invalid configuration name"}

if self.status.calibration_active:
return {"success": False, "message": "Calibration already active"}

Expand Down
14 changes: 14 additions & 0 deletions tests/test_calibrate.py
Original file line number Diff line number Diff line change
Expand Up @@ -292,6 +292,20 @@ def test_calibration_manager_rejects_double_start_via_message() -> None:
assert "already" in result.get("message", "").lower()


@pytest.mark.parametrize("config_file", ["", "../escape", "..", "a/b", "a\\b", " padded"])
def test_calibration_manager_rejects_unsafe_config_name(config_file) -> None:
"""config_file names the calibration JSON on disk; unsafe names never start a worker."""
from lelab.calibrate import CalibrationManager, CalibrationRequest

mgr = CalibrationManager()
result = mgr.start_calibration(
CalibrationRequest(device_type="robot", port="/dev/null", config_file=config_file)
)
assert result == {"success": False, "message": "Invalid configuration name"}
assert mgr.status.calibration_active is False
assert mgr.calibration_thread is None


def test_cleanup_device_force_releases_and_clears_when_disconnect_fails() -> None:
"""A failed device.disconnect() must still force-close the port and clear the
device handle — otherwise the COM port stays busy and blocks the next run."""
Expand Down