#!/usr/bin/env python3 """Line Discipline's oracle: type at a real terminal and record what read() returns. The page's engine says what a canonical-mode terminal does with a run of keystrokes. This asks the kernel instead. For every case it opens a fresh pseudo-terminal, writes the keystroke bytes into the master side, exactly as a terminal emulator does when a key is pressed, and calls read() on the slave side, exactly as a program reading its standard input does. What read() returns, in order, is the transcript. line-discipline-pty.py --fixture write tests/line-discipline-pty.json line-discipline-pty.py --emit the same sweep, as JSON, to stdout line-discipline-pty.py --emit --wide and every sequence of five keys too Knowing when the kernel has finished. A partial line never becomes readable, so "read until nothing is left" would wait forever on exactly the cases that matter. Each case therefore ends with one extra byte, 0x1e, and the slave's VEOL is set to it for the duration: VEOL is POSIX's additional line delimiter, disabled by default, and 0x1e is in no case's alphabet. The last read() of a case is the line that sentinel closed, so it holds whatever was still waiting in the line buffer, and every read() before it is what the keystrokes alone produced. Canonical reads never cross a line, so with a small read size the sentinel's line is the trailing reads that add up to its length, which a 4096-byte run of the same keys measures first. That VEOL change is the only setting altered from what the kernel hands a new pty, apart from VERASE in the second setting, and the fixture records the complete termios state so a reader can see it. """ from __future__ import annotations import argparse import datetime import itertools import json import os import platform import pty import random import re import select import sys import termios from pathlib import Path ROOT = Path(__file__).resolve().parent.parent FIXTURE = ROOT / "tests" / "line-discipline-pty.json" # The most keys the page lets a reader press, read from the page's own controls # so the sweep reaches the longest line a reader can make. MAX_KEYS = int(re.search(r"var MAX_KEYS = (\d+);", (ROOT / "site/js/line-discipline-ui.js").read_text()).group(1)) SENTINEL = 0x1E # What a key sends. Letters, a backslash (POSIX lets an XSI system honour it as # an escape before ERASE, KILL and EOF; Seventh Edition did), DEL and BS (the # two bytes a Backspace key can send), Ctrl-U, Ctrl-D, Enter's carriage return # and a bare line feed. ALPHABET = [0x61, 0x62, 0x5C, 0x7F, 0x08, 0x15, 0x04, 0x0D, 0x0A] # The keys that edit or release a line. Each one is sent after a long line of # letters, so a rule that changed with the length of the line would show. EDITS = [0x7F, 0x08, 0x15, 0x04, 0x0D, 0x0A] LETTERS = [0x61, 0x62, 0x5C] # The two settings: the kernel's own default, and the same terminal after # `stty erase ^H`, which is what a terminal whose Backspace sends BS needs. SETTINGS = [ {"name": "the kernel's default", "erase": None}, {"name": "after stty erase ^H", "erase": 0x08}, ] READ_SIZES = [4096, 1, 2, 3] # The witnesses the roster was admitted on, and the cases the page names. NAMED = [ b"\x04", # Ctrl-D on an empty line b"hi\x04", # Ctrl-D after text b"hi\x04\x04", # Ctrl-D twice b"hi\n\x04", # newline, then Ctrl-D b"hi\r\x04", # Enter, then Ctrl-D b"hellx\x7fo\r", # an erase b"hellx\x08o\r", # the other Backspace byte b"abc\x15hi\r", # a kill b"\x7f\x7f\x15\x04", # erase and kill on nothing, then EOF b"hi\x04\x7f\x7f\r", # an erase cannot reach back past a flush b"a\\\x7f\r", # a backslash does not escape the erase here b"a\\\x04", # nor the EOF b"\x04\x04\x04", # three ends of file in a row b"\x0f\r", # Ctrl-O, which Linux does not implement ] FLAG_NAMES = { "iflag": ["IGNBRK", "BRKINT", "IGNPAR", "PARMRK", "INPCK", "ISTRIP", "INLCR", "IGNCR", "ICRNL", "IUCLC", "IXON", "IXANY", "IXOFF", "IMAXBEL", "IUTF8"], "oflag": ["OPOST", "OLCUC", "ONLCR", "OCRNL", "ONOCR", "ONLRET"], "lflag": ["ISIG", "ICANON", "XCASE", "ECHO", "ECHOE", "ECHOK", "ECHONL", "ECHOCTL", "ECHOPRT", "ECHOKE", "NOFLSH", "TOSTOP", "IEXTEN", "EXTPROC"], } CC_NAMES = ["VINTR", "VQUIT", "VERASE", "VKILL", "VEOF", "VTIME", "VMIN", "VSWTC", "VSTART", "VSTOP", "VSUSP", "VEOL", "VREPRINT", "VDISCARD", "VWERASE", "VLNEXT", "VEOL2"] def _cc(value) -> int: return value[0] if isinstance(value, (bytes, bytearray)) else int(value) def describe_termios(attrs) -> dict: iflag, oflag, cflag, lflag, _, _, cc = attrs out = {"iflag": hex(iflag), "oflag": hex(oflag), "cflag": hex(cflag), "lflag": hex(lflag), "set": {}, "cc": {}} for field, value in (("iflag", iflag), ("oflag", oflag), ("lflag", lflag)): out["set"][field] = [n for n in FLAG_NAMES[field] if hasattr(termios, n) and value & getattr(termios, n)] for n in CC_NAMES: if hasattr(termios, n): out["cc"][n] = _cc(cc[getattr(termios, n)]) return out def run(keys: bytes, setting: dict, size: int) -> list[bytes]: """Every read() the slave returns for these keys, sentinel line last.""" master, slave = pty.openpty() try: attrs = termios.tcgetattr(slave) attrs[6][termios.VEOL] = bytes([SENTINEL]) if setting["erase"] is not None: attrs[6][termios.VERASE] = bytes([setting["erase"]]) termios.tcsetattr(slave, termios.TCSANOW, attrs) os.write(master, keys + bytes([SENTINEL])) reads: list[bytes] = [] while True: ready, _, _ = select.select([slave], [], [], 5.0) if not ready: raise TimeoutError(f"no read() completed for {keys!r}") data = os.read(slave, size) reads.append(data) if data.endswith(bytes([SENTINEL])): return reads if len(reads) > 64: raise RuntimeError(f"runaway transcript for {keys!r}") finally: os.close(master) os.close(slave) def transcript(keys: bytes, setting: dict, size: int) -> tuple[list[bytes], bytes]: """(the reads the keys produced, the bytes still waiting in the line).""" whole = run(keys, setting, 4096) pending = whole[-1][:-1] reads = whole if size == 4096 else run(keys, setting, size) # Walk back over the sentinel's line, which is len(pending) + 1 bytes. need, cut = len(pending) + 1, len(reads) while need > 0: cut -= 1 need -= len(reads[cut]) if need != 0 or b"".join(reads[cut:]) != pending + bytes([SENTINEL]): raise RuntimeError(f"could not separate the sentinel line for {keys!r}") return reads[:cut], pending def encode(keys: bytes, reads: list[bytes], pending: bytes) -> str: """keys:read.read.read|pending, all hex. A zero-byte read is an empty field.""" return f"{keys.hex()}:{'.'.join(r.hex() for r in reads)}{'.' if reads else ''}|{pending.hex()}" def sweep(wide: bool = False) -> list[dict]: rng = random.Random(1979) groups = [] for si, setting in enumerate(SETTINGS): for size in READ_SIZES: longest = 4 if (si == 0 and size == 4096) else 3 seqs = [bytes(c) for n in range(1, longest + 1) for c in itertools.product(ALPHABET, repeat=n)] if wide and si == 0 and size == 4096: seqs += [bytes(c) for c in itertools.product(ALPHABET, repeat=5)] seqs += NAMED # Longer runs, so the sweep is not only short sequences: 150 per # group. A random run seldom builds a long line, because a third of # the alphabet releases or kills it, so most are built on purpose: # a line of 6 to MAX_KEYS - 1 letters, then each edit key. The rest # are a fixed-seed sample of 6 to MAX_KEYS keys. for edit in EDITS: for n in range(6, MAX_KEYS): seqs.append(bytes(rng.choice(LETTERS) for _ in range(n)) + bytes([edit])) for _ in range(150 - len(EDITS) * (MAX_KEYS - 6)): seqs.append(bytes(rng.choice(ALPHABET) for _ in range(rng.randint(6, MAX_KEYS)))) cases = [encode(k, *transcript(k, setting, size)) for k in seqs] groups.append({"setting": si, "read": size, "cases": cases}) return groups def environment() -> dict: master, slave = pty.openpty() try: default = describe_termios(termios.tcgetattr(slave)) finally: os.close(master) os.close(slave) u = platform.uname() return {"kernel": {"sysname": u.system, "release": u.release, "version": u.version, "machine": u.machine}, "python": platform.python_version(), "termios_of_a_new_pty": default} def overlong(typed: int = 5000) -> dict: """Type more than a line can hold, then a newline, and see what is read. Not modelled by the page; measured so its ledger can say what this kernel does with a line too long to keep, rather than quote a constant. The letters cycle a to z, so the record can say which of them were kept.""" pattern = bytes(0x61 + i % 26 for i in range(typed)) master, slave = pty.openpty() try: attrs = termios.tcgetattr(slave) attrs[6][termios.VEOL] = bytes([SENTINEL]) termios.tcsetattr(slave, termios.TCSANOW, attrs) os.set_blocking(master, False) sent = 0 while sent < typed: sent += os.write(master, pattern[sent:sent + min(100, typed - sent)]) try: # drain the echo so the output side never fills while os.read(master, 65536): pass except BlockingIOError: pass os.write(master, b"\n" + bytes([SENTINEL])) reads = [] while True: ready, _, _ = select.select([slave], [], [], 5.0) if not ready: raise TimeoutError("overlong line never completed") data = os.read(slave, 1 << 16) reads.append(data) if data.endswith(bytes([SENTINEL])): break line = reads[0] kept = line[:-1] if line.endswith(b"\n") else line return {"typed": typed, "letters": "a to z, repeating", "then": "0a", "first_read_bytes": len(line), "letters_kept": len(kept), "kept_is_prefix": kept == pattern[:len(kept)], "ends_with_newline": line.endswith(b"\n"), "left_waiting": len(b"".join(reads[1:])) - 1} finally: os.close(master) os.close(slave) def main() -> int: ap = argparse.ArgumentParser() mode = ap.add_mutually_exclusive_group(required=True) mode.add_argument("--fixture", action="store_true") mode.add_argument("--emit", action="store_true") ap.add_argument("--wide", action="store_true") args = ap.parse_args() doc = { "_comment": "Written by scripts/line-discipline-pty.py from a real pty. " "Each case is keys:read.read.|pending in hex; a zero-byte read() " "is an empty field. Do not edit by hand: regenerate.", "captured": datetime.date.today().isoformat(), **environment(), "sentinel": {"byte": SENTINEL, "set_as": "VEOL", "why": "VEOL is disabled on a new pty and 0x1e is in no case, so the " "sentinel only closes the line left waiting at the end"}, "settings": [{"name": s["name"], "VERASE": s["erase"]} for s in SETTINGS], "alphabet": ALPHABET, "groups": sweep(args.wide), "overlong_line": overlong(), } text = json.dumps(doc, indent=1) + "\n" if args.fixture: FIXTURE.write_text(text) n = sum(len(g["cases"]) for g in doc["groups"]) print(f"wrote {FIXTURE.relative_to(ROOT)}: {n} cases on " f"{doc['kernel']['sysname']} {doc['kernel']['release']}") else: sys.stdout.write(text) return 0 if __name__ == "__main__": raise SystemExit(main())