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 # # 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 . import transaction from dnf.pycomp import is_py3bytes import dnf.const import dnf.exceptions import rpm # used by ansible (dnf.rpm.rpm.labelCompare in lib/ansible/modules/packaging/os/dnf.py) def detect_releasevers(installroot): # :api """Calculate the release version for the system, including releasever_major and releasever_minor if they are overriden by the system-release-major or system-release-minor provides.""" ts = transaction.initReadOnlyTransaction(root=installroot) ts.pushVSFlags(~(rpm._RPMVSF_NOSIGNATURES \| rpm._RPMVSF_NODIGESTS)) distrover_major_pkg = dnf.const.DISTROVER_MAJOR_PKG distrover_minor_pkg = dnf.const.DISTROVER_MINOR_PKG if dnf.pycomp.PY3: distrover_major_pkg = bytes(distrover_major_pkg, 'utf-8') distrover_minor_pkg = bytes(distrover_minor_pkg, 'utf-8') for distroverpkg in dnf.const.DISTROVERPKG: if dnf.pycomp.PY3: distroverpkg = bytes(distroverpkg, 'utf-8') try: idx = ts.dbMatch('provides', distroverpkg) except (TypeError, rpm.error) as e: raise dnf.exceptions.Error('Error: %s' % str(e)) if not len(idx): continue try: hdr = next(idx) except StopIteration: msg = 'Error: rpmdb failed to list provides. Try: rpm --rebuilddb' raise dnf.exceptions.Error(msg) releasever = hdr['version'] releasever_major = None releasever_minor = None try: try: # header returns bytes -> look for bytes # it may fail because rpm returns a decoded string since 10 Apr 2019 off = hdr[rpm.RPMTAG_PROVIDENAME].index(distroverpkg) except ValueError: # header returns a string -> look for a string off = hdr[rpm.RPMTAG_PROVIDENAME].index(distroverpkg.decode("utf8")) flag = hdr[rpm.RPMTAG_PROVIDEFLAGS][off] ver = hdr[rpm.RPMTAG_PROVIDEVERSION][off] if flag == rpm.RPMSENSE_EQUAL and ver: if hdr['name'] not in (distroverpkg, distroverpkg.decode("utf8")): # override the package version releasever = ver for provide, flag, ver in zip( hdr[rpm.RPMTAG_PROVIDENAME], hdr[rpm.RPMTAG_PROVIDEFLAGS], hdr[rpm.RPMTAG_PROVIDEVERSION]): if isinstance(provide, str): provide = bytes(provide, "utf-8") if provide == distrover_major_pkg and flag == rpm.RPMSENSE_EQUAL and ver: releasever_major = ver if provide == distrover_minor_pkg and flag == rpm.RPMSENSE_EQUAL and ver: releasever_minor = ver except (ValueError, KeyError, IndexError): pass if is_py3bytes(releasever): releasever = str(releasever, "utf-8") if is_py3bytes(releasever_major): releasever_major = str(releasever_major, "utf-8") if is_py3bytes(releasever_minor): releasever_minor = str(releasever_minor, "utf-8") return releasever, releasever_major, releasever_minor return (None, None, None) def detect_releasever(installroot): # :api """Calculate the release version for the system.""" releasever, _, _ = detect_releasevers(installroot) return releasever def _header(path): """Return RPM header of the file.""" ts = transaction.initReadOnlyTransaction() with open(path) as package: fdno = package.fileno() try: hdr = ts.hdrFromFdno(fdno) except rpm.error as e: raise dnf.exceptions.Error("{0}: '{1}'".format(e, path)) return hdr def _invert(dct): return {v: k for k in dct for v in dct[k]} _BASEARCH_MAP = _invert({ 'aarch64': ('aarch64',), 'alpha': ('alpha', 'alphaev4', 'alphaev45', 'alphaev5', 'alphaev56', 'alphaev6', 'alphaev67', 'alphaev68', 'alphaev7', 'alphapca56'), 'arm': ('armv5tejl', 'armv5tel', 'armv5tl', 'armv6l', 'armv7l', 'armv8l'), 'armhfp': ('armv6hl', 'armv7hl', 'armv7hnl', 'armv8hl'), 'i386': ('i386', 'athlon', 'geode', 'i386', 'i486', 'i586', 'i686'), 'ia64': ('ia64',), 'mips': ('mips',), 'mipsel': ('mipsel',), 'mips64': ('mips64',), 'mips64el': ('mips64el',), 'loongarch64': ('loongarch64',), 'noarch': ('noarch',), 'ppc': ('ppc',), 'ppc64': ('ppc64', 'ppc64iseries', 'ppc64p7', 'ppc64pseries'), 'ppc64le': ('ppc64le',), 'riscv32' : ('riscv32',), 'riscv64' : ('riscv64',), 'riscv128' : ('riscv128',), 's390': ('s390',), 's390x': ('s390x',), 'sh3': ('sh3',), 'sh4': ('sh4', 'sh4a'), 'sparc': ('sparc', 'sparc64', 'sparc64v', 'sparcv8', 'sparcv9', 'sparcv9v'), 'x86_64': ('x86_64', 'amd64', 'ia32e'), }) def basearch(arch): # :api return _BASEARCH_MAP[arch] def getheader(rpm_hdr, key): ''' Returns value of rpm_hdr[key] as a string. Rpm has switched from bytes to str and we need to handle both properly. ''' value = rpm_hdr[key] if is_py3bytes(value): value = str(value, "utf-8") return value	6.17 KB	2025-09-22 11:27
📁 __pycache__	-	2026-03-21 15:47
📄 connection.py # connection.py # Maintain RPMDB connections. # # 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 .transaction import initReadOnlyTransaction import dnf.util class RpmConnection(object): def __init__(self, root): self.root = root @property @dnf.util.lazyattr("_readonly_ts") def readonly_ts(self): return initReadOnlyTransaction(self.root)	1.34 KB	2022-09-09 11:21
📄 error.py # error.py # RpmUtilsError # # 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. # class RpmUtilsError(Exception): pass	1.01 KB	2022-09-09 11:21
📄 miscutils.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 2003 Duke University from __future__ import print_function, absolute_import, unicode_literals import os import subprocess import logging from shutil import which from dnf.i18n import _ _logger = logging.getLogger('dnf') _rpmkeys_binary = None def _find_rpmkeys_binary(): global _rpmkeys_binary if _rpmkeys_binary is None: _rpmkeys_binary = which("rpmkeys") _logger.debug(_('Using rpmkeys executable at %s to verify signatures'), _rpmkeys_binary) return _rpmkeys_binary def _process_rpm_output(data): # No signatures or digests = corrupt package. # There is at least one line for -: and another (empty) entry after the # last newline. if len(data) < 3 or data[0] != b'-:' or data[-1]: return 2 seen_sig, missing_key, not_trusted, not_signed = False, False, False, False for i in data[1:-1]: if b': BAD' in i: return 2 elif i.endswith(b': NOKEY'): missing_key = True elif i.endswith(b': NOTTRUSTED'): not_trusted = True elif i.endswith(b': NOTFOUND'): not_signed = True elif not i.endswith(b': OK'): return 2 if not_trusted: return 3 elif missing_key: return 1 elif not_signed: return 4 # we still check return code, so this is safe return 0 def _verifyPackageUsingRpmkeys(package, installroot): rpmkeys_binary = _find_rpmkeys_binary() if rpmkeys_binary is None or not os.path.isfile(rpmkeys_binary): _logger.critical(_('Cannot find rpmkeys executable to verify signatures.')) return 2 # "--define=_pkgverify_level signature" enforces signature checking; # "--define=_pkgverify_flags 0x0" ensures that all signatures are checked. args = ('rpmkeys', '--checksig', '--root', installroot, '--verbose', '--define=_pkgverify_level signature', '--define=_pkgverify_flags 0x0', '-') env = dict(os.environ) env['LC_ALL'] = 'C' with subprocess.Popen( args=args, executable=rpmkeys_binary, env=env, stdout=subprocess.PIPE, cwd='/', stdin=package) as p: data = p.communicate()[0] returncode = p.returncode if type(returncode) is not int: raise AssertionError('Popen set return code to non-int') # rpmkeys can return something other than 0 or 1 in the case of a # fatal error (OOM, abort() called, SIGSEGV, etc) if returncode >= 2 or returncode < 0: return 2 ret = _process_rpm_output(data.split(b'\n')) if ret: return ret return 2 if returncode else 0 def checkSig(ts, package): """Takes a transaction set and a package, check it's sigs, return 0 if they are all fine return 1 if the gpg key can't be found return 2 if the header is in someway damaged return 3 if the key is not trusted return 4 if the pkg is not gpg or pgp signed""" fdno = os.open(package, os.O_RDONLY\|os.O_NOCTTY\|os.O_CLOEXEC) try: value = _verifyPackageUsingRpmkeys(fdno, ts.ts.rootDir) finally: os.close(fdno) return value	3.81 KB	2022-09-09 11:21
📄 transaction.py # # Client code for Update Agent # Copyright (c) 1999-2002 Red Hat, Inc. Distributed under GPL. # # Adrian Likins <alikins@redhat.com> # Some Edits by Seth Vidal <skvidal@phy.duke.edu> # # a couple of classes wrapping up transactions so that we # can share transactions instead of creating new ones all over # from __future__ import absolute_import from __future__ import unicode_literals from dnf.i18n import _ import logging import rpm _logger = logging.getLogger('dnf') read_ts = None ts = None # wrapper/proxy class for rpm.Transaction so we can # instrument it, etc easily class TransactionWrapper(object): def __init__(self, root='/'): self.ts = rpm.TransactionSet(root) self._methods = ['check', 'order', 'addErase', 'addInstall', 'addReinstall', 'run', 'pgpImportPubkey', 'pgpPrtPkts', 'problems', 'setFlags', 'setVSFlags', 'setProbFilter', 'hdrFromFdno', 'next', 'clean'] self.tsflags = [] self.open = True def __del__(self): # Automatically close the rpm transaction when the reference is lost self.close() def close(self): if self.open: self.ts.closeDB() self.ts = None self.open = False def dbMatch(self, args, kwds): if 'patterns' in kwds: patterns = kwds.pop('patterns') else: patterns = [] mi = self.ts.dbMatch(args, **kwds) for (tag, tp, pat) in patterns: mi.pattern(tag, tp, pat) return mi def dbCookie(self): # dbCookie() does not support lazy opening of rpm database. # The following line opens the database if it is not already open. if self.ts.openDB() != 0: _logger.error(_('The openDB() function cannot open rpm database.')) return '' cookie = self.ts.dbCookie() if not cookie: _logger.error(_('The dbCookie() function did not return cookie of rpm database.')) return '' return cookie def __getattr__(self, attr): if attr in self._methods: return self.getMethod(attr) else: raise AttributeError(attr) def __iter__(self): return self.ts def getMethod(self, method): # in theory, we can override this with # profile/etc info return getattr(self.ts, method) # push/pop methods so we don't lose the previous # set value, and we can potentially debug a bit # easier def pushVSFlags(self, flags): self.tsflags.append(flags) self.ts.setVSFlags(self.tsflags[-1]) def addTsFlag(self, flag): curflags = self.ts.setFlags(0) self.ts.setFlags(curflags \| flag) def getTsFlags(self): curflags = self.ts.setFlags(0) self.ts.setFlags(curflags) return curflags def isTsFlagSet(self, flag): val = self.getTsFlags() return bool(flag & val) def setScriptFd(self, fd): self.ts.scriptFd = fd.fileno() def test(self, cb, conf={}): """tests the ts we've setup, takes a callback function and a conf dict for flags and what not""" origflags = self.getTsFlags() self.addTsFlag(rpm.RPMTRANS_FLAG_TEST) # FIXME GARBAGE - remove once this is reimplemented elsewhere # KEEPING FOR API COMPLIANCE ONLY if conf.get('diskspacecheck') == 0: self.ts.setProbFilter(rpm.RPMPROB_FILTER_DISKSPACE) tserrors = self.ts.run(cb.callback, '') self.ts.setFlags(origflags) reserrors = [] if tserrors is not None: for (descr, (etype, mount, need)) in tserrors: reserrors.append(descr) if not reserrors: reserrors.append(_('Errors occurred during test transaction.')) return reserrors def initReadOnlyTransaction(root='/'): read_ts = TransactionWrapper(root=root) read_ts.pushVSFlags((rpm._RPMVSF_NOSIGNATURES\|rpm._RPMVSF_NODIGESTS)) return read_ts	4.27 KB	2022-09-09 11:21

Terminal