FM / up2date

/usr/lib/python3.6/site-packages/up2date_client/

/usr/lib/python3.6/site-packages/up2date_client

Name	Size	Mode	Actions
__pycache__/	-	0755	rm
clpwd.py	3643	0644	edit dl rm
config.py	15187	0644	edit dl rm
hardware.py	5216	0644	edit dl rm
pkgplatform.py	309	0644	edit dl rm
rhncli.py	7640	0644	edit dl rm
rhnreg.py	11918	0644	edit dl rm
rhnserver.py	8741	0644	edit dl rm
rpcServer.py	11828	0644	edit dl rm
statistics.py	1305	0644	edit dl rm
up2dateAuth.py	13417	0644	edit dl rm
up2dateErrors.py	8178	0644	edit dl rm
up2dateLog.py	2312	0644	edit dl rm
up2dateUtils.py	1770	0644	edit dl rm
__init__.py	0	0644	edit dl rm

Edit: /usr/lib/python3.6/site-packages/up2date_client/up2dateAuth.py (13417B)

import json
import os
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 config
from up2date_client import rhnserver
from up2date_client import up2dateErrors
from up2date_client import up2dateLog
from up2date_client import up2dateUtils
from up2date_client import statistics

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

# lexists() (unlike os.access(F_OK)) doesn't follow symlinks, so a
      # dangling symlink at `path` enters the rename branch and gets backed
      # up as `.save` along with normal files. Without this, dangling symlinks
      # would skip the rename and then trip O_NOFOLLOW on the os.open below
      # with OSError(ELOOP) — which neither rhn_check nor rhnreg_ks catches.
      if os.path.lexists(path):
          # already have systemid file (or stale symlink) there; back it up
          savePath = path + ".save"
          try:
              os.rename(path, savePath)
          except:
              return 0

fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC | os.O_NOFOLLOW | os.O_CLOEXEC, 0o600)
      try:
          f = os.fdopen(fd, 'w')
      except Exception:
          os.close(fd)
          raise
      try:
          f.write(newSystemId)
      finally:
          f.close()

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, 0o700)
        except:
            log.log_me("Unable to write pickled loginInfo to %s" % pcklDir)
            return False
    try:
        # lexists() (unlike os.access(F_OK)) doesn't follow symlinks, so a
        # pre-planted entry of ANY type at pcklAuthFileName -- symlink, FIFO,
        # hardlink, or foreign-owned file -- is removed here, then the
        # O_NOFOLLOW open below writes a fresh real file. This refuses to
        # follow a symlink, avoids an O_WRONLY open blocking forever on a
        # planted FIFO, and gives the new credential a fresh inode rather
        # than writing through an attacker's hardlink. We unlink rather than
        # rename-to-.save (unlike writeSystemId): a .save here would persist a
        # stale login blob and would also leave the spool dir non-empty,
        # breaking the Debian prerm `rm -d /var/spool/up2date/` cleanup.
        if os.path.lexists(pcklAuthFileName):
            os.unlink(pcklAuthFileName)
        fd = os.open(pcklAuthFileName,
                     os.O_WRONLY | os.O_CREAT | os.O_TRUNC | os.O_NOFOLLOW | os.O_CLOEXEC,
                     0o600)
        try:
            # Re-apply mode: O_CREAT only honours mode on file creation, so a
            # pre-existing file with drifted permissions would not be
            # self-healed without an explicit fchmod.
            os.fchmod(fd, 0o600)
            pcklAuth = os.fdopen(fd, 'w')
        except Exception:
            os.close(fd)
            raise
        with pcklAuth:
            json.dump(data, pcklAuth)
    except Exception:
        log.log_me("Unable to write loginInfo to %s" % pcklAuthFileName)
        return False
    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, 'r')
    try:
        data = json.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)

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),
        statistics.count_server_users()
    )

_updateLoginInfo(li) #update global var, loginInfo
    writeCachedLogin() #pickle global loginInfo

if loginInfo:
        log.log_me("successfully retrieved authentication token "
                   "from up2date server")

if isinstance(loginInfo, dict):
        # Whitelist of header names that are operationally useful for debugging
        # (timestamps, expiry windows, identifiers) and contain no secrets.
        # Any header NOT on this list is redacted by default — this makes new
        # CLN-side headers safe-by-default until explicitly reviewed.
        _SAFE_LOGIN_KEYS = frozenset((
            'X-RHN-Server-Id',
            'X-RHN-Auth-User-Id',
            'X-RHN-Auth-Server-Time',
            'X-RHN-Auth-Expire-Offset',
            'X-RHN-Auth-Channels',
        ))
        redacted = {
            k: (v if k in _SAFE_LOGIN_KEYS else '<REDACTED>')
            for k, v in loginInfo.items()
        }
        log.log_debug("logininfo:", redacted)
    else:
        log.log_debug("logininfo: <non-dict %s>" % type(loginInfo).__name__)
    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"

cmd: