-
-
Notifications
You must be signed in to change notification settings - Fork 274
Expand file tree
/
Copy pathiotprotocol.py
More file actions
executable file
·171 lines (146 loc) · 5.52 KB
/
iotprotocol.py
File metadata and controls
executable file
·171 lines (146 loc) · 5.52 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
"""Module for the IOT legacy IOT KASA protocol."""
from __future__ import annotations
import asyncio
import logging
from collections.abc import Callable
from pprint import pformat as pf
from typing import TYPE_CHECKING, Any
from ..deviceconfig import DeviceConfig
from ..exceptions import (
AuthenticationError,
KasaException,
TimeoutError,
_ConnectionError,
_RetryableError,
)
from ..json import dumps as json_dumps
from ..transports import XorEncryption, XorTransport
from .protocol import BaseProtocol, mask_mac, redact_data
if TYPE_CHECKING:
from ..transports import BaseTransport
_LOGGER = logging.getLogger(__name__)
REDACTORS: dict[str, Callable[[Any], Any] | None] = {
"latitude": lambda x: 0,
"longitude": lambda x: 0,
"latitude_i": lambda x: 0,
"longitude_i": lambda x: 0,
"deviceId": lambda x: "REDACTED_" + x[9::],
"id": lambda x: "REDACTED_" + x[9::],
"alias": lambda x: "#MASKED_NAME#" if x else "",
"mac": mask_mac,
"mic_mac": mask_mac,
"ssid": lambda x: "#MASKED_SSID#" if x else "",
"oemId": lambda x: "REDACTED_" + x[9::],
}
class IotProtocol(BaseProtocol):
"""Class for the legacy TPLink IOT KASA Protocol."""
BACKOFF_SECONDS_AFTER_TIMEOUT = 1
def __init__(
self,
*,
transport: BaseTransport,
) -> None:
"""Create a protocol object."""
super().__init__(transport=transport)
self._query_lock = asyncio.Lock()
self._redact_data = True
async def query(self, request: str | dict, retry_count: int = 3) -> dict:
"""Query the device retrying for retry_count on failure."""
if isinstance(request, dict):
request = json_dumps(request)
assert isinstance(request, str) # noqa: S101
async with self._query_lock:
return await self._query(request, retry_count)
async def _query(self, request: str, retry_count: int = 3) -> dict:
for retry in range(retry_count + 1):
try:
return await self._execute_query(request, retry)
except _ConnectionError as sdex:
if retry >= retry_count:
_LOGGER.debug("Giving up on %s after %s retries", self._host, retry)
raise sdex
continue
except AuthenticationError as auex:
await self._transport.reset()
_LOGGER.debug(
"Unable to authenticate with %s, not retrying", self._host
)
raise auex
except _RetryableError as ex:
await self._transport.reset()
if retry >= retry_count:
_LOGGER.debug("Giving up on %s after %s retries", self._host, retry)
raise ex
continue
except TimeoutError as ex:
await self._transport.reset()
if retry >= retry_count:
_LOGGER.debug("Giving up on %s after %s retries", self._host, retry)
raise ex
await asyncio.sleep(self.BACKOFF_SECONDS_AFTER_TIMEOUT)
continue
except KasaException as ex:
await self._transport.reset()
_LOGGER.debug(
"Unable to query the device: %s, not retrying: %s",
self._host,
ex,
)
raise ex
# make mypy happy, this should never be reached..
raise KasaException("Query reached somehow to unreachable")
async def _execute_query(self, request: str, retry_count: int) -> dict:
debug_enabled = _LOGGER.isEnabledFor(logging.DEBUG)
if debug_enabled:
_LOGGER.debug(
"%s >> %s",
self._host,
request,
)
resp = await self._transport.send(request)
if debug_enabled:
data = redact_data(resp, REDACTORS) if self._redact_data else resp
_LOGGER.debug(
"%s << %s",
self._host,
pf(data),
)
return resp
async def close(self) -> None:
"""Close the underlying transport."""
await self._transport.close()
class _deprecated_TPLinkSmartHomeProtocol(IotProtocol):
def __init__(
self,
host: str | None = None,
*,
port: int | None = None,
timeout: int | None = None,
transport: BaseTransport | None = None,
) -> None:
"""Create a protocol object."""
if not host and not transport:
raise KasaException("host or transport must be supplied")
if not transport:
config = DeviceConfig(
host=host, # type: ignore[arg-type]
port_override=port,
timeout=timeout or XorTransport.DEFAULT_TIMEOUT,
)
transport = XorTransport(config=config)
super().__init__(transport=transport)
@staticmethod
def encrypt(request: str) -> bytes:
"""Encrypt a request for a TP-Link Smart Home Device.
:param request: plaintext request data
:return: ciphertext to be send over wire, in bytes
"""
return XorEncryption.encrypt(request)
@staticmethod
def decrypt(ciphertext: bytes) -> str:
"""Decrypt a response of a TP-Link Smart Home Device.
:param ciphertext: encrypted response data
:return: plaintext response
"""
return XorEncryption.decrypt(ciphertext)