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.
2026-10-06 21:03:50 +00:00 · 2026-10-04 10:58:14 -04:00
parent b36c795cbc
commit 99716fc3b9
2 changed files with 282 additions and 8 deletions
--- 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
--- 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