1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
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"/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('/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('/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())
|
