codekingpro/portable-devtools

1"""Pseudo terminal utilities."""2 3# Bugs: No signal handling.  Doesn't set slave termios and window size.4#       Only tested on Linux, FreeBSD, and macOS.5# See:  W. Richard Stevens. 1992.  Advanced Programming in the6#       UNIX Environment.  Chapter 19.7# Author: Steen Lumholt -- with additions by Guido.8 9from select import select10import os11import sys12import tty13 14# names imported directly for test mocking purposes15from os import close, waitpid16from tty import setraw, tcgetattr, tcsetattr17 18__all__ = ["openpty", "fork", "spawn"]19 20STDIN_FILENO = 021STDOUT_FILENO = 122STDERR_FILENO = 223 24CHILD = 025 26def openpty():27    """openpty() -> (master_fd, slave_fd)28    Open a pty master/slave pair, using os.openpty() if possible."""29 30    try:31        return os.openpty()32    except (AttributeError, OSError):33        pass34    master_fd, slave_name = _open_terminal()35 36    slave_fd = os.open(slave_name, os.O_RDWR)37    try:38        from fcntl import ioctl, I_PUSH39    except ImportError:40         return master_fd, slave_fd41    try:42        ioctl(slave_fd, I_PUSH, "ptem")43        ioctl(slave_fd, I_PUSH, "ldterm")44    except OSError:45        pass46    return master_fd, slave_fd47 48def _open_terminal():49    """Open pty master and return (master_fd, tty_name)."""50    for x in 'pqrstuvwxyzPQRST':51        for y in '0123456789abcdef':52            pty_name = '/dev/pty' + x + y53            try:54                fd = os.open(pty_name, os.O_RDWR)55            except OSError:56                continue57            return (fd, '/dev/tty' + x + y)58    raise OSError('out of pty devices')59 60 61def fork():62    """fork() -> (pid, master_fd)63    Fork and make the child a session leader with a controlling terminal."""64 65    try:66        pid, fd = os.forkpty()67    except (AttributeError, OSError):68        pass69    else:70        if pid == CHILD:71            try:72                os.setsid()73            except OSError:74                # os.forkpty() already set us session leader75                pass76        return pid, fd77 78    master_fd, slave_fd = openpty()79    pid = os.fork()80    if pid == CHILD:81        os.close(master_fd)82        os.login_tty(slave_fd)83    else:84        os.close(slave_fd)85 86    # Parent and child process.87    return pid, master_fd88 89def _read(fd):90    """Default read function."""91    return os.read(fd, 1024)92 93def _copy(master_fd, master_read=_read, stdin_read=_read):94    """Parent copy loop.95    Copies96            pty master -> standard output   (master_read)97            standard input -> pty master    (stdin_read)"""98    if os.get_blocking(master_fd):99        # If we write more than tty/ndisc is willing to buffer, we may block100        # indefinitely. So we set master_fd to non-blocking temporarily during101        # the copy operation.102        os.set_blocking(master_fd, False)103        try:104            _copy(master_fd, master_read=master_read, stdin_read=stdin_read)105        finally:106            # restore blocking mode for backwards compatibility107            os.set_blocking(master_fd, True)108        return109    high_waterlevel = 4096110    stdin_avail = master_fd != STDIN_FILENO111    stdout_avail = master_fd != STDOUT_FILENO112    i_buf = b''113    o_buf = b''114    while 1:115        rfds = []116        wfds = []117        if stdin_avail and len(i_buf) < high_waterlevel:118            rfds.append(STDIN_FILENO)119        if stdout_avail and len(o_buf) < high_waterlevel:120            rfds.append(master_fd)121        if stdout_avail and len(o_buf) > 0:122            wfds.append(STDOUT_FILENO)123        if len(i_buf) > 0:124            wfds.append(master_fd)125 126        rfds, wfds, _xfds = select(rfds, wfds, [])127 128        if STDOUT_FILENO in wfds:129            try:130                n = os.write(STDOUT_FILENO, o_buf)131                o_buf = o_buf[n:]132            except OSError:133                stdout_avail = False134 135        if master_fd in rfds:136            # Some OSes signal EOF by returning an empty byte string,137            # some throw OSErrors.138            try:139                data = master_read(master_fd)140            except OSError:141                data = b""142            if not data:  # Reached EOF.143                return    # Assume the child process has exited and is144                          # unreachable, so we clean up.145            o_buf += data146 147        if master_fd in wfds:148            n = os.write(master_fd, i_buf)149            i_buf = i_buf[n:]150 151        if stdin_avail and STDIN_FILENO in rfds:152            data = stdin_read(STDIN_FILENO)153            if not data:154                stdin_avail = False155            else:156                i_buf += data157 158def spawn(argv, master_read=_read, stdin_read=_read):159    """Create a spawned process."""160    if isinstance(argv, str):161        argv = (argv,)162    sys.audit('pty.spawn', argv)163 164    pid, master_fd = fork()165    if pid == CHILD:166        os.execlp(argv[0], *argv)167 168    try:169        mode = tcgetattr(STDIN_FILENO)170        setraw(STDIN_FILENO)171        restore = True172    except tty.error:    # This is the same as termios.error173        restore = False174 175    try:176        _copy(master_fd, master_read, stdin_read)177    finally:178        if restore:179            tcsetattr(STDIN_FILENO, tty.TCSAFLUSH, mode)180 181    close(master_fd)182    return waitpid(pid, 0)[1]183