diff options
Diffstat (limited to 'test')
| -rw-r--r-- | test/fs.py | 40 | ||||
| -rw-r--r-- | test/panes.zig | 4 | ||||
| -rw-r--r-- | test/v9fs.py | 181 | ||||
| -rw-r--r-- | test/v9fs_driver_test.py | 26 | ||||
| -rw-r--r-- | test/v9fs_terminal.py | 108 |
5 files changed, 356 insertions, 3 deletions
@@ -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()) |