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