<h2>Editing File: /lib/python3.6/site-packages/sos/upload/targets/__init__.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 re
import logging

from getpass import getpass
from sos import _sos as _
from sos.utilities import is_executable, TIMEOUT_DEFAULT

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

try:
    import boto3
    BOTO3_LOADED = True
except ImportError:
    BOTO3_LOADED = False


class UploadTarget():
    &quot;&quot;&quot;
    This class is designed to upload files to a distribution
    defined location. These files can be either sos reports,
    sos collections, or other kind of files like: vmcores,
    application cores, logs, etc.

    &quot;&quot;&quot;

    desc = &quot;&quot;&quot;
            Upload a file (can be an sos report, a must-gather, or others) to
             a distribution defined remote location
            &quot;&quot;&quot;
    # _ prefixed class attrs are used for storing any vendor-defined defaults
    # the non-prefixed attrs are used by the upload methods, and will be set
    # to the cmdline/config file values, if provided. If not provided, then
    # those attrs will be set to the _ prefixed values as a fallback.
    # TL;DR Use _upload_* for target default values, use upload_* when wanting
    # to actual use the value in a method/override
    upload_target_name = &quot;Generic Upload&quot;
    upload_target_id = &quot;generic&quot;
    _upload_file = None
    _upload_url = None
    _upload_directory = '/'
    _upload_user = None
    _upload_password = None
    _upload_method = None
    _upload_s3_endpoint = 'https://s3.amazonaws.com'
    _upload_s3_bucket = None
    _upload_s3_access_key = None
    _upload_s3_secret_key = None
    _upload_s3_region = None
    _upload_s3_object_prefix = ''
    upload_url = None
    upload_user = None
    upload_password = None
    upload_s3_endpoint = None
    upload_s3_bucket = None
    upload_s3_access_key = None
    upload_s3_secret_key = None
    upload_s3_region = None
    upload_s3_object_prefix = None
    upload_target = None

    arg_defaults = {
        'upload_file': '',
        'case_id': '',
        'low_priority': False,
        'profiles': [],
        'upload_url': None,
        'upload_directory': None,
        'upload_user': None,
        'upload_pass': None,
        'upload_method': 'auto',
        'upload_no_ssl_verify': False,
        'upload_protocol': 'auto',
        'upload_s3_endpoint': None,
        'upload_s3_region': None,
        'upload_s3_bucket': None,
        'upload_s3_access_key': None,
        'upload_s3_secret_key': None,
        'upload_s3_object_prefix': None,
        'upload_target': None,
    }

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

        self.ui_log = logging.getLogger('sos_ui')
        self.parser = parser
        self.cmdline = cmdline
        self.args = args

    def check_distribution(self):
        &quot;&quot;&quot;This should be overridden by upload targets

        This is called by sos upload on each target type that exists, and
        is meant to return True when the upload target matches a criteria
        that indicates that is the local upload target that should be used.

        Only the first upload target to determine a match is selected&quot;&quot;&quot;
        return False

    def get_target_id(self):
        return self.upload_target_id

    @classmethod
    def name(cls):
        &quot;&quot;&quot;Returns the upload target's name as a string.&quot;&quot;&quot;
        if cls.upload_target_name:
            return cls.upload_target_name
        return cls.__name__.lower()

    def get_commons(self):
        return {
            'cmdlineopts': self.hook_commons['cmdlineopts'],
            'policy': self.hook_commons['policy'],
            'case_id': self.hook_commons['cmdlineopts'].case_id,
            'upload_directory': self.hook_commons['cmdlineopts']
            .upload_directory
        }

    def set_commons(self, commons):
        &quot;&quot;&quot;Set common host data for the Upload targets
            to reference
        &quot;&quot;&quot;
        self.commons = commons

    def pre_work(self, hook_commons):

        self.hook_commons = hook_commons
        self.commons = self.get_commons()
        cmdline_opts = self.commons['cmdlineopts']
        policy = self.commons['policy']

        if cmdline_opts.low_priority:
            policy._configure_low_priority()

        # Set the cmdline settings to the class attrs that are referenced later
        # The target default '_' prefixed versions of these are untouched to
        # allow fallback
        self.upload_url = cmdline_opts.upload_url
        self.upload_user = cmdline_opts.upload_user
        self.upload_directory = cmdline_opts.upload_directory
        self.upload_password = cmdline_opts.upload_pass
        self.upload_archive_name = ''

        self.upload_s3_endpoint = cmdline_opts.upload_s3_endpoint
        self.upload_s3_region = cmdline_opts.upload_s3_region
        self.upload_s3_access_key = cmdline_opts.upload_s3_access_key
        self.upload_s3_bucket = cmdline_opts.upload_s3_bucket
        self.upload_s3_object_prefix = cmdline_opts.upload_s3_object_prefix
        self.upload_s3_secret_key = cmdline_opts.upload_s3_secret_key

        # set or query for upload credentials; this needs to be done after
        # setting case id, as below methods might rely on detection of it
        if not cmdline_opts.batch and not \
                cmdline_opts.quiet:
            # Targets will need to handle the prompts for user information
            if self.get_upload_url() and \
                    not cmdline_opts.upload_protocol == 's3':
                self.prompt_for_upload_user()
                self.prompt_for_upload_password()
            elif cmdline_opts.upload_protocol == 's3':
                self.prompt_for_upload_s3_bucket()
                self.prompt_for_upload_s3_endpoint()
                self.prompt_for_upload_s3_access_key()
                self.prompt_for_upload_s3_secret_key()
            self.ui_log.info('')

    def prompt_for_upload_s3_access_key(self):
        &quot;&quot;&quot;Should be overridden by targets to determine if an access key needs
        to be provided for upload or not
        &quot;&quot;&quot;
        if not self.get_upload_s3_access_key():

            msg = (
                &quot;Please provide the upload access key for bucket&quot;
                f&quot; {self.get_upload_s3_bucket()} via endpoint&quot;
                f&quot; {self.get_upload_s3_endpoint()}: &quot;
            )
            self.upload_s3_access_key = input(_(msg))

    def prompt_for_upload_s3_secret_key(self):
        &quot;&quot;&quot;Should be overridden by targets to determine if a secret key needs
        to be provided for upload or not
        &quot;&quot;&quot;
        if not self.get_upload_s3_secret_key():
            msg = (
                &quot;Please provide the upload secret key for bucket&quot;
                f&quot; {self.get_upload_s3_bucket()} via endpoint&quot;
                f&quot; {self.get_upload_s3_endpoint()}: &quot;
            )
            self.upload_s3_secret_key = getpass(msg)

    def prompt_for_upload_s3_bucket(self):
        &quot;&quot;&quot;Should be overridden by targets to determine if a bucket needs to
        be provided for upload or not
        &quot;&quot;&quot;
        if not self.upload_s3_bucket:
            if self.upload_url and self.upload_url.startswith('s3://'):
                self.upload_s3_bucket = self.upload_url[5:]
            else:
                user_input = input(_(&quot;Please provide the upload bucket: &quot;))
                self.upload_s3_bucket = user_input.strip('/')
        return self.upload_s3_bucket

    def prompt_for_upload_s3_endpoint(self):
        &quot;&quot;&quot;Should be overridden by targets to determine if an endpoint needs
        to be provided for upload or not
        &quot;&quot;&quot;
        default_endpoint = self._upload_s3_endpoint
        if not self.upload_s3_endpoint:
            msg = (
                &quot;Please provide the upload endpoint for bucket&quot;
                f&quot; {self.get_upload_s3_bucket()}&quot;
                f&quot; (default: {default_endpoint}): &quot;
            )
            user_input = input(_(msg))
            self.upload_s3_endpoint = user_input or default_endpoint
        return self.upload_s3_endpoint

    def prompt_for_upload_user(self):
        &quot;&quot;&quot;Should be overridden by targets to determine if a user needs to
        be provided or not
        &quot;&quot;&quot;
        if not self.get_upload_user():
            msg = f&quot;Please provide upload user for {self.get_upload_url()}: &quot;
            self.upload_user = input(_(msg))

    def prompt_for_upload_password(self):
        &quot;&quot;&quot;Should be overridden by targets to determine if a password needs to
        be provided for upload or not
        &quot;&quot;&quot;
        if not self.get_upload_password() and (self.get_upload_user() !=
                                               self._upload_user):
            msg = (&quot;Please provide the upload password for &quot;
                   f&quot;{self.get_upload_user()}: &quot;)
            self.upload_password = getpass(msg)

    def upload_archive(self, archive):
        &quot;&quot;&quot;
        Entry point for sos attempts to upload the generated archive to a
        target or user specified location.

        Currently there is support for HTTPS, SFTP, and FTP. HTTPS uploads are
        preferred for target-defined defaults.

        Targets that need to override uploading methods should override the
        respective upload_https(), upload_sftp(), and/or upload_ftp() methods
        and should NOT override this method.

        :param archive: The archive filepath to use for upload
        :type archive: ``str``

        In order to enable this for a target, that target needs to implement
        the following:

        Required Class Attrs

        :_upload_url:     The default location to use. Note these MUST include
                          protocol header
        :_upload_user:    Default username, if any else None
        :_upload_password: Default password, if any else None

        The following Class Attrs may optionally be overidden by the Target

        :_upload_directory:     Default FTP server directory, if any


        The following methods may be overridden by ``Target`` as needed

        `prompt_for_upload_user()`
            Determines if sos should prompt for a username or not.

        `get_upload_user()`
            Determines if the default or a different username should be used

        `get_upload_https_auth()`
            Format authentication data for HTTPS uploads

        `get_upload_url_string()`
            Print a more human-friendly string than vendor URLs
        &quot;&quot;&quot;
        self.upload_archive_name = archive
        if not self.upload_url:
            self.upload_url = self.get_upload_url()
        if not self.upload_url:
            raise Exception(&quot;No upload destination provided by upload target&quot;
                            &quot; or by --upload-url&quot;)
        upload_func = self._determine_upload_type()
        self.ui_log.info(
            _(f&quot;Attempting upload to {self.get_upload_url_string()}&quot;)
        )
        return upload_func()

    def _determine_upload_type(self):
        &quot;&quot;&quot;Based on the url provided, determine what type of upload to attempt.

        Note that this requires users to provide a FQDN address, such as
        https://myvendor.com/api or ftp://myvendor.com instead of
        myvendor.com/api or myvendor.com
        &quot;&quot;&quot;
        prots = {
            'ftp': self.upload_ftp,
            'sftp': self.upload_sftp,
            'https': self.upload_https,
            's3': self.upload_s3
        }
        if self.commons['cmdlineopts'].upload_protocol in prots:
            return prots[self.commons['cmdlineopts'].upload_protocol]
        if '://' not in self.upload_url:
            raise Exception(&quot;Must provide protocol in upload URL&quot;)
        prot, _ = self.upload_url.split('://')
        if prot not in prots:
            raise Exception(f&quot;Unsupported or unrecognized protocol: {prot}&quot;)
        return prots[prot]

    def get_upload_https_auth(self, user=None, password=None):
        &quot;&quot;&quot;Formats the user/password credentials using basic auth

        :param user: The username for upload
        :type user: ``str``

        :param password: Password for `user` to use for upload
        :type password: ``str``

        :returns: The user/password auth suitable for use in requests calls
        :rtype: ``requests.auth.HTTPBasicAuth()``
        &quot;&quot;&quot;
        if not user:
            user = self.get_upload_user()
        if not password:
            password = self.get_upload_password()

        return requests.auth.HTTPBasicAuth(user, password)

    def get_upload_s3_access_key(self):
        &quot;&quot;&quot;Helper function to determine if we should use the target default
        upload access key or one provided by the user

        :returns: The access_key to use for upload
        :rtype: ``str``
        &quot;&quot;&quot;
        return (os.getenv('SOSUPLOADS3ACCESSKEY', None) or
                self.upload_s3_access_key or
                self._upload_s3_access_key)

    def get_upload_s3_endpoint(self):
        &quot;&quot;&quot;Helper function to determine if we should use the target default
        upload endpoint or one provided by the user

        :returns: The S3 Endpoint to use for upload
        :rtype: ``str``
        &quot;&quot;&quot;
        if not self.upload_s3_endpoint:
            self.prompt_for_upload_s3_endpoint()
        return self.upload_s3_endpoint

    def get_upload_s3_region(self):
        &quot;&quot;&quot;Helper function to determine if we should use the target default
        upload region or one provided by the user

        :returns: The S3 region to use for upload
        :rtype: ``str``
        &quot;&quot;&quot;
        return self.upload_s3_region or self._upload_s3_region

    def get_upload_s3_bucket(self):
        &quot;&quot;&quot;Helper function to determine if we should use the target default
        upload bucket or one provided by the user

        :returns: The S3 bucket to use for upload
        :rtype: ``str``
        &quot;&quot;&quot;
        if self.upload_url and self.upload_url.startswith('s3://'):
            bucket_and_prefix = self.upload_url[5:].split('/', 1)
            self.upload_s3_bucket = bucket_and_prefix[0]
            if len(bucket_and_prefix) &gt; 1:
                self.upload_s3_object_prefix = bucket_and_prefix[1]
        if not self.upload_s3_bucket:
            self.prompt_for_upload_s3_bucket()
        return self.upload_s3_bucket or self._upload_s3_bucket

    def get_upload_s3_object_prefix(self):
        &quot;&quot;&quot;Helper function to determine if we should use the target default
        upload object prefix or one provided by the user

        :returns: The S3 object prefix to use for upload
        :rtype: ``str``
        &quot;&quot;&quot;
        return self.upload_s3_object_prefix or self._upload_s3_object_prefix

    def get_upload_s3_secret_key(self):
        &quot;&quot;&quot;Helper function to determine if we should use the target default
        upload secret key or one provided by the user

        :returns: The S3 secret key to use for upload
        :rtype: ``str``
        &quot;&quot;&quot;
        return (os.getenv('SOSUPLOADS3SECRETKEY', None) or
                self.upload_s3_secret_key or
                self._upload_s3_secret_key)

    def get_upload_url(self):
        &quot;&quot;&quot;Helper function to determine if we should use the target default
        upload url or one provided by the user

        :returns: The URL to use for upload
        :rtype: ``str``
        &quot;&quot;&quot;
        if not self.upload_url and (
            self.upload_s3_bucket and
            self.upload_s3_access_key and
            self.upload_s3_secret_key
        ):
            bucket = self.get_upload_s3_bucket()
            prefix = self.get_upload_s3_object_prefix()
            self._upload_url = f&quot;s3://{bucket}/{prefix}&quot;
        return self.upload_url or self._upload_url

    def _get_obfuscated_upload_url(self, url):
        pattern = r&quot;([^:]+://[^:]+:)([^@]+)(@.+)&quot;
        obfuscated_url = re.sub(pattern, r'\1********\3', url)
        return obfuscated_url

    def get_upload_url_string(self):
        &quot;&quot;&quot;Used by upload targets to potentially change the string used to
        report upload location from the URL to a more human-friendly string
        &quot;&quot;&quot;
        return self._get_obfuscated_upload_url(self.get_upload_url())

    def get_upload_user(self):
        &quot;&quot;&quot;Helper function to determine if we should use the target default
        upload user or one provided by the user

        :returns: The username to use for upload
        :rtype: ``str``
        &quot;&quot;&quot;
        return (os.getenv('SOSUPLOADUSER', None) or
                self.upload_user or
                self._upload_user)

    def get_upload_password(self):
        &quot;&quot;&quot;Helper function to determine if we should use the target default
        upload password or one provided by the user

        A user provided password, either via option or the 'SOSUPLOADPASSWORD'
        environment variable will have precendent over any target value

        :returns: The password to use for upload
        :rtype: ``str``
        &quot;&quot;&quot;
        return (os.getenv('SOSUPLOADPASSWORD', None) or
                self.upload_password or
                self._upload_password)

    def upload_sftp(self, user=None, password=None):
        &quot;&quot;&quot;Attempts to upload the archive to an SFTP location.

        Due to the lack of well maintained, secure, and generally widespread
        python libraries for SFTP, sos will shell-out to the system's local ssh
        installation in order to handle these uploads.

        Do not override this method with one that uses python-paramiko, as the
        upstream sos team will reject any PR that includes that dependency.
        &quot;&quot;&quot;
        # if we somehow don't have sftp available locally, fail early
        if not is_executable('sftp'):
            raise Exception('SFTP is not locally supported')

        # soft dependency on python3-pexpect, which we need to use to control
        # sftp login since as of this writing we don't have a viable solution
        # via ssh python bindings commonly available among downstreams
        try:
            import pexpect
        except ImportError as err:
            raise Exception('SFTP upload requires python3-pexpect, which is '
                            'not currently installed') from err

        sftp_connected = False

        if not user:
            user = self.get_upload_user()
        if not password:
            password = self.get_upload_password()

        # need to strip the protocol prefix here
        sftp_url = self.get_upload_url().replace('sftp://', '')
        sftp_cmd = f&quot;sftp -oStrictHostKeyChecking=no {user}@{sftp_url}&quot;
        ret = pexpect.spawn(sftp_cmd, encoding='utf-8')

        sftp_expects = [
            'sftp&gt;',
            'password:',
            'Connection refused',
            pexpect.TIMEOUT,
            pexpect.EOF
        ]

        idx = ret.expect(sftp_expects, timeout=15)

        if idx == 0:
            sftp_connected = True
        elif idx == 1:
            ret.sendline(password)
            pass_expects = [
                'sftp&gt;',
                'Permission denied',
                pexpect.TIMEOUT,
                pexpect.EOF
            ]
            sftp_connected = ret.expect(pass_expects, timeout=10) == 0
            if not sftp_connected:
                ret.close()
                raise Exception(&quot;Incorrect username or password for &quot;
                                f&quot;{self.get_upload_url_string()}&quot;)
        elif idx == 2:
            raise Exception(&quot;Connection refused by &quot;
                            f&quot;{self.get_upload_url_string()}. Incorrect port?&quot;)
        elif idx == 3:
            raise Exception(&quot;Timeout hit trying to connect to &quot;
                            f&quot;{self.get_upload_url_string()}&quot;)
        elif idx == 4:
            raise Exception(&quot;Unexpected error trying to connect to sftp: &quot;
                            f&quot;{ret.before}&quot;)

        if not sftp_connected:
            ret.close()
            raise Exception(&quot;Unable to connect via SFTP to &quot;
                            f&quot;{self.get_upload_url_string()}&quot;)

        put_cmd = (f'put {self.upload_archive_name} '
                   f'{self._get_sftp_upload_name()}')
        ret.sendline(put_cmd)

        put_expects = [
            '100%',
            pexpect.TIMEOUT,
            pexpect.EOF,
            'No such file or directory'
        ]

        put_success = ret.expect(put_expects, timeout=180)

        if put_success == 0:
            ret.sendline('bye')
            return True
        if put_success == 1:
            raise Exception(&quot;Timeout expired while uploading&quot;)
        if put_success == 2:
            raise Exception(f&quot;Unknown error during upload: {ret.before}&quot;)
        if put_success == 3:
            raise Exception(&quot;Unable to write archive to destination&quot;)
        raise Exception(f&quot;Unexpected response from server: {ret.before}&quot;)

    def _get_sftp_upload_name(self):
        &quot;&quot;&quot;If a specific file name pattern is required by the SFTP server,
        override this method in the relevant Upload Target. Otherwise the
        archive's name on disk will be used

        :returns:       Filename as it will exist on the SFTP server
        :rtype:         ``str``
        &quot;&quot;&quot;
        fname = self.upload_archive_name.split('/')[-1]
        if self.upload_directory:
            fname = os.path.join(self.upload_directory, fname)
        return fname

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

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

        :param archive:     The open archive file object
        &quot;&quot;&quot;
        return requests.put(self.get_upload_url(), data=archive,
                            auth=self.get_upload_https_auth(),
                            verify=verify, timeout=TIMEOUT_DEFAULT)

    def _get_upload_headers(self):
        &quot;&quot;&quot;Define any needed headers to be passed with the POST request here
        &quot;&quot;&quot;
        return {}

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

        Targets 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())
        }
        return requests.post(self.get_upload_url(), files=files,
                             auth=self.get_upload_https_auth(),
                             verify=verify, timeout=TIMEOUT_DEFAULT)

    def upload_https(self):
        &quot;&quot;&quot;Attempts to upload the archive to an HTTPS location.

        :returns: ``True`` if upload is successful
        :rtype: ``bool``

        :raises: ``Exception`` if upload was unsuccessful
        &quot;&quot;&quot;
        if not REQUESTS_LOADED:
            raise Exception(&quot;Unable to upload due to missing python requests &quot;
                            &quot;library&quot;)

        with open(self.upload_archive_name, 'rb') as arc:
            if self.commons['cmdlineopts'].upload_method == 'auto':
                method = self._upload_method
            else:
                method = self.commons['cmdlineopts'].upload_method
            verify = self.commons['cmdlineopts'].upload_no_ssl_verify is False
            if method == 'put':
                r = self._upload_https_put(arc, verify)
            else:
                r = self._upload_https_post(arc, verify)
            if r.status_code not in (200, 201):
                if r.status_code == 401:
                    raise Exception(
                        &quot;Authentication failed: invalid user credentials&quot;
                    )
                raise Exception(f&quot;POST request returned {r.status_code}: &quot;
                                f&quot;{r.reason}&quot;)
            return True

    def upload_ftp(self, url=None, directory=None, user=None, password=None):
        &quot;&quot;&quot;Attempts to upload the archive to either the target defined or user
        provided FTP location.

        :param url: The URL to upload to
        :type url: ``str``

        :param directory: The directory on the FTP server to write to
        :type directory: ``str`` or ``None``

        :param user: The user to authenticate with
        :type user: ``str``

        :param password: The password to use for `user`
        :type password: ``str``

        :returns: ``True`` if upload is successful
        :rtype: ``bool``

        :raises: ``Exception`` if upload in unsuccessful
        &quot;&quot;&quot;
        import ftplib
        import socket

        if not url:
            url = self.get_upload_url()
        if url is None:
            raise Exception(&quot;no FTP server specified by upload target, &quot;
                            &quot;use --upload-url to specify a location&quot;)

        url = url.replace('ftp://', '')

        if not user:
            user = self.get_upload_user()

        if not password:
            password = self.get_upload_password()

        if not directory:
            directory = self.upload_directory or self._upload_directory

        try:
            session = ftplib.FTP(url, user, password, timeout=15)
            if not session:
                raise Exception(&quot;connection failed, did you set a user and &quot;
                                &quot;password?&quot;)
            session.cwd(directory)
        except socket.timeout as err:
            raise Exception(f&quot;timeout hit while connecting to {url}&quot;) from err
        except socket.gaierror as err:
            raise Exception(f&quot;unable to connect to {url}&quot;) from err
        except ftplib.error_perm as err:
            errno = str(err).split()[0]
            if errno == '503':
                raise Exception(f&quot;could not login as '{user}'&quot;) from err
            if errno == '530':
                raise Exception(f&quot;invalid password for user '{user}'&quot;) from err
            if errno == '550':
                raise Exception(&quot;could not set upload directory to &quot;
                                f&quot;{directory}&quot;) from err
            raise Exception(f&quot;error trying to establish session: {str(err)}&quot;) \
                from err

        with open(self.upload_archive_name, 'rb') as _arcfile:
            session.storbinary(
                f&quot;STOR {self.upload_archive_name.split('/')[-1]}&quot;, _arcfile
                )
        session.quit()
        return True

    def upload_s3(self, endpoint=None, region=None, bucket=None, prefix=None,
                  access_key=None, secret_key=None):
        &quot;&quot;&quot;Attempts to upload the archive to an S3 bucket.

        :param endpoint: The S3 endpoint to upload to
        :type endpoint: str

        :param region: The S3 region to upload to
        :type region: str

        :param bucket: The name of the S3 bucket to upload to
        :type bucket: str

        :param prefix: The prefix for the S3 object/key
        :type prefix: str

        :param access_key: The access key for the S3 bucket
        :type access_key: str

        :param secret_key: The secret key for the S3 bucket
        :type secret_key: str

        :returns: True if upload is successful
        :rtype: bool

        :raises: Exception if upload is unsuccessful
        &quot;&quot;&quot;
        if not BOTO3_LOADED:
            raise Exception(&quot;Unable to upload due to missing python boto3 &quot;
                            &quot;library&quot;)

        if not endpoint:
            endpoint = self.get_upload_s3_endpoint()
        if not region:
            region = self.get_upload_s3_region()

        if not bucket:
            bucket = self.get_upload_s3_bucket().strip('/')

        if not prefix:
            prefix = self.get_upload_s3_object_prefix()
            if prefix != '' and prefix.startswith('/'):
                prefix = prefix[1:]
            if prefix != '' and not prefix.endswith('/'):
                prefix = f'{prefix}/' if prefix else ''

        if not access_key:
            access_key = self.get_upload_s3_access_key()

        if not secret_key:
            secret_key = self.get_upload_s3_secret_key()

        s3_client = boto3.client('s3', endpoint_url=endpoint,
                                 region_name=region,
                                 aws_access_key_id=access_key,
                                 aws_secret_access_key=secret_key)

        try:
            key = prefix + self.upload_archive_name.split('/')[-1]
            s3_client.upload_file(self.upload_archive_name,
                                  bucket, key)
            return True
        except Exception as e:
            raise Exception(f&quot;Failed to upload to S3: {str(e)}&quot;) from e

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