Use credentials_hash for smartcamera rtsp url (#1293)

kasa/smartcamera/modules/camera.py

@@ -2,13 +2,18 @@
 
 from __future__ import annotations
 
+import base64
+import logging
 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
 
+_LOGGER = logging.getLogger(__name__)
+
 LOCAL_STREAMING_PORT = 554
 
 
@@ -38,6 +43,27 @@ class Camera(SmartCameraModule):
         """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:
+                _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)
+
+        return None
+
     def stream_rtsp_url(self, credentials: Credentials | None = None) -> str | None:
         """Return the local rtsp streaming url.
 
@@ -51,7 +77,8 @@ class Camera(SmartCameraModule):
             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)

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,17 +37,41 @@ 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(
+    with patch.object(dev.config, "credentials", Credentials("bar", "foo")):
-        dev.protocol._transport, "_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 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
 
@@ -54,9 +80,7 @@ async def test_stream_rtsp_url(dev: Device):
     await dev.update()
     url = camera_module.stream_rtsp_url(Credentials("foo", "bar"))
     assert url is None
-    with patch.object(
+    with patch.object(dev.config, "credentials", Credentials("bar", "foo")):
-        dev.protocol._transport, "_credentials", Credentials("bar", "foo")
-    ):
         url = camera_module.stream_rtsp_url()
     assert url is None