Browse Source

Make the {finish_photo_url} link open with authentication on

    With authentication on, {finish_photo_url} pointed at the archive photo
    route, which needs a media token. A link tapped in Telegram, CallMeBot
    or a Home Assistant notification has none, so it only ever answered 401.

    _finish_photo_for_notification() now builds the link and the attachment
    bytes. With authentication off the link is the archive URL, unchanged.
    With it on, the photo is also saved through the notification photo
    store from #3199, and the link points there. That is an unguessable
    name that opens this one photo, for 3 days. If the auth check fails,
    it is treated as on. With auth on and the photo missing, no link is
    set rather than one that 401s. Photos over 2.5 MB are still linked but
    not attached, as before.
maziggy 1 day ago
parent
commit
888f6c4325
2 changed files with 152 additions and 31 deletions
  1. 66 31
      backend/app/main.py
  2. 86 0
      backend/tests/unit/test_finish_photo_link_auth.py

+ 66 - 31
backend/app/main.py

@@ -3135,6 +3135,65 @@ async def on_ams_change(printer_id: int, ams_data: list):
         logging.getLogger(__name__).error("Spoolman AMS sync failed for printer %s: %s", printer_id, e)
 
 
+# Largest finish photo attached to a notification; a bigger one is still linked.
+_FINISH_PHOTO_ATTACH_MAX_BYTES = 2_500_000
+
+
+async def _finish_photo_for_notification(
+    db, archive, archive_id: int, filename: str
+) -> tuple[str | None, bytes | None]:
+    """The ``{finish_photo_url}`` link and the bytes to attach, for a print's finish photo.
+
+    With authentication off the link is the archive's own photo route, as it
+    always was: it opens without a login and doesn't expire. With
+    authentication on that route needs a media token, which nothing tapping a
+    link in Telegram, CallMeBot or a Home Assistant notification has, so the
+    link only ever answered 401. The photo is then saved as a notification
+    photo as well (utils/notification_photos.py) and the link points there: an
+    unguessable name that opens this one photo and nothing else, for 3 days.
+
+    The link is relative when no External URL is set. Bytes over
+    ``_FINISH_PHOTO_ATTACH_MAX_BYTES`` are linked but not attached. Returns
+    ``(None, None)`` when the photo can't be found.
+    """
+    from backend.app.api.routes.settings import get_setting
+    from backend.app.core.auth import is_auth_enabled
+    from backend.app.utils.archive_paths import find_archive_photo
+    from backend.app.utils.notification_photos import save_notification_photo
+
+    log = logging.getLogger(__name__)
+    base = ((await get_setting(db, "external_url")) or "").strip().rstrip("/")
+    url: str | None = f"{base}/api/v1/archives/{archive_id}/photos/{filename}"
+
+    photo_bytes: bytes | None = None
+    try:
+        photo_path = find_archive_photo(archive, filename)
+        if photo_path is not None:
+            photo_bytes = await asyncio.to_thread(photo_path.read_bytes)
+    except Exception as e:
+        log.warning("[NOTIFY-BG] Failed to read finish photo bytes: %s", e)
+
+    try:
+        auth_on = await is_auth_enabled(db)
+    except Exception:
+        auth_on = True  # Unknown: the archive link may need a login, so don't rely on it.
+    if auth_on:
+        url = None
+        if photo_bytes:
+            try:
+                name = await asyncio.to_thread(save_notification_photo, photo_bytes, "print_complete")
+                url = f"{base}/api/v1/notifications/photos/{name}"
+            except Exception as e:
+                log.warning("[NOTIFY-BG] Failed to save finish photo for its link: %s", e)
+
+    if photo_bytes is not None and len(photo_bytes) > _FINISH_PHOTO_ATTACH_MAX_BYTES:
+        log.warning("[NOTIFY-BG] Finish photo too large for attachment: %s bytes", len(photo_bytes))
+        return url, None
+    if photo_bytes:
+        log.info("[NOTIFY-BG] Loaded finish photo bytes: %s bytes", len(photo_bytes))
+    return url, photo_bytes
+
+
 async def _capture_snapshot_for_notification(printer_id: int, printer, logger) -> bytes | None:
     """Capture a camera snapshot for notification image attachment.
 
@@ -8206,37 +8265,13 @@ async def on_print_complete(printer_id: int, data: dict):
                             archive_data["usage_results"] = usage_results
                         # Add finish photo URL and image bytes if available
                         if finish_photo_filename:
-                            from backend.app.api.routes.settings import get_setting
-
-                            external_url = await get_setting(db, "external_url")
-                            if external_url:
-                                external_url = external_url.rstrip("/")
-                                archive_data["finish_photo_url"] = (
-                                    f"{external_url}/api/v1/archives/{archive_id}/photos/{finish_photo_filename}"
-                                )
-                            else:
-                                # Fallback to relative URL (won't work for external services)
-                                archive_data["finish_photo_url"] = (
-                                    f"/api/v1/archives/{archive_id}/photos/{finish_photo_filename}"
-                                )
-
-                            # Read finish photo bytes for image attachment (e.g. Pushover)
-                            try:
-                                from backend.app.utils.archive_paths import find_archive_photo
-
-                                photo_path = find_archive_photo(archive, finish_photo_filename)
-                                if photo_path is not None:
-                                    photo_bytes = await asyncio.to_thread(photo_path.read_bytes)
-                                    if len(photo_bytes) <= 2_500_000:
-                                        archive_data["image_data"] = photo_bytes
-                                        logger.info("[NOTIFY-BG] Loaded finish photo bytes: %s bytes", len(photo_bytes))
-                                    else:
-                                        logger.warning(
-                                            f"[NOTIFY-BG] Finish photo too large for attachment: "
-                                            f"{len(photo_bytes)} bytes"
-                                        )
-                            except Exception as e:
-                                logger.warning("[NOTIFY-BG] Failed to read finish photo bytes: %s", e)
+                            photo_url, photo_bytes = await _finish_photo_for_notification(
+                                db, archive, archive_id, finish_photo_filename
+                            )
+                            if photo_url:
+                                archive_data["finish_photo_url"] = photo_url
+                            if photo_bytes:
+                                archive_data["image_data"] = photo_bytes
 
                 if not await _kill_switch_notification_already_sent(kill_switch_notification_task):
                     await notification_service.on_print_complete(

+ 86 - 0
backend/tests/unit/test_finish_photo_link_auth.py

@@ -0,0 +1,86 @@
+"""The {finish_photo_url} link opens with authentication on too.
+
+The archive photo route needs a media token when authentication is on, and
+nothing tapping a link in Telegram, CallMeBot or a Home Assistant notification
+has one, so the link answered 401. With authentication on, the link now points
+at a notification photo instead: an unguessable name that opens that one photo
+for 3 days. With authentication off it stays the archive link, which needs no
+login and doesn't expire.
+"""
+
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, patch
+
+import pytest
+
+from backend.app.utils import notification_photos
+
+ARCHIVE_ID = 42
+FILENAME = "finish_20260101_120000_abcd1234.jpg"
+
+
+@pytest.fixture
+def photo(tmp_path, monkeypatch):
+    """A finish photo on disk, and the notification photo store under tmp_path."""
+    monkeypatch.setattr(notification_photos.settings, "base_dir", tmp_path)
+    path = tmp_path / FILENAME
+    path.write_bytes(b"\xff\xd8finish-photo")
+    return path
+
+
+async def _call(*, auth_on, external_url="https://bambuddy.example", photo_path=None, auth_error=False):
+    from backend.app.main import _finish_photo_for_notification
+
+    auth = AsyncMock(side_effect=RuntimeError("db down")) if auth_error else AsyncMock(return_value=auth_on)
+    with (
+        patch("backend.app.api.routes.settings.get_setting", AsyncMock(return_value=external_url)),
+        patch("backend.app.core.auth.is_auth_enabled", auth),
+        patch("backend.app.utils.archive_paths.find_archive_photo", return_value=photo_path),
+    ):
+        return await _finish_photo_for_notification(AsyncMock(), SimpleNamespace(), ARCHIVE_ID, FILENAME)
+
+
+class TestFinishPhotoLink:
+    @pytest.mark.asyncio
+    async def test_auth_off_keeps_the_archive_link(self, photo):
+        url, data = await _call(auth_on=False, photo_path=photo)
+
+        assert url == f"https://bambuddy.example/api/v1/archives/{ARCHIVE_ID}/photos/{FILENAME}"
+        assert data == photo.read_bytes()
+
+    @pytest.mark.asyncio
+    async def test_auth_off_without_external_url_is_relative(self, photo):
+        url, _ = await _call(auth_on=False, external_url=None, photo_path=photo)
+
+        assert url == f"/api/v1/archives/{ARCHIVE_ID}/photos/{FILENAME}"
+
+    @pytest.mark.asyncio
+    async def test_auth_on_links_a_notification_photo_that_serves_the_same_bytes(self, photo):
+        url, data = await _call(auth_on=True, photo_path=photo)
+
+        prefix = "https://bambuddy.example/api/v1/notifications/photos/"
+        assert url.startswith(prefix)
+        served = notification_photos.find_notification_photo(url[len(prefix) :])
+        assert served is not None
+        assert served.read_bytes() == photo.read_bytes()
+        assert data == photo.read_bytes()
+
+    @pytest.mark.asyncio
+    async def test_auth_check_failing_is_treated_as_auth_on(self, photo):
+        url, _ = await _call(auth_on=False, auth_error=True, photo_path=photo)
+
+        assert "/api/v1/notifications/photos/" in url
+
+    @pytest.mark.asyncio
+    async def test_auth_on_without_the_photo_gives_no_link(self, photo):
+        # A link to the archive route would only answer 401, so none at all.
+        assert await _call(auth_on=True, photo_path=None) == (None, None)
+
+    @pytest.mark.asyncio
+    async def test_oversized_photo_is_linked_but_not_attached(self, photo):
+        photo.write_bytes(b"\xff" * 2_500_001)
+
+        url, data = await _call(auth_on=True, photo_path=photo)
+
+        assert url is not None
+        assert data is None