# The software in this package is distributed under the GNU General # Public License version 2 (with a special exception described below). # # A copy of GNU General Public License (GPL) is included in this distribution, # in the file COPYING.GPL. # # As a special exception, if other files instantiate templates or use macros # or inline functions from this file, or you compile this file and link it # with other works to produce a work based on this file, this file # does not by itself cause the resulting work to be covered # by the GNU General Public License. # # However the source code for this file must still be made available # in accordance with section (3) of the GNU General Public License. # # This exception does not invalidate any other reasons why a work based # on this file might be covered by the GNU General Public License. # # Copyright (c) 2016-2022 Intra2net AG """ binary: wrappers around binaries that work with the CNF store. Featuring: - CnfBinary: stateless class that can be used to invoke the different binaries in a Python-friendly way. .. note:: It is written as a class on purpose for it to be easily extended to support invoking non-local binaries, but methods are all class methods since the class is stateless. .. seealso:: Overview Diagram linked to from doc main page .. codeauthor:: Intra2net """ import subprocess import logging import shlex import html import re log = logging.getLogger("pyi2ncommon.cnfvar.binary") #: default get_cnf binary BIN_GET_CNF = "/usr/intranator/bin/get_cnf" #: default set_cnf binary BIN_SET_CNF = "/usr/intranator/bin/set_cnf" #: encoding used by the get_cnf and set_cnf binaries ENCODING = "latin1" class CnfBinary: """Provide wrappers around the multiple binaries to handle the CNF store.""" @classmethod def run_cmd(cls, cmd, cmd_input=None, ignore_errors=False, timeout=60, encoding=ENCODING): """ Run commands on the local machine with input via stdin. :param cmd: command to run :type cmd: str or list :param str cmd_input: input to give to the command :param bool ignore_errors: whether to ignore the exit code or raise if not zero :param int timeout: amount of seconds to wait for the command to complete :param str encoding: encoding to use (pass latin1 when not working with JSON) :returns: command result :rtype: :py:class:`subprocess.CompletedProcess` """ if isinstance(cmd, str): cmd = shlex.split(cmd) log.debug("Running binary cmd `%s` with input:\n%s", ' '.join(cmd), cmd_input) retval = subprocess.run(cmd, input=cmd_input, check=not ignore_errors, capture_output=True, encoding=encoding, timeout=timeout) log.debug("Command exited with \n" "\treturncode=%d\n" "\tstderr=%s\n" "\tstdout=%s", retval.returncode, retval.stderr, retval.stdout) return retval @classmethod def get_cnf(cls, name=None, instance=None, no_children=False, as_json=False): """ Wrapper around the `get_cnf` binary. :param str name: optional name of the CNFs to get :param instance: CNF instance :type instance: str or int :param bool no_children: whether to return child CNFs :param bool as_json: whether to return a JSON-formatted string :returns: output of the tool :rtype: str .. note:: being a wrapper, this function does not do anything extra like checking if arnied is running or waiting for generate. """ if name is None and instance is not None: raise ValueError("cannot pass `instance` without a `name`") if isinstance(instance, str): instance = int(instance) # validate elif instance is not None and not isinstance(instance, int): raise TypeError(f"`instance` is of wrong type {type(instance)}") # make sure 0 instance cnfvars like e.g. NICs can also be filtered by instance cmd = f"{BIN_GET_CNF} {name or ''} {instance if instance is not None else ''}" encoding = ENCODING if no_children: cmd += " --no-childs" if as_json: cmd += " --json" encoding = "utf8" # TODO: should error handling be improved so error messages # from the binary are converted into specific exceptions types? output = cls.run_cmd(cmd, encoding=encoding).stdout.strip() return output @classmethod def set_cnf(cls, input_str=None, input_file=None, delete=False, delete_file=False, as_json=False, fix_problems=False, **kwargs): """ Wrapper around the `set_cnf` binary. :param str input_str: string to send as input to the binary :param str input_file: path to a file to pass to the binary :param bool delete: whether to delete the corresponding CNFs :param bool delete_file: whether to delete the file passed as input :param bool as_json: whether to interpret input as JSON :param bool fix_problems: whether to automatically fix errors in the vars :param kwargs: extra arguments to pass to the binary - underscores are replaced by dash, e.g. set_cnf(..., enable_queue=True) becomes `/usr/intranator/bin/set_cnf --enable-queue` :raises: :py:class:`SetCnfException` in case the binary errors out .. note:: being a wrapper, this function does not do anything extra like checking if arnied is running or waiting for generate. """ if input_str is None and input_file is None: raise ValueError("Need to pass either a string or a file") if delete_file is True and input_file is None: raise ValueError("Cannot delete an unspecified file") cmd = f"{BIN_SET_CNF}" encoding = ENCODING if as_json: cmd += " --json" encoding = "utf8" if delete: cmd += " -x" if delete_file: cmd += " --delete-file" if fix_problems: cmd += " --fix-problems" for k, v in kwargs.items(): if v is True: k = "-".join(k.split("_")) cmd += f" --{k}" if input_file: cmd += f" {input_file}" try: cls.run_cmd(cmd, cmd_input=input_str, encoding=encoding) except subprocess.CalledProcessError as ex: # clean up the error message ex.stderr = cls._clean_set_cnf_error(ex.stderr) raise @classmethod def _clean_set_cnf_error(cls, err): """ Turn the error from the set_cnf binary into a more readable message. :param str err: error message from the binary :returns: the clean error message :rtype: str Keep only offending lines and strip out HTML tags. """ def get_error_lines(output_lines): for cnfline in output_lines: buffer = "" quoted = False parts = [] for c in cnfline: last_chr = buffer[-1] if buffer else None # this is a literal (unescaped) quote if c == "\"" and last_chr != "\\": quoted = not quoted if c == " " and not quoted: parts.append(buffer) buffer = "" else: buffer += c parts.append(buffer) if parts[-2] == "0": # no errors, ignore continue lineno, name, instance, parent, value, exit_code, error = parts # the binary outputs HTML strings error = html.unescape(error) # replace line breaks for readability error = re.sub(r"()+", " ", error) # clean up HTML tags error = re.sub(r"<[^<]+?>", "", error) if parent == "-1": message = f"`` {lineno} {name},{instance}: {value} ``" else: message = f"`` {lineno} ({parent}) {name},{instance}: {value} ``" yield f"Error in {message}: {error} (code={exit_code})" lines = err.splitlines() if len(lines) == 0: return err if "fatal errors in changes" not in lines[0]: return err errors = list(get_error_lines(lines[1:])) return "\n".join(errors)