ptyprocess

scrapli.transport.plugins.system.ptyprocess

PtyProcess

Source code in transport/plugins/system/ptyprocess.py

class PtyProcess:
    def __init__(self, pid: int, fd: int) -> None:
        """
        This class represents a process running in a pseudoterminal.

        The main constructor is the `spawn` method.

        Args:
            pid: integer value of pid
            fd: integer value of fd

        Returns:
            None

        Raises:
            N/A

        """
        _make_eof_intr()  # Ensure _EOF and _INTR are calculated
        self.pid = pid
        self.fd = fd
        readf = io.open(fd, "rb", buffering=0)
        writef = io.open(fd, "wb", buffering=0, closefd=False)
        self.fileobj = io.BufferedRWPair(readf, writef)  # type: ignore

        self.terminated = False
        self.closed = False
        self.exitstatus: Optional[int] = None
        self.signalstatus: Optional[int] = None
        # status returned by os.waitpid
        self.status: Optional[int] = None
        self.flag_eof = False
        # Used by close() to give kernel time to update process status.
        # Time in seconds.
        self.delayafterclose = 0.1
        # Used by terminate() to give kernel time to update process status.
        # Time in seconds.
        self.delayafterterminate = 0.1

    @classmethod
    def spawn(
        cls: Type[PtyProcessType],
        spawn_command: List[str],
        echo: bool = True,
        rows: int = 80,
        cols: int = 256,
    ) -> PtyProcessType:
        """
        Start the given command in a child process in a pseudo terminal.

        This does all the fork/exec type of stuff for a pty, and returns an instance of PtyProcess.
        For some devices setting terminal width strictly in the operating system (the actual network
        operating system) does not seem to be sufficient by itself for setting terminal length or
        width -- so we have optional values for rows/cols that can be passed here as well.

        Args:
            spawn_command: command to execute with arguments (if applicable), as a list
            echo: enable/disable echo -- defaults to True, should be left as True for "normal"
                scrapli operations, optionally disable for scrapli_netconf operations.
            rows: integer number of rows for ptyprocess "window"
            cols: integer number of cols for ptyprocess "window"

        Returns:
            PtyProcessType: instantiated PtyProcess object

        Raises:
            ScrapliValueError: if no ssh binary found on PATH
            Exception: IOError - if unable to set window size of child process
            Exception: OSError - if unable to spawn command in child process
            IOError: failing to reset window size
            exception: if we get an exception decoding output

        """
        # Note that it is difficult for this method to fail.
        # You cannot detect if the child process cannot start.
        # So the only way you can tell if the child process started
        # or not is to try to read from the file descriptor. If you get
        # EOF immediately then it means that the child is already dead.
        # That may not necessarily be bad because you may have spawned a child
        # that performs some task; creates no stdout output; and then dies.

        import fcntl
        import pty
        import resource
        import termios
        from pty import CHILD, STDIN_FILENO

        spawn_executable = which(spawn_command[0])
        if spawn_executable is None:
            raise ScrapliValueError("ssh executable not found!")
        spawn_command[0] = spawn_executable

        # [issue #119] To prevent the case where exec fails and the user is
        # stuck interacting with a python child process instead of whatever
        # was expected, we implement the solution from
        # http://stackoverflow.com/a/3703179 to pass the exception to the
        # parent process
        # [issue #119] 1. Before forking, open a pipe in the parent process.
        exec_err_pipe_read, exec_err_pipe_write = os.pipe()

        pid, fd = pty.fork()

        # Some platforms must call setwinsize() and setecho() from the
        # child process, and others from the master process. We do both,
        # allowing IOError for either.
        if pid == CHILD:
            try:
                _setwinsize(fd=STDIN_FILENO, rows=rows, cols=cols)
            except IOError as err:
                if err.args[0] not in (errno.EINVAL, errno.ENOTTY):
                    raise

            # disable echo if requested
            if echo is False:
                try:
                    _setecho(STDIN_FILENO, False)
                except (IOError, termios.error) as err:
                    if err.args[0] not in (errno.EINVAL, errno.ENOTTY):
                        raise

            # [issue #119] 3. The child closes the reading end and sets the
            # close-on-exec flag for the writing end.
            os.close(exec_err_pipe_read)
            fcntl.fcntl(exec_err_pipe_write, fcntl.F_SETFD, fcntl.FD_CLOEXEC)

            # Do not allow child to inherit open file descriptors from parent,
            # with the exception of the exec_err_pipe_write of the pipe.
            # Impose ceiling on max_fd: AIX bugfix for users with unlimited
            # nofiles where resource.RLIMIT_NOFILE is 2^63-1 and os.closerange()
            # occasionally raises out of range error
            max_fd = min(1048576, resource.getrlimit(resource.RLIMIT_NOFILE)[0])
            pass_fds = sorted({exec_err_pipe_write})
            for pair in zip([2] + pass_fds, pass_fds + [max_fd]):
                os.closerange(pair[0] + 1, pair[1])

            try:
                os.execv(spawn_executable, spawn_command)
            except OSError as err:
                # [issue #119] 5. If exec fails, the child writes the error
                # code back to the parent using the pipe, then exits.
                tosend = f"OSError:{err.errno}:{err}".encode()
                os.write(exec_err_pipe_write, tosend)
                os.close(exec_err_pipe_write)
                os._exit(os.EX_OSERR)

        # Parent
        inst = cls(pid, fd)

        # [issue #119] 2. After forking, the parent closes the writing end
        # of the pipe and reads from the reading end.
        os.close(exec_err_pipe_write)
        exec_err_data = os.read(exec_err_pipe_read, 4096)
        os.close(exec_err_pipe_read)

        # [issue #119] 6. The parent reads eof (a zero-length read) if the
        # child successfully performed exec, since close-on-exec made
        # successful exec close the writing end of the pipe. Or, if exec
        # failed, the parent reads the error code and can proceed
        # accordingly. Either way, the parent blocks until the child calls
        # exec.
        if len(exec_err_data) != 0:
            try:
                errclass, errno_s, errmsg = exec_err_data.split(b":", 2)
                exctype = getattr(builtins, errclass.decode("ascii"), Exception)

                exception = exctype(errmsg.decode("utf-8", "replace"))
                if exctype is OSError:
                    exception.errno = int(errno_s)
            except Exception:
                raise Exception("Subprocess failed, got bad error data: %r" % exec_err_data)
            else:
                raise exception

        try:
            inst.setwinsize(rows=rows, cols=cols)
        except IOError as err:
            if err.args[0] not in (errno.EINVAL, errno.ENOTTY, errno.ENXIO):
                raise

        return inst

    def __repr__(self) -> str:
        """
        Magic repr method for PtyProcess

        Args:
            N/A

        Returns:
            str: str repr of object

        Raises:
            N/A

        """
        return f"{type(self).__name__}(pid={self.pid}, fd={self.fd})"

    def __del__(self) -> None:
        """
        Magic delete method for PtyProcess

        This makes sure that no system resources are left open. Python only
        garbage collects Python objects. OS file descriptors are not Python
        objects, so they must be handled explicitly. If the child file
        descriptor was opened outside of this class (passed to the constructor)
        then this does not close it.

        Args:
            N/A

        Returns:
            None

        Raises:
            N/A

        """
        if not self.closed:
            # It is possible for __del__ methods to execute during the
            # teardown of the Python VM itself. Thus self.close() may
            # trigger an exception because os.close may be None.
            with suppress(Exception):
                self.close()

    def close(self) -> None:
        """
        Close the instance

        This closes the connection with the child application. Note that
        calling close() more than once is valid. This emulates standard Python
        behavior with files. Set force to True if you want to make sure that
        the child is terminated (SIGKILL is sent if the child ignores SIGHUP
        and SIGINT).

        Args:
            N/A

        Returns:
            None

        Raises:
            PtyProcessError: if child cannot be terminated

        """
        if not self.closed:
            # in the original ptyprocess vendor'd code the file object is "gracefully" closed,
            # however in some situations it seemed to hang forever on the close call... given that
            # as soon as this connection is closed it will need to be re-opened, and that will of
            # course re-create the fileobject this seems like an ok workaround because for reasons
            # unknown to me... this does not hang (even though in theory delete method just closes
            # things...?)
            with suppress(AttributeError):
                del self.fileobj

            # Give kernel time to update process status.
            time.sleep(self.delayafterclose)
            if self.isalive():
                if not self.terminate(force=True):
                    raise PtyProcessError("Could not terminate the child.")
            self.fd = -1
            self.closed = True
            self.pid = None

    def flush(self) -> None:
        """
        This does nothing.

        It is here to support the interface for a File-like object.

        Args:
            N/A

        Returns:
            None

        Raises:
            N/A

        """

    def read(self, size: int = 1024) -> bytes:
        """
        Read and return at most ``size`` bytes from the pty.

        Can block if there is nothing to read. Raises :exc:`EOFError` if the
        terminal was closed.

        Unlike Pexpect's ``read_nonblocking`` method, this doesn't try to deal
        with the vagaries of EOF on platforms that do strange things, like IRIX
        or older Solaris systems. It handles the errno=EIO pattern used on
        Linux, and the empty-string return used on BSD platforms and (seemingly)
        on recent Solaris.

        Args:
            size: bytes to read

        Returns:
            bytes: bytes read

        Raises:
            EOFError: eof encountered
            EOFError: eof encountered

        """
        try:
            s = self.fileobj.read1(size)
        except (OSError, IOError) as err:
            if err.args[0] == errno.EIO:
                # Linux-style EOF
                self.flag_eof = True
                raise EOFError("End Of File (EOF). Exception style platform.")
            raise
        if s == b"":
            # BSD-style EOF (also appears to work on recent Solaris (OpenIndiana))
            self.flag_eof = True
            raise EOFError("End Of File (EOF). Empty string style platform.")

        return s

    def write(self, bytes_to_write: bytes, flush: bool = True) -> int:
        """
        Write bytes to the pseudoterminal.

        Returns the number of bytes written.

        Args:
            bytes_to_write: bytes to write to the terminal
            flush: flush the terminal or not

        Returns:
            int: number of bytes written

        Raises:
            N/A

        """
        n = self.fileobj.write(bytes_to_write)
        if flush:
            self.fileobj.flush()
        return n

    def eof(self) -> bool:
        """
        This returns True if the EOF exception was ever raised.

        Args:
            N/A

        Returns:
            bool: if eof

        Raises:
            N/A

        """
        return self.flag_eof

    def terminate(self, force: bool = False) -> bool:
        """
        This forces a child process to terminate.

        It starts nicely with SIGHUP and SIGINT. If "force" is True then moves onto SIGKILL. This
        returns True if the child was terminated. This returns False if the child could not be
        terminated.

        Args:
            force: bool; force termination

        Returns:
            bool: terminate succeeded or failed

        Raises:
            N/A

        """
        if not self.isalive():
            return True
        try:
            self.kill(signal.SIGHUP)
            time.sleep(self.delayafterterminate)
            if not self.isalive():
                return True
            self.kill(signal.SIGCONT)
            time.sleep(self.delayafterterminate)
            if not self.isalive():
                return True
            self.kill(signal.SIGINT)
            time.sleep(self.delayafterterminate)
            if not self.isalive():
                return True
            if force:
                self.kill(signal.SIGKILL)
                time.sleep(self.delayafterterminate)
                if not self.isalive():
                    return True
        except OSError:
            # I think there are kernel timing issues that sometimes cause
            # this to happen. I think isalive() reports True, but the
            # process is dead to the kernel.
            # Make one last attempt to see if the kernel is up to date.
            time.sleep(self.delayafterterminate)
            if not self.isalive():
                return True
        return False

    def isalive(self) -> bool:
        """
        This tests if the child process is running or not. This is
        non-blocking. If the child was terminated then this will read the
        exitstatus or signalstatus of the child. This returns True if the child
        process appears to be running or False if not. It can take literally
        SECONDS for Solaris to return the right status.

        Args:
            N/A

        Returns:
            bool: if alive

        Raises:
            PtyProcessError: an error occurred with the process
            PtyProcessError: an error occurred with the process
            PtyProcessError: an error occurred with the process

        """

        if self.terminated:
            return False

        if self.flag_eof:
            # This is for Linux, which requires the blocking form
            # of waitpid to get the status of a defunct process.
            # This is super-lame. The flag_eof would have been set
            # in read_nonblocking(), so this should be safe.
            waitpid_options = 0
        else:
            waitpid_options = os.WNOHANG

        try:
            pid, status = os.waitpid(self.pid, waitpid_options)
        except OSError as e:
            # No child processes
            if e.errno == errno.ECHILD:
                raise PtyProcessError(
                    "isalive() encountered condition "
                    + 'where "terminated" is 0, but there was no child '
                    + "process. Did someone else call waitpid() "
                    + "on our process?"
                )
            raise

        # I have to do this twice for Solaris.
        # I can't even believe that I figured this out...
        # If waitpid() returns 0 it means that no child process
        # wishes to report, and the value of status is undefined.
        if pid == 0:
            try:
                ### os.WNOHANG) # Solaris!
                pid, status = os.waitpid(self.pid, waitpid_options)
            except OSError as e:  # pragma: no cover
                # This should never happen...
                if e.errno == errno.ECHILD:
                    raise PtyProcessError(
                        "isalive() encountered condition "
                        + "that should never happen. There was no child "
                        + "process. Did someone else call waitpid() "
                        + "on our process?"
                    )
                raise

            # If pid is still 0 after two calls to waitpid() then the process
            # really is alive. This seems to work on all platforms, except for
            # Irix which seems to require a blocking call on waitpid or select,
            # so I let read_nonblocking take care of this situation
            # (unfortunately, this requires waiting through the timeout).
            if pid == 0:
                return True

        if pid == 0:
            return True

        if os.WIFEXITED(status):
            self.status = status
            self.exitstatus = os.WEXITSTATUS(status)
            self.signalstatus = None
            self.terminated = True
        elif os.WIFSIGNALED(status):
            self.status = status
            self.exitstatus = None
            self.signalstatus = os.WTERMSIG(status)
            self.terminated = True
        elif os.WIFSTOPPED(status):
            raise PtyProcessError(
                "isalive() encountered condition "
                + "where child process is stopped. This is not "
                + "supported. Is some other process attempting "
                + "job control with our child pid?"
            )
        return False

    def kill(self, sig: int) -> None:
        """
        Send the given signal to the child application.

        In keeping with UNIX tradition it has a misleading name. It does not
        necessarily kill the child unless you send the right signal. See the
        :mod:`signal` module for constants representing signal numbers.

        Args:
            sig: signal to send to kill

        Returns:
            None

        Raises:
            N/A

        """

        # Same as os.kill, but the pid is given for you.
        if self.isalive():
            os.kill(self.pid, sig)

    def setwinsize(self, rows: int = 24, cols: int = 80) -> None:
        """
        Set window size.

        This will cause a SIGWINCH signal to be sent to the child. This does not change the physical
        window size. It changes the size reported to TTY-aware applications like vi or curses --
        applications that respond to the SIGWINCH signal.

        Args:
            rows: int number of rows for terminal
            cols: int number of cols for terminal

        Returns:
            None

        Raises:
            N/A

        """
        return _setwinsize(self.fd, rows, cols)

__del__() -> None

Magic delete method for PtyProcess

This makes sure that no system resources are left open. Python only garbage collects Python objects. OS file descriptors are not Python objects, so they must be handled explicitly. If the child file descriptor was opened outside of this class (passed to the constructor) then this does not close it.

Returns:

Type	Description
None	None

Source code in transport/plugins/system/ptyprocess.py

def __del__(self) -> None:
    """
    Magic delete method for PtyProcess

    This makes sure that no system resources are left open. Python only
    garbage collects Python objects. OS file descriptors are not Python
    objects, so they must be handled explicitly. If the child file
    descriptor was opened outside of this class (passed to the constructor)
    then this does not close it.

    Args:
        N/A

    Returns:
        None

    Raises:
        N/A

    """
    if not self.closed:
        # It is possible for __del__ methods to execute during the
        # teardown of the Python VM itself. Thus self.close() may
        # trigger an exception because os.close may be None.
        with suppress(Exception):
            self.close()

__init__(pid: int, fd: int) -> None

This class represents a process running in a pseudoterminal.

The main constructor is the spawn method.

Parameters:

Name	Type	Description	Default
pid	int	integer value of pid	required
fd	int	integer value of fd	required

Returns:

Type	Description
None	None

Source code in transport/plugins/system/ptyprocess.py

def __init__(self, pid: int, fd: int) -> None:
    """
    This class represents a process running in a pseudoterminal.

    The main constructor is the `spawn` method.

    Args:
        pid: integer value of pid
        fd: integer value of fd

    Returns:
        None

    Raises:
        N/A

    """
    _make_eof_intr()  # Ensure _EOF and _INTR are calculated
    self.pid = pid
    self.fd = fd
    readf = io.open(fd, "rb", buffering=0)
    writef = io.open(fd, "wb", buffering=0, closefd=False)
    self.fileobj = io.BufferedRWPair(readf, writef)  # type: ignore

    self.terminated = False
    self.closed = False
    self.exitstatus: Optional[int] = None
    self.signalstatus: Optional[int] = None
    # status returned by os.waitpid
    self.status: Optional[int] = None
    self.flag_eof = False
    # Used by close() to give kernel time to update process status.
    # Time in seconds.
    self.delayafterclose = 0.1
    # Used by terminate() to give kernel time to update process status.
    # Time in seconds.
    self.delayafterterminate = 0.1

__repr__() -> str

Magic repr method for PtyProcess

Returns:

Name	Type	Description
str	str	str repr of object

Source code in transport/plugins/system/ptyprocess.py

def __repr__(self) -> str:
    """
    Magic repr method for PtyProcess

    Args:
        N/A

    Returns:
        str: str repr of object

    Raises:
        N/A

    """
    return f"{type(self).__name__}(pid={self.pid}, fd={self.fd})"

close() -> None

Close the instance

This closes the connection with the child application. Note that calling close() more than once is valid. This emulates standard Python behavior with files. Set force to True if you want to make sure that the child is terminated (SIGKILL is sent if the child ignores SIGHUP and SIGINT).

Returns:

Type	Description
None	None

Raises:

Type	Description
PtyProcessError	if child cannot be terminated

Source code in transport/plugins/system/ptyprocess.py

def close(self) -> None:
    """
    Close the instance

    This closes the connection with the child application. Note that
    calling close() more than once is valid. This emulates standard Python
    behavior with files. Set force to True if you want to make sure that
    the child is terminated (SIGKILL is sent if the child ignores SIGHUP
    and SIGINT).

    Args:
        N/A

    Returns:
        None

    Raises:
        PtyProcessError: if child cannot be terminated

    """
    if not self.closed:
        # in the original ptyprocess vendor'd code the file object is "gracefully" closed,
        # however in some situations it seemed to hang forever on the close call... given that
        # as soon as this connection is closed it will need to be re-opened, and that will of
        # course re-create the fileobject this seems like an ok workaround because for reasons
        # unknown to me... this does not hang (even though in theory delete method just closes
        # things...?)
        with suppress(AttributeError):
            del self.fileobj

        # Give kernel time to update process status.
        time.sleep(self.delayafterclose)
        if self.isalive():
            if not self.terminate(force=True):
                raise PtyProcessError("Could not terminate the child.")
        self.fd = -1
        self.closed = True
        self.pid = None

eof() -> bool

This returns True if the EOF exception was ever raised.

Returns:

Name	Type	Description
bool	bool	if eof

Source code in transport/plugins/system/ptyprocess.py

def eof(self) -> bool:
    """
    This returns True if the EOF exception was ever raised.

    Args:
        N/A

    Returns:
        bool: if eof

    Raises:
        N/A

    """
    return self.flag_eof

flush() -> None

This does nothing.

It is here to support the interface for a File-like object.

Returns:

Type	Description
None	None

Source code in transport/plugins/system/ptyprocess.py

def flush(self) -> None:
    """
    This does nothing.

    It is here to support the interface for a File-like object.

    Args:
        N/A

    Returns:
        None

    Raises:
        N/A

    """

isalive() -> bool

This tests if the child process is running or not. This is non-blocking. If the child was terminated then this will read the exitstatus or signalstatus of the child. This returns True if the child process appears to be running or False if not. It can take literally SECONDS for Solaris to return the right status.

Returns:

Name	Type	Description
bool	bool	if alive

Raises:

Type	Description
PtyProcessError	an error occurred with the process
PtyProcessError	an error occurred with the process
PtyProcessError	an error occurred with the process

Source code in transport/plugins/system/ptyprocess.py

def isalive(self) -> bool:
    """
    This tests if the child process is running or not. This is
    non-blocking. If the child was terminated then this will read the
    exitstatus or signalstatus of the child. This returns True if the child
    process appears to be running or False if not. It can take literally
    SECONDS for Solaris to return the right status.

    Args:
        N/A

    Returns:
        bool: if alive

    Raises:
        PtyProcessError: an error occurred with the process
        PtyProcessError: an error occurred with the process
        PtyProcessError: an error occurred with the process

    """

    if self.terminated:
        return False

    if self.flag_eof:
        # This is for Linux, which requires the blocking form
        # of waitpid to get the status of a defunct process.
        # This is super-lame. The flag_eof would have been set
        # in read_nonblocking(), so this should be safe.
        waitpid_options = 0
    else:
        waitpid_options = os.WNOHANG

    try:
        pid, status = os.waitpid(self.pid, waitpid_options)
    except OSError as e:
        # No child processes
        if e.errno == errno.ECHILD:
            raise PtyProcessError(
                "isalive() encountered condition "
                + 'where "terminated" is 0, but there was no child '
                + "process. Did someone else call waitpid() "
                + "on our process?"
            )
        raise

    # I have to do this twice for Solaris.
    # I can't even believe that I figured this out...
    # If waitpid() returns 0 it means that no child process
    # wishes to report, and the value of status is undefined.
    if pid == 0:
        try:
            ### os.WNOHANG) # Solaris!
            pid, status = os.waitpid(self.pid, waitpid_options)
        except OSError as e:  # pragma: no cover
            # This should never happen...
            if e.errno == errno.ECHILD:
                raise PtyProcessError(
                    "isalive() encountered condition "
                    + "that should never happen. There was no child "
                    + "process. Did someone else call waitpid() "
                    + "on our process?"
                )
            raise

        # If pid is still 0 after two calls to waitpid() then the process
        # really is alive. This seems to work on all platforms, except for
        # Irix which seems to require a blocking call on waitpid or select,
        # so I let read_nonblocking take care of this situation
        # (unfortunately, this requires waiting through the timeout).
        if pid == 0:
            return True

    if pid == 0:
        return True

    if os.WIFEXITED(status):
        self.status = status
        self.exitstatus = os.WEXITSTATUS(status)
        self.signalstatus = None
        self.terminated = True
    elif os.WIFSIGNALED(status):
        self.status = status
        self.exitstatus = None
        self.signalstatus = os.WTERMSIG(status)
        self.terminated = True
    elif os.WIFSTOPPED(status):
        raise PtyProcessError(
            "isalive() encountered condition "
            + "where child process is stopped. This is not "
            + "supported. Is some other process attempting "
            + "job control with our child pid?"
        )
    return False

kill(sig: int) -> None

Send the given signal to the child application.

In keeping with UNIX tradition it has a misleading name. It does not necessarily kill the child unless you send the right signal. See the :mod:signal module for constants representing signal numbers.

Parameters:

Name	Type	Description	Default
sig	int	signal to send to kill	required

Returns:

Type	Description
None	None

Source code in transport/plugins/system/ptyprocess.py

def kill(self, sig: int) -> None:
    """
    Send the given signal to the child application.

    In keeping with UNIX tradition it has a misleading name. It does not
    necessarily kill the child unless you send the right signal. See the
    :mod:`signal` module for constants representing signal numbers.

    Args:
        sig: signal to send to kill

    Returns:
        None

    Raises:
        N/A

    """

    # Same as os.kill, but the pid is given for you.
    if self.isalive():
        os.kill(self.pid, sig)

read(size: int = 1024) -> bytes

Read and return at most size bytes from the pty.

Can block if there is nothing to read. Raises :exc:EOFError if the terminal was closed.

Unlike Pexpect's read_nonblocking method, this doesn't try to deal with the vagaries of EOF on platforms that do strange things, like IRIX or older Solaris systems. It handles the errno=EIO pattern used on Linux, and the empty-string return used on BSD platforms and (seemingly) on recent Solaris.

Parameters:

Name	Type	Description	Default
size	int	bytes to read	1024

Returns:

Name	Type	Description
bytes	bytes	bytes read

Raises:

Type	Description
EOFError	eof encountered
EOFError	eof encountered

Source code in transport/plugins/system/ptyprocess.py

def read(self, size: int = 1024) -> bytes:
    """
    Read and return at most ``size`` bytes from the pty.

    Can block if there is nothing to read. Raises :exc:`EOFError` if the
    terminal was closed.

    Unlike Pexpect's ``read_nonblocking`` method, this doesn't try to deal
    with the vagaries of EOF on platforms that do strange things, like IRIX
    or older Solaris systems. It handles the errno=EIO pattern used on
    Linux, and the empty-string return used on BSD platforms and (seemingly)
    on recent Solaris.

    Args:
        size: bytes to read

    Returns:
        bytes: bytes read

    Raises:
        EOFError: eof encountered
        EOFError: eof encountered

    """
    try:
        s = self.fileobj.read1(size)
    except (OSError, IOError) as err:
        if err.args[0] == errno.EIO:
            # Linux-style EOF
            self.flag_eof = True
            raise EOFError("End Of File (EOF). Exception style platform.")
        raise
    if s == b"":
        # BSD-style EOF (also appears to work on recent Solaris (OpenIndiana))
        self.flag_eof = True
        raise EOFError("End Of File (EOF). Empty string style platform.")

    return s

setwinsize(rows: int = 24, cols: int = 80) -> None

Set window size.

This will cause a SIGWINCH signal to be sent to the child. This does not change the physical window size. It changes the size reported to TTY-aware applications like vi or curses -- applications that respond to the SIGWINCH signal.

Parameters:

Name	Type	Description	Default
rows	int	int number of rows for terminal	24
cols	int	int number of cols for terminal	80

Returns:

Type	Description
None	None

Source code in transport/plugins/system/ptyprocess.py

def setwinsize(self, rows: int = 24, cols: int = 80) -> None:
    """
    Set window size.

    This will cause a SIGWINCH signal to be sent to the child. This does not change the physical
    window size. It changes the size reported to TTY-aware applications like vi or curses --
    applications that respond to the SIGWINCH signal.

    Args:
        rows: int number of rows for terminal
        cols: int number of cols for terminal

    Returns:
        None

    Raises:
        N/A

    """
    return _setwinsize(self.fd, rows, cols)

spawn(spawn_command: List[str], echo: bool = True, rows: int = 80, cols: int = 256) -> PtyProcessType classmethod

Start the given command in a child process in a pseudo terminal.

This does all the fork/exec type of stuff for a pty, and returns an instance of PtyProcess. For some devices setting terminal width strictly in the operating system (the actual network operating system) does not seem to be sufficient by itself for setting terminal length or width -- so we have optional values for rows/cols that can be passed here as well.

Parameters:

Name	Type	Description	Default
spawn_command	List[str]	command to execute with arguments (if applicable), as a list	required
echo	bool	enable/disable echo -- defaults to True, should be left as True for "normal" scrapli operations, optionally disable for scrapli_netconf operations.	True
rows	int	integer number of rows for ptyprocess "window"	80
cols	int	integer number of cols for ptyprocess "window"	256

Returns:

Name	Type	Description
PtyProcessType	PtyProcessType	instantiated PtyProcess object

Raises:

Type	Description
ScrapliValueError	if no ssh binary found on PATH
Exception	IOError - if unable to set window size of child process
Exception	OSError - if unable to spawn command in child process
IOError	failing to reset window size
exception	if we get an exception decoding output

Source code in transport/plugins/system/ptyprocess.py

@classmethod
def spawn(
    cls: Type[PtyProcessType],
    spawn_command: List[str],
    echo: bool = True,
    rows: int = 80,
    cols: int = 256,
) -> PtyProcessType:
    """
    Start the given command in a child process in a pseudo terminal.

    This does all the fork/exec type of stuff for a pty, and returns an instance of PtyProcess.
    For some devices setting terminal width strictly in the operating system (the actual network
    operating system) does not seem to be sufficient by itself for setting terminal length or
    width -- so we have optional values for rows/cols that can be passed here as well.

    Args:
        spawn_command: command to execute with arguments (if applicable), as a list
        echo: enable/disable echo -- defaults to True, should be left as True for "normal"
            scrapli operations, optionally disable for scrapli_netconf operations.
        rows: integer number of rows for ptyprocess "window"
        cols: integer number of cols for ptyprocess "window"

    Returns:
        PtyProcessType: instantiated PtyProcess object

    Raises:
        ScrapliValueError: if no ssh binary found on PATH
        Exception: IOError - if unable to set window size of child process
        Exception: OSError - if unable to spawn command in child process
        IOError: failing to reset window size
        exception: if we get an exception decoding output

    """
    # Note that it is difficult for this method to fail.
    # You cannot detect if the child process cannot start.
    # So the only way you can tell if the child process started
    # or not is to try to read from the file descriptor. If you get
    # EOF immediately then it means that the child is already dead.
    # That may not necessarily be bad because you may have spawned a child
    # that performs some task; creates no stdout output; and then dies.

    import fcntl
    import pty
    import resource
    import termios
    from pty import CHILD, STDIN_FILENO

    spawn_executable = which(spawn_command[0])
    if spawn_executable is None:
        raise ScrapliValueError("ssh executable not found!")
    spawn_command[0] = spawn_executable

    # [issue #119] To prevent the case where exec fails and the user is
    # stuck interacting with a python child process instead of whatever
    # was expected, we implement the solution from
    # http://stackoverflow.com/a/3703179 to pass the exception to the
    # parent process
    # [issue #119] 1. Before forking, open a pipe in the parent process.
    exec_err_pipe_read, exec_err_pipe_write = os.pipe()

    pid, fd = pty.fork()

    # Some platforms must call setwinsize() and setecho() from the
    # child process, and others from the master process. We do both,
    # allowing IOError for either.
    if pid == CHILD:
        try:
            _setwinsize(fd=STDIN_FILENO, rows=rows, cols=cols)
        except IOError as err:
            if err.args[0] not in (errno.EINVAL, errno.ENOTTY):
                raise

        # disable echo if requested
        if echo is False:
            try:
                _setecho(STDIN_FILENO, False)
            except (IOError, termios.error) as err:
                if err.args[0] not in (errno.EINVAL, errno.ENOTTY):
                    raise

        # [issue #119] 3. The child closes the reading end and sets the
        # close-on-exec flag for the writing end.
        os.close(exec_err_pipe_read)
        fcntl.fcntl(exec_err_pipe_write, fcntl.F_SETFD, fcntl.FD_CLOEXEC)

        # Do not allow child to inherit open file descriptors from parent,
        # with the exception of the exec_err_pipe_write of the pipe.
        # Impose ceiling on max_fd: AIX bugfix for users with unlimited
        # nofiles where resource.RLIMIT_NOFILE is 2^63-1 and os.closerange()
        # occasionally raises out of range error
        max_fd = min(1048576, resource.getrlimit(resource.RLIMIT_NOFILE)[0])
        pass_fds = sorted({exec_err_pipe_write})
        for pair in zip([2] + pass_fds, pass_fds + [max_fd]):
            os.closerange(pair[0] + 1, pair[1])

        try:
            os.execv(spawn_executable, spawn_command)
        except OSError as err:
            # [issue #119] 5. If exec fails, the child writes the error
            # code back to the parent using the pipe, then exits.
            tosend = f"OSError:{err.errno}:{err}".encode()
            os.write(exec_err_pipe_write, tosend)
            os.close(exec_err_pipe_write)
            os._exit(os.EX_OSERR)

    # Parent
    inst = cls(pid, fd)

    # [issue #119] 2. After forking, the parent closes the writing end
    # of the pipe and reads from the reading end.
    os.close(exec_err_pipe_write)
    exec_err_data = os.read(exec_err_pipe_read, 4096)
    os.close(exec_err_pipe_read)

    # [issue #119] 6. The parent reads eof (a zero-length read) if the
    # child successfully performed exec, since close-on-exec made
    # successful exec close the writing end of the pipe. Or, if exec
    # failed, the parent reads the error code and can proceed
    # accordingly. Either way, the parent blocks until the child calls
    # exec.
    if len(exec_err_data) != 0:
        try:
            errclass, errno_s, errmsg = exec_err_data.split(b":", 2)
            exctype = getattr(builtins, errclass.decode("ascii"), Exception)

            exception = exctype(errmsg.decode("utf-8", "replace"))
            if exctype is OSError:
                exception.errno = int(errno_s)
        except Exception:
            raise Exception("Subprocess failed, got bad error data: %r" % exec_err_data)
        else:
            raise exception

    try:
        inst.setwinsize(rows=rows, cols=cols)
    except IOError as err:
        if err.args[0] not in (errno.EINVAL, errno.ENOTTY, errno.ENXIO):
            raise

    return inst

terminate(force: bool = False) -> bool

This forces a child process to terminate.

It starts nicely with SIGHUP and SIGINT. If "force" is True then moves onto SIGKILL. This returns True if the child was terminated. This returns False if the child could not be terminated.

Parameters:

Name	Type	Description	Default
force	bool	bool; force termination	False

Returns:

Name	Type	Description
bool	bool	terminate succeeded or failed

Source code in transport/plugins/system/ptyprocess.py

def terminate(self, force: bool = False) -> bool:
    """
    This forces a child process to terminate.

    It starts nicely with SIGHUP and SIGINT. If "force" is True then moves onto SIGKILL. This
    returns True if the child was terminated. This returns False if the child could not be
    terminated.

    Args:
        force: bool; force termination

    Returns:
        bool: terminate succeeded or failed

    Raises:
        N/A

    """
    if not self.isalive():
        return True
    try:
        self.kill(signal.SIGHUP)
        time.sleep(self.delayafterterminate)
        if not self.isalive():
            return True
        self.kill(signal.SIGCONT)
        time.sleep(self.delayafterterminate)
        if not self.isalive():
            return True
        self.kill(signal.SIGINT)
        time.sleep(self.delayafterterminate)
        if not self.isalive():
            return True
        if force:
            self.kill(signal.SIGKILL)
            time.sleep(self.delayafterterminate)
            if not self.isalive():
                return True
    except OSError:
        # I think there are kernel timing issues that sometimes cause
        # this to happen. I think isalive() reports True, but the
        # process is dead to the kernel.
        # Make one last attempt to see if the kernel is up to date.
        time.sleep(self.delayafterterminate)
        if not self.isalive():
            return True
    return False

write(bytes_to_write: bytes, flush: bool = True) -> int

Write bytes to the pseudoterminal.

Returns the number of bytes written.

Parameters:

Name	Type	Description	Default
bytes_to_write	bytes	bytes to write to the terminal	required
flush	bool	flush the terminal or not	True

Returns:

Name	Type	Description
int	int	number of bytes written

Source code in transport/plugins/system/ptyprocess.py

def write(self, bytes_to_write: bytes, flush: bool = True) -> int:
    """
    Write bytes to the pseudoterminal.

    Returns the number of bytes written.

    Args:
        bytes_to_write: bytes to write to the terminal
        flush: flush the terminal or not

    Returns:
        int: number of bytes written

    Raises:
        N/A

    """
    n = self.fileobj.write(bytes_to_write)
    if flush:
        self.fileobj.flush()
    return n

PtyProcessError

Bases: Exception

Generic error class for this package.

Source code in transport/plugins/system/ptyprocess.py

class PtyProcessError(Exception):
    """Generic error class for this package."""