From bf7158a6e8ff380ef5dae6c6f0f267c7a57c0db7 Mon Sep 17 00:00:00 2001 From: git_admin Date: Mon, 27 Apr 2026 08:15:50 +0000 Subject: [PATCH] Tower: upload cetmix_tower_server 16.0.3.0.1 (via marketplace) --- .../models/cx_tower_command.py | 657 ++++++++++++++++++ 1 file changed, 657 insertions(+) create mode 100644 addons/cetmix_tower_server/models/cx_tower_command.py diff --git a/addons/cetmix_tower_server/models/cx_tower_command.py b/addons/cetmix_tower_server/models/cx_tower_command.py new file mode 100644 index 0000000..807103b --- /dev/null +++ b/addons/cetmix_tower_server/models/cx_tower_command.py @@ -0,0 +1,657 @@ +# Copyright (C) 2022 Cetmix OÜ +# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl). +import logging +from types import SimpleNamespace +from urllib import parse + +from dns import exception, resolver, reversename +from pytz import timezone + +from odoo import _, api, fields, models, tools +from odoo.exceptions import UserError +from odoo.tools import ormcache +from odoo.tools.float_utils import float_compare +from odoo.tools.safe_eval import wrap_module + +from .constants import DEFAULT_PYTHON_CODE, DEFAULT_PYTHON_CODE_HELP + +_logger = logging.getLogger(__name__) + +requests = wrap_module(__import__("requests"), ["post", "get", "delete", "request"]) +json = wrap_module(__import__("json"), ["dumps"]) +hashlib = wrap_module( + __import__("hashlib"), + [ + "sha1", + "sha224", + "sha256", + "sha384", + "sha512", + "sha3_224", + "sha3_256", + "sha3_384", + "sha3_512", + "shake_128", + "shake_256", + "blake2b", + "blake2s", + "md5", + "new", + ], +) +re = wrap_module( + __import__("re"), + [ + "match", + "fullmatch", + "search", + "sub", + "subn", + "split", + "findall", + "finditer", + "compile", + "template", + "escape", + "error", + ], +) +hmac = wrap_module( + __import__("hmac"), + ["new", "compare_digest"], +) +urllib_parse = wrap_module( + parse, + [ + "urlparse", + "urljoin", + "urlunparse", + "urlencode", + "urlsplit", + "urlunsplit", + "parse_qs", + "parse_qsl", + "quote", + "quote_plus", + "quote_from_bytes", + "unquote", + "unquote_plus", + "unquote_to_bytes", + ], +) +tldextract = wrap_module(__import__("tldextract"), ["extract"]) +dns_resolver = wrap_module(resolver, ["resolve", "query"]) +dns_reversename = wrap_module(reversename, ["from_address", "to_address"]) +dns_exception = wrap_module(exception, ["DNSException"]) + + +dns = SimpleNamespace( + resolver=dns_resolver, + reversename=dns_reversename, + exception=dns_exception, +) + + +class CxTowerCommand(models.Model): + """Command to run on a server""" + + _name = "cx.tower.command" + _inherit = [ + "cx.tower.template.mixin", + "cx.tower.reference.mixin", + "cx.tower.access.mixin", + "cx.tower.access.role.mixin", + "cx.tower.key.mixin", + "cx.tower.tag.mixin", + ] + _description = "Cetmix Tower Command" + _order = "name" + + active = fields.Boolean(default=True) + allow_parallel_run = fields.Boolean( + help="If enabled, multiple instances of the same command " + "can be run on the same server at the same time.\n" + "Otherwise, ANOTHER_COMMAND_RUNNING status will be returned if another" + " instance of the same command is already running" + ) + server_ids = fields.Many2many( + comodel_name="cx.tower.server", + relation="cx_tower_server_command_rel", + column1="command_id", + column2="server_id", + string="Servers", + help="Servers on which the command will be run.\n" + "If empty, command can be run on all servers", + ) + tag_ids = fields.Many2many( + relation="cx_tower_command_tag_rel", + column1="command_id", + column2="tag_id", + ) + os_ids = fields.Many2many( + comodel_name="cx.tower.os", + relation="cx_tower_os_command_rel", + column1="command_id", + column2="os_id", + string="OSes", + ) + note = fields.Text() + + action = fields.Selection( + selection=lambda self: self._selection_action(), + required=True, + default=lambda self: self._selection_action()[0][0], + ) + path = fields.Char( + string="Default Path", + help="Location where command will be run. " + "You can use {{ variables }} in path", + ) + file_template_id = fields.Many2one( + comodel_name="cx.tower.file.template", + help="This template will be used to create or update the pushed file", + ) + template_code = fields.Text( + string="Template Code", + related="file_template_id.code", + readonly=True, + help="Code of the associated file template", + ) + flight_plan_line_ids = fields.One2many( + comodel_name="cx.tower.plan.line", + related="flight_plan_id.line_ids", + readonly=True, + help="Lines of the associated flight plan", + ) + code = fields.Text( + compute="_compute_code", + store=True, + readonly=False, + ) + command_help = fields.Html( + compute="_compute_command_help", + compute_sudo=True, + ) + flight_plan_id = fields.Many2one( + comodel_name="cx.tower.plan", + help="Flight plan run by the command", + ) + flight_plan_used_ids = fields.Many2many( + comodel_name="cx.tower.plan", + help="Flight plan this command is used in", + relation="cx_tower_command_flight_plan_used_id_rel", + column1="command_id", + column2="plan_id", + store=True, + copy=False, + ) + flight_plan_used_ids_count = fields.Integer( + compute="_compute_flight_plan_used_ids_count", + help="Flight plan this command is used in", + ) + server_status = fields.Selection( + selection=lambda self: self.env["cx.tower.server"]._selection_status(), + help="Set the following status if command finishes with success. " + "Leave 'Undefined' if you don't need to update the status", + ) + no_split_for_sudo = fields.Boolean( + string="No Split for sudo", + help="If enabled, do not split command on '&&' when using sudo." + "Prepend sudo once to the whole command.", + ) + variable_ids = fields.Many2many( + comodel_name="cx.tower.variable", + relation="cx_tower_command_variable_rel", + column1="command_id", + column2="variable_id", + ) + + if_file_exists = fields.Selection( + selection=[ + ("skip", "Skip"), + ("overwrite", "Overwrite"), + ("raise", "Raise Error"), + ], + default="skip", + help="What to do if file already exists on the server.\n" + "- Skip: Do not create or update the file.\n" + "- Overwrite: Replace the existing file with the new one.\n" + "- Raise Error: Raise an error if the file already exists.", + ) + disconnect_file = fields.Boolean( + string="Disconnect from Template", + help=( + "If enabled, disconnects the file from its template " + "after running the command.\n" + ), + ) + # -- Jets + jet_template_id = fields.Many2one( + comodel_name="cx.tower.jet.template", + help="Action will be triggered for all dependent jets" " of this template", + ) + jet_action_id = fields.Many2one( + comodel_name="cx.tower.jet.action", + help="Action to trigger", + domain="[('jet_template_id', '=', jet_template_id)]", + ) + # -- Waypoints + waypoint_template_id = fields.Many2one( + comodel_name="cx.tower.jet.waypoint.template", + string="Waypoint Template", + help="Waypoint template to create the waypoint from. Used when action is " + "Create a Waypoint.", + ) + fly_here = fields.Boolean( + default=False, + help="When enabled, the created waypoint is set as current (fly to) " + "after creation.", + ) + + # ---- Access. Add relation for mixin fields + user_ids = fields.Many2many( + relation="cx_tower_command_user_rel", + ) + manager_ids = fields.Many2many( + relation="cx_tower_command_manager_rel", + ) + + @classmethod + def _get_depends_fields(cls): + """ + Define dependent fields for computing `variable_ids` in command-related models. + + This implementation specifies that the fields `code` and `path` + are used to determine the variables associated with a command. + + Returns: + list: A list of field names (str) representing the dependencies. + + Example: + The following fields trigger recomputation of `variable_ids`: + - `code`: The command's script or running logic. + - `path`: The default running path for the command. + """ + return ["code", "path"] + + # -- Selection + def _selection_action(self): + """Actions that can be run by a command. + + Returns: + List of tuples: available options. + """ + return [ + ("ssh_command", "SSH Command"), + ("python_code", "Python Code"), + ("file_using_template", "Create/Update File"), + ("plan", "Run Flight Plan"), + ("jet_action", "Trigger Jet Action"), + ("create_waypoint", "Create Waypoint"), + ] + + # -- Defaults + def _get_default_python_code(self): + """ + Default python command code + """ + return DEFAULT_PYTHON_CODE + + def _get_default_python_code_help(self): + """ + Default python code help + """ + + # Available libraries are Odoo objects + Python libraries + available_libraries = self._get_python_command_odoo_objects() + available_libraries.update(self._get_python_command_libraries()) + help_text_fragments = [] + for key, value in available_libraries.items(): + help_text_fragments.append(f"
  • {key}: {value['help']}
    • ") + + help_text_fragments.append( + f"
  • custom_values: {_('Flight plan custom values')}
    • " + ) + + help_text = "" + return f"{DEFAULT_PYTHON_CODE_HELP}{help_text}" + + # -- Computes + @api.depends("action") + def _compute_code(self): + """ + Compute default code + """ + default_python_code = self._get_default_python_code() + for command in self: + if command.action == "python_code": + command.code = default_python_code + continue + command.code = False + + @api.depends("action") + def _compute_command_help(self): + """ + Compute command help + """ + default_python_code_help = self._get_default_python_code_help() + for command in self: + if command.action == "python_code": + command.command_help = default_python_code_help + else: + command.command_help = False + + @api.depends("flight_plan_used_ids") + def _compute_flight_plan_used_ids_count(self): + """ + Compute flight plan ids count + """ + for command in self: + command.flight_plan_used_ids_count = len(command.flight_plan_used_ids) + + def action_open_command_logs(self): + """ + Open current current command log records + """ + action = self.env["ir.actions.actions"]._for_xml_id( + "cetmix_tower_server.action_cx_tower_command_log" + ) + action["domain"] = [("command_id", "=", self.id)] + return action + + def action_open_plans(self): + """ + Open plans this command is used in + """ + action = self.env["ir.actions.actions"]._for_xml_id( + "cetmix_tower_server.action_cx_tower_plan" + ) + action["domain"] = [("id", "in", self.flight_plan_used_ids.ids)] + return action + + def _check_server_compatibility(self, server): + """Check if the command is compatible with the server + Args: + server (cx.tower.server()): Server object + + Returns: + bool: True if the command is compatible with the server, False otherwise + """ + self.ensure_one() + return not self.server_ids or server.id in self.server_ids.ids + + # -- Business logic + @ormcache() + @api.model + def _get_python_command_libraries(self): + """ + Get available python imports. Use this method to import python libraries. + Please be advised, that this method is cached. + If you need to use a non-cached import, eg for Odoo objects, + use the `_get_python_command_odoo_objects` method instead. + + + Returns: + dict: Available libraries: + {"": { + "import": , + "help": + }} + """ + python_libraries = { + "_logger": { + "import": _logger, + "help": _( + "Logger object. Use with caution! Only for debugging purposes." + ), + }, + "re": { + "import": re, + "help": _("Python 're' library for regex operations"), + }, + "time": { + "import": tools.safe_eval.time, + "help": _("Python 'time' library"), + }, + "datetime": { + "import": tools.safe_eval.datetime, + "help": _("Python 'datetime' library"), + }, + "dateutil": { + "import": tools.safe_eval.dateutil, + "help": _("Python 'dateutil' library"), + }, + "timezone": { + "import": timezone, + "help": _("Python 'timezone' library"), + }, + "requests": { + "import": requests, + "help": _( + "Python 'requests' library. Available methods: 'post', 'get'," + " 'delete', 'request'" + ), + }, + "urllib_parse": { + "import": urllib_parse, + "help": _("Python 'urllib.parse' library methods."), + }, + "json": { + "import": json, + "help": _("Python 'json' library. Available methods: 'dumps'"), + }, + "float_compare": { + "import": float_compare, + "help": _("Float compare. Odoo helper function to compare floats."), + }, + "UserError": { + "import": UserError, + "help": _("UserError. Helper to raise UserError."), + }, + "hashlib": { + "import": hashlib, + "help": _( + "Python 'hashlib' library. " + "Documentation. " + "Available methods: 'sha1', 'sha224', " + "'sha256', 'sha384'," + " 'sha512', 'sha3_224', 'sha3_256', 'sha3_384', 'sha3_512', " + "'shake_128', 'shake_256'," + " 'blake2b', 'blake2s', 'md5', 'new'" + ), + }, + "hmac": { + "import": hmac, + "help": _( + "Python 'hmac' library. " + "Documentation. " + "Use 'new' to create HMAC objects. " + "Available methods on the HMAC *object*: 'update', 'copy'," + " 'digest', 'hexdigest'. " + " Module-level function: 'compare_digest'." + ), + }, + "tldextract": { + "import": tldextract, + "help": _( + "Python 'tldextract' library. Use " + "tldextract.extract() to parse domains. " + "Check tldextract for more information." + ), + }, + "dns": { + "import": dns, + "help": _( + "Python 'dnspython' library. " + "Documentation." + "
    • dns.resolver: " + "wrapped dnspython. Use " + 'dns.resolver.resolve(hostname, "A") for ' + "DNS lookups.
      • " + "
    • dns.reversename: wrapped dnspython. " + 'Use dns.reversename.from_address("8.8.8.8")' + " to build and reverse PTR records.
      • " + "
    • dns.exception: wrapped dnspython. " + "Catch " + "dns.exception.DNSException to handle " + "DNS-related errors.
      • " + "
    " + ), + }, + } + custom_python_libraries = self._custom_python_libraries() + for libraries in custom_python_libraries.values(): + python_libraries.update(libraries) + return python_libraries + + def _get_python_command_odoo_objects( + self, server=None, jet_template=None, jet=None, waypoint=None + ): + """ + This method is used to import Odoo objects. + Because Odoo objects can be records, this method is not cached. + Use this method to import Odoo objects that are not cached. + If you need to import some static objects, use the + `_get_python_command_libraries` method instead. + + Args: + server: Server to get the Odoo objects for. + jet_template: Jet template to get the Odoo objects for. + jet: Jet to get the Odoo objects for. + waypoint: Waypoint to get the Odoo objects for. + + Returns: + dict: Available Odoo objects: + {"": { + "import": , + "help": + }} + """ + return { + "uid": {"import": self._uid, "help": _("Current Odoo user ID")}, + "user": {"import": self.env.user, "help": _("Current Odoo user")}, + "env": {"import": self.env, "help": _("Odoo Environment")}, + "server": { + "import": server, + "help": _("Current Cetmix Tower server this command is running on"), + }, + "jet_template": { + "import": jet_template, + "help": _( + "Current Cetmix Tower jet template this command is running on" + ), + }, + "jet": { + "import": jet, + "help": _("Current Cetmix Tower jet this command is running on"), + }, + "waypoint": { + "import": waypoint, + "help": _( + "Current Cetmix Tower Jet waypoint this command is running on" + ), + }, + "tower": { + "import": self.env["cetmix.tower"], + "help": _( + "Cetmix Tower " + "helper class shortcut" + ), + }, + "tower_servers": { + "import": self.env["cx.tower.server"], + "help": _("A helper shortcut to env['cx.tower.server']"), + }, + "tower_jets": { + "import": self.env["cx.tower.jet"], + "help": _("A helper shortcut to env['cx.tower.jet']"), + }, + "tower_commands": { + "import": self.env["cx.tower.command"], + "help": _("A helper shortcut to env['cx.tower.command']"), + }, + "tower_plans": { + "import": self.env["cx.tower.plan"], + "help": _("A helper shortcut to env['cx.tower.plan']"), + }, + "tower_waypoints": { + "import": self.env["cx.tower.jet.waypoint"], + "help": _( + "A helper shortcut to env['cx.tower.jet.waypoint']" + ), + }, + } + + def _custom_python_libraries(self): + """ + This function is designed to be used in custom modules + extending Cetmix Tower to add custom python libraries + to the evaluation context. + + Returns: + Dict: Custom python libraries. + + The following format is used: + { + : {"": { + "import": , + "help": + } + } + + Where: + + Odoo module technical name. + is the name of the library how it will be used in the code. + + : The library object to expose. + : Help text (HTML) shown in the "Help" tab. + """ + return {} + + def _get_python_command_eval_context(self, server=None, **kwargs): + """ + Get the evaluation context for the python command. + This method is used to get the evaluation context for the python command. + + Args: + server: Server to get the evaluation context for. + **kwargs: Additional keyword arguments. + Returns: + dict: Evaluation context for the python command. + """ + + # Get the jet template, jet and waypoint from kwargs + jet_template = kwargs.get("jet_template") + jet = kwargs.get("jet") + waypoint = kwargs.get("waypoint") + + # Get the Odoo objects first + imports = self._get_python_command_odoo_objects( + server=server, + jet_template=jet_template, + jet=jet, + waypoint=waypoint, + ) + + # Update with the libraries + imports.update(self._get_python_command_libraries()) + eval_context = {key: value["import"] for key, value in imports.items()} + + eval_context["custom_values"] = kwargs.get("variable_values") or {} + return eval_context + + def _get_banned_python_code_keywords(self): + """ + Get the banned python code keywords for the python command. + Extend this method to add banned keywords to the list. + + Returns: + list: Banned python code keywords. + """ + return ["_set_secret_values(", "_get_secret_value(", "_get_secret_values("]