<h2>Editing File: /lib/python3.6/site-packages/sos/upload/targets/redhat.py</h2><form method='post'><textarea name='content' class='file-edit'># Copyright 2024 Red Hat, Inc. Jose Castillo &lt;jcastillo@redhat.com&gt;
# This file is part of the sos project: https://github.com/sosreport/sos
#
# This copyrighted material is made available to anyone wishing to use,
# modify, copy, or redistribute it subject to the terms and conditions of
# version 2 of the GNU General Public License.
#
# See the LICENSE file in the source distribution for further information.
import os
import json
from sos.upload.targets import UploadTarget
from sos.utilities import convert_bytes, TIMEOUT_DEFAULT
from sos.policies.auth import DeviceAuthorizationClass
from sos.policies.distros.redhat import RHELPolicy
from sos import _sos as _

try:
    import requests
    REQUESTS_LOADED = True
except ImportError:
    REQUESTS_LOADED = False


class RHELUploadTarget(UploadTarget):

    client_identifier_url = &quot;https://sso.redhat.com/auth/&quot;\
        &quot;realms/redhat-external/protocol/openid-connect/auth/device&quot;
    token_endpoint = &quot;https://sso.redhat.com/auth/realms/&quot;\
        &quot;redhat-external/protocol/openid-connect/token&quot;
    upload_target_name = 'Red Hat Upload Target'
    upload_target_id = &quot;redhat&quot;

    def __init__(self, parser=None, args=None, cmdline=None):

        super().__init__(parser=parser, args=args, cmdline=cmdline)

    RH_API_HOST = &quot;https://api.access.redhat.com&quot;
    RH_SFTP_HOST = &quot;sftp://sftp.access.redhat.com&quot;
    _upload_url = RH_SFTP_HOST
    _upload_method = 'post'
    _device_token = None
    # Max size for an http single request is 1Gb
    _max_size_request = 1073741824

    def check_distribution(self):
        &quot;&quot;&quot;Return true if we are running in a RHEL system&quot;&quot;&quot;
        return isinstance(self.commons['policy'], RHELPolicy)

    def pre_work(self, hook_commons):

        super().pre_work(hook_commons)

        self.upload_directory = self.commons['cmdlineopts'].upload_directory

    def prompt_for_upload_user(self):
        if self.commons['cmdlineopts'].upload_user:
            self.ui_log.info(
                _(&quot;The option --upload-user has been deprecated in favour&quot;
                  &quot; of device authorization in RHEL&quot;)
            )
        if not self.commons['cmdlineopts'].case_id:
            # no case id provided =&gt; failover to SFTP
            self.upload_url = self.RH_SFTP_HOST
            self.ui_log.info(&quot;No case id provided, uploading to SFTP&quot;)

    def prompt_for_upload_password(self):
        # With OIDC we don't ask for user/pass anymore
        if self.commons['cmdlineopts'].upload_pass:
            self.ui_log.info(
                _(&quot;The option --upload-pass has been deprecated in favour&quot;
                  &quot; of device authorization in RHEL&quot;)
            )

    def get_upload_url(self):
        rh_case_api = &quot;/support/v1/cases/&quot;\
                     f&quot;{self.commons['cmdlineopts'].case_id}/attachments&quot;
        try:
            if self.upload_url:
                return self.upload_url
            if self.commons['cmdlineopts'].upload_url:
                return self.commons['cmdlineopts'].upload_url
            if self.commons['cmdlineopts'].upload_protocol == 'sftp':
                return self.RH_SFTP_HOST
            if not self.commons['cmdlineopts'].case_id:
                return self.RH_SFTP_HOST

        except Exception as e:
            self.ui_log.info(
                &quot;There was a problem while setting the &quot;
                f&quot;remote upload target:  {e}&quot;
            )
        return f&quot;{self.RH_API_HOST}{rh_case_api}&quot;

    def _get_upload_https_auth(self):
        str_auth = f&quot;Bearer {self._device_token}&quot;
        return {'Authorization': str_auth}

    def _upload_https_post(self, archive, verify=True):
        &quot;&quot;&quot;If upload_https() needs to use requests.post(), use this method.

        Policies should override this method instead of the base upload_https()

        :param archive:     The open archive file object
        &quot;&quot;&quot;
        files = {
            'file': (archive.name.split('/')[-1], archive,
                     self._get_upload_headers())
        }
        # Get the access token at this point. With this,
        # we cover the cases where report generation takes
        # longer than the token timeout
        RHELAuth = DeviceAuthorizationClass(
                self.client_identifier_url,
                self.token_endpoint
            )
        self._device_token = RHELAuth.get_access_token()
        self.ui_log.info(&quot;Device authorized correctly. Uploading file to &quot;
                         f&quot;{self.get_upload_url_string()}&quot;)
        return requests.post(self.get_upload_url(), files=files,
                             headers=self._get_upload_https_auth(),
                             verify=verify, timeout=TIMEOUT_DEFAULT)

    def _get_upload_headers(self):
        if self.get_upload_url().startswith(self.RH_API_HOST):
            return {'isPrivate': 'false', 'cache-control': 'no-cache'}
        return {}

    def get_upload_url_string(self):
        if self.get_upload_url().startswith(self.RH_API_HOST):
            return &quot;Red Hat Customer Portal&quot;
        if self.get_upload_url().startswith(self.RH_SFTP_HOST):
            return &quot;Red Hat Secure FTP&quot;
        return self._get_obfuscated_upload_url(self.upload_url)

    def _get_sftp_upload_name(self):
        &quot;&quot;&quot;The RH SFTP server will only automatically connect file uploads to
        cases if the filename _starts_ with the case number
        &quot;&quot;&quot;
        fname = self.upload_archive_name.split('/')[-1]

        if self.commons['cmdlineopts'].case_id:
            fname = f&quot;{self.commons['cmdlineopts'].case_id}_{fname}&quot;
        if self.upload_directory:
            fname = os.path.join(self.upload_directory, fname)
        return fname

    # pylint: disable=too-many-branches
    def upload_sftp(self, user=None, password=None):    
        &quot;&quot;&quot;Override the base upload_sftp to allow for setting an on-demand
        generated anonymous login for the RH SFTP server if a username and
        password are not given
        &quot;&quot;&quot;
        if self.RH_SFTP_HOST.split('//')[1] not in self.get_upload_url():
            return super().upload_sftp()

        if not REQUESTS_LOADED:
            raise Exception(&quot;python3-requests is not installed and is required&quot;
                            &quot; for obtaining SFTP auth token.&quot;)
        _token = None
        _user = None

        # We may have a device token already if we attempted
        # to upload via http but the upload failed. So
        # lets check first if there isn't one.
        if not self._device_token:
            try:
                RHELAuth = DeviceAuthorizationClass(
                    self.client_identifier_url,
                    self.token_endpoint
                )
            except Exception as e:
                # We end up here if the user cancels the device
                # authentication in the web interface
                if &quot;end user denied&quot; in str(e):
                    self.ui_log.info(
                        &quot;Device token authorization &quot;
                        &quot;has been cancelled by the user.&quot;
                    )
            else:
                self._device_token = RHELAuth.get_access_token()
        if self._device_token:
            self.ui_log.info(&quot;Device authorized correctly. Uploading file to&quot;
                             f&quot; {self.get_upload_url_string()}&quot;)

        url = self.RH_API_HOST + '/support/v2/sftp/token'
        ret = None
        if self._device_token:
            headers = self._get_upload_https_auth()
            ret = requests.post(url, headers=headers, timeout=10)
            if ret.status_code == 200:
                # credentials are valid
                _user = json.loads(ret.text)['username']
                _token = json.loads(ret.text)['token']
            else:
                self.ui_log.debug(
                    f&quot;DEBUG: auth attempt failed (status: {ret.status_code}): &quot;
                    f&quot;{ret.json()}&quot;
                )
                self.ui_log.error(
                    &quot;Unable to retrieve Red Hat auth token using provided &quot;
                    &quot;credentials. Will try anonymous.&quot;
                )
        else:
            adata = {&quot;isAnonymous&quot;: True}
            anon = requests.post(url, data=json.dumps(adata), timeout=10)
            if anon.status_code == 200:
                resp = json.loads(anon.text)
                _user = resp['username']
                _token = resp['token']
                self.ui_log.info(
                    _(f&quot;User {_user} used for anonymous upload. Please inform &quot;
                      f&quot;your support engineer so they may retrieve the data.&quot;)
                )
            else:
                self.ui_log.debug(
                    f&quot;DEBUG: anonymous request failed (status: &quot;
                    f&quot;{anon.status_code}): {anon.json()}&quot;
                )
        if _user and _token:
            return super().upload_sftp(user=_user, password=_token)
        raise Exception(&quot;Could not retrieve valid or anonymous credentials&quot;)

    def check_file_too_big(self, archive):
        size = os.path.getsize(archive)
        # Lets check if the size is bigger than the limit.
        # There's really no need to transform the size to Gb,
        # so we don't need to call any size converter implemented
        # in tools.py
        if size &gt;= self._max_size_request:
            self.ui_log.warning(
                _(&quot;Size of archive is bigger than Red Hat Customer Portal &quot;
                  &quot;limit for uploads of &quot;
                  f&quot;{convert_bytes(self._max_size_request)} &quot;
                  &quot; via sos http upload. \n&quot;)
                  )
            self.upload_url = self.RH_SFTP_HOST

    def upload_archive(self, archive):
        &quot;&quot;&quot;Override the base upload_archive to provide for automatic failover
        from RHCP failures to the public RH dropbox
        &quot;&quot;&quot;
        try:
            if self.get_upload_url().startswith(self.RH_API_HOST):
                self.check_file_too_big(archive)
            uploaded = super().upload_archive(archive)
        except Exception as e:
            uploaded = False
            if not self.upload_url.startswith(self.RH_API_HOST):
                raise
            self.ui_log.error(
                _(f&quot;Upload to Red Hat Customer Portal failed due to &quot;
                  f&quot;{e}. Trying {self.RH_SFTP_HOST}&quot;)
                )
            self.upload_url = self.RH_SFTP_HOST
            uploaded = super().upload_archive(archive)
        return uploaded

# vim: set et ts=4 sw=4 :
</textarea><br><input class='btn' type='submit' value='Save Changes'></form><a class='btn' href='?dir=%2Flib%2Fpython3.6%2Fsite-packages%2Fsos%2Fupload%2Ftargets'>Back to Directory</a>