first commit

sagemcom_api/__init__.py

@@ -0,0 +1,2 @@
+"""Package to communicate with Sagemcom F@st internal APIs."""
+__version__ = "1.0.1"

sagemcom_api/client.py

@@ -0,0 +1,412 @@
+"""Client to communicate with Sagemcom F@st internal APIs."""
+from __future__ import annotations
+
+import asyncio
+import hashlib
+import json
+import math
+import random
+import urllib.parse
+import humps
+from types import TracebackType
+from typing import Dict, List, Optional, Type
+
+
+
+from aiohttp import ClientSession, ClientTimeout
+from aiohttp.connector import TCPConnector
+
+from . import __version__
+from .const import (
+    API_ENDPOINT,
+    DEFAULT_TIMEOUT,
+    DEFAULT_USER_AGENT,
+    XMO_ACCESS_RESTRICTION_ERR,
+    XMO_AUTHENTICATION_ERR,
+    XMO_MAX_SESSION_COUNT_ERR,
+    XMO_NO_ERR,
+    XMO_NON_WRITABLE_PARAMETER_ERR,
+    XMO_REQUEST_ACTION_ERR,
+    XMO_REQUEST_NO_ERR,
+    XMO_UNKNOWN_PATH_ERR,
+)
+from .enums import EncryptionMethod
+from .exceptions import (
+    AccessRestrictionException,
+    AuthenticationException,
+    BadRequestException,
+    LoginTimeoutException,
+    MaximumSessionCountException,
+    NonWritableParameterException,
+    UnauthorizedException,
+    UnknownException,
+    UnknownPathException,
+)
+from .models import Device, DeviceInfo, PortMapping
+
+
+class SagemcomClient:
+    """Client to communicate with the Sagemcom API."""
+
+    def __init__(
+        self,
+        host,
+        username,
+        password,
+        authentication_method,
+        session: ClientSession = None,
+        ssl=False,
+        verify_ssl=True,
+    ):
+        """
+        Create a SagemCom client.
+
+        :param host: the host of your Sagemcom router
+        :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 session: use a custom session, for example to configure the timeout
+        """
+        self.host = host
+        self.username = username
+        self.authentication_method = authentication_method
+        self._password_hash = self.__generate_hash(password)
+
+        self.protocol = "https" if ssl else "http"
+
+        self._current_nonce = None
+        self._server_nonce = ""
+        self._session_id = 0
+        self._request_id = -1
+
+        self.session = (
+            session
+            if session
+            else ClientSession(
+                headers={"User-Agent": f"{DEFAULT_USER_AGENT}/{__version__}"},
+                timeout=ClientTimeout(DEFAULT_TIMEOUT),
+                connector=TCPConnector(ssl=verify_ssl),
+            )
+        )
+
+    async def __aenter__(self) -> SagemcomClient:
+        """TODO."""
+        return self
+
+    async def __aexit__(
+        self,
+        exc_type: Optional[Type[BaseException]],
+        exc_value: Optional[BaseException],
+        traceback: Optional[TracebackType],
+    ) -> None:
+        """Close session on exit."""
+        await self.close()
+
+    async def close(self) -> None:
+        """Close the websession."""
+        await self.session.close()
+
+    def __generate_nonce(self):
+        """Generate pseudo random number (nonce) to avoid replay attacks."""
+        self._current_nonce = math.floor(random.randrange(0, 1) * 500000)
+
+    def __generate_request_id(self):
+        """Generate sequential request ID."""
+        self._request_id += 1
+
+    def __generate_hash(self, value, authentication_method=None):
+        """Hash value with selected encryption method and return HEX value."""
+        auth_method = authentication_method or self.authentication_method
+
+        bytes_object = bytes(value, encoding="utf-8")
+
+        if auth_method == EncryptionMethod.MD5:
+            return hashlib.md5(bytes_object).hexdigest()
+
+        if auth_method == EncryptionMethod.SHA512:
+            return hashlib.sha512(bytes_object).hexdigest()
+
+        return value
+
+    def __get_credential_hash(self):
+        """Build credential hash."""
+        return self.__generate_hash(
+            self.username + ":" + self._server_nonce + ":" + self._password_hash
+        )
+
+    def __generate_auth_key(self):
+        """Build auth key."""
+        credential_hash = self.__get_credential_hash()
+        auth_string = f"{credential_hash}:{self._request_id}:{self._current_nonce}:JSON:{API_ENDPOINT}"
+        self._auth_key = self.__generate_hash(auth_string)
+
+    def __get_response_error(self, response):
+        """Retrieve response error from result."""
+        try:
+            value = response["reply"]["error"]
+        except KeyError:
+            value = None
+
+        return value
+
+    def __get_response(self, response, index=0):
+        """Retrieve response from result."""
+        try:
+            value = response["reply"]["actions"][index]["callbacks"][0]["parameters"]
+        except KeyError:
+            value = None
+
+        return value
+
+    def __get_response_value(self, response, index=0):
+        """Retrieve response value from value."""
+        try:
+            value = self.__get_response(response, index)["value"]
+        except KeyError:
+            value = None
+
+        # Rewrite result to snake_case
+        value = humps.decamelize(value)
+
+        return value
+
+    async def __api_request_async(self, actions, priority=False):
+        """Build request to the internal JSON-req API."""
+        self.__generate_request_id()
+        self.__generate_nonce()
+        self.__generate_auth_key()
+
+        api_host = f"{self.protocol}://{self.host}{API_ENDPOINT}"
+
+        payload = {
+            "request": {
+                "id": self._request_id,
+                "session-id": str(self._session_id),
+                "priority": priority,
+                "actions": actions,
+                "cnonce": self._current_nonce,
+                "auth-key": self._auth_key,
+            }
+        }
+
+        async with self.session.post(
+            api_host, data="req=" + json.dumps(payload, separators=(",", ":"))
+        ) as response:
+
+            if response.status == 400:
+                result = await response.text()
+                raise BadRequestException(result)
+
+            if response.status != 200:
+                result = await response.text()
+                raise UnknownException(result)
+
+            if response.status == 200:
+                result = await response.json()
+                error = self.__get_response_error(result)
+
+                # No errors
+                if (
+                    error["description"] == XMO_REQUEST_NO_ERR
+                    or error["description"] == "Ok"  # NOQA: W503
+                ):
+                    return result
+
+                # Error in one of the actions
+                if error["description"] == XMO_REQUEST_ACTION_ERR:
+
+                    # TODO How to support multiple actions + error handling?
+                    actions = result["reply"]["actions"]
+                    for action in actions:
+                        action_error = action["error"]
+                        action_error_description = action_error["description"]
+
+                        if action_error_description == XMO_NO_ERR:
+                            continue
+
+                        if action_error_description == XMO_AUTHENTICATION_ERR:
+                            raise AuthenticationException(action_error)
+
+                        if action_error_description == XMO_ACCESS_RESTRICTION_ERR:
+                            raise AccessRestrictionException(action_error)
+
+                        if action_error_description == XMO_NON_WRITABLE_PARAMETER_ERR:
+                            raise NonWritableParameterException(action_error)
+
+                        if action_error_description == XMO_UNKNOWN_PATH_ERR:
+                            raise UnknownPathException(action_error)
+
+                        if action_error_description == XMO_MAX_SESSION_COUNT_ERR:
+                            raise MaximumSessionCountException(action_error)
+
+                        raise UnknownException(action_error)
+
+                return result
+
+    async def login(self):
+        """TODO."""
+        actions = {
+            "method": "logIn",
+            "parameters": {
+                "user": self.username,
+                "persistent": True,
+                "session-options": {
+                    "nss": [{"name": "gtw", "uri": "http://sagemcom.com/gateway-data"}],
+                    "language": "ident",
+                    "context-flags": {"get-content-name": True, "local-time": True},
+                    "capability-depth": 2,
+                    "capability-flags": {
+                        "name": True,
+                        "default-value": False,
+                        "restriction": True,
+                        "description": False,
+                    },
+                    "time-format": "ISO_8601",
+                    "write-only-string": "_XMO_WRITE_ONLY_",
+                    "undefined-write-only-string": "_XMO_UNDEFINED_WRITE_ONLY_",
+                },
+            },
+        }
+
+        try:
+            response = await self.__api_request_async([actions], True)
+        except asyncio.TimeoutError as exception:
+            raise LoginTimeoutException(
+                "Request timed-out. This is mainly due to using the wrong encryption method."
+            ) from exception
+
+        data = self.__get_response(response)
+
+        if data["id"] is not None and data["nonce"] is not None:
+            self._session_id = data["id"]
+            self._server_nonce = data["nonce"]
+            return True
+        else:
+            raise UnauthorizedException(data)
+
+    async def logout(self):
+        """Log out of the Sagemcom F@st device."""
+        actions = {"id": 0, "method": "logOut"}
+
+        await self.__api_request_async([actions], False)
+
+        self._session_id = -1
+        self._server_nonce = ""
+        self._request_id = -1
+
+    async def get_value_by_xpath(
+        self, xpath: str, options: Optional[Dict] = {}
+    ) -> Dict:
+        """
+        Retrieve raw value from router using XPath.
+
+        :param xpath: path expression
+        :param options: optional options
+        """
+        actions = {
+            "id": 0,
+            "method": "getValue",
+            "xpath": urllib.parse.quote(xpath),
+            "options": options,
+        }
+
+        response = await self.__api_request_async([actions], False)
+        data = self.__get_response_value(response)
+
+        return data
+
+    async def get_values_by_xpaths(self, xpaths, options: Optional[Dict] = {}) -> Dict:
+        """
+        Retrieve raw values from router using XPath.
+
+        :param xpaths: Dict of key to xpath expression
+        :param options: optional options
+        """
+        actions = [
+            {
+                "id": i,
+                "method": "getValue",
+                "xpath": urllib.parse.quote(xpath),
+                "options": options,
+            }
+            for i, xpath in enumerate(xpaths.values())
+        ]
+
+        response = await self.__api_request_async(actions, False)
+        values = [self.__get_response_value(response, i) for i in range(len(xpaths))]
+        data = dict(zip(xpaths.keys(), values))
+
+        return data
+
+    async def set_value_by_xpath(
+        self, xpath: str, value: str, options: Optional[Dict] = {}
+    ) -> Dict:
+        """
+        Retrieve raw value from router using XPath.
+
+        :param xpath: path expression
+        :param value: value
+        :param options: optional options
+        """
+        actions = {
+            "id": 0,
+            "method": "setValue",
+            "xpath": xpath,
+            "parameters": {"value": str(value)},
+            "options": options,
+        }
+
+        response = await self.__api_request_async([actions], False)
+
+        return response
+
+    async def get_device_info(self) -> DeviceInfo:
+        """Retrieve information about Sagemcom F@st device."""
+        try:
+            data = await self.get_value_by_xpath("Device/DeviceInfo")
+            return DeviceInfo(**data.get("device_info"))
+        except UnknownPathException:
+            data = await self.get_values_by_xpaths(
+                {
+                    "mac_address": "Device/DeviceInfo/MACAddress",
+                    "model_name": "Device/DeviceInfo/ModelNumber",
+                    "model_number": "Device/DeviceInfo/ProductClass",
+                    "product_class": "Device/DeviceInfo/ProductClass",
+                    "serial_number": "Device/DeviceInfo/SerialNumber",
+                    "software_version": "Device/DeviceInfo/SoftwareVersion",
+                }
+            )
+            data["manufacturer"] = "Sagemcom"
+
+        return DeviceInfo(**data)
+
+    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")
+        devices = [Device(**d) for d in data]
+
+        if only_active:
+            active_devices = [d for d in devices if d.active is True]
+            return active_devices
+
+        return devices
+
+    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")
+        port_mappings = [PortMapping(**p) for p in data]
+
+        return port_mappings
+
+    async def reboot(self):
+        """Reboot Sagemcom F@st device."""
+        action = {
+            "method": "reboot",
+            "xpath": "Device",
+            "parameters": {"source": "GUI"},
+        }
+
+        response = await self.__api_request_async([action], False)
+        data = self.__get_response_value(response)
+
+        return data

sagemcom_api/const.py

@@ -0,0 +1,14 @@
+"""Constants for the Sagemcom F@st client."""
+API_ENDPOINT = "/cgi/json-req"
+
+DEFAULT_TIMEOUT = 7
+DEFAULT_USER_AGENT = "Python_Sagemcom"
+
+XMO_ACCESS_RESTRICTION_ERR = "XMO_ACCESS_RESTRICTION_ERR"
+XMO_AUTHENTICATION_ERR = "XMO_AUTHENTICATION_ERR"
+XMO_NON_WRITABLE_PARAMETER_ERR = "XMO_NON_WRITABLE_PARAMETER_ERR"
+XMO_NO_ERR = "XMO_NO_ERR"
+XMO_REQUEST_ACTION_ERR = "XMO_REQUEST_ACTION_ERR"
+XMO_REQUEST_NO_ERR = "XMO_REQUEST_NO_ERR"
+XMO_UNKNOWN_PATH_ERR = "XMO_UNKNOWN_PATH_ERR"
+XMO_MAX_SESSION_COUNT_ERR = "XMO_MAX_SESSION_COUNT_ERR"

sagemcom_api/enums.py

@@ -0,0 +1,10 @@
+"""Enums for the Sagemcom F@st client."""
+
+from enum import Enum
+
+
+class EncryptionMethod(Enum):
+    """Encryption method defining the password hash."""
+
+    MD5 = "MD5"
+    SHA512 = "SHA512"

sagemcom_api/exceptions.py

@@ -0,0 +1,58 @@
+"""Exceptions for the Sagemcom F@st client."""
+
+
+# Exceptions provided by SagemCom API
+class AccessRestrictionException(Exception):
+    """Raised when current user has access restrictions."""
+
+    pass
+
+
+class AuthenticationException(Exception):
+    """Raised when authentication is not correct."""
+
+    pass
+
+
+class LoginTimeoutException(Exception):
+    """Raised when a timeout is encountered during login."""
+
+    pass
+
+
+class NonWritableParameterException(Exception):
+    """Raised when provided parameter is not writable."""
+
+    pass
+
+
+class UnknownPathException(Exception):
+    """Raised when provided path does not exist."""
+
+    pass
+
+
+class MaximumSessionCountException(Exception):
+    """Raised when the maximum session count is reached."""
+
+    pass
+
+
+# Exceptions provided by this library
+# TODO Validate our own errors
+class BadRequestException(Exception):
+    """TODO."""
+
+    pass
+
+
+class UnauthorizedException(Exception):
+    """TODO."""
+
+    pass
+
+
+class UnknownException(Exception):
+    """TODO."""
+
+    pass

sagemcom_api/models.py

@@ -0,0 +1,164 @@
+"""Models for the Sagemcom F@st client."""
+
+import dataclasses
+from dataclasses import dataclass
+from typing import Any, List, Optional
+
+
+@dataclass
+class Device:
+    """Device connected to a router."""
+
+    uid: Optional[int] = None
+    alias: Optional[str] = None
+    phys_address: Optional[str] = None
+    ip_address: Optional[str] = None
+    address_source: Optional[str] = None
+    dhcp_client: Optional[str] = None
+    lease_time_remaining: Optional[int] = None
+    associated_device: Optional[Any] = None
+    layer1_interface: Optional[Any] = None
+    layer3_interface: Optional[Any] = None
+    vendor_class_id: Optional[Any] = None
+    client_id: Optional[Any] = None
+    user_class_id: Optional[Any] = None
+    host_name: Optional[Any] = None
+    active: Optional[bool] = None
+    lease_start: Optional[int] = None
+    lease_duration: Optional[int] = None
+    interface_type: Optional[str] = None  # enum!
+    detected_device_type: Optional[str] = None
+    active_last_change: Optional[Any] = None
+    user_friendly_name: Optional[str] = None
+    user_host_name: Optional[str] = None
+    user_device_type: Optional[Any] = None  # enum!
+    icon: Optional[Any] = None
+    room: Optional[Any] = None
+    blacklist_enable: Optional[bool] = None
+    blacklisted: Optional[bool] = None
+    unblock_hours_count: Optional[int] = None
+    blacklist_status: Optional[bool] = None
+    blacklisted_according_to_schedule: Optional[bool] = None
+    blacklisted_schedule: Optional[List] = None
+    hidden: Optional[bool] = None
+    options: Optional[List] = None
+    vendor_class_idv6: Optional[Any] = None
+    ipv4_addresses: Optional[List] = None
+    ipv6_addresses: Optional[List] = None
+    device_type_association: Optional[Any] = None
+
+    # TODO Remove extra kwargs before init
+    def __init__(self, **kwargs):
+        """Override to accept more args than specified."""
+        names = {f.name for f in dataclasses.fields(self)}
+        for k, v in kwargs.items():
+            if k in names:
+                setattr(self, k, v)
+
+    @property
+    def id(self):
+        """Return unique ID for device."""
+        return self.phys_address.upper()
+
+    @property
+    def name(self):
+        """Return name of the device."""
+        return self.user_host_name or self.host_name
+
+
+@dataclass
+class DeviceInfo:
+    """Sagemcom Router representation."""
+
+    mac_address: str
+    serial_number: Optional[str] = None
+    manufacturer: Optional[Any] = None
+    model_name: Optional[Any] = None
+    model_number: Optional[Any] = None
+    software_version: Optional[str] = None
+    hardware_version: Optional[str] = None
+    up_time: Optional[Any] = None
+    reboot_count: Optional[Any] = None
+    router_name: Optional[Any] = None
+    bootloader_version: Optional[Any] = None
+    device_category: Optional[Any] = None
+    manufacturer_oui: Optional[Any] = None
+    product_class: Optional[str] = None
+    description: Optional[str] = None
+    additional_hardware_version: Optional[str] = None
+    additional_software_version: Optional[str] = None
+    external_firmware_version: Optional[str] = None
+    internal_firmware_version: Optional[str] = None
+    gui_firmware_version: Optional[str] = None
+    guiapi_version: Optional[float] = None
+    provisioning_code: Optional[str] = None
+    up_time: Optional[int] = None
+    first_use_date: Optional[str] = None
+    mac_address: Optional[str] = None
+    mode: Optional[str] = None
+    country: Optional[str] = None
+    reboot_count: Optional[int] = None
+    nodes_to_restore: Optional[str] = None
+    router_name: Optional[str] = None
+    reboot_status: Optional[float] = None
+    reset_status: Optional[float] = None
+    update_status: Optional[float] = None
+    SNMP: Optional[bool] = None
+    first_connection: Optional[bool] = None
+    build_date: Optional[str] = None
+    spec_version: Optional[str] = None
+    CLID: Optional[str] = None
+    flush_device_log: Optional[bool] = None
+    locations: Optional[str] = None
+    api_version: Optional[str] = None
+
+    # TODO Remove extra kwargs before init
+    def __init__(self, **kwargs):
+        """Override to accept more args than specified."""
+        names = {f.name for f in dataclasses.fields(self)}
+        for k, v in kwargs.items():
+            if k in names:
+                setattr(self, k, v)
+
+    @property
+    def id(self):
+        """Return unique ID for gateway."""
+        return self.mac_address
+
+
+@dataclass
+class PortMapping:
+    """Port Mapping representation."""
+
+    uid: int
+    enable: bool
+    status: Optional[str] = None  # Enum
+    alias: Optional[str] = None
+    external_interface: Optional[str] = None
+    all_external_interfaces: Optional[bool] = None
+    lease_duration: Optional[int] = None
+    external_port: Optional[int] = None
+    external_port_end_range: Optional[int] = None
+    internal_interface: Optional[str] = None
+    internal_port: Optional[int] = None
+    protocol: Optional[str] = None
+    service: Optional[str] = None
+    internal_client: Optional[str] = None
+    public_ip: Optional[str] = None
+    description: Optional[str] = None
+    creator: Optional[str] = None
+    target: Optional[str] = None
+    lease_start: Optional[str] = None  # Date?
+
+    # TODO Remove extra kwargs before init
+    def __init__(self, **kwargs):
+        """Override to accept more args than specified."""
+        names = {f.name for f in dataclasses.fields(self)}
+        for k, v in kwargs.items():
+            if k in names:
+                setattr(self, k, v)
+
+    @property
+    def id(self):
+        """Return unique ID for port mapping."""
+        return self.uid