"""Implementation of the TP-Link Klap Home Protocol.
Encryption/Decryption methods based on the works of
Simon Wilkinson and Chris Weeldon
Klap devices that have never been connected to the kasa
cloud should work with blank credentials.
Devices that have been connected to the kasa cloud will
switch intermittently between the users cloud credentials
and default kasa credentials that are hardcoded.
This appears to be an issue with the devices.
The protocol works by doing a two stage handshake to obtain
and encryption key and session id cookie.
Authentication uses an auth_hash which is
md5(md5(username),md5(password))
handshake1: client sends a random 16 byte local_seed to the
device and receives a random 16 bytes remote_seed, followed
by sha256(local_seed + auth_hash). It also returns a
TP_SESSIONID in the cookie header. This implementation
then checks this value against the possible auth_hashes
described above (user cloud, kasa hardcoded, blank). If it
finds a match it moves onto handshake2
handshake2: client sends sha25(remote_seed + auth_hash) to
the device along with the TP_SESSIONID. Device responds with
200 if successful. It generally will be because this
implementation checks the auth_hash it received during handshake1
encryption: local_seed, remote_seed and auth_hash are now used
for encryption. The last 4 bytes of the initialization vector
are used as a sequence number that increments every time the
client calls encrypt and this sequence number is sent as a
url parameter to the device along with the encrypted payload
https://gist.github.com/chriswheeldon/3b17d974db3817613c69191c0480fe55
https://github.com/python-kasa/python-kasa/pull/117
"""
from __future__ import annotations
import asyncio
import base64
import datetime
import hashlib
import logging
import secrets
import struct
import time
from pprint import pformat as pf
from typing import Any, cast
from cryptography.hazmat.primitives import padding
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
from yarl import URL
from .credentials import Credentials
from .deviceconfig import DeviceConfig
from .exceptions import AuthenticationError, KasaException, _RetryableError
from .httpclient import HttpClient
from .json import loads as json_loads
from .protocol import DEFAULT_CREDENTIALS, BaseTransport, get_default_credentials, md5
_LOGGER = logging.getLogger(__name__)
ONE_DAY_SECONDS = 86400
SESSION_EXPIRE_BUFFER_SECONDS = 60 * 20
PACK_SIGNED_LONG = struct.Struct(">l").pack
def _sha256(payload: bytes) -> bytes:
return hashlib.sha256(payload).digest() # noqa: S324
def _sha1(payload: bytes) -> bytes:
return hashlib.sha1(payload).digest() # noqa: S324
[docs]class KlapTransport(BaseTransport):
"""Implementation of the KLAP encryption protocol.
KLAP is the name used in device discovery for TP-Link's new encryption
protocol, used by newer firmware versions.
"""
DEFAULT_PORT: int = 80
DISCOVERY_QUERY = {"system": {"get_sysinfo": None}}
SESSION_COOKIE_NAME = "TP_SESSIONID"
TIMEOUT_COOKIE_NAME = "TIMEOUT"
def __init__(
self,
*,
config: DeviceConfig,
) -> None:
super().__init__(config=config)
self._http_client = HttpClient(config)
self._local_seed: bytes | None = None
if (
not self._credentials or self._credentials.username is None
) and not self._credentials_hash:
self._credentials = Credentials()
if self._credentials:
self._local_auth_hash = self.generate_auth_hash(self._credentials)
self._local_auth_owner = self.generate_owner_hash(self._credentials).hex()
else:
self._local_auth_hash = base64.b64decode(self._credentials_hash.encode()) # type: ignore[union-attr]
self._default_credentials_auth_hash: dict[str, bytes] = {}
self._blank_auth_hash = None
self._handshake_lock = asyncio.Lock()
self._query_lock = asyncio.Lock()
self._handshake_done = False
self._encryption_session: KlapEncryptionSession | None = None
self._session_expire_at: float | None = None
self._session_cookie: dict[str, Any] | None = None
_LOGGER.debug("Created KLAP transport for %s", self._host)
self._app_url = URL(f"http://{self._host}:{self._port}/app")
self._request_url = self._app_url / "request"
@property
def default_port(self):
"""Default port for the transport."""
return self.DEFAULT_PORT
@property
def credentials_hash(self) -> str:
"""The hashed credentials used by the transport."""
return base64.b64encode(self._local_auth_hash).decode()
def _handshake_session_expired(self):
"""Return true if session has expired."""
return (
self._session_expire_at is None
or self._session_expire_at - time.time() <= 0
)
[docs] async def send(self, request: str):
"""Send the request."""
if not self._handshake_done or self._handshake_session_expired():
await self.perform_handshake()
# Check for mypy
if self._encryption_session is not None:
payload, seq = self._encryption_session.encrypt(request.encode())
response_status, response_data = await self._http_client.post(
self._request_url,
params={"seq": seq},
data=payload,
cookies_dict=self._session_cookie,
)
msg = (
f"Host is {self._host}, "
+ f"Sequence is {seq}, "
+ f"Response status is {response_status}, Request was {request}"
)
if response_status != 200:
_LOGGER.error("Query failed after successful authentication " + msg)
# If we failed with a security error, force a new handshake next time.
if response_status == 403:
self._handshake_done = False
raise _RetryableError(
f"Got a security error from {self._host} after handshake "
+ "completed"
)
else:
raise KasaException(
f"Device {self._host} responded with {response_status} to"
+ f"request with seq {seq}"
)
else:
_LOGGER.debug("Query posted " + msg)
# Check for mypy
if self._encryption_session is not None:
decrypted_response = self._encryption_session.decrypt(response_data)
json_payload = json_loads(decrypted_response)
_LOGGER.debug(
"%s << %s",
self._host,
_LOGGER.isEnabledFor(logging.DEBUG) and pf(json_payload),
)
return json_payload
[docs] async def close(self) -> None:
"""Close the http client and reset internal state."""
await self.reset()
await self._http_client.close()
[docs] async def reset(self) -> None:
"""Reset internal handshake state."""
self._handshake_done = False
[docs] @staticmethod
def generate_auth_hash(creds: Credentials):
"""Generate an md5 auth hash for the protocol on the supplied credentials."""
un = creds.username
pw = creds.password
return md5(md5(un.encode()) + md5(pw.encode()))
[docs] @staticmethod
def handshake1_seed_auth_hash(
local_seed: bytes, remote_seed: bytes, auth_hash: bytes
):
"""Generate an md5 auth hash for the protocol on the supplied credentials."""
return _sha256(local_seed + auth_hash)
[docs] @staticmethod
def handshake2_seed_auth_hash(
local_seed: bytes, remote_seed: bytes, auth_hash: bytes
):
"""Generate an md5 auth hash for the protocol on the supplied credentials."""
return _sha256(remote_seed + auth_hash)
[docs] @staticmethod
def generate_owner_hash(creds: Credentials):
"""Return the MD5 hash of the username in this object."""
un = creds.username
return md5(un.encode())
[docs]class KlapTransportV2(KlapTransport):
"""Implementation of the KLAP encryption protocol with v2 hanshake hashes."""
[docs] @staticmethod
def generate_auth_hash(creds: Credentials):
"""Generate an md5 auth hash for the protocol on the supplied credentials."""
un = creds.username
pw = creds.password
return _sha256(_sha1(un.encode()) + _sha1(pw.encode()))
[docs] @staticmethod
def handshake1_seed_auth_hash(
local_seed: bytes, remote_seed: bytes, auth_hash: bytes
):
"""Generate an md5 auth hash for the protocol on the supplied credentials."""
return _sha256(local_seed + remote_seed + auth_hash)
[docs] @staticmethod
def handshake2_seed_auth_hash(
local_seed: bytes, remote_seed: bytes, auth_hash: bytes
):
"""Generate an md5 auth hash for the protocol on the supplied credentials."""
return _sha256(remote_seed + local_seed + auth_hash)
class KlapEncryptionSession:
"""Class to represent an encryption session and it's internal state.
i.e. sequence number which the device expects to increment.
"""
_cipher: Cipher
def __init__(self, local_seed, remote_seed, user_hash):
self.local_seed = local_seed
self.remote_seed = remote_seed
self.user_hash = user_hash
self._key = self._key_derive(local_seed, remote_seed, user_hash)
(self._iv, self._seq) = self._iv_derive(local_seed, remote_seed, user_hash)
self._aes = algorithms.AES(self._key)
self._sig = self._sig_derive(local_seed, remote_seed, user_hash)
def _key_derive(self, local_seed, remote_seed, user_hash):
payload = b"lsk" + local_seed + remote_seed + user_hash
return hashlib.sha256(payload).digest()[:16]
def _iv_derive(self, local_seed, remote_seed, user_hash):
# iv is first 16 bytes of sha256, where the last 4 bytes forms the
# sequence number used in requests and is incremented on each request
payload = b"iv" + local_seed + remote_seed + user_hash
fulliv = hashlib.sha256(payload).digest()
seq = int.from_bytes(fulliv[-4:], "big", signed=True)
return (fulliv[:12], seq)
def _sig_derive(self, local_seed, remote_seed, user_hash):
# used to create a hash with which to prefix each request
payload = b"ldk" + local_seed + remote_seed + user_hash
return hashlib.sha256(payload).digest()[:28]
def _generate_cipher(self):
iv_seq = self._iv + PACK_SIGNED_LONG(self._seq)
cbc = modes.CBC(iv_seq)
self._cipher = Cipher(self._aes, cbc)
def encrypt(self, msg):
"""Encrypt the data and increment the sequence number."""
self._seq += 1
self._generate_cipher()
if isinstance(msg, str):
msg = msg.encode("utf-8")
encryptor = self._cipher.encryptor()
padder = padding.PKCS7(128).padder()
padded_data = padder.update(msg) + padder.finalize()
ciphertext = encryptor.update(padded_data) + encryptor.finalize()
signature = hashlib.sha256(
self._sig + PACK_SIGNED_LONG(self._seq) + ciphertext
).digest()
return (signature + ciphertext, self._seq)
def decrypt(self, msg):
"""Decrypt the data."""
decryptor = self._cipher.decryptor()
dp = decryptor.update(msg[32:]) + decryptor.finalize()
unpadder = padding.PKCS7(128).unpadder()
plaintextbytes = unpadder.update(dp) + unpadder.finalize()
return plaintextbytes.decode()