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)