File: //opt/saltstack/salt/lib/python3.10/site-packages/salt/modules/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