library.py

Go to the documentation of this file.

#  Copyright 2008-2015 Nokia Networks
#  Copyright 2016-     Robot Framework Foundation
#
#  Licensed 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.
 
from __future__ import print_function
 
import re
import os
from .deco import keyword
try:
    from robot.api import logger
except ImportError:
    logger = None
 
from .sshconnectioncache import SSHConnectionCache
from .abstractclient import SSHClientException
from .client import SSHClient
from .config import (Configuration, IntegerEntry, LogLevelEntry, NewlineEntry,
                     StringEntry, TimeEntry)
from .utils import ConnectionCache, is_string, is_truthy, plural_or_not
from .version import VERSION
 
 
__version__ = VERSION
 
 
class SSHLibrary():
    ROBOT_LIBRARY_SCOPE = 'GLOBAL'
    ROBOT_LIBRARY_VERSION = __version__
 
    DEFAULT_TIMEOUT = '3 seconds'
    DEFAULT_NEWLINE = 'LF'
    DEFAULT_PROMPT = None
    DEFAULT_LOGLEVEL = 'INFO'
    DEFAULT_TERM_TYPE = 'vt100'
    DEFAULT_TERM_WIDTH = 80
    DEFAULT_TERM_HEIGHT = 24
    DEFAULT_PATH_SEPARATOR = '/'
    DEFAULT_ENCODING = 'UTF-8'
    DEFAULT_ESCAPE_ANSI = False
    DEFAULT_ENCODING_ERRORS = 'strict'
 
    
    def __init__(self,
                 timeout=DEFAULT_TIMEOUT,
                 newline=DEFAULT_NEWLINE,
                 prompt=DEFAULT_PROMPT,
                 loglevel=DEFAULT_LOGLEVEL,
                 term_type=DEFAULT_TERM_TYPE,
                 width=DEFAULT_TERM_WIDTH,
                 height=DEFAULT_TERM_HEIGHT,
                 path_separator=DEFAULT_PATH_SEPARATOR,
                 encoding=DEFAULT_ENCODING,
                 escape_ansi=DEFAULT_ESCAPE_ANSI,
                 encoding_errors=DEFAULT_ENCODING_ERRORS):
        self._connections_connections = SSHConnectionCache()
        self._config_config = _DefaultConfiguration(
            timeout or self.DEFAULT_TIMEOUTDEFAULT_TIMEOUT,
            newline or self.DEFAULT_NEWLINEDEFAULT_NEWLINE,
            prompt or self.DEFAULT_PROMPTDEFAULT_PROMPT,
            loglevel or self.DEFAULT_LOGLEVELDEFAULT_LOGLEVEL,
            term_type or self.DEFAULT_TERM_TYPEDEFAULT_TERM_TYPE,
            width or self.DEFAULT_TERM_WIDTHDEFAULT_TERM_WIDTH,
            height or self.DEFAULT_TERM_HEIGHTDEFAULT_TERM_HEIGHT,
            path_separator or self.DEFAULT_PATH_SEPARATORDEFAULT_PATH_SEPARATOR,
            encoding or self.DEFAULT_ENCODINGDEFAULT_ENCODING,
            escape_ansi or self.DEFAULT_ESCAPE_ANSIDEFAULT_ESCAPE_ANSI,
            encoding_errors or self.DEFAULT_ENCODING_ERRORSDEFAULT_ENCODING_ERRORS
        )
        self._last_commands_last_commands = dict()
 
    @property
    current = property
    
    def current(self):
        return self._connections_connections.current
 
    @keyword(types=None)
    
    def set_default_configuration(self, timeout=None, newline=None, prompt=None,
                                  loglevel=None, term_type=None, width=None,
                                  height=None, path_separator=None,
                                  encoding=None, escape_ansi=None, encoding_errors=None):
        self._config_config.update(timeout=timeout, newline=newline, prompt=prompt,
                            loglevel=loglevel, term_type=term_type, width=width,
                            height=height, path_separator=path_separator,
                            encoding=encoding, escape_ansi=escape_ansi, encoding_errors=encoding_errors)
 
    
    def set_client_configuration(self, timeout=None, newline=None, prompt=None,
                                 term_type=None, width=None, height=None,
                                 path_separator=None, encoding=None, escape_ansi=None, encoding_errors=None):
        self.currentcurrentcurrent.config.update(timeout=timeout, newline=newline,
                                   prompt=prompt, term_type=term_type,
                                   width=width, height=height,
                                   path_separator=path_separator,
                                   encoding=encoding, escape_ansi=escape_ansi,
                                   encoding_errors=encoding_errors)
 
    
    def enable_ssh_logging(self, logfile):
        if SSHClient.enable_logging(logfile):
            self._log_log('SSH log is written to <a href="%s">file</a>.' % logfile,
                      'HTML')
 
    
    def open_connection(self, host, alias=None, port=22, timeout=None,
                        newline=None, prompt=None, term_type=None, width=None,
                        height=None, path_separator=None, encoding=None, escape_ansi=None, encoding_errors=None):
        timeout = timeout or self._config_config.timeout
        newline = newline or self._config_config.newline
        prompt = prompt or self._config_config.prompt
        term_type = term_type or self._config_config.term_type
        width = width or self._config_config.width
        height = height or self._config_config.height
        path_separator = path_separator or self._config_config.path_separator
        encoding = encoding or self._config_config.encoding
        escape_ansi = escape_ansi or self._config_config.escape_ansi
        encoding_errors = encoding_errors or self._config_config.encoding_errors
        client = SSHClient(host, alias, port, timeout, newline, prompt,
                           term_type, width, height, path_separator, encoding, escape_ansi, encoding_errors)
        connection_index = self._connections_connections.register(client, alias)
        client.config.update(index=connection_index)
        return connection_index
 
    
    def switch_connection(self, index_or_alias):
        old_index = self._connections_connections.current_index
        if index_or_alias is None:
            self.close_connectionclose_connection()
        else:
            self._connections_connections.switch(index_or_alias)
        return old_index
 
    
    def close_connection(self):
        connections = self._connections_connections
        connections.close_current()
 
 
    
    def close_all_connections(self):
        self._connections_connections.close_all()
 
    
    def get_connection(self, index_or_alias=None, index=False, host=False,
                       alias=False, port=False, timeout=False, newline=False,
                       prompt=False, term_type=False, width=False, height=False,
                       encoding=False, escape_ansi=False):
        if not index_or_alias:
            index_or_alias = self._connections_connections.current_index
        try:
            config = self._connections_connections.get_connection(index_or_alias).config
        except RuntimeError:
            config = SSHClient(None).config
        except AttributeError:
            config = SSHClient(None).config
        self._log_log(str(config), self._config_config.loglevel)
        return_values = tuple(self._get_config_values_get_config_values(config, index, host,
                                                      alias, port, timeout,
                                                      newline, prompt,
                                                      term_type, width, height,
                                                      encoding, escape_ansi))
        if not return_values:
            return config
        if len(return_values) == 1:
            return return_values[0]
        return return_values
 
    def _log(self, msg, level='INFO'):
        level = self._active_loglevel_active_loglevel(level)
        if level != 'NONE':
            msg = msg.strip()
            if not msg:
                return
            if logger:
                logger.write(msg, level)
            else:
                print('*%s* %s' % (level, msg))
 
    def _active_loglevel(self, level):
        if level is None:
            return self._config_config.loglevel
        if is_string(level) and \
                level.upper() in ['TRACE', 'DEBUG', 'INFO', 'WARN', 'HTML', 'NONE']:
            return level.upper()
        raise AssertionError("Invalid log level '%s'." % level)
 
    def _get_config_values(self, config, index, host, alias, port, timeout,
                           newline, prompt, term_type, width, height, encoding, escape_ansi):
        if is_truthy(index):
            yield config.index
        if is_truthy(host):
            yield config.host
        if is_truthy(alias):
            yield config.alias
        if is_truthy(port):
            yield config.port
        if is_truthy(timeout):
            yield config.timeout
        if is_truthy(newline):
            yield config.newline
        if is_truthy(prompt):
            yield config.prompt
        if is_truthy(term_type):
            yield config.term_type
        if is_truthy(width):
            yield config.width
        if is_truthy(height):
            yield config.height
        if is_truthy(encoding):
            yield config.encoding
        if is_truthy(escape_ansi):
            yield config.escape_ansi
 
    
    def get_connections(self):
        configs = [c.config for c in self._connections_connections._connections if c]
        for c in configs:
            self._log_log(str(c), self._config_config.loglevel)
        return configs
 
    
    def login(self, username=None, password=None, allow_agent=False, look_for_keys=False, delay='0.5 seconds',
              proxy_cmd=None, read_config=False, jumphost_index_or_alias=None, keep_alive_interval='0 seconds'):
        jumphost_connection_conf = self.get_connectionget_connection(index_or_alias=jumphost_index_or_alias) \
            if jumphost_index_or_alias else None
        jumphost_connection = self._connections_connections.connections[jumphost_connection_conf.index-1] \
            if jumphost_connection_conf and jumphost_connection_conf.index else None
 
        return self._login_login(self.currentcurrentcurrent.login, username, password, is_truthy(allow_agent),
                           is_truthy(look_for_keys), delay, proxy_cmd, is_truthy(read_config),
                           jumphost_connection, keep_alive_interval)
 
    
    def login_with_public_key(self, username=None, keyfile=None, password='',
                              allow_agent=False, look_for_keys=False,
                              delay='0.5 seconds', proxy_cmd=None,
                              jumphost_index_or_alias=None,
                              read_config=False, keep_alive_interval='0 seconds'):
        if proxy_cmd and jumphost_index_or_alias:
            raise ValueError("`proxy_cmd` and `jumphost_connection` are mutually exclusive SSH features.")
        jumphost_connection_conf = self.get_connectionget_connection(index_or_alias=jumphost_index_or_alias) if jumphost_index_or_alias else None
        jumphost_connection = self._connections_connections.connections[jumphost_connection_conf.index-1] if jumphost_connection_conf and jumphost_connection_conf.index else None
        return self._login_login(self.currentcurrentcurrent.login_with_public_key, username,
                           keyfile, password, is_truthy(allow_agent),
                           is_truthy(look_for_keys), delay, proxy_cmd,
                           jumphost_connection, is_truthy(read_config), keep_alive_interval)
 
    def _login(self, login_method, username, *args):
        self._log_log("Logging into '%s:%s' as '%s'."
                   % (self.currentcurrentcurrent.config.host, self.currentcurrentcurrent.config.port,
                      username), self._config_config.loglevel)
        try:
            login_output = login_method(username, *args)
            if is_truthy(self.currentcurrentcurrent.config.escape_ansi):
                login_output = self._escape_ansi_sequences_escape_ansi_sequences(login_output)
            self._log_log('Read output: %s' % login_output, self._config_config.loglevel)
            return login_output
        except SSHClientException as e:
            raise RuntimeError(e)
 
    
    def get_pre_login_banner(self, host=None, port=22):
        if host:
            banner = SSHClient.get_banner_without_login(host, port)
        elif self.currentcurrentcurrent:
            banner = self.currentcurrentcurrent.get_banner()
        else:
            raise RuntimeError("'host' argument is mandatory if there is no open connection.")
        return banner.decode(self.DEFAULT_ENCODINGDEFAULT_ENCODING)
 
    
    def execute_command(self, command, return_stdout=True, return_stderr=False,
                        return_rc=False, sudo=False,  sudo_password=None, timeout=None, output_during_execution=False,
                        output_if_timeout=False, invoke_subsystem=False, forward_agent=False):
        if not is_truthy(sudo):
            self._log_log("Executing command '%s'." % command, self._config_config.loglevel)
        else:
            self._log_log("Executing command 'sudo %s'." % command, self._config_config.loglevel)
        opts = self._legacy_output_options_legacy_output_options(return_stdout, return_stderr,
                                           return_rc)
        stdout, stderr, rc = self.currentcurrentcurrent.execute_command(command, sudo, sudo_password,
                                                          timeout, output_during_execution, output_if_timeout,
                                                          is_truthy(invoke_subsystem), forward_agent)
        return self._return_command_output_return_command_output(stdout, stderr, rc, *opts)
 
    
    def start_command(self, command, sudo=False,  sudo_password=None, invoke_subsystem=False, forward_agent=False):
        if not is_truthy(sudo):
            self._log_log("Starting command '%s'." % command, self._config_config.loglevel)
        else:
            self._log_log("Starting command 'sudo %s'." % command, self._config_config.loglevel)
        if self.currentcurrentcurrent.config.index not in self._last_commands_last_commands.keys():
            self._last_commands_last_commands[self.currentcurrentcurrent.config.index] = command
        else:
            temp_dict = {self.currentcurrentcurrent.config.index: command}
            self._last_commands_last_commands.update(temp_dict)
        self.currentcurrentcurrent.start_command(command, sudo, sudo_password, is_truthy(invoke_subsystem), is_truthy(forward_agent))
 
    
    def read_command_output(self, return_stdout=True, return_stderr=False,
                            return_rc=False, timeout=None):
        self._log_log("Reading output of command '%s'." % self._last_commands_last_commands.get(self.currentcurrentcurrent.config.index), self._config_config.loglevel)
        opts = self._legacy_output_options_legacy_output_options(return_stdout, return_stderr,
                                           return_rc)
        try:
            stdout, stderr, rc = self.currentcurrentcurrent.read_command_output(timeout=timeout)
        except SSHClientException as msg:
            raise RuntimeError(msg)
        return self._return_command_output_return_command_output(stdout, stderr, rc, *opts)
 
    
    def create_local_ssh_tunnel(self, local_port, remote_host, remote_port=22, bind_address=None):
        self.currentcurrentcurrent.create_local_ssh_tunnel(local_port, remote_host, remote_port, bind_address)
 
    def _legacy_output_options(self, stdout, stderr, rc):
        if not is_string(stdout):
            return stdout, stderr, rc
        stdout = stdout.lower()
        if stdout == 'stderr':
            return False, True, rc
        if stdout == 'both':
            return True, True, rc
        return stdout, stderr, rc
 
    def _return_command_output(self, stdout, stderr, rc, return_stdout,
                               return_stderr, return_rc):
        self._log_log("Command exited with return code %d." % rc, self._config_config.loglevel)
        ret = []
        if is_truthy(return_stdout):
            ret.append(stdout.rstrip('\n'))
        if is_truthy(return_stderr):
            ret.append(stderr.rstrip('\n'))
        if is_truthy(return_rc):
            ret.append(rc)
        if len(ret) == 1:
            return ret[0]
        return ret
 
    
    def write(self, text, loglevel=None):
        self._write_write(text, add_newline=True)
        return self._read_and_log_read_and_log(loglevel, self.currentcurrentcurrent.read_until_newline)
 
    
    def write_bare(self, text):
        self._write_write(text)
 
    def _write(self, text, add_newline=False):
        try:
            self.currentcurrentcurrent.write(text, is_truthy(add_newline))
        except SSHClientException as e:
            raise RuntimeError(e)
 
    
    def write_until_expected_output(self, text, expected, timeout,
                                    retry_interval, loglevel=None):
        self._read_and_log_read_and_log(loglevel, self.currentcurrentcurrent.write_until_expected, text,
                           expected, timeout, retry_interval)
 
    
    def read(self, loglevel=None, delay=None):
        return self._read_and_log_read_and_log(loglevel, self.currentcurrentcurrent.read, delay)
 
    
    def read_until(self, expected, loglevel=None):
        return self._read_and_log_read_and_log(loglevel, self.currentcurrentcurrent.read_until, expected)
 
    
    def read_until_prompt(self, loglevel=None, strip_prompt=False):
        return self._read_and_log_read_and_log(loglevel, self.currentcurrentcurrent.read_until_prompt, is_truthy(strip_prompt))
 
    
    def read_until_regexp(self, regexp, loglevel=None):
        return self._read_and_log_read_and_log(loglevel, self.currentcurrentcurrent.read_until_regexp,
                                  regexp)
 
    def _read_and_log(self, loglevel, reader, *args):
        try:
            output = reader(*args)
        except SSHClientException as e:
            if is_truthy(self.currentcurrentcurrent.config.escape_ansi):
                message = self._escape_ansi_sequences_escape_ansi_sequences(e.args[0])
                raise RuntimeError(message)
            raise RuntimeError(e)
        if is_truthy(self.currentcurrentcurrent.config.escape_ansi):
            output = self._escape_ansi_sequences_escape_ansi_sequences(output)
        self._log_log(output, loglevel)
        return output
 
    @staticmethod
    def _escape_ansi_sequences(output):
        ansi_escape = re.compile(r'(?:\x1B[@-_]|[\x80-\x9F])[0-?]*[ -/]*[@-~]', flags=re.IGNORECASE)
        output = ansi_escape.sub('', output)
        return ("%r" % output)[1:-1].encode().decode('unicode-escape')
 
    
    def get_file(self, source, destination='.', scp='OFF', scp_preserve_times=False):
        return self._run_command_run_command(self.currentcurrentcurrent.get_file, source,
                                 destination, scp, scp_preserve_times)
 
    
    def get_directory(self, source, destination='.', recursive=False,
                      scp='OFF', scp_preserve_times=False):
        return self._run_command_run_command(self.currentcurrentcurrent.get_directory, source,
                                 destination, is_truthy(recursive), scp, scp_preserve_times)
 
    
    def put_file(self, source, destination='.', mode='0744', newline='',
                 scp='OFF', scp_preserve_times=False):
        return self._run_command_run_command(self.currentcurrentcurrent.put_file, source,
                                 destination, mode, newline, scp, scp_preserve_times)
 
    
    def put_directory(self, source, destination='.', mode='0744', newline='',
                      recursive=False, scp='OFF', scp_preserve_times=False):
        return self._run_command_run_command(self.currentcurrentcurrent.put_directory, source,
                                 destination, mode, newline,
                                 is_truthy(recursive), scp, scp_preserve_times)
 
    def _run_command(self, command, *args):
        try:
            files = command(*args)
        except SSHClientException as e:
            raise RuntimeError(e)
        if files:
            for src, dst in files:
                self._log_log("'%s' -> '%s'" % (src, dst), self._config_config.loglevel)
 
    
    def file_should_exist(self, path):
        if not self.currentcurrentcurrent.is_file(path):
            raise AssertionError("File '%s' does not exist." % path)
 
    
    def file_should_not_exist(self, path):
        if self.currentcurrentcurrent.is_file(path):
            raise AssertionError("File '%s' exists." % path)
 
    
    def directory_should_exist(self, path):
        if not self.currentcurrentcurrent.is_dir(path):
            raise AssertionError("Directory '%s' does not exist." % path)
 
    
    def directory_should_not_exist(self, path):
        if self.currentcurrentcurrent.is_dir(path):
            raise AssertionError("Directory '%s' exists." % path)
 
    
    def list_directory(self, path, pattern=None, absolute=False):
        try:
            items = self.currentcurrentcurrent.list_dir(path, pattern, is_truthy(absolute))
        except SSHClientException as msg:
            raise RuntimeError(msg)
        self._log_log('%d item%s:\n%s' % (len(items), plural_or_not(items),
                                       '\n'.join(items)), self._config_config.loglevel)
        return items
 
    
    def list_files_in_directory(self, path, pattern=None, absolute=False):
        absolute = is_truthy(absolute)
        try:
            files = self.currentcurrentcurrent.list_files_in_dir(path, pattern, absolute)
        except SSHClientException as msg:
            raise RuntimeError(msg)
        files = self.currentcurrentcurrent.list_files_in_dir(path, pattern, absolute)
        self._log_log('%d file%s:\n%s' % (len(files), plural_or_not(files),
                                       '\n'.join(files)), self._config_config.loglevel)
        return files
 
    
    def list_directories_in_directory(self, path, pattern=None, absolute=False):
        try:
            dirs = self.currentcurrentcurrent.list_dirs_in_dir(path, pattern, is_truthy(absolute))
        except SSHClientException as msg:
            raise RuntimeError(msg)
        self._log_log('%d director%s:\n%s' % (len(dirs),
                                           'y' if len(dirs) == 1 else 'ies',
                                           '\n'.join(dirs)), self._config_config.loglevel)
        return dirs
 
 
class _DefaultConfiguration(Configuration):
 
    def __init__(self, timeout, newline, prompt, loglevel, term_type, width,
                 height, path_separator, encoding, escape_ansi, encoding_errors):
        super(_DefaultConfiguration, self).__init__(
            timeout=TimeEntry(timeout),
            newline=NewlineEntry(newline),
            prompt=StringEntry(prompt),
            loglevel=LogLevelEntry(loglevel),
            term_type=StringEntry(term_type),
            width=IntegerEntry(width),
            height=IntegerEntry(height),
            path_separator=StringEntry(path_separator),
            encoding=StringEntry(encoding),
            escape_ansi=StringEntry(escape_ansi),
            encoding_errors=StringEntry(encoding_errors)
        )