From 99716fc3b979e478bb1720d87964ada2797a767c Mon Sep 17 00:00:00 2001 From: ZeliardM <140266236+ZeliardM@users.noreply.github.com> Date: Sun, 4 Oct 2026 10:58:14 -0400 Subject: [PATCH] Update smartcam credential handling and tests (#1664) This PR adds update_admin_password in the library and CLI for SMARTCAM devices. It also updates the update_credentials function to correctly handle the local_account for local device authentication. It updates all of the required tests and code coverage as well. --- kasa/smartcam/smartcamdevice.py | 73 ++++++++- tests/smartcam/test_smartcamdevice.py | 217 +++++++++++++++++++++++++- 2 files changed, 282 insertions(+), 8 deletions(-) diff --git a/kasa/smartcam/smartcamdevice.py b/kasa/smartcam/smartcamdevice.py index 7bc6184f..649a5a29 100644 --- a/kasa/smartcam/smartcamdevice.py +++ b/kasa/smartcam/smartcamdevice.py @@ -3,6 +3,7 @@ from __future__ import annotations import base64 +import hashlib import logging from typing import Any, cast @@ -10,6 +11,7 @@ from cryptography.hazmat.primitives import serialization from cryptography.hazmat.primitives.asymmetric import padding from cryptography.hazmat.primitives.asymmetric.rsa import RSAPublicKey +from ..credentials import DEFAULT_CREDENTIALS, get_default_credentials from ..device import DeviceInfo, WifiNetwork from ..device_type import DeviceType from ..deviceconfig import DeviceConfig @@ -357,13 +359,7 @@ class SmartCamDevice(SmartDevice): if net is None: raise DeviceError(f"Network with SSID '{ssid}' not found.") - public_key_b64 = self._public_key or self.STATIC_PUBLIC_KEY_B64 - key_bytes = base64.b64decode(public_key_b64) - public_key = serialization.load_der_public_key(key_bytes) - if not isinstance(public_key, RSAPublicKey): - raise TypeError("Loaded public key is not an RSA public key") - encrypted = public_key.encrypt(password.encode(), padding.PKCS1v15()) - encrypted_password = base64.b64encode(encrypted).decode() + encrypted_password = self._encrypt_password(password) payload = { "onboarding": { @@ -390,3 +386,66 @@ class SmartCamDevice(SmartDevice): "Received a kasa exception for wifi join, but this is expected" ) return {} + + def _encrypt_password(self, password: str) -> str: + public_key_b64 = self._public_key or self.STATIC_PUBLIC_KEY_B64 + key_bytes = base64.b64decode(public_key_b64) + public_key = serialization.load_der_public_key(key_bytes) + if not isinstance(public_key, RSAPublicKey): + raise TypeError("Loaded public key is not an RSA public key") + encrypted = public_key.encrypt(password.encode(), padding.PKCS1v15()) + return base64.b64encode(encrypted).decode() + + async def update_credentials(self, username: str, password: str) -> dict: + """Update smart camera credentials.""" + login_version = self.config.connection_type.login_version + last_error: DeviceError | None = None + for old_password_candidate in self._password_candidates(): + try: + return await self._try_change_password( + old_password_candidate, password, login_version + ) + except DeviceError as ex: + last_error = ex + + if last_error is not None: + raise last_error + raise KasaException("Unable to determine current admin password.") + + def _password_candidates(self) -> list[str]: + password_candidates = [ + get_default_credentials(default_credentials).password + for key, default_credentials in DEFAULT_CREDENTIALS.items() + if key.startswith("TAPOCAMERA") + ] + if self.credentials and self.credentials.password: + password_candidates.append(self.credentials.password) + + return list(dict.fromkeys(password_candidates)) + + async def _try_change_password( + self, old_password: str, new_password: str, login_version: int | None + ) -> dict: + new_password_hash = self._hash_password(new_password, login_version) + old_password_hash = self._hash_password(old_password, login_version) + + change_admin_password_payload: dict[str, str] = { + "secname": "root", + "username": "admin", + "old_passwd": old_password_hash, + "passwd": new_password_hash, + "ciphertext": self._encrypt_password(new_password_hash), + } + if login_version == 3: + change_admin_password_payload["encrypt_type"] = "3" + + payload = { + "user_management": {"change_admin_password": change_admin_password_payload} + } + return await self.protocol.query({"changeAdminPassword": payload}) + + @staticmethod + def _hash_password(password: str, login_version: int | None) -> str: + if login_version == 3: + return hashlib.sha256(password.encode()).hexdigest().upper() # noqa: S324 + return hashlib.md5(password.encode()).hexdigest().upper() # noqa: S324 diff --git a/tests/smartcam/test_smartcamdevice.py b/tests/smartcam/test_smartcamdevice.py index d3e38e54..46f74946 100644 --- a/tests/smartcam/test_smartcamdevice.py +++ b/tests/smartcam/test_smartcamdevice.py @@ -5,13 +5,15 @@ from __future__ import annotations import base64 import logging from datetime import UTC, datetime, timedelta +from typing import NotRequired, TypedDict from unittest.mock import AsyncMock, PropertyMock, patch from zoneinfo import ZoneInfo import pytest from freezegun.api import FrozenDateTimeFactory -from kasa import Device, DeviceType, Module +from kasa import Credentials, Device, DeviceType, Module +from kasa.credentials import DEFAULT_CREDENTIALS, get_default_credentials from kasa.exceptions import AuthenticationError, DeviceError, KasaException from kasa.smartcam import SmartCamDevice from kasa.smartcam.modules.time import Time @@ -19,6 +21,33 @@ from kasa.smartcam.modules.time import Time from ..conftest import device_smartcam, hub_smartcam +class _ChangeAdminPasswordPayload(TypedDict): + secname: str + username: str + old_passwd: str + passwd: str + ciphertext: str + encrypt_type: NotRequired[str] + + +class _UserManagementPayload(TypedDict): + change_admin_password: _ChangeAdminPasswordPayload + + +class _ChangeAdminPasswordRequest(TypedDict): + user_management: _UserManagementPayload + + +class _SmartCamPasswordRequest(TypedDict): + changeAdminPassword: _ChangeAdminPasswordRequest + + +def _change_admin_password_payload( + payload: _SmartCamPasswordRequest, +) -> _ChangeAdminPasswordPayload: + return payload["changeAdminPassword"]["user_management"]["change_admin_password"] + + @device_smartcam async def test_state(dev: Device) -> None: if dev.device_type is DeviceType.Hub: @@ -307,3 +336,189 @@ async def test_wifi_join_typeerror_on_non_rsa_key(dev: SmartCamDevice) -> None: ), ): await dev.wifi_join("TestSSID", "password123") + + +@device_smartcam +async def test_update_credentials_non_lv3_request(dev: SmartCamDevice) -> None: + dev.config.connection_type.login_version = 2 + + query_mock = AsyncMock(return_value={}) + with ( + patch.object(type(dev), "credentials", new_callable=PropertyMock) as cred_mock, + patch.object(dev.protocol, "query", query_mock), + patch.object( + dev, "_encrypt_password", return_value="encrypted-ciphertext" + ) as encrypt_mock, + ): + cred_mock.return_value = Credentials(username="admin", password="old-password") # noqa: S106 + result = await dev.update_credentials("new-user", "new-password") + + assert result == {} + encrypt_mock.assert_called_once() + query_mock.assert_awaited_once() + request: _SmartCamPasswordRequest = query_mock.await_args_list[0].args[0] + change_password_payload = _change_admin_password_payload(request) + assert change_password_payload["secname"] == "root" + assert change_password_payload["username"] == "admin" + assert change_password_payload["old_passwd"] + assert change_password_payload["passwd"] + assert change_password_payload["ciphertext"] == "encrypted-ciphertext" + assert "encrypt_type" not in change_password_payload + encrypt_mock.assert_called_once_with(change_password_payload["passwd"]) + + +@device_smartcam +async def test_update_credentials_lv3_request(dev: SmartCamDevice) -> None: + dev.config.connection_type.login_version = 3 + + query_mock = AsyncMock(side_effect=[DeviceError("bad default"), {}]) + with ( + patch.object(type(dev), "credentials", new_callable=PropertyMock) as cred_mock, + patch.object(dev.protocol, "query", query_mock), + patch.object( + dev, "_encrypt_password", return_value="encrypted-ciphertext" + ) as encrypt_mock, + ): + cred_mock.return_value = Credentials(username="admin", password="old-password") # noqa: S106 + result = await dev.update_credentials("new-user", "new-password") + + assert result == {} + assert encrypt_mock.call_count == 2 + assert query_mock.await_count == 2 + first_request: _SmartCamPasswordRequest = query_mock.await_args_list[0].args[0] + second_request: _SmartCamPasswordRequest = query_mock.await_args_list[1].args[0] + first_payload = _change_admin_password_payload(first_request) + second_payload = _change_admin_password_payload(second_request) + for payload in (first_payload, second_payload): + assert payload["secname"] == "root" + assert payload["username"] == "admin" + assert payload["old_passwd"] + assert payload["passwd"] + assert payload["ciphertext"] == "encrypted-ciphertext" + assert payload["encrypt_type"] == "3" + assert first_payload["old_passwd"] != second_payload["old_passwd"] + assert first_payload["passwd"] == second_payload["passwd"] + + +@device_smartcam +async def test_update_credentials_falls_back_to_current_password_when_default_fails( + dev: SmartCamDevice, +) -> None: + dev.config.connection_type.login_version = 2 + + query_mock = AsyncMock( + side_effect=[DeviceError("bad default"), DeviceError("bad fallback"), {}] + ) + with ( + patch.object(type(dev), "credentials", new_callable=PropertyMock) as cred_mock, + patch.object(dev.protocol, "query", query_mock), + patch.object(dev, "_encrypt_password", return_value="encrypted-ciphertext"), + ): + cred_mock.return_value = Credentials(username="admin", password="old-password") # noqa: S106 + result = await dev.update_credentials("new-user@example.com", "new-password") + + assert result == {} + assert query_mock.await_count == 3 + request: _SmartCamPasswordRequest = query_mock.await_args_list[2].args[0] + change_password_payload = _change_admin_password_payload(request) + assert change_password_payload["old_passwd"] + assert change_password_payload["passwd"] + + +@device_smartcam +async def test_update_credentials_returns_last_error_after_candidates_fail( + dev: SmartCamDevice, +) -> None: + dev.config.connection_type.login_version = 2 + + query_mock = AsyncMock( + side_effect=[ + DeviceError("bad default"), + DeviceError("bad fallback"), + DeviceError("bad current"), + ] + ) + with ( + patch.object(type(dev), "credentials", new_callable=PropertyMock) as cred_mock, + patch.object(dev.protocol, "query", query_mock), + patch.object(dev, "_encrypt_password", return_value="encrypted-ciphertext"), + ): + cred_mock.return_value = Credentials(username="admin", password="old-password") # noqa: S106 + with pytest.raises(DeviceError, match="bad current"): + await dev.update_credentials("new-user@example.com", "new-password") + + assert query_mock.await_count == 3 + + +@device_smartcam +async def test_update_credentials_all_candidates_fail(dev: SmartCamDevice) -> None: + dev.config.connection_type.login_version = 2 + error = DeviceError("always fails") + query_mock = AsyncMock(side_effect=error) + with ( + patch.object(type(dev), "credentials", new_callable=PropertyMock) as cred_mock, + patch.object(dev.protocol, "query", query_mock), + patch.object(dev, "_encrypt_password", return_value="encrypted-ciphertext"), + ): + cred_mock.return_value = Credentials(username="admin", password="old-password") # noqa: S106 + with pytest.raises(DeviceError): + await dev.update_credentials("new-user@example.com", "new-password") + assert query_mock.await_count == 3 + + +@device_smartcam +async def test_update_credentials_with_no_credentials(dev: SmartCamDevice) -> None: + dev.config.connection_type.login_version = 2 + query_mock = AsyncMock(return_value={}) + with ( + patch.object(type(dev), "credentials", new_callable=PropertyMock) as cred_mock, + patch.object(dev.protocol, "query", query_mock), + patch.object(dev, "_encrypt_password", return_value="encrypted-ciphertext"), + ): + cred_mock.return_value = None + result = await dev.update_credentials("new-user@example.com", "new-password") + + assert result == {} + # The matching default succeeds before any fallback candidate is attempted. + assert query_mock.await_count == 1 + request: _SmartCamPasswordRequest = query_mock.await_args_list[0].args[0] + payload = _change_admin_password_payload(request) + assert payload["old_passwd"] + + +@device_smartcam +async def test_update_credentials_with_no_password_candidates( + dev: SmartCamDevice, +) -> None: + with ( + patch.object(dev, "_password_candidates", return_value=[]), + pytest.raises( + KasaException, match="Unable to determine current admin password." + ), + ): + await dev.update_credentials("new-user@example.com", "new-password") + + +@device_smartcam +async def test_update_credentials_current_password_equals_default( + dev: SmartCamDevice, +) -> None: + dev.config.connection_type.login_version = 2 + default_old_password = get_default_credentials( + DEFAULT_CREDENTIALS["TAPOCAMERA"] + ).password + query_mock = AsyncMock(return_value={}) + with ( + patch.object(type(dev), "credentials", new_callable=PropertyMock) as cred_mock, + patch.object(dev.protocol, "query", query_mock), + patch.object(dev, "_encrypt_password", return_value="encrypted-ciphertext"), + ): + # current password is same as default — should not be added a second time + cred_mock.return_value = Credentials( + username="admin", password=default_old_password + ) + result = await dev.update_credentials("new-user@example.com", "new-password") + + assert result == {} + # The matching default succeeds before the fallback default is attempted. + assert query_mock.await_count == 1