Repository: vanbalken/arris-tg2492lg (language: python)

Issue:
Validate login-response
Using this library on an unsupported router can result in un clear errors. The login call can result in an OK because the result is not validated. The result should be a Base 64 encoded JSON object. When the result is a webpage the webpage contents are stored as the token. Additional calls will then fail.


Discussion / hints:

Fix patch:
diff --git a/arris_tg2492lg/__init__.py b/arris_tg2492lg/__init__.py
index ae64911..36f0c9d 100644
--- a/arris_tg2492lg/__init__.py
+++ b/arris_tg2492lg/__init__.py
@@ -1,7 +1,8 @@
-from .connect_box import ConnectBox
+from .connect_box import ConnectBox, RouterInformation
 from .device import Device
+from .exception import ConnectBoxError, InvalidCredentialError
 
-__all__ = ["ConnectBox", "Device"]
+__all__ = ["ConnectBox", "RouterInformation", "Device", "ConnectBoxError", "InvalidCredentialError"]
 
 import logging
 from logging import NullHandler
diff --git a/arris_tg2492lg/connect_box.py b/arris_tg2492lg/connect_box.py
index 6ca6e98..7cf718b 100644
--- a/arris_tg2492lg/connect_box.py
+++ b/arris_tg2492lg/connect_box.py
@@ -16,11 +16,11 @@
     MAC_ADDRESS_OID,
     SERIAL_NUMBER_OID,
     SOFTWARE_VERSION_OID,
+    TOKEN_EXPIRATION,
     USERNAME,
-    TOKEN_EXPIRY_TIME,
 )
 from .device import Device
-from .exception import ConnectBoxError
+from .exception import ConnectBoxError, InvalidCredentialError
 from .mib_mapper import format_mac, to_devices
 
 _LOGGER = logging.getLogger(__name__)
@@ -46,6 +46,9 @@ async def async_login(self) -> str:
 
             _LOGGER.debug("Received token: %s", token)
 
+            self._credential = Credential(token)
+            self._credential.validate()
+
             return token
 
     async def async_logout(self) -> None:
@@ -101,11 +104,10 @@ async def async_get_router_information(self) -> RouterInformation:
         return router_information
 
     async def _async_get_credential(self) -> Credential:
-        if self._credential is None or self._credential.expiration_time <= datetime.now().timestamp():
-            token = await self.async_login()
-            self._credential = Credential(token, datetime.now().timestamp() + TOKEN_EXPIRY_TIME)
+        if self._credential is None or self._credential.is_expired():
+            await self.async_login()
 
-        return self._credential
+        return self._credential  # type: ignore
 
     async def _async_snmp_get(self, oids: List[str]) -> Any:
         credential = await self._async_get_credential()
@@ -130,10 +132,59 @@ async def _async_snmp_get(self, oids: List[str]) -> Any:
             return json.loads(data)
 
 
-@dataclass
 class Credential:
-    token: str
-    expiration_time: float
+    def __init__(self, token: str):
+        self._token = token
+        self._created_at = datetime.now()
+
+    def is_expired(self) -> bool:
+        expires = self._created_at + TOKEN_EXPIRATION
+        return datetime.now() > expires
+
+    def validate(self) -> None:
+        try:
+            self._token_as_json()
+        except Exception as exc:
+            raise InvalidCredentialError from exc
+
+    def is_multi_login(self) -> bool:
+        """Return if the user is already logged in."""
+
+        is_logged_in_value = self._is_logged_in()
+
+        return is_logged_in_value > 0
+
+    @property
+    def token(self) -> str:
+        return self._token
+
+    def _is_logged_in(self) -> int:
+        """Return if the user is already logged in.
+
+        Reverse engineered from a javascript library of the routers web application. The response of this method doesn't
+        indicate if a consecutive call will fail or succeed. Method is kept for reference.
+        """
+        token_json = self._token_as_json()
+
+        if "muti" in token_json:
+            muti = token_json["muti"]
+            con_type = token_json["conType"]
+            gw_wan = token_json["gwWan"]
+
+            if gw_wan == "f" and con_type == "LAN" and muti == "GW_WAN":
+                return 2  # Remote user already login.
+            elif gw_wan == "t" and muti == "LAN":
+                return 1  # Local user already login.
+            if gw_wan == "f" and con_type == "LAN" and muti == "LAN":
+                return 3  # Other local user already login.
+            elif gw_wan == "t" and muti == "GW_WAN":
+                return 4  # Other remote user already login.
+
+        return 0
+
+    def _token_as_json(self) -> Any:
+        decoded_token = base64.b64decode(self._token)
+        return json.loads(decoded_token)
 
 
 @dataclass
diff --git a/arris_tg2492lg/const.py b/arris_tg2492lg/const.py
index 47d7c98..96e314c 100644
--- a/arris_tg2492lg/const.py
+++ b/arris_tg2492lg/const.py
@@ -1,7 +1,10 @@
 __version__ = "2.0.0"
 
+from datetime import timedelta
+
+
 USERNAME = "admin"
-TOKEN_EXPIRY_TIME = 5 * 60 * 1000
+TOKEN_EXPIRATION = timedelta(minutes=5)
 
 # For an overview of all Arris OIDs see: https://mibs.observium.org/mib/ARRIS-ROUTER-DEVICE-MIB/#
 ARRIS_ROUTER_MIB = "1.3.6.1.4.1.4115.1.20.1"
diff --git a/arris_tg2492lg/exception.py b/arris_tg2492lg/exception.py
index 00a20fa..6167d05 100644
--- a/arris_tg2492lg/exception.py
+++ b/arris_tg2492lg/exception.py
@@ -1,2 +1,6 @@
 class ConnectBoxError(Exception):
     """General Connect Box exception occurred."""
+
+
+class InvalidCredentialError(ConnectBoxError):
+    """Login credential is invalid."""


Test patch:
diff --git a/tests/test_connect_box.py b/tests/test_connect_box.py
index b49acb0..1b34dbb 100644
--- a/tests/test_connect_box.py
+++ b/tests/test_connect_box.py
@@ -1,3 +1,4 @@
+import base64
 import os
 import pytest
 
@@ -5,25 +6,63 @@
 from pathlib import Path
 
 from arris_tg2492lg.connect_box import ConnectBox
+from arris_tg2492lg.exception import InvalidCredentialError
 
 
-async def test_async_get_credential(aiohttp_client, loop):
+async def test_async_login_ok(aiohttp_client, loop):
     app = web.Application(loop=loop)
-    app.router.add_get("/login", get_credential)
+    app.router.add_get("/login", _get_credential)
     client = await aiohttp_client(app)
 
     connect_box = ConnectBox(client.session, f"http://{client.host}:{client.port}", "secret")
-    credential = await connect_box._async_get_credential()
+    token = await connect_box.async_login()
 
-    print(credential.token)
+    assert token == "eyJuYW1lIjogImFkbWluIn0="  # base64 representation of: {"name": "admin"}
 
-    assert credential.token == 'dummy_token'
+
+async def test_async_login_nok_wrong_password(aiohttp_client, loop):
+    """Test when an invalid password is provided.
+
+    The router responds with an Internal Server Error when a wrong password is provided.
+    """
+
+    def internal_server_error(request):
+        return web.Response(status=500)
+
+    app = web.Application(loop=loop)
+    app.router.add_get("/login", internal_server_error)
+    client = await aiohttp_client(app)
+
+    connect_box = ConnectBox(client.session, f"http://{client.host}:{client.port}", "wrong")
+
+    with pytest.raises(ClientResponseError):
+        await connect_box.async_login()
+
+
+async def test_async_login_nok_html_response(aiohttp_client, loop):
+    """Test when an invalid endpoint is used.
+
+    An unsupported browser could respond with html.
+    """
+
+    def get_html_response(request):
+        html_response = "<!DOCTYPE html><html><body>hello</body></html>"
+        return web.Response(text=html_response)
+
+    app = web.Application(loop=loop)
+    app.router.add_get("/login", get_html_response)
+    client = await aiohttp_client(app)
+
+    connect_box = ConnectBox(client.session, f"http://{client.host}:{client.port}", "secret")
+
+    with pytest.raises(InvalidCredentialError):
+        await connect_box.async_login()
 
 
 async def test_get_connected_devices(aiohttp_client, loop):
     app = web.Application(loop=loop)
-    app.router.add_get("/login", get_credential)
-    app.router.add_get("/getConnDevices", get_mock_data)
+    app.router.add_get("/login", _get_credential)
+    app.router.add_get("/getConnDevices", _get_mock_data)
     client = await aiohttp_client(app)
 
     connect_box = ConnectBox(client.session, f"http://{client.host}:{client.port}", "secret")
@@ -35,14 +74,14 @@ async def test_get_connected_devices(aiohttp_client, loop):
 async def test_get_connected_devices_throws_401_once(aiohttp_client, loop):
     def login_result(request):
         login_result.call_count += 1
-        return get_credential(request)
+        return _get_credential(request)
 
     def conn_devices_result(request):
         conn_devices_result.call_count += 1
         if conn_devices_result.call_count == 1:
             return web.Response(status=401)
         else:
-            return get_mock_data(request)
+            return _get_mock_data(request)
 
     login_result.call_count = 0
     conn_devices_result.call_count = 0
@@ -64,7 +103,7 @@ def conn_devices_result(request):
 async def test_get_connected_devices_throws_401_twice(aiohttp_client, loop):
     def login_result(request):
         login_result.call_count += 1
-        return get_credential(request)
+        return _get_credential(request)
 
     def conn_devices_result(request):
         conn_devices_result.call_count += 1
@@ -95,7 +134,7 @@ def get_logout_success(request):
     get_logout_success.call_count = 0
 
     app = web.Application(loop=loop)
-    app.router.add_get("/login", get_credential)
+    app.router.add_get("/login", _get_credential)
     app.router.add_get("/logout", get_logout_success)
     client = await aiohttp_client(app)
 
@@ -113,7 +152,7 @@ async def get_logout_failure(request):
     get_logout_failure.call_count = 0
 
     app = web.Application(loop=loop)
-    app.router.add_get("/login", get_credential)
+    app.router.add_get("/login", _get_credential)
     app.router.add_get("/logout", get_logout_failure)
     client = await aiohttp_client(app)
 
@@ -127,8 +166,8 @@ async def get_logout_failure(request):
 
 async def test_get_router_information(aiohttp_client, loop):
     app = web.Application(loop=loop)
-    app.router.add_get("/login", get_credential)
-    app.router.add_get("/snmpGet", get_mock_router_information)
+    app.router.add_get("/login", _get_credential)
+    app.router.add_get("/snmpGet", _get_mock_router_information)
     client = await aiohttp_client(app)
 
     connect_box = ConnectBox(client.session, f"http://{client.host}:{client.port}", "secret")
@@ -140,18 +179,20 @@ async def test_get_router_information(aiohttp_client, loop):
     assert router_information.serial_number == "ABCD12345678"
 
 
-def get_credential(request):
-    return web.Response(text='dummy_token')
+def _get_credential(request):
+    dummy_token = base64.b64encode('{"name": "admin"}'.encode("utf-8")).decode("ascii")
+
+    return web.Response(text=dummy_token)
 
 
-def get_mock_data(request):
+def _get_mock_data(request):
     current_path = Path(os.path.dirname(os.path.realpath(__file__)))
     test_data_path = current_path / "getConnDevices-response.json"
 
     return web.Response(text=test_data_path.read_text())
 
 
-def get_mock_router_information(request):
+def _get_mock_router_information(request):
     current_path = Path(os.path.dirname(os.path.realpath(__file__)))
     test_data_path = current_path / "snmpGet-response.json"