Add Linux Tty9p mounted terminals and forward raw TTY keys

1 files changed, 108 insertions, 0 deletions

diff --git a/test/v9fs_terminal.py b/test/v9fs_terminal.py
new file mode 100644
index 00000000..cc273a40
--- /dev/null
+++ b/test/v9fs_terminal.py
@@ -0,0 +1,108 @@
+#!/usr/bin/env python3
+"""Exercise Tty9p and its real launcher with a nonprivileged sudo stand-in.
+
+This checks PTY routing and lifecycle, not kernel-mount compatibility (v9fs.py).
+"""
+import json
+import os
+from pathlib import Path
+import signal
+import shutil
+import sys
+import tempfile
+import time
+
+from fs import execute, new_pane, session
+
+
+def until(probe, description):
+    deadline = time.monotonic() + 8
+    while time.monotonic() < deadline:
+        result = probe()
+        if result:
+            return result
+        time.sleep(.02)
+    raise AssertionError(description)
+
+
+def test(binary, helper):
+    with tempfile.TemporaryDirectory(prefix='pardes-v9fs-terminal-') as directory:
+        root = Path(directory)
+        # Exercise shell quoting with a real executable path containing both
+        # whitespace and an apostrophe, under bash and fish where available.
+        quoted_helper = root / "helper's quoted path"
+        shutil.copy2(helper, quoted_helper)
+        helper = quoted_helper
+        sudo = root / 'sudo'
+        # A fresh PATH in the owned fixture selects this stand-in. It never
+        # invokes real sudo, mounts anything, or handles a real password.
+        sudo.write_text(f'#!{sys.executable}\n' + '''
+import json, os, signal, sys, termios
+from pathlib import Path
+tty = os.open('/dev/tty', os.O_RDWR)
+assert os.isatty(tty)
+report = dict(argv=sys.argv[1:], pane=os.environ['PARDES_PANE'],
+              socket=os.environ['PARDES_9P'], mount=os.environ['PARDES_MOUNT'],
+              path=os.environ['PARDES_V9FS_PATH'], launcher=os.getppid(), pid=os.getpid())
+parent_status = Path(f"/proc/{report['launcher']}/status").read_text().splitlines()
+shell_pid = next(line.split()[1] for line in parent_status if line.startswith('PPid:'))
+report['shell'] = os.readlink(f'/proc/{shell_pid}/exe')
+Path(os.environ['V9FS_REPORT']).write_text(json.dumps(report))
+before = termios.tcgetattr(tty)
+hidden = termios.tcgetattr(tty)
+hidden[3] &= ~termios.ECHO
+termios.tcsetattr(tty, termios.TCSANOW, hidden)
+os.write(tty, b'V9FS_AUTH_READY: ')
+response = os.read(tty, 100)
+termios.tcsetattr(tty, termios.TCSANOW, before)
+os.write(tty, b'V9FS_INPUT_RECEIVED\\n')
+os.write(tty, b'sudo stand-in: authentication refused\\n')
+sys.exit(1)
+''')
+        sudo.chmod(0o700)
+        shells = [shutil.which('bash') or '/bin/sh', shutil.which('fish') or '/bin/sh']
+        for shell, interrupted in zip(shells, (False, True)):
+            report_path = root / ('interrupted.json' if interrupted else 'normal.json')
+            path = str(root) + ':' + os.environ.get('PATH', '/usr/bin:/bin')
+            name = 'interrupted' if interrupted else 'normal'
+            with session(str(binary), root, name, inherited={
+                'PARDES_V9FS_HELPER': str(helper), 'PATH': path,
+                'V9FS_REPORT': str(report_path), 'SHELL': shell,
+            }) as (client, address):
+                control = new_pane(client, b'')
+                execute(client, control, 'Shell ' + shell)
+                execute(client, control, 'Tty9p')
+                until(report_path.exists, 'Tty9p did not reach the sudo stand-in')
+                report = json.loads(report_path.read_text())
+                pane = f"/self/pane/{report['pane']}"
+                until(lambda: b'V9FS_AUTH_READY' in client.read(pane + '/body'), 'prompt not visible')
+                assert report['socket'] == str(address)
+                assert report['argv'][:2] == ['-E', '--'], report
+                assert report['argv'][2] == str(helper), report
+                assert report['argv'][3] == str(address), report
+                assert report['argv'][4] == report['mount'], report
+                assert report['argv'][5:8] == [str(os.getuid()), str(os.getgid()), '--'], report
+                assert report['path'] == path
+                assert report['shell'] == str(Path(shell).resolve()), report
+                # The detached core remains responsive during authentication.
+                assert client.read('/self/pane/1/body') == b'initial\n'
+                target = Path(report['mount'])
+                assert target.is_dir() and list(target.iterdir()) == []
+                if interrupted:
+                    os.kill(report['launcher'], signal.SIGTERM)
+                else:
+                    client.write(pane + '/pty/data', b'probe-response\r')
+                    until(lambda: b'V9FS_INPUT_RECEIVED' in client.read(pane + '/body'), 'input did not reach new PTY')
+                    assert b'probe-response' not in client.read(pane + '/body'), 'password input was echoed'
+                until(lambda: not target.exists(), 'launcher left its temporary mountpoint')
+                # Failure/cancellation returns to the original interactive
+                # shell, so the pane is useful and its errors remain visible.
+                client.write(pane + '/pty/data', b'printf "OUTER_READY:%s\\n" "$PARDES_PANE"\r')
+                expected = ('OUTER_READY:' + report['pane']).encode()
+                until(lambda: expected in client.read(pane + '/body'), 'original shell did not remain usable')
+                assert client.read('/self/pane/1/body') == b'initial\n'
+    print('Tty9p: shell-first startup, quoted paths, hidden password input, failure recovery and cleanup passed')
+
+
+if __name__ == '__main__':
+    test(Path(sys.argv[1]).resolve(), Path(sys.argv[2]).resolve())