Security hardening (maziggy/bambuddy-security #29)

.basedpyright/baseline.json

@@ -75,14 +75,6 @@
                     "lineCount": 1
                 }
             },
-            {
-                "code": "reportArgumentType",
-                "range": {
-                    "startColumn": 30,
-                    "endColumn": 67,
-                    "lineCount": 1
-                }
-            },
             {
                 "code": "reportOptionalMemberAccess",
                 "range": {

backend/app/api/routes/archives.py

@@ -14,6 +14,7 @@ from fastapi import APIRouter, Depends, File, Form, HTTPException, Query, Reques
 from fastapi.responses import FileResponse, Response
 from sqlalchemy import and_, case, func, or_, select
 from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.sql.elements import ColumnElement
 
 from backend.app.core import database
 from backend.app.core.auth import (
@@ -177,6 +178,17 @@ async def _claimed_timelapse_stems(db, printer_id: int | None, exclude_archive_i
     return {Path(p).stem for p in rows.scalars().all() if p}
 
 
+def _visible_archive_clauses(user, can_read_all: bool, printer_scope: PrinterScope) -> list:
+    """WHERE clauses for the archives a caller may see, as _ensure_archive_visible decides one by one."""
+    clauses = []
+    if (scope_clause := printer_scope.where(PrintArchive.printer_id)) is not None:
+        clauses.append(scope_clause)
+    if not can_read_all:
+        # Own archives only; an ownerless one needs read-all, and no user sees none.
+        clauses.append(PrintArchive.created_by_id == user.id if user is not None else PrintArchive.id.is_(None))
+    return clauses
+
+
 def _ensure_archive_visible(
     archive: PrintArchive | None,
     user: User | None,
@@ -450,8 +462,10 @@ async def list_archives(
         printer_scope=printer_scope,
     )
 
-    # Get sets of duplicate hashes and duplicate (name, hash) pairs (efficient single queries)
-    duplicate_hashes, duplicate_name_hash_pairs = await service.get_duplicate_hashes_and_names()
+    # Get sets of duplicate hashes and duplicate (name, hash) pairs (efficient single queries),
+    # counted among the archives this caller may see, as the duplicates view lists them.
+    visible = _visible_archive_clauses(user, can_read_all, printer_scope)
+    duplicate_hashes, duplicate_name_hash_pairs = await service.get_duplicate_hashes_and_names(visible)
 
     # Batch-load duplicate groups once for the current page keys.
     duplicate_hashes_in_page = {
@@ -482,7 +496,7 @@ async def list_archives(
                 PrintArchive.created_at,
                 PrintArchive.content_hash,
                 func.lower(PrintArchive.print_name).label("print_name_lower"),
-            ).where(or_(*duplicate_group_conditions), PrintArchive.deleted_at.is_(None))
+            ).where(or_(*duplicate_group_conditions), PrintArchive.deleted_at.is_(None), *visible)
         )
 
         duplicate_groups_by_hash: dict[str, list[tuple[int, datetime]]] = defaultdict(list)
@@ -617,6 +631,7 @@ async def no_3mf_warning(
             Permission.ARCHIVES_READ_OWN,
         )
     ),
+    printer_scope: PrinterScope = RequestPrinterScope,
 ):
     """Whether to nudge the user about a print that archived without its 3MF,
     and why. True iff any archive in the last 30 days was created via the
@@ -654,6 +669,8 @@ async def no_3mf_warning(
     ]
     if user is not None and not can_read_all:
         conditions.append(PrintArchive.created_by_id == user.id)
+    if (scope_clause := printer_scope.where(PrintArchive.printer_id)) is not None:
+        conditions.append(scope_clause)
     result = await db.execute(select(PrintArchive.extra_data).where(*conditions))
     reasons: set[str] = set()
     has_fallback = False
@@ -1633,6 +1650,7 @@ async def get_all_tags(
             Permission.ARCHIVES_READ_OWN,
         )
     ),
+    printer_scope: PrinterScope = RequestPrinterScope,
 ):
     """List all unique tags with usage counts.
 
@@ -1640,9 +1658,11 @@ async def get_all_tags(
     """
     user, can_read_all = auth_result
     # Query all archives with non-null tags
-    tag_conditions = [PrintArchive.tags.isnot(None), PrintArchive.deleted_at.is_(None)]
+    tag_conditions: list[ColumnElement[bool]] = [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)
+    if (scope_clause := printer_scope.where(PrintArchive.printer_id)) is not None:
+        tag_conditions.append(scope_clause)
     result = await db.execute(select(PrintArchive.tags).where(*tag_conditions))
     all_tags_rows = result.all()
 
@@ -1668,6 +1688,7 @@ async def rename_tag(
     request: Request,
     db: AsyncSession = Depends(get_db),
     _: User | None = RequirePermissionIfAuthEnabled(Permission.ARCHIVES_UPDATE_ALL),
+    printer_scope: PrinterScope = RequestPrinterScope,
 ):
     """Rename a tag across all archives.
 
@@ -1685,7 +1706,11 @@ async def rename_tag(
 
     # Find all archives containing the old tag
     result = await db.execute(
-        select(PrintArchive).where(PrintArchive.tags.isnot(None), PrintArchive.deleted_at.is_(None))
+        select(PrintArchive).where(
+            PrintArchive.tags.isnot(None),
+            PrintArchive.deleted_at.is_(None),
+            *_visible_archive_clauses(None, True, printer_scope),
+        )
     )
     archives = list(result.scalars().all())
 
@@ -1716,6 +1741,7 @@ async def delete_tag(
     tag_name: str,
     db: AsyncSession = Depends(get_db),
     _: User | None = RequirePermissionIfAuthEnabled(Permission.ARCHIVES_UPDATE_ALL),
+    printer_scope: PrinterScope = RequestPrinterScope,
 ):
     """Delete a tag from all archives.
 
@@ -1723,7 +1749,11 @@ async def delete_tag(
     """
     # Find all archives containing the tag
     result = await db.execute(
-        select(PrintArchive).where(PrintArchive.tags.isnot(None), PrintArchive.deleted_at.is_(None))
+        select(PrintArchive).where(
+            PrintArchive.tags.isnot(None),
+            PrintArchive.deleted_at.is_(None),
+            *_visible_archive_clauses(None, True, printer_scope),
+        )
     )
     archives = list(result.scalars().all())
 
@@ -1766,6 +1796,7 @@ async def get_archive(
         content_hash=archive.content_hash,
         print_name=archive.print_name,
         makerworld_model_id=makerworld_id,
+        visible=_visible_archive_clauses(user, can_read_all, printer_scope),
     )
     run_aggregates = await _load_run_aggregates(db, [archive.id])
     return archive_to_response(archive, duplicates, run_aggregate=run_aggregates.get(archive.id))
@@ -1860,7 +1891,9 @@ async def find_similar_archives(
 
     service = ArchiveComparisonService(db)
     try:
-        return await service.find_similar_archives(archive_id, limit=limit)
+        return await service.find_similar_archives(
+            archive_id, limit=limit, visible=_visible_archive_clauses(user, can_read_all, printer_scope)
+        )
     except ValueError as e:
         raise HTTPException(404, str(e))
 
@@ -2133,12 +2166,17 @@ async def rescan_archive(
 async def recalculate_all_costs(
     db: AsyncSession = Depends(get_db),
     _: User | None = RequirePermissionIfAuthEnabled(Permission.ARCHIVES_UPDATE_ALL),
+    printer_scope: PrinterScope = RequestPrinterScope,
 ):
     """Recalculate costs for all archives based on filament usage and prices."""
 
     from backend.app.api.routes.settings import get_setting
 
-    result = await db.execute(select(PrintArchive))
+    # Only the archives of the caller's printers, as the archive list shows them.
+    query = select(PrintArchive)
+    if (clause := printer_scope.where(PrintArchive.printer_id)) is not None:
+        query = query.where(clause)
+    result = await db.execute(query)
     archives = list(result.scalars().all())
 
     # Load all filaments for lookup
@@ -2211,11 +2249,16 @@ async def recalculate_all_costs(
 async def rescan_all_archives(
     db: AsyncSession = Depends(get_db),
     _: User | None = RequirePermissionIfAuthEnabled(Permission.ARCHIVES_UPDATE_ALL),
+    printer_scope: PrinterScope = RequestPrinterScope,
 ):
     """Rescan all archives and update their metadata."""
     from backend.app.services.archive import ThreeMFParser
 
-    result = await db.execute(select(PrintArchive))
+    # Only the archives of the caller's printers, as the archive list shows them.
+    query = select(PrintArchive)
+    if (clause := printer_scope.where(PrintArchive.printer_id)) is not None:
+        query = query.where(clause)
+    result = await db.execute(query)
     archives = list(result.scalars().all())
 
     updated = 0
@@ -2280,6 +2323,7 @@ async def get_archive_duplicates(
         content_hash=archive.content_hash,
         print_name=archive.print_name,
         makerworld_model_id=makerworld_id,
+        visible=_visible_archive_clauses(user, can_read_all, printer_scope),
     )
     return {"duplicates": duplicates, "count": len(duplicates)}
 
@@ -2288,9 +2332,14 @@ async def get_archive_duplicates(
 async def backfill_content_hashes(
     db: AsyncSession = Depends(get_db),
     _: User | None = RequirePermissionIfAuthEnabled(Permission.ARCHIVES_UPDATE_ALL),
+    printer_scope: PrinterScope = RequestPrinterScope,
 ):
     """Compute and store content hashes for all archives missing them."""
-    result = await db.execute(select(PrintArchive).where(PrintArchive.content_hash.is_(None)))
+    result = await db.execute(
+        select(PrintArchive).where(
+            PrintArchive.content_hash.is_(None), *_visible_archive_clauses(None, True, printer_scope)
+        )
+    )
     archives = list(result.scalars().all())
 
     updated = 0
@@ -5468,6 +5517,7 @@ async def upload_source_3mf_by_name(
     print_name: str = Query(None, description="Match archive by print name"),
     db: AsyncSession = Depends(get_db),
     _: User | None = RequirePermissionIfAuthEnabled(Permission.ARCHIVES_UPDATE_ALL),
+    printer_scope: PrinterScope = RequestPrinterScope,
 ):
     """Upload source 3MF and match to archive by print name.
 
@@ -5488,9 +5538,11 @@ async def upload_source_3mf_by_name(
             print_name = print_name[:-7]
 
     # Find matching archive - try exact match first, then fuzzy
+    # Only an archive on the caller's printers can be matched, as it is listed.
+    visible = _visible_archive_clauses(None, True, printer_scope)
     result = await db.execute(
         select(PrintArchive)
-        .where(PrintArchive.print_name == print_name)
+        .where(PrintArchive.print_name == print_name, *visible)
         .order_by(PrintArchive.created_at.desc())
         .limit(1)
     )
@@ -5500,7 +5552,7 @@ async def upload_source_3mf_by_name(
         # Try matching filename without .gcode.3mf
         result = await db.execute(
             select(PrintArchive)
-            .where(PrintArchive.filename.like(f"{print_name}%"))
+            .where(PrintArchive.filename.like(f"{print_name}%"), *visible)
             .order_by(PrintArchive.created_at.desc())
             .limit(1)
         )
@@ -5510,7 +5562,7 @@ async def upload_source_3mf_by_name(
         # Try case-insensitive partial match on print_name
         result = await db.execute(
             select(PrintArchive)
-            .where(PrintArchive.print_name.ilike(f"%{print_name}%"))
+            .where(PrintArchive.print_name.ilike(f"%{print_name}%"), *visible)
             .order_by(PrintArchive.created_at.desc())
             .limit(1)
         )

backend/app/core/websocket.py

@@ -31,6 +31,31 @@ def _may_receive(connection: WebSocket, printer_id: int | None) -> bool:
     return scope is not None and scope.allows(printer_id)
 
 
+# A printer id no limited scope holds: what an archive event carries when its
+# printer cannot be told, so only sockets that see every printer receive it.
+_UNKNOWN_PRINTER = -1
+
+
+async def _archive_printer_id(archive_id: Any) -> int | None:
+    """The printer an archive belongs to (None: printer-less, seen by everyone)."""
+    from sqlalchemy import select
+
+    from backend.app.core.database import async_session
+    from backend.app.models.archive import PrintArchive
+
+    if not isinstance(archive_id, int):
+        return _UNKNOWN_PRINTER
+    try:
+        async with async_session() as db:
+            row = (await db.execute(select(PrintArchive.printer_id).where(PrintArchive.id == archive_id))).first()
+    except Exception:
+        logger.exception("Could not look up the printer of archive %s for a live update", archive_id)
+        return _UNKNOWN_PRINTER
+    if row is None:
+        return _UNKNOWN_PRINTER
+    return row[0]
+
+
 class ConnectionManager:
     """Manages WebSocket connections and broadcasts."""
 
@@ -257,13 +282,15 @@ class ConnectionManager:
         )
 
     async def send_archive_updated(self, archive: dict):
-        """Notify clients that an archive was updated."""
-        await self.broadcast(
-            {
-                "type": "archive_updated",
-                "data": archive,
-            }
-        )
+        """Notify clients that an archive was updated.
+
+        Only sockets that may see the archive's printer receive it: callers send
+        just the changed fields, so the printer is looked up here when missing.
+        """
+        message: dict[str, Any] = {"type": "archive_updated", "data": archive}
+        if _message_printer_id(message) is None:
+            message["printer_id"] = await _archive_printer_id(archive.get("id"))
+        await self.broadcast(message)
 
     async def send_queue_item_uploading(
         self,

backend/app/services/archive.py

@@ -6,8 +6,10 @@ import os
 import re
 import shutil
 import zipfile
+from collections.abc import Sequence
 from datetime import date, datetime, time, timezone
 from pathlib import Path
+from typing import Any
 
 from defusedxml import ElementTree as ET
 from sqlalchemy import and_, or_, select, text
@@ -1110,9 +1112,13 @@ class ArchiveService:
                 sha256.update(chunk)
         return sha256.hexdigest()
 
-    async def get_duplicate_hashes_and_names(self) -> tuple[set[str], set[tuple[str, str]]]:
+    async def get_duplicate_hashes_and_names(
+        self, visible: Sequence[Any] = ()
+    ) -> tuple[set[str], set[tuple[str, str]]]:
         """Get all content hashes and (print name, hash) pairs that appear more than once.
 
+        ``visible`` limits the count to archives the caller may see.
+
         For hashes: returns all hashes with > 1 archive (true duplicates).
         For name/hash pairs: returns only pairs that have > 1 archive
                      (i.e., same file archived multiple times, not different files with same name).
@@ -1127,7 +1133,7 @@ class ArchiveService:
         # duplicate even though the user only sees one row.
         result = await self.db.execute(
             select(PrintArchive.content_hash)
-            .where(PrintArchive.content_hash.isnot(None), PrintArchive.deleted_at.is_(None))
+            .where(PrintArchive.content_hash.isnot(None), PrintArchive.deleted_at.is_(None), *visible)
             .group_by(PrintArchive.content_hash)
             .having(func.count(PrintArchive.id) > 1)
         )
@@ -1141,6 +1147,7 @@ class ArchiveService:
                 PrintArchive.print_name.isnot(None),
                 PrintArchive.content_hash.isnot(None),
                 PrintArchive.deleted_at.is_(None),
+                *visible,
             )
             .group_by(func.lower(PrintArchive.print_name), PrintArchive.content_hash)
             .having(func.count(PrintArchive.id) > 1)
@@ -1155,9 +1162,12 @@ class ArchiveService:
         content_hash: str | None = None,
         print_name: str | None = None,
         makerworld_model_id: str | None = None,
+        visible: Sequence[Any] = (),
     ) -> list[dict]:
         """Find duplicate archives based on hash or name matching.
 
+        ``visible`` holds WHERE clauses limiting matches to archives the caller may see.
+
         Returns list of dicts with id, print_name, created_at, match_type.
         """
         duplicates = []
@@ -1171,6 +1181,7 @@ class ArchiveService:
                         PrintArchive.content_hash == content_hash,
                         PrintArchive.id != archive_id,
                         PrintArchive.deleted_at.is_(None),
+                        *visible,
                     )
                 )
                 .order_by(PrintArchive.created_at.desc())
@@ -1190,7 +1201,7 @@ class ArchiveService:
         # Prefer strict name+hash matching when hash exists; fallback to name-only for legacy/manual
         # archives that may not have a content_hash.
         if print_name or makerworld_model_id:
-            conditions = [PrintArchive.id != archive_id, PrintArchive.deleted_at.is_(None)]
+            conditions = [PrintArchive.id != archive_id, PrintArchive.deleted_at.is_(None), *visible]
 
             name_conditions = []
             if print_name:

backend/app/services/archive_comparison.py

@@ -1,3 +1,6 @@
+from collections.abc import Sequence
+from typing import Any
+
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import selectinload
@@ -181,12 +184,14 @@ class ArchiveComparisonService:
         self,
         archive_id: int,
         limit: int = 10,
+        visible: Sequence[Any] = (),
     ) -> list[dict]:
         """Find archives with similar settings for comparison.
 
         Args:
             archive_id: The archive to find similar ones for
             limit: Maximum number of results
+            visible: WHERE clauses limiting matches to archives the caller may see
 
         Returns:
             List of similar archives with match reasons
@@ -210,6 +215,7 @@ class ArchiveComparisonService:
                     PrintArchive.id != archive_id,
                     PrintArchive.print_name == reference.print_name,
                     PrintArchive.deleted_at.is_(None),
+                    *visible,
                 )
                 .order_by(PrintArchive.created_at.desc())
                 .limit(limit)
@@ -236,6 +242,7 @@ class ArchiveComparisonService:
                     PrintArchive.id != archive_id,
                     PrintArchive.content_hash == reference.content_hash,
                     PrintArchive.deleted_at.is_(None),
+                    *visible,
                 )
                 .order_by(PrintArchive.created_at.desc())
                 .limit(limit - len(similar))
@@ -262,6 +269,8 @@ class ArchiveComparisonService:
                 .where(
                     PrintArchive.id != archive_id,
                     PrintArchive.filament_type == reference.filament_type,
+                    PrintArchive.deleted_at.is_(None),
+                    *visible,
                 )
                 .order_by(PrintArchive.created_at.desc())
                 .limit(limit - len(similar))

backend/tests/integration/test_bulk_archive_routes_scope.py

@@ -0,0 +1,152 @@
+"""Bulk archive actions keep to the caller's printers.
+
+Recalculating costs and rescanning all archives went over every archive,
+so a member of a group limited to some printers rewrote other printers'
+archives and got their ids back in the error list.
+"""
+
+from __future__ import annotations
+
+import pytest
+from httpx import AsyncClient
+from sqlalchemy import select
+
+from backend.app.models.archive import PrintArchive
+from backend.tests.integration.test_printer_scope_1727 import TEAM_PERMISSIONS, _admin_token, _auth, _group, _user
+
+pytestmark = [pytest.mark.asyncio, pytest.mark.integration]
+
+
+async def _member(client: AsyncClient, printer_ids: list[int]) -> dict[str, str]:
+    admin = await _admin_token(client)
+    perms = await _group(client, admin, "perms", permissions=[*TEAM_PERMISSIONS, "archives:update_all"])
+    team = await _group(client, admin, "team", printer_ids=printer_ids)
+    return _auth((await _user(client, admin, "member", [perms, team]))[0])
+
+
+async def test_rescan_all_reports_only_the_team_printers_archives(
+    async_client: AsyncClient, printer_factory, archive_factory
+):
+    a, b = await printer_factory(name="A"), await printer_factory(name="B")
+    # No file behind either archive, so each one lands in the error list.
+    on_a = await archive_factory(a.id, file_path="archives/missing-a.3mf")
+    await archive_factory(b.id, file_path="archives/missing-b.3mf")
+    member = await _member(async_client, [a.id])
+
+    response = await async_client.post("/api/v1/archives/rescan-all", headers=member)
+
+    assert response.status_code == 200, response.text
+    assert [error["id"] for error in response.json()["errors"]] == [on_a.id]
+
+
+async def test_recalculating_costs_leaves_other_printers_alone(
+    async_client: AsyncClient, db_session, printer_factory, archive_factory
+):
+    a, b = await printer_factory(name="A"), await printer_factory(name="B")
+    on_a = await archive_factory(a.id, cost=99.0, filament_used_grams=100, filament_type="PLA")
+    on_b = await archive_factory(b.id, cost=99.0, filament_used_grams=100, filament_type="PLA")
+    ids = (on_a.id, on_b.id)
+    member = await _member(async_client, [a.id])
+
+    response = await async_client.post("/api/v1/archives/recalculate-costs", headers=member)
+
+    assert response.status_code == 200, response.text
+    db_session.expire_all()
+    costs = dict(
+        (await db_session.execute(select(PrintArchive.id, PrintArchive.cost).where(PrintArchive.id.in_(ids)))).all()
+    )
+    assert costs[ids[0]] == 2.5  # 100 g at the default 25.00 per kg
+    assert costs[ids[1]] == 99.0
+
+
+async def test_tags_keep_to_the_team_printers(async_client: AsyncClient, db_session, printer_factory, archive_factory):
+    a, b = await printer_factory(name="A"), await printer_factory(name="B")
+    on_a = await archive_factory(a.id, tags="shared, alpha")
+    on_b = await archive_factory(b.id, tags="shared, beta")
+    ids = (on_a.id, on_b.id)
+    member = await _member(async_client, [a.id])
+
+    listed = await async_client.get("/api/v1/archives/tags", headers=member)
+    renamed = await async_client.put("/api/v1/archives/tags/shared", headers=member, json={"new_name": "common"})
+    deleted = await async_client.delete("/api/v1/archives/tags/common", headers=member)
+
+    assert sorted(t["name"] for t in listed.json()) == ["alpha", "shared"]
+    assert (renamed.json()["affected"], deleted.json()["affected"]) == (1, 1)
+    db_session.expire_all()
+    tags = dict(
+        (await db_session.execute(select(PrintArchive.id, PrintArchive.tags).where(PrintArchive.id.in_(ids)))).all()
+    )
+    assert tags[ids[1]] == "shared, beta"
+
+
+async def test_duplicates_and_similar_show_only_visible_archives(
+    async_client: AsyncClient, printer_factory, archive_factory
+):
+    a, b = await printer_factory(name="A"), await printer_factory(name="B")
+    mine = await archive_factory(a.id, print_name="Benchy", content_hash="h1")
+    same_team = await archive_factory(a.id, print_name="Benchy", content_hash="h1")
+    await archive_factory(b.id, print_name="Benchy", content_hash="h1")
+    member = await _member(async_client, [a.id])
+
+    duplicates = await async_client.get(f"/api/v1/archives/{mine.id}/duplicates", headers=member)
+    detail = await async_client.get(f"/api/v1/archives/{mine.id}", headers=member)
+    similar = await async_client.get(f"/api/v1/archives/{mine.id}/similar", headers=member)
+
+    assert [d["id"] for d in duplicates.json()["duplicates"]] == [same_team.id]
+    assert [d["id"] for d in detail.json()["duplicates"]] == [same_team.id]
+    assert {s["archive"]["id"] for s in similar.json()} == {same_team.id}
+
+
+async def test_backfill_hashes_reports_only_the_team_printers_archives(
+    async_client: AsyncClient, printer_factory, archive_factory
+):
+    a, b = await printer_factory(name="A"), await printer_factory(name="B")
+    on_a = await archive_factory(a.id, file_path="archives/none-a.3mf", content_hash=None)
+    await archive_factory(b.id, file_path="archives/none-b.3mf", content_hash=None)
+    member = await _member(async_client, [a.id])
+
+    response = await async_client.post("/api/v1/archives/backfill-hashes", headers=member)
+
+    assert [e["id"] for e in response.json()["errors"]] == [on_a.id]
+
+
+async def test_uploading_a_source_by_name_matches_only_the_team_printers(
+    async_client: AsyncClient, printer_factory, archive_factory
+):
+    a, b = await printer_factory(name="A"), await printer_factory(name="B")
+    await archive_factory(b.id, print_name="Secret Part")
+    member = await _member(async_client, [a.id])
+
+    response = await async_client.post(
+        "/api/v1/archives/upload-source",
+        headers=member,
+        params={"print_name": "Secret"},
+        files={"file": ("Secret Part.3mf", b"PK\x03\x04", "application/octet-stream")},
+    )
+
+    assert response.status_code == 404, response.text
+    assert "Secret Part" not in response.text
+
+
+async def test_the_list_counts_duplicates_among_visible_archives_only(
+    async_client: AsyncClient, printer_factory, archive_factory
+):
+    a, b = await printer_factory(name="A"), await printer_factory(name="B")
+    await archive_factory(b.id, print_name="Benchy", content_hash="h2")  # older, on B
+    mine = await archive_factory(a.id, print_name="Benchy", content_hash="h2")
+    member = await _member(async_client, [a.id])
+
+    listed = {row["id"]: row for row in (await async_client.get("/api/v1/archives/", headers=member)).json()}
+
+    assert listed[mine.id]["duplicate_count"] in (0, None)
+    assert listed[mine.id]["original_archive_id"] is None
+
+
+async def test_the_no_3mf_warning_ignores_other_printers(async_client: AsyncClient, printer_factory, archive_factory):
+    a, b = await printer_factory(name="A"), await printer_factory(name="B")
+    await archive_factory(b.id, extra_data={"no_3mf_available": True, "no_3mf_reason": "internal_storage"})
+    member = await _member(async_client, [a.id])
+
+    response = await async_client.get("/api/v1/archives/no-3mf-warning", headers=member)
+
+    assert response.json() == {"has_fallback": False, "reason": None}

backend/tests/unit/test_archive_updated_scope.py

@@ -0,0 +1,43 @@
+"""Live archive updates reach only sockets that may see the archive's printer."""
+
+from __future__ import annotations
+
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, patch
+
+import pytest
+
+from backend.app.core import websocket
+from backend.app.core.printer_scope import PrinterScope
+
+pytestmark = pytest.mark.asyncio
+
+
+def _socket(printer_ids):
+    scope = PrinterScope() if printer_ids is None else PrinterScope(frozenset(printer_ids))
+    return SimpleNamespace(state=SimpleNamespace(bambuddy_printer_scope=scope), send_text=AsyncMock())
+
+
+@pytest.mark.parametrize(
+    ("archive_printer", "expected"),
+    [
+        (1, {"team_a": True, "team_b": False, "everyone": True}),
+        (None, {"team_a": True, "team_b": True, "everyone": True}),  # printer-less: in every scope
+        (websocket._UNKNOWN_PRINTER, {"team_a": False, "team_b": False, "everyone": True}),
+    ],
+)
+async def test_an_update_follows_the_archive_printer(archive_printer, expected):
+    manager = websocket.ConnectionManager()
+    sockets = {"team_a": _socket({1}), "team_b": _socket({2}), "everyone": _socket(None)}
+    manager.active_connections = list(sockets.values())
+
+    with patch.object(websocket, "_archive_printer_id", new=AsyncMock(return_value=archive_printer)):
+        await manager.send_archive_updated({"id": 7, "cost": 1.5})
+
+    assert {name: s.send_text.await_count == 1 for name, s in sockets.items()} == expected
+
+
+async def test_an_archive_that_cannot_be_looked_up_reaches_only_unrestricted_sockets():
+    with patch("backend.app.core.database.async_session", side_effect=RuntimeError("db down")):
+        assert await websocket._archive_printer_id(7) == websocket._UNKNOWN_PRINTER
+    assert await websocket._archive_printer_id(None) == websocket._UNKNOWN_PRINTER