From 3e3018af71a57984e36d262849cf03d0e272ff10 Mon Sep 17 00:00:00 2001 From: Pablo Espinosa Date: Sun, 22 Feb 2026 00:58:42 +0000 Subject: [PATCH 01/22] Add API mode support with legacy/rest auto-detection --- sagemcom_api/client.py | 174 +++++++++++++++++++++++++++++++++++++++-- sagemcom_api/enums.py | 9 +++ 2 files changed, 177 insertions(+), 6 deletions(-) diff --git a/sagemcom_api/client.py b/sagemcom_api/client.py index 3adf14d..e0aa801 100644 --- a/sagemcom_api/client.py +++ b/sagemcom_api/client.py @@ -18,6 +18,7 @@ ClientOSError, ClientSession, ClientTimeout, + ContentTypeError, ServerDisconnectedError, TCPConnector, ) @@ -31,7 +32,7 @@ XMO_INVALID_SESSION_ERR, XMO_REQUEST_NO_ERR, ) -from .enums import EncryptionMethod +from .enums import ApiMode, EncryptionMethod from .exceptions import ( AuthenticationException, BadRequestException, @@ -65,6 +66,7 @@ def __init__( username: str, password: str, authentication_method: EncryptionMethod | None = None, + api_mode: ApiMode | str = ApiMode.AUTO, session: ClientSession | None = None, ssl: bool | None = False, verify_ssl: bool | None = True, @@ -75,11 +77,14 @@ def __init__( :param username: the username for your Sagemcom router :param password: the password for your Sagemcom router :param authentication_method: the auth method of your Sagemcom router + :param api_mode: one of auto, legacy or rest :param session: use a custom session, for example to configure the timeout """ 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) @@ -99,6 +104,11 @@ def __init__( ) ) + @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 +299,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,22 +343,107 @@ 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}" + payload = urllib.parse.urlencode(data or {}) + request_headers = {"Content-Type": "application/x-www-form-urlencoded"} + + async with self.session.request(method, url, data=payload, headers=request_headers) 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 AuthenticationException(result) + + raise UnknownException(result) + + async def __rest_login(self): + """Login to routers exposing the newer REST API.""" + await self.__rest_request( + "POST", + "/api/v1/login", + data={"login": self.username, "password": self.password}, + ) + return True + + 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 " 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 data or not isinstance(data, list): + raise UnknownException("Invalid response from /api/v1/device") + + device = data[0].get("device", {}) + 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=device.get("running", {}).get("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 +624,47 @@ 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: + data = await self.__rest_request("GET", "/api/v1/home") + if not data or not isinstance(data, list): + raise UnknownException("Invalid response from /api/v1/home") + + home = data[0] + devices: list[Device] = [] + + for entry in home.get("wirelessListDevice", []): + devices.append( + Device( + uid=entry.get("id"), + phys_address=entry.get("macAddress"), + ip_address=entry.get("ipAddress"), + host_name=entry.get("hostname"), + user_host_name=entry.get("friendlyname"), + active=entry.get("active", True), + interface_type="wifi", + detected_device_type=entry.get("devicetype"), + ) + ) + + for entry in home.get("ethernetListDevice", []): + devices.append( + Device( + uid=entry.get("id"), + phys_address=entry.get("macAddress"), + ip_address=entry.get("ipAddress"), + host_name=entry.get("hostname"), + user_host_name=entry.get("friendlyname"), + active=entry.get("active", True), + interface_type="ethernet", + detected_device_type=entry.get("devicetype"), + ) + ) + + if only_active: + return [d for d in devices if d.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 +687,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 +722,7 @@ async def get_logs(self) -> str: async def reboot(self): """Reboot Sagemcom F@st device.""" + self.__ensure_legacy_api() action = { "id": 0, "method": "reboot", diff --git a/sagemcom_api/enums.py b/sagemcom_api/enums.py index 99ac8ae..a9adf32 100644 --- a/sagemcom_api/enums.py +++ b/sagemcom_api/enums.py @@ -10,3 +10,12 @@ class EncryptionMethod(StrEnum): MD5 = "MD5" MD5_NONCE = "MD5_NONCE" SHA512 = "SHA512" + + +@unique +class ApiMode(StrEnum): + """API mode to use when communicating with the router.""" + + AUTO = "auto" + LEGACY = "legacy" + REST = "rest" From cfcc7e8e82131581be0e4327c2ec68e049929190 Mon Sep 17 00:00:00 2001 From: Pablo Espinosa Date: Sun, 22 Feb 2026 00:58:44 +0000 Subject: [PATCH 02/22] Add unit tests for REST mode and auto fallback --- tests/unit/test_client_basic.py | 103 +++++++++++++++++++++++++++++++- 1 file changed, 102 insertions(+), 1 deletion(-) diff --git a/tests/unit/test_client_basic.py b/tests/unit/test_client_basic.py index 7b24cd5..3b85bc3 100644 --- a/tests/unit/test_client_basic.py +++ b/tests/unit/test_client_basic.py @@ -9,7 +9,7 @@ from aiohttp import ClientSession 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, InvalidSessionException, @@ -182,6 +182,107 @@ async def test_login_with_preconfigured_fixture(mock_client_sha512): assert client._server_nonce == "abcdef1234567890" +@pytest.mark.asyncio +async def test_login_auto_fallbacks_to_rest_when_legacy_503(): + """Auto mode should switch to REST when legacy endpoint is unavailable.""" + mock_session = MagicMock(spec=ClientSession) + mock_session.close = AsyncMock() + + 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 mock_session.post.call_count == 1 + assert mock_session.request.call_count == 1 + + +@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 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.""" From f81a502725f77930fecdacbb9e093f0e29148a38 Mon Sep 17 00:00:00 2001 From: Pablo Espinosa Date: Sun, 22 Feb 2026 00:58:49 +0000 Subject: [PATCH 03/22] Document API mode usage and REST limitations --- README.md | 22 ++++++++++++++++++++-- 1 file changed, 20 insertions(+), 2 deletions(-) diff --git a/README.md b/README.md index fc66845..905832c 100644 --- a/README.md +++ b/README.md @@ -63,7 +63,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 +71,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,6 +127,16 @@ 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()` and `get_hosts()` are supported. XPath-based methods (`get_value_by_xpath`, `set_value_by_xpath`, `get_values_by_xpaths`) are legacy-only. + ### 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. From abe949a9b5a1bda60cdec253e1c10cfba1c8d8a1 Mon Sep 17 00:00:00 2001 From: Pablo Espinosa Date: Sun, 22 Feb 2026 02:24:28 +0000 Subject: [PATCH 04/22] Add REST reboot support and hosts endpoint fallback --- README.md | 2 +- sagemcom_api/client.py | 161 +++++++++++++++++++++++++------- tests/unit/test_client_basic.py | 93 ++++++++++++++++++ 3 files changed, 222 insertions(+), 34 deletions(-) diff --git a/README.md b/README.md index 905832c..b618fcb 100644 --- a/README.md +++ b/README.md @@ -135,7 +135,7 @@ The client supports two API variants: - `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()` and `get_hosts()` are supported. XPath-based methods (`get_value_by_xpath`, `set_value_by_xpath`, `get_values_by_xpaths`) are legacy-only. +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. ### Determine the EncryptionMethod diff --git a/sagemcom_api/client.py b/sagemcom_api/client.py index e0aa801..069f533 100644 --- a/sagemcom_api/client.py +++ b/sagemcom_api/client.py @@ -384,6 +384,116 @@ async def __rest_login(self): ) return True + @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]: + """Parse /api/v1/home hosts payload.""" + if isinstance(data, list): + if not data: + return [] + home = data[0] + elif isinstance(data, dict): + home = data + else: + raise UnknownException("Invalid response from /api/v1/home") + + if not isinstance(home, dict): + raise UnknownException("Invalid response from /api/v1/home") + + devices: list[Device] = [] + for entry in home.get("wirelessListDevice", []): + if isinstance(entry, dict): + devices.append(self.__build_rest_device(entry, "wifi")) + + for entry in home.get("ethernetListDevice", []): + if isinstance(entry, dict): + devices.append(self.__build_rest_device(entry, "ethernet")) + + return devices + + 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] + def __should_fallback_to_rest(self, exception: Exception) -> bool: """Return True when legacy API failure indicates a REST-only router.""" if isinstance(exception, UnsupportedHostException): @@ -625,40 +735,23 @@ 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: - data = await self.__rest_request("GET", "/api/v1/home") - if not data or not isinstance(data, list): - raise UnknownException("Invalid response from /api/v1/home") - - home = data[0] + rest_errors: list[Exception] = [] devices: list[Device] = [] - for entry in home.get("wirelessListDevice", []): - devices.append( - Device( - uid=entry.get("id"), - phys_address=entry.get("macAddress"), - ip_address=entry.get("ipAddress"), - host_name=entry.get("hostname"), - user_host_name=entry.get("friendlyname"), - active=entry.get("active", True), - interface_type="wifi", - detected_device_type=entry.get("devicetype"), - ) - ) - - for entry in home.get("ethernetListDevice", []): - devices.append( - Device( - uid=entry.get("id"), - phys_address=entry.get("macAddress"), - ip_address=entry.get("ipAddress"), - host_name=entry.get("hostname"), - user_host_name=entry.get("friendlyname"), - active=entry.get("active", True), - interface_type="ethernet", - detected_device_type=entry.get("devicetype"), - ) - ) + for endpoint, parser in ( + ("/api/v1/home", self.__extract_rest_home_hosts), + ("/api/v1/hosts", self.__extract_rest_hosts), + ): + try: + data = await self.__rest_request("GET", endpoint) + devices = parser(data) + break + except (UnknownException, UnsupportedHostException) as exception: + rest_errors.append(exception) + else: + if rest_errors: + raise rest_errors[-1] + raise UnknownException("Unable to retrieve hosts using REST endpoints") if only_active: return [d for d in devices if d.active is True] @@ -722,7 +815,9 @@ async def get_logs(self) -> str: async def reboot(self): """Reboot Sagemcom F@st device.""" - self.__ensure_legacy_api() + 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/tests/unit/test_client_basic.py b/tests/unit/test_client_basic.py index 3b85bc3..91af384 100644 --- a/tests/unit/test_client_basic.py +++ b/tests/unit/test_client_basic.py @@ -283,6 +283,99 @@ async def test_get_hosts_rest_mode(): assert devices[1].interface_type == "ethernet" +@pytest.mark.asyncio +async def test_get_hosts_rest_fallbacks_to_hosts_endpoint(): + """/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=[{"unexpected": "shape"}]) + 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_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.""" From 00c5f05c71d2d8b04e3bcda24ad98396578a0d1f Mon Sep 17 00:00:00 2001 From: Pablo Espinosa Date: Sun, 22 Feb 2026 04:08:40 +0000 Subject: [PATCH 05/22] Accept REST cookies from IP hosts with unsafe CookieJar --- sagemcom_api/client.py | 13 +++++++++++-- 1 file changed, 11 insertions(+), 2 deletions(-) diff --git a/sagemcom_api/client.py b/sagemcom_api/client.py index 069f533..6868d3d 100644 --- a/sagemcom_api/client.py +++ b/sagemcom_api/client.py @@ -14,10 +14,12 @@ import backoff import humps from aiohttp import ( + CookieJar, ClientConnectorError, ClientOSError, ClientSession, ClientTimeout, + CookieJar, ContentTypeError, ServerDisconnectedError, TCPConnector, @@ -100,7 +102,10 @@ def __init__( 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), + cookie_jar=CookieJar(unsafe=True), + connector=TCPConnector( + verify_ssl=verify_ssl if verify_ssl is not None else True + ), ) ) @@ -746,7 +751,11 @@ async def get_hosts(self, only_active: bool | None = False) -> list[Device]: data = await self.__rest_request("GET", endpoint) devices = parser(data) break - except (UnknownException, UnsupportedHostException) as exception: + except ( + UnknownException, + UnsupportedHostException, + AuthenticationException, + ) as exception: rest_errors.append(exception) else: if rest_errors: From 0352eb61b3cff485fee2d7b67e554f3d0c1d84e1 Mon Sep 17 00:00:00 2001 From: Pablo Espinosa Date: Sun, 22 Feb 2026 04:08:43 +0000 Subject: [PATCH 06/22] Add test for IP-cookie acceptance in default session --- tests/unit/test_client_basic.py | 69 +++++++++++++++++++++++++++++++++ 1 file changed, 69 insertions(+) diff --git a/tests/unit/test_client_basic.py b/tests/unit/test_client_basic.py index 91af384..9ac9444 100644 --- a/tests/unit/test_client_basic.py +++ b/tests/unit/test_client_basic.py @@ -42,6 +42,21 @@ async def test_response_with_invalid_utf8_is_parsed(): assert result["reply"]["value"] == "invalid � byte" +@pytest.mark.asyncio +async def test_default_session_accepts_ip_cookies(): + """Default aiohttp session should accept cookies from IP hosts.""" + client = SagemcomClient( + host="192.168.1.1", + username="admin", + password="admin", + authentication_method=EncryptionMethod.MD5, + ) + try: + assert getattr(client.session.cookie_jar, "_unsafe", False) is True + finally: + await client.close() + + @pytest.mark.asyncio async def test_login_success(mock_session_factory, login_success_response): """Test successful login with mocked session. @@ -338,6 +353,60 @@ async def test_get_hosts_rest_fallbacks_to_hosts_endpoint(): 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_reboot_rest_mode(): """reboot should call REST endpoint on REST firmware.""" From 7fdf78ecd97529752068d1e5af82bc1f371afd8b Mon Sep 17 00:00:00 2001 From: Pablo Espinosa Date: Mon, 23 Feb 2026 01:46:39 +0000 Subject: [PATCH 07/22] Expose EncryptionMethod.NONE results --- sagemcom_api/client.py | 8 +++++--- sagemcom_api/enums.py | 1 + tests/unit/test_client_basic.py | 16 ++++++++++++++++ 3 files changed, 22 insertions(+), 3 deletions(-) diff --git a/sagemcom_api/client.py b/sagemcom_api/client.py index 6868d3d..c0d9a69 100644 --- a/sagemcom_api/client.py +++ b/sagemcom_api/client.py @@ -550,17 +550,19 @@ def __ensure_legacy_api(self): "This method is not available with REST API mode. Use helper methods supported for REST firmware instead." ) - async def get_encryption_method(self): + async def get_encryption_method(self) -> EncryptionMethod: """Determine which encryption method to use for authentication and set it directly. :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 None + return EncryptionMethod.NONE for encryption_method in EncryptionMethod: try: + if encryption_method == EncryptionMethod.NONE: + continue self.authentication_method = encryption_method self._password_hash = self.__generate_hash(self.password, encryption_method) @@ -577,7 +579,7 @@ async def get_encryption_method(self): ): pass - return None + return EncryptionMethod.NONE async def get_value_by_xpath(self, xpath: str, options: dict | None = None, ignore_unknown_paths: bool = False) -> Any: """Retrieve raw value from router using XPath. diff --git a/sagemcom_api/enums.py b/sagemcom_api/enums.py index a9adf32..594620b 100644 --- a/sagemcom_api/enums.py +++ b/sagemcom_api/enums.py @@ -10,6 +10,7 @@ class EncryptionMethod(StrEnum): MD5 = "MD5" MD5_NONCE = "MD5_NONCE" SHA512 = "SHA512" + NONE = "NONE" @unique diff --git a/tests/unit/test_client_basic.py b/tests/unit/test_client_basic.py index 9ac9444..80b3305 100644 --- a/tests/unit/test_client_basic.py +++ b/tests/unit/test_client_basic.py @@ -234,6 +234,22 @@ async def test_login_auto_fallbacks_to_rest_when_legacy_503(): assert mock_session.request.call_count == 1 +@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_get_hosts_rest_mode(): """get_hosts should parse wifi and ethernet devices on REST firmware.""" From 37668b2a08f1e0e0c7fd8c56cb2b94344ec0c36c Mon Sep 17 00:00:00 2001 From: Pablo Espinosa Date: Mon, 23 Feb 2026 01:48:14 +0000 Subject: [PATCH 08/22] Log API mode outcome and logout response --- README.md | 2 +- sagemcom_api/client.py | 19 ++++++++++++++----- 2 files changed, 15 insertions(+), 6 deletions(-) diff --git a/README.md b/README.md index b618fcb..58d6b42 100644 --- a/README.md +++ b/README.md @@ -141,7 +141,7 @@ When REST mode is active, high-level helpers like `get_device_info()`, `get_host 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 and `EncryptionMethod.NONE` if no method matches or REST mode is configured. 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 c0d9a69..b726828 100644 --- a/sagemcom_api/client.py +++ b/sagemcom_api/client.py @@ -13,13 +13,13 @@ import backoff import humps +import logging from aiohttp import ( CookieJar, ClientConnectorError, ClientOSError, ClientSession, ClientTimeout, - CookieJar, ContentTypeError, ServerDisconnectedError, TCPConnector, @@ -49,6 +49,8 @@ ) from .models import Device, DeviceInfo, PortMapping, SpeedTestResult +LOGGER = logging.getLogger(__name__) + async def retry_login(invocation: Mapping[str, Any]) -> None: """Retry login via backoff if an exception occurs.""" @@ -67,7 +69,7 @@ def __init__( host: str, username: str, password: str, - authentication_method: EncryptionMethod | None = None, + authentication_method: EncryptionMethod, api_mode: ApiMode | str = ApiMode.AUTO, session: ClientSession | None = None, ssl: bool | None = False, @@ -514,30 +516,37 @@ async def login(self): """Login to the router using configured API mode.""" if self.api_mode == ApiMode.REST: self._active_api_mode = ApiMode.REST + LOGGER.info("API mode forced to REST") return await self.__rest_login() if self.api_mode == ApiMode.LEGACY: self._active_api_mode = ApiMode.LEGACY + LOGGER.info("API mode forced to LEGACY") return await self.__legacy_login() # Auto-detect mode: try legacy first, then fall back to REST for newer firmwares. try: self._active_api_mode = ApiMode.LEGACY - return await self.__legacy_login() + result = await self.__legacy_login() + LOGGER.info("API mode auto-detected as LEGACY") + return result except Exception as exception: # pylint: disable=broad-except if not self.__should_fallback_to_rest(exception): raise self._active_api_mode = ApiMode.REST + LOGGER.info("API mode auto-detected as REST") return await self.__rest_login() async def logout(self): """Log out of the Sagemcom F@st device.""" if self._active_api_mode == ApiMode.REST: - await self.__rest_request("POST", "/api/v1/logout") + response = await self.__rest_request("POST", "/api/v1/logout", data={"_": ""}) + LOGGER.info("REST logout response: %s", response) else: actions = {"id": 0, "method": "logOut"} - await self.__api_request_async([actions], False) + response = await self.__api_request_async([actions], False) + LOGGER.info("JSON-REQ logout response: %s", response) self._session_id = -1 self._server_nonce = "" From 55577496cd0bf12a646939b245978ae03b0886e2 Mon Sep 17 00:00:00 2001 From: Pablo Espinosa Date: Mon, 23 Feb 2026 01:56:35 +0000 Subject: [PATCH 09/22] Make authentication_method optional again --- sagemcom_api/client.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/sagemcom_api/client.py b/sagemcom_api/client.py index b726828..be61bad 100644 --- a/sagemcom_api/client.py +++ b/sagemcom_api/client.py @@ -69,7 +69,7 @@ def __init__( host: str, username: str, password: str, - authentication_method: EncryptionMethod, + authentication_method: EncryptionMethod | None = None, api_mode: ApiMode | str = ApiMode.AUTO, session: ClientSession | None = None, ssl: bool | None = False, From 577a198754e93030fca53a42cb9bac2087f92931 Mon Sep 17 00:00:00 2001 From: Pablo Espinosa Date: Mon, 23 Feb 2026 02:06:10 +0000 Subject: [PATCH 10/22] Remove client logging instrumentation --- sagemcom_api/client.py | 13 ++----------- 1 file changed, 2 insertions(+), 11 deletions(-) diff --git a/sagemcom_api/client.py b/sagemcom_api/client.py index be61bad..19b8cbc 100644 --- a/sagemcom_api/client.py +++ b/sagemcom_api/client.py @@ -13,7 +13,6 @@ import backoff import humps -import logging from aiohttp import ( CookieJar, ClientConnectorError, @@ -49,8 +48,6 @@ ) from .models import Device, DeviceInfo, PortMapping, SpeedTestResult -LOGGER = logging.getLogger(__name__) - async def retry_login(invocation: Mapping[str, Any]) -> None: """Retry login via backoff if an exception occurs.""" @@ -516,37 +513,31 @@ async def login(self): """Login to the router using configured API mode.""" if self.api_mode == ApiMode.REST: self._active_api_mode = ApiMode.REST - LOGGER.info("API mode forced to REST") return await self.__rest_login() if self.api_mode == ApiMode.LEGACY: self._active_api_mode = ApiMode.LEGACY - LOGGER.info("API mode forced to LEGACY") return await self.__legacy_login() # Auto-detect mode: try legacy first, then fall back to REST for newer firmwares. try: self._active_api_mode = ApiMode.LEGACY result = await self.__legacy_login() - LOGGER.info("API mode auto-detected as LEGACY") return result except Exception as exception: # pylint: disable=broad-except if not self.__should_fallback_to_rest(exception): raise self._active_api_mode = ApiMode.REST - LOGGER.info("API mode auto-detected as REST") return await self.__rest_login() async def logout(self): """Log out of the Sagemcom F@st device.""" if self._active_api_mode == ApiMode.REST: - response = await self.__rest_request("POST", "/api/v1/logout", data={"_": ""}) - LOGGER.info("REST logout response: %s", response) + await self.__rest_request("POST", "/api/v1/logout", data={"_": ""}) else: actions = {"id": 0, "method": "logOut"} - response = await self.__api_request_async([actions], False) - LOGGER.info("JSON-REQ logout response: %s", response) + await self.__api_request_async([actions], False) self._session_id = -1 self._server_nonce = "" From 33a412bf381864d3d26010a0d8082818f89f7bdf Mon Sep 17 00:00:00 2001 From: Pablo Espinosa Date: Tue, 24 Feb 2026 00:28:01 +0000 Subject: [PATCH 11/22] Probe REST when auto detecting encryption --- sagemcom_api/client.py | 22 ++++++++++++++++++++++ 1 file changed, 22 insertions(+) diff --git a/sagemcom_api/client.py b/sagemcom_api/client.py index 19b8cbc..81cfab3 100644 --- a/sagemcom_api/client.py +++ b/sagemcom_api/client.py @@ -388,6 +388,25 @@ async def __rest_login(self): ) return True + async def __probe_rest_availability(self) -> bool: + """Try a REST login/logout sequence to detect REST-only firmware.""" + try: + await self.__rest_request( + "POST", + "/api/v1/login", + data={"login": self.username, "password": self.password}, + ) + except ( + AuthenticationException, + UnauthorizedException, + UnsupportedHostException, + UnknownException, + ): + return False + + await self.__rest_request("POST", "/api/v1/logout", data={"_": ""}) + return True + @staticmethod def __first_value(data: dict[str, Any], *keys: str) -> Any: """Return the first non-None value from data for the given keys.""" @@ -559,6 +578,9 @@ async def get_encryption_method(self) -> EncryptionMethod: if self.api_mode == ApiMode.REST: return EncryptionMethod.NONE + if self.api_mode == ApiMode.AUTO and await self.__probe_rest_availability(): + return EncryptionMethod.NONE + for encryption_method in EncryptionMethod: try: if encryption_method == EncryptionMethod.NONE: From 376e09c1be97f40c3da903562299b7d4846ea71d Mon Sep 17 00:00:00 2001 From: Pablo Espinosa Date: Tue, 24 Feb 2026 00:52:52 +0000 Subject: [PATCH 12/22] Document YouFibre 5598 entry --- README.md | 1 + 1 file changed, 1 insertion(+) diff --git a/README.md b/README.md index 58d6b42..bdec9e1 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: "" | From 99e89959b93be3fec43b4a22ad8741f153ef89c4 Mon Sep 17 00:00:00 2001 From: Mick Vleeshouwer Date: Mon, 6 Apr 2026 22:48:26 +0000 Subject: [PATCH 13/22] Fix post-rebase prek issues --- sagemcom_api/client.py | 26 +++++++------------------- tests/unit/test_client_basic.py | 2 +- 2 files changed, 8 insertions(+), 20 deletions(-) diff --git a/sagemcom_api/client.py b/sagemcom_api/client.py index 81cfab3..23ecaed 100644 --- a/sagemcom_api/client.py +++ b/sagemcom_api/client.py @@ -14,12 +14,12 @@ import backoff import humps from aiohttp import ( - CookieJar, ClientConnectorError, ClientOSError, ClientSession, ClientTimeout, ContentTypeError, + CookieJar, ServerDisconnectedError, TCPConnector, ) @@ -102,9 +102,7 @@ def __init__( headers={"User-Agent": f"{DEFAULT_USER_AGENT}"}, timeout=ClientTimeout(DEFAULT_TIMEOUT), cookie_jar=CookieJar(unsafe=True), - connector=TCPConnector( - verify_ssl=verify_ssl if verify_ssl is not None else True - ), + connector=TCPConnector(verify_ssl=verify_ssl if verify_ssl is not None else True), ) ) @@ -428,9 +426,7 @@ def __to_bool(value: Any, default: bool = True) -> bool: 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: + 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, @@ -452,21 +448,13 @@ def __build_rest_device( return Device( uid=self.__first_value(entry, "id", "uid"), - phys_address=self.__first_value( - entry, "macAddress", "mac_address", "phys_address" - ), + 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 - ), + 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" - ), + detected_device_type=self.__first_value(entry, "devicetype", "deviceType", "detected_device_type"), ) def __extract_rest_home_hosts(self, data: Any) -> list[Device]: diff --git a/tests/unit/test_client_basic.py b/tests/unit/test_client_basic.py index 80b3305..82284e0 100644 --- a/tests/unit/test_client_basic.py +++ b/tests/unit/test_client_basic.py @@ -425,7 +425,7 @@ async def test_get_hosts_rest_fallbacks_on_home_400(): @pytest.mark.asyncio async def test_reboot_rest_mode(): - """reboot should call REST endpoint on REST firmware.""" + """Reboot should call REST endpoint on REST firmware.""" mock_session = MagicMock(spec=ClientSession) mock_session.close = AsyncMock() From 177e648bc88c62d45bee10b422bfc674d0325822 Mon Sep 17 00:00:00 2001 From: Mick Vleeshouwer Date: Tue, 28 Jul 2026 23:07:42 +0000 Subject: [PATCH 14/22] Correct REST request semantics --- sagemcom_api/client.py | 14 +++-- tests/unit/test_client_basic.py | 103 ++++++++++++++++++++++++++++++++ 2 files changed, 112 insertions(+), 5 deletions(-) diff --git a/sagemcom_api/client.py b/sagemcom_api/client.py index 23ecaed..5988895 100644 --- a/sagemcom_api/client.py +++ b/sagemcom_api/client.py @@ -353,10 +353,14 @@ async def __legacy_login(self): 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}" - payload = urllib.parse.urlencode(data or {}) - request_headers = {"Content-Type": "application/x-www-form-urlencoded"} - - async with self.session.request(method, url, data=payload, headers=request_headers) as response: + 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 @@ -373,7 +377,7 @@ async def __rest_request(self, method: str, endpoint: str, data: dict[str, Any] raise UnsupportedHostException(result) if response.status == 400: - raise AuthenticationException(result) + raise BadRequestException(result) raise UnknownException(result) diff --git a/tests/unit/test_client_basic.py b/tests/unit/test_client_basic.py index 82284e0..e4a3e70 100644 --- a/tests/unit/test_client_basic.py +++ b/tests/unit/test_client_basic.py @@ -12,9 +12,11 @@ from sagemcom_api.enums import ApiMode, EncryptionMethod from sagemcom_api.exceptions import ( AuthenticationException, + BadRequestException, InvalidSessionException, LoginRetryErrorException, NonWritableParameterException, + UnauthorizedException, UnknownPathException, ) @@ -42,6 +44,107 @@ 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 async def test_default_session_accepts_ip_cookies(): """Default aiohttp session should accept cookies from IP hosts.""" From 9f0459cd365efbc26e6a5a78619208d3b4adc980 Mon Sep 17 00:00:00 2001 From: Mick Vleeshouwer Date: Tue, 28 Jul 2026 23:10:29 +0000 Subject: [PATCH 15/22] Handle rejected REST availability probes --- sagemcom_api/client.py | 1 + tests/unit/test_client_basic.py | 26 ++++++++++++++++++++++++++ 2 files changed, 27 insertions(+) diff --git a/sagemcom_api/client.py b/sagemcom_api/client.py index 5988895..c0e8a42 100644 --- a/sagemcom_api/client.py +++ b/sagemcom_api/client.py @@ -400,6 +400,7 @@ async def __probe_rest_availability(self) -> bool: ) except ( AuthenticationException, + BadRequestException, UnauthorizedException, UnsupportedHostException, UnknownException, diff --git a/tests/unit/test_client_basic.py b/tests/unit/test_client_basic.py index e4a3e70..86a269b 100644 --- a/tests/unit/test_client_basic.py +++ b/tests/unit/test_client_basic.py @@ -353,6 +353,32 @@ async def test_get_encryption_method_rest_returns_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_get_hosts_rest_mode(): """get_hosts should parse wifi and ethernet devices on REST firmware.""" From 85c457d3cb98f40a9aab7ac37bb2dcf740c7e8d1 Mon Sep 17 00:00:00 2001 From: Mick Vleeshouwer Date: Tue, 28 Jul 2026 23:18:49 +0000 Subject: [PATCH 16/22] Isolate REST cookie and mode selection --- sagemcom_api/client.py | 52 ++++++++++----- tests/unit/test_client_basic.py | 114 ++++++++++++++++++++++++++++++-- 2 files changed, 145 insertions(+), 21 deletions(-) diff --git a/sagemcom_api/client.py b/sagemcom_api/client.py index c0e8a42..605e84a 100644 --- a/sagemcom_api/client.py +++ b/sagemcom_api/client.py @@ -95,17 +95,30 @@ 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), - cookie_jar=CookieJar(unsafe=True), - 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._owned_session_accepts_ip_cookies = self.api_mode == ApiMode.REST if self._session_owned else None + self.session = session or self.__create_session(allow_ip_cookies=self.api_mode == ApiMode.REST) + + 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 __activate_rest_mode(self) -> None: + """Select REST and replace only a library-owned safe session.""" + self._active_api_mode = ApiMode.REST + if not self._session_owned or self._owned_session_accepts_ip_cookies is True: + return + + await self.session.close() + self.session = self.__create_session(allow_ip_cookies=True) + self._owned_session_accepts_ip_cookies = True + @property def active_api_mode(self) -> ApiMode: """Return the API mode that is currently active.""" @@ -524,7 +537,7 @@ def __should_fallback_to_rest(self, exception: Exception) -> bool: async def login(self): """Login to the router using configured API mode.""" if self.api_mode == ApiMode.REST: - self._active_api_mode = ApiMode.REST + await self.__activate_rest_mode() return await self.__rest_login() if self.api_mode == ApiMode.LEGACY: @@ -540,7 +553,7 @@ async def login(self): if not self.__should_fallback_to_rest(exception): raise - self._active_api_mode = ApiMode.REST + await self.__activate_rest_mode() return await self.__rest_login() async def logout(self): @@ -571,9 +584,6 @@ async def get_encryption_method(self) -> EncryptionMethod: if self.api_mode == ApiMode.REST: return EncryptionMethod.NONE - if self.api_mode == ApiMode.AUTO and await self.__probe_rest_availability(): - return EncryptionMethod.NONE - for encryption_method in EncryptionMethod: try: if encryption_method == EncryptionMethod.NONE: @@ -581,7 +591,11 @@ async def get_encryption_method(self) -> EncryptionMethod: 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 @@ -593,6 +607,14 @@ async def get_encryption_method(self) -> EncryptionMethod: AuthenticationException, ): pass + except Exception as exception: + if not self.__should_fallback_to_rest(exception): + raise + + await self.__activate_rest_mode() + if await self.__probe_rest_availability(): + return EncryptionMethod.NONE + raise return EncryptionMethod.NONE diff --git a/tests/unit/test_client_basic.py b/tests/unit/test_client_basic.py index 86a269b..b9aeb65 100644 --- a/tests/unit/test_client_basic.py +++ b/tests/unit/test_client_basic.py @@ -6,7 +6,8 @@ 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 ApiMode, EncryptionMethod @@ -17,6 +18,7 @@ LoginRetryErrorException, NonWritableParameterException, UnauthorizedException, + UnknownException, UnknownPathException, ) @@ -146,16 +148,59 @@ async def test_rest_request_unauthorized_statuses_raise_unauthorized(status): @pytest.mark.asyncio -async def test_default_session_accepts_ip_cookies(): - """Default aiohttp session should accept cookies from IP hosts.""" +@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: - assert getattr(client.session.cookie_jar, "_unsafe", False) is True + 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() @@ -301,10 +346,12 @@ async def test_login_with_preconfigured_fixture(mock_client_sha512): @pytest.mark.asyncio -async def test_login_auto_fallbacks_to_rest_when_legacy_503(): - """Auto mode should switch to REST when legacy endpoint is unavailable.""" +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 @@ -333,10 +380,40 @@ async def test_login_auto_fallbacks_to_rest_when_legacy_503(): 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_get_encryption_method_rest_returns_none(): """REST mode should immediately signal that no encryption method is needed.""" @@ -379,6 +456,31 @@ async def test_probe_rest_bad_request_allows_legacy_encryption_detection(): 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_get_hosts_rest_mode(): """get_hosts should parse wifi and ethernet devices on REST firmware.""" From 75fa3c798660320b6316e9ea72deed23da0ec059 Mon Sep 17 00:00:00 2001 From: Mick Vleeshouwer Date: Tue, 28 Jul 2026 23:20:25 +0000 Subject: [PATCH 17/22] Keep explicit legacy mode isolated --- sagemcom_api/client.py | 2 +- tests/unit/test_client_basic.py | 21 +++++++++++++++++++++ 2 files changed, 22 insertions(+), 1 deletion(-) diff --git a/sagemcom_api/client.py b/sagemcom_api/client.py index 605e84a..0a8a409 100644 --- a/sagemcom_api/client.py +++ b/sagemcom_api/client.py @@ -608,7 +608,7 @@ async def get_encryption_method(self) -> EncryptionMethod: ): pass except Exception as exception: - if not self.__should_fallback_to_rest(exception): + if self.api_mode != ApiMode.AUTO or not self.__should_fallback_to_rest(exception): raise await self.__activate_rest_mode() diff --git a/tests/unit/test_client_basic.py b/tests/unit/test_client_basic.py index b9aeb65..37d4983 100644 --- a/tests/unit/test_client_basic.py +++ b/tests/unit/test_client_basic.py @@ -20,6 +20,7 @@ UnauthorizedException, UnknownException, UnknownPathException, + UnsupportedHostException, ) @@ -481,6 +482,26 @@ async def test_dual_stack_auto_prefers_working_legacy_encryption(): 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__probe_rest_availability = AsyncMock(return_value=True) + + with pytest.raises(UnsupportedHostException, match="legacy unavailable"): + await client.get_encryption_method() + + client._SagemcomClient__probe_rest_availability.assert_not_awaited() + + @pytest.mark.asyncio async def test_get_hosts_rest_mode(): """get_hosts should parse wifi and ethernet devices on REST firmware.""" From 77739d1ecb9638625f85d55c8136458de1727889 Mon Sep 17 00:00:00 2001 From: Mick Vleeshouwer Date: Tue, 28 Jul 2026 23:27:13 +0000 Subject: [PATCH 18/22] Harden REST response handling --- sagemcom_api/client.py | 67 ++++++++++------ tests/unit/test_client_basic.py | 135 ++++++++++++++++++++++++++++++++ 2 files changed, 178 insertions(+), 24 deletions(-) diff --git a/sagemcom_api/client.py b/sagemcom_api/client.py index 0a8a409..8d3d9bb 100644 --- a/sagemcom_api/client.py +++ b/sagemcom_api/client.py @@ -523,6 +523,22 @@ def __extract_rest_hosts(self, data: Any) -> list[Device]: 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): @@ -723,17 +739,22 @@ 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 data or not isinstance(data, list): + 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", {}) + 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=device.get("running", {}).get("version"), + software_version=software_version, hardware_version=device.get("hardware_version"), manufacturer="Sagemcom", up_time=device.get("uptime"), @@ -779,30 +800,28 @@ 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: - rest_errors: list[Exception] = [] - devices: list[Device] = [] - - for endpoint, parser in ( - ("/api/v1/home", self.__extract_rest_home_hosts), - ("/api/v1/hosts", self.__extract_rest_hosts), - ): - try: - data = await self.__rest_request("GET", endpoint) - devices = parser(data) - break - except ( - UnknownException, - UnsupportedHostException, - AuthenticationException, - ) as exception: - rest_errors.append(exception) + home_endpoint = "/api/v1/home" + 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: - if rest_errors: - raise rest_errors[-1] - raise UnknownException("Unable to retrieve hosts using REST endpoints") + try: + devices = self.__extract_rest_home_hosts(home_data) + except UnknownException: + devices = [] + + if devices: + if only_active: + return [device for device in devices if device.active is True] + return devices + + hosts_data = await self.__rest_request("GET", "/api/v1/hosts") + devices = self.__extract_rest_hosts(hosts_data) if only_active: - return [d for d in devices if d.active is True] + return [device for device in devices if device.active is True] return devices diff --git a/tests/unit/test_client_basic.py b/tests/unit/test_client_basic.py index 37d4983..ad4806f 100644 --- a/tests/unit/test_client_basic.py +++ b/tests/unit/test_client_basic.py @@ -502,6 +502,72 @@ async def test_legacy_mode_never_probes_rest_during_encryption_detection(): client._SagemcomClient__probe_rest_availability.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.""" @@ -675,6 +741,75 @@ async def test_get_hosts_rest_fallbacks_on_home_400(): assert devices[0].interface_type == "wifi" +@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.""" From 0b9cae1fd64375c92a3489a07705659b3029dd74 Mon Sep 17 00:00:00 2001 From: Mick Vleeshouwer Date: Tue, 28 Jul 2026 23:28:04 +0000 Subject: [PATCH 19/22] Handle malformed REST host collections --- sagemcom_api/client.py | 17 ++++++++++------- tests/unit/test_client_basic.py | 11 +++++++++-- 2 files changed, 19 insertions(+), 9 deletions(-) diff --git a/sagemcom_api/client.py b/sagemcom_api/client.py index 8d3d9bb..7a15db9 100644 --- a/sagemcom_api/client.py +++ b/sagemcom_api/client.py @@ -490,13 +490,16 @@ def __extract_rest_home_hosts(self, data: Any) -> list[Device]: raise UnknownException("Invalid response from /api/v1/home") devices: list[Device] = [] - for entry in home.get("wirelessListDevice", []): - if isinstance(entry, dict): - devices.append(self.__build_rest_device(entry, "wifi")) - - for entry in home.get("ethernetListDevice", []): - if isinstance(entry, dict): - devices.append(self.__build_rest_device(entry, "ethernet")) + for key, interface_type in ( + ("wirelessListDevice", "wifi"), + ("ethernetListDevice", "ethernet"), + ): + entries = home.get(key, []) + if not isinstance(entries, list): + raise UnknownException("Invalid response from /api/v1/home") + for entry in entries: + if isinstance(entry, dict): + devices.append(self.__build_rest_device(entry, interface_type)) return devices diff --git a/tests/unit/test_client_basic.py b/tests/unit/test_client_basic.py index ad4806f..7487ac6 100644 --- a/tests/unit/test_client_basic.py +++ b/tests/unit/test_client_basic.py @@ -633,7 +633,14 @@ async def test_get_hosts_rest_mode(): @pytest.mark.asyncio -async def test_get_hosts_rest_fallbacks_to_hosts_endpoint(): +@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() @@ -646,7 +653,7 @@ async def test_get_hosts_rest_fallbacks_to_hosts_endpoint(): home_response = AsyncMock() home_response.status = 200 - home_response.json = AsyncMock(return_value=[{"unexpected": "shape"}]) + home_response.json = AsyncMock(return_value=home_payload) home_response.__aenter__ = AsyncMock(return_value=home_response) home_response.__aexit__ = AsyncMock(return_value=None) From 841b581a2c883469bd4fe5b5bb113728a27b2ffb Mon Sep 17 00:00:00 2001 From: Mick Vleeshouwer Date: Tue, 28 Jul 2026 23:30:52 +0000 Subject: [PATCH 20/22] Preserve constructor positional compatibility --- sagemcom_api/client.py | 4 ++-- tests/unit/test_client_basic.py | 19 +++++++++++++++++++ 2 files changed, 21 insertions(+), 2 deletions(-) diff --git a/sagemcom_api/client.py b/sagemcom_api/client.py index 7a15db9..dd217ed 100644 --- a/sagemcom_api/client.py +++ b/sagemcom_api/client.py @@ -67,10 +67,10 @@ def __init__( username: str, password: str, authentication_method: EncryptionMethod | None = None, - api_mode: ApiMode | str = ApiMode.AUTO, session: ClientSession | None = None, ssl: bool | None = False, verify_ssl: bool | None = True, + api_mode: ApiMode | str = ApiMode.AUTO, ): """Create a SagemCom client. @@ -78,8 +78,8 @@ def __init__( :param username: the username for your Sagemcom router :param password: the password for your Sagemcom router :param authentication_method: the auth method of your Sagemcom router - :param api_mode: one of auto, legacy or rest :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 diff --git a/tests/unit/test_client_basic.py b/tests/unit/test_client_basic.py index 7487ac6..4dc30cc 100644 --- a/tests/unit/test_client_basic.py +++ b/tests/unit/test_client_basic.py @@ -24,6 +24,25 @@ ) +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.""" From 5092b7d053fedd75c33b23842043065d7d2f9f5e Mon Sep 17 00:00:00 2001 From: Mick Vleeshouwer Date: Wed, 28 Oct 2026 00:26:43 +0000 Subject: [PATCH 21/22] Restore legacy state when AUTO REST fallback fails A failed REST attempt in AUTO mode now switches back to legacy and its safe cookie jar and re-raises the original legacy error, so legacy users keep the exceptions they had before. REST login rejections raise AuthenticationException. get_encryption_method returns None again when no legacy method matches and NONE only once a REST login succeeds. REST get_hosts trusts a recognised /api/v1/home payload, even an empty one, and only falls back to /api/v1/hosts when the payload has no host lists. --- README.md | 4 +- sagemcom_api/client.py | 148 ++++++++++++++++---------------- tests/unit/test_client_basic.py | 107 ++++++++++++++++++++++- 3 files changed, 180 insertions(+), 79 deletions(-) diff --git a/README.md b/README.md index bdec9e1..4312279 100644 --- a/README.md +++ b/README.md @@ -138,11 +138,13 @@ The client supports two API variants: 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 and `EncryptionMethod.NONE` if no method matches or REST mode is configured. 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 dd217ed..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 @@ -37,6 +38,7 @@ from .exceptions import ( AuthenticationException, BadRequestException, + BaseSagemcomException, InvalidSessionException, LoginConnectionException, LoginRetryErrorException, @@ -97,8 +99,8 @@ def __init__( self._session_owned = session is None self._verify_ssl = verify_ssl if verify_ssl is not None else True - self._owned_session_accepts_ip_cookies = self.api_mode == ApiMode.REST if self._session_owned else None - self.session = session or self.__create_session(allow_ip_cookies=self.api_mode == ApiMode.REST) + 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.""" @@ -109,15 +111,20 @@ def __create_session(self, allow_ip_cookies: bool) -> ClientSession: connector=TCPConnector(verify_ssl=self._verify_ssl), ) - async def __activate_rest_mode(self) -> None: - """Select REST and replace only a library-owned safe session.""" - self._active_api_mode = ApiMode.REST - if not self._session_owned or self._owned_session_accepts_ip_cookies is True: + 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=True) - self._owned_session_accepts_ip_cookies = True + 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: @@ -396,33 +403,29 @@ async def __rest_request(self, method: str, endpoint: str, data: dict[str, Any] async def __rest_login(self): """Login to routers exposing the newer REST API.""" - await self.__rest_request( - "POST", - "/api/v1/login", - data={"login": self.username, "password": self.password}, - ) - return True - - async def __probe_rest_availability(self) -> bool: - """Try a REST login/logout sequence to detect REST-only firmware.""" try: await self.__rest_request( "POST", "/api/v1/login", data={"login": self.username, "password": self.password}, ) - except ( - AuthenticationException, - BadRequestException, - UnauthorizedException, - UnsupportedHostException, - UnknownException, - ): - return False - - await self.__rest_request("POST", "/api/v1/logout", data={"_": ""}) + 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.""" @@ -475,33 +478,29 @@ def __build_rest_device(self, entry: dict[str, Any], interface_type: str | None) detected_device_type=self.__first_value(entry, "devicetype", "deviceType", "detected_device_type"), ) - def __extract_rest_home_hosts(self, data: Any) -> list[Device]: - """Parse /api/v1/home hosts payload.""" - if isinstance(data, list): - if not data: - return [] - home = data[0] - elif isinstance(data, dict): - home = data - else: - raise UnknownException("Invalid response from /api/v1/home") - + 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): - raise UnknownException("Invalid response from /api/v1/home") - - devices: list[Device] = [] - for key, interface_type in ( - ("wirelessListDevice", "wifi"), - ("ethernetListDevice", "ethernet"), - ): - entries = home.get(key, []) - if not isinstance(entries, list): - raise UnknownException("Invalid response from /api/v1/home") - for entry in entries: - if isinstance(entry, dict): - devices.append(self.__build_rest_device(entry, interface_type)) + return None - return devices + 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.""" @@ -556,7 +555,7 @@ def __should_fallback_to_rest(self, exception: Exception) -> bool: async def login(self): """Login to the router using configured API mode.""" if self.api_mode == ApiMode.REST: - await self.__activate_rest_mode() + await self.__switch_api_mode(ApiMode.REST) return await self.__rest_login() if self.api_mode == ApiMode.LEGACY: @@ -566,14 +565,12 @@ async def login(self): # Auto-detect mode: try legacy first, then fall back to REST for newer firmwares. try: self._active_api_mode = ApiMode.LEGACY - result = await self.__legacy_login() - return result + return await self.__legacy_login() except Exception as exception: # pylint: disable=broad-except if not self.__should_fallback_to_rest(exception): raise - await self.__activate_rest_mode() - return await self.__rest_login() + return await self.__fallback_to_rest_login(exception) async def logout(self): """Log out of the Sagemcom F@st device.""" @@ -594,9 +591,11 @@ def __ensure_legacy_api(self): "This method is not available with REST API mode. Use helper methods supported for REST firmware instead." ) - async def get_encryption_method(self) -> EncryptionMethod: + async def get_encryption_method(self) -> 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. """ @@ -604,9 +603,9 @@ async def get_encryption_method(self) -> EncryptionMethod: return EncryptionMethod.NONE for encryption_method in EncryptionMethod: + if encryption_method == EncryptionMethod.NONE: + continue try: - if encryption_method == EncryptionMethod.NONE: - continue self.authentication_method = encryption_method self._password_hash = self.__generate_hash(self.password, encryption_method) @@ -630,12 +629,15 @@ async def get_encryption_method(self) -> EncryptionMethod: if self.api_mode != ApiMode.AUTO or not self.__should_fallback_to_rest(exception): raise - await self.__activate_rest_mode() - if await self.__probe_rest_availability(): - return EncryptionMethod.NONE - 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 EncryptionMethod.NONE + + return None async def get_value_by_xpath(self, xpath: str, options: dict | None = None, ignore_unknown_paths: bool = False) -> Any: """Retrieve raw value from router using XPath. @@ -804,24 +806,18 @@ 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: - try: - devices = self.__extract_rest_home_hosts(home_data) - except UnknownException: - devices = [] - - if devices: - if only_active: - return [device for device in devices if device.active is True] - return devices + devices = self.__extract_rest_home_hosts(home_data) - hosts_data = await self.__rest_request("GET", "/api/v1/hosts") - devices = self.__extract_rest_hosts(hosts_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] diff --git a/tests/unit/test_client_basic.py b/tests/unit/test_client_basic.py index 4dc30cc..1f8feae 100644 --- a/tests/unit/test_client_basic.py +++ b/tests/unit/test_client_basic.py @@ -434,6 +434,92 @@ async def test_auto_fallback_replaces_library_owned_session_for_ip_cookies(): 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 +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.""" @@ -513,12 +599,12 @@ async def test_legacy_mode_never_probes_rest_during_encryption_detection(): api_mode=ApiMode.LEGACY, ) client._SagemcomClient__legacy_login = AsyncMock(side_effect=UnsupportedHostException("legacy unavailable")) - client._SagemcomClient__probe_rest_availability = AsyncMock(return_value=True) + client._SagemcomClient__rest_login = AsyncMock(return_value=True) with pytest.raises(UnsupportedHostException, match="legacy unavailable"): await client.get_encryption_method() - client._SagemcomClient__probe_rest_availability.assert_not_awaited() + client._SagemcomClient__rest_login.assert_not_awaited() @pytest.mark.asyncio @@ -767,6 +853,23 @@ async def test_get_hosts_rest_fallbacks_on_home_400(): 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.""" From 0add81e4b6ee891f4dd087d7ec47adfc70bfb2af Mon Sep 17 00:00:00 2001 From: Mick Vleeshouwer Date: Wed, 28 Oct 2026 08:02:08 +0000 Subject: [PATCH 22/22] Test that AUTO mode stops on a legacy login lockout --- tests/unit/test_client_basic.py | 16 ++++++++++++++++ 1 file changed, 16 insertions(+) diff --git a/tests/unit/test_client_basic.py b/tests/unit/test_client_basic.py index 1f8feae..699e5f8 100644 --- a/tests/unit/test_client_basic.py +++ b/tests/unit/test_client_basic.py @@ -499,6 +499,22 @@ async def test_get_encryption_method_returns_none_when_no_legacy_method_matches( 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."""