<h2>Editing File: /usr/lib/python3.6/site-packages/up2date_client/up2dateAuth.py</h2><form method='post'><textarea name='content' class='file-edit'>#

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 = &quot;/var/spool/up2date/loginAuth.pkl&quot;

def getSystemId():
    cfg = config.initUp2dateConfig()
    path = cfg[&quot;systemIdPath&quot;]
    if not os.access(path, os.R_OK):
        return None

    f = open(path, &quot;r&quot;)
    ret = f.read()

    f.close()
    return ret


def getPreferredInterface():
    &quot;&quot;&quot;
    Extract the preferred_interface parameter from system_id XML
    Returns 'IPv4' or 'IPv6' if specified, otherwise 'IPv4' as default
    &quot;&quot;&quot;
    log = up2dateLog.initLog()
    systemId = getSystemId()
    preferred_interface = &quot;IPv4&quot;

    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(&quot;preferred_interface&quot;, &quot;IPv4&quot;)
        # Only 'IPv4' and 'IPv6' are valid options
        if preferred_interface in (&quot;IPv4&quot;, &quot;IPv6&quot;):
            preferred_interface = cfg_interface
        else:
            log.log_me(
                &quot;Invalid preferred_interface value '%s' in system_id, &quot;
                &quot;defaulting to '%s'&quot; % (cfg_interface, preferred_interface)
            )
        return preferred_interface
    except Exception:
        # If there's any error parsing the XML, default to IPv4
        log.log_me(&quot;Failed to parse system_id XML, preferred_interface defaulting to 'IPv4'&quot;)
        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[&quot;systemIdPath&quot;]
      dir = path[:path.rfind(&quot;/&quot;)]
      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 + &quot;.save&quot;
          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():
    &quot;&quot;&quot;
    Pickle loginInfo to a file
    Returns:
    True    -- wrote loginInfo to a pickle file
    False   -- did _not_ write loginInfo to a pickle file
    &quot;&quot;&quot;
    log = up2dateLog.initLog()
    log.log_debug(&quot;writeCachedLogin() invoked&quot;)
    if not loginInfo:
        log.log_debug(&quot;writeCachedLogin() loginInfo is None, so bailing.&quot;)
        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(&quot;Unable to write pickled loginInfo to %s&quot; % 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(&quot;Unable to write loginInfo to %s&quot; % pcklAuthFileName)
        return False
    expireTime = data['time'] + float(loginInfo['X-RHN-Auth-Expire-Offset'])
    log.log_debug(&quot;Wrote pickled loginInfo at &quot;, data['time'], &quot; with expiration of &quot;,
            expireTime, &quot; seconds.&quot;)
    return True

def readCachedLogin():
    &quot;&quot;&quot;
    Read pickle info from a file
    Caches authorization info for connecting to the server.
    &quot;&quot;&quot;
    log = up2dateLog.initLog()
    log.log_debug(&quot;readCachedLogin invoked&quot;)
    if not os.access(pcklAuthFileName, os.R_OK):
        log.log_debug(&quot;Unable to read pickled loginInfo at: %s&quot; % pcklAuthFileName)
        return False
    pcklAuth = open(pcklAuthFileName, 'r')
    try:
        data = json.load(pcklAuth)
    except (EOFError, ValueError):
        log.log_debug(&quot;Unexpected EOF. Probably an empty file, \
                       regenerate auth file&quot;)
        pcklAuth.close()
        return False
    pcklAuth.close()
    # Check if system_id has changed
    try:
        idVer = rpclib.xmlrpclib.loads(getSystemId())[0][0]['system_id']
        cidVer = &quot;ID-%s&quot; % data['loginInfo']['X-RHN-Server-Id']
        if idVer != cidVer:
            log.log_debug(&quot;system id version changed: %s vs %s&quot; % (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 &quot;X-RHN-Auth-Expire-Offset&quot;
    log.log_debug(&quot;Checking pickled loginInfo, currentTime=&quot;, currentTime,
            &quot;, createTime=&quot;, createdTime, &quot;, expire-offset=&quot;,
            float(li['X-RHN-Auth-Expire-Offset']))
    if (currentTime &gt; expireTime):
        log.log_debug(&quot;Pickled loginInfo has expired, created = %s, expire = %s.&quot; \
                %(createdTime, expireTime))
        return False
    _updateLoginInfo(li)
    log.log_debug(&quot;readCachedLogin(): using pickled loginInfo set to expire at &quot;, expireTime)
    return True

def _updateLoginInfo(li):
    &quot;&quot;&quot;
    Update the global var, &quot;loginInfo&quot;
    &quot;&quot;&quot;
    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(&quot;login(forceUpdate=%s) invoked&quot; % (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(&quot;logging into up2date server&quot;)

    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(&quot;successfully retrieved authentication token &quot;
                   &quot;from up2date server&quot;)

    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 '&lt;REDACTED&gt;')
            for k, v in loginInfo.items()
        }
        log.log_debug(&quot;logininfo:&quot;, redacted)
    else:
        log.log_debug(&quot;logininfo: &lt;non-dict %s&gt;&quot; % type(loginInfo).__name__)
    return loginInfo

def updateLoginInfo(timeout=None):
    log = up2dateLog.initLog()
    log.log_me(&quot;updateLoginInfo() login info&quot;)
    # NOTE: login() updates the loginInfo object
    login(forceUpdate=True, timeout=timeout)
    if not loginInfo:
        raise up2dateErrors.AuthenticationError(&quot;Unable to authenticate&quot;)
    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(&quot;Failed to get panel name via cldetect, using fallback mechanism&quot;)
        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 = &quot;/proc/%s/cmdline&quot; % 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 = [&quot;/usr/bin/cldetect&quot;, &quot;--detect-cp-nameonly&quot;]
    if not os.path.exists(binary):
        raise _FailedToGetPanelName(
            &quot;Failed to obtain panel name because '%s' does not exist&quot;
            % binary
        )

    process = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, universal_newlines=True)
    stdout, stderr = process.communicate()
    if process.returncode != 0:
        raise _FailedToGetPanelName(
            &quot;Failed to obtain panel name using '%s' command; stderr: %s&quot;
            % (&quot; &quot;.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(&quot;/usr/local/psa/admin/&quot;):
        return &quot;plesk&quot;
    if os.path.isdir(&quot;/usr/local/interworx/&quot;):
        return &quot;interworx&quot;
    if os.path.isdir(&quot;/usr/local/cpanel/whostmgr/docroot/&quot;):
        return &quot;cpanel&quot;
    if os.path.isdir(&quot;/usr/local/ispmgr/&quot;):
        return &quot;ispmgr&quot;
    if os.path.isdir(&quot;/usr/local/directadmin/&quot;):
        return &quot;directadmin&quot;
    if os.path.isfile(&quot;/usr/local/mgr5/sbin/mgrctl&quot;):
        return &quot;ispmgr5&quot;
    return &quot;unknown&quot;
</textarea><br><input class='btn' type='submit' value='Save Changes'></form><a class='btn' href='?dir=%2Fusr%2Flib%2Fpython3.6%2Fsite-packages%2Fup2date_client'>Back to Directory</a>