From d8a33287af9662c43cb1f148172a69778c7369ff Mon Sep 17 00:00:00 2001 From: Steven B <51370195+sdb9696@users.noreply.github.com> Date: Thu, 21 Nov 2024 10:13:20 +0000 Subject: [PATCH 1/3] Use credentials_hash for smartcamera rtsp url --- kasa/smartcamera/modules/camera.py | 23 ++++++++++++++++++++++- tests/smartcamera/test_smartcamera.py | 24 ++++++++++++++++-------- 2 files changed, 38 insertions(+), 9 deletions(-) diff --git a/kasa/smartcamera/modules/camera.py b/kasa/smartcamera/modules/camera.py index ecd7fff70..47b12392f 100644 --- a/kasa/smartcamera/modules/camera.py +++ b/kasa/smartcamera/modules/camera.py @@ -2,11 +2,13 @@ from __future__ import annotations +import base64 from urllib.parse import quote_plus from ...credentials import Credentials from ...device_type import DeviceType from ...feature import Feature +from ...json import loads as json_loads from ..smartcameramodule import SmartCameraModule LOCAL_STREAMING_PORT = 554 @@ -38,6 +40,24 @@ def is_on(self) -> bool: """Return the device id.""" return self.data["lens_mask_info"]["enabled"] == "off" + def _get_credentials(self) -> Credentials | None: + """Get credentials from .""" + config = self._device.config + if credentials := config.credentials: + return credentials + + if credentials_hash := config.credentials_hash: + try: + decoded = json_loads( + base64.b64decode(credentials_hash.encode()).decode() + ) + except Exception: + return None + if (username := decoded.get("un")) and (password := decoded.get("pwd")): + return Credentials(username, password) + + return None + def stream_rtsp_url(self, credentials: Credentials | None = None) -> str | None: """Return the local rtsp streaming url. @@ -51,7 +71,8 @@ def stream_rtsp_url(self, credentials: Credentials | None = None) -> str | None: return None dev = self._device if not credentials: - credentials = dev.credentials + credentials = self._get_credentials() + if not credentials or not credentials.username or not credentials.password: return None username = quote_plus(credentials.username) diff --git a/tests/smartcamera/test_smartcamera.py b/tests/smartcamera/test_smartcamera.py index 6b69cbb77..04f8283f4 100644 --- a/tests/smartcamera/test_smartcamera.py +++ b/tests/smartcamera/test_smartcamera.py @@ -2,6 +2,8 @@ from __future__ import annotations +import base64 +import json from datetime import UTC, datetime from unittest.mock import patch @@ -35,28 +37,34 @@ async def test_stream_rtsp_url(dev: Device): url = camera_module.stream_rtsp_url(Credentials("foo", "bar")) assert url == "rtsp://foo:bar@127.0.0.123:554/stream1" - with patch.object( - dev.protocol._transport, "_credentials", Credentials("bar", "foo") - ): + with patch.object(dev.config, "credentials", Credentials("bar", "foo")): url = camera_module.stream_rtsp_url() assert url == "rtsp://bar:foo@127.0.0.123:554/stream1" - with patch.object(dev.protocol._transport, "_credentials", Credentials("bar", "")): + with patch.object(dev.config, "credentials", Credentials("bar", "")): url = camera_module.stream_rtsp_url() assert url is None - with patch.object(dev.protocol._transport, "_credentials", Credentials("", "Foo")): + with patch.object(dev.config, "credentials", Credentials("", "Foo")): url = camera_module.stream_rtsp_url() assert url is None + # Test with credentials_hash + cred = json.dumps({"un": "bar", "pwd": "foobar"}) + cred_hash = base64.b64encode(cred.encode()).decode() + with ( + patch.object(dev.config, "credentials", None), + patch.object(dev.config, "credentials_hash", cred_hash), + ): + url = camera_module.stream_rtsp_url() + assert url == "rtsp://bar:foobar@127.0.0.123:554/stream1" + # Test with camera off await camera_module.set_state(False) await dev.update() url = camera_module.stream_rtsp_url(Credentials("foo", "bar")) assert url is None - with patch.object( - dev.protocol._transport, "_credentials", Credentials("bar", "foo") - ): + with patch.object(dev.config, "credentials", Credentials("bar", "foo")): url = camera_module.stream_rtsp_url() assert url is None From b109c5f621ed8adf0c042cdb1c623b5510b83b10 Mon Sep 17 00:00:00 2001 From: Steven B <51370195+sdb9696@users.noreply.github.com> Date: Thu, 21 Nov 2024 14:55:36 +0000 Subject: [PATCH 2/3] Test coverage --- tests/smartcamera/test_smartcamera.py | 16 ++++++++++++++++ 1 file changed, 16 insertions(+) diff --git a/tests/smartcamera/test_smartcamera.py b/tests/smartcamera/test_smartcamera.py index 04f8283f4..6f1d82d35 100644 --- a/tests/smartcamera/test_smartcamera.py +++ b/tests/smartcamera/test_smartcamera.py @@ -59,6 +59,22 @@ async def test_stream_rtsp_url(dev: Device): url = camera_module.stream_rtsp_url() assert url == "rtsp://bar:foobar@127.0.0.123:554/stream1" + # Test with invalid credentials_hash + with ( + patch.object(dev.config, "credentials", None), + patch.object(dev.config, "credentials_hash", b"238472871"), + ): + url = camera_module.stream_rtsp_url() + assert url is None + + # Test with no credentials + with ( + patch.object(dev.config, "credentials", None), + patch.object(dev.config, "credentials_hash", None), + ): + url = camera_module.stream_rtsp_url() + assert url is None + # Test with camera off await camera_module.set_state(False) await dev.update() From eb6e03feb573b22df9e82635383c3fd6ccdfb978 Mon Sep 17 00:00:00 2001 From: Steven B <51370195+sdb9696@users.noreply.github.com> Date: Thu, 21 Nov 2024 18:31:19 +0000 Subject: [PATCH 3/3] Log warning when unable to deserialize credentials_hash --- kasa/smartcamera/modules/camera.py | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/kasa/smartcamera/modules/camera.py b/kasa/smartcamera/modules/camera.py index 47b12392f..65f434d15 100644 --- a/kasa/smartcamera/modules/camera.py +++ b/kasa/smartcamera/modules/camera.py @@ -3,6 +3,7 @@ from __future__ import annotations import base64 +import logging from urllib.parse import quote_plus from ...credentials import Credentials @@ -11,6 +12,8 @@ from ...json import loads as json_loads from ..smartcameramodule import SmartCameraModule +_LOGGER = logging.getLogger(__name__) + LOCAL_STREAMING_PORT = 554 @@ -52,6 +55,9 @@ def _get_credentials(self) -> Credentials | None: base64.b64decode(credentials_hash.encode()).decode() ) except Exception: + _LOGGER.warning( + "Unable to deserialize credentials_hash: %s", credentials_hash + ) return None if (username := decoded.get("un")) and (password := decoded.get("pwd")): return Credentials(username, password)