Add Linux Tty9p mounted terminals and forward raw TTY keys

5 files changed, 356 insertions, 3 deletions

diff --git a/test/fs.py b/test/fs.py
index 85cf6855..2ffd5098 100644
--- a/test/fs.py
+++ b/test/fs.py
@@ -122,6 +122,41 @@ def execute(client, serial, command):
     client.write(f'/self/pane/{serial}/event', f'MX0 {len(text)}\n'.encode())
 
 
+def discovery(binary):
+    with tempfile.TemporaryDirectory(prefix='pardes-discovery-') as directory:
+        with session(binary, Path(directory), 'discovery') as (client, _):
+            before = client.read('/self/index')
+            entries = client.list('/self/new')
+            assert {'README', 'ctl', 'body', 'addr', 'data', 'event'} <= set(entries)
+            guide = client.read('/self/README')
+            assert guide == client.read('/self/new/README')
+            assert b'Each open creates a separate pane' in guide
+            # Walk and stat each entry, as a filesystem browser does, without open.
+            from ninep import string
+            for name in entries:
+                client.fid += 1
+                fid = client.fid
+                client.rpc(110, struct.pack('<IIH', 1, fid, 3) +
+                           string('self') + string('new') + string(name))
+                try:
+                    client.rpc(124, struct.pack('<I', fid))
+                finally:
+                    client.close(fid)
+            assert client.read('/self/index') == before, 'browsing created a pane'
+            first = int(client.read('/self/new/ctl').split()[0])
+            second = int(client.read('/self/new/ctl').split()[0])
+            assert first != second
+            client.write(f'/self/pane/{first}/body', b'first pane', truncate=True)
+            assert client.read(f'/self/pane/{first}/body') == b'first pane'
+            assert client.read(f'/self/pane/{second}/body') == b''
+            before_ids = set(client.list('/self/pane'))
+            client.write('/self/new/body', b'created with text', truncate=True)
+            made = set(client.list('/self/pane')) - before_ids
+            assert len(made) == 1
+            assert client.read('/self/pane/' + made.pop() + '/body') == b'created with text'
+    print('9P discovery: listing/stat/README are inert; opens create independent panes')
+
+
 def test(binary, quic=False):
     started = time.monotonic()
     for options, message in [
@@ -549,6 +584,9 @@ def quic_test(binary):
 
 
 if __name__ == '__main__':
+    if len(sys.argv) == 3 and sys.argv[2] == '--discovery':
+        discovery(str(Path(sys.argv[1]).resolve()))
+        raise SystemExit(0)
     if len(sys.argv) not in [2, 3] or (len(sys.argv) == 3 and sys.argv[2] != '--quic'):
-        raise SystemExit('usage: fs.py <pardes> [--quic]')
+        raise SystemExit('usage: fs.py <pardes> [--quic | --discovery]')
     test(str(Path(sys.argv[1]).resolve()), quic=len(sys.argv) == 3)
diff --git a/test/panes.zig b/test/panes.zig
index 2d57ecd8..a324d74a 100644
--- a/test/panes.zig
+++ b/test/panes.zig
@@ -1701,9 +1701,9 @@ const TerminalTests = struct {
             .{ .key = .{ .cp = '@', .text = "@", .ctrl = true }, .expected = "\x00" },
             .{ .key = .{ .cp = '_', .text = "_", .ctrl = true }, .expected = "\x1f" },
             .{ .key = .{ .cp = '1', .text = "1", .ctrl = true }, .expected = "1" },
-            .{ .key = .{ .cp = Key.up, .alt = true, .shift = true }, .expected = "\x1b[A" },
+            .{ .key = .{ .cp = Key.up, .alt = true, .shift = true }, .expected = "\x1b[1;4A" },
             .{ .key = .{ .cp = Key.delete }, .expected = "\x1b[3~" },
-            .{ .key = .{ .cp = Key.home }, .expected = null },
+            .{ .key = .{ .cp = Key.home }, .expected = "\x1b[H" },
         };
         for (cases) |case| {
             forwardKey(p, 0, case.key);
diff --git a/test/v9fs.py b/test/v9fs.py
new file mode 100644
index 00000000..7dc88195
--- /dev/null
+++ b/test/v9fs.py
@@ -0,0 +1,181 @@
+#!/usr/bin/env python3
+"""Opt-in Linux kernel-mount probe; the editor always runs unprivileged.
+
+Run after `sudo -v` with a native Pardes binary and the runtime v9fs helper.
+The helper alone runs through sudo, creates a private mount namespace, mounts
+9P, drops privileges, and execs this file's worker. No FUSE or global mount.
+"""
+import argparse
+import json
+import os
+import pwd
+from pathlib import Path
+import subprocess
+import sys
+import tempfile
+import time
+import traceback
+
+from fs import session
+from ninep import Client
+
+
+def write_existing(path, data, *, truncate=False):
+    flags = os.O_WRONLY | (os.O_TRUNC if truncate else 0)
+    fd = os.open(path, flags)
+    try:
+        assert os.write(fd, data) == len(data), str(path)
+    finally:
+        os.close(fd)
+
+
+def worker(mountpoint, socket, uid, gid, original_namespace):
+    assert os.getresuid() == (uid, uid, uid), os.getresuid()
+    assert os.getresgid() == (gid, gid, gid), os.getresgid()
+    assert set(os.getgroups()) == set(os.getgrouplist(pwd.getpwuid(uid).pw_name, gid)), os.getgroups()
+    assert os.readlink('/proc/self/ns/mnt') != original_namespace
+    status = dict(line.split(':', 1) for line in Path('/proc/self/status').read_text().splitlines())
+    for field in ('CapEff', 'CapPrm', 'CapAmb'):
+        assert int(status[field].strip(), 16) == 0, (field, status[field])
+    entries = Path('/proc/self/mountinfo').read_text().splitlines()
+    mounted = [line for line in entries if line.split()[4] == str(mountpoint)]
+    assert len(mounted) == 1 and ' - 9p ' in mounted[0], mounted
+    assert not any(field.startswith(('shared:', 'master:')) for field in mounted[0].split()[6:])
+
+    assert set(os.listdir(mountpoint)) == {'os', 'self'}
+    tree = mountpoint / 'self'
+    assert {'index', 'pane', 'new', 'screen'} <= set(os.listdir(tree))
+    # A direct connection provides independent evidence for VFS reads/writes.
+    with Client(socket) as client:
+        assert (tree / 'index').read_bytes() == client.read('/self/index')
+        assert (tree / 'pane/1/body').read_bytes() == b'initial\n'
+
+        before = client.read('/self/index')
+        subprocess.run(['ls', '-l', str(tree / 'new')], check=True, capture_output=True, timeout=5)
+        assert {'README', 'ctl', 'body'} <= set(os.listdir(tree / 'new'))
+        assert (tree / 'new/README').read_bytes() == (tree / 'README').read_bytes()
+        assert client.read('/self/index') == before, 'browsing created panes'
+        serial = int((tree / 'new/ctl').read_bytes().split()[0])
+        another = int((tree / 'new/ctl').read_bytes().split()[0])
+        assert serial != another, 'cached factory reused a pane'
+        client.write(f'/self/pane/{another}/ctl', b'delete\n')
+        pane = tree / 'pane' / str(serial)
+        wire = f'/self/pane/{serial}'
+        assert str(serial) in os.listdir(tree / 'pane')
+        assert serial in [int(row.split()[0]) for row in (tree / 'index').read_bytes().splitlines()]
+
+        write_existing(pane / 'body', b'kernel body\n', truncate=True)
+        assert client.read(wire + '/body') == b'kernel body\n'
+        # Reopen must observe changes made through another 9P connection.
+        client.write(wire + '/body', b'wire update\n', truncate=True)
+        assert (pane / 'body').read_bytes() == b'wire update\n'
+        write_existing(pane / 'body', b'appended\n')
+        assert client.read(wire + '/body') == b'wire update\nappended\n'
+        write_existing(pane / 'addr', b'#0,#4')
+        write_existing(pane / 'data', b'v9fs')
+        assert client.read(wire + '/body') == b'v9fs update\nappended\n'
+
+        # Exercise an actual shell redirection, including its O_TRUNC open.
+        subprocess.run(['/bin/sh', '-c', 'printf "name kernel-probe\\n" > "$1/ctl"',
+                        'v9fs-probe', str(pane)], check=True, timeout=5)
+        tag = client.read(wire + '/tag')
+        assert tag.split(maxsplit=1)[0] == str(socket.parent / 'kernel-probe').encode(), tag
+        # Children inherit this single mount and can use ordinary tools.
+        copied = subprocess.run(['/bin/cat', str(pane / 'body')],
+                                check=True, capture_output=True, timeout=5).stdout
+        assert copied == b'v9fs update\nappended\n'
+
+        command = b'Msg kernel-v9fs-ready'
+        write_existing(pane / 'body', command, truncate=True)
+        write_existing(pane / 'event', f'MX0 {len(command)}\n'.encode())
+        # Event writes acknowledge dispatch, so reopen screen until rendered.
+        deadline = time.monotonic() + 5
+        while True:
+            screen = json.loads((tree / 'screen').read_bytes())
+            if 'kernel-v9fs-ready' in ''.join(cell[0] for cell in screen['cells']):
+                break
+            assert time.monotonic() < deadline, 'Exec result was not rendered'
+            time.sleep(.01)
+
+        # Reading OS files through the exported tree does not recurse through
+        # the mount: the core still lives in the supervisor's namespace.
+        assert (mountpoint / 'os' / str(socket.parent).lstrip('/') / 'kernel.txt').read_bytes() == b'initial\n'
+        write_existing(pane / 'ctl', b'delete\n')
+        assert serial not in [int(row.split()[0]) for row in (tree / 'index').read_bytes().splitlines()]
+
+    print('v9fs: namespace isolation, privilege drop, inherited mount, directory refresh, '
+          'text edits, control writes, Exec and screen checks passed', flush=True)
+
+
+def run_helper(command, timeout=30):
+    # Keep the caller's controlling terminal: sudo credentials are commonly
+    # scoped to it. setsid/start_new_session makes an earlier sudo -v useless.
+    # The elevated helper itself creates the separate *mount* namespace.
+    child = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
+                             text=True)
+    try:
+        stdout, stderr = child.communicate(timeout=timeout)
+    except subprocess.TimeoutExpired:
+        # sudo forwards signals to its command. Keep the server alive while
+        # stopping the mount user, then let session() clean up.
+        child.terminate()
+        try:
+            child.communicate(timeout=5)
+        except subprocess.TimeoutExpired:
+            child.kill()
+            child.communicate(timeout=5)
+        raise RuntimeError('kernel probe timed out; no compatibility result')
+    if child.returncode:
+        raise RuntimeError(f'kernel probe failed ({child.returncode})\n{stdout}{stderr}')
+    return stdout
+
+
+def run(binary, helper):
+    if sys.platform != 'linux':
+        raise RuntimeError('this probe requires Linux v9fs')
+    if os.getuid() == 0:
+        raise RuntimeError('run the driver as your normal user; only the mount helper uses sudo')
+    # Never prompt from a build step. An unavailable prerequisite is a failure,
+    # not a skipped test that might be mistaken for mounted-filesystem coverage.
+    available = subprocess.run(['sudo', '-n', '-v'], capture_output=True, text=True, timeout=5)
+    if available.returncode:
+        raise RuntimeError('mount authorization unavailable: run sudo -v in your terminal, then retry\n'
+                           + available.stderr.strip())
+    namespace = os.readlink('/proc/self/ns/mnt')
+    with tempfile.TemporaryDirectory(prefix='pardes-v9fs-') as directory:
+        root = Path(directory)
+        target = root / 'mount'
+        target.mkdir()
+        with session(str(binary), root, 'kernel') as (client, address):
+            command = ['sudo', '-n', '--', str(helper), str(address), str(target),
+                       str(os.getuid()), str(os.getgid()), '--', sys.executable, '-B',
+                       str(Path(__file__).resolve()), '--worker', str(target), str(address),
+                       str(os.getuid()), str(os.getgid()), namespace]
+            print(run_helper(command), end='')
+            assert os.readlink('/proc/self/ns/mnt') == namespace
+            assert list(target.iterdir()) == [], 'mount escaped its private namespace'
+            assert client.read('/self/pane/1/body') == b'initial\n', 'core stopped serving after probe exit'
+    print('v9fs: supervisor namespace unchanged and session cleanup passed')
+
+
+def main():
+    if len(sys.argv) > 1 and sys.argv[1] == '--worker':
+        if len(sys.argv) != 7:
+            raise RuntimeError('invalid internal worker arguments')
+        worker(Path(sys.argv[2]), Path(sys.argv[3]), int(sys.argv[4]), int(sys.argv[5]), sys.argv[6])
+        return
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument('binary', type=lambda text: Path(text).resolve(strict=True))
+    parser.add_argument('helper', type=lambda text: Path(text).resolve(strict=True))
+    args = parser.parse_args()
+    run(args.binary, args.helper)
+
+
+if __name__ == '__main__':
+    try:
+        main()
+    except (AssertionError, OSError, RuntimeError, subprocess.SubprocessError) as error:
+        if len(sys.argv) > 1 and sys.argv[1] == '--worker':
+            traceback.print_exc()
+        print(f'v9fs: {error}', file=sys.stderr)
+        sys.exit(1)
diff --git a/test/v9fs_driver_test.py b/test/v9fs_driver_test.py
new file mode 100644
index 00000000..9781c4fe
--- /dev/null
+++ b/test/v9fs_driver_test.py
@@ -0,0 +1,26 @@
+#!/usr/bin/env python3
+"""Unprivileged regression checks for the v9fs helper launcher."""
+import os
+import sys
+import unittest
+
+from v9fs import run_helper
+
+
+class HelperLaunchTests(unittest.TestCase):
+    def test_preserves_session_for_terminal_scoped_sudo_credentials(self):
+        output = run_helper([sys.executable, '-c', 'import os; print(os.getsid(0))'])
+        self.assertEqual(int(output), os.getsid(0))
+
+    def test_failure_preserves_diagnostics(self):
+        with self.assertRaisesRegex(RuntimeError, r'(?s)failed \(7\).*mount refused'):
+            run_helper([sys.executable, '-c',
+                        'import sys; print("mount refused", file=sys.stderr); sys.exit(7)'])
+
+    def test_timeout_is_not_a_passing_probe(self):
+        with self.assertRaisesRegex(RuntimeError, 'timed out; no compatibility result'):
+            run_helper([sys.executable, '-c', 'import time; time.sleep(60)'], timeout=.1)
+
+
+if __name__ == '__main__':
+    unittest.main()
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())