Add TL-WR844N router client

Author: DeadmaroZ-TLOTL

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

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()

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

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')

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,