Security hardening (maziggy/bambuddy-security #32)

backend/app/api/routes/ha_sensors.py

@@ -6,9 +6,14 @@ from fastapi import APIRouter, Depends, HTTPException
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from backend.app.core.auth import RequirePermissionIfAuthEnabled, RequirePrinterPermissionIfAuthEnabled
+from backend.app.core.auth import (
+    RequestPrinterScope,
+    RequirePermissionIfAuthEnabled,
+    RequirePrinterPermissionIfAuthEnabled,
+)
 from backend.app.core.database import get_db
 from backend.app.core.permissions import Permission
+from backend.app.core.printer_scope import PrinterScope
 from backend.app.models.printer import Printer
 from backend.app.models.printer_ha_sensor import PrinterHASensor
 from backend.app.models.user import User
@@ -49,14 +54,25 @@ async def _refresh_quietly(sensor: PrinterHASensor, db: AsyncSession) -> None:
         logger.warning("Could not read %s right after saving it: %s", sensor.entity_id, e)
 
 
+async def _load_sensor(db: AsyncSession, sensor_id: int, printer_scope: PrinterScope) -> PrinterHASensor:
+    """The sensor, if its printer is one the caller may use; otherwise the 404 of a missing one."""
+    sensor = await db.get(PrinterHASensor, sensor_id)
+    if not sensor or not printer_scope.allows(sensor.printer_id):
+        raise HTTPException(404, "Sensor not found")
+    return sensor
+
+
 @router.get("/", response_model=list[PrinterHASensorResponse])
 async def list_ha_sensors(
     printer_id: int | None = None,
     db: AsyncSession = Depends(get_db),
     _: User | None = _READ,
+    printer_scope: PrinterScope = RequestPrinterScope,
 ):
     """List configured sensors, grouped by printer and in display order."""
     query = select(PrinterHASensor)
+    if (scope_clause := printer_scope.where(PrinterHASensor.printer_id)) is not None:
+        query = query.where(scope_clause)
     if printer_id is not None:
         query = query.where(PrinterHASensor.printer_id == printer_id)
     result = await db.execute(query.order_by(PrinterHASensor.printer_id, PrinterHASensor.sort_order))
@@ -133,8 +149,10 @@ async def create_ha_sensor(
     data: PrinterHASensorCreate,
     db: AsyncSession = Depends(get_db),
     _: User | None = _CREATE,
+    printer_scope: PrinterScope = RequestPrinterScope,
 ):
     """Bind a Home Assistant entity to a printer."""
+    printer_scope.ensure(data.printer_id)
     printer = await db.get(Printer, data.printer_id)
     if not printer:
         raise HTTPException(404, "Printer not found")
@@ -167,11 +185,9 @@ async def get_ha_sensor(
     sensor_id: int,
     db: AsyncSession = Depends(get_db),
     _: User | None = _READ,
+    printer_scope: PrinterScope = RequestPrinterScope,
 ):
-    sensor = await db.get(PrinterHASensor, sensor_id)
-    if not sensor:
-        raise HTTPException(404, "Sensor not found")
-    return sensor
+    return await _load_sensor(db, sensor_id, printer_scope)
 
 
 @router.patch("/{sensor_id}", response_model=PrinterHASensorResponse)
@@ -180,10 +196,9 @@ async def update_ha_sensor(
     data: PrinterHASensorUpdate,
     db: AsyncSession = Depends(get_db),
     _: User | None = _UPDATE,
+    printer_scope: PrinterScope = RequestPrinterScope,
 ):
-    sensor = await db.get(PrinterHASensor, sensor_id)
-    if not sensor:
-        raise HTTPException(404, "Sensor not found")
+    sensor = await _load_sensor(db, sensor_id, printer_scope)
 
     updates = data.model_dump(exclude_unset=True)
 
@@ -226,10 +241,9 @@ async def delete_ha_sensor(
     sensor_id: int,
     db: AsyncSession = Depends(get_db),
     _: User | None = _DELETE,
+    printer_scope: PrinterScope = RequestPrinterScope,
 ):
-    sensor = await db.get(PrinterHASensor, sensor_id)
-    if not sensor:
-        raise HTTPException(404, "Sensor not found")
+    sensor = await _load_sensor(db, sensor_id, printer_scope)
 
     name = sensor.name
     await db.delete(sensor)

backend/tests/integration/test_ha_sensor_scope.py

@@ -0,0 +1,100 @@
+"""Home Assistant printer sensors keep to the caller's printers."""
+
+from __future__ import annotations
+
+from unittest.mock import AsyncMock, patch
+
+import pytest
+from httpx import AsyncClient
+
+from backend.app.services.ha_sensor_manager import ha_sensor_manager
+from backend.tests.integration.test_printer_scope_1727 import _admin_token, _auth, _group, _user
+
+pytestmark = [pytest.mark.asyncio, pytest.mark.integration]
+
+PERMS = ["printers:read", "smart_plugs:read", "smart_plugs:create", "smart_plugs:update", "smart_plugs:delete"]
+
+
+@pytest.fixture(autouse=True)
+def _no_live_ha():
+    with patch.object(ha_sensor_manager, "refresh_one", AsyncMock()):
+        yield
+
+
+def _door(printer_id: int, entity: str = "binary_sensor.door") -> dict:
+    return {"name": "Door", "entity_id": entity, "kind": "binary", "alert_state": "on", "printer_id": printer_id}
+
+
+async def _setup(client: AsyncClient, printer_factory):
+    a, b = await printer_factory(name="A"), await printer_factory(name="B")
+    token = await _admin_token(client)
+    admin = _auth(token)
+    perms = await _group(client, token, "perms", permissions=PERMS)
+    team = await _group(client, token, "team", printer_ids=[a.id])
+    member = _auth((await _user(client, token, "member", [perms, team]))[0])
+    ids = {}
+    for name, printer in (("ours", a.id), ("theirs", b.id)):
+        created = await client.post("/api/v1/ha-sensors/", headers=admin, json=_door(printer))
+        assert created.status_code == 200, created.text
+        ids[name] = created.json()["id"]
+    return a.id, b.id, admin, member, ids
+
+
+async def test_a_member_lists_and_reads_only_sensors_of_their_printers(async_client: AsyncClient, printer_factory):
+    a_id, b_id, admin, member, ids = await _setup(async_client, printer_factory)
+
+    listed = await async_client.get("/api/v1/ha-sensors/", headers=member)
+    filtered = await async_client.get("/api/v1/ha-sensors/", headers=member, params={"printer_id": b_id})
+    ours = await async_client.get(f"/api/v1/ha-sensors/{ids['ours']}", headers=member)
+    theirs = await async_client.get(f"/api/v1/ha-sensors/{ids['theirs']}", headers=member)
+    missing = await async_client.get("/api/v1/ha-sensors/999999", headers=member)
+    everything = await async_client.get("/api/v1/ha-sensors/", headers=admin)
+
+    assert [s["id"] for s in listed.json()] == [ids["ours"]]
+    assert filtered.json() == []
+    assert ours.status_code == 200
+    assert (
+        (theirs.status_code, theirs.json())
+        == (missing.status_code, missing.json())
+        == (
+            404,
+            {"detail": "Sensor not found"},
+        )
+    )
+    assert len(everything.json()) == 2
+
+
+async def test_a_member_changes_only_sensors_of_their_printers(async_client: AsyncClient, printer_factory):
+    a_id, b_id, admin, member, ids = await _setup(async_client, printer_factory)
+
+    patched_theirs = await async_client.patch(
+        f"/api/v1/ha-sensors/{ids['theirs']}", headers=member, json={"block_print": True}
+    )
+    deleted_theirs = await async_client.delete(f"/api/v1/ha-sensors/{ids['theirs']}", headers=member)
+    patched_ours = await async_client.patch(f"/api/v1/ha-sensors/{ids['ours']}", headers=member, json={"name": "Lid"})
+    deleted_ours = await async_client.delete(f"/api/v1/ha-sensors/{ids['ours']}", headers=member)
+
+    assert (patched_theirs.status_code, deleted_theirs.status_code) == (404, 404)
+    assert (patched_ours.status_code, deleted_ours.status_code) == (200, 200)
+    left = (await async_client.get(f"/api/v1/ha-sensors/{ids['theirs']}", headers=admin)).json()
+    assert left["block_print"] is False
+
+
+async def test_a_member_binds_sensors_to_their_printers_only(async_client: AsyncClient, printer_factory):
+    a_id, b_id, admin, member, ids = await _setup(async_client, printer_factory)
+
+    theirs = await async_client.post("/api/v1/ha-sensors/", headers=member, json=_door(b_id, "binary_sensor.lid"))
+    missing = await async_client.post("/api/v1/ha-sensors/", headers=member, json=_door(999_999, "binary_sensor.lid"))
+    ours = await async_client.post("/api/v1/ha-sensors/", headers=member, json=_door(a_id, "binary_sensor.lid"))
+
+    assert (
+        (theirs.status_code, theirs.json())
+        == (missing.status_code, missing.json())
+        == (
+            404,
+            {"detail": "Printer not found"},
+        )
+    )
+    assert ours.status_code == 200
+    on_b = (await async_client.get("/api/v1/ha-sensors/", headers=admin, params={"printer_id": b_id})).json()
+    assert [s["entity_id"] for s in on_b] == ["binary_sensor.door"]