diff --git a/README.md b/README.md index d7b7c4c..58c9501 100644 --- a/README.md +++ b/README.md @@ -43,6 +43,7 @@ from tplinkrouterc6u import ( TPLinkCPE210Client, TPLinkSG108EClient, TplinkC80Router, + TplinkWR844NRouter, TplinkWDRRouter, TPLinkC50Client, TPLinkWR841NClient, @@ -431,6 +432,7 @@ or you have TP-link C5400X or similar router you need to get web encrypted passw - TL-WA1201 3.0 - TL-WA3001 v1.0 - TL-WR841N v14 +- TL-WR844N v1.0 - TL-WR3002X v1.0 - TL-WDR3600 V1 - TL-XDR3010 V2 diff --git a/test/test_client_wr844n.py b/test/test_client_wr844n.py new file mode 100644 index 0000000..f49afe7 --- /dev/null +++ b/test/test_client_wr844n.py @@ -0,0 +1,109 @@ +from ipaddress import IPv4Address +from unittest import TestCase, main + +from macaddress import EUI48 + +from tplinkrouterc6u import Connection +from tplinkrouterc6u.client.wr844n import TplinkWR844NRouter +from tplinkrouterc6u.common.dataclass import Device, Firmware, Status +from tplinkrouterc6u.common.exception import ClientException + + +FIRMWARE_RESPONSE = ('00000\r\nid 0|1,0,0\r\nfullName 300Mbps%20Wi-Fi%20Router\r\nfacturer TP-Link\r\n' + 'modelName TL-WR844N\r\nmodelVer 1.0\r\n' + 'softVer 1.10.0%20Build%20211011%20Rel.66152n(4555)\r\n' + 'hardVer TL-WR844N%201.0\r\nprodId 0x8440001') + +STATUS_RESPONSE = ('00000\r\nid 1|1,0,0\r\nauthKey token\r\nreserved\r\nsetWzd 1\r\nmode 4\r\n' + 'mac 0 40-ae-30-af-c8-72\r\nmac 1 40-ae-30-af-c8-73\r\nwanMacType 0\r\n' + 'id 4|1,0,0\r\nip 192.168.2.1\r\nmask 255.255.255.0\r\nmode 0\r\n' + 'smartIp 0\r\ngateway 0.0.0.0\r\n' + 'id 9|1,0,0\r\nhostName 0 LGwebOSTV\r\nhostName 1 Kaspars\r\n' + 'hostName 2 Redmi-Note-13-Pro-5G\r\nhostName 3\r\n' + 'mac 0 60-75-6c-a1-4a-82\r\nmac 1 70-08-10-0b-c6-24\r\n' + 'mac 2 d6-02-54-6b-67-c4\r\nmac 3 00-00-00-00-00-00\r\n' + 'ip 0 192.168.2.100\r\nip 1 192.168.2.101\r\nip 2 192.168.2.102\r\nip 3 0.0.0.0\r\n' + 'state 0 5\r\nstate 1 5\r\nstate 2 5\r\nstate 3 0\r\n' + 'expires 0 4979\r\nexpires 1 5514\r\nexpires 2 6296\r\nexpires 3 0\r\n' + 'id 0|1,0,0\r\nmodelName TL-WR844N\r\nhardVer TL-WR844N%201.0') + + +class ResponseMock: + def __init__(self, text, status_code=0): + self.text = text + self.status_code = status_code + + +class TplinkWR844NRouterTest(TplinkWR844NRouter): + response = '' + + def request(self, code: int, asyn: int, use_token: bool = False, data: str = None) -> ResponseMock | None: + if code == 2 and asyn == 1: + if use_token is False: + if data == '0|1,0,0': + return ResponseMock(FIRMWARE_RESPONSE, 200) + return ResponseMock('blabla\r\nblabla\r\nblabla\r\nauthinfo1\r\nauthinfo2') + return ResponseMock(self.response) + if (code == 16 or code == 7) and asyn == 0: + if use_token is False: + return ResponseMock('00000\r\n010001\r\nBC97577E65233B3E1137C61091D64176C334E52AD78FFBDDABC826B685435E' + '9D3DE83FE70C2AC62D6B13BD8EADA10B5623F9354DA0E99636A4F5519CA2DC2DC3\r\n12345656') + return ResponseMock('00000') + + raise ClientException() + + +class TestTPLinkWR844NClient(TestCase): + def test_supports(self) -> None: + client = TplinkWR844NRouterTest('', '') + + self.assertTrue(client.supports()) + + def test_get_firmware_handles_plaintext_response(self) -> None: + client = TplinkWR844NRouterTest('', '') + client.authorize() + client.response = FIRMWARE_RESPONSE + + firmware = client.get_firmware() + + self.assertIsInstance(firmware, Firmware) + self.assertEqual(firmware.hardware_version, 'TL-WR844N 1.0') + self.assertEqual(firmware.model, 'TL-WR844N') + self.assertEqual(firmware.firmware_version, '1.10.0 Build 211011 Rel.66152n(4555)') + + def test_get_status_uses_dhcp_client_block(self) -> None: + client = TplinkWR844NRouterTest('http://192.168.0.68', '') + client.authorize() + client.response = STATUS_RESPONSE + + status = client.get_status() + + self.assertIsInstance(status, Status) + self.assertEqual(status.lan_macaddr, '40-AE-30-AF-C8-72') + self.assertIsInstance(status.lan_macaddress, EUI48) + self.assertEqual(status.wan_macaddr, '40-AE-30-AF-C8-73') + self.assertEqual(status.lan_ipv4_addr, '192.168.2.1') + self.assertIsInstance(status.lan_ipv4_address, IPv4Address) + self.assertEqual(status.wan_ipv4_addr, '192.168.0.68') + self.assertEqual(status.conn_type, 'Router/AP') + self.assertTrue(status.wifi_2g_enable) + self.assertEqual(status.clients_total, 3) + self.assertEqual(status.wifi_clients_total, 3) + self.assertEqual(status.wired_total, 0) + self.assertEqual(len(status.devices), 3) + + device = status.devices[0] + self.assertIsInstance(device, Device) + self.assertEqual(device.type, Connection.HOST_2G) + self.assertEqual(device.hostname, 'LGwebOSTV') + self.assertEqual(device.ipaddr, '192.168.2.100') + self.assertEqual(device.macaddr, '60-75-6C-A1-4A-82') + + device = status.devices[2] + self.assertEqual(device.hostname, 'Redmi-Note-13-Pro-5G') + self.assertEqual(device.ipaddr, '192.168.2.102') + self.assertEqual(device.macaddr, 'D6-02-54-6B-67-C4') + + +if __name__ == '__main__': + main() diff --git a/tplinkrouterc6u/__init__.py b/tplinkrouterc6u/__init__.py index bba8193..17ec3c0 100644 --- a/tplinkrouterc6u/__init__.py +++ b/tplinkrouterc6u/__init__.py @@ -11,6 +11,7 @@ from tplinkrouterc6u.client.vr import TPLinkVRClient from tplinkrouterc6u.client.vr400v2 import TPLinkVR400v2Client from tplinkrouterc6u.client.c80 import TplinkC80Router +from tplinkrouterc6u.client.wr844n import TplinkWR844NRouter from tplinkrouterc6u.client.c5400x import TplinkC5400XRouter from tplinkrouterc6u.client.c3200 import TplinkC3200Router from tplinkrouterc6u.client.c1200 import TplinkC1200Router diff --git a/tplinkrouterc6u/client/wr844n.py b/tplinkrouterc6u/client/wr844n.py new file mode 100644 index 0000000..036ada9 --- /dev/null +++ b/tplinkrouterc6u/client/wr844n.py @@ -0,0 +1,112 @@ +import re +from logging import Logger +from urllib import parse +from urllib.parse import urlparse + +from tplinkrouterc6u.client.c80 import TplinkC80Router +from tplinkrouterc6u.common.dataclass import Device, Status +from tplinkrouterc6u.common.helper import get_ip, get_mac +from tplinkrouterc6u.common.package_enum import Connection + + +class TplinkWR844NRouter(TplinkC80Router): + """Client for TL-WR844N firmware that returns plaintext C80-style data.""" + + BLOCK_REGEX = re.compile(r'id (\d+\|\d,\d,\d)\r\n(.*?)(?=\r\nid \d+\||$)', re.DOTALL) + + def __init__(self, host: str, password: str, username: str = 'admin', logger: Logger = None, + verify_ssl: bool = True, timeout: int = 30) -> None: + super().__init__(host, password, username, logger, verify_ssl, timeout) + + def supports(self) -> bool: + try: + response = self.request(2, 1, data='0|1,0,0') + return response.status_code == 200 and self._is_wr844n_response(response.text) + except Exception: + return False + + def get_status(self) -> Status: + request_text = '#'.join([ + '1|1,0,0', + '4|1,0,0', + '9|1,0,0', + '23|1,0,0', + '0|1,0,0', + ]) + response_text = self._request_plaintext(request_text) + data_blocks = self._parse_data_blocks(response_text) + + mac_info = self._parse_last_values_from_block(data_blocks.get('1|1,0,0', [])) + lan_info = self._parse_last_values_from_block(data_blocks.get('4|1,0,0', [])) + wan_info = self._parse_last_values_from_block(data_blocks.get('23|1,0,0', [])) + + devices = self._parse_dhcp_devices(data_blocks.get('9|1,0,0', [])) + + status = Status() + status._lan_macaddr = get_mac(mac_info.get('mac 0', '00-00-00-00-00-00')) + status._wan_macaddr = get_mac(mac_info.get('mac 1', '00-00-00-00-00-00')) + status._lan_ipv4_addr = get_ip(lan_info.get('ip') or self._host_ip()) + status._wan_ipv4_addr = get_ip(wan_info.get('ip') or self._host_ip()) + + gateway = wan_info.get('gateway') or lan_info.get('gateway') + if gateway and gateway != '0.0.0.0': + status._wan_ipv4_gateway = get_ip(gateway) + + uptime = wan_info.get('upTime') + status.wan_ipv4_uptime = int(uptime) // 100 if uptime and uptime.isdigit() else None + status.devices = devices + status.wired_total = 0 + status.wifi_clients_total = len(devices) + status.clients_total = len(devices) + status.wifi_2g_enable = True + status.conn_type = 'Router/AP' + return status + + def _request_plaintext(self, text: str) -> str: + body = self._encrypt_body(text) + response = self.request(2, 1, True, data=body) + return self._decrypt_data(response.text) + + def _decrypt_data(self, encrypted_text: str) -> str: + if self._is_plain_response(encrypted_text): + return encrypted_text + return super()._decrypt_data(encrypted_text) + + def _parse_data_blocks(self, response_text: str) -> dict[str, list[str]]: + matches = TplinkWR844NRouter.BLOCK_REGEX.findall(response_text) + return {match[0]: [line for line in match[1].strip().split('\r\n') if line] for match in matches} + + def _parse_dhcp_devices(self, response_data: list[str]) -> list[Device]: + devices: list[Device] = [] + for item in self._parse_response_to_dict(response_data): + ip = item.get('ip') + mac = item.get('mac') + if not ip or not mac or mac == '00-00-00-00-00-00': + continue + devices.append(Device(Connection.HOST_2G, get_mac(mac), get_ip(ip), + parse.unquote(item.get('hostName', '')))) + return devices + + def _parse_last_values(self, text: str) -> dict[str, str]: + return self._parse_last_values_from_block([line for line in text.splitlines() if line]) + + def _parse_last_values_from_block(self, lines: list[str]) -> dict[str, str]: + values: dict[str, str] = {} + for line in lines: + if line == '00000' or line.startswith('id '): + continue + key, _, value = line.rpartition(' ') + if key: + values[key] = value.strip() + return values + + def _host_ip(self) -> str: + return urlparse(self.host).hostname or '0.0.0.0' + + @staticmethod + def _is_wr844n_response(text: str) -> bool: + return TplinkWR844NRouter._is_plain_response(text) and 'modelName TL-WR844N' in text + + @staticmethod + def _is_plain_response(text: str) -> bool: + return isinstance(text, str) and text.startswith('00000\r\n') diff --git a/tplinkrouterc6u/provider.py b/tplinkrouterc6u/provider.py index 5cce1ff..9c6fed4 100644 --- a/tplinkrouterc6u/provider.py +++ b/tplinkrouterc6u/provider.py @@ -16,6 +16,7 @@ from tplinkrouterc6u.client.c3200 import TplinkC3200Router from tplinkrouterc6u.client.c1200 import TplinkC1200Router from tplinkrouterc6u.client.c80 import TplinkC80Router +from tplinkrouterc6u.client.wr844n import TplinkWR844NRouter from tplinkrouterc6u.client.vr import TPLinkVRClient from tplinkrouterc6u.client.vr400v2 import TPLinkVR400v2Client from tplinkrouterc6u.client.r import TPLinkRClient @@ -81,6 +82,7 @@ def get_clients() -> dict[str, type[AbstractRouter]]: TplinkRouterSG.__name__: TplinkRouterSG, TplinkRouterV1_11.__name__: TplinkRouterV1_11, TplinkRouter.__name__: TplinkRouter, + TplinkWR844NRouter.__name__: TplinkWR844NRouter, TplinkC80Router.__name__: TplinkC80Router, TplinkWDRRouter.__name__: TplinkWDRRouter, TplinkRE330Router.__name__: TplinkRE330Router,