From f2131310729049803af1dacd4b882090ff1e1cf6 Mon Sep 17 00:00:00 2001 From: piekstra Date: Sun, 8 Feb 2026 01:35:50 -0500 Subject: [PATCH 1/2] Add Tapo cloud device support alongside Kasa Support TP-Link Tapo devices by authenticating against both the Kasa and Tapo cloud APIs in parallel. Both use the same V2 HMAC-SHA1 signing protocol but with different host, signing keys, and app identifiers (discovered via Tapo APK reverse engineering). Key changes: - signing.py: Add Tapo signing keys, make functions accept configurable keys - client.py: Support dual cloud types with separate host/keys/appType - device_client.py: Handle Tapo V2 passthrough format (/api/v2/common/passthrough) - device_info.py: Track which cloud a device belongs to (cloud_type) - device_manager.py: Login to both clouds, merge device lists with deduplication - New include_tapo parameter (default True) on TPLinkDeviceManager Tapo cloud differences from Kasa: - Host: n-wap.i.tplinkcloud.com (vs n-wap.tplinkcloud.com) - Passthrough: POST /api/v2/common/passthrough with flat body - App type: TP-Link_Tapo_Android --- tests/test_device_manager.py | 16 +- tests/test_tapo_devices.py | 159 +++++++++++++++++ .../tapo_get_device_list_response.json | 61 +++++++ ...apo_passthrough_get_sys_info_response.json | 6 + .../tapo_v2_account_status_response.json | 7 + .../__files/tapo_v2_login_response.json | 11 ++ .../tapo_get_device_list_request.json | 25 +++ ...tapo_passthrough_get_sys_info_request.json | 25 +++ .../tapo_v2_account_status_request.json | 22 +++ .../mappings/tapo_v2_login_request.json | 30 ++++ tplinkcloud/client.py | 102 ++++++++--- tplinkcloud/device_client.py | 50 ++++-- tplinkcloud/device_info.py | 3 +- tplinkcloud/device_manager.py | 160 +++++++++++++----- tplinkcloud/signing.py | 46 ++++- 15 files changed, 626 insertions(+), 97 deletions(-) create mode 100644 tests/test_tapo_devices.py create mode 100644 tests/wiremock/__files/tapo_get_device_list_response.json create mode 100644 tests/wiremock/__files/tapo_passthrough_get_sys_info_response.json create mode 100644 tests/wiremock/__files/tapo_v2_account_status_response.json create mode 100644 tests/wiremock/__files/tapo_v2_login_response.json create mode 100644 tests/wiremock/mappings/tapo_get_device_list_request.json create mode 100644 tests/wiremock/mappings/tapo_passthrough_get_sys_info_request.json create mode 100644 tests/wiremock/mappings/tapo_v2_account_status_request.json create mode 100644 tests/wiremock/mappings/tapo_v2_login_request.json diff --git a/tests/test_device_manager.py b/tests/test_device_manager.py index d4ff0d3..0f1d7cf 100644 --- a/tests/test_device_manager.py +++ b/tests/test_device_manager.py @@ -25,10 +25,12 @@ class TestGetDevices(object): async def test_gets_devices(self, client): device_list = await client.get_devices() assert device_list is not None - # 9 devices: HS103, HS105, HS110, HS300 (6 children), KL430, - # HS200, KP200 (2 children), KP400 (2 children), KL420L5 - # Total: 9 parents + 6 + 2 + 2 children = 19 - assert len(device_list) == 19 + # Kasa: 9 parents + 10 children = 19 + # HS103, HS105, HS110, HS300 (6 children), KL430, + # HS200, KP200 (2 children), KP400 (2 children), KL420L5 + # Tapo: 3 devices (P100, P110, L530) + # Total: 19 + 3 = 22 + assert len(device_list) == 22 @pytest.mark.usefixtures('client') class TestFindDevice(object): @@ -295,11 +297,11 @@ async def test_auth_token_initialized_without_credentials(self, client): verbose=False, term_id=os.environ.get('TPLINK_KASA_TERM_ID') ) - # Should be able to access _auth_token without AttributeError - assert device_manager._auth_token is None + # Should be able to access token without AttributeError + assert device_manager.get_token() is None # Should be able to set auth token manually device_manager.set_auth_token('test_token') - assert device_manager._auth_token == 'test_token' + assert device_manager.get_token() == 'test_token' @pytest.mark.asyncio async def test_auth_no_username(self, client): diff --git a/tests/test_tapo_devices.py b/tests/test_tapo_devices.py new file mode 100644 index 0000000..717868f --- /dev/null +++ b/tests/test_tapo_devices.py @@ -0,0 +1,159 @@ +"""Tests for Tapo device support via unified cloud listing.""" + +import os +import pytest +from tplinkcloud import TPLinkDeviceManager + + +class TestTapoDeviceListing: + + @pytest.fixture(scope="class") + async def device_manager(self): + return await TPLinkDeviceManager( + username=os.environ.get('TPLINK_KASA_USERNAME'), + password=os.environ.get('TPLINK_KASA_PASSWORD'), + prefetch=False, + cache_devices=False, + tplink_cloud_api_host=os.environ.get('TPLINK_KASA_API_URL'), + verbose=False, + term_id=os.environ.get('TPLINK_KASA_TERM_ID'), + include_tapo=True, + ) + + @pytest.mark.asyncio + async def test_gets_both_kasa_and_tapo_devices(self, device_manager): + devices = await device_manager.get_devices() + assert devices is not None + assert len(devices) > 0 + + cloud_types = {d.cloud_type for d in devices} + assert "kasa" in cloud_types + assert "tapo" in cloud_types + + @pytest.mark.asyncio + async def test_tapo_devices_have_tapo_cloud_type(self, device_manager): + devices = await device_manager.get_devices() + tapo_devices = [d for d in devices if d.cloud_type == "tapo"] + assert len(tapo_devices) > 0 + # Only check device_info.cloud_type on non-child devices + # (child devices use HS300ChildSysInfo, not TPLinkDeviceInfo) + for device in tapo_devices: + if hasattr(device.device_info, 'cloud_type'): + assert device.device_info.cloud_type == "tapo" + + @pytest.mark.asyncio + async def test_kasa_devices_have_kasa_cloud_type(self, device_manager): + devices = await device_manager.get_devices() + kasa_devices = [d for d in devices if d.cloud_type == "kasa"] + assert len(kasa_devices) > 0 + for device in kasa_devices: + if hasattr(device.device_info, 'cloud_type'): + assert device.device_info.cloud_type == "kasa" + + @pytest.mark.asyncio + async def test_tapo_device_models_present(self, device_manager): + devices = await device_manager.get_devices() + tapo_models = {d.device_info.device_model for d in devices if d.cloud_type == "tapo"} + assert "P100" in tapo_models + assert "P110" in tapo_models + assert "L530" in tapo_models + + @pytest.mark.asyncio + async def test_find_tapo_device_by_name(self, device_manager): + device = await device_manager.find_device("Kitchen Tapo Plug") + assert device is not None + assert device.cloud_type == "tapo" + assert device.device_info.device_model == "P100" + + @pytest.mark.asyncio + async def test_find_devices_returns_both_types(self, device_manager): + # "Light" and "Lamp" appear in both Kasa and Tapo device names + devices = await device_manager.find_devices("Tapo") + assert len(devices) > 0 + assert all(d.cloud_type == "tapo" for d in devices) + + @pytest.mark.asyncio + async def test_tapo_device_get_sys_info(self, device_manager): + device = await device_manager.find_device("Kitchen Tapo Plug") + assert device is not None + sys_info = await device.get_sys_info() + assert sys_info is not None + assert sys_info.get('model') == 'P100' + + +class TestTapoAuth: + + @pytest.mark.asyncio + async def test_tapo_token_set_on_login(self): + device_manager = TPLinkDeviceManager( + username=os.environ.get('TPLINK_KASA_USERNAME'), + password=os.environ.get('TPLINK_KASA_PASSWORD'), + prefetch=False, + tplink_cloud_api_host=os.environ.get('TPLINK_KASA_API_URL'), + term_id=os.environ.get('TPLINK_KASA_TERM_ID'), + include_tapo=True, + ) + assert device_manager.get_token() is not None + assert device_manager.get_tapo_token() is not None + + @pytest.mark.asyncio + async def test_tapo_refresh_token_set_on_login(self): + device_manager = TPLinkDeviceManager( + username=os.environ.get('TPLINK_KASA_USERNAME'), + password=os.environ.get('TPLINK_KASA_PASSWORD'), + prefetch=False, + tplink_cloud_api_host=os.environ.get('TPLINK_KASA_API_URL'), + term_id=os.environ.get('TPLINK_KASA_TERM_ID'), + include_tapo=True, + ) + assert device_manager.get_refresh_token() is not None + assert device_manager.get_tapo_refresh_token() is not None + + +class TestIncludeTapoFalse: + + @pytest.mark.asyncio + async def test_no_tapo_devices_when_disabled(self): + device_manager = await TPLinkDeviceManager( + username=os.environ.get('TPLINK_KASA_USERNAME'), + password=os.environ.get('TPLINK_KASA_PASSWORD'), + prefetch=False, + cache_devices=False, + tplink_cloud_api_host=os.environ.get('TPLINK_KASA_API_URL'), + term_id=os.environ.get('TPLINK_KASA_TERM_ID'), + include_tapo=False, + ) + devices = await device_manager.get_devices() + assert all(d.cloud_type == "kasa" for d in devices) + + @pytest.mark.asyncio + async def test_tapo_token_none_when_disabled(self): + device_manager = TPLinkDeviceManager( + username=os.environ.get('TPLINK_KASA_USERNAME'), + password=os.environ.get('TPLINK_KASA_PASSWORD'), + prefetch=False, + tplink_cloud_api_host=os.environ.get('TPLINK_KASA_API_URL'), + term_id=os.environ.get('TPLINK_KASA_TERM_ID'), + include_tapo=False, + ) + assert device_manager.get_token() is not None + assert device_manager.get_tapo_token() is None + + +class TestDeviceDeduplication: + + @pytest.mark.asyncio + async def test_duplicate_devices_not_listed_twice(self): + """If a device appears in both clouds, it should only appear once.""" + device_manager = await TPLinkDeviceManager( + username=os.environ.get('TPLINK_KASA_USERNAME'), + password=os.environ.get('TPLINK_KASA_PASSWORD'), + prefetch=False, + cache_devices=False, + tplink_cloud_api_host=os.environ.get('TPLINK_KASA_API_URL'), + term_id=os.environ.get('TPLINK_KASA_TERM_ID'), + include_tapo=True, + ) + devices = await device_manager.get_devices() + device_keys = [(d.device_id, getattr(d, 'child_id', None)) for d in devices] + assert len(device_keys) == len(set(device_keys)) diff --git a/tests/wiremock/__files/tapo_get_device_list_response.json b/tests/wiremock/__files/tapo_get_device_list_response.json new file mode 100644 index 0000000..8b02d6c --- /dev/null +++ b/tests/wiremock/__files/tapo_get_device_list_response.json @@ -0,0 +1,61 @@ +{ + "error_code": 0, + "result": { + "deviceList": [ + { + "deviceType": "SMART.TAPOPLUG", + "role": 0, + "fwVer": "1.3.0 Build 230425 Rel.163532", + "appServerUrl": "http://127.0.0.1:8080", + "deviceRegion": "us-east-1", + "deviceId": "TAPO_P100_ABCDEF1234567890ABCDEF1234567890", + "deviceName": "Tapo Mini Smart Wi-Fi Plug", + "deviceHwVer": "1.0", + "alias": "Kitchen Tapo Plug", + "deviceMac": "TAPO11223344", + "oemId": "TAPO1234EFGH5678ABCD1234EFGH5678", + "deviceModel": "P100", + "hwId": "TAPOABCD5678EFGH1234ABCD5678EFGH", + "fwId": "00000000000000000000000000000000", + "isSameRegion": true, + "status": 1 + }, + { + "deviceType": "SMART.TAPOPLUG", + "role": 0, + "fwVer": "1.2.1 Build 230425 Rel.163532", + "appServerUrl": "http://127.0.0.1:8080", + "deviceRegion": "us-east-1", + "deviceId": "TAPO_P110_BCDEF1234567890ABCDEF1234567890A", + "deviceName": "Tapo Mini Smart Wi-Fi Plug with Energy Monitoring", + "deviceHwVer": "1.0", + "alias": "Office Tapo Energy Plug", + "deviceMac": "TAPO55667788", + "oemId": "TAPO5678ABCD1234EFGH5678ABCD1234", + "deviceModel": "P110", + "hwId": "TAPOEFGH1234ABCD5678EFGH1234ABCD", + "fwId": "00000000000000000000000000000000", + "isSameRegion": true, + "status": 1 + }, + { + "deviceType": "SMART.TAPOBULB", + "role": 0, + "fwVer": "1.1.0 Build 230602 Rel.145203", + "appServerUrl": "http://127.0.0.1:8080", + "deviceRegion": "us-east-1", + "deviceId": "TAPO_L530_CDEF1234567890ABCDEF1234567890AB", + "deviceName": "Tapo Smart Wi-Fi Light Bulb, Multicolor", + "deviceHwVer": "2.0", + "alias": "Living Room Tapo Bulb", + "deviceMac": "TAPO99AABBCC", + "oemId": "TAPOBULB1234EFGH5678ABCD1234EFGH", + "deviceModel": "L530", + "hwId": "TAPOBULBABCD5678EFGH1234ABCD5678", + "fwId": "00000000000000000000000000000000", + "isSameRegion": true, + "status": 1 + } + ] + } +} diff --git a/tests/wiremock/__files/tapo_passthrough_get_sys_info_response.json b/tests/wiremock/__files/tapo_passthrough_get_sys_info_response.json new file mode 100644 index 0000000..128f3ab --- /dev/null +++ b/tests/wiremock/__files/tapo_passthrough_get_sys_info_response.json @@ -0,0 +1,6 @@ +{ + "error_code": 0, + "result": { + "responseData": "{\"system\":{\"get_sysinfo\":{\"sw_ver\":\"1.3.0 Build 230425 Rel.163532\",\"hw_ver\":\"1.0\",\"model\":\"P100\",\"deviceId\":\"TAPO_P100_ABCDEF1234567890ABCDEF1234567890\",\"alias\":\"Kitchen Tapo Plug\",\"relay_state\":1,\"on_time\":12345,\"feature\":\"TIM\",\"rssi\":-45,\"err_code\":0}}}" + } +} diff --git a/tests/wiremock/__files/tapo_v2_account_status_response.json b/tests/wiremock/__files/tapo_v2_account_status_response.json new file mode 100644 index 0000000..34e204b --- /dev/null +++ b/tests/wiremock/__files/tapo_v2_account_status_response.json @@ -0,0 +1,7 @@ +{ + "error_code": 0, + "result": { + "accountStatus": 0, + "appServerUrl": "http://127.0.0.1:8080" + } +} diff --git a/tests/wiremock/__files/tapo_v2_login_response.json b/tests/wiremock/__files/tapo_v2_login_response.json new file mode 100644 index 0000000..f85a2cf --- /dev/null +++ b/tests/wiremock/__files/tapo_v2_login_response.json @@ -0,0 +1,11 @@ +{ + "error_code": 0, + "result": { + "accountId": "123456", + "regTime": "2017-11-27 23:21:40", + "countryCode": "US", + "email": "kasa_docker", + "token": "tapo-token-mock-xyz789", + "refreshToken": "tapo-refresh-token-mock-xyz789" + } +} diff --git a/tests/wiremock/mappings/tapo_get_device_list_request.json b/tests/wiremock/mappings/tapo_get_device_list_request.json new file mode 100644 index 0000000..07547d7 --- /dev/null +++ b/tests/wiremock/mappings/tapo_get_device_list_request.json @@ -0,0 +1,25 @@ +{ + "priority": 1, + "request": { + "method": "POST", + "urlPath": "/", + "queryParameters": { + "appName": { "equalTo": "TP-Link_Tapo_Android" } + }, + "bodyPatterns": [ + { + "equalToJson": { + "method": "getDeviceList" + } + } + ], + "headers": { + "User-Agent": { "matches": "Dalvik/.*" }, + "Content-Type": { "matches": "application/json.*" } + } + }, + "response": { + "status": 200, + "bodyFileName": "tapo_get_device_list_response.json" + } +} diff --git a/tests/wiremock/mappings/tapo_passthrough_get_sys_info_request.json b/tests/wiremock/mappings/tapo_passthrough_get_sys_info_request.json new file mode 100644 index 0000000..9cbcbd5 --- /dev/null +++ b/tests/wiremock/mappings/tapo_passthrough_get_sys_info_request.json @@ -0,0 +1,25 @@ +{ + "request": { + "method": "POST", + "urlPath": "/api/v2/common/passthrough", + "bodyPatterns": [ + { + "matchesJsonPath": "$.deviceId" + }, + { + "matchesJsonPath": { + "expression": "$.requestData", + "contains": "get_sysinfo" + } + } + ], + "headers": { + "User-Agent": { "matches": "Dalvik/.*" }, + "Content-Type": { "matches": "application/json.*" } + } + }, + "response": { + "status": 200, + "bodyFileName": "tapo_passthrough_get_sys_info_response.json" + } +} diff --git a/tests/wiremock/mappings/tapo_v2_account_status_request.json b/tests/wiremock/mappings/tapo_v2_account_status_request.json new file mode 100644 index 0000000..afc0e61 --- /dev/null +++ b/tests/wiremock/mappings/tapo_v2_account_status_request.json @@ -0,0 +1,22 @@ +{ + "request": { + "method": "POST", + "urlPath": "/api/v2/account/getAccountStatusAndUrl", + "bodyPatterns": [ + { + "equalToJson": { + "appType": "TP-Link_Tapo_Android", + "cloudUserName": "kasa_docker" + } + } + ], + "headers": { + "User-Agent": { "matches": "Dalvik/.*" }, + "Content-Type": { "matches": "application/json.*" } + } + }, + "response": { + "status": 200, + "bodyFileName": "tapo_v2_account_status_response.json" + } +} diff --git a/tests/wiremock/mappings/tapo_v2_login_request.json b/tests/wiremock/mappings/tapo_v2_login_request.json new file mode 100644 index 0000000..9078eca --- /dev/null +++ b/tests/wiremock/mappings/tapo_v2_login_request.json @@ -0,0 +1,30 @@ +{ + "request": { + "method": "POST", + "urlPath": "/api/v2/account/login", + "bodyPatterns": [ + { + "equalToJson": { + "appType": "TP-Link_Tapo_Android", + "appVersion": "3.4.451", + "cloudPassword": "kasa_password", + "cloudUserName": "kasa_docker", + "platform": "Android", + "refreshTokenNeeded": true, + "supportBindAccount": false, + "terminalUUID": "2a8ced52-f200-4b79-a1fe-2f6b58193c4c", + "terminalName": "Pixel", + "terminalMeta": "Pixel" + } + } + ], + "headers": { + "User-Agent": { "matches": "Dalvik/.*" }, + "Content-Type": { "matches": "application/json.*" } + } + }, + "response": { + "status": 200, + "bodyFileName": "tapo_v2_login_response.json" + } +} diff --git a/tplinkcloud/client.py b/tplinkcloud/client.py index 63098b0..8a5655e 100644 --- a/tplinkcloud/client.py +++ b/tplinkcloud/client.py @@ -1,13 +1,11 @@ """Synchronous HTTP client for TP-Link Cloud API authentication and device listing. -Supports both V1 (legacy) and V2 (current) API protocols: +Supports both Kasa and Tapo cloud APIs using the V2 protocol: -V1: POST https://wap.tplinkcloud.com/?appName=Kasa_Android&... - Body: {"method": "login", "params": {"cloudUserName": "...", ...}} +Kasa: POST https://n-wap.tplinkcloud.com/api/v2/account/login?appName=Kasa_Android_Mix&... +Tapo: POST https://n-wap.i.tplinkcloud.com/api/v2/account/login?appName=TP-Link_Tapo_Android&... -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 +Both use the same signing algorithm (HMAC-SHA1) but with different key pairs. """ import json @@ -23,7 +21,13 @@ TPLinkMFARequiredError, TPLinkTokenExpiredError, ) -from .signing import get_signing_headers +from .signing import ( + KASA_ACCESS_KEY, + KASA_SECRET_KEY, + TAPO_ACCESS_KEY, + TAPO_SECRET_KEY, + get_signing_headers, +) # V2 API error codes _ERR_MFA_REQUIRED = -20677 @@ -32,11 +36,19 @@ _ERR_WRONG_CREDENTIALS = -20601 _ERR_ACCOUNT_LOCKED = -20675 -# Default API hosts -_V1_HOST = "https://wap.tplinkcloud.com" -_V2_HOST = "https://n-wap.tplinkcloud.com" +# Kasa cloud +KASA_HOST = "https://n-wap.tplinkcloud.com" +KASA_APP_TYPE = "Kasa_Android_Mix" +KASA_APP_NAME = "Kasa_Android_Mix" +KASA_APP_VER = "3.4.451" + +# Tapo cloud +TAPO_HOST = "https://n-wap.i.tplinkcloud.com" +TAPO_APP_TYPE = "TP-Link_Tapo_Android" +TAPO_APP_NAME = "TP-Link_Tapo_Android" +TAPO_APP_VER = "3.4.451" -# V2 API paths +# V2 API paths (shared by both Kasa and Tapo) _PATH_ACCOUNT_STATUS = "/api/v2/account/getAccountStatusAndUrl" _PATH_LOGIN = "/api/v2/account/login" _PATH_REFRESH_TOKEN = "/api/v2/account/refreshToken" @@ -44,18 +56,34 @@ class TPLinkApi: - def __init__(self, host=None, verbose=False, term_id=None): + def __init__(self, host=None, verbose=False, term_id=None, + cloud_type="kasa"): self._verbose = verbose 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._cloud_type = cloud_type + + if cloud_type == "tapo": + self._access_key = TAPO_ACCESS_KEY + self._secret_key = TAPO_SECRET_KEY + self._app_type = TAPO_APP_TYPE + self._app_name = TAPO_APP_NAME + self._app_ver = TAPO_APP_VER + default_host = TAPO_HOST + else: + self._access_key = KASA_ACCESS_KEY + self._secret_key = KASA_SECRET_KEY + self._app_type = KASA_APP_TYPE + self._app_name = KASA_APP_NAME + self._app_ver = KASA_APP_VER + default_host = KASA_HOST + + self.host = host or default_host + + # V2 query parameters (sent on all requests) self._query_params = { - "appName": "Kasa_Android_Mix", - "appVer": "3.4.451", + "appName": self._app_name, + "appVer": self._app_ver, "netType": "wifi", "termID": self._term_id, "ospf": "Android 14", @@ -71,6 +99,18 @@ def __init__(self, host=None, verbose=False, term_id=None): "Content-Type": "application/json;charset=UTF-8", } + @property + def cloud_type(self): + return self._cloud_type + + @property + def access_key(self): + return self._access_key + + @property + def secret_key(self): + return self._secret_key + def _request_post_v2(self, base_url, url_path, body, token=None): """Make a signed V2 API request. @@ -90,7 +130,11 @@ def _request_post_v2(self, base_url, url_path, body, token=None): if token: params["token"] = token - signing_headers = get_signing_headers(body_json, url_path) + signing_headers = get_signing_headers( + body_json, url_path, + access_key=self._access_key, + secret_key=self._secret_key, + ) headers = {**self._headers, **signing_headers} if self._verbose: @@ -121,7 +165,7 @@ def _request_post_v2(self, base_url, url_path, body, token=None): 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, + Kasa device operations use the V1 JSON format on the root path, but with V2 signing headers and query parameters. """ url_path = "/" @@ -131,7 +175,11 @@ def _request_post_v1(self, body, token=None): if token: params["token"] = token - signing_headers = get_signing_headers(body_json, url_path) + signing_headers = get_signing_headers( + body_json, url_path, + access_key=self._access_key, + secret_key=self._secret_key, + ) headers = {**self._headers, **signing_headers} if self._verbose: @@ -166,7 +214,7 @@ def _get_regional_url(self, username): The regional appServerUrl string. """ body = { - "appType": "Kasa_Android_Mix", + "appType": self._app_type, "cloudUserName": username, } response = self._request_post_v2( @@ -210,8 +258,8 @@ def login(self, username, password, mfa_callback=None): # Step 2: Login login_body = { - "appType": "Kasa_Android_Mix", - "appVersion": "3.4.451", + "appType": self._app_type, + "appVersion": self._app_ver, "cloudPassword": password, "cloudUserName": username, "platform": "Android", @@ -262,7 +310,7 @@ def _verify_mfa(self, regional_url, username, password, mfa_code): Dict with 'token' and optionally 'refreshToken'. """ body = { - "appType": "Kasa_Android_Mix", + "appType": self._app_type, "cloudPassword": password, "cloudUserName": username, "code": mfa_code, @@ -292,7 +340,7 @@ def refresh_login(self, refresh_token): TPLinkTokenExpiredError: If the refresh token itself has expired. """ body = { - "appType": "Kasa_Android_Mix", + "appType": self._app_type, "refreshToken": refresh_token, "terminalUUID": self._term_id, } diff --git a/tplinkcloud/device_client.py b/tplinkcloud/device_client.py index d5f1f4b..a3de951 100644 --- a/tplinkcloud/device_client.py +++ b/tplinkcloud/device_client.py @@ -4,18 +4,23 @@ from .api_response import TPLinkApiResponse from .certs import get_ca_cert_path -from .signing import get_signing_headers +from .signing import KASA_ACCESS_KEY, KASA_SECRET_KEY, get_signing_headers import ssl class TPLinkDeviceClient: - def __init__(self, host, token, verbose=False, term_id=None): + def __init__(self, host, token, verbose=False, term_id=None, + access_key=None, secret_key=None, app_name=None, + cloud_type="kasa"): self.host = host self._verbose = verbose self._term_id = term_id or str(uuid.uuid4()) + self._access_key = access_key or KASA_ACCESS_KEY + self._secret_key = secret_key or KASA_SECRET_KEY + self._cloud_type = cloud_type self._params = { - "appName": "Kasa_Android_Mix", + "appName": app_name or "Kasa_Android_Mix", "appVer": "3.4.451", "netType": "wifi", "termID": self._term_id, @@ -35,19 +40,24 @@ def __init__(self, host, token, verbose=False, term_id=None): # 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): + async def _request_post(self, body, url_path="/"): if self._verbose: - print('POST', self.host, body) + print('POST', self.host + url_path, body) - url_path = "/" body_json = json.dumps(body) - signing_headers = get_signing_headers(body_json, url_path) + signing_headers = get_signing_headers( + body_json, url_path, + access_key=self._access_key, + secret_key=self._secret_key, + ) headers = {**self._headers, **signing_headers} + url = self.host if url_path == "/" else f"{self.host}{url_path}" + async with aiohttp.ClientSession() as session: async with session.post( - self.host, + url, data=body_json, params=self._params, headers=headers, @@ -66,14 +76,26 @@ async def _request_post(self, body): raise Exception(str(response.status) + ': ' + response.reason) async def pass_through_request(self, device_id, request_data): - body = { - 'method': 'passthrough', - 'params': { + if self._cloud_type == "tapo": + # Tapo uses V2-style passthrough endpoint with flat body + body = { 'deviceId': device_id, - 'requestData': json.dumps(request_data) + 'requestData': json.dumps(request_data), } - } - response = await self._request_post(body) + response = await self._request_post( + body, url_path="/api/v2/common/passthrough" + ) + else: + # Kasa uses V1-style method/params wrapper on root path + body = { + 'method': 'passthrough', + 'params': { + 'deviceId': device_id, + 'requestData': json.dumps(request_data) + } + } + response = await self._request_post(body) + if response.successful: response_data = response.result.get('responseData') # Some devices (e.g., Archer routers) return responseData as a dict diff --git a/tplinkcloud/device_info.py b/tplinkcloud/device_info.py index 3e4f798..8c3c751 100644 --- a/tplinkcloud/device_info.py +++ b/tplinkcloud/device_info.py @@ -1,6 +1,6 @@ class TPLinkDeviceInfo: - def __init__(self, device_info): + def __init__(self, device_info, cloud_type="kasa"): self.device_type = device_info.get('deviceType') self.role = device_info.get('role') self.fw_ver = device_info.get('fwVer') @@ -17,3 +17,4 @@ def __init__(self, device_info): self.fw_id = device_info.get('fwId') self.is_same_region = device_info.get('isSameRegion') self.status = device_info.get('status') + self.cloud_type = cloud_type diff --git a/tplinkcloud/device_manager.py b/tplinkcloud/device_manager.py index f74fe48..d527693 100644 --- a/tplinkcloud/device_manager.py +++ b/tplinkcloud/device_manager.py @@ -51,27 +51,65 @@ def __init__( verbose=False, term_id=None, mfa_callback=None, + include_tapo=True, ): 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._include_tapo = include_tapo - self._tplink_api = TPLinkApi( - tplink_cloud_api_host, verbose=self._verbose, term_id=self._term_id) + # Kasa cloud API (always present) + self._kasa_api = TPLinkApi( + tplink_cloud_api_host, verbose=self._verbose, + term_id=self._term_id, cloud_type="kasa", + ) + self._kasa_token = None + self._kasa_refresh_token = None + + # Tapo cloud API (optional, enabled by default) + self._tapo_api = None + self._tapo_token = None + self._tapo_refresh_token = None + if self._include_tapo: + self._tapo_api = TPLinkApi( + tplink_cloud_api_host, verbose=self._verbose, + term_id=self._term_id, cloud_type="tapo", + ) if username and password: - self.login(username, password, mfa_callback=mfa_callback) + self._login_all(username, password, mfa_callback=mfa_callback) self._prefetch = prefetch + def _login_all(self, username, password, mfa_callback=None): + """Login to Kasa cloud and optionally Tapo cloud.""" + # Kasa login + result = self._kasa_api.login( + username, password, mfa_callback=mfa_callback + ) + if result: + self._kasa_token = result.get('token') + self._kasa_refresh_token = result.get('refreshToken') + + # Tapo login (separate cloud, same credentials) + if self._tapo_api: + try: + tapo_result = self._tapo_api.login( + username, password, mfa_callback=mfa_callback + ) + if tapo_result: + self._tapo_token = tapo_result.get('token') + self._tapo_refresh_token = tapo_result.get('refreshToken') + except Exception: + if self._verbose: + print("Tapo cloud login failed, continuing with Kasa only") + async def async_init(self): # Fetch the devices up front if prefetch and cache them if caching - if self._prefetch and self._cache_devices and self._auth_token: + if self._prefetch and self._cache_devices and self._kasa_token: await self.get_devices() return self @@ -82,80 +120,122 @@ async def get_devices(self): if self._cached_devices: return self._cached_devices + devices = [] + + # Get Kasa devices + kasa_devices = await self._get_cloud_devices( + self._kasa_api, self._kasa_token, self._kasa_refresh_token, + "kasa", + ) + devices.extend(kasa_devices) + + # Get Tapo devices + if self._tapo_api and self._tapo_token: + tapo_devices = await self._get_cloud_devices( + self._tapo_api, self._tapo_token, self._tapo_refresh_token, + "tapo", + ) + # Deduplicate: if a device appears in both clouds, keep the + # Kasa version (it's already in the list) + kasa_device_ids = {d.device_id for d in devices} + for device in tapo_devices: + if device.device_id not in kasa_device_ids: + devices.append(device) + + if self._cache_devices: + self._cached_devices = devices + + return devices + + async def _get_cloud_devices(self, api, token, refresh_token, cloud_type): + """Get devices from a specific cloud (Kasa or Tapo).""" try: - device_info_list = self._tplink_api.get_device_info_list( - self._auth_token) + device_info_list = api.get_device_info_list(token) except TPLinkTokenExpiredError: - if self._refresh_token: - self._do_refresh() - device_info_list = self._tplink_api.get_device_info_list( - self._auth_token) + if refresh_token: + result = api.refresh_login(refresh_token) + if result: + if cloud_type == "kasa": + self._kasa_token = result.get('token') + self._kasa_refresh_token = result.get('refreshToken') + token = self._kasa_token + else: + self._tapo_token = result.get('token') + self._tapo_refresh_token = result.get('refreshToken') + token = self._tapo_token + device_info_list = api.get_device_info_list(token) else: raise devices = [] children_gather_tasks = [] for device_info in device_info_list: - device = self._construct_device(device_info) + device = self._construct_device(device_info, api, token, cloud_type) devices.append(device) if device.has_children(): children_gather_tasks.append(device.get_children_async()) devices_children = await asyncio.gather(*children_gather_tasks) for device_children in devices_children: + for child in device_children: + child.cloud_type = cloud_type devices.extend(device_children) - if self._cache_devices: - self._device_info_list = device_info_list - self._cached_devices = devices - return devices - def _construct_device(self, device_info): - tplink_device_info = TPLinkDeviceInfo(device_info) + def _construct_device(self, device_info, api, token, cloud_type): + tplink_device_info = TPLinkDeviceInfo(device_info, cloud_type=cloud_type) client = TPLinkDeviceClient( tplink_device_info.app_server_url, - self._auth_token, + token, verbose=self._verbose, - term_id=self._term_id + term_id=self._term_id, + access_key=api.access_key, + secret_key=api.secret_key, + app_name=api._app_name, + cloud_type=cloud_type, ) model = tplink_device_info.device_model device_cls = next( (cls for prefix, cls in DEVICE_MODEL_MAP.items() if model.startswith(prefix)), TPLinkDevice, ) - return device_cls(client, tplink_device_info.device_id, tplink_device_info) + device = device_cls(client, tplink_device_info.device_id, tplink_device_info) + device.cloud_type = cloud_type + return device def login(self, username, password, mfa_callback=None): - result = self._tplink_api.login( + """Login to Kasa cloud. For backward compatibility.""" + result = self._kasa_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') + self._kasa_token = result.get('token') + self._kasa_refresh_token = result.get('refreshToken') + return self._kasa_token def set_auth_token(self, auth_token): - self._auth_token = auth_token + self._kasa_token = auth_token def get_token(self): - """Get the current auth token.""" - return self._auth_token + """Get the current Kasa auth token.""" + return self._kasa_token def set_refresh_token(self, refresh_token): - """Set the refresh token (e.g. when resuming a session).""" - self._refresh_token = refresh_token + """Set the Kasa refresh token (e.g. when resuming a session).""" + self._kasa_refresh_token = refresh_token def get_refresh_token(self): - """Get the current refresh token.""" - return self._refresh_token + """Get the current Kasa refresh token.""" + return self._kasa_refresh_token + + def get_tapo_token(self): + """Get the current Tapo auth token.""" + return self._tapo_token + + def get_tapo_refresh_token(self): + """Get the current Tapo refresh token.""" + return self._tapo_refresh_token async def find_device(self, device_name): devices = await self.get_devices() diff --git a/tplinkcloud/signing.py b/tplinkcloud/signing.py index 3c60e62..f6ca79b 100644 --- a/tplinkcloud/signing.py +++ b/tplinkcloud/signing.py @@ -4,6 +4,9 @@ The signing protocol uses an AccessKey/SecretKey pair that identifies the client application (not the user). +Both Kasa and Tapo apps use the same signing algorithm but with +different key pairs. + Signature format: sig_string = "{content_md5}\\n{timestamp}\\n{nonce}\\n{url_path}" signature = HMAC-SHA1(secret_key, sig_string) @@ -19,10 +22,18 @@ # App-level keys from Kasa Android APK (identify the app, not the user) -ACCESS_KEY = "e37525375f8845999bcc56d5e6faa76d" -SECRET_KEY = "314bc6700b3140ca80bc655e527cb062" +KASA_ACCESS_KEY = "e37525375f8845999bcc56d5e6faa76d" +KASA_SECRET_KEY = "314bc6700b3140ca80bc655e527cb062" + +# App-level keys from Tapo Android APK (identify the app, not the user) +TAPO_ACCESS_KEY = "4d11b6b9d5ea4d19a829adbb9714b057" +TAPO_SECRET_KEY = "6ed7d97f3e73467f8a5bab90b577ba4c" + +# Default to Kasa keys for backward compatibility +ACCESS_KEY = KASA_ACCESS_KEY +SECRET_KEY = KASA_SECRET_KEY -# The Kasa app uses a hardcoded timestamp for signing +# Both apps use the same hardcoded timestamp for signing SIGNING_TIMESTAMP = "9999999999" @@ -33,23 +44,33 @@ def compute_content_md5(body: str) -> str: ).decode() -def compute_signature(body_json: str, url_path: str) -> tuple[str, str]: +def compute_signature( + body_json: str, + url_path: str, + access_key: str | None = None, + secret_key: str | None = None, +) -> 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. + access_key: Override the default AccessKey (for Tapo support). + secret_key: Override the default SecretKey (for Tapo support). Returns: A tuple of (content_md5, x_authorization_header). """ + ak = access_key or ACCESS_KEY + sk = secret_key or SECRET_KEY + 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(), + sk.encode(), sig_string.encode(), hashlib.sha1, ).hexdigest() @@ -57,24 +78,33 @@ def compute_signature(body_json: str, url_path: str) -> tuple[str, str]: authorization = ( f"Timestamp={SIGNING_TIMESTAMP}, " f"Nonce={nonce}, " - f"AccessKey={ACCESS_KEY}, " + f"AccessKey={ak}, " f"Signature={signature}" ) return content_md5, authorization -def get_signing_headers(body_json: str, url_path: str) -> dict[str, str]: +def get_signing_headers( + body_json: str, + url_path: str, + access_key: str | None = None, + secret_key: str | None = None, +) -> 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). + access_key: Override the default AccessKey (for Tapo support). + secret_key: Override the default SecretKey (for Tapo support). Returns: Dict with Content-MD5 and X-Authorization headers. """ - content_md5, authorization = compute_signature(body_json, url_path) + content_md5, authorization = compute_signature( + body_json, url_path, access_key, secret_key + ) return { "Content-MD5": content_md5, "X-Authorization": authorization, From 048b71759dfffb080d0c5dd117bb9e17e9d6fd6d Mon Sep 17 00:00:00 2001 From: piekstra Date: Sun, 8 Feb 2026 01:39:29 -0500 Subject: [PATCH 2/2] Update README and CLAUDE.md with Tapo device support Add documentation for the new dual-cloud architecture, Tapo device compatibility list, cloud_type attribute, and include_tapo parameter. --- CLAUDE.md | 28 ++++++++++++++++++++++------ README.md | 27 +++++++++++++++++++++++---- 2 files changed, 45 insertions(+), 10 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index 9a06496..6f2ea58 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -1,6 +1,6 @@ # tplink-cloud-api -Python library for controlling TP-Link Kasa smart home devices remotely via the TP-Link Cloud API. Unlike local-network libraries, this works from anywhere with an internet connection. +Python library for controlling TP-Link Kasa and Tapo smart home devices remotely via the TP-Link Cloud API. Unlike local-network libraries, this works from anywhere with an internet connection. ## Quick start @@ -21,16 +21,32 @@ pytest --verbose 2. **Login** (`/api/v2/account/login`) — authenticates with TP-Link credentials, returns token + refresh token 3. **Device operations** — POST to `/` on the device's `appServerUrl` with V2 signing headers -All V2 requests use HMAC-SHA1 signing (see `tplinkcloud/signing.py`). The signing keys are app-level constants extracted from the Kasa Android APK — they are not secrets. +All V2 requests use HMAC-SHA1 signing (see `tplinkcloud/signing.py`). The signing keys are app-level constants extracted from the Kasa and Tapo Android APKs — they are not secrets. + +### Dual-cloud architecture (Kasa + Tapo) + +The library authenticates against both Kasa and Tapo clouds in parallel using the same TP-Link credentials. Each cloud has its own host, signing keys, and app identifiers: + +| Aspect | Kasa | Tapo | +|--------|------|------| +| Cloud host | `n-wap.tplinkcloud.com` | `n-wap.i.tplinkcloud.com` | +| App type | `Kasa_Android_Mix` | `TP-Link_Tapo_Android` | +| Passthrough URL | POST `/` with `{"method":"passthrough","params":{...}}` | POST `/api/v2/common/passthrough` with flat `{deviceId, requestData}` | +| Signing algorithm | HMAC-SHA1 (same) | HMAC-SHA1 (same) | + +`TPLinkDeviceManager` creates two `TPLinkApi` instances, logs into both, and merges device lists with deduplication by device ID. Each device has a `cloud_type` attribute ("kasa" or "tapo"). Tapo login failure is silently caught — the library falls back to Kasa-only. + +Set `include_tapo=False` on `TPLinkDeviceManager` to disable Tapo cloud support. ### Key modules | Module | Purpose | |---|---| -| `client.py` | Synchronous HTTP client for auth (login, MFA, refresh, device list) | -| `device_client.py` | Async HTTP client (aiohttp) for device operations with V2 signing | -| `device_manager.py` | Main entry point — `TPLinkDeviceManager` handles auth + device construction | -| `signing.py` | HMAC-SHA1 request signing for V2 API | +| `client.py` | Synchronous HTTP client for auth (login, MFA, refresh, device list); supports both Kasa and Tapo cloud types | +| `device_client.py` | Async HTTP client (aiohttp) for device operations with V2 signing; handles Kasa and Tapo passthrough formats | +| `device_manager.py` | Main entry point — `TPLinkDeviceManager` handles dual-cloud auth + unified device construction | +| `device_info.py` | Device metadata including `cloud_type` ("kasa" or "tapo") | +| `signing.py` | HMAC-SHA1 request signing for V2 API with configurable key pairs | | `exceptions.py` | `TPLinkAuthError`, `TPLinkMFARequiredError`, `TPLinkTokenExpiredError`, `TPLinkCloudError`, `TPLinkDeviceOfflineError` | | `certs/` | TP-Link private CA cert chain (V2 API servers use their own CA) | diff --git a/README.md b/README.md index 884e8f7..2266cf7 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,6 @@ # tplink-cloud-api -Control TP-Link Kasa smart home devices from anywhere over the internet using TP-Link's cloud API — no local network access required. +Control TP-Link Kasa and Tapo smart home devices from anywhere over the internet using TP-Link's cloud API — no local network access required. ## Why cloud control? @@ -17,7 +17,7 @@ If you just need local control on the same network as your devices, [python-kasa ## How it works -The library authenticates with your TP-Link / Kasa account credentials using the **V2 TP-Link Cloud API** with HMAC-SHA1 request signing. It supports MFA (two-factor authentication) and automatic refresh token management. +The library authenticates with your TP-Link / Kasa account credentials using the **V2 TP-Link Cloud API** with HMAC-SHA1 request signing. It automatically connects to both the Kasa and Tapo clouds in parallel, returning a unified device list. It supports MFA (two-factor authentication) and automatic refresh token management. Originally a Python port of [Adumont's Node.js module](https://github.com/adumont/tplink-cloud-api). @@ -25,6 +25,8 @@ Originally a Python port of [Adumont's Node.js module](https://github.com/adumon The following devices are _officially_ supported by the library at this time: +### Kasa Devices + **Smart Plugs** * HS100 (Smart Plug - Blocks two outlets as a single outlet) * HS103 (Smart Plug Lite - 12 Amp) @@ -49,6 +51,17 @@ The following devices are _officially_ supported by the library at this time: * KL420L5 (Smart LED Light Strip) * KL430 (Smart Light Strip, Multicolor) +### Tapo Devices + +Tapo devices are supported through the Tapo cloud API. The library automatically authenticates against both clouds and returns a unified device list. All Tapo devices support basic on/off control through the generic `TPLinkDevice` class: + +* P100 (Mini Smart Wi-Fi Plug) +* P110 (Mini Smart Wi-Fi Plug with Energy Monitoring) +* L530 (Smart Wi-Fi Light Bulb, Multicolor) +* Any other Tapo device registered to your TP-Link account + +Each device has a `cloud_type` attribute ("kasa" or "tapo") so you can identify which cloud it belongs to. + Devices not explicitly listed above will still work with basic on/off functionality through the generic `TPLinkDevice` class. ## Requirements @@ -194,7 +207,13 @@ devices = await device_manager.get_devices() if devices: print(f'Found {len(devices)} devices') for device in devices: - print(f'{device.model_type.name} device called {device.get_alias()}') + print(f'[{device.cloud_type}] {device.model_type.name} device called {device.get_alias()}') +``` + +By default, `get_devices()` returns devices from both the Kasa and Tapo clouds. To disable Tapo device discovery: + +```python +device_manager = TPLinkDeviceManager(username, password, include_tapo=False) ``` ### Control your devices @@ -213,7 +232,7 @@ else: print(f'Could not find {device_name}') ``` -Replace `My Smart Plug` with the alias you gave to your plug in the Kasa app (be sure to give a different alias to each device). Instead of `toggle()`, you can also use `power_on()` or `power_off()`. +Replace `My Smart Plug` with the alias you gave to your plug in the Kasa or Tapo app (be sure to give a different alias to each device). Instead of `toggle()`, you can also use `power_on()` or `power_off()`. To retrieve power consumption data for one of the individual plugs on an HS300 power strip (KP303 does not support power usage data):