diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 8ec6458..1680844 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -1,6 +1,6 @@ repos: - repo: https://github.com/ambv/black - rev: 21.7b0 + rev: 22.3.0 hooks: - id: black - repo: https://github.com/PyCQA/flake8 diff --git a/arris_tg2492lg/connect_box.py b/arris_tg2492lg/connect_box.py index 7cf718b..5e40a23 100644 --- a/arris_tg2492lg/connect_box.py +++ b/arris_tg2492lg/connect_box.py @@ -9,6 +9,7 @@ from dataclasses import dataclass from datetime import datetime from typing import Any, List, Optional +from urllib.parse import quote from yarl import URL from .const import ( @@ -35,11 +36,10 @@ def __init__(self, websession: ClientSession, hostname: str, password: str): self._credential: Optional[Credential] = None async def async_login(self) -> str: - arg_string = f"{USERNAME}:{self._password}" + arg_string = f"{quote(USERNAME)}:{quote(self._password)}" arg = base64.b64encode(arg_string.encode("utf-8")).decode("ascii") - params = {"arg": arg, "_n": self._nonce} - async with self._websession.get(f"{self._hostname}/login", params=params) as response: + async with self._websession.get(f"{self._hostname}/login?arg={arg}&_n={self._nonce}") as response: response.raise_for_status() token = await response.text() diff --git a/tests/test_connect_box.py b/tests/test_connect_box.py index 80153c7..e57b720 100644 --- a/tests/test_connect_box.py +++ b/tests/test_connect_box.py @@ -59,6 +59,51 @@ async def get_html_response(request): await connect_box.async_login() +async def test_login_url_replaces_special_characters_in_password(aiohttp_client): + """Validates that special characters in the password are replaced. + The webpage of the router replaces special characters in both the username and password using the '%xx' escape. + """ + + async def login_result(request): + login_result.url = str(request.url) + return await _get_credential(request) + + login_result.url = "" + + app = web.Application() + app.router.add_get("/login", login_result) + client = await aiohttp_client(app) + + connect_box = ConnectBox(client.session, f"http://{client.host}:{client.port}", "&=") + + token = await connect_box.async_login() + + assert token == "eyJuYW1lIjogImFkbWluIn0=" # base64 representation of: {"name": "admin"} + assert login_result.url.startswith(f"http://{client.host}:{client.port}/login?arg=YWRtaW46JTI2JTNE&_n=") + + +async def test_login_does_not_url_encode_base64(aiohttp_client): + """Validate that the characters in the base64 arg parameter are not replaced. + The portal of the router does not replace special characters (=) in the base64 encoded parameter. + """ + + async def login_result(request): + login_result.url = str(request.url) + return await _get_credential(request) + + login_result.url = "" + + app = web.Application() + app.router.add_get("/login", login_result) + client = await aiohttp_client(app) + + connect_box = ConnectBox(client.session, f"http://{client.host}:{client.port}", "secret2") + + await connect_box.async_login() + + assert login_result.url.startswith(f"http://{client.host}:{client.port}/login?arg=YWRtaW46c2VjcmV0Mg==&_n=") + + async def test_get_connected_devices(aiohttp_client): app = web.Application() app.router.add_get("/login", _get_credential)