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 the 6# UNIX Environment. Chapter 19. 7# Author: Steen Lumholt -- with additions by Guido. 8 9from select import select 10import os 11import sys 12import tty 13 14# names imported directly for test mocking purposes 15from os import close, waitpid 16from tty import setraw, tcgetattr, tcsetattr 17 18__all__ = ["openpty", "fork", "spawn"] 19 20STDIN_FILENO = 0 21STDOUT_FILENO = 1 22STDERR_FILENO = 2 23 24CHILD = 0 25 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 pass 34 master_fd, slave_name = _open_terminal() 35 slave_fd = slave_open(slave_name) 36 return master_fd, slave_fd 37 38def master_open(): 39 """master_open() -> (master_fd, slave_name) 40 Open a pty master and return the fd, and the filename of the slave end. 41 Deprecated, use openpty() instead.""" 42 43 import warnings 44 warnings.warn("Use pty.openpty() instead.", DeprecationWarning, stacklevel=2) # Remove API in 3.14 45 46 try: 47 master_fd, slave_fd = os.openpty() 48 except (AttributeError, OSError): 49 pass 50 else: 51 slave_name = os.ttyname(slave_fd) 52 os.close(slave_fd) 53 return master_fd, slave_name 54 55 return _open_terminal() 56 57def _open_terminal(): 58 """Open pty master and return (master_fd, tty_name).""" 59 for x in 'pqrstuvwxyzPQRST': 60 for y in '0123456789abcdef': 61 pty_name = '/dev/pty' + x + y 62 try: 63 fd = os.open(pty_name, os.O_RDWR) 64 except OSError: 65 continue 66 return (fd, '/dev/tty' + x + y) 67 raise OSError('out of pty devices') 68 69def slave_open(tty_name): 70 """slave_open(tty_name) -> slave_fd 71 Open the pty slave and acquire the controlling terminal, returning 72 opened filedescriptor. 73 Deprecated, use openpty() instead.""" 74 75 import warnings 76 warnings.warn("Use pty.openpty() instead.", DeprecationWarning, stacklevel=2) # Remove API in 3.14 77 78 result = os.open(tty_name, os.O_RDWR) 79 try: 80 from fcntl import ioctl, I_PUSH 81 except ImportError: 82 return result 83 try: 84 ioctl(result, I_PUSH, "ptem") 85 ioctl(result, I_PUSH, "ldterm") 86 except OSError: 87 pass 88 return result 89 90def fork(): 91 """fork() -> (pid, master_fd) 92 Fork and make the child a session leader with a controlling terminal.""" 93 94 try: 95 pid, fd = os.forkpty() 96 except (AttributeError, OSError): 97 pass 98 else: 99 if pid == CHILD: 100 try: 101 os.setsid() 102 except OSError: 103 # os.forkpty() already set us session leader 104 pass 105 return pid, fd 106 107 master_fd, slave_fd = openpty() 108 pid = os.fork() 109 if pid == CHILD: 110 os.close(master_fd) 111 os.login_tty(slave_fd) 112 else: 113 os.close(slave_fd) 114 115 # Parent and child process. 116 return pid, master_fd 117 118def _read(fd): 119 """Default read function.""" 120 return os.read(fd, 1024) 121 122def _copy(master_fd, master_read=_read, stdin_read=_read): 123 """Parent copy loop. 124 Copies 125 pty master -> standard output (master_read) 126 standard input -> pty master (stdin_read)""" 127 if os.get_blocking(master_fd): 128 # If we write more than tty/ndisc is willing to buffer, we may block 129 # indefinitely. So we set master_fd to non-blocking temporarily during 130 # the copy operation. 131 os.set_blocking(master_fd, False) 132 try: 133 _copy(master_fd, master_read=master_read, stdin_read=stdin_read) 134 finally: 135 # restore blocking mode for backwards compatibility 136 os.set_blocking(master_fd, True) 137 return 138 high_waterlevel = 4096 139 stdin_avail = master_fd != STDIN_FILENO 140 stdout_avail = master_fd != STDOUT_FILENO 141 i_buf = b'' 142 o_buf = b'' 143 while 1: 144 rfds = [] 145 wfds = [] 146 if stdin_avail and len(i_buf) < high_waterlevel: 147 rfds.append(STDIN_FILENO) 148 if stdout_avail and len(o_buf) < high_waterlevel: 149 rfds.append(master_fd) 150 if stdout_avail and len(o_buf) > 0: 151 wfds.append(STDOUT_FILENO) 152 if len(i_buf) > 0: 153 wfds.append(master_fd) 154 155 rfds, wfds, _xfds = select(rfds, wfds, []) 156 157 if STDOUT_FILENO in wfds: 158 try: 159 n = os.write(STDOUT_FILENO, o_buf) 160 o_buf = o_buf[n:] 161 except OSError: 162 stdout_avail = False 163 164 if master_fd in rfds: 165 # Some OSes signal EOF by returning an empty byte string, 166 # some throw OSErrors. 167 try: 168 data = master_read(master_fd) 169 except OSError: 170 data = b"" 171 if not data: # Reached EOF. 172 return # Assume the child process has exited and is 173 # unreachable, so we clean up. 174 o_buf += data 175 176 if master_fd in wfds: 177 n = os.write(master_fd, i_buf) 178 i_buf = i_buf[n:] 179 180 if stdin_avail and STDIN_FILENO in rfds: 181 data = stdin_read(STDIN_FILENO) 182 if not data: 183 stdin_avail = False 184 else: 185 i_buf += data 186 187def spawn(argv, master_read=_read, stdin_read=_read): 188 """Create a spawned process.""" 189 if isinstance(argv, str): 190 argv = (argv,) 191 sys.audit('pty.spawn', argv) 192 193 pid, master_fd = fork() 194 if pid == CHILD: 195 os.execlp(argv[0], *argv) 196 197 try: 198 mode = tcgetattr(STDIN_FILENO) 199 setraw(STDIN_FILENO) 200 restore = True 201 except tty.error: # This is the same as termios.error 202 restore = False 203 204 try: 205 _copy(master_fd, master_read, stdin_read) 206 finally: 207 if restore: 208 tcsetattr(STDIN_FILENO, tty.TCSAFLUSH, mode) 209 210 close(master_fd) 211 return waitpid(pid, 0)[1] 212