8000 Use credentials_hash for smartcamera rtsp url by sdb9696 · Pull Request #1293 · python-kasa/python-kasa · GitHub
[go: up one dir, main page]
Skip to content

Use credentials_hash for smartcamera rtsp url #1293

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 3 commits into from
Nov 21, 2024
Merged

Use credentials_hash for smartcamera rtsp url #1293

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
d8a3328
Use credentials_hash for smartcamera rtsp url
sdb9696 Nov 21, 2024
b109c5f
Test coverage
sdb9696 Nov 21, 2024
eb6e03f
Log warning when unable to deserialize credentials_hash
sdb9696 Nov 21, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
29 changes: 28 additions & 1 deletion kasa/smartcamera/modules/camera.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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


Expand Down Expand Up @@ -38,6 +43,27 @@ 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:
_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.

Expand All @@ -51,7 +77,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)
Expand Down
40 changes: 32 additions & 8 deletions tests/smartcamera/test_smartcamera.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,8 @@

from __future__ import annotations

import base64
import json
from datetime import UTC, datetime
from unittest.mock import patch

Expand Down Expand Up @@ -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(
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 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

Expand All @@ -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(
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

Expand Down
Loading
0