Wala shell

Advanced PHP File Manager

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

Name	Size	Last Modified	Actions
📁 ..	-	2026-03-21 15:47
📄 __init__.py # __init__.py # The legacy YUM subpackage. # # Copyright (C) 2013 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. #	1000 B	2022-09-09 11:21
📁 __pycache__	-	2026-03-21 15:47
📄 misc.py # misc.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. # """ Assorted utility functions for yum. """ from __future__ import print_function, absolute_import from __future__ import unicode_literals from dnf.pycomp import base64_decodebytes, basestring, unicode from stat import * import libdnf.utils import dnf.const import dnf.crypto import dnf.exceptions import dnf.i18n import errno import glob import io import os import os.path import pwd import re import shutil import tempfile _default_checksums = ['sha256'] _re_compiled_glob_match = None def re_glob(s): """ Tests if a string is a shell wildcard. """ global _re_compiled_glob_match if _re_compiled_glob_match is None: _re_compiled_glob_match = re.compile(r'[?]\|\[.+\]').search return _re_compiled_glob_match(s) _re_compiled_full_match = None def re_full_search_needed(s): """ Tests if a string needs a full nevra match, instead of just name. """ global _re_compiled_full_match if _re_compiled_full_match is None: # A glob, or a "." or "-" separator, followed by something (the ".") one = re.compile(r'.([-.?]\|\[.+\]).').match # Any epoch, for envra two = re.compile('[0-9]+:').match _re_compiled_full_match = (one, two) for rec in _re_compiled_full_match: if rec(s): return True return False def get_default_chksum_type(): return _default_checksums[0] class GenericHolder(object): """Generic Holder class used to hold other objects of known types It exists purely to be able to do object.somestuff, object.someotherstuff or object[key] and pass object to another function that will understand it""" def __init__(self, iter=None): self.__iter = iter def __iter__(self): if self.__iter is not None: return iter(self[self.__iter]) def __getitem__(self, item): if hasattr(self, item): return getattr(self, item) else: raise KeyError(item) def all_lists(self): """Return a dictionary of all lists.""" return {key: list_ for key, list_ in vars(self).items() if type(list_) is list} def merge_lists(self, other): """ Concatenate the list attributes from 'other' to ours. """ for (key, val) in other.all_lists().items(): vars(self).setdefault(key, []).extend(val) return self def procgpgkey(rawkey): '''Convert ASCII-armored GPG key to binary ''' # Normalise newlines rawkey = re.sub(b'\r\n?', b'\n', rawkey) # Extract block block = io.BytesIO() inblock = 0 pastheaders = 0 for line in rawkey.split(b'\n'): if line.startswith(b'-----BEGIN PGP PUBLIC KEY BLOCK-----'): inblock = 1 elif inblock and line.strip() == b'': pastheaders = 1 elif inblock and line.startswith(b'-----END PGP PUBLIC KEY BLOCK-----'): # Hit the end of the block, get out break elif pastheaders and line.startswith(b'='): # Hit the CRC line, don't include this and stop break elif pastheaders: block.write(line + b'\n') # Decode and return return base64_decodebytes(block.getvalue()) def keyInstalled(ts, keyid, timestamp): ''' Return if the GPG key described by the given keyid and timestamp are installed in the rpmdb. The keyid and timestamp should both be passed as integers. The ts is an rpm transaction set object Return values: - -1 key is not installed - 0 key with matching ID and timestamp is installed - 1 key with matching ID is installed but has an older timestamp - 2 key with matching ID is installed but has a newer timestamp No effort is made to handle duplicates. The first matching keyid is used to calculate the return result. ''' # Search for hdr in ts.dbMatch('name', 'gpg-pubkey'): if hdr['version'] == keyid: installedts = int(hdr['release'], 16) if installedts == timestamp: return 0 elif installedts < timestamp: return 1 else: return 2 return -1 def import_key_to_pubring(rawkey, keyid, gpgdir=None, make_ro_copy=True): if not os.path.exists(gpgdir): os.makedirs(gpgdir) with dnf.crypto.pubring_dir(gpgdir), dnf.crypto.Context() as ctx: # import the key with open(os.path.join(gpgdir, 'gpg.conf'), 'wb') as fp: fp.write(b'') ctx.op_import(rawkey) if make_ro_copy: rodir = gpgdir + '-ro' if not os.path.exists(rodir): os.makedirs(rodir, mode=0o755) for f in glob.glob(gpgdir + '/'): basename = os.path.basename(f) ro_f = rodir + '/' + basename shutil.copy(f, ro_f) os.chmod(ro_f, 0o755) # yes it is this stupid, why do you ask? opts = """lock-never no-auto-check-trustdb trust-model direct no-expensive-trust-checks no-permission-warning preserve-permissions """ with open(os.path.join(rodir, 'gpg.conf'), 'w', 0o755) as fp: fp.write(opts) return True def getCacheDir(): """return a path to a valid and safe cachedir - only used when not running as root or when --tempcache is set""" uid = os.geteuid() try: usertup = pwd.getpwuid(uid) username = dnf.i18n.ucd(usertup[0]) prefix = '%s-%s-' % (dnf.const.PREFIX, username) except KeyError: prefix = '%s-%s-' % (dnf.const.PREFIX, uid) # check for /var/tmp/prefix-* - dirpath = '%s/%s' % (dnf.const.TMPDIR, prefix) cachedirs = sorted(glob.glob(dirpath)) for thisdir in cachedirs: stats = os.lstat(thisdir) if S_ISDIR(stats[0]) and S_IMODE(stats[0]) == 448 and stats[4] == uid: return thisdir # make the dir (tempfile.mkdtemp()) cachedir = tempfile.mkdtemp(prefix=prefix, dir=dnf.const.TMPDIR) return cachedir def seq_max_split(seq, max_entries): """ Given a seq, split into a list of lists of length max_entries each. """ ret = [] num = len(seq) seq = list(seq) # Trying to use a set/etc. here is bad beg = 0 while num > max_entries: end = beg + max_entries ret.append(seq[beg:end]) beg += max_entries num -= max_entries ret.append(seq[beg:]) return ret def unlink_f(filename): """ Call os.unlink, but don't die if the file isn't there. This is the main difference between "rm -f" and plain "rm". """ try: os.unlink(filename) except OSError as e: if e.errno != errno.ENOENT: raise def stat_f(filename, ignore_EACCES=False): """ Call os.stat(), but don't die if the file isn't there. Returns None. """ try: return os.stat(filename) except OSError as e: if e.errno in (errno.ENOENT, errno.ENOTDIR): return None if ignore_EACCES and e.errno == errno.EACCES: return None raise def _getloginuid(): """ Get the audit-uid/login-uid, if available. os.getuid() is returned instead if there was a problem. Note that no caching is done here. """ # We might normally call audit.audit_getloginuid(), except that requires # importing all of the audit module. And it doesn't work anyway: BZ 518721 try: with open("/proc/self/loginuid") as fo: data = fo.read() return int(data) except (IOError, ValueError): return os.getuid() _cached_getloginuid = None def getloginuid(): """ Get the audit-uid/login-uid, if available. os.getuid() is returned instead if there was a problem. The value is cached, so you don't have to save it. """ global _cached_getloginuid if _cached_getloginuid is None: _cached_getloginuid = _getloginuid() return _cached_getloginuid def decompress(filename, dest=None, check_timestamps=False): """take a filename and decompress it into the same relative location. When the compression type is not recognized (or file is not compressed), the content of the file is copied to the destination""" if dest: out = dest else: out = None dot_pos = filename.rfind('.') if dot_pos > 0: ext = filename[dot_pos:] if ext in ('.zck', '.xz', '.bz2', '.gz', '.lzma', '.zst'): out = filename[:dot_pos] if out is None: raise dnf.exceptions.MiscError("Could not determine destination filename") if check_timestamps: fi = stat_f(filename) fo = stat_f(out) if fi and fo and fo.st_mtime == fi.st_mtime: return out try: # libdnf.utils.decompress either decompress file to the destination or # copy the content if the compression type is not recognized libdnf.utils.decompress(filename, out, 0o644) except RuntimeError as e: raise dnf.exceptions.MiscError(str(e)) if check_timestamps and fi: os.utime(out, (fi.st_mtime, fi.st_mtime)) return out def read_in_items_from_dot_dir(thisglob, line_as_list=True): """ Takes a glob of a dir (like /etc/foo.d/\\.foo) returns a list of all the lines in all the files matching that glob, ignores comments and blank lines, optional paramater 'line_as_list tells whether to treat each line as a space or comma-separated list, defaults to True. """ results = [] for fname in glob.glob(thisglob): with open(fname) as f: for line in f: if re.match(r'\s*(#\|$)', line): continue line = line.rstrip() # no more trailing \n's line = line.lstrip() # be nice if not line: continue if line_as_list: line = line.replace('\n', ' ') line = line.replace(',', ' ') results.extend(line.split()) continue results.append(line) return results	10.99 KB	2022-09-09 11:21
📄 rpmtrans.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 2005 Duke University # Parts Copyright 2007 Red Hat, Inc from __future__ import print_function, absolute_import from __future__ import unicode_literals import libdnf.transaction from dnf.i18n import _, ucd import dnf.callback import dnf.transaction import dnf.util import rpm import os import logging import sys import tempfile import traceback import warnings # TODO: merge w/ libdnf # transaction set states TS_UPDATE = 10 TS_INSTALL = 20 TS_ERASE = 40 TS_OBSOLETED = 50 TS_OBSOLETING = 60 TS_AVAILABLE = 70 TS_UPDATED = 90 TS_FAILED = 100 TS_INSTALL_STATES = [TS_INSTALL, TS_UPDATE, TS_OBSOLETING] TS_REMOVE_STATES = [TS_ERASE, TS_OBSOLETED, TS_UPDATED] RPM_ACTIONS_SET = {libdnf.transaction.TransactionItemAction_INSTALL, libdnf.transaction.TransactionItemAction_DOWNGRADE, libdnf.transaction.TransactionItemAction_DOWNGRADED, libdnf.transaction.TransactionItemAction_OBSOLETE, libdnf.transaction.TransactionItemAction_OBSOLETED, libdnf.transaction.TransactionItemAction_UPGRADE, libdnf.transaction.TransactionItemAction_UPGRADED, libdnf.transaction.TransactionItemAction_REMOVE, libdnf.transaction.TransactionItemAction_REINSTALLED} logger = logging.getLogger('dnf') def _add_deprecated_action(name): """ Wrapper to return a deprecated action constant while printing a deprecation warning. """ @property def _func(self): msg = "%s.%s is deprecated. Use dnf.callback.%s instead." \ % (self.__class__.__name__, name, name) warnings.warn(msg, DeprecationWarning, stacklevel=2) value = getattr(dnf.callback, name) return value return _func class TransactionDisplay(object): # :api def __init__(self): # :api pass # use constants from dnf.callback which are the official API PKG_CLEANUP = _add_deprecated_action("PKG_CLEANUP") PKG_DOWNGRADE = _add_deprecated_action("PKG_DOWNGRADE") PKG_REMOVE = _add_deprecated_action("PKG_REMOVE") PKG_ERASE = PKG_REMOVE PKG_INSTALL = _add_deprecated_action("PKG_INSTALL") PKG_OBSOLETE = _add_deprecated_action("PKG_OBSOLETE") PKG_REINSTALL = _add_deprecated_action("PKG_REINSTALL") PKG_UPGRADE = _add_deprecated_action("PKG_UPGRADE") PKG_VERIFY = _add_deprecated_action("PKG_VERIFY") TRANS_PREPARATION = _add_deprecated_action("TRANS_PREPARATION") PKG_SCRIPTLET = _add_deprecated_action("PKG_SCRIPTLET") TRANS_POST = _add_deprecated_action("TRANS_POST") def progress(self, package, action, ti_done, ti_total, ts_done, ts_total): """Report ongoing progress on a transaction item. :api :param package: a package being processed :param action: the action being performed :param ti_done: number of processed bytes of the transaction item being processed :param ti_total: total number of bytes of the transaction item being processed :param ts_done: number of actions processed in the whole transaction :param ts_total: total number of actions in the whole transaction """ pass def scriptout(self, msgs): """Hook for reporting an rpm scriptlet output. :param msgs: the scriptlet output """ pass def error(self, message): """Report an error that occurred during the transaction. :api""" pass def filelog(self, package, action): # check package object type - if it is a string - just output it """package is the same as in progress() - a package object or simple string action is also the same as in progress()""" pass def verify_tsi_package(self, pkg, count, total): # TODO: replace with verify_tsi? self.progress(pkg, dnf.transaction.PKG_VERIFY, 100, 100, count, total) class ErrorTransactionDisplay(TransactionDisplay): """An RPMTransaction display that prints errors to standard output.""" def error(self, message): super(ErrorTransactionDisplay, self).error(message) dnf.util._terminal_messenger('print', message, sys.stderr) class LoggingTransactionDisplay(TransactionDisplay): ''' Base class for a RPMTransaction display callback class ''' def __init__(self): super(LoggingTransactionDisplay, self).__init__() self.rpm_logger = logging.getLogger('dnf.rpm') def error(self, message): self.rpm_logger.error(message) def filelog(self, package, action): action_str = dnf.transaction.FILE_ACTIONS[action] msg = '%s: %s' % (action_str, package) self.rpm_logger.log(dnf.logging.SUBDEBUG, msg) def scriptout(self, msgs): if msgs: self.rpm_logger.info(ucd(msgs)) class RPMTransaction(object): def __init__(self, base, test=False, displays=()): if not displays: displays = [ErrorTransactionDisplay()] self.displays = displays self.base = base self.test = test # are we a test? self.trans_running = False self.fd = None self.total_actions = 0 self.total_installed = 0 self.complete_actions = 0 self.installed_pkg_names = set() self.total_removed = 0 self._setupOutputLogging(base.conf.rpmverbosity) self._te_list = [] # Index in _te_list of the transaction element being processed (for use # in callbacks) self._te_index = 0 self._tsi_cache = None def _setupOutputLogging(self, rpmverbosity="info"): # UGLY... set up the transaction to record output from scriptlets io_r = tempfile.NamedTemporaryFile() self._readpipe = io_r self._writepipe = open(io_r.name, 'w+b') self.base._ts.setScriptFd(self._writepipe) rpmverbosity = {'critical' : 'crit', 'emergency' : 'emerg', 'error' : 'err', 'information' : 'info', 'warn' : 'warning'}.get(rpmverbosity, rpmverbosity) rpmverbosity = 'RPMLOG_' + rpmverbosity.upper() if not hasattr(rpm, rpmverbosity): rpmverbosity = 'RPMLOG_INFO' rpm.setVerbosity(getattr(rpm, rpmverbosity)) rpm.setLogFile(self._writepipe) def _shutdownOutputLogging(self): # reset rpm bits from recording output rpm.setVerbosity(rpm.RPMLOG_NOTICE) rpm.setLogFile(sys.stderr) try: self._writepipe.close() except: pass def _scriptOutput(self): try: # XXX ugly workaround of problem which started after upgrading glibc # from glibc-2.27-32.fc28.x86_64 to glibc-2.28-9.fc29.x86_64 # After this upgrade nothing is read from _readpipe, so every # posttrans and postun scriptlet output is lost. The problem # only occurs when using dnf-2, dnf-3 is OK. # I did not find the root cause of this error yet. self._readpipe.seek(self._readpipe.tell()) out = self._readpipe.read() if not out: return None return out except IOError: pass def messages(self): messages = self._scriptOutput() if messages: for line in messages.splitlines(): yield ucd(line) def _scriptout(self): msgs = self._scriptOutput() for display in self.displays: display.scriptout(msgs) self.base.history.log_scriptlet_output(msgs) def __del__(self): self._shutdownOutputLogging() def _extract_cbkey(self, cbkey): """Obtain the package related to the calling callback.""" if hasattr(cbkey, "pkg"): tsi = cbkey return [tsi] te = self._te_list[self._te_index] te_nevra = dnf.util._te_nevra(te) if self._tsi_cache: if str(self._tsi_cache[0]) == te_nevra: return self._tsi_cache items = [] for tsi in self.base.transaction: if tsi.action not in RPM_ACTIONS_SET: # skip REINSTALL in order to return REINSTALLED, or REASON_CHANGE to avoid crash continue if str(tsi) == te_nevra: items.append(tsi) if items: self._tsi_cache = items return items raise RuntimeError("TransactionItem not found for key: %s" % cbkey) def callback(self, what, amount, total, key, client_data): try: if isinstance(key, str): key = ucd(key) if what == rpm.RPMCALLBACK_TRANS_START: self._transStart(total) elif what == rpm.RPMCALLBACK_TRANS_STOP: pass elif what == rpm.RPMCALLBACK_TRANS_PROGRESS: self._trans_progress(amount, total) elif what == rpm.RPMCALLBACK_ELEM_PROGRESS: # This callback type is issued every time the next transaction # element is about to be processed by RPM, before any other # callbacks are issued. "amount" carries the index of the element. self._elemProgress(key, amount) elif what == rpm.RPMCALLBACK_INST_OPEN_FILE: return self._instOpenFile(key) elif what == rpm.RPMCALLBACK_INST_CLOSE_FILE: self._instCloseFile(key) elif what == rpm.RPMCALLBACK_INST_START: self._inst_start(key) elif what == rpm.RPMCALLBACK_INST_STOP: self._inst_stop(key) elif what == rpm.RPMCALLBACK_INST_PROGRESS: self._instProgress(amount, total, key) elif what == rpm.RPMCALLBACK_UNINST_START: self._uninst_start(key) elif what == rpm.RPMCALLBACK_UNINST_STOP: self._unInstStop(key) elif what == rpm.RPMCALLBACK_UNINST_PROGRESS: self._uninst_progress(amount, total, key) elif what == rpm.RPMCALLBACK_CPIO_ERROR: self._cpioError(key) elif what == rpm.RPMCALLBACK_UNPACK_ERROR: self._unpackError(key) elif what == rpm.RPMCALLBACK_SCRIPT_ERROR: self._scriptError(amount, total, key) elif what == rpm.RPMCALLBACK_SCRIPT_START: self._script_start(key) elif what == rpm.RPMCALLBACK_SCRIPT_STOP: self._scriptStop() 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 _transStart(self, total): self.total_actions = total if self.test: return self.trans_running = True self._te_list = list(self.base._ts) def _trans_progress(self, amount, total): action = dnf.transaction.TRANS_PREPARATION for display in self.displays: display.progress('', action, amount + 1, total, 1, 1) def _elemProgress(self, key, index): self._te_index = index self.complete_actions += 1 if not self.test: transaction_list = self._extract_cbkey(key) for display in self.displays: display.filelog(transaction_list[0].pkg, transaction_list[0].action) def _instOpenFile(self, key): self.lastmsg = None transaction_list = self._extract_cbkey(key) pkg = transaction_list[0].pkg rpmloc = pkg.localPkg() try: self.fd = open(rpmloc) except IOError as e: for display in self.displays: display.error("Error: Cannot open file %s: %s" % (rpmloc, e)) else: if self.trans_running: self.total_installed += 1 self.installed_pkg_names.add(pkg.name) return self.fd.fileno() def _instCloseFile(self, key): self.fd.close() self.fd = None def _inst_start(self, key): pass def _inst_stop(self, key): if self.test or not self.trans_running: return self._scriptout() if self.complete_actions == self.total_actions: # RPM doesn't explicitly report when post-trans phase starts action = dnf.transaction.TRANS_POST for display in self.displays: display.progress(None, action, None, None, None, None) def _instProgress(self, amount, total, key): transaction_list = self._extract_cbkey(key) pkg = transaction_list[0].pkg action = transaction_list[0].action for display in self.displays: display.progress(pkg, action, amount, total, self.complete_actions, self.total_actions) def _uninst_start(self, key): self.total_removed += 1 def _uninst_progress(self, amount, total, key): transaction_list = self._extract_cbkey(key) pkg = transaction_list[0].pkg action = transaction_list[0].action for display in self.displays: display.progress(pkg, action, amount, total, self.complete_actions, self.total_actions) def _unInstStop(self, key): if self.test: return self._scriptout() def _cpioError(self, key): transaction_list = self._extract_cbkey(key) msg = "Error in cpio payload of rpm package %s" % transaction_list[0].pkg for display in self.displays: display.error(msg) def _unpackError(self, key): transaction_list = self._extract_cbkey(key) msg = "Error unpacking rpm package %s" % transaction_list[0].pkg for display in self.displays: display.error(msg) def _scriptError(self, amount, total, key): # "amount" carries the failed scriptlet tag, # "total" carries fatal/non-fatal status scriptlet_name = rpm.tagnames.get(amount, "<unknown>") transaction_list = self._extract_cbkey(key) name = transaction_list[0].pkg.name msg = ("Error in %s scriptlet in rpm package %s" % (scriptlet_name, name)) for display in self.displays: display.error(msg) def _script_start(self, key): # TODO: this doesn't fit into libdnf TransactionItem use cases action = dnf.transaction.PKG_SCRIPTLET if key is None and self._te_list == []: pkg = 'None' else: transaction_list = self._extract_cbkey(key) pkg = transaction_list[0].pkg complete = self.complete_actions if self.total_actions != 0 and self.complete_actions != 0 \ else 1 total = self.total_actions if self.total_actions != 0 and self.complete_actions != 0 else 1 for display in self.displays: display.progress(pkg, action, 100, 100, complete, total) def _scriptStop(self): self._scriptout() def verify_tsi_package(self, pkg, count, total): for display in self.displays: display.verify_tsi_package(pkg, count, total)	15.63 KB	2022-09-09 11:21

Terminal