From 53823ea9a50391f0f283357a02af13ad75474695 Mon Sep 17 00:00:00 2001 From: nopoz Date: Sun, 4 Oct 2026 06:30:01 -0700 Subject: [PATCH] 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/credentials.py | 24 +++ kasa/transports/aestransport.py | 9 + kasa/transports/basetransport.py | 22 +++ kasa/transports/klaptransport.py | 18 ++ kasa/transports/sslaestransport.py | 9 + kasa/transports/ssltransport.py | 9 + tests/transports/test_credentials_hash.py | 228 ++++++++++++++++++++++ 7 files changed, 319 insertions(+) create mode 100644 tests/transports/test_credentials_hash.py diff --git a/kasa/credentials.py b/kasa/credentials.py index 3497b76a..b57cc292 100644 --- a/kasa/credentials.py +++ b/kasa/credentials.py @@ -5,6 +5,8 @@ from __future__ import annotations import base64 from dataclasses import dataclass, field +from kasa.json import loads as json_loads + @dataclass class Credentials: @@ -15,6 +17,28 @@ class Credentials: #: Password of the cloud account password: str = field(default="", repr=False) + @classmethod + def _from_plaintext_hash(cls, credentials_hash: str) -> Credentials | None: + """Recover the credentials from a hash that stores them in plaintext. + + The ssl aes and tpap transports store base64 json of the plaintext + credentials, so a transport handed one of those after a device changed + its encryption type can derive its own hash rather than failing to + authenticate. Klap and aes hashes are one way, so this only works in + that direction. + """ + try: + decoded = json_loads(base64.b64decode(credentials_hash.encode())) + except (ValueError, UnicodeDecodeError): + return None + if not isinstance(decoded, dict): + return None + username = decoded.get("un") + password = decoded.get("pwd") + if isinstance(username, str) and isinstance(password, str): + return cls(username, password) + return None + def get_default_credentials(crdentials: tuple[str, str]) -> Credentials: """Return decoded default credentials.""" diff --git a/kasa/transports/aestransport.py b/kasa/transports/aestransport.py index 45b963fe..979c7d33 100644 --- a/kasa/transports/aestransport.py +++ b/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) diff --git a/kasa/transports/basetransport.py b/kasa/transports/basetransport.py index 1f1ed7d9..04485c48 100644 --- a/kasa/transports/basetransport.py +++ b/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.""" diff --git a/kasa/transports/klaptransport.py b/kasa/transports/klaptransport.py index 8253e0ae..7f0c1ff3 100644 --- a/kasa/transports/klaptransport.py +++ b/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) diff --git a/kasa/transports/sslaestransport.py b/kasa/transports/sslaestransport.py index ba7e8322..918ec656 100644 --- a/kasa/transports/sslaestransport.py +++ b/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.""" diff --git a/kasa/transports/ssltransport.py b/kasa/transports/ssltransport.py index e4fef9a3..f4abe487 100644 --- a/kasa/transports/ssltransport.py +++ b/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) diff --git a/tests/transports/test_credentials_hash.py b/tests/transports/test_credentials_hash.py new file mode 100644 index 00000000..d3441130 --- /dev/null +++ b/tests/transports/test_credentials_hash.py @@ -0,0 +1,228 @@ +"""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, + DeviceConnectionParameters, + DeviceEncryptionType, + DeviceFamily, +) +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") + +# The aes hash depends on the login version, so pin it rather than relying +# on the DeviceConfig default. +AES_LV2 = DeviceConnectionParameters( + device_family=DeviceFamily.SmartTapoPlug, + encryption_type=DeviceEncryptionType.Aes, + login_version=2, +) + + +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), + connection_type=AES_LV2, + ) + ) + + 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, + connection_type=AES_LV2, + ) + ) + + assert transport.credentials_hash == credentials_hash + + +async def test_klap_recovers_credentials_from_a_plaintext_hash(): + """A TPAP or SSL-AES hash carries plaintext, so KLAP can rederive its own.""" + transport = KlapTransportV2( + config=DeviceConfig("127.0.0.1", credentials_hash=plaintext_hash(CREDENTIALS)) + ) + + assert transport.credentials_hash == klap_hash(CREDENTIALS) + + +async def test_aes_recovers_credentials_from_a_plaintext_hash(): + """A TPAP or SSL-AES hash carries plaintext, so AES can rederive its own.""" + transport = AesTransport( + config=DeviceConfig( + "127.0.0.1", + credentials_hash=plaintext_hash(CREDENTIALS), + connection_type=AES_LV2, + ) + ) + + assert transport.credentials_hash == aes_hash(CREDENTIALS) + + +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() + ) + + +async def test_ssl_recovers_credentials_from_a_plaintext_hash(): + """A TPAP or SSL-AES hash carries plaintext, so ssl can rederive its own.""" + transport = SslTransport( + config=DeviceConfig("127.0.0.1", credentials_hash=plaintext_hash(CREDENTIALS)) + ) + + assert transport._login_params == SslTransport._get_login_params( + transport, CREDENTIALS + ) + + +async def test_klap_ignores_a_malformed_credentials_hash(): + """A hash that is not even base64 is treated as absent.""" + transport = KlapTransportV2( + config=DeviceConfig("127.0.0.1", credentials_hash="not!valid!base64!") + ) + + assert transport._local_auth_hash == KlapTransportV2.generate_auth_hash( + Credentials() + ) + assert transport.credentials_hash is None + + +async def test_klap_ignores_a_hash_that_is_not_a_json_object(): + """Recovery only applies to a json object, not to any decodable json.""" + credentials_hash = base64.b64encode(json_dumps(["user", "pass"]).encode()).decode() + transport = KlapTransportV2( + config=DeviceConfig("127.0.0.1", credentials_hash=credentials_hash) + ) + + assert transport._local_auth_hash == KlapTransportV2.generate_auth_hash( + Credentials() + ) + + +async def test_klap_prefers_credentials_over_a_foreign_hash(): + """Configured credentials win, so nothing is recovered from the hash.""" + other = Credentials("other@example.com", "other_password") + transport = KlapTransportV2( + config=DeviceConfig( + "127.0.0.1", + credentials=CREDENTIALS, + credentials_hash=plaintext_hash(other), + ) + ) + + assert transport.credentials_hash == klap_hash(CREDENTIALS) + + +async def test_aes_prefers_credentials_over_a_foreign_hash(): + """Configured credentials win, so nothing is recovered from the hash.""" + other = Credentials("other@example.com", "other_password") + transport = AesTransport( + config=DeviceConfig( + "127.0.0.1", + credentials=CREDENTIALS, + credentials_hash=plaintext_hash(other), + connection_type=AES_LV2, + ) + ) + + assert transport.credentials_hash == aes_hash(CREDENTIALS)