diff options
Diffstat (limited to 'pyvyos')
| -rw-r--r-- | pyvyos/__init__.py | 12 | ||||
| -rw-r--r-- | pyvyos/core/__init__.py | 7 | ||||
| -rw-r--r-- | pyvyos/core/device.py | 276 | ||||
| -rw-r--r-- | pyvyos/core/rest_client.py | 381 | ||||
| -rw-r--r-- | pyvyos/device.py | 282 | ||||
| -rw-r--r-- | pyvyos/exceptions.py | 33 | ||||
| -rw-r--r-- | pyvyos/rest.py | 322 | ||||
| -rw-r--r-- | pyvyos/specs/__init__.py | 12 | ||||
| -rw-r--r-- | pyvyos/specs/commands/__init__.py | 13 | ||||
| -rw-r--r-- | pyvyos/specs/commands/config_file.py | 36 | ||||
| -rw-r--r-- | pyvyos/specs/commands/configure.py | 38 | ||||
| -rw-r--r-- | pyvyos/specs/commands/generate.py | 20 | ||||
| -rw-r--r-- | pyvyos/specs/commands/image.py | 33 | ||||
| -rw-r--r-- | pyvyos/specs/commands/reset.py | 20 | ||||
| -rw-r--r-- | pyvyos/specs/commands/retrieve.py | 27 | ||||
| -rw-r--r-- | pyvyos/specs/commands/show.py | 20 | ||||
| -rw-r--r-- | pyvyos/specs/commands/system.py | 27 | ||||
| -rw-r--r-- | pyvyos/specs/models.py | 31 | ||||
| -rw-r--r-- | pyvyos/utils/__init__.py | 8 | ||||
| -rw-r--r-- | pyvyos/utils/ids.py | 14 | ||||
| -rw-r--r-- | pyvyos/utils/json.py | 47 | ||||
| -rw-r--r-- | pyvyos/utils/paths.py | 30 |
22 files changed, 1102 insertions, 587 deletions
diff --git a/pyvyos/__init__.py b/pyvyos/__init__.py index 44bd563..b0f867f 100644 --- a/pyvyos/__init__.py +++ b/pyvyos/__init__.py @@ -1,2 +1,10 @@ -from .device import VyDevice -from .device import ApiResponse +""" +PyVyOS - Python SDK for VyOS REST API + +Public API exports maintain backward compatibility. +Internal structure has been refactored to pyvyos.core for better organization. +""" +from .core.device import VyDevice +from .core.rest_client import ApiResponse + +__all__ = ["VyDevice", "ApiResponse"] diff --git a/pyvyos/core/__init__.py b/pyvyos/core/__init__.py new file mode 100644 index 0000000..f4b831c --- /dev/null +++ b/pyvyos/core/__init__.py @@ -0,0 +1,7 @@ +"""Core modules for PyVyOS - internal refactored structure.""" + +from .rest_client import ApiResponse, RestClient +from .device import VyDevice + +__all__ = ["ApiResponse", "RestClient", "VyDevice"] + diff --git a/pyvyos/core/device.py b/pyvyos/core/device.py new file mode 100644 index 0000000..6dc21e9 --- /dev/null +++ b/pyvyos/core/device.py @@ -0,0 +1,276 @@ +import warnings +from typing import List, Literal + +from .rest_client import ApiResponse, RestClient + + +class VyDevice(RestClient): + """ + Represents a device for interacting with the VyOS API. + + 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). + + Attributes: + hostname (str): The hostname or IP address of the VyOS device. + apikey (str): The API key for authentication. + protocol (str): The protocol used for communication. + port (int): The port used for communication. + verify (bool): Whether SSL certificate verification is enabled. + timeout (int): The request timeout in seconds. + + Methods: + _get_url(command): Get the full URL for a given API command. + _get_payload(op, path=[], file=None, url=None, name=None): Generate the API request payload. + _api_request(command, op, path=[], method='POST', file=None, url=None, name=None): Make an API request. + retrieve_show_config(path=[]): Retrieve and show the device configuration. + retrieve_return_values(path=[]): Retrieve and return specific configuration values. + reset(path=[]): Reset a specific configuration element. + image_add(url=None, file=None, path=[]): Add an image from a URL or file. + 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 + 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. + config_file_load(file=None): Load the configuration from a file. + reboot(path=["now"]): Reboot the device. + poweroff(path=["now"]): Power off the device. + """ + + 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. + + Args: + path (list, optional): The path elements for the configuration retrieval (default is an empty list). + + 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: List = None): + """ + Retrieve and return specific configuration values. + + Args: + path (list, optional): The path elements for the configuration retrieval (default is an empty list). + + Returns: + ApiResponse: An ApiResponse object representing the API response. + """ + return self._api_request( + command="retrieve", op="returnValues", path=path, method="POST" + ) + + def reset(self, path: List = None): + """ + Reset a specific configuration element. + + Args: + path (list, optional): The path elements for the configuration reset (default is an empty list). + + Returns: + ApiResponse: An ApiResponse object representing the API response. + """ + return self._api_request(command="reset", op="reset", path=path, method="POST") + + def image_add(self, url=None, file=None, path=[]): + """ + Add an image from a URL or file. + + Args: + url (str, optional): The URL of the image to add (default is None). + file (str, optional): The path to the local image file to add (default is None). + path (list, optional): The path elements for the image addition (default is an empty list). + + Returns: + ApiResponse: An ApiResponse object representing the API response. + """ + return self._api_request(command="image", op="add", resource_url=url, method="POST") + + def image_delete(self, name, url=None, file=None, path=[]): + """ + Delete a specific image. + + Args: + name (str): The name of the image to delete. + url (str, optional): The URL of the image to delete (default is None). + file (str, optional): The path to the local image file to delete (default is None). + path (list, optional): The path elements for the image deletion (default is an empty list). + + Returns: + ApiResponse: An ApiResponse object representing the API response. + """ + return self._api_request(command="image", op="delete", name=name, method="POST") + + def show(self, path: List = None): + """ + Show configuration information. + + Args: + path (list, optional): The path elements for the configuration display (default is an empty list). + + Returns: + ApiResponse: An ApiResponse object representing the API response. + """ + return self._api_request(command="show", op="show", path=path, method="POST") + + def generate(self, path: List = None): + """ + Generate configuration based on the given path. + + Args: + path (list, optional): The path elements for configuration generation (default is an empty list). + + Returns: + ApiResponse: An ApiResponse object representing the API response. + """ + return self._api_request( + command="generate", op="generate", path=path, method="POST" + ) + + def configure_set(self, path: List = None): + """ + Set configuration based on the given path. + + Args: + path (list, optional): The path elements for configuration setting (default is an empty list). + + Returns: + ApiResponse: An ApiResponse object representing the API response. + """ + return self._api_request( + command="configure", op="set", path=path, method="POST" + ) + + def configure_delete(self, path: List = None): + """ + Delete configuration based on the given path. + + Args: + path (list, optional): The path elements for configuration deletion (default is an empty list). + + Returns: + ApiResponse: An ApiResponse object representing the API response. + """ + 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): + """ + Save the configuration to a file. + + Args: + file (str, optional): The path to the file where the configuration will be saved (default is None). + + Returns: + ApiResponse: An ApiResponse object representing the API response. + """ + return self._api_request( + command="config-file", op="save", path=[], file=file, method="POST" + ) + + def config_file_load(self, file=None): + """ + Load the configuration from a file. + + Args: + file (str, optional): The path to the file from which the configuration will be loaded (default is None). + + Returns: + ApiResponse: An ApiResponse object representing the API response. + """ + return self._api_request( + command="config-file", op="load", path=[], file=file, method="POST" + ) + + def reboot(self, path: List = None): + """ + Reboot the device. + + Args: + path (list, optional): The path elements for the reboot operation (default is ["now"]). + + Returns: + ApiResponse: An ApiResponse object representing the API response. + """ + 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. + + Args: + path (list, optional): The path elements for the power off operation (default is ["now"]). + + Returns: + ApiResponse: An ApiResponse object representing the API response. + """ + if path is None: + path = ["now"] + + return self._api_request( + command="poweroff", op="poweroff", path=path, method="POST" + ) + diff --git a/pyvyos/core/rest_client.py b/pyvyos/core/rest_client.py new file mode 100644 index 0000000..eb77d9b --- /dev/null +++ b/pyvyos/core/rest_client.py @@ -0,0 +1,381 @@ +import json +import logging +import os +import time +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, +) + +logger = logging.getLogger("pyvyos") + +# Enable DEBUG logs if PYVYOS_DEBUG=1 +if os.getenv("PYVYOS_DEBUG") == "1": + logging.basicConfig(level=logging.DEBUG) + + +@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, + include_empty_path: bool = True, + ) -> 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 + include_empty_path (bool, optional): Whether to include empty path in payload. + Default True. Set to False to omit path when empty (except for config-file). + + 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 [] + + # Skip path if empty and include_empty_path is False + if not normalized_paths and not include_empty_path: + if not op: + raise ValueError( + "Must provide either 'op' or pre-formatted operations in 'path'" + ) + return {"op": op} + + 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") + + # Special case: config-file requires path to be present (even if empty) + # Other commands can omit path when empty + include_empty_path = (command == "config-file") + + 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, + include_empty_path=include_empty_path + ), + "headers": {}, + } + + # Initialize mutable defaults safely + path = path or [] + + # Generate request ID for tracing + try: + from ..utils.ids import request_id + req_id = request_id() + except ImportError: + req_id = None + + # Log request start (DEBUG level) + logger.debug( + "Request start", + extra={ + "request_id": req_id, + "command": command, + "op": op, + "has_path": bool(path), + } + ) + + # Request execution flow + start_time = time.time() + request_components = _prepare_request() + response = self._execute_request(**request_components) + elapsed_ms = int((time.time() - start_time) * 1000) + status, result, error = self._validate_response(response) + + # Log response (INFO level) + logger.info( + "Request completed", + extra={ + "request_id": req_id, + "command": command, + "op": op, + "status": status, + "success": not error, + "elapsed_ms": elapsed_ms, + } + ) + + # Sanitize sensitive data before returning (use utils if available) + try: + from ..utils.json import redact_key + sanitized_payload = redact_key(request_components["payload"]) + except ImportError: + 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 + diff --git a/pyvyos/device.py b/pyvyos/device.py index 246a913..89d880f 100644 --- a/pyvyos/device.py +++ b/pyvyos/device.py @@ -1,275 +1,13 @@ -import warnings -from typing import List, Literal +""" +Device module - backward compatibility shim. -from .rest import ApiResponse, RestClient +This module maintains backward compatibility by re-exporting classes +from the refactored core module structure. +Public API imports (from pyvyos import ...) continue to work unchanged. +Direct imports from this module may show deprecation warnings in future versions. +""" +from .core.device import VyDevice +from .core.rest_client import ApiResponse -class VyDevice(RestClient): - """ - Represents a device for interacting with the VyOS API. - - 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). - - Attributes: - hostname (str): The hostname or IP address of the VyOS device. - apikey (str): The API key for authentication. - protocol (str): The protocol used for communication. - port (int): The port used for communication. - verify (bool): Whether SSL certificate verification is enabled. - timeout (int): The request timeout in seconds. - - Methods: - _get_url(command): Get the full URL for a given API command. - _get_payload(op, path=[], file=None, url=None, name=None): Generate the API request payload. - _api_request(command, op, path=[], method='POST', file=None, url=None, name=None): Make an API request. - retrieve_show_config(path=[]): Retrieve and show the device configuration. - retrieve_return_values(path=[]): Retrieve and return specific configuration values. - reset(path=[]): Reset a specific configuration element. - image_add(url=None, file=None, path=[]): Add an image from a URL or file. - 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 - 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. - config_file_load(file=None): Load the configuration from a file. - reboot(path=["now"]): Reboot the device. - poweroff(path=["now"]): Power off the device. - """ - - 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. - - Args: - path (list, optional): The path elements for the configuration retrieval (default is an empty list). - - 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: List = None): - """ - Retrieve and return specific configuration values. - - Args: - path (list, optional): The path elements for the configuration retrieval (default is an empty list). - - Returns: - ApiResponse: An ApiResponse object representing the API response. - """ - return self._api_request( - command="retrieve", op="returnValues", path=path, method="POST" - ) - - def reset(self, path: List = None): - """ - Reset a specific configuration element. - - Args: - path (list, optional): The path elements for the configuration reset (default is an empty list). - - Returns: - ApiResponse: An ApiResponse object representing the API response. - """ - return self._api_request(command="reset", op="reset", path=path, method="POST") - - def image_add(self, url=None, file=None, path=[]): - """ - Add an image from a URL or file. - - Args: - url (str, optional): The URL of the image to add (default is None). - file (str, optional): The path to the local image file to add (default is None). - path (list, optional): The path elements for the image addition (default is an empty list). - - Returns: - ApiResponse: An ApiResponse object representing the API response. - """ - return self._api_request(command="image", op="add", url=url, method="POST") - - def image_delete(self, name, url=None, file=None, path=[]): - """ - Delete a specific image. - - Args: - name (str): The name of the image to delete. - url (str, optional): The URL of the image to delete (default is None). - file (str, optional): The path to the local image file to delete (default is None). - path (list, optional): The path elements for the image deletion (default is an empty list). - - Returns: - ApiResponse: An ApiResponse object representing the API response. - """ - return self._api_request(command="image", op="delete", name=name, method="POST") - - def show(self, path: List = None): - """ - Show configuration information. - - Args: - path (list, optional): The path elements for the configuration display (default is an empty list). - - Returns: - ApiResponse: An ApiResponse object representing the API response. - """ - return self._api_request(command="show", op="show", path=path, method="POST") - - def generate(self, path: List = None): - """ - Generate configuration based on the given path. - - Args: - path (list, optional): The path elements for configuration generation (default is an empty list). - - Returns: - ApiResponse: An ApiResponse object representing the API response. - """ - return self._api_request( - command="generate", op="generate", path=path, method="POST" - ) - - def configure_set(self, path: List = None): - """ - Set configuration based on the given path. - - Args: - path (list, optional): The path elements for configuration setting (default is an empty list). - - Returns: - ApiResponse: An ApiResponse object representing the API response. - """ - return self._api_request( - command="configure", op="set", path=path, method="POST" - ) - - def configure_delete(self, path: List = None): - """ - Delete configuration based on the given path. - - Args: - path (list, optional): The path elements for configuration deletion (default is an empty list). - - Returns: - ApiResponse: An ApiResponse object representing the API response. - """ - 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): - """ - Save the configuration to a file. - - Args: - file (str, optional): The path to the file where the configuration will be saved (default is None). - - Returns: - ApiResponse: An ApiResponse object representing the API response. - """ - return self._api_request( - command="config-file", op="save", file=file, method="POST" - ) - - def config_file_load(self, file=None): - """ - Load the configuration from a file. - - Args: - file (str, optional): The path to the file from which the configuration will be loaded (default is None). - - Returns: - ApiResponse: An ApiResponse object representing the API response. - """ - return self._api_request( - command="config-file", op="load", file=file, method="POST" - ) - - def reboot(self, path: List = None): - """ - Reboot the device. - - Args: - path (list, optional): The path elements for the reboot operation (default is ["now"]). - - Returns: - ApiResponse: An ApiResponse object representing the API response. - """ - 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. - - Args: - path (list, optional): The path elements for the power off operation (default is ["now"]). - - Returns: - ApiResponse: An ApiResponse object representing the API response. - """ - if path is None: - path = ["now"] - - return self._api_request( - command="poweroff", op="poweroff", path=path, method="POST" - ) +__all__ = ["VyDevice", "ApiResponse"] diff --git a/pyvyos/exceptions.py b/pyvyos/exceptions.py new file mode 100644 index 0000000..5a79733 --- /dev/null +++ b/pyvyos/exceptions.py @@ -0,0 +1,33 @@ +"""PyVyOS exception hierarchy for typed error handling.""" + + +class SDKError(Exception): + """Base exception for all PyVyOS SDK errors.""" + pass + + +class HttpError(SDKError): + """HTTP-level errors (network, timeout, status codes).""" + + def __init__(self, status: int, message: str): + self.status = status + self.message = message + super().__init__(f"HTTP Error {status}: {message}") + + +class ApiError(SDKError): + """API-level errors (when VyOS API returns success=False).""" + + def __init__(self, message: str, details: dict = None): + self.message = message + self.details = details + super().__init__(message) + + +class ValidationError(SDKError): + """Client-side validation errors (invalid parameters).""" + + def __init__(self, message: str): + self.message = message + super().__init__(message) + diff --git a/pyvyos/rest.py b/pyvyos/rest.py index 04c786e..3a48486 100644 --- a/pyvyos/rest.py +++ b/pyvyos/rest.py @@ -1,316 +1,12 @@ -import json -from abc import ABC -from dataclasses import dataclass -from typing import Tuple, List, Union, Dict, Any, Optional +""" +REST client module - backward compatibility shim. -import requests -from requests import Response -from requests.exceptions import ( - HTTPError, - ConnectionError, - Timeout, - RequestException, - JSONDecodeError, -) +This module maintains backward compatibility by re-exporting classes +from the refactored core module structure. +Public API imports (from pyvyos import ...) continue to work unchanged. +Direct imports from this module may show deprecation warnings in future versions. +""" +from .core.rest_client import ApiResponse, RestClient -@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 +__all__ = ["ApiResponse", "RestClient"] diff --git a/pyvyos/specs/__init__.py b/pyvyos/specs/__init__.py new file mode 100644 index 0000000..d12c764 --- /dev/null +++ b/pyvyos/specs/__init__.py @@ -0,0 +1,12 @@ +"""PyVyOS Pydantic specs for optional request/response validation.""" + +# Feature detection: gracefully degrade if pydantic not available +try: + from pydantic import BaseModel + PYDANTIC_AVAILABLE = True +except ImportError: + PYDANTIC_AVAILABLE = False + BaseModel = None # type: ignore + +__all__ = ["PYDANTIC_AVAILABLE"] + diff --git a/pyvyos/specs/commands/__init__.py b/pyvyos/specs/commands/__init__.py new file mode 100644 index 0000000..1d8e8b3 --- /dev/null +++ b/pyvyos/specs/commands/__init__.py @@ -0,0 +1,13 @@ +"""Pydantic models for individual VyOS API commands.""" + +from .config_file import ConfigFileSaveRequest, ConfigFileLoadRequest +from .configure import ConfigureSetRequest, ConfigureDeleteRequest, ConfigureMultipleOpRequest + +__all__ = [ + "ConfigFileSaveRequest", + "ConfigFileLoadRequest", + "ConfigureSetRequest", + "ConfigureDeleteRequest", + "ConfigureMultipleOpRequest", +] + diff --git a/pyvyos/specs/commands/config_file.py b/pyvyos/specs/commands/config_file.py new file mode 100644 index 0000000..911b094 --- /dev/null +++ b/pyvyos/specs/commands/config_file.py @@ -0,0 +1,36 @@ +"""Pydantic models for config-file operations.""" + +from typing import List, Literal, Optional + +try: + from pydantic import BaseModel, Field + from ..models import ApiRequest +except ImportError: + BaseModel = None # type: ignore + Field = None # type: ignore + ApiRequest = None # type: ignore + + +if BaseModel: + + class ConfigFileSaveRequest(ApiRequest): + """Request model for config-file save operation.""" + + op: Literal["save"] = "save" + path: List[str] = Field( + default_factory=list, + description="Required by VyOS API, typically empty array" + ) + file: Optional[str] = None + + + class ConfigFileLoadRequest(ApiRequest): + """Request model for config-file load operation.""" + + op: Literal["load"] = "load" + path: List[str] = Field( + default_factory=list, + description="Required by VyOS API, typically empty array" + ) + file: str # Required for load + diff --git a/pyvyos/specs/commands/configure.py b/pyvyos/specs/commands/configure.py new file mode 100644 index 0000000..d80aa3f --- /dev/null +++ b/pyvyos/specs/commands/configure.py @@ -0,0 +1,38 @@ +"""Pydantic models for configure operations.""" + +from typing import Dict, List, Literal, Union, Any + +try: + from pydantic import BaseModel, Field + from ..models import ApiRequest +except ImportError: + BaseModel = None # type: ignore + Field = None # type: ignore + ApiRequest = None # type: ignore + + +if BaseModel: + + class ConfigureSetRequest(ApiRequest): + """Request model for configure set operation.""" + + op: Literal["set"] = "set" + path: Union[List[str], List[List[str]]] # Required + + + class ConfigureDeleteRequest(ApiRequest): + """Request model for configure delete operation.""" + + op: Literal["delete"] = "delete" + path: List[str] # Required + + + class ConfigureMultipleOpRequest(BaseModel): + """Request model for configure multiple operations.""" + + op: Literal["set", "delete"] + path: Union[List[str], List[List[str]]] + + class Config: + extra = "allow" + diff --git a/pyvyos/specs/commands/generate.py b/pyvyos/specs/commands/generate.py new file mode 100644 index 0000000..40aa2a4 --- /dev/null +++ b/pyvyos/specs/commands/generate.py @@ -0,0 +1,20 @@ +"""Pydantic models for generate operations.""" + +from typing import List, Literal, Optional + +try: + from pydantic import BaseModel + from ..models import ApiRequest +except ImportError: + BaseModel = None # type: ignore + ApiRequest = None # type: ignore + + +if BaseModel: + + class GenerateRequest(ApiRequest): + """Request model for generate operation.""" + + op: Literal["generate"] = "generate" + path: Optional[List[str]] = None + diff --git a/pyvyos/specs/commands/image.py b/pyvyos/specs/commands/image.py new file mode 100644 index 0000000..973bae4 --- /dev/null +++ b/pyvyos/specs/commands/image.py @@ -0,0 +1,33 @@ +"""Pydantic models for image operations.""" + +from typing import List, Literal, Optional + +try: + from pydantic import BaseModel + from ..models import ApiRequest +except ImportError: + BaseModel = None # type: ignore + ApiRequest = None # type: ignore + + +if BaseModel: + + class ImageAddRequest(ApiRequest): + """Request model for image add operation.""" + + op: Literal["add"] = "add" + path: Optional[List[str]] = None + url: Optional[str] = None + file: Optional[str] = None + name: Optional[str] = None + + + class ImageDeleteRequest(ApiRequest): + """Request model for image delete operation.""" + + op: Literal["delete"] = "delete" + path: Optional[List[str]] = None + url: Optional[str] = None + file: Optional[str] = None + name: Optional[str] = None + diff --git a/pyvyos/specs/commands/reset.py b/pyvyos/specs/commands/reset.py new file mode 100644 index 0000000..ac02bfe --- /dev/null +++ b/pyvyos/specs/commands/reset.py @@ -0,0 +1,20 @@ +"""Pydantic models for reset operations.""" + +from typing import List, Literal, Optional + +try: + from pydantic import BaseModel + from ..models import ApiRequest +except ImportError: + BaseModel = None # type: ignore + ApiRequest = None # type: ignore + + +if BaseModel: + + class ResetRequest(ApiRequest): + """Request model for reset operation.""" + + op: Literal["reset"] = "reset" + path: Optional[List[str]] = None + diff --git a/pyvyos/specs/commands/retrieve.py b/pyvyos/specs/commands/retrieve.py new file mode 100644 index 0000000..65efbbf --- /dev/null +++ b/pyvyos/specs/commands/retrieve.py @@ -0,0 +1,27 @@ +"""Pydantic models for retrieve operations.""" + +from typing import List, Literal, Optional + +try: + from pydantic import BaseModel + from ..models import ApiRequest +except ImportError: + BaseModel = None # type: ignore + ApiRequest = None # type: ignore + + +if BaseModel: + + class RetrieveShowConfigRequest(ApiRequest): + """Request model for retrieve showConfig operation.""" + + op: Literal["showConfig"] = "showConfig" + path: Optional[List[str]] = None + + + class RetrieveReturnValuesRequest(ApiRequest): + """Request model for retrieve returnValues operation.""" + + op: Literal["returnValues"] = "returnValues" + path: Optional[List[str]] = None + diff --git a/pyvyos/specs/commands/show.py b/pyvyos/specs/commands/show.py new file mode 100644 index 0000000..2fb2b97 --- /dev/null +++ b/pyvyos/specs/commands/show.py @@ -0,0 +1,20 @@ +"""Pydantic models for show operations.""" + +from typing import List, Literal, Optional + +try: + from pydantic import BaseModel + from ..models import ApiRequest +except ImportError: + BaseModel = None # type: ignore + ApiRequest = None # type: ignore + + +if BaseModel: + + class ShowRequest(ApiRequest): + """Request model for show operation.""" + + op: Literal["show"] = "show" + path: Optional[List[str]] = None + diff --git a/pyvyos/specs/commands/system.py b/pyvyos/specs/commands/system.py new file mode 100644 index 0000000..9f5e5c9 --- /dev/null +++ b/pyvyos/specs/commands/system.py @@ -0,0 +1,27 @@ +"""Pydantic models for system operations (reboot, poweroff).""" + +from typing import List, Literal + +try: + from pydantic import BaseModel + from ..models import ApiRequest +except ImportError: + BaseModel = None # type: ignore + ApiRequest = None # type: ignore + + +if BaseModel: + + class RebootRequest(ApiRequest): + """Request model for reboot operation.""" + + op: Literal["reboot"] = "reboot" + path: List[str] = ["now"] + + + class PoweroffRequest(ApiRequest): + """Request model for poweroff operation.""" + + op: Literal["poweroff"] = "poweroff" + path: List[str] = ["now"] + diff --git a/pyvyos/specs/models.py b/pyvyos/specs/models.py new file mode 100644 index 0000000..a57ac82 --- /dev/null +++ b/pyvyos/specs/models.py @@ -0,0 +1,31 @@ +"""Base Pydantic models for PyVyOS API requests and responses.""" + +from typing import Any, Dict, List, Optional, Union + +if False: # type checking only + from pydantic import BaseModel +else: + try: + from pydantic import BaseModel, Field + except ImportError: + BaseModel = None # type: ignore + Field = None # type: ignore + + +class ApiRequest(BaseModel): + """Base request model for VyOS API operations.""" + + op: str + path: Optional[Union[List[str], List[List[str]], List[Dict[str, Any]]]] = None + + class Config: + extra = "allow" # Allow additional fields (file, url, name, etc.) + + +class ApiResponse(BaseModel): + """Base response model for VyOS API responses.""" + + success: bool + data: Union[Dict[str, Any], List[Any], str, None] + error: Optional[str] = None + diff --git a/pyvyos/utils/__init__.py b/pyvyos/utils/__init__.py new file mode 100644 index 0000000..adff044 --- /dev/null +++ b/pyvyos/utils/__init__.py @@ -0,0 +1,8 @@ +"""Utility functions for PyVyOS.""" + +from .json import redact_key, safe_dumps +from .ids import request_id +from .paths import build_path + +__all__ = ["redact_key", "safe_dumps", "request_id", "build_path"] + diff --git a/pyvyos/utils/ids.py b/pyvyos/utils/ids.py new file mode 100644 index 0000000..5807ddf --- /dev/null +++ b/pyvyos/utils/ids.py @@ -0,0 +1,14 @@ +"""ID generation utilities.""" + +import uuid + + +def request_id() -> str: + """ + Generate a unique request ID for tracing. + + Returns: + UUID4 string formatted for logging + """ + return str(uuid.uuid4()) + diff --git a/pyvyos/utils/json.py b/pyvyos/utils/json.py new file mode 100644 index 0000000..6d9a7c0 --- /dev/null +++ b/pyvyos/utils/json.py @@ -0,0 +1,47 @@ +"""JSON utilities with security (redaction) support.""" + +import json +from typing import Any, Dict, List + + +def redact_key(data: Dict[str, Any], keys: List[str] = None) -> Dict[str, Any]: + """ + Redact sensitive keys from a dictionary. + + Args: + data: Dictionary to redact + keys: List of keys to redact (default: ["key", "apikey", "password"]) + + Returns: + Copy of data with specified keys redacted as "***REDACTED***" + """ + if keys is None: + keys = ["key", "apikey", "password"] + + if not isinstance(data, dict): + return data + + redacted = data.copy() + for key in keys: + if key in redacted: + redacted[key] = "***REDACTED***" + + return redacted + + +def safe_dumps(obj: Any, redact_keys: List[str] = None) -> str: + """ + Safely serialize object to JSON with key redaction. + + Args: + obj: Object to serialize + redact_keys: Keys to redact if obj is a dict + + Returns: + JSON string with sensitive data redacted + """ + if isinstance(obj, dict): + obj = redact_key(obj, redact_keys) + + return json.dumps(obj, default=str) + diff --git a/pyvyos/utils/paths.py b/pyvyos/utils/paths.py new file mode 100644 index 0000000..b591b09 --- /dev/null +++ b/pyvyos/utils/paths.py @@ -0,0 +1,30 @@ +"""Path building utilities for VyOS configuration paths.""" + +from typing import List, Union + + +def build_path(*segments: Union[str, List[str]]) -> List[str]: + """ + Build a configuration path from segments. + + Args: + *segments: Path segments (strings or lists of strings) + + Returns: + Flattened list of path segments + + Examples: + build_path("interfaces", "ethernet", "eth0") + -> ["interfaces", "ethernet", "eth0"] + + build_path(["interfaces", "ethernet"], "address", "192.168.1.1/24") + -> ["interfaces", "ethernet", "address", "192.168.1.1/24"] + """ + result = [] + for segment in segments: + if isinstance(segment, list): + result.extend(segment) + else: + result.append(str(segment)) + return result + |