Security hardening (maziggy/bambuddy-security #43)

backend/app/main.py

@@ -4316,6 +4316,49 @@ async def _announce_plate_check_failed(printer_id: int, printer_name: str, reaso
         logging.getLogger(__name__).warning("[PLATE CHECK] Could not announce the failed check: %s", e)
 
 
+def _expected_print_keys(printer_id: int, filename: str | None, subtask_name: str | None) -> list[tuple[int, str]]:
+    """The ``_expected_prints`` keys a print start may be registered under."""
+    keys: list[tuple[int, str]] = []
+    if subtask_name:
+        keys.append((printer_id, subtask_name))
+        keys.append((printer_id, f"{subtask_name}.3mf"))
+        keys.append((printer_id, f"{subtask_name}.gcode.3mf"))
+    if filename:
+        fname = filename.split("/")[-1] if "/" in filename else filename
+        keys.append((printer_id, fname))
+        # Strip extensions to match
+        base = fname.replace(".gcode", "").replace(".3mf", "")
+        keys.append((printer_id, base))
+        keys.append((printer_id, f"{base}.3mf"))
+    return keys
+
+
+async def _dispatched_by_bambuddy(db, printer_id: int, filename: str | None, subtask_name: str | None) -> bool:
+    """Whether this print start is one Bambuddy sent to the printer."""
+    if any(key in _expected_prints for key in _expected_print_keys(printer_id, filename, subtask_name)):
+        return True
+    from backend.app.models.archive import PrintArchive
+    from backend.app.models.print_queue import PrintQueueItem
+
+    # The queue row the scheduler sets to printing before it sends -- but only
+    # when it was sent as this print: a row left printing by a run whose end
+    # was missed names another file.
+    dispatched_names = (
+        await db.execute(
+            select(PrintArchive.filename)
+            .join(PrintQueueItem, PrintQueueItem.archive_id == PrintArchive.id)
+            .where(PrintQueueItem.printer_id == printer_id, PrintQueueItem.status == "printing")
+        )
+    ).scalars()
+    observed = [name for name in (subtask_name, (filename or "").rsplit("/", 1)[-1]) if name]
+    return any(
+        _subtask_names_match(_subtask_name_from_filename(dispatched), _subtask_name_from_filename(seen))
+        for dispatched in dispatched_names
+        if dispatched
+        for seen in observed
+    )
+
+
 async def on_print_start(printer_id: int, data: dict):
     """Handle print start - archive the 3MF file immediately."""
     logger = logging.getLogger(__name__)
@@ -4581,7 +4624,14 @@ async def on_print_start(printer_id: int, data: dict):
         # matching completion is suppressed in on_print_complete for the same
         # reason. Checked before the auto-archive switches, whose "off" path
         # still notifies.
-        if is_internal_printer_job(data.get("filename", ""), data.get("subtask_name", "")):
+        # A /usr/ path is the printer's own file. A name alone is not proof: a
+        # print Bambuddy dispatched is never one, whatever it is called.
+        # Dispatched = an expected print, or, after a restart that lost those,
+        # a queue job printing on the printer.
+        if is_internal_printer_job(data.get("filename", ""), data.get("subtask_name", "")) and (
+            (data.get("filename") or "").startswith("/usr/")
+            or not await _dispatched_by_bambuddy(db, printer_id, data.get("filename", ""), data.get("subtask_name", ""))
+        ):
             logger.info(
                 "[CALLBACK] Skipping archive — internal printer job detected: filename=%s, subtask=%s",
                 data.get("filename", ""),
@@ -4665,19 +4715,7 @@ async def on_print_start(printer_id: int, data: dict):
             return
 
         # Check if this is an expected print from reprint/scheduled
-        # Build list of possible keys to check
-        expected_keys = []
-        if subtask_name:
-            expected_keys.append((printer_id, subtask_name))
-            expected_keys.append((printer_id, f"{subtask_name}.3mf"))
-            expected_keys.append((printer_id, f"{subtask_name}.gcode.3mf"))
-        if filename:
-            fname = filename.split("/")[-1] if "/" in filename else filename
-            expected_keys.append((printer_id, fname))
-            # Strip extensions to match
-            base = fname.replace(".gcode", "").replace(".3mf", "")
-            expected_keys.append((printer_id, base))
-            expected_keys.append((printer_id, f"{base}.3mf"))
+        expected_keys = _expected_print_keys(printer_id, filename, subtask_name)
 
         expected_archive_id = None
         for key in expected_keys:

backend/tests/integration/test_dispatched_print_named_like_calibration.py

@@ -0,0 +1,137 @@
+"""A print Bambuddy dispatched is adopted whatever it is called; the printer's
+own calibration runs are still left out of the archive."""
+
+import pytest
+
+from backend.app import main as main_module
+from backend.app.main import register_expected_print
+from backend.app.models.archive import PrintArchive
+from backend.tests.integration import test_external_print_confirmation_1898 as harness
+from backend.tests.integration.test_external_print_confirmation_1898 import (  # noqa: F401
+    _clear_print_state,
+    _drive_print_start,
+)
+
+pytestmark = [pytest.mark.asyncio, pytest.mark.integration]
+
+NAME = "pa_line_calib_mode"
+
+
+@pytest.fixture(autouse=True)
+def _calibration_name(monkeypatch):
+    monkeypatch.setattr(harness, "SUBTASK", NAME)
+    monkeypatch.setattr(main_module.printer_manager, "get_current_print_user", lambda printer_id: None)
+    main_module._owned_print_archive_ids.clear()
+    yield
+    main_module._owned_print_archive_ids.clear()
+
+
+async def test_a_dispatched_print_named_like_a_calibration_is_adopted(test_engine, db_session, printer_factory):
+    printer_id = (await printer_factory()).id
+    archive = PrintArchive(
+        printer_id=printer_id,
+        filename=f"{NAME}.gcode.3mf",
+        file_path=f"archives/test/{NAME}.gcode.3mf",
+        file_size=1,
+        print_name=NAME,
+        status="pending",
+    )
+    db_session.add(archive)
+    await db_session.commit()
+    archive_id = archive.id
+    register_expected_print(printer_id, f"{NAME}.gcode.3mf", archive_id)
+
+    await _drive_print_start(test_engine, type("P", (), {"id": printer_id})(), download_ok=False)
+
+    db_session.expire_all()
+    adopted = await db_session.get(PrintArchive, archive_id)
+    assert adopted.status == "printing"
+    assert archive_id in main_module._owned_print_archive_ids
+
+
+async def test_the_printers_own_calibration_is_still_skipped(test_engine, db_session, printer_factory):
+    """Nothing dispatched: a calibration run started on the printer or from the
+    slicer is not archived, as before."""
+    printer_id = (await printer_factory()).id
+
+    await _drive_print_start(test_engine, type("P", (), {"id": printer_id})(), download_ok=False)
+
+    db_session.expire_all()
+    from sqlalchemy import select
+
+    rows = (await db_session.execute(select(PrintArchive).where(PrintArchive.printer_id == printer_id))).scalars().all()
+    assert rows == []
+
+
+async def test_after_a_restart_the_printing_queue_job_says_it_is_bambuddys(test_engine, db_session, printer_factory):
+    """The expected entry is in memory only; the queue row is what survives."""
+    from sqlalchemy import select
+
+    from backend.app.models.print_queue import PrintQueueItem
+
+    printer_id = (await printer_factory()).id
+    archive = PrintArchive(
+        printer_id=printer_id,
+        filename=f"{NAME}.gcode.3mf",
+        file_path="archives/x.3mf",
+        file_size=1,
+        status="printing",
+        billing_run_id="run-9",
+    )
+    db_session.add(archive)
+    await db_session.flush()
+    db_session.add(PrintQueueItem(printer_id=printer_id, status="printing", position=1, archive_id=archive.id))
+    await db_session.commit()
+
+    await _drive_print_start(
+        test_engine, type("P", (), {"id": printer_id})(), download_ok=False, state_fields={"progress": 40}
+    )
+
+    db_session.expire_all()
+    rows = (await db_session.execute(select(PrintArchive).where(PrintArchive.printer_id == printer_id))).scalars().all()
+    assert [row.status for row in rows] == ["printing"]
+    assert rows[0].id in main_module._owned_print_archive_ids
+
+
+async def test_a_queue_job_of_another_file_does_not_count(test_engine, db_session, printer_factory):
+    """A row left printing by a run whose end was missed names another file."""
+    from sqlalchemy import select
+
+    from backend.app.models.print_queue import PrintQueueItem
+
+    printer_id = (await printer_factory()).id
+    other = PrintArchive(
+        printer_id=printer_id, filename="Bracket.gcode.3mf", file_path="archives/b.3mf", file_size=1, status="printing"
+    )
+    db_session.add(other)
+    await db_session.flush()
+    db_session.add(PrintQueueItem(printer_id=printer_id, status="printing", position=1, archive_id=other.id))
+    await db_session.commit()
+    other_id = other.id
+
+    await _drive_print_start(test_engine, type("P", (), {"id": printer_id})(), download_ok=False)
+
+    db_session.expire_all()
+    rows = (await db_session.execute(select(PrintArchive).where(PrintArchive.printer_id == printer_id))).scalars().all()
+    assert [row.id for row in rows] == [other_id]
+
+
+async def test_a_system_partition_file_is_the_printers_whatever_the_queue_says(
+    test_engine, db_session, printer_factory, monkeypatch
+):
+    """Nothing a user prints lives under /usr/: Bambuddy's own Calibrate runs it."""
+    from sqlalchemy import select
+
+    from backend.app.models.print_queue import PrintQueueItem
+
+    monkeypatch.setattr(harness, "SUBTASK", "auto_cali_for_user_param")
+    monkeypatch.setattr(harness, "DISPATCH", "/usr/etc/print/O1C2/auto_cali_for_user_param.gcode")
+    printer_id = (await printer_factory()).id
+    db_session.add(PrintQueueItem(printer_id=printer_id, status="printing", position=1))
+    await db_session.commit()
+
+    await _drive_print_start(test_engine, type("P", (), {"id": printer_id})(), download_ok=False)
+
+    db_session.expire_all()
+    rows = (await db_session.execute(select(PrintArchive).where(PrintArchive.printer_id == printer_id))).scalars().all()
+    assert rows == []

backend/tests/unit/test_internal_printer_jobs.py

@@ -153,6 +153,7 @@ class TestPrintStartSkipsTheCalibration:
             mock_session = AsyncMock()
             mock_session.__aenter__ = AsyncMock(return_value=mock_session)
             mock_session.__aexit__ = AsyncMock()
+            mock_session.scalar = AsyncMock(return_value=None)  # no queue job of Bambuddy's printing
             mock_session.execute = AsyncMock(
                 return_value=MagicMock(scalar_one_or_none=MagicMock(return_value=mock_printer))
             )
@@ -184,6 +185,7 @@ class TestPrintStartSkipsTheCalibration:
             mock_session = AsyncMock()
             mock_session.__aenter__ = AsyncMock(return_value=mock_session)
             mock_session.__aexit__ = AsyncMock()
+            mock_session.scalar = AsyncMock(return_value=None)  # no queue job of Bambuddy's printing
             mock_session.execute = AsyncMock(
                 return_value=MagicMock(scalar_one_or_none=MagicMock(return_value=mock_printer))
             )