diff --git a/.gitignore b/.gitignore index b6e4761..5d039d2 100644 --- a/.gitignore +++ b/.gitignore @@ -127,3 +127,4 @@ dmypy.json # Pyre type checker .pyre/ +.vscode/ \ No newline at end of file diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 0859882..0be6386 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -20,7 +20,7 @@ repos: - --skip="./.*,*.csv,*.json,*.md" - --quiet-level=2 exclude_types: [csv, json] - - repo: https://gitlab.com/pycqa/flake8 + - repo: https://github.com/pycqa/flake8 rev: 3.8.4 hooks: - id: flake8 @@ -34,6 +34,6 @@ repos: hooks: - id: yamllint - repo: https://github.com/PyCQA/isort - rev: 5.5.3 + rev: 5.12.0 hooks: - id: isort diff --git a/README.md b/README.md index 47349ac..e7e1661 100644 --- a/README.md +++ b/README.md @@ -28,6 +28,8 @@ The Sagemcom F@st series is used by multiple cable companies, where some cable c | Sagemcom F@st 5370e | Telia | sha512 | | | Sagemcom F@st 5566 | Bell (Home Hub 3000) | md5 | username: guest, password: "" | | Sagemcom F@st 5689 | Bell (Home Hub 4000) | md5 | username: admin, password: "" | +| Sagemcom F@st 5689E | Bell (Giga Hub) | sha512 | username: admin, password: "" | +| Sagemcom F@st 5690 | Bell (Giga Hub) | sha512 | username: admin, password: "" | | Sagemcom F@st 5655V2 | MásMóvil | md5 | | | Sagemcom F@st 5657IL | | md5 | | | Speedport Pro | Telekom | md5 | username: admin | diff --git a/poetry.lock b/poetry.lock index e23ac13..2b0fde7 100644 --- a/poetry.lock +++ b/poetry.lock @@ -14,6 +14,7 @@ charset-normalizer = ">=2.0,<3.0" frozenlist = ">=1.1.1" multidict = ">=4.5,<7.0" yarl = ">=1.0,<2.0" +requests = ">2.0" [package.extras] speedups = ["aiodns", "brotli", "cchardet"] @@ -425,7 +426,7 @@ py = ">=1.8.2" tomli = ">=1.0.0" [package.extras] -testing = ["argcomplete", "hypothesis (>=3.56)", "mock", "nose", "pygments (>=2.7.2)", "requests", "xmlschema"] +testing = ["argcomplete", "hypothesis (>=3.56)", "mock", "nose", "pygments (>=2.7.2)", "xmlschema"] [[package]] name = "pyupgrade" diff --git a/pyproject.toml b/pyproject.toml index 1f41e03..ad23294 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [tool.poetry] name = "sagemcom_api" -version = "1.0.8" +version = "1.0.9" description = "Python client to interact with SagemCom F@st routers via internal API's." authors = ["Mick Vleeshouwer "] license = "MIT" @@ -20,6 +20,7 @@ packages = [ python = ">=3.9,<4.0" aiohttp = "^3.7.3" pyhumps = "^3.0.2" +requests = ">2.0" [tool.poetry.dev-dependencies] pytest = "^7.1" diff --git a/sagemcom_api/client.py b/sagemcom_api/client.py index 044b8ed..2f95129 100644 --- a/sagemcom_api/client.py +++ b/sagemcom_api/client.py @@ -10,7 +10,7 @@ from typing import Dict, List, Optional, Type import urllib.parse -from aiohttp import ClientSession, ClientTimeout +from aiohttp import ClientConnectionError, ClientSession, ClientTimeout from aiohttp.connector import TCPConnector import humps @@ -33,6 +33,7 @@ AccessRestrictionException, AuthenticationException, BadRequestException, + LoginConnectionException, LoginTimeoutException, MaximumSessionCountException, NonWritableParameterException, @@ -40,7 +41,7 @@ UnknownException, UnknownPathException, ) -from .models import Device, DeviceInfo, PortMapping +from .models import Device, DeviceInfo, PortMapping, SpeedTestResult class SagemcomClient: @@ -55,6 +56,7 @@ def __init__( session: ClientSession = None, ssl=False, verify_ssl=True, + keep_keys=False, ): """ Create a SagemCom client. @@ -69,6 +71,7 @@ def __init__( self.username = username self.authentication_method = authentication_method self._password_hash = self.__generate_hash(password) + self.keep_keys = keep_keys self.protocol = "https" if ssl else "http" @@ -156,7 +159,7 @@ def __get_response(self, response, index=0): return value - def __get_response_value(self, response, index=0): + def __get_response_value(self, response, index=0, keep_keys = None): """Retrieve response value from value.""" try: value = self.__get_response(response, index)["value"] @@ -164,11 +167,12 @@ def __get_response_value(self, response, index=0): value = None # Rewrite result to snake_case - value = humps.decamelize(value) + if (keep_keys is not None and not keep_keys) or (keep_keys is None and not self.keep_keys): + value = humps.decamelize(value) return value - async def __api_request_async(self, actions, priority=False): + async def __api_request_async(self, actions, priority=False, **request_kwargs): """Build request to the internal JSON-req API.""" self.__generate_request_id() self.__generate_nonce() @@ -188,7 +192,9 @@ async def __api_request_async(self, actions, priority=False): } async with self.session.post( - api_host, data="req=" + json.dumps(payload, separators=(",", ":")) + api_host, + data="req=" + json.dumps(payload, separators=(",", ":")), + **request_kwargs, ) as response: if response.status == 400: @@ -272,6 +278,10 @@ async def login(self): raise LoginTimeoutException( "Request timed-out. This is mainly due to using the wrong encryption method." ) from exception + except ClientConnectionError as exception: + raise LoginConnectionException( + "Unable to connect to the device. Please check the host address." + ) from exception data = self.__get_response(response) @@ -293,7 +303,7 @@ async def logout(self): self._request_id = -1 async def get_value_by_xpath( - self, xpath: str, options: Optional[Dict] = {} + self, xpath: str, options: Optional[Dict] = {}, keep_keys = None ) -> Dict: """ Retrieve raw value from router using XPath. @@ -309,11 +319,11 @@ async def get_value_by_xpath( } response = await self.__api_request_async([actions], False) - data = self.__get_response_value(response) + data = self.__get_response_value(response, keep_keys = keep_keys) return data - async def get_values_by_xpaths(self, xpaths, options: Optional[Dict] = {}) -> Dict: + async def get_values_by_xpaths(self, xpaths, options: Optional[Dict] = {}, keep_keys = None) -> Dict: """ Retrieve raw values from router using XPath. @@ -331,7 +341,7 @@ async def get_values_by_xpaths(self, xpaths, options: Optional[Dict] = {}) -> Di ] response = await self.__api_request_async(actions, False) - values = [self.__get_response_value(response, i) for i in range(len(xpaths))] + values = [self.__get_response_value(response, i, keep_keys = keep_keys) for i in range(len(xpaths))] data = dict(zip(xpaths.keys(), values)) return data @@ -361,7 +371,7 @@ async def set_value_by_xpath( async def get_device_info(self) -> DeviceInfo: """Retrieve information about Sagemcom F@st device.""" try: - data = await self.get_value_by_xpath("Device/DeviceInfo") + data = await self.get_value_by_xpath("Device/DeviceInfo", keep_keys = False) return DeviceInfo(**data.get("device_info")) except UnknownPathException: data = await self.get_values_by_xpaths( @@ -380,7 +390,7 @@ async def get_device_info(self) -> DeviceInfo: async def get_hosts(self, only_active: Optional[bool] = False) -> List[Device]: """Retrieve hosts connected to Sagemcom F@st device.""" - data = await self.get_value_by_xpath("Device/Hosts/Hosts") + data = await self.get_value_by_xpath("Device/Hosts/Hosts", keep_keys = False) devices = [Device(**d) for d in data] if only_active: @@ -391,11 +401,30 @@ async def get_hosts(self, only_active: Optional[bool] = False) -> List[Device]: async def get_port_mappings(self) -> List[PortMapping]: """Retrieve configured Port Mappings on Sagemcom F@st device.""" - data = await self.get_value_by_xpath("Device/NAT/PortMappings") + data = await self.get_value_by_xpath("Device/NAT/PortMappings", keep_keys = False) port_mappings = [PortMapping(**p) for p in data] return port_mappings + async def get_logs(self) -> List[str]: + """ + Retrieve system logs. + """ + + actions = { + "id": 0, + "method": "getVendorLogDownloadURI", + "xpath": urllib.parse.quote("Device/DeviceInfo/VendorLogFiles/VendorLogFile[@uid='1']"), + } + + response = await self.__api_request_async([actions], False) + log_path = response["reply"]["actions"][0]["callbacks"][0]["parameters"]["uri"] + + log_uri = f"{self.protocol}://{self.host}{log_path}" + response = await self.session.get(log_uri, timeout=10) + + return await response.text() + async def reboot(self): """Reboot Sagemcom F@st device.""" action = { @@ -405,6 +434,38 @@ async def reboot(self): } response = await self.__api_request_async([action], False) - data = self.__get_response_value(response) + data = self.__get_response_value(response, keep_keys = False) return data + + async def run_speed_test(self, block_traffic: bool = False): + """Run Speed Test on Sagemcom F@st device.""" + actions = [ + { + "id": 0, + "method": "speedTestClient", + "xpath": "Device/IP/Diagnostics/SpeedTest", + "parameters": {"BlockTraffic": block_traffic}, + } + ] + return await self.__api_request_async(actions, False, timeout=100) + + async def get_speed_test_results(self): + """Retrieve Speed Test results from Sagemcom F@st device.""" + ret = await self.get_value_by_xpath("Device/IP/Diagnostics/SpeedTest") + history = ret["speed_test"]["history"] + if history: + timestamps = (int(k) for k in history["timestamp"].split(",")) + server_address = history["selected_server_address"].split(",") + block_traffic = history["block_traffic"].split(",") + latency = history["latency"].split(",") + upload = (float(k) for k in history["upload"].split(",")) + download = (float(k) for k in history["download"].split(",")) + results = [ + SpeedTestResult(*data) + for data in zip( + timestamps, server_address, block_traffic, latency, upload, download + ) + ] + return results + return [] diff --git a/sagemcom_api/exceptions.py b/sagemcom_api/exceptions.py index 8c2eedd..c320c39 100644 --- a/sagemcom_api/exceptions.py +++ b/sagemcom_api/exceptions.py @@ -20,6 +20,12 @@ class LoginTimeoutException(Exception): pass +class LoginConnectionException(Exception): + """Raised when a connection error is encountered during login.""" + + pass + + class NonWritableParameterException(Exception): """Raised when provided parameter is not writable.""" diff --git a/sagemcom_api/models.py b/sagemcom_api/models.py index 979049c..5b0c37e 100644 --- a/sagemcom_api/models.py +++ b/sagemcom_api/models.py @@ -2,6 +2,7 @@ import dataclasses from dataclasses import dataclass +import time from typing import Any, List, Optional @@ -162,3 +163,30 @@ def __init__(self, **kwargs): def id(self): """Return unique ID for port mapping.""" return self.uid + + +@dataclass +class SpeedTestResult: + """Representation of a speedtest result.""" + + timestamp: str + selected_server_address: str + block_traffic: bool + latency: str + upload: str + download: str + + def __post_init__(self): + """Process data after init.""" + # Convert timestamp to datetime object. + self.timestamp = time.strftime( + "%Y-%m-%d %H:%M:%S", time.localtime(self.timestamp) + ) + self.block_traffic = bool(self.block_traffic) + + def __str__(self) -> str: + """Return string representation of speedtest result.""" + return ( + f"timestamp: {self.timestamp}, latency: {self.latency}, " + f"upload: {self.upload}, download: {self.download}" + )