Fix credential hash to return None on empty credentials (#1029)

If discovery is triggered without credentials and discovers devices requiring authentication,
blank credentials are used to initialise the protocols and no connection is actually made.
In this instance we should not return the credentials_hash for blank credentials as it will be invalid.
This commit is contained in:
Steven B 2024-07-02 13:43:37 +01:00 committed by GitHub
parent e5b959e4a9
commit b8a87f1c57
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
6 changed files with 74 additions and 8 deletions

View File

@ -117,8 +117,10 @@ class AesTransport(BaseTransport):
return self.DEFAULT_PORT
@property
def credentials_hash(self) -> str:
def credentials_hash(self) -> str | None:
"""The hashed credentials used by the transport."""
if self._credentials == Credentials():
return None
return base64.b64encode(json_dumps(self._login_params).encode()).decode()
def _get_login_params(self, credentials: Credentials) -> dict[str, str]:

View File

@ -132,8 +132,10 @@ class KlapTransport(BaseTransport):
return self.DEFAULT_PORT
@property
def credentials_hash(self) -> str:
def credentials_hash(self) -> str | None:
"""The hashed credentials used by the transport."""
if self._credentials == Credentials():
return None
return base64.b64encode(self._local_auth_hash).decode()
async def perform_handshake1(self) -> tuple[bytes, bytes, bytes]:

View File

@ -59,7 +59,7 @@ class BaseTransport(ABC):
@property
@abstractmethod
def credentials_hash(self) -> str:
def credentials_hash(self) -> str | None:
"""The hashed credentials used by the transport."""
@abstractmethod

View File

@ -234,8 +234,8 @@ class FakeIotTransport(BaseTransport):
return 9999
@property
def credentials_hash(self) -> str:
return ""
def credentials_hash(self) -> None:
return None
def set_alias(self, x, child_ids=None):
if child_ids is None:

View File

@ -13,6 +13,7 @@ import pytest
from ..aestransport import AesTransport
from ..credentials import Credentials
from ..device import Device
from ..deviceconfig import DeviceConfig
from ..exceptions import KasaException
from ..iotprotocol import IotProtocol, _deprecated_TPLinkSmartHomeProtocol
@ -512,11 +513,72 @@ def test_transport_init_signature(class_name_obj):
)
@pytest.mark.parametrize(
("transport_class", "login_version", "expected_hash"),
[
pytest.param(
AesTransport,
1,
"eyJwYXNzd29yZCI6IlFtRnkiLCJ1c2VybmFtZSI6Ik1qQXhZVFppTXpBMU0yTmpNVFF5TW1ReVl6TTJOekJpTmpJMk1UWXlNakZrTWpJNU1Ea3lPUT09In0=",
id="aes-lv-1",
),
pytest.param(
AesTransport,
2,
"eyJwYXNzd29yZDIiOiJaVFE1Tm1aa01qQXhNelprTkdKaU56Z3lPR1ZpWWpCaFlqa3lOV0l4WW1RNU56Y3lNRGhsTkE9PSIsInVzZXJuYW1lIjoiTWpBeFlUWmlNekExTTJOak1UUXlNbVF5WXpNMk56QmlOakkyTVRZeU1qRmtNakk1TURreU9RPT0ifQ==",
id="aes-lv-2",
),
pytest.param(KlapTransport, 1, "xBhMRGYWStVCVk9aSD8/6Q==", id="klap-lv-1"),
pytest.param(KlapTransport, 2, "xBhMRGYWStVCVk9aSD8/6Q==", id="klap-lv-2"),
pytest.param(
KlapTransportV2,
1,
"tEmiensOcZkP9twDEZKwU3JJl3asmseKCP7N9sfatVo=",
id="klapv2-lv-1",
),
pytest.param(
KlapTransportV2,
2,
"tEmiensOcZkP9twDEZKwU3JJl3asmseKCP7N9sfatVo=",
id="klapv2-lv-2",
),
pytest.param(XorTransport, None, None, id="xor"),
],
)
@pytest.mark.parametrize(
("credentials", "expected_blank"),
[
pytest.param(Credentials("Foo", "Bar"), False, id="credentials"),
pytest.param(None, True, id="no-credentials"),
pytest.param(Credentials(None, "Bar"), True, id="no-username"), # type: ignore[arg-type]
],
)
async def test_transport_credentials_hash(
mocker, transport_class, login_version, expected_hash, credentials, expected_blank
):
"""Test that the actual hashing doesn't break and empty credential returns an empty hash."""
host = "127.0.0.1"
params = Device.ConnectionParameters(
device_family=Device.Family.SmartTapoPlug,
encryption_type=Device.EncryptionType.Xor,
login_version=login_version,
)
config = DeviceConfig(host, credentials=credentials, connection_type=params)
transport = transport_class(config=config)
credentials_hash = transport.credentials_hash
expected = None if expected_blank else expected_hash
assert credentials_hash == expected
@pytest.mark.parametrize(
"transport_class",
[AesTransport, KlapTransport, KlapTransportV2, XorTransport, XorTransport],
)
async def test_transport_credentials_hash(mocker, transport_class):
async def test_transport_credentials_hash_from_config(mocker, transport_class):
"""Test that credentials_hash provided via config sets correctly."""
host = "127.0.0.1"
credentials = Credentials("Foo", "Bar")

View File

@ -54,9 +54,9 @@ class XorTransport(BaseTransport):
return self.DEFAULT_PORT
@property
def credentials_hash(self) -> str:
def credentials_hash(self) -> str | None:
"""The hashed credentials used by the transport."""
return ""
return None
async def _connect(self, timeout: int) -> None:
"""Try to connect or reconnect to the device."""