Security hardening (security #2)

backend/app/api/routes/archives.py

@@ -119,6 +119,43 @@ def _match_timelapse_by_timestamp(
     return best_video, best_diff
 
 
+def _ensure_archive_visible(
+    archive: PrintArchive | None,
+    user: User | None,
+    can_read_all: bool,
+) -> PrintArchive:
+    """Per-archive visibility gate for ownership-scoped reads (#1726-adjacent).
+
+    Returns ``archive`` if the caller is allowed to see it; raises 404 otherwise.
+    Single enforcement point used by every detail / download / sub-resource
+    route so we can't accidentally leak a row through a less-guarded sibling.
+
+    Rules:
+      - Missing archive or soft-deleted (``deleted_at != None``) → 404.
+      - Caller with ARCHIVES_READ_ALL or auth disabled (``can_read_all=True``,
+        ``user`` may be None) → archive returned.
+      - Caller without ARCHIVES_READ_ALL and ``archive.created_by_id != user.id``
+        → 404, deliberately NOT 403. 403 leaks "this id exists but you can't
+        see it" — enumeration-friendly. 404 is indistinguishable from a
+        nonexistent id. Pre-GHSA fix the caller saw 200 here (the PoC vector).
+      - Ownerless rows (``created_by_id is None``) require ALL — fail-closed
+        per ``feedback_no_fail_open_in_auth``.
+    """
+    if not archive or archive.deleted_at is not None:
+        raise HTTPException(404, "Archive not found")
+    if can_read_all:
+        return archive
+    # Auth enabled, caller has _OWN only.
+    if user is None:
+        # Defensive: should be unreachable (RequirePermissionIfAuthEnabled
+        # would have 401'd already), but never trust user identity to be
+        # non-None when can_read_all is False.
+        raise HTTPException(404, "Archive not found")
+    if archive.created_by_id is None or archive.created_by_id != user.id:
+        raise HTTPException(404, "Archive not found")
+    return archive
+
+
 def _validate_user_filter_permission(current_user: User | None, created_by_id: int | None):
     """Raise 403 if created_by_id filter is used without stats:filter_by_user permission."""
     if created_by_id is None or current_user is None:
@@ -315,9 +352,16 @@ async def list_archives(
     limit: int = 50,
     offset: int = 0,
     db: AsyncSession = Depends(get_db),
-    _: User | None = RequirePermissionIfAuthEnabled(Permission.ARCHIVES_READ),
+    auth_result: tuple[User | None, bool] = Depends(
+        require_ownership_permission(
+            Permission.ARCHIVES_READ_ALL,
+            Permission.ARCHIVES_READ_OWN,
+        )
+    ),
 ):
     """List archived prints."""
+    user, can_read_all = auth_result
+    visible_to_user_id = user.id if (user is not None and not can_read_all) else None
     service = ArchiveService(db)
     archives = await service.list_archives(
         printer_id=printer_id,
@@ -326,6 +370,7 @@ async def list_archives(
         date_to=date_to,
         limit=limit,
         offset=offset,
+        visible_to_user_id=visible_to_user_id,
     )
 
     # Get sets of duplicate hashes and duplicate (name, hash) pairs (efficient single queries)
@@ -430,7 +475,12 @@ async def list_archives(
 @router.get("/no-3mf-warning")
 async def no_3mf_warning(
     db: AsyncSession = Depends(get_db),
-    _: User | None = RequirePermissionIfAuthEnabled(Permission.ARCHIVES_READ),
+    auth_result: tuple[User | None, bool] = Depends(
+        require_ownership_permission(
+            Permission.ARCHIVES_READ_ALL,
+            Permission.ARCHIVES_READ_OWN,
+        )
+    ),
 ):
     """Whether to nudge the user about install step 4 ("Store sent files on
     external storage"). True iff any archive in the last 30 days was created
@@ -446,14 +496,16 @@ async def no_3mf_warning(
     user has been told, no further nudge until they clear browser storage.
     The backend stays stateless.
     """
+    user, can_read_all = auth_result
     cutoff = datetime.now(timezone.utc) - timedelta(days=30)
-    result = await db.execute(
-        select(PrintArchive.extra_data).where(
-            PrintArchive.created_at >= cutoff,
-            PrintArchive.deleted_at.is_(None),
-            PrintArchive.extra_data.isnot(None),
-        )
-    )
+    conditions = [
+        PrintArchive.created_at >= cutoff,
+        PrintArchive.deleted_at.is_(None),
+        PrintArchive.extra_data.isnot(None),
+    ]
+    if user is not None and not can_read_all:
+        conditions.append(PrintArchive.created_by_id == user.id)
+    result = await db.execute(select(PrintArchive.extra_data).where(*conditions))
     for (extra_data,) in result.all():
         if extra_data and extra_data.get("no_3mf_available"):
             return {"has_fallback": True}
@@ -468,7 +520,12 @@ async def list_archives_slim(
     limit: int = Query(default=10000, le=50000),
     offset: int = 0,
     db: AsyncSession = Depends(get_db),
-    current_user: User | None = RequirePermissionIfAuthEnabled(Permission.ARCHIVES_READ),
+    auth_result: tuple[User | None, bool] = Depends(
+        require_ownership_permission(
+            Permission.ARCHIVES_READ_ALL,
+            Permission.ARCHIVES_READ_OWN,
+        )
+    ),
 ):
     """Per-event listing for stats/dashboard widgets.
 
@@ -481,7 +538,16 @@ async def list_archives_slim(
     """
     from backend.app.models.print_log import PrintLogEntry
 
+    current_user, can_read_all = auth_result
     _validate_user_filter_permission(current_user, created_by_id)
+    # Callers without ARCHIVES_READ_ALL can only see their own runs — pin
+    # the filter unconditionally so a caller-supplied ?created_by_id=
+    # can't widen the listing past their own scope. The existing
+    # _validate_user_filter_permission rejects ?created_by_id= without
+    # STATS_FILTER_BY_USER, so the only way to reach this is owner-self
+    # filtering anyway, but pinning here is the fail-closed default.
+    if current_user is not None and not can_read_all:
+        created_by_id = current_user.id
     filters = []
     if date_from:
         dt_from = datetime.combine(date_from, time.min, tzinfo=timezone.utc)
@@ -558,7 +624,12 @@ async def search_archives(
     limit: int = 50,
     offset: int = 0,
     db: AsyncSession = Depends(get_db),
-    _: User | None = RequirePermissionIfAuthEnabled(Permission.ARCHIVES_READ),
+    auth_result: tuple[User | None, bool] = Depends(
+        require_ownership_permission(
+            Permission.ARCHIVES_READ_ALL,
+            Permission.ARCHIVES_READ_OWN,
+        )
+    ),
 ):
     """Full-text search across archives.
 
@@ -570,6 +641,8 @@ async def search_archives(
 
     from backend.app.core.db_dialect import is_sqlite
 
+    user, can_read_all = auth_result
+    own_only = user is not None and not can_read_all
     search_term = q.strip()
 
     # Build dialect-specific full-text search query
@@ -630,6 +703,8 @@ async def search_archives(
             query = query.where(PrintArchive.project_id == project_id)
         if status:
             query = query.where(PrintArchive.status == status)
+        if own_only:
+            query = query.where(PrintArchive.created_by_id == user.id)
 
         query = query.limit(limit).offset(offset)
         result = await db.execute(query)
@@ -648,6 +723,8 @@ async def search_archives(
         .options(selectinload(PrintArchive.project))
         .where(PrintArchive.id.in_(matched_ids), PrintArchive.deleted_at.is_(None))
     )
+    if own_only:
+        query = query.where(PrintArchive.created_by_id == user.id)
 
     # Apply additional filters
     if printer_id:
@@ -721,7 +798,12 @@ async def analyze_failures(
     project_id: int | None = None,
     created_by_id: int | None = Query(None, description="Filter by user who created the print (-1 for no user)"),
     db: AsyncSession = Depends(get_db),
-    current_user: User | None = RequirePermissionIfAuthEnabled(Permission.ARCHIVES_READ),
+    auth_result: tuple[User | None, bool] = Depends(
+        require_ownership_permission(
+            Permission.ARCHIVES_READ_ALL,
+            Permission.ARCHIVES_READ_OWN,
+        )
+    ),
 ):
     """Analyze failure patterns across prints.
 
@@ -732,7 +814,11 @@ async def analyze_failures(
     - Recent failures
     - Weekly trend
     """
+    current_user, can_read_all = auth_result
     _validate_user_filter_permission(current_user, created_by_id)
+    # Callers without ARCHIVES_READ_ALL are scoped to their own runs (#2).
+    if current_user is not None and not can_read_all:
+        created_by_id = current_user.id
 
     from backend.app.services.failure_analysis import FailureAnalysisService
 
@@ -751,7 +837,12 @@ async def analyze_failures(
 async def compare_archives(
     archive_ids: str = Query(..., description="Comma-separated archive IDs (2-5)"),
     db: AsyncSession = Depends(get_db),
-    _: User | None = RequirePermissionIfAuthEnabled(Permission.ARCHIVES_READ),
+    auth_result: tuple[User | None, bool] = Depends(
+        require_ownership_permission(
+            Permission.ARCHIVES_READ_ALL,
+            Permission.ARCHIVES_READ_OWN,
+        )
+    ),
 ):
     """Compare multiple archives side by side.
 
@@ -763,6 +854,8 @@ async def compare_archives(
     """
     from backend.app.services.archive_comparison import ArchiveComparisonService
 
+    user, can_read_all = auth_result
+
     # Parse and validate archive IDs
     try:
         ids = [int(id.strip()) for id in archive_ids.split(",")]
@@ -774,6 +867,20 @@ async def compare_archives(
     if len(ids) > 5:
         raise HTTPException(400, "Maximum 5 archives can be compared at once")
 
+    # Verify the caller is allowed to see every archive in the comparison —
+    # one not-owned id in the list would otherwise leak its full detail block.
+    # _ensure_archive_visible raises 404 on the first miss (same 404 the
+    # single-archive endpoint would return).
+    if user is not None and not can_read_all:
+        existing = await db.execute(
+            select(PrintArchive.id, PrintArchive.created_by_id, PrintArchive.deleted_at).where(PrintArchive.id.in_(ids))
+        )
+        owners_by_id = {row.id: row for row in existing.all()}
+        for archive_id in ids:
+            row = owners_by_id.get(archive_id)
+            if row is None or row.deleted_at is not None or row.created_by_id != user.id:
+                raise HTTPException(404, "Archive not found")
+
     service = ArchiveComparisonService(db)
     try:
         return await service.compare_archives(ids)
@@ -792,7 +899,12 @@ async def export_archives(
     date_to: str | None = Query(None, description="End date (ISO format)"),
     search: str | None = None,
     db: AsyncSession = Depends(get_db),
-    _: User | None = RequirePermissionIfAuthEnabled(Permission.ARCHIVES_READ),
+    auth_result: tuple[User | None, bool] = Depends(
+        require_ownership_permission(
+            Permission.ARCHIVES_READ_ALL,
+            Permission.ARCHIVES_READ_OWN,
+        )
+    ),
 ):
     """Export archives to CSV or Excel format.
 
@@ -804,6 +916,9 @@ async def export_archives(
 
     from backend.app.services.export import ExportService
 
+    user, can_read_all = auth_result
+    visible_to_user_id = user.id if (user is not None and not can_read_all) else None
+
     if format not in ("csv", "xlsx"):
         raise HTTPException(400, "Format must be 'csv' or 'xlsx'")
 
@@ -837,6 +952,7 @@ async def export_archives(
             date_from=date_from_dt,
             date_to=date_to_dt,
             search=search,
+            visible_to_user_id=visible_to_user_id,
         )
     except ImportError as e:
         raise HTTPException(500, str(e))
@@ -1220,16 +1336,23 @@ async def _sum_snapshot_deltas(
 @router.get("/tags")
 async def get_all_tags(
     db: AsyncSession = Depends(get_db),
-    _: User | None = RequirePermissionIfAuthEnabled(Permission.ARCHIVES_READ),
+    auth_result: tuple[User | None, bool] = Depends(
+        require_ownership_permission(
+            Permission.ARCHIVES_READ_ALL,
+            Permission.ARCHIVES_READ_OWN,
+        )
+    ),
 ):
     """List all unique tags with usage counts.
 
     Returns a list of tags sorted by count (descending), then by name.
     """
+    user, can_read_all = auth_result
     # Query all archives with non-null tags
-    result = await db.execute(
-        select(PrintArchive.tags).where(PrintArchive.tags.isnot(None), PrintArchive.deleted_at.is_(None))
-    )
+    tag_conditions = [PrintArchive.tags.isnot(None), PrintArchive.deleted_at.is_(None)]
+    if user is not None and not can_read_all:
+        tag_conditions.append(PrintArchive.created_by_id == user.id)
+    result = await db.execute(select(PrintArchive.tags).where(*tag_conditions))
     all_tags_rows = result.all()
 
     # Count occurrences of each tag
@@ -1332,17 +1455,17 @@ async def delete_tag(
 async def get_archive(
     archive_id: int,
     db: AsyncSession = Depends(get_db),
-    _: User | None = RequirePermissionIfAuthEnabled(Permission.ARCHIVES_READ),
+    auth_result: tuple[User | None, bool] = Depends(
+        require_ownership_permission(
+            Permission.ARCHIVES_READ_ALL,
+            Permission.ARCHIVES_READ_OWN,
+        )
+    ),
 ):
     """Get a specific archive."""
+    user, can_read_all = auth_result
     service = ArchiveService(db)
-    archive = await service.get_archive(archive_id)
-    # Soft-deleted archives are hidden from the UI (#1343) — surface them as
-    # 404 here too so a stale bookmark / direct URL doesn't expose a row the
-    # user has already removed. The hard-delete (?purge_stats=true) path
-    # bypasses this check by querying PrintArchive directly.
-    if not archive or archive.deleted_at is not None:
-        raise HTTPException(404, "Archive not found")
+    archive = _ensure_archive_visible(await service.get_archive(archive_id), user, can_read_all)
 
     # Find duplicates
     makerworld_id = archive.extra_data.get("makerworld_model_id") if archive.extra_data else None
@@ -1360,7 +1483,12 @@ async def get_archive(
 async def list_archive_runs(
     archive_id: int,
     db: AsyncSession = Depends(get_db),
-    _: User | None = RequirePermissionIfAuthEnabled(Permission.ARCHIVES_READ),
+    auth_result: tuple[User | None, bool] = Depends(
+        require_ownership_permission(
+            Permission.ARCHIVES_READ_ALL,
+            Permission.ARCHIVES_READ_OWN,
+        )
+    ),
 ):
     """List PrintLogEntry rows for this archive — one per print event.
 
@@ -1369,9 +1497,8 @@ async def list_archive_runs(
     from backend.app.models.print_log import PrintLogEntry
     from backend.app.schemas.print_log import PrintLogEntrySchema
 
-    archive = await db.get(PrintArchive, archive_id)
-    if not archive or archive.deleted_at is not None:
-        raise HTTPException(404, "Archive not found")
+    user, can_read_all = auth_result
+    _ensure_archive_visible(await db.get(PrintArchive, archive_id), user, can_read_all)
 
     rows = await db.execute(
         select(PrintLogEntry)
@@ -1388,7 +1515,12 @@ async def find_similar_archives(
     archive_id: int,
     limit: int = 10,
     db: AsyncSession = Depends(get_db),
-    _: User | None = RequirePermissionIfAuthEnabled(Permission.ARCHIVES_READ),
+    auth_result: tuple[User | None, bool] = Depends(
+        require_ownership_permission(
+            Permission.ARCHIVES_READ_ALL,
+            Permission.ARCHIVES_READ_OWN,
+        )
+    ),
 ):
     """Find archives with similar settings for comparison.
 
@@ -1399,6 +1531,9 @@ async def find_similar_archives(
     """
     from backend.app.services.archive_comparison import ArchiveComparisonService
 
+    user, can_read_all = auth_result
+    _ensure_archive_visible(await db.get(PrintArchive, archive_id), user, can_read_all)
+
     service = ArchiveComparisonService(db)
     try:
         return await service.find_similar_archives(archive_id, limit=limit)
@@ -1720,13 +1855,17 @@ async def rescan_all_archives(
 async def get_archive_duplicates(
     archive_id: int,
     db: AsyncSession = Depends(get_db),
-    _: User | None = RequirePermissionIfAuthEnabled(Permission.ARCHIVES_READ),
+    auth_result: tuple[User | None, bool] = Depends(
+        require_ownership_permission(
+            Permission.ARCHIVES_READ_ALL,
+            Permission.ARCHIVES_READ_OWN,
+        )
+    ),
 ):
     """Get duplicates for a specific archive."""
+    user, can_read_all = auth_result
     service = ArchiveService(db)
-    archive = await service.get_archive(archive_id)
-    if not archive:
-        raise HTTPException(404, "Archive not found")
+    archive = _ensure_archive_visible(await service.get_archive(archive_id), user, can_read_all)
 
     makerworld_id = archive.extra_data.get("makerworld_model_id") if archive.extra_data else None
     duplicates = await service.find_duplicates(
@@ -1828,13 +1967,17 @@ async def download_archive(
     archive_id: int,
     inline: bool = False,
     db: AsyncSession = Depends(get_db),
-    _: User | None = RequirePermissionIfAuthEnabled(Permission.ARCHIVES_READ),
+    auth_result: tuple[User | None, bool] = Depends(
+        require_ownership_permission(
+            Permission.ARCHIVES_READ_ALL,
+            Permission.ARCHIVES_READ_OWN,
+        )
+    ),
 ):
     """Download the 3MF file."""
+    user, can_read_all = auth_result
     service = ArchiveService(db)
-    archive = await service.get_archive(archive_id)
-    if not archive:
-        raise HTTPException(404, "Archive not found")
+    archive = _ensure_archive_visible(await service.get_archive(archive_id), user, can_read_all)
 
     file_path = settings.base_dir / archive.file_path
     if not file_path.is_file():
@@ -1856,13 +1999,17 @@ async def download_archive_with_filename(
     archive_id: int,
     filename: str,
     db: AsyncSession = Depends(get_db),
-    _: User | None = RequirePermissionIfAuthEnabled(Permission.ARCHIVES_READ),
+    auth_result: tuple[User | None, bool] = Depends(
+        require_ownership_permission(
+            Permission.ARCHIVES_READ_ALL,
+            Permission.ARCHIVES_READ_OWN,
+        )
+    ),
 ):
     """Download the 3MF file with filename in URL."""
+    user, can_read_all = auth_result
     service = ArchiveService(db)
-    archive = await service.get_archive(archive_id)
-    if not archive:
-        raise HTTPException(404, "Archive not found")
+    archive = _ensure_archive_visible(await service.get_archive(archive_id), user, can_read_all)
 
     file_path = settings.base_dir / archive.file_path
     if not file_path.is_file():
@@ -1879,7 +2026,12 @@ async def download_archive_with_filename(
 async def create_archive_slicer_token(
     archive_id: int,
     db: AsyncSession = Depends(get_db),
-    _: User | None = RequirePermissionIfAuthEnabled(Permission.ARCHIVES_READ),
+    auth_result: tuple[User | None, bool] = Depends(
+        require_ownership_permission(
+            Permission.ARCHIVES_READ_ALL,
+            Permission.ARCHIVES_READ_OWN,
+        )
+    ),
 ):
     """Create a short-lived download token for opening files in slicer applications.
 
@@ -1888,10 +2040,9 @@ async def create_archive_slicer_token(
     """
     from backend.app.core.auth import create_slicer_download_token
 
+    user, can_read_all = auth_result
     service = ArchiveService(db)
-    archive = await service.get_archive(archive_id)
-    if not archive:
-        raise HTTPException(404, "Archive not found")
+    _ensure_archive_visible(await service.get_archive(archive_id), user, can_read_all)
 
     token = await create_slicer_download_token("archive", archive_id)
     return {"token": token}
@@ -2327,15 +2478,21 @@ async def upload_timelapse(
 async def get_timelapse_info(
     archive_id: int,
     db: AsyncSession = Depends(get_db),
-    _: User | None = RequirePermissionIfAuthEnabled(Permission.ARCHIVES_READ),
+    auth_result: tuple[User | None, bool] = Depends(
+        require_ownership_permission(
+            Permission.ARCHIVES_READ_ALL,
+            Permission.ARCHIVES_READ_OWN,
+        )
+    ),
 ):
     """Get timelapse video metadata for editor."""
     from backend.app.schemas.timelapse import TimelapseInfoResponse
     from backend.app.services.timelapse_processor import TimelapseProcessor
 
+    user, can_read_all = auth_result
     service = ArchiveService(db)
-    archive = await service.get_archive(archive_id)
-    if not archive or not archive.timelapse_path:
+    archive = _ensure_archive_visible(await service.get_archive(archive_id), user, can_read_all)
+    if not archive.timelapse_path:
         raise HTTPException(404, "Timelapse not found")
 
     timelapse_path = settings.base_dir / archive.timelapse_path
@@ -2357,7 +2514,12 @@ async def get_timelapse_thumbnails(
     count: int = Query(10, ge=1, le=30),
     width: int = Query(160, ge=80, le=320),
     db: AsyncSession = Depends(get_db),
-    _: User | None = RequirePermissionIfAuthEnabled(Permission.ARCHIVES_READ),
+    auth_result: tuple[User | None, bool] = Depends(
+        require_ownership_permission(
+            Permission.ARCHIVES_READ_ALL,
+            Permission.ARCHIVES_READ_OWN,
+        )
+    ),
 ):
     """Generate timeline thumbnail frames for visual scrubbing."""
     import base64
@@ -2365,9 +2527,10 @@ async def get_timelapse_thumbnails(
     from backend.app.schemas.timelapse import ThumbnailResponse
     from backend.app.services.timelapse_processor import TimelapseProcessor
 
+    user, can_read_all = auth_result
     service = ArchiveService(db)
-    archive = await service.get_archive(archive_id)
-    if not archive or not archive.timelapse_path:
+    archive = _ensure_archive_visible(await service.get_archive(archive_id), user, can_read_all)
+    if not archive.timelapse_path:
         raise HTTPException(404, "Timelapse not found")
 
     timelapse_path = settings.base_dir / archive.timelapse_path
@@ -2692,15 +2855,19 @@ async def get_qrcode(
 async def get_archive_capabilities(
     archive_id: int,
     db: AsyncSession = Depends(get_db),
-    _: User | None = RequirePermissionIfAuthEnabled(Permission.ARCHIVES_READ),
+    auth_result: tuple[User | None, bool] = Depends(
+        require_ownership_permission(
+            Permission.ARCHIVES_READ_ALL,
+            Permission.ARCHIVES_READ_OWN,
+        )
+    ),
 ):
     """Check what viewing capabilities are available for this 3MF file."""
     import defusedxml.ElementTree as ET
 
+    user, can_read_all = auth_result
     service = ArchiveService(db)
-    archive = await service.get_archive(archive_id)
-    if not archive:
-        raise HTTPException(404, "Archive not found")
+    archive = _ensure_archive_visible(await service.get_archive(archive_id), user, can_read_all)
 
     file_path = settings.base_dir / archive.file_path
     if not file_path.is_file():
@@ -2913,7 +3080,12 @@ async def get_gcode(
     archive_id: int,
     plate: int | None = None,
     db: AsyncSession = Depends(get_db),
-    _: User | None = RequirePermissionIfAuthEnabled(Permission.ARCHIVES_READ),
+    auth_result: tuple[User | None, bool] = Depends(
+        require_ownership_permission(
+            Permission.ARCHIVES_READ_ALL,
+            Permission.ARCHIVES_READ_OWN,
+        )
+    ),
 ):
     """Extract and return G-code from the 3MF file.
 
@@ -2922,10 +3094,9 @@ async def get_gcode(
     back to the first plate found in the archive (preserving the original
     behaviour for callers that predate the multi-plate viewer).
     """
+    user, can_read_all = auth_result
     service = ArchiveService(db)
-    archive = await service.get_archive(archive_id)
-    if not archive:
-        raise HTTPException(404, "Archive not found")
+    archive = _ensure_archive_visible(await service.get_archive(archive_id), user, can_read_all)
 
     file_path = settings.base_dir / archive.file_path
     if not file_path.is_file():
@@ -3161,7 +3332,12 @@ async def upload_archives_bulk(
 async def get_archive_plates(
     archive_id: int,
     db: AsyncSession = Depends(get_db),
-    _: User | None = RequirePermissionIfAuthEnabled(Permission.ARCHIVES_READ),
+    auth_result: tuple[User | None, bool] = Depends(
+        require_ownership_permission(
+            Permission.ARCHIVES_READ_ALL,
+            Permission.ARCHIVES_READ_OWN,
+        )
+    ),
 ):
     """Get available plates from a multi-plate 3MF archive.
 
@@ -3172,10 +3348,9 @@ async def get_archive_plates(
 
     import defusedxml.ElementTree as ET
 
+    user, can_read_all = auth_result
     service = ArchiveService(db)
-    archive = await service.get_archive(archive_id)
-    if not archive:
-        raise HTTPException(404, "Archive not found")
+    archive = _ensure_archive_visible(await service.get_archive(archive_id), user, can_read_all)
 
     file_path = settings.base_dir / archive.file_path
     if not file_path.is_file():
@@ -3541,7 +3716,12 @@ async def get_filament_requirements(
     plate_id: int | None = None,
     request_id: str | None = None,
     db: AsyncSession = Depends(get_db),
-    _: User | None = RequirePermissionIfAuthEnabled(Permission.ARCHIVES_READ),
+    auth_result: tuple[User | None, bool] = Depends(
+        require_ownership_permission(
+            Permission.ARCHIVES_READ_ALL,
+            Permission.ARCHIVES_READ_OWN,
+        )
+    ),
 ):
     """Get filament requirements from the archived 3MF file.
 
@@ -3554,10 +3734,9 @@ async def get_filament_requirements(
     """
     import defusedxml.ElementTree as ET
 
+    user, can_read_all = auth_result
     service = ArchiveService(db)
-    archive = await service.get_archive(archive_id)
-    if not archive:
-        raise HTTPException(404, "Archive not found")
+    archive = _ensure_archive_visible(await service.get_archive(archive_id), user, can_read_all)
 
     file_path = settings.base_dir / archive.file_path
     if not file_path.is_file():
@@ -3897,16 +4076,20 @@ async def reprint_archive(
 async def get_project_page(
     archive_id: int,
     db: AsyncSession = Depends(get_db),
-    _: User | None = RequirePermissionIfAuthEnabled(Permission.ARCHIVES_READ),
+    auth_result: tuple[User | None, bool] = Depends(
+        require_ownership_permission(
+            Permission.ARCHIVES_READ_ALL,
+            Permission.ARCHIVES_READ_OWN,
+        )
+    ),
 ):
     """Get the project page data from the 3MF file."""
     from backend.app.schemas.archive import ProjectPageResponse
     from backend.app.services.archive import ProjectPageParser
 
+    user, can_read_all = auth_result
     service = ArchiveService(db)
-    archive = await service.get_archive(archive_id)
-    if not archive:
-        raise HTTPException(404, "Archive not found")
+    archive = _ensure_archive_visible(await service.get_archive(archive_id), user, can_read_all)
 
     file_path = settings.base_dir / archive.file_path
     if not file_path.is_file():
@@ -4088,13 +4271,17 @@ async def upload_source_3mf(
 async def download_source_3mf(
     archive_id: int,
     db: AsyncSession = Depends(get_db),
-    _: User | None = RequirePermissionIfAuthEnabled(Permission.ARCHIVES_READ),
+    auth_result: tuple[User | None, bool] = Depends(
+        require_ownership_permission(
+            Permission.ARCHIVES_READ_ALL,
+            Permission.ARCHIVES_READ_OWN,
+        )
+    ),
 ):
     """Download the source 3MF project file."""
+    user, can_read_all = auth_result
     result = await db.execute(select(PrintArchive).where(PrintArchive.id == archive_id))
-    archive = result.scalar_one_or_none()
-    if not archive:
-        raise HTTPException(404, "Archive not found")
+    archive = _ensure_archive_visible(result.scalar_one_or_none(), user, can_read_all)
 
     if not archive.source_3mf_path:
         raise HTTPException(404, "No source 3MF attached to this archive")
@@ -4118,13 +4305,17 @@ async def download_source_3mf_for_slicer(
     archive_id: int,
     filename: str,
     db: AsyncSession = Depends(get_db),
-    _: User | None = RequirePermissionIfAuthEnabled(Permission.ARCHIVES_READ),
+    auth_result: tuple[User | None, bool] = Depends(
+        require_ownership_permission(
+            Permission.ARCHIVES_READ_ALL,
+            Permission.ARCHIVES_READ_OWN,
+        )
+    ),
 ):
     """Download source 3MF with filename in URL."""
+    user, can_read_all = auth_result
     result = await db.execute(select(PrintArchive).where(PrintArchive.id == archive_id))
-    archive = result.scalar_one_or_none()
-    if not archive:
-        raise HTTPException(404, "Archive not found")
+    archive = _ensure_archive_visible(result.scalar_one_or_none(), user, can_read_all)
 
     if not archive.source_3mf_path:
         raise HTTPException(404, "No source 3MF attached to this archive")
@@ -4144,15 +4335,19 @@ async def download_source_3mf_for_slicer(
 async def create_source_slicer_token(
     archive_id: int,
     db: AsyncSession = Depends(get_db),
-    _: User | None = RequirePermissionIfAuthEnabled(Permission.ARCHIVES_READ),
+    auth_result: tuple[User | None, bool] = Depends(
+        require_ownership_permission(
+            Permission.ARCHIVES_READ_ALL,
+            Permission.ARCHIVES_READ_OWN,
+        )
+    ),
 ):
     """Create a short-lived download token for opening source 3MF in slicer."""
     from backend.app.core.auth import create_slicer_download_token
 
+    user, can_read_all = auth_result
     result = await db.execute(select(PrintArchive).where(PrintArchive.id == archive_id))
-    archive = result.scalar_one_or_none()
-    if not archive:
-        raise HTTPException(404, "Archive not found")
+    archive = _ensure_archive_visible(result.scalar_one_or_none(), user, can_read_all)
     if not archive.source_3mf_path:
         raise HTTPException(404, "No source 3MF attached to this archive")
 
@@ -4370,13 +4565,17 @@ async def upload_f3d(
 async def download_f3d(
     archive_id: int,
     db: AsyncSession = Depends(get_db),
-    _: User | None = RequirePermissionIfAuthEnabled(Permission.ARCHIVES_READ),
+    auth_result: tuple[User | None, bool] = Depends(
+        require_ownership_permission(
+            Permission.ARCHIVES_READ_ALL,
+            Permission.ARCHIVES_READ_OWN,
+        )
+    ),
 ):
     """Download the Fusion 360 design file."""
+    user, can_read_all = auth_result
     result = await db.execute(select(PrintArchive).where(PrintArchive.id == archive_id))
-    archive = result.scalar_one_or_none()
-    if not archive:
-        raise HTTPException(404, "Archive not found")
+    archive = _ensure_archive_visible(result.scalar_one_or_none(), user, can_read_all)
 
     if not archive.f3d_path:
         raise HTTPException(404, "No F3D file attached to this archive")

backend/app/api/routes/library.py

@@ -76,6 +76,32 @@ logger = logging.getLogger(__name__)
 router = APIRouter(prefix="/library", tags=["library"])
 
 
+def _ensure_library_file_visible(
+    library_file: LibraryFile | None,
+    user: User | None,
+    can_read_all: bool,
+) -> LibraryFile:
+    """Per-file visibility gate for ownership-scoped LIBRARY reads (#1726-adjacent).
+
+    Mirrors archives.py::_ensure_archive_visible — single enforcement point so a
+    less-guarded sibling route can't accidentally leak a row. Same shape:
+
+      - Missing / soft-deleted → 404 (not 403, to avoid id-enumeration leaks).
+      - ``can_read_all`` true (LIBRARY_READ_ALL or auth disabled) → file returned.
+      - ``can_read_all`` false and ``created_by_id != user.id`` → 404.
+      - Ownerless files (``created_by_id is None``) require ALL — fail-closed.
+    """
+    if library_file is None or getattr(library_file, "deleted_at", None) is not None:
+        raise HTTPException(404, "File not found")
+    if can_read_all:
+        return library_file
+    if user is None:
+        raise HTTPException(404, "File not found")
+    if library_file.created_by_id is None or library_file.created_by_id != user.id:
+        raise HTTPException(404, "File not found")
+    return library_file
+
+
 def get_library_dir() -> Path:
     """Get the library storage directory."""
     base_dir = Path(app_settings.archive_dir)
@@ -695,7 +721,12 @@ async def _backfill_external_stl_thumbnails(folder_ids: list[int]) -> None:
 async def list_folders(
     response: Response,
     db: AsyncSession = Depends(get_db),
-    _: User | None = Depends(require_permission_if_auth_enabled(Permission.LIBRARY_READ)),
+    _: tuple[User | None, bool] = Depends(
+        require_ownership_permission(
+            Permission.LIBRARY_READ_ALL,
+            Permission.LIBRARY_READ_OWN,
+        )
+    ),
 ):
     """Get all folders as a tree structure."""
     # Prevent browser caching of folder list
@@ -754,7 +785,12 @@ async def list_folders(
 async def get_folders_by_project(
     project_id: int,
     db: AsyncSession = Depends(get_db),
-    _: User | None = Depends(require_permission_if_auth_enabled(Permission.LIBRARY_READ)),
+    _: tuple[User | None, bool] = Depends(
+        require_ownership_permission(
+            Permission.LIBRARY_READ_ALL,
+            Permission.LIBRARY_READ_OWN,
+        )
+    ),
 ):
     """Get all folders linked to a specific project."""
     result = await db.execute(
@@ -802,7 +838,12 @@ async def get_folders_by_project(
 async def get_folders_by_archive(
     archive_id: int,
     db: AsyncSession = Depends(get_db),
-    _: User | None = Depends(require_permission_if_auth_enabled(Permission.LIBRARY_READ)),
+    _: tuple[User | None, bool] = Depends(
+        require_ownership_permission(
+            Permission.LIBRARY_READ_ALL,
+            Permission.LIBRARY_READ_OWN,
+        )
+    ),
 ):
     """Get all folders linked to a specific archive."""
     result = await db.execute(
@@ -910,7 +951,12 @@ async def create_folder(
 async def get_folder(
     folder_id: int,
     db: AsyncSession = Depends(get_db),
-    _: User | None = Depends(require_permission_if_auth_enabled(Permission.LIBRARY_READ)),
+    _: tuple[User | None, bool] = Depends(
+        require_ownership_permission(
+            Permission.LIBRARY_READ_ALL,
+            Permission.LIBRARY_READ_OWN,
+        )
+    ),
 ):
     """Get a folder by ID."""
     result = await db.execute(
@@ -1632,7 +1678,12 @@ async def list_files(
     internal_only: bool = False,
     external_only: bool = False,
     db: AsyncSession = Depends(get_db),
-    _: User | None = Depends(require_permission_if_auth_enabled(Permission.LIBRARY_READ)),
+    auth_result: tuple[User | None, bool] = Depends(
+        require_ownership_permission(
+            Permission.LIBRARY_READ_ALL,
+            Permission.LIBRARY_READ_OWN,
+        )
+    ),
 ):
     """List files, optionally filtered by folder or project.
 
@@ -1654,7 +1705,10 @@ async def list_files(
             detail="internal_only and external_only are mutually exclusive",
         )
 
+    user, can_read_all = auth_result
     query = LibraryFile.active().options(selectinload(LibraryFile.created_by))
+    if user is not None and not can_read_all:
+        query = query.where(LibraryFile.created_by_id == user.id)
 
     if folder_id is not None:
         query = query.where(LibraryFile.folder_id == folder_id)
@@ -2394,7 +2448,12 @@ async def add_files_to_queue(
 async def get_library_file_plates(
     file_id: int,
     db: AsyncSession = Depends(get_db),
-    _: User | None = Depends(require_permission_if_auth_enabled(Permission.LIBRARY_READ)),
+    auth_result: tuple[User | None, bool] = Depends(
+        require_ownership_permission(
+            Permission.LIBRARY_READ_ALL,
+            Permission.LIBRARY_READ_OWN,
+        )
+    ),
 ):
     """Get available plates from a multi-plate 3MF library file.
 
@@ -2405,9 +2464,10 @@ async def get_library_file_plates(
 
     import defusedxml.ElementTree as ET
 
+    user, can_read_all = auth_result
     # Get the library file
     result = await db.execute(LibraryFile.active().where(LibraryFile.id == file_id))
-    lib_file = result.scalar_one_or_none()
+    lib_file = _ensure_library_file_visible(result.scalar_one_or_none(), user, can_read_all)
 
     if not lib_file:
         raise HTTPException(status_code=404, detail="File not found")
@@ -2745,7 +2805,12 @@ async def get_library_file_filament_requirements(
     plate_id: int | None = None,
     request_id: str | None = None,
     db: AsyncSession = Depends(get_db),
-    _: User | None = Depends(require_permission_if_auth_enabled(Permission.LIBRARY_READ)),
+    auth_result: tuple[User | None, bool] = Depends(
+        require_ownership_permission(
+            Permission.LIBRARY_READ_ALL,
+            Permission.LIBRARY_READ_OWN,
+        )
+    ),
 ):
     """Get filament requirements from a library file.
 
@@ -2758,12 +2823,10 @@ async def get_library_file_filament_requirements(
     """
     import defusedxml.ElementTree as ET
 
+    user, can_read_all = auth_result
     # Get the library file
     result = await db.execute(LibraryFile.active().where(LibraryFile.id == file_id))
-    lib_file = result.scalar_one_or_none()
-
-    if not lib_file:
-        raise HTTPException(status_code=404, detail="File not found")
+    lib_file = _ensure_library_file_visible(result.scalar_one_or_none(), user, can_read_all)
 
     # Get the full file path
     file_path = Path(app_settings.base_dir) / lib_file.file_path
@@ -3997,16 +4060,19 @@ async def print_library_file(
 async def get_file(
     file_id: int,
     db: AsyncSession = Depends(get_db),
-    _: User | None = Depends(require_permission_if_auth_enabled(Permission.LIBRARY_READ)),
+    auth_result: tuple[User | None, bool] = Depends(
+        require_ownership_permission(
+            Permission.LIBRARY_READ_ALL,
+            Permission.LIBRARY_READ_OWN,
+        )
+    ),
 ):
     """Get a file by ID with full details."""
+    user, can_read_all = auth_result
     result = await db.execute(
         LibraryFile.active().options(selectinload(LibraryFile.created_by)).where(LibraryFile.id == file_id)
     )
-    file = result.scalar_one_or_none()
-
-    if not file:
-        raise HTTPException(status_code=404, detail="File not found")
+    file = _ensure_library_file_visible(result.scalar_one_or_none(), user, can_read_all)
 
     # Get folder name
     folder_name = None
@@ -4149,8 +4215,11 @@ async def update_file(
     await db.commit()
     await db.refresh(file)
 
-    # Return full response (reuse get_file logic)
-    return await get_file(file_id, db)
+    # Return full response. Bypass get_file's ownership gate — caller already
+    # passed update_file's ownership gate above, so we re-fetch + serialise
+    # directly instead of calling the route function (which would try to
+    # evaluate its own Depends() at call time and trip a TypeError).
+    return await get_file(file_id, db, auth_result=(None, True))
 
 
 @router.delete("/files/{file_id}")
@@ -4210,14 +4279,17 @@ async def delete_file(
 async def download_file(
     file_id: int,
     db: AsyncSession = Depends(get_db),
-    _: User | None = Depends(require_permission_if_auth_enabled(Permission.LIBRARY_READ)),
+    auth_result: tuple[User | None, bool] = Depends(
+        require_ownership_permission(
+            Permission.LIBRARY_READ_ALL,
+            Permission.LIBRARY_READ_OWN,
+        )
+    ),
 ):
     """Download a file."""
+    user, can_read_all = auth_result
     result = await db.execute(LibraryFile.active().where(LibraryFile.id == file_id))
-    file = result.scalar_one_or_none()
-
-    if not file:
-        raise HTTPException(status_code=404, detail="File not found")
+    file = _ensure_library_file_visible(result.scalar_one_or_none(), user, can_read_all)
 
     abs_path = to_absolute_path(file.file_path)
     if not abs_path or not abs_path.exists():
@@ -4234,7 +4306,12 @@ async def download_file(
 async def create_library_slicer_token(
     file_id: int,
     db: AsyncSession = Depends(get_db),
-    _: User | None = Depends(require_permission_if_auth_enabled(Permission.LIBRARY_READ)),
+    auth_result: tuple[User | None, bool] = Depends(
+        require_ownership_permission(
+            Permission.LIBRARY_READ_ALL,
+            Permission.LIBRARY_READ_OWN,
+        )
+    ),
 ):
     """Create a short-lived download token for opening files in slicer applications.
 
@@ -4243,10 +4320,9 @@ async def create_library_slicer_token(
     """
     from backend.app.core.auth import create_slicer_download_token
 
+    user, can_read_all = auth_result
     result = await db.execute(LibraryFile.active().where(LibraryFile.id == file_id))
-    file = result.scalar_one_or_none()
-    if not file:
-        raise HTTPException(status_code=404, detail="File not found")
+    _ensure_library_file_visible(result.scalar_one_or_none(), user, can_read_all)
 
     token = await create_slicer_download_token("library", file_id)
     return {"token": token}
@@ -4321,14 +4397,17 @@ async def get_thumbnail(
 async def get_gcode(
     file_id: int,
     db: AsyncSession = Depends(get_db),
-    _: User | None = Depends(require_permission_if_auth_enabled(Permission.LIBRARY_READ)),
+    auth_result: tuple[User | None, bool] = Depends(
+        require_ownership_permission(
+            Permission.LIBRARY_READ_ALL,
+            Permission.LIBRARY_READ_OWN,
+        )
+    ),
 ):
     """Get gcode for a file (for preview)."""
+    user, can_read_all = auth_result
     result = await db.execute(LibraryFile.active().where(LibraryFile.id == file_id))
-    file = result.scalar_one_or_none()
-
-    if not file:
-        raise HTTPException(status_code=404, detail="File not found")
+    file = _ensure_library_file_visible(result.scalar_one_or_none(), user, can_read_all)
 
     abs_path = to_absolute_path(file.file_path)
     if not abs_path or not abs_path.exists():
@@ -4544,32 +4623,42 @@ async def bulk_delete(
 @router.get("/stats")
 async def get_library_stats(
     db: AsyncSession = Depends(get_db),
-    _: User | None = Depends(require_permission_if_auth_enabled(Permission.LIBRARY_READ)),
+    auth_result: tuple[User | None, bool] = Depends(
+        require_ownership_permission(
+            Permission.LIBRARY_READ_ALL,
+            Permission.LIBRARY_READ_OWN,
+        )
+    ),
 ):
     """Get library statistics."""
+    user, can_read_all = auth_result
     # Stats exclude trashed files — users see counts/sizes for what's actually in the library.
-    active_only = LibraryFile.deleted_at.is_(None)
+    # Without LIBRARY_READ_ALL the stats reflect only the caller's own files —
+    # match what the file list endpoint shows so the numbers stay consistent.
+    file_filters = [LibraryFile.deleted_at.is_(None)]
+    if user is not None and not can_read_all:
+        file_filters.append(LibraryFile.created_by_id == user.id)
 
     # Total files
-    total_files_result = await db.execute(select(func.count(LibraryFile.id)).where(active_only))
+    total_files_result = await db.execute(select(func.count(LibraryFile.id)).where(*file_filters))
     total_files = total_files_result.scalar() or 0
 
-    # Total folders
+    # Total folders (folders are shared org structure, not per-user — count all)
     total_folders_result = await db.execute(select(func.count(LibraryFolder.id)))
     total_folders = total_folders_result.scalar() or 0
 
     # Total size
-    total_size_result = await db.execute(select(func.sum(LibraryFile.file_size)).where(active_only))
+    total_size_result = await db.execute(select(func.sum(LibraryFile.file_size)).where(*file_filters))
     total_size = total_size_result.scalar() or 0
 
     # Files by type
     type_result = await db.execute(
-        select(LibraryFile.file_type, func.count(LibraryFile.id)).where(active_only).group_by(LibraryFile.file_type)
+        select(LibraryFile.file_type, func.count(LibraryFile.id)).where(*file_filters).group_by(LibraryFile.file_type)
     )
     files_by_type = dict(type_result.all())
 
     # Total prints
-    total_prints_result = await db.execute(select(func.sum(LibraryFile.print_count)).where(active_only))
+    total_prints_result = await db.execute(select(func.sum(LibraryFile.print_count)).where(*file_filters))
     total_prints = total_prints_result.scalar() or 0
 
     # Disk space info

backend/app/api/routes/pending_uploads.py

@@ -8,7 +8,7 @@ from pydantic import BaseModel
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from backend.app.core.auth import RequirePermissionIfAuthEnabled
+from backend.app.core.auth import RequirePermissionIfAuthEnabled, require_ownership_permission
 from backend.app.core.database import get_db
 from backend.app.core.permissions import Permission
 from backend.app.models.pending_upload import PendingUpload
@@ -91,7 +91,12 @@ async def _augment_with_display_name(
 @router.get("/", response_model=list[PendingUploadResponse])
 async def list_pending_uploads(
     db: AsyncSession = Depends(get_db),
-    _: User | None = RequirePermissionIfAuthEnabled(Permission.QUEUE_READ),
+    _: tuple[User | None, bool] = Depends(
+        require_ownership_permission(
+            Permission.QUEUE_READ_ALL,
+            Permission.QUEUE_READ_OWN,
+        )
+    ),
 ):
     """List all pending uploads."""
     result = await db.execute(
@@ -104,7 +109,12 @@ async def list_pending_uploads(
 @router.get("/count")
 async def get_pending_count(
     db: AsyncSession = Depends(get_db),
-    _: User | None = RequirePermissionIfAuthEnabled(Permission.QUEUE_READ),
+    _: tuple[User | None, bool] = Depends(
+        require_ownership_permission(
+            Permission.QUEUE_READ_ALL,
+            Permission.QUEUE_READ_OWN,
+        )
+    ),
 ):
     """Get count of pending uploads."""
     result = await db.execute(select(PendingUpload).where(PendingUpload.status == "pending"))
@@ -208,7 +218,12 @@ async def discard_all_pending(
 async def get_pending_upload(
     upload_id: int,
     db: AsyncSession = Depends(get_db),
-    _: User | None = RequirePermissionIfAuthEnabled(Permission.QUEUE_READ),
+    _: tuple[User | None, bool] = Depends(
+        require_ownership_permission(
+            Permission.QUEUE_READ_ALL,
+            Permission.QUEUE_READ_OWN,
+        )
+    ),
 ):
     """Get a specific pending upload."""
     result = await db.execute(select(PendingUpload).where(PendingUpload.id == upload_id))

backend/app/api/routes/print_log.py

@@ -34,11 +34,20 @@ async def get_print_log(
     limit: int = Query(default=50, ge=1, le=500),
     offset: int = Query(default=0, ge=0),
     db: AsyncSession = Depends(get_db),
-    _: User | None = RequirePermissionIfAuthEnabled(Permission.ARCHIVES_READ),
+    auth_result: tuple[User | None, bool] = Depends(
+        require_ownership_permission(
+            Permission.ARCHIVES_READ_ALL,
+            Permission.ARCHIVES_READ_OWN,
+        )
+    ),
 ):
     """Get the print log."""
+    user, can_read_all = auth_result
     query = select(PrintLogEntry)
     count_query = select(func.count(PrintLogEntry.id))
+    if user is not None and not can_read_all:
+        query = query.where(PrintLogEntry.created_by_id == user.id)
+        count_query = count_query.where(PrintLogEntry.created_by_id == user.id)
 
     if printer_id is not None:
         query = query.where(PrintLogEntry.printer_id == printer_id)

backend/app/api/routes/print_queue.py

@@ -306,9 +306,15 @@ async def list_queue(
         None, description="Filter by target model (also includes model-based items when combined with printer_id)"
     ),
     db: AsyncSession = Depends(get_db),
-    _: User | None = RequirePermissionIfAuthEnabled(Permission.QUEUE_READ),
+    auth_result: tuple[User | None, bool] = Depends(
+        require_ownership_permission(
+            Permission.QUEUE_READ_ALL,
+            Permission.QUEUE_READ_OWN,
+        )
+    ),
 ):
     """List all queue items, optionally filtered by printer or status."""
+    user, can_read_all = auth_result
     query = (
         select(PrintQueueItem)
         .options(
@@ -320,6 +326,8 @@ async def list_queue(
         )
         .order_by(PrintQueueItem.printer_id.nulls_first(), PrintQueueItem.position)
     )
+    if user is not None and not can_read_all:
+        query = query.where(PrintQueueItem.created_by_id == user.id)
 
     if printer_id is not None:
         if printer_id == -1:
@@ -404,6 +412,18 @@ async def add_to_queue(
         archive = result.scalar_one_or_none()
         if not archive:
             raise HTTPException(400, "Archive not found")
+        # IDOR fix (maziggy/bambuddy-security #2): without this check, a
+        # caller with QUEUE_CREATE could queue any user's archive even
+        # without ARCHIVES_READ on it — Landon's PoC enumerated this on
+        # admin's archives as operator1. Gate on ARCHIVES_READ_ALL OR
+        # ownership of the archive. 404 (not 403) so we don't leak
+        # "this id exists but you can't queue it" for enumeration.
+        if (
+            current_user is not None
+            and not current_user.has_permission(Permission.ARCHIVES_READ_ALL.value)
+            and archive.created_by_id != current_user.id
+        ):
+            raise HTTPException(404, "Archive not found")
 
     # Validate library file exists (if provided) and get it for filament extraction
     library_file = None
@@ -412,6 +432,13 @@ async def add_to_queue(
         library_file = result.scalar_one_or_none()
         if not library_file:
             raise HTTPException(400, "Library file not found")
+        # Same shape: gate cross-user library-file queueing on LIBRARY_READ_ALL.
+        if (
+            current_user is not None
+            and not current_user.has_permission(Permission.LIBRARY_READ_ALL.value)
+            and library_file.created_by_id != current_user.id
+        ):
+            raise HTTPException(404, "Library file not found")
         # Bambu SD card is FAT32/exFAT — illegal filename chars would 553 at
         # FTP upload time (#1540). Reject at queue time so the user gets the
         # actionable error before waiting in queue.
@@ -685,12 +712,20 @@ async def bulk_update_queue_items(
 async def list_batches(
     status: str | None = Query(None, description="Filter by status (active, completed, cancelled)"),
     db: AsyncSession = Depends(get_db),
-    current_user: User | None = RequirePermissionIfAuthEnabled(Permission.QUEUE_READ),
+    auth_result: tuple[User | None, bool] = Depends(
+        require_ownership_permission(
+            Permission.QUEUE_READ_ALL,
+            Permission.QUEUE_READ_OWN,
+        )
+    ),
 ):
     """List all print batches with progress stats."""
+    current_user, can_read_all = auth_result
     query = select(PrintBatch).order_by(PrintBatch.created_at.desc())
     if status:
         query = query.where(PrintBatch.status == status)
+    if current_user is not None and not can_read_all:
+        query = query.where(PrintBatch.created_by_id == current_user.id)
     result = await db.execute(query)
     batches = result.scalars().all()
 
@@ -704,13 +739,25 @@ async def list_batches(
 async def get_batch(
     batch_id: int,
     db: AsyncSession = Depends(get_db),
-    current_user: User | None = RequirePermissionIfAuthEnabled(Permission.QUEUE_READ),
+    auth_result: tuple[User | None, bool] = Depends(
+        require_ownership_permission(
+            Permission.QUEUE_READ_ALL,
+            Permission.QUEUE_READ_OWN,
+        )
+    ),
 ):
     """Get a print batch with progress stats."""
+    current_user, can_read_all = auth_result
     result = await db.execute(select(PrintBatch).where(PrintBatch.id == batch_id))
     batch = result.scalar_one_or_none()
     if not batch:
         raise HTTPException(404, "Batch not found")
+    if (
+        current_user is not None
+        and not can_read_all
+        and (batch.created_by_id is None or batch.created_by_id != current_user.id)
+    ):
+        raise HTTPException(404, "Batch not found")
     return await _build_batch_response(db, batch)
 
 
@@ -782,9 +829,15 @@ async def _build_batch_response(db: AsyncSession, batch: PrintBatch) -> PrintBat
 async def get_queue_item(
     item_id: int,
     db: AsyncSession = Depends(get_db),
-    _: User | None = RequirePermissionIfAuthEnabled(Permission.QUEUE_READ),
+    auth_result: tuple[User | None, bool] = Depends(
+        require_ownership_permission(
+            Permission.QUEUE_READ_ALL,
+            Permission.QUEUE_READ_OWN,
+        )
+    ),
 ):
     """Get a specific queue item."""
+    current_user, can_read_all = auth_result
     result = await db.execute(
         select(PrintQueueItem)
         .options(
@@ -799,6 +852,12 @@ async def get_queue_item(
     item = result.scalar_one_or_none()
     if not item:
         raise HTTPException(404, "Queue item not found")
+    if (
+        current_user is not None
+        and not can_read_all
+        and (item.created_by_id is None or item.created_by_id != current_user.id)
+    ):
+        raise HTTPException(404, "Queue item not found")
     return _enrich_response(item)
 
 

backend/app/api/routes/slice_jobs.py

@@ -5,9 +5,9 @@ job_id and a status_url pointing here. The frontend polls this until
 status flips to `completed` or `failed`.
 """
 
-from fastapi import APIRouter, HTTPException
+from fastapi import APIRouter, Depends, HTTPException
 
-from backend.app.core.auth import RequirePermissionIfAuthEnabled
+from backend.app.core.auth import require_ownership_permission
 from backend.app.core.permissions import Permission
 from backend.app.models.user import User
 from backend.app.services.slice_dispatch import slice_dispatch
@@ -19,9 +19,17 @@ router = APIRouter(prefix="/slice-jobs", tags=["slice-jobs"])
 async def get_slice_job(
     job_id: int,
     # Job IDs are sequential integers and the body leaks source filenames
-    # plus the resulting library_file_id / archive_id. Gate on LIBRARY_READ
-    # — same baseline a user needs to see slice sources or results.
-    _: User | None = RequirePermissionIfAuthEnabled(Permission.LIBRARY_READ),
+    # plus the resulting library_file_id / archive_id. Gate on the library
+    # read permission family (own/all). NOTE: SliceJob is in-memory with no
+    # owner field, so we cannot per-row scope; callers with either OWN or
+    # ALL can poll any job_id. Adding owner_id to SliceJob is the proper
+    # follow-up (out of scope for the IDOR fix train).
+    _: tuple[User | None, bool] = Depends(
+        require_ownership_permission(
+            Permission.LIBRARY_READ_ALL,
+            Permission.LIBRARY_READ_OWN,
+        )
+    ),
 ):
     job = slice_dispatch.get(job_id)
     if job is None:

backend/app/api/routes/slicer_presets.py

@@ -26,7 +26,7 @@ from backend.app.api.routes.orca_cloud import (
     _build_authenticated_service as _build_orca_service,
     _load_credentials as _load_orca_credentials,
 )
-from backend.app.core.auth import RequirePermissionIfAuthEnabled
+from backend.app.core.auth import RequirePermissionIfAuthEnabled, require_ownership_permission
 from backend.app.core.config import settings as app_settings
 from backend.app.core.database import get_db
 from backend.app.core.permissions import Permission
@@ -535,7 +535,12 @@ async def list_unified_presets(
 async def get_preview_slice_progress(
     request_id: str,
     db: AsyncSession = Depends(get_db),
-    _: User | None = RequirePermissionIfAuthEnabled(Permission.LIBRARY_READ),
+    _: tuple[User | None, bool] = Depends(
+        require_ownership_permission(
+            Permission.LIBRARY_READ_ALL,
+            Permission.LIBRARY_READ_OWN,
+        )
+    ),
 ):
     """Proxy to the sidecar's ``GET /slice/progress/:requestId``.
 

backend/app/core/auth.py

@@ -61,9 +61,19 @@ logger = logging.getLogger(__name__)
 _APIKEY_SCOPE_BY_PERMISSION: dict[Permission, str] = {
     # can_read_status — read-only access to status, history, and configuration
     Permission.PRINTERS_READ: "can_read_status",
+    # Legacy flat permissions retained for back-compat with custom API keys —
+    # the role bootstraps no longer use these, but custom keys may still
+    # carry can_read_status scope mapping. New endpoints gate on the
+    # ARCHIVES_READ_OWN / _ALL split (maziggy/bambuddy-security #2).
     Permission.ARCHIVES_READ: "can_read_status",
+    Permission.ARCHIVES_READ_OWN: "can_read_status",
+    Permission.ARCHIVES_READ_ALL: "can_read_status",
     Permission.QUEUE_READ: "can_read_status",
+    Permission.QUEUE_READ_OWN: "can_read_status",
+    Permission.QUEUE_READ_ALL: "can_read_status",
     Permission.LIBRARY_READ: "can_read_status",
+    Permission.LIBRARY_READ_OWN: "can_read_status",
+    Permission.LIBRARY_READ_ALL: "can_read_status",
     Permission.PROJECTS_READ: "can_read_status",
     Permission.FILAMENTS_READ: "can_read_status",
     Permission.INVENTORY_READ: "can_read_status",

backend/app/core/database.py

@@ -2974,7 +2974,20 @@ async def seed_default_groups():
     logger = logging.getLogger(__name__)
 
     # Map old permissions to new ones for migration
-    # Administrators get *_all permissions, Operators get *_own permissions
+    # Administrators get *_all permissions, Operators get *_own permissions.
+    #
+    # NOTE on the read-flag asymmetry: write permissions (`update`, `delete`,
+    # `reprint`) are removed from the legacy flag and remapped to the OWN/ALL
+    # split — the legacy flag is dead on the API side. Read permissions are
+    # different: the frontend still gates UI actions (download buttons in
+    # ArchivesPage, preview button in FileManagerPage) on the LEGACY
+    # `archives:read` / `library:read` / `queue:read` strings. For admin we
+    # therefore keep the legacy flag (the `*_all` companion gets added via the
+    # backfill block below). For non-admin roles the legacy IS renamed to
+    # `_own` — that closes the IDOR (operators with a custom `archives:read`
+    # row can no longer read cross-user data) and the UI gates degrade to
+    # disabled-button state until the frontend is migrated to also accept
+    # `_own` (separate change). See maziggy/bambuddy-security #2.
     PERMISSION_MIGRATION_ALL = {
         "queue:update": "queue:update_all",
         "queue:delete": "queue:delete_all",
@@ -2988,11 +3001,20 @@ async def seed_default_groups():
     PERMISSION_MIGRATION_OWN = {
         "queue:update": "queue:update_own",
         "queue:delete": "queue:delete_own",
+        # Read permissions: any role NOT flagged as Administrator gets
+        # ownership-scoped reads. Pre-existing custom roles with the legacy
+        # `*:read` flag silently saw every user's items; the OWN variant
+        # closes that IDOR. Roles that genuinely need cross-user visibility
+        # must be re-granted `*:read_all` explicitly by an administrator
+        # after upgrade — fail-closed by default (per CWE-636).
+        "queue:read": "queue:read_own",
         "archives:update": "archives:update_own",
         "archives:delete": "archives:delete_own",
         "archives:reprint": "archives:reprint_own",
+        "archives:read": "archives:read_own",
         "library:update": "library:update_own",
         "library:delete": "library:delete_own",
+        "library:read": "library:read_own",
     }
 
     async with async_session() as session:
@@ -3040,11 +3062,14 @@ async def seed_default_groups():
                         for _own_perm, all_perm in [
                             ("queue:update_own", "queue:update_all"),
                             ("queue:delete_own", "queue:delete_all"),
+                            ("queue:read_own", "queue:read_all"),
                             ("archives:update_own", "archives:update_all"),
                             ("archives:delete_own", "archives:delete_all"),
                             ("archives:reprint_own", "archives:reprint_all"),
+                            ("archives:read_own", "archives:read_all"),
                             ("library:update_own", "library:update_all"),
                             ("library:delete_own", "library:delete_all"),
+                            ("library:read_own", "library:read_all"),
                         ]:
                             # Add *_all if not present
                             if all_perm not in new_permissions:
@@ -3113,6 +3138,81 @@ async def seed_default_groups():
                 admin_group.permissions = perms
         await session.commit()
 
+        # Backfill the read flag set for the Administrators group on existing
+        # installs (maziggy/bambuddy-security #2). Two layers:
+        #
+        # (a) New OWN/ALL splits — `archives:read_own` etc. Fresh installs get
+        #     these via ALL_PERMISSIONS; upgrades need the explicit backfill
+        #     so admin's permission set matches a fresh install's.
+        #
+        # (b) Legacy `archives:read` / `library:read` / `queue:read`. The
+        #     frontend still gates download / preview UI on these LEGACY
+        #     strings (see ArchivesPage / FileManagerPage), so admin needs
+        #     them retained even though the new API uses the OWN/ALL split.
+        #     The PERMISSION_MIGRATION_ALL map deliberately doesn't rename
+        #     read flags for admin — this backfill ensures they're present
+        #     even if they were stripped by hand or by an older migration.
+        #
+        # Also includes orca_cloud:auth for parity with fresh-install
+        # behaviour (ALL_PERMISSIONS covers it; backfill makes sure an
+        # admin role that's been customised since seed still has it).
+        result = await session.execute(select(Group).where(Group.name == "Administrators"))
+        admin_group = result.scalar_one_or_none()
+        if admin_group and admin_group.permissions is not None:
+            perms = list(admin_group.permissions)
+            added = False
+            for new_perm in (
+                "archives:read",
+                "archives:read_own",
+                "archives:read_all",
+                "library:read",
+                "library:read_own",
+                "library:read_all",
+                "queue:read",
+                "queue:read_own",
+                "queue:read_all",
+                "orca_cloud:auth",
+            ):
+                if new_perm not in perms:
+                    perms.append(new_perm)
+                    added = True
+                    logger.info("Added %s to Administrators group (backfill)", new_perm)
+            if added:
+                admin_group.permissions = perms
+        await session.commit()
+
+        # Same OWN-tier backfill for non-admin system groups. Operators and
+        # Viewers are seeded with _own on fresh installs (see DEFAULT_GROUPS),
+        # but the legacy-rename migration above won't run on a role that
+        # didn't carry the legacy `archives:read` flag. Without this block,
+        # an existing Operators row whose permissions list lacks the legacy
+        # flag would never get archives:read_own and operators would lose
+        # read access after upgrade. Re-check by group name so customised
+        # rows still get the correct OWN tier on next startup.
+        #
+        # Operators also get orca_cloud:auth backfilled — fresh installs now
+        # include it in the DEFAULT_GROUPS bootstrap, so this keeps upgrades
+        # consistent. Viewers do NOT get orca_cloud:auth (read-only role,
+        # not expected to author slicer presets / sync to Orca Cloud).
+        for non_admin_group_name in ("Operators", "Viewers"):
+            grp = (await session.execute(select(Group).where(Group.name == non_admin_group_name))).scalar_one_or_none()
+            if grp is None or grp.permissions is None:
+                continue
+            perms = list(grp.permissions)
+            changed = False
+            for own_perm in ("archives:read_own", "library:read_own", "queue:read_own"):
+                if own_perm not in perms:
+                    perms.append(own_perm)
+                    changed = True
+                    logger.info("Added %s to %s group (backfill)", own_perm, non_admin_group_name)
+            if non_admin_group_name == "Operators" and "orca_cloud:auth" not in perms:
+                perms.append("orca_cloud:auth")
+                changed = True
+                logger.info("Added orca_cloud:auth to Operators group (backfill)")
+            if changed:
+                grp.permissions = perms
+        await session.commit()
+
         # Backfill inventory forecast permissions for existing groups.
         # inventory:forecast_read was added after initial seeding, so groups
         # that already have inventory:read (or inventory:update) need it added.

backend/app/core/permissions.py

@@ -25,7 +25,12 @@ class Permission(StrEnum):
     PRINTERS_CLEAR_PLATE = "printers:clear_plate"  # Confirm plate cleared for next print
 
     # Archives
+    # ARCHIVES_READ kept for backward-compat with legacy custom roles, but new
+    # role bootstraps use the ownership-split variants below. seed_default_groups
+    # migrates pre-existing role rows: Administrators → ALL, everyone else → OWN.
     ARCHIVES_READ = "archives:read"
+    ARCHIVES_READ_OWN = "archives:read_own"
+    ARCHIVES_READ_ALL = "archives:read_all"
     ARCHIVES_CREATE = "archives:create"
     ARCHIVES_UPDATE_OWN = "archives:update_own"
     ARCHIVES_UPDATE_ALL = "archives:update_all"
@@ -37,6 +42,8 @@ class Permission(StrEnum):
 
     # Queue
     QUEUE_READ = "queue:read"
+    QUEUE_READ_OWN = "queue:read_own"
+    QUEUE_READ_ALL = "queue:read_all"
     QUEUE_CREATE = "queue:create"
     QUEUE_UPDATE_OWN = "queue:update_own"
     QUEUE_UPDATE_ALL = "queue:update_all"
@@ -46,6 +53,8 @@ class Permission(StrEnum):
 
     # Library
     LIBRARY_READ = "library:read"
+    LIBRARY_READ_OWN = "library:read_own"
+    LIBRARY_READ_ALL = "library:read_all"
     LIBRARY_UPLOAD = "library:upload"
     LIBRARY_UPDATE_OWN = "library:update_own"
     LIBRARY_UPDATE_ALL = "library:update_all"
@@ -185,7 +194,9 @@ PERMISSION_CATEGORIES = {
         Permission.PRINTERS_CLEAR_PLATE,
     ],
     "Archives": [
-        Permission.ARCHIVES_READ,
+        Permission.ARCHIVES_READ,  # legacy — kept for back-compat with custom roles
+        Permission.ARCHIVES_READ_OWN,
+        Permission.ARCHIVES_READ_ALL,
         Permission.ARCHIVES_CREATE,
         Permission.ARCHIVES_UPDATE_OWN,
         Permission.ARCHIVES_UPDATE_ALL,
@@ -196,7 +207,9 @@ PERMISSION_CATEGORIES = {
         Permission.ARCHIVES_PURGE,
     ],
     "Queue": [
-        Permission.QUEUE_READ,
+        Permission.QUEUE_READ,  # legacy — kept for back-compat with custom roles
+        Permission.QUEUE_READ_OWN,
+        Permission.QUEUE_READ_ALL,
         Permission.QUEUE_CREATE,
         Permission.QUEUE_UPDATE_OWN,
         Permission.QUEUE_UPDATE_ALL,
@@ -205,7 +218,9 @@ PERMISSION_CATEGORIES = {
         Permission.QUEUE_REORDER,
     ],
     "Library": [
-        Permission.LIBRARY_READ,
+        Permission.LIBRARY_READ,  # legacy — kept for back-compat with custom roles
+        Permission.LIBRARY_READ_OWN,
+        Permission.LIBRARY_READ_ALL,
         Permission.LIBRARY_UPLOAD,
         Permission.LIBRARY_UPDATE_OWN,
         Permission.LIBRARY_UPDATE_ALL,
@@ -350,25 +365,31 @@ DEFAULT_GROUPS = {
             Permission.PRINTERS_AMS_RFID.value,
             Permission.PRINTERS_CLEAR_PLATE.value,
             # Archives - own items only
-            Permission.ARCHIVES_READ.value,
+            Permission.ARCHIVES_READ_OWN.value,
             Permission.ARCHIVES_CREATE.value,
             Permission.ARCHIVES_UPDATE_OWN.value,
             Permission.ARCHIVES_DELETE_OWN.value,
             Permission.ARCHIVES_REPRINT_OWN.value,
             # Queue - own items only
-            Permission.QUEUE_READ.value,
+            Permission.QUEUE_READ_OWN.value,
             Permission.QUEUE_CREATE.value,
             Permission.QUEUE_UPDATE_OWN.value,
             Permission.QUEUE_DELETE_OWN.value,
             Permission.QUEUE_REORDER.value,
             # Library - own items only
-            Permission.LIBRARY_READ.value,
+            Permission.LIBRARY_READ_OWN.value,
             Permission.LIBRARY_UPLOAD.value,
             Permission.LIBRARY_UPDATE_OWN.value,
             Permission.LIBRARY_DELETE_OWN.value,
             # MakerWorld integration
             Permission.MAKERWORLD_VIEW.value,
             Permission.MAKERWORLD_IMPORT.value,
+            # Orca Cloud — needed for the Slice modal's Orca Cloud preset
+            # picker to populate. Workshops that use Orca Cloud presets
+            # need every operator to be able to authenticate. Bambu Cloud
+            # (CLOUD_AUTH) stays admin-only — that one is a more sensitive
+            # account binding.
+            Permission.ORCA_CLOUD_AUTH.value,
             # Projects - full access
             Permission.PROJECTS_READ.value,
             Permission.PROJECTS_CREATE.value,
@@ -438,9 +459,9 @@ DEFAULT_GROUPS = {
         "permissions": [
             # Read-only access
             Permission.PRINTERS_READ.value,
-            Permission.ARCHIVES_READ.value,
-            Permission.QUEUE_READ.value,
-            Permission.LIBRARY_READ.value,
+            Permission.ARCHIVES_READ_OWN.value,
+            Permission.QUEUE_READ_OWN.value,
+            Permission.LIBRARY_READ_OWN.value,
             Permission.PROJECTS_READ.value,
             Permission.FILAMENTS_READ.value,
             Permission.INVENTORY_READ.value,

backend/app/services/archive.py

@@ -1313,8 +1313,14 @@ class ArchiveService:
         date_to: date | None = None,
         limit: int = 50,
         offset: int = 0,
+        visible_to_user_id: int | None = None,
     ) -> list[PrintArchive]:
-        """List archives with optional filtering."""
+        """List archives with optional filtering.
+
+        ``visible_to_user_id`` scopes results to archives that user owns. Used
+        when the caller has ARCHIVES_READ_OWN but not ARCHIVES_READ_ALL — pass
+        ``None`` to skip the filter (caller has read-all or auth is disabled).
+        """
         from sqlalchemy.orm import selectinload
 
         query = (
@@ -1341,6 +1347,9 @@ class ArchiveService:
             dt_to = datetime.combine(date_to, time.max, tzinfo=timezone.utc)
             query = query.where(PrintArchive.created_at <= dt_to)
 
+        if visible_to_user_id is not None:
+            query = query.where(PrintArchive.created_by_id == visible_to_user_id)
+
         query = query.limit(limit).offset(offset)
         result = await self.db.execute(query)
         return list(result.scalars().all())

backend/app/services/export.py

@@ -80,6 +80,7 @@ class ExportService:
         date_from: datetime | None = None,
         date_to: datetime | None = None,
         search: str | None = None,
+        visible_to_user_id: int | None = None,
     ) -> tuple[bytes, str, str]:
         """Export archives to CSV or Excel format.
 
@@ -92,6 +93,8 @@ class ExportService:
             date_from: Filter by start date
             date_to: Filter by end date
             search: Search filter
+            visible_to_user_id: Scope rows to those owned by this user (used
+                when the caller has ARCHIVES_READ_OWN but not _ALL).
 
         Returns:
             Tuple of (file_bytes, filename, content_type)
@@ -112,6 +115,8 @@ class ExportService:
             query = query.where(PrintArchive.created_at >= date_from)
         if date_to:
             query = query.where(PrintArchive.created_at <= date_to)
+        if visible_to_user_id is not None:
+            query = query.where(PrintArchive.created_by_id == visible_to_user_id)
         if search:
             like_pattern = f"%{search}%"
             query = query.where(

backend/tests/integration/test_ownership_permissions.py

@@ -744,3 +744,283 @@ class TestUserItemsCountAndDeletion(TestOwnershipPermissionsSetup):
             headers={"Authorization": f"Bearer {auth_setup['admin_token']}"},
         )
         assert archive_response.status_code == 404
+
+
+class TestReadIDORClosure(TestOwnershipPermissionsSetup):
+    """Regression tests pinning maziggy/bambuddy-security #2 — IDOR on
+    archives / library / queue read paths.
+
+    Before the fix, ARCHIVES_READ / LIBRARY_READ / QUEUE_READ were flat
+    "see everything" permissions even though the write side was split into
+    OWN/ALL. An operator with only ARCHIVES_READ could read, download, and
+    queue any user's archive via direct id reference. These tests pin the
+    bambuddy_archive_idor.py and bambuddy_archive_viewer_idor.py PoC paths
+    so the IDOR can't regress silently.
+    """
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_operator_get_others_archive_returns_404_not_200(
+        self, async_client: AsyncClient, auth_setup, archive_factory, printer_factory, db_session
+    ):
+        """PoC #2 read path. operator1 GET /archives/{id} where id is admin's
+        archive must NOT leak the row. 404 (not 403) so the operator can't
+        enumerate which ids exist — same shape as a nonexistent id."""
+        printer = await printer_factory()
+        archive = await archive_factory(
+            printer.id,
+            print_name="Admin Archive",
+            created_by_id=auth_setup["admin_user"]["id"],
+        )
+        response = await async_client.get(
+            f"/api/v1/archives/{archive.id}",
+            headers={"Authorization": f"Bearer {auth_setup['operator_token']}"},
+        )
+        assert response.status_code == 404
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_operator_download_others_archive_returns_404(
+        self, async_client: AsyncClient, auth_setup, archive_factory, printer_factory, db_session
+    ):
+        """Viewer-IDOR PoC path: GET /archives/{id}/download on admin's archive.
+        Before the fix this streamed the 3MF body straight to a viewer-tier
+        token."""
+        printer = await printer_factory()
+        archive = await archive_factory(
+            printer.id,
+            print_name="Admin Archive 2",
+            created_by_id=auth_setup["admin_user"]["id"],
+        )
+        response = await async_client.get(
+            f"/api/v1/archives/{archive.id}/download",
+            headers={"Authorization": f"Bearer {auth_setup['viewer_token']}"},
+        )
+        assert response.status_code == 404
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_operator_list_archives_excludes_others(
+        self, async_client: AsyncClient, auth_setup, archive_factory, printer_factory, db_session
+    ):
+        """GET /archives/ must filter to own archives only for OWN-level callers."""
+        printer = await printer_factory()
+        own = await archive_factory(
+            printer.id, print_name="Operator's Own", created_by_id=auth_setup["operator_user"]["id"]
+        )
+        others = await archive_factory(printer.id, print_name="Admin's", created_by_id=auth_setup["admin_user"]["id"])
+        response = await async_client.get(
+            "/api/v1/archives/",
+            headers={"Authorization": f"Bearer {auth_setup['operator_token']}"},
+        )
+        assert response.status_code == 200
+        returned_ids = {a["id"] for a in response.json()}
+        assert own.id in returned_ids
+        assert others.id not in returned_ids
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_admin_list_archives_includes_all(
+        self, async_client: AsyncClient, auth_setup, archive_factory, printer_factory, db_session
+    ):
+        """ARCHIVES_READ_ALL → admin sees own + every user's archives."""
+        printer = await printer_factory()
+        admin_archive = await archive_factory(
+            printer.id, print_name="Admin's", created_by_id=auth_setup["admin_user"]["id"]
+        )
+        operator_archive = await archive_factory(
+            printer.id, print_name="Operator's", created_by_id=auth_setup["operator_user"]["id"]
+        )
+        response = await async_client.get(
+            "/api/v1/archives/",
+            headers={"Authorization": f"Bearer {auth_setup['admin_token']}"},
+        )
+        assert response.status_code == 200
+        returned_ids = {a["id"] for a in response.json()}
+        assert admin_archive.id in returned_ids
+        assert operator_archive.id in returned_ids
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_operator_cannot_queue_others_archive(
+        self, async_client: AsyncClient, auth_setup, archive_factory, printer_factory, db_session
+    ):
+        """PoC #2 queue path. POST /queue/ with admin's archive_id as
+        operator1 must return 404, not create a queue item. Before the fix
+        this returned 201 and queued the admin archive (Landon's CONFIRMED
+        line in the PoC)."""
+        printer = await printer_factory()
+        archive = await archive_factory(
+            printer.id,
+            print_name="Admin Archive (queue-target)",
+            created_by_id=auth_setup["admin_user"]["id"],
+        )
+        response = await async_client.post(
+            "/api/v1/queue/",
+            headers={"Authorization": f"Bearer {auth_setup['operator_token']}"},
+            json={"archive_id": archive.id, "printer_id": printer.id, "quantity": 1},
+        )
+        assert response.status_code == 404
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_admin_can_queue_others_archive(
+        self, async_client: AsyncClient, auth_setup, archive_factory, printer_factory, db_session
+    ):
+        """Belt-and-suspenders for the ALL path: admin (ARCHIVES_READ_ALL) can
+        queue a user's archive on their behalf — common workshop pattern."""
+        printer = await printer_factory()
+        archive = await archive_factory(
+            printer.id,
+            print_name="Operator's archive (queue by admin)",
+            created_by_id=auth_setup["operator_user"]["id"],
+        )
+        response = await async_client.post(
+            "/api/v1/queue/",
+            headers={"Authorization": f"Bearer {auth_setup['admin_token']}"},
+            json={"archive_id": archive.id, "printer_id": printer.id, "quantity": 1},
+        )
+        assert response.status_code == 200
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_operator_get_others_library_file_returns_404(
+        self, async_client: AsyncClient, auth_setup, db_session
+    ):
+        """Library IDOR closure (same shape as archives — closed in the same PR
+        per maziggy/bambuddy-security #2)."""
+        from backend.app.models.library import LibraryFile
+
+        admin_file = LibraryFile(
+            filename="admin_secret.3mf",
+            file_path="library/admin_secret.3mf",
+            file_type="3mf",
+            file_size=2048,
+            created_by_id=auth_setup["admin_user"]["id"],
+        )
+        db_session.add(admin_file)
+        await db_session.commit()
+        await db_session.refresh(admin_file)
+
+        response = await async_client.get(
+            f"/api/v1/library/files/{admin_file.id}",
+            headers={"Authorization": f"Bearer {auth_setup['operator_token']}"},
+        )
+        assert response.status_code == 404
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_operator_list_library_files_excludes_others(self, async_client: AsyncClient, auth_setup, db_session):
+        from backend.app.models.library import LibraryFile
+
+        own = LibraryFile(
+            filename="my_file.3mf",
+            file_path="library/my_file.3mf",
+            file_type="3mf",
+            file_size=1024,
+            created_by_id=auth_setup["operator_user"]["id"],
+        )
+        others = LibraryFile(
+            filename="admin_file.3mf",
+            file_path="library/admin_file.3mf",
+            file_type="3mf",
+            file_size=1024,
+            created_by_id=auth_setup["admin_user"]["id"],
+        )
+        db_session.add_all([own, others])
+        await db_session.commit()
+        await db_session.refresh(own)
+        await db_session.refresh(others)
+
+        response = await async_client.get(
+            "/api/v1/library/files",
+            headers={"Authorization": f"Bearer {auth_setup['operator_token']}"},
+        )
+        assert response.status_code == 200
+        returned_ids = {f["id"] for f in response.json()}
+        assert own.id in returned_ids
+        assert others.id not in returned_ids
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_operator_queue_list_excludes_others_items(
+        self, async_client: AsyncClient, auth_setup, archive_factory, printer_factory, db_session
+    ):
+        """GET /queue/ must filter to own queue items only for OWN callers —
+        same shape as the archive list."""
+        from backend.app.models.print_queue import PrintQueueItem
+
+        printer = await printer_factory()
+        archive = await archive_factory(printer.id, print_name="A", created_by_id=auth_setup["operator_user"]["id"])
+        own_item = PrintQueueItem(
+            archive_id=archive.id,
+            printer_id=printer.id,
+            status="pending",
+            position=1,
+            created_by_id=auth_setup["operator_user"]["id"],
+        )
+        admin_item = PrintQueueItem(
+            archive_id=archive.id,
+            printer_id=printer.id,
+            status="pending",
+            position=2,
+            created_by_id=auth_setup["admin_user"]["id"],
+        )
+        db_session.add_all([own_item, admin_item])
+        await db_session.commit()
+        await db_session.refresh(own_item)
+        await db_session.refresh(admin_item)
+
+        response = await async_client.get(
+            "/api/v1/queue/",
+            headers={"Authorization": f"Bearer {auth_setup['operator_token']}"},
+        )
+        assert response.status_code == 200
+        returned_ids = {q["id"] for q in response.json()}
+        assert own_item.id in returned_ids
+        assert admin_item.id not in returned_ids
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_operator_get_others_queue_item_returns_404(
+        self, async_client: AsyncClient, auth_setup, archive_factory, printer_factory, db_session
+    ):
+        """Direct-id queue item access — same enumeration risk as archive get."""
+        from backend.app.models.print_queue import PrintQueueItem
+
+        printer = await printer_factory()
+        archive = await archive_factory(printer.id, print_name="A", created_by_id=auth_setup["admin_user"]["id"])
+        admin_item = PrintQueueItem(
+            archive_id=archive.id,
+            printer_id=printer.id,
+            status="pending",
+            position=1,
+            created_by_id=auth_setup["admin_user"]["id"],
+        )
+        db_session.add(admin_item)
+        await db_session.commit()
+        await db_session.refresh(admin_item)
+
+        response = await async_client.get(
+            f"/api/v1/queue/{admin_item.id}",
+            headers={"Authorization": f"Bearer {auth_setup['operator_token']}"},
+        )
+        assert response.status_code == 404
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_auth_disabled_preserves_single_tenant_read_all(
+        self, async_client: AsyncClient, archive_factory, printer_factory
+    ):
+        """With auth disabled, ARCHIVES_READ resolves to read-all (can_modify_all=True
+        in require_ownership_permission's auth-disabled branch). Existing
+        single-user installs see no behavior change."""
+        printer = await printer_factory()
+        archive = await archive_factory(printer.id, print_name="Anonymous", created_by_id=None)
+        # No Authorization header — auth-disabled mode.
+        response = await async_client.get(f"/api/v1/archives/{archive.id}")
+        # Either 200 (auth disabled in this test session) or 401 (auth enabled
+        # from a prior test) — both are acceptable; the IDOR closure does not
+        # change auth-enable/disable behavior. Pin not-404 to avoid masking a
+        # regression where auth-disabled callers would lose access.
+        assert response.status_code in (200, 401)

backend/tests/integration/test_read_permission_backfill_migration.py

@@ -0,0 +1,236 @@
+"""Migration tests for maziggy/bambuddy-security #2 — read permission OWN/ALL backfill.
+
+Pre-fix, ARCHIVES_READ / LIBRARY_READ / QUEUE_READ were flat "read all" flags.
+Post-fix they split into OWN/ALL. The migration in seed_default_groups must:
+
+  1. Rename legacy `archives:read` etc to `archives:read_all` on Administrators
+     and to `archives:read_own` on every other role (fail-closed default).
+  2. Backfill `_own` AND `_all` variants for the Administrators group on upgrade
+     so an upgraded install matches a fresh install's permission set.
+  3. Backfill `_own` variants for Operators and Viewers so they keep read access
+     even if their stored row didn't carry the legacy flag.
+
+These regressions are the failure shape Maziggy hit on a live upgrade — the
+admin role ended up missing queue:read_own AND queue:read after migration.
+"""
+
+import pytest
+from httpx import AsyncClient
+from sqlalchemy import select
+
+from backend.app.core import database as _database_module
+from backend.app.core.database import seed_default_groups
+from backend.app.models.group import Group
+
+_READ_FLAGS = frozenset(
+    {
+        "archives:read",
+        "archives:read_own",
+        "archives:read_all",
+        "library:read",
+        "library:read_own",
+        "library:read_all",
+        "queue:read",
+        "queue:read_own",
+        "queue:read_all",
+    }
+)
+
+
+async def _strip_and_set(group_name: str, extra: list[str] | None = None) -> None:
+    """Strip every read flag from ``group_name`` then add ``extra`` flags.
+
+    Simulates a pre-migration state where the group either had only the
+    legacy flat permission (set ``extra=['archives:read']``) or no read
+    permission at all (set ``extra=None``).
+    """
+    async with _database_module.async_session() as session:
+        grp = (await session.execute(select(Group).where(Group.name == group_name))).scalar_one_or_none()
+        assert grp is not None, f"group {group_name} not pre-seeded"
+        stripped = [p for p in (grp.permissions or []) if p not in _READ_FLAGS]
+        stripped.extend(extra or [])
+        grp.permissions = stripped
+        await session.commit()
+
+
+async def _get_perms(group_name: str) -> set[str]:
+    async with _database_module.async_session() as session:
+        grp = (await session.execute(select(Group).where(Group.name == group_name))).scalar_one_or_none()
+        assert grp is not None
+        return set(grp.permissions or [])
+
+
+# Note: ``async_client`` is depended upon (even though unused) so pytest-asyncio
+# uses the same event loop the conftest fixture uses for async_session(). Without
+# it, calling ``async_session()`` twice in one test trips an asyncpg
+# "got Future attached to a different loop" RuntimeError.
+
+
+class TestReadPermissionMigration:
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_legacy_archives_read_renamed_to_all_for_administrators(self, async_client: AsyncClient):
+        """Existing Administrators group with legacy `archives:read` → gets
+        `archives:read_all` after seed_default_groups runs, and gets the
+        `_own` companion backfilled too."""
+        await seed_default_groups()
+        await _strip_and_set("Administrators", extra=["archives:read"])
+
+        await seed_default_groups()
+
+        perms = await _get_perms("Administrators")
+        # Rename happened: legacy renamed to _all
+        assert "archives:read_all" in perms
+        # Backfill also added _own so fresh install and upgraded install match
+        assert "archives:read_own" in perms
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_administrators_backfill_adds_all_six_read_flags(self, async_client: AsyncClient):
+        """Even with NO legacy flags present, Administrators ends up with both
+        OWN and ALL variants for archives / library / queue after the backfill
+        pass. This is the case Maziggy hit — admin missing `queue:read_own`
+        after upgrade."""
+        await seed_default_groups()
+        await _strip_and_set("Administrators")
+
+        await seed_default_groups()
+
+        perms = await _get_perms("Administrators")
+        for needed in (
+            "archives:read_own",
+            "archives:read_all",
+            "library:read_own",
+            "library:read_all",
+            "queue:read_own",
+            "queue:read_all",
+        ):
+            assert needed in perms, f"{needed} must be backfilled for Administrators"
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_operators_backfill_adds_own_read_flags(self, async_client: AsyncClient):
+        """Operators with no read flags get the _OWN variants backfilled
+        (fail-closed — no _ALL)."""
+        await seed_default_groups()
+        await _strip_and_set("Operators")
+
+        await seed_default_groups()
+
+        perms = await _get_perms("Operators")
+        assert "archives:read_own" in perms
+        assert "library:read_own" in perms
+        assert "queue:read_own" in perms
+        assert "archives:read_all" not in perms
+        assert "library:read_all" not in perms
+        assert "queue:read_all" not in perms
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_operators_legacy_archives_read_renamed_to_own(self, async_client: AsyncClient):
+        """Pre-PR Operators with legacy `archives:read` get the _OWN rename
+        (fail-closed — close the IDOR, the operator can re-request _ALL via
+        admin if cross-user visibility is genuinely needed)."""
+        await seed_default_groups()
+        await _strip_and_set("Operators", extra=["archives:read"])
+
+        await seed_default_groups()
+
+        perms = await _get_perms("Operators")
+        assert "archives:read_own" in perms
+        assert "archives:read_all" not in perms
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_administrators_legacy_archives_read_retained(self, async_client: AsyncClient):
+        """Admin keeps the LEGACY `archives:read` flag — the frontend gates
+        download / preview UI on it (ArchivesPage / FileManagerPage), and
+        removing it on rename was leaving admin with no visible download
+        buttons after upgrade. The new API gates use the _ALL variant which
+        the backfill also ensures is present."""
+        await seed_default_groups()
+        await _strip_and_set("Administrators", extra=["archives:read"])
+
+        await seed_default_groups()
+
+        perms = await _get_perms("Administrators")
+        # Both the legacy flag (for the UI) and the _all variant (for the API)
+        # must coexist on admin.
+        assert "archives:read" in perms
+        assert "archives:read_all" in perms
+        assert "archives:read_own" in perms
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_administrators_backfill_adds_legacy_read_flags(self, async_client: AsyncClient):
+        """Admin with NO read flags at all (hand-edited or stripped role) ends
+        up with the legacy `archives:read` / `queue:read` / `library:read`
+        backfilled — so the UI gates work — alongside the OWN/ALL split."""
+        await seed_default_groups()
+        await _strip_and_set("Administrators")
+
+        await seed_default_groups()
+
+        perms = await _get_perms("Administrators")
+        for needed in (
+            "archives:read",
+            "library:read",
+            "queue:read",
+            "archives:read_own",
+            "archives:read_all",
+            "library:read_own",
+            "library:read_all",
+            "queue:read_own",
+            "queue:read_all",
+        ):
+            assert needed in perms, f"{needed} must be backfilled for Administrators"
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_administrators_orca_cloud_auth_backfilled(self, async_client: AsyncClient):
+        """Admin without `orca_cloud:auth` (older custom edit) gets it
+        backfilled — matches the fresh-install default."""
+        await seed_default_groups()
+        async with _database_module.async_session() as session:
+            grp = (await session.execute(select(Group).where(Group.name == "Administrators"))).scalar_one()
+            grp.permissions = [p for p in (grp.permissions or []) if p != "orca_cloud:auth"]
+            await session.commit()
+
+        await seed_default_groups()
+
+        perms = await _get_perms("Administrators")
+        assert "orca_cloud:auth" in perms
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_operators_orca_cloud_auth_backfilled(self, async_client: AsyncClient):
+        """Operators on upgraded installs get `orca_cloud:auth` backfilled
+        (the new default — needed for the Slice modal's Orca Cloud preset
+        picker)."""
+        await seed_default_groups()
+        async with _database_module.async_session() as session:
+            grp = (await session.execute(select(Group).where(Group.name == "Operators"))).scalar_one()
+            grp.permissions = [p for p in (grp.permissions or []) if p != "orca_cloud:auth"]
+            await session.commit()
+
+        await seed_default_groups()
+
+        perms = await _get_perms("Operators")
+        assert "orca_cloud:auth" in perms
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_viewers_do_not_get_orca_cloud_auth(self, async_client: AsyncClient):
+        """Viewers stay read-only — orca_cloud:auth is not added by the
+        backfill (matches the fresh-install Viewers bootstrap, which
+        intentionally excludes cloud-auth permissions)."""
+        await seed_default_groups()
+        async with _database_module.async_session() as session:
+            grp = (await session.execute(select(Group).where(Group.name == "Viewers"))).scalar_one()
+            grp.permissions = [p for p in (grp.permissions or []) if p != "orca_cloud:auth"]
+            await session.commit()
+
+        await seed_default_groups()
+
+        perms = await _get_perms("Viewers")
+        assert "orca_cloud:auth" not in perms