st2actions.runners.ssh.ParamikoSSHClient.__init__() - Code Metrics - Inspection of "Merge pull request #2596 from StackStorm/paramiko-..." - StackStorm/st2 - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 5db0c6...7afbb0 )

by Edward

created 2016-04-08 12:28 UTC

init() F

↳ Parent: st2actions.runners.ssh.ParamikoSSHClient

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Metric	Value
dl	0
loc	37
rs	3
cc	9

# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License.  You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import os
import posixpath
from StringIO import StringIO
import time

import eventlet
from oslo_config import cfg

import paramiko

# Depending on your version of Paramiko, it may cause a deprecation
# warning on Python 2.6.
# Ref: https://bugs.launchpad.net/paramiko/+bug/392973

from st2common.log import logging
from st2common.util.misc import strip_shell_chars
from st2common.util.shell import quote_unix

__all__ = [
    'ParamikoSSHClient',

    'SSHCommandTimeoutError'
]

PRIVATE_KEY_HEADER = 'PRIVATE KEY-----'.lower()


class SSHCommandTimeoutError(Exception):
    """
    Exception which is raised when an SSH command times out.
    """

    def __init__(self, cmd, timeout, stdout=None, stderr=None):
        """
        :param stdout: Stdout which was consumed until the timeout occured.
        :type stdout: ``str``

        :param stdout: Stderr which was consumed until the timeout occured.
        :type stderr: ``str``
        """
        self.cmd = cmd
        self.timeout = timeout
        self.stdout = stdout
        self.stderr = stderr
        message = 'Command didn\'t finish in %s seconds' % (timeout)
        super(SSHCommandTimeoutError, self).__init__(message)

    def __repr__(self):
        return ('<SSHCommandTimeoutError: cmd="%s",timeout=%s)>' %
                (self.cmd, self.timeout))

    def __str__(self):
        return self.message


class ParamikoSSHClient(object):
    """
    A SSH Client powered by Paramiko.
    """

    # Maximum number of bytes to read at once from a socket
    CHUNK_SIZE = 1024
    # How long to sleep while waiting for command to finish
    SLEEP_DELAY = 1.5
    # Connect socket timeout
    CONNECT_TIMEOUT = 60

    def __init__(self, hostname, port=22, username=None, password=None, bastion_host=None,
                 key=None, key_files=None, key_material=None, timeout=None, passphrase=None):
        """
        Authentication is always attempted in the following order:

        - The key passed in (if key is provided)
        - Any key we can find through an SSH agent (only if no password and
          key is provided)
        - Any "id_rsa" or "id_dsa" key discoverable in ~/.ssh/ (only if no
          password and key is provided)
        - Plain username/password auth, if a password was given (if password is
          provided)
        """
        if key_files and key_material:
            raise ValueError(('key_files and key_material arguments are '
                              'mutually exclusive'))

        if passphrase and not key_material:
            raise ValueError('passphrase should accompany private key material')

        self.hostname = hostname
        self.port = port
        self.username = username if username else cfg.CONF.system_user
        self.password = password
        self.key = key if key else cfg.CONF.system_user.ssh_key_file
        self.key_files = key_files
        if not self.key_files and self.key:
            self.key_files = key  # `key` arg is deprecated.
        self.timeout = timeout or ParamikoSSHClient.CONNECT_TIMEOUT
        self.key_material = key_material
        self.client = None
        self.logger = logging.getLogger(__name__)
        self.sftp = None
        self.bastion_host = bastion_host
        self.bastion_client = None
        self.bastion_socket = None
        self.passphrase = passphrase

    def connect(self):
        """
        Connect to the remote node over SSH.

        :return: True if the connection has been successfully established,
                 False otherwise.
        :rtype: ``bool``
        """
        if self.bastion_host:
            self.logger.debug('Bastion host specified, connecting')
            self.bastion_client = self._connect(host=self.bastion_host)
            transport = self.bastion_client.get_transport()
            real_addr = (self.hostname, self.port)
            # fabric uses ('', 0) for direct-tcpip, this duplicates that behaviour
            # see https://github.com/fabric/fabric/commit/c2a9bbfd50f560df6c6f9675603fb405c4071cad
            local_addr = ('', 0)
            self.bastion_socket = transport.open_channel('direct-tcpip', real_addr, local_addr)

        self.client = self._connect(host=self.hostname, socket=self.bastion_socket)
        self.sftp = self.client.open_sftp()
        return True

    def put(self, local_path, remote_path, mode=None, mirror_local_mode=False):
        """
        Upload a file to the remote node.

        :type local_path: ``st``
        :param local_path: File path on the local node.

        :type remote_path: ``str``
        :param remote_path: File path on the remote node.

        :type mode: ``int``
        :param mode: Permissions mode for the file. E.g. 0744.

        :type mirror_local_mode: ``int``
        :param mirror_local_mode: Should remote file mirror local mode.

        :return: Attributes of the remote file.
        :rtype: :class:`posix.stat_result` or ``None``
        """

        if not local_path or not remote_path:
            raise Exception('Need both local_path and remote_path. local: %s, remote: %s' %
                            local_path, remote_path)
        local_path = quote_unix(local_path)
        remote_path = quote_unix(remote_path)

        extra = {'_local_path': local_path, '_remote_path': remote_path, '_mode': mode,
                 '_mirror_local_mode': mirror_local_mode}
        self.logger.debug('Uploading file', extra=extra)

        if not os.path.exists(local_path):
            raise Exception('Path %s does not exist locally.' % local_path)

        rattrs = self.sftp.put(local_path, remote_path)

        if mode or mirror_local_mode:
            local_mode = mode
            if not mode or mirror_local_mode:
                local_mode = os.stat(local_path).st_mode

            # Cast to octal integer in case of string
            if isinstance(local_mode, basestring):
                local_mode = int(local_mode, 8)
            local_mode = local_mode & 07777
            remote_mode = rattrs.st_mode
            # Only bitshift if we actually got an remote_mode
            if remote_mode is not None:
                remote_mode = (remote_mode & 07777)
            if local_mode != remote_mode:
                self.sftp.chmod(remote_path, local_mode)

        return rattrs

    def put_dir(self, local_path, remote_path, mode=None, mirror_local_mode=False):
        """
        Upload a dir to the remote node.

        :type local_path: ``str``
        :param local_path: Dir path on the local node.

        :type remote_path: ``str``
        :param remote_path: Base dir path on the remote node.

        :type mode: ``int``
        :param mode: Permissions mode for the file. E.g. 0744.

        :type mirror_local_mode: ``int``
        :param mirror_local_mode: Should remote file mirror local mode.

        :return: List of files created on remote node.
        :rtype: ``list`` of ``str``
        """

        extra = {'_local_path': local_path, '_remote_path': remote_path, '_mode': mode,
                 '_mirror_local_mode': mirror_local_mode}
        self.logger.debug('Uploading dir', extra=extra)

        if os.path.basename(local_path):
            strip = os.path.dirname(local_path)
        else:
            strip = os.path.dirname(os.path.dirname(local_path))

        remote_paths = []

        for context, dirs, files in os.walk(local_path):
            rcontext = context.replace(strip, '', 1)
            # normalize pathname separators with POSIX separator
            rcontext = rcontext.replace(os.sep, '/')
            rcontext = rcontext.lstrip('/')
            rcontext = posixpath.join(remote_path, rcontext)

            if not self.exists(rcontext):
                self.sftp.mkdir(rcontext)

            for d in dirs:
                n = posixpath.join(rcontext, d)
                if not self.exists(n):
                    self.sftp.mkdir(n)

            for f in files:
                local_path = os.path.join(context, f)
                n = posixpath.join(rcontext, f)
                # Note that quote_unix is done by put anyways.
                p = self.put(local_path=local_path, remote_path=n,
                             mirror_local_mode=mirror_local_mode, mode=mode)
                remote_paths.append(p)

        return remote_paths

    def exists(self, remote_path):
        """
        Validate whether a remote file or directory exists.

        :param remote_path: Path to remote file.
        :type remote_path: ``str``

        :rtype: ``bool``
        """
        try:
            self.sftp.lstat(remote_path).st_mode
        except IOError:
            return False

        return True

    def mkdir(self, dir_path):
        """
        Create a directory on remote box.

        :param dir_path: Path to remote directory to be created.
        :type dir_path: ``str``

        :return: Returns nothing if successful else raises IOError exception.

        :rtype: ``None``
        """

        dir_path = quote_unix(dir_path)
        extra = {'_dir_path': dir_path}
        self.logger.debug('mkdir', extra=extra)
        return self.sftp.mkdir(dir_path)

    def delete_file(self, path):
        """
        Delete a file on remote box.

        :param path: Path to remote file to be deleted.
        :type path: ``str``

        :return: True if the file has been successfully deleted, False
                 otherwise.
        :rtype: ``bool``
        """

        path = quote_unix(path)
        extra = {'_path': path}
        self.logger.debug('Deleting file', extra=extra)
        self.sftp.unlink(path)
        return True

    def delete_dir(self, path, force=False, timeout=None):
        """
        Delete a dir on remote box.

        :param path: Path to remote dir to be deleted.
        :type path: ``str``

        :param force: Optional Forcefully remove dir.
        :type force: ``bool``

        :param timeout: Optional Time to wait for dir to be deleted. Only relevant for force.
        :type timeout: ``int``

        :return: True if the file has been successfully deleted, False
                 otherwise.
        :rtype: ``bool``
        """

        path = quote_unix(path)
        extra = {'_path': path}
        if force:
            command = 'rm -rf %s' % path
            extra['_command'] = command
            extra['_force'] = force
            self.logger.debug('Deleting dir', extra=extra)
            return self.run(command, timeout=timeout)

        self.logger.debug('Deleting dir', extra=extra)
        return self.sftp.rmdir(path)

    def run(self, cmd, timeout=None, quote=False):
        """
        Note: This function is based on paramiko's exec_command()
        method.

        :param timeout: How long to wait (in seconds) for the command to
                        finish (optional).
        :type timeout: ``float``
        """

        if quote:
            cmd = quote_unix(cmd)

        extra = {'_cmd': cmd}
        self.logger.info('Executing command', extra=extra)

        # Use the system default buffer size
        bufsize = -1

        transport = self.client.get_transport()
        chan = transport.open_session()

        start_time = time.time()
        if cmd.startswith('sudo'):
            # Note that fabric does this as well. If you set pty, stdout and stderr
            # streams will be combined into one.
            chan.get_pty()
        chan.exec_command(cmd)

        stdout = StringIO()
        stderr = StringIO()

        # Create a stdin file and immediately close it to prevent any
        # interactive script from hanging the process.
        stdin = chan.makefile('wb', bufsize)
        stdin.close()

        # Receive all the output
        # Note #1: This is used instead of chan.makefile approach to prevent
        # buffering issues and hanging if the executed command produces a lot
        # of output.
        #
        # Note #2: If you are going to remove "ready" checks inside the loop
        # you are going to have a bad time. Trying to consume from a channel
        # which is not ready will block for indefinitely.
        exit_status_ready = chan.exit_status_ready()

        if exit_status_ready:
            stdout.write(self._consume_stdout(chan).getvalue())
            stderr.write(self._consume_stderr(chan).getvalue())

        while not exit_status_ready:
            current_time = time.time()
            elapsed_time = (current_time - start_time)

            if timeout and (elapsed_time > timeout):
                # TODO: Is this the right way to clean up?
                chan.close()

                stdout = strip_shell_chars(stdout.getvalue())
                stderr = strip_shell_chars(stderr.getvalue())
                raise SSHCommandTimeoutError(cmd=cmd, timeout=timeout, stdout=stdout,
                                             stderr=stderr)

            stdout.write(self._consume_stdout(chan).getvalue())
            stderr.write(self._consume_stderr(chan).getvalue())

            # We need to check the exist status here, because the command could
            # print some output and exit during this sleep bellow.
            exit_status_ready = chan.exit_status_ready()

            if exit_status_ready:
                break

            # Short sleep to prevent busy waiting
            eventlet.sleep(self.SLEEP_DELAY)
        # print('Wait over. Channel must be ready for host: %s' % self.hostname)

        # Receive the exit status code of the command we ran.
        status = chan.recv_exit_status()

        stdout = strip_shell_chars(stdout.getvalue())
        stderr = strip_shell_chars(stderr.getvalue())

        extra = {'_status': status, '_stdout': stdout, '_stderr': stderr}
        self.logger.debug('Command finished', extra=extra)

        return [stdout, stderr, status]

    def close(self):
        self.logger.debug('Closing server connection')

        self.client.close()
        if self.sftp:
            self.sftp.close()
        if self.bastion_client:
            self.bastion_client.close()
        return True

    def _consume_stdout(self, chan):
        """
        Try to consume stdout data from chan if it's receive ready.
        """

        out = bytearray()
        stdout = StringIO()
        if chan.recv_ready():
            data = chan.recv(self.CHUNK_SIZE)
            out += data

            while data:
                ready = chan.recv_ready()

                if not ready:
                    break

                data = chan.recv(self.CHUNK_SIZE)
                out += data

        stdout.write(self._get_decoded_data(out))
        return stdout

    def _consume_stderr(self, chan):
        """
        Try to consume stderr data from chan if it's receive ready.
        """

        out = bytearray()
        stderr = StringIO()
        if chan.recv_stderr_ready():
            data = chan.recv_stderr(self.CHUNK_SIZE)
            out += data

            while data:
                ready = chan.recv_stderr_ready()

                if not ready:
                    break

                data = chan.recv_stderr(self.CHUNK_SIZE)
                out += data

        stderr.write(self._get_decoded_data(out))
        return stderr

    def _get_decoded_data(self, data):
        try:
            return data.decode('utf-8')
        except:
            self.logger.exception('Non UTF-8 character found in data: %s', data)
            raise

    def _get_pkey_object(self, key_material, passphrase):
        """
        Try to detect private key type and return paramiko.PKey object.
        """

        for cls in [paramiko.RSAKey, paramiko.DSSKey, paramiko.ECDSAKey]:
            try:
                key = cls.from_private_key(StringIO(key_material), password=passphrase)
            except paramiko.ssh_exception.SSHException:
                # Invalid key, try other key type
                pass
            else:
                return key

        # If a user passes in something which looks like file path we throw a more friendly
        # exception letting the user know we expect the contents a not a path.
        # Note: We do it here and not up the stack to avoid false positives.
        contains_header = PRIVATE_KEY_HEADER in key_material.lower()
        if not contains_header and (key_material.count('/') >= 1 or key_material.count('\\') >= 1):
            msg = ('"private_key" parameter needs to contain private key data / content and not '
                   'a path')
        elif passphrase:
            msg = 'Invalid passphrase or invalid/unsupported key type'
        else:
            msg = 'Invalid or unsupported key type'

        raise paramiko.ssh_exception.SSHException(msg)

    def _connect(self, host, socket=None):
        """

        :type host: ``str``
        :param host: Host to connect to

        :type socket: :class:`paramiko.Channel` or an opened :class:`socket.socket`
        :param socket: If specified, won't open a socket for communication to the specified host
                       and will use this instead

        :return: A connected SSHClient
        :rtype: :class:`paramiko.SSHClient`
        """
        conninfo = {'hostname': host,
                    'port': self.port,
                    'username': self.username,
                    'allow_agent': False,
                    'look_for_keys': False,
                    'timeout': self.timeout}

        if self.password:
            conninfo['password'] = self.password

        if self.key_files:
            conninfo['key_filename'] = self.key_files

        if self.key_material:
            conninfo['pkey'] = self._get_pkey_object(key_material=self.key_material,
                                                     passphrase=self.passphrase)

        if not self.password and not (self.key_files or self.key_material):
            conninfo['allow_agent'] = True
            conninfo['look_for_keys'] = True

        extra = {'_hostname': host, '_port': self.port,
                 '_username': self.username, '_timeout': self.timeout}
        self.logger.debug('Connecting to server', extra=extra)

        if socket:
            conninfo['sock'] = socket

        client = paramiko.SSHClient()
        client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        client.connect(**conninfo)

        return client

    def __repr__(self):
        return ('<ParamikoSSHClient hostname=%s,port=%s,username=%s,id=%s>' %
                (self.hostname, self.port, self.username, id(self)))


1			# Licensed to the Apache Software Foundation (ASF) under one or more
2			# contributor license agreements. See the NOTICE file distributed with
3			# this work for additional information regarding copyright ownership.
4			# The ASF licenses this file to You under the Apache License, Version 2.0
5			# (the "License"); you may not use this file except in compliance with
6			# the License. You may obtain a copy of the License at
7			#
8			# http://www.apache.org/licenses/LICENSE-2.0
9			#
10			# Unless required by applicable law or agreed to in writing, software
11			# distributed under the License is distributed on an "AS IS" BASIS,
12			# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13			# See the License for the specific language governing permissions and
14			# limitations under the License.
15
16			import os
17			import posixpath
18			from StringIO import StringIO
19			import time
20
21			import eventlet
22			from oslo_config import cfg
23
24			import paramiko
25
26			# Depending on your version of Paramiko, it may cause a deprecation
27			# warning on Python 2.6.
28			# Ref: https://bugs.launchpad.net/paramiko/+bug/392973
29
30			from st2common.log import logging
31			from st2common.util.misc import strip_shell_chars
32			from st2common.util.shell import quote_unix
33
34			__all__ = [
35			'ParamikoSSHClient',
36
37			'SSHCommandTimeoutError'
38			]
39
40			PRIVATE_KEY_HEADER = 'PRIVATE KEY-----'.lower()
41
42
43			class SSHCommandTimeoutError(Exception):
44			"""
45			Exception which is raised when an SSH command times out.
46			"""
47
48			def __init__(self, cmd, timeout, stdout=None, stderr=None):
49			"""
50			:param stdout: Stdout which was consumed until the timeout occured.
51			:type stdout: ``str``
52
53			:param stdout: Stderr which was consumed until the timeout occured.
54			:type stderr: ``str``
55			"""
56			self.cmd = cmd
57			self.timeout = timeout
58			self.stdout = stdout
59			self.stderr = stderr
60			message = 'Command didn\'t finish in %s seconds' % (timeout)
61			super(SSHCommandTimeoutError, self).__init__(message)
62
63			def __repr__(self):
64			return ('<SSHCommandTimeoutError: cmd="%s",timeout=%s)>' %
65			(self.cmd, self.timeout))
66
67			def __str__(self):
68			return self.message
69
70
71			class ParamikoSSHClient(object):
72			"""
73			A SSH Client powered by Paramiko.
74			"""
75
76			# Maximum number of bytes to read at once from a socket
77			CHUNK_SIZE = 1024
78			# How long to sleep while waiting for command to finish
79			SLEEP_DELAY = 1.5
80			# Connect socket timeout
81			CONNECT_TIMEOUT = 60
82
83			def __init__(self, hostname, port=22, username=None, password=None, bastion_host=None,
84			key=None, key_files=None, key_material=None, timeout=None, passphrase=None):
85			"""
86			Authentication is always attempted in the following order:
87
88			- The key passed in (if key is provided)
89			- Any key we can find through an SSH agent (only if no password and
90			key is provided)
91			- Any "id_rsa" or "id_dsa" key discoverable in ~/.ssh/ (only if no
92			password and key is provided)
93			- Plain username/password auth, if a password was given (if password is
94			provided)
95			"""
96			if key_files and key_material:
97			raise ValueError(('key_files and key_material arguments are '
98			'mutually exclusive'))
99
100			if passphrase and not key_material:
101			raise ValueError('passphrase should accompany private key material')
102
103			self.hostname = hostname
104			self.port = port
105			self.username = username if username else cfg.CONF.system_user
106			self.password = password
107			self.key = key if key else cfg.CONF.system_user.ssh_key_file
108			self.key_files = key_files
109			if not self.key_files and self.key:
110			self.key_files = key # `key` arg is deprecated.
111			self.timeout = timeout or ParamikoSSHClient.CONNECT_TIMEOUT
112			self.key_material = key_material
113			self.client = None
114			self.logger = logging.getLogger(__name__)
115			self.sftp = None
116			self.bastion_host = bastion_host
117			self.bastion_client = None
118			self.bastion_socket = None
119			self.passphrase = passphrase
120
121			def connect(self):
122			"""
123			Connect to the remote node over SSH.
124
125			:return: True if the connection has been successfully established,
126			False otherwise.
127			:rtype: ``bool``
128			"""
129			if self.bastion_host:
130			self.logger.debug('Bastion host specified, connecting')
131			self.bastion_client = self._connect(host=self.bastion_host)
132			transport = self.bastion_client.get_transport()
133			real_addr = (self.hostname, self.port)
134			# fabric uses ('', 0) for direct-tcpip, this duplicates that behaviour
135			# see https://github.com/fabric/fabric/commit/c2a9bbfd50f560df6c6f9675603fb405c4071cad
136			local_addr = ('', 0)
137			self.bastion_socket = transport.open_channel('direct-tcpip', real_addr, local_addr)
138
139			self.client = self._connect(host=self.hostname, socket=self.bastion_socket)
140			self.sftp = self.client.open_sftp()
141			return True
142
143			def put(self, local_path, remote_path, mode=None, mirror_local_mode=False):
144			"""
145			Upload a file to the remote node.
146
147			:type local_path: ``st``
148			:param local_path: File path on the local node.
149
150			:type remote_path: ``str``
151			:param remote_path: File path on the remote node.
152
153			:type mode: ``int``
154			:param mode: Permissions mode for the file. E.g. 0744.
155
156			:type mirror_local_mode: ``int``
157			:param mirror_local_mode: Should remote file mirror local mode.
158
159			:return: Attributes of the remote file.
160			:rtype: :class:`posix.stat_result` or ``None``
161			"""
162
163			if not local_path or not remote_path:
164			raise Exception('Need both local_path and remote_path. local: %s, remote: %s' %
165			local_path, remote_path)
166			local_path = quote_unix(local_path)
167			remote_path = quote_unix(remote_path)
168
169			extra = {'_local_path': local_path, '_remote_path': remote_path, '_mode': mode,
170			'_mirror_local_mode': mirror_local_mode}
171			self.logger.debug('Uploading file', extra=extra)
172
173			if not os.path.exists(local_path):
174			raise Exception('Path %s does not exist locally.' % local_path)
175
176			rattrs = self.sftp.put(local_path, remote_path)
177
178			if mode or mirror_local_mode:
179			local_mode = mode
180			if not mode or mirror_local_mode:
181			local_mode = os.stat(local_path).st_mode
182
183			# Cast to octal integer in case of string
184			if isinstance(local_mode, basestring):
185			local_mode = int(local_mode, 8)
186			local_mode = local_mode & 07777
187			remote_mode = rattrs.st_mode
188			# Only bitshift if we actually got an remote_mode
189			if remote_mode is not None:
190			remote_mode = (remote_mode & 07777)
191			if local_mode != remote_mode:
192			self.sftp.chmod(remote_path, local_mode)
193
194			return rattrs
195
196			def put_dir(self, local_path, remote_path, mode=None, mirror_local_mode=False):
197			"""
198			Upload a dir to the remote node.
199
200			:type local_path: ``str``
201			:param local_path: Dir path on the local node.
202
203			:type remote_path: ``str``
204			:param remote_path: Base dir path on the remote node.
205
206			:type mode: ``int``
207			:param mode: Permissions mode for the file. E.g. 0744.
208
209			:type mirror_local_mode: ``int``
210			:param mirror_local_mode: Should remote file mirror local mode.
211
212			:return: List of files created on remote node.
213			:rtype: ``list`` of ``str``
214			"""
215
216			extra = {'_local_path': local_path, '_remote_path': remote_path, '_mode': mode,
217			'_mirror_local_mode': mirror_local_mode}
218			self.logger.debug('Uploading dir', extra=extra)
219
220			if os.path.basename(local_path):
221			strip = os.path.dirname(local_path)
222			else:
223			strip = os.path.dirname(os.path.dirname(local_path))
224
225			remote_paths = []
226
227			for context, dirs, files in os.walk(local_path):
228			rcontext = context.replace(strip, '', 1)
229			# normalize pathname separators with POSIX separator
230			rcontext = rcontext.replace(os.sep, '/')
231			rcontext = rcontext.lstrip('/')
232			rcontext = posixpath.join(remote_path, rcontext)
233
234			if not self.exists(rcontext):
235			self.sftp.mkdir(rcontext)
236
237			for d in dirs:
238			n = posixpath.join(rcontext, d)
239			if not self.exists(n):
240			self.sftp.mkdir(n)
241
242			for f in files:
243			local_path = os.path.join(context, f)
244			n = posixpath.join(rcontext, f)
245			# Note that quote_unix is done by put anyways.
246			p = self.put(local_path=local_path, remote_path=n,
247			mirror_local_mode=mirror_local_mode, mode=mode)
248			remote_paths.append(p)
249
250			return remote_paths
251
252			def exists(self, remote_path):
253			"""
254			Validate whether a remote file or directory exists.
255
256			:param remote_path: Path to remote file.
257			:type remote_path: ``str``
258
259			:rtype: ``bool``
260			"""
261			try:
262			self.sftp.lstat(remote_path).st_mode
263			except IOError:
264			return False
265
266			return True
267
268			def mkdir(self, dir_path):
269			"""
270			Create a directory on remote box.
271
272			:param dir_path: Path to remote directory to be created.
273			:type dir_path: ``str``
274
275			:return: Returns nothing if successful else raises IOError exception.
276
277			:rtype: ``None``
278			"""
279
280			dir_path = quote_unix(dir_path)
281			extra = {'_dir_path': dir_path}
282			self.logger.debug('mkdir', extra=extra)
283			return self.sftp.mkdir(dir_path)
284
285			def delete_file(self, path):
286			"""
287			Delete a file on remote box.
288
289			:param path: Path to remote file to be deleted.
290			:type path: ``str``
291
292			:return: True if the file has been successfully deleted, False
293			otherwise.
294			:rtype: ``bool``
295			"""
296
297			path = quote_unix(path)
298			extra = {'_path': path}
299			self.logger.debug('Deleting file', extra=extra)
300			self.sftp.unlink(path)
301			return True
302
303			def delete_dir(self, path, force=False, timeout=None):
304			"""
305			Delete a dir on remote box.
306
307			:param path: Path to remote dir to be deleted.
308			:type path: ``str``
309
310			:param force: Optional Forcefully remove dir.
311			:type force: ``bool``
312
313			:param timeout: Optional Time to wait for dir to be deleted. Only relevant for force.
314			:type timeout: ``int``
315
316			:return: True if the file has been successfully deleted, False
317			otherwise.
318			:rtype: ``bool``
319			"""
320
321			path = quote_unix(path)
322			extra = {'_path': path}
323			if force:
324			command = 'rm -rf %s' % path
325			extra['_command'] = command
326			extra['_force'] = force
327			self.logger.debug('Deleting dir', extra=extra)
328			return self.run(command, timeout=timeout)
329
330			self.logger.debug('Deleting dir', extra=extra)
331			return self.sftp.rmdir(path)
332
333			def run(self, cmd, timeout=None, quote=False):
334			"""
335			Note: This function is based on paramiko's exec_command()
336			method.
337
338			:param timeout: How long to wait (in seconds) for the command to
339			finish (optional).
340			:type timeout: ``float``
341			"""
342
343			if quote:
344			cmd = quote_unix(cmd)
345
346			extra = {'_cmd': cmd}
347			self.logger.info('Executing command', extra=extra)
348
349			# Use the system default buffer size
350			bufsize = -1
351
352			transport = self.client.get_transport()
353			chan = transport.open_session()
354
355			start_time = time.time()
356			if cmd.startswith('sudo'):
357			# Note that fabric does this as well. If you set pty, stdout and stderr
358			# streams will be combined into one.
359			chan.get_pty()
360			chan.exec_command(cmd)
361
362			stdout = StringIO()
363			stderr = StringIO()
364
365			# Create a stdin file and immediately close it to prevent any
366			# interactive script from hanging the process.
367			stdin = chan.makefile('wb', bufsize)
368			stdin.close()
369
370			# Receive all the output
371			# Note #1: This is used instead of chan.makefile approach to prevent
372			# buffering issues and hanging if the executed command produces a lot
373			# of output.
374			#
375			# Note #2: If you are going to remove "ready" checks inside the loop
376			# you are going to have a bad time. Trying to consume from a channel
377			# which is not ready will block for indefinitely.
378			exit_status_ready = chan.exit_status_ready()
379
380			if exit_status_ready:
381			stdout.write(self._consume_stdout(chan).getvalue())
382			stderr.write(self._consume_stderr(chan).getvalue())
383
384			while not exit_status_ready:
385			current_time = time.time()
386			elapsed_time = (current_time - start_time)
387
388			if timeout and (elapsed_time > timeout):
389			# TODO: Is this the right way to clean up?
390			chan.close()
391
392			stdout = strip_shell_chars(stdout.getvalue())
393			stderr = strip_shell_chars(stderr.getvalue())
394			raise SSHCommandTimeoutError(cmd=cmd, timeout=timeout, stdout=stdout,
395			stderr=stderr)
396
397			stdout.write(self._consume_stdout(chan).getvalue())
398			stderr.write(self._consume_stderr(chan).getvalue())
399
400			# We need to check the exist status here, because the command could
401			# print some output and exit during this sleep bellow.
402			exit_status_ready = chan.exit_status_ready()
403
404			if exit_status_ready:
405			break
406
407			# Short sleep to prevent busy waiting
408			eventlet.sleep(self.SLEEP_DELAY)
409			# print('Wait over. Channel must be ready for host: %s' % self.hostname)
410
411			# Receive the exit status code of the command we ran.
412			status = chan.recv_exit_status()
413
414			stdout = strip_shell_chars(stdout.getvalue())
415			stderr = strip_shell_chars(stderr.getvalue())
416
417			extra = {'_status': status, '_stdout': stdout, '_stderr': stderr}
418			self.logger.debug('Command finished', extra=extra)
419
420			return [stdout, stderr, status]
421
422			def close(self):
423			self.logger.debug('Closing server connection')
424
425			self.client.close()
426			if self.sftp:
427			self.sftp.close()
428			if self.bastion_client:
429			self.bastion_client.close()
430			return True
431
432			def _consume_stdout(self, chan):
433			"""
434			Try to consume stdout data from chan if it's receive ready.
435			"""
436
437			out = bytearray()
438			stdout = StringIO()
439			if chan.recv_ready():
440			data = chan.recv(self.CHUNK_SIZE)
441			out += data
442
443			while data:
444			ready = chan.recv_ready()
445
446			if not ready:
447			break
448
449			data = chan.recv(self.CHUNK_SIZE)
450			out += data
451
452			stdout.write(self._get_decoded_data(out))
453			return stdout
454
455			def _consume_stderr(self, chan):
456			"""
457			Try to consume stderr data from chan if it's receive ready.
458			"""
459
460			out = bytearray()
461			stderr = StringIO()
462			if chan.recv_stderr_ready():
463			data = chan.recv_stderr(self.CHUNK_SIZE)
464			out += data
465
466			while data:
467			ready = chan.recv_stderr_ready()
468
469			if not ready:
470			break
471
472			data = chan.recv_stderr(self.CHUNK_SIZE)
473			out += data
474
475			stderr.write(self._get_decoded_data(out))
476			return stderr
477
478			def _get_decoded_data(self, data):
479			try:
480			return data.decode('utf-8')
481			except:
482			self.logger.exception('Non UTF-8 character found in data: %s', data)
483			raise
484
485			def _get_pkey_object(self, key_material, passphrase):
486			"""
487			Try to detect private key type and return paramiko.PKey object.
488			"""
489
490			for cls in [paramiko.RSAKey, paramiko.DSSKey, paramiko.ECDSAKey]:
491			try:
492			key = cls.from_private_key(StringIO(key_material), password=passphrase)
493			except paramiko.ssh_exception.SSHException:
494			# Invalid key, try other key type
495			pass
496			else:
497			return key
498
499			# If a user passes in something which looks like file path we throw a more friendly
500			# exception letting the user know we expect the contents a not a path.
501			# Note: We do it here and not up the stack to avoid false positives.
502			contains_header = PRIVATE_KEY_HEADER in key_material.lower()
503			if not contains_header and (key_material.count('/') >= 1 or key_material.count('\\') >= 1):
504			msg = ('"private_key" parameter needs to contain private key data / content and not '
505			'a path')
506			elif passphrase:
507			msg = 'Invalid passphrase or invalid/unsupported key type'
508			else:
509			msg = 'Invalid or unsupported key type'
510
511			raise paramiko.ssh_exception.SSHException(msg)
512
513			def _connect(self, host, socket=None):
514			"""
515
516			:type host: ``str``
517			:param host: Host to connect to
518
519			:type socket: :class:`paramiko.Channel` or an opened :class:`socket.socket`
520			:param socket: If specified, won't open a socket for communication to the specified host
521			and will use this instead
522
523			:return: A connected SSHClient
524			:rtype: :class:`paramiko.SSHClient`
525			"""
526			conninfo = {'hostname': host,
527			'port': self.port,
528			'username': self.username,
529			'allow_agent': False,
530			'look_for_keys': False,
531			'timeout': self.timeout}
532
533			if self.password:
534			conninfo['password'] = self.password
535
536			if self.key_files:
537			conninfo['key_filename'] = self.key_files
538
539			if self.key_material:
540			conninfo['pkey'] = self._get_pkey_object(key_material=self.key_material,
541			passphrase=self.passphrase)
542
543			if not self.password and not (self.key_files or self.key_material):
544			conninfo['allow_agent'] = True
545			conninfo['look_for_keys'] = True
546
547			extra = {'_hostname': host, '_port': self.port,
548			'_username': self.username, '_timeout': self.timeout}
549			self.logger.debug('Connecting to server', extra=extra)
550
551			if socket:
552			conninfo['sock'] = socket
553
554			client = paramiko.SSHClient()
555			client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
556			client.connect(**conninfo)
557
558			return client
559
560			def __repr__(self):
561			return ('<ParamikoSSHClient hostname=%s,port=%s,username=%s,id=%s>' %
562			(self.hostname, self.port, self.username, id(self)))
563

StackStorm / st2

Push — master ( 5db0c6...7afbb0 )

__init__() F

Complexity

Size

Duplication

Duplication Side-by-Side

Filter issues like

init() F