78 lines
3.5 KiB
Python
78 lines
3.5 KiB
Python
"""Adopt the appointed open TTY; never reopen a device pathname.
|
|
|
|
Installed only by the fixed --archy-owned entry points. FD3 is the open TTY,
|
|
FD4 the retained runtime HOME. Both are supplied by the serial owner after its
|
|
credential drop. This adapter also applies in the frozen esptool descendant.
|
|
"""
|
|
import errno
|
|
import fcntl
|
|
import os
|
|
import stat
|
|
import struct
|
|
import termios
|
|
|
|
|
|
def install(path, fd=3):
|
|
import serial
|
|
import serial.tools.list_ports
|
|
from serial.tools.list_ports_common import ListPortInfo
|
|
|
|
if not stat.S_ISCHR(os.fstat(fd).st_mode) or not os.isatty(fd):
|
|
raise ValueError('Appointed serial FD is not a TTY')
|
|
if fcntl.fcntl(fd, fcntl.F_GETFL) & os.O_ACCMODE != os.O_RDWR:
|
|
raise ValueError('Appointed TTY is not read/write')
|
|
if struct.unpack('i', fcntl.ioctl(fd, 0x80045440, struct.pack('i', 0)))[0] != 1:
|
|
raise ValueError('Appointed TTY is not exclusive')
|
|
original = serial.Serial
|
|
# Disposable PTYs have no modem lines. The production appointer only admits
|
|
# /dev/ttyUSB<n> or ttyACM<n>; this seam does not mask real USB ioctl failures.
|
|
# ttyname() itself can fail when /dev is private and the original node is
|
|
# hidden. Identify UNIX98 PTYs from the kernel device number, not a locator.
|
|
fixture_pty = 136 <= os.major(os.fstat(fd).st_rdev) <= 143
|
|
|
|
class AppointedSerial(original):
|
|
def _update_dtr_state(self):
|
|
try: return super()._update_dtr_state()
|
|
except OSError as error:
|
|
if not fixture_pty or error.errno != errno.ENOTTY: raise
|
|
|
|
def _update_rts_state(self):
|
|
try: return super()._update_rts_state()
|
|
except OSError as error:
|
|
if not fixture_pty or error.errno != errno.ENOTTY: raise
|
|
|
|
def open(self):
|
|
if self._port != path or self.is_open:
|
|
raise serial.SerialException('Only the appointed TTY may be opened')
|
|
# A dup retains the same kernel open file description. A proc-fd
|
|
# pathname open would be a NEW open and is deliberately not used.
|
|
self.fd = os.dup(fd)
|
|
self.pipe_abort_read_r = self.pipe_abort_read_w = None
|
|
self.pipe_abort_write_r = self.pipe_abort_write_w = None
|
|
try:
|
|
self._reconfigure_port(force_update=True)
|
|
try:
|
|
if not self._dsrdtr: self._update_dtr_state()
|
|
if not self._rtscts: self._update_rts_state()
|
|
except OSError as error:
|
|
if error.errno not in (errno.EINVAL, errno.ENOTTY): raise
|
|
self._reset_input_buffer()
|
|
self.pipe_abort_read_r, self.pipe_abort_read_w = os.pipe()
|
|
self.pipe_abort_write_r, self.pipe_abort_write_w = os.pipe()
|
|
os.set_blocking(self.pipe_abort_read_r, False)
|
|
os.set_blocking(self.pipe_abort_write_r, False)
|
|
self.is_open = True
|
|
except BaseException:
|
|
os.close(self.fd); self.fd = None
|
|
for name in ('pipe_abort_read_r', 'pipe_abort_read_w', 'pipe_abort_write_r', 'pipe_abort_write_w'):
|
|
item = getattr(self, name)
|
|
if item is not None: os.close(item)
|
|
setattr(self, name, None)
|
|
raise
|
|
|
|
serial.Serial = AppointedSerial
|
|
port = ListPortInfo(path, skip_link_detection=True)
|
|
port.description = 'Externally appointed mesh radio'
|
|
# Inventory is exactly the granted FD's display label, never host enumeration.
|
|
serial.tools.list_ports.comports = lambda *a, **kw: [port]
|