编辑 config.py - 文件管理器

✏️ 编辑：config.py

路径：/usr/lib/python3.9/site-packages/up2date_client/config.py

大小：14.8 KB · 修改：2026-07-16 12:48:05 · 权限：0644 · 只读（保存会失败）

# 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"),
    "disallowConfChanges": (
        "Config options that can not be overwritten by a config update action",
        ["sslCACert", "serverURL", "disallowConfChanges"],
    ),
}

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

# CWE-93/CWE-113: refuse to serialize any key, comment, or value that
        # contains a CR or LF. This is a line-oriented config file, so an
        # embedded newline would inject an extra "key=value" directive into the
        # root-owned config (e.g. a forged serverURL/sslCACert). The reader
        # strips backslash line-continuations on load (see load()), so the
        # format never legitimately stores a literal newline -- fail closed
        # (reject before opening the file) rather than writing a corrupt or
        # attacker-controlled config.
        for _key in self.dict.keys():
            (_comment, _value) = self.dict[_key]
            _values = _value if type(_value) == type([]) else [_value]
            for _field in [_key, _comment] + list(_values):
                if "\n" in str(_field) or "\r" in str(_field):
                    raise ValueError(
                        "refusing to write config option %r to %s: a key, "
                        "comment, or value contains a CR/LF newline"
                        % (_key, self.fileName))

fd = os.open(
            self.fileName + ".new",
            os.O_WRONLY | os.O_CREAT | os.O_TRUNC | os.O_NOFOLLOW | os.O_CLOEXEC,
            0o600,
        )
        try:
            # O_CREAT only honours the mode argument on file creation. If
            # <fileName>.new already exists (e.g. left over from a previously
            # interrupted save), O_TRUNC truncates content but preserves the
            # existing inode's mode — fchmod is required to self-heal that.
            os.fchmod(fd, 0o600)
            f = os.fdopen(fd, "w")
        except Exception:
            os.close(fd)
            raise

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 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 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