diff --git a/README.md b/README.md index fc66845..4312279 100644 --- a/README.md +++ b/README.md @@ -33,6 +33,7 @@ The Sagemcom F@st series is used by multiple cable companies, where some cable c | Sagemcom F@st 5370e | Telia | sha512 | | | Sagemcom F@st 5380 | TDC | md5 | | | Sagemcom F@st 5566 | Bell (Home Hub 3000) | md5 | username: guest, password: "" | +| Sagemcom F@st 5598 | YouFibre | None / New API | username: admin, password: "" | | Sagemcom F@st 5688T | Salt (FibreBox_X6) | sha512 | username: admin | | Sagemcom F@st 5689 | Bell (Home Hub 4000) | md5 | username: admin, password: "" | | Sagemcom F@st 5689E | Bell (Giga Hub) | sha512 | username: admin, password: "" | @@ -63,7 +64,7 @@ The following script can be used as a quickstart. ```python import asyncio from sagemcom_api.client import SagemcomClient -from sagemcom_api.enums import EncryptionMethod +from sagemcom_api.enums import ApiMode, EncryptionMethod from sagemcom_api.exceptions import NonWritableParameterException HOST = "" @@ -71,9 +72,17 @@ USERNAME = "" PASSWORD = "" ENCRYPTION_METHOD = EncryptionMethod.SHA512 # or EncryptionMethod.MD5 VALIDATE_SSL_CERT = True +API_MODE = ApiMode.AUTO # auto, legacy or rest async def main() -> None: - async with SagemcomClient(HOST, USERNAME, PASSWORD, ENCRYPTION_METHOD, verify_ssl=VALIDATE_SSL_CERT) as client: + async with SagemcomClient( + HOST, + USERNAME, + PASSWORD, + ENCRYPTION_METHOD, + api_mode=API_MODE, + verify_ssl=VALIDATE_SSL_CERT, + ) as client: try: await client.login() except Exception as exception: # pylint: disable=broad-except @@ -119,11 +128,23 @@ asyncio.run(main()) ## Advanced +### API Mode + +The client supports two API variants: + +- `ApiMode.LEGACY`: original `/cgi/json-req` API with XPath support +- `ApiMode.REST`: newer `/api/v1/*` API used by newer firmwares +- `ApiMode.AUTO` (default): tries legacy first, then falls back to REST when the legacy endpoint is unavailable + +When REST mode is active, high-level helpers like `get_device_info()`, `get_hosts()` and `reboot()` are supported. XPath-based methods (`get_value_by_xpath`, `set_value_by_xpath`, `get_values_by_xpaths`) are legacy-only. + +REST firmware sets its session cookie on the router's IP address. Sessions created by the client accept such cookies in REST mode only; if you pass your own `session` for a REST router, create it with `CookieJar(unsafe=True)`. + ### Determine the EncryptionMethod If you are not sure which encryption method to use, you can leave it empty or pass `None` and use `get_encryption_method` to determine the encryption method. -`get_encryption_method` will return an `EncryptionMethod` when a match is found. Best would be to use this function only during your initial investigation. +`get_encryption_method` will return an `EncryptionMethod` when a match is found, `EncryptionMethod.NONE` when the router uses the REST API, and `None` if no method matches. Best would be to use this function only during your initial investigation. This function will throw a `LoginTimeoutException` when no match is found, since this is still a HTTP Time Out. This could caused by the wrong encryption method, but also by trying to connect to an inaccessible host. diff --git a/sagemcom_api/client.py b/sagemcom_api/client.py index 3adf14d..ae6d016 100644 --- a/sagemcom_api/client.py +++ b/sagemcom_api/client.py @@ -2,6 +2,7 @@ from __future__ import annotations +import contextlib import hashlib import json import math @@ -18,6 +19,8 @@ ClientOSError, ClientSession, ClientTimeout, + ContentTypeError, + CookieJar, ServerDisconnectedError, TCPConnector, ) @@ -31,10 +34,11 @@ XMO_INVALID_SESSION_ERR, XMO_REQUEST_NO_ERR, ) -from .enums import EncryptionMethod +from .enums import ApiMode, EncryptionMethod from .exceptions import ( AuthenticationException, BadRequestException, + BaseSagemcomException, InvalidSessionException, LoginConnectionException, LoginRetryErrorException, @@ -68,6 +72,7 @@ def __init__( session: ClientSession | None = None, ssl: bool | None = False, verify_ssl: bool | None = True, + api_mode: ApiMode | str = ApiMode.AUTO, ): """Create a SagemCom client. @@ -76,10 +81,13 @@ def __init__( :param password: the password for your Sagemcom router :param authentication_method: the auth method of your Sagemcom router :param session: use a custom session, for example to configure the timeout + :param api_mode: one of auto, legacy or rest """ self.host = host self.username = username self.authentication_method = authentication_method + self.api_mode = ApiMode(api_mode) + self._active_api_mode: ApiMode = self.api_mode if self.api_mode != ApiMode.AUTO else ApiMode.LEGACY self.password = password self._current_nonce = None self._password_hash = self.__generate_hash(password) @@ -89,16 +97,40 @@ def __init__( self._session_id = 0 self._request_id = -1 - self.session = ( - session - if session - else ClientSession( - headers={"User-Agent": f"{DEFAULT_USER_AGENT}"}, - timeout=ClientTimeout(DEFAULT_TIMEOUT), - connector=TCPConnector(verify_ssl=verify_ssl if verify_ssl is not None else True), - ) + self._session_owned = session is None + self._verify_ssl = verify_ssl if verify_ssl is not None else True + self._session_allows_ip_cookies = self.api_mode == ApiMode.REST + self.session = session or self.__create_session(allow_ip_cookies=self._session_allows_ip_cookies) + + def __create_session(self, allow_ip_cookies: bool) -> ClientSession: + """Create a library-owned session for the selected API's cookie policy.""" + return ClientSession( + headers={"User-Agent": f"{DEFAULT_USER_AGENT}"}, + timeout=ClientTimeout(DEFAULT_TIMEOUT), + cookie_jar=CookieJar(unsafe=allow_ip_cookies), + connector=TCPConnector(verify_ssl=self._verify_ssl), ) + async def __switch_api_mode(self, api_mode: ApiMode) -> None: + """Select the active API and match a library-owned session's cookie policy to it. + + REST firmware sets its session cookie on the IP host, which needs an unsafe cookie jar, + but legacy firmware breaks when that jar is used. Caller-provided sessions are never replaced. + """ + self._active_api_mode = api_mode + allow_ip_cookies = api_mode == ApiMode.REST + if not self._session_owned or self._session_allows_ip_cookies == allow_ip_cookies: + return + + await self.session.close() + self.session = self.__create_session(allow_ip_cookies=allow_ip_cookies) + self._session_allows_ip_cookies = allow_ip_cookies + + @property + def active_api_mode(self) -> ApiMode: + """Return the API mode that is currently active.""" + return self._active_api_mode + async def __aenter__(self) -> SagemcomClient: """TODO.""" return self @@ -289,8 +321,8 @@ async def __api_request_async(self, actions, priority=False, idempotent=True, ig ActionErrorHandler.throw_if_error(result, ignore_unknown_paths=ignore_unknown_paths) return result - async def login(self): - """Login to the SagemCom F@st router using a username and password.""" + async def __legacy_login(self): + """Login to the legacy JSON-REQ API.""" actions = { "id": 0, "method": "logIn", @@ -333,28 +365,255 @@ async def login(self): raise UnauthorizedException(data) + @backoff.on_exception( + backoff.expo, + (ClientConnectorError, ClientOSError, ServerDisconnectedError), + max_tries=5, + ) + async def __rest_request(self, method: str, endpoint: str, data: dict[str, Any] | None = None): + """Call the REST API using form-encoded payloads.""" + url = f"{self.protocol}://{self.host}{endpoint}" + method = method.upper() + request_kwargs: dict[str, Any] = {} + if method in ("GET", "HEAD"): + request_kwargs["params"] = data or {} + elif method in ("POST", "PUT", "PATCH"): + request_kwargs["data"] = data or {} + + async with self.session.request(method, url, **request_kwargs) as response: + if response.status in (200, 204): + if response.status == 204: + return None + try: + return await response.json() + except (json.JSONDecodeError, ContentTypeError): + return await response.text() + + result = await response.text() + if response.status in (401, 403): + raise UnauthorizedException(result) + + if response.status == 404: + raise UnsupportedHostException(result) + + if response.status == 400: + raise BadRequestException(result) + + raise UnknownException(result) + + async def __rest_login(self): + """Login to routers exposing the newer REST API.""" + try: + await self.__rest_request( + "POST", + "/api/v1/login", + data={"login": self.username, "password": self.password}, + ) + except UnauthorizedException as exception: + raise AuthenticationException(str(exception)) from exception + return True + + async def __fallback_to_rest_login(self, legacy_exception: Exception): + """Login over REST after legacy failed, restoring legacy state when REST fails too.""" + await self.__switch_api_mode(ApiMode.REST) + try: + return await self.__rest_login() + except AuthenticationException: + await self.__switch_api_mode(ApiMode.LEGACY) + raise + except Exception as rest_exception: + await self.__switch_api_mode(ApiMode.LEGACY) + # Without a REST endpoint this is a legacy router, so keep the error legacy users always got. + raise legacy_exception from rest_exception + + @staticmethod + def __first_value(data: dict[str, Any], *keys: str) -> Any: + """Return the first non-None value from data for the given keys.""" + for key in keys: + if key in data and data[key] is not None: + return data[key] + return None + + @staticmethod + def __to_bool(value: Any, default: bool = True) -> bool: + """Convert mixed payload boolean values to bool.""" + if value is None: + return default + if isinstance(value, bool): + return value + if isinstance(value, (int, float)): + return bool(value) + if isinstance(value, str): + return value.strip().lower() in ("1", "true", "yes", "on", "up") + return default + + def __build_rest_device(self, entry: dict[str, Any], interface_type: str | None) -> Device: + """Map a REST host entry to Device.""" + detected_interface = self.__first_value( + entry, + "interface_type", + "interfaceType", + "interface", + "connectionType", + "connection_type", + "type", + ) + if interface_type is None and isinstance(detected_interface, str): + normalized = detected_interface.lower() + if "wifi" in normalized or "wireless" in normalized or "wlan" in normalized: + interface_type = "wifi" + elif "ethernet" in normalized or "eth" in normalized or "lan" in normalized: + interface_type = "ethernet" + else: + interface_type = detected_interface + + return Device( + uid=self.__first_value(entry, "id", "uid"), + phys_address=self.__first_value(entry, "macAddress", "mac_address", "phys_address"), + ip_address=self.__first_value(entry, "ipAddress", "ip_address"), + host_name=self.__first_value(entry, "hostname", "host_name", "name"), + user_host_name=self.__first_value(entry, "friendlyname", "friendly_name", "user_host_name"), + active=self.__to_bool(self.__first_value(entry, "active", "isActive"), True), + interface_type=interface_type, + detected_device_type=self.__first_value(entry, "devicetype", "deviceType", "detected_device_type"), + ) + + def __extract_rest_home_hosts(self, data: Any) -> list[Device] | None: + """Parse /api/v1/home hosts payload, returning None when it carries no host lists.""" + home = data[0] if isinstance(data, list) and data else data + if not isinstance(home, dict): + return None + + host_lists = [ + (entries, interface_type) + for key, interface_type in ( + ("wirelessListDevice", "wifi"), + ("ethernetListDevice", "ethernet"), + ) + if isinstance(entries := home.get(key), list) + ] + if not host_lists: + return None + + return [ + self.__build_rest_device(entry, interface_type) + for entries, interface_type in host_lists + for entry in entries + if isinstance(entry, dict) + ] + + def __extract_rest_hosts(self, data: Any) -> list[Device]: + """Parse /api/v1/hosts payload.""" + hosts: list[dict[str, Any]] + if isinstance(data, list): + hosts = [entry for entry in data if isinstance(entry, dict)] + elif isinstance(data, dict): + raw_hosts = self.__first_value( + data, + "hosts", + "Hosts", + "list", + "listDevice", + "list_device", + "devices", + ) + if not isinstance(raw_hosts, list): + raise UnknownException("Invalid response from /api/v1/hosts") + hosts = [entry for entry in raw_hosts if isinstance(entry, dict)] + else: + raise UnknownException("Invalid response from /api/v1/hosts") + + return [self.__build_rest_device(entry, None) for entry in hosts] + + @staticmethod + def __is_unsupported_rest_endpoint(exception: Exception, endpoint: str) -> bool: + """Recognize responses that explicitly reject a REST endpoint.""" + if isinstance(exception, UnsupportedHostException): + return True + if not isinstance(exception, BadRequestException): + return False + + try: + payload = json.loads(str(exception)) + except (json.JSONDecodeError, TypeError): + return False + + details = payload.get("exception") if isinstance(payload, dict) else None + return isinstance(details, dict) and details.get("domain") == endpoint + + def __should_fallback_to_rest(self, exception: Exception) -> bool: + """Return True when legacy API failure indicates a REST-only router.""" + if isinstance(exception, UnsupportedHostException): + return True + + if isinstance(exception, (UnknownException, BadRequestException)): + content = str(exception).lower() + return "service unavailable" in content or " EncryptionMethod | None: """Determine which encryption method to use for authentication and set it directly. + Returns EncryptionMethod.NONE for REST firmware and None when no legacy method matches. + :raises LoginRetryErrorException: the router has locked logins after too many attempts. Probing further would only extend the lock, so detection stops here. """ + if self.api_mode == ApiMode.REST: + return EncryptionMethod.NONE + for encryption_method in EncryptionMethod: + if encryption_method == EncryptionMethod.NONE: + continue try: self.authentication_method = encryption_method self._password_hash = self.__generate_hash(self.password, encryption_method) - await self.login() + if self.api_mode == ApiMode.AUTO: + self._active_api_mode = ApiMode.LEGACY + await self.__legacy_login() + else: + await self.login() self._server_nonce = "" self._session_id = 0 @@ -366,6 +625,17 @@ async def get_encryption_method(self): AuthenticationException, ): pass + except Exception as exception: + if self.api_mode != ApiMode.AUTO or not self.__should_fallback_to_rest(exception): + raise + + await self.__fallback_to_rest_login(exception) + self.authentication_method = EncryptionMethod.NONE + self._password_hash = self.__generate_hash(self.password, EncryptionMethod.NONE) + with contextlib.suppress(BaseSagemcomException): + await self.logout() + + return EncryptionMethod.NONE return None @@ -399,6 +669,8 @@ async def get_values_by_xpaths(self, xpaths: dict[str, str], options: dict | Non :param ignore_unknown_paths: return None for each path that does not exist instead of raising UnknownPathException; other errors are still raised """ + self.__ensure_legacy_api() + actions = [ { "id": i, @@ -441,6 +713,8 @@ async def set_values_by_xpaths(self, xpaths: dict[str, str], options: dict | Non :param xpaths: Dict of xpath expression to value :param options: optional options """ + self.__ensure_legacy_api() + actions = [ { "id": i, @@ -468,6 +742,31 @@ async def set_values_by_xpaths(self, xpaths: dict[str, str], options: dict | Non ) async def get_device_info(self) -> DeviceInfo: """Retrieve information about Sagemcom F@st device.""" + if self._active_api_mode == ApiMode.REST: + data = await self.__rest_request("GET", "/api/v1/device") + if not isinstance(data, list) or not data or not isinstance(data[0], dict): + raise UnknownException("Invalid response from /api/v1/device") + + device = data[0].get("device") + if not isinstance(device, dict): + raise UnknownException("Invalid response from /api/v1/device") + + running = device.get("running") + software_version = running.get("version") if isinstance(running, dict) else None + return DeviceInfo( + mac_address=device.get("wan_mac_address"), + serial_number=device.get("serialnumber"), + model_name=device.get("modelname"), + model_number=device.get("modelname"), + product_class=device.get("modelname"), + software_version=software_version, + hardware_version=device.get("hardware_version"), + manufacturer="Sagemcom", + up_time=device.get("uptime"), + first_use_date=device.get("firstusedate"), + reboot_count=device.get("numberofboots"), + ) + try: data = await self.get_value_by_xpath("Device/DeviceInfo") return DeviceInfo(**data["device_info"]) @@ -505,6 +804,26 @@ async def get_device_info(self) -> DeviceInfo: ) async def get_hosts(self, only_active: bool | None = False) -> list[Device]: """Retrieve hosts connected to Sagemcom F@st device.""" + if self._active_api_mode == ApiMode.REST: + home_endpoint = "/api/v1/home" + devices: list[Device] | None = None + try: + home_data = await self.__rest_request("GET", home_endpoint) + except (BadRequestException, UnsupportedHostException) as exception: + if not self.__is_unsupported_rest_endpoint(exception, home_endpoint): + raise + else: + devices = self.__extract_rest_home_hosts(home_data) + + if devices is None: + hosts_data = await self.__rest_request("GET", "/api/v1/hosts") + devices = self.__extract_rest_hosts(hosts_data) + + if only_active: + return [device for device in devices if device.active is True] + + return devices + data = await self.get_value_by_xpath("Device/Hosts/Hosts", options={"capability-flags": {"interface": True}}) devices = [Device(**d) for d in data] @@ -527,6 +846,7 @@ async def get_hosts(self, only_active: bool | None = False) -> list[Device]: ) async def get_port_mappings(self) -> list[PortMapping]: """Retrieve configured Port Mappings on Sagemcom F@st device.""" + self.__ensure_legacy_api() data = await self.get_value_by_xpath("Device/NAT/PortMappings") port_mappings = [PortMapping(**p) for p in data] @@ -561,6 +881,9 @@ async def get_logs(self) -> str: async def reboot(self): """Reboot Sagemcom F@st device.""" + if self._active_api_mode == ApiMode.REST: + return await self.__rest_request("POST", "/api/v1/device/reboot") + action = { "id": 0, "method": "reboot", diff --git a/sagemcom_api/enums.py b/sagemcom_api/enums.py index 99ac8ae..594620b 100644 --- a/sagemcom_api/enums.py +++ b/sagemcom_api/enums.py @@ -10,3 +10,13 @@ class EncryptionMethod(StrEnum): MD5 = "MD5" MD5_NONCE = "MD5_NONCE" SHA512 = "SHA512" + NONE = "NONE" + + +@unique +class ApiMode(StrEnum): + """API mode to use when communicating with the router.""" + + AUTO = "auto" + LEGACY = "legacy" + REST = "rest" diff --git a/tests/unit/test_client_basic.py b/tests/unit/test_client_basic.py index 7b24cd5..699e5f8 100644 --- a/tests/unit/test_client_basic.py +++ b/tests/unit/test_client_basic.py @@ -6,19 +6,43 @@ from unittest.mock import AsyncMock, MagicMock import pytest -from aiohttp import ClientSession +from aiohttp import ClientSession, CookieJar +from yarl import URL from sagemcom_api.client import SagemcomClient -from sagemcom_api.enums import EncryptionMethod +from sagemcom_api.enums import ApiMode, EncryptionMethod from sagemcom_api.exceptions import ( AuthenticationException, + BadRequestException, InvalidSessionException, LoginRetryErrorException, NonWritableParameterException, + UnauthorizedException, + UnknownException, UnknownPathException, + UnsupportedHostException, ) +def test_constructor_preserves_legacy_positional_argument_order(): + """Existing positional session and TLS arguments should keep their meaning.""" + session = MagicMock(spec=ClientSession) + + client = SagemcomClient( + "192.168.1.1", + "admin", + "password", + EncryptionMethod.MD5, + session, + True, + False, + ) + + assert client.session is session + assert client.protocol == "https" + assert client.api_mode == ApiMode.AUTO + + @pytest.mark.asyncio async def test_response_with_invalid_utf8_is_parsed(): """Test malformed UTF-8 in a router response does not discard valid JSON.""" @@ -42,6 +66,165 @@ async def test_response_with_invalid_utf8_is_parsed(): assert result["reply"]["value"] == "invalid � byte" +@pytest.mark.asyncio +@pytest.mark.parametrize("method", ["get", "HEAD"]) +async def test_rest_request_query_methods_use_params(method): + """GET and HEAD requests should carry values in the query string.""" + response = AsyncMock() + response.status = 204 + response.__aenter__ = AsyncMock(return_value=response) + response.__aexit__ = AsyncMock(return_value=None) + + session = MagicMock(spec=ClientSession) + session.request.return_value = response + client = SagemcomClient( + host="192.168.1.1", + username="admin", + password="admin", + session=session, + ) + request_data = {"filter": "online devices"} + + await client._SagemcomClient__rest_request(method, "/api/v1/hosts", request_data) + + session.request.assert_called_once_with( + method.upper(), + "http://192.168.1.1/api/v1/hosts", + params=request_data, + ) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("method", ["post", "PUT", "patch"]) +async def test_rest_request_body_methods_use_form_data(method): + """Body methods should pass mappings as form data.""" + response = AsyncMock() + response.status = 204 + response.__aenter__ = AsyncMock(return_value=response) + response.__aexit__ = AsyncMock(return_value=None) + + session = MagicMock(spec=ClientSession) + session.request.return_value = response + client = SagemcomClient( + host="192.168.1.1", + username="admin", + password="admin", + session=session, + ) + request_data = {"login": "admin", "password": "secret value"} + + await client._SagemcomClient__rest_request(method, "/api/v1/login", request_data) + + session.request.assert_called_once_with( + method.upper(), + "http://192.168.1.1/api/v1/login", + data=request_data, + ) + + +@pytest.mark.asyncio +async def test_rest_request_bad_request_raises_bad_request_exception(): + """HTTP 400 should remain distinct from authentication failures.""" + response = AsyncMock() + response.status = 400 + response.text = AsyncMock(return_value="bad request") + response.__aenter__ = AsyncMock(return_value=response) + response.__aexit__ = AsyncMock(return_value=None) + + session = MagicMock(spec=ClientSession) + session.request.return_value = response + client = SagemcomClient( + host="192.168.1.1", + username="admin", + password="admin", + session=session, + ) + + with pytest.raises(BadRequestException, match="bad request"): + await client._SagemcomClient__rest_request("GET", "/api/v1/hosts") + + +@pytest.mark.asyncio +@pytest.mark.parametrize("status", [401, 403]) +async def test_rest_request_unauthorized_statuses_raise_unauthorized(status): + """HTTP authentication statuses should raise UnauthorizedException.""" + response = AsyncMock() + response.status = status + response.text = AsyncMock(return_value="unauthorized") + response.__aenter__ = AsyncMock(return_value=response) + response.__aexit__ = AsyncMock(return_value=None) + + session = MagicMock(spec=ClientSession) + session.request.return_value = response + client = SagemcomClient( + host="192.168.1.1", + username="admin", + password="admin", + session=session, + ) + + with pytest.raises(UnauthorizedException, match="unauthorized"): + await client._SagemcomClient__rest_request("GET", "/api/v1/hosts") + + +@pytest.mark.asyncio +@pytest.mark.parametrize("api_mode", [ApiMode.LEGACY, ApiMode.AUTO]) +async def test_default_legacy_cookie_policy_rejects_ip_cookies(api_mode): + """Legacy-capable sessions should retain aiohttp's safe cookie policy.""" + client = SagemcomClient( + host="192.168.1.1", + username="admin", + password="admin", + authentication_method=EncryptionMethod.MD5, + api_mode=api_mode, + ) + try: + router_url = URL("http://192.168.1.1/") + client.session.cookie_jar.update_cookies({"session": "legacy-cookie"}, response_url=router_url) + + assert "session" not in client.session.cookie_jar.filter_cookies(router_url) + finally: + await client.close() + + +@pytest.mark.asyncio +async def test_default_rest_cookie_policy_accepts_ip_cookies(): + """A library-owned REST session should retain cookies from an IP host.""" + client = SagemcomClient( + host="192.168.1.1", + username="admin", + password="admin", + api_mode=ApiMode.REST, + ) + try: + router_url = URL("http://192.168.1.1/") + client.session.cookie_jar.update_cookies({"session": "rest-cookie"}, response_url=router_url) + + cookies = client.session.cookie_jar.filter_cookies(router_url) + assert cookies["session"].value == "rest-cookie" + finally: + await client.close() + + +@pytest.mark.asyncio +async def test_supplied_rest_session_and_cookie_jar_are_unchanged(): + """REST mode should preserve caller-owned session policy and identity.""" + cookie_jar = CookieJar() + session = ClientSession(cookie_jar=cookie_jar) + client = SagemcomClient( + host="192.168.1.1", + username="admin", + password="admin", + api_mode=ApiMode.REST, + session=session, + ) + try: + assert client.session is session + assert client.session.cookie_jar is cookie_jar + finally: + await client.close() + + @pytest.mark.asyncio async def test_login_success(mock_session_factory, login_success_response): """Test successful login with mocked session. @@ -182,6 +365,634 @@ async def test_login_with_preconfigured_fixture(mock_client_sha512): assert client._server_nonce == "abcdef1234567890" +@pytest.mark.asyncio +async def test_login_auto_fallback_preserves_supplied_cookie_jar(): + """AUTO fallback should not mutate a caller-owned session or cookie jar.""" + mock_session = MagicMock(spec=ClientSession) + mock_session.close = AsyncMock() + supplied_cookie_jar = CookieJar() + mock_session.cookie_jar = supplied_cookie_jar + + legacy_response = AsyncMock() + legacy_response.status = 503 + legacy_response.text = AsyncMock(return_value="503 Service Unavailable") + legacy_response.__aenter__ = AsyncMock(return_value=legacy_response) + legacy_response.__aexit__ = AsyncMock(return_value=None) + mock_session.post.return_value = legacy_response + + rest_response = AsyncMock() + rest_response.status = 204 + rest_response.text = AsyncMock(return_value="") + rest_response.__aenter__ = AsyncMock(return_value=rest_response) + rest_response.__aexit__ = AsyncMock(return_value=None) + mock_session.request.return_value = rest_response + + client = SagemcomClient( + host="192.168.1.1", + username="admin", + password="admin", + authentication_method=EncryptionMethod.MD5, + session=mock_session, + api_mode=ApiMode.AUTO, + ) + + result = await client.login() + + assert result is True + assert client.active_api_mode == ApiMode.REST + assert client.session is mock_session + assert client.session.cookie_jar is supplied_cookie_jar + assert mock_session.post.call_count == 1 + assert mock_session.request.call_count == 1 + + +@pytest.mark.asyncio +async def test_auto_fallback_replaces_library_owned_session_for_ip_cookies(): + """AUTO should replace only its own safe session when REST is selected.""" + client = SagemcomClient( + host="192.168.1.1", + username="admin", + password="admin", + api_mode=ApiMode.AUTO, + ) + original_session = client.session + client._SagemcomClient__legacy_login = AsyncMock(side_effect=UnknownException("service unavailable")) + client._SagemcomClient__rest_login = AsyncMock(return_value=True) + + try: + result = await client.login() + + assert result is True + assert client.session is not original_session + assert original_session.closed + + router_url = URL("http://192.168.1.1/") + client.session.cookie_jar.update_cookies({"session": "rest-cookie"}, response_url=router_url) + cookies = client.session.cookie_jar.filter_cookies(router_url) + assert cookies["session"].value == "rest-cookie" + finally: + await client.close() + + +@pytest.mark.asyncio +async def test_auto_failed_rest_fallback_restores_legacy_session(): + """A legacy router whose REST login also fails should keep its legacy error and safe session.""" + client = SagemcomClient( + host="192.168.1.1", + username="admin", + password="admin", + api_mode=ApiMode.AUTO, + ) + legacy_error = UnknownException("service unavailable") + client._SagemcomClient__legacy_login = AsyncMock(side_effect=legacy_error) + client._SagemcomClient__rest_request = AsyncMock(side_effect=UnsupportedHostException("not found")) + + try: + with pytest.raises(UnknownException) as exc_info: + await client.login() + + assert exc_info.value is legacy_error + assert client.active_api_mode == ApiMode.LEGACY + + router_url = URL("http://192.168.1.1/") + client.session.cookie_jar.update_cookies({"session": "legacy-cookie"}, response_url=router_url) + assert "session" not in client.session.cookie_jar.filter_cookies(router_url) + finally: + await client.close() + + +@pytest.mark.asyncio +async def test_auto_rest_fallback_rejected_credentials_raise_authentication_error(): + """A REST login rejection after fallback should surface as an authentication failure.""" + session = MagicMock(spec=ClientSession) + client = SagemcomClient( + host="192.168.1.1", + username="admin", + password="wrong", + session=session, + api_mode=ApiMode.AUTO, + ) + client._SagemcomClient__legacy_login = AsyncMock(side_effect=UnknownException("service unavailable")) + client._SagemcomClient__rest_request = AsyncMock(side_effect=UnauthorizedException("unauthorized")) + + with pytest.raises(AuthenticationException): + await client.login() + + assert client.active_api_mode == ApiMode.LEGACY + + +@pytest.mark.asyncio +async def test_get_encryption_method_returns_none_when_no_legacy_method_matches(): + """Legacy detection should keep returning None when every method is rejected.""" + session = MagicMock(spec=ClientSession) + client = SagemcomClient( + host="192.168.1.1", + username="admin", + password="wrong", + session=session, + api_mode=ApiMode.AUTO, + ) + client._SagemcomClient__legacy_login = AsyncMock(side_effect=AuthenticationException("rejected")) + client._SagemcomClient__rest_request = AsyncMock() + + assert await client.get_encryption_method() is None + client._SagemcomClient__rest_request.assert_not_awaited() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("method_name", ["login", "get_encryption_method"]) +async def test_auto_login_lockout_never_falls_back_to_rest(method_name, mock_session_factory, login_retry_error_response): + """A legacy login lockout must stop AUTO mode instead of trying REST.""" + mock_session = mock_session_factory([login_retry_error_response] * 3) + client = SagemcomClient("192.168.1.1", "admin", "admin", session=mock_session) + client._SagemcomClient__rest_request = AsyncMock() + + with pytest.raises(LoginRetryErrorException): + await getattr(client, method_name)() + + assert mock_session.post.call_count == 1 + client._SagemcomClient__rest_request.assert_not_awaited() + assert client.active_api_mode == ApiMode.LEGACY + + +@pytest.mark.asyncio +async def test_get_encryption_method_auto_detects_rest_firmware(): + """AUTO detection should report NONE only after a REST login succeeds.""" + session = MagicMock(spec=ClientSession) + client = SagemcomClient( + host="192.168.1.1", + username="admin", + password="admin", + session=session, + api_mode=ApiMode.AUTO, + ) + client._SagemcomClient__legacy_login = AsyncMock(side_effect=UnknownException("service unavailable")) + client._SagemcomClient__rest_request = AsyncMock(return_value=None) + + assert await client.get_encryption_method() == EncryptionMethod.NONE + assert client.authentication_method == EncryptionMethod.NONE + assert client.active_api_mode == ApiMode.REST + endpoints = [call.args[1] for call in client._SagemcomClient__rest_request.await_args_list] + assert endpoints == ["/api/v1/login", "/api/v1/logout"] + + +@pytest.mark.asyncio +async def test_get_encryption_method_rest_returns_none(): + """REST mode should immediately signal that no encryption method is needed.""" + client = SagemcomClient( + host="192.168.1.1", + username="admin", + password="admin", + api_mode=ApiMode.REST, + ) + + result = await client.get_encryption_method() + + assert result == EncryptionMethod.NONE + await client.close() + + +@pytest.mark.asyncio +async def test_probe_rest_bad_request_allows_legacy_encryption_detection(): + """A rejected REST probe should not prevent legacy mode detection.""" + rest_response = AsyncMock() + rest_response.status = 400 + rest_response.text = AsyncMock(return_value="REST unavailable") + rest_response.__aenter__ = AsyncMock(return_value=rest_response) + rest_response.__aexit__ = AsyncMock(return_value=None) + + session = MagicMock(spec=ClientSession) + session.request.return_value = rest_response + client = SagemcomClient( + host="192.168.1.1", + username="admin", + password="admin", + session=session, + api_mode=ApiMode.AUTO, + ) + client._SagemcomClient__legacy_login = AsyncMock(return_value=True) + + result = await client.get_encryption_method() + + assert result == EncryptionMethod.MD5 + client._SagemcomClient__legacy_login.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_dual_stack_auto_prefers_working_legacy_encryption(): + """AUTO encryption detection should prefer legacy when both APIs work.""" + rest_response = AsyncMock() + rest_response.status = 204 + rest_response.__aenter__ = AsyncMock(return_value=rest_response) + rest_response.__aexit__ = AsyncMock(return_value=None) + + session = MagicMock(spec=ClientSession) + session.request.return_value = rest_response + client = SagemcomClient( + host="192.168.1.1", + username="admin", + password="admin", + session=session, + api_mode=ApiMode.AUTO, + ) + client._SagemcomClient__legacy_login = AsyncMock(return_value=True) + + result = await client.get_encryption_method() + + assert result == EncryptionMethod.MD5 + client._SagemcomClient__legacy_login.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_legacy_mode_never_probes_rest_during_encryption_detection(): + """Explicit legacy mode should not silently switch transports.""" + session = MagicMock(spec=ClientSession) + client = SagemcomClient( + host="192.168.1.1", + username="admin", + password="admin", + session=session, + api_mode=ApiMode.LEGACY, + ) + client._SagemcomClient__legacy_login = AsyncMock(side_effect=UnsupportedHostException("legacy unavailable")) + client._SagemcomClient__rest_login = AsyncMock(return_value=True) + + with pytest.raises(UnsupportedHostException, match="legacy unavailable"): + await client.get_encryption_method() + + client._SagemcomClient__rest_login.assert_not_awaited() + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "payload", + [ + {}, + [], + ["not a mapping"], + [{"device": []}], + ], +) +async def test_device_info_rest_rejects_malformed_shapes(payload): + """Malformed required REST device containers should raise a library error.""" + response = AsyncMock() + response.status = 200 + response.json = AsyncMock(return_value=payload) + response.__aenter__ = AsyncMock(return_value=response) + response.__aexit__ = AsyncMock(return_value=None) + + session = MagicMock(spec=ClientSession) + session.request.return_value = response + client = SagemcomClient( + host="192.168.1.1", + username="admin", + password="admin", + session=session, + api_mode=ApiMode.REST, + ) + + with pytest.raises(UnknownException, match="/api/v1/device"): + await client.get_device_info() + + +@pytest.mark.asyncio +async def test_device_info_rest_ignores_malformed_optional_running(): + """Malformed optional running data should omit the software version.""" + response = AsyncMock() + response.status = 200 + response.json = AsyncMock( + return_value=[ + { + "device": { + "wan_mac_address": "aa:bb:cc:dd:ee:ff", + "running": [], + } + } + ] + ) + response.__aenter__ = AsyncMock(return_value=response) + response.__aexit__ = AsyncMock(return_value=None) + + session = MagicMock(spec=ClientSession) + session.request.return_value = response + client = SagemcomClient( + host="192.168.1.1", + username="admin", + password="admin", + session=session, + api_mode=ApiMode.REST, + ) + + device_info = await client.get_device_info() + + assert device_info.mac_address == "aa:bb:cc:dd:ee:ff" + assert device_info.software_version is None + + +@pytest.mark.asyncio +async def test_get_hosts_rest_mode(): + """get_hosts should parse wifi and ethernet devices on REST firmware.""" + mock_session = MagicMock(spec=ClientSession) + mock_session.close = AsyncMock() + + login_response = AsyncMock() + login_response.status = 204 + login_response.text = AsyncMock(return_value="") + login_response.__aenter__ = AsyncMock(return_value=login_response) + login_response.__aexit__ = AsyncMock(return_value=None) + + home_payload = [ + { + "wirelessListDevice": [ + { + "id": 1, + "hostname": "wifi-device", + "friendlyname": "wifi-device", + "macAddress": "aa:bb:cc:dd:ee:ff", + "ipAddress": "192.168.1.2", + "active": True, + "devicetype": "MISCELLANEOUS", + } + ], + "ethernetListDevice": [ + { + "id": 2, + "hostname": "lan-device", + "friendlyname": "lan-device", + "macAddress": "11:22:33:44:55:66", + "ipAddress": "192.168.1.3", + "active": True, + "devicetype": "MISCELLANEOUS", + } + ], + } + ] + hosts_response = AsyncMock() + hosts_response.status = 200 + hosts_response.json = AsyncMock(return_value=home_payload) + hosts_response.__aenter__ = AsyncMock(return_value=hosts_response) + hosts_response.__aexit__ = AsyncMock(return_value=None) + + mock_session.request.side_effect = [login_response, hosts_response] + + client = SagemcomClient( + host="192.168.1.1", + username="admin", + password="admin", + session=mock_session, + api_mode=ApiMode.REST, + ) + + await client.login() + devices = await client.get_hosts() + + assert len(devices) == 2 + assert devices[0].host_name == "wifi-device" + assert devices[0].interface_type == "wifi" + assert devices[1].host_name == "lan-device" + assert devices[1].interface_type == "ethernet" + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "home_payload", + [ + [{"unexpected": "shape"}], + [{"wirelessListDevice": None}], + ], +) +async def test_get_hosts_rest_fallbacks_to_hosts_endpoint(home_payload): + """/api/v1/hosts should be used when /api/v1/home response is invalid.""" + mock_session = MagicMock(spec=ClientSession) + mock_session.close = AsyncMock() + + login_response = AsyncMock() + login_response.status = 204 + login_response.text = AsyncMock(return_value="") + login_response.__aenter__ = AsyncMock(return_value=login_response) + login_response.__aexit__ = AsyncMock(return_value=None) + + home_response = AsyncMock() + home_response.status = 200 + home_response.json = AsyncMock(return_value=home_payload) + home_response.__aenter__ = AsyncMock(return_value=home_response) + home_response.__aexit__ = AsyncMock(return_value=None) + + hosts_payload = [ + { + "id": 7, + "hostname": "tablet", + "friendlyname": "tablet", + "macAddress": "aa:aa:aa:aa:aa:aa", + "ipAddress": "192.168.1.50", + "active": "true", + "interfaceType": "wireless", + "devicetype": "TABLET", + } + ] + hosts_response = AsyncMock() + hosts_response.status = 200 + hosts_response.json = AsyncMock(return_value=hosts_payload) + hosts_response.__aenter__ = AsyncMock(return_value=hosts_response) + hosts_response.__aexit__ = AsyncMock(return_value=None) + + mock_session.request.side_effect = [login_response, home_response, hosts_response] + + client = SagemcomClient( + host="192.168.1.1", + username="admin", + password="admin", + session=mock_session, + api_mode=ApiMode.REST, + ) + + await client.login() + devices = await client.get_hosts(only_active=True) + + assert len(devices) == 1 + assert devices[0].host_name == "tablet" + assert devices[0].interface_type == "wifi" + assert devices[0].active is True + + +@pytest.mark.asyncio +async def test_get_hosts_rest_fallbacks_on_home_400(): + """/api/v1/hosts should be tried when /api/v1/home returns HTTP 400.""" + mock_session = MagicMock(spec=ClientSession) + mock_session.close = AsyncMock() + + login_response = AsyncMock() + login_response.status = 204 + login_response.text = AsyncMock(return_value="") + login_response.__aenter__ = AsyncMock(return_value=login_response) + login_response.__aexit__ = AsyncMock(return_value=None) + + home_response = AsyncMock() + home_response.status = 400 + home_response.text = AsyncMock(return_value='{"exception":{"domain":"/api/v1/home"}}') + home_response.__aenter__ = AsyncMock(return_value=home_response) + home_response.__aexit__ = AsyncMock(return_value=None) + + hosts_payload = [ + { + "id": 3, + "hostname": "phone", + "friendlyname": "phone", + "macAddress": "de:ad:be:ef:00:01", + "ipAddress": "192.168.1.25", + "active": True, + "interfaceType": "wireless", + "devicetype": "SMARTPHONE", + } + ] + hosts_response = AsyncMock() + hosts_response.status = 200 + hosts_response.json = AsyncMock(return_value=hosts_payload) + hosts_response.__aenter__ = AsyncMock(return_value=hosts_response) + hosts_response.__aexit__ = AsyncMock(return_value=None) + + mock_session.request.side_effect = [login_response, home_response, hosts_response] + + client = SagemcomClient( + host="192.168.1.1", + username="admin", + password="admin", + session=mock_session, + api_mode=ApiMode.REST, + ) + + await client.login() + devices = await client.get_hosts() + + assert len(devices) == 1 + assert devices[0].host_name == "phone" + assert devices[0].interface_type == "wifi" + + +@pytest.mark.asyncio +async def test_get_hosts_rest_trusts_empty_home_host_lists(): + """A recognised but empty /api/v1/home payload means no hosts, not an unsupported endpoint.""" + session = MagicMock(spec=ClientSession) + client = SagemcomClient( + host="192.168.1.1", + username="admin", + password="admin", + session=session, + api_mode=ApiMode.REST, + ) + client._SagemcomClient__rest_request = AsyncMock(return_value=[{"wirelessListDevice": [], "ethernetListDevice": []}]) + + assert await client.get_hosts() == [] + client._SagemcomClient__rest_request.assert_awaited_once_with("GET", "/api/v1/home") + + +@pytest.mark.asyncio +async def test_hosts_rest_fallback_does_not_swallow_authentication_errors(): + """Authentication failures from the preferred endpoint should propagate.""" + response = AsyncMock() + response.status = 401 + response.text = AsyncMock(return_value="unauthorized") + response.__aenter__ = AsyncMock(return_value=response) + response.__aexit__ = AsyncMock(return_value=None) + + session = MagicMock(spec=ClientSession) + session.request.return_value = response + client = SagemcomClient( + host="192.168.1.1", + username="admin", + password="admin", + session=session, + api_mode=ApiMode.REST, + ) + + with pytest.raises(UnauthorizedException, match="unauthorized"): + await client.get_hosts() + + assert session.request.call_count == 1 + + +@pytest.mark.asyncio +async def test_hosts_rest_fallback_does_not_swallow_transport_errors(): + """Transport failures from the preferred endpoint should propagate.""" + session = MagicMock(spec=ClientSession) + session.request.side_effect = ConnectionError("transport failed") + client = SagemcomClient( + host="192.168.1.1", + username="admin", + password="admin", + session=session, + api_mode=ApiMode.REST, + ) + + with pytest.raises(ConnectionError, match="transport failed"): + await client.get_hosts() + + assert session.request.call_count == 1 + + +@pytest.mark.asyncio +async def test_hosts_rest_fallback_does_not_swallow_unrelated_bad_request(): + """Only the known unsupported-home response should trigger fallback.""" + response = AsyncMock() + response.status = 400 + response.text = AsyncMock(return_value='{"exception":{"domain":"other"}}') + response.__aenter__ = AsyncMock(return_value=response) + response.__aexit__ = AsyncMock(return_value=None) + + session = MagicMock(spec=ClientSession) + session.request.return_value = response + client = SagemcomClient( + host="192.168.1.1", + username="admin", + password="admin", + session=session, + api_mode=ApiMode.REST, + ) + + with pytest.raises(BadRequestException): + await client.get_hosts() + + assert session.request.call_count == 1 + + +@pytest.mark.asyncio +async def test_reboot_rest_mode(): + """Reboot should call REST endpoint on REST firmware.""" + mock_session = MagicMock(spec=ClientSession) + mock_session.close = AsyncMock() + + login_response = AsyncMock() + login_response.status = 204 + login_response.text = AsyncMock(return_value="") + login_response.__aenter__ = AsyncMock(return_value=login_response) + login_response.__aexit__ = AsyncMock(return_value=None) + + reboot_response = AsyncMock() + reboot_response.status = 204 + reboot_response.text = AsyncMock(return_value="") + reboot_response.__aenter__ = AsyncMock(return_value=reboot_response) + reboot_response.__aexit__ = AsyncMock(return_value=None) + + mock_session.request.side_effect = [login_response, reboot_response] + + client = SagemcomClient( + host="192.168.1.1", + username="admin", + password="admin", + session=mock_session, + api_mode=ApiMode.REST, + ) + + await client.login() + result = await client.reboot() + + assert result is None + assert mock_session.request.call_count == 2 + reboot_call = mock_session.request.call_args_list[1] + assert reboot_call.args[0] == "POST" + assert reboot_call.args[1].endswith("/api/v1/device/reboot") + + @pytest.mark.asyncio async def test_get_value_by_xpath_suppresses_unknown_path(mock_session_factory, login_success_response, xpath_unknown_path_error_response): """Test that ignore_unknown_paths=True returns None for UnknownPathException."""