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
📄 __init__.py # __init__.py # The toplevel DNF package. # # Copyright (C) 2012-2016 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 unicode_literals import warnings import dnf.pycomp warnings.filterwarnings('once', category=DeprecationWarning, module=r'^dnf\..*$') from dnf.const import VERSION __version__ = VERSION # :api import dnf.base Base = dnf.base.Base # :api import dnf.plugin Plugin = dnf.plugin.Plugin # :api # setup libraries dnf.pycomp.urlparse.uses_fragment.append("media")	1.36 KB	2022-09-09 11:21
📁 __pycache__	-	2026-03-21 15:47
📄 base.py # Copyright 2005 Duke University # Copyright (C) 2012-2018 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 Library 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. """ Supplies the Base class. """ from __future__ import absolute_import from __future__ import division from __future__ import print_function from __future__ import unicode_literals import argparse import dnf import libdnf.transaction from copy import deepcopy from dnf.comps import CompsQuery from dnf.i18n import _, P_, ucd from dnf.util import _parse_specs from dnf.db.history import SwdbInterface from dnf.yum import misc try: from collections.abc import Sequence except ImportError: from collections import Sequence import datetime import dnf.callback import dnf.comps import dnf.conf import dnf.conf.read import dnf.crypto import dnf.dnssec import dnf.drpm import dnf.exceptions import dnf.goal import dnf.history import dnf.lock import dnf.logging # WITH_MODULES is used by ansible (lib/ansible/modules/packaging/os/dnf.py) try: import dnf.module.module_base WITH_MODULES = True except ImportError: WITH_MODULES = False import dnf.persistor import dnf.plugin import dnf.query import dnf.repo import dnf.repodict import dnf.rpm.connection import dnf.rpm.miscutils import dnf.rpm.transaction import dnf.sack import dnf.selector import dnf.subject import dnf.transaction import dnf.util import dnf.yum.rpmtrans import functools import gc import hawkey import itertools import logging import math import os import operator import re import rpm import tempfile import time import shutil logger = logging.getLogger("dnf") class Base(object): def __init__(self, conf=None): # :api self._closed = False self._conf = conf or self._setup_default_conf() self._goal = None self._repo_persistor = None self._sack = None self._transaction = None self._priv_ts = None self._comps = None self._comps_trans = dnf.comps.TransactionBunch() self._history = None self._tempfiles = set() self._trans_tempfiles = set() self._ds_callback = dnf.callback.Depsolve() self._logging = dnf.logging.Logging() self._repos = dnf.repodict.RepoDict() self._rpm_probfilter = set([rpm.RPMPROB_FILTER_OLDPACKAGE]) self._plugins = dnf.plugin.Plugins() self._trans_success = False self._trans_install_set = False self._tempfile_persistor = None # self._update_security_filters is used by ansible self._update_security_filters = [] self._update_security_options = {} self._allow_erasing = False self._repo_set_imported_gpg_keys = set() self._persistence = libdnf.transaction.TransactionPersistence_UNKNOWN self.output = None def __enter__(self): return self def __exit__(self, exc_args): self.close() def __del__(self): self.close() def _add_tempfiles(self, files): if self._transaction: self._trans_tempfiles.update(files) elif self.conf.destdir: pass else: self._tempfiles.update(files) def _add_repo_to_sack(self, repo): repo.load() mdload_flags = dict(load_filelists=True, load_presto=repo.deltarpm, load_updateinfo=True) if repo.load_metadata_other: mdload_flags["load_other"] = True try: self._sack.load_repo(repo._repo, build_cache=True, mdload_flags) except hawkey.Exception as e: logger.debug(_("loading repo '{}' failure: {}").format(repo.id, e)) raise dnf.exceptions.RepoError( _("Loading repository '{}' has failed").format(repo.id)) @staticmethod def _setup_default_conf(): conf = dnf.conf.Conf() subst = conf.substitutions if 'releasever' not in subst: releasever, major, minor = \ dnf.rpm.detect_releasevers(conf.installroot) subst['releasever'] = releasever if major is not None: subst['releasever_major'] = major if minor is not None: subst['releasever_minor'] = minor return conf def _setup_modular_excludes(self): hot_fix_repos = [i.id for i in self.repos.iter_enabled() if i.module_hotfixes] try: solver_errors = self.sack.filter_modules( self._moduleContainer, hot_fix_repos, self.conf.installroot, self.conf.module_platform_id, update_only=False, debugsolver=self.conf.debug_solver, module_obsoletes=self.conf.module_obsoletes) except hawkey.Exception as e: raise dnf.exceptions.Error(ucd(e)) if solver_errors: logger.warning( dnf.module.module_base.format_modular_solver_errors(solver_errors[0])) def _setup_excludes_includes(self, only_main=False): disabled = set(self.conf.disable_excludes) if 'all' in disabled and WITH_MODULES: self._setup_modular_excludes() return repo_includes = [] repo_excludes = [] # first evaluate repo specific includes/excludes if not only_main: for r in self.repos.iter_enabled(): if r.id in disabled: continue if len(r.includepkgs) > 0: incl_query = self.sack.query().filterm(empty=True) for incl in set(r.includepkgs): subj = dnf.subject.Subject(incl) incl_query = incl_query.union(subj.get_best_query( self.sack, with_nevra=True, with_provides=False, with_filenames=False)) incl_query.filterm(reponame=r.id) repo_includes.append((incl_query.apply(), r.id)) excl_query = self.sack.query().filterm(empty=True) for excl in set(r.excludepkgs): subj = dnf.subject.Subject(excl) excl_query = excl_query.union(subj.get_best_query( self.sack, with_nevra=True, with_provides=False, with_filenames=False)) excl_query.filterm(reponame=r.id) if excl_query: repo_excludes.append((excl_query, r.id)) # then main (global) includes/excludes because they can mask # repo specific settings if 'main' not in disabled: include_query = self.sack.query().filterm(empty=True) if len(self.conf.includepkgs) > 0: for incl in set(self.conf.includepkgs): subj = dnf.subject.Subject(incl) include_query = include_query.union(subj.get_best_query( self.sack, with_nevra=True, with_provides=False, with_filenames=False)) exclude_query = self.sack.query().filterm(empty=True) for excl in set(self.conf.excludepkgs): subj = dnf.subject.Subject(excl) exclude_query = exclude_query.union(subj.get_best_query( self.sack, with_nevra=True, with_provides=False, with_filenames=False)) if len(self.conf.includepkgs) > 0: self.sack.add_includes(include_query) self.sack.set_use_includes(True) if exclude_query: self.sack.add_excludes(exclude_query) if repo_includes: for query, repoid in repo_includes: self.sack.add_includes(query) self.sack.set_use_includes(True, repoid) if repo_excludes: for query, repoid in repo_excludes: self.sack.add_excludes(query) if not only_main and WITH_MODULES: self._setup_modular_excludes() def _store_persistent_data(self): if self._repo_persistor and not self.conf.cacheonly: expired = [r.id for r in self.repos.iter_enabled() if (r.metadata and r._repo.isExpired())] self._repo_persistor.expired_to_add.update(expired) self._repo_persistor.save() if self._tempfile_persistor: self._tempfile_persistor.save() @property def comps(self): # :api if self._comps is None: self.read_comps(arch_filter=True) return self._comps @property def conf(self): # :api return self._conf @property def repos(self): # :api return self._repos @repos.deleter def repos(self): # :api self._repos = None @property @dnf.util.lazyattr("_priv_rpmconn") def _rpmconn(self): return dnf.rpm.connection.RpmConnection(self.conf.installroot) @property def sack(self): # :api return self._sack @property def _moduleContainer(self): if self.sack is None: raise dnf.exceptions.Error("Sack was not initialized") if self.sack._moduleContainer is None: self.sack._moduleContainer = libdnf.module.ModulePackageContainer( False, self.conf.installroot, self.conf.substitutions["arch"], self.conf.persistdir) return self.sack._moduleContainer @property def transaction(self): # :api return self._transaction @transaction.setter def transaction(self, value): # :api if self._transaction: raise ValueError('transaction already set') self._transaction = value def _activate_persistor(self): self._repo_persistor = dnf.persistor.RepoPersistor(self.conf.cachedir) def init_plugins(self, disabled_glob=(), enable_plugins=(), cli=None): # :api """Load plugins and run their __init__().""" if self.conf.plugins: self._plugins._load(self.conf, disabled_glob, enable_plugins) self._plugins._run_init(self, cli) def pre_configure_plugins(self): # :api """Run plugins pre_configure() method.""" self._plugins._run_pre_config() def configure_plugins(self): # :api """Run plugins configure() method.""" self._plugins._run_config() def unload_plugins(self): # :api """Run plugins unload() method.""" self._plugins._unload() def update_cache(self, timer=False): # :api period = self.conf.metadata_timer_sync if self._repo_persistor is None: self._activate_persistor() persistor = self._repo_persistor if timer: if dnf.util.on_metered_connection(): msg = _('Metadata timer caching disabled ' 'when running on metered connection.') logger.info(msg) return False if dnf.util.on_ac_power() is False: msg = _('Metadata timer caching disabled ' 'when running on a battery.') logger.info(msg) return False if period <= 0: msg = _('Metadata timer caching disabled.') logger.info(msg) return False since_last_makecache = persistor.since_last_makecache() if since_last_makecache is not None and since_last_makecache < period: logger.info(_('Metadata cache refreshed recently.')) return False for repo in self.repos.values(): repo._repo.setMaxMirrorTries(1) if not self.repos._any_enabled(): logger.info(_('There are no enabled repositories in "{}".').format( '", "'.join(self.conf.reposdir))) return False for r in self.repos.iter_enabled(): (is_cache, expires_in) = r._metadata_expire_in() if expires_in is None: logger.info(_('%s: will never be expired and will not be refreshed.'), r.id) elif not is_cache or expires_in <= 0: logger.debug(_('%s: has expired and will be refreshed.'), r.id) r._repo.expire() elif timer and expires_in < period: # expires within the checking period: msg = _("%s: metadata will expire after %d seconds and will be refreshed now") logger.debug(msg, r.id, expires_in) r._repo.expire() else: logger.debug(_('%s: will expire after %d seconds.'), r.id, expires_in) if timer: persistor.reset_last_makecache = True self.fill_sack(load_system_repo=False, load_available_repos=True) # performs the md sync logger.info(_('Metadata cache created.')) return True def fill_sack(self, load_system_repo=True, load_available_repos=True): # :api """Prepare the Sack and the Goal objects. """ timer = dnf.logging.Timer('sack setup') self.reset(sack=True, goal=True) self._sack = dnf.sack._build_sack(self) lock = dnf.lock.build_metadata_lock(self.conf.cachedir, self.conf.exit_on_lock) with lock: if load_system_repo is not False: try: # FIXME: If build_cache=True, @System.solv is incorrectly updated in install- # remove loops self._sack.load_system_repo(build_cache=False) except IOError: if load_system_repo != 'auto': raise if load_available_repos: error_repos = [] mts = 0 age = time.time() # Iterate over installed GPG keys and check their validity using DNSSEC if self.conf.gpgkey_dns_verification: dnf.dnssec.RpmImportedKeys.check_imported_keys_validity() for r in self.repos.iter_enabled(): try: self._add_repo_to_sack(r) if r._repo.getTimestamp() > mts: mts = r._repo.getTimestamp() if r._repo.getAge() < age: age = r._repo.getAge() logger.debug(_("%s: using metadata from %s."), r.id, dnf.util.normalize_time( r._repo.getMaxTimestamp())) except dnf.exceptions.RepoError as e: r._repo.expire() if r.skip_if_unavailable is False: raise logger.warning("Error: %s", e) error_repos.append(r.id) r.disable() if error_repos: logger.warning( _("Ignoring repositories: %s"), ', '.join(error_repos)) if self.repos._any_enabled(): if age != 0 and mts != 0: logger.info(_("Last metadata expiration check: %s ago on %s."), datetime.timedelta(seconds=int(age)), dnf.util.normalize_time(mts)) else: self.repos.all().disable() conf = self.conf self._sack._configure(conf.installonlypkgs, conf.installonly_limit, conf.allow_vendor_change) self._setup_excludes_includes() timer() self._goal = dnf.goal.Goal(self._sack) self._goal.protect_running_kernel = conf.protect_running_kernel self._plugins.run_sack() return self._sack def fill_sack_from_repos_in_cache(self, load_system_repo=True): # :api """ Prepare Sack and Goal objects and also load all enabled repositories from cache only, it doesn't download anything and it doesn't check if metadata are expired. If there is not enough metadata present (repond.xml or both primary.xml and solv file are missing) given repo is either skipped or it throws a RepoError exception depending on skip_if_unavailable configuration. """ timer = dnf.logging.Timer('sack setup') self.reset(sack=True, goal=True) self._sack = dnf.sack._build_sack(self) lock = dnf.lock.build_metadata_lock(self.conf.cachedir, self.conf.exit_on_lock) with lock: if load_system_repo is not False: try: # FIXME: If build_cache=True, @System.solv is incorrectly updated in install- # remove loops self._sack.load_system_repo(build_cache=False) except IOError: if load_system_repo != 'auto': raise error_repos = [] # Iterate over installed GPG keys and check their validity using DNSSEC if self.conf.gpgkey_dns_verification: dnf.dnssec.RpmImportedKeys.check_imported_keys_validity() for repo in self.repos.iter_enabled(): try: repo._repo.loadCache(throwExcept=True, ignoreMissing=True) mdload_flags = dict(load_filelists=True, load_presto=repo.deltarpm, load_updateinfo=True) if repo.load_metadata_other: mdload_flags["load_other"] = True self._sack.load_repo(repo._repo, mdload_flags) logger.debug(_("%s: using metadata from %s."), repo.id, dnf.util.normalize_time( repo._repo.getMaxTimestamp())) except (RuntimeError, hawkey.Exception) as e: if repo.skip_if_unavailable is False: raise dnf.exceptions.RepoError( _("loading repo '{}' failure: {}").format(repo.id, e)) else: logger.debug(_("loading repo '{}' failure: {}").format(repo.id, e)) error_repos.append(repo.id) repo.disable() if error_repos: logger.warning( _("Ignoring repositories: %s"), ', '.join(error_repos)) conf = self.conf self._sack._configure(conf.installonlypkgs, conf.installonly_limit, conf.allow_vendor_change) self._setup_excludes_includes() timer() self._goal = dnf.goal.Goal(self._sack) self._goal.protect_running_kernel = conf.protect_running_kernel self._plugins.run_sack() return self._sack def _finalize_base(self): self._tempfile_persistor = dnf.persistor.TempfilePersistor( self.conf.cachedir) if not self.conf.keepcache: self._clean_packages(self._tempfiles) if self._trans_success: self._trans_tempfiles.update( self._tempfile_persistor.get_saved_tempfiles()) self._tempfile_persistor.empty() if self._trans_install_set: self._clean_packages(self._trans_tempfiles) else: self._tempfile_persistor.tempfiles_to_add.update( self._trans_tempfiles) if self._tempfile_persistor.tempfiles_to_add: logger.info(_("The downloaded packages were saved in cache " "until the next successful transaction.")) logger.info(_("You can remove cached packages by executing " "'%s'."), "{prog} clean packages".format(prog=dnf.util.MAIN_PROG)) # Do not trigger the lazy creation: if self._history is not None: self.history.close() self._store_persistent_data() self._closeRpmDB() self._trans_success = False def close(self): # :api """Close all potential handles and clean cache. Typically the handles are to data sources and sinks. """ if self._closed: return logger.log(dnf.logging.DDEBUG, 'Cleaning up.') self._closed = True self._finalize_base() self.reset(sack=True, repos=True, goal=True) self._plugins = None def read_all_repos(self, opts=None): # :api """Read repositories from the main conf file and from .repo files.""" reader = dnf.conf.read.RepoReader(self.conf, opts) for repo in reader: try: self.repos.add(repo) except dnf.exceptions.ConfigError as e: logger.warning(e) def reset(self, sack=False, repos=False, goal=False): # :api """Make the Base object forget about various things.""" if sack: self._sack = None if repos: self._repos = dnf.repodict.RepoDict() if goal: self._goal = None if self._sack is not None: self._goal = dnf.goal.Goal(self._sack) self._goal.protect_running_kernel = self.conf.protect_running_kernel if self._sack and self._moduleContainer: # sack must be set to enable operations on moduleContainer self._moduleContainer.rollback() if self._history is not None: self.history.close() self._comps_trans = dnf.comps.TransactionBunch() self._transaction = None self._update_security_filters = [] if sack and goal: # We've just done this, above: # # _sack _goal # \| \| # -- [CUT] -- -- [CUT] -- # \| \| # v \| v # +----------------+ [C] +-------------+ # \| DnfSack object \| <-[U]- \| Goal object \| # +----------------+ [T] +-------------+ # \|^ \|^ \|^ \| # \|\| \|\| \|\| # \|\| \|\| \|\| \| # +--\|\|----\|\|----\|\|---+ [C] # \| v\| v\| v\| \| <--[U]-- _transaction # \| Pkg1 Pkg2 PkgN \| [T] # \| \| \| # \| Transaction oject \| # +-------------------+ # # At this point, the DnfSack object would be released only # eventually, by Python's generational garbage collector, due to the # cyclic references DnfSack<->Pkg1 ... DnfSack<->PkgN. # # The delayed release is a problem: the DnfSack object may # (indirectly) own "page file" file descriptors in libsolv, via # libdnf. For example, # # sack->priv->pool->repos[1]->repodata[1]->store.pagefd = 7 # sack->priv->pool->repos[1]->repodata[2]->store.pagefd = 8 # # These file descriptors are closed when the DnfSack object is # eventually released, that is, when dnf_sack_finalize() (in libdnf) # calls pool_free() (in libsolv). # # We need that to happen right now, as callers may want to unmount # the filesystems which those file descriptors refer to immediately # after reset() returns. Therefore, force a garbage collection here. gc.collect() def _closeRpmDB(self): """Closes down the instances of rpmdb that could be open.""" del self._ts _TS_FLAGS_TO_RPM = {'noscripts': rpm.RPMTRANS_FLAG_NOSCRIPTS, 'notriggers': rpm.RPMTRANS_FLAG_NOTRIGGERS, 'nodocs': rpm.RPMTRANS_FLAG_NODOCS, 'test': rpm.RPMTRANS_FLAG_TEST, 'justdb': rpm.RPMTRANS_FLAG_JUSTDB, 'nocontexts': rpm.RPMTRANS_FLAG_NOCONTEXTS, 'nocrypto': rpm.RPMTRANS_FLAG_NOFILEDIGEST} if hasattr(rpm, 'RPMTRANS_FLAG_NOCAPS'): # Introduced in rpm-4.14 _TS_FLAGS_TO_RPM['nocaps'] = rpm.RPMTRANS_FLAG_NOCAPS _TS_VSFLAGS_TO_RPM = {'nocrypto': rpm._RPMVSF_NOSIGNATURES \| rpm._RPMVSF_NODIGESTS} @property def goal(self): return self._goal @property def _ts(self): """Set up the RPM transaction set that will be used for all the work.""" if self._priv_ts is not None: return self._priv_ts self._priv_ts = dnf.rpm.transaction.TransactionWrapper( self.conf.installroot) self._priv_ts.setFlags(0) # reset everything. for flag in self.conf.tsflags: rpm_flag = self._TS_FLAGS_TO_RPM.get(flag) if rpm_flag is None: logger.critical(_('Invalid tsflag in config file: %s'), flag) continue self._priv_ts.addTsFlag(rpm_flag) vs_flag = self._TS_VSFLAGS_TO_RPM.get(flag) if vs_flag is not None: self._priv_ts.pushVSFlags(vs_flag) if not self.conf.diskspacecheck: self._rpm_probfilter.add(rpm.RPMPROB_FILTER_DISKSPACE) if self.conf.ignorearch: self._rpm_probfilter.add(rpm.RPMPROB_FILTER_IGNOREARCH) probfilter = functools.reduce(operator.or_, self._rpm_probfilter, 0) self._priv_ts.setProbFilter(probfilter) return self._priv_ts @_ts.deleter def _ts(self): """Releases the RPM transaction set. """ if self._priv_ts is None: return self._priv_ts.close() del self._priv_ts self._priv_ts = None def read_comps(self, arch_filter=False): # :api """Create the groups object to access the comps metadata.""" timer = dnf.logging.Timer('loading comps') self._comps = dnf.comps.Comps() logger.log(dnf.logging.DDEBUG, 'Getting group metadata') for repo in self.repos.iter_enabled(): if not repo.enablegroups: continue if not repo.metadata: continue comps_fn = repo._repo.getCompsFn() if not comps_fn: continue logger.log(dnf.logging.DDEBUG, 'Adding group file from repository: %s', repo.id) gen_dir = os.path.join(os.path.dirname(comps_fn), 'gen') gen_file = os.path.join(gen_dir, 'groups.xml') temp_file = None try: if not os.path.exists(gen_dir): os.makedirs(gen_dir, mode=0o755) misc.decompress(comps_fn, dest=gen_file, check_timestamps=True) except (PermissionError, dnf.exceptions.MiscError): temp_file = tempfile.NamedTemporaryFile() gen_file = temp_file.name misc.decompress(comps_fn, dest=gen_file, check_timestamps=False) try: self._comps._add_from_xml_filename(gen_file) except dnf.exceptions.CompsError as e: msg = _('Failed to add groups file for repository: %s - %s') logger.critical(msg, repo.id, e) if temp_file: temp_file.close() if arch_filter: self._comps._i.arch_filter( [self._conf.substitutions['basearch']]) timer() return self._comps def _getHistory(self): """auto create the history object that to access/append the transaction history information. """ if self._history is None: releasever = self.conf.releasever self._history = SwdbInterface(self.conf.persistdir, releasever=releasever) return self._history history = property(fget=lambda self: self._getHistory(), fset=lambda self, value: setattr( self, "_history", value), fdel=lambda self: setattr(self, "_history", None), doc="DNF SWDB Interface Object") def _goal2transaction(self, goal): ts = self.history.rpm all_obsoleted = set(goal.list_obsoleted()) installonly_query = self._get_installonly_query() installonly_query.apply() installonly_query_installed = installonly_query.installed().apply() for pkg in goal.list_downgrades(): obs = goal.obsoleted_by_package(pkg) downgraded = obs[0] self._ds_callback.pkg_added(downgraded, 'dd') self._ds_callback.pkg_added(pkg, 'd') ts.add_downgrade(pkg, downgraded, obs[1:]) for pkg in goal.list_reinstalls(): self._ds_callback.pkg_added(pkg, 'r') obs = goal.obsoleted_by_package(pkg) nevra_pkg = str(pkg) # reinstall could obsolete multiple packages with the same NEVRA or different NEVRA # Set the package with the same NEVRA as reinstalled obsoletes = [] for obs_pkg in obs: if str(obs_pkg) == nevra_pkg: obsoletes.insert(0, obs_pkg) else: obsoletes.append(obs_pkg) reinstalled = obsoletes[0] ts.add_reinstall(pkg, reinstalled, obsoletes[1:]) for pkg in goal.list_installs(): self._ds_callback.pkg_added(pkg, 'i') obs = goal.obsoleted_by_package(pkg) # Skip obsoleted packages that are not part of all_obsoleted, # they are handled as upgrades/downgrades. # Also keep RPMs with the same name - they're not always in all_obsoleted. obs = [i for i in obs if i in all_obsoleted or i.name == pkg.name] reason = goal.get_reason(pkg) # Inherit reason if package is installonly an package with same name is installed # Use the same logic like upgrade # Upgrade of installonly packages result in install or install and remove step if pkg in installonly_query and installonly_query_installed.filter(name=pkg.name): reason = ts.get_reason(pkg) # inherit the best reason from obsoleted packages for obsolete in obs: reason_obsolete = ts.get_reason(obsolete) if libdnf.transaction.TransactionItemReasonCompare(reason, reason_obsolete) == -1: reason = reason_obsolete ts.add_install(pkg, obs, reason) cb = lambda pkg: self._ds_callback.pkg_added(pkg, 'od') dnf.util.mapall(cb, obs) for pkg in goal.list_upgrades(): obs = goal.obsoleted_by_package(pkg) upgraded = None for i in obs: # try to find a package with matching name as the upgrade if i.name == pkg.name: upgraded = i break if upgraded is None: # no matching name -> pick the first one upgraded = obs.pop(0) else: obs.remove(upgraded) # Skip obsoleted packages that are not part of all_obsoleted, # they are handled as upgrades/downgrades. # Also keep RPMs with the same name - they're not always in all_obsoleted. obs = [i for i in obs if i in all_obsoleted or i.name == pkg.name] cb = lambda pkg: self._ds_callback.pkg_added(pkg, 'od') dnf.util.mapall(cb, obs) if pkg in installonly_query: ts.add_install(pkg, obs) else: ts.add_upgrade(pkg, upgraded, obs) self._ds_callback.pkg_added(upgraded, 'ud') self._ds_callback.pkg_added(pkg, 'u') erasures = goal.list_erasures() if erasures: remaining_installed_query = self.sack.query(flags=hawkey.IGNORE_EXCLUDES).installed() remaining_installed_query.filterm(pkg__neq=erasures) remaining_installed_query.apply() for pkg in erasures: tmp_remaining_installed_query = remaining_installed_query.filter(name=pkg.name, arch=pkg.arch) if tmp_remaining_installed_query: remaining = tmp_remaining_installed_query[0] ts.get_reason(remaining) self.history.set_reason(remaining, ts.get_reason(remaining)) self._ds_callback.pkg_added(pkg, 'e') reason = goal.get_reason(pkg) ts.add_erase(pkg, reason) return ts def _query_matches_installed(self, q): """ See what packages in the query match packages (also in older versions, but always same architecture) that are already installed. Unlike in case of _sltr_matches_installed(), it is practical here to know even the packages in the original query that can still be installed. """ inst = q.installed() inst_per_arch = inst._na_dict() avail_per_arch = q.available()._na_dict() avail_l = [] inst_l = [] for na in avail_per_arch: if na in inst_per_arch: inst_l.append(inst_per_arch[na][0]) else: avail_l.append(avail_per_arch[na]) return inst_l, avail_l def _sltr_matches_installed(self, sltr): """ See if sltr matches a patches that is (in older version or different architecture perhaps) already installed. """ inst = self.sack.query().installed().filterm(pkg=sltr.matches()) return list(inst) def iter_userinstalled(self): """Get iterator over the packages installed by the user.""" return (pkg for pkg in self.sack.query().installed() if self.history.user_installed(pkg)) def _run_hawkey_goal(self, goal, allow_erasing): ret = goal.run( allow_uninstall=allow_erasing, force_best=self.conf.best, ignore_weak_deps=(not self.conf.install_weak_deps)) if self.conf.debug_solver: goal.write_debugdata('./debugdata/rpms') return ret def _set_excludes_from_weak_to_goal(self): """ Add exclude_from_weak from configuration and autodetect unmet weak deps exclude them from candidates to satisfy weak dependencies """ self._goal.reset_exclude_from_weak() if self.conf.exclude_from_weak_autodetect: self._goal.exclude_from_weak_autodetect() for weak_exclude in self.conf.exclude_from_weak: subj = dnf.subject.Subject(weak_exclude) query = subj.get_best_query(self.sack, with_nevra=True, with_provides=False, with_filenames=False) query = query.available() self._goal.add_exclude_from_weak(query) def resolve(self, allow_erasing=False): # :api """Build the transaction set.""" exc = None self._finalize_comps_trans() timer = dnf.logging.Timer('depsolve') self._ds_callback.start() goal = self._goal if goal.req_has_erase(): goal.push_userinstalled(self.sack.query().installed(), self.history) elif not self.conf.upgrade_group_objects_upgrade: # exclude packages installed from groups # these packages will be marked to installation # which could prevent them from upgrade, downgrade # to prevent "conflicting job" error it's not applied # to "remove" and "reinstall" commands solver = self._build_comps_solver() solver._exclude_packages_from_installed_groups(self) goal.add_protected(self.sack.query().filterm( name=self.conf.protected_packages)) self._set_excludes_from_weak_to_goal() if not self._run_hawkey_goal(goal, allow_erasing): if self.conf.debuglevel >= 6: goal.log_decisions() msg = dnf.util._format_resolve_problems(goal.problem_rules()) exc = dnf.exceptions.DepsolveError(msg) else: self._transaction = self._goal2transaction(goal) self._ds_callback.end() timer() got_transaction = self._transaction is not None and \ len(self._transaction) > 0 if got_transaction: msg = self._transaction._rpm_limitations() if msg: exc = dnf.exceptions.Error(msg) if exc is not None: raise exc self._plugins.run_resolved() # auto-enable module streams based on installed RPMs new_pkgs = self._goal.list_installs() new_pkgs += self._goal.list_upgrades() new_pkgs += self._goal.list_downgrades() new_pkgs += self._goal.list_reinstalls() self.sack.set_modules_enabled_by_pkgset(self._moduleContainer, new_pkgs) return got_transaction def do_transaction(self, display=()): # :api if not isinstance(display, Sequence): display = [display] display = \ [dnf.yum.rpmtrans.LoggingTransactionDisplay()] + list(display) if not self.transaction: # packages are not changed, but comps and modules changes need to be committed self._moduleContainer.save() self._moduleContainer.updateFailSafeData() if self._history and (self._history.group or self._history.env): cmdline = None if hasattr(self, 'args') and self.args: cmdline = ' '.join(self.args) elif hasattr(self, 'cmds') and self.cmds: cmdline = ' '.join(self.cmds) old = self.history.last() if old is None: rpmdb_version = self._ts.dbCookie() else: rpmdb_version = old.end_rpmdb_version self.history.beg(rpmdb_version, [], [], cmdline=cmdline, persistence=self._persistence) self.history.end(rpmdb_version) self._plugins.run_pre_transaction() self._plugins.run_transaction() self._trans_success = True return tid = None logger.info(_('Running transaction check')) lock = dnf.lock.build_rpmdb_lock(self.conf.persistdir, self.conf.exit_on_lock) with lock: self.transaction._populate_rpm_ts(self._ts) msgs = self._run_rpm_check() if msgs: msg = _('Error: transaction check vs depsolve:') logger.error(msg) for msg in msgs: logger.error(msg) raise dnf.exceptions.TransactionCheckError(msg) logger.info(_('Transaction check succeeded.')) timer = dnf.logging.Timer('transaction test') logger.info(_('Running transaction test')) self._ts.order() # order the transaction self._ts.clean() # release memory not needed beyond this point testcb = dnf.yum.rpmtrans.RPMTransaction(self, test=True) tserrors = self._ts.test(testcb) if len(tserrors) > 0: for msg in testcb.messages(): logger.critical(_('RPM: {}').format(msg)) errstring = _('Transaction test error:') + '\n' for descr in tserrors: errstring += ' %s\n' % ucd(descr) summary = self._trans_error_summary(errstring) if summary: errstring += '\n' + summary raise dnf.exceptions.Error(errstring) del testcb logger.info(_('Transaction test succeeded.')) # With RPMTRANS_FLAG_TEST return just before anything is stored permanently if self._ts.isTsFlagSet(rpm.RPMTRANS_FLAG_TEST): return timer() # save module states on disk right before entering rpm transaction, # because we want system in recoverable state if transaction gets interrupted self._moduleContainer.save() self._moduleContainer.updateFailSafeData() # unset the sigquit handler timer = dnf.logging.Timer('transaction') # setup our rpm ts callback cb = dnf.yum.rpmtrans.RPMTransaction(self, displays=display) if self.conf.debuglevel < 2: for display_ in cb.displays: display_.output = False self._plugins.run_pre_transaction() logger.info(_('Running transaction')) tid = self._run_transaction(cb=cb) timer() self._plugins.unload_removed_plugins(self.transaction) self._plugins.run_transaction() # log post transaction summary def _pto_callback(action, tsis): msgs = [] for tsi in tsis: msgs.append('{}: {}'.format(action, str(tsi))) return msgs for msg in dnf.util._post_transaction_output(self, self.transaction, _pto_callback): logger.debug(msg) return tid def _trans_error_summary(self, errstring): """Parse the error string for 'interesting' errors which can be grouped, such as disk space issues. :param errstring: the error string :return: a string containing a summary of the errors """ summary = '' # do disk space report first p = re.compile(r'needs (\d+)(K\|M)B(?: more space)? on the (\S+) filesystem') disk = {} for m in p.finditer(errstring): size_in_mb = int(m.group(1)) if m.group(2) == 'M' else math.ceil( int(m.group(1)) / 1024.0) if m.group(3) not in disk: disk[m.group(3)] = size_in_mb if disk[m.group(3)] < size_in_mb: disk[m.group(3)] = size_in_mb if disk: summary += _('Disk Requirements:') + "\n" for k in disk: summary += " " + P_( 'At least {0}MB more space needed on the {1} filesystem.', 'At least {0}MB more space needed on the {1} filesystem.', disk[k]).format(disk[k], k) + '\n' if not summary: return None summary = _('Error Summary') + '\n-------------\n' + summary return summary def _record_history(self): return self.conf.history_record and \ not self._ts.isTsFlagSet(rpm.RPMTRANS_FLAG_TEST) def _run_transaction(self, cb): """ Perform the RPM transaction. :return: history database transaction ID or None """ tid = None if self._record_history(): using_pkgs_pats = list(self.conf.history_record_packages) installed_query = self.sack.query().installed() using_pkgs = installed_query.filter(name=using_pkgs_pats).run() rpmdbv = self._ts.dbCookie() lastdbv = self.history.last() if lastdbv is not None: lastdbv = lastdbv.end_rpmdb_version if lastdbv is None or rpmdbv != lastdbv: logger.debug(_("RPMDB altered outside of {prog}.").format( prog=dnf.util.MAIN_PROG_UPPER)) cmdline = None if hasattr(self, 'args') and self.args: cmdline = ' '.join(self.args) elif hasattr(self, 'cmds') and self.cmds: cmdline = ' '.join(self.cmds) comment = self.conf.comment if self.conf.comment else "" tid = self.history.beg(rpmdbv, using_pkgs, [], cmdline=cmdline, comment=comment, persistence=self._persistence) if self.conf.reset_nice: onice = os.nice(0) if onice: try: os.nice(-onice) except: onice = 0 logger.log(dnf.logging.DDEBUG, 'RPM transaction start.') errors = self._ts.run(cb.callback, '') logger.log(dnf.logging.DDEBUG, 'RPM transaction over.') # ts.run() exit codes are, hmm, "creative": None means all ok, empty # list means some errors happened in the transaction and non-empty # list that there were errors preventing the ts from starting... if self.conf.reset_nice: try: os.nice(onice) except: pass dnf.util._sync_rpm_trans_with_swdb(self._ts, self._transaction) if errors is None: pass elif len(errors) == 0: # If there is no failing element it means that some "global" error # occurred (like rpm failed to obtain the transaction lock). Just pass # the rpm logs on to the user and raise an Error. # If there are failing elements the problem is related to those # elements and the Error is raised later, after saving the failure # to the history and printing out the transaction table to user. failed = [el for el in self._ts if el.Failed()] if not failed: for msg in cb.messages(): logger.critical(_('RPM: {}').format(msg)) msg = _('Could not run transaction.') raise dnf.exceptions.Error(msg) else: logger.critical(_("Transaction couldn't start:")) for e in errors: logger.critical(ucd(e[0])) if self._record_history() and not self._ts.isTsFlagSet(rpm.RPMTRANS_FLAG_TEST): self.history.end(rpmdbv) msg = _("Could not run transaction.") raise dnf.exceptions.Error(msg) for i in ('ts_all_fn', 'ts_done_fn'): if hasattr(cb, i): fn = getattr(cb, i) try: misc.unlink_f(fn) except (IOError, OSError): msg = _('Failed to remove transaction file %s') logger.critical(msg, fn) # keep install_set status because _verify_transaction will clean it self._trans_install_set = bool(self._transaction.install_set) # sync up what just happened versus what is in the rpmdb if not self._ts.isTsFlagSet(rpm.RPMTRANS_FLAG_TEST): self._verify_transaction(cb.verify_tsi_package) return tid def _verify_transaction(self, verify_pkg_cb=None): transaction_items = [ tsi for tsi in self.transaction if tsi.action != libdnf.transaction.TransactionItemAction_REASON_CHANGE] total = len(transaction_items) def display_banner(pkg, count): count += 1 if verify_pkg_cb is not None: verify_pkg_cb(pkg, count, total) return count timer = dnf.logging.Timer('verify transaction') count = 0 rpmdb_sack = dnf.sack.rpmdb_sack(self) # mark group packages that are installed on the system as installed in the db q = rpmdb_sack.query().installed() names = set([i.name for i in q]) for ti in self.history.group: g = ti.getCompsGroupItem() for p in g.getPackages(): if p.getName() in names: p.setInstalled(True) p.save() # TODO: installed groups in environments # Post-transaction verification is no longer needed, # because DNF trusts error codes returned by RPM. # Verification banner is displayed to preserve UX. # TODO: drop in future DNF for tsi in transaction_items: count = display_banner(tsi.pkg, count) rpmdbv = self._ts.dbCookie() self.history.end(rpmdbv) timer() self._trans_success = True def _download_remote_payloads(self, payloads, drpm, progress, callback_total, fail_fast=True): lock = dnf.lock.build_download_lock(self.conf.cachedir, self.conf.exit_on_lock) with lock: beg_download = time.time() est_remote_size = sum(pload.download_size for pload in payloads) total_drpm = len( [payload for payload in payloads if isinstance(payload, dnf.drpm.DeltaPayload)]) # compatibility part for tools that do not accept total_drpms keyword if progress.start.__code__.co_argcount == 4: progress.start(len(payloads), est_remote_size, total_drpms=total_drpm) else: progress.start(len(payloads), est_remote_size) errors = dnf.repo._download_payloads(payloads, drpm, fail_fast) if errors._irrecoverable(): raise dnf.exceptions.DownloadError(errors._irrecoverable()) remote_size = sum(errors._bandwidth_used(pload) for pload in payloads) saving = dnf.repo._update_saving((0, 0), payloads, errors._recoverable) retries = self.conf.retries forever = retries == 0 while errors._recoverable and (forever or retries > 0): if retries > 0: retries -= 1 msg = _("Some packages were not downloaded. Retrying.") logger.info(msg) remaining_pkgs = [pkg for pkg in errors._recoverable] payloads = \ [dnf.repo._pkg2payload(pkg, progress, dnf.repo.RPMPayload) for pkg in remaining_pkgs] est_remote_size = sum(pload.download_size for pload in payloads) progress.start(len(payloads), est_remote_size) errors = dnf.repo._download_payloads(payloads, drpm, fail_fast) if errors._irrecoverable(): raise dnf.exceptions.DownloadError(errors._irrecoverable()) remote_size += \ sum(errors._bandwidth_used(pload) for pload in payloads) saving = dnf.repo._update_saving(saving, payloads, {}) if errors._recoverable: msg = dnf.exceptions.DownloadError.errmap2str( errors._recoverable) logger.info(msg) if callback_total is not None: callback_total(remote_size, beg_download) (real, full) = saving if real != full: if real < full: msg = _("Delta RPMs reduced %.1f MB of updates to %.1f MB " "(%.1f%% saved)") percent = 100 - real / full 100 elif real > full: msg = _("Failed Delta RPMs increased %.1f MB of updates to %.1f MB " "(%.1f%% wasted)") percent = 100 - full / real * 100 logger.info(msg, full / 1024 2, real / 1024 2, percent) def download_packages(self, pkglist, progress=None, callback_total=None): # :api """Download the packages specified by the given list of packages. `pkglist` is a list of packages to download, `progress` is an optional DownloadProgress instance, `callback_total` an optional callback to output messages about the download operation. """ remote_pkgs, local_pkgs = self._select_remote_pkgs(pkglist) if remote_pkgs: if progress is None: progress = dnf.callback.NullDownloadProgress() drpm = dnf.drpm.DeltaInfo(self.sack.query().installed(), progress, self.conf.deltarpm_percentage) self._add_tempfiles([pkg.localPkg() for pkg in remote_pkgs]) payloads = [dnf.repo._pkg2payload(pkg, progress, drpm.delta_factory, dnf.repo.RPMPayload) for pkg in remote_pkgs] self._download_remote_payloads(payloads, drpm, progress, callback_total) if self.conf.destdir: for pkg in local_pkgs: if pkg.baseurl: location = os.path.join(pkg.get_local_baseurl(), pkg.location.lstrip("/")) else: location = os.path.join(pkg.repo.pkgdir, pkg.location.lstrip("/")) shutil.copy(location, self.conf.destdir) def add_remote_rpms(self, path_list, strict=True, progress=None): # :api pkgs = [] if not path_list: return pkgs if self._goal.req_length(): raise dnf.exceptions.Error( _("Cannot add local packages, because transaction job already exists")) pkgs_error = [] for path in path_list: if not os.path.exists(path) and '://' in path: # download remote rpm to a tempfile path = dnf.util._urlopen_progress(path, self.conf, progress) self._add_tempfiles([path]) try: pkgs.append(self.sack.add_cmdline_package(path)) except IOError as e: logger.warning(e) pkgs_error.append(path) self._setup_excludes_includes(only_main=True) if pkgs_error and strict: raise IOError(_("Could not open: {}").format(' '.join(pkgs_error))) return pkgs def _sig_check_pkg(self, po): """Verify the GPG signature of the given package object. :param po: the package object to verify the signature of :return: (result, error_string) where result is:: 0 = GPG signature verifies ok or verification is not required. 1 = GPG verification failed but installation of the right GPG key might help. 2 = Fatal GPG verification error, give up. """ if po._from_cmdline: check = self.conf.localpkg_gpgcheck hasgpgkey = 0 else: repo = self.repos[po.repoid] check = repo.gpgcheck hasgpgkey = not not repo.gpgkey if check: root = self.conf.installroot ts = dnf.rpm.transaction.initReadOnlyTransaction(root) sigresult = dnf.rpm.miscutils.checkSig(ts, po.localPkg()) localfn = os.path.basename(po.localPkg()) del ts if sigresult == 0: result = 0 msg = '' elif sigresult == 1: if hasgpgkey: result = 1 else: result = 2 msg = _('Public key for %s is not installed') % localfn elif sigresult == 2: result = 2 msg = _('Problem opening package %s') % localfn elif sigresult == 3: if hasgpgkey: result = 1 else: result = 2 result = 1 msg = _('Public key for %s is not trusted') % localfn elif sigresult == 4: result = 2 msg = _('Package %s is not signed') % localfn else: result = 0 msg = '' return result, msg def package_signature_check(self, pkg): # :api """Verify the GPG signature of the given package object. :param pkg: the package object to verify the signature of :return: (result, error_string) where result is:: 0 = GPG signature verifies ok or verification is not required. 1 = GPG verification failed but installation of the right GPG key might help. 2 = Fatal GPG verification error, give up. """ return self._sig_check_pkg(pkg) def _clean_packages(self, packages): for fn in packages: if not os.path.exists(fn): continue try: misc.unlink_f(fn) except OSError: logger.warning(_('Cannot remove %s'), fn) continue else: logger.log(dnf.logging.DDEBUG, _('%s removed'), fn) def _do_package_lists(self, pkgnarrow='all', patterns=None, showdups=None, ignore_case=False, reponame=None): """Return a :class:`misc.GenericHolder` containing lists of package objects. The contents of the lists are specified in various ways by the arguments. :param pkgnarrow: a string specifying which types of packages lists to produces, such as updates, installed, available, etc. :param patterns: a list of names or wildcards specifying packages to list :param showdups: whether to include duplicate packages in the lists :param ignore_case: whether to ignore case when searching by package names :param reponame: limit packages list to the given repository :return: a :class:`misc.GenericHolder` instance with the following lists defined:: available = list of packageObjects installed = list of packageObjects upgrades = tuples of packageObjects (updating, installed) extras = list of packageObjects obsoletes = tuples of packageObjects (obsoleting, installed) recent = list of packageObjects """ if showdups is None: showdups = self.conf.showdupesfromrepos if patterns is None: return self._list_pattern( pkgnarrow, patterns, showdups, ignore_case, reponame) assert not dnf.util.is_string_type(patterns) list_fn = functools.partial( self._list_pattern, pkgnarrow, showdups=showdups, ignore_case=ignore_case, reponame=reponame) if patterns is None or len(patterns) == 0: return list_fn(None) yghs = map(list_fn, patterns) return functools.reduce(lambda a, b: a.merge_lists(b), yghs) def _list_pattern(self, pkgnarrow, pattern, showdups, ignore_case, reponame=None): def is_from_repo(package): """Test whether given package originates from the repository.""" if reponame is None: return True return self.history.repo(package) == reponame def pkgs_from_repo(packages): """Filter out the packages which do not originate from the repo.""" return (package for package in packages if is_from_repo(package)) def query_for_repo(query): """Filter out the packages which do not originate from the repo.""" if reponame is None: return query return query.filter(reponame=reponame) ygh = misc.GenericHolder(iter=pkgnarrow) installed = [] available = [] reinstall_available = [] old_available = [] updates = [] obsoletes = [] obsoletesTuples = [] recent = [] extras = [] autoremove = [] # do the initial pre-selection ic = ignore_case q = self.sack.query() if pattern is not None: subj = dnf.subject.Subject(pattern, ignore_case=ic) q = subj.get_best_query(self.sack, with_provides=False) # list all packages - those installed and available: if pkgnarrow == 'all': dinst = {} ndinst = {} # Newest versions by name.arch for po in q.installed(): dinst[po.pkgtup] = po if showdups: continue key = (po.name, po.arch) if key not in ndinst or po > ndinst[key]: ndinst[key] = po installed = list(pkgs_from_repo(dinst.values())) avail = query_for_repo(q.available()) if not showdups: avail = avail.filterm(latest_per_arch_by_priority=True) for pkg in avail: if showdups: if pkg.pkgtup in dinst: reinstall_available.append(pkg) else: available.append(pkg) else: key = (pkg.name, pkg.arch) if pkg.pkgtup in dinst: reinstall_available.append(pkg) elif key not in ndinst or pkg.evr_gt(ndinst[key]): available.append(pkg) else: old_available.append(pkg) # produce the updates list of tuples elif pkgnarrow == 'upgrades': updates = query_for_repo(q).filterm(upgrades_by_priority=True) # reduce a query to security upgrades if they are specified updates = self._merge_update_filters(updates, upgrade=True) # reduce a query to remove src RPMs updates.filterm(arch__neq=['src', 'nosrc']) # reduce a query to latest packages updates = updates.latest().run() # installed only elif pkgnarrow == 'installed': installed = list(pkgs_from_repo(q.installed())) # available in a repository elif pkgnarrow == 'available': if showdups: avail = query_for_repo(q).available() installed_dict = q.installed()._na_dict() for avail_pkg in avail: key = (avail_pkg.name, avail_pkg.arch) installed_pkgs = installed_dict.get(key, []) same_ver = [pkg for pkg in installed_pkgs if pkg.evr == avail_pkg.evr] if len(same_ver) > 0: reinstall_available.append(avail_pkg) else: available.append(avail_pkg) else: # we will only look at the latest versions of packages: available_dict = query_for_repo( q).available().filterm(latest_per_arch_by_priority=True)._na_dict() installed_dict = q.installed().latest()._na_dict() for (name, arch) in available_dict: avail_pkg = available_dict[(name, arch)][0] inst_pkg = installed_dict.get((name, arch), [None])[0] if not inst_pkg or avail_pkg.evr_gt(inst_pkg): available.append(avail_pkg) elif avail_pkg.evr_eq(inst_pkg): reinstall_available.append(avail_pkg) else: old_available.append(avail_pkg) # packages to be removed by autoremove elif pkgnarrow == 'autoremove': autoremove_q = query_for_repo(q)._unneeded(self.history.swdb) autoremove = autoremove_q.run() # not in a repo but installed elif pkgnarrow == 'extras': extras = [pkg for pkg in q.extras() if is_from_repo(pkg)] # obsoleting packages (and what they obsolete) elif pkgnarrow == 'obsoletes': inst = q.installed() obsoletes = query_for_repo( self.sack.query()).filter(obsoletes_by_priority=inst) # reduce a query to security upgrades if they are specified obsoletes = self._merge_update_filters(obsoletes, warning=False, upgrade=True) # reduce a query to remove src RPMs obsoletes.filterm(arch__neq=['src', 'nosrc']) obsoletesTuples = [] for new in obsoletes: obsoleted_reldeps = new.obsoletes obsoletesTuples.extend( [(new, old) for old in inst.filter(provides=obsoleted_reldeps)]) # packages recently added to the repositories elif pkgnarrow == 'recent': avail = q.available() if not showdups: avail = avail.filterm(latest_per_arch_by_priority=True) recent = query_for_repo(avail)._recent(self.conf.recent) ygh.installed = installed ygh.available = available ygh.reinstall_available = reinstall_available ygh.old_available = old_available ygh.updates = updates ygh.obsoletes = obsoletes ygh.obsoletesTuples = obsoletesTuples ygh.recent = recent ygh.extras = extras ygh.autoremove = autoremove return ygh def _add_comps_trans(self, trans): self._comps_trans += trans return len(trans) def _remove_if_unneeded(self, query): """ Mark to remove packages that are not required by any user installed package (reason group or user) :param query: dnf.query.Query() object """ query = query.installed() if not query: return unneeded_pkgs = query._safe_to_remove(self.history.swdb, debug_solver=False) unneeded_pkgs_history = query.filter( pkg=[i for i in query if self.history.group.is_removable_pkg(i.name)]) pkg_with_dependent_pkgs = unneeded_pkgs_history.difference(unneeded_pkgs) # mark packages with dependent packages as a dependency to allow removal with dependent # package for pkg in pkg_with_dependent_pkgs: self.history.set_reason(pkg, libdnf.transaction.TransactionItemReason_DEPENDENCY) unneeded_pkgs = unneeded_pkgs.intersection(unneeded_pkgs_history) remove_packages = query.intersection(unneeded_pkgs) if remove_packages: for pkg in remove_packages: self._goal.erase(pkg, clean_deps=self.conf.clean_requirements_on_remove) def _finalize_comps_trans(self): trans = self._comps_trans basearch = self.conf.substitutions['basearch'] def trans_upgrade(query, remove_query, comps_pkg): sltr = dnf.selector.Selector(self.sack) sltr.set(pkg=query) self._goal.upgrade(select=sltr) return remove_query def trans_install(query, remove_query, comps_pkg, strict): if self.conf.multilib_policy == "all": if not comps_pkg.requires: self._install_multiarch(query, strict=strict) else: # it installs only one arch for conditional packages installed_query = query.installed().apply() self._report_already_installed(installed_query) sltr = dnf.selector.Selector(self.sack) sltr.set(provides="({} if {})".format(comps_pkg.name, comps_pkg.requires)) self._goal.install(select=sltr, optional=not strict) else: sltr = dnf.selector.Selector(self.sack) if comps_pkg.requires: sltr.set(provides="({} if {})".format(comps_pkg.name, comps_pkg.requires)) else: if self.conf.obsoletes: # If there is no installed package in the pkgs_list, add only # obsoleters of the latest versions. Otherwise behave # consistently with upgrade and add all obsoleters. # See https://bugzilla.redhat.com/show_bug.cgi?id=2176263 # for details of the problem. if query.installed(): query = query.union(self.sack.query().filterm(obsoletes=query)) else: query = query.union(self.sack.query().filterm( obsoletes=query.filter(latest_per_arch_by_priority=True))) sltr.set(pkg=query) self._goal.install(select=sltr, optional=not strict) return remove_query def trans_remove(query, remove_query, comps_pkg): remove_query = remove_query.union(query) return remove_query remove_query = self.sack.query().filterm(empty=True) attr_fn = ((trans.install, functools.partial(trans_install, strict=True)), (trans.install_opt, functools.partial(trans_install, strict=False)), (trans.upgrade, trans_upgrade), (trans.remove, trans_remove)) for (attr, fn) in attr_fn: for comps_pkg in attr: query_args = {'name': comps_pkg.name} if (comps_pkg.basearchonly): query_args.update({'arch': basearch}) q = self.sack.query().filterm(*query_args).apply() q.filterm(arch__neq=["src", "nosrc"]) if not q: package_string = comps_pkg.name if comps_pkg.basearchonly: package_string += '.' + basearch logger.warning(_('No match for group package "{}"').format(package_string)) continue remove_query = fn(q, remove_query, comps_pkg) self._goal.group_members.add(comps_pkg.name) self._remove_if_unneeded(remove_query) def _build_comps_solver(self): def reason_fn(pkgname): q = self.sack.query().installed().filterm(name=pkgname) if not q: return None try: return self.history.rpm.get_reason(q[0]) except AttributeError: return libdnf.transaction.TransactionItemReason_UNKNOWN return dnf.comps.Solver(self.history, self._comps, reason_fn) def environment_install(self, env_id, types, exclude=None, strict=True, exclude_groups=None): # :api """Installs packages of environment group identified by env_id. :param types: Types of packages to install. Either an integer as a logical conjunction of CompsPackageType ids or a list of string package type ids (conditional, default, mandatory, optional). """ assert dnf.util.is_string_type(env_id) solver = self._build_comps_solver() if not isinstance(types, int): types = libdnf.transaction.listToCompsPackageType(types) trans = solver._environment_install(env_id, types, exclude or set(), strict, exclude_groups) if not trans: return 0 return self._add_comps_trans(trans) def environment_remove(self, env_id): # :api assert dnf.util.is_string_type(env_id) solver = self._build_comps_solver() trans = solver._environment_remove(env_id) return self._add_comps_trans(trans) def group_install(self, grp_id, pkg_types, exclude=None, strict=True): # :api """Installs packages of selected group :param pkg_types: Types of packages to install. Either an integer as a logical conjunction of CompsPackageType ids or a list of string package type ids (conditional, default, mandatory, optional). :param exclude: list of package name glob patterns that will be excluded from install set :param strict: boolean indicating whether group packages that exist but are non-installable due to e.g. dependency issues should be skipped (False) or cause transaction to fail to resolve (True) """ def _pattern_to_pkgname(pattern): if dnf.util.is_glob_pattern(pattern): q = self.sack.query().filterm(name__glob=pattern) return map(lambda p: p.name, q) else: return (pattern,) assert dnf.util.is_string_type(grp_id) exclude_pkgnames = None if exclude: nested_excludes = [_pattern_to_pkgname(p) for p in exclude] exclude_pkgnames = itertools.chain.from_iterable(nested_excludes) solver = self._build_comps_solver() if not isinstance(pkg_types, int): pkg_types = libdnf.transaction.listToCompsPackageType(pkg_types) trans = solver._group_install(grp_id, pkg_types, exclude_pkgnames, strict) if not trans: return 0 if strict: instlog = trans.install else: instlog = trans.install_opt logger.debug(_("Adding packages from group '%s': %s"), grp_id, instlog) return self._add_comps_trans(trans) def env_group_install(self, patterns, types, strict=True, exclude=None, exclude_groups=None): q = CompsQuery(self.comps, self.history, CompsQuery.ENVIRONMENTS \| CompsQuery.GROUPS, CompsQuery.AVAILABLE) cnt = 0 done = True for pattern in patterns: try: res = q.get(pattern) except dnf.exceptions.CompsError as err: logger.error(ucd(err)) done = False continue for group_id in res.groups: if not exclude_groups or group_id not in exclude_groups: cnt += self.group_install(group_id, types, exclude=exclude, strict=strict) for env_id in res.environments: cnt += self.environment_install(env_id, types, exclude=exclude, strict=strict, exclude_groups=exclude_groups) if not done and strict: raise dnf.exceptions.Error(_('Nothing to do.')) return cnt def group_remove(self, grp_id): # :api assert dnf.util.is_string_type(grp_id) solver = self._build_comps_solver() trans = solver._group_remove(grp_id) return self._add_comps_trans(trans) def env_group_remove(self, patterns): q = CompsQuery(self.comps, self.history, CompsQuery.ENVIRONMENTS \| CompsQuery.GROUPS, CompsQuery.INSTALLED) try: res = q.get(patterns) except dnf.exceptions.CompsError as err: logger.error("Warning: %s", ucd(err)) raise dnf.exceptions.Error(_('No groups marked for removal.')) cnt = 0 for env in res.environments: cnt += self.environment_remove(env) for grp in res.groups: cnt += self.group_remove(grp) return cnt def env_group_upgrade(self, patterns): q = CompsQuery(self.comps, self.history, CompsQuery.GROUPS \| CompsQuery.ENVIRONMENTS, CompsQuery.INSTALLED) group_upgraded = False for pattern in patterns: try: res = q.get(pattern) except dnf.exceptions.CompsError as err: logger.error(ucd(err)) continue for env in res.environments: try: self.environment_upgrade(env) group_upgraded = True except dnf.exceptions.CompsError as err: logger.error(ucd(err)) continue for grp in res.groups: try: self.group_upgrade(grp) group_upgraded = True except dnf.exceptions.CompsError as err: logger.error(ucd(err)) continue if not group_upgraded: msg = _('No group marked for upgrade.') raise dnf.cli.CliError(msg) def environment_upgrade(self, env_id): # :api assert dnf.util.is_string_type(env_id) solver = self._build_comps_solver() trans = solver._environment_upgrade(env_id) return self._add_comps_trans(trans) def group_upgrade(self, grp_id): # :api assert dnf.util.is_string_type(grp_id) solver = self._build_comps_solver() trans = solver._group_upgrade(grp_id) return self._add_comps_trans(trans) def _gpg_key_check(self): """Checks for the presence of GPG keys in the rpmdb. :return: 0 if there are no GPG keys in the rpmdb, and 1 if there are keys """ gpgkeyschecked = self.conf.cachedir + '/.gpgkeyschecked.yum' if os.path.exists(gpgkeyschecked): return 1 installroot = self.conf.installroot myts = dnf.rpm.transaction.initReadOnlyTransaction(root=installroot) myts.pushVSFlags(~(rpm._RPMVSF_NOSIGNATURES \| rpm._RPMVSF_NODIGESTS)) idx = myts.dbMatch('name', 'gpg-pubkey') keys = len(idx) del idx del myts if keys == 0: return 0 else: mydir = os.path.dirname(gpgkeyschecked) if not os.path.exists(mydir): os.makedirs(mydir) fo = open(gpgkeyschecked, 'w') fo.close() del fo return 1 def _install_multiarch(self, query, reponame=None, strict=True): already_inst, available = self._query_matches_installed(query) self._report_already_installed(already_inst) for packages in available: sltr = dnf.selector.Selector(self.sack) q = self.sack.query().filterm(pkg=packages) if self.conf.obsoletes: # use only obsoletes of the latest versions # See https://bugzilla.redhat.com/show_bug.cgi?id=2176263 # for details of the problem. q = q.union(self.sack.query().filterm( obsoletes=q.filter(latest_per_arch_by_priority=True))) sltr = sltr.set(pkg=q) if reponame is not None: sltr = sltr.set(reponame=reponame) self._goal.install(select=sltr, optional=(not strict)) return len(available) def _categorize_specs(self, install, exclude): """ Categorize :param install and :param exclude list into two groups each (packages and groups) :param install: list of specs, whether packages ('foo') or groups/modules ('@bar') :param exclude: list of specs, whether packages ('foo') or groups/modules ('@bar') :return: categorized install and exclude specs (stored in argparse.Namespace class) To access packages use: specs.pkg_specs, to access groups use: specs.grp_specs """ install_specs = argparse.Namespace() exclude_specs = argparse.Namespace() _parse_specs(install_specs, install) _parse_specs(exclude_specs, exclude) return install_specs, exclude_specs def _exclude_package_specs(self, exclude_specs): glob_excludes = [exclude for exclude in exclude_specs.pkg_specs if dnf.util.is_glob_pattern(exclude)] excludes = [exclude for exclude in exclude_specs.pkg_specs if exclude not in glob_excludes] exclude_query = self.sack.query().filter(name=excludes) glob_exclude_query = self.sack.query().filter(name__glob=glob_excludes) self.sack.add_excludes(exclude_query) self.sack.add_excludes(glob_exclude_query) def _expand_groups(self, group_specs): groups = set() q = CompsQuery(self.comps, self.history, CompsQuery.ENVIRONMENTS \| CompsQuery.GROUPS, CompsQuery.AVAILABLE \| CompsQuery.INSTALLED) for pattern in group_specs: try: res = q.get(pattern) except dnf.exceptions.CompsError as err: logger.error("Warning: Module or %s", ucd(err)) continue groups.update(res.groups) groups.update(res.environments) for environment_id in res.environments: environment = self.comps._environment_by_id(environment_id) for group in environment.groups_iter(): groups.add(group.id) return list(groups) def _install_groups(self, group_specs, excludes, skipped, strict=True): for group_spec in group_specs: try: types = self.conf.group_package_types if '/' in group_spec: split = group_spec.split('/') group_spec = split[0] types = split[1].split(',') self.env_group_install([group_spec], types, strict, excludes.pkg_specs, excludes.grp_specs) except dnf.exceptions.Error: skipped.append("@" + group_spec) def install_specs(self, install, exclude=None, reponame=None, strict=True, forms=None): # :api if exclude is None: exclude = [] no_match_group_specs = [] error_group_specs = [] no_match_pkg_specs = [] error_pkg_specs = [] install_specs, exclude_specs = self._categorize_specs(install, exclude) self._exclude_package_specs(exclude_specs) for spec in install_specs.pkg_specs: try: self.install(spec, reponame=reponame, strict=strict, forms=forms) except dnf.exceptions.MarkingError as e: logger.error(str(e)) no_match_pkg_specs.append(spec) no_match_module_specs = [] module_depsolv_errors = () if WITH_MODULES and install_specs.grp_specs: try: module_base = dnf.module.module_base.ModuleBase(self) module_base.install(install_specs.grp_specs, strict) except dnf.exceptions.MarkingErrors as e: if e.no_match_group_specs: for e_spec in e.no_match_group_specs: no_match_module_specs.append(e_spec) if e.error_group_specs: for e_spec in e.error_group_specs: error_group_specs.append("@" + e_spec) module_depsolv_errors = e.module_depsolv_errors else: no_match_module_specs = install_specs.grp_specs if no_match_module_specs: exclude_specs.grp_specs = self._expand_groups(exclude_specs.grp_specs) self._install_groups(no_match_module_specs, exclude_specs, no_match_group_specs, strict) if no_match_group_specs or error_group_specs or no_match_pkg_specs or error_pkg_specs \ or module_depsolv_errors: raise dnf.exceptions.MarkingErrors(no_match_group_specs=no_match_group_specs, error_group_specs=error_group_specs, no_match_pkg_specs=no_match_pkg_specs, error_pkg_specs=error_pkg_specs, module_depsolv_errors=module_depsolv_errors) def install(self, pkg_spec, reponame=None, strict=True, forms=None): # :api """Mark package(s) given by pkg_spec and reponame for installation.""" kwargs = {'forms': forms, 'with_src': False} if forms: kwargs['with_nevra'] = True kwargs['with_provides'] = False kwargs['with_filenames'] = False subj = dnf.subject.Subject(pkg_spec) solution = subj.get_best_solution(self.sack, kwargs) if self.conf.multilib_policy == "all" or subj._is_arch_specified(solution): q = solution['query'] if reponame is not None: q.filterm(reponame=reponame) if not q: self._raise_package_not_found_error(pkg_spec, forms, reponame) return self._install_multiarch(q, reponame=reponame, strict=strict) elif self.conf.multilib_policy == "best": sltrs = subj._get_best_selectors(self, forms=forms, obsoletes=self.conf.obsoletes, reponame=reponame, reports=True, solution=solution) if not sltrs: self._raise_package_not_found_error(pkg_spec, forms, reponame) for sltr in sltrs: self._goal.install(select=sltr, optional=(not strict)) return 1 return 0 def package_downgrade(self, pkg, strict=False): # :api if pkg._from_system: msg = 'downgrade_package() for an installed package.' raise NotImplementedError(msg) q = self.sack.query().installed().filterm(name=pkg.name, arch=[pkg.arch, "noarch"]) if not q: msg = _("Package %s not installed, cannot downgrade it.") logger.warning(msg, pkg.name) raise dnf.exceptions.MarkingError(_('No match for argument: %s') % pkg.location, pkg.name) elif sorted(q)[0] > pkg: sltr = dnf.selector.Selector(self.sack) sltr.set(pkg=[pkg]) self._goal.install(select=sltr, optional=(not strict)) return 1 else: msg = _("Package %s of lower version already installed, " "cannot downgrade it.") logger.warning(msg, pkg.name) return 0 def package_install(self, pkg, strict=True): # :api q = self.sack.query()._nevra(pkg.name, pkg.evr, pkg.arch) already_inst, available = self._query_matches_installed(q) if pkg in already_inst: self._report_already_installed([pkg]) elif pkg not in itertools.chain.from_iterable(available): raise dnf.exceptions.PackageNotFoundError(_('No match for argument: %s') % pkg.location) else: sltr = dnf.selector.Selector(self.sack) sltr.set(pkg=[pkg]) self._goal.install(select=sltr, optional=(not strict)) return 1 def package_reinstall(self, pkg): if self.sack.query().installed().filterm(name=pkg.name, evr=pkg.evr, arch=pkg.arch): self._goal.install(pkg) return 1 msg = _("Package %s not installed, cannot reinstall it.") logger.warning(msg, str(pkg)) raise dnf.exceptions.MarkingError(_('No match for argument: %s') % pkg.location, pkg.name) def package_remove(self, pkg): self._goal.erase(pkg) return 1 def package_upgrade(self, pkg): # :api if pkg._from_system: msg = 'upgrade_package() for an installed package.' raise NotImplementedError(msg) if pkg.arch == 'src': msg = _("File %s is a source package and cannot be updated, ignoring.") logger.info(msg, pkg.location) return 0 installed = self.sack.query().installed().apply() if self.conf.obsoletes and self.sack.query().filterm(pkg=[pkg]).filterm(obsoletes=installed): sltr = dnf.selector.Selector(self.sack) sltr.set(pkg=[pkg]) self._goal.upgrade(select=sltr) return 1 # do not filter by arch if the package is noarch if pkg.arch == "noarch": q = installed.filter(name=pkg.name) else: q = installed.filter(name=pkg.name, arch=[pkg.arch, "noarch"]) if not q: msg = _("Package %s not installed, cannot update it.") logger.warning(msg, pkg.name) raise dnf.exceptions.MarkingError( _('No match for argument: %s') % pkg.location, pkg.name) elif sorted(q)[-1] < pkg: sltr = dnf.selector.Selector(self.sack) sltr.set(pkg=[pkg]) self._goal.upgrade(select=sltr) return 1 else: msg = _("The same or higher version of %s is already installed, " "cannot update it.") logger.warning(msg, pkg.name) return 0 def _upgrade_internal(self, query, obsoletes, reponame, pkg_spec=None): installed_all = self.sack.query().installed() # Add only relevant obsoletes to transaction => installed, upgrades q = query.intersection(self.sack.query().filterm(name=[pkg.name for pkg in installed_all])) installed_query = q.installed() if obsoletes: obsoletes = self.sack.query().available().filterm( obsoletes=installed_query.union(q.upgrades())) # add obsoletes into transaction query = query.union(obsoletes) if reponame is not None: query.filterm(reponame=reponame) query = self._merge_update_filters(query, pkg_spec=pkg_spec, upgrade=True) if query: # Given that we use libsolv's targeted transactions, we need to ensure that the transaction contains both # the new targeted version and also the current installed version (for the upgraded package). This is # because if it only contained the new version, libsolv would decide to reinstall the package even if it # had just a different buildtime or vendor but the same version # (https://github.com/openSUSE/libsolv/issues/287) # - In general, the query already contains both the new and installed versions but not always. # If repository-packages command is used, the installed packages are filtered out because they are from # the @system repo. We need to add them back in. # - However we need to add installed versions of just the packages that are being upgraded. We don't want # to add all installed packages because it could increase the number of solutions for the transaction # (especially without --best) and since libsolv prefers the smallest possible upgrade it could result # in no upgrade even if there is one available. This is a problem in general but its critical with # --security transactions (https://bugzilla.redhat.com/show_bug.cgi?id=2097757) # - We want to add only the latest versions of installed packages, this is specifically for installonly # packages. Otherwise if for example kernel-1 and kernel-3 were installed and present in the # transaction libsolv could decide to install kernel-2 because it is an upgrade for kernel-1 even # though we don't want it because there already is a newer version present. query = query.union(installed_all.latest().filter(name=[pkg.name for pkg in query])) sltr = dnf.selector.Selector(self.sack) sltr.set(pkg=query) self._goal.upgrade(select=sltr) return 1 def upgrade(self, pkg_spec, reponame=None): # :api subj = dnf.subject.Subject(pkg_spec) solution = subj.get_best_solution(self.sack) q = solution["query"] if q: wildcard = dnf.util.is_glob_pattern(pkg_spec) # wildcard shouldn't print not installed packages # only solution with nevra.name provide packages with same name if not wildcard and solution['nevra'] and solution['nevra'].name: pkg_name = solution['nevra'].name installed = self.sack.query().installed().apply() obsoleters = q.filter(obsoletes=installed) \ if self.conf.obsoletes else self.sack.query().filterm(empty=True) if not obsoleters: installed_name = installed.filter(name=pkg_name).apply() if not installed_name: msg = _('Package %s available, but not installed.') logger.warning(msg, pkg_name) raise dnf.exceptions.PackagesNotInstalledError( _('No match for argument: %s') % pkg_spec, pkg_spec) elif solution['nevra'].arch and not dnf.util.is_glob_pattern(solution['nevra'].arch): if not installed_name.filterm(arch=solution['nevra'].arch): msg = _('Package %s available, but installed for different architecture.') logger.warning(msg, "{}.{}".format(pkg_name, solution['nevra'].arch)) obsoletes = self.conf.obsoletes and solution['nevra'] \ and solution['nevra'].has_just_name() return self._upgrade_internal(q, obsoletes, reponame, pkg_spec) raise dnf.exceptions.MarkingError(_('No match for argument: %s') % pkg_spec, pkg_spec) def upgrade_all(self, reponame=None): # :api # provide only available packages to solver to trigger targeted upgrade # possibilities will be ignored # usage of selected packages will unify dnf behavior with other upgrade functions return self._upgrade_internal( self.sack.query(), self.conf.obsoletes, reponame, pkg_spec=None) def distro_sync(self, pkg_spec=None): if pkg_spec is None: self._goal.distupgrade_all() else: subject = dnf.subject.Subject(pkg_spec) solution = subject.get_best_solution(self.sack, with_src=False) solution["query"].filterm(reponame__neq=hawkey.SYSTEM_REPO_NAME) sltrs = subject._get_best_selectors(self, solution=solution, obsoletes=self.conf.obsoletes, reports=True) if not sltrs: logger.info(_('No package %s installed.'), pkg_spec) return 0 for sltr in sltrs: self._goal.distupgrade(select=sltr) return 1 def autoremove(self, forms=None, pkg_specs=None, grp_specs=None, filenames=None): # :api """Removes all 'leaf' packages from the system that were originally installed as dependencies of user-installed packages but which are no longer required by any such package.""" if any([grp_specs, pkg_specs, filenames]): pkg_specs += filenames done = False # Remove groups. if grp_specs and forms: for grp_spec in grp_specs: msg = _('Not a valid form: %s') logger.warning(msg, grp_spec) elif grp_specs: if self.env_group_remove(grp_specs): done = True for pkg_spec in pkg_specs: try: self.remove(pkg_spec, forms=forms) except dnf.exceptions.MarkingError as e: logger.info(str(e)) else: done = True if not done: logger.warning(_('No packages marked for removal.')) else: pkgs = self.sack.query()._unneeded(self.history.swdb, debug_solver=self.conf.debug_solver) for pkg in pkgs: self.package_remove(pkg) def remove(self, pkg_spec, reponame=None, forms=None): # :api """Mark the specified package for removal.""" kwargs = {'forms': forms} if forms: kwargs['with_nevra'] = True kwargs['with_provides'] = False kwargs['with_filenames'] = False matches = dnf.subject.Subject(pkg_spec).get_best_query(self.sack, kwargs) installed = [ pkg for pkg in matches.installed() if reponame is None or self.history.repo(pkg) == reponame] if not installed: self._raise_package_not_installed_error(pkg_spec, forms, reponame) clean_deps = self.conf.clean_requirements_on_remove for pkg in installed: self._goal.erase(pkg, clean_deps=clean_deps) return len(installed) def reinstall(self, pkg_spec, old_reponame=None, new_reponame=None, new_reponame_neq=None, remove_na=False): subj = dnf.subject.Subject(pkg_spec) q = subj.get_best_query(self.sack) installed_pkgs = [ pkg for pkg in q.installed() if old_reponame is None or self.history.repo(pkg) == old_reponame] available_q = q.available() if new_reponame is not None: available_q.filterm(reponame=new_reponame) if new_reponame_neq is not None: available_q.filterm(reponame__neq=new_reponame_neq) available_nevra2pkg = dnf.query._per_nevra_dict(available_q) if not installed_pkgs: raise dnf.exceptions.PackagesNotInstalledError( 'no package matched', pkg_spec, available_q.run()) cnt = 0 clean_deps = self.conf.clean_requirements_on_remove strict = self.conf.strict for installed_pkg in installed_pkgs: try: available_pkgs = available_nevra2pkg[ucd(installed_pkg)] except KeyError: if not remove_na: continue self._goal.erase(installed_pkg, clean_deps=clean_deps) else: sltr = dnf.selector.Selector(self.sack) sltr.set(pkg=available_pkgs) self._goal.install(select=sltr, optional=(not strict)) cnt += 1 if cnt == 0: raise dnf.exceptions.PackagesNotAvailableError( 'no package matched', pkg_spec, installed_pkgs) return cnt def downgrade(self, pkg_spec): # :api """Mark a package to be downgraded. This is equivalent to first removing the currently installed package, and then installing an older version. """ return self.downgrade_to(pkg_spec) def downgrade_to(self, pkg_spec, strict=False): """Downgrade to specific version if specified otherwise downgrades to one version lower than the package installed. """ subj = dnf.subject.Subject(pkg_spec) q = subj.get_best_query(self.sack) if not q: msg = _('No match for argument: %s') % pkg_spec raise dnf.exceptions.PackageNotFoundError(msg, pkg_spec) done = 0 available_pkgs = q.available() available_pkg_names = list(available_pkgs._name_dict().keys()) q_installed = self.sack.query().installed().filterm(name=available_pkg_names) if len(q_installed) == 0: msg = _('Packages for argument %s available, but not installed.') % pkg_spec raise dnf.exceptions.PackagesNotInstalledError(msg, pkg_spec, available_pkgs) for pkg_name in q_installed._name_dict().keys(): downgrade_pkgs = available_pkgs.downgrades().filter(name=pkg_name) if not downgrade_pkgs: msg = _("Package %s of lowest version already installed, cannot downgrade it.") logger.warning(msg, pkg_name) continue sltr = dnf.selector.Selector(self.sack) sltr.set(pkg=downgrade_pkgs) self._goal.install(select=sltr, optional=(not strict)) done = 1 return done def provides(self, provides_spec): providers = self.sack.query().filterm(file__glob=provides_spec) if providers: return providers, [provides_spec] providers = dnf.query._by_provides(self.sack, provides_spec) if providers: return providers, [provides_spec] if provides_spec.startswith('/bin/') or provides_spec.startswith('/sbin/'): # compatibility for packages that didn't do UsrMove binary_provides = ['/usr' + provides_spec] elif provides_spec.startswith('/'): # provides_spec is a file path return providers, [provides_spec] else: # suppose that provides_spec is a command, search in /usr/sbin/ binary_provides = [prefix + provides_spec for prefix in ['/bin/', '/sbin/', '/usr/bin/', '/usr/sbin/']] return self.sack.query().filterm(file__glob=binary_provides), binary_provides def add_security_filters(self, cmp_type, types=(), advisory=(), bugzilla=(), cves=(), severity=()): # :api """ It modifies results of install, upgrade, and distrosync methods according to provided filters. :param cmp_type: only 'eq' or 'gte' allowed :param types: List or tuple with strings. E.g. 'bugfix', 'enhancement', 'newpackage', 'security' :param advisory: List or tuple with strings. E.g.Eg. FEDORA-2201-123 :param bugzilla: List or tuple with strings. Include packages that fix a Bugzilla ID, Eg. 123123. :param cves: List or tuple with strings. Include packages that fix a CVE (Common Vulnerabilities and Exposures) ID. Eg. CVE-2201-0123 :param severity: List or tuple with strings. Includes packages that provide a fix for an issue of the specified severity. """ cmp_dict = {'eq': '__eqg', 'gte': '__eqg__gt'} if cmp_type not in cmp_dict: raise ValueError("Unsupported value for `cmp_type`") cmp = cmp_dict[cmp_type] if types: key = 'advisory_type' + cmp self._update_security_options.setdefault(key, set()).update(types) if advisory: key = 'advisory' + cmp self._update_security_options.setdefault(key, set()).update(advisory) if bugzilla: key = 'advisory_bug' + cmp self._update_security_options.setdefault(key, set()).update(bugzilla) if cves: key = 'advisory_cve' + cmp self._update_security_options.setdefault(key, set()).update(cves) if severity: key = 'advisory_severity' + cmp self._update_security_options.setdefault(key, set()).update(severity) def reset_security_filters(self): # :api """ Reset all security filters """ self._update_security_options = {} def _merge_update_filters(self, q, pkg_spec=None, warning=True, upgrade=False): """ Merge Queries in _update_filters and return intersection with q Query @param q: Query @return: Query """ if not (self._update_security_options or self._update_security_filters) or not q: return q merged_queries = self.sack.query().filterm(empty=True) if self._update_security_filters: for query in self._update_security_filters: merged_queries = merged_queries.union(query) self._update_security_filters = [merged_queries] if self._update_security_options: for filter_name, values in self._update_security_options.items(): if upgrade: filter_name = filter_name + '__upgrade' kwargs = {filter_name: values} merged_queries = merged_queries.union(q.filter(*kwargs)) merged_queries = q.intersection(merged_queries) if not merged_queries: if warning: q = q.upgrades() count = len(q._name_dict().keys()) if count > 0: if pkg_spec is None: msg1 = _("No security updates needed, but {} update " "available").format(count) msg2 = _("No security updates needed, but {} updates " "available").format(count) logger.warning(P_(msg1, msg2, count)) else: msg1 = _('No security updates needed for "{}", but {} ' 'update available').format(pkg_spec, count) msg2 = _('No security updates needed for "{}", but {} ' 'updates available').format(pkg_spec, count) logger.warning(P_(msg1, msg2, count)) return merged_queries def _get_key_for_package(self, po, askcb=None, fullaskcb=None): """Retrieve a key for a package. If needed, use the given callback to prompt whether the key should be imported. :param po: the package object to retrieve the key of :param askcb: Callback function to use to ask permission to import a key. The arguments askcb* should take are the package object, the userid of the key, and the keyid :param fullaskcb: Callback function to use to ask permission to import a key. This differs from askcb in that it gets passed a dictionary so that we can expand the values passed. :raises: :class:`dnf.exceptions.Error` if there are errors retrieving the keys """ if po._from_cmdline: # raise an exception, because po.repoid is not in self.repos msg = _('Unable to retrieve a key for a commandline package: %s') raise ValueError(msg % po) repo = self.repos[po.repoid] key_installed = repo.id in self._repo_set_imported_gpg_keys keyurls = [] if key_installed else repo.gpgkey def _prov_key_data(msg): msg += _('. Failing package is: %s') % (po) + '\n ' msg += _('GPG Keys are configured as: %s') % \ (', '.join(repo.gpgkey)) return msg user_cb_fail = False self._repo_set_imported_gpg_keys.add(repo.id) for keyurl in keyurls: keys = dnf.crypto.retrieve(keyurl, repo) for info in keys: # Check if key is already installed if misc.keyInstalled(self._ts, info.rpm_id, info.timestamp) >= 0: msg = _('GPG key at %s (0x%s) is already installed') logger.info(msg, keyurl, info.short_id) continue # DNS Extension: create a key object, pass it to the verification class # and print its result as an advice to the user. if self.conf.gpgkey_dns_verification: dns_input_key = dnf.dnssec.KeyInfo.from_rpm_key_object(info.userid, info.raw_key) dns_result = dnf.dnssec.DNSSECKeyVerification.verify(dns_input_key) logger.info(dnf.dnssec.nice_user_msg(dns_input_key, dns_result)) # Try installing/updating GPG key info.url = keyurl if self.conf.gpgkey_dns_verification: dnf.crypto.log_dns_key_import(info, dns_result) else: dnf.crypto.log_key_import(info) rc = False if self.conf.assumeno: rc = False elif self.conf.assumeyes: # DNS Extension: We assume, that the key is trusted in case it is valid, # its existence is explicitly denied or in case the domain is not signed # and therefore there is no way to know for sure (this is mainly for # backward compatibility) # FAQ: # * What is PROVEN_NONEXISTENCE? # In DNSSEC, your domain does not need to be signed, but this state # (not signed) has to be proven by the upper domain. e.g. when example.com. # is not signed, com. servers have to sign the message, that example.com. # does not have any signing key (KSK to be more precise). if self.conf.gpgkey_dns_verification: if dns_result in (dnf.dnssec.Validity.VALID, dnf.dnssec.Validity.PROVEN_NONEXISTENCE): rc = True logger.info(dnf.dnssec.any_msg(_("The key has been approved."))) else: rc = False logger.info(dnf.dnssec.any_msg(_("The key has been rejected."))) else: rc = True # grab the .sig/.asc for the keyurl, if it exists if it # does check the signature on the key if it is signed by # one of our ca-keys for this repo or the global one then # rc = True else ask as normal. elif fullaskcb: rc = fullaskcb({"po": po, "userid": info.userid, "hexkeyid": info.short_id, "keyurl": keyurl, "fingerprint": info.fingerprint, "timestamp": info.timestamp}) elif askcb: rc = askcb(po, info.userid, info.short_id) if not rc: user_cb_fail = True continue # Import the key # If rpm.RPMTRANS_FLAG_TEST in self._ts, gpg keys cannot be imported successfully # therefore the flag was removed for import operation test_flag = self._ts.isTsFlagSet(rpm.RPMTRANS_FLAG_TEST) if test_flag: orig_flags = self._ts.getTsFlags() self._ts.setFlags(orig_flags - rpm.RPMTRANS_FLAG_TEST) result = self._ts.pgpImportPubkey(misc.procgpgkey(info.raw_key)) if test_flag: self._ts.setFlags(orig_flags) if result != 0: msg = _('Key import failed (code %d)') % result raise dnf.exceptions.Error(_prov_key_data(msg)) logger.info(_('Key imported successfully')) key_installed = True if not key_installed and user_cb_fail: raise dnf.exceptions.Error(_("Didn't install any keys")) if not key_installed: msg = _('The GPG keys listed for the "%s" repository are ' 'already installed but they are not correct for this ' 'package.\n' 'Check that the correct key URLs are configured for ' 'this repository.') % repo.name raise dnf.exceptions.Error(_prov_key_data(msg)) # Check if the newly installed keys helped result, errmsg = self._sig_check_pkg(po) if result != 0: if keyurls: msg = _("Import of key(s) didn't help, wrong key(s)?") logger.info(msg) errmsg = ucd(errmsg) raise dnf.exceptions.Error(_prov_key_data(errmsg)) def package_import_key(self, pkg, askcb=None, fullaskcb=None): # :api """Retrieve a key for a package. If needed, use the given callback to prompt whether the key should be imported. :param pkg: the package object to retrieve the key of :param askcb: Callback function to use to ask permission to import a key. The arguments askcb should take are the package object, the userid of the key, and the keyid :param fullaskcb: Callback function to use to ask permission to import a key. This differs from askcb in that it gets passed a dictionary so that we can expand the values passed. :raises: :class:`dnf.exceptions.Error` if there are errors retrieving the keys """ self._get_key_for_package(pkg, askcb, fullaskcb) def _run_rpm_check(self): results = [] self._ts.check() for prob in self._ts.problems(): # Newer rpm (4.8.0+) has problem objects, older have just strings. # Should probably move to using the new objects, when we can. For # now just be compatible. results.append(ucd(prob)) return results def urlopen(self, url, repo=None, mode='w+b', kwargs): # :api """ Open the specified absolute url, return a file object which respects proxy setting even for non-repo downloads """ return dnf.util._urlopen(url, self.conf, repo, mode, kwargs) def _get_installonly_query(self, q=None): if q is None: q = self._sack.query(flags=hawkey.IGNORE_EXCLUDES) installonly = q.filter(provides=self.conf.installonlypkgs) return installonly def _report_icase_hint(self, pkg_spec): subj = dnf.subject.Subject(pkg_spec, ignore_case=True) solution = subj.get_best_solution(self.sack, with_nevra=True, with_provides=False, with_filenames=False) if solution['query'] and solution['nevra'] and solution['nevra'].name and \ pkg_spec != solution['query'][0].name: logger.info(_(" * Maybe you meant: {}").format(solution['query'][0].name)) def _select_remote_pkgs(self, install_pkgs): """ Check checksum of packages from local repositories and returns list packages from remote repositories that will be downloaded. Packages from commandline are skipped. :param install_pkgs: list of packages :return: list of remote pkgs """ def _verification_of_packages(pkg_list, logger_msg): all_packages_verified = True for pkg in pkg_list: pkg_successfully_verified = False try: pkg_successfully_verified = pkg.verifyLocalPkg() except Exception as e: logger.critical(str(e)) if pkg_successfully_verified is not True: logger.critical(logger_msg.format(pkg, pkg.reponame)) all_packages_verified = False return all_packages_verified remote_pkgs = [] local_repository_pkgs = [] for pkg in install_pkgs: if pkg._is_local_pkg(): if pkg.reponame != hawkey.CMDLINE_REPO_NAME: local_repository_pkgs.append(pkg) else: remote_pkgs.append(pkg) msg = _('Package "{}" from local repository "{}" has incorrect checksum') if not _verification_of_packages(local_repository_pkgs, msg): raise dnf.exceptions.Error( _("Some packages from local repository have incorrect checksum")) if self.conf.cacheonly: msg = _('Package "{}" from repository "{}" has incorrect checksum') if not _verification_of_packages(remote_pkgs, msg): raise dnf.exceptions.Error( _('Some packages have invalid cache, but cannot be downloaded due to ' '"--cacheonly" option')) remote_pkgs = [] return remote_pkgs, local_repository_pkgs def _report_already_installed(self, packages): for pkg in packages: _msg_installed(pkg) def _raise_package_not_found_error(self, pkg_spec, forms, reponame): all_query = self.sack.query(flags=hawkey.IGNORE_EXCLUDES) subject = dnf.subject.Subject(pkg_spec) solution = subject.get_best_solution( self.sack, forms=forms, with_src=False, query=all_query) if reponame is not None: solution['query'].filterm(reponame=reponame) if not solution['query']: raise dnf.exceptions.PackageNotFoundError(_('No match for argument'), pkg_spec) else: with_regular_query = self.sack.query(flags=hawkey.IGNORE_REGULAR_EXCLUDES) with_regular_query = solution['query'].intersection(with_regular_query) # Modular filtering is applied on a package set that already has regular excludes # filtered out. So if a package wasn't filtered out by regular excludes, it must have # been filtered out by modularity. if with_regular_query: msg = _('All matches were filtered out by exclude filtering for argument') else: msg = _('All matches were filtered out by modular filtering for argument') raise dnf.exceptions.PackageNotFoundError(msg, pkg_spec) def _raise_package_not_installed_error(self, pkg_spec, forms, reponame): all_query = self.sack.query(flags=hawkey.IGNORE_EXCLUDES).installed() subject = dnf.subject.Subject(pkg_spec) solution = subject.get_best_solution( self.sack, forms=forms, with_src=False, query=all_query) if not solution['query']: raise dnf.exceptions.PackagesNotInstalledError(_('No match for argument'), pkg_spec) if reponame is not None: installed = [pkg for pkg in solution['query'] if self.history.repo(pkg) == reponame] else: installed = solution['query'] if not installed: msg = _('All matches were installed from a different repository for argument') else: msg = _('All matches were filtered out by exclude filtering for argument') raise dnf.exceptions.PackagesNotInstalledError(msg, pkg_spec) def setup_loggers(self): # :api """ Setup DNF file loggers based on given configuration file. The loggers are set the same way as if DNF was run from CLI. """ self._logging._setup_from_dnf_conf(self.conf, file_loggers_only=True) def _skipped_packages(self, report_problems, transaction): """returns set of conflicting packages and set of packages with broken dependency that would be additionally installed when --best and --allowerasing""" if self._goal.actions & (hawkey.INSTALL \| hawkey.UPGRADE \| hawkey.UPGRADE_ALL): best = True else: best = False ng = deepcopy(self._goal) params = {"allow_uninstall": self._allow_erasing, "force_best": best, "ignore_weak": True} ret = ng.run(**params) if not ret and report_problems: msg = dnf.util._format_resolve_problems(ng.problem_rules()) logger.warning(msg) problem_conflicts = set(ng.problem_conflicts(available=True)) problem_dependency = set(ng.problem_broken_dependency(available=True)) - problem_conflicts def _nevra(item): return hawkey.NEVRA(name=item.name, epoch=item.epoch, version=item.version, release=item.release, arch=item.arch) # Sometimes, pkg is not in transaction item, therefore, comparing by nevra transaction_nevras = [_nevra(tsi) for tsi in transaction] skipped_conflicts = set( [pkg for pkg in problem_conflicts if _nevra(pkg) not in transaction_nevras]) skipped_dependency = set( [pkg for pkg in problem_dependency if _nevra(pkg) not in transaction_nevras]) return skipped_conflicts, skipped_dependency def reboot_needed(self): """Check whether a system reboot is recommended following the transaction :return: bool """ if not self.transaction: return False # List taken from DNF needs-restarting need_reboot = frozenset(('kernel', 'kernel-rt', 'glibc', 'linux-firmware', 'systemd', 'dbus', 'dbus-broker', 'dbus-daemon', 'microcode_ctl')) changed_pkgs = self.transaction.install_set \| self.transaction.remove_set return any(pkg.name in need_reboot for pkg in changed_pkgs) def _msg_installed(pkg): name = ucd(pkg) msg = _('Package %s is already installed.') logger.info(msg, name)	118.17 KB	2025-09-22 11:27
📄 callback.py # callbacks.py # Abstract interfaces to communicate progress on tasks. # # 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 unicode_literals import dnf.yum.rpmtrans import dnf.transaction PKG_DOWNGRADE = dnf.transaction.PKG_DOWNGRADE # :api PKG_DOWNGRADED = dnf.transaction.PKG_DOWNGRADED # :api PKG_INSTALL = dnf.transaction.PKG_INSTALL # :api PKG_OBSOLETE = dnf.transaction.PKG_OBSOLETE # :api PKG_OBSOLETED = dnf.transaction.PKG_OBSOLETED # :api PKG_REINSTALL = dnf.transaction.PKG_REINSTALL # :api PKG_REINSTALLED = dnf.transaction.PKG_REINSTALLED # :api PKG_REMOVE = dnf.transaction.PKG_ERASE # :api PKG_ERASE = PKG_REMOVE # deprecated, use PKG_REMOVE instead PKG_UPGRADE = dnf.transaction.PKG_UPGRADE # :api PKG_UPGRADED = dnf.transaction.PKG_UPGRADED # :api PKG_CLEANUP = dnf.transaction.PKG_CLEANUP # :api PKG_VERIFY = dnf.transaction.PKG_VERIFY # :api PKG_SCRIPTLET = dnf.transaction.PKG_SCRIPTLET # :api TRANS_PREPARATION = dnf.transaction.TRANS_PREPARATION # :api TRANS_POST = dnf.transaction.TRANS_POST # :api STATUS_OK = None # :api STATUS_FAILED = 1 # :api STATUS_ALREADY_EXISTS = 2 # :api STATUS_MIRROR = 3 # :api STATUS_DRPM = 4 # :api class KeyImport(object): def _confirm(self, id, userid, fingerprint, url, timestamp): """Ask the user if the key should be imported.""" return False class Payload(object): # :api def __init__(self, progress): self.progress = progress def __str__(self): """Nice, human-readable representation. :api""" pass @property def download_size(self): """Total size of the download. :api""" pass class DownloadProgress(object): # :api def end(self, payload, status, msg): """Communicate the information that `payload` has finished downloading. :api, `status` is a constant denoting the type of outcome, `err_msg` is an error message in case the outcome was an error. """ pass def message(self, msg): pass def progress(self, payload, done): """Update the progress display. :api `payload` is the payload this call reports progress for, `done` is how many bytes of this payload are already downloaded. """ pass def start(self, total_files, total_size, total_drpms=0): """Start new progress metering. :api `total_files` the number of files that will be downloaded, `total_size` total size of all files. """ pass class NullDownloadProgress(DownloadProgress): pass class Depsolve(object): def start(self): pass def pkg_added(self, pkg, mode): pass def end(self): pass TransactionProgress = dnf.yum.rpmtrans.TransactionDisplay # :api	3.64 KB	2022-09-09 11:21
📁 cli	-	2026-03-21 15:47
📄 comps.py # comps.py # Interface to libcomps. # # Copyright (C) 2013-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 print_function from __future__ import unicode_literals import libdnf.transaction from dnf.exceptions import CompsError from dnf.i18n import _, ucd from functools import reduce import dnf.i18n import dnf.util import fnmatch import gettext import itertools import libcomps import locale import logging import operator import re import sys logger = logging.getLogger("dnf") # :api :binformat CONDITIONAL = libdnf.transaction.CompsPackageType_CONDITIONAL DEFAULT = libdnf.transaction.CompsPackageType_DEFAULT MANDATORY = libdnf.transaction.CompsPackageType_MANDATORY OPTIONAL = libdnf.transaction.CompsPackageType_OPTIONAL ALL_TYPES = CONDITIONAL \| DEFAULT \| MANDATORY \| OPTIONAL def _internal_comps_length(comps): collections = (comps.categories, comps.groups, comps.environments) return reduce(operator.__add__, map(len, collections)) def _first_if_iterable(seq): if seq is None: return None return dnf.util.first(seq) def _by_pattern(pattern, case_sensitive, sqn): """Return items from sqn matching either exactly or glob-wise.""" pattern = dnf.i18n.ucd(pattern) exact = {g for g in sqn if g.name == pattern or g.id == pattern} if exact: return exact if case_sensitive: match = re.compile(fnmatch.translate(pattern)).match else: match = re.compile(fnmatch.translate(pattern), flags=re.I).match ret = set() for g in sqn: if match(g.id): ret.add(g) elif g.name is not None and match(g.name): ret.add(g) elif g.ui_name is not None and match(g.ui_name): ret.add(g) return ret def _fn_display_order(group): return sys.maxsize if group.display_order is None else group.display_order def install_or_skip(install_fnc, grp_or_env_id, types, exclude=None, strict=True, exclude_groups=None): """ Installs a group or an environment identified by grp_or_env_id. This method is preserved for API compatibility. It used to catch an exception thrown when a gorup or env was already installed, which is no longer thrown. `install_fnc` has to be Solver._group_install or Solver._environment_install. """ return install_fnc(grp_or_env_id, types, exclude, strict, exclude_groups) class _Langs(object): """Get all usable abbreviations for the current language.""" def __init__(self): self.last_locale = None self.cache = None @staticmethod def _dotted_locale_str(): lcl = locale.getlocale(locale.LC_MESSAGES) if lcl == (None, None): return 'C' return '.'.join(lcl) def get(self): current_locale = self._dotted_locale_str() if self.last_locale == current_locale: return self.cache self.cache = [] locales = [current_locale] if current_locale != 'C': locales.append('C') for l in locales: for nlang in gettext._expand_lang(l): if nlang not in self.cache: self.cache.append(nlang) self.last_locale = current_locale return self.cache class CompsQuery(object): AVAILABLE = 1 INSTALLED = 2 ENVIRONMENTS = 1 GROUPS = 2 def __init__(self, comps, history, kinds, status): self.comps = comps self.history = history self.kinds = kinds self.status = status def _get_groups(self, available, installed): result = set() if self.status & self.AVAILABLE: result.update({i.id for i in available}) if self.status & self.INSTALLED: for i in installed: group = i.getCompsGroupItem() if not group: continue result.add(group.getGroupId()) return result def _get_envs(self, available, installed): result = set() if self.status & self.AVAILABLE: result.update({i.id for i in available}) if self.status & self.INSTALLED: for i in installed: env = i.getCompsEnvironmentItem() if not env: continue result.add(env.getEnvironmentId()) return result def get(self, *patterns): res = dnf.util.Bunch() res.environments = [] res.groups = [] for pat in patterns: envs = grps = [] if self.kinds & self.ENVIRONMENTS: available = self.comps.environments_by_pattern(pat) installed = self.history.env.search_by_pattern(pat) envs = self._get_envs(available, installed) res.environments.extend(envs) if self.kinds & self.GROUPS: available = self.comps.groups_by_pattern(pat) installed = self.history.group.search_by_pattern(pat) grps = self._get_groups(available, installed) res.groups.extend(grps) if not envs and not grps: if self.status == self.INSTALLED: msg = _("Module or Group '%s' is not installed.") % ucd(pat) elif self.status == self.AVAILABLE: msg = _("Module or Group '%s' is not available.") % ucd(pat) else: msg = _("Module or Group '%s' does not exist.") % ucd(pat) raise CompsError(msg) return res class Forwarder(object): def __init__(self, iobj, langs): self._i = iobj self._langs = langs def __getattr__(self, name): return getattr(self._i, name) def _ui_text(self, default, dct): for l in self._langs.get(): t = dct.get(l) if t is not None: return t return default @property def ui_description(self): return self._ui_text(self.desc, self.desc_by_lang) @property def ui_name(self): return self._ui_text(self.name, self.name_by_lang) class Category(Forwarder): # :api def __init__(self, iobj, langs, group_factory): super(Category, self).__init__(iobj, langs) self._group_factory = group_factory def _build_group(self, grp_id): grp = self._group_factory(grp_id.name) if grp is None: msg = "no group '%s' from category '%s'" raise ValueError(msg % (grp_id.name, self.id)) return grp def groups_iter(self): for grp_id in self.group_ids: yield self._build_group(grp_id) @property def groups(self): return list(self.groups_iter()) class Environment(Forwarder): # :api def __init__(self, iobj, langs, group_factory): super(Environment, self).__init__(iobj, langs) self._group_factory = group_factory def _build_group(self, grp_id): grp = self._group_factory(grp_id.name) if grp is None: msg = "no group '%s' from environment '%s'" raise ValueError(msg % (grp_id.name, self.id)) return grp def _build_groups(self, ids): groups = [] for gi in ids: try: groups.append(self._build_group(gi)) except ValueError as e: logger.error(e) return groups def groups_iter(self): for grp_id in itertools.chain(self.group_ids, self.option_ids): try: yield self._build_group(grp_id) except ValueError as e: logger.error(e) @property def mandatory_groups(self): return self._build_groups(self.group_ids) @property def optional_groups(self): return self._build_groups(self.option_ids) class Group(Forwarder): # :api def __init__(self, iobj, langs, pkg_factory): super(Group, self).__init__(iobj, langs) self._pkg_factory = pkg_factory self.selected = iobj.default def _packages_of_type(self, type_): return [pkg for pkg in self.packages if pkg.type == type_] @property def conditional_packages(self): return self._packages_of_type(libcomps.PACKAGE_TYPE_CONDITIONAL) @property def default_packages(self): return self._packages_of_type(libcomps.PACKAGE_TYPE_DEFAULT) def packages_iter(self): # :api return map(self._pkg_factory, self.packages) @property def mandatory_packages(self): return self._packages_of_type(libcomps.PACKAGE_TYPE_MANDATORY) @property def optional_packages(self): return self._packages_of_type(libcomps.PACKAGE_TYPE_OPTIONAL) @property def visible(self): return self._i.uservisible class Package(Forwarder): """Represents comps package data. :api""" _OPT_MAP = { libcomps.PACKAGE_TYPE_CONDITIONAL : CONDITIONAL, libcomps.PACKAGE_TYPE_DEFAULT : DEFAULT, libcomps.PACKAGE_TYPE_MANDATORY : MANDATORY, libcomps.PACKAGE_TYPE_OPTIONAL : OPTIONAL, } def __init__(self, ipkg): self._i = ipkg @property def name(self): # :api return self._i.name @property def option_type(self): # :api return self._OPT_MAP[self.type] class Comps(object): # :api def __init__(self): self._i = libcomps.Comps() self._langs = _Langs() def __len__(self): return _internal_comps_length(self._i) def _build_category(self, icategory): return Category(icategory, self._langs, self._group_by_id) def _build_environment(self, ienvironment): return Environment(ienvironment, self._langs, self._group_by_id) def _build_group(self, igroup): return Group(igroup, self._langs, self._build_package) def _build_package(self, ipkg): return Package(ipkg) def _add_from_xml_filename(self, fn): comps = libcomps.Comps() try: comps.fromxml_f(fn) except libcomps.ParserError: errors = comps.get_last_errors() raise CompsError(' '.join(errors)) self._i += comps @property def categories(self): # :api return list(self.categories_iter()) def category_by_pattern(self, pattern, case_sensitive=False): # :api assert dnf.util.is_string_type(pattern) cats = self.categories_by_pattern(pattern, case_sensitive) return _first_if_iterable(cats) def categories_by_pattern(self, pattern, case_sensitive=False): # :api assert dnf.util.is_string_type(pattern) return _by_pattern(pattern, case_sensitive, self.categories) def categories_iter(self): # :api return (self._build_category(c) for c in self._i.categories) @property def environments(self): # :api return sorted(self.environments_iter(), key=_fn_display_order) def _environment_by_id(self, id): assert dnf.util.is_string_type(id) return dnf.util.first(g for g in self.environments_iter() if g.id == id) def environment_by_pattern(self, pattern, case_sensitive=False): # :api assert dnf.util.is_string_type(pattern) envs = self.environments_by_pattern(pattern, case_sensitive) return _first_if_iterable(envs) def environments_by_pattern(self, pattern, case_sensitive=False): # :api assert dnf.util.is_string_type(pattern) envs = list(self.environments_iter()) found_envs = _by_pattern(pattern, case_sensitive, envs) return sorted(found_envs, key=_fn_display_order) def environments_iter(self): # :api return (self._build_environment(e) for e in self._i.environments) @property def groups(self): # :api return sorted(self.groups_iter(), key=_fn_display_order) def _group_by_id(self, id_): assert dnf.util.is_string_type(id_) return dnf.util.first(g for g in self.groups_iter() if g.id == id_) def group_by_pattern(self, pattern, case_sensitive=False): # :api assert dnf.util.is_string_type(pattern) grps = self.groups_by_pattern(pattern, case_sensitive) return _first_if_iterable(grps) def groups_by_pattern(self, pattern, case_sensitive=False): # :api assert dnf.util.is_string_type(pattern) grps = _by_pattern(pattern, case_sensitive, list(self.groups_iter())) return sorted(grps, key=_fn_display_order) def groups_iter(self): # :api return (self._build_group(g) for g in self._i.groups) class CompsTransPkg(object): def __init__(self, pkg_or_name): if dnf.util.is_string_type(pkg_or_name): # from package name self.basearchonly = False self.name = pkg_or_name self.optional = True self.requires = None elif isinstance(pkg_or_name, libdnf.transaction.CompsGroupPackage): # from swdb package # TODO: self.basearchonly = False # self.basearchonly = pkg_or_name.basearchonly self.name = pkg_or_name.getName() self.optional = pkg_or_name.getPackageType() & libcomps.PACKAGE_TYPE_OPTIONAL # TODO: self.requires = None # self.requires = pkg_or_name.requires else: # from comps package self.basearchonly = pkg_or_name.basearchonly self.name = pkg_or_name.name self.optional = pkg_or_name.type & libcomps.PACKAGE_TYPE_OPTIONAL self.requires = pkg_or_name.requires def __eq__(self, other): return (self.name == other.name and self.basearchonly == other.basearchonly and self.optional == other.optional and self.requires == other.requires) def __str__(self): return self.name def __hash__(self): return hash((self.name, self.basearchonly, self.optional, self.requires)) class TransactionBunch(object): def __init__(self): self._install = set() self._install_opt = set() self._remove = set() self._upgrade = set() def __iadd__(self, other): self._install.update(other._install) self._install_opt.update(other._install_opt) self._upgrade.update(other._upgrade) self._remove = (self._remove \| other._remove) - \ self._install - self._install_opt - self._upgrade return self def __len__(self): return len(self.install) + len(self.install_opt) + len(self.upgrade) + len(self.remove) @staticmethod def _set_value(param, val): for item in val: if isinstance(item, CompsTransPkg): param.add(item) else: param.add(CompsTransPkg(item)) @property def install(self): """ Packages to be installed with strict=True - transaction will fail if they cannot be installed due to dependency errors etc. """ return self._install @install.setter def install(self, value): self._set_value(self._install, value) @property def install_opt(self): """ Packages to be installed with strict=False - they will be skipped if they cannot be installed """ return self._install_opt @install_opt.setter def install_opt(self, value): self._set_value(self._install_opt, value) @property def remove(self): return self._remove @remove.setter def remove(self, value): self._set_value(self._remove, value) @property def upgrade(self): return self._upgrade @upgrade.setter def upgrade(self, value): self._set_value(self._upgrade, value) class Solver(object): def __init__(self, history, comps, reason_fn): self.history = history self.comps = comps self._reason_fn = reason_fn @staticmethod def _mandatory_group_set(env): return {grp.id for grp in env.mandatory_groups} @staticmethod def _full_package_set(grp): return {pkg.getName() for pkg in grp.mandatory_packages + grp.default_packages + grp.optional_packages + grp.conditional_packages} @staticmethod def _pkgs_of_type(group, pkg_types, exclude=[]): def filter(pkgs): return [pkg for pkg in pkgs if pkg.name not in exclude] pkgs = set() if pkg_types & MANDATORY: pkgs.update(filter(group.mandatory_packages)) if pkg_types & DEFAULT: pkgs.update(filter(group.default_packages)) if pkg_types & OPTIONAL: pkgs.update(filter(group.optional_packages)) if pkg_types & CONDITIONAL: pkgs.update(filter(group.conditional_packages)) return pkgs def _removable_pkg(self, pkg_name): assert dnf.util.is_string_type(pkg_name) return self.history.group.is_removable_pkg(pkg_name) def _removable_grp(self, group_id): assert dnf.util.is_string_type(group_id) return self.history.env.is_removable_group(group_id) def _environment_install(self, env_id, pkg_types, exclude=None, strict=True, exclude_groups=None): assert dnf.util.is_string_type(env_id) comps_env = self.comps._environment_by_id(env_id) if not comps_env: raise CompsError(_("Environment id '%s' does not exist.") % ucd(env_id)) swdb_env = self.history.env.new(env_id, comps_env.name, comps_env.ui_name, pkg_types) self.history.env.install(swdb_env) trans = TransactionBunch() for comps_group in comps_env.mandatory_groups: if exclude_groups and comps_group.id in exclude_groups: continue trans += self._group_install(comps_group.id, pkg_types, exclude, strict) swdb_env.addGroup(comps_group.id, True, MANDATORY) for comps_group in comps_env.optional_groups: if exclude_groups and comps_group.id in exclude_groups: continue swdb_env.addGroup(comps_group.id, False, OPTIONAL) # TODO: if a group is already installed, mark it as installed? return trans def _environment_remove(self, env_id): assert dnf.util.is_string_type(env_id) is True swdb_env = self.history.env.get(env_id) if not swdb_env: raise CompsError(_("Environment id '%s' is not installed.") % env_id) self.history.env.remove(swdb_env) trans = TransactionBunch() group_ids = set([i.getGroupId() for i in swdb_env.getGroups()]) for group_id in group_ids: if not self._removable_grp(group_id): continue trans += self._group_remove(group_id) return trans def _environment_upgrade(self, env_id): assert dnf.util.is_string_type(env_id) comps_env = self.comps._environment_by_id(env_id) swdb_env = self.history.env.get(env_id) if not swdb_env: raise CompsError(_("Environment '%s' is not installed.") % env_id) if not comps_env: raise CompsError(_("Environment '%s' is not available.") % env_id) old_set = set([i.getGroupId() for i in swdb_env.getGroups()]) pkg_types = swdb_env.getPackageTypes() # create a new record for current transaction swdb_env = self.history.env.new(comps_env.id, comps_env.name, comps_env.ui_name, pkg_types) trans = TransactionBunch() for comps_group in comps_env.mandatory_groups: if comps_group.id in old_set: if self.history.group.get(comps_group.id): # upgrade installed group trans += self._group_upgrade(comps_group.id) else: # install new group trans += self._group_install(comps_group.id, pkg_types) swdb_env.addGroup(comps_group.id, True, MANDATORY) for comps_group in comps_env.optional_groups: if comps_group.id in old_set and self.history.group.get(comps_group.id): # upgrade installed group trans += self._group_upgrade(comps_group.id) swdb_env.addGroup(comps_group.id, False, OPTIONAL) # TODO: if a group is already installed, mark it as installed? self.history.env.upgrade(swdb_env) return trans def _group_install(self, group_id, pkg_types, exclude=None, strict=True, exclude_groups=None): assert dnf.util.is_string_type(group_id) comps_group = self.comps._group_by_id(group_id) if not comps_group: raise CompsError(_("Group id '%s' does not exist.") % ucd(group_id)) swdb_group = self.history.group.new(group_id, comps_group.name, comps_group.ui_name, pkg_types) for i in comps_group.packages_iter(): swdb_group.addPackage(i.name, False, Package._OPT_MAP[i.type]) self.history.group.install(swdb_group) trans = TransactionBunch() # TODO: remove exclude if strict: trans.install.update(self._pkgs_of_type(comps_group, pkg_types, exclude=[])) else: trans.install_opt.update(self._pkgs_of_type(comps_group, pkg_types, exclude=[])) return trans def _group_remove(self, group_id): assert dnf.util.is_string_type(group_id) swdb_group = self.history.group.get(group_id) if not swdb_group: raise CompsError(_("Module or Group '%s' is not installed.") % group_id) self.history.group.remove(swdb_group) trans = TransactionBunch() trans.remove = {pkg for pkg in swdb_group.getPackages() if self._removable_pkg(pkg.getName())} return trans def _group_upgrade(self, group_id): assert dnf.util.is_string_type(group_id) comps_group = self.comps._group_by_id(group_id) swdb_group = self.history.group.get(group_id) exclude = [] if not swdb_group: argument = comps_group.ui_name if comps_group else group_id raise CompsError(_("Module or Group '%s' is not installed.") % argument) if not comps_group: raise CompsError(_("Module or Group '%s' is not available.") % group_id) pkg_types = swdb_group.getPackageTypes() old_set = set([i.getName() for i in swdb_group.getPackages()]) new_set = self._pkgs_of_type(comps_group, pkg_types, exclude) # create a new record for current transaction swdb_group = self.history.group.new(group_id, comps_group.name, comps_group.ui_name, pkg_types) for i in comps_group.packages_iter(): swdb_group.addPackage(i.name, False, Package._OPT_MAP[i.type]) self.history.group.upgrade(swdb_group) trans = TransactionBunch() trans.install = {pkg for pkg in new_set if pkg.name not in old_set} trans.remove = {name for name in old_set if name not in [pkg.name for pkg in new_set]} trans.upgrade = {pkg for pkg in new_set if pkg.name in old_set} return trans def _exclude_packages_from_installed_groups(self, base): for group in self.persistor.groups: p_grp = self.persistor.group(group) if p_grp.installed: installed_pkg_names = \ set(p_grp.full_list) - set(p_grp.pkg_exclude) installed_pkgs = base.sack.query().installed().filterm(name=installed_pkg_names) for pkg in installed_pkgs: base._goal.install(pkg)	24.24 KB	2022-09-09 11:21
📁 conf	-	2026-03-21 15:47
📄 const.py # const.py # dnf constants. # # Copyright (C) 2012-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 unicode_literals CONF_FILENAME='/etc/dnf/dnf.conf' # :api CONF_AUTOMATIC_FILENAME='/etc/dnf/automatic.conf' DISTROVERPKG=('system-release(releasever)', 'system-release', 'distribution-release(releasever)', 'distribution-release', 'redhat-release', 'suse-release') DISTROVER_MAJOR_PKG='system-release(releasever_major)' DISTROVER_MINOR_PKG='system-release(releasever_minor)' GROUP_PACKAGE_TYPES = ('mandatory', 'default', 'conditional') # :api INSTALLONLYPKGS=['kernel', 'kernel-PAE', 'installonlypkg(kernel)', 'installonlypkg(kernel-module)', 'installonlypkg(vm)', 'multiversion(kernel)'] LOG='dnf.log' LOG_HAWKEY='hawkey.log' LOG_LIBREPO='dnf.librepo.log' LOG_MARKER='--- logging initialized ---' LOG_RPM='dnf.rpm.log' NAME='DNF' PERSISTDIR='/var/lib/dnf' # :api PID_FILENAME = '/var/run/dnf.pid' RUNDIR='/run' USER_RUNDIR='/run/user' SYSTEM_CACHEDIR='/var/cache/dnf' TMPDIR='/var/tmp/' # CLI verbose values greater or equal to this are considered "verbose": VERBOSE_LEVEL=6 PREFIX=NAME.lower() PROGRAM_NAME=NAME.lower() # Deprecated - no longer used, Argparser prints program name based on sys.argv PLUGINCONFPATH = '/etc/dnf/plugins' # :api PLUGINPATH = '/usr/lib/python3.9/site-packages/dnf-plugins' VERSION='4.14.0' USER_AGENT = "dnf/%s" % VERSION BUGTRACKER_COMPONENT=NAME.lower() BUGTRACKER='https://bugs.almalinux.org/'	2.39 KB	2025-09-22 11:27
📄 crypto.py # crypto.py # Keys and signatures. # # 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 print_function from __future__ import absolute_import from __future__ import unicode_literals from dnf.i18n import _ import contextlib import dnf.pycomp import dnf.util import dnf.yum.misc import io import logging import os import tempfile try: from gpg import Context from gpg import Data except ImportError: import gpgme class Context(object): def __init__(self): self.__dict__["ctx"] = gpgme.Context() def __enter__(self): return self def __exit__(self, type, value, tb): pass @property def armor(self): return self.ctx.armor @armor.setter def armor(self, value): self.ctx.armor = value def op_import(self, key_fo): if isinstance(key_fo, basestring): key_fo = io.BytesIO(key_fo) self.ctx.import_(key_fo) def op_export(self, pattern, mode, keydata): self.ctx.export(pattern, keydata) def __getattr__(self, name): return getattr(self.ctx, name) class Data(object): def __init__(self): self.__dict__["buf"] = io.BytesIO() def __enter__(self): return self def __exit__(self, type, value, tb): pass def read(self): return self.buf.getvalue() def __getattr__(self, name): return getattr(self.buf, name) GPG_HOME_ENV = 'GNUPGHOME' logger = logging.getLogger('dnf') def _extract_signing_subkey(key): return dnf.util.first(subkey for subkey in key.subkeys if subkey.can_sign) def _printable_fingerprint(fpr_hex): segments = (fpr_hex[i:i + 4] for i in range(0, len(fpr_hex), 4)) return " ".join(segments) def import_repo_keys(repo): gpgdir = repo._pubring_dir known_keys = keyids_from_pubring(gpgdir) for keyurl in repo.gpgkey: for keyinfo in retrieve(keyurl, repo): keyid = keyinfo.id_ if keyid in known_keys: logger.debug(_('repo %s: 0x%s already imported'), repo.id, keyid) continue if not repo._key_import._confirm(keyinfo): continue dnf.yum.misc.import_key_to_pubring( keyinfo.raw_key, keyinfo.short_id, gpgdir=gpgdir, make_ro_copy=False) logger.debug(_('repo %s: imported key 0x%s.'), repo.id, keyid) def keyids_from_pubring(gpgdir): if not os.path.exists(gpgdir): return [] with pubring_dir(gpgdir), Context() as ctx: keyids = [] for k in ctx.keylist(): subkey = _extract_signing_subkey(k) if subkey is not None: keyids.append(subkey.keyid) return keyids def log_key_import(keyinfo): msg = (_('Importing GPG key 0x%s:\n' ' Userid : "%s"\n' ' Fingerprint: %s\n' ' From : %s') % (keyinfo.short_id, keyinfo.userid, _printable_fingerprint(keyinfo.fingerprint), keyinfo.url.replace("file://", ""))) logger.critical("%s", msg) def log_dns_key_import(keyinfo, dns_result): log_key_import(keyinfo) if dns_result == dnf.dnssec.Validity.VALID: logger.critical(_('Verified using DNS record with DNSSEC signature.')) else: logger.critical(_('NOT verified using DNS record.')) @contextlib.contextmanager def pubring_dir(pubring_dir): orig = os.environ.get(GPG_HOME_ENV, None) os.environ[GPG_HOME_ENV] = pubring_dir try: yield finally: if orig is None: del os.environ[GPG_HOME_ENV] else: os.environ[GPG_HOME_ENV] = orig def rawkey2infos(key_fo): pb_dir = tempfile.mkdtemp() keyinfos = [] with pubring_dir(pb_dir), Context() as ctx: ctx.op_import(key_fo) for key in ctx.keylist(): subkey = _extract_signing_subkey(key) if subkey is None: continue keyinfos.append(Key(key, subkey)) ctx.armor = True for info in keyinfos: with Data() as sink: ctx.op_export(info.id_, 0, sink) sink.seek(0, os.SEEK_SET) info.raw_key = sink.read() dnf.util.rm_rf(pb_dir) return keyinfos def retrieve(keyurl, repo=None): if keyurl.startswith('http:'): logger.warning(_("retrieving repo key for %s unencrypted from %s"), repo.id, keyurl) with dnf.util._urlopen(keyurl, repo=repo) as handle: keyinfos = rawkey2infos(handle) for keyinfo in keyinfos: keyinfo.url = keyurl return keyinfos class Key(object): def __init__(self, key, subkey): self.id_ = subkey.keyid self.fingerprint = subkey.fpr self.raw_key = None self.timestamp = subkey.timestamp self.url = None self.userid = key.uids[0].uid @property def short_id(self): rj = '0' if dnf.pycomp.PY3 else b'0' return self.id_[-8:].rjust(8, rj) @property def rpm_id(self): return self.short_id.lower()	5.98 KB	2022-09-09 11:21
📁 db	-	2026-03-21 15:47
📄 dnssec.py # dnssec.py # DNS extension for automatic GPG key verification # # Copyright (C) 2012-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 print_function from __future__ import absolute_import from __future__ import unicode_literals from enum import Enum import base64 import hashlib import logging import re from dnf.i18n import _ import dnf.rpm import dnf.exceptions logger = logging.getLogger("dnf") RR_TYPE_OPENPGPKEY = 61 class DnssecError(dnf.exceptions.Error): """ Exception used in the dnssec module """ def __repr__(self): return "<DnssecError, value='{}'>"\ .format(self.value if self.value is not None else "Not specified") def email2location(email_address, tag="_openpgpkey"): # type: (str, str) -> str """ Implements RFC 7929, section 3 https://tools.ietf.org/html/rfc7929#section-3 :param email_address: :param tag: :return: """ split = email_address.rsplit("@", 1) if len(split) != 2: msg = "Email address must contain exactly one '@' sign." raise DnssecError(msg) local = split[0] domain = split[1] hash = hashlib.sha256() hash.update(local.encode('utf-8')) digest = base64.b16encode(hash.digest()[0:28])\ .decode("utf-8")\ .lower() return digest + "." + tag + "." + domain class Validity(Enum): """ Output of the verification algorithm. TODO: this type might be simplified in order to less reflect the underlying DNS layer. TODO: more specifically the variants from 3 to 5 should have more understandable names """ VALID = 1 REVOKED = 2 PROVEN_NONEXISTENCE = 3 RESULT_NOT_SECURE = 4 BOGUS_RESULT = 5 ERROR = 9 class NoKey: """ This class represents an absence of a key in the cache. It is an expression of non-existence using the Python's type system. """ pass class KeyInfo: """ Wrapper class for email and associated verification key, where both are represented in form of a string. """ def __init__(self, email=None, key=None): self.email = email self.key = key def __repr__(self): return 'KeyInfo("{}", "{}...")'.format(self.email, self.key.decode('ascii')[:6]) @staticmethod def from_rpm_key_object(userid, raw_key): # type: (str, bytes) -> KeyInfo """ Since dnf uses different format of the key than the one used in DNS RR, I need to convert the former one into the new one. """ input_email = re.search('<(.@.)>', userid) if input_email is None: raise DnssecError email = input_email.group(1) key = raw_key.decode('ascii').split('\n') start = 0 stop = 0 for i in range(0, len(key)): if key[i] == '-----BEGIN PGP PUBLIC KEY BLOCK-----': start = i if key[i] == '-----END PGP PUBLIC KEY BLOCK-----': stop = i cat_key = ''.join(key[start + 2:stop - 1]).encode('ascii') return KeyInfo(email, cat_key) class DNSSECKeyVerification: """ The main class when it comes to verification itself. It wraps Unbound context and a cache with already obtained results. """ # Mapping from email address to b64 encoded public key or NoKey in case of proven nonexistence _cache = {} # type: Dict[str, Union[str, NoKey]] @staticmethod def _cache_hit(key_union, input_key_string): # type: (Union[str, NoKey], str) -> Validity """ Compare the key in case it was found in the cache. """ if key_union == input_key_string: logger.debug("Cache hit, valid key") return Validity.VALID elif key_union is NoKey: logger.debug("Cache hit, proven non-existence") return Validity.PROVEN_NONEXISTENCE else: logger.debug("Key in cache: {}".format(key_union)) logger.debug("Input key : {}".format(input_key_string)) return Validity.REVOKED @staticmethod def _cache_miss(input_key): # type: (KeyInfo) -> Validity """ In case the key was not found in the cache, create an Unbound context and contact the DNS system """ try: import unbound except ImportError as e: msg = _("Configuration option 'gpgkey_dns_verification' requires " "python3-unbound ({})".format(e)) raise dnf.exceptions.Error(msg) ctx = unbound.ub_ctx() if ctx.set_option("verbosity:", "0") != 0: logger.debug("Unbound context: Failed to set verbosity") if ctx.set_option("qname-minimisation:", "yes") != 0: logger.debug("Unbound context: Failed to set qname minimisation") if ctx.resolvconf() != 0: logger.debug("Unbound context: Failed to read resolv.conf") if ctx.add_ta_file("/var/lib/unbound/root.key") != 0: logger.debug("Unbound context: Failed to add trust anchor file") status, result = ctx.resolve(email2location(input_key.email), RR_TYPE_OPENPGPKEY, unbound.RR_CLASS_IN) if status != 0: logger.debug("Communication with DNS servers failed") return Validity.ERROR if result.bogus: logger.debug("DNSSEC signatures are wrong ({})".format(result.why_bogus)) return Validity.BOGUS_RESULT if not result.secure: logger.debug("Result is not secured with DNSSEC") return Validity.RESULT_NOT_SECURE if result.nxdomain or (result.rcode == unbound.RCODE_NOERROR and not result.havedata): logger.debug("Non-existence of this record was proven by DNSSEC") return Validity.PROVEN_NONEXISTENCE if not result.havedata: # TODO: This is weird result, but there is no way to perform validation, so just return # an error # Should handle only SERVFAIL, REFUSED and similar rcodes logger.debug("Unknown error in DNS communication: {}".format(result.rcode_str)) return Validity.ERROR else: data = result.data.as_raw_data()[0] dns_data_b64 = base64.b64encode(data) if dns_data_b64 == input_key.key: return Validity.VALID else: # In case it is different, print the keys for further examination in debug mode logger.debug("Key from DNS: {}".format(dns_data_b64)) logger.debug("Input key : {}".format(input_key.key)) return Validity.REVOKED @staticmethod def verify(input_key): # type: (KeyInfo) -> Validity """ Public API. Use this method to verify a KeyInfo object. """ logger.debug("Running verification for key with id: {}".format(input_key.email)) key_union = DNSSECKeyVerification._cache.get(input_key.email) if key_union is not None: return DNSSECKeyVerification._cache_hit(key_union, input_key.key) else: result = DNSSECKeyVerification._cache_miss(input_key) if result == Validity.VALID: DNSSECKeyVerification._cache[input_key.email] = input_key.key elif result == Validity.PROVEN_NONEXISTENCE: DNSSECKeyVerification._cache[input_key.email] = NoKey() return result def nice_user_msg(ki, v): # type: (KeyInfo, Validity) -> str """ Inform the user about key validity in a human readable way. """ prefix = _("DNSSEC extension: Key for user ") + ki.email + " " if v == Validity.VALID: return prefix + _("is valid.") else: return prefix + _("has unknown status.") def any_msg(m): # type: (str) -> str """ Label any given message with DNSSEC extension tag """ return _("DNSSEC extension: ") + m class RpmImportedKeys: """ Wrapper around keys, that are imported in the RPM database. The keys are stored in packages with name gpg-pubkey, where the version and release is different for each of them. The key content itself is stored as an ASCII armored string in the package description, so it needs to be parsed before it can be used. """ @staticmethod def _query_db_for_gpg_keys(): # type: () -> List[KeyInfo] # TODO: base.conf.installroot ?? -----------------------\ transaction_set = dnf.rpm.transaction.TransactionWrapper() packages = transaction_set.dbMatch("name", "gpg-pubkey") return_list = [] for pkg in packages: packager = dnf.rpm.getheader(pkg, 'packager') email = re.search('<(.@.)>', packager).group(1) description = dnf.rpm.getheader(pkg, 'description') key_lines = description.split('\n')[3:-3] key_str = ''.join(key_lines) return_list += [KeyInfo(email, key_str.encode('ascii'))] return return_list @staticmethod def check_imported_keys_validity(): keys = RpmImportedKeys._query_db_for_gpg_keys() logger.info(any_msg(_("Testing already imported keys for their validity."))) for key in keys: try: result = DNSSECKeyVerification.verify(key) except DnssecError as e: # Errors in this exception should not be fatal, print it and just continue logger.warning("DNSSEC extension error (email={}): {}" .format(key.email, e.value)) continue # TODO: remove revoked keys automatically and possibly ask user to confirm if result == Validity.VALID: logger.debug(any_msg("GPG Key {} is valid".format(key.email))) pass elif result == Validity.PROVEN_NONEXISTENCE: logger.debug(any_msg("GPG Key {} does not support DNS" " verification".format(key.email))) elif result == Validity.BOGUS_RESULT: logger.info(any_msg("GPG Key {} could not be verified, because DNSSEC signatures" " are bogus. Possible causes: wrong configuration of the DNS" " server, MITM attack".format(key.email))) elif result == Validity.REVOKED: logger.info(any_msg("GPG Key {} has been revoked and should" " be removed immediately".format(key.email))) else: logger.debug(any_msg("GPG Key {} could not be tested".format(key.email)))	11.36 KB	2022-09-09 11:21
📄 drpm.py # drpm.py # Delta RPM support # # Copyright (C) 2012-2016 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 binascii import hexlify from dnf.yum.misc import unlink_f from dnf.i18n import _ import dnf.callback import dnf.logging import dnf.repo import hawkey import logging import libdnf.repo import os APPLYDELTA = '/usr/bin/applydeltarpm' logger = logging.getLogger("dnf") class DeltaPayload(dnf.repo.PackagePayload): def __init__(self, delta_info, delta, pkg, progress): super(DeltaPayload, self).__init__(pkg, progress) self.delta_info = delta_info self.delta = delta def __str__(self): return os.path.basename(self.delta.location) def _end_cb(self, cbdata, lr_status, msg): super(DeltaPayload, self)._end_cb(cbdata, lr_status, msg) if lr_status != libdnf.repo.PackageTargetCB.TransferStatus_ERROR: self.delta_info.enqueue(self) def _target_params(self): delta = self.delta ctype, csum = delta.chksum ctype = hawkey.chksum_name(ctype) chksum = hexlify(csum).decode() ctype_code = libdnf.repo.PackageTarget.checksumType(ctype) if ctype_code == libdnf.repo.PackageTarget.ChecksumType_UNKNOWN: logger.warning(_("unsupported checksum type: %s"), ctype) return { 'relative_url' : delta.location, 'checksum_type' : ctype_code, 'checksum' : chksum, 'expectedsize' : delta.downloadsize, 'base_url' : delta.baseurl, } @property def download_size(self): return self.delta.downloadsize @property def _full_size(self): return self.pkg.downloadsize def localPkg(self): location = self.delta.location return os.path.join(self.pkg.repo.pkgdir, os.path.basename(location)) class DeltaInfo(object): def __init__(self, query, progress, deltarpm_percentage=None): '''A delta lookup and rebuild context query -- installed packages to use when looking up deltas progress -- progress obj to display finished delta rebuilds ''' self.deltarpm_installed = False if os.access(APPLYDELTA, os.X_OK): self.deltarpm_installed = True try: self.deltarpm_jobs = os.sysconf('SC_NPROCESSORS_ONLN') except (TypeError, ValueError): self.deltarpm_jobs = 4 if deltarpm_percentage is None: self.deltarpm_percentage = dnf.conf.Conf().deltarpm_percentage else: self.deltarpm_percentage = deltarpm_percentage self.query = query self.progress = progress self.queue = [] self.jobs = {} self.err = {} def delta_factory(self, po, progress): '''Turn a po to Delta RPM po, if possible''' if not self.deltarpm_installed: # deltarpm is not installed return None if not po.repo.deltarpm or not self.deltarpm_percentage: # drpm disabled return None if po._is_local_pkg(): # drpm disabled for local return None if os.path.exists(po.localPkg()): # already there return None best = po._size * self.deltarpm_percentage / 100 best_delta = None for ipo in self.query.filter(name=po.name, arch=po.arch): delta = po.get_delta_from_evr(ipo.evr) if delta and delta.downloadsize < best: best = delta.downloadsize best_delta = delta if best_delta: return DeltaPayload(self, best_delta, po, progress) return None def job_done(self, pid, code): # handle a finished delta rebuild logger.log(dnf.logging.SUBDEBUG, 'drpm: %d: return code: %d, %d', pid, code >> 8, code & 0xff) pload = self.jobs.pop(pid) pkg = pload.pkg if code != 0: unlink_f(pload.pkg.localPkg()) self.err[pkg] = [_('Delta RPM rebuild failed')] elif not pload.pkg.verifyLocalPkg(): self.err[pkg] = [_('Checksum of the delta-rebuilt RPM failed')] else: os.unlink(pload.localPkg()) self.progress.end(pload, dnf.callback.STATUS_DRPM, _('done')) def start_job(self, pload): # spawn a delta rebuild job spawn_args = [APPLYDELTA, APPLYDELTA, '-a', pload.pkg.arch, pload.localPkg(), pload.pkg.localPkg()] pid = os.spawnl(os.P_NOWAIT, *spawn_args) logger.log(dnf.logging.SUBDEBUG, 'drpm: spawned %d: %s', pid, ' '.join(spawn_args[1:])) self.jobs[pid] = pload def enqueue(self, pload): # process finished jobs, start new ones while self.jobs: pid, code = os.waitpid(-1, os.WNOHANG) if not pid: break self.job_done(pid, code) self.queue.append(pload) while len(self.jobs) < self.deltarpm_jobs: self.start_job(self.queue.pop(0)) if not self.queue: break def wait(self): '''Wait until all jobs have finished''' while self.jobs: pid, code = os.wait() self.job_done(pid, code) if self.queue: self.start_job(self.queue.pop(0))	6.2 KB	2022-09-09 11:21
📄 exceptions.py # 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 Library 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. # Copyright 2004 Duke University """ Core DNF Errors. """ from __future__ import unicode_literals from dnf.i18n import ucd, _, P_ import dnf.util import libdnf import warnings class DeprecationWarning(DeprecationWarning): # :api pass class Error(Exception): # :api """Base Error. All other Errors thrown by DNF should inherit from this. :api """ def __init__(self, value=None): super(Error, self).__init__() self.value = None if value is None else ucd(value) def __str__(self): return "{}".format(self.value) def __unicode__(self): return ucd(self.__str__()) class CompsError(Error): # :api pass class ConfigError(Error): def __init__(self, value=None, raw_error=None): super(ConfigError, self).__init__(value) self.raw_error = ucd(raw_error) if raw_error is not None else None class DatabaseError(Error): pass class DepsolveError(Error): # :api pass class DownloadError(Error): # :api def __init__(self, errmap): super(DownloadError, self).__init__() self.errmap = errmap @staticmethod def errmap2str(errmap): errstrings = [] for key in errmap: for error in errmap[key]: msg = '%s: %s' % (key, error) if key else '%s' % error errstrings.append(msg) return '\n'.join(errstrings) def __str__(self): return self.errmap2str(self.errmap) class LockError(Error): pass class MarkingError(Error): # :api def __init__(self, value=None, pkg_spec=None): """Initialize the marking error instance.""" super(MarkingError, self).__init__(value) self.pkg_spec = None if pkg_spec is None else ucd(pkg_spec) def __str__(self): string = super(MarkingError, self).__str__() if self.pkg_spec: string += ': ' + self.pkg_spec return string class MarkingErrors(Error): # :api def __init__(self, no_match_group_specs=(), error_group_specs=(), no_match_pkg_specs=(), error_pkg_specs=(), module_depsolv_errors=()): """Initialize the marking error instance.""" msg = _("Problems in request:") if (no_match_pkg_specs): msg += "\n" + _("missing packages: ") + ", ".join(no_match_pkg_specs) if (error_pkg_specs): msg += "\n" + _("broken packages: ") + ", ".join(error_pkg_specs) if (no_match_group_specs): msg += "\n" + _("missing groups or modules: ") + ", ".join(no_match_group_specs) if (error_group_specs): msg += "\n" + _("broken groups or modules: ") + ", ".join(error_group_specs) if (module_depsolv_errors): msg_mod = dnf.util._format_resolve_problems(module_depsolv_errors[0]) if module_depsolv_errors[1] == \ libdnf.module.ModulePackageContainer.ModuleErrorType_ERROR_IN_DEFAULTS: msg += "\n" + "\n".join([P_('Modular dependency problem with Defaults:', 'Modular dependency problems with Defaults:', len(module_depsolv_errors)), msg_mod]) else: msg += "\n" + "\n".join([P_('Modular dependency problem:', 'Modular dependency problems:', len(module_depsolv_errors)), msg_mod]) super(MarkingErrors, self).__init__(msg) self.no_match_group_specs = no_match_group_specs self.error_group_specs = error_group_specs self.no_match_pkg_specs = no_match_pkg_specs self.error_pkg_specs = error_pkg_specs self.module_depsolv_errors = module_depsolv_errors @property def module_debsolv_errors(self): msg = "Attribute module_debsolv_errors is deprecated. Use module_depsolv_errors " \ "attribute instead." warnings.warn(msg, DeprecationWarning, stacklevel=2) return self.module_depsolv_errors class MetadataError(Error): pass class MiscError(Error): pass class PackagesNotAvailableError(MarkingError): def __init__(self, value=None, pkg_spec=None, packages=None): super(PackagesNotAvailableError, self).__init__(value, pkg_spec) self.packages = packages or [] class PackageNotFoundError(MarkingError): pass class PackagesNotInstalledError(MarkingError): def __init__(self, value=None, pkg_spec=None, packages=None): super(PackagesNotInstalledError, self).__init__(value, pkg_spec) self.packages = packages or [] class ProcessLockError(LockError): def __init__(self, value, pid): super(ProcessLockError, self).__init__(value) self.pid = pid def __reduce__(self): """Pickling support.""" return (ProcessLockError, (self.value, self.pid)) class ReadOnlyVariableError(Error): def __init__(self, value, variable_name): super(ReadOnlyVariableError, self).__init__(value) self.variable_name = variable_name class RepoError(Error): # :api pass class ThreadLockError(LockError): pass class TransactionCheckError(Error): pass	5.89 KB	2025-09-22 11:27
📄 goal.py # goal.py # Customized hawkey.Goal # # Copyright (C) 2014-2016 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 hawkey import Goal	1.08 KB	2022-09-09 11:21
📄 history.py # history.py # Interfaces to the history of transactions. # # Copyright (C) 2013-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. # """Interfaces to the history of transactions.""" from __future__ import absolute_import from __future__ import unicode_literals	1.12 KB	2022-09-09 11:21
📄 i18n.py # i18n.py # # Copyright (C) 2012-2016 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 from __future__ import unicode_literals from dnf.pycomp import unicode import dnf import locale import os import signal import sys import unicodedata """ Centralize i18n stuff here. Must be unittested. """ class UnicodeStream(object): def __init__(self, stream, encoding): self.stream = stream self.encoding = encoding def write(self, s): if not isinstance(s, str): s = (s.decode(self.encoding, 'replace') if dnf.pycomp.PY3 else s.encode(self.encoding, 'replace')) try: self.stream.write(s) except UnicodeEncodeError: s_bytes = s.encode(self.stream.encoding, 'backslashreplace') if hasattr(self.stream, 'buffer'): self.stream.buffer.write(s_bytes) else: s = s_bytes.decode(self.stream.encoding, 'ignore') self.stream.write(s) def __getattr__(self, name): return getattr(self.stream, name) def _full_ucd_support(encoding): """Return true if encoding can express any Unicode character. Even if an encoding can express all accented letters in the given language, we can't generally settle for it in DNF since sometimes we output special characters like the registered trademark symbol (U+00AE) and surprisingly many national non-unicode encodings, including e.g. ASCII and ISO-8859-2, don't contain it. """ if encoding is None: return False lower = encoding.lower() if lower.startswith('utf-') or lower.startswith('utf_'): return True return False def _guess_encoding(): """ Take the best shot at the current system's string encoding. """ encoding = locale.getpreferredencoding(False) return 'utf-8' if encoding.startswith("ANSI") else encoding def setup_locale(): try: dnf.pycomp.setlocale(locale.LC_ALL, '') except locale.Error: # default to C.UTF-8 or C locale if we got a failure. try: dnf.pycomp.setlocale(locale.LC_ALL, 'C.UTF-8') os.environ['LC_ALL'] = 'C.UTF-8' except locale.Error: dnf.pycomp.setlocale(locale.LC_ALL, 'C') os.environ['LC_ALL'] = 'C' print('Failed to set locale, defaulting to {}'.format(os.environ['LC_ALL']), file=sys.stderr) def setup_stdout(): """ Check that stdout is of suitable encoding and handle the situation if not. Returns True if stdout was of suitable encoding already and no changes were needed. """ stdout = sys.stdout if not stdout.isatty(): signal.signal(signal.SIGPIPE, signal.SIG_DFL) try: encoding = stdout.encoding except AttributeError: encoding = None if not _full_ucd_support(encoding): sys.stdout = UnicodeStream(stdout, _guess_encoding()) return False return True def ucd_input(ucstring): # :api, deprecated in 2.0.0, will be erased when python2 is abandoned """ It uses print instead of passing the prompt to raw_input. raw_input doesn't encode the passed string and the output goes into stderr """ print(ucstring, end='') return dnf.pycomp.raw_input() def ucd(obj): # :api, deprecated in 2.0.0, will be erased when python2 is abandoned """ Like the builtin unicode() but tries to use a reasonable encoding. """ if dnf.pycomp.PY3: if dnf.pycomp.is_py3bytes(obj): return str(obj, _guess_encoding(), errors='ignore') elif isinstance(obj, str): return obj return str(obj) else: if isinstance(obj, dnf.pycomp.unicode): return obj if hasattr(obj, '__unicode__'): # see the doc for the unicode() built-in. The logic here is: if obj # implements __unicode__, let it take a crack at it, but handle the # situation if it fails: try: return dnf.pycomp.unicode(obj) except UnicodeError: pass return dnf.pycomp.unicode(str(obj), _guess_encoding(), errors='ignore') # functions for formatting output according to terminal width, # They should be used instead of build-in functions to count on different # widths of Unicode characters def _exact_width_char(uchar): return 2 if unicodedata.east_asian_width(uchar) in ('W', 'F') else 1 def chop_str(msg, chop=None): """ Return the textual width of a Unicode string, chopping it to a specified value. This is what you want to use instead of %.s, as it does the "right" thing with regard to different Unicode character width Eg. "%.s" % (10, msg) <= becomes => "%s" % (chop_str(msg, 10)) """ if chop is None: return exact_width(msg), msg width = 0 chopped_msg = "" for char in msg: char_width = _exact_width_char(char) if width + char_width > chop: break chopped_msg += char width += char_width return width, chopped_msg def exact_width(msg): """ Calculates width of char at terminal screen (Asian char counts for two) """ return sum(_exact_width_char(c) for c in msg) def fill_exact_width(msg, fill, chop=None, left=True, prefix='', suffix=''): """ Expand a msg to a specified "width" or chop to same. Expansion can be left or right. This is what you want to use instead of %.s, as it does the "right" thing with regard to different Unicode character width. prefix and suffix should be used for "invisible" bytes, like highlighting. Examples: ``"%-.s" % (10, 20, msg)`` becomes ``"%s" % (fill_exact_width(msg, 10, 20))``. ``"%20.10s" % (msg)`` becomes ``"%s" % (fill_exact_width(msg, 20, 10, left=False))``. ``"%s%.10s%s" % (pre, msg, suf)`` becomes ``"%s" % (fill_exact_width(msg, 0, 10, prefix=pre, suffix=suf))``. """ width, msg = chop_str(msg, chop) if width >= fill: if prefix or suffix: msg = ''.join([prefix, msg, suffix]) else: extra = " " * (fill - width) if left: msg = ''.join([prefix, msg, suffix, extra]) else: msg = ''.join([extra, prefix, msg, suffix]) return msg def textwrap_fill(text, width=70, initial_indent='', subsequent_indent=''): """ Works like we want textwrap.wrap() to work, uses Unicode strings and doesn't screw up lists/blocks/etc. """ def _indent_at_beg(line): count = 0 byte = 'X' for byte in line: if byte != ' ': break count += 1 if byte not in ("-", "", ".", "o", '\xe2'): return count, 0 list_chr = chop_str(line[count:], 1)[1] if list_chr in ("-", "", ".", "o", "\u2022", "\u2023", "\u2218"): nxt = _indent_at_beg(line[count+len(list_chr):]) nxt = nxt[1] or nxt[0] if nxt: return count, count + 1 + nxt return count, 0 text = text.rstrip('\n') lines = text.replace('\t', ' ' * 8).split('\n') ret = [] indent = initial_indent wrap_last = False csab = 0 cspc_indent = 0 for line in lines: line = line.rstrip(' ') (lsab, lspc_indent) = (csab, cspc_indent) (csab, cspc_indent) = _indent_at_beg(line) force_nl = False # We want to stop wrapping under "certain" conditions: if wrap_last and cspc_indent: # if line starts a list or force_nl = True if wrap_last and csab == len(line): # is empty line force_nl = True # if line doesn't continue a list and is "block indented" if wrap_last and not lspc_indent: if csab >= 4 and csab != lsab: force_nl = True if force_nl: ret.append(indent.rstrip(' ')) indent = subsequent_indent wrap_last = False if csab == len(line): # empty line, remove spaces to make it easier. line = '' if wrap_last: line = line.lstrip(' ') cspc_indent = lspc_indent if exact_width(indent + line) <= width: wrap_last = False ret.append(indent + line) indent = subsequent_indent continue wrap_last = True words = line.split(' ') line = indent spcs = cspc_indent if not spcs and csab >= 4: spcs = csab for word in words: if (width < exact_width(line + word)) and \ (exact_width(line) > exact_width(subsequent_indent)): ret.append(line.rstrip(' ')) line = subsequent_indent + ' ' * spcs line += word line += ' ' indent = line.rstrip(' ') + ' ' if wrap_last: ret.append(indent.rstrip(' ')) return '\n'.join(ret) def select_short_long(width, msg_short, msg_long): """ Automatically selects the short (abbreviated) or long (full) message depending on whether we have enough screen space to display the full message or not. If a caller by mistake passes a long string as msg_short and a short string as a msg_long this function recognizes the mistake and swaps the arguments. This function is especially useful in the i18n context when you cannot predict how long are the translated messages. Limitations: 1. If msg_short is longer than width you will still get an overflow. This function does not abbreviate the string. 2. You are not obliged to provide an actually abbreviated string, it is perfectly correct to pass the same string twice if you don't want any abbreviation. However, if you provide two different strings but having the same width this function is unable to recognize which one is correct and you should assume that it is unpredictable which one is returned. Example: ``select_short_long (10, _("Repo"), _("Repository"))`` will return "Repository" in English but the results in other languages may be different. """ width_short = exact_width(msg_short) width_long = exact_width(msg_long) # If we have two strings of the same width: if width_short == width_long: return msg_long # If the short string is wider than the long string: elif width_short > width_long: return msg_short if width_short <= width else msg_long # The regular case: else: return msg_long if width_long <= width else msg_short def translation(name): # :api, deprecated in 2.0.0, will be erased when python2 is abandoned """ Easy gettext translations setup based on given domain name """ setup_locale() def ucd_wrapper(fnc): return lambda w: ucd(fnc(w)) t = dnf.pycomp.gettext.translation(name, fallback=True) return map(ucd_wrapper, dnf.pycomp.gettext_setup(t)) def pgettext(context, message): result = _(context + chr(4) + message) if "\004" in result: return message else: return result # setup translations _, P_ = translation("dnf") C_ = pgettext	12.03 KB	2022-09-09 11:21
📄 lock.py # lock.py # DNF Locking Subsystem. # # Copyright (C) 2013-2016 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.exceptions import ProcessLockError, ThreadLockError, LockError from dnf.i18n import _ from dnf.yum import misc import dnf.logging import dnf.util import errno import fcntl import hashlib import logging import os import threading import time logger = logging.getLogger("dnf") def _fit_lock_dir(dir_): if not dnf.util.am_i_root(): # for regular users the best we currently do is not to clash with # another DNF process of the same user. Since dir_ is quite definitely # not writable for us, yet significant, use its hash: hexdir = hashlib.sha1(dir_.encode('utf-8')).hexdigest() dir_ = os.path.join(misc.getCacheDir(), 'locks', hexdir) return dir_ def build_download_lock(cachedir, exit_on_lock): return ProcessLock(os.path.join(_fit_lock_dir(cachedir), 'download_lock.pid'), 'cachedir', not exit_on_lock) def build_metadata_lock(cachedir, exit_on_lock): return ProcessLock(os.path.join(_fit_lock_dir(cachedir), 'metadata_lock.pid'), 'metadata', not exit_on_lock) def build_rpmdb_lock(persistdir, exit_on_lock): return ProcessLock(os.path.join(_fit_lock_dir(persistdir), 'rpmdb_lock.pid'), 'RPMDB', not exit_on_lock) def build_log_lock(logdir, exit_on_lock): return ProcessLock(os.path.join(_fit_lock_dir(logdir), 'log_lock.pid'), 'log', not exit_on_lock) class ProcessLock(object): def __init__(self, target, description, blocking=False): self.blocking = blocking self.count = 0 self.description = description self.target = target self.thread_lock = threading.RLock() def _lock_thread(self): if not self.thread_lock.acquire(blocking=False): msg = '%s already locked by a different thread' % self.description raise ThreadLockError(msg) self.count += 1 def _try_lock(self, pid): fd = os.open(self.target, os.O_CREAT \| os.O_RDWR, 0o644) try: try: fcntl.flock(fd, fcntl.LOCK_EX \| fcntl.LOCK_NB) except OSError as e: if e.errno == errno.EWOULDBLOCK: return -1 raise old_pid = os.read(fd, 20) if len(old_pid) == 0: # empty file, write our pid os.write(fd, str(pid).encode('utf-8')) return pid try: old_pid = int(old_pid) except ValueError: msg = _('Malformed lock file found: %s.\n' 'Ensure no other dnf/yum process is running and ' 'remove the lock file manually or run ' 'systemd-tmpfiles --remove dnf.conf.') % (self.target) raise LockError(msg) if old_pid == pid: # already locked by this process return pid if not os.access('/proc/%d/stat' % old_pid, os.F_OK): # locked by a dead process, write our pid os.lseek(fd, 0, os.SEEK_SET) os.ftruncate(fd, 0) os.write(fd, str(pid).encode('utf-8')) return pid return old_pid finally: os.close(fd) def _unlock_thread(self): self.count -= 1 self.thread_lock.release() def __enter__(self): dnf.util.ensure_dir(os.path.dirname(self.target)) self._lock_thread() prev_pid = -1 my_pid = os.getpid() pid = self._try_lock(my_pid) while pid != my_pid: if pid != -1: if not self.blocking: self._unlock_thread() msg = '%s already locked by %d' % (self.description, pid) raise ProcessLockError(msg, pid) if prev_pid != pid: msg = _('Waiting for process with pid %d to finish.') % (pid) logger.info(msg) prev_pid = pid time.sleep(1) pid = self._try_lock(my_pid) def __exit__(self, *exc_args): if self.count == 1: os.unlink(self.target) self._unlock_thread()	5.2 KB	2022-09-09 11:21
📄 logging.py # logging.py # DNF Logging Subsystem. # # Copyright (C) 2013-2016 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.exceptions import dnf.const import dnf.lock import dnf.util import libdnf.repo import logging import logging.handlers import os import sys import time import warnings import gzip # :api loggers are: 'dnf', 'dnf.plugin', 'dnf.rpm' SUPERCRITICAL = 100 # do not use this for logging CRITICAL = logging.CRITICAL ERROR = logging.ERROR WARNING = logging.WARNING INFO = logging.INFO DEBUG = logging.DEBUG DDEBUG = 8 # used by anaconda (pyanaconda/payload/dnfpayload.py) SUBDEBUG = 6 TRACE = 4 ALL = 2 def only_once(func): """Method decorator turning the method into noop on second or later calls.""" def noop(_args, _kwargs): pass def swan_song(self, args, *kwargs): func(self, args, *kwargs) setattr(self, func.__name__, noop) return swan_song class _MaxLevelFilter(object): def __init__(self, max_level): self.max_level = max_level def filter(self, record): if record.levelno >= self.max_level: return 0 return 1 _VERBOSE_VAL_MAPPING = { 0 : SUPERCRITICAL, 1 : logging.INFO, 2 : logging.INFO, # the default 3 : logging.DEBUG, 4 : logging.DEBUG, 5 : logging.DEBUG, 6 : logging.DEBUG, # verbose value 7 : DDEBUG, 8 : SUBDEBUG, 9 : TRACE, 10: ALL, # more verbous librepo and hawkey } def _cfg_verbose_val2level(cfg_errval): assert 0 <= cfg_errval <= 10 return _VERBOSE_VAL_MAPPING.get(cfg_errval, TRACE) # Both the DNF default and the verbose default are WARNING. Note that ERROR has # no specific level. _ERR_VAL_MAPPING = { 0: SUPERCRITICAL, 1: logging.CRITICAL, 2: logging.ERROR } def _cfg_err_val2level(cfg_errval): assert 0 <= cfg_errval <= 10 return _ERR_VAL_MAPPING.get(cfg_errval, logging.WARNING) def compression_namer(name): return name + ".gz" CHUNK_SIZE = 128 1024 # 128 KB def compression_rotator(source, dest): with open(source, "rb") as sf: with gzip.open(dest, 'wb') as wf: while True: data = sf.read(CHUNK_SIZE) if not data: break wf.write(data) os.remove(source) class MultiprocessRotatingFileHandler(logging.handlers.RotatingFileHandler): def __init__(self, filename, mode='a', maxBytes=0, backupCount=0, encoding=None, delay=False): super(MultiprocessRotatingFileHandler, self).__init__( filename, mode, maxBytes, backupCount, encoding, delay) self.rotate_lock = dnf.lock.build_log_lock("/var/log/", True) def emit(self, record): while True: try: if self.shouldRollover(record): with self.rotate_lock: # Do rollover while preserving the mode of the new log file mode = os.stat(self.baseFilename).st_mode self.doRollover() os.chmod(self.baseFilename, mode) logging.FileHandler.emit(self, record) return except (dnf.exceptions.ProcessLockError, dnf.exceptions.ThreadLockError): time.sleep(0.01) except Exception: self.handleError(record) return def _create_filehandler(logfile, log_size, log_rotate, log_compress): if not os.path.exists(logfile): dnf.util.ensure_dir(os.path.dirname(logfile)) dnf.util.touch(logfile) handler = MultiprocessRotatingFileHandler(logfile, maxBytes=log_size, backupCount=log_rotate) formatter = logging.Formatter("%(asctime)s %(levelname)s %(message)s", "%Y-%m-%dT%H:%M:%S%z") formatter.converter = time.localtime handler.setFormatter(formatter) if log_compress: handler.rotator = compression_rotator handler.namer = compression_namer return handler def _paint_mark(logger): logger.log(INFO, dnf.const.LOG_MARKER) class Logging(object): def __init__(self): self.stdout_handler = self.stderr_handler = None logging.addLevelName(DDEBUG, "DDEBUG") logging.addLevelName(SUBDEBUG, "SUBDEBUG") logging.addLevelName(TRACE, "TRACE") logging.addLevelName(ALL, "ALL") logging.captureWarnings(True) logging.raiseExceptions = False @only_once def _presetup(self): logger_dnf = logging.getLogger("dnf") logger_dnf.setLevel(TRACE) # setup stdout stdout = logging.StreamHandler(sys.stdout) stdout.setLevel(INFO) stdout.addFilter(_MaxLevelFilter(logging.WARNING)) logger_dnf.addHandler(stdout) self.stdout_handler = stdout # setup stderr stderr = logging.StreamHandler(sys.stderr) stderr.setLevel(WARNING) logger_dnf.addHandler(stderr) self.stderr_handler = stderr @only_once def _setup_file_loggers(self, logfile_level, logdir, log_size, log_rotate, log_compress): logger_dnf = logging.getLogger("dnf") logger_dnf.setLevel(TRACE) # setup file logger logfile = os.path.join(logdir, dnf.const.LOG) handler = _create_filehandler(logfile, log_size, log_rotate, log_compress) handler.setLevel(logfile_level) logger_dnf.addHandler(handler) # setup Python warnings logger_warnings = logging.getLogger("py.warnings") logger_warnings.addHandler(handler) logger_librepo = logging.getLogger("librepo") logger_librepo.setLevel(TRACE) logfile = os.path.join(logdir, dnf.const.LOG_LIBREPO) handler = _create_filehandler(logfile, log_size, log_rotate, log_compress) logger_librepo.addHandler(handler) libdnf.repo.LibrepoLog.addHandler(logfile, logfile_level <= ALL) # setup RPM callbacks logger logger_rpm = logging.getLogger("dnf.rpm") logger_rpm.propagate = False logger_rpm.setLevel(SUBDEBUG) logfile = os.path.join(logdir, dnf.const.LOG_RPM) handler = _create_filehandler(logfile, log_size, log_rotate, log_compress) logger_rpm.addHandler(handler) @only_once def _setup(self, verbose_level, error_level, logfile_level, logdir, log_size, log_rotate, log_compress): self._presetup() self._setup_file_loggers(logfile_level, logdir, log_size, log_rotate, log_compress) logger_warnings = logging.getLogger("py.warnings") logger_warnings.addHandler(self.stderr_handler) # setup RPM callbacks logger logger_rpm = logging.getLogger("dnf.rpm") logger_rpm.addHandler(self.stdout_handler) logger_rpm.addHandler(self.stderr_handler) logger_dnf = logging.getLogger("dnf") # temporarily turn off stdout/stderr handlers: self.stdout_handler.setLevel(WARNING) self.stderr_handler.setLevel(WARNING) _paint_mark(logger_dnf) _paint_mark(logger_rpm) # bring std handlers to the preferred level self.stdout_handler.setLevel(verbose_level) self.stderr_handler.setLevel(error_level) def _setup_from_dnf_conf(self, conf, file_loggers_only=False): verbose_level_r = _cfg_verbose_val2level(conf.debuglevel) error_level_r = _cfg_err_val2level(conf.errorlevel) logfile_level_r = _cfg_verbose_val2level(conf.logfilelevel) logdir = conf.logdir log_size = conf.log_size log_rotate = conf.log_rotate log_compress = conf.log_compress if file_loggers_only: return self._setup_file_loggers(logfile_level_r, logdir, log_size, log_rotate, log_compress) else: return self._setup( verbose_level_r, error_level_r, logfile_level_r, logdir, log_size, log_rotate, log_compress) class Timer(object): def __init__(self, what): self.what = what self.start = time.time() def __call__(self): diff = time.time() - self.start msg = 'timer: %s: %d ms' % (self.what, diff * 1000) logging.getLogger("dnf").log(DDEBUG, msg) _LIBDNF_TO_DNF_LOGLEVEL_MAPPING = { libdnf.utils.Logger.Level_CRITICAL: CRITICAL, libdnf.utils.Logger.Level_ERROR: ERROR, libdnf.utils.Logger.Level_WARNING: WARNING, libdnf.utils.Logger.Level_NOTICE: INFO, libdnf.utils.Logger.Level_INFO: INFO, libdnf.utils.Logger.Level_DEBUG: DEBUG, libdnf.utils.Logger.Level_TRACE: TRACE } class LibdnfLoggerCB(libdnf.utils.Logger): def __init__(self): super(LibdnfLoggerCB, self).__init__() self._dnf_logger = logging.getLogger("dnf") self._librepo_logger = logging.getLogger("librepo") def write(self, source, *args): """Log message. source -- integer, defines origin (libdnf, librepo, ...) of message, 0 - unknown """ if len(args) == 2: level, message = args elif len(args) == 4: time, pid, level, message = args if source == libdnf.utils.Logger.LOG_SOURCE_LIBREPO: self._librepo_logger.log(_LIBDNF_TO_DNF_LOGLEVEL_MAPPING[level], message) else: self._dnf_logger.log(_LIBDNF_TO_DNF_LOGLEVEL_MAPPING[level], message) libdnfLoggerCB = LibdnfLoggerCB() libdnf.utils.Log.setLogger(libdnfLoggerCB)	10.11 KB	2022-09-09 11:21
📄 match_counter.py # match_counter.py # Implements class MatchCounter. # # Copyright (C) 2012-2016 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 print_function from __future__ import unicode_literals from functools import reduce WEIGHTS = { 'name' : 7, 'summary' : 4, 'description' : 2, 'url' : 1, } def _canonize_string_set(sset, length): """ Ordered sset with empty strings prepended. """ current = len(sset) l = [''] * (length - current) + sorted(sset) return l class MatchCounter(dict): """Map packages to which of their attributes matched in a search against what values. The mapping is: ``package -> [(key, needle), ... ]``. """ @staticmethod def _eval_weights(pkg, matches): # how much is each match worth and return their sum: def weight(match): key = match[0] needle = match[1] haystack = getattr(pkg, key) if key == "name" and haystack == needle: # if package matches exactly by name, increase weight return 2 * WEIGHTS[key] return WEIGHTS[key] return sum(map(weight, matches)) def _key_func(self): """Get the key function used for sorting matches. It is not enough to only look at the matches and order them by the sum of their weighted hits. In case this number is the same we have to ensure that the same matched needles are next to each other in the result. Returned function is: pkg -> (weights_sum, canonized_needles_set, -distance) """ def get_key(pkg): return ( # use negative value to make sure packages with the highest weight come first - self._eval_weights(pkg, self[pkg]), # then order packages alphabetically pkg.name, ) return get_key def _max_needles(self): """Return the max count of needles of all packages.""" if self: return max(len(self.matched_needles(pkg)) for pkg in self) return 0 def add(self, pkg, key, needle): self.setdefault(pkg, []).append((key, needle)) def dump(self): for pkg in self: print('%s\t%s' % (pkg, self[pkg])) def matched_haystacks(self, pkg): return set(getattr(pkg, m[0]) for m in self[pkg]) def matched_keys(self, pkg): # return keys in the same order they appear in the list result = [] for i in self[pkg]: if i[0] in result: continue result.append(i[0]) return result def matched_needles(self, pkg): return set(m[1] for m in self[pkg]) def sorted(self, reverse=False, limit_to=None): keys = limit_to if limit_to else self.keys() return sorted(keys, key=self._key_func()) def total(self): return reduce(lambda total, pkg: total + len(self[pkg]), self, 0)	3.83 KB	2022-09-09 11:21
📁 module	-	2026-03-21 15:47
📄 package.py # package.py # Module defining the dnf.Package class. # # Copyright (C) 2012-2016 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. # """ Contains the dnf.Package class. """ from __future__ import absolute_import from __future__ import unicode_literals from dnf.i18n import _ import binascii import dnf.exceptions import dnf.rpm import dnf.yum.misc import hawkey import libdnf.error import libdnf.utils import logging import os import rpm logger = logging.getLogger("dnf") class Package(hawkey.Package): """ Represents a package. #:api """ DEBUGINFO_SUFFIX = "-debuginfo" # :api DEBUGSOURCE_SUFFIX = "-debugsource" # :api def __init__(self, initobject, base): super(Package, self).__init__(initobject) self.base = base self._priv_chksum = None self._repo = None self._priv_size = None @property def _chksum(self): if self._priv_chksum: return self._priv_chksum if self._from_cmdline: chksum_type = dnf.yum.misc.get_default_chksum_type() try: chksum_val = libdnf.utils.checksum_value(chksum_type, self.location) except libdnf.error.Error as e: raise dnf.exceptions.MiscError(str(e)) return (hawkey.chksum_type(chksum_type), binascii.unhexlify(chksum_val)) return super(Package, self).chksum @_chksum.setter def _chksum(self, val): self._priv_chksum = val @property def _from_cmdline(self): return self.reponame == hawkey.CMDLINE_REPO_NAME @property def _from_system(self): return self.reponame == hawkey.SYSTEM_REPO_NAME @property def _from_repo(self): """ For installed packages returns id of repository from which the package was installed prefixed with '@' (if such information is available in the history database). Otherwise returns id of repository the package belongs to (@System for installed packages of unknown origin) """ pkgrepo = None if self._from_system: pkgrepo = self.base.history.repo(self) if pkgrepo: return '@' + pkgrepo return self.reponame @property def from_repo(self): # :api if self._from_system: return self.base.history.repo(self) return "" @property def _header(self): """ Returns the header of a locally present rpm package file. As opposed to self.get_header(), which retrieves the header of an installed package from rpmdb. """ return dnf.rpm._header(self.localPkg()) @property def _size(self): if self._priv_size: return self._priv_size return super(Package, self).size @_size.setter def _size(self, val): self._priv_size = val @property def _pkgid(self): if self.hdr_chksum is None: return None (_, chksum) = self.hdr_chksum return binascii.hexlify(chksum) @property def source_name(self): # :api """ returns name of source package e.g. krb5-libs -> krb5 """ if self.sourcerpm is not None: # trim suffix first srcname = dnf.util.rtrim(self.sourcerpm, ".src.rpm") # sourcerpm should be in form of name-version-release now, so we # will strip the two rightmost parts separated by dash. # Using rtrim with version and release of self is not sufficient # because the package can have different version to the source # package. srcname = srcname.rsplit('-', 2)[0] else: srcname = None return srcname @property def debug_name(self): # :api """ Returns name of the debuginfo package for this package. If this package is a debuginfo package, returns its name. If this package is a debugsource package, returns the debuginfo package for the base package. e.g. kernel-PAE -> kernel-PAE-debuginfo """ if self.name.endswith(self.DEBUGINFO_SUFFIX): return self.name name = self.name if self.name.endswith(self.DEBUGSOURCE_SUFFIX): name = name[:-len(self.DEBUGSOURCE_SUFFIX)] return name + self.DEBUGINFO_SUFFIX @property def debugsource_name(self): # :api """ Returns name of the debugsource package for this package. e.g. krb5-libs -> krb5-debugsource """ # assuming self.source_name is None only for a source package src_name = self.source_name if self.source_name is not None else self.name return src_name + self.DEBUGSOURCE_SUFFIX def get_header(self): """ Returns the rpm header of the package if it is installed. If not installed, returns None. The header is not cached, it is retrieved from rpmdb on every call. In case of a failure (e.g. when the rpmdb changes between loading the data and calling this method), raises an instance of PackageNotFoundError. """ if not self._from_system: return None try: # RPMDBI_PACKAGES stands for the header of the package return next(self.base._ts.dbMatch(rpm.RPMDBI_PACKAGES, self.rpmdbid)) except StopIteration: raise dnf.exceptions.PackageNotFoundError("Package not found when attempting to retrieve header", str(self)) @property def source_debug_name(self): # :api """ returns name of debuginfo package for source package of given package e.g. krb5-libs -> krb5-debuginfo """ # assuming self.source_name is None only for a source package src_name = self.source_name if self.source_name is not None else self.name return src_name + self.DEBUGINFO_SUFFIX @property # yum compatibility attribute def idx(self): """ Always type it to int, rpm bindings expect it like that. """ return int(self.rpmdbid) @property # yum compatibility attribute def repoid(self): return self.reponame @property # yum compatibility attribute def pkgtup(self): return (self.name, self.arch, str(self.e), self.v, self.r) @property # yum compatibility attribute def repo(self): if self._repo: return self._repo return self.base.repos[self.reponame] @repo.setter def repo(self, val): self._repo = val @property def reason(self): if self.repoid != hawkey.SYSTEM_REPO_NAME: return None return self.base.history.rpm.get_reason_name(self) @property # yum compatibility attribute def relativepath(self): return self.location @property # yum compatibility attribute def a(self): return self.arch @property # yum compatibility attribute def e(self): return self.epoch @property # yum compatibility attribute def v(self): return self.version @property # yum compatibility attribute def r(self): return self.release @property # yum compatibility attribute def ui_from_repo(self): return self.reponame # yum compatibility method def evr_eq(self, pkg): return self.evr_cmp(pkg) == 0 # yum compatibility method def evr_gt(self, pkg): return self.evr_cmp(pkg) > 0 # yum compatibility method def evr_lt(self, pkg): return self.evr_cmp(pkg) < 0 # yum compatibility method def getDiscNum(self): return self.medianr # yum compatibility method def localPkg(self): """ Package's location in the filesystem. For packages in remote repo returns where the package will be/has been downloaded. """ if self._from_cmdline: return self.location loc = self.location if self.repo._repo.isLocal() and self.baseurl and self.baseurl.startswith('file://'): return os.path.join(self.get_local_baseurl(), loc.lstrip("/")) if not self._is_local_pkg(): loc = os.path.basename(loc) return os.path.join(self.pkgdir, loc.lstrip("/")) def remote_location(self, schemes=('http', 'ftp', 'file', 'https')): # :api """ The location from where the package can be downloaded from. Returns None for installed and commandline packages. :param schemes: list of allowed protocols. Default is ('http', 'ftp', 'file', 'https') :return: location (string) or None """ if self._from_system or self._from_cmdline: return None if self.baseurl: return os.path.join(self.baseurl, self.location.lstrip("/")) return self.repo.remote_location(self.location, schemes) def _is_local_pkg(self): if self._from_system: return True if '://' in self.location and not self.location.startswith('file://'): # the package has a remote URL as its location return False return self._from_cmdline or \ (self.repo._repo.isLocal() and (not self.baseurl or self.baseurl.startswith('file://'))) @property def pkgdir(self): if (self.repo._repo.isLocal() and not self._is_local_pkg()): return self.repo.cache_pkgdir() else: return self.repo.pkgdir # yum compatibility method def returnIdSum(self): """ Return the chksum type and chksum string how the legacy yum expects it. """ if self._chksum is None: return (None, None) (chksum_type, chksum) = self._chksum return (hawkey.chksum_name(chksum_type), binascii.hexlify(chksum).decode()) # yum compatibility method def verifyLocalPkg(self): if self._from_system: raise ValueError("Can not verify an installed package.") if self._from_cmdline: return True # local package always verifies against itself (chksum_type, chksum) = self.returnIdSum() try: return libdnf.utils.checksum_check(chksum_type, self.localPkg(), chksum) except libdnf.error.Error as e: raise dnf.exceptions.MiscError(str(e))	11.04 KB	2025-09-22 11:27
📄 persistor.py # persistor.py # Persistence data container. # # Copyright (C) 2013-2016 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 current implementation is storing to files in persistdir. Do not depend on # specific files existing, instead use the persistor API. The underlying # implementation can change, e.g. for one general file with a serialized dict of # data etc. from __future__ import absolute_import from __future__ import unicode_literals from dnf.i18n import _ import dnf.util import errno import fnmatch import json import logging import os import re logger = logging.getLogger("dnf") class JSONDB(object): def _check_json_db(self, json_path): if not os.path.isfile(json_path): # initialize new db dnf.util.ensure_dir(os.path.dirname(json_path)) self._write_json_db(json_path, []) def _get_json_db(self, json_path, default=[]): with open(json_path, 'r') as f: content = f.read() if content == "": # empty file is invalid json format logger.warning(_("%s is empty file"), json_path) self._write_json_db(json_path, default) else: try: default = json.loads(content) except ValueError as e: logger.warning(e) return default @staticmethod def _write_json_db(json_path, content): with open(json_path, 'w') as f: json.dump(content, f) class RepoPersistor(JSONDB): """Persistent data kept for repositories. Is arch/releasever specific and stores to cachedir. """ def __init__(self, cachedir): self.cachedir = cachedir self.db_path = os.path.join(self.cachedir, "expired_repos.json") self.expired_to_add = set() self.reset_last_makecache = False @property def _last_makecache_path(self): return os.path.join(self.cachedir, "last_makecache") def get_expired_repos(self): try: self._check_json_db(self.db_path) return set(self._get_json_db(self.db_path)) except OSError as e: logger.warning(_("Failed to load expired repos cache: %s"), e) return None def save(self): try: self._check_json_db(self.db_path) self._write_json_db(self.db_path, list(self.expired_to_add)) except OSError as e: logger.warning(_("Failed to store expired repos cache: %s"), e) return False if self.reset_last_makecache: try: dnf.util.touch(self._last_makecache_path) return True except IOError: logger.warning(_("Failed storing last makecache time.")) return False def since_last_makecache(self): try: return int(dnf.util.file_age(self._last_makecache_path)) except OSError: logger.warning(_("Failed determining last makecache time.")) return None class TempfilePersistor(JSONDB): def __init__(self, cachedir): self.db_path = os.path.join(cachedir, "tempfiles.json") self.tempfiles_to_add = set() self._empty = False def get_saved_tempfiles(self): self._check_json_db(self.db_path) return self._get_json_db(self.db_path) def save(self): if not self._empty and not self.tempfiles_to_add: return self._check_json_db(self.db_path) if self._empty: self._write_json_db(self.db_path, []) return if self.tempfiles_to_add: data = set(self._get_json_db(self.db_path)) data.update(self.tempfiles_to_add) self._write_json_db(self.db_path, list(data)) def empty(self): self._empty = True	4.58 KB	2022-09-09 11:21
📄 plugin.py # plugin.py # The interface for building DNF plugins. # # Copyright (C) 2012-2016 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 print_function from __future__ import unicode_literals import fnmatch import glob import importlib import inspect import logging import operator import os import sys import traceback import libdnf import dnf.logging import dnf.pycomp import dnf.util from dnf.i18n import _ logger = logging.getLogger('dnf') DYNAMIC_PACKAGE = 'dnf.plugin.dynamic' class Plugin(object): """The base class custom plugins must derive from. #:api""" name = '<invalid>' config_name = None @classmethod def read_config(cls, conf): # :api parser = libdnf.conf.ConfigParser() name = cls.config_name if cls.config_name else cls.name files = ['%s/%s.conf' % (path, name) for path in conf.pluginconfpath] for file in files: if os.path.isfile(file): try: parser.read(file) except Exception as e: raise dnf.exceptions.ConfigError(_("Parsing file failed: %s") % str(e)) return parser def __init__(self, base, cli): # :api self.base = base self.cli = cli def pre_config(self): # :api pass def config(self): # :api pass def resolved(self): # :api pass def sack(self): # :api pass def pre_transaction(self): # :api pass def transaction(self): # :api pass class Plugins(object): def __init__(self): self.plugin_cls = [] self.plugins = [] def __del__(self): self._unload() def _caller(self, method): for plugin in self.plugins: try: getattr(plugin, method)() except dnf.exceptions.Error: raise except Exception: exc_type, exc_value, exc_traceback = sys.exc_info() except_list = traceback.format_exception(exc_type, exc_value, exc_traceback) logger.critical(''.join(except_list)) def _check_enabled(self, conf, enable_plugins): """Checks whether plugins are enabled or disabled in configuration files and removes disabled plugins from list""" for plug_cls in self.plugin_cls[:]: name = plug_cls.name if any(fnmatch.fnmatch(name, pattern) for pattern in enable_plugins): continue parser = plug_cls.read_config(conf) # has it enabled = False? disabled = (parser.has_section('main') and parser.has_option('main', 'enabled') and not parser.getboolean('main', 'enabled')) if disabled: self.plugin_cls.remove(plug_cls) def _load(self, conf, skips, enable_plugins): """Dynamically load relevant plugin modules.""" if DYNAMIC_PACKAGE in sys.modules: raise RuntimeError("load_plugins() called twice") sys.modules[DYNAMIC_PACKAGE] = package = dnf.pycomp.ModuleType(DYNAMIC_PACKAGE) package.__path__ = [] files = _get_plugins_files(conf.pluginpath, skips, enable_plugins) _import_modules(package, files) self.plugin_cls = _plugin_classes()[:] self._check_enabled(conf, enable_plugins) if len(self.plugin_cls) > 0: names = sorted(plugin.name for plugin in self.plugin_cls) logger.debug(_('Loaded plugins: %s'), ', '.join(names)) def _run_pre_config(self): self._caller('pre_config') def _run_config(self): self._caller('config') def _run_init(self, base, cli=None): for p_cls in self.plugin_cls: plugin = p_cls(base, cli) self.plugins.append(plugin) def run_sack(self): self._caller('sack') def run_resolved(self): self._caller('resolved') def run_pre_transaction(self): self._caller('pre_transaction') def run_transaction(self): self._caller('transaction') def _unload(self): if DYNAMIC_PACKAGE in sys.modules: logger.log(dnf.logging.DDEBUG, 'Plugins were unloaded.') del sys.modules[DYNAMIC_PACKAGE] def unload_removed_plugins(self, transaction): """ Unload plugins that were removed in the `transaction`. """ if not transaction.remove_set: return # gather all installed plugins and their files plugins = dict() for plugin in self.plugins: plugins[inspect.getfile(plugin.__class__)] = plugin # gather all removed files that are plugin files plugin_files = set(plugins.keys()) erased_plugin_files = set() for pkg in transaction.remove_set: erased_plugin_files.update(plugin_files.intersection(pkg.files)) if not erased_plugin_files: return # check whether removed plugin file is added at the same time (upgrade of a plugin) for pkg in transaction.install_set: erased_plugin_files.difference_update(pkg.files) # unload plugins that were removed in transaction for plugin_file in erased_plugin_files: self.plugins.remove(plugins[plugin_file]) def _plugin_classes(): return Plugin.__subclasses__() def _import_modules(package, py_files): for fn in py_files: path, module = os.path.split(fn) package.__path__.append(path) (module, ext) = os.path.splitext(module) name = '%s.%s' % (package.__name__, module) try: module = importlib.import_module(name) except Exception as e: logger.error(_('Failed loading plugin "%s": %s'), module, e) logger.log(dnf.logging.SUBDEBUG, '', exc_info=True) def _get_plugins_files(paths, disable_plugins, enable_plugins): plugins = [] disable_plugins = set(disable_plugins) enable_plugins = set(enable_plugins) pattern_enable_found = set() pattern_disable_found = set() for p in paths: for fn in glob.glob('%s/*.py' % p): (plugin_name, dummy) = os.path.splitext(os.path.basename(fn)) matched = True enable_pattern_tested = False for pattern_skip in disable_plugins: if _plugin_name_matches_pattern(plugin_name, pattern_skip): pattern_disable_found.add(pattern_skip) matched = False for pattern_enable in enable_plugins: if _plugin_name_matches_pattern(plugin_name, pattern_enable): matched = True pattern_enable_found.add(pattern_enable) enable_pattern_tested = True if not enable_pattern_tested: for pattern_enable in enable_plugins: if _plugin_name_matches_pattern(plugin_name, pattern_enable): pattern_enable_found.add(pattern_enable) if matched: plugins.append(fn) enable_not_found = enable_plugins.difference(pattern_enable_found) if enable_not_found: logger.warning(_("No matches found for the following enable plugin patterns: {}").format( ", ".join(sorted(enable_not_found)))) disable_not_found = disable_plugins.difference(pattern_disable_found) if disable_not_found: logger.warning(_("No matches found for the following disable plugin patterns: {}").format( ", ".join(sorted(disable_not_found)))) return plugins def _plugin_name_matches_pattern(plugin_name, pattern): """ Checks plugin name matches the pattern. The alternative plugin name using dashes instead of underscores is tried in case of original name is not matched. (see https://bugzilla.redhat.com/show_bug.cgi?id=1980712) """ try_names = set((plugin_name, plugin_name.replace('_', '-'))) return any(fnmatch.fnmatch(name, pattern) for name in try_names) def register_command(command_class): # :api """A class decorator for automatic command registration.""" def __init__(self, base, cli): if cli: cli.register_command(command_class) plugin_class = type(str(command_class.__name__ + 'Plugin'), (dnf.Plugin,), {"__init__": __init__, "name": command_class.aliases[0]}) command_class._plugin = plugin_class return command_class	9.33 KB	2025-09-22 11:27
📄 pycomp.py # pycomp.py # Python 2 and Python 3 compatibility module # # Copyright (C) 2013-2016 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 gettext import NullTranslations from sys import version_info import base64 import email.mime.text import gettext import itertools import locale import sys import types PY3 = version_info.major >= 3 if PY3: from io import StringIO from configparser import ConfigParser import queue import urllib.parse import shlex # functions renamed in py3 Queue = queue.Queue basestring = unicode = str filterfalse = itertools.filterfalse long = int NullTranslations.ugettext = NullTranslations.gettext NullTranslations.ungettext = NullTranslations.ngettext xrange = range raw_input = input base64_decodebytes = base64.decodebytes urlparse = urllib.parse urllib_quote = urlparse.quote shlex_quote = shlex.quote sys_maxsize = sys.maxsize def gettext_setup(t): _ = t.gettext P_ = t.ngettext return (_, P_) # string helpers def is_py2str_py3bytes(o): return isinstance(o, bytes) def is_py3bytes(o): return isinstance(o, bytes) # functions that don't take unicode arguments in py2 ModuleType = lambda m: types.ModuleType(m) format = locale.format_string def setlocale(category, loc=None): locale.setlocale(category, loc) def write_to_file(f, content): f.write(content) def email_mime(body): return email.mime.text.MIMEText(body) else: # functions renamed in py3 from __builtin__ import unicode, basestring, long, xrange, raw_input from StringIO import StringIO from ConfigParser import ConfigParser import Queue import urllib import urlparse import pipes Queue = Queue.Queue filterfalse = itertools.ifilterfalse base64_decodebytes = base64.decodestring urllib_quote = urllib.quote shlex_quote = pipes.quote sys_maxsize = sys.maxint def gettext_setup(t): _ = t.ugettext P_ = t.ungettext return (_, P_) # string helpers def is_py2str_py3bytes(o): return isinstance(o, str) def is_py3bytes(o): return False # functions that don't take unicode arguments in py2 ModuleType = lambda m: types.ModuleType(m.encode('utf-8')) def format(percent, args, kwargs): return locale.format(percent.encode('utf-8'), args, **kwargs) def setlocale(category, loc=None): locale.setlocale(category, loc.encode('utf-8')) def write_to_file(f, content): f.write(content.encode('utf-8')) def email_mime(body): return email.mime.text.MIMEText(body.encode('utf-8'))	3.52 KB	2022-09-09 11:21
📄 query.py # query.py # Implements Query. # # Copyright (C) 2012-2016 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 from hawkey import Query from dnf.i18n import ucd from dnf.pycomp import basestring def _by_provides(sack, patterns, ignore_case=False, get_query=False): if isinstance(patterns, basestring): patterns = [patterns] q = sack.query() flags = [] if ignore_case: flags.append(hawkey.ICASE) q.filterm(*flags, provides__glob=patterns) if get_query: return q return q.run() def _per_nevra_dict(pkg_list): nevra_dic = {} for pkg in pkg_list: nevra_dic.setdefault(ucd(pkg), []).append(pkg) return nevra_dic	1.62 KB	2025-09-22 11:27
📄 repo.py # repo.py # DNF Repository objects. # # Copyright (C) 2013-2016 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, _ import dnf.callback import dnf.conf import dnf.conf.substitutions import dnf.const import dnf.crypto import dnf.exceptions import dnf.logging import dnf.pycomp import dnf.util import dnf.yum.misc import libdnf.error import libdnf.repo import functools import hashlib import hawkey import logging import operator import os import re import shutil import string import sys import time import traceback import urllib _PACKAGES_RELATIVE_DIR = "packages" _MIRRORLIST_FILENAME = "mirrorlist" # Chars allowed in a repo ID _REPOID_CHARS = string.ascii_letters + string.digits + '-_.:' # Regex pattern that matches a repo cachedir and captures the repo ID _CACHEDIR_RE = r'(?P<repoid>[%s]+)\-[%s]{16}' % (re.escape(_REPOID_CHARS), string.hexdigits) # Regex patterns matching any filename that is repo-specific cache data of a # particular type. The filename is expected to not contain the base cachedir # path components. CACHE_FILES = { 'metadata': r'^%s\/.((xml\|yaml)(\.gz\|\.xz\|\.bz2\|.zck)?\|asc\|cachecookie\|%s)$' % (_CACHEDIR_RE, _MIRRORLIST_FILENAME), 'packages': r'^%s\/%s\/.+rpm$' % (_CACHEDIR_RE, _PACKAGES_RELATIVE_DIR), 'dbcache': r'^.+(solv\|solvx)$', } logger = logging.getLogger("dnf") def repo_id_invalid(repo_id): # :api """Return index of an invalid character in the repo ID (if present).""" first_invalid = libdnf.repo.Repo.verifyId(repo_id) return None if first_invalid < 0 else first_invalid def _pkg2payload(pkg, progress, factories): for fn in factories: pload = fn(pkg, progress) if pload is not None: return pload raise ValueError(_('no matching payload factory for %s') % pkg) def _download_payloads(payloads, drpm, fail_fast=True): # download packages def _download_sort_key(payload): return not hasattr(payload, 'delta') drpm.err.clear() targets = [pload._librepo_target() for pload in sorted(payloads, key=_download_sort_key)] errs = _DownloadErrors() try: libdnf.repo.PackageTarget.downloadPackages(libdnf.repo.VectorPPackageTarget(targets), fail_fast) except RuntimeError as e: errs._fatal = str(e) drpm.wait() # process downloading errors errs._recoverable = drpm.err.copy() for tgt in targets: err = tgt.getErr() if err is None or err.startswith('Not finished'): continue callbacks = tgt.getCallbacks() payload = callbacks.package_pload pkg = payload.pkg if err == 'Already downloaded': errs._skipped.add(pkg) continue pkg.repo._repo.expire() errs._pkg_irrecoverable[pkg] = [err] return errs def _update_saving(saving, payloads, errs): real, full = saving for pload in payloads: pkg = pload.pkg if pkg in errs: real += pload.download_size continue real += pload.download_size full += pload._full_size return real, full class _DownloadErrors(object): def __init__(self): self._pkg_irrecoverable = {} self._val_recoverable = {} self._fatal = None self._skipped = set() def _irrecoverable(self): if self._pkg_irrecoverable: return self._pkg_irrecoverable if self._fatal: return {'': [self._fatal]} return {} @property def _recoverable(self): return self._val_recoverable @_recoverable.setter def _recoverable(self, new_dct): self._val_recoverable = new_dct def _bandwidth_used(self, pload): if pload.pkg in self._skipped: return 0 return pload.download_size class _DetailedLibrepoError(Exception): def __init__(self, librepo_err, source_url): Exception.__init__(self) self.librepo_code = librepo_err.args[0] self.librepo_msg = librepo_err.args[1] self.source_url = source_url class _NullKeyImport(dnf.callback.KeyImport): def _confirm(self, id, userid, fingerprint, url, timestamp): return True class Metadata(object): def __init__(self, repo): self._repo = repo @property def fresh(self): # :api return self._repo.fresh() class PackageTargetCallbacks(libdnf.repo.PackageTargetCB): def __init__(self, package_pload): super(PackageTargetCallbacks, self).__init__() self.package_pload = package_pload def end(self, status, msg): self.package_pload._end_cb(None, status, msg) return 0 def progress(self, totalToDownload, downloaded): self.package_pload._progress_cb(None, totalToDownload, downloaded) return 0 def mirrorFailure(self, msg, url): self.package_pload._mirrorfail_cb(None, msg, url) return 0 class PackagePayload(dnf.callback.Payload): def __init__(self, pkg, progress): super(PackagePayload, self).__init__(progress) self.callbacks = PackageTargetCallbacks(self) self.pkg = pkg def _end_cb(self, cbdata, lr_status, msg): """End callback to librepo operation.""" status = dnf.callback.STATUS_FAILED if msg is None: status = dnf.callback.STATUS_OK elif msg.startswith('Not finished'): return elif lr_status == libdnf.repo.PackageTargetCB.TransferStatus_ALREADYEXISTS: status = dnf.callback.STATUS_ALREADY_EXISTS self.progress.end(self, status, msg) def _mirrorfail_cb(self, cbdata, err, url): self.progress.end(self, dnf.callback.STATUS_MIRROR, err) def _progress_cb(self, cbdata, total, done): try: self.progress.progress(self, done) except Exception: exc_type, exc_value, exc_traceback = sys.exc_info() except_list = traceback.format_exception(exc_type, exc_value, exc_traceback) logger.critical(''.join(except_list)) @property def _full_size(self): return self.download_size def _librepo_target(self): pkg = self.pkg pkgdir = pkg.pkgdir dnf.util.ensure_dir(pkgdir) target_dct = { 'dest': pkgdir, 'resume': True, 'cbdata': self, 'progresscb': self._progress_cb, 'endcb': self._end_cb, 'mirrorfailurecb': self._mirrorfail_cb, } target_dct.update(self._target_params()) return libdnf.repo.PackageTarget( pkg.repo._repo, target_dct['relative_url'], target_dct['dest'], target_dct['checksum_type'], target_dct['checksum'], target_dct['expectedsize'], target_dct['base_url'], target_dct['resume'], 0, 0, self.callbacks) class RPMPayload(PackagePayload): def __str__(self): return os.path.basename(self.pkg.location) def _target_params(self): pkg = self.pkg ctype, csum = pkg.returnIdSum() ctype_code = libdnf.repo.PackageTarget.checksumType(ctype) if ctype_code == libdnf.repo.PackageTarget.ChecksumType_UNKNOWN: logger.warning(_("unsupported checksum type: %s"), ctype) return { 'relative_url': pkg.location, 'checksum_type': ctype_code, 'checksum': csum, 'expectedsize': pkg.downloadsize, 'base_url': pkg.baseurl, } @property def download_size(self): """Total size of the download.""" return self.pkg.downloadsize class RemoteRPMPayload(PackagePayload): def __init__(self, remote_location, conf, progress): super(RemoteRPMPayload, self).__init__("unused_object", progress) self.remote_location = remote_location self.remote_size = 0 self.conf = conf s = (self.conf.releasever or "") + self.conf.substitutions.get('basearch') digest = hashlib.sha256(s.encode('utf8')).hexdigest()[:16] repodir = "commandline-" + digest self.pkgdir = os.path.join(self.conf.cachedir, repodir, "packages") dnf.util.ensure_dir(self.pkgdir) self.local_path = os.path.join(self.pkgdir, self.__str__().lstrip("/")) def __str__(self): return os.path.basename(urllib.parse.unquote(self.remote_location)) def _progress_cb(self, cbdata, total, done): self.remote_size = total try: self.progress.progress(self, done) except Exception: exc_type, exc_value, exc_traceback = sys.exc_info() except_list = traceback.format_exception(exc_type, exc_value, exc_traceback) logger.critical(''.join(except_list)) def _librepo_target(self): return libdnf.repo.PackageTarget( self.conf._config, self.remote_location, self.pkgdir, 0, None, 0, None, True, 0, 0, self.callbacks) @property def download_size(self): """Total size of the download.""" return self.remote_size class MDPayload(dnf.callback.Payload): def __init__(self, progress): super(MDPayload, self).__init__(progress) self._text = "" self._download_size = 0 self.fastest_mirror_running = False self.mirror_failures = set() def __str__(self): if dnf.pycomp.PY3: return self._text else: return self._text.encode('utf-8') def __unicode__(self): return self._text def _progress_cb(self, cbdata, total, done): self._download_size = total self.progress.progress(self, done) def _fastestmirror_cb(self, cbdata, stage, data): if stage == libdnf.repo.RepoCB.FastestMirrorStage_DETECTION: # pinging mirrors, this might take a while msg = _('determining the fastest mirror (%s hosts).. ') % data self.fastest_mirror_running = True elif stage == libdnf.repo.RepoCB.FastestMirrorStage_STATUS and self.fastest_mirror_running: # done.. report but ignore any errors msg = 'error: %s\n' % data if data else 'done.\n' else: return self.progress.message(msg) def _mirror_failure_cb(self, cbdata, msg, url, metadata): self.mirror_failures.add(msg) msg = 'error: %s (%s).' % (msg, url) logger.debug(msg) @property def download_size(self): return self._download_size @property def progress(self): return self._progress @progress.setter def progress(self, progress): if progress is None: progress = dnf.callback.NullDownloadProgress() self._progress = progress def start(self, text): self._text = text self.progress.start(1, 0) def end(self): self._download_size = 0 self.progress.end(self, None, None) # use the local cache even if it's expired. download if there's no cache. SYNC_LAZY = libdnf.repo.Repo.SyncStrategy_LAZY # use the local cache, even if it's expired, never download. SYNC_ONLY_CACHE = libdnf.repo.Repo.SyncStrategy_ONLY_CACHE # try the cache, if it is expired download new md. SYNC_TRY_CACHE = libdnf.repo.Repo.SyncStrategy_TRY_CACHE class RepoCallbacks(libdnf.repo.RepoCB): def __init__(self, repo): super(RepoCallbacks, self).__init__() self._repo = repo self._md_pload = repo._md_pload def start(self, what): self._md_pload.start(what) def end(self): self._md_pload.end() def progress(self, totalToDownload, downloaded): self._md_pload._progress_cb(None, totalToDownload, downloaded) return 0 def fastestMirror(self, stage, ptr): self._md_pload._fastestmirror_cb(None, stage, ptr) def handleMirrorFailure(self, msg, url, metadata): self._md_pload._mirror_failure_cb(None, msg, url, metadata) return 0 def repokeyImport(self, id, userid, fingerprint, url, timestamp): return self._repo._key_import._confirm(id, userid, fingerprint, url, timestamp) class Repo(dnf.conf.RepoConf): # :api DEFAULT_SYNC = SYNC_TRY_CACHE def __init__(self, name=None, parent_conf=None): # :api super(Repo, self).__init__(section=name, parent=parent_conf) self._config.this.disown() # _repo will be the owner of _config self._repo = libdnf.repo.Repo(name if name else "", self._config) self._md_pload = MDPayload(dnf.callback.NullDownloadProgress()) self._callbacks = RepoCallbacks(self) self._callbacks.this.disown() # _repo will be the owner of callbacks self._repo.setCallbacks(self._callbacks) self._pkgdir = None self._key_import = _NullKeyImport() self.metadata = None # :api self._repo.setSyncStrategy(SYNC_ONLY_CACHE if parent_conf and parent_conf.cacheonly else self.DEFAULT_SYNC) if parent_conf: self._repo.setSubstitutions(parent_conf.substitutions) self._substitutions = dnf.conf.substitutions.Substitutions() self._check_config_file_age = parent_conf.check_config_file_age \ if parent_conf is not None else True @property def id(self): # :api return self._repo.getId() @property def repofile(self): # :api return self._repo.getRepoFilePath() @repofile.setter def repofile(self, value): self._repo.setRepoFilePath(value) @property def pkgdir(self): # :api if self._repo.isLocal(): return self._repo.getLocalBaseurl() return self.cache_pkgdir() def cache_pkgdir(self): if self._pkgdir is not None: return self._pkgdir return os.path.join(self._repo.getCachedir(), _PACKAGES_RELATIVE_DIR) @pkgdir.setter def pkgdir(self, val): # :api self._pkgdir = val @property def _pubring_dir(self): return os.path.join(self._repo.getCachedir(), 'pubring') @property def load_metadata_other(self): return self._repo.getLoadMetadataOther() @load_metadata_other.setter def load_metadata_other(self, val): self._repo.setLoadMetadataOther(val) def __lt__(self, other): return self.id < other.id def __repr__(self): return "<%s %s>" % (self.__class__.__name__, self.id) def __setattr__(self, name, value): super(Repo, self).__setattr__(name, value) def disable(self): # :api self._repo.disable() def enable(self): # :api self._repo.enable() def add_metadata_type_to_download(self, metadata_type): # :api """Ask for additional repository metadata type to download. Given metadata_type is appended to the default metadata set when repository is downloaded. Parameters ---------- metadata_type: string Example: add_metadata_type_to_download("productid") """ self._repo.addMetadataTypeToDownload(metadata_type) def remove_metadata_type_from_download(self, metadata_type): # :api """Stop asking for this additional repository metadata type in download. Given metadata_type is no longer downloaded by default when this repository is downloaded. Parameters ---------- metadata_type: string Example: remove_metadata_type_from_download("productid") """ self._repo.removeMetadataTypeFromDownload(metadata_type) def get_metadata_path(self, metadata_type): # :api """Return path to the file with downloaded repository metadata of given type. Parameters ---------- metadata_type: string """ return self._repo.getMetadataPath(metadata_type) def get_metadata_content(self, metadata_type): # :api """Return content of the file with downloaded repository metadata of given type. Content of compressed metadata file is returned uncompressed. Parameters ---------- metadata_type: string """ return self._repo.getMetadataContent(metadata_type) def load(self): # :api """Load the metadata for this repo. Depending on the configuration and the age and consistence of data available on the disk cache, either loads the metadata from the cache or downloads them from the mirror, baseurl or metalink. This method will by default not try to refresh already loaded data if called repeatedly. Returns True if this call to load() caused a fresh metadata download. """ ret = False try: ret = self._repo.load() except (libdnf.error.Error, RuntimeError) as e: if self._md_pload.mirror_failures: msg = "Errors during downloading metadata for repository '%s':" % self.id for failure in self._md_pload.mirror_failures: msg += "\n - %s" % failure logger.warning(msg) raise dnf.exceptions.RepoError(str(e)) finally: self._md_pload.mirror_failures = set() self.metadata = Metadata(self._repo) return ret def _metadata_expire_in(self): """Get the number of seconds after which the cached metadata will expire. Returns a tuple, boolean whether there even is cached metadata and the number of seconds it will expire in. Negative number means the metadata has expired already, None that it never expires. """ if not self.metadata: self._repo.loadCache(False) if self.metadata: if self.metadata_expire == -1: return True, None expiration = self._repo.getExpiresIn() if self._repo.isExpired(): expiration = min(0, expiration) return True, expiration return False, 0 def _set_key_import(self, key_import): self._key_import = key_import def set_progress_bar(self, progress): # :api self._md_pload.progress = progress def get_http_headers(self): # :api """Returns user defined http headers. Returns ------- headers : tuple of strings """ return self._repo.getHttpHeaders() def set_http_headers(self, headers): # :api """Sets http headers. Sets new http headers and rewrites existing ones. Parameters ---------- headers : tuple or list of strings Example: set_http_headers(["User-Agent: Agent007", "MyFieldName: MyFieldValue"]) """ self._repo.setHttpHeaders(headers) def remote_location(self, location, schemes=('http', 'ftp', 'file', 'https')): """ :param location: relative location inside the repo :param schemes: list of allowed protocols. Default is ('http', 'ftp', 'file', 'https') :return: absolute url (string) or None """ def schemes_filter(url_list): for url in url_list: if schemes: s = dnf.pycomp.urlparse.urlparse(url)[0] if s in schemes: return os.path.join(url, location.lstrip('/')) else: return os.path.join(url, location.lstrip('/')) return None if not location: return None mirrors = self._repo.getMirrors() if mirrors: return schemes_filter(mirrors) elif self.baseurl: return schemes_filter(self.baseurl)	20.31 KB	2025-09-22 11:27
📄 repodict.py # repodict.py # Managing repo configuration in DNF. # # Copyright (C) 2013-2016 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 unicode_literals from dnf.exceptions import ConfigError from dnf.i18n import _ import dnf.util import libdnf.conf import fnmatch import os logger = dnf.util.logger class RepoDict(dict): # :api def add(self, repo): # :api id_ = repo.id if id_ in self: msg = 'Repository %s is listed more than once in the configuration' raise ConfigError(msg % id_) try: repo._repo.verify() except RuntimeError as e: raise ConfigError("{0}".format(e)) self[id_] = repo def all(self): # :api return dnf.util.MultiCallList(self.values()) def _any_enabled(self): return not dnf.util.empty(self.iter_enabled()) def _enable_sub_repos(self, sub_name_fn): for repo in self.iter_enabled(): for found in self.get_matching(sub_name_fn(repo.id)): if not found.enabled: logger.info(_('enabling %s repository'), found.id) found.enable() def add_new_repo(self, repoid, conf, baseurl=(), **kwargs): # :api """ Creates new repo object and add it into RepoDict. Variables in provided values will be automatically substituted using conf.substitutions (like $releasever, ...) @param repoid: Repo ID - string @param conf: dnf Base().conf object @param baseurl: List of strings @param kwargs: keys and values that will be used to setattr on dnf.repo.Repo() object @return: dnf.repo.Repo() object """ def substitute(values): if isinstance(values, str): return libdnf.conf.ConfigParser.substitute(values, conf.substitutions) elif isinstance(values, list) or isinstance(values, tuple): substituted = [] for value in values: if isinstance(value, str): substituted.append( libdnf.conf.ConfigParser.substitute(value, conf.substitutions)) if substituted: return substituted return values repo = dnf.repo.Repo(repoid, conf) for path in baseurl: if '://' not in path: path = 'file://{}'.format(os.path.abspath(path)) repo.baseurl += [substitute(path)] for (key, value) in kwargs.items(): setattr(repo, key, substitute(value)) self.add(repo) logger.info(_("Added %s repo from %s"), repoid, ', '.join(baseurl)) return repo def enable_debug_repos(self): # :api """enable debug repos corresponding to already enabled binary repos""" def debug_name(name): return ("{}-debug-rpms".format(name[:-5]) if name.endswith("-rpms") else "{}-debuginfo".format(name)) self._enable_sub_repos(debug_name) def enable_source_repos(self): # :api """enable source repos corresponding to already enabled binary repos""" def source_name(name): return ("{}-source-rpms".format(name[:-5]) if name.endswith("-rpms") else "{}-source".format(name)) self._enable_sub_repos(source_name) def get_matching(self, key): # :api if dnf.util.is_glob_pattern(key): l = [self[k] for k in self if fnmatch.fnmatch(k, key)] return dnf.util.MultiCallList(l) repo = self.get(key, None) if repo is None: return dnf.util.MultiCallList([]) return dnf.util.MultiCallList([repo]) def iter_enabled(self): # :api return (r for r in self.values() if r.enabled) def items(self): """return repos sorted by priority""" return (item for item in sorted(super(RepoDict, self).items(), key=lambda x: (x[1].priority, x[1].cost))) def __iter__(self): return self.keys() def keys(self): return (k for k, v in self.items()) def values(self): return (v for k, v in self.items())	5.04 KB	2025-09-22 11:27
📁 rpm	-	2026-03-21 15:47
📄 sack.py # sack.py # The dnf.Sack class, derived from hawkey.Sack # # Copyright (C) 2012-2016 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.util import dnf.package import dnf.query import logging import hawkey import os from dnf.pycomp import basestring from dnf.i18n import _ logger = logging.getLogger("dnf") class Sack(hawkey.Sack): # :api def __init__(self, args, kwargs): super(Sack, self).__init__(args, **kwargs) def _configure(self, installonly=None, installonly_limit=0, allow_vendor_change=None): if installonly: self.installonly = installonly self.installonly_limit = installonly_limit if allow_vendor_change is not None: self.allow_vendor_change = allow_vendor_change if allow_vendor_change is False: logger.warning(_("allow_vendor_change is disabled. This option is currently not supported for downgrade and distro-sync commands")) def query(self, flags=0): # :api """Factory function returning a DNF Query.""" return dnf.query.Query(self, flags) def _build_sack(base): cachedir = base.conf.cachedir # create the dir ourselves so we have the permissions under control: dnf.util.ensure_dir(cachedir) return Sack(pkgcls=dnf.package.Package, pkginitval=base, arch=base.conf.substitutions["arch"], cachedir=cachedir, rootdir=base.conf.installroot, logfile=os.path.join(base.conf.logdir, dnf.const.LOG_HAWKEY), logdebug=base.conf.logfilelevel > 9) def _rpmdb_sack(base): # used by subscription-manager (src/dnf-plugins/product-id.py) sack = _build_sack(base) try: # It can fail if rpmDB is not present sack.load_system_repo(build_cache=False) except IOError: pass return sack def rpmdb_sack(base): # :api """ Returns a new instance of sack containing only installed packages (@System repo) Useful to get list of the installed RPMs after transaction. """ return _rpmdb_sack(base)	2.97 KB	2022-09-09 11:21
📄 selector.py # selector.py # DNF specific hawkey.Selector handling. # # Copyright (C) 2012-2016 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 hawkey import Selector	1.1 KB	2022-09-09 11:21
📄 subject.py # subject.py # Implements Subject. # # Copyright (C) 2012-2016 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 print_function from __future__ import unicode_literals from hawkey import Subject # :api	1.12 KB	2022-09-09 11:21
📄 transaction.py # -- coding: utf-8 -- # transaction.py # Managing the transaction to be passed to RPM. # # Copyright (C) 2013-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 libdnf.transaction from dnf.i18n import _, C_ # :api - all action constants are considered an API # per-package actions - from libdnf PKG_DOWNGRADE = libdnf.transaction.TransactionItemAction_DOWNGRADE PKG_DOWNGRADED = libdnf.transaction.TransactionItemAction_DOWNGRADED PKG_INSTALL = libdnf.transaction.TransactionItemAction_INSTALL PKG_OBSOLETE = libdnf.transaction.TransactionItemAction_OBSOLETE PKG_OBSOLETED = libdnf.transaction.TransactionItemAction_OBSOLETED PKG_REINSTALL = libdnf.transaction.TransactionItemAction_REINSTALL PKG_REINSTALLED = libdnf.transaction.TransactionItemAction_REINSTALLED PKG_REMOVE = libdnf.transaction.TransactionItemAction_REMOVE PKG_UPGRADE = libdnf.transaction.TransactionItemAction_UPGRADE PKG_UPGRADED = libdnf.transaction.TransactionItemAction_UPGRADED # compatibility PKG_ERASE = PKG_REMOVE # per-package actions - additional PKG_CLEANUP = 101 PKG_VERIFY = 102 PKG_SCRIPTLET = 103 # transaction-wide actions TRANS_PREPARATION = 201 TRANS_POST = 202 # packages that appeared on the system FORWARD_ACTIONS = [ libdnf.transaction.TransactionItemAction_INSTALL, libdnf.transaction.TransactionItemAction_DOWNGRADE, libdnf.transaction.TransactionItemAction_OBSOLETE, libdnf.transaction.TransactionItemAction_UPGRADE, libdnf.transaction.TransactionItemAction_REINSTALL, ] # packages that got removed from the system BACKWARD_ACTIONS = [ libdnf.transaction.TransactionItemAction_DOWNGRADED, libdnf.transaction.TransactionItemAction_OBSOLETED, libdnf.transaction.TransactionItemAction_UPGRADED, libdnf.transaction.TransactionItemAction_REMOVE, # TODO: REINSTALLED may and may not belong here; the same NEVRA is in FORWARD_ACTIONS already # libdnf.transaction.TransactionItemAction_REINSTALLED, ] ACTIONS = { # TRANSLATORS: This is for a single package currently being downgraded. PKG_DOWNGRADE: C_('currently', 'Downgrading'), PKG_DOWNGRADED: _('Cleanup'), # TRANSLATORS: This is for a single package currently being installed. PKG_INSTALL: C_('currently', 'Installing'), PKG_OBSOLETE: _('Obsoleting'), PKG_OBSOLETED: _('Obsoleting'), # TRANSLATORS: This is for a single package currently being reinstalled. PKG_REINSTALL: C_('currently', 'Reinstalling'), PKG_REINSTALLED: _('Cleanup'), # TODO: 'Removing'? PKG_REMOVE: _('Erasing'), # TRANSLATORS: This is for a single package currently being upgraded. PKG_UPGRADE: C_('currently', 'Upgrading'), PKG_UPGRADED: _('Cleanup'), PKG_CLEANUP: _('Cleanup'), PKG_VERIFY: _('Verifying'), PKG_SCRIPTLET: _('Running scriptlet'), TRANS_PREPARATION: _('Preparing'), # TODO: TRANS_POST } # untranslated strings, logging to /var/log/dnf/dnf.rpm.log FILE_ACTIONS = { PKG_DOWNGRADE: 'Downgrade', PKG_DOWNGRADED: 'Downgraded', PKG_INSTALL: 'Installed', PKG_OBSOLETE: 'Obsolete', PKG_OBSOLETED: 'Obsoleted', PKG_REINSTALL: 'Reinstall', PKG_REINSTALLED: 'Reinstalled', # TODO: 'Removed'? PKG_REMOVE: 'Erase', PKG_UPGRADE: 'Upgrade', PKG_UPGRADED: 'Upgraded', PKG_CLEANUP: 'Cleanup', PKG_VERIFY: 'Verified', PKG_SCRIPTLET: 'Running scriptlet', TRANS_PREPARATION: 'Preparing', # TODO: TRANS_POST }	4.29 KB	2022-09-09 11:21
📄 transaction_sr.py # Copyright (C) 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 Library 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 print_function from __future__ import unicode_literals import libdnf import hawkey from dnf.i18n import _ import dnf.exceptions import json VERSION_MAJOR = 0 VERSION_MINOR = 0 VERSION = "%s.%s" % (VERSION_MAJOR, VERSION_MINOR) """ The version of the stored transaction. MAJOR version denotes backwards incompatible changes (old dnf won't work with new transaction JSON). MINOR version denotes extending the format without breaking backwards compatibility (old dnf can work with new transaction JSON). Forwards compatibility needs to be handled by being able to process the old format as well as the new one. """ class TransactionError(dnf.exceptions.Error): def __init__(self, msg): super(TransactionError, self).__init__(msg) class TransactionReplayError(dnf.exceptions.Error): def __init__(self, filename, errors): """ :param filename: The name of the transaction file being replayed :param errors: a list of error classes or a string with an error description """ # store args in case someone wants to read them from a caught exception self.filename = filename if isinstance(errors, (list, tuple)): self.errors = errors else: self.errors = [errors] if filename: msg = _('The following problems occurred while replaying the transaction from file "{filename}":').format(filename=filename) else: msg = _('The following problems occurred while running a transaction:') for error in self.errors: msg += "\n " + str(error) super(TransactionReplayError, self).__init__(msg) class IncompatibleTransactionVersionError(TransactionReplayError): def __init__(self, filename, msg): super(IncompatibleTransactionVersionError, self).__init__(filename, msg) def _check_version(version, filename): major, minor = version.split('.') try: major = int(major) except ValueError as e: raise TransactionReplayError( filename, _('Invalid major version "{major}", number expected.').format(major=major) ) try: int(minor) # minor is unused, just check it's a number except ValueError as e: raise TransactionReplayError( filename, _('Invalid minor version "{minor}", number expected.').format(minor=minor) ) if major != VERSION_MAJOR: raise IncompatibleTransactionVersionError( filename, _('Incompatible major version "{major}", supported major version is "{major_supp}".') .format(major=major, major_supp=VERSION_MAJOR) ) def serialize_transaction(transaction): """ Serializes a transaction to a data structure that is equivalent to the stored JSON format. :param transaction: the transaction to serialize (an instance of dnf.db.history.TransactionWrapper) """ data = { "version": VERSION, } rpms = [] groups = [] environments = [] if transaction is None: return data for tsi in transaction.packages(): if tsi.is_package(): rpms.append({ "action": tsi.action_name, "nevra": tsi.nevra, "reason": libdnf.transaction.TransactionItemReasonToString(tsi.reason), "repo_id": tsi.from_repo }) elif tsi.is_group(): group = tsi.get_group() group_data = { "action": tsi.action_name, "id": group.getGroupId(), "packages": [], "package_types": libdnf.transaction.compsPackageTypeToString(group.getPackageTypes()) } for pkg in group.getPackages(): group_data["packages"].append({ "name": pkg.getName(), "installed": pkg.getInstalled(), "package_type": libdnf.transaction.compsPackageTypeToString(pkg.getPackageType()) }) groups.append(group_data) elif tsi.is_environment(): env = tsi.get_environment() env_data = { "action": tsi.action_name, "id": env.getEnvironmentId(), "groups": [], "package_types": libdnf.transaction.compsPackageTypeToString(env.getPackageTypes()) } for grp in env.getGroups(): env_data["groups"].append({ "id": grp.getGroupId(), "installed": grp.getInstalled(), "group_type": libdnf.transaction.compsPackageTypeToString(grp.getGroupType()) }) environments.append(env_data) if rpms: data["rpms"] = rpms if groups: data["groups"] = groups if environments: data["environments"] = environments return data class TransactionReplay(object): """ A class that encapsulates replaying a transaction. The transaction data are loaded and stored when the class is initialized. The transaction is run by calling the `run()` method, after the transaction is created (but before it is performed), the `post_transaction()` method needs to be called to verify no extra packages were pulled in and also to fix the reasons. """ def __init__( self, base, filename="", data=None, ignore_extras=False, ignore_installed=False, skip_unavailable=False ): """ :param base: the dnf base :param filename: the filename to load the transaction from (conflicts with the 'data' argument) :param data: the dictionary to load the transaction from (conflicts with the 'filename' argument) :param ignore_extras: whether to ignore extra package pulled into the transaction :param ignore_installed: whether to ignore installed versions of packages :param skip_unavailable: whether to skip transaction packages that aren't available """ self._base = base self._filename = filename self._ignore_installed = ignore_installed self._ignore_extras = ignore_extras self._skip_unavailable = skip_unavailable if not self._base.conf.strict: self._skip_unavailable = True self._nevra_cache = set() self._nevra_reason_cache = {} self._warnings = [] if filename and data: raise ValueError(_("Conflicting TransactionReplay arguments have been specified: filename, data")) elif filename: self._load_from_file(filename) else: self._load_from_data(data) def _load_from_file(self, fn): self._filename = fn with open(fn, "r") as f: try: replay_data = json.load(f) except json.decoder.JSONDecodeError as e: raise TransactionReplayError(fn, str(e) + ".") try: self._load_from_data(replay_data) except TransactionError as e: raise TransactionReplayError(fn, e) def _load_from_data(self, data): self._replay_data = data self._verify_toplevel_json(self._replay_data) self._rpms = self._replay_data.get("rpms", []) self._assert_type(self._rpms, list, "rpms", "array") self._groups = self._replay_data.get("groups", []) self._assert_type(self._groups, list, "groups", "array") self._environments = self._replay_data.get("environments", []) self._assert_type(self._environments, list, "environments", "array") def _raise_or_warn(self, warn_only, msg): if warn_only: self._warnings.append(msg) else: raise TransactionError(msg) def _assert_type(self, value, t, id, expected): if not isinstance(value, t): raise TransactionError(_('Unexpected type of "{id}", {exp} expected.').format(id=id, exp=expected)) def _verify_toplevel_json(self, replay_data): fn = self._filename if "version" not in replay_data: raise TransactionReplayError(fn, _('Missing key "{key}".'.format(key="version"))) self._assert_type(replay_data["version"], str, "version", "string") _check_version(replay_data["version"], fn) def _replay_pkg_action(self, pkg_data): try: action = pkg_data["action"] nevra = pkg_data["nevra"] repo_id = pkg_data["repo_id"] reason = libdnf.transaction.StringToTransactionItemReason(pkg_data["reason"]) except KeyError as e: raise TransactionError( _('Missing object key "{key}" in an rpm.').format(key=e.args[0]) ) except IndexError as e: raise TransactionError( _('Unexpected value of package reason "{reason}" for rpm nevra "{nevra}".') .format(reason=pkg_data["reason"], nevra=nevra) ) subj = hawkey.Subject(nevra) parsed_nevras = subj.get_nevra_possibilities(forms=[hawkey.FORM_NEVRA]) if len(parsed_nevras) != 1: raise TransactionError(_('Cannot parse NEVRA for package "{nevra}".').format(nevra=nevra)) parsed_nevra = parsed_nevras[0] na = "%s.%s" % (parsed_nevra.name, parsed_nevra.arch) query_na = self._base.sack.query().filter(name=parsed_nevra.name, arch=parsed_nevra.arch) epoch = parsed_nevra.epoch if parsed_nevra.epoch is not None else 0 query = query_na.filter(epoch=epoch, version=parsed_nevra.version, release=parsed_nevra.release) # In case the package is found in the same repo as in the original # transaction, limit the query to that plus installed packages. IOW # remove packages with the same NEVRA in case they are found in # multiple repos and the repo the package came from originally is one # of them. # This can e.g. make a difference in the system-upgrade plugin, in case # the same NEVRA is in two repos, this makes sure the same repo is used # for both download and upgrade steps of the plugin. if repo_id: query_repo = query.filter(reponame=repo_id) if query_repo: query = query_repo.union(query.installed()) if not query: self._raise_or_warn(self._skip_unavailable, _('Cannot find rpm nevra "{nevra}".').format(nevra=nevra)) return # a cache to check no extra packages were pulled into the transaction if action != "Reason Change": self._nevra_cache.add(nevra) # store reasons for forward actions and "Removed", the rest of the # actions reasons should stay as they were determined by the transaction if action in ("Install", "Upgrade", "Downgrade", "Reinstall", "Removed"): self._nevra_reason_cache[nevra] = reason if action in ("Install", "Upgrade", "Downgrade"): if action == "Install" and query_na.installed() and not self._base._get_installonly_query(query_na): self._raise_or_warn(self._ignore_installed, _('Package "{na}" is already installed for action "{action}".').format(na=na, action=action)) sltr = dnf.selector.Selector(self._base.sack).set(pkg=query) self._base.goal.install(select=sltr, optional=not self._base.conf.strict) elif action == "Reinstall": query = query.available() if not query: self._raise_or_warn(self._skip_unavailable, _('Package nevra "{nevra}" not available in repositories for action "{action}".') .format(nevra=nevra, action=action)) return sltr = dnf.selector.Selector(self._base.sack).set(pkg=query) self._base.goal.install(select=sltr, optional=not self._base.conf.strict) elif action in ("Upgraded", "Downgraded", "Reinstalled", "Removed", "Obsoleted"): query = query.installed() if not query: self._raise_or_warn(self._ignore_installed, _('Package nevra "{nevra}" not installed for action "{action}".').format(nevra=nevra, action=action)) return # erasing the original version (the reverse part of an action like # e.g. upgrade) is more robust, but we can't do it if # skip_unavailable is True, because if the forward part of the # action is skipped, we would simply remove the package here if not self._skip_unavailable or action == "Removed": for pkg in query: self._base.goal.erase(pkg, clean_deps=False) elif action == "Reason Change": self._base.history.set_reason(query[0], reason) else: raise TransactionError( _('Unexpected value of package action "{action}" for rpm nevra "{nevra}".') .format(action=action, nevra=nevra) ) def _create_swdb_group(self, group_id, pkg_types, pkgs): comps_group = self._base.comps._group_by_id(group_id) if not comps_group: self._raise_or_warn(self._skip_unavailable, _("Group id '%s' is not available.") % group_id) return None swdb_group = self._base.history.group.new(group_id, comps_group.name, comps_group.ui_name, pkg_types) try: for pkg in pkgs: name = pkg["name"] self._assert_type(name, str, "groups.packages.name", "string") installed = pkg["installed"] self._assert_type(installed, bool, "groups.packages.installed", "boolean") package_type = pkg["package_type"] self._assert_type(package_type, str, "groups.packages.package_type", "string") try: swdb_group.addPackage(name, installed, libdnf.transaction.stringToCompsPackageType(package_type)) except libdnf.error.Error as e: raise TransactionError(str(e)) except KeyError as e: raise TransactionError( _('Missing object key "{key}" in groups.packages.').format(key=e.args[0]) ) return swdb_group def _swdb_group_install(self, group_id, pkg_types, pkgs): swdb_group = self._create_swdb_group(group_id, pkg_types, pkgs) if swdb_group is not None: self._base.history.group.install(swdb_group) def _swdb_group_upgrade(self, group_id, pkg_types, pkgs): if not self._base.history.group.get(group_id): self._raise_or_warn( self._ignore_installed, _("Group id '%s' is not installed.") % group_id) return swdb_group = self._create_swdb_group(group_id, pkg_types, pkgs) if swdb_group is not None: self._base.history.group.upgrade(swdb_group) def _swdb_group_downgrade(self, group_id, pkg_types, pkgs): if not self._base.history.group.get(group_id): self._raise_or_warn(self._ignore_installed, _("Group id '%s' is not installed.") % group_id) return swdb_group = self._create_swdb_group(group_id, pkg_types, pkgs) if swdb_group is not None: self._base.history.group.downgrade(swdb_group) def _swdb_group_remove(self, group_id, pkg_types, pkgs): if not self._base.history.group.get(group_id): self._raise_or_warn(self._ignore_installed, _("Group id '%s' is not installed.") % group_id) return swdb_group = self._create_swdb_group(group_id, pkg_types, pkgs) if swdb_group is not None: self._base.history.group.remove(swdb_group) def _create_swdb_environment(self, env_id, pkg_types, groups): comps_env = self._base.comps._environment_by_id(env_id) if not comps_env: self._raise_or_warn(self._skip_unavailable, _("Environment id '%s' is not available.") % env_id) return None swdb_env = self._base.history.env.new(env_id, comps_env.name, comps_env.ui_name, pkg_types) try: for grp in groups: id = grp["id"] self._assert_type(id, str, "environments.groups.id", "string") installed = grp["installed"] self._assert_type(installed, bool, "environments.groups.installed", "boolean") group_type = grp["group_type"] self._assert_type(group_type, str, "environments.groups.group_type", "string") try: group_type = libdnf.transaction.stringToCompsPackageType(group_type) except libdnf.error.Error as e: raise TransactionError(str(e)) if group_type not in ( libdnf.transaction.CompsPackageType_MANDATORY, libdnf.transaction.CompsPackageType_OPTIONAL ): raise TransactionError( _('Invalid value "{group_type}" of environments.groups.group_type, ' 'only "mandatory" or "optional" is supported.' ).format(group_type=grp["group_type"]) ) swdb_env.addGroup(id, installed, group_type) except KeyError as e: raise TransactionError( _('Missing object key "{key}" in environments.groups.').format(key=e.args[0]) ) return swdb_env def _swdb_environment_install(self, env_id, pkg_types, groups): swdb_env = self._create_swdb_environment(env_id, pkg_types, groups) if swdb_env is not None: self._base.history.env.install(swdb_env) def _swdb_environment_upgrade(self, env_id, pkg_types, groups): if not self._base.history.env.get(env_id): self._raise_or_warn(self._ignore_installed,_("Environment id '%s' is not installed.") % env_id) return swdb_env = self._create_swdb_environment(env_id, pkg_types, groups) if swdb_env is not None: self._base.history.env.upgrade(swdb_env) def _swdb_environment_downgrade(self, env_id, pkg_types, groups): if not self._base.history.env.get(env_id): self._raise_or_warn(self._ignore_installed, _("Environment id '%s' is not installed.") % env_id) return swdb_env = self._create_swdb_environment(env_id, pkg_types, groups) if swdb_env is not None: self._base.history.env.downgrade(swdb_env) def _swdb_environment_remove(self, env_id, pkg_types, groups): if not self._base.history.env.get(env_id): self._raise_or_warn(self._ignore_installed, _("Environment id '%s' is not installed.") % env_id) return swdb_env = self._create_swdb_environment(env_id, pkg_types, groups) if swdb_env is not None: self._base.history.env.remove(swdb_env) def get_data(self): """ :returns: the loaded data of the transaction """ return self._replay_data def get_warnings(self): """ :returns: an array of warnings gathered during the transaction replay """ return self._warnings def run(self): """ Replays the transaction. """ fn = self._filename errors = [] for pkg_data in self._rpms: try: self._replay_pkg_action(pkg_data) except TransactionError as e: errors.append(e) for group_data in self._groups: try: action = group_data["action"] group_id = group_data["id"] try: pkg_types = libdnf.transaction.stringToCompsPackageType(group_data["package_types"]) except libdnf.error.Error as e: errors.append(TransactionError(str(e))) continue if action == "Install": self._swdb_group_install(group_id, pkg_types, group_data["packages"]) elif action == "Upgrade": self._swdb_group_upgrade(group_id, pkg_types, group_data["packages"]) elif action == "Downgraded": self._swdb_group_downgrade(group_id, pkg_types, group_data["packages"]) elif action == "Removed": self._swdb_group_remove(group_id, pkg_types, group_data["packages"]) else: errors.append(TransactionError( _('Unexpected value of group action "{action}" for group "{group}".') .format(action=action, group=group_id) )) except KeyError as e: errors.append(TransactionError( _('Missing object key "{key}" in a group.').format(key=e.args[0]) )) except TransactionError as e: errors.append(e) for env_data in self._environments: try: action = env_data["action"] env_id = env_data["id"] try: pkg_types = libdnf.transaction.stringToCompsPackageType(env_data["package_types"]) except libdnf.error.Error as e: errors.append(TransactionError(str(e))) continue if action == "Install": self._swdb_environment_install(env_id, pkg_types, env_data["groups"]) elif action == "Upgrade": self._swdb_environment_upgrade(env_id, pkg_types, env_data["groups"]) elif action == "Downgraded": self._swdb_environment_downgrade(env_id, pkg_types, env_data["groups"]) elif action == "Removed": self._swdb_environment_remove(env_id, pkg_types, env_data["groups"]) else: errors.append(TransactionError( _('Unexpected value of environment action "{action}" for environment "{env}".') .format(action=action, env=env_id) )) except KeyError as e: errors.append(TransactionError( _('Missing object key "{key}" in an environment.').format(key=e.args[0]) )) except TransactionError as e: errors.append(e) if errors: raise TransactionReplayError(fn, errors) def post_transaction(self): """ Sets reasons in the transaction history to values from the stored transaction. Also serves to check whether additional packages were pulled in by the transaction, which results in an error (unless ignore_extras is True). """ if not self._base.transaction: return errors = [] for tsi in self._base.transaction: try: pkg = tsi.pkg except KeyError as e: # the transaction item has no package, happens for action == "Reason Change" continue nevra = str(pkg) if nevra not in self._nevra_cache: # if ignore_installed is True, we don't want to check for # Upgraded/Downgraded/Reinstalled extras in the transaction, # basically those may be installed and we are ignoring them if not self._ignore_installed or not tsi.action in ( libdnf.transaction.TransactionItemAction_UPGRADED, libdnf.transaction.TransactionItemAction_DOWNGRADED, libdnf.transaction.TransactionItemAction_REINSTALLED ): msg = _('Package nevra "{nevra}", which is not present in the transaction file, was pulled ' 'into the transaction.' ).format(nevra=nevra) if not self._ignore_extras: errors.append(TransactionError(msg)) else: self._warnings.append(msg) try: replay_reason = self._nevra_reason_cache[nevra] if tsi.action in ( libdnf.transaction.TransactionItemAction_INSTALL, libdnf.transaction.TransactionItemAction_REMOVE ) or libdnf.transaction.TransactionItemReasonCompare(replay_reason, tsi.reason) > 0: tsi.reason = replay_reason except KeyError as e: # if the pkg nevra wasn't found, we don't want to change the reason pass if errors: raise TransactionReplayError(self._filename, errors)	25.2 KB	2022-09-09 11:21
📄 util.py # util.py # Basic dnf utils. # # Copyright (C) 2012-2016 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 from __future__ import absolute_import from __future__ import unicode_literals from .pycomp import PY3, basestring from dnf.i18n import _, ucd import argparse import ctypes import dnf import dnf.callback import dnf.const import dnf.pycomp import errno import functools import hawkey import itertools import locale import logging import os import pwd import shutil import subprocess import sys import tempfile import time import libdnf.repo import libdnf.transaction logger = logging.getLogger('dnf') MAIN_PROG = argparse.ArgumentParser().prog if argparse.ArgumentParser().prog == "yum" else "dnf" MAIN_PROG_UPPER = MAIN_PROG.upper() """DNF Utilities.""" def _parse_specs(namespace, values): """ Categorize :param values list into packages, groups and filenames :param namespace: argparse.Namespace, where specs will be stored :param values: list of specs, whether packages ('foo') or groups/modules ('@bar') or filenames ('.rmp', 'http://', ...) To access packages use: specs.pkg_specs, to access groups use: specs.grp_specs, to access filenames use: specs.filenames """ setattr(namespace, "filenames", []) setattr(namespace, "grp_specs", []) setattr(namespace, "pkg_specs", []) tmp_set = set() for value in values: if value in tmp_set: continue tmp_set.add(value) schemes = dnf.pycomp.urlparse.urlparse(value)[0] if value.endswith('.rpm'): namespace.filenames.append(value) elif schemes and schemes in ('http', 'ftp', 'file', 'https'): namespace.filenames.append(value) elif value.startswith('@'): namespace.grp_specs.append(value[1:]) else: namespace.pkg_specs.append(value) def _urlopen_progress(url, conf, progress=None): if progress is None: progress = dnf.callback.NullDownloadProgress() pload = dnf.repo.RemoteRPMPayload(url, conf, progress) est_remote_size = sum([pload.download_size]) progress.start(1, est_remote_size) targets = [pload._librepo_target()] try: libdnf.repo.PackageTarget.downloadPackages(libdnf.repo.VectorPPackageTarget(targets), True) except RuntimeError as e: if conf.strict: raise IOError(str(e)) logger.error(str(e)) return pload.local_path def _urlopen(url, conf=None, repo=None, mode='w+b', kwargs): """ Open the specified absolute url, return a file object which respects proxy setting even for non-repo downloads """ if PY3 and 'b' not in mode: kwargs.setdefault('encoding', 'utf-8') fo = tempfile.NamedTemporaryFile(mode, kwargs) try: if repo: repo._repo.downloadUrl(url, fo.fileno()) else: libdnf.repo.Downloader.downloadURL(conf._config if conf else None, url, fo.fileno()) except RuntimeError as e: raise IOError(str(e)) fo.seek(0) return fo def rtrim(s, r): if s.endswith(r): s = s[:-len(r)] return s def am_i_root(): # used by ansible (lib/ansible/modules/packaging/os/dnf.py) return os.geteuid() == 0 def clear_dir(path): """Remove all files and dirs under `path` Also see rm_rf() """ for entry in os.listdir(path): contained_path = os.path.join(path, entry) rm_rf(contained_path) def ensure_dir(dname): # used by ansible (lib/ansible/modules/packaging/os/dnf.py) try: os.makedirs(dname, mode=0o755) except OSError as e: if e.errno != errno.EEXIST or not os.path.isdir(dname): raise e def split_path(path): """ Split path by path separators. Use os.path.join() to join the path back to string. """ result = [] head = path while True: head, tail = os.path.split(head) if not tail: if head or not result: # if not result: make sure result is [""] so os.path.join(result) can be called result.insert(0, head) break result.insert(0, tail) return result def empty(iterable): try: l = len(iterable) except TypeError: l = len(list(iterable)) return l == 0 def first(iterable): """Returns the first item from an iterable or None if it has no elements.""" it = iter(iterable) try: return next(it) except StopIteration: return None def first_not_none(iterable): it = iter(iterable) try: return next(item for item in it if item is not None) except StopIteration: return None def file_age(fn): return time.time() - file_timestamp(fn) def file_timestamp(fn): return os.stat(fn).st_mtime def get_effective_login(): try: return pwd.getpwuid(os.geteuid())[0] except KeyError: return "UID: %s" % os.geteuid() def get_in(dct, keys, not_found): """Like dict.get() for nested dicts.""" for k in keys: dct = dct.get(k) if dct is None: return not_found return dct def group_by_filter(fn, iterable): def splitter(acc, item): acc[not bool(fn(item))].append(item) return acc return functools.reduce(splitter, iterable, ([], [])) def insert_if(item, iterable, condition): """Insert an item into an iterable by a condition.""" for original_item in iterable: if condition(original_item): yield item yield original_item def is_exhausted(iterator): """Test whether an iterator is exhausted.""" try: next(iterator) except StopIteration: return True else: return False def is_glob_pattern(pattern): if is_string_type(pattern): pattern = [pattern] return (isinstance(pattern, list) and any(set(p) & set("[?") for p in pattern)) def is_string_type(obj): if PY3: return isinstance(obj, str) else: return isinstance(obj, basestring) def lazyattr(attrname): """Decorator to get lazy attribute initialization. Composes with @property. Force reinitialization by deleting the <attrname>. """ def get_decorated(fn): def cached_getter(obj): try: return getattr(obj, attrname) except AttributeError: val = fn(obj) setattr(obj, attrname, val) return val return cached_getter return get_decorated def mapall(fn, seq): """Like functools.map(), but return a list instead of an iterator. This means all side effects of fn take place even without iterating the result. """ return list(map(fn, seq)) def normalize_time(timestamp): """Convert time into locale aware datetime string object.""" t = time.strftime("%c", time.localtime(timestamp)) if not dnf.pycomp.PY3: current_locale_setting = locale.getlocale()[1] if current_locale_setting: t = t.decode(current_locale_setting) return t def on_ac_power(): """Decide whether we are on line power. Returns True if we are on line power, False if not, None if it can not be decided. """ try: ps_folder = "/sys/class/power_supply" ac_nodes = [node for node in os.listdir(ps_folder) if node.startswith("AC")] if len(ac_nodes) > 0: ac_node = ac_nodes[0] with open("{}/{}/online".format(ps_folder, ac_node)) as ac_status: data = ac_status.read() return int(data) == 1 return None except (IOError, ValueError): return None def on_metered_connection(): """Decide whether we are on metered connection. Returns: True: if on metered connection False: if not None: if it can not be decided """ try: import dbus except ImportError: return None try: bus = dbus.SystemBus() proxy = bus.get_object("org.freedesktop.NetworkManager", "/org/freedesktop/NetworkManager") iface = dbus.Interface(proxy, "org.freedesktop.DBus.Properties") metered = iface.Get("org.freedesktop.NetworkManager", "Metered") except dbus.DBusException: return None if metered == 0: # NM_METERED_UNKNOWN return None elif metered in (1, 3): # NM_METERED_YES, NM_METERED_GUESS_YES return True elif metered in (2, 4): # NM_METERED_NO, NM_METERED_GUESS_NO return False else: # Something undocumented (at least at this moment) raise ValueError("Unknown value for metered property: %r", metered) def partition(pred, iterable): """Use a predicate to partition entries into false entries and true entries. Credit: Python library itertools' documentation. """ t1, t2 = itertools.tee(iterable) return dnf.pycomp.filterfalse(pred, t1), filter(pred, t2) def rm_rf(path): try: shutil.rmtree(path) except OSError: pass def split_by(iterable, condition): """Split an iterable into tuples by a condition. Inserts a separator before each item which meets the condition and then cuts the iterable by these separators. """ separator = object() # A unique object. # Create a function returning tuple of objects before the separator. def next_subsequence(it): return tuple(itertools.takewhile(lambda e: e != separator, it)) # Mark each place where the condition is met by the separator. marked = insert_if(separator, iterable, condition) # The 1st subsequence may be empty if the 1st item meets the condition. yield next_subsequence(marked) while True: subsequence = next_subsequence(marked) if not subsequence: break yield subsequence def strip_prefix(s, prefix): if s.startswith(prefix): return s[len(prefix):] return None def touch(path, no_create=False): """Create an empty file if it doesn't exist or bump it's timestamps. If no_create is True only bumps the timestamps. """ if no_create or os.access(path, os.F_OK): return os.utime(path, None) with open(path, 'a'): pass def _terminal_messenger(tp='write', msg="", out=sys.stdout): try: if tp == 'write': out.write(msg) elif tp == 'flush': out.flush() elif tp == 'write_flush': out.write(msg) out.flush() elif tp == 'print': print(msg, file=out) else: raise ValueError('Unsupported type: ' + tp) except IOError as e: logger.critical('{}: {}'.format(type(e).__name__, ucd(e))) pass def _format_resolve_problems(resolve_problems): """ Format string about problems in resolve :param resolve_problems: list with list of strings (output of goal.problem_rules()) :return: string """ msg = "" count_problems = (len(resolve_problems) > 1) for i, rs in enumerate(resolve_problems, start=1): if count_problems: msg += "\n " + _("Problem") + " %d: " % i else: msg += "\n " + _("Problem") + ": " msg += "\n - ".join(rs) return msg def _te_nevra(te): nevra = te.N() + '-' if te.E() is not None and te.E() != '0': nevra += te.E() + ':' return nevra + te.V() + '-' + te.R() + '.' + te.A() def _log_rpm_trans_with_swdb(rpm_transaction, swdb_transaction): logger.debug("Logging transaction elements") for rpm_el in rpm_transaction: tsi = rpm_el.Key() tsi_state = None if tsi is not None: tsi_state = tsi.state msg = "RPM element: '{}', Key(): '{}', Key state: '{}', Failed() '{}': ".format( _te_nevra(rpm_el), tsi, tsi_state, rpm_el.Failed()) logger.debug(msg) for tsi in swdb_transaction: msg = "SWDB element: '{}', State: '{}', Action: '{}', From repo: '{}', Reason: '{}', " \ "Get reason: '{}'".format(str(tsi), tsi.state, tsi.action, tsi.from_repo, tsi.reason, tsi.get_reason()) logger.debug(msg) def _sync_rpm_trans_with_swdb(rpm_transaction, swdb_transaction): revert_actions = {libdnf.transaction.TransactionItemAction_DOWNGRADED, libdnf.transaction.TransactionItemAction_OBSOLETED, libdnf.transaction.TransactionItemAction_REMOVE, libdnf.transaction.TransactionItemAction_UPGRADED, libdnf.transaction.TransactionItemAction_REINSTALLED} cached_tsi = [tsi for tsi in swdb_transaction] el_not_found = False error = False for rpm_el in rpm_transaction: te_nevra = _te_nevra(rpm_el) tsi = rpm_el.Key() if tsi is None or not hasattr(tsi, "pkg"): for tsi_candidate in cached_tsi: if tsi_candidate.state != libdnf.transaction.TransactionItemState_UNKNOWN: continue if tsi_candidate.action not in revert_actions: continue if str(tsi_candidate) == te_nevra: tsi = tsi_candidate break if tsi is None or not hasattr(tsi, "pkg"): logger.critical(_("TransactionItem not found for key: {}").format(te_nevra)) el_not_found = True continue if rpm_el.Failed(): tsi.state = libdnf.transaction.TransactionItemState_ERROR error = True else: tsi.state = libdnf.transaction.TransactionItemState_DONE for tsi in cached_tsi: if tsi.state == libdnf.transaction.TransactionItemState_UNKNOWN: logger.critical(_("TransactionSWDBItem not found for key: {}").format(str(tsi))) el_not_found = True if error: logger.debug(_('Errors occurred during transaction.')) if el_not_found: _log_rpm_trans_with_swdb(rpm_transaction, cached_tsi) class tmpdir(object): # used by subscription-manager (src/dnf-plugins/product-id.py) def __init__(self): prefix = '%s-' % dnf.const.PREFIX self.path = tempfile.mkdtemp(prefix=prefix) def __enter__(self): return self.path def __exit__(self, exc_type, exc_value, traceback): rm_rf(self.path) class Bunch(dict): """Dictionary with attribute accessing syntax. In DNF, prefer using this over dnf.yum.misc.GenericHolder. Credit: Alex Martelli, Doug Hudgeon """ def __init__(self, args, kwds): super(Bunch, self).__init__(args, *kwds) self.__dict__ = self def __hash__(self): return id(self) class MultiCallList(list): def __init__(self, iterable): super(MultiCallList, self).__init__() self.extend(iterable) def __getattr__(self, what): def fn(args, *kwargs): def call_what(v): method = getattr(v, what) return method(args, **kwargs) return list(map(call_what, self)) return fn def __setattr__(self, what, val): def setter(item): setattr(item, what, val) return list(map(setter, self)) def _make_lists(transaction): b = Bunch({ 'downgraded': [], 'erased': [], 'erased_clean': [], 'erased_dep': [], 'installed': [], 'installed_group': [], 'installed_dep': [], 'installed_weak': [], 'reinstalled': [], 'upgraded': [], 'failed': [], }) for tsi in transaction: if tsi.state == libdnf.transaction.TransactionItemState_ERROR: b.failed.append(tsi) elif tsi.action == libdnf.transaction.TransactionItemAction_DOWNGRADE: b.downgraded.append(tsi) elif tsi.action == libdnf.transaction.TransactionItemAction_INSTALL: if tsi.reason == libdnf.transaction.TransactionItemReason_GROUP: b.installed_group.append(tsi) elif tsi.reason == libdnf.transaction.TransactionItemReason_DEPENDENCY: b.installed_dep.append(tsi) elif tsi.reason == libdnf.transaction.TransactionItemReason_WEAK_DEPENDENCY: b.installed_weak.append(tsi) else: # TransactionItemReason_USER b.installed.append(tsi) elif tsi.action == libdnf.transaction.TransactionItemAction_REINSTALL: b.reinstalled.append(tsi) elif tsi.action == libdnf.transaction.TransactionItemAction_REMOVE: if tsi.reason == libdnf.transaction.TransactionItemReason_CLEAN: b.erased_clean.append(tsi) elif tsi.reason == libdnf.transaction.TransactionItemReason_DEPENDENCY: b.erased_dep.append(tsi) else: b.erased.append(tsi) elif tsi.action == libdnf.transaction.TransactionItemAction_UPGRADE: b.upgraded.append(tsi) return b def _post_transaction_output(base, transaction, action_callback): """Returns a human-readable summary of the results of the transaction. :param action_callback: function generating output for specific action. It takes two parameters - action as a string and list of affected packages for this action :return: a list of lines containing a human-readable summary of the results of the transaction """ def _tsi_or_pkg_nevra_cmp(item1, item2): """Compares two transaction items or packages by nevra. Used as a fallback when tsi does not contain package object. """ ret = (item1.name > item2.name) - (item1.name < item2.name) if ret != 0: return ret nevra1 = hawkey.NEVRA(name=item1.name, epoch=item1.epoch, version=item1.version, release=item1.release, arch=item1.arch) nevra2 = hawkey.NEVRA(name=item2.name, epoch=item2.epoch, version=item2.version, release=item2.release, arch=item2.arch) ret = nevra1.evr_cmp(nevra2, base.sack) if ret != 0: return ret return (item1.arch > item2.arch) - (item1.arch < item2.arch) list_bunch = dnf.util._make_lists(transaction) skipped_conflicts, skipped_broken = base._skipped_packages( report_problems=False, transaction=transaction) skipped = skipped_conflicts.union(skipped_broken) out = [] for (action, tsis) in [(_('Upgraded'), list_bunch.upgraded), (_('Downgraded'), list_bunch.downgraded), (_('Installed'), list_bunch.installed + list_bunch.installed_group + list_bunch.installed_weak + list_bunch.installed_dep), (_('Reinstalled'), list_bunch.reinstalled), (_('Skipped'), skipped), (_('Removed'), list_bunch.erased + list_bunch.erased_dep + list_bunch.erased_clean), (_('Failed'), list_bunch.failed)]: out.extend(action_callback( action, sorted(tsis, key=functools.cmp_to_key(_tsi_or_pkg_nevra_cmp)))) return out def _name_unset_wrapper(input_name): # returns <name-unset> for everything that evaluates to False (None, empty..) return input_name if input_name else _("<name-unset>") class _BootcSystem: usr = "/usr" CLONE_NEWNS = 0x00020000 # defined in linux/include/uapi/linux/sched.h def __init__(self): if not self.is_bootc_system(): raise RuntimeError(_("Not running on a bootc system.")) import gi self._gi = gi gi.require_version("OSTree", "1.0") from gi.repository import OSTree self._OSTree = OSTree self._sysroot = self._OSTree.Sysroot.new_default() assert self._sysroot.load(None) self._booted_deployment = self._sysroot.require_booted_deployment() assert self._booted_deployment is not None @staticmethod def is_bootc_system(): """Returns true is the system is managed as an immutable container, false otherwise.""" ostree_booted = "/run/ostree-booted" return os.path.isfile(ostree_booted) @classmethod def is_writable(cls): """Returns true if and only if /usr is writable.""" return os.access(cls.usr, os.W_OK) def _get_unlocked_state(self): return self._booted_deployment.get_unlocked() def is_unlocked_transient(self): """Returns true if and only if the bootc system is unlocked in a transient state, i.e. a overlayfs is mounted as read-only on /usr. Changes can be made to the overlayfs by remounting /usr as read/write in a private mount namespace.""" return self._get_unlocked_state() == self._OSTree.DeploymentUnlockedState.TRANSIENT @classmethod def _set_up_mountns(cls): # os.unshare is only available in Python >= 3.12. # Access symbols in libraries loaded by the Python interpreter, # which will include libc. See https://bugs.python.org/issue34592. libc = ctypes.CDLL(None) if libc.unshare(cls.CLONE_NEWNS) != 0: raise OSError("Failed to unshare mount namespace") mount_command = ["mount", "--options-source=disable", "-o", "remount,rw", cls.usr] try: completed_process = subprocess.run(mount_command, text=True) completed_process.check_returncode() except FileNotFoundError: raise dnf.exceptions.Error(_("%s: command not found.") % mount_command[0]) except subprocess.CalledProcessError: raise dnf.exceptions.Error(_("Failed to mount %s as read/write: %s", cls.usr, completed_process.stderr)) @staticmethod def _unlock(): unlock_command = ["ostree", "admin", "unlock", "--transient"] try: completed_process = subprocess.run(unlock_command, text=True) completed_process.check_returncode() except FileNotFoundError: raise dnf.exceptions.Error(_("%s: command not found. Is this a bootc system?") % unlock_command[0]) except subprocess.CalledProcessError: raise dnf.exceptions.Error(_("Failed to unlock system: %s", completed_process.stderr)) def make_writable(self): """Set up a writable overlay on bootc systems.""" bootc_unlocked_state = self._get_unlocked_state() valid_bootc_unlocked_states = ( self._OSTree.DeploymentUnlockedState.NONE, self._OSTree.DeploymentUnlockedState.DEVELOPMENT, self._OSTree.DeploymentUnlockedState.TRANSIENT, self._OSTree.DeploymentUnlockedState.HOTFIX, ) if bootc_unlocked_state not in valid_bootc_unlocked_states: raise ValueError(_("Unhandled bootc unlocked state: %s") % bootc_unlocked_state.value_nick) writable_unlocked_states = ( self._OSTree.DeploymentUnlockedState.DEVELOPMENT, self._OSTree.DeploymentUnlockedState.HOTFIX, ) if bootc_unlocked_state in writable_unlocked_states: # System is already unlocked in development mode, and usr is # already mounted read/write. pass elif bootc_unlocked_state == self._OSTree.DeploymentUnlockedState.NONE: # System is not unlocked. Unlock it in transient mode, then set up # a mount namespace for DNF. self._unlock() self._set_up_mountns() elif bootc_unlocked_state == self._OSTree.DeploymentUnlockedState.TRANSIENT: # System is unlocked in transient mode, so usr is mounted # read-only. Set up a mount namespace for DNF. self._set_up_mountns() assert os.access(self.usr, os.W_OK)	24.4 KB	2025-09-22 11:27
📁 yum	-	2026-03-21 15:47

Edit File:

Terminal