Security hardening (maziggy/bambuddy-security #28)

.basedpyright/baseline.json

@@ -10,16 +10,6 @@
                 }
             }
         ],
-        "./backend/app/api/routes/ams_history.py": [
-            {
-                "code": "reportAttributeAccessIssue",
-                "range": {
-                    "startColumn": 35,
-                    "endColumn": 41,
-                    "lineCount": 1
-                }
-            }
-        ],
         "./backend/app/api/routes/archives.py": [
             {
                 "code": "reportArgumentType",
@@ -1224,16 +1214,6 @@
                 }
             }
         ],
-        "./backend/app/api/routes/printer_sensor_history.py": [
-            {
-                "code": "reportAttributeAccessIssue",
-                "range": {
-                    "startColumn": 39,
-                    "endColumn": 45,
-                    "lineCount": 1
-                }
-            }
-        ],
         "./backend/app/api/routes/printers.py": [
             {
                 "code": "reportArgumentType",

backend/app/api/routes/ams_history.py

@@ -102,36 +102,3 @@ async def get_ams_history(
         max_temperature=stats.max_temp,
         avg_temperature=round(stats.avg_temp, 1) if stats.avg_temp is not None else None,
     )
-
-
-@router.delete("/{printer_id}")
-async def delete_old_history(
-    printer_id: int,
-    days: int = Query(default=30, ge=1, le=365, description="Delete data older than X days"),
-    db: AsyncSession = Depends(get_db),
-    _: User | None = RequirePrinterPermissionIfAuthEnabled(Permission.AMS_HISTORY_READ),
-):
-    """Delete old AMS history data for a printer."""
-    cutoff = datetime.now(timezone.utc) - timedelta(days=days)
-
-    result = await db.execute(
-        select(func.count(AMSSensorHistory.id)).where(
-            and_(
-                AMSSensorHistory.printer_id == printer_id,
-                AMSSensorHistory.recorded_at < cutoff,
-            )
-        )
-    )
-    count = result.scalar()
-
-    await db.execute(
-        AMSSensorHistory.__table__.delete().where(
-            and_(
-                AMSSensorHistory.printer_id == printer_id,
-                AMSSensorHistory.recorded_at < cutoff,
-            )
-        )
-    )
-    await db.commit()
-
-    return {"deleted": count, "message": f"Deleted {count} records older than {days} days"}

backend/app/api/routes/bug_report.py

@@ -73,7 +73,7 @@ async def start_logging(
 @router.post("/stop-logging", response_model=StopLoggingResponse)
 async def stop_logging(
     was_debug: bool = Query(default=False),
-    _: User | None = RequirePermissionIfAuthEnabled(Permission.SETTINGS_READ),
+    _: User | None = RequirePermissionIfAuthEnabled(Permission.SETTINGS_UPDATE),
 ):
     """Collect logs and restore previous log level."""
     logs = await _get_recent_sanitized_logs()

backend/app/api/routes/camera.py

@@ -1894,7 +1894,7 @@ async def calibrate_plate_detection(
     label: str | None = None,
     use_external: bool | None = None,
     db: AsyncSession = Depends(get_db),
-    _: User | None = RequirePrinterPermissionIfAuthEnabled(Permission.CAMERA_VIEW),
+    _: User | None = RequirePrinterPermissionIfAuthEnabled(Permission.PRINTERS_UPDATE),
 ):
     """Calibrate plate detection by capturing a reference image of the empty plate.
 
@@ -1967,7 +1967,7 @@ async def delete_plate_calibration(
     printer_id: int,
     plate_type: str | None = None,
     db: AsyncSession = Depends(get_db),
-    _: User | None = RequirePrinterPermissionIfAuthEnabled(Permission.CAMERA_VIEW),
+    _: User | None = RequirePrinterPermissionIfAuthEnabled(Permission.PRINTERS_UPDATE),
 ):
     """Delete the plate detection calibration for a printer and plate type.
 
@@ -2117,7 +2117,7 @@ async def update_reference_label(
     index: int,
     label: str,
     db: AsyncSession = Depends(get_db),
-    _: User | None = RequirePrinterPermissionIfAuthEnabled(Permission.CAMERA_VIEW),
+    _: User | None = RequirePrinterPermissionIfAuthEnabled(Permission.PRINTERS_UPDATE),
 ):
     """Update the label for a calibration reference."""
     from backend.app.services.plate_detection import PlateDetector, is_plate_detection_available
@@ -2142,7 +2142,7 @@ async def delete_reference(
     printer_id: int,
     index: int,
     db: AsyncSession = Depends(get_db),
-    _: User | None = RequirePrinterPermissionIfAuthEnabled(Permission.CAMERA_VIEW),
+    _: User | None = RequirePrinterPermissionIfAuthEnabled(Permission.PRINTERS_UPDATE),
 ):
     """Delete a specific calibration reference."""
     from backend.app.services.plate_detection import PlateDetector, is_plate_detection_available

backend/app/api/routes/printer_sensor_history.py

@@ -105,36 +105,3 @@ async def get_printer_sensor_history(
         )
 
     return PrinterSensorHistoryResponse(printer_id=printer_id, series=series_out)
-
-
-@router.delete("/{printer_id}")
-async def delete_old_history(
-    printer_id: int,
-    days: int = Query(default=30, ge=1, le=365, description="Delete data older than X days"),
-    db: AsyncSession = Depends(get_db),
-    _: User | None = RequirePrinterPermissionIfAuthEnabled(Permission.PRINTER_SENSOR_HISTORY_READ),
-):
-    """Delete old printer sensor history for a printer."""
-    cutoff = datetime.now(timezone.utc) - timedelta(days=days)
-
-    result = await db.execute(
-        select(func.count(PrinterSensorHistory.id)).where(
-            and_(
-                PrinterSensorHistory.printer_id == printer_id,
-                PrinterSensorHistory.recorded_at < cutoff,
-            )
-        )
-    )
-    count = result.scalar()
-
-    await db.execute(
-        PrinterSensorHistory.__table__.delete().where(
-            and_(
-                PrinterSensorHistory.printer_id == printer_id,
-                PrinterSensorHistory.recorded_at < cutoff,
-            )
-        )
-    )
-    await db.commit()
-
-    return {"deleted": count, "message": f"Deleted {count} records older than {days} days"}

backend/app/api/routes/smart_plugs.py

@@ -396,7 +396,7 @@ class DiscoveredTasmotaDevice(BaseModel):
 @router.post("/discover/scan", response_model=TasmotaScanStatus)
 async def start_tasmota_scan(
     request: TasmotaScanRequest | None = Body(default=None),
-    _: User | None = RequirePermissionIfAuthEnabled(Permission.SMART_PLUGS_READ),
+    _: User | None = RequirePermissionIfAuthEnabled(Permission.SMART_PLUGS_CREATE),
 ):
     """Start an IP range scan for Tasmota devices.
 
@@ -437,7 +437,7 @@ async def get_tasmota_scan_status(
 
 @router.post("/discover/stop", response_model=TasmotaScanStatus)
 async def stop_tasmota_scan(
-    _: User | None = RequirePermissionIfAuthEnabled(Permission.SMART_PLUGS_READ),
+    _: User | None = RequirePermissionIfAuthEnabled(Permission.SMART_PLUGS_CREATE),
 ):
     """Stop the current Tasmota scan."""
     tasmota_scanner.stop()

backend/app/api/routes/spoolbuddy.py

@@ -1341,7 +1341,7 @@ async def queue_diagnostic(
     device_id: str,
     diagnostic: str,
     db: AsyncSession = Depends(get_db),
-    _: User | None = RequirePermissionIfAuthEnabled(Permission.INVENTORY_READ),
+    _: User | None = RequirePermissionIfAuthEnabled(Permission.INVENTORY_UPDATE),
 ):
     """Queue a hardware diagnostic to run on the SpoolBuddy device.
 

backend/tests/integration/test_ams_history_api.py

@@ -178,26 +178,13 @@ class TestAMSHistoryAPI:
 
     @pytest.mark.asyncio
     @pytest.mark.integration
-    async def test_delete_old_history(
+    async def test_history_cannot_be_deleted_through_the_api(
         self, async_client: AsyncClient, ams_history_factory, printer_factory, db_session
     ):
-        """Verify old history can be deleted."""
+        """Old history goes by the retention setting only; there is no delete route."""
         printer = await printer_factory()
-        # Create an old record
         await ams_history_factory(printer_id=printer.id, recorded_at=datetime.now() - timedelta(days=60))
 
-        # Delete records older than 30 days
         response = await async_client.delete(f"/api/v1/ams-history/{printer.id}", params={"days": 30})
-        assert response.status_code == 200
-        data = response.json()
-        assert data["deleted"] >= 1
 
-    @pytest.mark.asyncio
-    @pytest.mark.integration
-    async def test_delete_old_history_no_records(self, async_client: AsyncClient, printer_factory, db_session):
-        """Verify delete with no old records returns 0."""
-        printer = await printer_factory()
-        response = await async_client.delete(f"/api/v1/ams-history/{printer.id}", params={"days": 30})
-        assert response.status_code == 200
-        data = response.json()
-        assert data["deleted"] == 0
+        assert response.status_code == 405

backend/tests/integration/test_writes_need_write_permissions.py

@@ -0,0 +1,81 @@
+"""Routes that change shared state need a write permission, not a read one.
+
+Each route is called by a user holding only the permission it used to accept
+(refused) and by one holding the permission it needs now (let through; it may
+still fail later for its own reasons, e.g. no camera or no such device).
+"""
+
+from __future__ import annotations
+
+from unittest.mock import AsyncMock, patch
+
+import pytest
+from httpx import AsyncClient
+
+from backend.tests.integration.test_printer_scope_1727 import _admin_token, _auth, _group, _user
+
+pytestmark = [pytest.mark.asyncio, pytest.mark.integration]
+
+# (method, path template, permission that used to suffice, permission needed now)
+ROUTES = [
+    ("POST", "/api/v1/printers/{printer}/camera/plate-detection/calibrate", "camera:view", "printers:update"),
+    ("DELETE", "/api/v1/printers/{printer}/camera/plate-detection/calibrate", "camera:view", "printers:update"),
+    ("PUT", "/api/v1/printers/{printer}/camera/plate-detection/references/0", "camera:view", "printers:update"),
+    ("DELETE", "/api/v1/printers/{printer}/camera/plate-detection/references/0", "camera:view", "printers:update"),
+    ("POST", "/api/v1/bug-report/stop-logging?was_debug=true", "settings:read", "settings:update"),
+    ("POST", "/api/v1/smart-plugs/discover/scan", "smart_plugs:read", "smart_plugs:create"),
+    ("POST", "/api/v1/smart-plugs/discover/stop", "smart_plugs:read", "smart_plugs:create"),
+    ("POST", "/api/v1/spoolbuddy/diagnostics/no-such-device/run", "inventory:read", "inventory:update"),
+]
+
+
+async def _call(client: AsyncClient, headers: dict[str, str], method: str, path: str):
+    kwargs = {"json": {"label": "x"}} if method == "PUT" else {}
+    return await client.request(method, path, headers=headers, **kwargs)
+
+
+@pytest.mark.parametrize(("method", "path", "read_permission", "write_permission"), ROUTES)
+async def test_a_read_permission_no_longer_changes_anything(
+    async_client: AsyncClient, printer_factory, method, path, read_permission, write_permission
+):
+    printer = await printer_factory()
+    admin = await _admin_token(async_client)
+    reader = _auth(
+        (
+            await _user(
+                async_client,
+                admin,
+                "reader",
+                [await _group(async_client, admin, "r", permissions=["printers:read", read_permission])],
+            )
+        )[0]
+    )
+    writer = _auth(
+        (
+            await _user(
+                async_client,
+                admin,
+                "writer",
+                [
+                    await _group(
+                        async_client, admin, "w", permissions=["printers:read", read_permission, write_permission]
+                    )
+                ],
+            )
+        )[0]
+    )
+    url = path.format(printer=printer.id)
+
+    # Keep the let-through calls from touching the network or the log level.
+    with (
+        patch("backend.app.api.routes.smart_plugs.tasmota_scanner") as scanner,
+        patch("backend.app.api.routes.bug_report._get_recent_sanitized_logs", new=AsyncMock(return_value="")),
+    ):
+        scanner.scan_range = AsyncMock()
+        scanner.progress = (0, 0)
+        scanner.is_running = False
+        refused = await _call(async_client, reader, method, url)
+        allowed = await _call(async_client, writer, method, url)
+
+    assert refused.status_code == 403, refused.text
+    assert allowed.status_code != 403, allowed.text

backend/tests/unit/test_printer_sensor_history.py

@@ -116,27 +116,17 @@ async def test_get_clamps_to_hours_window(async_client: AsyncClient, db_session:
 
 
 @pytest.mark.asyncio
-async def test_delete_removes_old_rows(async_client: AsyncClient, db_session: AsyncSession):
-    """DELETE removes rows older than `days` for the given printer only."""
-    keep_printer = Printer(name="Keep", serial_number="KEEP-001", ip_address="10.0.0.20", access_code="12345678")
-    other_printer = Printer(name="Other", serial_number="OTHER-001", ip_address="10.0.0.21", access_code="12345678")
-    db_session.add_all([keep_printer, other_printer])
+async def test_history_cannot_be_deleted_through_the_api(async_client: AsyncClient, db_session: AsyncSession):
+    """Old history goes by the retention setting only; there is no delete route."""
+    printer = Printer(name="Keep", serial_number="KEEP-001", ip_address="10.0.0.20", access_code="12345678")
+    db_session.add(printer)
     await db_session.commit()
-    await db_session.refresh(keep_printer)
-    await db_session.refresh(other_printer)
-
-    now = datetime.now(timezone.utc)
-    old = now - timedelta(days=40)
-    db_session.add(PrinterSensorHistory(printer_id=keep_printer.id, sensor_kind="bed", value=60.0, recorded_at=old))
-    db_session.add(PrinterSensorHistory(printer_id=keep_printer.id, sensor_kind="bed", value=60.0, recorded_at=now))
-    db_session.add(PrinterSensorHistory(printer_id=other_printer.id, sensor_kind="bed", value=60.0, recorded_at=old))
+    await db_session.refresh(printer)
+    old = datetime.now(timezone.utc) - timedelta(days=40)
+    db_session.add(PrinterSensorHistory(printer_id=printer.id, sensor_kind="bed", value=60.0, recorded_at=old))
     await db_session.commit()
 
-    response = await async_client.delete(f"/api/v1/printer-sensor-history/{keep_printer.id}?days=30")
-    assert response.status_code == 200
-    assert response.json()["deleted"] == 1
+    response = await async_client.delete(f"/api/v1/printer-sensor-history/{printer.id}?days=30")
 
-    # other printer's old row untouched.
-    rows = (await db_session.execute(select(PrinterSensorHistory))).scalars().all()
-    kinds_left = sorted((r.printer_id, r.value) for r in rows)
-    assert kinds_left == sorted([(keep_printer.id, 60.0), (other_printer.id, 60.0)])
+    assert response.status_code == 405
+    assert len((await db_session.execute(select(PrinterSensorHistory))).scalars().all()) == 1