# Copyright (C) 2024 Cetmix OÜ # License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl). import logging import time import warnings from odoo import _, api, models from odoo.exceptions import ValidationError from . import tools from .constants import NOT_FOUND, SSH_CONNECTION_ERROR _logger = logging.getLogger(__name__) class CetmixTower(models.AbstractModel): """Generic model used to simplify Odoo automation. Used to keep main integration function in a single place. """ _name = "cetmix.tower" _description = "Cetmix Tower Odoo Automation" @api.model def server_create_from_template(self, template_reference, server_name, **kwargs): """ THIS METHOD IS DEPRECATED. USE THE 'cx.tower.server.template' MODEL DIRECTLY. """ _logger.warning( "server_create_from_template: This method is deprecated " "and will be removed in the future. " "Use the 'cx.tower.server.template' model directly instead." ) return self.env["cx.tower.server.template"].create_server_from_template( template_reference=template_reference, server_name=server_name, **kwargs ) @api.model def server_run_command( self, server_reference, command_reference, get_result=True, **variable_values ): """ THIS METHOD IS DEPRECATED. USE THE 'cx.tower.server' MODEL DIRECTLY. """ _logger.warning( "server_run_command: This method is deprecated and " "will be removed in the future. " "Use the 'cx.tower.server' model directly instead." ) server = self.env["cx.tower.server"].get_by_reference(server_reference) if not server: return {"exit_code": NOT_FOUND, "message": _("Server not found")} command = self.env["cx.tower.command"].get_by_reference(command_reference) if not command: return {"exit_code": NOT_FOUND, "message": _("Command not found")} # Will return command result if get_result is True # Otherwise will save to log and return None command_result = server.with_context(no_command_log=get_result).run_command( command, **{"variable_values": variable_values} if variable_values else {} ) # Return command result if get_result is True if command_result: status = command_result.get("status") response = command_result.get("response", "") error = command_result.get("error", "") return { "exit_code": status, "message": response or error, } def server_run_flight_plan( self, server_reference, flight_plan_reference, **variable_values ): """THIS METHOD IS DEPRECATED. USE THE 'cx.tower.server' MODEL DIRECTLY.""" _logger.warning( "server_run_flight_plan: This method is deprecated and " "will be removed in the future. " "Use the 'cx.tower.server' model directly instead." ) server = self.env["cx.tower.server"].get_by_reference(server_reference) if not server: # This is not the best way to handle this, but it's the only way to # avoid complex response handling return False flight_plan = self.env["cx.tower.plan"].get_by_reference(flight_plan_reference) if not flight_plan: # This is not the best way to handle this, but it's the only way to # avoid complex response handling return False return server.run_flight_plan( flight_plan, **{"variable_values": variable_values} if variable_values else {}, ) @api.model def server_set_variable_value(self, server_reference, variable_reference, value): """THIS METHOD IS DEPRECATED. USE THE 'cx.tower.server' MODEL DIRECTLY.""" _logger.warning( "server_set_variable_value: This method is deprecated and " "will be removed in the future. " "Use the 'cx.tower.server' model directly instead." ) server = self.env["cx.tower.server"].get_by_reference(server_reference) if not server: return {"exit_code": NOT_FOUND, "message": _("Server not found")} variable = self.env["cx.tower.variable"].get_by_reference(variable_reference) if not variable: return {"exit_code": NOT_FOUND, "message": _("Variable not found")} # Check if variable is already defined for the server variable_value_record = variable.value_ids.filtered( lambda v: v.server_id == server ) if variable_value_record: variable_value_record.value_char = value result = {"exit_code": 0, "message": _("Variable value updated")} else: self.env["cx.tower.variable.value"].create( { "variable_id": variable.id, "server_id": server.id, "value_char": value, } ) result = {"exit_code": 0, "message": _("Variable value created")} return result @api.model def server_get_variable_value( self, server_reference, variable_reference, check_global=True ): """THIS METHOD IS DEPRECATED. USE THE 'cx.tower.server' MODEL DIRECTLY.""" _logger.warning( "server_get_variable_value: This method is deprecated and " "will be removed in the future. " "Use the 'cx.tower.server' model directly instead." ) if not check_global: warnings.warn( "server_get_variable_value: 'check_global' is deprecated and " "will be removed in the future. " "Global values are always checked.", DeprecationWarning, stacklevel=2, ) # Get server by reference server = self.env["cx.tower.server"].get_by_reference(server_reference) if not server: _logger.warning( "server_get_variable_value: Server not found for reference '%s'", server_reference, ) return None return ( self.env["cx.tower.variable"] ._get_variable_values_by_references( variable_references=[variable_reference], server=server ) .get(variable_reference) ) @api.model def server_check_ssh_connection( self, server_reference, attempts=5, wait_time=10, try_command=True, try_file=True, ): """ Check if SSH connection to the server is available. This method uses the `test_ssh_connection` method of the 'cx.tower.server' model. It tries to connect to the server multiple times and is designed to be used in the Python commands or Odoo automated actions. Args: server_reference (Char): Server reference. attempts (int): Number of attempts to try the connection. Default is 5. wait_time (int): Wait time in seconds between connection attempts. Default is 10 seconds. try_command (bool): Try to execute a command. Default is True. try_file (bool): Try file operations. Default is True. Raises: ValidationError: If the provided server reference is invalid or the server cannot be found. Returns: dict: { "exit_code": int, 0 for success, error code for failure "message": str # Description of the result } """ server = self.env["cx.tower.server"].get_by_reference(server_reference) if not server: raise ValidationError(_("No server found for the provided reference.")) # Try connecting multiple times for attempt in range(1, attempts + 1): try: _logger.info( "Attempt %s of %s to connect to server %s", attempt, attempts, server_reference, ) result = server.test_ssh_connection( raise_on_error=True, return_notification=False, try_command=try_command, try_file=try_file, ) if result.get("status") == 0: return { "exit_code": 0, "message": _("Connection successful."), } if attempt == attempts: return { "exit_code": SSH_CONNECTION_ERROR, "message": _( "Failed to connect after %(attempts)s attempts. " "Error: %(err)s", attempts=attempts, err=result.get("error", ""), ), } except Exception as e: # pylint: disable=broad-except if attempt == attempts: return { "exit_code": SSH_CONNECTION_ERROR, "message": _("Failed to connect. Error: %(err)s", err=e), } time.sleep(wait_time) @api.model def server_validate_secret( self, secret_value, secret_reference, server_reference=None ): """ Validates the provided secret value against the actual secret. Accepts either a full inline reference (e.g. #!cxtower.secret.!#) or just a . Args: secret_value (Char): Value to validate secret_reference (Char): Reference code or inline reference server_reference (Char, optional): Reference code of the server Returns: Bool: True if the value matches the secret, False otherwise """ server = self.env["cx.tower.server"] if server_reference: server = server.get_by_reference(server_reference) # Try to extract reference from inline format using _extract_key_parts key_parts = self.env["cx.tower.key"]._extract_key_parts(secret_reference) if key_parts: # _extract_key_parts returns a tuple: (key_type, reference). # We only need the reference part here. secret_reference = key_parts[1] value = self.env["cx.tower.key"]._resolve_key_type_secret( secret_reference, server_id=server.id ) return value == secret_value @api.model def generate_random_id(self, sections=1, population=4, separator="-"): """ Helper method that allows to generate a random id with customizable sections and population. Such ids are more human readable and less likely to collide. Args: sections (int): Number of sections to generate. population (int): Population of the sections. separator (str): Separator between sections. Returns: str: Random id """ return tools.generate_random_id( sections=sections, population=population, separator=separator ) @api.model def is_valid_url(self, url, no_scheme_check=False): """ Check if the provided URL is a valid URL. The `urlparse` function from the `urllib.parse` module is used. Args: url (str): URL to check no_scheme_check (bool): If True, the scheme check will be skipped. Defaults to False. Returns: bool: True if the URL is valid, False otherwise """ return tools.is_valid_url(url=url, no_scheme_check=no_scheme_check)