#!/usr/bin/python
# Note that the shebang above is ignored when run in policy
# See lib/packages.cf `package_module zypper` use of the
# `interpreter` attribute to use cfengine-selected-python.
#####################################################################################
# Copyright 2016 Normation SAS
#####################################################################################
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, Version 3.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see .
#
#####################################################################################
# This script is based on the CFEngine's masterfiles yum script:
# https://github.com/cfengine/masterfiles/blob/master/modules/packages/yum
# Licensed under:
# MIT Public License
# Copyright 2022 Northern.tech AS
import sys
import os
import subprocess
import re
from distutils.version import StrictVersion
rpm_cmd = os.environ.get('CFENGINE_TEST_RPM_CMD', "/bin/rpm")
rpm_quiet_option = ["--quiet"]
rpm_output_format = "Name=%{name}\nVersion=%|EPOCH?{%{epoch}:}:{}|%{version}-%{release}\nArchitecture=%{arch}\n"
zypper_cmd = os.environ.get('CFENGINE_TEST_ZYPPER_CMD', "/usr/bin/zypper")
zypper_options = ["--quiet", "-n"]
NULLFILE = open(os.devnull, 'w')
redirection_is_broken_cached = -1
zypper_supports_oldpackage_cached = -1
def zypper_supports_oldpackage():
# Suse zypper "--oldpackage" option is only supported greater that 1.6.169
global zypper_supports_oldpackage_cached
if zypper_supports_oldpackage_cached == -1:
zypper_version = subprocess.Popen([zypper_cmd, "-n", "--version"], stdout=subprocess.PIPE).communicate()[0]
zypper_version = zypper_version.decode().split(" ", 1)[1].rstrip()
if StrictVersion(zypper_version) >= StrictVersion("1.6.169"):
zypper_supports_oldpackage_cached = 1
else:
zypper_supports_oldpackage_cached = 0
return zypper_supports_oldpackage_cached
def redirection_is_broken():
# Older versions of Python have a bug where it is impossible to redirect
# stderr using subprocess, and any attempt at redirecting *anything*, not
# necessarily stderr, will result in it being closed instead. This is very
# bad, because RPM may then open its RPM database on file descriptor 2
# (stderr), and will cause it to output error messages directly into the
# database file. Fortunately "stdout=subprocess.PIPE" doesn't have the bug,
# and that's good, because it would have been much more tricky to solve.
global redirection_is_broken_cached
if redirection_is_broken_cached == -1:
cmd_line = [sys.executable, sys.argv[0], "internal-test-stderr"]
if subprocess.call(cmd_line, stdout=sys.stderr) == 0:
redirection_is_broken_cached = 0
else:
redirection_is_broken_cached = 1
return redirection_is_broken_cached
def subprocess_Popen(cmd, stdout=None, stderr=None):
if not redirection_is_broken() or (stdout is None and stderr is None) or stdout == subprocess.PIPE or stderr == subprocess.PIPE:
return subprocess.Popen(cmd, stdout=stdout, stderr=stderr)
old_stdout_fd = -1
old_stderr_fd = -1
if stdout is not None:
old_stdout_fd = os.dup(1)
os.dup2(stdout.fileno(), 1)
if stderr is not None:
old_stderr_fd = os.dup(2)
os.dup2(stderr.fileno(), 2)
result = subprocess.Popen(cmd)
if old_stdout_fd >= 0:
os.dup2(old_stdout_fd, 1)
os.close(old_stdout_fd)
if old_stderr_fd >= 0:
os.dup2(old_stderr_fd, 2)
os.close(old_stderr_fd)
return result
# If subprocess.PIPE is passed as stderr, it will re-write it in an ErrorMessage
def subprocess_call(cmd, stdout=None, stderr=None):
process = subprocess_Popen(cmd, stdout, stderr)
outs, errs = process.communicate()
if stderr == subprocess.PIPE:
lines = [line for line in errs.decode().splitlines()]
if len(lines):
printed_error = "ErrorMessage=" + " ".join(lines)
sys.stdout.write(printed_error)
sys.stdout.flush()
return process.returncode
def get_package_data():
pkg_string = ""
for line in sys.stdin:
if line.startswith("File="):
pkg_string = line.split("=", 1)[1].rstrip()
# Don't break, we need to exhaust stdin.
if not pkg_string:
return 1
if pkg_string.startswith("/"):
# Absolute file.
sys.stdout.write("PackageType=file\n")
sys.stdout.flush()
subprocess_call([rpm_cmd, "--qf", rpm_output_format, "-qp", pkg_string], stderr=subprocess.PIPE)
return 0
elif re.search("[:,]", pkg_string):
# Contains an illegal symbol.
sys.stdout.write(line + "ErrorMessage: Package string with illegal format\n")
return 1
else:
sys.stdout.write("PackageType=repo\n")
sys.stdout.write("Name=" + pkg_string + "\n")
return 0
def list_installed():
# Ignore everything.
sys.stdin.readlines()
subprocess_call([rpm_cmd, "-qa", "--qf", rpm_output_format])
return 0
def list_updates(online):
# Ignore everything.
sys.stdin.readlines()
online_flag = []
if not online:
online_flag = ["--no-refresh"]
process = subprocess_Popen([zypper_cmd] + zypper_options + online_flag + ["list-updates"], stdout=subprocess.PIPE)
for line in (line.decode() for line in process.stdout):
# Zypper's output looks like:
#
# S | Repository | Name | Current Version | Available Version | Arch
# --+-------------------+--------------+------------------------------------+------------------------------------------------+-------
# v | Rudder repository | rudder-agent | 1398866025:3.2.6.release-1.SLES.11 | 1398866025:3.2.7.rc1.git201609190419-1.SLES.11 | x86_64
#
# Which gives:
#
# v | Rudder repository | rudder-agent | 1398866025:3.2.6.release-1.SLES.11 | 1398866025:3.2.7.rc1.git201609190419-1.SLES.11 | x86_64
# may contain package name old version, ignore it version available architecture
# special chars
# v\s+\|[^\|]+\ |\s+(?P\S+)\s+\|\s+\S+\s+\ |\s+(?P\S+)\s+\ |\s+(?P\S+)\s*$
# The first char will always be "v" which means there is a new version available on search outputs.
match = re.match("v\s+\|[^\|]+\|\s+(?P\S+)\s+\|\s+\S+\s+\|\s+(?P\S+)\s+\|\s+(?P\S+)\s*$", line)
if match is not None:
sys.stdout.write("Name=" + match.group("name") + "\n")
sys.stdout.write("Version=" + match.group("version") + "\n")
sys.stdout.write("Architecture=" + match.group("arch") + "\n")
return 0
# Returns a pair:
# List 1: Contains arguments for a single command line.
# List 2: Contains arguments for multiple command lines (see comments in
# repo_install()).
def one_package_argument(name, arch, version, is_zypper_install):
args = []
archs = []
exists = False
if arch:
archs.append(arch)
if is_zypper_install:
process = subprocess_Popen([rpm_cmd, "--qf", "%{arch}\n",
"-q", name], stdout=subprocess.PIPE)
existing_archs = [line.decode().rstrip() for line in process.stdout]
process.wait()
if process.returncode == 0 and existing_archs:
exists = True
if not arch:
# Here we have no specified architecture and we are
# installing. If we have existing versions, operate
# on those, instead of the platform default.
archs += existing_archs
version_suffix = ""
if version:
version_suffix = "=" + version
if archs:
args += [name + "." + arch + version_suffix for arch in archs]
else:
args.append(name + version_suffix)
if exists and version:
return [], args
else:
return args, []
# Returns a pair:
# List 1: Contains arguments for a single command line.
# List 2: Contains arguments for multiple command lines (see comments in
# repo_install()). This is a list of lists, where the logic is:
# list
# | +---- package1:amd64 -+
# +- sublist ---+ +--- Do these together
# | +---- package1:i386 -+
# |
# |
# | +---- package2:amd64 -+
# +- sublist ---+ +--- And these together
# +---- package2:i386 -+
def package_arguments_builder(is_zypper_install):
name = ""
version = ""
arch = ""
single_cmd_args = [] # List of arguments
multi_cmd_args = [] # List of lists of arguments
old_name = ""
for line in sys.stdin:
if line.startswith("Name="):
if name:
# Each new "Name=" triggers a new entry.
single_list, multi_list = one_package_argument(name, arch, version, is_zypper_install)
single_cmd_args += single_list
if name == old_name:
# Packages that differ only by architecture should be
# processed together
multi_cmd_args[-1] += multi_list
elif multi_list:
# Otherwise we process them individually.
multi_cmd_args += [multi_list]
version = ""
arch = ""
old_name = name
name = line.split("=", 1)[1].rstrip()
elif line.startswith("Version="):
version = line.split("=", 1)[1].rstrip()
elif line.startswith("Architecture="):
arch = line.split("=", 1)[1].rstrip()
if name:
single_list, multi_list = one_package_argument(name, arch, version, is_zypper_install)
single_cmd_args += single_list
if name == old_name:
# Packages that differ only by architecture should be
# processed together
multi_cmd_args[-1] += multi_list
elif multi_list:
# Otherwise we process them individually.
multi_cmd_args += [multi_list]
return single_cmd_args, multi_cmd_args
def repo_install():
# Due to how zypper works we need to split repo installs into several
# components.
#
# 1. Installation of fresh packages is easy, we add all of them on one
# command line.
# 2. Upgrade of existing packages where no version has been specified is
# also easy, we add that to the same command line.
# 3. Up/downgrade of existing packages where version is specified is
# tricky, for several reasons:
# a) There is no one zypper command that will do both, "install" or
# "upgrade" will only upgrade, and "downgrade" will only downgrade.
# b) There is no way rpm or zypper will tell you which version is higher
# than the other, and we know from experience with the old package
# promise implementation that we don't want to try to do such a
# comparison ourselves.
# c) zypper has no dry-run mode, so we cannot tell in advance which
# operation will succeed.
# d) zypper will not even tell you whether operation succeeded when you
# run it for real
#
# So here's what we need to do. We start by querying each package to find
# out whether that exact version is installed. If it fulfills 1. or 2. we
# add it to that single command line.
#
# If we end up at 3. we need to split the work and do each package
# separately. We do:
#
# 1. Try to upgrade using "zypper upgrade".
# 2. Query the package again, see if it is the right version now.
# 3. If not, try to downgrade using "zypper downgrade".
# 4. Query the package again, see if it is the right version now.
# 5. Final safeguard, try installing using "zypper install". This may happen
# in case we have one architecture already, but we are installing a
# second one. In this case only install will work.
# 6. (No need to check again, CFEngine will do the final check)
#
# This is considerably more expensive than what we do for apt, but it's the
# only way to cover all bases. In apt it will be one apt call for any number
# of packages, with zypper it will in the worst case be:
# 1 + 5 * number_of_packages
# although a more common case will probably be:
# 1 + 2 * number_of_packages
# since it's unlikely that people will do a whole lot of downgrades
# simultaneously.
single_cmd_args, multi_cmd_args = package_arguments_builder(True)
if single_cmd_args:
cmd_line = [zypper_cmd] + zypper_options + ["install"]
if zypper_supports_oldpackage():
cmd_line += ["--oldpackage"]
cmd_line.extend(single_cmd_args)
subprocess_call(cmd_line, stdout=NULLFILE, stderr=subprocess.PIPE)
if multi_cmd_args:
for block in multi_cmd_args:
# Try to upgrade.
cmd_line = [zypper_cmd] + zypper_options + ["update"] + block
subprocess_call(cmd_line, stdout=NULLFILE, stderr=subprocess.PIPE)
# See if it succeeded.
success = True
for item in block:
cmd_line = [rpm_cmd] + rpm_quiet_option + ["-q", item]
if subprocess_call(cmd_line, stdout=NULLFILE, stderr=subprocess.PIPE) != 0:
success = False
break
if success:
continue
# Try to plain install.
cmd_line = [zypper_cmd] + zypper_options + ["install"]
if zypper_supports_oldpackage():
cmd_line += ["--oldpackage"]
cmd_line += block
subprocess_call(cmd_line, stdout=NULLFILE, stderr=subprocess.PIPE)
# No final check. CFEngine will figure out that it's missing
# if it failed.
# return 0 doesn't mean we succeeded with everything, but it's expensive to
# check, so let CFEngine do that.
return 0
def remove():
cmd_line = [zypper_cmd] + zypper_options + ["remove"]
# package_arguments_builder will always return empty second element in case
# of removals, so just drop it. |
# V
args = package_arguments_builder(False)[0]
if args:
return subprocess_call(cmd_line + args, stdout=NULLFILE, stderr=subprocess.PIPE)
return 0
def file_install():
cmd_line = [rpm_cmd] + rpm_quiet_option + ["--force", "-U"]
found = False
for line in sys.stdin:
if line.startswith("File="):
found = True
cmd_line.append(line.split("=", 1)[1].rstrip())
if not found:
return 0
subprocess_call(cmd_line, stdout=NULLFILE, stderr=subprocess.PIPE)
return 0
def main():
if len(sys.argv) < 2:
sys.stderr.write("Need to provide argument\n")
return 2
if sys.argv[1] == "internal-test-stderr":
# This will cause an exception if stderr is closed.
try:
os.fstat(2)
except OSError:
return 1
return 0
elif sys.argv[1] == "supports-api-version":
sys.stdout.write("1\n")
return 0
elif sys.argv[1] == "get-package-data":
return get_package_data()
elif sys.argv[1] == "list-installed":
return list_installed()
elif sys.argv[1] == "list-updates":
return list_updates(True)
elif sys.argv[1] == "list-updates-local":
return list_updates(False)
elif sys.argv[1] == "repo-install":
return repo_install()
elif sys.argv[1] == "remove":
return remove()
elif sys.argv[1] == "file-install":
return file_install()
else:
sys.stderr.write("Invalid operation\n")
return 2
sys.exit(main())