devlib/utils/ssh.py

#    Copyright 2014-2015 ARM Limited
#
# 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.
#


import os
import stat
import logging
import subprocess
import re
import threading
import tempfile
import shutil
import socket
import time

import pexpect
from distutils.version import StrictVersion as V
if V(pexpect.__version__) < V('4.0.0'):
    import pxssh
else:
    from pexpect import pxssh
from pexpect import EOF, TIMEOUT, spawn

from devlib.exception import HostError, TargetError, TimeoutError
from devlib.utils.misc import which, strip_bash_colors, escape_single_quotes, check_output
from devlib.utils.types import boolean


ssh = None
scp = None
sshpass = None


logger = logging.getLogger('ssh')
gem5_logger = logging.getLogger('gem5-connection')

def ssh_get_shell(host, username, password=None, keyfile=None, port=None, timeout=10, telnet=False, original_prompt=None):
    _check_env()
    start_time = time.time()
    while True:
        if telnet:
            if keyfile:
                raise ValueError('keyfile may not be used with a telnet connection.')
            conn = TelnetPxssh(original_prompt=original_prompt)
        else:  # ssh
            conn = pxssh.pxssh()

        try:
            if keyfile:
                conn.login(host, username, ssh_key=keyfile, port=port, login_timeout=timeout)
            else:
                conn.login(host, username, password, port=port, login_timeout=timeout)
            break
        except EOF:
            timeout -= time.time() - start_time
            if timeout <= 0:
                message = 'Could not connect to {}; is the host name correct?'
                raise TargetError(message.format(host))
            time.sleep(5)

    conn.setwinsize(500,200)
    conn.sendline('')
    conn.prompt()
    conn.setecho(False)
    return conn


class TelnetPxssh(pxssh.pxssh):
    # pylint: disable=arguments-differ

    def __init__(self, original_prompt):
        super(TelnetPxssh, self).__init__()
        self.original_prompt = original_prompt or r'[#$]'

    def login(self, server, username, password='', login_timeout=10,
              auto_prompt_reset=True, sync_multiplier=1, port=23):
        args = ['telnet']
        if username is not None:
            args += ['-l', username]
        args += [server, str(port)]
        cmd = ' '.join(args)

        spawn._spawn(self, cmd)  # pylint: disable=protected-access

        try:
            i = self.expect('(?i)(?:password)', timeout=login_timeout)
            if i == 0:
                self.sendline(password)
                i = self.expect([self.original_prompt, 'Login incorrect'], timeout=login_timeout)
            if i:
                raise pxssh.ExceptionPxssh('could not log in: password was incorrect')
        except TIMEOUT:
            if not password:
                # No password promt before TIMEOUT & no password provided
                # so assume everything is okay
                pass
            else:
                raise pxssh.ExceptionPxssh('could not log in: did not see a password prompt')

        if not self.sync_original_prompt(sync_multiplier):
            self.close()
            raise pxssh.ExceptionPxssh('could not synchronize with original prompt')

        if auto_prompt_reset:
            if not self.set_unique_prompt():
                self.close()
                message = 'could not set shell prompt (recieved: {}, expected: {}).'
                raise pxssh.ExceptionPxssh(message.format(self.before, self.PROMPT))
        return True


def check_keyfile(keyfile):
    """
    keyfile must have the right access premissions in order to be useable. If the specified
    file doesn't, create a temporary copy and set the right permissions for that.

    Returns either the ``keyfile`` (if the permissions on it are correct) or the path to a
    temporary copy with the right permissions.
    """
    desired_mask = stat.S_IWUSR | stat.S_IRUSR
    actual_mask = os.stat(keyfile).st_mode & 0xFF
    if actual_mask != desired_mask:
        tmp_file = os.path.join(tempfile.gettempdir(), os.path.basename(keyfile))
        shutil.copy(keyfile, tmp_file)
        os.chmod(tmp_file, desired_mask)
        return tmp_file
    else:  # permissions on keyfile are OK
        return keyfile


class SshConnection(object):

    default_password_prompt = '[sudo] password'
    max_cancel_attempts = 5
    default_timeout=10

    @property
    def name(self):
        return self.host

    def __init__(self,
                 host,
                 username,
                 password=None,
                 keyfile=None,
                 port=None,
                 timeout=None,
                 telnet=False,
                 password_prompt=None,
                 original_prompt=None,
                 platform=None
                 ):
        self.host = host
        self.username = username
        self.password = password
        self.keyfile = check_keyfile(keyfile) if keyfile else keyfile
        self.port = port
        self.lock = threading.Lock()
        self.password_prompt = password_prompt if password_prompt is not None else self.default_password_prompt
        logger.debug('Logging in {}@{}'.format(username, host))
        timeout = timeout if timeout is not None else self.default_timeout
        self.conn = ssh_get_shell(host, username, password, self.keyfile, port, timeout, False, None)

    def push(self, source, dest, timeout=30):
        dest = '{}@{}:{}'.format(self.username, self.host, dest)
        return self._scp(source, dest, timeout)

    def pull(self, source, dest, timeout=30):
        source = '{}@{}:{}'.format(self.username, self.host, source)
        return self._scp(source, dest, timeout)

    def execute(self, command, timeout=None, check_exit_code=True,
                as_root=False, strip_colors=True): #pylint: disable=unused-argument
        if command == '':
            # Empty command is valid but the __devlib_ec stuff below will
            # produce a syntax error with bash. Treat as a special case.
            return ''
        try:
            with self.lock:
                _command = '({}); __devlib_ec=$?; echo; echo $__devlib_ec'.format(command)
                raw_output = self._execute_and_wait_for_prompt(
                    _command, timeout, as_root, strip_colors)
                output, exit_code_text, _ = raw_output.rsplit('\r\n', 2)
                if check_exit_code:
                    try:
                        exit_code = int(exit_code_text)
                        if exit_code:
                            message = 'Got exit code {}\nfrom: {}\nOUTPUT: {}'
                            raise TargetError(message.format(exit_code, command, output))
                    except (ValueError, IndexError):
                        logger.warning(
                            'Could not get exit code for "{}",\ngot: "{}"'\
                            .format(command, exit_code_text))
                return output
        except EOF:
            raise TargetError('Connection lost.')

    def background(self, command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, as_root=False):
        try:
            port_string = '-p {}'.format(self.port) if self.port else ''
            keyfile_string = '-i {}'.format(self.keyfile) if self.keyfile else ''
            if as_root:
                command = "sudo -- sh -c '{}'".format(command)
            command = '{} {} {} {}@{} {}'.format(ssh, keyfile_string, port_string, self.username, self.host, command)
            logger.debug(command)
            if self.password:
                command = _give_password(self.password, command)
            return subprocess.Popen(command, stdout=stdout, stderr=stderr, shell=True)
        except EOF:
            raise TargetError('Connection lost.')

    def close(self):
        logger.debug('Logging out {}@{}'.format(self.username, self.host))
        self.conn.logout()

    def cancel_running_command(self):
        # simulate impatiently hitting ^C until command prompt appears
        logger.debug('Sending ^C')
        for _ in xrange(self.max_cancel_attempts):
            self.conn.sendline(chr(3))
            if self.conn.prompt(0.1):
                return True
        return False

    def _execute_and_wait_for_prompt(self, command, timeout=None, as_root=False, strip_colors=True, log=True):
        self.conn.prompt(0.1)  # clear an existing prompt if there is one.
        if self.username == 'root':
            # As we're already root, there is no need to use sudo.
            as_root = False
        if as_root:
            command = "sudo -- sh -c '{}'".format(escape_single_quotes(command))
            if log:
                logger.debug(command)
            self.conn.sendline(command)
            if self.password:
                index = self.conn.expect_exact([self.password_prompt, TIMEOUT], timeout=0.5)
                if index == 0:
                    self.conn.sendline(self.password)
        else:  # not as_root
            if log:
                logger.debug(command)
            self.conn.sendline(command)
        timed_out = self._wait_for_prompt(timeout)
        # the regex removes line breaks potential introduced when writing
        # command to shell.
        output = process_backspaces(self.conn.before)
        output = re.sub(r'\r([^\n])', r'\1', output)
        if '\r\n' in output: # strip the echoed command
            output = output.split('\r\n', 1)[1]
        if timed_out:
            self.cancel_running_command()
            raise TimeoutError(command, output)
        if strip_colors:
            output = strip_bash_colors(output)
        return output

    def _wait_for_prompt(self, timeout=None):
        if timeout:
            return not self.conn.prompt(timeout)
        else:  # cannot timeout; wait forever
            while not self.conn.prompt(1):
                pass
            return False

    def _scp(self, source, dest, timeout=30):
        # NOTE: the version of scp in Ubuntu 12.04 occasionally (and bizarrely)
        # fails to connect to a device if port is explicitly specified using -P
        # option, even if it is the default port, 22. To minimize this problem,
        # only specify -P for scp if the port is *not* the default.
        port_string = '-P {}'.format(self.port) if (self.port and self.port != 22) else ''
        keyfile_string = '-i {}'.format(self.keyfile) if self.keyfile else ''
        command = '{} -r {} {} {} {}'.format(scp, keyfile_string, port_string, source, dest)
        pass_string = ''
        logger.debug(command)
        if self.password:
            command = _give_password(self.password, command)
        try:
            check_output(command, timeout=timeout, shell=True)
        except subprocess.CalledProcessError as e:
            raise subprocess.CalledProcessError(e.returncode, e.cmd.replace(pass_string, ''), e.output)
        except TimeoutError as e:
            raise TimeoutError(e.command.replace(pass_string, ''), e.output)


class TelnetConnection(SshConnection):

    def __init__(self,
                 host,
                 username,
                 password=None,
                 port=None,
                 timeout=None,
                 password_prompt=None,
                 original_prompt=None,
                 platform=None):
        self.host = host
        self.username = username
        self.password = password
        self.port = port
        self.keyfile = None
        self.lock = threading.Lock()
        self.password_prompt = password_prompt if password_prompt is not None else self.default_password_prompt
        logger.debug('Logging in {}@{}'.format(username, host))
        timeout = timeout if timeout is not None else self.default_timeout
        self.conn = ssh_get_shell(host, username, password, None, port, timeout, True, original_prompt)


class Gem5Connection(TelnetConnection):

    def __init__(self,
                 platform,
                 host=None,
                 username=None,
                 password=None,
                 port=None,
                 timeout=None,
                 password_prompt=None,
                 original_prompt=None,
                 ):
        if host is not None:
            host_system = socket.gethostname()
            if host_system != host:
                raise TargetError("Gem5Connection can only connect to gem5 "
                                   "simulations on your current host, which "
                                   "differs from the one given {}!"
                                   .format(host_system, host))
        if username is not None and username != 'root':
            raise ValueError('User should be root in gem5!')
        if password is not None and password != '':
            raise ValueError('No password needed in gem5!')
        self.username = 'root'
        self.is_rooted = True
        self.password = None
        self.port = None
        # Long timeouts to account for gem5 being slow
        # Can be overriden if the given timeout is longer
        self.default_timeout = 3600
        if timeout is not None:
            if timeout > self.default_timeout:
                logger.info('Overwriting the default timeout of gem5 ({})'
                                 ' to {}'.format(self.default_timeout, timeout))
                self.default_timeout = timeout
            else:
                logger.info('Ignoring the given timeout --> gem5 needs longer timeouts')
        self.ready_timeout = self.default_timeout * 3
        # Counterpart in gem5_interact_dir
        self.gem5_input_dir = '/mnt/host/'
        # Location of m5 binary in the gem5 simulated system
        self.m5_path = None
        # Actual telnet connection to gem5 simulation
        self.conn = None
        # Flag to indicate the gem5 device is ready to interact with the
        # outer world
        self.ready = False
        # Lock file to prevent multiple connections to same gem5 simulation
        # (gem5 does not allow this)
        self.lock_directory = '/tmp/'
        self.lock_file_name = None # Will be set once connected to gem5

        # These parameters will be set by either the method to connect to the
        # gem5 platform or directly to the gem5 simulation
        # Intermediate directory to push things to gem5 using VirtIO
        self.gem5_interact_dir = None
        # Directory to store output  from gem5 on the host
        self.gem5_out_dir = None
        # Actual gem5 simulation
        self.gem5simulation = None

        # Connect to gem5
        if platform:
            self._connect_gem5_platform(platform)

        # Wait for boot
        self._wait_for_boot()

        # Mount the virtIO to transfer files in/out gem5 system
        self._mount_virtio()

    def set_hostinteractdir(self, indir):
        logger.info('Setting hostinteractdir  from {} to {}'
                    .format(self.gem5_input_dir, indir))
        self.gem5_input_dir = indir

    def push(self, source, dest, timeout=None):
        """
        Push a file to the gem5 device using VirtIO

        The file to push to the device is copied to the temporary directory on
        the host, before being copied within the simulation to the destination.
        Checks, in the form of 'ls' with error code checking, are performed to
        ensure that the file is copied to the destination.
        """
        # First check if the connection is set up to interact with gem5
        self._check_ready()

        filename = os.path.basename(source)
        logger.debug("Pushing {} to device.".format(source))
        logger.debug("gem5interactdir: {}".format(self.gem5_interact_dir))
        logger.debug("dest: {}".format(dest))
        logger.debug("filename: {}".format(filename))

        # We need to copy the file to copy to the temporary directory
        self._move_to_temp_dir(source)

        # Dest in gem5 world is a file rather than directory
        if os.path.basename(dest) != filename:
            dest = os.path.join(dest, filename)
        # Back to the gem5 world
        self._gem5_shell("ls -al {}{}".format(self.gem5_input_dir, filename))
        self._gem5_shell("cat '{}''{}' > '{}'".format(self.gem5_input_dir,
                                                     filename,
                                                     dest))
        self._gem5_shell("sync")
        self._gem5_shell("ls -al {}".format(dest))
        self._gem5_shell("ls -al {}".format(self.gem5_input_dir))
        logger.debug("Push complete.")

    def pull(self, source, dest, timeout=0): #pylint: disable=unused-argument
        """
        Pull a file from the gem5 device using m5 writefile

        The file is copied to the local directory within the guest as the m5
        writefile command assumes that the file is local. The file is then
        written out to the host system using writefile, prior to being moved to
        the destination on the host.
        """
        # First check if the connection is set up to interact with gem5
        self._check_ready()

        result = self._gem5_shell("ls {}".format(source))
        files = result.split()

        for filename in files:
            dest_file = os.path.basename(filename)
            logger.debug("pull_file {} {}".format(filename, dest_file))
            # writefile needs the file to be copied to be in the current
            # working directory so if needed, copy to the working directory
            # We don't check the exit code here because it is non-zero if the
            # source and destination are the same. The ls below will cause an
            # error if the file was not where we expected it to be.
            if os.path.isabs(source):
                if os.path.dirname(source) != self.execute('pwd',
                                              check_exit_code=False):
                    self._gem5_shell("cat '{}' > '{}'".format(filename,
                                                              dest_file))
            self._gem5_shell("sync")
            self._gem5_shell("ls -la {}".format(dest_file))
            logger.debug('Finished the copy in the simulator')
            self._gem5_util("writefile {}".format(dest_file))

            if 'cpu' not in filename:
                while not os.path.exists(os.path.join(self.gem5_out_dir,
                                                      dest_file)):
                    time.sleep(1)

            # Perform the local move
            if os.path.exists(os.path.join(dest, dest_file)):
                logger.warning(
                            'Destination file {} already exists!'\
                            .format(dest_file))
            else:
                shutil.move(os.path.join(self.gem5_out_dir, dest_file), dest)
            logger.debug("Pull complete.")

    def execute(self, command, timeout=1000, check_exit_code=True,
                as_root=False, strip_colors=True):
        """
        Execute a command on the gem5 platform
        """
        # First check if the connection is set up to interact with gem5
        self._check_ready()

        output = self._gem5_shell(command,
                                  check_exit_code=check_exit_code,
                                  as_root=as_root)
        if strip_colors:
            output = strip_bash_colors(output)
        return output

    def background(self, command, stdout=subprocess.PIPE,
                   stderr=subprocess.PIPE, as_root=False):
        # First check if the connection is set up to interact with gem5
        self._check_ready()

        # Create the logfile for stderr/stdout redirection
        command_name = command.split(' ')[0].split('/')[-1]
        redirection_file = 'BACKGROUND_{}.log'.format(command_name)
        trial = 0
        while os.path.isfile(redirection_file):
            # Log file already exists so add to name
           redirection_file = 'BACKGROUND_{}{}.log'.format(command_name, trial)
           trial += 1

        # Create the command to pass on to gem5 shell
        complete_command = '{} >> {} 2>&1 &'.format(command, redirection_file)
        output = self._gem5_shell(complete_command, as_root=as_root)
        output = strip_bash_colors(output)
        gem5_logger.info('STDERR/STDOUT of background command will be '
                         'redirected to {}. Use target.pull() to '
                         'get this file'.format(redirection_file))
        return output

    def close(self):
        """
        Close and disconnect from the gem5 simulation. Additionally, we remove
        the temporary directory used to pass files into the simulation.
        """
        gem5_logger.info("Gracefully terminating the gem5 simulation.")
        try:
            self._gem5_util("exit")
            self.gem5simulation.wait()
        except EOF:
            pass
        gem5_logger.info("Removing the temporary directory")
        try:
            shutil.rmtree(self.gem5_interact_dir)
        except OSError:
            gem5_logger.warn("Failed to remove the temporary directory!")

        # Delete the lock file
        os.remove(self.lock_file_name)

    # Functions only to be called by the Gem5 connection itself
    def _connect_gem5_platform(self, platform):
        port = platform.gem5_port
        gem5_simulation = platform.gem5
        gem5_interact_dir = platform.gem5_interact_dir
        gem5_out_dir = platform.gem5_out_dir

        self.connect_gem5(port, gem5_simulation, gem5_interact_dir, gem5_out_dir)

    # This function connects to the gem5 simulation
    def connect_gem5(self, port, gem5_simulation, gem5_interact_dir,
                      gem5_out_dir):
        """
        Connect to the telnet port of the gem5 simulation.

        We connect, and wait for the prompt to be found. We do not use a timeout
        for this, and wait for the prompt in a while loop as the gem5 simulation
        can take many hours to reach a prompt when booting the system. We also
        inject some newlines periodically to try and force gem5 to show a
        prompt. Once the prompt has been found, we replace it with a unique
        prompt to ensure that we are able to match it properly. We also disable
        the echo as this simplifies parsing the output when executing commands
        on the device.
        """
        host = socket.gethostname()
        gem5_logger.info("Connecting to the gem5 simulation on port {}".format(port))

        # Check if there is no on-going connection yet
        lock_file_name = '{}{}_{}.LOCK'.format(self.lock_directory, host, port)
        if os.path.isfile(lock_file_name):
            # There is already a connection to this gem5 simulation
            raise TargetError('There is already a connection to the gem5 '
                              'simulation using port {} on {}!'
                              .format(port, host))

        # Connect to the gem5 telnet port. Use a short timeout here.
        attempts = 0
        while attempts < 10:
            attempts += 1
            try:
                self.conn = TelnetPxssh(original_prompt=None)
                self.conn.login(host, self.username, port=port,
                                login_timeout=10, auto_prompt_reset=False)
                break
            except pxssh.ExceptionPxssh:
                pass
        else:
            gem5_simulation.kill()
            raise TargetError("Failed to connect to the gem5 telnet session.")

        gem5_logger.info("Connected! Waiting for prompt...")

        # Create the lock file
        self.lock_file_name = lock_file_name
        open(self.lock_file_name, 'w').close() # Similar to touch
        gem5_logger.info("Created lock file {} to prevent reconnecting to "
                         "same simulation".format(self.lock_file_name))

        # We need to find the prompt. It might be different if we are resuming
        # from a checkpoint. Therefore, we test multiple options here.
        prompt_found = False
        while not prompt_found:
            try:
                self._login_to_device()
            except TIMEOUT:
                pass
            try:
                # Try and force a prompt to be shown
                self.conn.send('\n')
                self.conn.expect([r'# ', self.conn.UNIQUE_PROMPT, r'\[PEXPECT\][\\\$\#]+ '], timeout=60)
                prompt_found = True
            except TIMEOUT:
                pass

        gem5_logger.info("Successfully logged in")
        gem5_logger.info("Setting unique prompt...")

        self.conn.set_unique_prompt()
        self.conn.prompt()
        gem5_logger.info("Prompt found and replaced with a unique string")

        # We check that the prompt is what we think it should be. If not, we
        # need to update the regex we use to match.
        self._find_prompt()

        self.conn.setecho(False)
        self._sync_gem5_shell()

        # Fully connected to gem5 simulation
        self.gem5_interact_dir = gem5_interact_dir
        self.gem5_out_dir = gem5_out_dir
        self.gem5simulation = gem5_simulation

        # Ready for interaction now
        self.ready = True

    def _login_to_device(self):
        """
        Login to device, will be overwritten if there is an actual login
        """
        pass

    def _find_prompt(self):
        prompt = r'\[PEXPECT\][\\\$\#]+ '
        synced = False
        while not synced:
            self.conn.send('\n')
            i = self.conn.expect([prompt, self.conn.UNIQUE_PROMPT, r'[\$\#] '], timeout=self.default_timeout)
            if i == 0:
                synced = True
            elif i == 1:
                prompt = self.conn.UNIQUE_PROMPT
                synced = True
            else:
                prompt = re.sub(r'\$', r'\\\$', self.conn.before.strip() + self.conn.after.strip())
                prompt = re.sub(r'\#', r'\\\#', prompt)
                prompt = re.sub(r'\[', r'\[', prompt)
                prompt = re.sub(r'\]', r'\]', prompt)

        self.conn.PROMPT = prompt

    def _sync_gem5_shell(self):
        """
        Synchronise with the gem5 shell.

        Write some unique text to the gem5 device to allow us to synchronise
        with the shell output. We actually get two prompts so we need to match
        both of these.
        """
        gem5_logger.debug("Sending Sync")
        self.conn.send("echo \*\*sync\*\*\n")
        self.conn.expect(r"\*\*sync\*\*", timeout=self.default_timeout)
        self.conn.expect([self.conn.UNIQUE_PROMPT, self.conn.PROMPT], timeout=self.default_timeout)
        self.conn.expect([self.conn.UNIQUE_PROMPT, self.conn.PROMPT], timeout=self.default_timeout)

    def _gem5_util(self, command):
        """ Execute a gem5 utility command using the m5 binary on the device """
        if self.m5_path is None:
            raise TargetError('Path to m5 binary on simulated system  is not set!')
        self._gem5_shell('{} {}'.format(self.m5_path, command))

    def _gem5_shell(self, command, as_root=False, timeout=None, check_exit_code=True, sync=True):  # pylint: disable=R0912
        """
        Execute a command in the gem5 shell

        This wraps the telnet connection to gem5 and processes the raw output.

        This method waits for the shell to return, and then will try and
        separate the output from the command from the command itself. If this
        fails, warn, but continue with the potentially wrong output.

        The exit code is also checked by default, and non-zero exit codes will
        raise a TargetError.
        """
        if sync:
            self._sync_gem5_shell()

        gem5_logger.debug("gem5_shell command: {}".format(command))

        # Send the actual command
        self.conn.send("{}\n".format(command))

        # Wait for the response. We just sit here and wait for the prompt to
        # appear, as gem5 might take a long time to provide the output. This
        # avoids timeout issues.
        command_index = -1
        while command_index == -1:
            if self.conn.prompt():
                output = re.sub(r' \r([^\n])', r'\1', self.conn.before)
                output = re.sub(r'[\b]', r'', output)
                # Deal with line wrapping
                output = re.sub(r'[\r].+?<', r'', output)
                command_index = output.find(command)

                # If we have -1, then we cannot match the command, but the
                # prompt has returned. Hence, we have a bit of an issue. We
                # warn, and return the whole output.
                if command_index == -1:
                    gem5_logger.warn("gem5_shell: Unable to match command in "
                                     "command output. Expect parsing errors!")
                    command_index = 0

        output = output[command_index + len(command):].strip()

        # It is possible that gem5 will echo the command. Therefore, we need to
        # remove that too!
        command_index = output.find(command)
        if command_index != -1:
            output = output[command_index + len(command):].strip()

        gem5_logger.debug("gem5_shell output: {}".format(output))

        # We get a second prompt. Hence, we need to eat one to make sure that we
        # stay in sync. If we do not do this, we risk getting out of sync for
        # slower simulations.
        self.conn.expect([self.conn.UNIQUE_PROMPT, self.conn.PROMPT], timeout=self.default_timeout)

        if check_exit_code:
            exit_code_text = self._gem5_shell('echo $?', as_root=as_root,
                                             timeout=timeout, check_exit_code=False,
                                             sync=False)
            try:
                exit_code = int(exit_code_text.split()[0])
                if exit_code:
                    message = 'Got exit code {}\nfrom: {}\nOUTPUT: {}'
                    raise TargetError(message.format(exit_code, command, output))
            except (ValueError, IndexError):
                gem5_logger.warning('Could not get exit code for "{}",\ngot: "{}"'.format(command, exit_code_text))

        return output

    def _mount_virtio(self):
        """
        Mount the VirtIO device in the simulated system.
        """
        gem5_logger.info("Mounting VirtIO device in simulated system")

        self._gem5_shell('su -c "mkdir -p {}" root'.format(self.gem5_input_dir))
        mount_command = "mount -t 9p -o trans=virtio,version=9p2000.L,aname={} gem5 {}".format(self.gem5_interact_dir, self.gem5_input_dir)
        self._gem5_shell(mount_command)

    def _move_to_temp_dir(self, source):
        """
        Move a file to the temporary directory on the host for copying to the
        gem5 device
        """
        command = "cp {} {}".format(source, self.gem5_interact_dir)
        gem5_logger.debug("Local copy command: {}".format(command))
        subprocess.call(command.split())
        subprocess.call("sync".split())

    def _check_ready(self):
        """
        Check if the gem5 platform is ready
        """
        if not self.ready:
            raise TargetError('Gem5 is not ready to interact yet')

    def _wait_for_boot(self):
        pass

    def _probe_file(self, filepath):
        """
        Internal method to check if the target has a certain file
        """
        command = 'if [ -e \'{}\' ]; then echo 1; else echo 0; fi'
        output = self.execute(command.format(filepath), as_root=self.is_rooted)
        return boolean(output.strip())


class LinuxGem5Connection(Gem5Connection):

    def _login_to_device(self):
        gem5_logger.info("Trying to log in to gem5 device")
        login_prompt = ['login:', 'AEL login:', 'username:', 'aarch64-gem5 login:']
        login_password_prompt = ['password:']
        # Wait for the login prompt
        prompt = login_prompt + [self.conn.UNIQUE_PROMPT]
        i = self.conn.expect(prompt, timeout=10)
        # Check if we are already at a prompt, or if we need to log in.
        if i < len(prompt) - 1:
            self.conn.sendline("{}".format(self.username))
            password_prompt = login_password_prompt + [r'# ', self.conn.UNIQUE_PROMPT]
            j = self.conn.expect(password_prompt, timeout=self.default_timeout)
            if j < len(password_prompt) - 2:
                self.conn.sendline("{}".format(self.password))
                self.conn.expect([r'# ', self.conn.UNIQUE_PROMPT], timeout=self.default_timeout)



class AndroidGem5Connection(Gem5Connection):

    def _wait_for_boot(self):
        """
        Wait for the system to boot

        We monitor the sys.boot_completed and service.bootanim.exit system
        properties to determine when the system has finished booting. In the
        event that we cannot coerce the result of service.bootanim.exit to an
        integer, we assume that the boot animation was disabled and do not wait
        for it to finish.

        """
        gem5_logger.info("Waiting for Android to boot...")
        while True:
            booted = False
            anim_finished = True  # Assume boot animation was disabled on except
            try:
                booted = (int('0' + self._gem5_shell('getprop sys.boot_completed', check_exit_code=False).strip()) == 1)
                anim_finished = (int(self._gem5_shell('getprop service.bootanim.exit', check_exit_code=False).strip()) == 1)
            except ValueError:
                pass
            if booted and anim_finished:
                break
            time.sleep(60)

        gem5_logger.info("Android booted")

def _give_password(password, command):
    if not sshpass:
        raise HostError('Must have sshpass installed on the host in order to use password-based auth.')
    pass_string = "sshpass -p '{}' ".format(password)
    return pass_string + command


def _check_env():
    global ssh, scp, sshpass  # pylint: disable=global-statement
    if not ssh:
        ssh = which('ssh')
        scp = which('scp')
        sshpass = which('sshpass')
    if not (ssh and scp):
        raise HostError('OpenSSH must be installed on the host.')


def process_backspaces(text):
    chars = []
    for c in text:
        if c == chr(8) and chars:  # backspace
            chars.pop()
        else:
            chars.append(c)
    return ''.join(chars)