Skip to content
Merged
Show file tree
Hide file tree
Changes from 2 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
24 changes: 24 additions & 0 deletions kasa/credentials.py
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 @@ -16,6 +18,28 @@ class Credentials:
password: str = field(default="", repr=False)


def _credentials_from_plaintext_hash(credentials_hash: str) -> Credentials | None:
Comment thread
rytilahti marked this conversation as resolved.
Outdated
"""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 Credentials(username, password)
return None


def get_default_credentials(crdentials: tuple[str, str]) -> Credentials:
"""Return decoded default credentials."""
un = base64.b64decode(crdentials[0].encode()).decode()
Expand Down
26 changes: 25 additions & 1 deletion kasa/transports/aestransport.py
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,12 @@
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
from yarl import URL

from kasa.credentials import DEFAULT_CREDENTIALS, Credentials, get_default_credentials
from kasa.credentials import (
DEFAULT_CREDENTIALS,
Credentials,
_credentials_from_plaintext_hash,
get_default_credentials,
)
from kasa.deviceconfig import DeviceConfig
from kasa.exceptions import (
SMART_AUTHENTICATION_ERRORS,
Expand Down Expand Up @@ -86,6 +91,16 @@ def __init__(
super().__init__(config=config)

self._login_version = config.connection_type.login_version
# A hash another transport produced is not a bad password.
if self._credentials_hash and not self._is_transport_credentials_hash(
self._credentials_hash
):
if not self._credentials:
self._credentials = _credentials_from_plaintext_hash(
self._credentials_hash
)
self._credentials_hash = None

if (
not self._credentials or self._credentials.username is None
) and not self._credentials_hash:
Expand Down Expand Up @@ -131,6 +146,15 @@ def credentials_hash(self) -> str | None:
return None
return base64.b64encode(json_dumps(self._login_params).encode()).decode()

@staticmethod
def _is_transport_credentials_hash(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
37 changes: 36 additions & 1 deletion kasa/transports/klaptransport.py
Original file line number Diff line number Diff line change
Expand Up @@ -59,7 +59,12 @@
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
from yarl import URL

from kasa.credentials import DEFAULT_CREDENTIALS, Credentials, get_default_credentials
from kasa.credentials import (
DEFAULT_CREDENTIALS,
Credentials,
_credentials_from_plaintext_hash,
get_default_credentials,
)
from kasa.deviceconfig import DeviceConfig
from kasa.exceptions import AuthenticationError, KasaException, _RetryableError
from kasa.httpclient import HttpClient
Expand Down Expand Up @@ -118,6 +123,16 @@ def __init__(

self._http_client = HttpClient(config)
self._local_seed: bytes | None = None
# A hash another transport produced is not a bad password.
if self._credentials_hash and not self._is_transport_credentials_hash(
self._credentials_hash
):
if not self._credentials:
self._credentials = _credentials_from_plaintext_hash(
self._credentials_hash
)
self._credentials_hash = None

if (
not self._credentials or self._credentials.username is None
) and not self._credentials_hash:
Expand Down Expand Up @@ -162,6 +177,26 @@ 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 device can change its encryption type without the credentials
changing, so a stored hash may be one that another transport wrote.
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
16 changes: 16 additions & 0 deletions kasa/transports/sslaestransport.py
Original file line number Diff line number Diff line change
Expand Up @@ -91,6 +91,13 @@ def __init__(
super().__init__(config=config)

self._login_version = config.connection_type.login_version

# A hash another transport produced is not a bad password.
if self._credentials_hash and not self._is_transport_credentials_hash(
self._credentials_hash
):
self._credentials_hash = None

if (
not self._credentials or self._credentials.username is None
) and not self._credentials_hash:
Expand Down Expand Up @@ -147,6 +154,15 @@ def _create_b64_credentials(credentials: Credentials) -> str:
ch = {"un": credentials.username, "pwd": credentials.password}
return base64.b64encode(json_dumps(ch).encode()).decode()

@staticmethod
def _is_transport_credentials_hash(credentials_hash: str) -> bool:
Comment thread
nopoz marked this conversation as resolved.
Outdated
"""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
15 changes: 15 additions & 0 deletions kasa/transports/ssltransport.py
Original file line number Diff line number Diff line change
Expand Up @@ -69,6 +69,12 @@ def __init__(
) -> None:
super().__init__(config=config)

# A hash another transport produced is not a bad password.
if self._credentials_hash and not self._is_transport_credentials_hash(
Comment thread
nopoz marked this conversation as resolved.
Outdated
self._credentials_hash
):
self._credentials_hash = None

if (
not self._credentials or self._credentials.username is None
) and not self._credentials_hash:
Expand Down Expand Up @@ -103,6 +109,15 @@ def credentials_hash(self) -> str:
"""The hashed credentials used by the transport."""
return base64.b64encode(json_dumps(self._login_params).encode()).decode()

@staticmethod
def _is_transport_credentials_hash(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
Loading