Ignore a credentials_hash produced by a different transport

A credentials_hash is specific to the transport that made it. Klap stores the base64 of a raw digest, aes and ssltransport store base64 json of hashed credentials, and sslaestransport stores base64 json of the plaintext. Nothing checked that the hash a transport was handed was its own. A device can change its encryption type without the credentials changing. Toggling Third-Party Compatibility on a Tapo device moves it between tpap and klap, and discovery then hands the transport a stored hash the other transport wrote. Klap decodes any base64 successfully, so it built _local_auth_hash out of another transport's json and failed handshake1, reporting "Device response did not match our challenge". Callers cannot tell that from a wrong password; Home Assistant acts on it by deleting the stored hash and asking the user to reauthenticate. aestransport and ssltransport are worse and raise UnicodeDecodeError from the constructor, and sslaestransport raises KeyError. Check the hash has the shape the transport produces and treat a foreign one as absent instead.
rytilahti · Oct 4, 2026 · Aug 31, 2026 · Aug 31, 2026 · Sep 19, 2026 · Aug 31, 2026
commit da1e72587f54225899c946794a7d441b8932cba1
diff --git a/kasa/transports/aestransport.py b/kasa/transports/aestransport.py
@@ -86,6 +86,12 @@ def __init__(
         super().__init__(config=config)
 
         self._login_version = config.connection_type.login_version
+        # A hash another transport produced is not a bad password.
+        if self._credentials_hash and not self._is_transport_credentials_hash(
+            self._credentials_hash
+        ):
+            self._credentials_hash = None
+
         if (
             not self._credentials or self._credentials.username is None
         ) and not self._credentials_hash:
@@ -131,6 +137,15 @@ def credentials_hash(self) -> str | None:
             return None
         return base64.b64encode(json_dumps(self._login_params).encode()).decode()
 
+    @staticmethod
+    def _is_transport_credentials_hash(credentials_hash: str) -> bool:
+        """Whether the hash has the shape this transport produces."""
+        try:
+            decoded = json_loads(base64.b64decode(credentials_hash.encode()))
+        except (ValueError, UnicodeDecodeError):
+            return False
+        return isinstance(decoded, dict) and "username" in decoded
+
     def _get_login_params(self, credentials: Credentials) -> dict[str, str]:
         """Get the login parameters based on the login_version."""
         un, pw = self.hash_credentials(self._login_version == 2, credentials)

diff --git a/kasa/transports/klaptransport.py b/kasa/transports/klaptransport.py
@@ -118,6 +118,12 @@ def __init__(
 
         self._http_client = HttpClient(config)
         self._local_seed: bytes | None = None
+        # A hash another transport produced is not a bad password.
+        if self._credentials_hash and not self._is_transport_credentials_hash(
+            self._credentials_hash
+        ):
+            self._credentials_hash = None
+
         if (
             not self._credentials or self._credentials.username is None
         ) and not self._credentials_hash:
@@ -162,6 +168,26 @@ def credentials_hash(self) -> str | None:
             return None
         return base64.b64encode(self._local_auth_hash).decode()
 
+    @classmethod
+    def _is_transport_credentials_hash(cls, credentials_hash: str) -> bool:
+        """Whether the hash has the shape this transport produces.
+
+        A device can change its encryption type without the credentials
+        changing, so a stored hash may be one that another transport wrote.
+        A klap hash is the base64 of a raw digest, so it is the right length
+        and, unlike the json hashes other transports store, not decodable.
+        """
+        try:
+            decoded = base64.b64decode(credentials_hash.encode(), validate=True)
+        except ValueError:
+            return False
+        if len(decoded) != len(cls.generate_auth_hash(Credentials())):
+            return False
+        try:
+            return not isinstance(json_loads(decoded), dict)
+        except (ValueError, UnicodeDecodeError):
+            return True
+
     async def perform_handshake1(self) -> tuple[bytes, bytes, bytes]:
         """Perform handshake1."""
         local_seed: bytes = secrets.token_bytes(16)

diff --git a/kasa/transports/sslaestransport.py b/kasa/transports/sslaestransport.py
@@ -91,6 +91,13 @@ def __init__(
         super().__init__(config=config)
 
         self._login_version = config.connection_type.login_version
+
+        # A hash another transport produced is not a bad password.
+        if self._credentials_hash and not self._is_transport_credentials_hash(
+            self._credentials_hash
+        ):
+            self._credentials_hash = None
+
         if (
             not self._credentials or self._credentials.username is None
         ) and not self._credentials_hash:
@@ -147,6 +154,15 @@ def _create_b64_credentials(credentials: Credentials) -> str:
         ch = {"un": credentials.username, "pwd": credentials.password}
         return base64.b64encode(json_dumps(ch).encode()).decode()
 
+    @staticmethod
+    def _is_transport_credentials_hash(credentials_hash: str) -> bool:
+        """Whether the hash has the shape this transport produces."""
+        try:
+            decoded = json_loads(base64.b64decode(credentials_hash.encode()))
+        except (ValueError, UnicodeDecodeError):
+            return False
+        return isinstance(decoded, dict) and "un" in decoded and "pwd" in decoded
+
     @property
     def credentials_hash(self) -> str | None:
         """The hashed credentials used by the transport."""

diff --git a/kasa/transports/ssltransport.py b/kasa/transports/ssltransport.py
@@ -69,6 +69,12 @@ def __init__(
     ) -> None:
         super().__init__(config=config)
 
+        # A hash another transport produced is not a bad password.
+        if self._credentials_hash and not self._is_transport_credentials_hash(
+            self._credentials_hash
+        ):
+            self._credentials_hash = None
+
         if (
             not self._credentials or self._credentials.username is None
         ) and not self._credentials_hash:
@@ -103,6 +109,15 @@ def credentials_hash(self) -> str:
         """The hashed credentials used by the transport."""
         return base64.b64encode(json_dumps(self._login_params).encode()).decode()
 
+    @staticmethod
+    def _is_transport_credentials_hash(credentials_hash: str) -> bool:
+        """Whether the hash has the shape this transport produces."""
+        try:
+            decoded = json_loads(base64.b64decode(credentials_hash.encode()))
+        except (ValueError, UnicodeDecodeError):
+            return False
+        return isinstance(decoded, dict) and "username" in decoded
+
     def _get_login_params(self, credentials: Credentials) -> dict[str, str]:
         """Get the login parameters based on the login_version."""
         un, pw = self.hash_credentials(credentials)

diff --git a/tests/transports/test_credentials_hash.py b/tests/transports/test_credentials_hash.py
@@ -0,0 +1,121 @@
+"""Tests for credentials_hash handling across transports.
+
+A credentials_hash is transport specific. Devices can change their encryption
+type without the credentials changing, for example when Third-Party
+Compatibility is toggled on a Tapo device, so a transport can be handed a hash
+that a different transport produced. It must not treat that as a bad password.
+"""
+
+from __future__ import annotations
+
+import base64
+
+import pytest
+
+from kasa.credentials import Credentials
+from kasa.deviceconfig import DeviceConfig
+from kasa.json import dumps as json_dumps
+from kasa.transports.aestransport import AesTransport
+from kasa.transports.klaptransport import KlapTransportV2
+from kasa.transports.sslaestransport import SslAesTransport
+from kasa.transports.ssltransport import SslTransport
+
+pytestmark = [pytest.mark.requires_dummy]
+
+CREDENTIALS = Credentials("user@example.com", "great_password")
+
+
+def klap_hash(credentials: Credentials) -> str:
+    """Build a KLAP v2 credentials_hash: base64 of a raw sha256 digest."""
+    return base64.b64encode(KlapTransportV2.generate_auth_hash(credentials)).decode()
+
+
+def aes_hash(credentials: Credentials) -> str:
+    """Build an AES credentials_hash: base64 json of sha1'd credentials."""
+    un, pw = AesTransport.hash_credentials(True, credentials)
+    return base64.b64encode(
+        json_dumps({"password2": pw, "username": un}).encode()
+    ).decode()
+
+
+def plaintext_hash(credentials: Credentials) -> str:
+    """Build a TPAP or SSL-AES credentials_hash: base64 json of plaintext."""
+    return base64.b64encode(
+        json_dumps({"un": credentials.username, "pwd": credentials.password}).encode()
+    ).decode()
+
+
+async def test_klap_ignores_an_aes_credentials_hash():
+    """KLAP must not build an auth hash out of another transport's hash."""
+    transport = KlapTransportV2(
+        config=DeviceConfig("127.0.0.1", credentials_hash=aes_hash(CREDENTIALS))
+    )
+
+    assert transport._local_auth_hash == KlapTransportV2.generate_auth_hash(
+        Credentials()
+    )
+    assert transport.credentials_hash is None
+
+
+async def test_aes_ignores_a_klap_credentials_hash():
+    """AES must not crash or authenticate on another transport's hash."""
+    transport = AesTransport(
+        config=DeviceConfig("127.0.0.1", credentials_hash=klap_hash(CREDENTIALS))
+    )
+
+    assert transport._login_params == AesTransport._get_login_params(
+        transport, Credentials()
+    )
+    assert transport.credentials_hash is None
+
+
+async def test_klap_keeps_its_own_credentials_hash():
+    """A hash the transport itself produced is still used."""
+    credentials_hash = klap_hash(CREDENTIALS)
+    transport = KlapTransportV2(
+        config=DeviceConfig("127.0.0.1", credentials_hash=credentials_hash)
+    )
+
+    assert transport.credentials_hash == credentials_hash
+
+
+async def test_aes_keeps_its_own_credentials_hash():
+    """A hash the transport itself produced is still used."""
+    credentials_hash = aes_hash(CREDENTIALS)
+    transport = AesTransport(
+        config=DeviceConfig("127.0.0.1", credentials_hash=credentials_hash)
+    )
+
+    assert transport.credentials_hash == credentials_hash
+
+
+async def test_sslaes_ignores_a_klap_credentials_hash():
+    """SSL-AES must not crash on another transport's hash."""
+    transport = SslAesTransport(
+        config=DeviceConfig("127.0.0.1", credentials_hash=klap_hash(CREDENTIALS))
+    )
+
+    assert transport._username is None
+    assert transport._password is None
+    assert transport.credentials_hash is None
+
+
+async def test_sslaes_keeps_its_own_credentials_hash():
+    """A hash the transport itself produced is still used."""
+    credentials_hash = plaintext_hash(CREDENTIALS)
+    transport = SslAesTransport(
+        config=DeviceConfig("127.0.0.1", credentials_hash=credentials_hash)
+    )
+
+    assert transport.credentials_hash == credentials_hash
+
+
+async def test_ssl_ignores_a_klap_credentials_hash():
+    """The ssl transport must not crash on another transport's hash."""
+    transport = SslTransport(
+        config=DeviceConfig("127.0.0.1", credentials_hash=klap_hash(CREDENTIALS))
+    )
+
+    assert transport._login_params == SslTransport._get_login_params(
+        transport, Credentials()
+    )