Merge pull request #18 from eduardormorais/release/0.3.0

Updating version release 0.3.0

3 files changed, 415 insertions, 181 deletions

diff --git a/pyvyos/__init__.py b/pyvyos/__init__.py
index 7c327e5..44bd563 100644
--- a/pyvyos/__init__.py
+++ b/pyvyos/__init__.py
@@ -1,2 +1,2 @@
 from .device import VyDevice
-from .device import ApiResponse
\ No newline at end of file
+from .device import ApiResponse
diff --git a/pyvyos/device.py b/pyvyos/device.py
index d168b65..246a913 100644
--- a/pyvyos/device.py
+++ b/pyvyos/device.py
@@ -1,26 +1,10 @@
-import urllib3
-import requests
-import json
-import pprint
-from dataclasses import dataclass
-
-@dataclass
-class ApiResponse:
-    """
-    Represents an API response.
+import warnings
+from typing import List, Literal
 
-    Attributes:
-        status (int): The HTTP status code of the response.
-        request (dict): The request payload sent to the API.
-        result (dict): The data result of the API response.
-        error (str): Any error message in case of a failed response.
-    """
-    status: int
-    request: dict
-    result: dict
-    error: str
+from .rest import ApiResponse, RestClient
 
-class VyDevice:
+
+class VyDevice(RestClient):
     """
     Represents a device for interacting with the VyOS API.
 
@@ -51,8 +35,8 @@ class VyDevice:
         image_delete(name, url=None, file=None, path=[]): Delete a specific image.
         show(path=[]): Show configuration information.
         generate(path=[]): Generate configuration based on specified path.
-        configure_set(path=[]): Sets configuration based on the specified path. This method is versatile, accepting 
-        either a single configuration path or a list of configuration paths. This flexibility 
+        configure_set(path=[]): Sets configuration based on the specified path. This method is versatile, accepting
+        either a single configuration path or a list of configuration paths. This flexibility
         allows for setting both individual and multiple configurations in a single operation.
         configure_delete(path=[]): Delete configuration based on specified path.
         config_file_save(file=None): Save the configuration to a file.
@@ -61,140 +45,37 @@ class VyDevice:
         poweroff(path=["now"]): Power off the device.
     """
 
-    def __init__(self, hostname, apikey, protocol='https', port=443, verify=True, timeout=10):
-        """
-        Initializes a VyDevice instance.
-
-        Args:
-            hostname (str): The hostname or IP address of the VyOS device.
-            apikey (str): The API key for authentication.
-            protocol (str, optional): The protocol to use (default is 'https').
-            port (int, optional): The port to use (default is 443).
-            verify (bool, optional): Whether to verify SSL certificates (default is True).
-            timeout (int, optional): The request timeout in seconds (default is 10).
-        """
-        self.hostname = hostname
-        self.apikey = apikey
-        self.protocol = protocol
-        self.port = port
-        self.verify = verify
-        self.timeout = timeout
-
-    def _get_url(self, command):
-        """
-        Get the full URL for a specific API command.
-
-        Args:
-            command (str): The API command to construct the URL for.
-
-        Returns:
-            str: The full URL for the API command.
-        """
-        return f"{self.protocol}://{self.hostname}:{self.port}/{command}"
-
-    def _get_payload(self, op, path=[], file=None, url=None, name=None):
-        """
-        Generate the payload for an API request.
-
-        Args:
-            op (str): The operation to perform in the API request.
-            path (list, optional): The path elements for the API request. This can be a single list for a single
-                                configuration path or a list of lists for multiple configuration paths.
-            file (str, optional): The file to include in the request (default is None).
-            url (str, optional): The URL to include in the request (default is None).
-            name (str, optional): The name to include in the request (default is None).
-
-        Returns:
-            dict: The payload for the API request.
-        """
-        # Adjusting the data structure based on whether path is single or multiple
-        if isinstance(path[0], list):  # Handling multiple paths
-            data = [{'op': op, 'path': p} for p in path]
-        else:  # Handling a single path
-            data = {'op': op, 'path': path}
-
-        # Including the optional parameters if provided
-        if file:
-            if isinstance(data, list):  # If data is a list of dicts (multiple paths)
-                for d in data:
-                    d['file'] = file
-            else:  # If data is a single dict (single path)
-                data['file'] = file
-                
-        if url:
-            if isinstance(data, list):
-                for d in data:
-                    d['url'] = url
-            else:
-                data['url'] = url
-            
-        if name:
-            if isinstance(data, list):
-                for d in data:
-                    d['name'] = name
-            else:
-                data['name'] = name
-                
-        payload = {
-            'data': json.dumps(data),
-            'key': self.apikey
-        }
-
-        return payload
-
-
-    def _api_request(self, command, op, path=[], method='POST', file=None, url=None, name=None):
-        """
-        Make an API request.
-
-        Args:
-            command (str): The API command to execute.
-            op (str): The operation to perform in the API request.
-            path (list, optional): The path elements for the API request (default is an empty list).
-            method (str, optional): The HTTP method to use for the request (default is 'POST').
-            file (str, optional): The file to include in the request (default is None).
-            url (str, optional): The URL to include in the request (default is None).
-            name (str, optional): The name to include in the request (default is None).
-
-        Returns:
-            ApiResponse: An ApiResponse object representing the API response.
-        """
-        url = self._get_url(command)
-        payload = self._get_payload(op, path=path, file=file, url=url, name=name)
-        
-        headers = {}
-        error = False      
-        result = {}
-
-        try:
-            resp = requests.post(url, verify=self.verify, data=payload, timeout=self.timeout, headers=headers)
-
-            if resp.status_code == 200:
-                try:
-                    resp_decoded = resp.json()
-                    
-                    if resp_decoded['success'] == True:
-                        result = resp_decoded['data']
-                        error = False
-                    else:   
-                        error = resp_decoded['error']
-                   
-                except json.JSONDecodeError:
-                    error = 'json decode error'
-            else:
-                error = 'http error'
-
-            status = resp.status_code
-
-        except requests.exceptions.ConnectionError as e:
-            error = 'connection error: ' + str(e)
-            status = 0
-  
-        # Removing apikey from payload for security reasons
-        del(payload['key'])
-        return ApiResponse(status=status, request=payload, result=result, error=error)
-
-    def retrieve_show_config(self, path=[]):
+    def __init__(
+        self,
+        hostname: str,
+        apikey: str,
+        protocol: Literal["http", "https"] = "https",
+        port: int = 443,
+        verify: bool = True,
+        timeout: int = 10,
+    ):
+        super().__init__(
+            hostname, apikey, protocol, int(port), bool(verify), int(timeout)
+        )
+        self._validate_params()
+
+    def _validate_params(
+        self,
+    ) -> None:
+        """Validação centralizada de parâmetros"""
+        if not isinstance(self.hostname, str) or len(self.hostname) < 3:
+            raise ValueError("Invalid hostname")
+
+        if self.protocol not in ("http", "https"):
+            raise ValueError("The protocol must be http or https")
+
+        if not 1 <= self.port <= 65535:
+            raise ValueError("Port out of valid range (1-65535)")
+
+        if self.timeout and self.timeout < 1:
+            warnings.warn("Timeout below 1s may cause instability", UserWarning)
+
+    def retrieve_show_config(self, path: List = None):
         """
         Retrieve and show the device configuration.
 
@@ -204,9 +85,12 @@ class VyDevice:
         Returns:
             ApiResponse: An ApiResponse object representing the API response.
         """
-        return self._api_request(command="retrieve", op='showConfig', path=path, method="POST")
 
-    def retrieve_return_values(self, path=[]):
+        return self._api_request(
+            command="retrieve", op="showConfig", path=path, method="POST"
+        )
+
+    def retrieve_return_values(self, path: List = None):
         """
         Retrieve and return specific configuration values.
 
@@ -216,9 +100,11 @@ class VyDevice:
         Returns:
             ApiResponse: An ApiResponse object representing the API response.
         """
-        return self._api_request(command="retrieve", op='returnValues', path=path, method="POST")
+        return self._api_request(
+            command="retrieve", op="returnValues", path=path, method="POST"
+        )
 
-    def reset(self, path=[]):
+    def reset(self, path: List = None):
         """
         Reset a specific configuration element.
 
@@ -228,7 +114,7 @@ class VyDevice:
         Returns:
             ApiResponse: An ApiResponse object representing the API response.
         """
-        return self._api_request(command="reset", op='reset', path=path, method="POST")
+        return self._api_request(command="reset", op="reset", path=path, method="POST")
 
     def image_add(self, url=None, file=None, path=[]):
         """
@@ -242,7 +128,7 @@ class VyDevice:
         Returns:
             ApiResponse: An ApiResponse object representing the API response.
         """
-        return self._api_request(command="image", op='add', url=url, method="POST")
+        return self._api_request(command="image", op="add", url=url, method="POST")
 
     def image_delete(self, name, url=None, file=None, path=[]):
         """
@@ -257,9 +143,9 @@ class VyDevice:
         Returns:
             ApiResponse: An ApiResponse object representing the API response.
         """
-        return self._api_request(command="image", op='delete', name=name, method="POST")
+        return self._api_request(command="image", op="delete", name=name, method="POST")
 
-    def show(self, path=[]):
+    def show(self, path: List = None):
         """
         Show configuration information.
 
@@ -269,9 +155,9 @@ class VyDevice:
         Returns:
             ApiResponse: An ApiResponse object representing the API response.
         """
-        return self._api_request(command="show", op='show', path=path, method="POST")
+        return self._api_request(command="show", op="show", path=path, method="POST")
 
-    def generate(self, path=[]):
+    def generate(self, path: List = None):
         """
         Generate configuration based on the given path.
 
@@ -281,9 +167,11 @@ class VyDevice:
         Returns:
             ApiResponse: An ApiResponse object representing the API response.
         """
-        return self._api_request(command="generate", op='generate', path=path, method="POST")
+        return self._api_request(
+            command="generate", op="generate", path=path, method="POST"
+        )
 
-    def configure_set(self, path=[]):
+    def configure_set(self, path: List = None):
         """
         Set configuration based on the given path.
 
@@ -293,10 +181,11 @@ class VyDevice:
         Returns:
             ApiResponse: An ApiResponse object representing the API response.
         """
-        return self._api_request(command="configure", op='set', path=path, method="POST")
+        return self._api_request(
+            command="configure", op="set", path=path, method="POST"
+        )
 
-
-    def configure_delete(self, path=[]):
+    def configure_delete(self, path: List = None):
         """
         Delete configuration based on the given path.
 
@@ -306,7 +195,22 @@ class VyDevice:
         Returns:
             ApiResponse: An ApiResponse object representing the API response.
         """
-        return self._api_request(command="configure", op='delete', path=path, method="POST")
+        return self._api_request(
+            command="configure", op="delete", path=path, method="POST"
+        )
+
+    def configure_multiple_op(self, op_path: List = None):
+        """
+        Set configuration based on the given {operation : path} for multiple operation.
+
+        Args:
+            op_path (list): The path elements for configuration deletion  or/and setting.
+            eg: [{'op': 'delete', 'path': [...]}, {'op': 'set', 'path': [...]}]
+
+        Returns:
+            ApiResponse: An ApiResponse object representing the API response.
+        """
+        return self._api_request(command="configure", op="", path=op_path)
 
     def config_file_save(self, file=None):
         """
@@ -318,7 +222,9 @@ class VyDevice:
         Returns:
             ApiResponse: An ApiResponse object representing the API response.
         """
-        return self._api_request(command="config-file", op='save', file=file, method="POST")
+        return self._api_request(
+            command="config-file", op="save", file=file, method="POST"
+        )
 
     def config_file_load(self, file=None):
         """
@@ -330,9 +236,11 @@ class VyDevice:
         Returns:
             ApiResponse: An ApiResponse object representing the API response.
         """
-        return self._api_request(command="config-file", op='load', file=file, method="POST")
+        return self._api_request(
+            command="config-file", op="load", file=file, method="POST"
+        )
 
-    def reboot(self, path=["now"]):
+    def reboot(self, path: List = None):
         """
         Reboot the device.
 
@@ -342,9 +250,14 @@ class VyDevice:
         Returns:
             ApiResponse: An ApiResponse object representing the API response.
         """
-        return self._api_request(command="reboot", op='reboot', path=path, method="POST")
-    
-    def poweroff(self, path=["now"]):
+        if path is None:
+            path = ["now"]
+
+        return self._api_request(
+            command="reboot", op="reboot", path=path, method="POST"
+        )
+
+    def poweroff(self, path: List = None):
         """
         Power off the device.
 
@@ -354,4 +267,9 @@ class VyDevice:
         Returns:
             ApiResponse: An ApiResponse object representing the API response.
         """
-        return self._api_request(command="poweroff", op='poweroff', path=path, method="POST")
+        if path is None:
+            path = ["now"]
+
+        return self._api_request(
+            command="poweroff", op="poweroff", path=path, method="POST"
+        )
diff --git a/pyvyos/rest.py b/pyvyos/rest.py
new file mode 100644
index 0000000..04c786e
--- /dev/null
+++ b/pyvyos/rest.py
@@ -0,0 +1,316 @@
+import json
+from abc import ABC
+from dataclasses import dataclass
+from typing import Tuple, List, Union, Dict, Any, Optional
+
+import requests
+from requests import Response
+from requests.exceptions import (
+    HTTPError,
+    ConnectionError,
+    Timeout,
+    RequestException,
+    JSONDecodeError,
+)
+
+
+@dataclass
+class ApiResponse:
+    """
+    Represents an API response.
+
+    Attributes:
+        status (int): The HTTP status code of the response.
+        request (dict): The request payload sent to the API.
+        result (dict): The data result of the API response.
+        error (str): Any error message in case of a failed response.
+    """
+
+    status: int
+    request: dict
+    result: dict
+    error: str
+
+
+class RestClient(ABC):
+    """Secure REST client for integration with VyOS device APIs"""
+
+    hostname: str
+    apikey: str
+    protocol: str
+    port: int
+    verify: bool
+    timeout: int
+
+    def __init__(
+        self,
+        hostname: str,
+        apikey: str,
+        protocol: str = "https",
+        port: int = 443,
+        verify: bool = False,
+        timeout: int = 10,
+    ):
+        """
+        Args:
+            hostname: VyOS device address
+            apikey: API key for authentication
+            protocol: Protocol (http/https)
+            port: Access port
+            verify: Verify SSL certificates
+            timeout: Request timeout in seconds
+        """
+        super().__init__()
+        self.hostname = hostname
+        self.apikey = apikey
+        self.protocol = protocol
+        self.port = port
+        self.verify = verify
+        self.timeout = timeout
+
+    def _get_url(self, command):
+        """
+        Get the full URL for a specific API command.
+
+        Args:
+            command (str): The API command to construct the URL for.
+
+        Returns:
+            str: The full URL for the API command.
+        """
+        return f"{self.protocol}://{self.hostname}:{self.port}/{command}"
+
+    def _get_payload(
+        self,
+        op: Optional[str] = None,
+        path: Union[List[str], List[List[str]]] = None,
+        file: Optional[str] = None,
+        url: Optional[str] = None,
+        name: Optional[str] = None,
+    ) -> Dict[str, Any]:
+        """
+        Generates API request payload based on specified operations and parameters.
+
+        Parameters:
+            op (str, optional): Operation to perform (e.g., 'set', 'delete')
+            path (Union[List[str], List[List[str]]], optional):
+                Configuration path(s) for the API. Can be:
+                - Single path as string list
+                - Multiple paths as list of string lists
+            file (str, optional): File path for upload
+            url (str, optional): External resource URL
+            name (str, optional): Resource name
+
+        Returns:
+            Dict: Formatted API payload containing:
+                - data: JSON-serialized operations
+                - key: API key
+
+        Raises:
+            ValueError: If required parameters are missing or invalid
+        """
+
+        def _create_operations() -> Union[List[Dict], Dict]:
+            """Creates operation structure based on parameters."""
+            if not op:
+                if not all(isinstance(p, dict) for p in path):
+                    raise ValueError(
+                        "Path must contain dictionaries when no operation is specified"
+                    )
+                return path
+
+            normalized_paths = path or []
+            is_multiple = (
+                isinstance(normalized_paths[0], list) if normalized_paths else False
+            )
+
+            if is_multiple:
+                return [{"op": op, "path": p} for p in normalized_paths]
+            return {"op": op, "path": normalized_paths}
+
+        def _add_optional_params(
+            data: Union[List[Dict], Dict], params: Dict[str, str]
+        ) -> Union[List[Dict], Dict]:
+            """Adds optional parameters to operation structure."""
+            if isinstance(data, list):
+                return [{**item, **params} for item in data]
+            return {**data, **params}
+
+        # Initial validation
+        if not op and not path:
+            raise ValueError(
+                "Must provide either 'op' or pre-formatted operations in 'path'"
+            )
+
+        operations = _create_operations()
+        optional_params = {
+            k: v for k, v in zip(["file", "url", "name"], [file, url, name]) if v
+        }
+
+        if optional_params:
+            operations = _add_optional_params(operations, optional_params)
+
+        return {"data": json.dumps(operations), "key": self.apikey}
+
+    def _api_request(
+        self,
+        command: str,
+        op: Optional[str] = None,
+        path: Optional[List[str]] = None,
+        method: str = "POST",
+        file: Optional[str] = None,
+        resource_url: Optional[str] = None,
+        name: Optional[str] = None,
+    ):
+        """
+        Executes an API request with proper error handling and security measures.
+
+        Parameters:
+            command (str): API endpoint command to execute
+            op (str, optional): Operation type (e.g., 'create', 'update', 'delete')
+            path (List[str], optional): Hierarchical path for resource location
+            method (str): HTTP method (GET/POST/PUT/DELETE). Default: POST
+            file (str, optional): Local file path for file uploads
+            resource_url (str, optional): External resource URL reference
+            name (str, optional): Resource identifier name
+
+        Returns:
+            ApiResponse: Structured response containing:
+                - status: HTTP status code
+                - request: Sanitized request payload
+                - result: Parsed response data
+                - error: Error message if applicable
+
+        Raises:
+            ConnectionError: Network communication failures
+            Timeout: Server response timeout
+            ValueError: Invalid parameter combinations
+        """
+
+        def _prepare_request() -> Dict[str, Any]:
+            """Constructs request components with validation."""
+            if not command:
+                raise ValueError("API command is required")
+            return {
+                "url": self._get_url(command),
+                "method": method,
+                "verify": self.verify,
+                "timeout": self.timeout,
+                "payload": self._get_payload(
+                    op, path=path, file=file, url=resource_url, name=name
+                ),
+                "headers": {},
+            }
+
+        # Initialize mutable defaults safely
+        path = path or []
+
+        # Request execution flow
+        request_components = _prepare_request()
+        response = self._execute_request(**request_components)
+        status, result, error = self._validate_response(response)
+
+        # Sanitize sensitive data before returning
+        sanitized_payload = request_components["payload"].copy()
+        sanitized_payload.pop("key", None)
+
+        return ApiResponse(
+            status=status, request=sanitized_payload, result=result, error=error
+        )
+
+    @classmethod
+    def _execute_request(
+        cls,
+        url: str,
+        method: str,
+        verify: bool,
+        timeout: int,
+        payload: Dict,
+        headers: Dict,
+    ) -> requests.Response:
+        """Sends HTTP request with error handling."""
+        try:
+            return requests.request(
+                method=method.upper(),
+                url=url,
+                verify=verify,
+                data=payload,
+                timeout=timeout,
+                headers=headers,
+            )
+        except Timeout:
+            raise Timeout(f"Request timed out after {timeout} seconds")
+        except RequestException as e:
+            raise ConnectionError(f"Network error: {str(e)}")
+
+    @classmethod
+    def _validate_response(
+        cls, resp: Response
+    ) -> Tuple[Optional[int], Dict[str, Any], Union[str, bool]]:
+        """
+        Validates and processes API responses with comprehensive error handling.
+
+        Parameters:
+            resp (Response): HTTP response object from requests library
+
+        Returns:
+            Tuple containing:
+            - status (int | None): HTTP status code
+            - result (dict): Parsed successful response data
+            - error (str | bool): Error message (False indicates success)
+
+        Raises:
+            ValueError: For invalid response structures
+            RuntimeError: For unexpected parsing failures
+
+        Processing Flow:
+            1. HTTP Status Code Validation
+            2. Response Body Parsing
+            3. API Success/Failure Flag Check
+            4. Error Message Extraction
+            5. Fallback Error Handling
+        """
+        status: Optional[int] = None
+        result: Dict[str, Any] = {}
+        error: Union[str, bool] = False
+
+        def _validate_schema(response_json: Dict[str, Any]) -> None:
+            """Validates response structure against API contract."""
+            required_keys = {"success", "data", "error"}
+            if isinstance(response_json, dict) and not required_keys.issubset(response_json.keys()):
+                missing = required_keys - response_json.keys()
+                raise ValueError(f"Invalid response structure. Missing keys: {missing}")
+
+        try:
+            # Validate HTTP status code
+            resp.raise_for_status()
+            status = resp.status_code
+
+            # Parse and validate JSON structure
+            resp_decoded = resp.json()
+            _validate_schema(resp_decoded)
+
+            # Process API business logic
+            if resp_decoded["success"]:
+                result = resp_decoded["data"]
+            else:
+                error = f"API Error {status}: {resp_decoded['error']}"
+
+        except JSONDecodeError as exc:
+            error = f"Invalid response format: {str(exc)}"
+            status = resp.status_code if resp is not None and isinstance(resp, Response) else 500
+
+
+        except HTTPError as exc:
+            response = exc.response
+            status = response.status_code if response is not None and isinstance(response, Response) else 500
+            error = f"HTTP Error {status}: {response.text[:200] if response else 'Unknown error'}"
+
+        except ValueError as exc:
+            error = f"Validation Error: {str(exc)}"
+
+        except Exception as exc:
+            error = f"Unexpected error: {str(exc)}"
+            status = 500
+
+        return status, result, error