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	0 B	2025-05-29 13:37
📁 __pycache__	-	2026-03-21 15:47
📄 capabilities.py from up2date_client import config from up2date_client import up2dateErrors try: # python2 import UserDict except ImportError: # python3 import collections as UserDict import gettext t = gettext.translation('rhn-client-tools', fallback=True) # Python 3 translations don't have a ugettext method if not hasattr(t, 'ugettext'): t.ugettext = t.gettext _ = t.ugettext # a dict with "capability name" as the key, and the version # as the value. neededCaps = {"caneatCheese": {'version':"21"}, "supportsAutoUp2dateOption": {'version': "1"}, "registration.finish_message": {'version': "1"}, "xmlrpc.packages.extended_profile": {'version':"1"}, "registration.delta_packages": {'version':"1"}, "registration.update_contact_info": {'version': "1"}, "registration.extended_update_support": {"version" : "1"}, "registration.smbios": {"version" : "1"}} def parseCap(capstring): value = None caps = capstring.split(',') capslist = [] for cap in caps: try: (key_version, value) = [i.strip() for i in cap.split("=", 1)] except ValueError: # Bad directive: not in 'a = b' format continue # parse out the version # lets give it a shot sans regex's first... (key,version) = key_version.split("(", 1) # just to be paranoid if version[-1] != ")": print("something broke in parsing the capabilited headers") #FIXME: raise an approriate exception here... # trim off the trailing paren version = version[:-1] data = {'version': version, 'value': value} capslist.append((key, data)) return capslist class Capabilities(UserDict.UserDict): def __init__(self): UserDict.UserDict.__init__(self) self.missingCaps = {} #self.populate() # self.validate() self.neededCaps = neededCaps self.cfg = config.initUp2dateConfig() def populate(self, headers): for key, val in headers.items(): if key.lower() == "x-rhn-server-capability": capslist = parseCap(val) for (cap,data) in capslist: self.data[cap] = data def parseCapVersion(self, versionString): index = versionString.find('-') # version of "-" is bogus, ditto for "1-" if index > 0: rng = versionString.split("-") start = rng[0] end = rng[1] versions = range(int(start), int(end)+1) return versions vers = versionString.split(':') if len(vers) > 1: versions = [int(a) for a in vers] return versions return [int(versionString)] def validateCap(self, cap, capvalue): if not cap in self.data: errstr = _("This client requires the server to support %s, which the current " \ "server does not support") % cap self.missingCaps[cap] = None else: data = self.data[cap] # DOES the server have the version we need if int(capvalue['version']) not in self.parseCapVersion(data['version']): self.missingCaps[cap] = self.neededCaps[cap] def validate(self): for key in self.neededCaps.keys(): self.validateCap(key, self.neededCaps[key]) self.workaroundMissingCaps() def setConfig(self, key, configItem): if key in self.tmpCaps: self.cfg[configItem] = 0 del self.tmpCaps[key] else: self.cfg[configItem] = 1 def workaroundMissingCaps(self): # if we have caps that we know we want, but we can # can work around, setup config variables here so # that we know to do just that self.tmpCaps = self.missingCaps # this is an example of how to work around it key = 'caneatCheese' if key in self.tmpCaps: # do whatevers needed to workaround del self.tmpCaps[key] else: # we support this, set a config option to # indicate that possibly pass # dict of key to configItem, and the config item that # corresponds with it capsConfigMap = {'supportsAutoUp2dateOption': 'supportsAutoUp2dateOption', 'registration.finish_message': 'supportsFinishMessage', "registration.update_contact_info" : 'supportsUpdateContactInfo', "registration.delta_packages" : 'supportsDeltaPackages', "xmlrpc.packages.extended_profile" : 'supportsExtendedPackageProfile', "registration.extended_update_support" : "supportsEUS", "registration.smbios" : "supportsSMBIOS"} for key in capsConfigMap.keys(): self.setConfig(key, capsConfigMap[key]) # if we want to blow up on missing caps we cant eat around missingCaps = [] wrongVersionCaps = [] if len(self.tmpCaps): for cap in self.tmpCaps: capInfo = self.tmpCaps[cap] if capInfo == None: # it's completly mssing missingCaps.append((cap, capInfo)) else: wrongVersionCaps.append((cap, capInfo)) errString = "" errorList = [] if len(wrongVersionCaps): for (cap, capInfo) in wrongVersionCaps: errString = errString + "Needs %s of version: %s but server has version: %s\n" % (cap, capInfo['version'], self.data[cap]['version']) errorList.append({"capName":cap, "capInfo":capInfo, "serverVersion":self.data[cap]}) if len(missingCaps): for (cap, capInfo) in missingCaps: errString = errString + "Needs %s but server does not support that capability\n" % (cap) errorList.append({"capName":cap, "capInfo":capInfo, "serverVersion":""}) if len(errString): raise up2dateErrors.ServerCapabilityError(errString, errorList) def hasCapability(self, capability, version=None): """Checks if the server supports a capability and optionally a version. Returns True or False. This complements the neededCaps mechanism provided by this module. Using hasCapability makes it easier to do something only if the server supports it or to put workaround code in the user of this class. The neededCaps mechanism makes it easier to put workaround code in this module, which makes sense if it is to be shared. 'capability' should be a string such as 'registration.foobar'. It can be a capability in 'neededCaps' above or one that isn't there. 'version' can be a string (where isdigit() is True) or an int. """ assert version is None or str(version).isdigit() if not capability in self.data: return False if version: data = self.data[capability] if int(version) not in self.parseCapVersion(data['version']): return False return True	7.28 KB	2025-05-29 13:37
📄 clientCaps.py # a dict with "capability name" as the key, and the version # as the value. import glob import os from up2date_client.capabilities import parseCap try: # python2 import UserDict except ImportError: # python3 import collections as UserDict class ClientCapabilities(UserDict.UserDict): def __init__(self): UserDict.UserDict.__init__(self) self.populate() def populate(self, capsToPopulate=None): # FIXME: at some point, this will be # intelligently populated... localcaps = { "caneatCheese":{'version':1, 'value': 1} } if capsToPopulate: localcaps = capsToPopulate self.data = localcaps def headerFormat(self): headerList = [] for key in self.data.keys(): headerName = "X-RHN-Client-Capability" value = "%s(%s)=%s" % (key, self.data[key]['version'], self.data[key]['value']) headerList.append((headerName, value)) return headerList caps = ClientCapabilities() def loadLocalCaps(capsDir = "/etc/sysconfig/rhn/clientCaps.d"): capsFiles = glob.glob("%s/*" % capsDir) for capsFile in capsFiles: if os.path.isdir(capsFile): continue if not os.access(capsFile, os.R_OK): continue fd = open(capsFile, "r") for line in fd.readlines(): line = line.strip() if not line or line[0] == "#": continue caplist = parseCap(line) for (cap,data) in caplist: caps.data[cap] = data # print(caps.data) loadLocalCaps() # register local caps we require. def registerCap(cap, data): caps.data[cap] = data # figure out something pretty here registerCap("packages.runTransaction", {'version':'1', 'value':'1'}) registerCap("packages.rollBack", {'version':'1', 'value':'1'}) registerCap("packages.verify", {'version':'1', 'value':'1'}) registerCap("packages.extended_profile", {'version':'2', 'value':'1'}) registerCap("reboot.reboot", {'version':'1', 'value':'1'}) registerCap("packages.update", {'version':'2', 'value':'2'})	2.16 KB	2025-05-29 13:37
📄 clpwd.py import pwd import os class ClPwd: class NoSuchUserException(Exception): def __init__(self, user): Exception.__init__(self, "No such user (%s)" % (user,)) def __init__(self, min_uid = None): self._user_key_map = {} self._uid_key_map = {} self._user_full_map = {} self._uid_full_map = {} if min_uid is None: self._min_uid = self.get_sys_min_uid(500) else: self._min_uid = min_uid def get_user_dict(self): self._load_passwd_database() return self._user_key_map def get_uid_dict(self): self._load_passwd_database() return self._uid_key_map def get_user_full_dict(self): self._load_passwd_database() return self._user_full_map def get_uid_full_dict(self): self._load_passwd_database() return self._uid_full_map def get_pw_by_name(self, user): """ Return pw_entry for user """ try: return self.get_user_full_dict()[user] except KeyError: raise ClPwd.NoSuchUserException(user) def get_pw_by_uid(self, uid): """ Return list of passwd entries for uid """ try: return self.get_uid_full_dict()[uid] except KeyError: raise ClPwd.NoSuchUserException(uid) def get_uid(self, user): """ Returns uid for user """ try: return self.get_user_full_dict()[user].pw_uid except KeyError: raise ClPwd.NoSuchUserException(user) def get_homedir(self, user): """ Returns homedir for a user @param user: string @return: string """ try: return self.get_user_full_dict()[user].pw_dir except KeyError: raise ClPwd.NoSuchUserException(user) def _load_passwd_database(self): """ Loads the passwd database and fills user_to_uid and user_to_homedir maps """ if not self._uid_full_map: for entry in pwd.getpwall(): self._user_full_map[entry.pw_name] = entry if entry.pw_uid not in self._uid_full_map: self._uid_full_map[entry.pw_uid] = [] self._uid_full_map[entry.pw_uid].append(entry) if entry.pw_uid >= self._min_uid: self._user_key_map[entry.pw_name] = entry if entry.pw_uid not in self._uid_key_map: self._uid_key_map[entry.pw_uid] = [] self._uid_key_map[entry.pw_uid].append(entry) def get_names(self, uid): """ Return names of users with uid specified @param uid: int @return: list of strings """ try: entries = self.get_uid_full_dict()[uid] except KeyError: raise ClPwd.NoSuchUserException(uid) return [entry.pw_name for entry in entries] def get_sys_min_uid(self, def_min_uid = 500): """ Return system defined MIN_UID from /etc/login.def or def_min_uid @param def_min_uid: int @return: MIN_UID: int """ LOGIN_DEF_FILE = '/etc/login.defs' if (os.path.exists(LOGIN_DEF_FILE)): with open(LOGIN_DEF_FILE, 'r') as lines: for line in lines: if line.startswith('UID_MIN'): try: return int(line.split('UID_MIN')[1].strip()) except ValueError: pass return def_min_uid	3.56 KB	2025-05-29 13:37
📄 config.py # This file is a portion of the Red Hat Update Agent # Copyright (c) 1999--2020 Red Hat, Inc. Distributed under GPL # # Authors: # Cristian Gafton <gafton@redhat.com> # Adrian Likins <alikins@redhat.com> # """ This module includes the Config and Up2date Config classes use by the up2date agent to hold config info. """ import os import sys import locale import requests from rhn.connections import idn_ascii_to_puny, idn_puny_to_unicode from rhn.i18n import ustr, sstr try: # python2 from urlparse import urlsplit, urlunsplit except ImportError: # python3 from urllib.parse import urlsplit, urlunsplit import gettext t = gettext.translation("rhn-client-tools", fallback=True) # Python 3 translations don't have a ugettext method if not hasattr(t, "ugettext"): t.ugettext = t.gettext _ = t.ugettext # XXX: This could be moved in a more "static" location if it is too # much of an eye sore Defaults = { "enableProxy": ("Use a HTTP Proxy", 0), "serverURL": ("Remote server URL", "https://xmlrpc.cln.cloudlinux.com/XMLRPC/"), "serverURLipv6": ("Remote server URL for access over IPv6", "https://ipv6.xmlrpc.cln.cloudlinux.com/XMLRPC/"), "mirrorURL": ("Mirror list URL", "https://repo.cloudlinux.com/cloudlinux/mirrorlists/cln-mirrors"), "debug": ("Whether or not debugging is enabled", 0), "systemIdPath": ("Location of system id", "/etc/sysconfig/rhn/systemid"), "versionOverride": ("Override the automatically determined system version", ""), "httpProxy": ("HTTP proxy in host:port format, e.g. squid.example.com:3128", ""), "proxyUser": ("The username for an authenticated proxy", ""), "proxyPassword": ("The password to use for an authenticated proxy", ""), "enableProxyAuth": ("To use an authenticated proxy or not", 0), "networkRetries": ("Number of attempts to make at network connections before giving up", 1), "sslCACert": ("The CA cert used to verify the ssl server", "/usr/share/rhn/CLN-ORG-TRUSTED-SSL-CERT"), "noReboot": ("Disable the reboot action", 0), "disallowConfChanges": ( "Config options that can not be overwritten by a config update action", ["sslCACert", "serverURL", "disallowConfChanges", "noReboot"], ), } FileOptions = [ "systemIdPath", "sslCACert", "tmpDir", ] # a peristent configuration storage class class ConfigFile: "class for handling persistent config options for the client" def __init__(self, filename=None): self.dict = {} self.fileName = filename if self.fileName: self.load() def load(self, filename=None): if filename: self.fileName = filename if self.fileName is None: return if not os.access(self.fileName, os.R_OK): # print("warning: can't access %s" % self.fileName) return f = open(self.fileName, "r") multiline = "" for line in f.readlines(): # strip comments if line.find("#") == 0: continue line = multiline + line.strip() if not line: continue # if line ends in '\', append the next line before parsing if line[-1] == "\\": multiline = line[:-1].strip() continue else: multiline = "" split = line.split("=", 1) if len(split) != 2: # not in 'a = b' format. we should log this # or maybe error. continue key = split[0].strip() value = ustr(split[1].strip()) # decode a comment line comment = None pos = key.find("[comment]") if pos != -1: key = key[:pos] comment = value value = None # figure out if we need to parse the value further if value: # possibly split value into a list values = value.split(";") if key in ["proxyUser", "proxyPassword"]: value = sstr(value.encode(locale.getpreferredencoding())) elif len(values) == 1: try: value = int(value) except ValueError: pass elif values[0] == "": value = [] else: # there could be whitespace between the values on # one line, let's strip it out value = [val.strip() for val in values if val.strip()] # now insert the (comment, value) in the dictionary newval = (comment, value) if key in self.dict: # do we need to update newval = self.dict[key] if comment is not None: # override comment newval = (comment, newval[1]) if value is not None: # override value newval = (newval[0], value) self.dict[key] = newval f.close() def save(self): if self.fileName is None: return # this really shouldn't happen, since it means that the # /etc/sysconfig/rhn directory doesn't exist, which is way broken # and note the attempted fix breaks useage of this by the applet # since it reuses this code to create its config file, and therefore # tries to makedirs() the users home dir again (with a specific perms) # and fails (see #130391) if not os.access(self.fileName, os.R_OK): if not os.access(os.path.dirname(self.fileName), os.R_OK): print(_("%s was not found" % os.path.dirname(self.fileName))) return f = open(self.fileName + ".new", "w") os.chmod(self.fileName + ".new", int("0644", 8)) f.write("# Automatically generated Red Hat Update Agent config file, do not edit.\n") f.write("# Format: 1.0\n") f.write("") for key in self.dict.keys(): (comment, value) = self.dict[key] f.write(sstr("%s[comment]=%s\n" % (key, comment))) if type(value) != type([]): value = [value] if key in FileOptions: value = map(os.path.abspath, value) f.write(sstr("%s=%s\n" % (key, ";".join(map(str, value))))) f.write("\n") f.close() os.rename(self.fileName + ".new", self.fileName) # dictionary interface def __contains__(self, name): return name in self.dict def has_key(self, name): # obsoleted, left for compatibility with older python return name in self def keys(self): return self.dict.keys() def values(self): return [a[1] for a in self.dict.values()] def update(self, dict): self.dict.update(dict) # we return None when we reference an invalid key instead of # raising an exception def __getitem__(self, name): if name in self.dict: return self.dict[name][1] return None def __setitem__(self, name, value): if name in self.dict: val = self.dict[name] else: val = (None, None) self.dict[name] = (val[0], value) # we might need to expose the comments... def info(self, name): if name in self.dict: return self.dict[name][0] return "" # a superclass for the ConfigFile that also handles runtime-only # config values class Config: def __init__(self, filename=None): self.stored = ConfigFile() self.stored.update(Defaults) if filename: self.stored.load(filename) self.runtime = {} # classic dictionary interface: we prefer values from the runtime # dictionary over the ones from the stored config def __contains__(self, name): if name in self.runtime: return True if name in self.stored: return True return False def has_key(self, name): # obsoleted, left for compatibility with older python return name in self def keys(self): ret = list(self.runtime.keys()) for k in self.stored.keys(): if k not in ret: ret.append(k) return ret def values(self): ret = [] for k in self.keys(): ret.append(self.__getitem__(k)) return ret def items(self): ret = [] for k in self.keys(): ret.append((k, self.__getitem__(k))) return ret def __len__(self): return len(self.keys()) def __setitem__(self, name, value): self.runtime[name] = value # we return None when nothing is found instead of raising and exception def __getitem__(self, name): if name in self.runtime: return self.runtime[name] if name in self.stored: return self.stored[name] return None # These function expose access to the peristent storage for # updates and saves def info(self, name): # retrieve comments return self.stored.info(name) def save(self): self.stored.save() def load(self, filename): self.stored.load(filename) # make sure the runtime cache is not polluted for k in self.stored.keys(): if k not in self.runtime: continue # allow this one to pass through del self.runtime[k] # save straight in the persistent storage def set(self, name, value): self.stored[name] = value # clean up the runtime cache if name in self.runtime: del self.runtime[name] def getProxySetting(): """returns proxy string in format hostname:port hostname is converted to Punycode (RFC3492) if needed """ cfg = initUp2dateConfig() proxy = None proxyHost = cfg["httpProxy"] if proxyHost: if proxyHost[:7] == "http://": proxyHost = proxyHost[7:] parts = proxyHost.split(":") parts[0] = str(idn_ascii_to_puny(parts[0])) proxy = ":".join(parts) return proxy def convert_url_to_puny(url): """returns url where hostname is converted to Punycode (RFC3492)""" s = urlsplit(url) return sstr(urlunsplit((s[0], ustr(idn_ascii_to_puny(s[1])), s[2], s[3], s[4]))) def convert_url_from_puny(url): """returns url where hostname is converted from Punycode (RFC3492). Returns unicode string.""" s = urlsplit(url) return ustr(urlunsplit((s[0], idn_puny_to_unicode(s[1]), s[2], s[3], s[4]))) def getServerlURLFromMirror(): url = cfg["mirrorURL"] if url is None: url = "https://repo.cloudlinux.com/cloudlinux/mirrorlists/cln-mirrors" if url.startswith("file://"): with open(url.replace("file://", ""), "r") as mirrorlist: mirrors = map(str.strip, mirrorlist.readlines()) return [convert_url_to_puny(mirror) for mirror in mirrors if mirror] request = requests.get(url) return [convert_url_to_puny(mirror) for mirror in request.text.split("\n") if mirror] def processServerURL(serverUrl): """ Internal function to process server URL to Punycode format. Processes both single URLs and lists of URLs. :param serverUrl: URL or list of URLs to process. :return: List of processed URLs in Punycode format. """ if isinstance(serverUrl, list): return [convert_url_to_puny(i) for i in serverUrl] else: return [convert_url_to_puny(serverUrl)] def getServerURLPair(registration=False): """ Return a pair of server URLs (primary and fallback) based on the preferred interface. :param registration: If True, it indicates that the URLs are being fetched for registration purposes. If that is the case, the function will always use IPv4 URLs with IPv6 as fallback, regardless of the preferred_interface setting. Assumed false by default, expected to be specified explicitly when needed. :return: Pair of server URL configs, first is the primary, second is the fallback. """ cfg = initUp2dateConfig() ipv4_url = processServerURL(cfg["serverURL"]) ipv6_url = processServerURL(cfg["serverURLipv6"]) # Note that we don't consider the case where there's no IPv6 URL configured, # we expect that the configuration will always have both URLs set. ipv4_primary_pair = (ipv4_url, ipv6_url) ipv6_primary_pair = (ipv6_url, ipv4_url) # If registration is True, we always use IPv4 URLs with IPv6 as fallback if registration: return ipv4_primary_pair # Import here to avoid circular import from up2date_client import up2dateAuth # Check if preferred_interface is set to IPv6 preferred_interface = up2dateAuth.getPreferredInterface() # If IPv6 is preferred, use it if preferred_interface == "IPv6" and ipv6_url: return ipv6_primary_pair # Otherwise use the normal URL return ipv4_primary_pair def getServerURL(registration=False): """ Return the primary server URL from config based on preferred_interface. If preferred_interface=IPv6 in system_id, returns serverURLipv6 if available. Otherwise returns normal serverURL. Note: the config may have one value or a list of values, but this function always returns a list. :return: List of server URLs with hostnames converted to Punycode. """ return getServerURLPair(registration=registration)[0] # Return the primary URL def getFallbackServerURL(registration=False): """ Determine the fallback server URL from system_id. The fallback server URL is the one that is not set as preferred_interface. If preferred_interface is IPv6, it returns serverURL (i.e. IPv4 host), otherwise serverURLipv6. """ return getServerURLPair(registration=registration)[1] # Return the fallback URL def setServerURL(serverURL): """Set serverURL in config""" cfg = initUp2dateConfig() cfg.set("serverURL", serverURL) def setServerURLipv6(serverURL): """Set serverURLipv6 in config""" cfg = initUp2dateConfig() cfg.set("serverURLipv6", serverURL) def setSSLCACert(sslCACert): """Set sslCACert in config""" cfg = initUp2dateConfig() cfg.set("sslCACert", sslCACert) def initUp2dateConfig(cfg_file="/etc/sysconfig/rhn/up2date"): """This function is the right way to get at the up2date config.""" global cfg try: cfg except NameError: cfg = None if cfg == None: cfg = Config(cfg_file) cfg["isatty"] = False if sys.stdout.isatty(): cfg["isatty"] = True return cfg	14.46 KB	2025-05-29 13:37
📄 debUtils.py # Client code for Update Agent # Copyright (c) 2011--2016 Red Hat, Inc. Distributed under GPLv2. # # Author: Simon Lukasik # Lukas Durfina # import os import apt import gettext t = gettext.translation('rhn-client-tools', fallback=True) # Python 3 translations don't have a ugettext method if not hasattr(t, 'ugettext'): t.ugettext = t.gettext _ = t.ugettext # FIXME: After Debian bug 187019 is resolved def verifyPackages(packages): cache = apt.Cache() missing_packages = [] for package in packages: pkg = cache[package[0]] if pkg == None or not pkg.is_installed: missing_packages.append(package) return [], missing_packages def parseVRE(version): epoch = '' release = 'X' if version.find(':') != -1: epoch, version = version.split(':') if version.find('-') != -1: tmp = version.split('-') version = '-'.join(tmp[:-1]) release = tmp[-1] return version, release, epoch def installTime(pkg_name, pkg_arch): dir = '/var/lib/dpkg/info' files = [ '%s.list' % pkg_name, '%s:%s.list' % (pkg_name, pkg_arch) ] # In edge cases, pkg_name can include the arch but the .list file does not if ':' in pkg_name: files.append('%s.list' % (pkg_name[:pkg_name.index(':')])) for f in files: path = os.path.join(dir,f) if os.path.isfile(path): return os.path.getmtime(path) return 0 #FIXME: Using Apt cache might not be an ultimate solution. # It could be better to parse /var/lib/dpkg/status manually. # Apt cache might not contain all the packages. def getInstalledPackageList(msgCallback = None, progressCallback = None, getArch=None, getInfo = None): """ Return list of packages. Package is dict with following keys: name, epoch, version, release and optionaly arch. """ if msgCallback != None: msgCallback(_("Getting list of packages installed on the system")) cache = apt.Cache() total = 0 for pkg in cache: if pkg.installed != None: total += 1 count = 0 pkg_list = [] for pkg in cache: if pkg.installed == None: continue version, release, epoch = parseVRE(pkg.installed.version) package = { 'name': pkg.name, 'epoch': epoch, 'version': version, 'release': release, 'arch': pkg.installed.architecture + '-deb', 'installtime': installTime(pkg.name, pkg.installed.architecture) } pkg_list.append(package) if progressCallback != None: progressCallback(count, total) count = count + 1 pkg_list.sort(key=lambda package: package['version']) return pkg_list def setDebugVerbosity(): pass	2.76 KB	2025-05-29 13:37
📄 getMethod.py # Retrieve action method name given queued action information. # # Client code for Update Agent # Copyright (c) 1999--2016 Red Hat, Inc. Distributed under GPLv2. # # An allowable xmlrpc method is retrieved given a base location, a # hierarchical route to the class/module, and method name. # import os import string import sys from rhn.tb import raise_with_tb try: # python2 from types import ClassType except ImportError: # python3 ClassType = type class GetMethodException(Exception): """Exception class""" pass def sanity(methodNameComps): #""" # Verifies if all the components have proper names #""" # Allowed characters in each string alpha = string.ascii_lowercase + string.ascii_uppercase allowedChars = alpha + string.digits + '_' for comp in methodNameComps: if not len(comp): raise GetMethodException("Empty method component") for c in comp: if c not in allowedChars: raise GetMethodException( "Invalid character '%s' in the method name" % c) # Can only begin with a letter if comp[0] not in alpha: raise GetMethodException( "Method names should start with an alphabetic character") def getMethod(methodName, baseClass): #""" #Retreive method given methodName, path to base of tree, and class/module #route/label. #""" # First split the method name methodNameComps = baseClass.split('.') + methodName.split('.') # Sanity checks sanity(methodNameComps) # Look for the module, start with the most specific for index in range(len(methodNameComps), 0, -1): modulename = '.'.join(methodNameComps[:index]) try: actions = __import__(modulename) except ImportError: # does not exist, try next one continue except Exception: raise_with_tb(GetMethodException("Could not import module %s" % modulename)) # found one, skip the rest break else: # no module found. die raise GetMethodException("Action %s could not be imported" % methodName) # The position of the file fIndex = index className = actions # Iterate through the list of components and try to load that specific # module/method for index in range(1, len(methodNameComps)): comp = methodNameComps[index] if index < fIndex: # This is a directory or a file we have to load if not hasattr(className, comp): # Hmmm... Not there raise GetMethodException("Class %s has no attribute %s" % ( '.'.join(methodNameComps[:index]), comp)) className = getattr(className, comp) #print(type(className)) continue # A file or method # We look for the special __rhnexport__ array if not hasattr(className, '__rhnexport__'): raise GetMethodException("Class %s is not RHN-compliant" % \ '.'.join(methodNameComps[:index])) export = getattr(className, '__rhnexport__') if comp not in export: raise GetMethodException("Class %s does not export '%s'" % ( '.'.join(methodNameComps[:index]), comp)) className = getattr(className, comp) if type(className) is ClassType: # Try to instantiate it className = className() #print(type(className)) return className #----------------------------------------------------------------------------- if __name__ == '__main__': # Two valid ones and a bogus one methods = [ 'a.b.c.d.e.f', 'a.b.c.d.e.foo.h', 'a.b.c.d.e.g.h', 'a.b.d.d.e.g.h', 'a.b.d.d._e.g.h', 'a.b.d.d.e_.g.h', 'a.b.d.d.e-.g.h', 'a.b.d.d..g.h', ] for m in methods: print("----Running method %s: " % m) try: method = getMethod(m, 'Actions') except GetMethodException: e = sys.exc_info()[1] print("Error getting the method %s: %s" % (m, ''.join(map(str, e.args)))) else: method() #-----------------------------------------------------------------------------	4.2 KB	2025-05-29 13:37
📄 haltree.py # HalTree Purpose: # # HalTree is a way to organize the mess of data you get from hal. In general, # if you want to get all the information about every device in the system, you # end up with a list of dicts, where each dict contains the property name/values # for a device. This list isn't very useful as the hal data is actually # organized into a tree. For example, you have the computer as the head, then # there may be a scsi card plugged in. That in turn will have scsi channels # and luns, which scsi devices may be connected to. So this module will help # you reorganize your hal data back to the way they were intended. # # HalTree Usage: # # The tree gets built one device at a time. Once you've created a HalTree # object, devices get added to the tree with HalTree.add(hw_dev_dict). The # devices can be added in any particular order, and the tree gets properly # structured as the devices get added. But the tree structure isn't likely # to be ready until all the devices have been added. Those devices without a # parent get stuck in the no_parent_yet list. # # When a device gets added, it is no longer a plain dict. It is stored in a # HalDevice. The original dict can be found in HalDevice.properties. try: # python2 from types import StringType, IntType except ImportError: # python3 StringType = bytes IntType = int class HalDevice: "An object containing its udi, properties and children" def __init__ (self, properties): self.udi = properties['info.udi'] self.properties = properties self.children = [] self.classification = None if 'info.parent' in properties: self.parent_udi = properties['info.parent'] else: self.parent_udi = None self.parent = None def print_properties(self): print(self.udi, ":") for property, value in self.properties.items(): print(" ", property," ==> ", value) class HalTree: def __init__ (self): self.head = None self.no_parent_yet = [] def add(self, hal_device): if hal_device.parent_udi: parent = self.__find_node(hal_device.parent_udi) if parent: parent.children.append(hal_device) hal_device.parent = parent else: #parent isn't in the main tree yet, stick it in waiting self.no_parent_yet.append(hal_device) else: #if it doesn't have a parent, it must be the head 'computer' self.head = hal_device #check to see if there are any children waiting for this dev self.__get_lost_children(hal_device) def __get_lost_children(self, hal_device): found_list = [] indexes = [] no_parent_yet_copy = self.no_parent_yet[:] for dev in no_parent_yet_copy: if dev.parent_udi == hal_device.udi: dev.parent = hal_device hal_device.children.append(dev) self.no_parent_yet.remove(dev) def __find_node(self, udi): """ This takes a node in the HalDevice tree and returns the HalDevice with the given udi. """ if self.head: node = HalTree.__find_node_worker(self.head, udi) if node: return node for node in self.no_parent_yet: found_node = HalTree.__find_node_worker(node, udi) if found_node: return found_node return None @staticmethod def __find_node_worker(node, udi): if node.udi == udi: return node for device in node.children: res = HalTree.__find_node_worker(device, udi) if res: return res return None def print_tree(self): self.__print_dev_tree(self.head, "") def __print_dev_tree(self, node, indent): print(indent, node.udi) print(indent, "CLASS:", node.classification) for name, property in node.properties.items(): if (type(property) == StringType): if property.isdigit(): print(indent + " ", "%-20s ==> %s" % (name, hex(int(property)))) else: print(indent + " ", "%-20s ==> %s" % (name, property)) elif (type(property) == IntType): print(indent + " ", "%-20s ==> %s" % (name, hex(int(property)))) else: print(indent + " ", "%-20s ==> %s" % (name, property)) print for child in node.children: self.__print_dev_tree(child, indent + " ")	4.53 KB	2025-05-29 13:37
📄 hardware.py # # Copyright (c) 1999--2018 Red Hat, Inc. # # This software is licensed to you under the GNU General Public License, # version 2 (GPLv2). There is NO WARRANTY for this software, express or # implied, including the implied warranties of MERCHANTABILITY or FITNESS # FOR A PARTICULAR PURPOSE. You should have received a copy of GPLv2 # along with this software; if not, see # http://www.gnu.org/licenses/old-licenses/gpl-2.0.txt. # # Red Hat trademarks are not licensed under GPLv2. No permission is # granted to use or replicate Red Hat trademarks that are incorporated # in this software or its documentation. # # This thing gets the hardware configuraion out of a system """Used to read hardware info from kudzu, /proc, etc""" from socket import gethostname, getaddrinfo, AF_INET, AF_INET6 import socket import re import os import sys from up2date_client import config from up2date_client import rhnserver from up2date_client import up2dateUtils from rhn.i18n import ustr try: long except NameError: # long is not defined in python3 long = int # network module for python 2 try: import ethtool ethtool_present = True except ImportError: ethtool_present = False # network module for python3 (Fedora >= 23) try: import netifaces import ipaddress netifaces_present = True except ImportError: netifaces_present = False import gettext t = gettext.translation('rhn-client-tools', fallback=True) # Python 3 translations don't have a ugettext method if not hasattr(t, 'ugettext'): t.ugettext = t.gettext _ = t.ugettext import dbus with up2dateUtils.suppress_errors(['Failed to save log entry', 'SMBIOS.: entry point at']): import dmidecode from up2date_client import up2dateLog try: from up2date_client.hardware_udev import get_devices, get_computer_info using_gudev = 1 except ImportError: try: # F13 and EL6 from up2date_client.hardware_gudev import get_devices, get_computer_info using_gudev = 1 except ImportError: from up2date_client.hardware_hal import check_hal_dbus_status, get_hal_computer, read_hal using_gudev = 0 # Some systems don't have the _locale module installed try: import locale except ImportError: locale = None sys.path.append("/usr/share/rhsm") try: from subscription_manager.hwprobe import Hardware as SubManHardware subscription_manager_available = True except ImportError: subscription_manager_available = False # this does not change, we can cache it _dmi_data = None _dmi_not_available = 0 def dmi_warnings(): if not hasattr(dmidecode, 'get_warnings'): return None return dmidecode.get_warnings() dmi_warn = dmi_warnings() if dmi_warn: dmidecode.clear_warnings() log = up2dateLog.initLog() log.log_debug("Warnings collected during dmidecode import: %s" % dmi_warn) def _initialize_dmi_data(): """ Initialize _dmi_data unless it already exist and returns it """ global _dmi_data, _dmi_not_available if _dmi_data is None: if _dmi_not_available: # do not try to initialize it again and again if not available return None else : dmixml = dmidecode.dmidecodeXML() dmixml.SetResultType(dmidecode.DMIXML_DOC) # Get all the DMI data and prepare a XPath context try: data = dmixml.QuerySection('all') dmi_warn = dmi_warnings() if dmi_warn: dmidecode.clear_warnings() log = up2dateLog.initLog() log.log_debug("dmidecode warnings: %s" % dmi_warn) except: # DMI decode FAIL, this can happend e.g in PV guest _dmi_not_available = 1 dmi_warn = dmi_warnings() if dmi_warn: dmidecode.clear_warnings() return None _dmi_data = data.xpathNewContext() return _dmi_data def get_dmi_data(path): """ Fetch DMI data from given section using given path. If data could not be retrieved, returns empty string. General method and should not be used outside of this module. """ dmi_data = _initialize_dmi_data() if dmi_data is None: return '' data = dmi_data.xpathEval(path) if data != []: return data[0].content else: # The path do not exist return '' def dmi_vendor(): """ Return Vendor from dmidecode bios information. If this value could not be fetch, returns empty string. """ return get_dmi_data('/dmidecode/BIOSinfo/Vendor') def dmi_system_uuid(): """ Return UUID from dmidecode system information. If this value could not be fetch, returns empty string. """ # if guest was created manualy it can have empty UUID, in this # case dmidecode set attribute unavailable to 1 uuid = get_dmi_data("/dmidecode/SystemInfo/SystemUUID[not(@unavailable='1')]") if not uuid: uuid = '' return uuid def read_installinfo(): if not os.access("/etc/sysconfig/installinfo", os.R_OK): return {} installinfo = open("/etc/sysconfig/installinfo", "r").readlines() installdict = {} installdict['class'] = "INSTALLINFO" for info in installinfo: if not len(info): continue vals = info.split('=') if len(vals) <= 1: continue strippedstring = vals[0].strip() vals[0] = strippedstring installdict[vals[0]] = ''.join(vals[1:]).strip() return installdict def cpu_count(): """ returns number of CPU in system Beware that it can be different from number of active CPU (e.g. on s390x architecture """ try: cpu_dir = os.listdir('/sys/devices/system/cpu/') except OSError: cpu_dir = [] re_cpu = re.compile(r"^cpu[0-9]+$") return len([i for i in cpu_dir if re_cpu.match(i)]) # get the number of sockets available on this machine def __get_number_sockets(): try: if subscription_manager_available: return SubManHardware().getCpuInfo()['cpu.cpu_socket(s)'] except: pass # something went wrong, let's figure it out ourselves number_sockets = 0 # Try lscpu command if available if os.access("/usr/bin/lscpu", os.X_OK): try: lines = os.popen("/usr/bin/lscpu -p").readlines() max_socket_index = -1 for line in lines: if line.startswith('#'): continue # get the socket index from the output socket_index = int(line.split(',')[2]) if socket_index > max_socket_index: max_socket_index = socket_index if max_socket_index > -1: return 1 + max_socket_index except: pass # Next try parsing /proc/cpuinfo if os.access("/proc/cpuinfo", os.R_OK): try: lines = open("/proc/cpuinfo", 'r').readlines() socket_ids = set() for line in lines: if 'physical id' in line: socket_index = int(line.split(':')[1].strip()) socket_ids.add(socket_index) if len(socket_ids) > 0: return len(socket_ids) except: pass # Next try dmidecode if os.access("/usr/sbin/dmidecode", os.X_OK): try: lines = os.popen("/usr/sbin/dmidecode -t processor").readlines() count = 0 for line in lines: if 'Processor Information' in line: count += 1 if count > 0: return count except: pass return None # This has got to be one of the ugliest fucntions alive def read_cpuinfo(): def get_entry(a, entry): e = entry.lower() if not e in a: return "" return a[e] # read cpu list and return number of cpus and list as dictionary def get_cpulist_as_dict(cpulist): count = 0 tmpdict = {} for cpu in cpulist.split("\n\n"): if not len(cpu): continue count = count + 1 if count > 1: break # no need to parse rest for cpu_attr in cpu.split("\n"): if not len(cpu_attr): continue vals = cpu_attr.split(":") if len(vals) != 2: # XXX: make at least some effort to recover this data... continue name, value = vals[0].strip(), vals[1].strip() tmpdict[name.lower()] = value return tmpdict if not os.access("/proc/cpuinfo", os.R_OK): return {} # Okay, the kernel likes to give us the information we need in the # standard "C" locale. if locale: # not really needed if you don't plan on using atof() locale.setlocale(locale.LC_NUMERIC, "C") cpulist = open("/proc/cpuinfo", "r").read() uname = os.uname()[4].lower() count = cpu_count() # This thing should return a hwdict that has the following # members: # # class, desc (required to identify the hardware device) # count, type, model, model_number, model_ver, model_rev # bogomips, platform, speed, cache hwdict = { 'class': "CPU", "desc" : "Processor", } if uname[0] == "i" and uname[-2:] == "86" or (uname == "x86_64"): # IA32 compatible enough tmpdict = get_cpulist_as_dict(cpulist) if uname == "x86_64": hwdict['platform'] = 'x86_64' else: hwdict['platform'] = "i386" hwdict['count'] = count hwdict['type'] = get_entry(tmpdict, 'vendor_id') hwdict['model'] = get_entry(tmpdict, 'model name') hwdict['model_number'] = get_entry(tmpdict, 'cpu family') hwdict['model_ver'] = get_entry(tmpdict, 'model') hwdict['model_rev'] = get_entry(tmpdict, 'stepping') hwdict['cache'] = get_entry(tmpdict, 'cache size') hwdict['bogomips'] = get_entry(tmpdict, 'bogomips') hwdict['other'] = get_entry(tmpdict, 'flags') mhz_speed = get_entry(tmpdict, 'cpu mhz') if mhz_speed == "": # damn, some machines don't report this mhz_speed = "-1" try: hwdict['speed'] = int(round(float(mhz_speed)) - 1) except ValueError: hwdict['speed'] = -1 elif uname in["alpha", "alphaev6"]: # Treat it as an an Alpha tmpdict = get_cpulist_as_dict(cpulist) hwdict['platform'] = "alpha" hwdict['count'] = get_entry(tmpdict, 'cpus detected') hwdict['type'] = get_entry(tmpdict, 'cpu') hwdict['model'] = get_entry(tmpdict, 'cpu model') hwdict['model_number'] = get_entry(tmpdict, 'cpu variation') hwdict['model_version'] = "%s/%s" % (get_entry(tmpdict, 'system type'), get_entry(tmpdict,'system variation')) hwdict['model_rev'] = get_entry(tmpdict, 'cpu revision') hwdict['cache'] = "" # pitty the kernel doesn't tell us this. hwdict['bogomips'] = get_entry(tmpdict, 'bogomips') hwdict['other'] = get_entry(tmpdict, 'platform string') hz_speed = get_entry(tmpdict, 'cycle frequency [Hz]') # some funky alphas actually report in the form "462375000 est." hz_speed = hz_speed.split() try: hwdict['speed'] = int(round(float(hz_speed[0]))) / 1000000 except ValueError: hwdict['speed'] = -1 elif uname in ["ia64"]: tmpdict = get_cpulist_as_dict(cpulist) hwdict['platform'] = uname hwdict['count'] = count hwdict['type'] = get_entry(tmpdict, 'vendor') hwdict['model'] = get_entry(tmpdict, 'family') hwdict['model_ver'] = get_entry(tmpdict, 'archrev') hwdict['model_rev'] = get_entry(tmpdict, 'revision') hwdict['bogomips'] = get_entry(tmpdict, 'bogomips') mhz_speed = tmpdict['cpu mhz'] try: hwdict['speed'] = int(round(float(mhz_speed)) - 1) except ValueError: hwdict['speed'] = -1 hwdict['other'] = get_entry(tmpdict, 'features') elif uname in ['ppc64']: tmpdict = get_cpulist_as_dict(cpulist) hwdict['platform'] = uname hwdict['count'] = count hwdict['model'] = get_entry(tmpdict, "cpu") hwdict['model_ver'] = get_entry(tmpdict, 'revision') hwdict['bogomips'] = get_entry(tmpdict, 'bogomips') hwdict['type'] = get_entry(tmpdict, 'machine') # strings are postpended with "mhz" mhz_speed = get_entry(tmpdict, 'clock')[:-3] try: hwdict['speed'] = int(round(float(mhz_speed)) - 1) except ValueError: hwdict['speed'] = -1 elif uname in ['s390', 's390x']: tmpdict = {} for cpu in cpulist.split("\n"): vals = cpu.split(": ") if len(vals) != 2: continue tmpdict[vals[0].strip()] = vals[1].strip() hwdict['platform'] = uname hwdict['type'] = get_entry(tmpdict,'vendor_id') hwdict['model'] = uname hwdict['count'] = count hwdict['bogomips'] = get_entry(tmpdict, 'bogomips per cpu') hwdict['model_number'] = "" hwdict['model_ver'] = "" hwdict['model_rev'] = "" hwdict['cache'] = "" hwdict['other'] = get_entry(tmpdict, 'features') hwdict['speed'] = 0 else: # XXX: expand me. Be nice to others hwdict['platform'] = uname hwdict['count'] = count hwdict['type'] = uname hwdict['model'] = uname hwdict['model_number'] = "" hwdict['model_ver'] = "" hwdict['model_rev'] = "" hwdict['cache'] = "" hwdict['bogomips'] = "" hwdict['other'] = "" hwdict['speed'] = 0 # make sure we get the right number here if not hwdict["count"]: hwdict["count"] = 1 else: try: hwdict["count"] = int(hwdict["count"]) except: hwdict["count"] = 1 else: if hwdict["count"] == 0: # we have at least one hwdict["count"] = 1 # Network communication doesn't really belong in here. Sadly though # this is the only single place we can put this check. If it's not # here then it would need to be in five or six other places, which # is not good from a DRY and quality-assurance perspective. s = rhnserver.RhnServer() if s.capabilities.hasCapability('cpu_sockets'): # If we know it add in the number of sockets number_sockets = __get_number_sockets() if number_sockets: hwdict['socket_count'] = number_sockets # This whole things hurts a lot. return hwdict def read_memory(): un = os.uname() kernel = un[2] if kernel[:3] >= "2.6": return read_memory_2_6() if kernel[:3] == "2.4": return read_memory_2_4() def read_memory_2_4(): if not os.access("/proc/meminfo", os.R_OK): return {} meminfo = open("/proc/meminfo", "r").read() lines = meminfo.split("\n") curline = lines[1] memlist = curline.split() memdict = {} memdict['class'] = "MEMORY" megs = int(long(memlist[1])/(10241024)) if megs < 32: megs = megs + (4 - (megs % 4)) else: megs = megs + (16 - (megs % 16)) memdict['ram'] = str(megs) curline = lines[2] memlist = curline.split() # otherwise, it breaks on > ~4gigs of swap megs = int(long(memlist[1])/(1024*1024)) memdict['swap'] = str(megs) return memdict def read_memory_2_6(): if not os.access("/proc/meminfo", os.R_OK): return {} meminfo = open("/proc/meminfo", "r").read() lines = meminfo.split("\n") meminfo_dict = {} for line in lines: blobs = line.split(":", 1) key = blobs[0] if len(blobs) == 1: continue #print(blobs) value = blobs[1].strip() meminfo_dict[key] = value memdict = {} memdict["class"] = "MEMORY" total_str = meminfo_dict['MemTotal'] blips = total_str.split(" ") total_k = long(blips[0]) megs = long(total_k/(1024)) swap_str = meminfo_dict['SwapTotal'] blips = swap_str.split(' ') swap_k = long(blips[0]) swap_megs = long(swap_k/(1024)) memdict['ram'] = str(megs) memdict['swap'] = str(swap_megs) return memdict def findHostByRoute(): """ returns [hostname, intf, intf6] Where hostname is you FQDN of this machine. And intf is numeric IPv4 address. And intf6 is IPv6 address. """ cfg = config.initUp2dateConfig() sl = config.getServerURL() st = {'https':443, 'http':80} hostname = None intf = None intf6 = None for serverUrl in sl: server = serverUrl.split('/')[2] servertype = serverUrl.split(':')[0] port = st[servertype] for family in (AF_INET6, AF_INET): try: s = socket.socket(family) except socket.error: continue if cfg['enableProxy']: server_port = config.getProxySetting() (server, port) = server_port.split(':') port = int(port) try: s.settimeout(5) s.connect((server, port)) intf_tmp = s.getsockname()[0] if family == AF_INET: intf = intf_tmp else: intf6 = intf_tmp hostname_tmp = socket.getfqdn(intf_tmp) if hostname_tmp != intf_tmp: hostname = hostname_tmp except socket.error: s.close() continue s.close() # Override hostname with the value from /etc/hostname if os.path.isfile("/etc/hostname") and os.access("/etc/hostname", os.R_OK): hostnameinfo = open("/etc/hostname", "r").readlines() for info in hostnameinfo: if not len(info): continue hostname = info.strip() # Override hostname with the one in /etc/sysconfig/network # for bz# 457953 elif os.path.isfile("/etc/sysconfig/network") and os.access("/etc/sysconfig/network", os.R_OK): networkinfo = open("/etc/sysconfig/network", "r").readlines() for info in networkinfo: if not len(info): continue vals = info.split('=') if len(vals) <= 1: continue if vals[0].strip() == "HOSTNAME": # /etc/sysconfig/network is of shell syntax, # so values can be quoted hostname = ''.join(vals[1:]).strip('"\' \t\n') break if hostname == None or hostname == 'localhost.localdomain': hostname = "unknown" return hostname, intf, intf6 def get_slave_hwaddr(master, slave): hwaddr = "" try: bonding = open('/proc/net/bonding/%s' % master, "r") except: return hwaddr slave_found = False for line in bonding.readlines(): if slave_found and line.find("Permanent HW addr: ") != -1: hwaddr = line.split()[3] break if line.find("Slave Interface: ") != -1: ifname = line.split()[2] if ifname == slave: slave_found = True bonding.close() return hwaddr def read_network(): netdict = {} netdict['class'] = "NETINFO" netdict['hostname'], netdict['ipaddr'], netdict['ip6addr'] = findHostByRoute() if netdict['hostname'] == "unknown": netdict['hostname'] = gethostname() if "." not in netdict['hostname']: netdict['hostname'] = socket.getfqdn() if netdict['ipaddr'] is None: try: list_of_addrs = getaddrinfo(netdict['hostname'], None) ipv4_addrs = filter(lambda x:x[0]==socket.AF_INET, list_of_addrs) # take first ipv4 addr netdict['ipaddr'] = ipv4_addrs[0][4][0] except: netdict['ipaddr'] = "127.0.0.1" if netdict['ip6addr'] is None: try: list_of_addrs = getaddrinfo(netdict['hostname'], None) ipv6_addrs = filter(lambda x:x[0]==socket.AF_INET6, list_of_addrs) # take first ipv6 addr netdict['ip6addr'] = ipv6_addrs[0][4][0] except: netdict['ip6addr'] = "::1" if netdict['ipaddr'] is None: netdict['ipaddr'] = '' if netdict['ip6addr'] is None: netdict['ip6addr'] = '' return netdict def read_network_interfaces(): intDict = {} intDict['class'] = "NETINTERFACES" if not ethtool_present and not netifaces_present: # ethtool is not available on non-linux platforms (as kfreebsd), skip it sys.stderr.write("Warning: information about network interfaces could not be retrieved on this platform.\n") return intDict if ethtool_present: interfaces = list(set(ethtool.get_devices() + ethtool.get_active_devices())) else: interfaces = netifaces.interfaces() for interface in interfaces: try: if ethtool_present: hwaddr = ethtool.get_hwaddr(interface) else: hwaddr = netifaces.ifaddresses(interface)[netifaces.AF_LINK][0]['addr'] except: hwaddr = "" # slave devices can have their hwaddr changed try: master = os.readlink('/sys/class/net/%s/master' % interface) except: master = None if master: master_interface = os.path.basename(master) hwaddr = get_slave_hwaddr(master_interface, interface) try: if ethtool_present: module = ethtool.get_module(interface) else: driver_file = open('/sys/class/net/%s/device/uevent' % interface, 'r') module = driver_file.readline().split('=')[1].strip() driver_file.close() except: if interface == 'lo': module = "loopback" else: module = "Unknown" try: if ethtool_present: ipaddr = ethtool.get_ipaddr(interface) else: ipaddr = netifaces.ifaddresses(interface)[netifaces.AF_INET][0]['addr'] except: ipaddr = "" try: if ethtool_present: netmask = ethtool.get_netmask(interface) else: netmask = netifaces.ifaddresses(interface)[netifaces.AF_INET][0]['netmask'] except: netmask = "" try: if ethtool_present: broadcast = ethtool.get_broadcast(interface) else: broadcast = netifaces.ifaddresses(interface)[netifaces.AF_INET][0]['broadcast'] except: broadcast = "" ip6_list = [] if ethtool_present: dev_info = ethtool.get_interfaces_info(interface) for info in dev_info: # one interface may have more IPv6 addresses for ip6 in info.get_ipv6_addresses(): scope = ip6.scope if scope == 'global': scope = 'universe' ip6_list.append({ 'scope': scope, 'addr': ip6.address, 'netmask': ip6.netmask }) else: try: for dev_info in netifaces.ifaddresses(interface)[netifaces.AF_INET6]: ip6_addr = dev_info['addr'].split('%')[0] scope_info = ipaddress.IPv6Address(ip6_addr) if scope_info.is_global: scope = 'universe' elif scope_info.is_link_local: scope = 'link' elif scope_info.is_loopback: scope = 'host' elif scope_info.is_site_local: scope = 'site' # count number of '1' bits in netmask returned by netifaces ip6_netmask = dev_info['netmask'] netmask_bits = 0 for two_octets in ip6_netmask.split(':'): if not two_octets: break elif two_octets.lower() == 'ffff': netmask_bits += 16 else: # remove '0b' from begin and find last '1' in the string netmask_bits += 1 + bin(int(two_octets.split('/')[0], 16))[2:].rindex('1') ip6_list.append({ 'scope': scope, 'addr': ip6_addr, 'netmask': netmask_bits }) except KeyError: pass # no ipv6 for this interface intDict[interface] = {'hwaddr': hwaddr, 'ipaddr': ipaddr, 'netmask': netmask, 'broadcast': broadcast, 'module': module, 'ipv6': ip6_list} return intDict # Read DMI information via hal. def read_dmi(): dmidict = {} dmidict["class"] = "DMI" # Try to obtain DMI info if architecture is i386, x86_64 or ia64 uname = os.uname()[4].lower() if not (uname[0] == "i" and uname[-2:] == "86") and not (uname == "x86_64"): return dmidict # System Information vendor = dmi_vendor() if vendor: dmidict["vendor"] = vendor product = get_dmi_data('/dmidecode/SystemInfo/ProductName') if product: dmidict["product"] = product version = get_dmi_data('/dmidecode/SystemInfo/Version') if version: system = product + " " + version dmidict["system"] = system # BaseBoard Information dmidict["board"] = get_dmi_data('/dmidecode/BaseBoardInfo/Manufacturer') # Bios Information vendor = get_dmi_data('/dmidecode/BIOSinfo/Vendor') if vendor: dmidict["bios_vendor"] = vendor version = get_dmi_data('/dmidecode/BIOSinfo/Version') if version: dmidict["bios_version"] = version release = get_dmi_data('/dmidecode/BIOSinfo/ReleaseDate') if release: dmidict["bios_release"] = release # Chassis Information # The hairy part is figuring out if there is an asset tag/serial number of importance chassis_serial = get_dmi_data('/dmidecode/ChassisInfo/SerialNumber') chassis_tag = get_dmi_data('/dmidecode/ChassisInfo/AssetTag') board_serial = get_dmi_data('/dmidecode/BaseBoardInfo/SerialNumber') system_serial = get_dmi_data('/dmidecode/SystemInfo/SerialNumber') dmidict["asset"] = "(%s: %s) (%s: %s) (%s: %s) (%s: %s)" % ("chassis", chassis_serial, "chassis", chassis_tag, "board", board_serial, "system", system_serial) # Clean up empty entries for k in list(dmidict.keys()): if dmidict[k] is None: del dmidict[k] # Finished return dmidict def get_hal_system_and_smbios(): try: if using_gudev: props = get_computer_info() else: computer = get_hal_computer() props = computer.GetAllProperties() except Exception: log = up2dateLog.initLog() msg = "Error reading system and smbios information: %s\n" % (sys.exc_info()[1]) log.log_debug(msg) return {} system_and_smbios = {} for key in props: if key.startswith('system'): system_and_smbios[ustr(key)] = ustr(props[key]) system_and_smbios.update(get_smbios()) return system_and_smbios def get_smbios(): """ Returns dictionary with values we are interested for. For historical reason it is in format, which use HAL. Currently in dictionary are keys: smbios.system.uuid, smbios.bios.vendor, smbios.system.serial, smbios.system.manufacturer. """ _initialize_dmi_data() if _dmi_not_available: return {} else: return { 'smbios.system.uuid': dmi_system_uuid(), 'smbios.bios.vendor': dmi_vendor(), 'smbios.system.serial': get_dmi_data('/dmidecode/SystemInfo/SerialNumber'), 'smbios.system.manufacturer': get_dmi_data('/dmidecode/SystemInfo/Manufacturer'), 'smbios.system.product': get_dmi_data('/dmidecode/SystemInfo/ProductName'), 'smbios.system.skunumber': get_dmi_data('/dmidecode/SystemInfo/SKUnumber'), 'smbios.system.family': get_dmi_data('/dmidecode/SystemInfo/Family'), 'smbios.system.version': get_dmi_data('/dmidecode/SystemInfo/Version'), } # this one reads it all def Hardware(): if using_gudev: allhw = get_devices() else: hal_status, dbus_status = check_hal_dbus_status() hwdaemon = 1 if hal_status or dbus_status: # if status != 0 haldaemon or messagebus service not running. # set flag and dont try probing hardware and DMI info # and warn the user. log = up2dateLog.initLog() msg = "Warning: haldaemon or messagebus service not running. Cannot probe hardware and DMI information.\n" log.log_me(msg) hwdaemon = 0 allhw = [] if hwdaemon: try: ret = read_hal() if ret: allhw = ret except: # bz253596 : Logging Dbus Error messages instead of printing on stdout log = up2dateLog.initLog() msg = "Error reading hardware information: %s\n" % (sys.exc_info()[0]) log.log_me(msg) # all others return individual arrays # cpu info try: ret = read_cpuinfo() if ret: allhw.append(ret) except: print(_("Error reading cpu information:"), sys.exc_info()[0]) # memory size info try: ret = read_memory() if ret: allhw.append(ret) except: print(_("Error reading system memory information:"), sys.exc_info()[0]) cfg = config.initUp2dateConfig() if not cfg["skipNetwork"]: # minimal networking info try: ret = read_network() if ret: allhw.append(ret) except: print(_("Error reading networking information:"), sys.exc_info()[0]) # dont like catchall exceptions but theres not # really anything useful we could do at this point # and its been trouble prone enough # minimal DMI info try: ret = read_dmi() if ret: allhw.append(ret) except: # bz253596 : Logging Dbus Error messages instead of printing on stdout log = up2dateLog.initLog() msg = "Error reading DMI information: %s\n" % (sys.exc_info()[0]) log.log_me(msg) try: ret = read_installinfo() if ret: allhw.append(ret) except: print(_("Error reading install method information:"), sys.exc_info()[0]) if not cfg["skipNetwork"]: try: ret = read_network_interfaces() if ret: allhw.append(ret) except: print(_("Error reading network interface information:"), sys.exc_info()[0]) # all Done. return allhw # XXX: Need more functions here: # - filesystems layout (/proc.mounts and /proc/mdstat) # - is the kudzu config enough or should we strat chasing lscpi and try to parse that # piece of crap output? # # Main program # if __name__ == '__main__': for hw in Hardware(): for k in hw.keys(): print("'%s' : '%s'" % (k, hw[k])) print	31.91 KB	2025-05-29 13:37
📄 hardware_gudev.py # Copyright (c) 2010--2016 Red Hat, Inc. # # This software is licensed to you under the GNU General Public License, # version 2 (GPLv2). There is NO WARRANTY for this software, express or # implied, including the implied warranties of MERCHANTABILITY or FITNESS # FOR A PARTICULAR PURPOSE. You should have received a copy of GPLv2 # along with this software; if not, see # http://www.gnu.org/licenses/old-licenses/gpl-2.0.txt. # # Red Hat trademarks are not licensed under GPLv2. No permission is # granted to use or replicate Red Hat trademarks that are incorporated # in this software or its documentation. # try: import gi gi.require_version('GUdev', '1.0') from gi.repository import GUdev gi_gudev = True except (ImportError, ValueError): import gudev import glib gi_gudev = False import os import re from hwdata import PCI, USB def get_devices(): """ Returns list of dictionaries with keys for every device in system (values are provide as example): 'bus' : 'pci' 'driver' : 'pcieport-driver' 'pciType' : '1' 'detached' : '0' 'class' : 'OTHER' 'desc' : 'Intel Corporation\|5000 Series Chipset PCI Express x4 Port 2' """ # listen to uevents on all subsystems if gi_gudev: client = GUdev.Client() else: client = gudev.Client([""]) # FIX ME - change to None to list all devices once it is fixed in gudev devices = client.query_by_subsystem("pci") + client.query_by_subsystem("usb") + client.query_by_subsystem("block") + client.query_by_subsystem("ccw") + client.query_by_subsystem("scsi") result = [] for device in devices: subsystem = device.get_subsystem() result_item = { 'bus': subsystem, 'driver': device.get_driver(), 'pciType': _clasify_pci_type(subsystem), 'detached': '0', # always zero? 'class': _clasify_class(device), 'desc': _get_device_desc(device), } if result_item['class'] is None: result_item['class'] = 'OTHER' if result_item['driver'] is None: result_item['driver'] = 'unknown' if subsystem == 'block': if device.has_property('ID_BUS'): result_item['bus'] = device.get_property('ID_BUS') result_item['device'] = device.get_name() if device.get_devtype() == 'partition': # do not report partitions, just whole disks continue if device.get_property('DM_NAME'): # LVM device continue if device.get_property('MAJOR') == '1': # ram device continue if device.get_property('MAJOR') == '7': # character devices for virtual console terminals continue # This is interpreted as Physical. But what to do with it? # result_item['prop1'] = '' # This is interpreted as Logical. But what to do with it? # result_item['prop2'] = '' elif subsystem == 'pci': pci_class = device.get_property('PCI_ID') if pci_class: (result_item['prop1'], result_item['prop2']) = pci_class.split(':') pci_subsys = device.get_property('PCI_SUBSYS_ID') if pci_subsys: (result_item['prop3'], result_item['prop4']) = pci_subsys.split(':') elif subsystem == 'usb': if device.has_property('ID_VENDOR_ID'): result_item['prop1'] = device.get_property('ID_VENDOR_ID') if device.has_property('ID_MODEL_ID'): result_item['prop2'] = device.get_property('ID_MODEL_ID') if device.has_property('ID_BUS') and device.get_property('ID_BUS') == 'scsi': if device.has_property('ID_PATH') or device.has_property('DEVPATH'): if device.has_property('ID_PATH'): path = device.get_property('ID_PATH') m = re.search('.scsi-(\d+):(\d+):(\d+):(\d+)', path) else: # device.has_property('DEVPATH') path = device.get_property('DEVPATH') m = re.search('./(\d+):(\d+):(\d+):(\d+)/block/', path) if m: # do not fail, if regexp did not match result_item['prop1'] = m.group(1) # DEV_HOST result_item['prop2'] = m.group(2) # DEV_ID result_item['prop3'] = m.group(3) # DEV_CHANNEL result_item['prop4'] = m.group(4) # DEV_LUN result.append(result_item) return result def get_computer_info(): """ Return dictionaries with keys (values are provided as example): 'system.formfactor': 'unknown' 'system.kernel.version': '2.6.18-128.1.6.el5xen' 'system.kernel.machine': 'i686' 'system.kernel.name': 'Linux' """ uname = os.uname() result = { 'system.kernel.name': uname[0], 'system.kernel.version': uname[2], 'system.kernel.machine': uname[4], } return result #PCI DEVICE DEFINES # These are taken from pci_ids.h in the linux kernel source and used to # properly identify the hardware PCI_BASE_CLASS_STORAGE = '1' PCI_CLASS_STORAGE_SCSI = '00' PCI_CLASS_STORAGE_IDE = '01' PCI_CLASS_STORAGE_FLOPPY = '02' PCI_CLASS_STORAGE_IPI = '03' PCI_CLASS_STORAGE_RAID = '04' PCI_CLASS_STORAGE_OTHER = '80' PCI_BASE_CLASS_NETWORK = '2' PCI_CLASS_NETWORK_ETHERNET = '00' PCI_CLASS_NETWORK_TOKEN_RING = '01' PCI_CLASS_NETWORK_FDDI = '02' PCI_CLASS_NETWORK_ATM = '03' PCI_CLASS_NETWORK_OTHER = '80' PCI_BASE_CLASS_DISPLAY = '3' PCI_CLASS_DISPLAY_VGA = '00' PCI_CLASS_DISPLAY_XGA = '01' PCI_CLASS_DISPLAY_3D = '02' PCI_CLASS_DISPLAY_OTHER = '80' PCI_BASE_CLASS_MULTIMEDIA = '4' PCI_CLASS_MULTIMEDIA_VIDEO = '00' PCI_CLASS_MULTIMEDIA_AUDIO = '01' PCI_CLASS_MULTIMEDIA_PHONE = '02' PCI_CLASS_MULTIMEDIA_OTHER = '80' PCI_BASE_CLASS_BRIDGE = '6' PCI_CLASS_BRIDGE_HOST = '00' PCI_CLASS_BRIDGE_ISA = '01' PCI_CLASS_BRIDGE_EISA = '02' PCI_CLASS_BRIDGE_MC = '03' PCI_CLASS_BRIDGE_PCI = '04' PCI_CLASS_BRIDGE_PCMCIA = '05' PCI_CLASS_BRIDGE_NUBUS = '06' PCI_CLASS_BRIDGE_CARDBUS = '07' PCI_CLASS_BRIDGE_RACEWAY = '08' PCI_CLASS_BRIDGE_OTHER = '80' PCI_BASE_CLASS_COMMUNICATION = '7' PCI_CLASS_COMMUNICATION_SERIAL = '00' PCI_CLASS_COMMUNICATION_PARALLEL = '01' PCI_CLASS_COMMUNICATION_MULTISERIAL = '02' PCI_CLASS_COMMUNICATION_MODEM = '03' PCI_CLASS_COMMUNICATION_OTHER = '80' PCI_BASE_CLASS_INPUT = '9' PCI_CLASS_INPUT_KEYBOARD = '00' PCI_CLASS_INPUT_PEN = '01' PCI_CLASS_INPUT_MOUSE = '02' PCI_CLASS_INPUT_SCANNER = '03' PCI_CLASS_INPUT_GAMEPORT = '04' PCI_CLASS_INPUT_OTHER = '80' PCI_BASE_CLASS_SERIAL = 'C' PCI_CLASS_SERIAL_FIREWIRE = '00' PCI_CLASS_SERIAL_ACCESS = '01' PCI_CLASS_SERIAL_SSA = '02' PCI_CLASS_SERIAL_USB = '03' PCI_CLASS_SERIAL_FIBER = '04' PCI_CLASS_SERIAL_SMBUS = '05' def _clasify_pci_type(subsystem): """ return 1 if device is PCI, otherwise -1 """ if subsystem == 'pci': return '1' else: return '-1' def _clasify_class(device): """ Clasify type of device. Returned value is one of following string: NETWORK, KEYBOARD, MOUSE, VIDEO, USB, IDE, SCSI, RAID, MODEM, SCANNER CAPTURE, AUDIO, FIREWIRE, SOCKET, CDROM, HD, FLOPPY, TAPE, PRINTER, OTHER or None if it is neither PCI nor USB device. """ (base_class, sub_class) = _parse_pci_class(device.get_property('PCI_CLASS')) subsystem = device.get_subsystem() # network devices if base_class == PCI_BASE_CLASS_NETWORK: return 'OTHER' # if set as 'NETWORK' it will not display in HW tab # input devices # pci if base_class == PCI_BASE_CLASS_INPUT: if sub_class == PCI_CLASS_INPUT_KEYBOARD: return 'KEYBOARD' elif sub_class == PCI_CLASS_INPUT_MOUSE: return 'MOUSE' # usb id_serial = device.get_property('ID_SERIAL') if id_serial: id_serial = id_serial.lower() # KEYBOARD <-- do this before mouse, some keyboards have built-in mice if 'keyboard' in id_serial: return 'KEYBOARD' # MOUSE if 'mouse' in id_serial: return 'MOUSE' if base_class: # PCI Devices if base_class == PCI_BASE_CLASS_DISPLAY: return 'VIDEO' elif base_class == PCI_BASE_CLASS_SERIAL: if sub_class == PCI_CLASS_SERIAL_USB: return 'USB' elif sub_class == PCI_CLASS_SERIAL_FIREWIRE: return 'FIREWIRE' elif base_class == PCI_BASE_CLASS_STORAGE: if sub_class == PCI_CLASS_STORAGE_IDE: return 'IDE' elif sub_class == PCI_CLASS_STORAGE_SCSI: return 'SCSI' elif sub_class == PCI_CLASS_STORAGE_RAID: return 'RAID' elif sub_class == PCI_CLASS_STORAGE_FLOPPY: return 'FLOPPY' elif base_class == PCI_BASE_CLASS_COMMUNICATION and sub_class == PCI_CLASS_COMMUNICATION_MODEM: return 'MODEM' elif base_class == PCI_BASE_CLASS_INPUT and sub_class == PCI_CLASS_INPUT_SCANNER: return 'SCANNER' elif base_class == PCI_BASE_CLASS_MULTIMEDIA: if sub_class == PCI_CLASS_MULTIMEDIA_VIDEO: return 'CAPTURE' elif sub_class == PCI_CLASS_MULTIMEDIA_AUDIO: return 'AUDIO' elif base_class == PCI_BASE_CLASS_BRIDGE and ( sub_class == PCI_CLASS_BRIDGE_PCMCIA or sub_class == PCI_CLASS_BRIDGE_CARDBUS ): return 'SOCKET' if subsystem == 'block': if device.has_property('ID_CDROM') or ( device.has_property('ID_TYPE') and device.get_property('ID_TYPE') == 'cd'): return 'CDROM' else: return 'HD' elif subsystem == 'sound': return 'AUDIO' if subsystem =='scsi': if device.get_devtype =='scsi_device': dev_type = _get_scsi_dev_type(device) if dev_type == 0 or dev_type == 14: return 'HD' elif dev_type == 1: return 'TAPE' elif dev_type == 5: return 'CDROM' else: return 'OTHER' # PRINTER m = re.search('.*/lp\d+$', device.get_sysfs_path()) if m: return 'PRINTER' if subsystem == 'scsi': return 'SCSI' # Catchall for specific devices, only do this after all the others if subsystem == 'pci' or subsystem == 'usb': return 'OTHER' # No class found return None def _get_device_desc(device): """ Return human readable description of device. """ subsystem = device.get_subsystem() command = None result = None if subsystem == 'pci': (vendor_id, device_id) = device.get_property('PCI_ID').split(':') pci = PCI() result = "%s\|%s" % (pci.get_vendor(vendor_id), pci.get_device(vendor_id, device_id)) elif subsystem == 'usb': vendor_id = device.get_property('ID_VENDOR_ID') usb = USB() if vendor_id: result = "%s\|%s" % (usb.get_vendor(vendor_id), usb.get_device(vendor_id, device.get_property('ID_MODEL_ID'))) elif device.get_devtype() == 'usb_interface': if device.get_driver() == 'usbhid': result = 'USB HID Interface' elif device.get_driver() == 'hub': result = 'USB Hub Interface' else: result = 'USB Interface' elif device.get_devtype() == 'usb_device' and device.get_property('PRODUCT'): (vendor_id, model_id) = device.get_property('PRODUCT').split('/')[:2] # left pad it with 0 to 4 digits vendor_id = '%.4x' % int(vendor_id, 16) model_id = '%.4x' % int(model_id, 16) result = "%s\|%s" % (usb.get_vendor(vendor_id), usb.get_device(vendor_id, model_id)) elif subsystem == 'block': result = device.get_property('ID_MODEL') if result: return result else: return '' def _parse_pci_class(pci_class): """ Parse Class Code. Return touple of [base class code, sub-class code] You are usually interested to only first two. The third - "specific register-level programming interface" is ignored. For details, see the PCI Local Bus Specification 2.1/2.2 Section 6.2.1 Device Identification """ if pci_class is None: return (None, None) else: return (pci_class[-6:-4], pci_class[-4:-2]) def _get_scsi_dev_type(device): """ Return SCSI type of device in raw format as presented in /sys/...devpath../type """ try: f = open("%s/type" % device.get_sysfs_path(), 'r') except IOError: return -1 result = f.readline() f.close() return result	12.91 KB	2025-05-29 13:37
📄 hardware_hal.py # # Copyright (c) 1999--2016 Red Hat, Inc. # # This software is licensed to you under the GNU General Public License, # version 2 (GPLv2). There is NO WARRANTY for this software, express or # implied, including the implied warranties of MERCHANTABILITY or FITNESS # FOR A PARTICULAR PURPOSE. You should have received a copy of GPLv2 # along with this software; if not, see # http://www.gnu.org/licenses/old-licenses/gpl-2.0.txt. # # Red Hat trademarks are not licensed under GPLv2. No permission is # granted to use or replicate Red Hat trademarks that are incorporated # in this software or its documentation. # """ Get hardware info using HAL """ from .haltree import HalTree, HalDevice import dbus #PCI DEVICE DEFINES # These are taken from pci_ids.h in the linux kernel source and used to # properly identify the hardware PCI_BASE_CLASS_STORAGE = 1 PCI_CLASS_STORAGE_SCSI = 0 PCI_CLASS_STORAGE_IDE = 1 PCI_CLASS_STORAGE_FLOPPY = 2 PCI_CLASS_STORAGE_IPI = 3 PCI_CLASS_STORAGE_RAID = 4 PCI_CLASS_STORAGE_OTHER = 80 PCI_BASE_CLASS_NETWORK = 2 PCI_CLASS_NETWORK_ETHERNET = 0 PCI_CLASS_NETWORK_TOKEN_RING = 1 PCI_CLASS_NETWORK_FDDI = 2 PCI_CLASS_NETWORK_ATM = 3 PCI_CLASS_NETWORK_OTHER = 80 PCI_BASE_CLASS_DISPLAY = 3 PCI_CLASS_DISPLAY_VGA = 0 PCI_CLASS_DISPLAY_XGA = 1 PCI_CLASS_DISPLAY_3D = 2 PCI_CLASS_DISPLAY_OTHER = 80 PCI_BASE_CLASS_MULTIMEDIA = 4 PCI_CLASS_MULTIMEDIA_VIDEO = 0 PCI_CLASS_MULTIMEDIA_AUDIO = 1 PCI_CLASS_MULTIMEDIA_PHONE = 2 PCI_CLASS_MULTIMEDIA_OTHER = 80 PCI_BASE_CLASS_BRIDGE = 6 PCI_CLASS_BRIDGE_HOST = 0 PCI_CLASS_BRIDGE_ISA = 1 PCI_CLASS_BRIDGE_EISA = 2 PCI_CLASS_BRIDGE_MC = 3 PCI_CLASS_BRIDGE_PCI = 4 PCI_CLASS_BRIDGE_PCMCIA = 5 PCI_CLASS_BRIDGE_NUBUS = 6 PCI_CLASS_BRIDGE_CARDBUS = 7 PCI_CLASS_BRIDGE_RACEWAY = 8 PCI_CLASS_BRIDGE_OTHER = 80 PCI_BASE_CLASS_COMMUNICATION = 7 PCI_CLASS_COMMUNICATION_SERIAL = 0 PCI_CLASS_COMMUNICATION_PARALLEL = 1 PCI_CLASS_COMMUNICATION_MULTISERIAL = 2 PCI_CLASS_COMMUNICATION_MODEM = 3 PCI_CLASS_COMMUNICATION_OTHER = 80 PCI_BASE_CLASS_INPUT = 9 PCI_CLASS_INPUT_KEYBOARD = 0 PCI_CLASS_INPUT_PEN = 1 PCI_CLASS_INPUT_MOUSE = 2 PCI_CLASS_INPUT_SCANNER = 3 PCI_CLASS_INPUT_GAMEPORT = 4 PCI_CLASS_INPUT_OTHER = 80 PCI_BASE_CLASS_SERIAL = 12 PCI_CLASS_SERIAL_FIREWIRE = 0 PCI_CLASS_SERIAL_ACCESS = 1 PCI_CLASS_SERIAL_SSA = 2 PCI_CLASS_SERIAL_USB = 3 PCI_CLASS_SERIAL_FIBER = 4 PCI_CLASS_SERIAL_SMBUS = 5 # read_hal() # # This reads in all the properties for each device from HAL, storing the # property names & values into a dict. A list of dicts is returned. # # This only works on newer versions of dbus & HAL (as found in RHEL5) def read_hal(): ret = [] bus = dbus.SystemBus() hal_manager_obj = bus.get_object('org.freedesktop.Hal', '/org/freedesktop/Hal/Manager') hal_manager = dbus.Interface(hal_manager_obj, 'org.freedesktop.Hal.Manager') device_list = hal_manager.GetAllDevices() hal_tree = HalTree() for udi in device_list: device_obj = bus.get_object ('org.freedesktop.Hal', udi) device = dbus.Interface(device_obj, 'org.freedesktop.Hal.Device') properties = device.GetAllProperties() haldev = HalDevice(properties) hal_tree.add(haldev) kudzu_list = process_hal_nodes(hal_tree.head) return kudzu_list # Recursive function, does all the dirty work for add_hal_hardware def process_hal_nodes(node): kudzu_list = [] node.classification = classify_hal(node) if node.classification: parent = node.parent dev = {} dev['class'] = node.classification #get bus dev['bus'] = str(get_device_bus(node)) #get scsi info if dev['bus'] == 'scsi': if 'scsi.host' in parent.properties: dev['prop1'] = parent.properties['scsi.host'] if 'scsi.target' in parent.properties: dev['prop2'] = parent.properties['scsi.target'] if 'scsi.bus' in parent.properties: dev['prop3'] = parent.properties['scsi.bus'] if 'scsi.lun' in parent.properties: dev['prop4'] = parent.properties['scsi.lun'] dev['driver'] = str(get_device_driver(node)) device_path = get_device_path(node) if device_path: dev['device'] = str(device_path) dev['desc'] = str(get_device_description(node)) dev['pciType'] = str(get_device_pcitype(node)) dev['detached'] = 0 kudzu_list.append(dev) for child in node.children: child_list = process_hal_nodes(child) kudzu_list.extend(child_list) return kudzu_list def classify_hal(node): # NETWORK if 'net.interface' in node.properties: return 'NETWORK' if 'info.product' in node.properties and 'info.category' in node.properties: if node.properties['info.category'] == 'input': # KEYBOARD <-- do this before mouse, some keyboards have built-in mice if 'keyboard' in node.properties['info.product'].lower(): return 'KEYBOARD' # MOUSE if 'mouse' in node.properties['info.product'].lower(): return 'MOUSE' if 'pci.device_class' in node.properties: #VIDEO if node.properties['pci.device_class'] == PCI_BASE_CLASS_DISPLAY: return 'VIDEO' #USB if (node.properties['pci.device_class'] == PCI_BASE_CLASS_SERIAL and node.properties['pci.device_subclass'] == PCI_CLASS_SERIAL_USB): return 'USB' if node.properties['pci.device_class'] == PCI_BASE_CLASS_STORAGE: #IDE if node.properties['pci.device_subclass'] == PCI_CLASS_STORAGE_IDE: return 'IDE' #SCSI if node.properties['pci.device_subclass'] == PCI_CLASS_STORAGE_SCSI: return 'SCSI' #RAID if node.properties['pci.device_subclass'] == PCI_CLASS_STORAGE_RAID: return 'RAID' #MODEM if (node.properties['pci.device_class'] == PCI_BASE_CLASS_COMMUNICATION and node.properties['pci.device_subclass'] == PCI_CLASS_COMMUNICATION_MODEM): return 'MODEM' #SCANNER if (node.properties['pci.device_class'] == PCI_BASE_CLASS_INPUT and node.properties['pci.device_subclass'] == PCI_CLASS_INPUT_SCANNER): return 'SCANNER' if node.properties['pci.device_class'] == PCI_BASE_CLASS_MULTIMEDIA: #CAPTURE -- video capture card if node.properties['pci.device_subclass'] == PCI_CLASS_MULTIMEDIA_VIDEO: return 'CAPTURE' #AUDIO if node.properties['pci.device_subclass'] == PCI_CLASS_MULTIMEDIA_AUDIO: return 'AUDIO' #FIREWIRE if (node.properties['pci.device_class'] == PCI_BASE_CLASS_SERIAL and node.properties['pci.device_subclass'] == PCI_CLASS_SERIAL_FIREWIRE): return 'FIREWIRE' #SOCKET -- PCMCIA yenta socket stuff if (node.properties['pci.device_class'] == PCI_BASE_CLASS_BRIDGE and (node.properties['pci.device_subclass'] == PCI_CLASS_BRIDGE_PCMCIA or node.properties['pci.device_subclass'] == PCI_CLASS_BRIDGE_CARDBUS)): return 'SOCKET' if 'storage.drive_type' in node.properties: #CDROM if node.properties['storage.drive_type'] == 'cdrom': return 'CDROM' #HD if node.properties['storage.drive_type'] == 'disk': return 'HD' #FLOPPY if node.properties['storage.drive_type'] == 'floppy': return 'FLOPPY' #TAPE if node.properties['storage.drive_type'] == 'tape': return 'TAPE' if 'xen.type' in node.properties: if node.properties['xen.type'] == 'vbd': return 'HD' #PRINTER if 'printer.product' in node.properties: return 'PRINTER' #Catchall for specific devices, only do this after all the others if ('pci.product_id' in node.properties or 'usb.product_id' in node.properties): return 'OTHER' # No class found return None def get_device_bus(node): if 'storage.bus' in node.properties: bus = node.properties['storage.bus'] elif 'info.bus' in node.properties: if node.properties['info.bus'] == 'platform': bus = 'MISC' else: bus = node.properties['info.bus'] else: bus = 'MISC' return bus def get_device_driver(node): if 'info.linux.driver' in node.properties: driver = node.properties['info.linux.driver'] elif 'net.linux.driver' in node.properties: driver = node.properties['net.linux.driver'] else: driver = 'unknown' return driver def get_device_path(node): """ Return the device file path. As kudzu did not return a string with the /dev/ prefix, this function will not, either. RHN's DB has a limit of 16 characters for the device path. If the path is longer than that, return None. If no device path is found, return None. """ dev = None if 'block.device' in node.properties: dev = node.properties['block.device'] elif 'linux.device_file' in node.properties: dev = node.properties['linux.device_file'] elif (node.classification == 'NETWORK' and 'net.interface' in node.properties): dev = node.properties['net.interface'] if dev: if dev.startswith('/dev/'): dev = dev[5:] if len(dev) > 16: dev = None return dev def get_device_description(node): if ('info.vendor' in node.properties and 'info.product' in node.properties): desc = node.properties['info.vendor'] + '\|' + node.properties['info.product'] elif ('info.vendor' in node.properties): desc = node.properties['info.vendor'] elif 'info.product' in node.properties: desc = node.properties['info.product'] else: desc = "" return desc def get_device_pcitype(node): PCI_TYPE_PCMCIA = 2 PCI_TYPE_PCI = 1 PCI_TYPE_NOT_PCI = -1 if 'info.bus' in (node.properties and node.properties['info.bus'] == 'pci'): parent = node.parent if ('pci.device_class' in parent.properties and (parent.properties['pci.device_class'] == 6 and (parent.properties['pci.device_subclass'] == 5 or parent.properties['pci.device_subclass'] == 7))): pcitype = PCI_TYPE_PCMCIA else: pcitype = PCI_TYPE_PCI else: pcitype = PCI_TYPE_NOT_PCI return pcitype def get_hal_computer(): bus = dbus.SystemBus() computer_obj = bus.get_object("org.freedesktop.Hal", "/org/freedesktop/Hal/devices/computer") computer = dbus.Interface(computer_obj, "org.freedesktop.Hal.Device") return computer def check_hal_dbus_status(): # check if hal and messagebus are running, if not warn the user import subprocess as commands hal_status, msg = commands.getstatusoutput('/etc/init.d/haldaemon status') dbus_status, msg = commands.getstatusoutput('/etc/init.d/messagebus status') return hal_status, dbus_status	11.33 KB	2025-05-29 13:37
📄 hardware_udev.py # Copyright (c) 2010--2016 Red Hat, Inc. # # This software is licensed to you under the GNU General Public License, # version 2 (GPLv2). There is NO WARRANTY for this software, express or # implied, including the implied warranties of MERCHANTABILITY or FITNESS # FOR A PARTICULAR PURPOSE. You should have received a copy of GPLv2 # along with this software; if not, see # http://www.gnu.org/licenses/old-licenses/gpl-2.0.txt. # # Red Hat trademarks are not licensed under GPLv2. No permission is # granted to use or replicate Red Hat trademarks that are incorporated # in this software or its documentation. # import pyudev import os import re from hwdata import PCI, USB def get_devices(): """ Returns list of dictionaries with keys for every device in system (values are provide as example): 'bus' : 'pci' 'driver' : 'pcieport-driver' 'pciType' : '1' 'detached' : '0' 'class' : 'OTHER' 'desc' : 'Intel Corporation\|5000 Series Chipset PCI Express x4 Port 2' """ # listen to uevents on all subsystems context = pyudev.Context() devices = context.list_devices() result = [] for device in devices.match_subsystem("pci").match_subsystem("usb").match_subsystem("block").match_subsystem("ccw").match_subsystem("scsi"): subsystem = device.subsystem result_item = { 'bus': subsystem, 'driver': device.driver, 'pciType': _clasify_pci_type(subsystem), 'detached': '0', # always zero? 'class': _clasify_class(device), 'desc': _get_device_desc(device), } if result_item['class'] is None: result_item['class'] = 'OTHER' if result_item['driver'] is None: result_item['driver'] = 'unknown' if subsystem == 'block': if _get_device_property(device, 'ID_BUS'): result_item['bus'] = _get_device_property(device, 'ID_BUS') result_item['device'] = device.sys_name if device.device_type == 'partition': # do not report partitions, just whole disks continue if _get_device_property(device, 'DM_NAME'): # LVM device continue if _get_device_property(device, 'MAJOR') == '1': # ram device continue if _get_device_property(device, 'MAJOR') == '7': # character devices for virtual console terminals continue # This is interpreted as Physical. But what to do with it? # result_item['prop1'] = '' # This is interpreted as Logical. But what to do with it? # result_item['prop2'] = '' elif subsystem == 'pci': pci_class = _get_device_property(device, 'PCI_ID') if pci_class: (result_item['prop1'], result_item['prop2']) = pci_class.split(':') pci_subsys = _get_device_property(device, 'PCI_SUBSYS_ID') if pci_subsys: (result_item['prop3'], result_item['prop4']) = pci_subsys.split(':') elif subsystem == 'usb': if _get_device_property(device, 'ID_VENDOR_ID'): result_item['prop1'] = _get_device_property(device, 'ID_VENDOR_ID') if _get_device_property(device, 'ID_MODEL_ID'): result_item['prop2'] = _get_device_property(device, 'ID_MODEL_ID') if _get_device_property(device, 'ID_BUS') and _get_device_property(device, 'ID_BUS') == 'scsi': if _get_device_property(device, 'ID_PATH') or _get_device_property(device, 'DEVPATH'): if _get_device_property(device, 'ID_PATH'): path = _get_device_property(device, 'ID_PATH') m = re.search('.scsi-(\d+):(\d+):(\d+):(\d+)', path) else: # device.has_property('DEVPATH') path = _get_device_property(device, 'DEVPATH') m = re.search('./(\d+):(\d+):(\d+):(\d+)/block/', path) if m: # do not fail, if regexp did not match result_item['prop1'] = m.group(1) # DEV_HOST result_item['prop2'] = m.group(2) # DEV_ID result_item['prop3'] = m.group(3) # DEV_CHANNEL result_item['prop4'] = m.group(4) # DEV_LUN result.append(result_item) return result def get_computer_info(): """ Return dictionaries with keys (values are provided as example): 'system.formfactor': 'unknown' 'system.kernel.version': '2.6.18-128.1.6.el5xen' 'system.kernel.machine': 'i686' 'system.kernel.name': 'Linux' """ uname = os.uname() result = { 'system.kernel.name': uname[0], 'system.kernel.version': uname[2], 'system.kernel.machine': uname[4], } return result #PCI DEVICE DEFINES # These are taken from pci_ids.h in the linux kernel source and used to # properly identify the hardware PCI_BASE_CLASS_STORAGE = '1' PCI_CLASS_STORAGE_SCSI = '00' PCI_CLASS_STORAGE_IDE = '01' PCI_CLASS_STORAGE_FLOPPY = '02' PCI_CLASS_STORAGE_IPI = '03' PCI_CLASS_STORAGE_RAID = '04' PCI_CLASS_STORAGE_OTHER = '80' PCI_BASE_CLASS_NETWORK = '2' PCI_CLASS_NETWORK_ETHERNET = '00' PCI_CLASS_NETWORK_TOKEN_RING = '01' PCI_CLASS_NETWORK_FDDI = '02' PCI_CLASS_NETWORK_ATM = '03' PCI_CLASS_NETWORK_OTHER = '80' PCI_BASE_CLASS_DISPLAY = '3' PCI_CLASS_DISPLAY_VGA = '00' PCI_CLASS_DISPLAY_XGA = '01' PCI_CLASS_DISPLAY_3D = '02' PCI_CLASS_DISPLAY_OTHER = '80' PCI_BASE_CLASS_MULTIMEDIA = '4' PCI_CLASS_MULTIMEDIA_VIDEO = '00' PCI_CLASS_MULTIMEDIA_AUDIO = '01' PCI_CLASS_MULTIMEDIA_PHONE = '02' PCI_CLASS_MULTIMEDIA_OTHER = '80' PCI_BASE_CLASS_BRIDGE = '6' PCI_CLASS_BRIDGE_HOST = '00' PCI_CLASS_BRIDGE_ISA = '01' PCI_CLASS_BRIDGE_EISA = '02' PCI_CLASS_BRIDGE_MC = '03' PCI_CLASS_BRIDGE_PCI = '04' PCI_CLASS_BRIDGE_PCMCIA = '05' PCI_CLASS_BRIDGE_NUBUS = '06' PCI_CLASS_BRIDGE_CARDBUS = '07' PCI_CLASS_BRIDGE_RACEWAY = '08' PCI_CLASS_BRIDGE_OTHER = '80' PCI_BASE_CLASS_COMMUNICATION = '7' PCI_CLASS_COMMUNICATION_SERIAL = '00' PCI_CLASS_COMMUNICATION_PARALLEL = '01' PCI_CLASS_COMMUNICATION_MULTISERIAL = '02' PCI_CLASS_COMMUNICATION_MODEM = '03' PCI_CLASS_COMMUNICATION_OTHER = '80' PCI_BASE_CLASS_INPUT = '9' PCI_CLASS_INPUT_KEYBOARD = '00' PCI_CLASS_INPUT_PEN = '01' PCI_CLASS_INPUT_MOUSE = '02' PCI_CLASS_INPUT_SCANNER = '03' PCI_CLASS_INPUT_GAMEPORT = '04' PCI_CLASS_INPUT_OTHER = '80' PCI_BASE_CLASS_SERIAL = 'C' PCI_CLASS_SERIAL_FIREWIRE = '00' PCI_CLASS_SERIAL_ACCESS = '01' PCI_CLASS_SERIAL_SSA = '02' PCI_CLASS_SERIAL_USB = '03' PCI_CLASS_SERIAL_FIBER = '04' PCI_CLASS_SERIAL_SMBUS = '05' def _clasify_pci_type(subsystem): """ return 1 if device is PCI, otherwise -1 """ if subsystem == 'pci': return '1' else: return '-1' def _get_device_property(device, prop): """ return the property of the given device independent of the implementation version """ try: return device.properties.get(prop) except AttributeError: return device.get(prop) def _clasify_class(device): """ Clasify type of device. Returned value is one of following string: NETWORK, KEYBOARD, MOUSE, VIDEO, USB, IDE, SCSI, RAID, MODEM, SCANNER CAPTURE, AUDIO, FIREWIRE, SOCKET, CDROM, HD, FLOPPY, TAPE, PRINTER, OTHER or None if it is neither PCI nor USB device. """ (base_class, sub_class) = _parse_pci_class(_get_device_property(device, 'PCI_CLASS')) subsystem = device.subsystem # network devices if base_class == PCI_BASE_CLASS_NETWORK: return 'OTHER' # if set as 'NETWORK' it will not display in HW tab # input devices # pci if base_class == PCI_BASE_CLASS_INPUT: if sub_class == PCI_CLASS_INPUT_KEYBOARD: return 'KEYBOARD' elif sub_class == PCI_CLASS_INPUT_MOUSE: return 'MOUSE' # usb id_serial = _get_device_property(device, 'ID_SERIAL') if id_serial: id_serial = id_serial.lower() # KEYBOARD <-- do this before mouse, some keyboards have built-in mice if 'keyboard' in id_serial: return 'KEYBOARD' # MOUSE if 'mouse' in id_serial: return 'MOUSE' if base_class: # PCI Devices if base_class == PCI_BASE_CLASS_DISPLAY: return 'VIDEO' elif base_class == PCI_BASE_CLASS_SERIAL: if sub_class == PCI_CLASS_SERIAL_USB: return 'USB' elif sub_class == PCI_CLASS_SERIAL_FIREWIRE: return 'FIREWIRE' elif base_class == PCI_BASE_CLASS_STORAGE: if sub_class == PCI_CLASS_STORAGE_IDE: return 'IDE' elif sub_class == PCI_CLASS_STORAGE_SCSI: return 'SCSI' elif sub_class == PCI_CLASS_STORAGE_RAID: return 'RAID' elif sub_class == PCI_CLASS_STORAGE_FLOPPY: return 'FLOPPY' elif base_class == PCI_BASE_CLASS_COMMUNICATION and sub_class == PCI_CLASS_COMMUNICATION_MODEM: return 'MODEM' elif base_class == PCI_BASE_CLASS_INPUT and sub_class == PCI_CLASS_INPUT_SCANNER: return 'SCANNER' elif base_class == PCI_BASE_CLASS_MULTIMEDIA: if sub_class == PCI_CLASS_MULTIMEDIA_VIDEO: return 'CAPTURE' elif sub_class == PCI_CLASS_MULTIMEDIA_AUDIO: return 'AUDIO' elif base_class == PCI_BASE_CLASS_BRIDGE and ( sub_class == PCI_CLASS_BRIDGE_PCMCIA or sub_class == PCI_CLASS_BRIDGE_CARDBUS ): return 'SOCKET' if subsystem == 'block': if _get_device_property(device, 'ID_CDROM') or ( _get_device_property(device, 'ID_TYPE') and _get_device_property(device, 'ID_TYPE') == 'cd'): return 'CDROM' else: return 'HD' elif subsystem == 'sound': return 'AUDIO' if subsystem =='scsi': if device.device_type =='scsi_device': dev_type = _get_scsi_dev_type(device) if dev_type == 0 or dev_type == 14: return 'HD' elif dev_type == 1: return 'TAPE' elif dev_type == 5: return 'CDROM' else: return 'OTHER' # PRINTER m = re.search('.*/lp\d+$', device.sys_path) if m: return 'PRINTER' if subsystem == 'scsi': return 'SCSI' # Catchall for specific devices, only do this after all the others if subsystem == 'pci' or subsystem == 'usb': return 'OTHER' # No class found return None def _get_device_desc(device): """ Return human readable description of device. """ subsystem = device.subsystem command = None result = None if subsystem == 'pci': (vendor_id, device_id) = _get_device_property(device, 'PCI_ID').split(':') pci = PCI() result = "%s\|%s" % (pci.get_vendor(vendor_id), pci.get_device(vendor_id, device_id)) elif subsystem == 'usb': vendor_id = _get_device_property(device, 'ID_VENDOR_ID') usb = USB() if vendor_id: result = "%s\|%s" % (usb.get_vendor(vendor_id), usb.get_device(vendor_id, _get_device_property(device, 'ID_MODEL_ID'))) elif device.device_type == 'usb_interface': if device.driver == 'usbhid': result = 'USB HID Interface' elif device.driver == 'hub': result = 'USB Hub Interface' else: result = 'USB Interface' elif device.device_type == 'usb_device' and _get_device_property(device, 'PRODUCT'): (vendor_id, model_id) = _get_device_property(device, 'PRODUCT').split('/')[:2] # left pad it with 0 to 4 digits vendor_id = '%.4x' % int(vendor_id, 16) model_id = '%.4x' % int(model_id, 16) result = "%s\|%s" % (usb.get_vendor(vendor_id), usb.get_device(vendor_id, model_id)) elif subsystem == 'block': result = _get_device_property(device, 'ID_MODEL') if result: return result else: return '' def _parse_pci_class(pci_class): """ Parse Class Code. Return touple of [base class code, sub-class code] You are usually interested to only first two. The third - "specific register-level programming interface" is ignored. For details, see the PCI Local Bus Specification 2.1/2.2 Section 6.2.1 Device Identification """ if pci_class is None: return (None, None) else: return (pci_class[-6:-4], pci_class[-4:-2]) def _get_scsi_dev_type(device): """ Return SCSI type of device in raw format as presented in /sys/...devpath../type """ try: f = open("%s/type" % device.sys_path, 'r') except IOError: return -1 result = f.readline() f.close() return result	12.99 KB	2025-05-29 13:37
📄 pkgUtils.py # Client code for Update Agent # Copyright (c) 2011--2016 Red Hat, Inc. Distributed under GPLv2. # # Author: Simon Lukasik from up2date_client.pkgplatform import getPlatform if getPlatform() == 'deb': from up2date_client.debUtils import * else: from up2date_client.rpmUtils import *	295 B	2025-05-29 13:37
📄 pkgplatform.py # Client code for Update Agent # Copyright (c) 2011--2012 Red Hat, Inc. Distributed under GPLv2. # # Author: Simon Lukasik # # substituted to the prefered platfrom by Makefile _platform='rpm' def getPlatform(): if _platform != '@PLAT' + 'FORM@': return _platform else: return 'rpm'	309 B	2025-05-29 13:52
📄 pmPlugin.py # Client code for enabling plugin # Copyright (c) 2000--2016 Red Hat, Inc. import os import re import rpm # global variables try: from dnf import __version__ PM_PLUGIN_CONF = '/etc/dnf/plugins/spacewalk.conf' PM_PLUGIN_NAME = 'dnf-plugin-spacewalk' PM_NAME = 'dnf' except ImportError: PM_PLUGIN_CONF = '/etc/yum/pluginconf.d/rhnplugin.conf' PM_PLUGIN_NAME = 'yum-rhn-plugin' PM_NAME = 'yum' def pluginEnable(): """Enables plugin, may throw IOError""" conf_changed = 0 plugin_present = 0 if PluginPackagePresent(): plugin_present = 1 if PluginConfPresent(): if not PluginEnabled(): enablePlugin() conf_changed = 1 else: createDefaultPluginConf() conf_changed = 1 elif os.path.exists("/usr/lib/zypp/plugins/services/spacewalk"): """SUSE zypp plugin is installed""" plugin_present = 1 return plugin_present, conf_changed def PluginPackagePresent(): """ Returns positive number if plugin package is installed, otherwise it return 0 """ ts = rpm.TransactionSet() headers = ts.dbMatch('providename', PM_PLUGIN_NAME) return headers.count() def PluginConfPresent(): """ Returns true if PM_PLUGIN_CONF is presented """ try: os.stat(PM_PLUGIN_CONF) return True except OSError: return False def createDefaultPluginConf(): """ Create file PM_PLUGIN_CONF, with default values """ f = open(PM_PLUGIN_CONF, 'w') f.write("""[main] enabled = 1 gpgcheck = 1""") f.close() def PluginEnabled(): """ Returns True if plugin is enabled Can thrown IOError exception. """ f = open(PM_PLUGIN_CONF, 'r') lines = f.readlines() f.close() main_section = False result = False for line in lines: if re.match("^\[.]", line): if re.match("^\[main]", line): main_section = True else: main_section = False if main_section: m = re.match('^\senabled\s=\s([0-9])', line) if m: if int(m.group(1)): result = True else: result = False return result def enablePlugin(): """ enable plugin by setting enabled=1 in file PM_PLUGIN_CONF Can thrown IOError exception. """ f = open(PM_PLUGIN_CONF, 'r') lines = f.readlines() f.close() main_section = False f = open(PM_PLUGIN_CONF, 'w') for line in lines: if re.match("^\[.]", line): if re.match("^\[main]", line): main_section = True else: main_section = False if main_section: line = re.sub('^(\s)enabled\s*=.+', r'\1enabled = 1', line) f.write(line) f.close()	2.79 KB	2025-05-29 13:37
📄 rhnChannel.py # all the crap that is stored on the rhn side of stuff # updating/fetching package lists, channels, etc from up2date_client import up2dateAuth from up2date_client import up2dateLog from up2date_client import up2dateErrors from up2date_client import config from up2date_client import rhnserver import gettext t = gettext.translation('rhn-client-tools', fallback=True) # Python 3 translations don't have a ugettext method if not hasattr(t, 'ugettext'): t.ugettext = t.gettext _ = t.ugettext # heh, dont get much more generic than this... class rhnChannel: # shrug, use attributes for thetime being def __init__(self, **kwargs): self.dict = {} for kw in kwargs.keys(): self.dict[kw] = kwargs[kw] def __getitem__(self, item): return self.dict[item] def __setitem__(self, item, value): self.dict[item] = value def __lt__(self, other): return (self.dict["name"] > other.dict["name"]) def keys(self): return self.dict.keys() def values(self): return self.dict.values() def items(self): return self.dict.items() class rhnChannelList: def __init__(self): # probabaly need to keep these in order for #precedence self.list = [] def addChannel(self, channel): self.list.append(channel) def channels(self): return self.list def getByLabel(self, channelname): for channel in self.list: if channel['label'] == channelname: return channel def getByName(self, channelname): return self.getByLabel(channelname) def getByType(self, type): channels = [] for channel in self.list: if channel['type'] == type: channels.append(channel) return channels # for the gui client that needs to show more info # maybe we should always make this call? If nothing # else, wrapper should have a way to show extended channel info def getChannelDetails(timeout=None): channels = [] sourceChannels = getChannels(timeout=timeout) for sourceChannel in sourceChannels.channels(): if sourceChannel['type'] != 'up2date': # FIMXE: kluge since we dont have a good name, maybe be able to fix sourceChannel['name'] = sourceChannel['label'] sourceChannel['description'] = "%s channel %s from %s" % (sourceChannel['type'], sourceChannel['label'], sourceChannel['url']) channels.append(sourceChannel) return channels cmdline_pkgs = [] global selected_channels selected_channels = None def getChannels(force=None, label_whitelist=None, timeout=None): """ return rhnChannelList containing list of channel we are subscribed to """ cfg = config.initUp2dateConfig() log = up2dateLog.initLog() global selected_channels if not selected_channels and not force: selected_channels = rhnChannelList() s = rhnserver.RhnServer(timeout=timeout) if not up2dateAuth.getSystemId(): raise up2dateErrors.NoSystemIdError(_("Unable to Locate SystemId")) up2dateChannels = s.up2date.listChannels(up2dateAuth.getSystemId()) for chan in up2dateChannels: if label_whitelist and not chan['label'] in label_whitelist: continue channel = rhnChannel(type = 'up2date', url = config.getServerlURLFromMirror()) for key in chan.keys(): if key == "last_modified": channel['version'] = chan['last_modified'] else: channel[key] = chan[key] if cfg['channelOverride']: channel['name'] = channel['label'] = \ channel['description'] = \ channel['summary'] = cfg['channelOverride'] log.log_me('WARNING: Channel overridden locally: %s' % channel['name']) log.log_me('-------: If you see this warning outside of the leapp process') log.log_me('-------: contact CloudLinux support.') selected_channels.addChannel(channel) if len(selected_channels.list) == 0: raise up2dateErrors.NoChannelsError(_("This system may not be updated until it is associated with a channel.")) return selected_channels def setChannels(tempchannels): global selected_channels selected_channels = None whitelist = dict(map(lambda x: (x,1), tempchannels)) return getChannels(label_whitelist=whitelist) def subscribeChannels(channels,username,passwd): s = rhnserver.RhnServer() return s.up2date.subscribeChannels(up2dateAuth.getSystemId(), channels, username, passwd) def unsubscribeChannels(channels,username,passwd): s = rhnserver.RhnServer() return s.up2date.unsubscribeChannels(up2dateAuth.getSystemId(), channels, username, passwd)	4.91 KB	2025-05-29 13:37
📄 rhnHardware.py from up2date_client import up2dateAuth from up2date_client import rpcServer from up2date_client import hardware def updateHardware(): s = rpcServer.getServer() hardwareList = hardware.Hardware() s.registration.refresh_hw_profile(up2dateAuth.getSystemId(), hardwareList)	328 B	2025-05-29 13:37
📄 rhnPackageInfo.py # all the crap that is stored on the rhn side of stuff # updating/fetching package lists, channels, etc from up2date_client import up2dateAuth from up2date_client import up2dateLog from up2date_client import rhnserver from up2date_client import pkgUtils def logDeltaPackages(pkgs): log = up2dateLog.initLog() log.log_me("Adding packages to package profile: %s" % pprint_pkglist(pkgs['added'])) log.log_me("Removing packages from package profile: %s" % pprint_pkglist(pkgs['removed'])) # pylint: disable=unused-argument def updatePackageProfile(timeout=None): """ get a list of installed packages and send it to rhnServer """ # The endpoint that is called in this function (registration.update_packages) # is disabled on the CLN side, and can throw errors if called (CLOS-3032). return # Original code preserved below in case we ever need to re-enable this function. # log = up2dateLog.initLog() # log.log_me("Updating package profile") # packages = pkgUtils.getInstalledPackageList(getArch=1) # s = rhnserver.RhnServer(timeout=timeout) # if not s.capabilities.hasCapability('xmlrpc.packages.extended_profile', 2): # # for older satellites and hosted - convert to old format # packages = convertPackagesFromHashToList(packages) # s.registration.update_packages(up2dateAuth.getSystemId(), packages) def pprint_pkglist(pkglist): if type(pkglist) == type([]): output = ["%s-%s-%s" % (a[0],a[1],a[2]) for a in pkglist] else: output = "%s-%s-%s" % (pkglist[0], pkglist[1], pkglist[2]) return output def convertPackagesFromHashToList(packages): """ takes list of hashes and covert it to list of lists resulting strucure is: [[name, version, release, epoch, arch, cookie], ... ] """ result = [] for package in packages: if 'arch' in package and 'cookie' in package: result.append([package['name'], package['version'], package['release'], package['epoch'], package['arch'], package['cookie']]) elif 'arch' in package: result.append([package['name'], package['version'], package['release'], package['epoch'], package['arch']]) else: result.append([package['name'], package['version'], package['release'], package['epoch']]) return result	2.34 KB	2025-05-29 13:37
📄 rhncli.py # # Common cli related functions for RHN Client Tools # Copyright (c) 1999--2016 Red Hat, Inc. # # This software is licensed to you under the GNU General Public License, # version 2 (GPLv2). There is NO WARRANTY for this software, express or # implied, including the implied warranties of MERCHANTABILITY or FITNESS # FOR A PARTICULAR PURPOSE. You should have received a copy of GPLv2 # along with this software; if not, see # http://www.gnu.org/licenses/old-licenses/gpl-2.0.txt. # # Red Hat trademarks are not licensed under GPLv2. No permission is # granted to use or replicate Red Hat trademarks that are incorporated # in this software or its documentation. # # Authors: # Adrian Likins <alikins@redhat.com> # Preston Brown <pbrown@redhat.com> # James Bowes <jbowes@redhat.com> # # In addition, as a special exception, the copyright holders give # permission to link the code of portions of this program with the # OpenSSL library under certain conditions as described in each # individual source file, and distribute linked combinations # including the two. # You must obey the GNU General Public License in all respects # for all of the code used other than OpenSSL. If you modify # file(s) with this exception, you may extend this exception to your # version of the file(s), but you are not obligated to do so. If you # do not wish to do so, delete this exception statement from your # version. If you delete this exception statement from all source # files in the program, then also delete it here. import sys import os import socket from optparse import Option from optparse import OptionParser from OpenSSL import SSL from OpenSSL import crypto from rhn import rpclib from rhn.i18n import sstr try: # python2 import xmlrpclib except ImportError: # python3 import xmlrpc.client as xmlrpclib import gettext t = gettext.translation('rhn-client-tools', fallback=True) # Python 3 translations don't have a ugettext method if not hasattr(t, 'ugettext'): t.ugettext = t.gettext _ = t.ugettext from up2date_client import config from up2date_client import up2dateAuth from up2date_client import up2dateErrors from up2date_client import up2dateLog from up2date_client import up2dateUtils from up2date_client import pkgUtils _optionsTable = [ Option("-v", "--verbose", action="count", default=0, help=_("Show additional output")), Option("--proxy", action="store", help=_("Specify an http proxy to use")), Option("--proxyUser", action="store", help=_("Specify a username to use with an authenticated http proxy")), Option("--proxyPassword", action="store", help=_("Specify a password to use with an authenticated http proxy")), Option("--debug", action="store_true", help=_("Enable debug output (network requests)")), ] class RhnCli(object): def __init__(self): self.optparser = OptionParser(option_list = _optionsTable, version = RhnCli.__versionString()) self.options = None self.args = None self.hasGui = False def run(self): # catch any uncaught exceptions and handle them nicely sys.excepthook = exceptionHandler # main loop call try: self.initialize() sys.exit(self.main() or 0) except KeyboardInterrupt: sys.stderr.write(sstr(_("\nAborted.\n"))) sys.exit(1) except OSError: sys.stderr.write(sstr(_("An unexpected OS error occurred: %s\n") % sys.exc_info()[1])) sys.exit(1) except rpclib.MalformedURIError: # Subclass of IOError so must come 1st? e = sys.exc_info()[1] if e is None or len(str(e)) == 0: sys.stderr.write(sstr(_("A connection was attempted with a malformed URI.\n"))) else: sys.stderr.write(sstr(_("A connection was attempted with a malformed URI: %s.\n") % e)) except IOError: sys.stderr.write(sstr(_("There was some sort of I/O error: %s\n") % sys.exc_info()[1])) sys.exit(1) except SSL.Error: sys.stderr.write(sstr(_("There was an SSL error: %s\n") % sys.exc_info()[1])) sys.stderr.write(sstr(_("A common cause of this error is the system time being incorrect. " \ "Verify that the time on this system is correct.\n"))) sys.exit(1) except (SSL.SysCallError, socket.error): sys.stderr.write(sstr("OpenSSL.SSL.SysCallError: %s\n" % str(sys.exc_info()[1]))) sys.exit(2) except crypto.Error: sys.stderr.write(sstr(_("There was a SSL crypto error: %s\n") % sys.exc_info()[1])) except SystemExit: raise except up2dateErrors.AuthenticationError: sys.stderr.write(sstr(_("There was an authentication error: %s\n") % sys.exc_info()[1])) sys.exit(1) except up2dateErrors.RpmError: sys.stderr.write(sstr("%s\n" % sys.exc_info()[1])) sys.exit(1) except xmlrpclib.ProtocolError: sys.stderr.write(sstr("XMLRPC ProtocolError: %s\n" % str(sys.exc_info()[1]))) sys.exit(3) def initialize(self): (self.options, self.args) = self.optparser.parse_args() RhnCli.__setDebugLevel(self.options.verbose) # see if were running as root if os.geteuid() != 0: rootWarningMsg = _("You must be root to run %s") % sys.argv[0] self._warning_dialog(rootWarningMsg) sys.exit(1) if self.options.debug: self._initialize_debug_network_logs() self.__updateProxyConfig() def main(self): raise NotImplementedError def _initialize_debug_network_logs(self): """ Enables logging of all all https requests to console """ import logging try: from http.client import HTTPConnection # py3 except ImportError: from httplib import HTTPConnection # py2 HTTPConnection.debuglevel = 1 logging.basicConfig() logging.getLogger().setLevel(logging.DEBUG) requests_log = logging.getLogger("requests.packages.urllib3") requests_log.setLevel(logging.DEBUG) requests_log.propagate = True def _testRhnLogin(self): try: up2dateAuth.updateLoginInfo() return True except up2dateErrors.ServerCapabilityError: print(sys.exc_info()[1]) return False except up2dateErrors.AuthenticationError: return False except up2dateErrors.RhnServerException: log = up2dateLog.initLog() log.log_me('There was a RhnServerException while testing login:\n') log.log_exception(*sys.exc_info()) return False def _warning_dialog(self, message): if self.hasGui: try: from up2date_client import gui gui.errorWindow(message) except: print(_("Unable to open gui. Try `up2date --nox`")) print(message) else: print(message) def __updateProxyConfig(self): """Update potential proxy configuration. Note: this will _not_ save the info to up2date's configuration file A separate call to config.initUp2dateConfig.save() is needed. """ cfg = config.initUp2dateConfig() if self.options.proxy: cfg.set("httpProxy", self.options.proxy) cfg.set("enableProxy", 1) if self.options.proxyUser: cfg.set("proxyUser", self.options.proxyUser) cfg.set("enableProxyAuth", 1) if self.options.proxyPassword: cfg.set("proxyPassword", self.options.proxyPassword) cfg.set("enableProxyAuth", 1) def saveConfig(self): """ Saves the current up2date configuration being used to disk. """ cfg = config.initUp2dateConfig() cfg.save() def __faultError(self, errMsg): if self.hasGui: try: from up2date_client import gui gui.errorWindow(errMsg) except: print(_("Unable to open gui. Try `up2date --nox`")) print(errMsg) else: print(errMsg) @staticmethod def __versionString(): versionString = _("%%prog (Spacewalk Client Tools) %s\n" "Copyright (C) 1999--2014 Red Hat, Inc.\n" "Licensed under the terms of the GPLv2.") % up2dateUtils.version() return versionString @staticmethod def __setDebugLevel(level): cfg = config.initUp2dateConfig() # figure out the debug level cfg["debug"] = cfg["debug"] + level if cfg["debug"] > 2: pkgUtils.setDebugVerbosity() def exceptionHandler(type, value, tb): log = up2dateLog.initLog() sys.stderr.write(sstr(_("An error has occurred:") + "\n")) if hasattr(value, "errmsg"): sys.stderr.write(sstr(value.errmsg) + "\n") log.log_exception(type, value, tb) else: sys.stderr.write(sstr(str(type) + "\n")) log.log_exception(type, value, tb) sys.stderr.write(sstr(_("See /var/log/up2date for more information") + "\n"))	9.12 KB	2025-05-29 13:37
📄 rhnreg.py # # RHN Registration Client # Copyright (c) 2000--2016 Red Hat, Inc. # # Authors: # Adrian Likins <alikins@redhat.com> # Preston Brown <pbrown@redhat.com> # Daniel Benamy <dbenamy@redhat.com> import os import sys import dbus import base64 import libxml2 from up2date_client import up2dateUtils from up2date_client import up2dateErrors from up2date_client import up2dateAuth from up2date_client import rhnserver from up2date_client import pkgUtils from up2date_client import up2dateLog from up2date_client import rhnreg_constants from up2date_client import hardware from up2date_client.rhnPackageInfo import convertPackagesFromHashToList from up2date_client.pkgplatform import getPlatform from rhn.i18n import ustr, sstr from rhn.tb import raise_with_tb try: # python2 import urlparse import xmlrpclib from types import ListType, TupleType, StringType, UnicodeType, DictType, DictionaryType except ImportError: # python3 import urllib.parse as urlparse import xmlrpc.client as xmlrpclib ListType = list TupleType = tuple StringType = bytes UnicodeType = str DictType = dict DictionaryType = dict long = int try: from virtualization import support except ImportError: support = None import gettext t = gettext.translation('rhn-client-tools', fallback=True) # Python 3 translations don't have a ugettext method if not hasattr(t, 'ugettext'): t.ugettext = t.gettext _ = t.ugettext # global variables #SYSID_DIR = /tmp SYSID_DIR = "/etc/sysconfig/rhn" REMIND_FILE = "%s/rhn_register_remind" % SYSID_DIR HW_CODE_FILE = "%s/hw-activation-code" % SYSID_DIR RHSM_FILE = "/etc/pki/consumer/cert.pem" JWT_TOKEN = '/etc/sysconfig/rhn/jwt.token' _human_readable_to_product = { 'CloudLinux OS Shared Pro': 'shared_pro', 'CloudLinux OS Shared': 'shared', 'CloudLinux OS Solo': 'solo', 'CloudLinux OS Admin': 'admin', } _product_to_human_readable = { v: k for k, v in _human_readable_to_product.items()} from up2date_client import config cfg = config.initUp2dateConfig() log = up2dateLog.initLog() def startRhnsd(): # successful registration. Try to start rhnsd if it isn't running. if os.access("/usr/sbin/rhnsd", os.R_OK\|os.X_OK): # test for UsrMerge systemd environment systemd_system_unitdir = "/usr/lib/systemd/system" systemd_systemctl = "/usr/bin/systemctl" if not os.access(systemd_systemctl, os.R_OK\|os.X_OK): if os.access("/bin/systemctl", os.R_OK\|os.X_OK): systemd_systemctl = "/bin/systemctl" systemd_system_unitdir = "/lib/systemd/system" if os.access("%s/rhnsd.service" % systemd_system_unitdir, os.R_OK): # systemd if os.access(systemd_systemctl, os.R_OK\|os.X_OK): os.system("%s enable rhnsd > /dev/null" % systemd_systemctl); os.system("%s start rhnsd > /dev/null" % systemd_systemctl); else: print(_("Warning: unable to enable rhnsd with systemd")) else: # SysV init scripts if os.access("/sbin/chkconfig", os.R_OK\|os.X_OK): os.system("/sbin/chkconfig rhnsd on > /dev/null"); else: print(_("Warning: unable to enable rhnsd with chkconfig")) service_path = "/sbin/service" if not os.access(service_path, os.R_OK\|os.X_OK): if os.access("/usr/sbin/service", os.R_OK\|os.X_OK): service_path = "/usr/sbin/service" rc = os.system("%s rhnsd status > /dev/null" % service_path) if rc: os.system("%s rhnsd start > /dev/null" % service_path) def getOemInfo(): configFile = cfg["oemInfoFile"] or "/etc/sysconfig/rhn/oeminfo" if not os.access(configFile, os.R_OK): return {} fd = open(configFile, "r") L = fd.readlines() info = {} for i in L: i = i.strip() if i == "": continue try: (key, value) = i.split(':') except ValueError: raise_with_tb(up2dateErrors.OemInfoFileError(i)) info[key] = value.strip() return info def rhsm_registered(): """ Returns true if system is registred using subscription manager """ if os.access(RHSM_FILE, os.R_OK): statinfo = os.stat(RHSM_FILE) return statinfo.st_size > 0 else: return False def registered(): return os.access(cfg['systemIdPath'], os.R_OK) def createSystemRegisterRemindFile(): if not os.access(REMIND_FILE, os.R_OK): # touch the file to tell the applet it needs to remind # the user to register fd = open(REMIND_FILE, "w+") fd.close() def removeSystemRegisterRemindFile(): if os.access(REMIND_FILE, os.R_OK): os.unlink(REMIND_FILE) def _write_secure_file(secure_file, file_contents): """ Write a file to disk that is not readable by other users. """ dir_name = os.path.dirname(secure_file) if not os.access(dir_name, os.W_OK): return False if os.access(secure_file, os.F_OK): # already have file there; let's back it up try: os.rename(secure_file, secure_file + '.save') except: return False fd = os.open(secure_file, os.O_WRONLY \| os.O_CREAT, int('0600', 8)) fd_file = os.fdopen(fd, 'w') try: fd_file.write(sstr(file_contents)) finally: fd_file.close() return True def writeSystemId(systemId): res = _write_secure_file(cfg['systemIdPath'], systemId) # newer registratio clients will create a file indicating that # we need to remind the user to register, this removes it if res: removeSystemRegisterRemindFile() updateRhsmStatus() return res def extract_system_id(): xpath_str = "//member[name='system_id']/value/string" systemId = up2dateAuth.getSystemId() if systemId is None: return try: result = libxml2.parseDoc(systemId) context = result.xpathNewContext() system_id = context.xpathEval(xpath_str)[0].content # remove `ID-` and convert to integer system_id = int(system_id[3:]) except (IndexError, libxml2.parserError): log.log_me('systemID file doesn\'t have system_id field or the file is broken') return return system_id def _execute_pre_jwt_update_hook(token: str, allowTransition: bool): """ Execute binary file which we use as hook for jwt token updates """ import subprocess, os # during cldeploy we don't have python or any other packages, so just exit if not os.path.exists('/opt/cloudlinux/venv/bin/python3'): return cmd = ['/usr/sbin/cl-pre-jwt-update', '--new-token', token] if allowTransition: cmd.append('--allow-transition') p = subprocess.Popen(cmd) stdout, stderr = p.communicate() if p.returncode != 0: log.log_me("Pre jwt update hook failed with stdout=%s and stderr=%s" % (stdout, stderr)) def _execute_post_jwt_update_hook(allowTransition: bool): """ Execute binary file which we use as hook for jwt token updates """ import subprocess, os # during cldeploy we don't have python or any other packages, so just exit if not os.path.exists('/opt/cloudlinux/venv/bin/python3'): return cmd = ['/usr/sbin/cl-post-jwt-update'] if allowTransition: cmd.append('--allow-transition') p = subprocess.Popen(cmd) stdout, stderr = p.communicate() if p.returncode != 0: log.log_me("Post jwt update hook failed with stdout=%s and stderr=%s" % (stdout, stderr)) def getAndWriteJWTTokenToFile(systemId, allowTransition = False): """ Get a JWT token from CLN and save it to the file :param systemId: content of file `/etc/sysconfig/rhn/systemid` :return: None """ xmlrpm_server = rhnserver.RhnServer() try: result = xmlrpm_server.up2date.getJWTToken(systemId) except up2dateErrors.UnknownMethodException: # if CLN doesn't have this method we do nothing return except (up2dateErrors.AuthenticationTicketError, up2dateErrors.RhnUuidUniquenessError, up2dateErrors.CommunicationError, up2dateErrors.AuthenticationOrAccountCreationError): log.log_exception(sys.exc_info()) return _execute_pre_jwt_update_hook(result, allowTransition) _write_secure_file(JWT_TOKEN, result) _execute_post_jwt_update_hook(allowTransition) def writeHWCode(hw_activation_code): """Returns True if the write is successful or False if it fails.""" return _write_secure_file(HW_CODE_FILE, hw_activation_code + '\n') def get_virt_info(): """ This function returns the UUID and virtualization type of this system, if it is a guest. Otherwise, it returns None. To figure this out, we'll use a number of heuristics (list in order of precedence): 1. Check /proc/xen/xsd_port. If exists, we know the system is a host; exit. 2. Check SMBIOS. If vendor='Xen' and UUID is non-zero, we know the system is a fully-virt guest; exit. 3. Check /sys/hypervisor/uuid. If exists and is non-zero, we know the system is a para-virt guest; exit. 4. If non of the above checks worked; we know we have a non-xen-enabled system; exit. """ # First, check whether /proc/xen/xsd_port exists. If so, we know this is # a host system. try: if os.path.exists("/proc/xen/xsd_port"): # Ok, we know this is at least* a host system. However, it may # also be a fully-virt guest. Check for that next. If it is, we'll # just report that instead since we only support one level of # virtualization. (uuid, virt_type) = get_fully_virt_info() return (uuid, virt_type) except IOError: # Failed. Move on to next strategy. pass # This is not a virt host system. Check if it's a fully-virt guest. (uuid, virt_type) = get_fully_virt_info() if uuid is not None: return (uuid, virt_type) # This is not a fully virt guest system. Check if it's a para-virt guest. (uuid, virt_type) = get_para_virt_info() if uuid is not None: return (uuid, virt_type) # If we got here, we have a system that does not have virtualization # enabled. return (None, None) def get_para_virt_info(): """ This function checks /sys/hypervisor/uuid to see if the system is a para-virt guest. It returns a (uuid, virt_type) tuple. """ try: uuid_file = open('/sys/hypervisor/uuid', 'r') uuid = uuid_file.read() uuid_file.close() uuid = uuid.lower().replace('-', '').rstrip("\r\n") virt_type = "para" return (uuid, virt_type) except IOError: # Failed; must not be para-virt. pass return (None, None) def get_fully_virt_info(): """ This function looks in the SMBIOS area to determine if this is a fully-virt guest. It returns a (uuid, virt_type) tuple. """ vendor = hardware.dmi_vendor() uuid = hardware.dmi_system_uuid() if vendor.lower() == "xen": uuid = uuid.lower().replace('-', '') virt_type = "fully" return (uuid, virt_type) else: return (None, None) def _is_host_uuid(uuid): uuid = eval('0x%s' % uuid) return long(uuid) == long(0) def welcomeText(): s = rhnserver.RegistrationRhnServer() return s.registration.welcome_message() def getCaps(): s = rhnserver.RegistrationRhnServer() # figure out if were missing any needed caps s.capabilities.validate() def reserveUser(username, password): s = rhnserver.RegistrationRhnServer() return s.registration.reserve_user(username, password) class RegistrationResult: def __init__(self, systemId, channels, failedChannels, systemSlots, failedSystemSlots, universalActivationKey, rawDict=None): # TODO Get rid of rawDict self._systemId = systemId self._channels = channels self._failedChannels = failedChannels self._systemSlots = systemSlots self._failedSystemSlots = failedSystemSlots if len(universalActivationKey) > 0: self._universalActivationKey = universalActivationKey else: self._universalActivationKey = None self.rawDict = rawDict def getSystemId(self): return self._systemId def getChannels(self): return self._channels def getFailedChannels(self): return self._failedChannels def getSystemSlots(self): return self._systemSlots def getSystemSlotDescriptions(self): return [self._getSlotDescription(s) for s in self._systemSlots] def getFailedSystemSlotDescriptions(self): return [self._getFailedSlotDescription(s) for s in self._failedSystemSlots] def getUniversalActivationKey(self): """Returns None if no universal activation key was used.""" return self._universalActivationKey def hasBaseAndUpdates(self): """Returns True if the system was subscribed to at least one channel and was given any type of system slot so it will get updates. In other words, returns True if the system will be getting at least basic updates. """ # If it was subscribed to at least one channel, that must include a # base channel. return len(self._channels) > 0 and len(self._systemSlots) > 0 def _getFailedSlotDescription(self, slot): if slot == 'virtualization_host': return rhnreg_constants.VIRT + " " + rhnreg_constants.VIRT_FAILED else: return self._getSlotDescription(slot) def _getSlotDescription(self, slot): if slot == 'enterprise_entitled': return rhnreg_constants.MANAGEMENT elif slot == 'virtualization_host': return rhnreg_constants.VIRT else: return slot def getServerEdition(human_readable: bool = False): edition_cache_file = '/opt/cloudlinux/cl_edition' # edition cache does not exist only in the case when # we did not register server yet if not os.path.exists(edition_cache_file): return 'shared' with open(edition_cache_file) as f: raw_edition = f.read().strip('\n') if human_readable: return raw_edition return _human_readable_to_product[raw_edition] def get_users_count_from_cllib(): from subprocess import Popen, PIPE if not os.path.exists('/opt/cloudlinux/venv/bin'): raise ValueError() cmd = '/opt/cloudlinux/venv/bin/python3 -c "from clcommon.cpapi import cpusers; print(cpusers())"' process = Popen(cmd, shell=True, stdout=PIPE, stderr=PIPE) output, errors = [result.decode().strip() for result in process.communicate()] if errors: raise ValueError() return len(output[1:-1].split(', ')) def get_users_count_generic(): from up2date_client.clpwd import ClPwd pwd = ClPwd() return len(pwd.get_uid_dict()) def countServerUsers(): try: users_count = get_users_count_from_cllib() except Exception: users_count = get_users_count_generic() return users_count def checkLicenseKey(activationKey, strictEdition, silentMigration): try: licenseInformation = checkKey(activationKey) except up2dateErrors.CommunicationError as e: print("%s" % e.errmsg) sys.exit(1) except up2dateErrors.UnknownMethodException: return currentEdition = getServerEdition() licenseEdition = licenseInformation['edition'] if licenseEdition == currentEdition: return if strictEdition: print( "WARNING: Automatic registration in yum transactions is " "only available when edition matches the provided license. " "Your current edition is {current_edition} and your license is {new_edition}.".format( current_edition=_product_to_human_readable[currentEdition], new_edition=_product_to_human_readable[licenseEdition] ) ) print("Run clnreg_ks manually to complete registration.") sys.exit(1) if not silentMigration: if not sys.stdin.isatty(): print('Error: interactive input required for edition migration, but tool ' 'is running in non-interactive mode. Please try running the tool again ' 'in interactive shell or add `--migrate-silently` flag to accept all' 'questions and perform the edition migration silently.') exit(1) message = (f"{_product_to_human_readable[currentEdition]} edition installed on your server " f"does not match license you are trying to register server with: " f"{_product_to_human_readable[licenseEdition]}. Migration is required. " f"You may lose access to the services which are not supported by the new edition.") edition_to_users_limit = { 'admin': 5, 'solo': 1 } license_users_limit = edition_to_users_limit.get(licenseEdition) if license_users_limit is not None: users_on_server = countServerUsers() if users_on_server > license_users_limit: print(f"The license you are trying to register with allows a maximum of " f"{edition_to_users_limit[licenseEdition]} hosting accounts which is less " f"than {users_on_server} users detected on this server. Aborting.") sys.exit(1) else: message = (f"{message} Also, the license you are trying to register with allows a maximum of " f"{edition_to_users_limit[licenseEdition]} hosting accounts. " f"Make sure that your system complies with this requirement.") _askConfirmation(message) def _askConfirmation(confirmationMessage: str): """ Prints message and makes sure that client is ready for edition migration. """ print(confirmationMessage) response = input("Do you want to continue? [N/y]: ", ) if response.lower() != 'y': print('Aborted.') sys.exit(1) def registerSystem(username = None, password = None, profileName = None, token = None, other = None, edition=None): """Wrapper for the old xmlrpc to register a system. Activates subscriptions if a reg num is given. """ assert username is None and password is None, \ "username and password usage is deprecated" auth_dict = { "profile_name" : profileName, "os_release" : up2dateUtils.getVersion(), "release_name" : up2dateUtils.getOSRelease(), "architecture" : up2dateUtils.getArch() } # send information about previous registration system_id_xml = up2dateAuth.getSystemId() if system_id_xml is not None: auth_dict["system_id"] = system_id_xml # dict of other bits to send if other: for (key, item) in other.items(): auth_dict[key] = item if token: auth_dict["token"] = token else: auth_dict["username"] = username auth_dict["password"] = password if edition is not None: auth_dict['edition'] = edition else: auth_dict['edition'] = 'solo' \ if os.path.exists('/etc/cloudlinux-edition-solo') \ else 'admin' if os.path.exists('/etc/cloudlinux-edition-admin') else 'shared' if cfg['supportsSMBIOS']: auth_dict["smbios"] = _encode_characters(hardware.get_smbios()) s = rhnserver.RegistrationRhnServer() ret = s.registration.new_system(auth_dict) return ret def checkKey(activationKey): """ Check the activation key and return it's edition and customer """ s = rhnserver.RegistrationRhnServer() ret = s.registration.license_check(activationKey) # {'customerId': 10000327, 'edition': 'solo'} return ret def updateRhsmStatus(): try: bus = dbus.SystemBus() validity_obj = bus.ProxyObjectClass(bus, 'com.redhat.SubscriptionManager', '/EntitlementStatus', introspect=False) validity_iface = dbus.Interface(validity_obj, dbus_interface='com.redhat.SubscriptionManager.EntitlementStatus') except dbus.DBusException: # we can't connect to dbus. it's not running, likely from a minimal # install. we can't do anything here, so just ignore it. return try: validity_iface.check_status() except dbus.DBusException: # the call timed out, or something similar. we don't really care # about a timely reply or what the result might be, we just want # the method to run. So we can safely ignore this. pass def getAvailableChannels(username, password): s = rhnserver.RegistrationRhnServer() server_arch = up2dateUtils.getArch() server_version = up2dateUtils.getVersion() server_release = up2dateUtils.getRelease() availableChannels = None try: availableChannels = s.registration.available_eus_channels( username, password, server_arch, server_version, server_release) except xmlrpclib.Fault: f = sys.exc_info()[1] if f.faultCode == 99: raise_with_tb(up2dateErrors.DelayError(f.faultString)) else: raise return availableChannels def registerSystem2(username = None, password = None, profileName = None, packages = None, activationKey = None, other = {}): """Uses the new xmlrpcs to register a system. Returns a dict instead of just system id. The main differences between this and registerSystem and that this doesn't do activation and does child channel subscriptions if possible. See the documentation for the xmlrpc handlers in backend for more detail. If nothing is going to be in other, it can be {} or None. New in RHEL 5. """ if other is None: other = {} if activationKey: assert username is None assert password is None assert activationKey is not None else: assert username is not None assert password is not None assert activationKey is None for key in other.keys(): assert key in ['registration_number', 'org_id', 'virt_uuid', 'virt_type', 'channel'] if cfg['supportsSMBIOS']: other["smbios"] = _encode_characters(hardware.get_smbios()) s = rhnserver.RegistrationRhnServer() if activationKey: info = s.registration.new_system_activation_key(profileName, up2dateUtils.getOSRelease(), up2dateUtils.getVersion(), up2dateUtils.getArch(), activationKey, other) else: info = s.registration.new_system_user_pass(profileName, up2dateUtils.getOSRelease(), up2dateUtils.getVersion(), up2dateUtils.getArch(), username, password, other) log.log_debug("Returned:\n%s" % info) result = RegistrationResult(info['system_id'], info['channels'], info['failed_channels'], info['system_slots'], info['failed_system_slots'], info['universal_activation_key'], rawDict=info) return result def server_supports_eus(): return cfg["supportsEUS"] # pylint: disable=unused-argument def sendHardware(systemId, hardwareList): # The endpoint that is called in this function (registration.add_hw_profile) # is disabled on the CLN side, and does not do anything. # Therefore, we can safely disable this function. return # Original code preserved below in case we ever need to re-enable this function. # def remove_ip6addr(x): # if x['class'] == 'NETINFO' and 'ip6addr' in x: # del x['ip6addr'] # return x # s = rhnserver.RegistrationRhnServer() # if not s.capabilities.hasCapability('ipv6', 1): # hardwareList = [remove_ip6addr(i) for i in hardwareList] # s.registration.add_hw_profile(systemId, _encode_characters(hardwareList)) # pylint: disable=unused-argument def sendPackages(systemId, packageList): # The endpoint that is called in this function (registration.add_packages) # is disabled on the CLN side, and does not do anything. # Therefore, we can safely disable this function. return # Original code preserved below in case we ever need to re-enable this function. # s = rhnserver.RegistrationRhnServer() # if not s.capabilities.hasCapability('xmlrpc.packages.extended_profile', 2): # # for older satellites and hosted - convert to old format # packageList = convertPackagesFromHashToList(packageList) # s.registration.add_packages(systemId, packageList) def sendVirtInfo(systemId): if support is not None: support.refresh() def listPackages(systemId): s = rhnserver.RegistrationRhnServer() print(s.registration.list_packages, systemId()) def makeNiceServerUrl(server): """Raises up2dateErrors.InvalidProtocolError if the server url has a protocol specified and it's not http or https. """ protocol, host, path, parameters, query, fragmentIdentifier = urlparse.urlparse(server) if protocol is None or protocol == '': server = 'https://' + server # We must call it again because if there wasn't a protocol the # host will be in path protocol, host, path, parameters, query, fragmentIdentifier = urlparse.urlparse(server) if protocol not in ['https', 'http']: raise up2dateErrors.InvalidProtocolError("You specified an invalid " "protocol. Only https and " "http are allowed.") if path is None or path == '' or path == '/': path = '/XMLRPC' server = urlparse.urlunparse((protocol, host, path, parameters, query, fragmentIdentifier)) # TODO Raise an exception if url isn't valid return server def getServerType(serverUrl=None): """Returns 'hosted' if the url points to a known hosted server. Otherwise returns 'satellite'. """ return 'satellite' class ActivationResult: ACTIVATED_NOW = 0 ALREADY_USED = 1 def __init__(self, status, registrationNumber, channels={}, systemSlots={}): """channels and systemSlots are dicts where the key/value pairs are label (string) / quantity (int). """ self._status = status # TODO Validate reg num self._regNum = registrationNumber self._channels = channels self._systemSlots = systemSlots def getStatus(self): return self._status def getRegistrationNumber(self): return self._regNum def getChannelsActivated(self): """Returns a dict- the key/value pairs are label/quantity.""" return self._channels def getSystemSlotsActivated(self): """Returns a dict- the key/value pairs are label/quantity.""" return self._systemSlots def _encode_characters(args): """ All the data we gathered from dmi, bios, gudev are in utf-8, we need to convert characters beyond ord(127) - e.g \xae to unicode. """ result=[] for item in args: item_type = type(item) if item_type == StringType: item = ustr(item) elif item_type == TupleType: item = tuple(_encode_characters(i) for i in item) elif item_type == ListType: item = [_encode_characters(i) for i in item] elif item_type == DictType or item_type == DictionaryType: item = dict([(_encode_characters(name, val)) for name, val in item.items()]) # else: numbers or UnicodeType - are safe result.append(item) if len(result) == 1: return result[0] else: return tuple(result) def _activate_hardware(login, password): # Read the asset code from the hardware. activateHWResult = None hardwareInfo = None hw_activation_code = None try: hardwareInfo = hardware.get_hal_system_and_smbios() hardwareInfo = _encode_characters(hardwareInfo) except: log.log_me("There was an error while reading the hardware " "info from the bios. Traceback:\n") log.log_exception(sys.exc_info()) if hardwareInfo is not None: try: activateHWResult = activateHardwareInfo( login, password, hardwareInfo) if activateHWResult.getStatus() == ActivationResult.ACTIVATED_NOW: hw_activation_code = activateHWResult.getRegistrationNumber() writeHWCode(hw_activation_code) except up2dateErrors.NotEntitlingError: log.log_debug('There are are no entitlements associated ' 'with this hardware.') except up2dateErrors.InvalidRegistrationNumberError: log.log_debug('The hardware id was not recognized as valid.') return hw_activation_code def activateHardwareInfo(username, password, hardwareInfo, orgId=None): """Tries to activate an entitlement linked to the hardware info that we read from the bios. Returns an ActivationResult. Can raise: Invalid number. Hardware info is not entitling. Communication errors, etc """ ## import pprint ## pprint.pprint(hardwareInfo) other = {} if orgId: other = {'org_id': orgId} server = rhnserver.RegistrationRhnServer() result = server.registration.activate_hardware_info(username, password, hardwareInfo, other) statusCode = result['status_code'] regNum = result['registration_number'] log.log_debug('Server returned status code %s' % statusCode) if statusCode == 0: return ActivationResult(ActivationResult.ACTIVATED_NOW, regNum) elif statusCode == 1: return ActivationResult(ActivationResult.ALREADY_USED, regNum) else: message = "The server returned unknown status code %s while activating" \ " the hardware info." % statusCode raise up2dateErrors.CommunicationError(message) def spawnRhnCheckForUI(): if os.access("/usr/sbin/rhn_check", os.R_OK\|os.X_OK): from subprocess import Popen, PIPE p = Popen(["/usr/sbin/rhn_check"], stdin=PIPE, stdout=PIPE, \ stderr=PIPE) map(lambda x:log.log_me(x), p.stdout.readlines() + \ p.stderr.readlines()) else: log.log_me("Warning: unable to run rhn_check") if getPlatform() == 'deb': def pluginEnable(): """On Debian no extra action for plugin is needed""" return 1, 0 else: from up2date_client.pmPlugin import pluginEnable	31.35 KB	2025-05-29 13:37
📄 rhnreg_constants.py # -- coding: utf-8 -- # # String constants for the RHN Register TUI/GUI. # Copyright (c) 2000--2016 Red Hat, Inc. # # Author: # James Slagle <jslagle@redhat.com> from up2date_client.pmPlugin import PM_PLUGIN_NAME, PM_NAME import gettext t = gettext.translation('rhn-client-tools', fallback=True) # Python 3 translations don't have a ugettext method if not hasattr(t, 'ugettext'): t.ugettext = t.gettext _ = t.ugettext COPYRIGHT_TEXT = _(u"Copyright © 2006--2014 Red Hat, Inc. All rights reserved.") # Satellite URL Window SATELLITE_URL_WINDOW = _("Enter your CloudLinux Network URL.") SATELLITE_URL_TEXT = _("Please enter the location of your CloudLinux Network " "server and of its SSL " "certificate. The SSL certificate is only required " "if you will be connecting over https (recommended).") SATELLITE_URL_PROMPT = _("Satellite URL:") SATELLITE_URL_PROMPT2 = _("SSL certificate:") SATELLITE_REQUIRED = _("You must enter a valid Satellite URL.") SSL_REQUIRED = _("If you are using https you must enter the location " "of a valid SSL certificate.") # Connect Window CONNECT_WINDOW = _("Attempting to contact the Spacewalk server.") CONNECT_WINDOW_TEXT = _("We are attempting to contact the CloudLinux Network " "Network server at %s.") CONNECT_WINDOW_TEXT2 = _("A proxy was specified at %s.") # Start Window START_REGISTER_WINDOW = _("System Registration") START_REGISTER_TEXT = _("This assistant will guide you through " "connecting your system to " "CloudLinux Network to receive software " "updates, including " "security updates, to keep your system supported " "and compliant. " "You will need the following at this time:\n\n" " * A network connection\n" " * Your CloudLinux Login & password\n" " * The location of a CloudLinux Network " "or Proxy\n\n") # Why Register Window WHY_REGISTER = _("Why Should I Connect to CLN? ...") WHY_REGISTER_WINDOW = _("Why Register") WHY_REGISTER_TEXT = _("Connecting your system to CloudLinux Network allows you to take full " "advantage of the benefits of a paid subscription, including:") WHY_REGISTER_SEC = _("Security & Updates:") WHY_REGISTER_DLD = _("Downloads & Upgrades:") WHY_REGISTER_SUPP = _("Support:") WHY_REGISTER_COMP = _("Compliance:") WHY_REGISTER_SEC_TXT = _("Receive the latest software updates, including security updates, keeping this " "CloudLinux system updated and secure.") WHY_REGISTER_DLD_TXT = _("Download installation images for CloudLinux releases, " "including new releases.") WHY_REGISTER_SUPP_TXT = _("Access to the technical support experts at CloudLinux for help " "with any issues you might encounter with this system.") WHY_REGISTER_COMP_TXT = _("Stay in compliance with your subscription agreement " "and manage subscriptions " "for systems connected to your account.") WHY_REGISTER_TIP = _("Tip: CloudLinux values your privacy: " "http://www.cloudlinux.com/company/privacy/") BACK_REGISTER = _("Take me back to the registration") # Confirm Quit Window CONFIRM_QUIT = _("Software Update Not Set Up") CONFIRM_QUIT_SURE = _("Are you sure you don't want to connect your system to CloudLinux Network? " "You'll miss out on the benefits of a CloudLinux subscription:\n") CONFIRM_QUIT_WILLNOT = _("You will not be able to take advantage of these subscription privileges without connecting " "your system to CloudLinux Network.\n") CONTINUE_REGISTERING = _("Take me back to the setup process.") REGISTER_LATER2 = _("I'll register later.") # Info Window REGISTER_WINDOW = _("CloudLinux Account") LOGIN_PROMPT = _("Please enter your login information for the %s " "CloudLinux Network:\n\n") HOSTED_LOGIN = _("CloudLinux Login:") LOGIN = _("Login:") PASSWORD = _("Password:") LOGIN_TIP = _("Tip: Forgot your login or password? Contact your " "Satellite's Organization Administrator.") USER_REQUIRED = _("Please enter a desired login.") PASSWORD_REQUIRED = _("Please enter and verify a password.") # OS Release Window SELECT_OSRELEASE = _("Operating System Release Version") OS_VERSION = _("Operating System version:") MINOR_RELEASE = _(" Minor Release: ") LIMITED_UPDATES = _("Limited Updates Only") ALL_UPDATES = _("All available updates") CONFIRM_OS_RELEASE_SELECTION = _("Confirm operating system release selection") CONFIRM_OS_ALL = _("Your system will be subscribed to the base" " software channel to receive all available" " updates.") # Hardware Window HARDWARE_WINDOW = _("Create Profile - Hardware") HARDWARE_WINDOW_DESC1 = _("A Profile Name is a descriptive name that" " you choose to identify this System Profile" " on the CloudLinux Network web pages. Optionally," " include a computer serial or identification number.") HARDWARE_WINDOW_DESC2 = _("Additional hardware information including PCI" " devices, disk sizes and mount points will be" " included in the profile.") HARDWARE_WINDOW_CHECKBOX = _("Include the following information about hardware" " and network:") # Packages Window PACKAGES_WINDOW = _("Create Profile - Packages") PACKAGES_WINDOW_DESC1 = _("RPM information is important to determine what" " updated software packages are relevant to this" " system.") PACKAGES_WINDOW_DESC2 = _("Include RPM packages installed on this system" " in my System Profile") PACKAGES_WINDOW_UNCHECK = _("You may deselect individual packages by" " unchecking them below.") PACKAGES_WINDOW_PKGLIST = _("Building Package List") # Product Window EMAIL = _("Email Address:") SYSTEM_ALREADY_SETUP = _("System Already Registered") SYSTEM_ALREADY_REGISTERED = _("It appears this system has already been set up for software updates:") SYSTEM_ALREADY_REGISTERED_CONT = _("Are you sure you would like to continue?") RHSM_SYSTEM_ALREADY_REGISTERED = _("This system has already been registered using CloudLinux Subscription Management.\n\n" "Your system is being registered again using CloudLinux Network" " or CloudLinux Network Proxy technology. CloudLinux recommends that customers only register once.\n\n" "To learn more about RHN Classic/CloudLinux Network registration and technologies please consult this" " Knowledge Base Article: https://access.redhat.com/kb/docs/DOC-45563") # Send Window SEND_WINDOW = _("Send Profile Information to CloudLinux Network") SEND_WINDOW_DESC = _("We are finished collecting information for the System Profile.\n\n" "Press \"Next\" to send this System Profile to CloudLinux Network. " "Click \"Cancel\" and no information will be sent. " "You can run the registration program later by " "typing `rhn_register` at the command line.") # Sending Window SENDING_WINDOW = _("Sending Profile to CloudLinux Network") # Finish Window FINISH_WINDOW = _("Updates Configured") FINISH_WINDOW_TEXT_TUI = _("You may now run '%s update' from this system's " "command line to get the latest " "software updates from CloudLinux Network. You will need to run this " "periodically to " "get the latest updates. Alternatively, you may configure this " "system for automatic software updates (also known as 'auto errata update') " "via the CloudLinux Network web interface. (Instructions for this are in chapter 6 " "of the CLN Reference Guide, available from the 'Help' button in the main Red " "Hat Network Satellite web interface.)") % PM_NAME # Review Window REVIEW_WINDOW = _("Review Subscription") REVIEW_WINDOW_PROMPT = _("Please review the subscription details below:") SUB_NUM = _("The installation number %s was activated during " "this system's initial connection to CloudLinux Network.") SUB_NUM_RESULT = _("Subscriptions have been activated for the following " "CloudLinux products/services:") CHANNELS_TITLE = _("Software Channel Subscriptions:") OK_CHANNELS = _("This system will receive updates from the " "following software channels:") CHANNELS_SAT_WARNING = _("Warning: Only installed product listed above will receive " "updates and support. If you would like " "to receive updates for additional products, please " "login to your satellite web interface " "and subscribe this system to the appropriate " "software channels. See Kbase article " "for more details. " "(http://kbase.redhat.com/faq/docs/DOC-11313)") PM_PLUGIN_WARNING = _("Warning: %s is not present, could not enable it.\n" "Automatic updates will not work.") % (PM_PLUGIN_NAME) PM_PLUGIN_CONF_CHANGED = _("Note: %s has been enabled.") % (PM_PLUGIN_NAME) PM_PLUGIN_CONF_ERROR = _("Warning: An error occurred during enabling %s.\n" "%s is not enabled.\n" "Automatic updates will not work.") % (PM_PLUGIN_NAME, PM_PLUGIN_NAME) FAILED_CHANNELS = _("You were unable to be subscribed to the following " "software channels because there were insufficient " "subscriptions available in your account:") NO_BASE_CHANNEL = _( "This system was unable to subscribe to any software channels. Your system " "will not receive any software updates to keep it secure and supported. " "Contact your Satellite administrator about this problem. Once you make the " "appropriate active subscriptions available in your account, you may browse " "to this system's profile in the CLN web interface and subscribe this system " "to software channels via the software > software channels tab.") SLOTS_TITLE = _("Service Level:") OK_SLOTS = _("Depending on what CloudLinux Network modules are associated with a system, you'll " "enjoy different benefits. The following are the " "CloudLinux Network modules associated with this system:") SLOTS = SLOTS_TITLE + "\n" + OK_SLOTS + "\n%s" FAILED_SLOTS = _("This system was unable to be associated with the " "following CLN module(s) because there were " "insufficient subscriptions available in your account:") MANAGEMENT = _("Management module: automatic updates, systems " "grouping, systems permissions, system package profiling, " "bare-metal provisioning, existing state provisioning, " "rollbacks, configuration management") VIRT = _("Virtualization module: software updates for a limited number of " "virtual guests on this system.") VIRT_FAILED = _("<b>Warning:</b> Any guest systems you create on this system " "and register to RHN will consume CloudLinux " "subscriptions beyond this host system's subscription. You will need " "to: (1) make a virtualization system entitlement available and " "(2) apply that system entitlement to this " "system in CLN's web interface if you do not want virtual guests of " "this system to consume additional subscriptions.") NO_SYS_ENTITLEMENT = _("This system was unable to be associated with " "any CLN service level modules. This system will not receive any software " "updates to keep it secure and supported. Contace your Satellite administrator " "about this problem. Once you make the " "appropriate active subscriptions available in your account, you may browse " "to this system's profile in the CLN web interface, delete the profile, and " "re-connect this system to CloudLinux Network.") ACTIVATION_KEY = _("Universal default activation key detected\n" "A universal default activation key was detected in your account. " "This means that a set of properties (software channel subscriptions, " "package installations, system group memberships, etc.) " "for your system's connection to CloudLinux Network or CloudLinux Network Proxy" "have been determined by the activation key rather than your " "installation number. " "You may also refer to the RHN Reference Guide, section 6.4.6 for more details " "about activation keys (http://access.redhat.com/knowledge/docs/Red_Hat_Network/)\n" "Universal Default activation key: %s") # Error Messages. FATAL_ERROR = _("Fatal Error") WARNING = _("Warning") HOSTED_CONNECTION_ERROR = _("We can't contact the CloudLinux Network.\n\n" "Double check the location provided - is '%s' correct?\n" "If not, you can correct it and try again.\n\n" "Make sure that the network connection on this system is operational.\n\n" "This system will not be able to successfully receive software updates " "from CloudLinux without connecting to a CloudLinux Network server") BASECHANNELERROR = _("Architecture: %s, OS Release: %s, OS " "Version: %s") SERVER_TOO_OLD = _("This server doesn't support functionality " "needed by this version of the software update" " setup client. Please try again with a newer " "server.") SSL_CERT_ERROR_MSG = _("<b><span size=\"16000\">Incompatible Certificate File</span></b>\n\n" "The certificate you provided, <b>%s</b>, is not compatible with " " the CloudLinux Network server at <b>%s</b>. You may want to double-check" " that you have provided a valid certificate file." " Are you sure you have provided the correct certificate, and that" " the certificate file has not been corrupted?\n\n" "Please try again with a different certificate file.") SSL_CERT_EXPIRED = _("<b><span size=\"12000\">Incompatible Certificate File</span></b>\n\n" " The certificate is expired. Please ensure you have the correct " " certificate and your system time is correct.") SSL_CERT_FILE_NOT_FOUND_ERRER = _("Please verify the values of sslCACert and serverURL in " "/etc/sysconfig/rhn/up2date. You can either make the " "serverURL use http instead of https, or you can " "download the SSL cert from your Satellite, place it " "in /usr/share/rhn, and ensure sslCACert points to it.") ACT_KEY_USAGE_LIMIT_ERROR = _("Problem registering system.\n\n" "A universal default activation key limits the " "number of systems which can connect to " "the CLN organization associated with your " "login. To allow this system to connect, " "please contact your CLN organization " "administrator to increase the number of " "systems allowed to connect or to disable " "this universal default activation key. " "More details can be found in CloudLinux " "Knowledgebase Article #7924 at " "http://kbase.redhat.com/faq/FAQ_61_7924.shtm ") CHANNEL_PAGE_TIP = _("\n Tip: Minor releases with a '' are currently" " supported by CloudLinux.\n\n") CHANNEL_PAGE_WARNING = _("Warning:You will not be able to limit this" " system to minor release that is older than" " the recent minor release if you select this" " option.\n") CONFIRM_OS_WARNING = _("Your system will be subscribed to %s \n" "base software channel. You will not be\n" "able to move this system to an earlier release\n" "(you will be able to move to a newer release).\n" "Are you sure you would like to continue?") # Navigation OK = _("OK") ERROR = _("Error") NEXT = _("Next") BACK = _("Back") CANCEL = _("Cancel") NO_CANCEL = _("No, Cancel") YES_CONT = _("Yes, Continue") DESELECT = _("Press <space> to deselect the option.")	18.13 KB	2025-05-29 13:37
📄 rhnserver.py # rhn-client-tools # # Copyright (c) 2006--2016 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; version 2 of the License. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program; if not, write to the Free Software # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA # 02110-1301 USA # # In addition, as a special exception, the copyright holders give # permission to link the code of portions of this program with the # OpenSSL library under certain conditions as described in each # individual source file, and distribute linked combinations # including the two. # You must obey the GNU General Public License in all respects # for all of the code used other than OpenSSL. If you modify # file(s) with this exception, you may extend this exception to your # version of the file(s), but you are not obligated to do so. If you # do not wish to do so, delete this exception statement from your # version. If you delete this exception statement from all source # files in the program, then also delete it here. from rhn.tb import raise_with_tb from up2date_client import rpcServer from up2date_client import up2dateErrors from up2date_client import capabilities import sys import OpenSSL try: # python2 import xmlrpclib except ImportError: # python3 import xmlrpc.client as xmlrpclib class _DoCallWrapper(object): """ A callable object that will handle multiple levels of attributes, and catch exceptions. """ def __init__(self, server, method_name): self._server = server self._method_name = method_name def __getattr__(self, method_name): """ Recursively build up the method name to pass to the server. """ return _DoCallWrapper(self._server, "%s.%s" % (self._method_name, method_name)) def __call__(self, args, kwargs): """ Call the method. Catch faults and translate them. """ method = getattr(self._server, self._method_name) try: return rpcServer.doCall(method, args, **kwargs) except xmlrpclib.Fault: raise_with_tb(self.__exception_from_fault(sys.exc_info()[1])) except OpenSSL.SSL.Error: # TODO This should probably be moved to rhnlib and raise an # exception that subclasses OpenSSL.SSL.Error # TODO Is there a better way to detect cert failures? error = str(sys.exc_info()[1]) error = error.strip("[()]") pieces = error.split(',') message = "" if len(pieces) > 2: message = pieces[2] elif len(pieces) == 2: message = pieces[1] message = message.strip(" '") if message == 'certificate verify failed': raise_with_tb(up2dateErrors.SSLCertificateVerifyFailedError()) else: raise_with_tb(up2dateErrors.NetworkError(message)) def __exception_from_fault(self, fault): if fault.faultCode == -3: # This username is already taken, or the password is incorrect. exception = up2dateErrors.AuthenticationOrAccountCreationError(fault.faultString) elif fault.faultCode == -2: # Invalid username and password combination. exception = up2dateErrors.AuthenticationOrAccountCreationError(fault.faultString) elif fault.faultCode == -110: # Account is disabled exception = up2dateErrors.AuthenticationOrAccountCreationError(fault.faultString) elif fault.faultCode == -1: exception = up2dateErrors.UnknownMethodException(fault.faultString) elif fault.faultCode == -13: # Username is too short. exception = up2dateErrors.LoginMinLengthError(fault.faultString) elif fault.faultCode == -14: # too short password exception = up2dateErrors.PasswordMinLengthError( fault.faultString) elif fault.faultCode == -15: # bad chars in username exception = up2dateErrors.ValidationError(fault.faultString) elif fault.faultCode == -16: # Invalid product registration code. # TODO Should this really be a validation error? exception = up2dateErrors.ValidationError(fault.faultString) elif fault.faultCode == -19: # invalid exception = up2dateErrors.NoBaseChannelError(fault.faultString) elif fault.faultCode == -31: # No entitlement exception = up2dateErrors.InsuffMgmntEntsError(fault.faultString) elif fault.faultCode == -36: # rhnException.py says this means "Invalid action." # TODO find out which is right exception = up2dateErrors.PasswordError(fault.faultString) elif abs(fault.faultCode) == 49: exception = up2dateErrors.AbuseError(fault.faultString) elif abs(fault.faultCode) == 60: exception = up2dateErrors.AuthenticationTicketError(fault.faultString) elif abs(fault.faultCode) == 74: exception = up2dateErrors.RegistrationDeniedError() elif abs(fault.faultCode) == 105: exception = up2dateErrors.RhnUuidUniquenessError(fault.faultString) elif fault.faultCode == 99: exception = up2dateErrors.DelayError(fault.faultString) elif abs(fault.faultCode) == 91: exception = up2dateErrors.InsuffMgmntEntsError(fault.faultString) elif fault.faultCode == -106: # Invalid username. exception = up2dateErrors.ValidationError(fault.faultString) elif fault.faultCode == -600: # Invalid username. exception = up2dateErrors.InvalidRegistrationNumberError(fault.faultString) elif fault.faultCode == -601: # No entitlements associated with given hardware info exception = up2dateErrors.NotEntitlingError(fault.faultString) elif fault.faultCode == -602: # No entitlements associated with reg num exception = up2dateErrors.NotEntitlingError(fault.faultString) elif fault.faultCode == -2001 or fault.faultCode == -700: exception = up2dateErrors.AuthenticationOrAccountCreationError( fault.faultString) elif fault.faultCode == -701: exception = up2dateErrors.PasswordMaxLengthError( fault.faultString) elif fault.faultCode == -61: exception = up2dateErrors.ActivationKeyUsageLimitError( fault.faultString) elif fault.faultCode == -5: exception = up2dateErrors.UnableToCreateUser( fault.faultString) else: exception = up2dateErrors.CommunicationError(fault.faultString) return exception class RhnServer(object): """ An rpc server object that calls doCall for you, and catches lower level exceptions """ def __init__( self, serverOverride=None, timeout=None, rpcServerOverride=None, registration=False, ): if rpcServerOverride is None: self._server = rpcServer.getServer( serverOverride=serverOverride, timeout=timeout, registration=registration, ) else: self._server = rpcServerOverride self._capabilities = None def __get_capabilities(self): if self._capabilities is None: headers = self._server.get_response_headers() if headers is None: self.registration.welcome_message() headers = self._server.get_response_headers() self._capabilities = capabilities.Capabilities() self._capabilities.populate(headers) return self._capabilities capabilities = property(__get_capabilities) def add_header(self, key, value): self._server.add_header(key, value) def __getattr__(self, method_name): """Return a callable object that will do the work for us.""" return _DoCallWrapper(self._server, method_name) class RegistrationRhnServer(RhnServer): """ A specialized RhnServer subclass for handling registration-related calls. Intended to be used as a shorthand for registration tasks instead of the plain RhnServer. """ def __init__(self, serverOverride=None, timeout=None, rpcServerOverride=None): super(RegistrationRhnServer, self).__init__( serverOverride=serverOverride, timeout=timeout, rpcServerOverride=rpcServerOverride, registration=True, )	9.31 KB	2025-05-29 13:37
📄 rpcServer.py # import os import sys import socket import time from up2date_client import config from up2date_client import clientCaps from up2date_client import up2dateLog from up2date_client import up2dateErrors from up2date_client import up2dateUtils from rhn import SSL from rhn import rpclib from rhn.tb import raise_with_tb try: # python2 import httplib import urllib2 import urlparse import xmlrpclib except ImportError: # python3 import http.client as httplib import urllib.request as urllib2 import urllib.parse as urlparse import xmlrpc.client as xmlrpclib import gettext t = gettext.translation("rhn-client-tools", fallback=True) # Python 3 translations don't have a ugettext method if not hasattr(t, "ugettext"): t.ugettext = t.gettext _ = t.ugettext def stdoutMsgCallback(msg): print(msg) class RetryServer(rpclib.Server): # Dict of server: error_message pairs to keep track of failure messages and display them if needed _error_messages = {} def addServerList(self, serverList): self.serverList = serverList def _request1(self, methodname, params): self.log = up2dateLog.initLog() while 1: try: ret = self._request(methodname, params) except rpclib.InvalidRedirectionError: raise except xmlrpclib.Fault: raise except httplib.BadStatusLine: self.log.log_me("Error: Server Unavailable. Please try later.") stdoutMsgCallback(_("Error: Server Unavailable. Please try later.")) sys.exit(-1) except: server = self.serverList.next() if server is None: # since just because we failed, the server list could # change (aka, firstboot, they get an option to reset the # the server configuration) so reset the serverList self.serverList.resetServerIndex() # Display the error messages for all servers to stdout # This only happens if all servers in the list fail if self._error_messages: error_combined_msg = "\n".join( ["%s:\n%s" % (host, error) for host, error in self._error_messages.items()] ) # No need to log this, these are already logged at the same point where the error occurred stdoutMsgCallback(_("Errors occurred while trying to connect to the remote servers.")) stdoutMsgCallback(error_combined_msg) raise msg = "An error occurred talking to %s:\n" % self._host exception_msg = "%s\n%s\n" % (sys.exc_info()[0], sys.exc_info()[1]) # Store the error message for the failed server self._error_messages[self._host] = exception_msg msg = msg + exception_msg msg = msg + "Trying the next serverURL: %s\n" % self.serverList.server() self.log.log_me(msg) # try a different url # use the next serverURL parse_res = urlparse.urlsplit(self.serverList.server()) typ = parse_res[0] # scheme self._host = parse_res[1] # netloc self._handler = parse_res[2] # path typ = typ.lower() if typ not in ("http", "https"): raise_with_tb(rpclib.InvalidRedirectionError("Redirected to unsupported protocol %s" % typ)) self._orig_handler = self._handler self._type = typ self._uri = self.serverList.server() if not self._handler: self._handler = "/RPC2" self._allow_redirect = 1 continue # if we get this far, we succedded break return ret def __getattr__(self, name): # magic method dispatcher return rpclib.xmlrpclib._Method(self._request1, name) # uh, yeah, this could be an iterator, but we need it to work on 1.5 as well class ServerList: def __init__(self, serverlist=[]): self.serverList = serverlist self.index = 0 def server(self): self.serverurl = self.serverList[self.index] return self.serverurl def next(self): self.index = self.index + 1 if self.index >= len(self.serverList): return None return self.server() def resetServerIndex(self): self.index = 0 def getServer( refreshCallback=None, serverOverride=None, timeout=None, registration=False, ): log = up2dateLog.initLog() cfg = config.initUp2dateConfig() # Where do we keep the CA certificate for RHNS? # The servers we're talking to need to have their certs # signed by one of these CA. ca = cfg["sslCACert"] if not isinstance(ca, list): ca = [ca] rhns_ca_certs = ca or ["/usr/share/rhn/CLN-ORG-TRUSTED-SSL-CERT"] if cfg["enableProxy"]: proxyHost = config.getProxySetting() else: proxyHost = None if not serverOverride: serverUrls = config.getServerURL(registration=registration) # Check for a fallback server URL and add it to the list if available fallbackURL = config.getFallbackServerURL(registration=registration) # Ensure that none of the URLs from the fallback list are already in the serverUrls list if fallbackURL: for url in fallbackURL: if url not in serverUrls: serverUrls.append(url) else: serverUrls = serverOverride serverList = ServerList(serverUrls) proxyUser = None proxyPassword = None if cfg["enableProxyAuth"]: proxyUser = cfg["proxyUser"] or None proxyPassword = cfg["proxyPassword"] or None lang = None for env in "LANGUAGE", "LC_ALL", "LC_MESSAGES", "LANG": if env in os.environ: if not os.environ[env]: # sometimes unset continue lang = os.environ[env].split(":")[0] lang = lang.split(".")[0] break s = RetryServer( serverList.server(), refreshCallback=refreshCallback, proxy=proxyHost, username=proxyUser, password=proxyPassword, timeout=timeout, ) s.addServerList(serverList) s.add_header("X-Up2date-Version", up2dateUtils.version()) if lang: s.setlang(lang) # require CLN-ORG-TRUSTED-SSL-CERT file to be able to authenticate the SSL connections need_ca = [True for i in s.serverList.serverList if urlparse.urlparse(i)[0] == "https"] if need_ca: for rhns_ca_cert in rhns_ca_certs: if not os.access(rhns_ca_cert, os.R_OK): msg = "%s: %s" % (_("ERROR: can not find RHNS CA file"), rhns_ca_cert) log.log_me("%s" % msg) raise up2dateErrors.SSLCertificateFileNotFound(msg) # force the validation of the SSL cert s.add_trusted_cert(rhns_ca_cert) clientCaps.loadLocalCaps() # send up the capabality info headerlist = clientCaps.caps.headerFormat() for headerName, value in headerlist: s.add_header(headerName, value) return s def doCall(method, args, kwargs): log = up2dateLog.initLog() log.log_debug("rpcServer: Calling XMLRPC %s" % method.__dict__["_Method__name"]) cfg = config.initUp2dateConfig() ret = None attempt_count = 1 try: attempts = int(cfg["networkRetries"]) except ValueError: attempts = 1 if attempts <= 0: attempts = 1 while 1: failure = 0 ret = None try: ret = method(args, **kwargs) except KeyboardInterrupt: raise_with_tb(up2dateErrors.CommunicationError(_("Connection aborted by the user"))) # if we get a socket error, keep tryingx2 except (socket.error, SSL.socket_error): log.log_me("A socket error occurred: %s, attempt #%s" % (sys.exc_info()[1], attempt_count)) if attempt_count >= attempts: e = sys.exc_info()[1] if len(e.args) > 1: raise_with_tb(up2dateErrors.CommunicationError(e.args[1])) else: raise_with_tb(up2dateErrors.CommunicationError(e.args[0])) else: failure = 1 except httplib.IncompleteRead: print("httplib.IncompleteRead") raise_with_tb(up2dateErrors.CommunicationError("httplib.IncompleteRead")) except urllib2.HTTPError: e = sys.exc_info()[1] msg = "\nAn HTTP error occurred:\n" msg = msg + "URL: %s\n" % e.filename msg = msg + "Status Code: %s\n" % e.code msg = msg + "Error Message: %s\n" % e.msg log.log_me(msg) raise_with_tb(up2dateErrors.CommunicationError(msg)) except xmlrpclib.ProtocolError: e = sys.exc_info()[1] log.log_me("A protocol error occurred: %s , attempt #%s," % (e.errmsg, attempt_count)) if e.errcode == 404: log.log_me("Could not find URL, %s" % (e.url)) log.log_me("Check server name and/or URL, then retry\n") (errCode, errMsg) = rpclib.reportError(e.headers) reset = 0 if abs(errCode) == 34: log.log_me("Auth token timeout occurred\n errmsg: %s" % errMsg) # this calls login, which in tern calls doCall (ie, # this function) but login should never get a 34, so # should be safe from recursion from up2date_client import up2dateAuth up2dateAuth.updateLoginInfo() # the servers are being throttle to pay users only, catch the # exceptions and display a nice error message if abs(errCode) == 51: log.log_me(_("Server has refused connection due to high load")) raise_with_tb(up2dateErrors.CommunicationError(e.errmsg)) # if we get a 404 from our server, thats pretty # fatal... no point in retrying over and over. Note that # errCode == 17 is specific to our servers, if the # serverURL is just pointing somewhere random they will # get a 0 for errcode and will raise a CommunicationError if abs(errCode) == 17: # in this case, the args are the package string, so lets try to # build a useful error message if type(args[0]) == type([]): pkg = args[0] else: pkg = args[1] if type(pkg) == type([]): pkgName = "%s-%s-%s.%s" % (pkg[0], pkg[1], pkg[2], pkg[4]) else: pkgName = pkg msg = "File Not Found: %s\n%s" % (pkgName, errMsg) log.log_me(msg) raise_with_tb(up2dateErrors.FileNotFoundError(msg)) if not reset: if attempt_count >= attempts: raise_with_tb(up2dateErrors.CommunicationError(e.errmsg)) else: failure = 1 except xmlrpclib.ResponseError: raise_with_tb(up2dateErrors.CommunicationError("Broken response from the server.")) if ret != None: break else: failure = 1 if failure: # rest for five seconds before trying again time.sleep(5) attempt_count = attempt_count + 1 if attempt_count > attempts: raise up2dateErrors.CommunicationError("The data returned from the server was incomplete") return ret	11.72 KB	2025-05-29 13:37
📄 rpmUtils.py # some high level utility stuff for rpm handling # Client code for Update Agent # Copyright (c) 1999--2016 Red Hat, Inc. Distributed under GPLv2. # # Author: Preston Brown <pbrown@redhat.com> # Adrian Likins <alikins@redhat.com> # # # FIXME: Some exceptions in here are currently in up2date.py # fix by moving to up2dateErrors.py and importing from there # # import os import rpm from rhn.i18n import sstr from up2date_client import transaction import gettext t = gettext.translation('rhn-client-tools', fallback=True) # Python 3 translations don't have a ugettext method if not hasattr(t, 'ugettext'): t.ugettext = t.gettext _ = t.ugettext def installedHeaderByKeyword(kwargs): """ just cause this is such a potentially useful looking method... """ _ts = transaction.initReadOnlyTransaction() mi = _ts.dbMatch() for keyword in kwargs.keys(): mi.pattern(keyword, rpm.RPMMIRE_GLOB, kwargs[keyword]) # we really shouldnt be getting multiples here, but what the heck headerList = [] for h in mi: headerList.append(h) return headerList def verifyPackages(packages): """ given a list of package labels, run rpm -V on them and return a dict keyed off that data """ data = {} missing_packages = [] # data structure is keyed off package # label, with value being an array of each # line of the output from -V retlist = [] for package in packages: # we have to have at least name... # Note: we cant reliable match on epoch, so just # skip it... two packages that only diff by epoch is # way broken anyway keywords = {'name': package[0], 'version': package[1], 'release': package[2], # we left our epoch 'arch': package[4] } for key in (keywords.keys()): if (keywords[key] == None) or (keywords[key] == ""): del(keywords[key]) headers = installedHeaderByKeyword(keywords) if len(headers) == 0: missing_packages.append(package) for header in headers: epoch = header['epoch'] if epoch == None: epoch = "" # gpg-pubkey "packages" can have an arch of None, see bz #162701 arch = header["arch"] if arch == None: arch = "" pkg = (header['name'], header['version'], header['release'], epoch, arch) # dont include arch in the label if it's a None arch, #162701 if header["arch"] == "": packageLabel = "%s-%s-%s" % (pkg[0], pkg[1], pkg[2]) else: packageLabel = "%s-%s-%s.%s" % (pkg[0], pkg[1], pkg[2], pkg[4]) verifystring = "/usr/bin/rpmverify -V %s" % packageLabel fd = os.popen(verifystring) res = fd.readlines() fd.close() reslist = [] for line in res: reslist.append(line.strip()) retlist.append([pkg, reslist]) return retlist, missing_packages #FIXME: this looks like a good candidate for caching, since it takes a second # or two to run, and I can call it a couple of times def getInstalledPackageList(msgCallback = None, progressCallback = None, getArch=None, getInfo = None): """ Return list of packages. Package is hash with keys name, epoch, version, release and optionaly arch and cookie """ pkg_list = [] if msgCallback != None: msgCallback(_("Getting list of packages installed on the system")) _ts = transaction.initReadOnlyTransaction() count = 0 total = 0 for h in _ts.dbMatch(): if h == None: break count = count + 1 total = count count = 0 for h in _ts.dbMatch(): if h == None: break package = { 'name': sstr(h['name']), 'epoch': h['epoch'], 'version': sstr(h['version']), 'release': sstr(h['release']), 'installtime': h['installtime'] } if package['epoch'] == None: package['epoch'] = "" else: # convert it to string package['epoch'] = "%s" % package['epoch'] if getArch: package['arch'] = h['arch'] # the arch on gpg-pubkeys is "None"... if package['arch']: package['arch'] = sstr(package['arch']) pkg_list.append(package) elif getInfo: if h['arch']: package['arch'] = sstr(h['arch']) if h['cookie']: package['cookie'] = sstr(h['cookie']) pkg_list.append(package) else: pkg_list.append(package) if progressCallback != None: progressCallback(count, total) count = count + 1 pkg_list.sort(key=lambda x:(x['name'], x['epoch'], x['version'], x['release'])) return pkg_list def setDebugVerbosity(): """Set rpm's verbosity mode """ try: rpm.setVerbosity(rpm.RPMLOG_DEBUG) except AttributeError: print("extra verbosity not supported in this version of rpm")	5.2 KB	2025-05-29 13:37
📄 transaction.py # # Client code for Update Agent # Copyright (c) 1999--2016 Red Hat, Inc. Distributed under GPLv2. # # Adrian Likins <alikins@redhat.com # # # a couple of classes wrapping up transactions so that we # can share transactions instead of creating new ones all over # import rpm read_ts = None ts = None # *********** NOTE: **********# # for the sake of clarity, the names "added/removed" as used here # are indicative of what happened when the original transaction was # ran. Aka, if you "up2date foobar" and it updates foobar-1-0 with # foobar-2-0, you added foobar-2-0 and removed foobar-1-0 # # The reason I mention this explicitly is the trouble of describing # what happens when you rollback the transaction, which is basically # the opposite, and leads to plenty of confusion # class TransactionData: # simple data structure designed to transport info # about rpm transactions around def __init__(self): self.data = {} # a list of tuples of pkg info, and mode ('e', 'i', 'u') # the pkgInfo is tuple of [name, version, release, epoch, arch] # size is never used directly for this, it's here as a place holder # arch is optional, if the server specifies it, go with what # removed packages only need [n,v,r,e,arch] self.data['packages'] = [] # list of flags to set for the transaction self.data['flags'] = [] self.data['vsflags'] = [] self.data['probFilterFlags'] = [] def display(self): out = "" removed = [] installed = [] updated = [] misc = [] for (pkgInfo, mode) in self.data['packages']: if mode == 'u': updated.append(pkgInfo) elif mode == 'i': installed.append(pkgInfo) elif mode == 'e': removed.append(pkgInfo) else: misc.append(pkgInfo) for pkgInfo in removed: out = out + "\t\t[e] %s-%s-%s:%s\n" % (pkgInfo[0], pkgInfo[1], pkgInfo[2], pkgInfo[3]) for pkgInfo in installed: out = out + "\t\t[i] %s-%s-%s:%s\n" % (pkgInfo[0], pkgInfo[1], pkgInfo[2], pkgInfo[3]) for pkgInfo in updated: out = out + "\t\t[u] %s-%s-%s:%s\n" % (pkgInfo[0], pkgInfo[1], pkgInfo[2], pkgInfo[3]) for pkgInfo in misc: out = out + "\t\t[%s] %s-%s-%s:%s\n" % (pkgInfo[5], pkgInfo[0], pkgInfo[1], pkgInfo[2], pkgInfo[3]) return out # wrapper/proxy class for rpm.Transaction so we can # instrument it, etc easily class Up2dateTransaction: def __init__(self): self.ts = rpm.TransactionSet() self._methods = ['dbMatch', 'check', 'order', 'addErase', 'addInstall', 'run', 'IDTXload', 'IDTXglob', 'rollback', 'pgpImportPubkey', 'pgpPrtPkts', 'Debug', 'setFlags', 'setVSFlags', 'setProbFilter', 'hdrFromFdno'] self.tsflags = [] def __getattr__(self, attr): if attr in self._methods: return self.getMethod(attr) else: raise AttributeError(attr) def getMethod(self, method): # in theory, we can override this with # profile/etc info return getattr(self.ts, method) # push/pop methods so we dont lose the previous # set value, and we can potentiall debug a bit # easier def pushVSFlags(self, flags): self.tsflags.append(flags) self.ts.setVSFlags(self.tsflags[-1]) def popVSFlags(self): del self.tsflags[-1] self.ts.setVSFlags(self.tsflags[-1]) def initReadOnlyTransaction(): global read_ts if read_ts == None: read_ts = Up2dateTransaction() # FIXME: replace with macro defination read_ts.pushVSFlags(-1) return read_ts	4.09 KB	2025-05-29 13:37
📄 tui.py # # TUI for RHN Registration # Copyright (c) 2000--2020 Red Hat, Inc. # # This software is licensed to you under the GNU General Public License, # version 2 (GPLv2). There is NO WARRANTY for this software, express or # implied, including the implied warranties of MERCHANTABILITY or FITNESS # FOR A PARTICULAR PURPOSE. You should have received a copy of GPLv2 # along with this software; if not, see # http://www.gnu.org/licenses/old-licenses/gpl-2.0.txt. # # Red Hat trademarks are not licensed under GPLv2. No permission is # granted to use or replicate Red Hat trademarks that are incorporated # in this software or its documentation. # from os import geteuid import sys import gettext t = gettext.translation('rhn-client-tools', fallback=True) # Python 3 translations don't have a ugettext method if not hasattr(t, 'ugettext'): t.ugettext = t.gettext _ = t.ugettext import snack import signal from up2date_client import rhnreg, hardware from up2date_client import up2dateErrors from up2date_client import up2dateUtils from up2date_client import pkgUtils from up2date_client import up2dateLog from up2date_client import config from up2date_client.config import convert_url_from_puny from up2date_client import up2dateAuth from rhn import rpclib from rhn.connections import idn_puny_to_unicode from rhn.i18n import sstr from up2date_client.pmPlugin import PM_PLUGIN_NAME, PM_PLUGIN_CONF from up2date_client.rhnreg_constants import * log = up2dateLog.initLog() cfg = config.initUp2dateConfig() def ErrorWindow(screen, errmsg): snack.ButtonChoiceWindow(screen, sstr(ERROR), sstr(u"%s" % errmsg), [sstr(BACK)]) def FatalErrorWindow(screen, errmsg): snack.ButtonChoiceWindow(screen, sstr(FATAL_ERROR), sstr(u"%s" % errmsg), [sstr(OK)]) screen.finish() sys.exit(1) def WarningWindow(screen, errmsg): snack.ButtonChoiceWindow(screen, sstr(WARNING), sstr("%s" % errmsg), [sstr(OK)]) screen.finish() def ConfirmQuitWindow(screen): button = snack.ButtonChoiceWindow(screen, sstr(CONFIRM_QUIT), sstr(CONFIRM_QUIT_SURE) + "\n" + \ sstr(WHY_REGISTER_SEC) + "\n" + \ sstr(WHY_REGISTER_SEC_TXT) + "\n\n" + \ sstr(WHY_REGISTER_DLD) + "\n" + \ sstr(WHY_REGISTER_DLD_TXT) + "\n\n" + \ sstr(WHY_REGISTER_SUPP) + "\n" + \ sstr(WHY_REGISTER_SUPP_TXT) + "\n\n" + \ sstr(WHY_REGISTER_COMP) + "\n" + \ sstr(WHY_REGISTER_COMP_TXT) + "\n\n" + \ sstr(CONFIRM_QUIT_WILLNOT) + "\n" + \ sstr(WHY_REGISTER_TIP), [sstr(CONTINUE_REGISTERING), sstr(REGISTER_LATER2)], width = 70) if button == sstr(REGISTER_LATER2).lower(): screen.finish() return 1 else: return 0 def tui_call_wrapper(screen, func, params): try: results = func(params) except up2dateErrors.CommunicationError: ErrorWindow(screen, HOSTED_CONNECTION_ERROR % config.getServerURL()[0]) raise sys.exc_info()[1] except up2dateErrors.SSLCertificateVerifyFailedError: ErrorWindow(screen, e.errmsg) raise sys.exc_info()[1] except up2dateErrors.NoBaseChannelError: e = sys.exc_info()[1] FatalErrorWindow(screen, e.errmsg + '\n' + BASECHANNELERROR % (up2dateUtils.getArch(), up2dateUtils.getOSRelease(), up2dateUtils.getVersion())) except up2dateErrors.SSLCertificateFileNotFound: e = sys.exc_info()[1] ErrorWindow(screen, e.errmsg + '\n\n' + SSL_CERT_FILE_NOT_FOUND_ERRER) raise e return results class WindowSkipException(Exception): pass class AlreadyRegisteredWindow: name = "AlreadyRegisteredWindow" def __init__(self, screen, tui): if not rhnreg.registered() or tui.test: raise WindowSkipException() self.screen = screen self.tui = tui size = snack._snack.size() systemIdXml = rpclib.xmlrpclib.loads(up2dateAuth.getSystemId()) oldUsername = systemIdXml[0][0]['username'] oldsystemId = systemIdXml[0][0]['system_id'] toplevel = snack.GridForm(self.screen, sstr(SYSTEM_ALREADY_SETUP), 1, 2) self.bb = snack.ButtonBar(self.screen, [(sstr(YES_CONT), "next"), (sstr(NO_CANCEL), "exit")]) toplevel.add(self.bb, 0, 1, growx = 1) tb = snack.Textbox(size[0]-30, size[1]-20, sstr(SYSTEM_ALREADY_REGISTERED + "\n\n" + _("Spacewalk Location:") + " " + convert_url_from_puny(self.tui.serverURL) + "\n" + _("Login:") + " " + oldUsername + "\n" + _("System ID:") + " " + oldsystemId + "\n\n" + SYSTEM_ALREADY_REGISTERED_CONT + "\n"), 1, 1) toplevel.add(tb, 0, 0, padding = (0, 0, 0, 1)) self.g = toplevel def saveResults(self): pass def run(self): log.log_debug("Running %s" % self.name) result = self.g.runOnce() button = self.bb.buttonPressed(result) if result == "F12": return "next" return button class SatelliteUrlWindow: name = "SatelliteUrlWindow" def __init__(self, screen, tui): self.screen = screen self.tui = tui self.tui.alreadyRegistered = 0 self.server = convert_url_from_puny(self.tui.serverURL) fixed_server_url = rhnreg.makeNiceServerUrl(self.server) #Save the config only if the url is different if fixed_server_url != self.server: self.server = fixed_server_url config.setServerURL(self.server) cfg.save() size = snack._snack.size() toplevel = snack.GridForm(screen, sstr(SATELLITE_URL_WINDOW), 1, 4) prompt_text = SATELLITE_URL_TEXT url_label = SATELLITE_URL_PROMPT ssl_label = SATELLITE_URL_PROMPT2 label = snack.Textbox(size[0]-10, 3, sstr(prompt_text), scroll = 0, wrap = 1) toplevel.add(label, 0, 0, anchorLeft = 1) # spacer label = snack.Label(sstr("")) toplevel.add(label, 0, 1) grid = snack.Grid(2, 3) label = snack.Label(sstr(url_label)) grid.setField(label, 0, 0, padding = (0, 0, 1, 0), anchorRight = 1) self.urlEntry = snack.Entry(40) self.urlEntry.set(self.server) grid.setField(self.urlEntry, 1, 0, anchorLeft = 1) label = snack.Label(sstr(ssl_label)) grid.setField(label, 0, 1, padding = (0, 0, 1, 0), anchorRight = 1) self.sslEntry = snack.Entry(40) self.sslEntry.set(tui.sslCACert) grid.setField(self.sslEntry, 1, 1, anchorLeft = 1) toplevel.add(grid, 0, 2) # BUTTON BAR self.bb = snack.ButtonBar(screen, [(sstr(NEXT), "next"), (sstr(BACK), "back"), (sstr(CANCEL), "cancel")]) toplevel.add(self.bb, 0, 3, padding = (0, 1, 0, 0), growx = 1) self.g = toplevel def validateFields(self): if self.urlEntry.value() == "": snack.ButtonChoiceWindow(self.screen, sstr(ERROR), sstr(SATELLITE_REQUIRED), buttons = [sstr(OK)]) self.g.setCurrent(self.urlEntry) return 0 if (self.urlEntry.value()[:5] == 'https' and self.sslEntry.value() == ""): snack.ButtonChoiceWindow(self.screen, sstr(ERROR), sstr(SSL_REQUIRED), buttons = [sstr(OK)]) self.g.setCurrent(self.sslEntry) return 0 return 1 def saveResults(self): serverEntry = self.urlEntry.value() # fix up the server url, E.G. if someone left off /XMLRPC fixed_server_url = rhnreg.makeNiceServerUrl(serverEntry) if fixed_server_url != serverEntry: serverEntry = fixed_server_url self.tui.serverURL = serverEntry self.tui.sslCACert = self.sslEntry.value() config.setServerURL(serverEntry) config.setSSLCACert(self.sslEntry.value()) cfg.save() def run(self): log.log_debug("Running %s" % self.name) self.screen.refresh() valid = 0 while not valid: result = self.g.run() button = self.bb.buttonPressed(result) if result == "F12": button = "next" if button == "next": valid = self.validateFields() else: break self.screen.popWindow() return button class AlreadyRegisteredSubscriptionManagerWindow: name = "AlreadyRegisteredSubscriptionManagerWindow" def __init__(self, screen, tui): if not rhnreg.rhsm_registered() or tui.test: raise WindowSkipException() self.screen = screen self.tui = tui size = snack._snack.size() toplevel = snack.GridForm(self.screen, sstr(SYSTEM_ALREADY_SETUP), 1, 2) self.bb = snack.ButtonBar(self.screen, [(sstr(YES_CONT), "next"), (sstr(NO_CANCEL), "exit")]) toplevel.add(self.bb, 0, 1, growx = 1) tb = snack.Textbox(size[0]-30, size[1]-20, sstr(WARNING + "\n\n" + RHSM_SYSTEM_ALREADY_REGISTERED + "\n\n" + SYSTEM_ALREADY_REGISTERED_CONT + "\n"), 1, 1) toplevel.add(tb, 0, 0, padding = (0, 0, 0, 1)) self.g = toplevel def saveResults(self): pass def run(self): log.log_debug("Running %s" % self.name) result = self.g.runOnce() button = self.bb.buttonPressed(result) if result == "F12": return "next" return button class ConnectWindow: name = "ConnectWindow" def __init__(self, screen, tui): self.screen = screen self.tui = tui size = snack._snack.size() self.server = convert_url_from_puny(self.tui.serverURL) self.proxy = cfg['httpProxy'] toplevel = snack.GridForm(self.screen, sstr(CONNECT_WINDOW), 1, 1) text = CONNECT_WINDOW_TEXT % self.server + "\n\n" if self.proxy: text += CONNECT_WINDOW_TEXT2 % self.proxy tb = snack.Textbox(size[0]-30, size[1]-20, sstr(text), 1, 1) toplevel.add(tb, 0, 0, padding = (0, 0, 0, 1)) self.g = toplevel def run(self): log.log_debug("Running %s" % self.name) # We draw and display the window. The window gets displayed as # long as we are attempting to connect to the server. Once we # connect the window is gone. result = self.g.draw() self.screen.refresh() # try to connect given the server url and ssl cert provided. If # unsuccessful, return to previous screen to allow user to fix. try: tui_call_wrapper(self.screen, rhnreg.getCaps) except: return "back" self.screen.popWindow() # Just return next, although the user wouldn't have actually pressed # anything. return "next" def saveResults(self): pass class StartWindow: name = "StartWindow" def __init__(self, screen, tui): self.screen = screen self.tui = tui size = snack._snack.size() toplevel = snack.GridForm(self.screen, sstr(START_REGISTER_WINDOW), 1, 2) start_register_text = sstr(START_REGISTER_TEXT) tb = snack.Textbox(size[0]-10, size[1]-14, start_register_text, 1, 1) toplevel.add(tb, 0, 0, padding = (0, 0, 0, 1)) self.bb = snack.ButtonBar(self.screen, [(sstr(WHY_REGISTER), "why_register"), (sstr(NEXT), "next"), (sstr(CANCEL), "cancel")]) toplevel.add(self.bb, 0, 1, growx = 1) self.g = toplevel def saveResults(self): pass def run(self): log.log_debug("Running %s" % self.name) result = self.g.runOnce() button = self.bb.buttonPressed(result) if result == "F12": return "next" elif button == "why_register": why_reg_win = WhyRegisterWindow(self.screen, self.tui) why_reg_win.run() return button return button class WhyRegisterWindow: name = "WhyRegisterWindow" def __init__(self, screen, tui): self.screen = screen self.tui = tui size = snack._snack.size() toplevel = snack.GridForm(self.screen, sstr(WHY_REGISTER_WINDOW), 1, 2) why_register_text = WHY_REGISTER_TEXT + "\n\n" + \ WHY_REGISTER_SEC + "\n" + \ WHY_REGISTER_SEC_TXT + "\n\n" + \ WHY_REGISTER_DLD + "\n" + \ WHY_REGISTER_DLD_TXT + "\n\n" + \ WHY_REGISTER_SUPP + "\n" + \ WHY_REGISTER_SUPP_TXT + "\n\n" + \ WHY_REGISTER_COMP + "\n" + \ WHY_REGISTER_COMP_TXT + "\n\n" + \ WHY_REGISTER_TIP tb = snack.Textbox(size[0]-10, size[1]-14, sstr(why_register_text), 1, 1) toplevel.add(tb, 0, 0, padding = (0, 0, 0, 1)) self.bb = snack.ButtonBar(self.screen, [(sstr(BACK_REGISTER), "back")]) toplevel.add(self.bb, 0, 1, growx = 1) self.g = toplevel def run(self): log.log_debug("Running %s" % self.name) result = self.g.runOnce() button = self.bb.buttonPressed(result) return button class InfoWindow: name = "InfoWindow" def __init__(self, screen, tui): self.screen = screen self.tui = tui self.tui.alreadyRegistered = 0 self.server = self.tui.serverURL size = snack._snack.size() toplevel = snack.GridForm(screen, sstr(REGISTER_WINDOW), 1, 4) decoded_server = convert_url_from_puny(self.server) url = self.server if decoded_server != self.server: url += " (%s)" % decoded_server login_prompt = LOGIN_PROMPT % url login_label = LOGIN login_tip = LOGIN_TIP label = snack.Textbox(size[0]-10, 3, sstr(login_prompt), scroll = 0, wrap = 1) toplevel.add(label, 0, 0, anchorLeft = 1) grid = snack.Grid(2, 3) label = snack.Label(sstr(login_label)) grid.setField(label, 0, 0, padding = (0, 0, 1, 0), anchorRight = 1) self.userNameEntry = snack.Entry(20) self.userNameEntry.set(tui.userName) grid.setField(self.userNameEntry, 1, 0, anchorLeft = 1) label = snack.Label(sstr(PASSWORD)) grid.setField(label, 0, 1, padding = (0, 0, 1, 0), anchorRight = 1) try: self.passwordEntry = snack.Entry(20, password = 1) except TypeError: self.passwordEntry = snack.Entry(20, hidden = 1) self.passwordEntry.set(tui.password) grid.setField(self.passwordEntry, 1, 1, anchorLeft = 1) toplevel.add(grid, 0, 1) label = snack.TextboxReflowed(size[0]-10, sstr(login_tip)) toplevel.add(label, 0, 2, anchorLeft=1) # BUTTON BAR self.bb = snack.ButtonBar(screen, [(sstr(NEXT), "next"), (sstr(BACK), "back"), (sstr(CANCEL), "cancel")]) toplevel.add(self.bb, 0, 3, padding = (0, 1, 0, 0), growx = 1) self.g = toplevel def validateFields(self): if self.userNameEntry.value() == "": snack.ButtonChoiceWindow(self.screen, sstr(ERROR), sstr(USER_REQUIRED), buttons = [sstr(OK)]) self.g.setCurrent(self.userNameEntry) return 0 if self.passwordEntry.value() == "": snack.ButtonChoiceWindow(self.screen, sstr(ERROR), sstr(PASSWORD_REQUIRED), buttons = [sstr(OK)]) self.g.setCurrent(self.passwordEntry) return 0 try: self.tui.alreadyRegistered = rhnreg.reserveUser(self.userNameEntry.value(), self.passwordEntry.value()) except up2dateErrors.ValidationError: e = sys.exc_info()[1] snack.ButtonChoiceWindow(self.screen, sstr(_("Error")), sstr(_("The server indicated an error:\n")) + sstr(e.errmsg), buttons = [sstr(_("OK"))]) self.g.setCurrent(self.userNameEntry) return 0 except up2dateErrors.CommunicationError: e = sys.exc_info()[1] FatalErrorWindow(self.screen, _("There was an error communicating with the registration server:\n") + e.errmsg) return 1 def saveResults(self): self.tui.userName = self.userNameEntry.value() self.tui.password = self.passwordEntry.value() def run(self): log.log_debug("Running %s" % self.name) self.screen.refresh() valid = 0 while not valid: result = self.g.run() button = self.bb.buttonPressed(result) if result == "F12": button = "next" if button == "next": valid = self.validateFields() else: break self.screen.popWindow() return button class OSReleaseWindow: name = "OSReleaseWindow" def __init__(self, screen, tui): self.tui = tui if not rhnreg.server_supports_eus(): log.log_debug("Server does not support EUS, skipping OSReleaseWindow") raise WindowSkipException() self.available_channels = rhnreg.getAvailableChannels( tui.userName, tui.password) if len(self.available_channels['channels']) < 1: log.log_debug("No available EUS channels, skipping OSReleaseWindow") raise WindowSkipException() self.screen = screen self.size = snack._snack.size() self.selectChannel = False toplevel = snack.GridForm(self.screen, sstr(SELECT_OSRELEASE), 1, 7) self.g = toplevel self.ostext = snack.TextboxReflowed(self.size[0]-10, sstr(OS_VERSION)) toplevel.add(self.ostext, 0, 0, anchorLeft = 1) optiontext1 = sstr(LIMITED_UPDATES) if self.tui.limited_updates_button: self.limited_updates_button = snack.SingleRadioButton(optiontext1, None, isOn = 1) else: self.limited_updates_button = snack.SingleRadioButton(optiontext1, None) toplevel.add(self.limited_updates_button, 0, 1, padding = (0, 1, 0, 1), anchorLeft = 1) self.sublabel = snack.Label(sstr(MINOR_RELEASE)) toplevel.add(self.sublabel, 0, 2, anchorLeft = 1) self.channelList = snack.Listbox(self.size[1]-22, 1, width = self.size[0]-10) toplevel.add(self.channelList, 0, 3) for key, value in sorted(self.available_channels['channels'].items(), key=lambda a:a[0]): if key in self.available_channels['receiving_updates']: value = value + "" self.channelList.append(" " + value, key) self.tip = snack.TextboxReflowed(self.size[0]-10, sstr(CHANNEL_PAGE_TIP)) toplevel.add(self.tip, 0, 4, anchorLeft = 1) optiontext2 = sstr(ALL_UPDATES) if self.tui.all_updates_button: self.all_updates_button = snack.SingleRadioButton(optiontext2, self.limited_updates_button, isOn=1) else: self.all_updates_button = snack.SingleRadioButton(optiontext2, self.limited_updates_button) toplevel.add(self.all_updates_button, 0, 5, padding = (0, 0, 0, 1), anchorLeft = 1) #self.warning = snack.TextboxReflowed(self.size[0]-10, # sstr(CHANNEL_PAGE_WARNING)) #toplevel.add(self.warning, 0, 9, anchorLeft = 1) self.bb = snack.ButtonBar(screen, [(sstr(NEXT), "next"), (sstr(BACK), "back"), (sstr(CANCEL), "cancel")]) toplevel.add(self.bb, 0, 6, growx = 1) self.screen.refresh() def run(self): log.log_debug("Running %s" % self.name) self.screen.refresh() valid = "cancel" while valid == "cancel": result = self.g.run() button = self.bb.buttonPressed(result) if result == "F12": button = "next" if button == "next": valid = self.validateFields() else: break self.screen.popWindow() return button def validateFields(self): msgbox = "ok" later_release = False if self.limited_updates_button.selected(): later_release = self.channelList.current() != \ self.available_channels['default_channel'] title = sstr(CONFIRM_OS_RELEASE_SELECTION) if later_release: msgbox = snack.ButtonChoiceWindow(self.screen, title, sstr(CONFIRM_OS_WARNING) % self.channelList.current(), buttons =[sstr(OK), sstr(CANCEL)]) return msgbox if self.all_updates_button.selected() or later_release: msgbox = snack.ButtonChoiceWindow(self.screen, title, sstr(CONFIRM_OS_ALL), buttons =[sstr(OK), sstr(CANCEL)]) return msgbox return msgbox def saveResults(self): # if limited updates save the channel and selction # for future use self.tui.other.pop('channel', None) if self.limited_updates_button.selected(): log.log_debug("Selected Channel %s" % self.channelList.current()) self.tui.other['channel'] = self.channelList.current() self.tui.limited_updates_button = self.limited_updates_button.selected() self.tui.all_updates_button = 0 # saving data for all updates button if self.all_updates_button.selected(): self.tui.all_updates_button = self.all_updates_button.selected() self.tui.limited_updates_button = 0 class HardwareWindow: name = "HardwareWindow" def __init__(self, screen, tui): self.screen = screen self.tui = tui size = snack._snack.size() #get the virtualization uuid and set it to other. (virt_uuid, virt_type) = rhnreg.get_virt_info() if not virt_uuid is None: self.tui.other['virt_uuid'] = virt_uuid self.tui.other['virt_type'] = virt_type # read all hardware in tui.hardware = hardware.Hardware() toplevel = snack.GridForm(screen, sstr(HARDWARE_WINDOW), 1, 7) text = snack.TextboxReflowed(70, sstr(HARDWARE_WINDOW_DESC1)) toplevel.add(text, 0, 0, anchorLeft = 1) grid = snack.Grid(2, 2) label = snack.Label(sstr(_("Profile name:"))) grid.setField(label, 0, 0, padding = (0, 0, 1, 0), anchorRight = 1) self.profileEntry = snack.Entry(40) grid.setField(self.profileEntry, 1, 0, anchorLeft = 1) toplevel.add(grid, 0, 1, anchorLeft = 1) if tui.includeHardware: self.hardwareButton = snack.Checkbox(sstr(HARDWARE_WINDOW_CHECKBOX), isOn = 1) else: self.hardwareButton = snack.Checkbox(sstr(HARDWARE_WINDOW_CHECKBOX)) toplevel.add(self.hardwareButton, 0, 2, padding = (0, 1, 0, 0), anchorLeft = 1) label = snack.Label(sstr(DESELECT)) toplevel.add(label, 0, 3, anchorLeft = 1, padding = (0, 0, 0, 1)) grid = snack.Grid(4, 3) hardware_text = '' hardware_text += _("Version: ") + sstr(up2dateUtils.getVersion()) + " " self.versionLabel = snack.Label(sstr(_("Version: "))) grid.setField(self.versionLabel, 0, 0, padding = (0, 0, 1, 0), anchorLeft = 1) self.versionLabel2 = snack.Label(sstr(up2dateUtils.getVersion())) grid.setField(self.versionLabel2, 1, 0, anchorLeft = 1) hardware_text += _("CPU model: ") for hw in tui.hardware: if hw['class'] == 'CPU': hardware_text += hw['model'] +"\n" hardware_text += _("Hostname: ") for hw in tui.hardware: if hw['class'] == 'NETINFO': unicode_hostname = idn_puny_to_unicode(hw['hostname']) hardware_text += unicode_hostname + "\n" if tui.profileName != "": self.profileEntry.set(tui.profileName) else: self.profileEntry.set(sstr(unicode_hostname)) hardware_text += _("CPU speed: ") for hw in tui.hardware: if hw['class'] == 'CPU': hardware_text += _("%d MHz") % hw['speed'] + " " hardware_text += _("IP Address: ") for hw in tui.hardware: if hw['class'] == 'NETINFO': if hw['ipaddr']: hardware_text += hw['ipaddr'] + " " elif hw['ip6addr']: hardware_text += hw['ip6addr'] + " " hardware_text += _("Memory: ") for hw in tui.hardware: if hw['class'] == 'MEMORY': hardware_text += _("%s megabytes") % hw['ram'] tb = snack.TextboxReflowed(80, sstr(hardware_text)) toplevel.add(tb, 0, 4) self.additionalHWLabel = snack.TextboxReflowed(size[0]-10, sstr(HARDWARE_WINDOW_DESC2)) toplevel.add(self.additionalHWLabel, 0, 5, padding = (0, 1, 0, 0), anchorLeft = 1) # BUTTON BAR self.bb = snack.ButtonBar(screen, [(sstr(NEXT), "next"), (sstr(BACK), "back"), (sstr(CANCEL), "cancel")]) toplevel.add(self.bb, 0, 6, padding = (0, 1, 0, 0), growx = 1) self.g = toplevel # self.screen.gridWrappedWindow(toplevel, 'HardwareWindow', 80, 14) def saveResults(self): self.tui.profileName = self.profileEntry.value() self.tui.includeHardware = self.hardwareButton.selected() def run(self): log.log_debug("Running %s" % self.name) result = self.g.runOnce() button = self.bb.buttonPressed(result) if result == "F12": return "next" return button class PackagesWindow: name = "PackagesWindow" def __init__(self, screen, tui): self.screen = screen self.tui = tui size = snack._snack.size() toplevel = snack.GridForm(screen, sstr(PACKAGES_WINDOW), 1, 5) self.g = toplevel text = snack.TextboxReflowed(size[0]-10, sstr(PACKAGES_WINDOW_DESC1)) toplevel.add(text, 0, 0, anchorLeft = 1) self.packagesButton = snack.Checkbox(sstr(PACKAGES_WINDOW_DESC2), 1) toplevel.add(self.packagesButton, 0, 1, padding = (0, 1, 0, 1), anchorLeft = 1) label = snack.Label(sstr(PACKAGES_WINDOW_UNCHECK)) toplevel.add(label, 0, 2, anchorLeft = 1) #self.packageList = snack.Listbox(size[1]-18, 1, width = size[0]-10) self.packageList = snack.CheckboxTree(size[1]-18, 1) toplevel.add(self.packageList, 0, 3) # do we need to read the packages from disk? if tui.packageList == []: self.pwin = snack.GridForm(screen, sstr(PACKAGES_WINDOW_PKGLIST), 1, 1) self.scale = snack.Scale(40, 100) self.pwin.add(self.scale, 0, 0) self.pwin.draw() self.screen.refresh() getArch = 0 if rhnreg.cfg['supportsExtendedPackageProfile']: getArch = 1 tui.packageList = pkgUtils.getInstalledPackageList(getArch=getArch) self.screen.popWindow() for package in tui.packageList: self.packageList.append("%s-%s-%s" % (sstr(package['name']), sstr(package['version']), sstr(package['release'])), item = sstr(package['name']), selected = 1) # BUTTON BAR self.bb = snack.ButtonBar(screen, [(sstr(NEXT), "next"), (sstr(BACK), "back"), (sstr(CANCEL), "cancel")]) toplevel.add(self.bb, 0, 4, padding = (0, 1, 0, 0), growx = 1) def setScale(self, amount, total): self.scale.set(int(((amount 1.0)/ total) * 100)) self.pwin.draw() self.screen.refresh() def saveResults(self): self.tui.includePackages = self.packagesButton.selected() selection = self.packageList.getSelection() for pkg in self.tui.packageList: if pkg['name'] in selection: self.tui.selectedPackages.append(pkg) def run(self): log.log_debug("Running %s" % self.name) result = self.g.runOnce() button = self.bb.buttonPressed(result) if result == "F12": return "next" return button class SendWindow: name = "SendWindow" def __init__(self, screen, tui): self.screen = screen self.tui = tui size = snack._snack.size() toplevel = snack.GridForm(screen, sstr(SEND_WINDOW), 1, 2) text = snack.TextboxReflowed(size[0]-15, sstr(SEND_WINDOW_DESC)) toplevel.add(text, 0, 0) # BUTTON BAR self.bb = snack.ButtonBar(screen, [(sstr(NEXT), "next"), (sstr(BACK), "back"), (sstr(CANCEL), "cancel")]) toplevel.add(self.bb, 0, 1, padding = (0, 1, 0, 0), growx = 1) self.g = toplevel def saveResults(self): pass def run(self): log.log_debug("Running %s" % self.name) result = self.g.runOnce() button = self.bb.buttonPressed(result) if result == "F12": return "next" return button class SendingWindow: name = "SendingWindow" def __init__(self, screen, tui): self.screen = screen self.tui = tui size = snack._snack.size() self.pwin = snack.GridForm(screen, sstr(SENDING_WINDOW), 1, 1) self.scale = snack.Scale(40, 100) self.pwin.add(self.scale, 0, 0) def run(self): log.log_debug("Running %s" % self.name) self.pwin.draw() self.screen.refresh() reg_info = None try: # reg_info dict contains: 'system_id', 'channels', # 'failed_channels', 'slots', 'failed_slots' log.log_debug('other is %s' % str(self.tui.other)) reg_info = rhnreg.registerSystem2(self.tui.userName, self.tui.password, self.tui.profileName, other = self.tui.other) reg_info = reg_info.rawDict systemId = sstr(reg_info['system_id']) except up2dateErrors.CommunicationError: e = sys.exc_info()[1] FatalErrorWindow(self.screen, _("Problem registering system:\n") + e.errmsg) except up2dateErrors.RhnUuidUniquenessError: e = sys.exc_info()[1] FatalErrorWindow(self.screen, _("Problem registering system:\n") + e.errmsg) except up2dateErrors.InsuffMgmntEntsError: e = sys.exc_info()[1] FatalErrorWindow(self.screen, _("Problem registering system:\n") + e.errmsg) except up2dateErrors.RegistrationDeniedError: e = sys.exc_info()[1] FatalErrorWindow(self.screen, _("Problem registering system:\n") + e.errmsg) except up2dateErrors.ActivationKeyUsageLimitError: FatalErrorWindow(self.screen, ACT_KEY_USAGE_LIMIT_ERROR) except: log.log_exception(sys.exc_info()) FatalErrorWindow(self.screen, _("Problem registering system.")) # write the system id out. if not rhnreg.writeSystemId(systemId): FatalErrorWindow(self.screen, _("Problem writing out system id to disk.")) self.setScale(1, 4) # include the info from the oeminfo file as well self.oemInfo = rhnreg.getOemInfo() self.setScale(2, 4) # maybe upload hardware profile if self.tui.includeHardware: try: rhnreg.sendHardware(systemId, self.tui.hardware) except up2dateErrors.CommunicationError: e = sys.exc_info()[1] FatalErrorWindow(self.screen, _("Problem sending hardware profile:\n") + e.errmsg) except: log.log_exception(sys.exc_info()) FatalErrorWindow(self.screen, _("Problem sending hardware profile.")) self.setScale(3, 4) # build up package list if necessary if self.tui.includePackages: try: rhnreg.sendPackages(systemId, self.tui.selectedPackages) except up2dateErrors.CommunicationError: e = sys.exc_info()[1] FatalErrorWindow(self.screen, _("Problem sending package list:\n") + e.errmsg) except: log.log_exception(sys.exc_info()) FatalErrorWindow(self.screen, _("Problem sending package list.")) li = None try: li = up2dateAuth.updateLoginInfo() except up2dateErrors.InsuffMgmntEntsError: FatalErrorWindow(self.screen, sys.exc_info()[1]) # Send virtualization information to the server. rhnreg.sendVirtInfo(systemId) # enable yum-rhn-plugin / dnf-plugin-spacewalk try: self.tui.pm_plugin_present, self.tui.pm_plugin_conf_changed = rhnreg.pluginEnable() except IOError: e = sys.exc_info()[1] WarningWindow(self.screen, _("Could not open %s\n%s is not enabled.\n") % (PM_PLUGIN_CONF, PM_PLUGIN_NAME) + e.errmsg) self.tui.pm_plugin_conf_error = 1 rhnreg.spawnRhnCheckForUI() self.setScale(4, 4) # Pop the pwin (Progress bar window) self.screen.popWindow() self.tui.reg_info = reg_info return "next" def saveResults(self): pass def setScale(self, amount, total): self.scale.set(int(((amount 1.0)/ total) * 100)) self.pwin.draw() self.screen.refresh() class FinishWindow: name = "FinishWindow" def __init__(self, screen, tui): self.screen = screen self.tui = tui size = snack._snack.size() toplevel = snack.GridForm(screen, sstr(FINISH_WINDOW), 1, 2) text = snack.TextboxReflowed(size[0]-11, sstr(FINISH_WINDOW_TEXT_TUI)) toplevel.add(text, 0, 0) # BUTTON BAR self.bb = snack.ButtonBar(screen, [(sstr(_("Finish")), "next")]) toplevel.add(self.bb, 0, 1, padding = (0, 1, 0, 0), growx = 1) self.g = toplevel def saveResults(self): pass def run(self): log.log_debug("Running %s" % self.name) result = self.g.runOnce() button = self.bb.buttonPressed(result) if result == "F12": return "next" return button class ReviewWindow: name = "ReviewWindow" def __init__(self, screen, tui): self.screen = screen self.tui = tui self.reg_info = tui.reg_info size = snack._snack.size() toplevel = snack.GridForm(screen, sstr(REVIEW_WINDOW), 1, 2) review_window_text = '' if not self.tui.pm_plugin_present: review_window_text += PM_PLUGIN_WARNING + "\n\n" if self.tui.pm_plugin_conf_error: review_window_text += PM_PLUGIN_CONF_ERROR + "\n\n" if self.tui.pm_plugin_conf_changed: review_window_text += PM_PLUGIN_CONF_CHANGED + "\n\n" # Build up the review_window_text based on the data in self.reg_info review_window_text += REVIEW_WINDOW_PROMPT + "\n\n" # Create and add the text for what channels the system was # subscribed to. if len(self.reg_info['channels']) > 0: channel_list = "" for channel in self.reg_info['channels']: channel_list += channel + "\n" channels = CHANNELS_TITLE + "\n" + \ OK_CHANNELS + "\n" + \ "%s\n" log.log_debug("server type is %s " % self.tui.serverType) channels += CHANNELS_SAT_WARNING review_window_text += channels % channel_list + "\n\n" if len(self.reg_info['system_slots']) > 0: slot_list = "" for slot in self.reg_info['system_slots']: if slot == 'enterprise_entitled': slot_list += MANAGEMENT + "\n" elif slot == 'virtualization_host': slot_list += VIRT + "\n" else: slot_list += slot + "\n" review_window_text += SLOTS % slot_list + "\n\n" if len(self.reg_info['universal_activation_key']) > 0: act_key_list = "" for act_key in self.reg_info['universal_activation_key']: act_key_list += act_key review_window_text += ACTIVATION_KEY % (act_key_list) self.review_window = snack.Textbox(size[0]-10, size[1]-14, sstr(review_window_text), 1, 1) toplevel.add(self.review_window, 0, 0, padding = (0, 1, 0, 0)) # BUTTON BAR self.bb = snack.ButtonBar(screen, [(sstr(OK), "next")]) toplevel.add(self.bb, 0, 1, padding = (0, 1, 0, 0), growx = 1) self.g = toplevel def saveResults(self): return 1 def run(self): log.log_debug("Running %s" % self.name) result = self.g.runOnce() button = self.bb.buttonPressed(result) if result == "F12": button = "next" if not self.tui.pm_plugin_present: button = "exit" if self.tui.pm_plugin_conf_error: button = "exit" return button class Tui: name = "RHN_REGISTER_TUI" def __init__(self, screen, test): self.screen = screen self.test = test self.size = snack._snack.size() self.drawFrame() self.alreadyRegistered = 0 try: self.serverType = rhnreg.getServerType() except up2dateErrors.InvalidProtocolError: FatalErrorWindow(screen, _("You specified an invalid protocol." + "Only https and http are allowed.")) self.windows = [ AlreadyRegisteredSubscriptionManagerWindow, AlreadyRegisteredWindow, StartWindow, SatelliteUrlWindow, ConnectWindow, InfoWindow, OSReleaseWindow, HardwareWindow, PackagesWindow, SendWindow, SendingWindow, ReviewWindow, FinishWindow ] self.serverURL = config.getServerURL()[0] if not cfg['sslCACert']: cfg.set('sslCACert', '/usr/share/rhn/CLN-ORG-TRUSTED-SSL-CERT') self.sslCACert = cfg['sslCACert'] def __del__(self): self.screen.finish() def drawFrame(self): self.welcomeText = COPYRIGHT_TEXT self.screen.drawRootText(0, 0, sstr(self.welcomeText)) self.screen.pushHelpLine(sstr(_(" <Tab>/<Alt-Tab> between elements \| <Space> selects \| <F12> next screen"))) def initResults(self): self.userName = "" self.password = "" self.oemInfo = {} self.productInfo = { "entitlement_num" : "", "registration_num" : "", "first_name" : "", "last_name" : "", "company" : "", "address" : "", "city" : "", "state" : "", "zip" : "", "country" : "", } self.other = {} self.other['registration_number'] = '' self.profileName = "" self.includeHardware = 1 self.limited_updates_button = 1 self.all_updates_button = 0 self.includePackages = 0 self.packageList = [] self.selectedPackages = [] self.pm_plugin_present = 1 self.pm_plugin_conf_error = 0 self.pm_plugin_conf_changed = 0 def run(self): log.log_debug("Running %s" % self.name) self.initResults() direction = "forward" try: index = 0 while index < len(self.windows): win = None try: win = self.windows[index](self.screen, self) except WindowSkipException: if direction == "forward": index = index + 1 else: index = index - 1 continue log.log_debug("index is %s" % index) result = win.run() log.log_debug("Result %s" % result) if result == "back": if index > 0: index = index - 1 # If we're on the info window, "back" means go back # to the satellite url window, not back to the # temporary connection test window. if (index > 0 and self.windows[index].name == ConnectWindow.name): index -= 1 direction = "backward" elif result == "exit": return elif result == "cancel": log.log_debug("Caught a cancel request") # Show the confirm quit window if ConfirmQuitWindow(self.screen) == 1: return elif result == "next": index = index + 1 win.saveResults() direction = "forward" finally: self.screen.finish() def main(): test = 0 signal.signal(signal.SIGINT, signal.SIG_IGN) if len(sys.argv) > 1: if sys.argv[1] == "-t" or sys.argv[1] == "--test": test = 1 screen = snack.SnackScreen() if geteuid() != 0 and not test: FatalErrorWindow(screen, _("You must run the RHN registration program as root.")) tui = Tui(screen, test) tui.run() if __name__ == "__main__": main()	43.7 KB	2025-05-29 13:37
📄 up2dateAuth.py # import os import pickle import sys import time import socket import subprocess from up2date_client import rpcServer try: # python2 from types import DictType except ImportError: # python3 DictType = dict from rhn import rpclib from up2date_client import clientCaps from up2date_client import config from up2date_client import rhnserver from up2date_client import up2dateErrors from up2date_client import up2dateLog from up2date_client import up2dateUtils loginInfo = None pcklAuthFileName = "/var/spool/up2date/loginAuth.pkl" def getSystemId(): cfg = config.initUp2dateConfig() path = cfg["systemIdPath"] if not os.access(path, os.R_OK): return None f = open(path, "r") ret = f.read() f.close() return ret def getPreferredInterface(): """ Extract the preferred_interface parameter from system_id XML Returns 'IPv4' or 'IPv6' if specified, otherwise 'IPv4' as default """ log = up2dateLog.initLog() systemId = getSystemId() preferred_interface = "IPv4" if systemId is None: return preferred_interface try: # Read the system ID XML and parse it # rpclib.xmlrpclib is the standard way to do this across the project params = rpclib.xmlrpclib.loads(systemId)[0][0] cfg_interface = params.get("preferred_interface", "IPv4") # Only 'IPv4' and 'IPv6' are valid options if preferred_interface in ("IPv4", "IPv6"): preferred_interface = cfg_interface else: log.log_me( "Invalid preferred_interface value '%s' in system_id, " "defaulting to '%s'" % (cfg_interface, preferred_interface) ) return preferred_interface except Exception: # If there's any error parsing the XML, default to IPv4 log.log_me("Failed to parse system_id XML, preferred_interface defaulting to 'IPv4'") log.log_exception(sys.exc_info()) return preferred_interface # if a user has upgraded to a newer release of Red Hat but still # has a systemid from their older release, they need to get an updated # systemid from the RHN servers. This takes care of that. def maybeUpdateVersion(): # if process is in the middle of system upgrade, do not switch channel automatically if os.environ.get('LEAPP_IPU_IN_PROGRESS'): return cfg = config.initUp2dateConfig() try: idVer = rpclib.xmlrpclib.loads(getSystemId())[0][0]['os_release'] except: # they may not even have a system id yet. return 0 systemVer = up2dateUtils.getVersion() if not cfg['channelOverride'] and idVer != systemVer: s = rhnserver.RegistrationRhnServer() newSystemId = s.registration.upgrade_version(getSystemId(), systemVer) path = cfg["systemIdPath"] dir = path[:path.rfind("/")] if not os.access(dir, os.W_OK): try: os.mkdir(dir) except: return 0 if not os.access(dir, os.W_OK): return 0 if os.access(path, os.F_OK): # already have systemid file there; let's back it up savePath = path + ".save" try: os.rename(path, savePath) except: return 0 f = open(path, "w") f.write(newSystemId) f.close() try: os.chmod(path, int('0600', 8)) except: pass def writeCachedLogin(): """ Pickle loginInfo to a file Returns: True -- wrote loginInfo to a pickle file False -- did _not_ write loginInfo to a pickle file """ log = up2dateLog.initLog() log.log_debug("writeCachedLogin() invoked") if not loginInfo: log.log_debug("writeCachedLogin() loginInfo is None, so bailing.") return False data = {'time': time.time(), 'loginInfo': loginInfo} pcklDir = os.path.dirname(pcklAuthFileName) if not os.access(pcklDir, os.W_OK): try: os.mkdir(pcklDir) os.chmod(pcklDir, int('0700', 8)) except: log.log_me("Unable to write pickled loginInfo to %s" % pcklDir) return False pcklAuth = open(pcklAuthFileName, 'wb') os.chmod(pcklAuthFileName, int('0600', 8)) pickle.dump(data, pcklAuth) pcklAuth.close() expireTime = data['time'] + float(loginInfo['X-RHN-Auth-Expire-Offset']) log.log_debug("Wrote pickled loginInfo at ", data['time'], " with expiration of ", expireTime, " seconds.") return True def readCachedLogin(): """ Read pickle info from a file Caches authorization info for connecting to the server. """ log = up2dateLog.initLog() log.log_debug("readCachedLogin invoked") if not os.access(pcklAuthFileName, os.R_OK): log.log_debug("Unable to read pickled loginInfo at: %s" % pcklAuthFileName) return False pcklAuth = open(pcklAuthFileName, 'rb') try: data = pickle.load(pcklAuth) except (EOFError, ValueError): log.log_debug("Unexpected EOF. Probably an empty file, \ regenerate auth file") pcklAuth.close() return False pcklAuth.close() # Check if system_id has changed try: idVer = rpclib.xmlrpclib.loads(getSystemId())[0][0]['system_id'] cidVer = "ID-%s" % data['loginInfo']['X-RHN-Server-Id'] if idVer != cidVer: log.log_debug("system id version changed: %s vs %s" % (idVer, cidVer)) return False except: pass createdTime = data['time'] li = data['loginInfo'] currentTime = time.time() expireTime = createdTime + float(li['X-RHN-Auth-Expire-Offset']) #Check if expired, offset is stored in "X-RHN-Auth-Expire-Offset" log.log_debug("Checking pickled loginInfo, currentTime=", currentTime, ", createTime=", createdTime, ", expire-offset=", float(li['X-RHN-Auth-Expire-Offset'])) if (currentTime > expireTime): log.log_debug("Pickled loginInfo has expired, created = %s, expire = %s." \ %(createdTime, expireTime)) return False _updateLoginInfo(li) log.log_debug("readCachedLogin(): using pickled loginInfo set to expire at ", expireTime) return True def _updateLoginInfo(li): """ Update the global var, "loginInfo" """ global loginInfo if type(li) == DictType: if type(loginInfo) == DictType: # must retain the reference. loginInfo.update(li) else: # this had better be the initial login or we lose the reference. loginInfo = li else: loginInfo = None # allow to pass in a system id for use in rhnreg # a bit of a kluge to make caps work correctly def login(systemId=None, forceUpdate=False, timeout=None): log = up2dateLog.initLog() log.log_debug("login(forceUpdate=%s) invoked" % (forceUpdate)) if not forceUpdate and not loginInfo: if readCachedLogin(): return loginInfo server = rhnserver.RhnServer(timeout=timeout) # send up the capabality info headerlist = clientCaps.caps.headerFormat() for (headerName, value) in headerlist: server.add_header(headerName, value) if systemId == None: systemId = getSystemId() if not systemId: return None maybeUpdateVersion() log.log_me("logging into up2date server") li = server.up2date.login(systemId, socket.getfqdn(), _get_panel_name(log)) # figure out if were missing any needed caps server.capabilities.validate() _updateLoginInfo(li) #update global var, loginInfo writeCachedLogin() #pickle global loginInfo if loginInfo: log.log_me("successfully retrieved authentication token " "from up2date server") log.log_debug("logininfo:", loginInfo) return loginInfo def updateLoginInfo(timeout=None): log = up2dateLog.initLog() log.log_me("updateLoginInfo() login info") # NOTE: login() updates the loginInfo object login(forceUpdate=True, timeout=timeout) if not loginInfo: raise up2dateErrors.AuthenticationError("Unable to authenticate") return loginInfo def getLoginInfo(timeout=None): global loginInfo try: loginInfo = loginInfo except NameError: loginInfo = None if loginInfo: return loginInfo # NOTE: login() updates the loginInfo object login(timeout=timeout) return loginInfo class _FailedToGetPanelName(Exception): pass def _get_panel_name(log): try: panel_name = _get_panel_name_via_cldetect() except _FailedToGetPanelName: # Do not spam logs when we're executed from under cldeploy script # and no cloudlinux packages are yet installed. if not _is_cldeploy_running(): log.log_exception(sys.exc_info()) log.log_me("Failed to get panel name via cldetect, using fallback mechanism") panel_name = _fallback_get_panel_name() return panel_name.lower() def _is_cldeploy_running(): # NOTE(vlebedev): This is more or less a direct port of shell code from cldeploy script. lock_file_path = '/var/lock/cldeploy.lck' cldeploy_running = False pid = None if os.path.exists(lock_file_path): with open(lock_file_path) as f: pid = f.read().strip() if pid: pid_cmdline_path = "/proc/%s/cmdline" % pid if os.path.exists(pid_cmdline_path): with open(pid_cmdline_path) as f: cldeploy_running = 'cldeploy' in f.read() return cldeploy_running def _get_panel_name_via_cldetect(): binary, _ = cmd = ["/usr/bin/cldetect", "--detect-cp-nameonly"] if not os.path.exists(binary): raise _FailedToGetPanelName( "Failed to obtain panel name because '%s' does not exist" % binary ) process = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, universal_newlines=True) stdout, stderr = process.communicate() if process.returncode != 0: raise _FailedToGetPanelName( "Failed to obtain panel name using '%s' command; stderr: %s" % (" ".join(cmd), stderr) ) return stdout.strip() def _fallback_get_panel_name(): # NOTE(vlebedev): This is more or less a direct port of shell code from cldeploy script. # TODO(vlebedev): Pass $PANEL from cldeploy instead of duplication? if os.path.isdir("/usr/local/psa/admin/"): return "plesk" if os.path.isdir("/usr/local/interworx/"): return "interworx" if os.path.isdir("/usr/local/cpanel/whostmgr/docroot/"): return "cpanel" if os.path.isdir("/usr/local/ispmgr/"): return "ispmgr" if os.path.isdir("/usr/local/directadmin/"): return "directadmin" if os.path.isfile("/usr/local/mgr5/sbin/mgrctl"): return "ispmgr5" return "unknown"	10.69 KB	2025-05-29 13:37
📄 up2dateErrors.py # # Client code for Update Agent # Copyright (c) 1999--2016 Red Hat, Inc. Distributed under GPLv2. # # Author: Preston Brown <pbrown@redhat.com> # Adrian Likins <alikins@redhat.com # Cristian Gafton <gafton@redhat.com> # # # In addition, as a special exception, the copyright holders give # permission to link the code of portions of this program with the # OpenSSL library under certain conditions as described in each # individual source file, and distribute linked combinations # including the two. # You must obey the GNU General Public License in all respects # for all of the code used other than OpenSSL. If you modify # file(s) with this exception, you may extend this exception to your # version of the file(s), but you are not obligated to do so. If you # do not wish to do so, delete this exception statement from your # version. If you delete this exception statement from all source # files in the program, then also delete it here. import gettext t = gettext.translation('rhn-client-tools', fallback=True) # Python 3 translations don't have a ugettext method if not hasattr(t, 'ugettext'): t.ugettext = t.gettext _ = t.ugettext import OpenSSL from rhn.i18n import ustr from up2date_client import config from up2date_client import up2dateLog from up2date_client.pkgplatform import getPlatform import sys sys.path = sys.path[1:] + sys.path[:1] try: from yum.Errors import YumBaseError as PmBaseError except ImportError: try: from dnf.exceptions import Error as PmBaseError except ImportError: class PmBaseError(Exception): def __init__(self, errmsg): self.value = errmsg def __getattr__(self, name): raise AttributeError(_("class %s has no attribute '%s'") % (self.__class__.__name__, name)) def __setattr__(self, name, value): if name in ['errmsg', 'value']: self.__dict__['value'] = value else: self.__dict__[name] = value class Error(PmBaseError): """base class for errors""" premsg = '' def __init__(self, errmsg): errmsg = ustr(errmsg) PmBaseError.__init__(self, errmsg) self.value = 'rhn-plugin: ' + self.premsg + errmsg self.log = up2dateLog.initLog() def __repr__(self): self.log.log_me(self.value) return self.value def __getattr__(self, name): """ Spacewalk backend still use errmsg, let have errmsg as alias to value """ if name == 'errmsg': return self.value else: if hasattr(PmBaseError, '__getattr__'): # rhel5 has no __getattribute__() return PmBaseError.__getattr__(self, name) else: if name in self.__dict__: return self.__dict__[name] else: raise AttributeError(_("class %s has no attribute '%s'") % (self.__class__.__name__, name)) def __setattr__(self, name, value): """ Spacewalk backend still use errmsg, let have errmsg as alias to value """ if name == 'errmsg': self.__dict__['value'] = value else: if hasattr(PmBaseError, '__setattr__'): # rhel5 has no __setattr__() PmBaseError.__setattr__(self, name, value) else: self.__dict__[name] = value class DebAndSuseRepoError(Error): pass try: from yum.Errors import RepoError except ImportError: try: from dnf.exceptions import RepoError except ImportError: RepoError = DebAndSuseRepoError class RpmError(Error): """rpm itself raised an error condition""" premsg = _("RPM error. The message was:\n") class RhnServerException(Error): pass class PasswordError(RhnServerException): """Raise when the server responds with that a password is incorrect""" premsg = _("Password error. The message was:\n") class DependencyError(Error): """Raise when a rpm transaction set has a dependency error""" premsg = _("RPM dependency error. The message was:\n") def __init__(self, msg, deps=None): Error.__init__(self, msg) # just tag on the whole deps tuple, so we have plenty of info # to play with self.deps = deps class CommunicationError(RhnServerException): """Indicates a problem doing xml-rpc http communication with the server""" premsg = _("Error communicating with server. "\ "The message was:\n") class FileNotFoundError(Error): """ Raise when a package or header that is requested returns a 404 error code""" premsg = _("File Not Found: \n") class DelayError(RhnServerException): """ Raise when the expected response from a xml-rpc call exceeds a timeout""" premsg = _("Delay error from server. The message was:\n") class RpmRemoveError(Error): """ Raise when we can't remove a package for some reason (failed deps, etc)""" def __init__(self, args): Error.__init__(self, "") self.args = args for key in self.args.keys(): self.args[key] = ustr(self.args[key]) self.value = self.value + "%s failed because of %s\n" % ( key, self.args[key]) self.data = self.args def __repr__(self): return self.value class NoLogError(Error): def __init__(self, msg): msg = ustr(msg) self.value = self.premsg + msg def __repr__(self): return self.value class AbuseError(Error): pass class AuthenticationTicketError(NoLogError, RhnServerException): pass class AuthenticationError(NoLogError): pass class ValidationError(NoLogError, RhnServerException): """indicates an error during server input validation""" premsg = _("Error validating data at server:\n") class InvalidRegistrationNumberError(ValidationError): pass class RegistrationDeniedError(RhnServerException): def __init__(self): RhnServerException.__init__(self, self.changeExplanation()) def __repr__(self): return self.value def changeExplanation(self): return _(""" CloudLinux Network Classic is not supported. To register with CloudLinux Subscription Management please run: subscription-manager register --auto-attach Get more information at www.cloudlinux.com """) class InvalidProductRegistrationError(NoLogError): """indicates an error during server input validation""" premsg = _("The installation number is invalid") class OemInfoFileError(NoLogError): premsg = _("Error parsing the oemInfo file at field:\n") class NoBaseChannelError(NoLogError, RhnServerException): """No valid base channel was found for this system""" pass class UnknownMethodException(NoLogError, RhnServerException): pass class RhnUuidUniquenessError(NoLogError, RhnServerException): pass class ServerCapabilityError(Error): def __init__(self, msg, errorlist=None): Error.__init__(self, msg) self.errorlist = [] if errorlist: self.errorlist=errorlist def __repr__(self): return self.value class NoChannelsError(NoLogError): pass class NetworkError(Error): """ some generic network error occurred, e.g. connection reset by peer """ premsg = _("Network error: ") class SSLCertificateVerifyFailedError(RepoError, Error): def __init__(self): # Need to override __init__ because the base class requires a message arg # and this exception shouldn't. up2dateConfig = config.initUp2dateConfig() certFile = up2dateConfig['sslCACert'] f = open(certFile, "r") buf = f.read() f.close() tempCert = OpenSSL.crypto.load_certificate(OpenSSL.crypto.FILETYPE_PEM, buf) if tempCert.has_expired(): RepoError.__init__(self ,"The certificate %s is expired. Please ensure you have the correct" " certificate and your system time is correct." % certFile) else: RepoError.__init__(self, "The SSL certificate failed verification.") __getattr__ = Error.__getattr__ class SSLCertificateFileNotFound(Error): pass class AuthenticationOrAccountCreationError(ValidationError): """Class that can represent different things depending on context: While logging in with an existing user it represents a username or password being incorrect. While creating a new account, it represents the username already being taken or the user not being allowed to create an account. Optimally these different things would be different exceptions, but there are single fault codes the server can return to the client that can mean more than one of them so we have no way of knowing which is actually intended. """ pass class NotEntitlingError(Error): pass class InvalidProtocolError(Error): pass class UnableToCreateUser(NoLogError): pass class ActivationKeyUsageLimitError(NoLogError): pass class LoginMinLengthError(NoLogError): pass class PasswordMinLengthError(NoLogError): pass class PasswordMaxLengthError(NoLogError): pass class InsuffMgmntEntsError(RhnServerException): def __init__(self, msg ): RhnServerException.__init__(self, self.changeExplanation(msg)) def __repr__(self): return self.value def changeExplanation(self, msg): newExpln = _(""" Your organization does not have enough Management entitlements to register this system to CloudLinux Network. Please notify your organization administrator of this error. You should be able to register this system after your organization frees existing or purchases additional entitlements. Additional entitlements may be purchased by your organization administrator by logging into CloudLinux Network and visiting A common cause of this error code is due to having mistakenly setup an Activation Key which is set as the universal default. If an activation key is set on the account as a universal default, you can disable this key and retry to avoid requiring a Management entitlement.""") term = "Explanation:" loc = msg.rindex(term) + len(term) return msg[:loc] + newExpln class NoSystemIdError(NoLogError): pass class InvalidRedirectionError(NoLogError): """ Raise when redirect requests could'nt return a package""" pass	10.26 KB	2025-05-29 13:37
📄 up2dateLog.py # import time import traceback from rhn.i18n import ustr, sstr from up2date_client import config class Log: """ attempt to log all interesting stuff, namely, anything that hits the network any error messages, package installs, etc """ # " emacs sucks def __init__(self): self.app = "up2date" self.cfg = config.initUp2dateConfig() self.log_info = '' def set_app_name(self, name): self.app = str(name) def log_debug(self, args): if self.cfg["debug"] > 1: self.log_me("D: ", args) def log_me(self, *args): """General logging function. Eg: log_me("I am a banana.") """ self.log_info = "[%s] %s" % (time.ctime(time.time()), self.app) s = u"" for i in args: # we really need unicode(str(i)) here, because i can be anything # from string or int to list, dict or even class i = ustr(str(i)) s += i if self.cfg["debug"] > 1: print(s) self.write_log(s) def trace_me(self): self.log_info = "[%s] %s" % (time.ctime(time.time()), self.app) x = traceback.extract_stack() msg = ''.join(traceback.format_list(x)) self.write_log(msg) def log_exception(self, logtype, value, tb): self.log_info = "[%s] %s" % (time.ctime(time.time()), self.app) output = ["\n"] # Accumulate the strings in a list output.append("Traceback (most recent call last):\n") output = output + traceback.format_list(traceback.extract_tb(tb)) output.append("%s: %s\n" % (logtype, value)) self.write_log("".join(output)) def write_log(self, s): log_name = self.cfg["logFile"] or "/var/log/up2date" log_file = open(log_name, 'a') msg = u"%s %s\n" % (ustr(self.log_info), ustr(s)) log_file.write(sstr(msg)) log_file.flush() log_file.close() def initLog(): global log try: log = log except NameError: log = None if log == None: log = Log() return log	2.06 KB	2025-05-29 13:37
📄 up2dateUtils.py # Client code for Update Agent # Copyright (c) 1999--2018 Red Hat, Inc. Distributed under GPLv2. # # Author: Preston Brown <pbrown@redhat.com> # Adrian Likins <alikins@redhat.com> # """utility functions for up2date""" import contextlib import os import sys import re import gettext from up2date_client import up2dateErrors from up2date_client import config from up2date_client.pkgplatform import getPlatform from rhn.i18n import sstr t = gettext.translation('rhn-client-tools', fallback=True) # Python 3 translations don't have a ugettext method if not hasattr(t, 'ugettext'): t.ugettext = t.gettext _ = t.ugettext if getPlatform() == 'deb': import lsb_release def _getOSVersionAndRelease(): dist_info = lsb_release.get_distro_information() os_name = dist_info['ID'] os_version = 'n/a' if 'CODENAME' in dist_info: os_version = dist_info['CODENAME'] os_release = dist_info['RELEASE'] return os_name, os_version, os_release else: from up2date_client import transaction def _getOSVersionAndRelease(): ts = transaction.initReadOnlyTransaction() for h in ts.dbMatch('Providename', "oraclelinux-release"): SYSRELVER = 'system-release(releasever)' version = sstr(h['version']) release = sstr(h['release']) if SYSRELVER in (sstr(provide) for provide in h['providename']): provides = dict((sstr(n), sstr(v)) for n,v in zip(h['providename'], h['provideversion'])) release = '%s-%s' % (version, release) version = provides[SYSRELVER] osVersionRelease = (sstr(h['name']), version, release) return osVersionRelease else: for h in ts.dbMatch('Providename', "redhat-release"): SYSRELVER = 'system-release(releasever)' version = sstr(h['version']) release = sstr(h['release']) if SYSRELVER in (sstr(provide) for provide in h['providename']): provides = dict((sstr(n), sstr(v)) for n,v in zip(h['providename'], h['provideversion'])) release = '%s-%s' % (version, release) version = provides[SYSRELVER] osVersionRelease = (sstr(h['name']), version, release) return osVersionRelease else: for h in ts.dbMatch('Providename', "distribution-release"): osVersionRelease = (sstr(h['name']), sstr(h['version']), sstr(h['release'])) # zypper requires a exclusive lock on the rpmdb. So we need # to close it here. ts.ts.closeDB() return osVersionRelease else: raise up2dateErrors.RpmError( "Could not determine what version of CloudLinux you "\ "are running.\nIf you get this error, try running \n\n"\ "\t\trpm --rebuilddb\n\n") def getVersion(): ''' Returns the version of redhat-release rpm ''' cfg = config.initUp2dateConfig() if cfg["versionOverride"]: return str(cfg["versionOverride"]) os_release, version, release = _getOSVersionAndRelease() return version def getOSRelease(): ''' Returns the name of the redhat-release rpm ''' os_release, version, release = _getOSVersionAndRelease() return os_release def getRelease(): ''' Returns the release of the redhat-release rpm ''' os_release, version, release = _getOSVersionAndRelease() return release def getArch(): if os.access("/etc/rpm/platform", os.R_OK): fd = open("/etc/rpm/platform", "r") platform = fd.read().strip() #bz 216225 #handle some replacements.. replace = {"ia32e-redhat-linux": "x86_64-redhat-linux"} if platform in replace: platform = replace[platform] return platform arch = os.uname()[4] if getPlatform() == 'deb': # On debian we only support i386 if arch in ['i486', 'i586', 'i686']: arch = 'i386' if arch == 'x86_64': arch = 'amd64' arch += '-debian-linux' return arch def version(): # substituted to the real version by the Makefile at installation time. return "2.12.5-1.el9.cloudlinux" @contextlib.contextmanager def suppress_errors(error_patterns): ''' Context manager to suppress errors matching the specified patterns ''' read_end, write_end = os.pipe() old_stdout = os.dup(1) old_stderr = os.dup(2) os.dup2(write_end, 1) os.dup2(write_end, 2) try: yield finally: # Restore stdout and stderr os.dup2(old_stdout, 1) os.dup2(old_stderr, 2) os.close(write_end) with os.fdopen(read_end) as f: combined_pattern = re.compile('\|'.join(error_patterns)) for line in f: if not combined_pattern.search(line): print(line, file=sys.stderr)	5.03 KB	2025-05-29 13:52

Edit File:

Terminal