Do not authenticate with a credentials_hash from a different transport (#1749)

A `credentials_hash` is specific to the transport that produced it, but
nothing checked that the hash a transport was handed was its own.

This matters because a device can change its encryption type without the
credentials changing. For example, toggling **Third-Party Compatibility** in the Tapo
app moves a plug or strip between TPAP and KLAP, and firmware updates are known to introduce new transports.

This introduces a per-transport check that the hash has the shape that transport produces, and treats a foreign hash as absent rather than as a bad password. This makes the failure honest but still ends in a reauth.

This also changes how credentials are read back out when the hash is one of the plaintext forms and lets the transport derive its own, so the change of encryption type needs no reauth in all cases.

kasa/transports/aestransport.py

@@ -131,6 +131,15 @@ class AesTransport(BaseTransport):
             return None
         return base64.b64encode(json_dumps(self._login_params).encode()).decode()
 
+    @classmethod
+    def is_transport_credentials_hash(cls, 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)

kasa/transports/basetransport.py

@@ -8,6 +8,8 @@ from __future__ import annotations
 from abc import ABC, abstractmethod
 from typing import TYPE_CHECKING
 
+from kasa.credentials import Credentials
+
 if TYPE_CHECKING:
     from kasa import DeviceConfig
 
@@ -28,6 +30,16 @@ class BaseTransport(ABC):
         self._port = config.port_override or self.default_port
         self._credentials = config.credentials
         self._credentials_hash = config.credentials_hash
+        if self._credentials_hash and not self.is_transport_credentials_hash(
+            self._credentials_hash
+        ):
+            # A hash another transport produced is not a bad password, so drop
+            # it, recovering the credentials from it first if it holds them.
+            if not self._credentials:
+                self._credentials = Credentials._from_plaintext_hash(
+                    self._credentials_hash
+                )
+            self._credentials_hash = None
         if not config.timeout:
             config.timeout = self.DEFAULT_TIMEOUT
         self._timeout = config.timeout
@@ -42,6 +54,16 @@ class BaseTransport(ABC):
     def credentials_hash(self) -> str | None:
         """The hashed credentials used by the transport."""
 
+    @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.
+        Transports that can recognise their own hashes override this.
+        """
+        return True
+
     @abstractmethod
     async def send(self, request: str) -> dict:
         """Send a message to the device and return a response."""

kasa/transports/klaptransport.py

@@ -162,6 +162,24 @@ class KlapTransport(BaseTransport):
             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 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)

kasa/transports/sslaestransport.py

@@ -148,6 +148,15 @@ class SslAesTransport(BaseTransport):
         ch = {"un": credentials.username, "pwd": credentials.password}
         return base64.b64encode(json_dumps(ch).encode()).decode()
 
+    @classmethod
+    def is_transport_credentials_hash(cls, 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."""

kasa/transports/ssltransport.py

@@ -103,6 +103,15 @@ class SslTransport(BaseTransport):
         """The hashed credentials used by the transport."""
         return base64.b64encode(json_dumps(self._login_params).encode()).decode()
 
+    @classmethod
+    def is_transport_credentials_hash(cls, 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)