diff --git a/homeassistant/components/unifiprotect/__init__.py b/homeassistant/components/unifiprotect/__init__.py index 8d6dd30a2b3..9f7876091ee 100644 --- a/homeassistant/components/unifiprotect/__init__.py +++ b/homeassistant/components/unifiprotect/__init__.py @@ -34,7 +34,6 @@ from .const import ( ) from .data import ProtectData from .services import async_cleanup_services, async_setup_services -from .views import ThumbnailProxyView _LOGGER = logging.getLogger(__name__) @@ -84,7 +83,6 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: hass.data.setdefault(DOMAIN, {})[entry.entry_id] = data_service hass.config_entries.async_setup_platforms(entry, PLATFORMS) async_setup_services(hass) - hass.http.register_view(ThumbnailProxyView(hass)) entry.async_on_unload(entry.add_update_listener(_async_options_updated)) entry.async_on_unload( diff --git a/homeassistant/components/unifiprotect/const.py b/homeassistant/components/unifiprotect/const.py index f424c99cff0..230f728e145 100644 --- a/homeassistant/components/unifiprotect/const.py +++ b/homeassistant/components/unifiprotect/const.py @@ -7,7 +7,6 @@ from homeassistant.const import Platform DOMAIN = "unifiprotect" ATTR_EVENT_SCORE = "event_score" -ATTR_EVENT_THUMB = "event_thumbnail" ATTR_WIDTH = "width" ATTR_HEIGHT = "height" ATTR_FPS = "fps" diff --git a/homeassistant/components/unifiprotect/data.py b/homeassistant/components/unifiprotect/data.py index f7f05b5bd4c..cb58e21fb54 100644 --- a/homeassistant/components/unifiprotect/data.py +++ b/homeassistant/components/unifiprotect/data.py @@ -1,7 +1,6 @@ """Base class for protect data.""" from __future__ import annotations -from collections import deque from collections.abc import Generator, Iterable from datetime import timedelta import logging @@ -43,7 +42,6 @@ class ProtectData: self._unsub_websocket: CALLBACK_TYPE | None = None self.last_update_success = False - self.access_tokens: dict[str, deque] = {} self.api = protect @property @@ -177,10 +175,3 @@ class ProtectData: _LOGGER.debug("Updating device: %s", device_id) for update_callback in self._subscriptions[device_id]: update_callback() - - @callback - def async_get_or_create_access_tokens(self, entity_id: str) -> deque: - """Wrap access_tokens to automatically create underlying data structure if missing.""" - if entity_id not in self.access_tokens: - self.access_tokens[entity_id] = deque([], 2) - return self.access_tokens[entity_id] diff --git a/homeassistant/components/unifiprotect/entity.py b/homeassistant/components/unifiprotect/entity.py index c7da55e19f0..03d4458c0b0 100644 --- a/homeassistant/components/unifiprotect/entity.py +++ b/homeassistant/components/unifiprotect/entity.py @@ -1,14 +1,9 @@ """Shared Entity definition for UniFi Protect Integration.""" from __future__ import annotations -from collections import deque from collections.abc import Sequence -from datetime import datetime, timedelta -import hashlib import logging -from random import SystemRandom -from typing import Any, Final -from urllib.parse import urlencode +from typing import Any from pyunifiprotect.data import ( Camera, @@ -26,21 +21,11 @@ from homeassistant.core import callback import homeassistant.helpers.device_registry as dr from homeassistant.helpers.entity import DeviceInfo, Entity, EntityDescription -from .const import ( - ATTR_EVENT_SCORE, - ATTR_EVENT_THUMB, - DEFAULT_ATTRIBUTION, - DEFAULT_BRAND, - DOMAIN, -) +from .const import ATTR_EVENT_SCORE, DEFAULT_ATTRIBUTION, DEFAULT_BRAND, DOMAIN from .data import ProtectData from .models import ProtectRequiredKeysMixin from .utils import get_nested_attr -from .views import ThumbnailProxyView -EVENT_UPDATE_TOKENS = "unifiprotect_update_tokens" -TOKEN_CHANGE_INTERVAL: Final = timedelta(minutes=1) -_RND: Final = SystemRandom() _LOGGER = logging.getLogger(__name__) @@ -219,50 +204,7 @@ class ProtectNVREntity(ProtectDeviceEntity): self._attr_available = self.data.last_update_success -class AccessTokenMixin(ProtectDeviceEntity): - """Adds access_token attribute and provides access tokens for use for anonymous views.""" - - @property - def access_tokens(self) -> deque[str]: - """Get valid access_tokens for current entity.""" - return self.data.async_get_or_create_access_tokens(self.entity_id) - - @callback - def _async_update_and_write_token(self, now: datetime) -> None: - _LOGGER.debug("Updating access tokens for %s", self.entity_id) - self.async_update_token() - self.async_write_ha_state() - - @callback - def async_update_token(self) -> None: - """Update the used token.""" - self.access_tokens.append( - hashlib.sha256(_RND.getrandbits(256).to_bytes(32, "little")).hexdigest() - ) - - @callback - def async_cleanup_tokens(self) -> None: - """Clean up any remaining tokens on removal.""" - if self.entity_id in self.data.access_tokens: - del self.data.access_tokens[self.entity_id] - - async def async_added_to_hass(self) -> None: - """Run when entity about to be added to hass. - - Injects callbacks to update access tokens automatically - """ - await super().async_added_to_hass() - - self.async_update_token() - self.async_on_remove( - self.hass.helpers.event.async_track_time_interval( - self._async_update_and_write_token, TOKEN_CHANGE_INTERVAL - ) - ) - self.async_on_remove(self.async_cleanup_tokens) - - -class EventThumbnailMixin(AccessTokenMixin): +class EventThumbnailMixin(ProtectDeviceEntity): """Adds motion event attributes to sensor.""" def __init__(self, *args: Any, **kwarg: Any) -> None: @@ -283,21 +225,12 @@ class EventThumbnailMixin(AccessTokenMixin): # Camera motion sensors with object detection attrs: dict[str, Any] = { ATTR_EVENT_SCORE: 0, - ATTR_EVENT_THUMB: None, } if self._event is None: return attrs attrs[ATTR_EVENT_SCORE] = self._event.score - if len(self.access_tokens) > 0: - params = urlencode( - {"entity_id": self.entity_id, "token": self.access_tokens[-1]} - ) - attrs[ATTR_EVENT_THUMB] = ( - ThumbnailProxyView.url.format(event_id=self._event.id) + f"?{params}" - ) - return attrs @callback diff --git a/homeassistant/components/unifiprotect/views.py b/homeassistant/components/unifiprotect/views.py deleted file mode 100644 index 4ec2f08ddca..00000000000 --- a/homeassistant/components/unifiprotect/views.py +++ /dev/null @@ -1,91 +0,0 @@ -"""UniFi Protect Integration views.""" -from __future__ import annotations - -import collections -from http import HTTPStatus -import logging -from typing import Any - -from aiohttp import web -from pyunifiprotect.api import ProtectApiClient -from pyunifiprotect.exceptions import NvrError - -from homeassistant.components.http import KEY_AUTHENTICATED, HomeAssistantView -from homeassistant.core import HomeAssistant - -from .const import DOMAIN -from .data import ProtectData - -_LOGGER = logging.getLogger(__name__) - - -def _404(message: Any) -> web.Response: - _LOGGER.error("Error on load thumbnail: %s", message) - return web.Response(status=HTTPStatus.NOT_FOUND) - - -class ThumbnailProxyView(HomeAssistantView): - """View to proxy event thumbnails from UniFi Protect.""" - - requires_auth = False - url = "/api/ufp/thumbnail/{event_id}" - name = "api:ufp_thumbnail" - - def __init__(self, hass: HomeAssistant) -> None: - """Initialize a thumbnail proxy view.""" - self.hass = hass - self.data = hass.data[DOMAIN] - - def _get_access_tokens( - self, entity_id: str - ) -> tuple[collections.deque, ProtectApiClient] | None: - - entries: list[ProtectData] = list(self.data.values()) - for entry in entries: - if entity_id in entry.access_tokens: - return entry.access_tokens[entity_id], entry.api - return None - - async def get(self, request: web.Request, event_id: str) -> web.Response: - """Start a get request.""" - - entity_id: str | None = request.query.get("entity_id") - width: int | str | None = request.query.get("w") - height: int | str | None = request.query.get("h") - token: str | None = request.query.get("token") - - if width is not None: - try: - width = int(width) - except ValueError: - return _404("Invalid width param") - if height is not None: - try: - height = int(height) - except ValueError: - return _404("Invalid height param") - - access_tokens: list[str] = [] - if entity_id is not None: - items = self._get_access_tokens(entity_id) - if items is None: - return _404(f"Could not find entity with entity_id {entity_id}") - - access_tokens = list(items[0]) - instance = items[1] - - authenticated = request[KEY_AUTHENTICATED] or token in access_tokens - if not authenticated: - raise web.HTTPUnauthorized() - - try: - thumbnail = await instance.get_event_thumbnail( - event_id, width=width, height=height - ) - except NvrError as err: - return _404(err) - - if thumbnail is None: - return _404("Event thumbnail not found") - - return web.Response(body=thumbnail, content_type="image/jpeg") diff --git a/tests/components/unifiprotect/test_binary_sensor.py b/tests/components/unifiprotect/test_binary_sensor.py index 6fc96d31359..89d974f9f5c 100644 --- a/tests/components/unifiprotect/test_binary_sensor.py +++ b/tests/components/unifiprotect/test_binary_sensor.py @@ -17,7 +17,6 @@ from homeassistant.components.unifiprotect.binary_sensor import ( ) from homeassistant.components.unifiprotect.const import ( ATTR_EVENT_SCORE, - ATTR_EVENT_THUMB, DEFAULT_ATTRIBUTION, ) from homeassistant.const import ( @@ -258,7 +257,6 @@ async def test_binary_sensor_setup_camera_all( assert state.state == STATE_OFF assert state.attributes[ATTR_ATTRIBUTION] == DEFAULT_ATTRIBUTION assert state.attributes[ATTR_EVENT_SCORE] == 0 - assert state.attributes[ATTR_EVENT_THUMB] is None async def test_binary_sensor_setup_camera_none( @@ -350,6 +348,3 @@ async def test_binary_sensor_update_motion( assert state.state == STATE_ON assert state.attributes[ATTR_ATTRIBUTION] == DEFAULT_ATTRIBUTION assert state.attributes[ATTR_EVENT_SCORE] == 100 - assert state.attributes[ATTR_EVENT_THUMB].startswith( - f"/api/ufp/thumbnail/test_event_id?entity_id={entity_id}&token=" - ) diff --git a/tests/components/unifiprotect/test_sensor.py b/tests/components/unifiprotect/test_sensor.py index e704fb8a6d6..38c883c15c6 100644 --- a/tests/components/unifiprotect/test_sensor.py +++ b/tests/components/unifiprotect/test_sensor.py @@ -13,7 +13,6 @@ from pyunifiprotect.data.types import EventType, SmartDetectObjectType from homeassistant.components.unifiprotect.const import ( ATTR_EVENT_SCORE, - ATTR_EVENT_THUMB, DEFAULT_ATTRIBUTION, ) from homeassistant.components.unifiprotect.sensor import ( @@ -407,7 +406,6 @@ async def test_sensor_setup_camera( assert state.state == DETECTED_OBJECT_NONE assert state.attributes[ATTR_ATTRIBUTION] == DEFAULT_ATTRIBUTION assert state.attributes[ATTR_EVENT_SCORE] == 0 - assert state.attributes[ATTR_EVENT_THUMB] is None async def test_sensor_update_motion( @@ -450,6 +448,3 @@ async def test_sensor_update_motion( assert state.state == SmartDetectObjectType.PERSON.value assert state.attributes[ATTR_ATTRIBUTION] == DEFAULT_ATTRIBUTION assert state.attributes[ATTR_EVENT_SCORE] == 100 - assert state.attributes[ATTR_EVENT_THUMB].startswith( - f"/api/ufp/thumbnail/test_event_id?entity_id={entity_id}&token=" - ) diff --git a/tests/components/unifiprotect/test_views.py b/tests/components/unifiprotect/test_views.py deleted file mode 100644 index f42923acd1e..00000000000 --- a/tests/components/unifiprotect/test_views.py +++ /dev/null @@ -1,236 +0,0 @@ -"""Test UniFi Protect views.""" -# pylint: disable=protected-access -from __future__ import annotations - -from datetime import datetime, timedelta -from unittest.mock import AsyncMock - -import pytest -from pyunifiprotect.data import Camera, Event, EventType -from pyunifiprotect.exceptions import NvrError - -from homeassistant.components.unifiprotect.binary_sensor import MOTION_SENSORS -from homeassistant.components.unifiprotect.const import ATTR_EVENT_THUMB -from homeassistant.components.unifiprotect.entity import TOKEN_CHANGE_INTERVAL -from homeassistant.const import STATE_ON, Platform -from homeassistant.core import HomeAssistant - -from .conftest import MockEntityFixture, ids_from_device_description, time_changed - - -@pytest.fixture(name="thumb_url") -async def thumb_url_fixture( - hass: HomeAssistant, - mock_entry: MockEntityFixture, - mock_camera: Camera, - now: datetime, -): - """Fixture for a single camera for testing the binary_sensor platform.""" - - # disable pydantic validation so mocking can happen - Camera.__config__.validate_assignment = False - - camera_obj = mock_camera.copy(deep=True) - camera_obj._api = mock_entry.api - camera_obj.channels[0]._api = mock_entry.api - camera_obj.channels[1]._api = mock_entry.api - camera_obj.channels[2]._api = mock_entry.api - camera_obj.name = "Test Camera" - camera_obj.is_motion_detected = True - - event = Event( - id="test_event_id", - type=EventType.MOTION, - start=now - timedelta(seconds=1), - end=None, - score=100, - smart_detect_types=[], - smart_detect_event_ids=[], - camera_id=camera_obj.id, - ) - camera_obj.last_motion_event_id = event.id - - mock_entry.api.bootstrap.reset_objects() - mock_entry.api.bootstrap.nvr.system_info.storage.devices = [] - mock_entry.api.bootstrap.cameras = { - camera_obj.id: camera_obj, - } - mock_entry.api.bootstrap.events = {event.id: event} - - await hass.config_entries.async_setup(mock_entry.entry.entry_id) - await hass.async_block_till_done() - - _, entity_id = ids_from_device_description( - Platform.BINARY_SENSOR, camera_obj, MOTION_SENSORS[0] - ) - - # make sure access tokens are generated - await time_changed(hass, 1) - - state = hass.states.get(entity_id) - assert state - assert state.state == STATE_ON - assert state.attributes[ATTR_EVENT_THUMB].startswith( - f"/api/ufp/thumbnail/test_event_id?entity_id={entity_id}&token=" - ) - - yield state.attributes[ATTR_EVENT_THUMB] - - Camera.__config__.validate_assignment = True - - -async def test_thumbnail_view_good( - thumb_url: str, - hass_client_no_auth, - mock_entry: MockEntityFixture, -): - """Test good result from thumbnail view.""" - - mock_entry.api.get_event_thumbnail = AsyncMock() - - client = await hass_client_no_auth() - - response = await client.get(thumb_url) - assert response.status == 200 - - mock_entry.api.get_event_thumbnail.assert_called_once_with( - "test_event_id", width=None, height=None - ) - - -async def test_thumbnail_view_good_args( - thumb_url: str, - hass_client_no_auth, - mock_entry: MockEntityFixture, -): - """Test good result from thumbnail view.""" - - mock_entry.api.get_event_thumbnail = AsyncMock() - - client = await hass_client_no_auth() - - response = await client.get(thumb_url + "&w=200&h=200") - assert response.status == 200 - - mock_entry.api.get_event_thumbnail.assert_called_once_with( - "test_event_id", width=200, height=200 - ) - - -async def test_thumbnail_view_bad_width( - thumb_url: str, - hass_client_no_auth, - mock_entry: MockEntityFixture, -): - """Test good result from thumbnail view.""" - - mock_entry.api.get_event_thumbnail = AsyncMock() - - client = await hass_client_no_auth() - - response = await client.get(thumb_url + "&w=safds&h=200") - assert response.status == 404 - - assert not mock_entry.api.get_event_thumbnail.called - - -async def test_thumbnail_view_bad_height( - thumb_url: str, - hass_client_no_auth, - mock_entry: MockEntityFixture, -): - """Test good result from thumbnail view.""" - - mock_entry.api.get_event_thumbnail = AsyncMock() - - client = await hass_client_no_auth() - - response = await client.get(thumb_url + "&w=200&h=asda") - assert response.status == 404 - - assert not mock_entry.api.get_event_thumbnail.called - - -async def test_thumbnail_view_bad_entity_id( - thumb_url: str, - hass_client_no_auth, - mock_entry: MockEntityFixture, -): - """Test good result from thumbnail view.""" - - mock_entry.api.get_event_thumbnail = AsyncMock() - - client = await hass_client_no_auth() - - response = await client.get("/api/ufp/thumbnail/test_event_id?entity_id=sdfsfd") - assert response.status == 404 - - assert not mock_entry.api.get_event_thumbnail.called - - -async def test_thumbnail_view_bad_access_token( - thumb_url: str, - hass_client_no_auth, - mock_entry: MockEntityFixture, -): - """Test good result from thumbnail view.""" - - mock_entry.api.get_event_thumbnail = AsyncMock() - - client = await hass_client_no_auth() - - thumb_url = thumb_url[:-1] - - response = await client.get(thumb_url) - assert response.status == 401 - - assert not mock_entry.api.get_event_thumbnail.called - - -async def test_thumbnail_view_upstream_error( - thumb_url: str, - hass_client_no_auth, - mock_entry: MockEntityFixture, -): - """Test good result from thumbnail view.""" - - mock_entry.api.get_event_thumbnail = AsyncMock(side_effect=NvrError) - - client = await hass_client_no_auth() - - response = await client.get(thumb_url) - assert response.status == 404 - - -async def test_thumbnail_view_no_thumb( - thumb_url: str, - hass_client_no_auth, - mock_entry: MockEntityFixture, -): - """Test good result from thumbnail view.""" - - mock_entry.api.get_event_thumbnail = AsyncMock(return_value=None) - - client = await hass_client_no_auth() - - response = await client.get(thumb_url) - assert response.status == 404 - - -async def test_thumbnail_view_expired_access_token( - hass: HomeAssistant, - thumb_url: str, - hass_client_no_auth, - mock_entry: MockEntityFixture, -): - """Test good result from thumbnail view.""" - - mock_entry.api.get_event_thumbnail = AsyncMock() - - await time_changed(hass, TOKEN_CHANGE_INTERVAL.total_seconds()) - await time_changed(hass, TOKEN_CHANGE_INTERVAL.total_seconds()) - - client = await hass_client_no_auth() - - response = await client.get(thumb_url) - assert response.status == 401