Wala shell

Advanced PHP File Manager

Server IP: 127.0.0.1 | PHP Version: 8.3.30 | Domains Hosted: 54

Name	Size	Last Modified
📁 ..	-	2026-03-21 15:56
📁 __pycache__	-	2026-03-21 15:53
📄 builddep.py # builddep.py # Install all the deps needed to build this package. # # Copyright (C) 2013-2015 Red Hat, Inc. # Copyright (C) 2015 Igor Gnatenko # # This copyrighted material is made available to anyone wishing to use, # modify, copy, or redistribute it subject to the terms and conditions of # the GNU General Public License v.2, or (at your option) any later version. # This program is distributed in the hope that it will be useful, but WITHOUT # ANY WARRANTY expressed or implied, including the implied warranties 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, write to the # Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA # 02110-1301, USA. Any Red Hat trademarks that are incorporated in the # source code or documentation are not subject to the GNU General Public # License and may only be used or replicated with the express permission of # Red Hat, Inc. # from __future__ import absolute_import from __future__ import unicode_literals from dnfpluginscore import _, logger import argparse import dnf import dnf.cli import dnf.exceptions import dnf.rpm.transaction import dnf.yum.rpmtrans import libdnf.repo import os import rpm import shutil import tempfile @dnf.plugin.register_command class BuildDepCommand(dnf.cli.Command): aliases = ('builddep', 'build-dep') msg = "Install build dependencies for package or spec file" summary = _(msg) usage = _("[PACKAGE\|PACKAGE.spec]") def __init__(self, cli): super(BuildDepCommand, self).__init__(cli) self._rpm_ts = dnf.rpm.transaction.initReadOnlyTransaction() self.tempdirs = [] def __del__(self): for temp_dir in self.tempdirs: shutil.rmtree(temp_dir) def _download_remote_file(self, pkgspec): """ In case pkgspec is a remote URL, download it to a temporary location and use the temporary file instead. """ location = dnf.pycomp.urlparse.urlparse(pkgspec) if location[0] in ('file', ''): # just strip the file:// prefix return location.path downloader = libdnf.repo.Downloader() temp_dir = tempfile.mkdtemp(prefix="dnf_builddep_") temp_file = os.path.join(temp_dir, os.path.basename(pkgspec)) self.tempdirs.append(temp_dir) temp_fo = open(temp_file, "wb+") try: downloader.downloadURL(self.base.conf._config, pkgspec, temp_fo.fileno()) except RuntimeError as ex: raise finally: temp_fo.close() return temp_file @staticmethod def set_argparser(parser): def macro_def(arg): arglist = arg.split(None, 1) if arg else [] if len(arglist) < 2: msg = _("'%s' is not of the format 'MACRO EXPR'") % arg raise argparse.ArgumentTypeError(msg) return arglist parser.add_argument('packages', nargs='+', metavar='package', help=_('packages with builddeps to install')) parser.add_argument('-D', '--define', action='append', default=[], metavar="'MACRO EXPR'", type=macro_def, help=_('define a macro for spec file parsing')) parser.add_argument('--skip-unavailable', action='store_true', default=False, help=_('skip build dependencies not available in repositories')) ptype = parser.add_mutually_exclusive_group() ptype.add_argument('--spec', action='store_true', help=_('treat commandline arguments as spec files')) ptype.add_argument('--srpm', action='store_true', help=_('treat commandline arguments as source rpm')) def pre_configure(self): if not self.opts.rpmverbosity: self.opts.rpmverbosity = 'error' def configure(self): demands = self.cli.demands demands.available_repos = True demands.resolving = True demands.root_user = True demands.sack_activation = True # enable source repos only if needed if not (self.opts.spec or self.opts.srpm): for pkgspec in self.opts.packages: if not (pkgspec.endswith('.src.rpm') or pkgspec.endswith('.nosrc.rpm') or pkgspec.endswith('.spec')): self.base.repos.enable_source_repos() break def run(self): rpmlog = dnf.yum.rpmtrans.RPMTransaction(self.base) # Push user-supplied macro definitions for spec parsing for macro in self.opts.define: rpm.addMacro(macro[0], macro[1]) pkg_errors = False for pkgspec in self.opts.packages: pkgspec = self._download_remote_file(pkgspec) try: if self.opts.srpm: self._src_deps(pkgspec) elif self.opts.spec: self._spec_deps(pkgspec) elif pkgspec.endswith('.src.rpm') or pkgspec.endswith('nosrc.rpm'): self._src_deps(pkgspec) elif pkgspec.endswith('.spec'): self._spec_deps(pkgspec) else: self._remote_deps(pkgspec) except dnf.exceptions.Error as e: for line in rpmlog.messages(): logger.error(_("RPM: {}").format(line)) logger.error(e) pkg_errors = True # Pop user macros so they don't affect future rpm calls for macro in self.opts.define: rpm.delMacro(macro[0]) if pkg_errors: raise dnf.exceptions.Error(_("Some packages could not be found.")) @staticmethod def _rpm_dep2reldep_str(rpm_dep): return rpm_dep.DNEVR()[2:] def _install(self, reldep_str): # Try to find something by provides sltr = dnf.selector.Selector(self.base.sack) sltr.set(provides=reldep_str) found = sltr.matches() if not found and reldep_str.startswith("/"): # Nothing matches by provides and since it's file, try by files sltr = dnf.selector.Selector(self.base.sack) sltr.set(file=reldep_str) found = sltr.matches() if not found and not reldep_str.startswith("("): # No provides, no files # Richdeps can have no matches but it could be correct (solver must decide later) msg = _("No matching package to install: '%s'") logger.warning(msg, reldep_str) return self.opts.skip_unavailable is True if found: already_inst = self.base._sltr_matches_installed(sltr) if already_inst: for package in already_inst: dnf.base._msg_installed(package) self.base._goal.install(select=sltr, optional=False) return True def _src_deps(self, src_fn): fd = os.open(src_fn, os.O_RDONLY) try: h = self._rpm_ts.hdrFromFdno(fd) except rpm.error as e: if str(e) == 'error reading package header': e = _("Failed to open: '%s', not a valid source rpm file.") % src_fn os.close(fd) raise dnf.exceptions.Error(e) os.close(fd) ds = h.dsFromHeader('requirename') done = True for dep in ds: reldep_str = self._rpm_dep2reldep_str(dep) if reldep_str.startswith('rpmlib('): continue done &= self._install(reldep_str) if not done: err = _("Not all dependencies satisfied") raise dnf.exceptions.Error(err) if self.opts.define: logger.warning(_("Warning: -D or --define arguments have no meaning " "for source rpm packages.")) def _spec_deps(self, spec_fn): try: spec = rpm.spec(spec_fn) except ValueError as ex: msg = _("Failed to open: '%s', not a valid spec file: %s") % ( spec_fn, ex) raise dnf.exceptions.Error(msg) done = True for dep in rpm.ds(spec.sourceHeader, 'requires'): reldep_str = self._rpm_dep2reldep_str(dep) done &= self._install(reldep_str) if not done: err = _("Not all dependencies satisfied") raise dnf.exceptions.Error(err) def _remote_deps(self, package): available = dnf.subject.Subject(package).get_best_query( self.base.sack).filter(arch__neq="src") sourcenames = list({pkg.source_name for pkg in available}) pkgs = self.base.sack.query().available().filter( name=(sourcenames + [package]), arch="src").latest().run() if not pkgs: raise dnf.exceptions.Error(_('no package matched: %s') % package) done = True for pkg in pkgs: for req in pkg.requires: done &= self._install(str(req)) if not done: err = _("Not all dependencies satisfied") raise dnf.exceptions.Error(err)	9.13 KB	2022-09-09 11:42
📄 changelog.py # changelog.py # DNF plugin adding a command changelog. # # Copyright (C) 2014 Red Hat, Inc. # # This copyrighted material is made available to anyone wishing to use, # modify, copy, or redistribute it subject to the terms and conditions of # the GNU General Public License v.2, or (at your option) any later version. # This program is distributed in the hope that it will be useful, but WITHOUT # ANY WARRANTY expressed or implied, including the implied warranties 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, write to the # Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA # 02110-1301, USA. Any Red Hat trademarks that are incorporated in the # source code or documentation are not subject to the GNU General Public # License and may only be used or replicated with the express permission of # Red Hat, Inc. # from __future__ import absolute_import from __future__ import unicode_literals import argparse import collections import dateutil.parser from dnfpluginscore import _, P_, logger import dnf import dnf.cli def validate_date(val): try: return dateutil.parser.parse(val, fuzzy=True) except (ValueError, TypeError, OverflowError): raise argparse.ArgumentTypeError(_('Not a valid date: "{0}".').format(val)) @dnf.plugin.register_command class ChangelogCommand(dnf.cli.Command): aliases = ('changelog',) summary = _('Show changelog data of packages') @staticmethod def set_argparser(parser): filter_group = parser.add_mutually_exclusive_group() filter_group.add_argument( '--since', metavar="DATE", default=None, type=validate_date, help=_('show changelog entries since DATE. To avoid ambiguosity, ' 'YYYY-MM-DD format is recommended.')) filter_group.add_argument( '--count', default=None, type=int, help=_('show given number of changelog entries per package')) filter_group.add_argument( '--upgrades', default=False, action='store_true', help=_('show only new changelog entries for packages, that provide an ' 'upgrade for some of already installed packages.')) parser.add_argument("package", nargs='*', metavar=_('PACKAGE')) def configure(self): demands = self.cli.demands demands.available_repos = True demands.sack_activation = True demands.changelogs = True def query(self): q = self.base.sack.query() if self.opts.package: q.filterm(empty=True) for pkg in self.opts.package: pkg_q = dnf.subject.Subject(pkg, ignore_case=True).get_best_query( self.base.sack, with_nevra=True, with_provides=False, with_filenames=False) if self.opts.repo: pkg_q.filterm(reponame=self.opts.repo) if pkg_q: q = q.union(pkg_q.latest()) else: logger.info(_('No match for argument: %s') % pkg) elif self.opts.repo: q.filterm(reponame=self.opts.repo) if self.opts.upgrades: q = q.upgrades() else: q = q.available() return q def by_srpm(self, packages): by_srpm = collections.OrderedDict() for pkg in sorted(packages): by_srpm.setdefault((pkg.source_name or pkg.name, pkg.evr), []).append(pkg) return by_srpm def filter_changelogs(self, package): if self.opts.upgrades: return self.base.latest_changelogs(package) elif self.opts.count: return package.changelogs[:self.opts.count] elif self.opts.since: return [chlog for chlog in package.changelogs if chlog['timestamp'] >= self.opts.since.date()] else: return package.changelogs def run(self): if self.opts.since: logger.info(_('Listing changelogs since {}').format(self.opts.since)) elif self.opts.count: logger.info(P_('Listing only latest changelog', 'Listing {} latest changelogs', self.opts.count).format(self.opts.count)) elif self.opts.upgrades: logger.info( _('Listing only new changelogs since installed version of the package')) else: logger.info(_('Listing all changelogs')) by_srpm = self.by_srpm(self.query()) for name in by_srpm: print(_('Changelogs for {}').format( ', '.join(sorted({str(pkg) for pkg in by_srpm[name]})))) for chlog in self.filter_changelogs(by_srpm[name][0]): print(self.base.format_changelog(chlog))	4.85 KB	2022-09-09 11:42
📄 config_manager.py # # Copyright (C) 2015 Red Hat, Inc. # # This copyrighted material is made available to anyone wishing to use, # modify, copy, or redistribute it subject to the terms and conditions of # the GNU General Public License v.2, or (at your option) any later version. # This program is distributed in the hope that it will be useful, but WITHOUT # ANY WARRANTY expressed or implied, including the implied warranties 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, write to the # Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA # 02110-1301, USA. Any Red Hat trademarks that are incorporated in the # source code or documentation are not subject to the GNU General Public # License and may only be used or replicated with the express permission of # Red Hat, Inc. # from __future__ import absolute_import from __future__ import unicode_literals from dnfpluginscore import _, logger, P_ import dnf import dnf.cli import dnf.pycomp import dnf.util import fnmatch import hashlib import os import re import shutil @dnf.plugin.register_command class ConfigManagerCommand(dnf.cli.Command): aliases = ['config-manager'] summary = _('manage {prog} configuration options and repositories').format( prog=dnf.util.MAIN_PROG) @staticmethod def set_argparser(parser): parser.add_argument( 'crepo', nargs='', metavar='repo', help=_('repo to modify')) parser.add_argument( '--save', default=False, action='store_true', help=_('save the current options (useful with --setopt)')) parser.add_argument( '--add-repo', default=[], action='append', metavar='URL', help=_('add (and enable) the repo from the specified file or url')) parser.add_argument( '--dump', default=False, action='store_true', help=_('print current configuration values to stdout')) parser.add_argument( '--dump-variables', default=False, action='store_true', help=_('print variable values to stdout')) enable_group = parser.add_mutually_exclusive_group() enable_group.add_argument("--set-enabled", default=False, dest="set_enabled", action="store_true", help=_("enable repos (automatically saves)")) enable_group.add_argument("--set-disabled", default=False, dest="set_disabled", action="store_true", help=_("disable repos (automatically saves)")) def configure(self): # setup sack and populate it with enabled repos demands = self.cli.demands demands.available_repos = True # if no argument was passed then error if (not (self.opts.add_repo != [] or self.opts.save or self.opts.dump or self.opts.dump_variables or self.opts.set_disabled or self.opts.set_enabled) ): self.cli.optparser.error(_("one of the following arguments is required: {}") .format(' '.join([ "--save", "--add-repo", "--dump", "--dump-variables", "--set-enabled", "--enable", "--set-disabled", "--disable"]))) # warn with hint if --enablerepo or --disablerepo argument was passed if self.opts.repos_ed != []: logger.warning(_("Warning: --enablerepo/--disablerepo arguments have no meaning" "with config manager. Use --set-enabled/--set-disabled instead.")) if (self.opts.save or self.opts.set_enabled or self.opts.set_disabled or self.opts.add_repo): demands.root_user = True # sanitize commas https://bugzilla.redhat.com/show_bug.cgi?id=1830530 temp_list = [x.split(',') for x in self.opts.crepo if x != ','] # flatten sublists self.opts.crepo = [item for sublist in temp_list for item in sublist if item != ''] def run(self): """Execute the util action here.""" if self.opts.add_repo: self.add_repo() else: self.modify_repo() def modify_repo(self): """ process --set-enabled, --set-disabled and --setopt options """ matching_repos = [] # list of matched repositories not_matching_repos_id = set() # IDs of not matched repositories def match_repos(key, add_matching_repos): matching = self.base.repos.get_matching(key) if not matching: not_matching_repos_id.add(name) elif add_matching_repos: matching_repos.extend(matching) if self.opts.crepo: for name in self.opts.crepo: match_repos(name, True) if hasattr(self.opts, 'repo_setopts'): for name in self.opts.repo_setopts.keys(): match_repos(name, False) else: if hasattr(self.opts, 'repo_setopts'): for name in self.opts.repo_setopts.keys(): match_repos(name, True) if not_matching_repos_id: raise dnf.exceptions.Error(_("No matching repo to modify: %s.") % ', '.join(not_matching_repos_id)) sbc = self.base.conf modify = {} if hasattr(self.opts, 'main_setopts') and self.opts.main_setopts: modify = self.opts.main_setopts if self.opts.dump_variables: for name, val in self.base.conf.substitutions.items(): print("%s = %s" % (name, val)) if not self.opts.crepo or 'main' in self.opts.crepo: if self.opts.save and modify: # modify [main] in global configuration file self.base.conf.write_raw_configfile(self.base.conf.config_file_path, 'main', sbc.substitutions, modify) if self.opts.dump: print(self.base.output.fmtSection('main')) print(self.base.conf.dump()) if not matching_repos: return if self.opts.set_enabled or self.opts.set_disabled: self.opts.save = True for repo in sorted(matching_repos): repo_modify = {} if self.opts.set_enabled: repo_modify['enabled'] = "1" elif self.opts.set_disabled: repo_modify['enabled'] = "0" if hasattr(self.opts, 'repo_setopts'): for repoid, setopts in self.opts.repo_setopts.items(): if fnmatch.fnmatch(repo.id, repoid): repo_modify.update(setopts) if self.opts.save and repo_modify: self.base.conf.write_raw_configfile(repo.repofile, repo.id, sbc.substitutions, repo_modify) if self.opts.dump: print(self.base.output.fmtSection('repo: ' + repo.id)) print(repo.dump()) def add_repo(self): """ process --add-repo option """ # Get the reposdir location myrepodir = self.base.conf.get_reposdir errors_count = 0 for url in self.opts.add_repo: if dnf.pycomp.urlparse.urlparse(url).scheme == '': url = 'file://' + os.path.abspath(url) logger.info(_('Adding repo from: %s'), url) if url.endswith('.repo'): # .repo file - download, put into reposdir and enable it destname = os.path.basename(url) destname = os.path.join(myrepodir, destname) try: f = self.base.urlopen(url, mode='w+') shutil.copy2(f.name, destname) os.chmod(destname, 0o644) f.close() except IOError as e: errors_count += 1 logger.error(e) continue else: # just url to repo, create .repo file on our own repoid = sanitize_url_to_fs(url) reponame = 'created by {} config-manager from {}'.format(dnf.util.MAIN_PROG, url) destname = os.path.join(myrepodir, "%s.repo" % repoid) content = "[%s]\nname=%s\nbaseurl=%s\nenabled=1\n" % \ (repoid, reponame, url) if not save_to_file(destname, content): continue if errors_count: raise dnf.exceptions.Error(P_("Configuration of repo failed", "Configuration of repos failed", errors_count)) def save_to_file(filename, content): try: with open(filename, 'w+') as fd: dnf.pycomp.write_to_file(fd, content) os.chmod(filename, 0o644) except (IOError, OSError) as e: logger.error(_('Could not save repo to repofile %s: %s'), filename, e) return False return True # Regular expressions to sanitise cache filenames RE_SCHEME = re.compile(r'^\w+:/(\w+:\|www\.)?') RE_SLASH = re.compile(r'[?/:&#\|~\\[\]\'\\]+') RE_BEGIN = re.compile(r'^[,.]') RE_FINAL = re.compile(r'[,.]*$') def sanitize_url_to_fs(url): """Return a filename suitable for the filesystem and for repo id Strips dangerous and common characters to create a filename we can use to store the cache in. """ try: if RE_SCHEME.match(url): if dnf.pycomp.PY3: url = url.encode('idna').decode('utf-8') else: if isinstance(url, str): url = url.decode('utf-8').encode('idna') else: url = url.encode('idna') if isinstance(url, unicode): url = url.encode('utf-8') except (UnicodeDecodeError, UnicodeEncodeError, UnicodeError, TypeError): pass url = RE_SCHEME.sub("", url) url = RE_SLASH.sub("_", url) url = RE_BEGIN.sub("", url) url = RE_FINAL.sub("", url) # limit length of url if len(url) > 250: parts = url[:185].split('_') lastindex = 185-len(parts[-1]) csum = hashlib.sha256() csum.update(url[lastindex:].encode('utf-8')) url = url[:lastindex] + '_' + csum.hexdigest() # remove all not allowed characters allowed_regex = "[^abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789_.:-]" return re.sub(allowed_regex, '', url)	10.63 KB	2022-09-09 11:42
📄 copr.py # supplies the 'copr' command. # # Copyright (C) 2014-2015 Red Hat, Inc. # # This copyrighted material is made available to anyone wishing to use, # modify, copy, or redistribute it subject to the terms and conditions of # the GNU General Public License v.2, or (at your option) any later version. # This program is distributed in the hope that it will be useful, but WITHOUT # ANY WARRANTY expressed or implied, including the implied warranties 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, write to the # Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA # 02110-1301, USA. Any Red Hat trademarks that are incorporated in the # source code or documentation are not subject to the GNU General Public # License and may only be used or replicated with the express permission of # Red Hat, Inc. # from __future__ import print_function import glob import itertools import json import os import re import shutil import stat import sys import base64 import json from dnfpluginscore import _, logger import dnf from dnf.pycomp import PY3 from dnf.i18n import ucd import rpm # Attempt importing the linux_distribution function from distro # If that fails, attempt to import the deprecated implementation # from the platform module. try: from distro import name, version, codename, os_release_attr # Re-implement distro.linux_distribution() to avoid a deprecation warning def linux_distribution(): return (name(), version(), codename()) except ImportError: def os_release_attr(_): return "" try: from platform import linux_distribution except ImportError: # Simple fallback for distributions that lack an implementation def linux_distribution(): with open('/etc/os-release') as os_release_file: os_release_data = {} for line in os_release_file: try: os_release_key, os_release_value = line.rstrip().split('=') os_release_data[os_release_key] = os_release_value.strip('"') except ValueError: # Skip empty lines and everything that is not a simple # variable assignment pass return (os_release_data['NAME'], os_release_data['VERSION_ID'], None) PLUGIN_CONF = 'copr' YES = set([_('yes'), _('y')]) NO = set([_('no'), _('n'), '']) if PY3: from configparser import ConfigParser, NoOptionError, NoSectionError from urllib.request import urlopen, HTTPError, URLError else: from ConfigParser import ConfigParser, NoOptionError, NoSectionError from urllib2 import urlopen, HTTPError, URLError @dnf.plugin.register_command class CoprCommand(dnf.cli.Command): """ Copr plugin for DNF """ chroot_config = None default_hostname = "copr.fedorainfracloud.org" default_hub = "fedora" default_protocol = "https" default_port = 443 default_url = default_protocol + "://" + default_hostname aliases = ("copr",) summary = _("Interact with Copr repositories.") first_warning = True usage = _(""" enable name/project [chroot] disable name/project remove name/project list --installed/enabled/disabled list --available-by-user=NAME search project Examples: copr enable rhscl/perl516 epel-6-x86_64 copr enable ignatenkobrain/ocltoys copr disable rhscl/perl516 copr remove rhscl/perl516 copr list --enabled copr list --available-by-user=ignatenkobrain copr search tests """) @staticmethod def set_argparser(parser): parser.add_argument('subcommand', nargs=1, choices=['help', 'enable', 'disable', 'remove', 'list', 'search']) list_option = parser.add_mutually_exclusive_group() list_option.add_argument('--installed', action='store_true', help=_('List all installed Copr repositories (default)')) list_option.add_argument('--enabled', action='store_true', help=_('List enabled Copr repositories')) list_option.add_argument('--disabled', action='store_true', help=_('List disabled Copr repositories')) list_option.add_argument('--available-by-user', metavar='NAME', help=_('List available Copr repositories by user NAME')) parser.add_argument('--hub', help=_('Specify an instance of Copr to work with')) parser.add_argument('arg', nargs='') def configure(self): if self.cli.command.opts.command != "copr": return copr_hub = None copr_plugin_config = ConfigParser() config_files = [] config_path = self.base.conf.pluginconfpath[0] default_config_file = os.path.join(config_path, PLUGIN_CONF + ".conf") if os.path.isfile(default_config_file): config_files.append(default_config_file) copr_plugin_config.read(default_config_file) if copr_plugin_config.has_option('main', 'distribution') and\ copr_plugin_config.has_option('main', 'releasever'): distribution = copr_plugin_config.get('main', 'distribution') releasever = copr_plugin_config.get('main', 'releasever') self.chroot_config = [distribution, releasever] else: self.chroot_config = [False, False] for filename in os.listdir(os.path.join(config_path, PLUGIN_CONF + ".d")): if filename.endswith('.conf'): config_file = os.path.join(config_path, PLUGIN_CONF + ".d", filename) config_files.append(config_file) project = [] if len(self.opts.arg): project = self.opts.arg[0].split("/") if len(project) == 3 and self.opts.hub: logger.critical( _('Error: ') + _('specify Copr hub either with `--hub` or using ' '`copr_hub/copr_username/copr_projectname` format') ) raise dnf.cli.CliError(_('multiple hubs specified')) # Copr hub was not specified, using default hub `fedora` elif not self.opts.hub and len(project) != 3: self.copr_hostname = self.default_hostname self.copr_url = self.default_url # Copr hub specified with hub/user/project format elif len(project) == 3: copr_hub = project[0] else: copr_hub = self.opts.hub # Try to find hub in a config file if config_files and copr_hub: self.copr_url = None copr_plugin_config.read(sorted(config_files, reverse=True)) hostname = self._read_config_item(copr_plugin_config, copr_hub, 'hostname', None) if hostname: protocol = self._read_config_item(copr_plugin_config, copr_hub, 'protocol', self.default_protocol) port = self._read_config_item(copr_plugin_config, copr_hub, 'port', self.default_port) self.copr_hostname = hostname self.copr_url = protocol + "://" + hostname if int(port) != self.default_port: self.copr_url += ":" + port self.copr_hostname += ":" + port if not self.copr_url: if '://' not in copr_hub: self.copr_hostname = copr_hub self.copr_url = self.default_protocol + "://" + copr_hub else: self.copr_hostname = copr_hub.split('://', 1)[1] self.copr_url = copr_hub def _read_config_item(self, config, hub, section, default): try: return config.get(hub, section) except (NoOptionError, NoSectionError): return default def _user_warning_before_prompt(self, text): sys.stderr.write("{0}\n".format(text.strip())) def run(self): subcommand = self.opts.subcommand[0] if subcommand == "help": self.cli.optparser.print_help(self) return 0 if subcommand == "list": if self.opts.available_by_user: self._list_user_projects(self.opts.available_by_user) return else: self._list_installed_repositories(self.base.conf.reposdir[0], self.opts.enabled, self.opts.disabled) return try: project_name = self.opts.arg[0] except (ValueError, IndexError): logger.critical( _('Error: ') + _('exactly two additional parameters to ' 'copr command are required')) self.cli.optparser.print_help(self) raise dnf.cli.CliError( _('exactly two additional parameters to ' 'copr command are required')) try: chroot = self.opts.arg[1] if len(self.opts.arg) > 2: raise dnf.exceptions.Error(_('Too many arguments.')) self.chroot_parts = chroot.split("-") if len(self.chroot_parts) < 3: raise dnf.exceptions.Error(_('Bad format of optional chroot. The format is ' 'distribution-version-architecture.')) except IndexError: chroot = self._guess_chroot() self.chroot_parts = chroot.split("-") # commands without defined copr_username/copr_projectname if subcommand == "search": self._search(project_name) return project = project_name.split("/") if len(project) not in [2, 3]: logger.critical( _('Error: ') + _('use format `copr_username/copr_projectname` ' 'to reference copr project')) raise dnf.cli.CliError(_('bad copr project format')) elif len(project) == 2: copr_username = project[0] copr_projectname = project[1] else: copr_username = project[1] copr_projectname = project[2] project_name = copr_username + "/" + copr_projectname repo_filename = "{0}/_copr:{1}:{2}:{3}.repo".format( self.base.conf.get_reposdir, self.copr_hostname, self._sanitize_username(copr_username), copr_projectname) if subcommand == "enable": self._need_root() info = _(""" Enabling a Copr repository. Please note that this repository is not part of the main distribution, and quality may vary. The Fedora Project does not exercise any power over the contents of this repository beyond the rules outlined in the Copr FAQ at <https://docs.pagure.org/copr.copr/user_documentation.html#what-i-can-build-in-copr>, and packages are not held to any quality or security level. Please do not file bug reports about these packages in Fedora Bugzilla. In case of problems, contact the owner of this repository. """) project = '/'.join([self.copr_hostname, copr_username, copr_projectname]) msg = "Do you really want to enable {0}?".format(project) self._ask_user(info, msg) self._download_repo(project_name, repo_filename) logger.info(_("Repository successfully enabled.")) self._runtime_deps_warning(copr_username, copr_projectname) elif subcommand == "disable": self._need_root() self._disable_repo(copr_username, copr_projectname) logger.info(_("Repository successfully disabled.")) elif subcommand == "remove": self._need_root() self._remove_repo(copr_username, copr_projectname) logger.info(_("Repository successfully removed.")) else: raise dnf.exceptions.Error( _('Unknown subcommand {}.').format(subcommand)) def _list_repo_file(self, repo_id, repo, enabled_only, disabled_only): file_name = repo.repofile.split('/')[-1] match_new = re.match("_copr:" + self.copr_hostname, file_name) match_old = self.copr_url == self.default_url and re.match("_copr_", file_name) match_any = re.match("_copr:\|^_copr_", file_name) if self.opts.hub: if not match_new and not match_old: return elif not match_any: return if re.match('copr:.:.:.:ml', repo_id): # We skip multilib repositories return if re.match('coprdep:.', repo_id): # Runtime dependencies are not listed. return enabled = repo.enabled if (enabled and disabled_only) or (not enabled and enabled_only): return old_repo = False # repo ID has copr:<hostname>:<user>:<copr_dir> format, while <copr_dir> # can contain more colons if re.match("copr:", repo_id): _, copr_hostname, copr_owner, copr_dir = repo_id.split(':', 3) msg = copr_hostname + '/' + copr_owner + "/" + copr_dir # repo ID has <user>-<project> format, try to get hub from file name elif re.match("_copr:", file_name): copr_name = repo_id.split('-', 1) copr_hostname = file_name.rsplit(':', 2)[0].split(':', 1)[1] msg = copr_hostname + '/' + copr_name[0] + '/' + copr_name[1] # no information about hub, assume the default one else: copr_name = repo_id.split('-', 1) msg = self.default_hostname + '/' + copr_name[0] + '/' + copr_name[1] old_repo = True if not enabled: msg += " (disabled)" if old_repo: msg += " " print(msg) return old_repo def _list_installed_repositories(self, directory, enabled_only, disabled_only): old_repo = False for repo_id, repo in self.base.repos.items(): if self._list_repo_file(repo_id, repo, enabled_only, disabled_only): old_repo = True if old_repo: print(_("* These coprs have repo file with an old format that contains " "no information about Copr hub - the default one was assumed. " "Re-enable the project to fix this.")) def _list_user_projects(self, user_name): # https://copr.fedorainfracloud.org/api_3/project/list?ownername=ignatenkobrain api_path = "/api_3/project/list?ownername={0}".format(user_name) url = self.copr_url + api_path res = self.base.urlopen(url, mode='w+') try: json_parse = json.loads(res.read()) except ValueError: raise dnf.exceptions.Error( _("Can't parse repositories for username '{}'.") .format(user_name)) self._check_json_output(json_parse) section_text = _("List of {} coprs").format(user_name) self._print_match_section(section_text) for item in json_parse["items"]: msg = "{0}/{1} : ".format(user_name, item["name"]) desc = item["description"] or _("No description given") msg = self.base.output.fmtKeyValFill(ucd(msg), desc) print(msg) def _search(self, query): # https://copr.fedorainfracloud.org/api_3/project/search?query=tests api_path = "/api_3/project/search?query={}".format(query) url = self.copr_url + api_path res = self.base.urlopen(url, mode='w+') try: json_parse = json.loads(res.read()) except ValueError: raise dnf.exceptions.Error(_("Can't parse search for '{}'." ).format(query)) self._check_json_output(json_parse) section_text = _("Matched: {}").format(query) self._print_match_section(section_text) for item in json_parse["items"]: msg = "{0} : ".format(item["full_name"]) desc = item["description"] or _("No description given.") msg = self.base.output.fmtKeyValFill(ucd(msg), desc) print(msg) def _print_match_section(self, text): formatted = self.base.output.fmtSection(text) print(formatted) def _ask_user_no_raise(self, info, msg): if not self.first_warning: sys.stderr.write("\n") self.first_warning = False sys.stderr.write("{0}\n".format(info.strip())) if self.base._promptWanted(): if self.base.conf.assumeno or not self.base.output.userconfirm( msg='\n{} [y/N]: '.format(msg), defaultyes_msg='\n{} [Y/n]: '.format(msg)): return False return True def _ask_user(self, info, msg): if not self._ask_user_no_raise(info, msg): raise dnf.exceptions.Error(_('Safe and good answer. Exiting.')) @classmethod def _need_root(cls): # FIXME this should do dnf itself (BZ#1062889) if os.geteuid() != 0: raise dnf.exceptions.Error( _('This command has to be run under the root user.')) def _guess_chroot(self): """ Guess which chroot is equivalent to this machine """ # FIXME Copr should generate non-specific arch repo dist = self.chroot_config if dist is None or (dist[0] is False) or (dist[1] is False): dist = linux_distribution() # Get distribution architecture distarch = self.base.conf.substitutions['basearch'] if "Fedora" in dist[0]: if "Rawhide" in dist: chroot = ("fedora-rawhide-" + distarch) # workaround for enabling repos in Rawhide when VERSION in os-release # contains a name other than Rawhide elif "rawhide" in os_release_attr("redhat_support_product_version"): chroot = ("fedora-rawhide-" + distarch) else: chroot = ("fedora-{0}-{1}".format(dist[1], distarch)) elif "Mageia" in dist: # Get distribution architecture (Mageia does not use $basearch) distarch = rpm.expandMacro("%{distro_arch}") # Set the chroot if "Cauldron" in dist: chroot = ("mageia-cauldron-{}".format(distarch)) else: chroot = ("mageia-{0}-{1}".format(dist[1], distarch)) elif "openSUSE" in dist: # Get distribution architecture (openSUSE does not use $basearch) distarch = rpm.expandMacro("%{_target_cpu}") # Set the chroot if "Tumbleweed" in dist: chroot = ("opensuse-tumbleweed-{}".format(distarch)) else: chroot = ("opensuse-leap-{0}-{1}".format(dist[1], distarch)) else: chroot = ("epel-{}-{}".format(dist[1].split(".", 1)[0], distarch if distarch else "x86_64")) return chroot def _download_repo(self, project_name, repo_filename): short_chroot = '-'.join(self.chroot_parts[:-1]) arch = self.chroot_parts[-1] api_path = "/coprs/{0}/repo/{1}/dnf.repo?arch={2}".format(project_name, short_chroot, arch) try: response = urlopen(self.copr_url + api_path) if os.path.exists(repo_filename): os.remove(repo_filename) except HTTPError as e: if e.code != 404: error_msg = _("Request to {0} failed: {1} - {2}").format(self.copr_url + api_path, e.code, str(e)) raise dnf.exceptions.Error(error_msg) error_msg = _("It wasn't possible to enable this project.\n") error_data = e.headers.get("Copr-Error-Data") if error_data: error_data_decoded = base64.b64decode(error_data).decode('utf-8') error_data_decoded = json.loads(error_data_decoded) error_msg += _("Repository '{0}' does not exist in project '{1}'.").format( '-'.join(self.chroot_parts), project_name) if error_data_decoded.get("available chroots"): error_msg += _("\nAvailable repositories: ") + ', '.join( "'{}'".format(x) for x in error_data_decoded["available chroots"]) error_msg += _("\n\nIf you want to enable a non-default repository, use the following command:\n" " 'dnf copr enable {0} <repository>'\n" "But note that the installed repo file will likely need a manual " "modification.").format(project_name) raise dnf.exceptions.Error(error_msg) else: error_msg += _("Project {0} does not exist.").format(project_name) raise dnf.exceptions.Error(error_msg) except URLError as e: error_msg = _("Failed to connect to {0}: {1}").format(self.copr_url + api_path, e.reason.strerror) raise dnf.exceptions.Error(error_msg) # Try to read the first line, and detect the repo_filename from that (override the repo_filename value). first_line = response.readline() line = first_line.decode("utf-8") if re.match(r"\[copr:", line): repo_filename = os.path.join(self.base.conf.get_reposdir, "_" + line[1:-2] + ".repo") # if using default hub, remove possible old repofile if self.copr_url == self.default_url: # copr:hub:user:project.repo => _copr_user_project.repo old_repo_filename = repo_filename.replace("_copr:", "_copr", 1)\ .replace(self.copr_hostname, "").replace(":", "_", 1).replace(":", "-")\ .replace("group_", "@") if os.path.exists(old_repo_filename): os.remove(old_repo_filename) with open(repo_filename, 'wb') as f: f.write(first_line) for line in response.readlines(): f.write(line) os.chmod(repo_filename, stat.S_IRUSR \| stat.S_IWUSR \| stat.S_IRGRP \| stat.S_IROTH) def _runtime_deps_warning(self, copr_username, copr_projectname): """ In addition to the main copr repo (that has repo ID prefixed with `copr:`), the repofile might contain additional repositories that serve as runtime dependencies. This method informs the user about the additional repos and provides an option to disable them. """ self.base.reset(repos=True) self.base.read_all_repos() repo = self._get_copr_repo(self._sanitize_username(copr_username), copr_projectname) runtime_deps = [] for repo_id in repo.cfg.sections(): if repo_id.startswith("copr:"): continue runtime_deps.append(repo_id) if not runtime_deps: return info = _( "Maintainer of the enabled Copr repository decided to make\n" "it dependent on other repositories. Such repositories are\n" "usually necessary for successful installation of RPMs from\n" "the main Copr repository (they provide runtime dependencies).\n\n" "Be aware that the note about quality and bug-reporting\n" "above applies here too, Fedora Project doesn't control the\n" "content. Please review the list:\n\n" "{0}\n\n" "These repositories have been enabled automatically." ) counter = itertools.count(1) info = info.format("\n\n".join([ "{num:2}. [{repoid}]\n baseurl={baseurl}".format( num=next(counter), repoid=repoid, baseurl=repo.cfg.getValue(repoid, "baseurl")) for repoid in runtime_deps ])) if not self._ask_user_no_raise(info, _("Do you want to keep them enabled?")): for dep in runtime_deps: self.base.conf.write_raw_configfile(repo.repofile, dep, self.base.conf.substitutions, {"enabled": "0"}) def _get_copr_repo(self, copr_username, copr_projectname): repo_id = "copr:{0}:{1}:{2}".format(self.copr_hostname.rsplit(':', 1)[0], self._sanitize_username(copr_username), copr_projectname) if repo_id not in self.base.repos: # check if there is a repo with old ID format repo_id = repo_id = "{0}-{1}".format(self._sanitize_username(copr_username), copr_projectname) if repo_id in self.base.repos and "_copr" in self.base.repos[repo_id].repofile: file_name = self.base.repos[repo_id].repofile.split('/')[-1] try: copr_hostname = file_name.rsplit(':', 2)[0].split(':', 1)[1] if copr_hostname != self.copr_hostname: return None except IndexError: # old filename format without hostname pass else: return None return self.base.repos[repo_id] def _remove_repo(self, copr_username, copr_projectname): # FIXME is it Copr repo ? repo = self._get_copr_repo(copr_username, copr_projectname) if not repo: raise dnf.exceptions.Error( _("Failed to remove copr repo {0}/{1}/{2}" .format(self.copr_hostname, copr_username, copr_projectname))) try: os.remove(repo.repofile) except OSError as e: raise dnf.exceptions.Error(str(e)) def _disable_repo(self, copr_username, copr_projectname): repo = self._get_copr_repo(copr_username, copr_projectname) if repo is None: raise dnf.exceptions.Error( _("Failed to disable copr repo {}/{}" .format(copr_username, copr_projectname))) # disable all repos provided by the repo file for repo_id in repo.cfg.sections(): self.base.conf.write_raw_configfile(repo.repofile, repo_id, self.base.conf.substitutions, {"enabled": "0"}) @classmethod def _get_data(cls, f): """ Wrapper around response from server check data and print nice error in case of some error (and return None) otherwise return json object. """ try: output = json.loads(f.read()) except ValueError: dnf.cli.CliError(_("Unknown response from server.")) return return output @classmethod def _check_json_output(cls, json_obj): if "error" in json_obj: raise dnf.exceptions.Error("{}".format(json_obj["error"])) @classmethod def _sanitize_username(cls, copr_username): if copr_username[0] == "@": return "group_{}".format(copr_username[1:]) else: return copr_username @dnf.plugin.register_command class PlaygroundCommand(CoprCommand): """ Playground plugin for DNF """ aliases = ("playground",) summary = _("Interact with Playground repository.") usage = " [enable\|disable\|upgrade]" def _cmd_enable(self, chroot): self._need_root() self._ask_user( _("Enabling a Playground repository."), _("Do you want to continue?"), ) api_url = "{0}/api/playground/list/".format( self.copr_url) f = self.base.urlopen(api_url, mode="w+") output = self._get_data(f) f.close() if output["output"] != "ok": raise dnf.cli.CliError(_("Unknown response from server.")) for repo in output["repos"]: project_name = "{0}/{1}".format(repo["username"], repo["coprname"]) repo_filename = "{}/_playground_{}.repo".format(self.base.conf.get_reposdir, project_name.replace("/", "-")) try: if chroot not in repo["chroots"]: continue api_url = "{0}/api/coprs/{1}/detail/{2}/".format( self.copr_url, project_name, chroot) f = self.base.urlopen(api_url, mode='w+') output2 = self._get_data(f) f.close() if (output2 and ("output" in output2) and (output2["output"] == "ok")): self._download_repo(project_name, repo_filename) except dnf.exceptions.Error: # likely 404 and that repo does not exist pass def _cmd_disable(self): self._need_root() for repo_filename in glob.glob("{}/_playground_*.repo".format(self.base.conf.get_reposdir)): self._remove_repo(repo_filename) @staticmethod def set_argparser(parser): parser.add_argument('subcommand', nargs=1, choices=['enable', 'disable', 'upgrade']) def run(self): raise dnf.exceptions.Error("Playground is temporarily unsupported") subcommand = self.opts.subcommand[0] chroot = self._guess_chroot() if subcommand == "enable": self._cmd_enable(chroot) logger.info(_("Playground repositories successfully enabled.")) elif subcommand == "disable": self._cmd_disable() logger.info(_("Playground repositories successfully disabled.")) elif subcommand == "upgrade": self._cmd_disable() self._cmd_enable(chroot) logger.info(_("Playground repositories successfully updated.")) else: raise dnf.exceptions.Error( _('Unknown subcommand {}.').format(subcommand))	29.59 KB	2025-11-12 04:27
📄 debug.py # # Copyright (C) 2015 Red Hat, Inc. # # This copyrighted material is made available to anyone wishing to use, # modify, copy, or redistribute it subject to the terms and conditions of # the GNU General Public License v.2, or (at your option) any later version. # This program is distributed in the hope that it will be useful, but WITHOUT # ANY WARRANTY expressed or implied, including the implied warranties 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, write to the # Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA # 02110-1301, USA. Any Red Hat trademarks that are incorporated in the # source code or documentation are not subject to the GNU General Public # License and may only be used or replicated with the express permission of # Red Hat, Inc. # from __future__ import absolute_import from __future__ import unicode_literals from dnf.i18n import ucd from dnfpluginscore import _, logger import dnf import dnf.cli import gzip import hawkey import os import rpm import sys import time DEBUG_VERSION = "dnf-debug-dump version 1\n" class Debug(dnf.Plugin): name = 'debug' def __init__(self, base, cli): super(Debug, self).__init__(base, cli) self.base = base self.cli = cli if self.cli is not None: self.cli.register_command(DebugDumpCommand) self.cli.register_command(DebugRestoreCommand) class DebugDumpCommand(dnf.cli.Command): aliases = ("debug-dump",) summary = _("dump information about installed rpm packages to file") def __init__(self, cli): super(DebugDumpCommand, self).__init__(cli) self.dump_file = None def configure(self): self.cli.demands.sack_activation = True self.cli.demands.available_repos = True @staticmethod def set_argparser(parser): parser.add_argument( "--norepos", action="store_true", default=False, help=_("do not attempt to dump the repository contents.")) parser.add_argument( "filename", nargs="?", help=_("optional name of dump file")) def run(self): """create debug txt file and compress it, if no filename specified use dnf_debug_dump-<timestamp>.txt.gz by default""" filename = self.opts.filename if not filename: now = time.strftime("%Y-%m-%d_%T", time.localtime(time.time())) filename = "dnf_debug_dump-%s-%s.txt.gz" % (os.uname()[1], now) filename = os.path.abspath(filename) if filename.endswith(".gz"): self.dump_file = gzip.GzipFile(filename, "w") else: self.dump_file = open(filename, "w") self.write(DEBUG_VERSION) self.dump_system_info() self.dump_dnf_config_info() self.dump_rpm_problems() self.dump_packages(not self.opts.norepos) self.dump_rpmdb_versions() self.dump_file.close() print(_("Output written to: %s") % filename) def write(self, msg): if dnf.pycomp.PY3 and isinstance(self.dump_file, gzip.GzipFile): msg = bytes(msg, "utf8") dnf.pycomp.write_to_file(self.dump_file, msg) def dump_system_info(self): self.write("%%%%SYSTEM INFO\n") uname = os.uname() self.write(" uname: %s, %s\n" % (uname[2], uname[4])) self.write(" rpm ver: %s\n" % rpm.__version__) self.write(" python ver: %s\n" % sys.version.replace("\n", "")) return def dump_dnf_config_info(self): var = self.base.conf.substitutions plugins = ",".join([p.name for p in self.base._plugins.plugins]) self.write("%%%%DNF INFO\n") self.write(" arch: %s\n" % var["arch"]) self.write(" basearch: %s\n" % var["basearch"]) self.write(" releasever: %s\n" % var["releasever"]) self.write(" dnf ver: %s\n" % dnf.const.VERSION) self.write(" enabled plugins: %s\n" % plugins) self.write(" global excludes: %s\n" % ",".join(self.base.conf.excludepkgs)) return def dump_rpm_problems(self): self.write("%%%%RPMDB PROBLEMS\n") (missing, conflicts) = rpm_problems(self.base) self.write("".join(["Package %s requires %s\n" % (ucd(pkg), ucd(req)) for (req, pkg) in missing])) self.write("".join(["Package %s conflicts with %s\n" % (ucd(pkg), ucd(conf)) for (conf, pkg) in conflicts])) def dump_packages(self, load_repos): q = self.base.sack.query() # packages from rpmdb self.write("%%%%RPMDB\n") for p in sorted(q.installed()): self.write(" %s\n" % pkgspec(p)) if not load_repos: return self.write("%%%%REPOS\n") available = q.available() for repo in sorted(self.base.repos.iter_enabled(), key=lambda x: x.id): try: url = None if repo.metalink is not None: url = repo.metalink elif repo.mirrorlist is not None: url = repo.mirrorlist elif len(repo.baseurl) > 0: url = repo.baseurl[0] self.write("%%%s - %s\n" % (repo.id, url)) self.write(" excludes: %s\n" % ",".join(repo.excludepkgs)) for po in sorted(available.filter(reponame=repo.id)): self.write(" %s\n" % pkgspec(po)) except dnf.exceptions.Error as e: self.write("Error accessing repo %s: %s\n" % (repo, str(e))) continue return def dump_rpmdb_versions(self): self.write("%%%%RPMDB VERSIONS\n") version = self.base._ts.dbCookie() self.write(" all: %s\n" % version) return class DebugRestoreCommand(dnf.cli.Command): aliases = ("debug-restore",) summary = _("restore packages recorded in debug-dump file") def configure(self): self.cli.demands.sack_activation = True self.cli.demands.available_repos = True self.cli.demands.root_user = True if not self.opts.output: self.cli.demands.resolving = True @staticmethod def set_argparser(parser): parser.add_argument( "--output", action="store_true", help=_("output commands that would be run to stdout.")) parser.add_argument( "--install-latest", action="store_true", help=_("Install the latest version of recorded packages.")) parser.add_argument( "--ignore-arch", action="store_true", help=_("Ignore architecture and install missing packages matching " "the name, epoch, version and release.")) parser.add_argument( "--filter-types", metavar="[install, remove, replace]", default="install, remove, replace", help=_("limit to specified type")) parser.add_argument( "--remove-installonly", action="store_true", help=_('Allow removing of install-only packages. Using this option may ' 'result in an attempt to remove the running kernel.')) parser.add_argument( "filename", nargs=1, help=_("name of dump file")) def run(self): """Execute the command action here.""" if self.opts.filter_types: self.opts.filter_types = set( self.opts.filter_types.replace(",", " ").split()) dump_pkgs = self.read_dump_file(self.opts.filename[0]) self.process_installed(dump_pkgs, self.opts) self.process_dump(dump_pkgs, self.opts) def process_installed(self, dump_pkgs, opts): installed = self.base.sack.query().installed() installonly_pkgs = self.base._get_installonly_query(installed) for pkg in installed: pkg_remove = False spec = pkgspec(pkg) dumped_versions = dump_pkgs.get((pkg.name, pkg.arch), None) if dumped_versions is not None: evr = (pkg.epoch, pkg.version, pkg.release) if evr in dumped_versions: # the correct version is already installed dumped_versions[evr] = 'skip' else: # other version is currently installed if pkg in installonly_pkgs: # package is install-only, should be removed pkg_remove = True else: # package should be upgraded / downgraded if "replace" in opts.filter_types: action = 'replace' else: action = 'skip' for d_evr in dumped_versions.keys(): dumped_versions[d_evr] = action else: # package should not be installed pkg_remove = True if pkg_remove and "remove" in opts.filter_types: if pkg not in installonly_pkgs or opts.remove_installonly: if opts.output: print("remove %s" % spec) else: self.base.package_remove(pkg) def process_dump(self, dump_pkgs, opts): for (n, a) in sorted(dump_pkgs.keys()): dumped_versions = dump_pkgs[(n, a)] for (e, v, r) in sorted(dumped_versions.keys()): action = dumped_versions[(e, v, r)] if action == 'skip': continue if opts.ignore_arch: arch = "" else: arch = "." + a if opts.install_latest and action == "install": pkg_spec = "%s%s" % (n, arch) else: pkg_spec = pkgtup2spec(n, arch, e, v, r) if action in opts.filter_types: if opts.output: print("%s %s" % (action, pkg_spec)) else: try: self.base.install(pkg_spec) except dnf.exceptions.MarkingError: logger.error(_("Package %s is not available"), pkg_spec) @staticmethod def read_dump_file(filename): if filename.endswith(".gz"): fobj = gzip.GzipFile(filename) else: fobj = open(filename) if ucd(fobj.readline()) != DEBUG_VERSION: logger.error(_("Bad dnf debug file: %s"), filename) raise dnf.exceptions.Error skip = True pkgs = {} for line in fobj: line = ucd(line) if skip: if line == "%%%%RPMDB\n": skip = False continue if not line or line[0] != " ": break pkg_spec = line.strip() nevra = hawkey.split_nevra(pkg_spec) # {(name, arch): {(epoch, version, release): action}} pkgs.setdefault((nevra.name, nevra.arch), {})[ (nevra.epoch, nevra.version, nevra.release)] = "install" return pkgs def rpm_problems(base): rpmdb = dnf.sack._rpmdb_sack(base) allpkgs = rpmdb.query().installed() requires = set() conflicts = set() for pkg in allpkgs: requires.update([(req, pkg) for req in pkg.requires if not str(req) == "solvable:prereqmarker" and not str(req).startswith("rpmlib(")]) conflicts.update([(conf, pkg) for conf in pkg.conflicts]) missing_requires = [(req, pkg) for (req, pkg) in requires if not allpkgs.filter(provides=req)] existing_conflicts = [(conf, pkg) for (conf, pkg) in conflicts if allpkgs.filter(provides=conf)] return missing_requires, existing_conflicts def pkgspec(pkg): return pkgtup2spec(pkg.name, pkg.arch, pkg.epoch, pkg.version, pkg.release) def pkgtup2spec(name, arch, epoch, version, release): a = "" if not arch else ".%s" % arch.lstrip('.') e = "" if epoch in (None, "") else "%s:" % epoch return "%s-%s%s-%s%s" % (name, e, version, release, a)	12.26 KB	2022-09-09 11:42
📄 debuginfo-install.py # debuginfo-install.py # Install the debuginfo of packages and their dependencies to debug this package. # # Copyright (C) 2014 Igor Gnatenko # Copyright (C) 2014-2019 Red Hat # # This copyrighted material is made available to anyone wishing to use, # modify, copy, or redistribute it subject to the terms and conditions of # the GNU General Public License v.2, or (at your option) any later version. # This program is distributed in the hope that it will be useful, but WITHOUT # ANY WARRANTY expressed or implied, including the implied warranties 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, write to the # Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA # 02110-1301, USA. Any Red Hat trademarks that are incorporated in the # source code or documentation are not subject to the GNU General Public # License and may only be used or replicated with the express permission of # Red Hat, Inc. # from dnfpluginscore import _, logger import dnf from dnf.package import Package class DebuginfoInstall(dnf.Plugin): """DNF plugin supplying the 'debuginfo-install' command.""" name = 'debuginfo-install' def __init__(self, base, cli): """Initialize the plugin instance.""" super(DebuginfoInstall, self).__init__(base, cli) self.base = base self.cli = cli if cli is not None: cli.register_command(DebuginfoInstallCommand) def config(self): cp = self.read_config(self.base.conf) autoupdate = (cp.has_section('main') and cp.has_option('main', 'autoupdate') and cp.getboolean('main', 'autoupdate')) if autoupdate: # allow update of already installed debuginfo packages dbginfo = dnf.sack._rpmdb_sack(self.base).query().filterm(name__glob="-debuginfo") if len(dbginfo): self.base.repos.enable_debug_repos() class DebuginfoInstallCommand(dnf.cli.Command): """ DebuginfoInstall plugin for DNF """ aliases = ("debuginfo-install",) summary = _('install debuginfo packages') def __init__(self, cli): super(DebuginfoInstallCommand, self).__init__(cli) self.available_debuginfo_missing = set() self.available_debugsource_missing = set() self.installed_debuginfo_missing = set() self.installed_debugsource_missing = set() @staticmethod def set_argparser(parser): parser.add_argument('package', nargs='+') def configure(self): demands = self.cli.demands demands.resolving = True demands.root_user = True demands.sack_activation = True demands.available_repos = True self.base.repos.enable_debug_repos() def run(self): errors_spec = [] debuginfo_suffix_len = len(Package.DEBUGINFO_SUFFIX) debugsource_suffix_len = len(Package.DEBUGSOURCE_SUFFIX) for pkgspec in self.opts.package: solution = dnf.subject.Subject(pkgspec).get_best_solution(self.base.sack, with_src=False) query = solution["query"] if not query: logger.info(_('No match for argument: %s'), self.base.output.term.bold(pkgspec)) errors_spec.append(pkgspec) continue package_dict = query.available()._name_dict() # installed versions of packages have priority, replace / add them to the dict package_dict.update(query.installed()._name_dict()) # Remove debuginfo packages if their base packages are in the query. # They can get there through globs and they break the installation # of debug packages with the same version as the installed base # packages. If the base package of a debuginfo package is not in # the query, the user specified a debug package on the command # line. We don't want to ignore those, so we will install them. # But, in this case the version will not be matched to the # installed version of the base package, as that would require # another query and is further complicated if the user specifies a # version themselves etc. for name in list(package_dict.keys()): if name.endswith(Package.DEBUGINFO_SUFFIX): if name[:-debuginfo_suffix_len] in package_dict: package_dict.pop(name) if name.endswith(Package.DEBUGSOURCE_SUFFIX): if name[:-debugsource_suffix_len] in package_dict: package_dict.pop(name) # attempt to install debuginfo and debugsource for the highest # listed version of the package (in case the package is installed, # only the installed version is listed) for pkgs in package_dict.values(): first_pkg = pkgs[0] # for packages from system (installed) there can be more # packages with different architectures listed and we want to # install debuginfo for all of them if first_pkg._from_system: # we need to split them by architectures and install the # latest version for each architecture arch_dict = {} for pkg in pkgs: arch_dict.setdefault(pkg.arch, []).append(pkg) for package_arch_list in arch_dict.values(): pkg = package_arch_list[0] if not self._install_debug_from_system(pkg.debug_name, pkg): if not self._install_debug_from_system(pkg.source_debug_name, pkg): self.installed_debuginfo_missing.add(str(pkg)) if not self._install_debug_from_system(pkg.debugsource_name, pkg): self.installed_debugsource_missing.add(str(pkg)) continue # if the package in question is -debuginfo or -debugsource, install it directly if first_pkg.name.endswith(Package.DEBUGINFO_SUFFIX) \ or first_pkg.name.endswith(Package.DEBUGSOURCE_SUFFIX): self._install(pkgs) # pass all pkgs to the solver, it will pick the best one continue # if we have NEVRA parsed from the pkgspec, use it to install the package if solution["nevra"] is not None: if not self._install_debug(first_pkg.debug_name, solution["nevra"]): if not self._install_debug(first_pkg.source_debug_name, solution["nevra"]): self.available_debuginfo_missing.add( "{}-{}".format(first_pkg.name, first_pkg.evr)) if not self._install_debug(first_pkg.debugsource_name, solution["nevra"]): self.available_debugsource_missing.add( "{}-{}".format(first_pkg.name, first_pkg.evr)) continue # if we don't have NEVRA from the pkgspec, pass nevras from # all packages that were found (while replacing the name with # the -debuginfo and -debugsource variant) to the solver, which # will pick the correct version and architecture if not self._install_debug_no_nevra(first_pkg.debug_name, pkgs): if not self._install_debug_no_nevra(first_pkg.source_debug_name, pkgs): self.available_debuginfo_missing.add( "{}-{}".format(first_pkg.name, first_pkg.evr)) if not self._install_debug_no_nevra(first_pkg.debugsource_name, pkgs): self.available_debugsource_missing.add( "{}-{}".format(first_pkg.name, first_pkg.evr)) if self.available_debuginfo_missing: logger.info( _("Could not find debuginfo package for the following available packages: %s"), ", ".join(sorted(self.available_debuginfo_missing))) if self.available_debugsource_missing: logger.info( _("Could not find debugsource package for the following available packages: %s"), ", ".join(sorted(self.available_debugsource_missing))) if self.installed_debuginfo_missing: logger.info( _("Could not find debuginfo package for the following installed packages: %s"), ", ".join(sorted(self.installed_debuginfo_missing))) if self.installed_debugsource_missing: logger.info( _("Could not find debugsource package for the following installed packages: %s"), ", ".join(sorted(self.installed_debugsource_missing))) if errors_spec and self.base.conf.strict: raise dnf.exceptions.PackagesNotAvailableError(_("Unable to find a match"), pkg_spec=' '.join(errors_spec)) def _install_debug_from_system(self, debug_name, pkg): query = self.base.sack.query().filter(name=debug_name, epoch=pkg.epoch, version=pkg.version, release=pkg.release, arch=pkg.arch) if query: self._install(query) return True return False def _install_debug(self, debug_name, base_nevra): kwargs = {} # if some part of EVRA was specified in the argument, add it as a filter if base_nevra.epoch is not None: kwargs["epoch__glob"] = base_nevra.epoch if base_nevra.version is not None: kwargs["version__glob"] = base_nevra.version if base_nevra.release is not None: kwargs["release__glob"] = base_nevra.release if base_nevra.arch is not None: kwargs["arch__glob"] = base_nevra.arch query = self.base.sack.query().filter(name=debug_name, *kwargs) if query: self._install(query) return True return False def _install_debug_no_nevra(self, debug_name, pkgs): query = self.base.sack.query().filterm( nevra_strict=["{}-{}.{}".format(debug_name, p.evr, p.arch) for p in pkgs]) if query: self._install(query) return True return False def _install(self, pkgs): selector = dnf.selector.Selector(self.base.sack) selector.set(pkg=pkgs) self.base.goal.install(select=selector, optional=not self.base.conf.strict)	10.82 KB	2022-09-09 11:42
📄 download.py # download.py, supplies the 'download' command. # # Copyright (C) 2013-2015 Red Hat, Inc. # # This copyrighted material is made available to anyone wishing to use, # modify, copy, or redistribute it subject to the terms and conditions of # the GNU General Public License v.2, or (at your option) any later version. # This program is distributed in the hope that it will be useful, but WITHOUT # ANY WARRANTY expressed or implied, including the implied warranties 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, write to the # Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA # 02110-1301, USA. Any Red Hat trademarks that are incorporated in the # source code or documentation are not subject to the GNU General Public # License and may only be used or replicated with the express permission of # Red Hat, Inc. # from __future__ import absolute_import from __future__ import unicode_literals from dnfpluginscore import _, logger from dnf.cli.option_parser import OptionParser import dnf import dnf.cli import dnf.exceptions import dnf.i18n import dnf.subject import dnf.util import hawkey import itertools import os import shutil @dnf.plugin.register_command class DownloadCommand(dnf.cli.Command): aliases = ['download'] summary = _('Download package to current directory') def __init__(self, cli): super(DownloadCommand, self).__init__(cli) self.opts = None self.parser = None @staticmethod def set_argparser(parser): parser.add_argument('packages', nargs='+', help=_('packages to download')) parser.add_argument("--source", action='store_true', help=_('download the src.rpm instead')) parser.add_argument("--debuginfo", action='store_true', help=_('download the -debuginfo package instead')) parser.add_argument("--debugsource", action='store_true', help=_('download the -debugsource package instead')) parser.add_argument("--arch", '--archlist', dest='arches', default=[], action=OptionParser._SplitCallback, metavar='[arch]', help=_("limit the query to packages of given architectures.")) parser.add_argument('--resolve', action='store_true', help=_('resolve and download needed dependencies')) parser.add_argument('--alldeps', action='store_true', help=_('when running with --resolve, download all dependencies ' '(do not exclude already installed ones)')) parser.add_argument('--url', '--urls', action='store_true', dest='url', help=_('print list of urls where the rpms ' 'can be downloaded instead of downloading')) parser.add_argument('--urlprotocols', action='append', choices=['http', 'https', 'rsync', 'ftp'], default=[], help=_('when running with --url, ' 'limit to specific protocols')) def configure(self): # setup sack and populate it with enabled repos demands = self.cli.demands demands.sack_activation = True demands.available_repos = True if self.opts.resolve and self.opts.alldeps: demands.load_system_repo = False if self.opts.source: self.base.repos.enable_source_repos() if self.opts.debuginfo or self.opts.debugsource: self.base.repos.enable_debug_repos() if self.opts.destdir: self.base.conf.destdir = self.opts.destdir else: self.base.conf.destdir = dnf.i18n.ucd(os.getcwd()) def run(self): """Execute the util action here.""" if (not self.opts.source and not self.opts.debuginfo and not self.opts.debugsource): pkgs = self._get_pkg_objs_rpms(self.opts.packages) else: pkgs = [] if self.opts.source: pkgs.extend(self._get_pkg_objs_source(self.opts.packages)) if self.opts.debuginfo: pkgs.extend(self._get_pkg_objs_debuginfo(self.opts.packages)) if self.opts.debugsource: pkgs.extend(self._get_pkg_objs_debugsource(self.opts.packages)) # If user asked for just urls then print them and we're done if self.opts.url: for pkg in pkgs: # command line repo packages do not have .remote_location if pkg.repoid != hawkey.CMDLINE_REPO_NAME: url = pkg.remote_location(schemes=self.opts.urlprotocols) if url: print(url) else: msg = _("Failed to get mirror for package: %s") % pkg.name if self.base.conf.strict: raise dnf.exceptions.Error(msg) logger.warning(msg) return else: self._do_downloads(pkgs) # download rpms def _do_downloads(self, pkgs): """ Perform the download for a list of packages """ pkg_dict = {} for pkg in pkgs: pkg_dict.setdefault(str(pkg), []).append(pkg) to_download = [] cmdline = [] for pkg_list in pkg_dict.values(): pkgs_cmdline = [pkg for pkg in pkg_list if pkg.repoid == hawkey.CMDLINE_REPO_NAME] if pkgs_cmdline: cmdline.append(pkgs_cmdline[0]) continue pkg_list.sort(key=lambda x: (x.repo.priority, x.repo.cost)) to_download.append(pkg_list[0]) if to_download: self.base.download_packages(to_download, self.base.output.progress) if cmdline: # command line repo packages are either local files or already downloaded urls # just copy them to the destination for pkg in cmdline: # python<3.4 shutil module does not raise SameFileError, check manually src = pkg.localPkg() dst = os.path.join(self.base.conf.destdir, os.path.basename(src)) if os.path.exists(dst) and os.path.samefile(src, dst): continue shutil.copy(src, self.base.conf.destdir) locations = sorted([pkg.localPkg() for pkg in to_download + cmdline]) return locations def _get_pkg_objs_rpms(self, pkg_specs): """ Return a list of dnf.Package objects that represent the rpms to download. """ if self.opts.resolve: pkgs = self._get_packages_with_deps(pkg_specs) else: pkgs = self._get_packages(pkg_specs) return pkgs def _get_pkg_objs_source(self, pkg_specs): """ Return a list of dnf.Package objects that represent the source rpms to download. """ pkgs = self._get_pkg_objs_rpms(pkg_specs) source_pkgs = self._get_source_packages(pkgs) pkgs = set(self._get_packages(source_pkgs, source=True)) return pkgs def _get_pkg_objs_debuginfo(self, pkg_specs): """ Return a list of dnf.Package objects that represent the debuginfo rpms to download. """ dbg_pkgs = set() q = self.base.sack.query().available() for pkg in self._get_packages(pkg_specs): for dbg_name in [pkg.debug_name, pkg.source_debug_name]: dbg_available = q.filter( name=dbg_name, epoch=int(pkg.epoch), version=pkg.version, release=pkg.release, arch=pkg.arch ) if not dbg_available: continue for p in dbg_available: dbg_pkgs.add(p) break return dbg_pkgs def _get_pkg_objs_debugsource(self, pkg_specs): """ Return a list of dnf.Package objects that represent the debugsource rpms to download. """ dbg_pkgs = set() q = self.base.sack.query().available() for pkg in self._get_packages(pkg_specs): dbg_available = q.filter( name=pkg.debugsource_name, epoch=int(pkg.epoch), version=pkg.version, release=pkg.release, arch=pkg.arch ) for p in dbg_available: dbg_pkgs.add(p) return dbg_pkgs def _get_packages(self, pkg_specs, source=False): """Get packages matching pkg_specs.""" func = self._get_query_source if source else self._get_query queries = [] for pkg_spec in pkg_specs: try: queries.append(func(pkg_spec)) except dnf.exceptions.PackageNotFoundError as e: logger.error(dnf.i18n.ucd(e)) if self.base.conf.strict: logger.error(_("Exiting due to strict setting.")) raise dnf.exceptions.Error(e) pkgs = list(itertools.chain(*queries)) return pkgs def _get_packages_with_deps(self, pkg_specs, source=False): """Get packages matching pkg_specs and the deps.""" pkgs = self._get_packages(pkg_specs) pkg_set = set(pkgs) for pkg in pkgs: goal = hawkey.Goal(self.base.sack) goal.install(pkg) rc = goal.run() if rc: pkg_set.update(goal.list_installs()) pkg_set.update(goal.list_upgrades()) else: msg = [_('Error in resolve of packages:')] logger.error("\n ".join(msg + [str(pkg) for pkg in pkgs])) logger.error(dnf.util._format_resolve_problems(goal.problem_rules())) raise dnf.exceptions.Error() return pkg_set @staticmethod def _get_source_packages(pkgs): """Get list of source rpm names for a list of packages.""" source_pkgs = set() for pkg in pkgs: if pkg.sourcerpm: source_pkgs.add(pkg.sourcerpm) logger.debug(' --> Package : %s Source : %s', str(pkg), pkg.sourcerpm) elif pkg.arch == 'src': source_pkgs.add("%s-%s.src.rpm" % (pkg.name, pkg.evr)) else: logger.info(_("No source rpm defined for %s"), str(pkg)) return list(source_pkgs) def _get_query(self, pkg_spec): """Return a query to match a pkg_spec.""" schemes = dnf.pycomp.urlparse.urlparse(pkg_spec)[0] is_url = schemes and schemes in ('http', 'ftp', 'file', 'https') if is_url or (pkg_spec.endswith('.rpm') and os.path.isfile(pkg_spec)): pkgs = self.base.add_remote_rpms([pkg_spec], progress=self.base.output.progress) return self.base.sack.query().filterm(pkg=pkgs) subj = dnf.subject.Subject(pkg_spec) q = subj.get_best_query(self.base.sack, with_src=self.opts.source) q = q.available() q = q.filterm(latest_per_arch_by_priority=True) if self.opts.arches: q = q.filter(arch=self.opts.arches) if len(q.run()) == 0: msg = _("No package %s available.") % (pkg_spec) raise dnf.exceptions.PackageNotFoundError(msg) return q def _get_query_source(self, pkg_spec): """Return a query to match a source rpm file name.""" pkg_spec = pkg_spec[:-4] # skip the .rpm subj = dnf.subject.Subject(pkg_spec) for nevra_obj in subj.get_nevra_possibilities(): tmp_query = nevra_obj.to_query(self.base.sack).available() if tmp_query: return tmp_query.latest() msg = _("No package %s available.") % (pkg_spec) raise dnf.exceptions.PackageNotFoundError(msg)	12.04 KB	2022-09-09 11:42
📄 generate_completion_cache.py # coding=utf-8 # generate_completion_cache.py - generate cache for dnf bash completion # Copyright © 2013 Elad Alfassa <elad@fedoraproject.org> # Copyright (C) 2014-2015 Igor Gnatenko <i.gnatenko.brain@gmail.com> # Copyright (C) 2015 Red Hat, Inc. # This copyrighted material is made available to anyone wishing to use, # modify, copy, or redistribute it subject to the terms and conditions of # the GNU General Public License v.2, or (at your option) any later version. # This program is distributed in the hope that it will be useful, but WITHOUT # ANY WARRANTY expressed or implied, including the implied warranties 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, write to the # Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA # 02110-1301, USA. from __future__ import absolute_import from __future__ import unicode_literals from dnf.i18n import ucd from dnfpluginscore import logger import dnf import os.path import sqlite3 class BashCompletionCache(dnf.Plugin): name = 'generate_completion_cache' def __init__(self, base, cli): super(BashCompletionCache, self).__init__(base, cli) self.base = base self.cache_file = "/var/cache/dnf/packages.db" @staticmethod def _out(msg): logger.debug('Completion plugin: %s', msg) def sack(self): ''' Generate cache of available packages ''' # We generate this cache only if the repos were just freshed or if the # cache file doesn't exist fresh = False for repo in self.base.repos.iter_enabled(): if repo.metadata is not None and repo.metadata.fresh: # One fresh repo is enough to cause a regen of the cache fresh = True break if not os.path.exists(self.cache_file) or fresh: try: with sqlite3.connect(self.cache_file) as conn: self._out('Generating completion cache...') cur = conn.cursor() cur.execute( "create table if not exists available (pkg TEXT)") cur.execute( "create unique index if not exists " "pkg_available ON available(pkg)") cur.execute("delete from available") avail_pkgs = self.base.sack.query().available() avail_pkgs_insert = [[str(x)] for x in avail_pkgs if x.arch != "src"] cur.executemany("insert or ignore into available values (?)", avail_pkgs_insert) conn.commit() except sqlite3.OperationalError as e: self._out("Can't write completion cache: %s" % ucd(e)) def transaction(self): ''' Generate cache of installed packages ''' if not self.transaction: return try: with sqlite3.connect(self.cache_file) as conn: self._out('Generating completion cache...') cur = conn.cursor() cur.execute("create table if not exists installed (pkg TEXT)") cur.execute( "create unique index if not exists " "pkg_installed ON installed(pkg)") cur.execute("delete from installed") inst_pkgs = dnf.sack._rpmdb_sack(self.base).query().installed() inst_pkgs_insert = [[str(x)] for x in inst_pkgs if x.arch != "src"] cur.executemany("insert or ignore into installed values (?)", inst_pkgs_insert) conn.commit() except sqlite3.OperationalError as e: self._out("Can't write completion cache: %s" % ucd(e))	3.86 KB	2022-09-09 11:42
📄 groups_manager.py # groups_manager.py # DNF plugin for managing comps groups metadata files # # Copyright (C) 2020 Red Hat, Inc. # # This copyrighted material is made available to anyone wishing to use, # modify, copy, or redistribute it subject to the terms and conditions of # the GNU General Public License v.2, or (at your option) any later version. # This program is distributed in the hope that it will be useful, but WITHOUT # ANY WARRANTY expressed or implied, including the implied warranties 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, write to the # Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA # 02110-1301, USA. Any Red Hat trademarks that are incorporated in the # source code or documentation are not subject to the GNU General Public # License and may only be used or replicated with the express permission of # Red Hat, Inc. # from __future__ import absolute_import from __future__ import unicode_literals import argparse import gzip import libcomps import os import re import shutil import tempfile from dnfpluginscore import _, logger import dnf import dnf.cli RE_GROUP_ID_VALID = '-a-z0-9_.:' RE_GROUP_ID = re.compile(r'^[{}]+$'.format(RE_GROUP_ID_VALID)) RE_LANG = re.compile(r'^[-a-zA-Z0-9_.@]+$') COMPS_XML_OPTIONS = { 'default_explicit': True, 'uservisible_explicit': True, 'empty_groups': True} def group_id_type(value): '''group id validator''' if not RE_GROUP_ID.match(value): raise argparse.ArgumentTypeError(_('Invalid group id')) return value def translation_type(value): '''translated texts validator''' data = value.split(':', 2) if len(data) != 2: raise argparse.ArgumentTypeError( _("Invalid translated data, should be in form 'lang:text'")) lang, text = data if not RE_LANG.match(lang): raise argparse.ArgumentTypeError(_('Invalid/empty language for translated data')) return lang, text def text_to_id(text): '''generate group id based on its name''' group_id = text.lower() group_id = re.sub('[^{}]'.format(RE_GROUP_ID_VALID), '', group_id) if not group_id: raise dnf.cli.CliError( _("Can't generate group id from '{}'. Please specify group id using --id.").format( text)) return group_id @dnf.plugin.register_command class GroupsManagerCommand(dnf.cli.Command): aliases = ('groups-manager',) summary = _('create and edit groups metadata file') def __init__(self, cli): super(GroupsManagerCommand, self).__init__(cli) self.comps = libcomps.Comps() @staticmethod def set_argparser(parser): # input / output options parser.add_argument('--load', action='append', default=[], metavar='COMPS.XML', help=_('load groups metadata from file')) parser.add_argument('--save', action='append', default=[], metavar='COMPS.XML', help=_('save groups metadata to file')) parser.add_argument('--merge', metavar='COMPS.XML', help=_('load and save groups metadata to file')) parser.add_argument('--print', action='store_true', default=False, help=_('print the result metadata to stdout')) # group options parser.add_argument('--id', type=group_id_type, help=_('group id')) parser.add_argument('-n', '--name', help=_('group name')) parser.add_argument('--description', help=_('group description')) parser.add_argument('--display-order', type=int, help=_('group display order')) parser.add_argument('--translated-name', action='append', default=[], metavar='LANG:TEXT', type=translation_type, help=_('translated name for the group')) parser.add_argument('--translated-description', action='append', default=[], metavar='LANG:TEXT', type=translation_type, help=_('translated description for the group')) visible = parser.add_mutually_exclusive_group() visible.add_argument('--user-visible', dest='user_visible', action='store_true', default=None, help=_('make the group user visible (default)')) visible.add_argument('--not-user-visible', dest='user_visible', action='store_false', default=None, help=_('make the group user invisible')) # package list options section = parser.add_mutually_exclusive_group() section.add_argument('--mandatory', action='store_true', help=_('add packages to the mandatory section')) section.add_argument('--optional', action='store_true', help=_('add packages to the optional section')) section.add_argument('--remove', action='store_true', default=False, help=_('remove packages from the group instead of adding them')) parser.add_argument('--dependencies', action='store_true', help=_('include also direct dependencies for packages')) parser.add_argument("packages", nargs='*', metavar='PACKAGE', help=_('package specification')) def configure(self): demands = self.cli.demands if self.opts.packages: demands.sack_activation = True demands.available_repos = True demands.load_system_repo = False # handle --merge option (shortcut to --load and --save the same file) if self.opts.merge: self.opts.load.insert(0, self.opts.merge) self.opts.save.append(self.opts.merge) # check that group is specified when editing is attempted if (self.opts.description or self.opts.display_order or self.opts.translated_name or self.opts.translated_description or self.opts.user_visible is not None or self.opts.packages): if not self.opts.id and not self.opts.name: raise dnf.cli.CliError( _("Can't edit group without specifying it (use --id or --name)")) def load_input_files(self): """ Loads all input xml files. Returns True if at least one file was successfuly loaded """ for file_name in self.opts.load: file_comps = libcomps.Comps() try: if file_name.endswith('.gz'): # libcomps does not support gzipped files - decompress to temporary # location with gzip.open(file_name) as gz_file: temp_file = tempfile.NamedTemporaryFile(delete=False) try: shutil.copyfileobj(gz_file, temp_file) # close temp_file to ensure the content is flushed to disk temp_file.close() file_comps.fromxml_f(temp_file.name) finally: os.unlink(temp_file.name) else: file_comps.fromxml_f(file_name) except (IOError, OSError, libcomps.ParserError) as err: # gzip module raises OSError on reading from malformed gz file # get_last_errors() output often contains duplicit lines, remove them seen = set() for error in file_comps.get_last_errors(): if error in seen: continue logger.error(error.strip()) seen.add(error) raise dnf.exceptions.Error( _("Can't load file \"{}\": {}").format(file_name, err)) else: self.comps += file_comps def save_output_files(self): for file_name in self.opts.save: try: # xml_f returns a list of errors / log entries errors = self.comps.xml_f(file_name, xml_options=COMPS_XML_OPTIONS) except libcomps.XMLGenError as err: errors = [err] if errors: # xml_f() method could return more than one error. In this case # raise the latest of them and log the others. for err in errors[:-1]: logger.error(err.strip()) raise dnf.exceptions.Error(_("Can't save file \"{}\": {}").format( file_name, errors[-1].strip())) def find_group(self, group_id, name): ''' Try to find group according to command line parameters - first by id then by name. ''' group = None if group_id: for grp in self.comps.groups: if grp.id == group_id: group = grp break if group is None and name: for grp in self.comps.groups: if grp.name == name: group = grp break return group def edit_group(self, group): ''' Set attributes and package lists for selected group ''' def langlist_to_strdict(lst): str_dict = libcomps.StrDict() for lang, text in lst: str_dict[lang] = text return str_dict # set group attributes if self.opts.name: group.name = self.opts.name if self.opts.description: group.desc = self.opts.description if self.opts.display_order: group.display_order = self.opts.display_order if self.opts.user_visible is not None: group.uservisible = self.opts.user_visible if self.opts.translated_name: group.name_by_lang = langlist_to_strdict(self.opts.translated_name) if self.opts.translated_description: group.desc_by_lang = langlist_to_strdict(self.opts.translated_description) # edit packages list if self.opts.packages: # find packages according to specifications from command line packages = set() for pkg_spec in self.opts.packages: subj = dnf.subject.Subject(pkg_spec) q = subj.get_best_query(self.base.sack, with_nevra=True, with_provides=False, with_filenames=False).latest() if not q: logger.warning(_("No match for argument: {}").format(pkg_spec)) continue packages.update(q) if self.opts.dependencies: # add packages that provide requirements requirements = set() for pkg in packages: requirements.update(pkg.requires) packages.update(self.base.sack.query().filterm(provides=requirements)) pkg_names = {pkg.name for pkg in packages} if self.opts.remove: for pkg_name in pkg_names: for pkg in group.packages_match(name=pkg_name, type=libcomps.PACKAGE_TYPE_UNKNOWN): group.packages.remove(pkg) else: if self.opts.mandatory: pkg_type = libcomps.PACKAGE_TYPE_MANDATORY elif self.opts.optional: pkg_type = libcomps.PACKAGE_TYPE_OPTIONAL else: pkg_type = libcomps.PACKAGE_TYPE_DEFAULT for pkg_name in sorted(pkg_names): if not group.packages_match(name=pkg_name, type=pkg_type): group.packages.append(libcomps.Package(name=pkg_name, type=pkg_type)) def run(self): self.load_input_files() if self.opts.id or self.opts.name: # we are adding / editing a group group = self.find_group(group_id=self.opts.id, name=self.opts.name) if group is None: # create a new group if self.opts.remove: raise dnf.exceptions.Error(_("Can't remove packages from non-existent group")) group = libcomps.Group() if self.opts.id: group.id = self.opts.id group.name = self.opts.id elif self.opts.name: group_id = text_to_id(self.opts.name) if self.find_group(group_id=group_id, name=None): raise dnf.cli.CliError( _("Group id '{}' generated from '{}' is duplicit. " "Please specify group id using --id.").format( group_id, self.opts.name)) group.id = group_id self.comps.groups.append(group) self.edit_group(group) self.save_output_files() if self.opts.print or (not self.opts.save): print(self.comps.xml_str(xml_options=COMPS_XML_OPTIONS))	13.21 KB	2022-09-09 11:42
📄 needs_restarting.py # needs_restarting.py # DNF plugin to check for running binaries in a need of restarting. # # Copyright (C) 2014 Red Hat, Inc. # # This copyrighted material is made available to anyone wishing to use, # modify, copy, or redistribute it subject to the terms and conditions of # the GNU General Public License v.2, or (at your option) any later version. # This program is distributed in the hope that it will be useful, but WITHOUT # ANY WARRANTY expressed or implied, including the implied warranties 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, write to the # Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA # 02110-1301, USA. Any Red Hat trademarks that are incorporated in the # source code or documentation are not subject to the GNU General Public # License and may only be used or replicated with the express permission of # Red Hat, Inc. # # the mechanism of scanning smaps for opened files and matching them back to # packages is heavily inspired by the original needs-restarting.py: # http://yum.baseurl.org/gitweb?p=yum-utils.git;a=blob;f=needs-restarting.py from __future__ import absolute_import from __future__ import division from __future__ import print_function from __future__ import unicode_literals from dnfpluginscore import logger, _ import dnf import dnf.cli import dbus import functools import os import re import stat import time # For which package updates we should recommend a reboot # Mostly taken from https://access.redhat.com/solutions/27943 NEED_REBOOT = ['kernel', 'kernel-rt', 'glibc', 'linux-firmware', 'systemd', 'dbus', 'dbus-broker', 'dbus-daemon', 'microcode_ctl'] def get_options_from_dir(filepath, base): """ Provide filepath as string if single dir or list of strings Return set of package names contained in files under filepath """ if not os.path.exists(filepath): return set() options = set() for file in os.listdir(filepath): if os.path.isdir(file) or not file.endswith('.conf'): continue with open(os.path.join(filepath, file)) as fp: for line in fp: options.add((line.rstrip(), file)) packages = set() for pkg in base.sack.query().installed().filter(name={x[0] for x in options}): packages.add(pkg.name) for name, file in {x for x in options if x[0] not in packages}: logger.warning( _('No installed package found for package name "{pkg}" ' 'specified in needs-restarting file "{file}".'.format(pkg=name, file=file))) return packages def list_opened_files(uid): for (pid, smaps) in list_smaps(): try: if uid is not None and uid != owner_uid(smaps): continue with open(smaps, 'r', errors='replace') as smaps_file: lines = smaps_file.readlines() except EnvironmentError: logger.warning("Failed to read PID %d's smaps.", pid) continue for line in lines: ofile = smap2opened_file(pid, line) if ofile is not None: yield ofile def list_smaps(): for dir_ in os.listdir('/proc'): try: pid = int(dir_) except ValueError: continue smaps = '/proc/%d/smaps' % pid yield (pid, smaps) def memoize(func): sentinel = object() cache = {} def wrapper(param): val = cache.get(param, sentinel) if val is not sentinel: return val val = func(param) cache[param] = val return val return wrapper def owner_uid(fname): return os.stat(fname)[stat.ST_UID] def owning_package(sack, fname): matches = sack.query().filter(file=fname).run() if matches: return matches[0] return None def print_cmd(pid): cmdline = '/proc/%d/cmdline' % pid with open(cmdline) as cmdline_file: command = dnf.i18n.ucd(cmdline_file.read()) command = ' '.join(command.split('\000')) print('%d : %s' % (pid, command)) def get_service_dbus(pid): bus = dbus.SystemBus() systemd_manager_object = bus.get_object( 'org.freedesktop.systemd1', '/org/freedesktop/systemd1' ) systemd_manager_interface = dbus.Interface( systemd_manager_object, 'org.freedesktop.systemd1.Manager' ) service_proxy = None try: service_proxy = bus.get_object( 'org.freedesktop.systemd1', systemd_manager_interface.GetUnitByPID(pid) ) except dbus.DBusException as e: # There is no unit for the pid. Usually error is 'NoUnitForPid'. # Considering what we do at the bottom (just return if not service) # Then there's really no reason to exit here on that exception. # Log what's happened then move on. msg = str(e) logger.warning("Failed to get systemd unit for PID {}: {}".format(pid, msg)) return service_properties = dbus.Interface( service_proxy, dbus_interface="org.freedesktop.DBus.Properties") name = service_properties.Get( "org.freedesktop.systemd1.Unit", 'Id' ) if name.endswith(".service"): return name return def smap2opened_file(pid, line): slash = line.find('/') if slash < 0: return None if line.find('00:') >= 0: # not a regular file return None fn = line[slash:].strip() suffix_index = fn.rfind(' (deleted)') if suffix_index < 0: return OpenedFile(pid, fn, False) else: return OpenedFile(pid, fn[:suffix_index], True) class OpenedFile(object): RE_TRANSACTION_FILE = re.compile('^(.+);[0-9A-Fa-f]{8,}$') def __init__(self, pid, name, deleted): self.deleted = deleted self.name = name self.pid = pid @property def presumed_name(self): """Calculate the name of the file pre-transaction. In case of a file that got deleted during the transactionm, possibly just because of an upgrade to a newer version of the same file, RPM renames the old file to the same name with a hexadecimal suffix just before delting it. """ if self.deleted: match = self.RE_TRANSACTION_FILE.match(self.name) if match: return match.group(1) return self.name class ProcessStart(object): def __init__(self): self.kernel_boot_time = ProcessStart.get_kernel_boot_time() self.boot_time = ProcessStart.get_boot_time(self.kernel_boot_time) self.sc_clk_tck = ProcessStart.get_sc_clk_tck() @staticmethod def get_kernel_boot_time(): try: with open('/proc/stat', 'r') as file: for line in file: if line.startswith("btime "): key, value = line.split() return float(value) except OSError as e: logger.debug("Couldn't read /proc/stat: %s", e) return 0 @staticmethod def get_boot_time(kernel_boot_time): """ We have three sources from which to derive the boot time. These values vary depending on containerization, existence of a Real Time Clock, etc. - UnitsLoadStartTimestamp property on /org/freedesktop/systemd1 The start time of the service manager, according to systemd itself. Seems to be more reliable than UserspaceTimestamp when the RTC is in local time. Works unless the system was not booted with systemd, such as in (most) containers. - st_mtime of /proc/1 Reflects the time the first process was run after booting. This works for all known cases except machines without a RTC---they awake at the start of the epoch. - btime field of /proc/stat Reflects the time when the kernel started. Works for machines without RTC iff the current time is reasonably correct. Does not work on containers which share their kernel with the host---there, the host kernel uptime is returned. """ units_load_start_timestamp = None try: # systemd timestamps are the preferred method to determine boot # time. max(proc_1_boot_time, kernel_boot_time) does not allow us # to disambiguate between an unreliable RTC (e.g. the RTC is in # UTC+1 instead of UTC) and a container with a proc_1_boot_time > # kernel_boot_time. So we use UnitsLoadStartTimestamp if it's # available, else fall back to the other methods. bus = dbus.SystemBus() systemd1 = bus.get_object( 'org.freedesktop.systemd1', '/org/freedesktop/systemd1' ) props = dbus.Interface( systemd1, dbus.PROPERTIES_IFACE ) units_load_start_timestamp = props.Get( 'org.freedesktop.systemd1.Manager', 'UnitsLoadStartTimestamp' ) if units_load_start_timestamp != 0: systemd_boot_time = units_load_start_timestamp / (1000 * 1000) logger.debug("Got boot time from systemd: %s", systemd_boot_time) return systemd_boot_time except dbus.exceptions.DBusException as e: logger.debug("D-Bus error getting boot time from systemd: %s", e) logger.debug("Couldn't get boot time from systemd, checking st_mtime of /proc/1 and btime field of /proc/stat.") try: proc_1_boot_time = float(os.stat('/proc/1').st_mtime) except OSError as e: logger.debug("Couldn't stat /proc/1: %s", e) proc_1_boot_time = 1 kernel_boot_time = kernel_boot_time boot_time = max(proc_1_boot_time, kernel_boot_time) logger.debug("st_mtime of /proc/1: %s", proc_1_boot_time) logger.debug("btime field of /proc/stat: %s", kernel_boot_time) logger.debug("Using %s as the system boot time.", boot_time) return boot_time @staticmethod def get_sc_clk_tck(): return os.sysconf(os.sysconf_names['SC_CLK_TCK']) def __call__(self, pid): stat_fn = '/proc/%d/stat' % pid with open(stat_fn) as stat_file: stats = stat_file.read().split() ticks_after_kernel_boot = int(stats[21]) secs_after_kernel_boot = ticks_after_kernel_boot / self.sc_clk_tck # The process's start time is always measured relative to the kernel # start time, not either of the other methods we use to get "boot time". return self.kernel_boot_time + secs_after_kernel_boot @dnf.plugin.register_command class NeedsRestartingCommand(dnf.cli.Command): aliases = ('needs-restarting',) summary = _('determine updated binaries that need restarting') @staticmethod def set_argparser(parser): parser.add_argument('-u', '--useronly', action='store_true', help=_("only consider this user's processes")) parser.add_argument('-r', '--reboothint', action='store_true', help=_("only report whether a reboot is required " "(exit code 1) or not (exit code 0)")) parser.add_argument('-s', '--services', action='store_true', help=_("only report affected systemd services")) def configure(self): demands = self.cli.demands demands.sack_activation = True def run(self): process_start = ProcessStart() owning_pkg_fn = functools.partial(owning_package, self.base.sack) owning_pkg_fn = memoize(owning_pkg_fn) opt = get_options_from_dir(os.path.join( self.base.conf.installroot, "etc/dnf/plugins/needs-restarting.d/"), self.base) NEED_REBOOT.extend(opt) if self.opts.reboothint: need_reboot = set() installed = self.base.sack.query().installed() for pkg in installed.filter(name=NEED_REBOOT): if pkg.installtime > process_start.boot_time: need_reboot.add(pkg.name) if need_reboot: print(_('Core libraries or services have been updated ' 'since boot-up:')) for name in sorted(need_reboot): print(' * %s' % name) print() print(_('Reboot is required to fully utilize these updates.')) print(_('More information:'), 'https://access.redhat.com/solutions/27943') raise dnf.exceptions.Error() # Sets exit code 1 else: print(_('No core libraries or services have been updated ' 'since boot-up.')) print(_('Reboot should not be necessary.')) return None stale_pids = set() uid = os.geteuid() if self.opts.useronly else None for ofile in list_opened_files(uid): pkg = owning_pkg_fn(ofile.presumed_name) if pkg is None: continue if pkg.installtime > process_start(ofile.pid): stale_pids.add(ofile.pid) if self.opts.services: names = set([get_service_dbus(pid) for pid in sorted(stale_pids)]) for name in names: if name is not None: print(name) return 0 for pid in sorted(stale_pids): print_cmd(pid)	13.52 KB	2025-11-12 04:27
📄 post-transaction-actions.py # Copyright (C) 2019 Red Hat, Inc. # # Licensed under the GNU Lesser General Public License Version 2.1 # # This library is free software; you can redistribute it and/or # modify it under the terms of the GNU Lesser General Public # License as published by the Free Software Foundation; either # version 2.1 of the License, or (at your option) any later version. # # This library 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 # Lesser General Public License for more details. # # You should have received a copy of the GNU Lesser General Public # License along with this library; if not, write to the Free Software # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA """ The plugin allows to define actions to be executed upon completing an RPM transaction. Each action may define a (glob-like) filtering rule on the package NEVRA or package files, as well as whether the package was installed or removed. Actions are defined in action files. The action can start any shell command. Commands can contain variables. The same command (after variables substitution) is executed only once per transaction. The order of execution of the commands may differ from the order of the packages in the transaction. """ from dnfpluginscore import _, logger import libdnf.conf import dnf import dnf.transaction import glob import os import subprocess class PostTransactionActions(dnf.Plugin): name = "post-transaction-actions" def __init__(self, base, cli): super(PostTransactionActions, self).__init__(base, cli) self.actiondir = "/etc/dnf/plugins/post-transaction-actions.d/" self.base = base self.logger = logger def config(self): conf = self.read_config(self.base.conf) if conf.has_section("main"): if conf.has_option("main", "actiondir"): self.actiondir = conf.get("main", "actiondir") def _parse_actions(self): """Parses .action files from self.actiondir path. Parsed actions are stored in a list of tuples.""" action_file_list = [] if os.access(self.actiondir, os.R_OK): action_file_list.extend(glob.glob(self.actiondir + ".action")) action_tuples = [] # (action key, action_state, shell command) for file_name in action_file_list: for line in open(file_name).readlines(): line = line.strip() if line and line[0] != "#": try: (action_key, action_state, action_command) = line.split(':', 2) except ValueError as e: self.logger.error(_('Bad Action Line "%s": %s') % (line, e)) continue else: action_tuples.append((action_key, action_state, action_command)) return action_tuples def _replace_vars(self, ts_item, command): """Replaces variables in the command. Variables: ${name}, ${arch}, ${ver}, ${rel}, ${epoch}, ${repoid}, ${state} or $name, $arch, $ver, $rel, $epoch, $repoid, $state""" action_dict = {dnf.transaction.PKG_DOWNGRADE: "downgrade", dnf.transaction.PKG_DOWNGRADED: "downgraded", dnf.transaction.PKG_INSTALL: "install", dnf.transaction.PKG_OBSOLETE: "obsolete", dnf.transaction.PKG_OBSOLETED: "obsoleted", dnf.transaction.PKG_REINSTALL: "reinstall", dnf.transaction.PKG_REINSTALLED: "reinstalled", dnf.transaction.PKG_REMOVE: "remove", dnf.transaction.PKG_UPGRADE: "upgrade", dnf.transaction.PKG_UPGRADED: "upgraded"} action = action_dict.get(ts_item.action, "unknown - %s" % ts_item.action) vardict = {"name": ts_item.name, "arch": ts_item.arch, "ver": ts_item.version, "rel": ts_item.release, "epoch": str(ts_item.epoch), "repoid": ts_item.from_repo, "state": action} result = libdnf.conf.ConfigParser_substitute(command, vardict) return result def transaction(self): action_tuples = self._parse_actions() in_ts_items = [] out_ts_items = [] all_ts_items = [] for ts_item in self.base.transaction: if ts_item.action in dnf.transaction.FORWARD_ACTIONS: in_ts_items.append(ts_item) elif ts_item.action in dnf.transaction.BACKWARD_ACTIONS: out_ts_items.append(ts_item) else: # The action is not rpm change. It can be a reason change, therefore we can skip that item continue all_ts_items.append(ts_item) commands_to_run = [] for (a_key, a_state, a_command) in action_tuples: if a_state == "in": ts_items = in_ts_items elif a_state == "out": ts_items = out_ts_items elif a_state == "any": ts_items = all_ts_items else: # unsupported state, skip it self.logger.error(_("Bad Transaction State: %s") % a_state) continue subj = dnf.subject.Subject(a_key) pkgs = [tsi.pkg for tsi in ts_items] query = self.base.sack.query().filterm(pkg=pkgs) query = subj.get_best_query(self.base.sack, with_nevra=True, with_provides=False, with_filenames=True, query=query) for pkg in query: ts_item_list = [tsi for tsi in ts_items if tsi.pkg == pkg and tsi.pkg.reponame == pkg.reponame] ts_item = ts_item_list[0] command = self._replace_vars(ts_item, a_command) commands_to_run.append(command) # de-dup the list seen = set() commands_to_run = [x for x in commands_to_run if x not in seen and not seen.add(x)] for command in commands_to_run: try: proc = subprocess.Popen(command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, close_fds=True, universal_newlines=True) stdout, stderr = proc.communicate() if stdout: self.logger.info(_('post-transaction-actions: %s') % (stdout)) if stderr: self.logger.error(_('post-transaction-actions: %s') % (stderr)) except Exception as e: self.logger.error(_('post-transaction-actions: Bad Command "%s": %s') % (command, e))	6.84 KB	2022-09-09 11:42
📄 repoclosure.py # repoclosure.py # DNF plugin adding a command to display a list of unresolved dependencies # for repositories. # # Copyright (C) 2015 Igor Gnatenko # # This copyrighted material is made available to anyone wishing to use, # modify, copy, or redistribute it subject to the terms and conditions of # the GNU General Public License v.2, or (at your option) any later version. # This program is distributed in the hope that it will be useful, but WITHOUT # ANY WARRANTY expressed or implied, including the implied warranties 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, write to the # Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA # 02110-1301, USA. Any Red Hat trademarks that are incorporated in the # source code or documentation are not subject to the GNU General Public # License and may only be used or replicated with the express permission of # Red Hat, Inc. # from __future__ import absolute_import from __future__ import unicode_literals from dnfpluginscore import _ import dnf.cli class RepoClosure(dnf.Plugin): name = "repoclosure" def __init__(self, base, cli): super(RepoClosure, self).__init__(base, cli) if cli is None: return cli.register_command(RepoClosureCommand) class RepoClosureCommand(dnf.cli.Command): aliases = ("repoclosure",) summary = _("Display a list of unresolved dependencies for repositories") def configure(self): demands = self.cli.demands demands.sack_activation = True demands.available_repos = True if self.opts.repo: for repo in self.base.repos.all(): if repo.id not in self.opts.repo and repo.id not in self.opts.check: repo.disable() else: repo.enable() def run(self): total_missing = 0 if self.opts.arches: unresolved = self._get_unresolved(self.opts.arches) else: unresolved = self._get_unresolved() for pkg in sorted(unresolved.keys()): print("package: {} from {}".format(str(pkg), pkg.reponame)) print(" unresolved deps ({}):".format(len(unresolved[pkg]))) total_missing += len(unresolved[pkg]) for dep in unresolved[pkg]: print(" {}".format(dep)) if len(unresolved) > 0: msg = _( "Repoclosure ended with unresolved dependencies ({}) across {} packages.".format( total_missing, len(unresolved) ) ) raise dnf.exceptions.Error(msg) def _get_unresolved(self, arch=None): unresolved = {} deps = set() # We have two sets of packages, available and to_check: # * available is the set of packages used to satisfy dependencies # * to_check is the set of packages we are checking the dependencies of # # to_check can be a subset of available if the --arch, --best, --check, # --newest, or --pkg options are used # # --arch: only packages matching arch are checked # --best: available only contains the latest packages per arch across all repos # --check: only check packages in the specified repo(s) # --newest: only consider the latest versions of a package from each repo # --pkg: only check the specified packages # # Relationship of --best and --newest: # # Pkg Set \| Neither \| --best \| --newest \| --best and --newest \| # available \| all \| latest in all repos \| latest per repo \| latest in all repos \| # to_check \| all \| all \| latest per repo \| latest per repo \| if self.opts.newest: available = self.base.sack.query().filter(empty=True) to_check = self.base.sack.query().filter(empty=True) for repo in self.base.repos.iter_enabled(): available = \ available.union(self.base.sack.query().filter(reponame=repo.id).latest()) to_check = \ to_check.union(self.base.sack.query().filter(reponame=repo.id).latest()) else: available = self.base.sack.query().available() to_check = self.base.sack.query().available() if self.opts.pkglist: pkglist_q = self.base.sack.query().filter(empty=True) errors = [] for pkg in self.opts.pkglist: subj = dnf.subject.Subject(pkg) pkg_q = to_check.intersection( subj.get_best_query(self.base.sack, with_nevra=True, with_provides=False, with_filenames=False)) if pkg_q: pkglist_q = pkglist_q.union(pkg_q) else: errors.append(pkg) if errors: raise dnf.exceptions.Error( _('no package matched: %s') % ', '.join(errors)) to_check = pkglist_q if self.opts.check: to_check.filterm(reponame=self.opts.check) if arch is not None: to_check.filterm(arch=arch) if self.base.conf.best: available.filterm(latest_per_arch=True) available.apply() to_check.apply() for pkg in to_check: unresolved[pkg] = set() for req in pkg.requires: reqname = str(req) # XXX: https://bugzilla.redhat.com/show_bug.cgi?id=1186721 if reqname.startswith("solvable:") or \ reqname.startswith("rpmlib("): continue deps.add(req) unresolved[pkg].add(req) unresolved_deps = set(x for x in deps if not available.filter(provides=x)) unresolved_transition = {k: set(x for x in v if x in unresolved_deps) for k, v in unresolved.items()} return {k: v for k, v in unresolved_transition.items() if v} @staticmethod def set_argparser(parser): parser.add_argument("--arch", default=[], action="append", dest='arches', help=_("check packages of the given archs, can be " "specified multiple times")) parser.add_argument("--check", default=[], action="append", help=_("Specify repositories to check")) parser.add_argument("-n", "--newest", action="store_true", help=_("Check only the newest packages in the " "repos")) parser.add_argument("--pkg", default=[], action="append", help=_("Check closure for this package only"), dest="pkglist")	6.89 KB	2022-09-09 11:42
📄 repodiff.py # repodiff.py # DNF plugin adding a command to show differencies between two sets # of repositories. # # Copyright (C) 2018 Red Hat, Inc. # # This copyrighted material is made available to anyone wishing to use, # modify, copy, or redistribute it subject to the terms and conditions of # the GNU General Public License v.2, or (at your option) any later version. # This program is distributed in the hope that it will be useful, but WITHOUT # ANY WARRANTY expressed or implied, including the implied warranties 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, write to the # Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA # 02110-1301, USA. Any Red Hat trademarks that are incorporated in the # source code or documentation are not subject to the GNU General Public # License and may only be used or replicated with the express permission of # Red Hat, Inc. # from __future__ import absolute_import from __future__ import unicode_literals import dnf.cli from dnf.cli.option_parser import OptionParser import hawkey from dnfpluginscore import _ class RepoDiff(dnf.Plugin): name = "repodiff" def __init__(self, base, cli): super(RepoDiff, self).__init__(base, cli) if cli is None: return cli.register_command(RepoDiffCommand) class RepoDiffCommand(dnf.cli.Command): aliases = ("repodiff",) summary = _("List differences between two sets of repositories") @staticmethod def set_argparser(parser): # I'd like to use --old and --new options like Yum did. # But ability to disable abbreviated long options is added # only in Python >= 3.5 # So in command arguments we are not able to use arguments, # which are prefixes of main arguments (i.w. --new would be # treated as --newpackage). This is because we run .parse_args # two times - for main and then for command arguments. # https://stackoverflow.com/questions/33900846 parser.add_argument("--repo-old", "-o", default=[], action="append", dest="old", help=_("Specify old repository, can be used multiple times")) parser.add_argument("--repo-new", "-n", default=[], action="append", dest="new", help=_("Specify new repository, can be used multiple times")) parser.add_argument("--arch", "--archlist", "-a", default=[], action=OptionParser._SplitCallback, dest="arches", help=_("Specify architectures to compare, can be used " "multiple times. By default, only source rpms are " "compared.")) parser.add_argument("--size", "-s", action="store_true", help=_("Output additional data about the size of the changes.")) parser.add_argument("--compare-arch", action="store_true", help=_("Compare packages also by arch. By default " "packages are compared just by name.")) parser.add_argument("--simple", action="store_true", help=_("Output a simple one line message for modified packages.")) parser.add_argument("--downgrade", action="store_true", help=_("Split the data for modified packages between " "upgraded and downgraded packages.")) def configure(self): demands = self.cli.demands demands.sack_activation = True demands.available_repos = True demands.changelogs = True self.base.conf.disable_excludes = ["all"] # TODO yum was able to handle mirrorlist in --new/--old arguments # Can be resolved by improving --repofrompath option if not self.opts.new or not self.opts.old: msg = _("Both old and new repositories must be set.") raise dnf.exceptions.Error(msg) for repo in self.base.repos.all(): if repo.id in self.opts.new + self.opts.old: repo.enable() else: repo.disable() if not self.opts.arches: self.opts.arches = ['src'] def _pkgkey(self, pkg): if self.opts.compare_arch: return (pkg.name, pkg.arch) return pkg.name def _repodiff(self, old, new): '''compares packagesets old and new, returns dictionary with packages: added: only in new set removed: only in old set upgraded: in both old and new, new has bigger evr downgraded: in both old and new, new has lower evr obsoletes: dictionary of which old package is obsoleted by which new ''' old_d = dict([(self._pkgkey(p), p) for p in old]) old_keys = set(old_d.keys()) new_d = dict([(self._pkgkey(p), p) for p in new]) new_keys = set(new_d.keys()) # mapping obsoleted_package_from_old: obsoleted_by_package_from_new obsoletes = dict() for obsoleter in new.filter(obsoletes=old): for obsoleted in old.filter(provides=obsoleter.obsoletes): obsoletes[self._pkgkey(obsoleted)] = obsoleter evr_cmp = self.base.sack.evr_cmp repodiff = dict( added=[new_d[k] for k in new_keys - old_keys], removed=[old_d[k] for k in old_keys - new_keys], obsoletes=obsoletes, upgraded=[], downgraded=[]) for k in old_keys.intersection(new_keys): pkg_old = old_d[k] pkg_new = new_d[k] if pkg_old.evr == pkg_new.evr: continue if evr_cmp(pkg_old.evr, pkg_new.evr) > 0: repodiff['downgraded'].append((pkg_old, pkg_new)) else: repodiff['upgraded'].append((pkg_old, pkg_new)) return repodiff def _report(self, repodiff): def pkgstr(pkg): if self.opts.compare_arch: return str(pkg) return "%s-%s" % (pkg.name, pkg.evr) def sizestr(num): msg = str(num) if num > 0: msg += " ({})".format(dnf.cli.format.format_number(num).strip()) elif num < 0: msg += " (-{})".format(dnf.cli.format.format_number(-num).strip()) return msg def report_modified(pkg_old, pkg_new): msgs = [] if self.opts.simple: msgs.append("%s -> %s" % (pkgstr(pkg_old), pkgstr(pkg_new))) else: msgs.append('') msgs.append("%s -> %s" % (pkgstr(pkg_old), pkgstr(pkg_new))) msgs.append('-' * len(msgs[-1])) if pkg_old.changelogs: old_chlog = pkg_old.changelogs[0] else: old_chlog = None for chlog in pkg_new.changelogs: if old_chlog: if chlog['timestamp'] < old_chlog['timestamp']: break elif (chlog['timestamp'] == old_chlog['timestamp'] and chlog['author'] == old_chlog['author'] and chlog['text'] == old_chlog['text']): break msgs.append('* %s %s\n%s' % ( chlog['timestamp'].strftime("%a %b %d %Y"), dnf.i18n.ucd(chlog['author']), dnf.i18n.ucd(chlog['text']))) if self.opts.size: msgs.append(_("Size change: {} bytes").format( pkg_new.size - pkg_old.size)) print('\n'.join(msgs)) sizes = dict(added=0, removed=0, upgraded=0, downgraded=0) for pkg in sorted(repodiff['added']): print(_("Added package : {}").format(pkgstr(pkg))) sizes['added'] += pkg.size for pkg in sorted(repodiff['removed']): print(_("Removed package: {}").format(pkgstr(pkg))) obsoletedby = repodiff['obsoletes'].get(self._pkgkey(pkg)) if obsoletedby: print(_("Obsoleted by : {}").format(pkgstr(obsoletedby))) sizes['removed'] += pkg.size if self.opts.downgrade: if repodiff['upgraded']: print(_("\nUpgraded packages")) for (pkg_old, pkg_new) in sorted(repodiff['upgraded']): sizes['upgraded'] += (pkg_new.size - pkg_old.size) report_modified(pkg_old, pkg_new) if repodiff['downgraded']: print(_("\nDowngraded packages")) for (pkg_old, pkg_new) in sorted(repodiff['downgraded']): sizes['downgraded'] += (pkg_new.size - pkg_old.size) report_modified(pkg_old, pkg_new) else: modified = repodiff['upgraded'] + repodiff['downgraded'] if modified: print(_("\nModified packages")) for (pkg_old, pkg_new) in sorted(modified): sizes['upgraded'] += (pkg_new.size - pkg_old.size) report_modified(pkg_old, pkg_new) print(_("\nSummary")) print(_("Added packages: {}").format(len(repodiff['added']))) print(_("Removed packages: {}").format(len(repodiff['removed']))) if self.opts.downgrade: print(_("Upgraded packages: {}").format(len(repodiff['upgraded']))) print(_("Downgraded packages: {}").format(len(repodiff['downgraded']))) else: print(_("Modified packages: {}").format( len(repodiff['upgraded']) + len(repodiff['downgraded']))) if self.opts.size: print(_("Size of added packages: {}").format(sizestr(sizes['added']))) print(_("Size of removed packages: {}").format(sizestr(sizes['removed']))) if not self.opts.downgrade: print(_("Size of modified packages: {}").format( sizestr(sizes['upgraded'] + sizes['downgraded']))) else: print(_("Size of upgraded packages: {}").format( sizestr(sizes['upgraded']))) print(_("Size of downgraded packages: {}").format( sizestr(sizes['downgraded']))) print(_("Size change: {}").format( sizestr(sizes['added'] + sizes['upgraded'] + sizes['downgraded'] - sizes['removed']))) def run(self): # prepare old and new packagesets based by given arguments q_new = self.base.sack.query(hawkey.IGNORE_EXCLUDES).filter( reponame=self.opts.new) q_old = self.base.sack.query(hawkey.IGNORE_EXCLUDES).filter( reponame=self.opts.old) if self.opts.arches and '*' not in self.opts.arches: q_new.filterm(arch=self.opts.arches) q_old.filterm(arch=self.opts.arches) if self.opts.compare_arch: q_new.filterm(latest_per_arch=1) q_old.filterm(latest_per_arch=1) else: q_new.filterm(latest=1) q_old.filterm(latest=1) q_new.apply() q_old.apply() self._report(self._repodiff(q_old, q_new))	11.21 KB	2022-09-09 11:42
📄 repograph.py # repograph.py # DNF plugin adding a command to Output a full package dependency graph in dot # format. # # Copyright (C) 2015 Igor Gnatenko # # This copyrighted material is made available to anyone wishing to use, # modify, copy, or redistribute it subject to the terms and conditions of # the GNU General Public License v.2, or (at your option) any later version. # This program is distributed in the hope that it will be useful, but WITHOUT # ANY WARRANTY expressed or implied, including the implied warranties 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, write to the # Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA # 02110-1301, USA. Any Red Hat trademarks that are incorporated in the # source code or documentation are not subject to the GNU General Public # License and may only be used or replicated with the express permission of # Red Hat, Inc. # from __future__ import absolute_import from __future__ import unicode_literals from dnfpluginscore import _, logger import dnf.cli DOT_HEADER = """ size="20.69,25.52"; ratio="fill"; rankdir="TB"; orientation=port; node[style="filled"]; """ class RepoGraph(dnf.Plugin): name = "repograph" def __init__(self, base, cli): super(RepoGraph, self).__init__(base, cli) if cli is None: return cli.register_command(RepoGraphCommand) class RepoGraphCommand(dnf.cli.Command): aliases = ("repograph", "repo-graph",) summary = _("Output a full package dependency graph in dot format") def configure(self): demands = self.cli.demands demands.sack_activation = True demands.available_repos = True if self.opts.repo: for repo in self.base.repos.all(): if repo.id not in self.opts.repo: repo.disable() else: repo.enable() def run(self): self.do_dot(DOT_HEADER) def do_dot(self, header): maxdeps = 0 deps = self._get_deps(self.base.sack) print("digraph packages {") print("{}".format(header)) for pkg in deps.keys(): if len(deps[pkg]) > maxdeps: maxdeps = len(deps[pkg]) # color calculations lifted from rpmgraph h = 0.5 + (0.6 / 23 * len(deps[pkg])) s = h + 0.1 b = 1.0 print('"{}" [color="{:.12g} {:.12g} {}"];'.format(pkg, h, s, b)) print('"{}" -> {{'.format(pkg)) for req in deps[pkg]: print('"{}"'.format(req)) print('}} [color="{:.12g} {:.12g} {}"];\n'.format(h, s, b)) print("}") @staticmethod def _get_deps(sack): requires = {} prov = {} skip = [] available = sack.query().available() for pkg in available: xx = {} for req in pkg.requires: reqname = str(req) if reqname in skip: continue # XXX: https://bugzilla.redhat.com/show_bug.cgi?id=1186721 if reqname.startswith("solvable:"): continue if reqname in prov: provider = prov[reqname] else: provider = available.filter(provides=reqname) if not provider: logger.debug(_("Nothing provides: '%s'"), reqname) skip.append(reqname) continue else: provider = provider[0].name prov[reqname] = provider if provider == pkg.name: xx[provider] = None if provider in xx or provider in skip: continue else: xx[provider] = None requires[pkg.name] = xx.keys() return requires	4 KB	2022-09-09 11:42
📄 repomanage.py # repomanage.py # DNF plugin adding a command to manage rpm packages from given directory. # # Copyright (C) 2015 Igor Gnatenko # # This copyrighted material is made available to anyone wishing to use, # modify, copy, or redistribute it subject to the terms and conditions of # the GNU General Public License v.2, or (at your option) any later version. # This program is distributed in the hope that it will be useful, but WITHOUT # ANY WARRANTY expressed or implied, including the implied warranties 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, write to the # Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA # 02110-1301, USA. Any Red Hat trademarks that are incorporated in the # source code or documentation are not subject to the GNU General Public # License and may only be used or replicated with the express permission of # Red Hat, Inc. # from __future__ import absolute_import from __future__ import unicode_literals from dnfpluginscore import _, logger import dnf import dnf.cli import logging import os import hawkey class RepoManage(dnf.Plugin): name = "repomanage" def __init__(self, base, cli): super(RepoManage, self).__init__(base, cli) if cli is None: return cli.register_command(RepoManageCommand) class RepoManageCommand(dnf.cli.Command): aliases = ("repomanage",) summary = _("Manage a directory of rpm packages") def pre_configure(self): if not self.opts.verbose and not self.opts.quiet: self.cli.redirect_logger(stdout=logging.WARNING, stderr=logging.INFO) def configure(self): if not self.opts.verbose and not self.opts.quiet: self.cli.redirect_repo_progress() demands = self.cli.demands demands.sack_activation = True def run(self): if self.opts.new and self.opts.old: raise dnf.exceptions.Error(_("Pass either --old or --new, not both!")) if self.opts.new and self.opts.oldonly: raise dnf.exceptions.Error(_("Pass either --oldonly or --new, not both!")) if self.opts.old and self.opts.oldonly: raise dnf.exceptions.Error(_("Pass either --old or --oldonly, not both!")) if not self.opts.old and not self.opts.oldonly: self.opts.new = True verfile = {} pkgdict = {} module_dict = {} # {NameStream: {Version: [modules]}} all_modular_artifacts = set() keepnum = int(self.opts.keep) # the number of items to keep try: REPOMANAGE_REPOID = "repomanage_repo" repo_conf = self.base.repos.add_new_repo(REPOMANAGE_REPOID, self.base.conf, baseurl=[self.opts.path]) # Always expire the repo, otherwise repomanage could use cached metadata and give identical results # for multiple runs even if the actual repo changed in the meantime repo_conf._repo.expire() self.base._add_repo_to_sack(repo_conf) if dnf.base.WITH_MODULES: self.base._setup_modular_excludes() # Prepare modules module_packages = self.base._moduleContainer.getModulePackages() for module_package in module_packages: # Even though we load only REPOMANAGE_REPOID other modules can be loaded from system # failsafe data automatically, we don't want them affecting repomanage results so ONLY # use modules from REPOMANAGE_REPOID. if module_package.getRepoID() == REPOMANAGE_REPOID: all_modular_artifacts.update(module_package.getArtifacts()) module_dict.setdefault(module_package.getNameStream(), {}).setdefault( module_package.getVersionNum(), []).append(module_package) except dnf.exceptions.RepoError: rpm_list = [] rpm_list = self._get_file_list(self.opts.path, ".rpm") if len(rpm_list) == 0: raise dnf.exceptions.Error(_("No files to process")) self.base.reset(sack=True, repos=True) self.base.fill_sack(load_system_repo=False, load_available_repos=False) try: self.base.add_remote_rpms(rpm_list, progress=self.base.output.progress) except IOError: logger.warning(_("Could not open {}").format(', '.join(rpm_list))) # Prepare regular packages query = self.base.sack.query(flags=hawkey.IGNORE_MODULAR_EXCLUDES).available() packages = [x for x in query.filter(pkg__neq=query.filter(nevra_strict=all_modular_artifacts)).available()] packages.sort() for pkg in packages: na = (pkg.name, pkg.arch) if na in pkgdict: if pkg not in pkgdict[na]: pkgdict[na].append(pkg) else: pkgdict[na] = [pkg] nevra = self._package_to_nevra(pkg) if nevra in verfile: verfile[nevra].append(self._package_to_path(pkg)) else: verfile[nevra] = [self._package_to_path(pkg)] outputpackages = [] # modular packages keepnum_latest_stream_artifacts = set() if self.opts.new: # regular packages for (n, a) in pkgdict.keys(): evrlist = pkgdict[(n, a)] newevrs = evrlist[-keepnum:] for package in newevrs: nevra = self._package_to_nevra(package) for fpkg in verfile[nevra]: outputpackages.append(fpkg) # modular packages for streams_by_version in module_dict.values(): sorted_stream_versions = sorted(streams_by_version.keys()) new_sorted_stream_versions = sorted_stream_versions[-keepnum:] for i in new_sorted_stream_versions: for stream in streams_by_version[i]: keepnum_latest_stream_artifacts.update(set(stream.getArtifacts())) if self.opts.old: # regular packages for (n, a) in pkgdict.keys(): evrlist = pkgdict[(n, a)] oldevrs = evrlist[:-keepnum] for package in oldevrs: nevra = self._package_to_nevra(package) for fpkg in verfile[nevra]: outputpackages.append(fpkg) # modular packages for streams_by_version in module_dict.values(): sorted_stream_versions = sorted(streams_by_version.keys()) old_sorted_stream_versions = sorted_stream_versions[:-keepnum] for i in old_sorted_stream_versions: for stream in streams_by_version[i]: keepnum_latest_stream_artifacts.update(set(stream.getArtifacts())) if self.opts.oldonly: # regular packages for (n, a) in pkgdict.keys(): evrlist = pkgdict[(n, a)] oldevrs = evrlist[:-keepnum] for package in oldevrs: nevra = self._package_to_nevra(package) for fpkg in verfile[nevra]: outputpackages.append(fpkg) # modular packages keepnum_newer_stream_artifacts = set() for streams_by_version in module_dict.values(): sorted_stream_versions = sorted(streams_by_version.keys()) new_sorted_stream_versions = sorted_stream_versions[-keepnum:] for i in new_sorted_stream_versions: for stream in streams_by_version[i]: keepnum_newer_stream_artifacts.update(set(stream.getArtifacts())) for streams_by_version in module_dict.values(): sorted_stream_versions = sorted(streams_by_version.keys()) old_sorted_stream_versions = sorted_stream_versions[:-keepnum] for i in old_sorted_stream_versions: for stream in streams_by_version[i]: for artifact in stream.getArtifacts(): if artifact not in keepnum_newer_stream_artifacts: keepnum_latest_stream_artifacts.add(artifact) modular_packages = [self._package_to_path(x) for x in query.filter(pkg__eq=query.filter(nevra_strict=keepnum_latest_stream_artifacts)).available()] outputpackages = outputpackages + modular_packages outputpackages.sort() if self.opts.space: print(" ".join(outputpackages)) else: for pkg in outputpackages: print(pkg) @staticmethod def set_argparser(parser): parser.add_argument("-o", "--old", action="store_true", help=_("Print the older packages")) parser.add_argument("-O", "--oldonly", action="store_true", help=_("Print the older packages. Exclude the newest packages.")) parser.add_argument("-n", "--new", action="store_true", help=_("Print the newest packages")) parser.add_argument("-s", "--space", action="store_true", help=_("Space separated output, not newline")) parser.add_argument("-k", "--keep", action="store", metavar="KEEP", help=_("Newest N packages to keep - defaults to 1"), default=1, type=int) parser.add_argument("path", action="store", help=_("Path to directory")) @staticmethod def _get_file_list(path, ext): """Return all files in path matching ext return list object """ filelist = [] for root, dirs, files in os.walk(path): for f in files: if os.path.splitext(f)[1].lower() == str(ext): filelist.append(os.path.join(root, f)) return filelist def _package_to_path(self, pkg): if len(self.base.repos): return os.path.join(self.opts.path, pkg.location) else: return pkg.location @staticmethod def _package_to_nevra(pkg): return (pkg.name, pkg.epoch, pkg.version, pkg.release, pkg.arch)	10.32 KB	2022-09-09 11:42
📄 reposync.py # reposync.py # DNF plugin adding a command to download all packages from given remote repo. # # Copyright (C) 2014 Red Hat, Inc. # # This copyrighted material is made available to anyone wishing to use, # modify, copy, or redistribute it subject to the terms and conditions of # the GNU General Public License v.2, or (at your option) any later version. # This program is distributed in the hope that it will be useful, but WITHOUT # ANY WARRANTY expressed or implied, including the implied warranties 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, write to the # Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA # 02110-1301, USA. Any Red Hat trademarks that are incorporated in the # source code or documentation are not subject to the GNU General Public # License and may only be used or replicated with the express permission of # Red Hat, Inc. # from __future__ import absolute_import from __future__ import unicode_literals import hawkey import os import shutil import types from dnfpluginscore import _, logger from dnf.cli.option_parser import OptionParser import dnf import dnf.cli def _pkgdir(intermediate, target): cwd = dnf.i18n.ucd(os.getcwd()) return os.path.realpath(os.path.join(cwd, intermediate, target)) class RPMPayloadLocation(dnf.repo.RPMPayload): def __init__(self, pkg, progress, pkg_location): super(RPMPayloadLocation, self).__init__(pkg, progress) self.package_dir = os.path.dirname(pkg_location) def _target_params(self): tp = super(RPMPayloadLocation, self)._target_params() dnf.util.ensure_dir(self.package_dir) tp['dest'] = self.package_dir return tp @dnf.plugin.register_command class RepoSyncCommand(dnf.cli.Command): aliases = ('reposync',) summary = _('download all packages from remote repo') def __init__(self, cli): super(RepoSyncCommand, self).__init__(cli) @staticmethod def set_argparser(parser): parser.add_argument('-a', '--arch', dest='arches', default=[], action=OptionParser._SplitCallback, metavar='[arch]', help=_('download only packages for this ARCH')) parser.add_argument('--delete', default=False, action='store_true', help=_('delete local packages no longer present in repository')) parser.add_argument('--download-metadata', default=False, action='store_true', help=_('download all the metadata.')) parser.add_argument('-g', '--gpgcheck', default=False, action='store_true', help=_('Remove packages that fail GPG signature checking ' 'after downloading')) parser.add_argument('-m', '--downloadcomps', default=False, action='store_true', help=_('also download and uncompress comps.xml')) parser.add_argument('--metadata-path', help=_('where to store downloaded repository metadata. ' 'Defaults to the value of --download-path.')) parser.add_argument('-n', '--newest-only', default=False, action='store_true', help=_('download only newest packages per-repo')) parser.add_argument('--norepopath', default=False, action='store_true', help=_("Don't add the reponame to the download path.")) parser.add_argument('-p', '--download-path', default='./', help=_('where to store downloaded repositories')) parser.add_argument('--remote-time', default=False, action='store_true', help=_('try to set local timestamps of local files by ' 'the one on the server')) parser.add_argument('--source', default=False, action='store_true', help=_('download only source packages')) parser.add_argument('-u', '--urls', default=False, action='store_true', help=_("Just list urls of what would be downloaded, " "don't download")) def configure(self): demands = self.cli.demands demands.available_repos = True demands.sack_activation = True repos = self.base.repos if self.opts.repo: repos.all().disable() for repoid in self.opts.repo: try: repo = repos[repoid] except KeyError: raise dnf.cli.CliError("Unknown repo: '%s'." % repoid) repo.enable() if self.opts.source: repos.enable_source_repos() if len(list(repos.iter_enabled())) > 1 and self.opts.norepopath: raise dnf.cli.CliError( _("Can't use --norepopath with multiple repositories")) for repo in repos.iter_enabled(): repo._repo.expire() repo.deltarpm = False def run(self): self.base.conf.keepcache = True gpgcheck_ok = True for repo in self.base.repos.iter_enabled(): if self.opts.remote_time: repo._repo.setPreserveRemoteTime(True) if self.opts.download_metadata: if self.opts.urls: for md_type, md_location in repo._repo.getMetadataLocations(): url = repo.remote_location(md_location) if url: print(url) else: msg = _("Failed to get mirror for metadata: %s") % md_type logger.warning(msg) else: self.download_metadata(repo) if self.opts.downloadcomps: if self.opts.urls: mdl = dict(repo._repo.getMetadataLocations()) group_locations = [mdl[md_type] for md_type in ('group', 'group_gz', 'group_gz_zck') if md_type in mdl] if group_locations: for group_location in group_locations: url = repo.remote_location(group_location) if url: print(url) break else: msg = _("Failed to get mirror for the group file.") logger.warning(msg) else: self.getcomps(repo) pkglist = self.get_pkglist(repo) if self.opts.urls: self.print_urls(pkglist) else: self.download_packages(pkglist) if self.opts.gpgcheck: for pkg in pkglist: local_path = self.pkg_download_path(pkg) # base.package_signature_check uses pkg.localPkg() to determine # the location of the package rpm file on the disk. # Set it to the correct download path. pkg.localPkg = types.MethodType( lambda s, local_path=local_path: local_path, pkg) result, error = self.base.package_signature_check(pkg) if result != 0: logger.warning(_("Removing {}: {}").format( os.path.basename(local_path), error)) os.unlink(local_path) gpgcheck_ok = False if self.opts.delete: self.delete_old_local_packages(repo, pkglist) if not gpgcheck_ok: raise dnf.exceptions.Error(_("GPG signature check failed.")) def repo_target(self, repo): return _pkgdir(self.opts.destdir or self.opts.download_path, repo.id if not self.opts.norepopath else '') def metadata_target(self, repo): if self.opts.metadata_path: return _pkgdir(self.opts.metadata_path, repo.id if not self.opts.norepopath else '') else: return self.repo_target(repo) def pkg_download_path(self, pkg): repo_target = self.repo_target(pkg.repo) pkg_download_path = os.path.realpath( os.path.join(repo_target, pkg.location)) # join() ensures repo_target ends with a path separator (otherwise the # check would pass if pkg_download_path was a "sibling" path component # of repo_target that has the same prefix). if not pkg_download_path.startswith(os.path.join(repo_target, '')): raise dnf.exceptions.Error( _("Download target '{}' is outside of download path '{}'.").format( pkg_download_path, repo_target)) return pkg_download_path def delete_old_local_packages(self, repo, pkglist): # delete any *.rpm file under target path, that was not downloaded from repository downloaded_files = set(self.pkg_download_path(pkg) for pkg in pkglist) for dirpath, dirnames, filenames in os.walk(self.repo_target(repo)): for filename in filenames: path = os.path.join(dirpath, filename) if filename.endswith('.rpm') and os.path.isfile(path): if path not in downloaded_files: # Delete disappeared or relocated file try: os.unlink(path) logger.info(_("[DELETED] %s"), path) except OSError: logger.error(_("failed to delete file %s"), path) def getcomps(self, repo): comps_fn = repo._repo.getCompsFn() if comps_fn: dest_path = self.metadata_target(repo) dnf.util.ensure_dir(dest_path) dest = os.path.join(dest_path, 'comps.xml') dnf.yum.misc.decompress(comps_fn, dest=dest) logger.info(_("comps.xml for repository %s saved"), repo.id) def download_metadata(self, repo): repo_target = self.metadata_target(repo) repo._repo.downloadMetadata(repo_target) return True def _get_latest(self, query): """ return union of these queries: - the latest NEVRAs from non-modular packages - all packages from stream version with the latest package NEVRA (this should not be needed but the latest package NEVRAs might be part of an older module version) - all packages from the latest stream version """ if not dnf.base.WITH_MODULES: return query.latest() query.apply() module_packages = self.base._moduleContainer.getModulePackages() all_artifacts = set() module_dict = {} # {NameStream: {Version: [modules]}} artifact_version = {} # {artifact: {NameStream: [Version]}} for module_package in module_packages: artifacts = module_package.getArtifacts() all_artifacts.update(artifacts) module_dict.setdefault(module_package.getNameStream(), {}).setdefault( module_package.getVersionNum(), []).append(module_package) for artifact in artifacts: artifact_version.setdefault(artifact, {}).setdefault( module_package.getNameStream(), []).append(module_package.getVersionNum()) # the latest NEVRAs from non-modular packages latest_query = query.filter( pkg__neq=query.filter(nevra_strict=all_artifacts)).latest() # artifacts from the newest version and those versions that contain an artifact # with the highest NEVRA latest_stream_artifacts = set() for namestream, version_dict in module_dict.items(): # versions that will be synchronized versions = set() # add the newest stream version versions.add(sorted(version_dict.keys(), reverse=True)[0]) # collect all artifacts in all stream versions stream_artifacts = set() for modules in version_dict.values(): for module in modules: stream_artifacts.update(module.getArtifacts()) # find versions to which the packages with the highest NEVRAs belong for latest_pkg in query.filter(nevra_strict=stream_artifacts).latest(): # here we depend on modules.yaml allways containing full NEVRA (including epoch) nevra = "{0.name}-{0.epoch}:{0.version}-{0.release}.{0.arch}".format(latest_pkg) # download only highest version containing the latest artifact versions.add(max(artifact_version[nevra][namestream])) # add all artifacts from selected versions for synchronization for version in versions: for module in version_dict[version]: latest_stream_artifacts.update(module.getArtifacts()) latest_query = latest_query.union(query.filter(nevra_strict=latest_stream_artifacts)) return latest_query def get_pkglist(self, repo): query = self.base.sack.query(flags=hawkey.IGNORE_MODULAR_EXCLUDES).available().filterm( reponame=repo.id) if self.opts.newest_only: query = self._get_latest(query) if self.opts.source: query.filterm(arch='src') elif self.opts.arches: query.filterm(arch=self.opts.arches) # skip packages that would have been downloaded to the same location pkglist = [] seen_paths = set() for pkg in query: download_path = self.pkg_download_path(pkg) if download_path not in seen_paths: pkglist.append(pkg) seen_paths.add(download_path) return pkglist def download_packages(self, pkglist): base = self.base progress = base.output.progress if progress is None: progress = dnf.callback.NullDownloadProgress() drpm = dnf.drpm.DeltaInfo(base.sack.query(flags=hawkey.IGNORE_MODULAR_EXCLUDES).installed(), progress, 0) payloads = [RPMPayloadLocation(pkg, progress, self.pkg_download_path(pkg)) for pkg in pkglist] base._download_remote_payloads(payloads, drpm, progress, None, False) def print_urls(self, pkglist): for pkg in pkglist: url = pkg.remote_location() if url: print(url) else: msg = _("Failed to get mirror for package: %s") % pkg.name logger.warning(msg)	14.67 KB	2025-11-12 04:27
📄 spacewalk.py # # Copyright (C) 2015 Red Hat, Inc. # # This copyrighted material is made available to anyone wishing to use, # modify, copy, or redistribute it subject to the terms and conditions of # the GNU General Public License v.2, or (at your option) any later version. # This program is distributed in the hope that it will be useful, but WITHOUT # ANY WARRANTY expressed or implied, including the implied warranties 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, write to the # Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA # 02110-1301, USA. Any Red Hat trademarks that are incorporated in the # source code or documentation are not subject to the GNU General Public # License and may only be used or replicated with the express permission of # Red Hat, Inc. # from __future__ import absolute_import from __future__ import unicode_literals from dnfpluginscore import _, logger import dnf import dnf.exceptions import errno import json import librepo import os from copy import copy from dnf.conf.config import PRIO_PLUGINCONFIG import up2date_client.up2dateAuth import up2date_client.config import up2date_client.rhnChannel import up2date_client.rhnPackageInfo from rhn.i18n import ustr from up2date_client import up2dateErrors STORED_CHANNELS_NAME = '_spacewalk.json' RHN_DISABLED = _("CloudLinux Network based repositories will be disabled.") CHANNELS_DISABLED = _("CloudLinux Network channel support will be disabled.") COMMUNICATION_ERROR = _("There was an error communicating with CloudLinux Network server.") NOT_REGISTERED_ERROR = _("This system is not registered with CloudLinux Network server.") NOT_SUBSCRIBED_ERROR = _("This system is not subscribed to any channels.") NO_SYSTEM_ID_ERROR = _("SystemId could not be acquired.") USE_RHNREGISTER = _("You can use rhn_register to register.") UPDATES_FROM_SPACEWALK = _("This system is receiving updates from CloudLinux Network server.") GPG_KEY_REJECTED = _("For security reasons packages from CloudLinux Network based repositories can be verified only with locally installed gpg keys. GPG key '%s' has been rejected.") PROFILE_NOT_SENT = _("Package profile information could not be sent.") MISSING_HEADER = _("Missing required login information for CloudLinux Network: %s") LEAPP_IN_PROGRESS = _("Leapp upgrade is running - using cache.") MUST_BE_ROOT = _('Spacewalk plugin has to be run under with the root privileges.') class Spacewalk(dnf.Plugin): name = 'spacewalk' def __init__(self, base, cli): super(Spacewalk, self).__init__(base, cli) self.base = base self.cli = cli self.stored_channels_path = os.path.join(self.base.conf.persistdir, STORED_CHANNELS_NAME) self.connected_to_spacewalk = False self.up2date_cfg = {} self.conf = copy(self.base.conf) self.parser = self.read_config(self.conf) if "main" in self.parser.sections(): options = self.parser.items("main") for (key, value) in options: self.conf._set_value(key, value, PRIO_PLUGINCONFIG) if not dnf.util.am_i_root(): logger.warning(MUST_BE_ROOT) self.conf.enabled = False if not self.conf.enabled: return logger.debug('initialized Spacewalk plugin') self.activate_channels() def config(self): if not self.conf.enabled: return # cli is None when plugin is executed # though automatic actions or other pluings # like in case of leapp which uses custom dnf plugin if not self.cli: return self.cli.demands.root_user = True def clnreg(self): os.system('/usr/sbin/clnreg_ks --strict-edition') def activate_channels(self, networking=True): enabled_channels = {} sslcacert = None force_http = 0 proxy_url = None login_info = None cached_channels = self._read_channels_file() if not networking: # no network communication, use list of channels from persistdir enabled_channels = cached_channels elif os.path.isfile("/etc/cln_leapp_in_progress"): # networking is true, but CLN urls won't be accessible, use cache logger.warning(LEAPP_IN_PROGRESS) enabled_channels = cached_channels else: # setup proxy according to up2date self.up2date_cfg = up2date_client.config.initUp2dateConfig() sslcacert = get_ssl_ca_cert(self.up2date_cfg) force_http = self.up2date_cfg['useNoSSLForPackages'], # trying to register system once in case of error while getLoginInfo clnreg_tried = False while not clnreg_tried: try: login_info = up2date_client.up2dateAuth.getLoginInfo(timeout=self.conf.timeout) clnreg_tried = True except up2dateErrors.RhnServerException as e: if clnreg_tried == False: self.clnreg() clnreg_tried = True continue logger.error("%s\n%s\n%s", COMMUNICATION_ERROR, RHN_DISABLED, e) return if not login_info: logger.error("%s\n%s", NOT_REGISTERED_ERROR, RHN_DISABLED) self._write_channels_file({}) return try: svrChannels = up2date_client.rhnChannel.getChannelDetails( timeout=self.conf.timeout) except up2dateErrors.CommunicationError as e: logger.error("%s\n%s\n%s", COMMUNICATION_ERROR, RHN_DISABLED, e) return except up2dateErrors.NoChannelsError: logger.error("%s\n%s", NOT_SUBSCRIBED_ERROR, CHANNELS_DISABLED) self._write_channels_file({}) return except up2dateErrors.NoSystemIdError: logger.error("%s %s\n%s\n%s", NOT_SUBSCRIBED_ERROR, NO_SYSTEM_ID_ERROR, USE_RHNREGISTER, RHN_DISABLED) return self.connected_to_spacewalk = True logger.info(UPDATES_FROM_SPACEWALK) for channel in svrChannels: if channel['version']: enabled_channels[channel['label']] = dict(channel.items()) self._write_channels_file(enabled_channels) repos = self.base.repos for (channel_id, channel_dict) in enabled_channels.items(): cached_channel = cached_channels.get(channel_id) cached_version = None if cached_channel: cached_version = cached_channel.get('version') conf = copy(self.conf) if channel_id in self.parser.sections(): options = self.parser.items(channel_id) for (key, value) in options: conf._set_value(key, value, PRIO_PLUGINCONFIG) repo = SpacewalkRepo(channel_dict, { 'conf' : self.base.conf, 'proxy' : proxy_url, 'timeout' : conf.timeout, 'sslcacert' : sslcacert, 'force_http': force_http, 'cached_version' : cached_version, 'login_info': login_info, 'gpgcheck': conf.gpgcheck, 'enabled': conf.enabled, }) repos.add(repo) # DEBUG logger.debug(enabled_channels) def transaction(self): """ Update system's profile after transaction. """ if not self.conf.enabled: return if not self.connected_to_spacewalk: # not connected so nothing to do here return if self.up2date_cfg['writeChangesToLog'] == 1: delta = self._make_package_delta() up2date_client.rhnPackageInfo.logDeltaPackages(delta) try: up2date_client.rhnPackageInfo.updatePackageProfile( timeout=self.conf.timeout) except up2dateErrors.RhnServerException as e: logger.error("%s\n%s\n%s", COMMUNICATION_ERROR, PROFILE_NOT_SENT, e) def _read_channels_file(self): try: with open(self.stored_channels_path, "r") as channels_file: content = channels_file.read() channels = json.loads(content) return channels except (FileNotFoundError, IOError) as e: if e.errno != errno.ENOENT: raise except json.decoder.JSONDecodeError as e: pass # ignore broken json and recreate it later return {} def _write_channels_file(self, var): try: with open(self.stored_channels_path, "w") as channels_file: json.dump(var, channels_file, indent=4) except (FileNotFoundError, IOError) as e: if e.errno != errno.ENOENT: raise def _make_package_delta(self): delta = {'added' : [(p.name, p.version, p. release, p.epoch, p.arch) for p in self.base.transaction.install_set], 'removed': [(p.name, p.version, p. release, p.epoch, p.arch) for p in self.base.transaction.remove_set], } return delta class SpacewalkRepo(dnf.repo.Repo): """ Repository object for Spacewalk. Uses up2date libraries. """ needed_headers = ['X-RHN-Server-Id', 'X-RHN-Auth-User-Id', 'X-RHN-Auth', 'X-RHN-Auth-Server-Time', 'X-RHN-Auth-Expire-Offset'] def __init__(self, channel, opts): super(SpacewalkRepo, self).__init__(ustr(channel['label']), opts.get('conf')) # dnf stuff self.name = ustr(channel['name']) self.baseurl = [ url + '/GET-REQ/' + self.id for url in channel['url']] self.sslcacert = opts.get('sslcacert') self.proxy = opts.get('proxy') try: self.gpgkey = get_gpg_key_urls(channel['gpg_key_url']) except InvalidGpgKeyLocation as e: logger.warning(GPG_KEY_REJECTED, dnf.i18n.ucd(e)) self.gpgkey = [] if channel['version'] != opts.get('cached_version'): self.metadata_expire = 1 # spacewalk stuff self.login_info = opts.get('login_info') self.keepalive = 0 self.bandwidth = 0 self.retries = 1 self.throttle = 0 self.timeout = opts.get('timeout') self.gpgcheck = opts.get('gpgcheck') self.force_http = opts.get('force_http') if opts.get('enabled'): self.enable() else: self.disable() if hasattr(self, 'set_http_headers'): # dnf > 4.0.9 on RHEL 8, Fedora 29/30 http_headers = self.create_http_headers() if http_headers: self.set_http_headers(http_headers) def create_http_headers(self): http_headers = [] if not self.login_info: return http_headers for header in self.needed_headers: if not header in self.login_info: error = MISSING_HEADER % header raise dnf.Error.RepoError(error) if self.login_info[header] in (None, ''): # This doesn't work due to bug in librepo (or even deeper in libcurl) # the workaround bellow can be removed once BZ#1211662 is fixed #http_headers.append("%s;" % header) http_headers.append("%s: \r\nX-libcurl-Empty-Header-Workaround: *" % header) else: http_headers.append("%s: %s" % (header, self.login_info[header])) if not self.force_http: http_headers.append("X-RHN-Transport-Capability: follow-redirects=3") return http_headers def _handle_new_remote(self, destdir, mirror_setup=True): # this function is called only on dnf < 3.6.0 (up to Fedora 29) handle = super(SpacewalkRepo, self)._handle_new_remote(destdir, mirror_setup) http_headers = self.create_http_headers() if http_headers: handle.setopt(librepo.LRO_HTTPHEADER, http_headers) return handle # FIXME # all rutines bellow should go to rhn-client-tools because they are share # between yum-rhn-plugin and dnf-plugin-spacewalk def get_gpg_key_urls(key_url_string): """ Parse the key urls and validate them. key_url_string is a space seperated list of gpg key urls that must be located in /etc/pkg/rpm-gpg/. Return a list of strings containing the key urls. Raises InvalidGpgKeyLocation if any of the key urls are invalid. """ key_urls = key_url_string.split() for key_url in key_urls: if not is_valid_gpg_key_url(key_url): raise InvalidGpgKeyLocation(key_url) return key_urls class InvalidGpgKeyLocation(Exception): pass def is_valid_gpg_key_url(key_url): proto_split = key_url.split('://') if len(proto_split) != 2: return False proto, path = proto_split if proto.lower() != 'file': return False path = os.path.normpath(path) if not path.startswith('/etc/pki/rpm-gpg/'): return False return True def get_ssl_ca_cert(up2date_cfg): if not ('sslCACert' in up2date_cfg and up2date_cfg['sslCACert']): raise BadSslCaCertConfig ca_certs = up2date_cfg['sslCACert'] if type(ca_certs) == list: return ca_certs[0] return ca_certs	13.88 KB	2025-02-26 09:32
📄 system_upgrade.py # -- coding: utf-8 -- # # Copyright (c) 2015-2020 Red Hat, Inc. # # 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; either version 2 of the License, or # (at your option) any later version. # # 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 <http://www.gnu.org/licenses/>. # # Author(s): Will Woods <wwoods@redhat.com> """system_upgrade.py - DNF plugin to handle major-version system upgrades.""" from subprocess import call, Popen, check_output, CalledProcessError import json import os import os.path import re import sys import uuid from systemd import journal from dnfpluginscore import _, logger import dnf import dnf.cli from dnf.cli import CliError from dnf.i18n import ucd import dnf.transaction from dnf.transaction_sr import serialize_transaction, TransactionReplay import libdnf.conf # Translators: This string is only used in unit tests. _("the color of the sky") DOWNLOAD_FINISHED_ID = uuid.UUID('9348174c5cc74001a71ef26bd79d302e') REBOOT_REQUESTED_ID = uuid.UUID('fef1cc509d5047268b83a3a553f54b43') UPGRADE_STARTED_ID = uuid.UUID('3e0a5636d16b4ca4bbe5321d06c6aa62') UPGRADE_FINISHED_ID = uuid.UUID('8cec00a1566f4d3594f116450395f06c') ID_TO_IDENTIFY_BOOTS = UPGRADE_STARTED_ID PLYMOUTH = '/usr/bin/plymouth' RELEASEVER_MSG = _( "Need a --releasever greater than the current system version.") DOWNLOAD_FINISHED_MSG = _( # Translators: do not change "reboot" here "Download complete! Use 'dnf {command} reboot' to start the upgrade.\n" "To remove cached metadata and transaction use 'dnf {command} clean'") CANT_RESET_RELEASEVER = _( "Sorry, you need to use 'download --releasever' instead of '--network'") STATE_VERSION = 3 # --- Miscellaneous helper functions ------------------------------------------ def reboot(poweroff = False): if os.getenv("DNF_SYSTEM_UPGRADE_NO_REBOOT", default=False): logger.info(_("Reboot turned off, not rebooting.")) else: if poweroff: Popen(["systemctl", "poweroff"]) else: Popen(["systemctl", "reboot"]) def get_url_from_os_release(): key = "UPGRADE_GUIDE_URL=" for path in ["/etc/os-release", "/usr/lib/os-release"]: try: with open(path) as release_file: for line in release_file: line = line.strip() if line.startswith(key): return line[len(key):].strip('"') except IOError: continue return None # DNF-FIXME: dnf.util.clear_dir() doesn't delete regular files :/ def clear_dir(path, ignore=[]): if not os.path.isdir(path): return for entry in os.listdir(path): fullpath = os.path.join(path, entry) if fullpath in ignore: continue try: if os.path.isdir(fullpath): dnf.util.rm_rf(fullpath) else: os.unlink(fullpath) except OSError: pass def check_release_ver(conf, target=None): if dnf.rpm.detect_releasever(conf.installroot) == conf.releasever: raise CliError(RELEASEVER_MSG) if target and target != conf.releasever: # it's too late to set releasever here, so this can't work. # (see https://bugzilla.redhat.com/show_bug.cgi?id=1212341) raise CliError(CANT_RESET_RELEASEVER) def disable_blanking(): try: tty = open('/dev/tty0', 'wb') tty.write(b'\33[9;0]') except Exception as e: print(_("Screen blanking can't be disabled: %s") % e) # --- State object - for tracking upgrade state between runs ------------------ # DNF-INTEGRATION-NOTE: basically the same thing as dnf.persistor.JSONDB class State(object): def __init__(self, statefile): self.statefile = statefile self._data = {} self._read() def _read(self): try: with open(self.statefile) as fp: self._data = json.load(fp) except IOError: self._data = {} except ValueError: self._data = {} logger.warning(_("Failed loading state file: %s, continuing with " "empty state."), self.statefile) def write(self): dnf.util.ensure_dir(os.path.dirname(self.statefile)) with open(self.statefile, 'w') as outf: json.dump(self._data, outf, indent=4, sort_keys=True) def clear(self): if os.path.exists(self.statefile): os.unlink(self.statefile) self._read() def __enter__(self): return self def __exit__(self, exc_type, exc_value, traceback): if exc_type is None: self.write() # helper function for creating properties. pylint: disable=protected-access def _prop(option): # pylint: disable=no-self-argument def setprop(self, value): self._data[option] = value def getprop(self): return self._data.get(option) return property(getprop, setprop) # !!! Increase STATE_VERSION for any changes in data structure like a new property or a new # data structure !!! state_version = _prop("state_version") download_status = _prop("download_status") destdir = _prop("destdir") target_releasever = _prop("target_releasever") system_releasever = _prop("system_releasever") gpgcheck = _prop("gpgcheck") # list of repos with gpgcheck=True gpgcheck_repos = _prop("gpgcheck_repos") # list of repos with repo_gpgcheck=True repo_gpgcheck_repos = _prop("repo_gpgcheck_repos") upgrade_status = _prop("upgrade_status") upgrade_command = _prop("upgrade_command") distro_sync = _prop("distro_sync") poweroff_after = _prop("poweroff_after") enable_disable_repos = _prop("enable_disable_repos") module_platform_id = _prop("module_platform_id") # --- Plymouth output helpers ------------------------------------------------- class PlymouthOutput(object): """A plymouth output helper class. Filters duplicate calls, and stops calling the plymouth binary if we fail to contact it. """ def __init__(self): self.alive = True self._last_args = dict() self._last_msg = None def _plymouth(self, cmd, args): dupe_cmd = (args == self._last_args.get(cmd)) if (self.alive and not dupe_cmd) or cmd == '--ping': try: self.alive = (call((PLYMOUTH, cmd) + args) == 0) except OSError: self.alive = False self._last_args[cmd] = args return self.alive def ping(self): return self._plymouth("--ping") def message(self, msg): if self._last_msg and self._last_msg != msg: self._plymouth("hide-message", "--text", self._last_msg) self._last_msg = msg return self._plymouth("display-message", "--text", msg) def set_mode(self): mode = 'updates' try: s = check_output([PLYMOUTH, '--help']) if re.search('--system-upgrade', ucd(s)): mode = 'system-upgrade' except (CalledProcessError, OSError): pass return self._plymouth("change-mode", "--" + mode) def progress(self, percent): return self._plymouth("system-update", "--progress", str(percent)) # A single PlymouthOutput instance for us to use within this module Plymouth = PlymouthOutput() # A TransactionProgress class that updates plymouth for us. class PlymouthTransactionProgress(dnf.callback.TransactionProgress): # pylint: disable=too-many-arguments def progress(self, package, action, ti_done, ti_total, ts_done, ts_total): self._update_plymouth(package, action, ts_done, ts_total) def _update_plymouth(self, package, action, current, total): # Prevents quick jumps of progressbar when pretrans scriptlets # and TRANS_PREPARATION are reported as 1/1 if total == 1: return # Verification goes through all the packages again, # which resets the "current" param value, this prevents # resetting of the progress bar as well. (Rhbug:1809096) if action != dnf.callback.PKG_VERIFY: Plymouth.progress(int(90.0 current / total)) else: Plymouth.progress(90 + int(10.0 * current / total)) Plymouth.message(self._fmt_event(package, action, current, total)) def _fmt_event(self, package, action, current, total): action = dnf.transaction.ACTIONS.get(action, action) return "[%d/%d] %s %s..." % (current, total, action, package) # --- journal helpers ------------------------------------------------- def find_boots(message_id): """Find all boots with this message id. Returns the entries of all found boots. """ j = journal.Reader() j.add_match(MESSAGE_ID=message_id.hex, # identify the message _UID=0) # prevent spoofing of logs oldboot = None for entry in j: boot = entry['_BOOT_ID'] if boot == oldboot: continue oldboot = boot yield entry def list_logs(): print(_('The following boots appear to contain upgrade logs:')) n = -1 for n, entry in enumerate(find_boots(ID_TO_IDENTIFY_BOOTS)): print('{} / {.hex}: {:%Y-%m-%d %H:%M:%S} {}→{}'.format( n + 1, entry['_BOOT_ID'], entry['__REALTIME_TIMESTAMP'], entry.get('SYSTEM_RELEASEVER', '??'), entry.get('TARGET_RELEASEVER', '??'))) if n == -1: print(_('-- no logs were found --')) def pick_boot(message_id, n): boots = list(find_boots(message_id)) # Positive indices index all found boots starting with 1 and going forward, # zero is the current boot, and -1, -2, -3 are previous going backwards. # This is the same as journalctl. try: if n == 0: raise IndexError if n > 0: n -= 1 return boots[n]['_BOOT_ID'] except IndexError: raise CliError(_("Cannot find logs with this index.")) def show_log(n): boot_id = pick_boot(ID_TO_IDENTIFY_BOOTS, n) process = Popen(['journalctl', '--boot', boot_id.hex]) process.wait() rc = process.returncode if rc == 1: raise dnf.exceptions.Error(_("Unable to match systemd journal entry")) CMDS = ['download', 'clean', 'reboot', 'upgrade', 'log'] # --- The actual Plugin and Command objects! ---------------------------------- class SystemUpgradePlugin(dnf.Plugin): name = 'system-upgrade' def __init__(self, base, cli): super(SystemUpgradePlugin, self).__init__(base, cli) if cli: cli.register_command(SystemUpgradeCommand) cli.register_command(OfflineUpgradeCommand) cli.register_command(OfflineDistrosyncCommand) class SystemUpgradeCommand(dnf.cli.Command): aliases = ('system-upgrade', 'fedup',) summary = _("Prepare system for upgrade to a new release") DATADIR = 'var/lib/dnf/system-upgrade' def __init__(self, cli): super(SystemUpgradeCommand, self).__init__(cli) self.datadir = os.path.join(cli.base.conf.installroot, self.DATADIR) self.transaction_file = os.path.join(self.datadir, 'system-upgrade-transaction.json') self.magic_symlink = os.path.join(cli.base.conf.installroot, 'system-update') self.state = State(os.path.join(self.datadir, 'system-upgrade-state.json')) @staticmethod def set_argparser(parser): parser.add_argument("--no-downgrade", dest='distro_sync', action='store_false', help=_("keep installed packages if the new " "release's version is older")) parser.add_argument('--poweroff', dest='poweroff_after', action='store_true', help=_("power off system after the operation " "is completed")) parser.add_argument('tid', nargs=1, choices=CMDS, metavar="[%s]" % "\|".join(CMDS)) parser.add_argument('--number', type=int, help=_('which logs to show')) def log_status(self, message, message_id): """Log directly to the journal.""" journal.send(message, MESSAGE_ID=message_id, PRIORITY=journal.LOG_NOTICE, SYSTEM_RELEASEVER=self.state.system_releasever, TARGET_RELEASEVER=self.state.target_releasever, DNF_VERSION=dnf.const.VERSION) def pre_configure(self): self._call_sub("check") self._call_sub("pre_configure") def configure(self): self._call_sub("configure") def run(self): self._call_sub("run") def run_transaction(self): self._call_sub("transaction") def run_resolved(self): self._call_sub("resolved") def _call_sub(self, name): subfunc = getattr(self, name + '_' + self.opts.tid[0], None) if callable(subfunc): subfunc() def _check_state_version(self, command): if self.state.state_version != STATE_VERSION: msg = _("Incompatible version of data. Rerun 'dnf {command} download [OPTIONS]'" "").format(command=command) raise CliError(msg) def _set_cachedir(self): # set download directories from json state file self.base.conf.cachedir = self.datadir self.base.conf.destdir = self.state.destdir if self.state.destdir else None def _get_forward_reverse_pkg_reason_pairs(self): """ forward = {repoid:{pkg_nevra: {tsi.action: tsi.reason}} reverse = {pkg_nevra: {tsi.action: tsi.reason}} :return: forward, reverse """ backward_action = set(dnf.transaction.BACKWARD_ACTIONS + [libdnf.transaction.TransactionItemAction_REINSTALLED]) forward_actions = set(dnf.transaction.FORWARD_ACTIONS) forward = {} reverse = {} for tsi in self.cli.base.transaction: if tsi.action in forward_actions: pkg = tsi.pkg forward.setdefault(pkg.repo.id, {}).setdefault( str(pkg), {})[tsi.action] = tsi.reason elif tsi.action in backward_action: reverse.setdefault(str(tsi.pkg), {})[tsi.action] = tsi.reason return forward, reverse # == pre_configure_: set up action-specific demands ========================== def pre_configure_download(self): # only download subcommand accepts --destdir command line option self.base.conf.cachedir = self.datadir self.base.conf.destdir = self.opts.destdir if self.opts.destdir else None if 'offline-distrosync' == self.opts.command and not self.opts.distro_sync: raise CliError( _("Command 'offline-distrosync' cannot be used with --no-downgrade option")) elif 'offline-upgrade' == self.opts.command: self.opts.distro_sync = False def pre_configure_reboot(self): self._set_cachedir() def pre_configure_upgrade(self): self._set_cachedir() if self.state.enable_disable_repos: self.opts.repos_ed = self.state.enable_disable_repos self.base.conf.releasever = self.state.target_releasever def pre_configure_clean(self): self._set_cachedir() # == configure_: set up action-specific demands ========================== def configure_download(self): if 'system-upgrade' == self.opts.command or 'fedup' == self.opts.command: logger.warning(_('WARNING: this operation is not supported on the RHEL distribution. ' 'Proceed at your own risk.')) help_url = get_url_from_os_release() if help_url: msg = _('Additional information for System Upgrade: {}') logger.info(msg.format(ucd(help_url))) if self.base._promptWanted(): msg = _('Before you continue ensure that your system is fully upgraded by running ' '"dnf --refresh upgrade". Do you want to continue') if self.base.conf.assumeno or not self.base.output.userconfirm( msg='{} [y/N]: '.format(msg), defaultyes_msg='{} [Y/n]: '.format(msg)): logger.error(_("Operation aborted.")) sys.exit(1) check_release_ver(self.base.conf, target=self.opts.releasever) elif 'offline-upgrade' == self.opts.command: self.cli._populate_update_security_filter(self.opts) self.cli.demands.root_user = True self.cli.demands.resolving = True self.cli.demands.available_repos = True self.cli.demands.sack_activation = True self.cli.demands.freshest_metadata = True # We want to do the depsolve / download / transaction-test, but not # run the actual RPM transaction to install the downloaded packages. # Setting the "test" flag makes the RPM transaction a test transaction, # so nothing actually gets installed. # (It also means that we run two test transactions in a row, which is # kind of silly, but that's something for DNF to fix...) self.base.conf.tsflags += ["test"] def configure_reboot(self): # FUTURE: add a --debug-shell option to enable debug shell: # systemctl add-wants system-update.target debug-shell.service self.cli.demands.root_user = True def configure_upgrade(self): # same as the download, but offline and non-interactive. so... self.cli.demands.root_user = True self.cli.demands.resolving = True self.cli.demands.available_repos = True self.cli.demands.sack_activation = True # use the saved value for --allowerasing, etc. self.opts.distro_sync = self.state.distro_sync if self.state.gpgcheck is not None: self.base.conf.gpgcheck = self.state.gpgcheck if self.state.gpgcheck_repos is not None: for repo in self.base.repos.values(): repo.gpgcheck = repo.id in self.state.gpgcheck_repos if self.state.repo_gpgcheck_repos is not None: for repo in self.base.repos.values(): repo.repo_gpgcheck = repo.id in self.state.repo_gpgcheck_repos self.base.conf.module_platform_id = self.state.module_platform_id # don't try to get new metadata, 'cuz we're offline self.cli.demands.cacheonly = True # and don't ask any questions (we confirmed all this beforehand) self.base.conf.assumeyes = True self.cli.demands.transaction_display = PlymouthTransactionProgress() # upgrade operation already removes all element that must be removed. Additional removal # could trigger unwanted changes in transaction. self.base.conf.clean_requirements_on_remove = False self.base.conf.install_weak_deps = False def configure_clean(self): self.cli.demands.root_user = True def configure_log(self): pass # == check_: do any action-specific checks =============================== def check_reboot(self): if not self.state.download_status == 'complete': raise CliError(_("system is not ready for upgrade")) self._check_state_version(self.opts.command) if self.state.upgrade_command != self.opts.command: msg = _("the transaction was not prepared for '{command}'. " "Rerun 'dnf {command} download [OPTIONS]'").format(command=self.opts.command) raise CliError(msg) if os.path.lexists(self.magic_symlink): raise CliError(_("upgrade is already scheduled")) dnf.util.ensure_dir(self.datadir) # FUTURE: checkRPMDBStatus(self.state.download_transaction_id) def check_upgrade(self): if not os.path.lexists(self.magic_symlink): logger.info(_("trigger file does not exist. exiting quietly.")) raise SystemExit(0) if os.readlink(self.magic_symlink) != self.datadir: logger.info(_("another upgrade tool is running. exiting quietly.")) raise SystemExit(0) # Delete symlink ASAP to avoid reboot loops dnf.yum.misc.unlink_f(self.magic_symlink) command = self.state.upgrade_command if not command: command = self.opts.command self._check_state_version(command) if not self.state.upgrade_status == 'ready': msg = _("use 'dnf {command} reboot' to begin the upgrade").format(command=command) raise CliError(msg) # == run_: run the action/prep the transaction =========================== def run_prepare(self): # make the magic symlink os.symlink(self.datadir, self.magic_symlink) # set upgrade_status so that the upgrade can run with self.state as state: state.upgrade_status = 'ready' def run_reboot(self): self.run_prepare() if not self.opts.tid[0] == "reboot": return self.state.poweroff_after = self.opts.poweroff_after self.log_status(_("Rebooting to perform upgrade."), REBOOT_REQUESTED_ID) # Explicit write since __exit__ doesn't seem to get called when rebooting self.state.write() reboot() def run_download(self): # Mark everything in the world for upgrade/sync if self.opts.distro_sync: self.base.distro_sync() else: self.base.upgrade_all() if self.opts.command not in ['offline-upgrade', 'offline-distrosync']: # Mark all installed groups and environments for upgrade self.base.read_comps() installed_groups = [g.id for g in self.base.comps.groups if self.base.history.group.get(g.id)] if installed_groups: self.base.env_group_upgrade(installed_groups) installed_environments = [g.id for g in self.base.comps.environments if self.base.history.env.get(g.id)] if installed_environments: self.base.env_group_upgrade(installed_environments) with self.state as state: state.download_status = 'downloading' state.target_releasever = self.base.conf.releasever state.destdir = self.base.conf.destdir def run_upgrade(self): # change the upgrade status (so we can detect crashed upgrades later) command = '' with self.state as state: state.upgrade_status = 'incomplete' command = state.upgrade_command if command == 'offline-upgrade': msg = _("Starting offline upgrade. This will take a while.") elif command == 'offline-distrosync': msg = _("Starting offline distrosync. This will take a while.") else: msg = _("Starting system upgrade. This will take a while.") self.log_status(msg, UPGRADE_STARTED_ID) # reset the splash mode and let the user know we're running Plymouth.set_mode() Plymouth.progress(0) Plymouth.message(msg) # disable screen blanking disable_blanking() self.replay = TransactionReplay(self.base, self.transaction_file) self.replay.run() def run_clean(self): logger.info(_("Cleaning up downloaded data...")) # Don't delete persistor, it contains paths for downloaded packages # that are used by dnf during finalizing base to clean them up clear_dir(self.base.conf.cachedir, [dnf.persistor.TempfilePersistor(self.base.conf.cachedir).db_path]) with self.state as state: state.download_status = None state.state_version = None state.upgrade_status = None state.upgrade_command = None state.destdir = None def run_log(self): if self.opts.number: show_log(self.opts.number) else: list_logs() # == resolved_: do staff after succesful resolvement ===================== def resolved_upgrade(self): """Adjust transaction reasons according to stored values""" self.replay.post_transaction() # == transaction_: do stuff after a successful transaction =============== def transaction_download(self): transaction = self.base.history.get_current() if not transaction.packages(): logger.info(_("The system-upgrade transaction is empty, your system is already up-to-date.")) return data = serialize_transaction(transaction) try: with open(self.transaction_file, "w") as f: json.dump(data, f, indent=4, sort_keys=True) f.write("\n") print(_("Transaction saved to {}.").format(self.transaction_file)) except OSError as e: raise dnf.cli.CliError(_('Error storing transaction: {}').format(str(e))) # Okay! Write out the state so the upgrade can use it. system_ver = dnf.rpm.detect_releasever(self.base.conf.installroot) with self.state as state: state.download_status = 'complete' state.state_version = STATE_VERSION state.distro_sync = self.opts.distro_sync state.gpgcheck = self.base.conf.gpgcheck state.gpgcheck_repos = [ repo.id for repo in self.base.repos.values() if repo.gpgcheck] state.repo_gpgcheck_repos = [ repo.id for repo in self.base.repos.values() if repo.repo_gpgcheck] state.system_releasever = system_ver state.target_releasever = self.base.conf.releasever state.module_platform_id = self.base.conf.module_platform_id state.enable_disable_repos = self.opts.repos_ed state.destdir = self.base.conf.destdir state.upgrade_command = self.opts.command msg = DOWNLOAD_FINISHED_MSG.format(command=self.opts.command) logger.info(msg) self.log_status(_("Download finished."), DOWNLOAD_FINISHED_ID) def transaction_upgrade(self): if self.state.poweroff_after: upgrade_complete_msg = _("Upgrade complete! Cleaning up and powering off...") else: upgrade_complete_msg = _("Upgrade complete! Cleaning up and rebooting...") Plymouth.message(upgrade_complete_msg) self.log_status(upgrade_complete_msg, UPGRADE_FINISHED_ID) self.run_clean() if self.opts.tid[0] == "upgrade": reboot(self.state.poweroff_after) class OfflineUpgradeCommand(SystemUpgradeCommand): aliases = ('offline-upgrade',) summary = _("Prepare offline upgrade of the system") class OfflineDistrosyncCommand(SystemUpgradeCommand): aliases = ('offline-distrosync',) summary = _("Prepare offline distrosync of the system")	26.88 KB	2025-11-12 04:27
📄 versionlock.py # # Copyright (C) 2015 Red Hat, Inc. # # This copyrighted material is made available to anyone wishing to use, # modify, copy, or redistribute it subject to the terms and conditions of # the GNU General Public License v.2, or (at your option) any later version. # This program is distributed in the hope that it will be useful, but WITHOUT # ANY WARRANTY expressed or implied, including the implied warranties 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, write to the # Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA # 02110-1301, USA. Any Red Hat trademarks that are incorporated in the # source code or documentation are not subject to the GNU General Public # License and may only be used or replicated with the express permission of # Red Hat, Inc. # from __future__ import absolute_import from __future__ import unicode_literals from dnfpluginscore import _, logger import dnf import dnf.cli import dnf.exceptions import fnmatch import hawkey import os import tempfile import time import warnings NOT_READABLE = _('Unable to read version lock configuration: %s') NO_LOCKLIST = _('Locklist not set') ADDING_SPEC = _('Adding versionlock on:') EXCLUDING_SPEC = _('Adding exclude on:') EXISTING_SPEC = _('Package already locked in equivalent form:') ALREADY_LOCKED = _('Package {} is already locked') ALREADY_EXCLUDED = _('Package {} is already excluded') DELETING_SPEC = _('Deleting versionlock for:') NOTFOUND_SPEC = _('No package found for:') NO_VERSIONLOCK = _('Excludes from versionlock plugin were not applied') APPLY_LOCK = _('Versionlock plugin: number of lock rules from file "{}" applied: {}') APPLY_EXCLUDE = _('Versionlock plugin: number of exclude rules from file "{}" applied: {}') NEVRA_ERROR = _('Versionlock plugin: could not parse pattern:') locklist_fn = None class VersionLock(dnf.Plugin): name = 'versionlock' def __init__(self, base, cli): super(VersionLock, self).__init__(base, cli) self.base = base self.cli = cli if self.cli is not None: self.cli.register_command(VersionLockCommand) def config(self): global locklist_fn cp = self.read_config(self.base.conf) locklist_fn = (cp.has_section('main') and cp.has_option('main', 'locklist') and cp.get('main', 'locklist')) def locking_enabled(self): if self.cli is None: enabled = True # loaded via the api, not called by cli else: enabled = self.cli.demands.plugin_filtering_enabled if enabled is None: enabled = self.cli.demands.resolving return enabled def sack(self): if not self.locking_enabled(): logger.debug(NO_VERSIONLOCK) return excludes_query = self.base.sack.query().filter(empty=True) locked_query = self.base.sack.query().filter(empty=True) locked_names = set() # counter of applied rules [locked_count, excluded_count] count = [0, 0] for pat in _read_locklist(): excl = 0 if pat and pat[0] == '!': pat = pat[1:] excl = 1 possible_nevras = dnf.subject.Subject(pat).get_nevra_possibilities( forms=[hawkey.FORM_NEVRA, hawkey.FORM_NEVR, hawkey.FORM_NEV, hawkey.FORM_NA, hawkey.FORM_NAME]) if possible_nevras: count[excl] += 1 else: logger.error("%s %s", NEVRA_ERROR, pat) continue for nevra in possible_nevras: pat_query = nevra.to_query(self.base.sack) if excl: excludes_query = excludes_query.union(pat_query) else: locked_names.add(nevra.name) locked_query = locked_query.union(pat_query) if pat_query: break if count[1]: logger.debug(APPLY_EXCLUDE.format(locklist_fn, count[1])) if count[0]: logger.debug(APPLY_LOCK.format(locklist_fn, count[0])) if locked_names: all_versions = self.base.sack.query().filter(name__glob=list(locked_names)) other_versions = all_versions.difference(locked_query) excludes_query = excludes_query.union(other_versions) # exclude also anything that obsoletes the locked versions of packages obsoletes_query = self.base.sack.query().filterm(obsoletes=locked_query) # leave out obsoleters that are also part of locked versions (otherwise the obsoleter package # would not be installable at all) excludes_query = excludes_query.union(obsoletes_query.difference(locked_query)) excludes_query.filterm(reponame__neq=hawkey.SYSTEM_REPO_NAME) if excludes_query: self.base.sack.add_excludes(excludes_query) EXC_CMDS = ['exclude', 'add-!', 'add!'] DEL_CMDS = ['delete', 'del'] DEP_EXC_CMDS = ['blacklist'] ALL_CMDS = ['add', 'clear', 'list'] + EXC_CMDS + DEL_CMDS + DEP_EXC_CMDS class VersionLockCommand(dnf.cli.Command): aliases = ("versionlock",) summary = _("control package version locks") usage = "[add\|exclude\|list\|delete\|clear] [<package-nevr-spec>]" @staticmethod def set_argparser(parser): parser.add_argument("--raw", default=False, action='store_true', help=_("Use package specifications as they are, do not " "try to parse them")) parser.add_argument("subcommand", nargs='?', metavar="[add\|exclude\|list\|delete\|clear]") parser.add_argument("package", nargs='', metavar="[<package-nevr-spec>]") def configure(self): self.cli.demands.sack_activation = True self.cli.demands.available_repos = True def run(self): cmd = 'list' if self.opts.subcommand: if self.opts.subcommand not in ALL_CMDS: cmd = 'add' self.opts.package.insert(0, self.opts.subcommand) elif self.opts.subcommand in EXC_CMDS: cmd = 'exclude' elif self.opts.subcommand in DEP_EXC_CMDS: msg = _("Subcommand '{}' is deprecated. Use 'exclude' subcommand instead.").format( self.opts.subcommand) warnings.warn(msg, dnf.exceptions.DeprecationWarning, stacklevel=2) cmd = 'exclude' elif self.opts.subcommand in DEL_CMDS: cmd = 'delete' else: cmd = self.opts.subcommand if cmd == 'add': results = _search_locklist(self.opts.package) for entry, entry_cmd in results: if entry_cmd == '': _write_locklist(self.base, [entry], self.opts.raw, True, "\n# Added lock on %s\n" % time.ctime(), ADDING_SPEC, '') elif cmd != entry_cmd: raise dnf.exceptions.Error(ALREADY_EXCLUDED.format(entry)) else: logger.info("%s %s", EXISTING_SPEC, entry) elif cmd == 'exclude': results = _search_locklist(self.opts.package) for entry, entry_cmd in results: if entry_cmd == '': _write_locklist(self.base, [entry], self.opts.raw, False, "\n# Added exclude on %s\n" % time.ctime(), EXCLUDING_SPEC, '!') elif cmd != entry_cmd: raise dnf.exceptions.Error(ALREADY_LOCKED.format(entry)) else: logger.info("%s %s", EXISTING_SPEC, entry) elif cmd == 'list': for pat in _read_locklist(): print(pat) elif cmd == 'clear': if not locklist_fn: raise dnf.exceptions.Error(NO_LOCKLIST) with open(locklist_fn, 'w') as f: # open in write mode truncates file pass elif cmd == 'delete': if not locklist_fn: raise dnf.exceptions.Error(NO_LOCKLIST) dirname = os.path.dirname(locklist_fn) (out, tmpfilename) = tempfile.mkstemp(dir=dirname, suffix='.tmp') locked_specs = _read_locklist() count = 0 with os.fdopen(out, 'w', -1) as out: for ent in locked_specs: if _match(ent, self.opts.package): print("%s %s" % (DELETING_SPEC, ent)) count += 1 continue out.write(ent) out.write('\n') if not count: os.unlink(tmpfilename) else: os.chmod(tmpfilename, 0o644) os.rename(tmpfilename, locklist_fn) def _read_locklist(): locklist = [] try: if not locklist_fn: raise dnf.exceptions.Error(NO_LOCKLIST) with open(locklist_fn) as llfile: for line in llfile.readlines(): if line.startswith('#') or line.strip() == '': continue locklist.append(line.strip()) except IOError as e: raise dnf.exceptions.Error(NOT_READABLE % e) return locklist def _search_locklist(package): results = [] found = action = '' locked_specs = _read_locklist() for pkg in package: match = False for ent in locked_specs: found = action = '' if _match(ent, [pkg]): found = ent action = 'exclude' if ent.startswith('!') else 'add' results.append((found, action)) match = True if not match: results.append((pkg, action)) return results def _write_locklist(base, args, raw, try_installed, comment, info, prefix): specs = set() for pat in args: if raw: specs.add(pat) continue subj = dnf.subject.Subject(pat) pkgs = None if try_installed: pkgs = subj.get_best_query(dnf.sack._rpmdb_sack(base), with_nevra=True, with_provides=False, with_filenames=False) if not pkgs: pkgs = subj.get_best_query(base.sack, with_nevra=True, with_provides=False, with_filenames=False) if not pkgs: print("%s %s" % (NOTFOUND_SPEC, pat)) for pkg in pkgs: specs.add(pkgtup2spec(pkg.pkgtup)) if specs: try: if not locklist_fn: raise dnf.exceptions.Error(NO_LOCKLIST) with open(locklist_fn, 'a') as f: f.write(comment) for spec in specs: print("%s %s" % (info, spec)) f.write("%s%s\n" % (prefix, spec)) except IOError as e: raise dnf.exceptions.Error(NOT_READABLE % e) def _match(ent, patterns): ent = ent.lstrip('!') for pat in patterns: if ent == pat: return True try: n = hawkey.split_nevra(ent) except hawkey.ValueException: return False for name in ( '%s' % n.name, '%s.%s' % (n.name, n.arch), '%s-%s' % (n.name, n.version), '%s-%s-%s' % (n.name, n.version, n.release), '%s-%s:%s' % (n.name, n.epoch, n.version), '%s-%s-%s.%s' % (n.name, n.version, n.release, n.arch), '%s-%s:%s-%s' % (n.name, n.epoch, n.version, n.release), '%s:%s-%s-%s.%s' % (n.epoch, n.name, n.version, n.release, n.arch), '%s-%s:%s-%s.%s' % (n.name, n.epoch, n.version, n.release, n.arch), ): for pat in patterns: if fnmatch.fnmatch(name, pat): return True return False def pkgtup2spec(name, arch, epoch, version, release): # we ignore arch return "%s-%s:%s-%s.*" % (name, epoch or "0", version, release)	12.03 KB	2022-09-09 11:42

Edit File:

Terminal