FazBrowse GitHub Viewer | Trending |
URL:
| Home
Tools: [Download Repo ZIP]   [Original HTTPS Page]

Do not authenticate with a credentials_hash from a different transport by nopoz · Pull Request #1749 · python-kasa/python-kasa · GitHub

Repository navigation

Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension .py  (7) All 1 file type selected
Viewed files
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Unified
Split
Hide whitespace
Diff view
Unified
Split
Hide whitespace
24 changes: 24 additions & 0 deletions kasa/credentials.py
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode characters
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,8 @@
import base64
from dataclasses import dataclass, field

from kasa.json import loads as json_loads


@dataclass
class Credentials:
Expand All @@ -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."""
Expand Down
9 changes: 9 additions & 0 deletions kasa/transports/aestransport.py
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode characters
Original file line number Diff line number Diff line change
Expand Up @@ -131,6 +131,15 @@ def credentials_hash(self) -> str | None:
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)
Expand Down
22 changes: 22 additions & 0 deletions kasa/transports/basetransport.py
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode characters
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,8 @@
from abc import ABC, abstractmethod
from typing import TYPE_CHECKING

from kasa.credentials import Credentials

if TYPE_CHECKING:
from kasa import DeviceConfig

Expand All @@ -28,6 +30,16 @@ def __init__(
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
Expand All @@ -42,6 +54,16 @@ def default_port(self) -> int:
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."""
Expand Down
18 changes: 18 additions & 0 deletions kasa/transports/klaptransport.py
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode characters
Original file line number Diff line number Diff line change
Expand Up @@ -162,6 +162,24 @@ 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 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)
Expand Down
9 changes: 9 additions & 0 deletions kasa/transports/sslaestransport.py
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode characters
Original file line number Diff line number Diff line change
Expand Up @@ -147,6 +147,15 @@ def _create_b64_credentials(credentials: Credentials) -> str:
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."""
Expand Down
9 changes: 9 additions & 0 deletions kasa/transports/ssltransport.py
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode characters
Original file line number Diff line number Diff line change
Expand Up @@ -103,6 +103,15 @@ def credentials_hash(self) -> str:
"""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)
Expand Down
228 changes: 228 additions & 0 deletions tests/transports/test_credentials_hash.py
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode characters
Original file line number Diff line number Diff line change
@@ -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)
Loading

Back | FazBrowse Home | New Git URL