Security hardening (maziggy/bambuddy-security #37)

backend/app/api/routes/archives.py

@@ -1782,7 +1782,7 @@ async def get_archive(
             Permission.ARCHIVES_READ_OWN,
         )
     ),
-    printer_scope: PrinterScope = MediaOrRequestPrinterScope,
+    printer_scope: PrinterScope = RequestPrinterScope,
 ):
     """Get a specific archive."""
     user, can_read_all = auth_result
@@ -1812,7 +1812,7 @@ async def get_archive_delete_impact(
             Permission.ARCHIVES_READ_OWN,
         )
     ),
-    printer_scope: PrinterScope = MediaOrRequestPrinterScope,
+    printer_scope: PrinterScope = RequestPrinterScope,
 ):
     """Pre-flight for the delete-confirm modal (#1734).
 
@@ -1842,7 +1842,7 @@ async def list_archive_runs(
             Permission.ARCHIVES_READ_OWN,
         )
     ),
-    printer_scope: PrinterScope = MediaOrRequestPrinterScope,
+    printer_scope: PrinterScope = RequestPrinterScope,
 ):
     """List PrintLogEntry rows for this archive — one per print event.
 
@@ -1875,7 +1875,7 @@ async def find_similar_archives(
             Permission.ARCHIVES_READ_OWN,
         )
     ),
-    printer_scope: PrinterScope = MediaOrRequestPrinterScope,
+    printer_scope: PrinterScope = RequestPrinterScope,
 ):
     """Find archives with similar settings for comparison.
 
@@ -2033,7 +2033,7 @@ async def toggle_favorite(
             Permission.ARCHIVES_UPDATE_OWN,
         )
     ),
-    printer_scope: PrinterScope = MediaOrRequestPrinterScope,
+    printer_scope: PrinterScope = RequestPrinterScope,
 ):
     """Toggle favorite status for an archive."""
     user, can_modify_all = auth_result
@@ -2310,7 +2310,7 @@ async def get_archive_duplicates(
             Permission.ARCHIVES_READ_OWN,
         )
     ),
-    printer_scope: PrinterScope = MediaOrRequestPrinterScope,
+    printer_scope: PrinterScope = RequestPrinterScope,
 ):
     """Get duplicates for a specific archive."""
     user, can_read_all = auth_result
@@ -2451,7 +2451,7 @@ async def download_archive(
             Permission.ARCHIVES_READ_OWN,
         )
     ),
-    printer_scope: PrinterScope = MediaOrRequestPrinterScope,
+    printer_scope: PrinterScope = RequestPrinterScope,
 ):
     """Download the 3MF file."""
     user, can_read_all = auth_result
@@ -2484,7 +2484,7 @@ async def download_archive_with_filename(
             Permission.ARCHIVES_READ_OWN,
         )
     ),
-    printer_scope: PrinterScope = MediaOrRequestPrinterScope,
+    printer_scope: PrinterScope = RequestPrinterScope,
 ):
     """Download the 3MF file with filename in URL."""
     user, can_read_all = auth_result
@@ -2512,7 +2512,7 @@ async def create_archive_slicer_token(
             Permission.ARCHIVES_READ_OWN,
         )
     ),
-    printer_scope: PrinterScope = MediaOrRequestPrinterScope,
+    printer_scope: PrinterScope = RequestPrinterScope,
 ):
     """Create a short-lived download token for opening files in slicer applications.
 
@@ -2756,7 +2756,7 @@ async def create_archive_media_download_token(
     auth_result: tuple[User | None, bool] = Depends(
         require_ownership_permission(Permission.ARCHIVES_READ_ALL, Permission.ARCHIVES_READ_OWN)
     ),
-    printer_scope: PrinterScope = MediaOrRequestPrinterScope,
+    printer_scope: PrinterScope = RequestPrinterScope,
 ):
     """Mint a single-use token bound to an archive's attached timelapse."""
 
@@ -2861,7 +2861,7 @@ async def delete_timelapse(
             Permission.ARCHIVES_DELETE_OWN,
         )
     ),
-    printer_scope: PrinterScope = MediaOrRequestPrinterScope,
+    printer_scope: PrinterScope = RequestPrinterScope,
 ):
     """Remove the timelapse video from an archive."""
     user, can_modify_all = auth_result
@@ -3307,7 +3307,7 @@ async def get_timelapse_info(
             Permission.ARCHIVES_READ_OWN,
         )
     ),
-    printer_scope: PrinterScope = MediaOrRequestPrinterScope,
+    printer_scope: PrinterScope = RequestPrinterScope,
 ):
     """Get timelapse video metadata for editor."""
     from backend.app.schemas.timelapse import TimelapseInfoResponse
@@ -3344,7 +3344,7 @@ async def get_timelapse_thumbnails(
             Permission.ARCHIVES_READ_OWN,
         )
     ),
-    printer_scope: PrinterScope = MediaOrRequestPrinterScope,
+    printer_scope: PrinterScope = RequestPrinterScope,
 ):
     """Generate timeline thumbnail frames for visual scrubbing."""
     import base64
@@ -3502,7 +3502,7 @@ async def upload_photo(
             Permission.ARCHIVES_UPDATE_OWN,
         )
     ),
-    printer_scope: PrinterScope = MediaOrRequestPrinterScope,
+    printer_scope: PrinterScope = RequestPrinterScope,
 ):
     """Upload a photo of the printed result."""
     user, can_modify_all = auth_result
@@ -3602,7 +3602,7 @@ async def delete_photo(
             Permission.ARCHIVES_DELETE_OWN,
         )
     ),
-    printer_scope: PrinterScope = MediaOrRequestPrinterScope,
+    printer_scope: PrinterScope = RequestPrinterScope,
 ):
     """Delete a photo."""
     user, can_modify_all = auth_result
@@ -3950,7 +3950,7 @@ async def get_archive_capabilities(
             Permission.ARCHIVES_READ_OWN,
         )
     ),
-    printer_scope: PrinterScope = MediaOrRequestPrinterScope,
+    printer_scope: PrinterScope = RequestPrinterScope,
 ):
     """Check what viewing capabilities are available for this 3MF file."""
     import defusedxml.ElementTree as ET
@@ -4178,7 +4178,7 @@ async def get_gcode(
             Permission.ARCHIVES_READ_OWN,
         )
     ),
-    printer_scope: PrinterScope = MediaOrRequestPrinterScope,
+    printer_scope: PrinterScope = RequestPrinterScope,
 ):
     """Extract and return G-code from the 3MF file.
 
@@ -4460,7 +4460,7 @@ async def get_archive_plates(
             Permission.ARCHIVES_READ_OWN,
         )
     ),
-    printer_scope: PrinterScope = MediaOrRequestPrinterScope,
+    printer_scope: PrinterScope = RequestPrinterScope,
 ):
     """Get available plates from a multi-plate 3MF archive.
 
@@ -4867,7 +4867,7 @@ async def get_filament_requirements(
             Permission.ARCHIVES_READ_OWN,
         )
     ),
-    printer_scope: PrinterScope = MediaOrRequestPrinterScope,
+    printer_scope: PrinterScope = RequestPrinterScope,
 ):
     """Get filament requirements from the archived 3MF file.
 
@@ -5033,7 +5033,7 @@ async def slice_archive(
     request: SliceRequest,
     db: AsyncSession = Depends(get_db),
     current_user: User | None = RequirePermissionIfAuthEnabled(Permission.LIBRARY_UPLOAD),
-    printer_scope: PrinterScope = MediaOrRequestPrinterScope,
+    printer_scope: PrinterScope = RequestPrinterScope,
     actor: User | ApiKeyActor | None = RequestActor,
 ):
     """Enqueue a slice job for an archive's source. Returns 202 + job_id;
@@ -5181,7 +5181,7 @@ async def get_project_page(
             Permission.ARCHIVES_READ_OWN,
         )
     ),
-    printer_scope: PrinterScope = MediaOrRequestPrinterScope,
+    printer_scope: PrinterScope = RequestPrinterScope,
 ):
     """Get the project page data from the 3MF file."""
     from backend.app.schemas.archive import ProjectPageResponse
@@ -5212,7 +5212,7 @@ async def update_project_page(
             Permission.ARCHIVES_UPDATE_OWN,
         )
     ),
-    printer_scope: PrinterScope = MediaOrRequestPrinterScope,
+    printer_scope: PrinterScope = RequestPrinterScope,
 ):
     """Update project page metadata in the 3MF file."""
     from backend.app.services.archive import ProjectPageParser
@@ -5341,7 +5341,7 @@ async def upload_source_3mf(
             Permission.ARCHIVES_UPDATE_OWN,
         )
     ),
-    printer_scope: PrinterScope = MediaOrRequestPrinterScope,
+    printer_scope: PrinterScope = RequestPrinterScope,
 ):
     """Upload the original source 3MF project file for an archive."""
     user, can_modify_all = auth_result
@@ -5393,7 +5393,7 @@ async def download_source_3mf(
             Permission.ARCHIVES_READ_OWN,
         )
     ),
-    printer_scope: PrinterScope = MediaOrRequestPrinterScope,
+    printer_scope: PrinterScope = RequestPrinterScope,
 ):
     """Download the source 3MF project file."""
     user, can_read_all = auth_result
@@ -5428,7 +5428,7 @@ async def download_source_3mf_for_slicer(
             Permission.ARCHIVES_READ_OWN,
         )
     ),
-    printer_scope: PrinterScope = MediaOrRequestPrinterScope,
+    printer_scope: PrinterScope = RequestPrinterScope,
 ):
     """Download source 3MF with filename in URL."""
     user, can_read_all = auth_result
@@ -5459,7 +5459,7 @@ async def create_source_slicer_token(
             Permission.ARCHIVES_READ_OWN,
         )
     ),
-    printer_scope: PrinterScope = MediaOrRequestPrinterScope,
+    printer_scope: PrinterScope = RequestPrinterScope,
 ):
     """Create a short-lived download token for opening source 3MF in slicer."""
     from backend.app.core.auth import create_slicer_download_token
@@ -5614,7 +5614,7 @@ async def delete_source_3mf(
             Permission.ARCHIVES_DELETE_OWN,
         )
     ),
-    printer_scope: PrinterScope = MediaOrRequestPrinterScope,
+    printer_scope: PrinterScope = RequestPrinterScope,
 ):
     """Delete the source 3MF project file from an archive."""
     user, can_modify_all = auth_result
@@ -5652,7 +5652,7 @@ async def upload_f3d(
             Permission.ARCHIVES_UPDATE_OWN,
         )
     ),
-    printer_scope: PrinterScope = MediaOrRequestPrinterScope,
+    printer_scope: PrinterScope = RequestPrinterScope,
 ):
     """Upload a Fusion 360 design file for an archive."""
     user, can_modify_all = auth_result
@@ -5704,7 +5704,7 @@ async def download_f3d(
             Permission.ARCHIVES_READ_OWN,
         )
     ),
-    printer_scope: PrinterScope = MediaOrRequestPrinterScope,
+    printer_scope: PrinterScope = RequestPrinterScope,
 ):
     """Download the Fusion 360 design file."""
     user, can_read_all = auth_result
@@ -5738,7 +5738,7 @@ async def delete_f3d(
             Permission.ARCHIVES_DELETE_OWN,
         )
     ),
-    printer_scope: PrinterScope = MediaOrRequestPrinterScope,
+    printer_scope: PrinterScope = RequestPrinterScope,
 ):
     """Delete the Fusion 360 design file from an archive."""
     user, can_modify_all = auth_result

backend/app/core/auth.py

@@ -2374,10 +2374,11 @@ async def get_media_or_request_printer_scope(
     """``RequestPrinterScope`` for routes an ``<img>`` may load with ``?token=``.
 
     Such a request carries a media token and no headers, so the header-based
-    lookup would find nobody and answer with no printers. With headers present
-    they win, exactly as in ``require_media_token_*``.
+    lookup would find nobody and answer with no printers. A media token wins
+    over any headers sent with it, exactly as in ``require_media_token_*``:
+    the printers are always those of the caller the permission check named.
     """
-    if token and credentials is None and x_api_key is None:
+    if token:
         async with async_session() as db:
             if not await is_auth_enabled(db):
                 return ALL_PRINTERS

backend/tests/integration/test_media_token_with_headers_scope.py

@@ -0,0 +1,58 @@
+"""A media token decides both who the caller is and which printers they see."""
+
+from __future__ import annotations
+
+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]
+
+
+async def test_headers_next_to_a_media_token_do_not_widen_its_printers(
+    async_client: AsyncClient, printer_factory, archive_factory
+):
+    a, b = await printer_factory(name="A"), await printer_factory(name="B")
+    theirs = await archive_factory(b.id, print_name="Theirs")
+    ours = await archive_factory(a.id, print_name="Ours")
+    token = await _admin_token(async_client)
+    perms = await _group(async_client, token, "perms", permissions=["printers:read", "archives:read_all"])
+    team = await _group(async_client, token, "team", printer_ids=[a.id])
+    member_jwt, _ = await _user(async_client, token, "member", [perms, team])
+    media = await async_client.post("/api/v1/auth/media-token", headers=_auth(member_jwt))
+    assert media.status_code == 200, media.text
+    key = await async_client.post(
+        "/api/v1/api-keys/", headers=_auth(token), json={"name": "k", "can_read_status": True}
+    )
+    assert key.status_code == 200, key.text
+    headers = {"X-API-Key": key.json()["key"]}
+
+    def qrcode(archive_id: int):
+        return async_client.get(
+            f"/api/v1/archives/{archive_id}/qrcode", params={"token": media.json()["token"]}, headers=headers
+        )
+
+    assert (await qrcode(ours.id)).status_code == 200
+    assert (await qrcode(theirs.id)).status_code == 404
+
+
+async def test_a_media_token_next_to_headers_does_not_widen_their_printers(
+    async_client: AsyncClient, printer_factory, archive_factory
+):
+    a, b = await printer_factory(name="A"), await printer_factory(name="B")
+    theirs = await archive_factory(b.id, print_name="Theirs")
+    ours = await archive_factory(a.id, print_name="Ours")
+    token = await _admin_token(async_client)
+    perms = await _group(async_client, token, "perms", permissions=["printers:read", "archives:read_all"])
+    team = await _group(async_client, token, "team", printer_ids=[a.id])
+    member_jwt, _ = await _user(async_client, token, "member", [perms, team])
+    admin_media = (await async_client.post("/api/v1/auth/media-token", headers=_auth(token))).json()["token"]
+
+    def get(archive_id: int):
+        return async_client.get(
+            f"/api/v1/archives/{archive_id}", params={"token": admin_media}, headers=_auth(member_jwt)
+        )
+
+    assert (await get(ours.id)).status_code == 200
+    assert (await get(theirs.id)).status_code == 404

backend/tests/unit/test_media_scope_matches_media_gate.py

@@ -0,0 +1,43 @@
+"""A route takes its printers from a media token only if its permission check does too."""
+
+from __future__ import annotations
+
+import ast
+from pathlib import Path
+
+API = Path(__file__).resolve().parents[2] / "app" / "api"
+MEDIA_SCOPE = {"MediaOrRequestPrinterScope", "get_media_or_request_printer_scope"}
+HEADER_PRINCIPAL = {"RequestPrinterScope", "get_printer_scope_if_auth_enabled", "RequestActor"}
+
+
+def _names(node: ast.AST) -> set[str]:
+    found = set()
+    for sub in ast.walk(node):
+        if isinstance(sub, ast.Name):
+            found.add(sub.id)
+        elif isinstance(sub, ast.Attribute):
+            found.add(sub.attr)
+    return found
+
+
+def _route_dependencies(path: Path):
+    for node in ast.walk(ast.parse(path.read_text())):
+        if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) or not node.decorator_list:
+            continue
+        args = node.args
+        parts = [*args.defaults, *[d for d in args.kw_defaults if d is not None]]
+        parts += [a.annotation for a in [*args.args, *args.kwonlyargs] if a.annotation is not None]
+        names = set().union(*(_names(p) for p in parts)) if parts else set()
+        yield f"{path.relative_to(API)}:{node.name}", names
+
+
+def test_a_route_takes_its_caller_from_one_place():
+    offenders = []
+    for path in sorted(API.rglob("*.py")):
+        for route, names in _route_dependencies(path):
+            media_gate = any(name.startswith("require_media_token_") for name in names)
+            if names & MEDIA_SCOPE and not media_gate:
+                offenders.append(f"{route}: media-token printers without a media-token check")
+            if media_gate and names & HEADER_PRINCIPAL:
+                offenders.append(f"{route}: media-token check with a header-based caller")
+    assert offenders == []