test(auth): pin disabled keys, ended sessions on every gate, own-only key owners, long-lived token scope, model targets of limited keys; stricter route-auth backstop

backend/tests/integration/test_api_key_queue_acts_as_owner.py

@@ -311,6 +311,62 @@ class TestBatches:
         assert batch.name == "theirs"
 
 
+class TestAnOwnOnlyOwner:
+    """An owner who may change only their own queue items.
+
+    Their key is refused at the permission gate (it needs the ``*_all``
+    permission of its owner), so it changes no other user's item, and an
+    owner's deactivation stops it.
+    """
+
+    async def _items(self, db_session, world) -> tuple[PrintQueueItem, PrintQueueItem]:
+        await _set(db_session, "billing_enabled", "false")
+        mine = await _archive(db_session, world["owner"].id, 1)
+        theirs = await _archive(db_session, world["other"].id, 2)
+        own = PrintQueueItem(
+            printer_id=world["printer"].id, archive_id=mine.id, status="pending", created_by_id=world["owner"].id
+        )
+        other = PrintQueueItem(
+            printer_id=world["printer"].id, archive_id=theirs.id, status="pending", created_by_id=world["other"].id
+        )
+        db_session.add_all([own, other])
+        await db_session.commit()
+        return own, other
+
+    async def test_another_users_item_is_left_alone(self, async_client, db_session, world):
+        own, other = await self._items(db_session, world)
+        key = await _key(db_session, world["owner"].id)
+
+        patched = await async_client.patch(f"/api/v1/queue/{other.id}", json={"manual_start": True}, headers=key)
+        bulk = await async_client.patch(
+            "/api/v1/queue/bulk", json={"item_ids": [own.id, other.id], "manual_start": True}, headers=key
+        )
+        deleted = await async_client.delete(f"/api/v1/queue/{other.id}", headers=key)
+
+        assert patched.status_code in (403, 404)
+        assert bulk.status_code in (200, 403), bulk.text
+        assert deleted.status_code in (403, 404)
+        await db_session.refresh(other)
+        assert other.manual_start is False
+        assert await db_session.get(PrintQueueItem, other.id) is not None
+
+    async def test_a_deactivated_owners_key_changes_nothing(self, async_client, db_session, world):
+        """An admin's key, so the deactivation alone is what refuses it."""
+        own, _other = await self._items(db_session, world)
+        key = await _key(db_session, world["admin"].id)
+        world["admin"].is_active = False
+        await db_session.commit()
+
+        patched = await async_client.patch(f"/api/v1/queue/{own.id}", json={"manual_start": True}, headers=key)
+        deleted = await async_client.delete(f"/api/v1/queue/{own.id}", headers=key)
+
+        assert patched.status_code in (401, 403)
+        assert deleted.status_code in (401, 403)
+        await db_session.refresh(own)
+        assert own.manual_start is False
+        assert await db_session.get(PrintQueueItem, own.id) is not None
+
+
 @pytest.fixture
 async def library_world(db_session, world):
     """The same owner, who may also upload, read stats and pipelines, and a

backend/tests/integration/test_auth_apikey_rbac.py

@@ -141,6 +141,10 @@ class TestApiKeyDenylistIntegrity:
             Permission.GITHUB_BACKUP,
             Permission.GITHUB_RESTORE,
             Permission.FIRMWARE_UPDATE,
+            # Destructive: drop prints from the stats, or scan the network
+            Permission.LIBRARY_PURGE,
+            Permission.ARCHIVES_PURGE,
+            Permission.DISCOVERY_SCAN,
         }
         missing = expected_denied - _APIKEY_DENIED_PERMISSIONS
         assert not missing, (
@@ -452,13 +456,18 @@ class TestCheckApiKeyPermissionsMatrix:
 
         from backend.app.core.auth import _check_apikey_permissions
         from backend.app.core.permissions import Permission
+        from backend.app.models.api_key import APIKey
 
         perm = Permission[perm_name].value
-        all_flags = _FakeApiKey(can_read_status=True, can_queue=True, can_control_printer=True, can_manage_library=True)
+        # Every scope flag the model has, so no flag can be what refuses it
+        all_flags = _FakeApiKey()
+        for column in APIKey.__table__.columns:
+            if column.name.startswith("can_"):
+                setattr(all_flags, column.name, True)
         with pytest.raises(HTTPException) as exc:
             _check_apikey_permissions(all_flags, [perm])
         assert exc.value.status_code == 403
-        assert "administrative" in exc.value.detail.lower() or "does not have" in exc.value.detail.lower()
+        assert "administrative" in exc.value.detail.lower()
 
     def test_unknown_permission_string_is_admin_denied(self):
         """An unrecognised permission string must fail closed, not silently pass."""

backend/tests/integration/test_ended_session_gates.py

@@ -0,0 +1,179 @@
+"""An ended session's JWT is refused by every kind of route gate.
+
+Each gate resolves the bearer token itself: the permission gate, the
+ownership gate (``*_own`` / ``*_all``), the any-of gate and the admin gate. A
+session ends at logout, at a password change (own or by an admin) and when
+the user is deactivated; before that the same token passes every gate.
+
+The API gateway (``auth_middleware``) refuses such a token before any route
+runs, and most routes stack more than one gate, so a broken gate would not show
+on a real route. Each gate is therefore also tried alone, on a route of its
+own that the gateway lets through.
+
+The session under test began 30 s ago: a token issued in the same second as a
+password change passes on purpose (JWT ``iat`` is whole seconds).
+"""
+
+from datetime import timedelta
+
+import jwt
+import pytest
+from httpx import AsyncClient
+from sqlalchemy import update
+
+from backend.app.core import auth
+from backend.app.models.user import User
+from backend.app.utils.local_time import utcnow_naive
+
+pytestmark = [pytest.mark.asyncio, pytest.mark.integration]
+
+PASSWORD = "Gatepass1!"
+
+
+def _auth(token: str) -> dict[str, str]:
+    return {"Authorization": f"Bearer {token}"}
+
+
+async def _login(client: AsyncClient, username: str, password: str) -> str:
+    response = await client.post("/api/v1/auth/login", json={"username": username, "password": password})
+    assert response.status_code == 200, response.text
+    return response.json()["access_token"]
+
+
+def _session_from_30_s_ago(username: str) -> str:
+    issued = utcnow_naive() - timedelta(seconds=30)
+    return jwt.encode(
+        {"sub": username, "iat": issued, "exp": issued + timedelta(days=1), "jti": f"gates-{username}"},
+        auth.SECRET_KEY,
+        algorithm=auth.ALGORITHM,
+    )
+
+
+@pytest.fixture
+async def users(async_client: AsyncClient, db_session) -> dict:
+    """The first admin, and a second admin whose session the tests end."""
+    await async_client.post(
+        "/api/v1/auth/setup",
+        json={"auth_enabled": True, "admin_username": "gateadmin", "admin_password": PASSWORD},
+    )
+    admin = await _login(async_client, "gateadmin", PASSWORD)
+    created = await async_client.post(
+        "/api/v1/users/",
+        headers=_auth(admin),
+        json={"username": "gateuser", "password": PASSWORD, "role": "admin"},
+    )
+    assert created.status_code == 201, created.text
+    await db_session.execute(
+        update(User)
+        .where(User.username == "gateuser")
+        .values(password_changed_at=utcnow_naive() - timedelta(minutes=1))
+    )
+    await db_session.commit()
+    return {"admin": admin, "user_id": created.json()["id"], "jwt": _session_from_30_s_ago("gateuser")}
+
+
+async def _gates(client: AsyncClient, token: str, name: str) -> dict[str, int]:
+    headers = _auth(token)
+    return {
+        "permission": (await client.get("/api/v1/printers/", headers=headers)).status_code,
+        "ownership": (await client.get("/api/v1/queue/", headers=headers)).status_code,
+        "any-of": (await client.get("/api/v1/inventory/spools/by-tag?tag_uid=00", headers=headers)).status_code,
+        "admin": (
+            await client.post("/api/v1/groups/", headers=headers, json={"name": name, "permissions": []})
+        ).status_code,
+    }
+
+
+async def _logout(client: AsyncClient, users: dict) -> None:
+    assert (await client.post("/api/v1/auth/logout", headers=_auth(users["jwt"]))).status_code == 200
+
+
+async def _own_password_change(client: AsyncClient, users: dict) -> None:
+    response = await client.post(
+        "/api/v1/users/me/change-password",
+        headers=_auth(users["jwt"]),
+        json={"current_password": PASSWORD, "new_password": "Changed1!pass"},
+    )
+    assert response.status_code == 200, response.text
+
+
+async def _admin_password_change(client: AsyncClient, users: dict) -> None:
+    response = await client.patch(
+        f"/api/v1/users/{users['user_id']}", headers=_auth(users["admin"]), json={"password": "Changed1!pass"}
+    )
+    assert response.status_code == 200, response.text
+
+
+async def _deactivation(client: AsyncClient, users: dict) -> None:
+    response = await client.patch(
+        f"/api/v1/users/{users['user_id']}", headers=_auth(users["admin"]), json={"is_active": False}
+    )
+    assert response.status_code == 200, response.text
+
+
+@pytest.mark.parametrize("end_session", [_logout, _own_password_change, _admin_password_change, _deactivation])
+async def test_every_gate_refuses_an_ended_session(async_client: AsyncClient, users, end_session):
+    before = await _gates(async_client, users["jwt"], "before")
+    assert all(code != 401 for code in before.values()), before
+
+    await end_session(async_client, users)
+    after = await _gates(async_client, users["jwt"], "after")
+
+    assert after == dict.fromkeys(after, 401)
+
+
+def _lone_gates():
+    from fastapi import Depends
+
+    from backend.app.core import auth as a
+    from backend.app.core.permissions import Permission
+
+    return {
+        "gateway": None,
+        "permission": a.RequirePermissionIfAuthEnabled(Permission.PRINTERS_READ),
+        "any-of": a.RequireAnyPermissionIfAuthEnabled(Permission.INVENTORY_READ, Permission.INVENTORY_UPDATE),
+        "ownership": Depends(a.require_ownership_permission(Permission.QUEUE_READ_ALL, Permission.QUEUE_READ_OWN)),
+        "admin": a.RequireAdminIfAuthEnabled(),
+        "user": Depends(a.get_current_user),
+        "optional user": Depends(a.get_current_user_optional),
+        "auth if enabled": Depends(a.require_auth_if_enabled),
+        "key scope": Depends(
+            a.require_api_key_scope("can_update_energy_cost", "update_energy_cost", Permission.SETTINGS_UPDATE)
+        ),
+    }
+
+
+LONE = "/api/v1/lone-gate-test/"
+
+
+@pytest.fixture
+def lone_gate_routes(monkeypatch):
+    """One route per gate. All but "gateway" are let through by the gateway."""
+    from backend.app import main
+
+    routes = []
+    for name, gate in _lone_gates().items():
+        path = LONE + name.replace(" ", "-")
+        main.app.add_api_route(path, lambda: {"ok": True}, dependencies=[gate] if gate else [])
+        # Ahead of the SPA's catch-all route
+        routes.append(main.app.router.routes.pop())
+        main.app.router.routes.insert(0, routes[-1])
+    monkeypatch.setattr(main, "PUBLIC_API_ROUTES", {*main.PUBLIC_API_ROUTES, *(r.path for r in routes[1:])})
+    yield {name: route.path for name, route in zip(_lone_gates(), routes, strict=True)}
+    for route in routes:
+        main.app.router.routes.remove(route)
+
+
+@pytest.mark.parametrize("end_session", [_logout, _own_password_change, _admin_password_change, _deactivation])
+@pytest.mark.parametrize("gate", list(_lone_gates()))
+async def test_each_gate_alone_refuses_an_ended_session(
+    async_client: AsyncClient, users, lone_gate_routes, end_session, gate
+):
+    path = lone_gate_routes[gate]
+    headers = _auth(users["jwt"])
+    assert (await async_client.get(path, headers=headers)).status_code == 200
+
+    await end_session(async_client, users)
+
+    # get_current_user answers a deactivated account with 403 "disabled"
+    assert (await async_client.get(path, headers=headers)).status_code in (401, 403)

backend/tests/integration/test_printer_scope_1727.py

@@ -296,6 +296,40 @@ class TestTokens:
         assert wall.status_code == 200, wall.text
         assert [p["id"] for p in wall.json()] == [a.id]
 
+    async def _long_lived(self, async_client, jwt: str, scope: str) -> str:
+        created = await async_client.post(
+            "/api/v1/auth/tokens", headers=_auth(jwt), json={"name": scope, "expires_in_days": 30, "scope": scope}
+        )
+        assert created.status_code in (200, 201), created.text
+        return created.json()["token"]
+
+    async def test_long_lived_overlay_token_keeps_to_its_owners_printers(
+        self, async_client, printer_factory, mock_printer_manager
+    ):
+        a, b = await printer_factory(name="A"), await printer_factory(name="B")
+        admin = await _admin_token(async_client)
+        jwt, _ = await _team_member(async_client, admin, "member", [a.id])
+        token = await self._long_lived(async_client, jwt, "overlay")
+
+        hidden = await async_client.get(f"/api/v1/printers/{b.id}/overlay-status?token={token}")
+        own = await async_client.get(f"/api/v1/printers/{a.id}/overlay-status?token={token}")
+
+        assert hidden.status_code == 404
+        assert own.status_code == 200, own.text
+
+    async def test_long_lived_camera_token_keeps_to_its_owners_printers(self, async_client, printer_factory):
+        from backend.app.core.auth import verify_camera_stream_token
+
+        a, b = await printer_factory(name="A"), await printer_factory(name="B")
+        admin = await _admin_token(async_client)
+        jwt, _ = await _team_member(async_client, admin, "member", [a.id])
+        token = await self._long_lived(async_client, jwt, "camera_stream")
+
+        scope = await verify_camera_stream_token(token)
+        assert scope is not None and scope.printer_ids == frozenset({a.id})
+        snapshot = await async_client.get(f"/api/v1/printers/{b.id}/camera/snapshot?token={token}")
+        assert snapshot.status_code == 404
+
     async def test_websocket_token_carries_its_minters_scope(self, async_client, printer_factory):
         from backend.app.core.auth import principal_printer_scope, verify_websocket_token_principal
         from backend.app.core.database import async_session
@@ -427,6 +461,120 @@ class TestQueueAndHistory:
         assert on_b.id not in ids
 
 
+class TestModelTargetsOfALimitedKey:
+    """A limited key may not reach "any printer of a model" by any route."""
+
+    async def _key(self, async_client, admin: str, printer_id: int) -> dict[str, str]:
+        created = await async_client.post(
+            "/api/v1/api-keys/",
+            headers=_auth(admin),
+            json={"name": "k", "can_queue": True, "can_manage_library": True, "printer_ids": [printer_id]},
+        )
+        assert created.status_code == 200, created.text
+        return {"X-API-Key": created.json()["key"]}
+
+    async def test_editing_a_job_onto_a_model(self, async_client, printer_factory, archive_factory, db_session):
+        from backend.app.models.print_queue import PrintQueueItem
+
+        a = await printer_factory(name="A", model="X1C")
+        await printer_factory(name="B", model="X1C")
+        archive = await archive_factory(a.id)
+        admin = await _admin_token(async_client)
+        key = await self._key(async_client, admin, a.id)
+        item = await async_client.post(
+            "/api/v1/queue/", headers=key, json={"archive_id": archive.id, "printer_id": a.id}
+        )
+        assert item.status_code == 200, item.text
+
+        moved = await async_client.patch(
+            f"/api/v1/queue/{item.json()['id']}", headers=key, json={"printer_id": None, "target_model": "X1C"}
+        )
+
+        assert moved.status_code == 400, moved.text
+        assert "specific printer" in moved.json()["detail"]
+        row = await db_session.get(PrintQueueItem, item.json()["id"])
+        await db_session.refresh(row)
+        assert (row.printer_id, row.target_model) == (a.id, None)
+
+    async def test_dispatching_a_batch_of_model_jobs(self, async_client, printer_factory, archive_factory):
+        a = await printer_factory(name="A", model="X1C")
+        await printer_factory(name="B", model="X1C")
+        archive = await archive_factory(a.id)
+        admin = await _admin_token(async_client)
+        batch = await async_client.post(
+            "/api/v1/queue/batches",
+            headers=_auth(admin),
+            json={"name": "o", "archive_id": archive.id, "plates": [{"plate_id": None, "quantity_target": 3}]},
+        )
+        assert batch.status_code in (200, 201), batch.text
+        batch_id = batch.json()["id"]
+        first = await async_client.post(
+            "/api/v1/queue/",
+            headers=_auth(admin),
+            json={"archive_id": archive.id, "target_model": "X1C", "batch_id": batch_id},
+        )
+        assert first.status_code == 200, first.text
+        key = await self._key(async_client, admin, a.id)
+
+        dispatched = await async_client.post(f"/api/v1/queue/batches/{batch_id}/dispatch", headers=key, json={})
+
+        assert dispatched.status_code == 400, dispatched.text
+        assert "specific printer" in dispatched.json()["detail"]
+        listed = await async_client.get("/api/v1/queue/", headers=_auth(admin))
+        assert len(listed.json()) == 1
+
+    async def test_running_a_pipeline_aimed_at_a_model(self, async_client, printer_factory, db_session, request):
+        import uuid
+        from pathlib import Path
+
+        from backend.app.core.config import settings as app_settings
+        from backend.app.models.library import LibraryFile
+
+        a = await printer_factory(name="A", model="X1C")
+        admin = await _admin_token(async_client)
+        pipeline = await async_client.post(
+            "/api/v1/slicer-pipelines/",
+            headers=_auth(admin),
+            json={
+                "name": "p",
+                "printer_preset": {"source": "local", "id": "1"},
+                "process_preset": {"source": "local", "id": "2"},
+                "filament_presets": [{"source": "local", "id": "3"}],
+            },
+        )
+        assert pipeline.status_code == 201, pipeline.text
+        aimed = await async_client.put(
+            f"/api/v1/slicer-pipelines/{pipeline.json()['id']}",
+            headers=_auth(admin),
+            json={"target_kind": "printer_class", "target_model_class": "X1C"},
+        )
+        assert aimed.status_code == 200, aimed.text
+        name = f"scope_pipeline_{uuid.uuid4().hex}.3mf"
+        source_path = Path(app_settings.base_dir) / name
+        source_path.write_bytes(b"")
+        request.addfinalizer(lambda: source_path.unlink(missing_ok=True))
+        source = LibraryFile(
+            filename="cube.3mf",
+            file_path=name,
+            file_type="3mf",
+            file_size=0,
+            file_hash="scope-pipeline",
+            source_type="uploaded",
+        )
+        db_session.add(source)
+        await db_session.commit()
+        key = await self._key(async_client, admin, a.id)
+
+        run = await async_client.post(
+            f"/api/v1/slicer-pipelines/{pipeline.json()['id']}/run",
+            headers=key,
+            json={"source_library_file_id": source.id},
+        )
+
+        assert run.status_code == 400, run.text
+        assert "specific printer" in run.json()["detail"]
+
+
 class TestScheduler:
     async def test_model_matching_stays_within_the_scope(self, db_session, printer_factory):
         from backend.app.core.printer_scope import PrinterScope

backend/tests/integration/test_webhook_api_key_check.py

@@ -130,3 +130,20 @@ async def test_a_wrong_key_and_no_key_are_401s(async_client: AsyncClient, key):
     assert wrong.status_code == 401
     assert wrong.json()["detail"] == "Invalid API key"
     assert missing.status_code == 401
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+@pytest.mark.parametrize("route", [ROUTE, "/api/v1/printers/"])
+@pytest.mark.parametrize("header", ["X-API-Key", "Authorization"])
+async def test_a_disabled_key_is_a_401(async_client: AsyncClient, key, db_session, route, header):
+    from backend.app.models.settings import Settings
+
+    db_session.add(Settings(key="auth_enabled", value="true"))
+    await db_session.execute(update(APIKey).values(enabled=False).where(APIKey.name == "webhook-key"))
+    await db_session.commit()
+    value = key if header == "X-API-Key" else f"Bearer {key}"
+
+    response = await async_client.get(route, headers={header: value})
+
+    assert response.status_code == 401, response.text

backend/tests/unit/test_route_auth_coverage.py

@@ -36,8 +36,6 @@ from backend.app.main import app
 _AUTH_QUALNAME_PATTERNS: tuple[str, ...] = (
     "require_",  # require_permission_if_auth_enabled, require_admin_*, require_auth_*, require_any_*, require_ownership_*, require_camera_stream_token_*, require_energy_cost_update
     "cloud_caller",  # cloud.py route-level dep
-    "_cloud_api_key_gate",  # cloud.py router-level dep
-    "resolve_api_key_cloud_owner",  # used by slicer routes that need the API key's owner
     "get_current_user",  # JWT identity resolution
     "get_current_active_user",  # JWT identity resolution
     "get_api_key",  # webhook routes use this directly
@@ -80,6 +78,14 @@ _PUBLIC_ROUTES: frozenset[tuple[str, str]] = frozenset(
         # Forgot-password (anonymous request) + confirm (signed token in the URL).
         ("POST", "/api/v1/auth/forgot-password"),
         ("POST", "/api/v1/auth/forgot-password/confirm"),
+        # Own password change — resolves the caller with get_current_user_optional
+        # and answers 401 inline when there is none; the current password in the
+        # body is checked too.
+        ("POST", "/api/v1/users/me/change-password"),
+        # Who am I — checks the JWT or API key inline (revocation, freshness,
+        # active user) and answers 401 without one; the auth middleware also
+        # guards it.
+        ("GET", "/api/v1/auth/me"),
         # ---- HTTP API: signed-URL routes (token in path is the auth) ----
         # Signed download URLs — token in path validated by the handler.
         ("GET", "/api/v1/archives/{archive_id}/dl/{token}/{filename}"),
@@ -158,9 +164,33 @@ def _walk_dependant_qualnames(dependant) -> list[str]:
     return names
 
 
+def _is_auth_qualname(qualname: str) -> bool:
+    """An auth-bearing callable. An ``*_optional`` resolver is not one: it
+    lets an anonymous caller through (``get_current_user_optional`` answers
+    None), so a route that relies on it checks inline and is listed."""
+    if "_optional" in qualname:
+        return False
+    return any(p in qualname for p in _AUTH_QUALNAME_PATTERNS)
+
+
 def _has_auth_dep(dependant) -> bool:
-    """True if any callable in the dependant tree matches an auth pattern."""
-    return any(any(p in qn for p in _AUTH_QUALNAME_PATTERNS) for qn in _walk_dependant_qualnames(dependant))
+    """True if any dependency in the tree matches an auth pattern. The route's
+    own handler is not one, whatever it is called."""
+    return any(_is_auth_qualname(qn) for sub in dependant.dependencies for qn in _walk_dependant_qualnames(sub))
+
+
+@pytest.mark.unit
+@pytest.mark.parametrize(
+    ("qualname", "is_auth"),
+    [
+        ("require_permission_if_auth_enabled.<locals>.permission_checker", True),
+        ("get_current_user", True),
+        ("get_current_user_optional", False),
+        ("some_helper", False),
+    ],
+)
+def test_the_matcher_does_not_count_an_optional_resolver(qualname, is_auth) -> None:
+    assert _is_auth_qualname(qualname) is is_auth
 
 
 def _ws_endpoint_does_inline_token_check(route: APIWebSocketRoute) -> bool: