| FazBrowse GitHub Viewer | Trending | | Home |
| Tools: [Download Repo ZIP] [Original HTTPS Page] |
| Original file line number | Diff line number | Diff line change | |
|---|---|---|---|
@@ -2,13 +2,18 @@ | |||
| 2 | 2 | ||
| 3 | 3 | from __future__ import annotations | |
| 4 | 4 | ||
| 5 | + import base64 | ||
| 6 | + import logging | ||
| 5 | 7 | from urllib.parse import quote_plus | |
| 6 | 8 | ||
| 7 | 9 | from ...credentials import Credentials | |
| 8 | 10 | from ...device_type import DeviceType | |
| 9 | 11 | from ...feature import Feature | |
| 12 | + from ...json import loads as json_loads | ||
| 10 | 13 | from ..smartcameramodule import SmartCameraModule | |
| 11 | 14 | ||
| 15 | + _LOGGER = logging.getLogger(__name__) | ||
| 16 | + | ||
| 12 | 17 | LOCAL_STREAMING_PORT = 554 | |
| 13 | 18 | ||
| 14 | 19 | ||
@@ -38,6 +43,27 @@ def is_on(self) -> bool: | |||
| 38 | 43 | """Return the device id.""" | |
| 39 | 44 | return self.data["lens_mask_info"]["enabled"] == "off" | |
| 40 | 45 | ||
| 46 | + def _get_credentials(self) -> Credentials | None: | ||
| 47 | + """Get credentials from .""" | ||
| 48 | + config = self._device.config | ||
| 49 | + if credentials := config.credentials: | ||
| 50 | + return credentials | ||
| 51 | + | ||
| 52 | + if credentials_hash := config.credentials_hash: | ||
| 53 | + try: | ||
| 54 | + decoded = json_loads( | ||
| 55 | + base64.b64decode(credentials_hash.encode()).decode() | ||
| 56 | + ) | ||
| 57 | + except Exception: | ||
| 58 | + _LOGGER.warning( | ||
| 59 | + "Unable to deserialize credentials_hash: %s", credentials_hash | ||
| 60 | + ) | ||
| 61 | + return None | ||
| 62 | + if (username := decoded.get("un")) and (password := decoded.get("pwd")): | ||
| 63 | + return Credentials(username, password) | ||
| 64 | + | ||
| 65 | + return None | ||
| 66 | + | ||
| 41 | 67 | def stream_rtsp_url(self, credentials: Credentials | None = None) -> str | None: | |
| 42 | 68 | """Return the local rtsp streaming url. | |
| 43 | 69 | ||
@@ -51,7 +77,8 @@ def stream_rtsp_url(self, credentials: Credentials | None = None) -> str | None: | |||
| 51 | 77 | return None | |
| 52 | 78 | dev = self._device | |
| 53 | 79 | if not credentials: | |
| 54 | - credentials = dev.credentials | ||
| 80 | + credentials = self._get_credentials() | ||
| 81 | + | ||
| 55 | 82 | if not credentials or not credentials.username or not credentials.password: | |
| 56 | 83 | return None | |
| 57 | 84 | username = quote_plus(credentials.username) | |
| Original file line number | Diff line number | Diff line change | |
|---|---|---|---|
@@ -2,6 +2,8 @@ | |||
| 2 | 2 | ||
| 3 | 3 | from __future__ import annotations | |
| 4 | 4 | ||
| 5 | + import base64 | ||
| 6 | + import json | ||
| 5 | 7 | from datetime import UTC, datetime | |
| 6 | 8 | from unittest.mock import patch | |
| 7 | 9 | ||
@@ -35,17 +37,41 @@ async def test_stream_rtsp_url(dev: Device): | |||
| 35 | 37 | url = camera_module.stream_rtsp_url(Credentials("foo", "bar")) | |
| 36 | 38 | assert url == "rtsp://foo:bar@127.0.0.123:554/stream1" | |
| 37 | 39 | ||
| 38 | - with patch.object( | ||
| 39 | - dev.protocol._transport, "_credentials", Credentials("bar", "foo") | ||
| 40 | - ): | ||
| 40 | + with patch.object(dev.config, "credentials", Credentials("bar", "foo")): | ||
| 41 | 41 | url = camera_module.stream_rtsp_url() | |
| 42 | 42 | assert url == "rtsp://bar:foo@127.0.0.123:554/stream1" | |
| 43 | 43 | ||
| 44 | - with patch.object(dev.protocol._transport, "_credentials", Credentials("bar", "")): | ||
| 44 | + with patch.object(dev.config, "credentials", Credentials("bar", "")): | ||
| 45 | 45 | url = camera_module.stream_rtsp_url() | |
| 46 | 46 | assert url is None | |
| 47 | 47 | ||
| 48 | - with patch.object(dev.protocol._transport, "_credentials", Credentials("", "Foo")): | ||
| 48 | + with patch.object(dev.config, "credentials", Credentials("", "Foo")): | ||
| 49 | + url = camera_module.stream_rtsp_url() | ||
| 50 | + assert url is None | ||
| 51 | + | ||
| 52 | + # Test with credentials_hash | ||
| 53 | + cred = json.dumps({"un": "bar", "pwd": "foobar"}) | ||
| 54 | + cred_hash = base64.b64encode(cred.encode()).decode() | ||
| 55 | + with ( | ||
| 56 | + patch.object(dev.config, "credentials", None), | ||
| 57 | + patch.object(dev.config, "credentials_hash", cred_hash), | ||
| 58 | + ): | ||
| 59 | + url = camera_module.stream_rtsp_url() | ||
| 60 | + assert url == "rtsp://bar:foobar@127.0.0.123:554/stream1" | ||
| 61 | + | ||
| 62 | + # Test with invalid credentials_hash | ||
| 63 | + with ( | ||
| 64 | + patch.object(dev.config, "credentials", None), | ||
| 65 | + patch.object(dev.config, "credentials_hash", b"238472871"), | ||
| 66 | + ): | ||
| 67 | + url = camera_module.stream_rtsp_url() | ||
| 68 | + assert url is None | ||
| 69 | + | ||
| 70 | + # Test with no credentials | ||
| 71 | + with ( | ||
| 72 | + patch.object(dev.config, "credentials", None), | ||
| 73 | + patch.object(dev.config, "credentials_hash", None), | ||
| 74 | + ): | ||
| 49 | 75 | url = camera_module.stream_rtsp_url() | |
| 50 | 76 | assert url is None | |
| 51 | 77 | ||
@@ -54,9 +80,7 @@ async def test_stream_rtsp_url(dev: Device): | |||
| 54 | 80 | await dev.update() | |
| 55 | 81 | url = camera_module.stream_rtsp_url(Credentials("foo", "bar")) | |
| 56 | 82 | assert url is None | |
| 57 | - with patch.object( | ||
| 58 | - dev.protocol._transport, "_credentials", Credentials("bar", "foo") | ||
| 59 | - ): | ||
| 83 | + with patch.object(dev.config, "credentials", Credentials("bar", "foo")): | ||
| 60 | 84 | url = camera_module.stream_rtsp_url() | |
| 61 | 85 | assert url is None | |
| 62 | 86 | ||
| Back | FazBrowse Home | New Git URL |
0 commit comments