fix: allow get snapshot with liveonly permissions (#285)

* feat(camera): allow snapshots with `READ_LIVE` permissions

* feat(camera): enable `READ_LIVE` permission for `get_package_snapshot`

* test(data): add test data for `READ_LIVE` permissions

* test(camera): add unit tests for get_snapshot method

* chore(pre-commit.ci): auto fixes

---------

Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>

src/uiprotect/data/devices.py

@@ -2013,13 +2013,18 @@ async def get_snapshot(
 
         Datetime of screenshot is approximate. It may be +/- a few seconds.
         """
+        # Use READ_LIVE if dt is None, otherwise READ_MEDIA
+        permission = (
+            PermissionNode.READ_LIVE if dt is None else PermissionNode.READ_MEDIA
+        )
         if not self._api.bootstrap.auth_user.can(
             ModelType.CAMERA,
-            PermissionNode.READ_MEDIA,
+            permission,
             self,
         ):
+            action = "read live" if dt is None else "read media"
             raise NotAuthorized(
-                f"Do not have permission to read media for camera: {self.id}",
+                f"Do not have permission to {action} for camera: {self.id}"
             )
 
         if height is None and width is None and self.high_camera_channel is not None:
@@ -2041,13 +2046,18 @@ async def get_package_snapshot(
         if not self.feature_flags.has_package_camera:
             raise BadRequest("Device does not have package camera")
 
+        # Use READ_LIVE if dt is None, otherwise READ_MEDIA
+        permission = (
+            PermissionNode.READ_LIVE if dt is None else PermissionNode.READ_MEDIA
+        )
         if not self._api.bootstrap.auth_user.can(
             ModelType.CAMERA,
-            PermissionNode.READ_MEDIA,
+            permission,
             self,
         ):
+            action = "read live" if dt is None else "read media"
             raise NotAuthorized(
-                f"Do not have permission to read media for camera: {self.id}",
+                f"Do not have permission to {action} for camera: {self.id}"
             )
 
         if height is None and width is None and self.package_camera_channel is not None:

tests/data/test_camera.py

@@ -3,12 +3,13 @@
 from __future__ import annotations
 
 from datetime import datetime, timedelta, timezone
-from unittest.mock import Mock, patch
+from unittest.mock import AsyncMock, MagicMock, Mock, patch
 
 import pytest
 from pydantic.v1 import ValidationError
 
 from tests.conftest import TEST_CAMERA_EXISTS
+from uiprotect import ProtectApiClient
 from uiprotect.data import (
     Camera,
     ChimeType,
@@ -24,7 +25,7 @@
 from uiprotect.data.devices import CameraZone, Hotplug, HotplugExtender
 from uiprotect.data.types import DEFAULT, SmartDetectObjectType
 from uiprotect.data.websocket import WSAction, WSSubscriptionMessage
-from uiprotect.exceptions import BadRequest
+from uiprotect.exceptions import BadRequest, NotAuthorized
 from uiprotect.utils import to_js_time
 
 
@@ -1333,3 +1334,114 @@ async def test_camera_set_ptz_home(ptz_camera: Camera | None):
         require_auth=True,
         raise_exception=True,
     )
+
+
+@pytest.mark.skipif(not TEST_CAMERA_EXISTS, reason="Missing test data")
+@pytest.mark.asyncio()
+async def test_get_snapshot_read_live(camera_obj: Camera | None):
+    camera_obj.api.api_request.reset_mock()
+    camera_obj._api = MagicMock(spec=ProtectApiClient)
+
+    camera_obj._api.get_camera_snapshot = AsyncMock(return_value=b"snapshot_data")
+
+    snapshot = await camera_obj.get_snapshot()
+
+    assert snapshot == b"snapshot_data"
+    camera_obj._api.get_camera_snapshot.assert_called_once_with(
+        camera_obj.id, None, camera_obj.high_camera_channel.height, dt=None
+    )
+
+
+@pytest.mark.skipif(not TEST_CAMERA_EXISTS, reason="Missing test data")
+@pytest.mark.asyncio()
+async def test_get_snapshot_read_live_no_perm(camera_obj: Camera | None):
+    camera_obj._api = MagicMock(spec=ProtectApiClient)
+    camera_obj._api.get_camera_snapshot = AsyncMock(return_value=b"snapshot_data")
+
+    auth_user = camera_obj._api.bootstrap.auth_user
+
+    with patch.object(auth_user, "can", return_value=False):
+        camera_obj.api.api_request.reset_mock()
+
+        with pytest.raises(
+            NotAuthorized,
+            match=f"Do not have permission to read live for camera: {camera_obj.id}",
+        ):
+            await camera_obj.get_snapshot()
+
+
+@pytest.mark.skipif(not TEST_CAMERA_EXISTS, reason="Missing test data")
+@pytest.mark.asyncio
+async def test_get_snapshot_with_dt(camera_obj: Camera):
+    camera_obj.api.api_request.reset_mock()
+    camera_obj._api = MagicMock(spec=ProtectApiClient)
+    camera_obj._api.get_camera_snapshot = AsyncMock(return_value=b"snapshot_data")
+
+    now = datetime.now(tz=timezone.utc)
+
+    snapshot = await camera_obj.get_snapshot(dt=now)
+
+    assert snapshot == b"snapshot_data"
+    camera_obj._api.get_camera_snapshot.assert_called_once_with(
+        camera_obj.id, None, camera_obj.high_camera_channel.height, dt=now
+    )
+
+
+@pytest.mark.skipif(not TEST_CAMERA_EXISTS, reason="Missing test data")
+@pytest.mark.asyncio()
+async def test_get_package_snapshot_read_live(camera_obj: Camera | None):
+    camera_obj.api.api_request.reset_mock()
+    camera_obj.feature_flags.has_package_camera = True
+    camera_obj._api = MagicMock(spec=ProtectApiClient)
+
+    camera_obj._api.get_package_camera_snapshot = AsyncMock(
+        return_value=b"snapshot_data"
+    )
+
+    snapshot = await camera_obj.get_package_snapshot()
+
+    assert snapshot == b"snapshot_data"
+    camera_obj._api.get_package_camera_snapshot.assert_called_once_with(
+        camera_obj.id, None, None, dt=None
+    )
+
+
+@pytest.mark.skipif(not TEST_CAMERA_EXISTS, reason="Missing test data")
+@pytest.mark.asyncio()
+async def test_get_package_snapshot_read_live_no_perm(camera_obj: Camera | None):
+    camera_obj._api = MagicMock(spec=ProtectApiClient)
+    camera_obj.feature_flags.has_package_camera = True
+    camera_obj._api.get_package_camera_snapshot = AsyncMock(
+        return_value=b"snapshot_data"
+    )
+
+    auth_user = camera_obj._api.bootstrap.auth_user
+
+    with patch.object(auth_user, "can", return_value=False):
+        camera_obj.api.api_request.reset_mock()
+
+        with pytest.raises(
+            NotAuthorized,
+            match=f"Do not have permission to read live for camera: {camera_obj.id}",
+        ):
+            await camera_obj.get_package_snapshot()
+
+
+@pytest.mark.skipif(not TEST_CAMERA_EXISTS, reason="Missing test data")
+@pytest.mark.asyncio
+async def test_get_package_snapshot_with_dt(camera_obj: Camera):
+    camera_obj.api.api_request.reset_mock()
+    camera_obj.feature_flags.has_package_camera = True
+    camera_obj._api = MagicMock(spec=ProtectApiClient)
+    camera_obj._api.get_package_camera_snapshot = AsyncMock(
+        return_value=b"snapshot_data"
+    )
+
+    now = datetime.now(tz=timezone.utc)
+
+    snapshot = await camera_obj.get_package_snapshot(dt=now)
+
+    assert snapshot == b"snapshot_data"
+    camera_obj._api.get_package_camera_snapshot.assert_called_once_with(
+        camera_obj.id, None, None, dt=now
+    )

tests/sample_data/sample_bootstrap.json

@@ -5676,7 +5676,7 @@
                 "schedule:create,read,write,delete:*",
                 "legacyUFV:read,write,delete:*",
                 "bridge:create,read,write,delete:*",
-                "camera:create,read,write,delete,readmedia,deletemedia:*",
+                "camera:create,read,write,delete,readmedia,readlive,deletemedia:*",
                 "light:create,read,write,delete:*",
                 "sensor:create,read,write,delete:*",
                 "doorlock:create,read,write,delete:*",
@@ -5740,7 +5740,7 @@
                 "liveview:create",
                 "user:read,write,delete:$",
                 "bridge:read:*",
-                "camera:read,readmedia:*",
+                "camera:read,readmedia,readlive:*",
                 "doorlock:read:*",
                 "light:read:*",
                 "sensor:read:*",
@@ -5756,7 +5756,7 @@
                 "schedule:create,read,write,delete:*",
                 "legacyUFV:read,write,delete:*",
                 "bridge:create,read,write,delete:*",
-                "camera:create,read,write,delete,readmedia,deletemedia:*",
+                "camera:create,read,write,delete,readmedia,readlive,deletemedia:*",
                 "light:create,read,write,delete:*",
                 "sensor:create,read,write,delete:*",
                 "doorlock:create,read,write,delete:*",
@@ -5870,7 +5870,7 @@
                 "schedule:create,read,write,delete:*",
                 "legacyUFV:read,write,delete:*",
                 "bridge:create,read,write,delete:*",
-                "camera:create,read,write,delete,readmedia,deletemedia:*",
+                "camera:create,read,write,delete,readmedia,readlive,deletemedia:*",
                 "light:create,read,write,delete:*",
                 "sensor:create,read,write,delete:*",
                 "doorlock:create,read,write,delete:*",