diff --git a/setup.py b/setup.py index eb1761a..cdeae41 100644 --- a/setup.py +++ b/setup.py @@ -12,6 +12,9 @@ author='Dev Piekstra', author_email='piekstra.dev@gmail.com', packages=find_packages(exclude=("tests",)), + package_data={ + 'tplinkcloud': ['certs/*.pem'], + }, version=__version__, description='Python library for communicating with the TP-Link Cloud API to manage TP-Link Kasa Smart Home devices', long_description=long_description, @@ -21,7 +24,7 @@ 'aiohttp>=3,<4' ], url='https://github.com/piekstra/tplink-cloud-api', - python_requires='>=3.7', + python_requires='>=3.10', zip_safe=False, license='GPL-3', classifiers=[ diff --git a/tplinkcloud/__init__.py b/tplinkcloud/__init__.py index 0918b8a..40300df 100644 --- a/tplinkcloud/__init__.py +++ b/tplinkcloud/__init__.py @@ -1,11 +1,23 @@ from .device_manager import TPLinkDeviceManager from .device_manager_power_tools import TPLinkDeviceManagerPowerTools from .device_schedule_rule_builder import TPLinkDeviceScheduleRuleBuilder +from .exceptions import ( + TPLinkAuthError, + TPLinkCloudError, + TPLinkDeviceOfflineError, + TPLinkMFARequiredError, + TPLinkTokenExpiredError, +) __all__ = [ 'TPLinkDeviceManager', 'TPLinkDeviceManagerPowerTools', 'TPLinkDeviceScheduleRuleBuilder', + 'TPLinkAuthError', + 'TPLinkCloudError', + 'TPLinkDeviceOfflineError', + 'TPLinkMFARequiredError', + 'TPLinkTokenExpiredError', ] # Windows OS-specific HACK to silence exception thrown on event loop being closed diff --git a/tplinkcloud/api_response.py b/tplinkcloud/api_response.py index 985d92f..7c336bb 100644 --- a/tplinkcloud/api_response.py +++ b/tplinkcloud/api_response.py @@ -6,6 +6,14 @@ def __init__(self, response): def successful(self): return self._response.get('error_code') == 0 + @property + def error_code(self): + return self._response.get('error_code') + @property def result(self): return self._response.get('result') + + @property + def msg(self): + return self._response.get('msg') diff --git a/tplinkcloud/certs/__init__.py b/tplinkcloud/certs/__init__.py new file mode 100644 index 0000000..8fcd5b9 --- /dev/null +++ b/tplinkcloud/certs/__init__.py @@ -0,0 +1,19 @@ +"""TP-Link CA certificate chain for V2 API SSL verification.""" + +from importlib import resources + + +def get_ca_cert_path() -> str: + """Get the filesystem path to the bundled TP-Link CA certificate chain. + + The V2 API servers (n-*.tplinkcloud.com) use TP-Link's private CA, + which is not in the system trust store. This chain includes: + - Root: tp-link-CA + - Intermediate: TP-LINK CA P1 + - Leaf: *.tplinkcloud.com + """ + ref = resources.files(__package__) / "tplink-ca-chain.pem" + # resources.as_file extracts to a temp location if needed (e.g. from a zip) + # but for a regular install the path is stable + with resources.as_file(ref) as path: + return str(path) diff --git a/tplinkcloud/certs/tplink-ca-chain.pem b/tplinkcloud/certs/tplink-ca-chain.pem new file mode 100644 index 0000000..c309c3c --- /dev/null +++ b/tplinkcloud/certs/tplink-ca-chain.pem @@ -0,0 +1,70 @@ +-----BEGIN CERTIFICATE----- +MIIEfTCCA2WgAwIBAgITZAAAAVRNT7bhOPHCcwABAAABVDANBgkqhkiG9w0BAQsF +ADBaMRIwEAYKCZImiZPyLGQBGRYCY24xEzARBgoJkiaJk/IsZAEZFgNjb20xFzAV +BgoJkiaJk/IsZAEZFgd0cC1saW5rMRYwFAYDVQQDEw1UUC1MSU5LIENBIFAxMB4X +DTI1MTAyODAwMjY1MVoXDTI2MTAyODAwMjY1MVowWDELMAkGA1UEBhMCVVMxDzAN +BgNVBAcTBklydmluZTEcMBoGA1UEChMTVFAtTElOSyBHTE9CQUwgSU5DLjEaMBgG +A1UEAwwRKi50cGxpbmtjbG91ZC5jb20wggEiMA0GCSqGSIb3DQEBAQUAA4IBDwAw +ggEKAoIBAQCtdXbGKixzsYaEjaqs7TM/zUxRx9RcK/sKXJxUjZuIfhzo4KJ51d6D +OTkXdgcYdDdYh+1jpcetitHw/Jd8rUT4xdvpBVSEVFnK1x97ADaPrxd+o7LGpAi8 +dlvPWO3d3zw4YdOTNVhcajCbuJ5sKV0HGSBzSPzYsqpghmwFB3Kfjtq/e4q1NVZw +WWFgzq3iZ2rmPYACrn/MoVRQlNzFdOZ/X0SrvjJbLTWbQ5TuUBpJEL21BEKHn0F+ +Yo3Z+QPGRGiCV574xMgiGhMlkxGScS4y809HTvIz4AFbOOy6M2yJmS2x+fIkt//J +wMEkydfMcAhaz0JrevqVPEsrz3zaco+5AgMBAAGjggE8MIIBODAOBgNVHQ8BAf8E +BAMCBaAwEwYDVR0lBAwwCgYIKwYBBQUHAwEwLQYDVR0RBCYwJIIRKi50cGxpbmtj +bG91ZC5jb22CD3RwbGlua2Nsb3VkLmNvbTAdBgNVHQ4EFgQUp67HKroMq2OsBpcm +Ne64r49PjQ8wHwYDVR0jBBgwFoAUVQi1z4hvsNPAlOf6v6KN9JRW0l4wRwYDVR0f +BEAwPjA8oDqgOIY2aHR0cDovL3RwY3JsLnRwLWxpbmsuY29tLmNuL2NlcnRkYXRh +L1RQLUxJTktfQ0FfUDEuY3JsMDwGCSsGAQQBgjcVBwQvMC0GJSsGAQQBgjcVCIap +mz6E6u1+hsWVAbeERIfEoHBHhcD9C4XOlVMCAWQCAQswGwYJKwYBBAGCNxUKBA4w +DDAKBggrBgEFBQcDATANBgkqhkiG9w0BAQsFAAOCAQEAr4tMuesR2EoVFkZryo5U +49aOZ6mbLr0YTtRx3ZFJehXEPtzUAK2yDyXHLtiUQ1e2uBzJnqNB6Qi6lEISL9fu +nvd+xcZvE4H40Woj0TwT9GJwbHIrCu+1QmtJYCTWOjkjTb+kN8ueEHlTGaCn5W9B +2oQ0M0+MC1iKjAjYLJrnXnO63KVhfKFaCsys2tY7A9COA9y/JSX2JAdMoFQrygM6 +X9lydmEhrzQT/n8uk6hiRDdeFm6jXEejQZ1GvFNwZDVIgUzlp8W9qQm3x6IDRTx7 +ZE/Fwq/Ixk8WKwfPlJOhNsqKguwFPXe6S0pEMbOGX067yObBarULC4CCW+GRrdHJ +EA== +-----END CERTIFICATE----- +-----BEGIN CERTIFICATE----- +MIID+DCCAuCgAwIBAgITdAAAAAdZrje7ktFgcAAAAAAABzANBgkqhkiG9w0BAQsF +ADAVMRMwEQYDVQQDEwp0cC1saW5rLUNBMCAXDTIwMDUxOTA5MjEwMloYDzIwNjgw +MTE5MDgzNzUyWjBaMRIwEAYKCZImiZPyLGQBGRYCY24xEzARBgoJkiaJk/IsZAEZ +FgNjb20xFzAVBgoJkiaJk/IsZAEZFgd0cC1saW5rMRYwFAYDVQQDEw1UUC1MSU5L +IENBIFAxMIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEAv4DMrSRNtjjt +P+1swNtCCthsOCiyy4+v5fFilnP5GPFbTgiOjUA1SMmWqrhDCHG8ZC2HrCIm7/Oj +O41M7IHbr0HHwMOrm4rkeucvuCu1Hmrh47L0YK872aWTlMcqL2Lt2sJJpl7kZhH0 +LVWlYPQSXRgOZE6I+UsgUfksf4H7lH1mReYjTa8uzQJzN14IrCUzMc6lAZxMR/fr +zQ2hXbn4P+hGDBtw2k8Zj5rZJm6ALTpe/JGiR04ocOqRhldJYYoijhfqcpp7racT +tbo4wrz4dpfmZQv1K7ZOkAi79lEjTFE6OiSMJAkxL3ObOB3h7qkcjlb9xXgpGWJQ +ylLhFNHnzQIDAQABo4H5MIH2MBAGCSsGAQQBgjcVAQQDAgEBMCMGCSsGAQQBgjcV +AgQWBBRskYoxJ4QIQK+6xIr4syb5p3hJ+zAdBgNVHQ4EFgQUVQi1z4hvsNPAlOf6 +v6KN9JRW0l4wGQYJKwYBBAGCNxQCBAweCgBTAHUAYgBDAEEwCwYDVR0PBAQDAgGG +MA8GA1UdEwEB/wQFMAMBAf8wHwYDVR0jBBgwFoAUxu2iBRTsef5iNnsADVhMJDQW +i6kwRAYDVR0fBD0wOzA5oDegNYYzaHR0cDovL3RwY3JsLnRwLWxpbmsuY29tLmNu +L2NlcnRkYXRhL3RwLWxpbmstQ0EuY3JsMA0GCSqGSIb3DQEBCwUAA4IBAQArzgXJ +ZORRDC3hSu0zoAWWn4bUaXy+1K2fbkkfceJc1yQUnx7WhVOtPiuCFYp5qraSUVHk +vFVFEUY2Jhdk1uP7lXQuU81XuCisMA2VoNcOL9blVyturewyN8YWmWhptjeYKEPy +SEUUTTkKjqdsWFwxqyDrwom70Ev+TOZfyF8O8IQCoj1Lv0wcEE1B+HGdVboitJUA +f4+tJg8bqahrZrFMSTVttuRxkIYupPlhFhY880/ptuUy2taRkhDP2717dUdB9KsQ +/xeRvYXgPbK1o4esNjmftMcziWanTLTT/DwQv/8j8J32gGojNtuX8NzYfqx8gxMQ +RUH9G1c+dc3etM0K +-----END CERTIFICATE----- +-----BEGIN CERTIFICATE----- +MIIDBzCCAe+gAwIBAgIQT5x0ma7QnINHCQvhnmzR9zANBgkqhkiG9w0BAQsFADAV +MRMwEQYDVQQDEwp0cC1saW5rLUNBMCAXDTE4MDExOTA4Mjc1MloYDzIwNjgwMTE5 +MDgzNzUyWjAVMRMwEQYDVQQDEwp0cC1saW5rLUNBMIIBIjANBgkqhkiG9w0BAQEF +AAOCAQ8AMIIBCgKCAQEAuGG8n5zEUN1j5wuvUz4pAIMurhKHbpfUUu+b2acFHKS6 +iU9hNJWvDyhXcihY5Wz6aq9m4D5SZcgW3k31YoNNtrztDjdg2qw7AaX85S99/G0B +VbIXktrhs34OW19WA/haDwut3dFhLem+gCRRKUXcmuqchZc84dY7JFVfhPcJci4m +sRjLCFNO0ho9OX+MZwfO4BLaeAqKVoAor6rf4BXVtO0xjYHDKO0fb3AWLLJ4EjGe +q6YieqPiYlPFEqRm5PrvBXTm0IuQogygyVpK4LHr/K207ZLyV33DxLLbsUgSEJVn +pZUv/WUujXjlIDgxIvyZZCYiXO3dle2/MEvpmZk6JQIDAQABo1EwTzALBgNVHQ8E +BAMCAYYwDwYDVR0TAQH/BAUwAwEB/zAdBgNVHQ4EFgQUxu2iBRTsef5iNnsADVhM +JDQWi6kwEAYJKwYBBAGCNxUBBAMCAQAwDQYJKoZIhvcNAQELBQADggEBAB52Majd ++wo3cb5BsTo63z2Psbbyl4ACMUaw68NxUMy61Oihx3mcLzLJqiIZcKePiHskLqLJ +F7QfT9TqjvizMjFJVgsLuVubUBXKBzqyN+3KKlQci0PO3mH+ObhyaE7BzV+qrS3P +dVTgsCWFv8DkgLTRudSWxL7VwVoedc7lRz5EroGgJ33nRGCR0ngcW919tLTARDQO +pULmzulcdWeZgG+0PLX0xjJQIjFEvbOxR1Z+gxMupBz0rWFokmWYrcga8eWiWzjQ +Ia3/ASBVJ69srV77trWlfLumkChbXk9i64NXBKnce0Jmll0Y9OC1nMPqrbQKnzcn +dSAA4fejD/qMQn0= +-----END CERTIFICATE----- diff --git a/tplinkcloud/client.py b/tplinkcloud/client.py index 57ea5ad..63098b0 100644 --- a/tplinkcloud/client.py +++ b/tplinkcloud/client.py @@ -1,48 +1,150 @@ -import requests +"""Synchronous HTTP client for TP-Link Cloud API authentication and device listing. + +Supports both V1 (legacy) and V2 (current) API protocols: + +V1: POST https://wap.tplinkcloud.com/?appName=Kasa_Android&... + Body: {"method": "login", "params": {"cloudUserName": "...", ...}} + +V2: POST https://n-wap.tplinkcloud.com/api/v2/account/login?appName=Kasa_Android_Mix&... + Body: {"cloudUserName": "...", "cloudPassword": "...", ...} (flat, no wrapper) + Headers: X-Authorization (HMAC-SHA1 signature), Content-MD5 +""" + import json import uuid +import requests + from .api_response import TPLinkApiResponse +from .certs import get_ca_cert_path +from .exceptions import ( + TPLinkAuthError, + TPLinkCloudError, + TPLinkMFARequiredError, + TPLinkTokenExpiredError, +) +from .signing import get_signing_headers + +# V2 API error codes +_ERR_MFA_REQUIRED = -20677 +_ERR_TOKEN_EXPIRED = -20651 +_ERR_REFRESH_TOKEN_EXPIRED = -20655 +_ERR_WRONG_CREDENTIALS = -20601 +_ERR_ACCOUNT_LOCKED = -20675 + +# Default API hosts +_V1_HOST = "https://wap.tplinkcloud.com" +_V2_HOST = "https://n-wap.tplinkcloud.com" + +# V2 API paths +_PATH_ACCOUNT_STATUS = "/api/v2/account/getAccountStatusAndUrl" +_PATH_LOGIN = "/api/v2/account/login" +_PATH_REFRESH_TOKEN = "/api/v2/account/refreshToken" +_PATH_MFA_LOGIN = "/api/v2/account/checkMFACodeAndLogin" class TPLinkApi: def __init__(self, host=None, verbose=False, term_id=None): - self.host = host if host else 'https://wap.tplinkcloud.com' self._verbose = verbose - self._termId = term_id if term_id else str(uuid.uuid4()) - self._default_params = { - 'appName': 'Kasa_Android', - 'termID': self._termId, - 'appVer': '1.4.4.607', - 'ospf': 'Android+6.0.1', - 'netType': 'wifi', - 'locale': 'es_ES' + self._term_id = term_id or str(uuid.uuid4()) + self._ca_cert_path = get_ca_cert_path() + + # V2 is the default; V1 host provided for backward compatibility + self.host = host or _V2_HOST + + # V2 query parameters (sent on all requests, matching C29914q interceptor) + self._query_params = { + "appName": "Kasa_Android_Mix", + "appVer": "3.4.451", + "netType": "wifi", + "termID": self._term_id, + "ospf": "Android 14", + "brand": "TPLINK", + "locale": "en_US", + "model": "Pixel", + "termName": "Pixel", + "termMeta": "Pixel", } + self._headers = { - 'User-Agent': - 'Dalvik/2.1.0 (Linux; U; Android 6.0.1; A0001 Build/M4B30X)', - 'Content-Type': 'application/json' + "User-Agent": "Dalvik/2.1.0 (Linux; U; Android 14; Pixel Build/UP1A)", + "Content-Type": "application/json;charset=UTF-8", } - def _request_post(self, body, token=None): + def _request_post_v2(self, base_url, url_path, body, token=None): + """Make a signed V2 API request. + + Args: + base_url: The base URL (e.g. "https://n-use1-wap.tplinkcloud.com"). + url_path: The API path (e.g. "/api/v2/account/login"). + body: The request body dict (flat format, no method/params wrapper). + token: Optional auth token to include in query params. + + Returns: + TPLinkApiResponse + """ + url = f"{base_url}{url_path}" + body_json = json.dumps(body) + + params = self._query_params.copy() + if token: + params["token"] = token + + signing_headers = get_signing_headers(body_json, url_path) + headers = {**self._headers, **signing_headers} + if self._verbose: - print('POST', self.host, body) + print(f"POST {url}") + print(f"Body: {body_json}") + + response = requests.post( + url, + data=body_json, + params=params, + headers=headers, + verify=self._ca_cert_path, + timeout=15, + ) + if response.status_code == 200: + response_json = response.json() + if self._verbose: + print(json.dumps(response_json, indent=2)) + return TPLinkApiResponse(response_json) + + if response.content: + raise TPLinkCloudError( + f"{response.status_code}: {response.reason}: {response.content!r}" + ) + raise TPLinkCloudError(f"{response.status_code}: {response.reason}") + + def _request_post_v1(self, body, token=None): + """Make a V1-style request (method/params wrapper) with V2 signing. + + Device operations still use the V1 JSON format on the root path, + but with V2 signing headers and query parameters. + """ + url_path = "/" body_json = json.dumps(body) + params = self._query_params.copy() if token: - params = self._default_params.copy() - params['token'] = token - else: - params = self._default_params - - s = requests.Session() - response = s.request( - 'POST', + params["token"] = token + + signing_headers = get_signing_headers(body_json, url_path) + headers = {**self._headers, **signing_headers} + + if self._verbose: + print(f"POST {self.host}/") + print(f"Body: {body_json}") + + response = requests.post( self.host, data=body_json, params=params, - headers=self._headers + headers=headers, + verify=self._ca_cert_path, + timeout=15, ) if response.status_code == 200: @@ -50,40 +152,183 @@ def _request_post(self, body, token=None): if self._verbose: print(json.dumps(response_json, indent=2)) return TPLinkApiResponse(response_json) - elif response.content: - raise Exception(str(response.status_code) + ': ' + - response.reason + ': ' + str(response.content)) - else: - raise Exception(str(response.status_code) + ': ' + response.reason) - - # Returns a token if properly authenticated - def login(self, username, password): + + if response.content: + raise TPLinkCloudError( + f"{response.status_code}: {response.reason}: {response.content!r}" + ) + raise TPLinkCloudError(f"{response.status_code}: {response.reason}") + + def _get_regional_url(self, username): + """Discover the regional API server URL for the given account. + + Returns: + The regional appServerUrl string. + """ + body = { + "appType": "Kasa_Android_Mix", + "cloudUserName": username, + } + response = self._request_post_v2( + self.host, _PATH_ACCOUNT_STATUS, body + ) + if response.successful: + return response.result.get("appServerUrl", self.host) + + return self.host + + def login(self, username, password, mfa_callback=None): + """Authenticate with the TP-Link Cloud V2 API. + + Flow: + 1. getAccountStatusAndUrl -> regional URL + 2. login on regional URL -> token (or MFA challenge) + 3. If MFA required and callback provided, handle MFA + + Args: + username: TP-Link / Kasa account email. + password: Account password. + mfa_callback: Optional callable(mfa_type, email) -> str that returns + the MFA verification code. If MFA is required and no + callback is provided, raises TPLinkMFARequiredError. + + Returns: + Dict with 'token' and optionally 'refreshToken'. + + Raises: + TPLinkAuthError: Wrong credentials or account locked. + TPLinkMFARequiredError: MFA required but no callback provided. + """ if not username: raise ValueError("Cannot login, username is not set") if not password: raise ValueError("Cannot login, password not set") + + # Step 1: Discover regional URL + regional_url = self._get_regional_url(username) + self.host = regional_url + + # Step 2: Login + login_body = { + "appType": "Kasa_Android_Mix", + "appVersion": "3.4.451", + "cloudPassword": password, + "cloudUserName": username, + "platform": "Android", + "refreshTokenNeeded": True, + "supportBindAccount": False, + "terminalUUID": self._term_id, + "terminalName": "Pixel", + "terminalMeta": "Pixel", + } + + response = self._request_post_v2( + regional_url, _PATH_LOGIN, login_body + ) + + error_code = response.error_code + if error_code == 0: + return response.result + + # Handle specific error codes + if error_code == _ERR_MFA_REQUIRED: + if mfa_callback is None: + raise TPLinkMFARequiredError( + "MFA verification required. Provide an mfa_callback.", + error_code=error_code, + mfa_type=response.result.get("mfaType") if response.result else None, + email=username, + ) + # Get MFA code from callback and verify + mfa_type = response.result.get("mfaType", "verifyCodeLogin") if response.result else "verifyCodeLogin" + mfa_code = mfa_callback(mfa_type, username) + return self._verify_mfa(regional_url, username, password, mfa_code) + + if error_code in (_ERR_WRONG_CREDENTIALS, _ERR_ACCOUNT_LOCKED): + raise TPLinkAuthError( + response.msg or "Authentication failed", + error_code=error_code, + ) + + raise TPLinkCloudError( + response.msg or f"Login failed with error code {error_code}", + error_code=error_code, + ) + + def _verify_mfa(self, regional_url, username, password, mfa_code): + """Complete MFA verification. + + Returns: + Dict with 'token' and optionally 'refreshToken'. + """ body = { - 'method': 'login', - 'url': self.host, - 'params': { - 'appType': 'Kasa_Android', - 'cloudUserName': username, - 'cloudPassword': password, - 'terminalUUID': self._termId - } + "appType": "Kasa_Android_Mix", + "cloudPassword": password, + "cloudUserName": username, + "code": mfa_code, + "terminalUUID": self._term_id, } - response = self._request_post(body) + response = self._request_post_v2( + regional_url, _PATH_MFA_LOGIN, body + ) if response.successful: - return response.result.get('token') + return response.result - return None + raise TPLinkAuthError( + response.msg or "MFA verification failed", + error_code=response.error_code, + ) + + def refresh_login(self, refresh_token): + """Refresh an expired auth token using a refresh token. + + Args: + refresh_token: The refresh token from a previous login. + + Returns: + Dict with new 'token' and 'refreshToken'. + + Raises: + TPLinkTokenExpiredError: If the refresh token itself has expired. + """ + body = { + "appType": "Kasa_Android_Mix", + "refreshToken": refresh_token, + "terminalUUID": self._term_id, + } + response = self._request_post_v2( + self.host, _PATH_REFRESH_TOKEN, body + ) + if response.successful: + return response.result + + if response.error_code == _ERR_REFRESH_TOKEN_EXPIRED: + raise TPLinkTokenExpiredError( + "Refresh token has expired. Full re-login required.", + error_code=response.error_code, + ) + + raise TPLinkCloudError( + response.msg or f"Token refresh failed with error code {response.error_code}", + error_code=response.error_code, + ) def get_device_info_list(self, token): + """Get the list of devices registered to the account. + + Uses V1-style request format with V2 signing. + """ body = { - 'method': 'getDeviceList' + "method": "getDeviceList", } - response = self._request_post(body, token) + response = self._request_post_v1(body, token) if response.successful: - return response.result.get('deviceList') + return response.result.get("deviceList", []) + + if response.error_code == _ERR_TOKEN_EXPIRED: + raise TPLinkTokenExpiredError( + "Auth token expired", + error_code=response.error_code, + ) return [] diff --git a/tplinkcloud/device_client.py b/tplinkcloud/device_client.py index e62444c..d5f1f4b 100644 --- a/tplinkcloud/device_client.py +++ b/tplinkcloud/device_client.py @@ -1,45 +1,58 @@ -import asyncio import aiohttp import json import uuid from .api_response import TPLinkApiResponse +from .certs import get_ca_cert_path +from .signing import get_signing_headers +import ssl class TPLinkDeviceClient: def __init__(self, host, token, verbose=False, term_id=None): self.host = host self._verbose = verbose - self._termId = term_id if term_id else str(uuid.uuid4()) + self._term_id = term_id or str(uuid.uuid4()) + self._params = { - 'appName': 'Kasa_Android', - 'termID': self._termId, - 'appVer': '1.4.4.607', - 'ospf': 'Android+6.0.1', - 'netType': 'wifi', - 'locale': 'es_ES', - 'token': token + "appName": "Kasa_Android_Mix", + "appVer": "3.4.451", + "netType": "wifi", + "termID": self._term_id, + "ospf": "Android 14", + "brand": "TPLINK", + "locale": "en_US", + "model": "Pixel", + "termName": "Pixel", + "termMeta": "Pixel", + "token": token, } self._headers = { - "cache-control": "no-cache", - 'User-Agent': - 'Dalvik/2.1.0 (Linux; U; Android 6.0.1; A0001 Build/M4B30X)', - 'Content-Type': 'application/json' + "User-Agent": "Dalvik/2.1.0 (Linux; U; Android 14; Pixel Build/UP1A)", + "Content-Type": "application/json;charset=UTF-8", } + # Build SSL context with TP-Link's private CA + self._ssl_context = ssl.create_default_context(cafile=get_ca_cert_path()) + async def _request_post(self, body): if self._verbose: print('POST', self.host, body) + url_path = "/" body_json = json.dumps(body) + signing_headers = get_signing_headers(body_json, url_path) + headers = {**self._headers, **signing_headers} + async with aiohttp.ClientSession() as session: async with session.post( self.host, data=body_json, params=self._params, - headers=self._headers, - timeout=600 + headers=headers, + ssl=self._ssl_context, + timeout=aiohttp.ClientTimeout(total=600), ) as response: if response.status == 200: response_json = await response.json(content_type=None) @@ -51,7 +64,7 @@ async def _request_post(self, body): response.reason + ': ' + str(response.content)) else: raise Exception(str(response.status) + ': ' + response.reason) - + async def pass_through_request(self, device_id, request_data): body = { 'method': 'passthrough', diff --git a/tplinkcloud/device_manager.py b/tplinkcloud/device_manager.py index 80e49a9..0a7ff70 100644 --- a/tplinkcloud/device_manager.py +++ b/tplinkcloud/device_manager.py @@ -3,6 +3,7 @@ from .device_info import TPLinkDeviceInfo from .device_client import TPLinkDeviceClient from .client import TPLinkApi +from .exceptions import TPLinkTokenExpiredError # Supported devices from .hs100 import HS100 @@ -27,19 +28,24 @@ def __init__( cache_devices=True, tplink_cloud_api_host=None, verbose=False, - term_id=None + term_id=None, + mfa_callback=None, ): self._verbose = verbose self._cache_devices = cache_devices self._cached_devices = None self._term_id = term_id self._auth_token = None + self._refresh_token = None + self._username = username + self._password = password + self._mfa_callback = mfa_callback self._tplink_api = TPLinkApi( tplink_cloud_api_host, verbose=self._verbose, term_id=self._term_id) if username and password: - self.login(username, password) + self.login(username, password, mfa_callback=mfa_callback) self._prefetch = prefetch async def async_init(self): @@ -55,8 +61,16 @@ async def get_devices(self): if self._cached_devices: return self._cached_devices - device_info_list = self._tplink_api.get_device_info_list( - self._auth_token) + try: + device_info_list = self._tplink_api.get_device_info_list( + self._auth_token) + except TPLinkTokenExpiredError: + if self._refresh_token: + self._do_refresh() + device_info_list = self._tplink_api.get_device_info_list( + self._auth_token) + else: + raise devices = [] children_gather_tasks = [] @@ -107,18 +121,37 @@ def _construct_device(self, device_info): else: return TPLinkDevice(client, tplink_device_info.device_id, tplink_device_info) - def login(self, username, password): - # Note that this token expires after some amount of time - # (Currently unkown what exactly that expiration time is) - auth_token = self._tplink_api.login(username, password) - self.set_auth_token(auth_token) - return auth_token + def login(self, username, password, mfa_callback=None): + result = self._tplink_api.login( + username, password, mfa_callback=mfa_callback + ) + if result: + self.set_auth_token(result.get('token')) + self._refresh_token = result.get('refreshToken') + return self._auth_token + + def _do_refresh(self): + """Refresh the auth token using the stored refresh token.""" + result = self._tplink_api.refresh_login(self._refresh_token) + if result: + self.set_auth_token(result.get('token')) + self._refresh_token = result.get('refreshToken') def set_auth_token(self, auth_token): - # this token is used for all requests to the TP-Link API - # where authentication is required self._auth_token = auth_token + def get_token(self): + """Get the current auth token.""" + return self._auth_token + + def set_refresh_token(self, refresh_token): + """Set the refresh token (e.g. when resuming a session).""" + self._refresh_token = refresh_token + + def get_refresh_token(self): + """Get the current refresh token.""" + return self._refresh_token + async def find_device(self, device_name): devices = await self.get_devices() # Just return the first match diff --git a/tplinkcloud/exceptions.py b/tplinkcloud/exceptions.py new file mode 100644 index 0000000..b3a90e0 --- /dev/null +++ b/tplinkcloud/exceptions.py @@ -0,0 +1,50 @@ +"""Custom exception classes for the TP-Link Cloud API. + +Error codes from the V2 API: + -20104 Parameter doesn't exist (malformed request) + -20601 Incorrect email or password + -20675 Account locked (too many failed attempts) + -20677 MFA code required + -20651 Token expired + -20655 Refresh token expired +""" + + +class TPLinkCloudError(Exception): + """Base exception for TP-Link Cloud API errors.""" + + def __init__(self, message: str, error_code: int | None = None): + self.error_code = error_code + super().__init__(message) + + +class TPLinkAuthError(TPLinkCloudError): + """Authentication failed (wrong credentials, account locked, etc.).""" + + +class TPLinkMFARequiredError(TPLinkCloudError): + """MFA verification code is required to complete login. + + Attributes: + mfa_type: The type of MFA (e.g. "verifyCodeLogin"). + email: The email address the code was sent to. + """ + + def __init__( + self, + message: str, + error_code: int | None = None, + mfa_type: str | None = None, + email: str | None = None, + ): + self.mfa_type = mfa_type + self.email = email + super().__init__(message, error_code) + + +class TPLinkTokenExpiredError(TPLinkCloudError): + """The auth token or refresh token has expired.""" + + +class TPLinkDeviceOfflineError(TPLinkCloudError): + """The target device is offline or unreachable.""" diff --git a/tplinkcloud/signing.py b/tplinkcloud/signing.py new file mode 100644 index 0000000..3c60e62 --- /dev/null +++ b/tplinkcloud/signing.py @@ -0,0 +1,81 @@ +"""HMAC-SHA1 request signing for TP-Link Cloud API V2. + +The V2 API requires all requests to be signed with HMAC-SHA1. +The signing protocol uses an AccessKey/SecretKey pair that identifies +the client application (not the user). + +Signature format: + sig_string = "{content_md5}\\n{timestamp}\\n{nonce}\\n{url_path}" + signature = HMAC-SHA1(secret_key, sig_string) + +The signature is sent via the X-Authorization header: + X-Authorization: Timestamp={ts}, Nonce={nonce}, AccessKey={ak}, Signature={sig} +""" + +import base64 +import hashlib +import hmac +import uuid + + +# App-level keys from Kasa Android APK (identify the app, not the user) +ACCESS_KEY = "e37525375f8845999bcc56d5e6faa76d" +SECRET_KEY = "314bc6700b3140ca80bc655e527cb062" + +# The Kasa app uses a hardcoded timestamp for signing +SIGNING_TIMESTAMP = "9999999999" + + +def compute_content_md5(body: str) -> str: + """Compute Base64-encoded MD5 hash of the request body.""" + return base64.b64encode( + hashlib.md5(body.encode()).digest() + ).decode() + + +def compute_signature(body_json: str, url_path: str) -> tuple[str, str]: + """Compute HMAC-SHA1 signature for a V2 API request. + + Args: + body_json: The JSON-serialized request body. + url_path: The URL path (e.g. "/api/v2/account/login"). Must not + include query parameters. + + Returns: + A tuple of (content_md5, x_authorization_header). + """ + content_md5 = compute_content_md5(body_json) + nonce = str(uuid.uuid4()) + + sig_string = f"{content_md5}\n{SIGNING_TIMESTAMP}\n{nonce}\n{url_path}" + signature = hmac.new( + SECRET_KEY.encode(), + sig_string.encode(), + hashlib.sha1, + ).hexdigest() + + authorization = ( + f"Timestamp={SIGNING_TIMESTAMP}, " + f"Nonce={nonce}, " + f"AccessKey={ACCESS_KEY}, " + f"Signature={signature}" + ) + + return content_md5, authorization + + +def get_signing_headers(body_json: str, url_path: str) -> dict[str, str]: + """Get the headers required for a signed V2 API request. + + Args: + body_json: The JSON-serialized request body. + url_path: The URL path (without query parameters). + + Returns: + Dict with Content-MD5 and X-Authorization headers. + """ + content_md5, authorization = compute_signature(body_json, url_path) + return { + "Content-MD5": content_md5, + "X-Authorization": authorization, + }