22

✏️ Editing: rpm_lowpkg.py

"""
Support for rpm
"""

import datetime
import logging
import os
import re

import salt.utils.decorators.path
import salt.utils.itertools
import salt.utils.path
import salt.utils.pkg.rpm
from salt.exceptions import CommandExecutionError, SaltInvocationError
from salt.utils.versions import LooseVersion

try:
    import rpm

HAS_RPM = True
except ImportError:
    HAS_RPM = False

log = logging.getLogger(__name__)

# Define the module's virtual name
__virtualname__ = "lowpkg"

def __virtual__():
    """
    Confine this module to rpm based systems
    """
    if not salt.utils.path.which("rpm"):
        return (
            False,
            "The rpm execution module failed to load: rpm binary is not in the path.",
        )
    try:
        os_grain = __grains__["os"].lower()
        os_family = __grains__["os_family"].lower()
    except Exception:  # pylint: disable=broad-except
        return (
            False,
            "The rpm execution module failed to load: failed to detect os or os_family"
            " grains.",
        )

enabled = (
        "amazon",
        "xcp",
        "xenserver",
        "virtuozzolinux",
        "virtuozzo",
        "issabel pbx",
        "openeuler",
    )

if os_family in ["redhat", "suse"] or os_grain in enabled:
        return __virtualname__
    return (
        False,
        "The rpm execution module failed to load: only available on redhat/suse type"
        " systems or amazon, xcp, xenserver, virtuozzolinux, virtuozzo, issabel pbx or openeuler.",
    )

def bin_pkg_info(path, saltenv="base"):
    """
    .. versionadded:: 2015.8.0

Parses RPM metadata and returns a dictionary of information about the
    package (name, version, etc.).

path
        Path to the file. Can either be an absolute path to a file on the
        minion, or a salt fileserver URL (e.g. ``salt://path/to/file.rpm``).
        If a salt fileserver URL is passed, the file will be cached to the
        minion so that it can be examined.

saltenv : base
        Salt fileserver environment from which to retrieve the package. Ignored
        if ``path`` is a local file path on the minion.

CLI Example:

.. code-block:: bash

salt '*' lowpkg.bin_pkg_info /root/salt-2015.5.1-2.el7.noarch.rpm
        salt '*' lowpkg.bin_pkg_info salt://salt-2015.5.1-2.el7.noarch.rpm
    """
    # If the path is a valid protocol, pull it down using cp.cache_file
    if __salt__["config.valid_fileproto"](path):
        newpath = __salt__["cp.cache_file"](path, saltenv)
        if not newpath:
            raise CommandExecutionError(
                f"Unable to retrieve {path} from saltenv '{saltenv}'"
            )
        path = newpath
    else:
        if not os.path.exists(path):
            raise CommandExecutionError(f"{path} does not exist on minion")
        elif not os.path.isabs(path):
            raise SaltInvocationError(f"{path} does not exist on minion")

# REPOID is not a valid tag for the rpm command. Remove it and replace it
    # with 'none'
    queryformat = salt.utils.pkg.rpm.QUERYFORMAT.replace("%{REPOID}", "none")
    output = __salt__["cmd.run_stdout"](
        ["rpm", "-qp", "--queryformat", queryformat, path],
        output_loglevel="trace",
        ignore_retcode=True,
        python_shell=False,
    )
    ret = {}
    pkginfo = salt.utils.pkg.rpm.parse_pkginfo(output, osarch=__grains__["osarch"])
    try:
        for field in pkginfo._fields:
            ret[field] = getattr(pkginfo, field)
    except AttributeError:
        # pkginfo is None
        return None
    return ret

def list_pkgs(*packages, **kwargs):
    """
    List the packages currently installed in a dict::

{'<package_name>': '<version>'}

root
        use root as top level directory (default: "/")

CLI Example:

.. code-block:: bash

salt '*' lowpkg.list_pkgs
    """
    pkgs = {}
    cmd = ["rpm"]
    if kwargs.get("root"):
        cmd.extend(["--root", kwargs["root"]])
    cmd.extend(["-q" if packages else "-qa", "--queryformat", r"%{NAME} %{VERSION}\n"])
    if packages:
        cmd.extend(packages)
    out = __salt__["cmd.run"](cmd, output_loglevel="trace", python_shell=False)
    for line in salt.utils.itertools.split(out, "\n"):
        if "is not installed" in line:
            continue
        comps = line.split()
        pkgs[comps[0]] = comps[1]
    return pkgs

def verify(*packages, **kwargs):
    """
    Runs an rpm -Va on a system, and returns the results in a dict

root
        use root as top level directory (default: "/")

Files with an attribute of config, doc, ghost, license or readme in the
    package header can be ignored using the ``ignore_types`` keyword argument

CLI Example:

.. code-block:: bash

salt '*' lowpkg.verify
        salt '*' lowpkg.verify httpd
        salt '*' lowpkg.verify httpd postfix
        salt '*' lowpkg.verify httpd postfix ignore_types=['config','doc']
    """
    ftypes = {"c": "config", "d": "doc", "g": "ghost", "l": "license", "r": "readme"}
    ret = {}
    ignore_types = kwargs.get("ignore_types", [])
    if not isinstance(ignore_types, (list, str)):
        raise SaltInvocationError(
            "ignore_types must be a list or a comma-separated string"
        )
    if isinstance(ignore_types, str):
        try:
            ignore_types = [x.strip() for x in ignore_types.split(",")]
        except AttributeError:
            ignore_types = [x.strip() for x in str(ignore_types).split(",")]

verify_options = kwargs.get("verify_options", [])
    if not isinstance(verify_options, (list, str)):
        raise SaltInvocationError(
            "verify_options must be a list or a comma-separated string"
        )
    if isinstance(verify_options, str):
        try:
            verify_options = [x.strip() for x in verify_options.split(",")]
        except AttributeError:
            verify_options = [x.strip() for x in str(verify_options).split(",")]

cmd = ["rpm"]
    if kwargs.get("root"):
        cmd.extend(["--root", kwargs["root"]])
    cmd.extend(["--" + x for x in verify_options])
    if packages:
        cmd.append("-V")
        # Can't concatenate a tuple, must do a list.extend()
        cmd.extend(packages)
    else:
        cmd.append("-Va")
    out = __salt__["cmd.run_all"](
        cmd, output_loglevel="trace", ignore_retcode=True, python_shell=False
    )

if not out["stdout"].strip() and out["retcode"] != 0:
        # If there is no stdout and the retcode is 0, then verification
        # succeeded, but if the retcode is nonzero, then the command failed.
        msg = "Failed to verify package(s)"
        if out["stderr"]:
            msg += ": {}".format(out["stderr"])
        raise CommandExecutionError(msg)

for line in salt.utils.itertools.split(out["stdout"], "\n"):
        fdict = {"mismatch": []}
        if "missing" in line:
            line = " " + line
            fdict["missing"] = True
            del fdict["mismatch"]
        fname = line[13:]
        if line[11:12] in ftypes:
            fdict["type"] = ftypes[line[11:12]]
        if "type" not in fdict or fdict["type"] not in ignore_types:
            if line[0:1] == "S":
                fdict["mismatch"].append("size")
            if line[1:2] == "M":
                fdict["mismatch"].append("mode")
            if line[2:3] == "5":
                fdict["mismatch"].append("md5sum")
            if line[3:4] == "D":
                fdict["mismatch"].append("device major/minor number")
            if line[4:5] == "L":
                fdict["mismatch"].append("readlink path")
            if line[5:6] == "U":
                fdict["mismatch"].append("user")
            if line[6:7] == "G":
                fdict["mismatch"].append("group")
            if line[7:8] == "T":
                fdict["mismatch"].append("mtime")
            if line[8:9] == "P":
                fdict["mismatch"].append("capabilities")
            ret[fname] = fdict
    return ret

def modified(*packages, **flags):
    """
    List the modified files that belong to a package. Not specifying any packages
    will return a list of _all_ modified files on the system's RPM database.

.. versionadded:: 2015.5.0

root
        use root as top level directory (default: "/")

CLI Examples:

.. code-block:: bash

salt '*' lowpkg.modified httpd
        salt '*' lowpkg.modified httpd postfix
        salt '*' lowpkg.modified
    """
    cmd = ["rpm"]
    if flags.get("root"):
        cmd.extend(["--root", flags.pop("root")])
    cmd.append("-Va")
    cmd.extend(packages)
    ret = __salt__["cmd.run_all"](cmd, output_loglevel="trace", python_shell=False)

data = {}

# If verification has an output, then it means it failed
    # and the return code will be 1. We are interested in any bigger
    # than 1 code.
    if ret["retcode"] > 1:
        del ret["stdout"]
        return ret
    elif not ret["retcode"]:
        return data

ptrn = re.compile(r"\s+")
    changes = cfg = f_name = None
    for f_info in salt.utils.itertools.split(ret["stdout"], "\n"):
        f_info = ptrn.split(f_info)
        if len(f_info) == 3:  # Config file
            changes, cfg, f_name = f_info
        else:
            changes, f_name = f_info
            cfg = None
        keys = [
            "size",
            "mode",
            "checksum",
            "device",
            "symlink",
            "owner",
            "group",
            "time",
            "capabilities",
        ]
        changes = list(changes)
        if len(changes) == 8:  # Older RPMs do not support capabilities
            changes.append(".")
        stats = []
        for k, v in zip(keys, changes):
            if v != ".":
                stats.append(k)
        if cfg is not None:
            stats.append("config")
        data[f_name] = stats

if not flags:
        return data

# Filtering
    filtered_data = {}
    for f_name, stats in data.items():
        include = True
        for param, pval in flags.items():
            if param.startswith("_"):
                continue
            if (not pval and param in stats) or (pval and param not in stats):
                include = False
                break
        if include:
            filtered_data[f_name] = stats

return filtered_data

def file_list(*packages, **kwargs):
    """
    List the files that belong to a package. Not specifying any packages will
    return a list of _every_ file on the system's rpm database (not generally
    recommended).

root
        use root as top level directory (default: "/")

CLI Examples:

.. code-block:: bash

salt '*' lowpkg.file_list httpd
        salt '*' lowpkg.file_list httpd postfix
        salt '*' lowpkg.file_list
    """
    cmd = ["rpm"]
    if kwargs.get("root"):
        cmd.extend(["--root", kwargs["root"]])

cmd.append("-ql" if packages else "-qla")
    if packages:
        # Can't concatenate a tuple, must do a list.extend()
        cmd.extend(packages)

ret = __salt__["cmd.run"](
        cmd, output_loglevel="trace", python_shell=False
    ).splitlines()
    return {"errors": [], "files": ret}

def file_dict(*packages, **kwargs):
    """
    List the files that belong to a package, sorted by group. Not specifying
    any packages will return a list of _every_ file on the system's rpm
    database (not generally recommended).

root
        use root as top level directory (default: "/")

CLI Examples:

.. code-block:: bash

salt '*' lowpkg.file_dict httpd
        salt '*' lowpkg.file_dict httpd postfix
        salt '*' lowpkg.file_dict
    """
    errors = []
    ret = {}
    pkgs = {}
    cmd = ["rpm"]
    if kwargs.get("root"):
        cmd.extend(["--root", kwargs["root"]])
    cmd.extend(["-q" if packages else "-qa", "--queryformat", r"%{NAME} %{VERSION}\n"])
    if packages:
        cmd.extend(packages)
    out = __salt__["cmd.run"](cmd, output_loglevel="trace", python_shell=False)
    for line in salt.utils.itertools.split(out, "\n"):
        if "is not installed" in line:
            errors.append(line)
            continue
        comps = line.split()
        pkgs[comps[0]] = {"version": comps[1]}
    for pkg in pkgs:
        cmd = ["rpm"]
        if kwargs.get("root"):
            cmd.extend(["--root", kwargs["root"]])
        cmd.extend(["-ql", pkg])
        out = __salt__["cmd.run"](
            ["rpm", "-ql", pkg], output_loglevel="trace", python_shell=False
        )
        ret[pkg] = out.splitlines()
    return {"errors": errors, "packages": ret}

def owner(*paths, **kwargs):
    """
    Return the name of the package that owns the file. Multiple file paths can
    be passed. If a single path is passed, a string will be returned,
    and if multiple paths are passed, a dictionary of file/package name pairs
    will be returned.

If the file is not owned by a package, or is not present on the minion,
    then an empty string will be returned for that path.

root
        use root as top level directory (default: "/")

CLI Examples:

.. code-block:: bash

salt '*' lowpkg.owner /usr/bin/apachectl
        salt '*' lowpkg.owner /usr/bin/apachectl /etc/httpd/conf/httpd.conf
    """
    if not paths:
        return ""
    ret = {}
    for path in paths:
        cmd = ["rpm"]
        if kwargs.get("root"):
            cmd.extend(["--root", kwargs["root"]])
        cmd.extend(["-qf", "--queryformat", "%{name}", path])
        ret[path] = __salt__["cmd.run_stdout"](
            cmd, output_loglevel="trace", python_shell=False
        )
        if "not owned" in ret[path].lower():
            ret[path] = ""
    if len(ret) == 1:
        return next(iter(ret.values()))
    return ret

@salt.utils.decorators.path.which("rpm2cpio")
@salt.utils.decorators.path.which("cpio")
@salt.utils.decorators.path.which("diff")
def diff(package_path, path):
    """
    Return a formatted diff between current file and original in a package.
    NOTE: this function includes all files (configuration and not), but does
    not work on binary content.

:param package: Full pack of the RPM file
    :param path: Full path to the installed file
    :return: Difference or empty string. For binary files only a notification.

CLI Example:

.. code-block:: bash

salt '*' lowpkg.diff /path/to/apache2.rpm /etc/apache2/httpd.conf
    """

cmd = (
        "rpm2cpio {0} "
        "| cpio -i --quiet --to-stdout .{1} "
        "| diff -u --label 'A {1}' --from-file=- --label 'B {1}' {1}"
    )
    res = __salt__["cmd.shell"](cmd.format(package_path, path), output_loglevel="trace")
    if res and res.startswith("Binary file"):
        return f"File '{path}' is binary and its content has been modified."

return res

def info(*packages, **kwargs):
    """
    Return a detailed package(s) summary information.
    If no packages specified, all packages will be returned.

:param packages:

:param attr:
        Comma-separated package attributes. If no 'attr' is specified, all available attributes returned.

Valid attributes are:
            version, vendor, release, build_date, build_date_time_t, install_date, install_date_time_t,
            build_host, group, source_rpm, arch, epoch, size, license, signature, packager, url, summary, description.

:param all_versions:
        Return information for all installed versions of the packages

:param root:
        use root as top level directory (default: "/")

:return:

CLI Example:

.. code-block:: bash

salt '*' lowpkg.info apache2 bash
        salt '*' lowpkg.info apache2 bash attr=version
        salt '*' lowpkg.info apache2 bash attr=version,build_date_iso,size
        salt '*' lowpkg.info apache2 bash attr=version,build_date_iso,size all_versions=True
    """
    all_versions = kwargs.get("all_versions", False)
    # LONGSIZE is not a valid tag for all versions of rpm. If LONGSIZE isn't
    # available, then we can just use SIZE for older versions. See Issue #31366.
    rpm_tags = __salt__["cmd.run_stdout"](
        ["rpm", "--querytags"], python_shell=False
    ).splitlines()
    if "LONGSIZE" in rpm_tags:
        size_tag = "%{LONGSIZE}"
    else:
        size_tag = "%{SIZE}"

cmd = ["rpm"]
    if kwargs.get("root"):
        cmd.extend(["--root", kwargs["root"]])
    if packages:
        cmd.append("-q")
        cmd.extend(packages)
    else:
        cmd.append("-qa")

# Construct query format
    attr_map = {
        "name": "name: %{NAME}\\n",
        "relocations": (
            "relocations: %|PREFIXES?{[%{PREFIXES} ]}:{(not relocatable)}|\\n"
        ),
        "version": "version: %{VERSION}\\n",
        "vendor": "vendor: %{VENDOR}\\n",
        "release": "release: %{RELEASE}\\n",
        "epoch": "%|EPOCH?{epoch: %{EPOCH}\\n}|",
        "build_date_time_t": "build_date_time_t: %{BUILDTIME}\\n",
        "build_date": "build_date: %{BUILDTIME}\\n",
        "install_date_time_t": (
            "install_date_time_t: %|INSTALLTIME?{%{INSTALLTIME}}:{(not installed)}|\\n"
        ),
        "install_date": (
            "install_date: %|INSTALLTIME?{%{INSTALLTIME}}:{(not installed)}|\\n"
        ),
        "build_host": "build_host: %{BUILDHOST}\\n",
        "group": "group: %{GROUP}\\n",
        "source_rpm": "source_rpm: %{SOURCERPM}\\n",
        "size": "size: " + size_tag + "\\n",
        "arch": "arch: %{ARCH}\\n",
        "license": "%|LICENSE?{license: %{LICENSE}\\n}|",
        "signature": (
            "signature:"
            " %|DSAHEADER?{%{DSAHEADER:pgpsig}}:{%|RSAHEADER?{%{RSAHEADER:pgpsig}}:"
            "{%|SIGGPG?{%{SIGGPG:pgpsig}}:{%|SIGPGP?{%{SIGPGP:pgpsig}}:{(none)}|}|}|}|\\n"
        ),
        "packager": "%|PACKAGER?{packager: %{PACKAGER}\\n}|",
        "url": "%|URL?{url: %{URL}\\n}|",
        "summary": "summary: %{SUMMARY}\\n",
        "description": "description:\\n%{DESCRIPTION}\\n",
        "edition": "edition: %|EPOCH?{%{EPOCH}:}|%{VERSION}-%{RELEASE}\\n",
    }

attr = kwargs.get("attr", None) and kwargs["attr"].split(",") or None
    query = list()
    if attr:
        for attr_k in attr:
            if attr_k in attr_map and attr_k != "description":
                query.append(attr_map[attr_k])
        if not query:
            raise CommandExecutionError("No valid attributes found.")
        if "name" not in attr:
            attr.append("name")
            query.append(attr_map["name"])
        if "edition" not in attr:
            attr.append("edition")
            query.append(attr_map["edition"])
    else:
        for attr_k, attr_v in attr_map.items():
            if attr_k != "description":
                query.append(attr_v)
    if attr and "description" in attr or not attr:
        query.append(attr_map["description"])
    query.append("-----\\n")

cmd = " ".join(cmd)
    call = __salt__["cmd.run_all"](
        cmd + " --queryformat '{}'".format("".join(query)),
        output_loglevel="trace",
        env={"TZ": "UTC"},
        clean_env=True,
        ignore_retcode=True,
    )
    out = call["stdout"]
    _ret = list()
    for pkg_info in re.split(r"----*", out):
        pkg_info = pkg_info.strip()
        if not pkg_info:
            continue
        pkg_info = pkg_info.split(os.linesep)
        if pkg_info[-1].lower().startswith("distribution"):
            pkg_info = pkg_info[:-1]

pkg_data = dict()
        pkg_name = None
        descr_marker = False
        descr = list()
        for line in pkg_info:
            if descr_marker:
                descr.append(line)
                continue
            line = [item.strip() for item in line.split(":", 1)]
            if len(line) != 2:
                continue
            key, value = line
            if key == "description":
                descr_marker = True
                continue
            if key == "name":
                pkg_name = value

# Convert Unix ticks into ISO time format
            if key in ["build_date", "install_date"]:
                try:
                    pkg_data[key] = (
                        datetime.datetime.utcfromtimestamp(int(value)).isoformat() + "Z"
                    )
                except ValueError:
                    log.warning('Could not convert "%s" into Unix time', value)
                continue

# Convert Unix ticks into an Integer
            if key in ["build_date_time_t", "install_date_time_t"]:
                try:
                    pkg_data[key] = int(value)
                except ValueError:
                    log.warning('Could not convert "%s" into Unix time', value)
                continue
            if key not in ["description", "name"] and value:
                pkg_data[key] = value
        if attr and "description" in attr or not attr:
            pkg_data["description"] = os.linesep.join(descr)
        if pkg_name:
            pkg_data["name"] = pkg_name
            _ret.append(pkg_data)

# Force-sort package data by version,
    # pick only latest versions
    # (in case multiple packages installed, e.g. kernel)
    ret = dict()
    for pkg_data in reversed(sorted(_ret, key=lambda x: LooseVersion(x["edition"]))):
        pkg_name = pkg_data.pop("name")
        # Filter out GPG public keys packages
        if pkg_name.startswith("gpg-pubkey"):
            continue
        if pkg_name not in ret:
            if all_versions:
                ret[pkg_name] = [pkg_data.copy()]
            else:
                ret[pkg_name] = pkg_data.copy()
                del ret[pkg_name]["edition"]
        elif all_versions:
            ret[pkg_name].append(pkg_data.copy())

return ret

def version_cmp(ver1, ver2, ignore_epoch=False):
    """
    .. versionadded:: 2015.8.9

Do a cmp-style comparison on two packages. Return -1 if ver1 < ver2, 0 if
    ver1 == ver2, and 1 if ver1 > ver2. Return None if there was a problem
    making the comparison.

ignore_epoch : False
        Set to ``True`` to ignore the epoch when comparing versions

.. versionadded:: 2015.8.10,2016.3.2

CLI Example:

.. code-block:: bash

salt '*' pkg.version_cmp '0.2-001' '0.2.0.1-002'
    """

def normalize(x):
        return str(x).split(":", maxsplit=1)[-1] if ignore_epoch else str(x)

ver1 = normalize(ver1)
    ver2 = normalize(ver2)

(ver1_e, ver1_v, ver1_r) = salt.utils.pkg.rpm.version_to_evr(ver1)
    (ver2_e, ver2_v, ver2_r) = salt.utils.pkg.rpm.version_to_evr(ver2)
    # If one EVR is missing a release but not the other and they
    # otherwise would be equal, ignore the release. This can happen if
    # e.g. you are checking if a package version 3.2 is satisfied by
    # 3.2-1.
    if not ver1_r or not ver2_r:
        ver1_r = ver2_r = ""

if HAS_RPM:
        try:
            cmp_func = None
            try:
                cmp_func = rpm.labelCompare
            except AttributeError:
                # Catches corner case where someone has a module named "rpm" in
                # their pythonpath.
                log.debug(
                    "rpm module imported, but it does not have the "
                    "labelCompare function. Not using rpm.labelCompare for "
                    "version comparison."
                )
            if cmp_func is not None:
                cmp_result = cmp_func(
                    (ver1_e, ver1_v, ver1_r), (ver2_e, ver2_v, ver2_r)
                )
                if cmp_result not in (-1, 0, 1):
                    raise CommandExecutionError(
                        f"Comparison result '{cmp_result}' is invalid"
                    )
                return cmp_result
        except Exception as exc:  # pylint: disable=broad-except
            log.warning(
                "Failed to compare version '%s' to '%s' using RPM: %s", ver1, ver2, exc
            )
    else:
        log.info(
            "Install a package that provides rpm.labelCompare for "
            "faster version comparisons."
        )
    return salt.utils.pkg.rpm.evr_compare(
        (ver1_e, ver1_v, ver1_r), (ver2_e, ver2_v, ver2_r)
    )

def checksum(*paths, **kwargs):
    """
    Return if the signature of a RPM file is valid.

root
        use root as top level directory (default: "/")

CLI Example:

.. code-block:: bash

salt '*' lowpkg.checksum /path/to/package1.rpm
        salt '*' lowpkg.checksum /path/to/package1.rpm /path/to/package2.rpm
    """
    ret = dict()

if not paths:
        raise CommandExecutionError("No package files has been specified.")

cmd = ["rpm"]
    if kwargs.get("root"):
        cmd.extend(["--root", kwargs["root"]])
    cmd.extend(["-K", "--quiet"])
    for package_file in paths:
        cmd_ = cmd + [package_file]
        ret[package_file] = bool(
            __salt__["file.file_exists"](package_file)
        ) and not __salt__["cmd.retcode"](
            cmd_, ignore_retcode=True, output_loglevel="trace", python_shell=False
        )

return ret

MR SKB