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
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
|
#!/usr/bin/env python3
import argparse
import hashlib
import json
from pathlib import Path
import shlex
import signal
import sys
import tempfile
import time
import uuid
from fs import session
def screen_text(client):
screen = client.screen()
cols = screen['cols']
return '\n'.join(''.join(cell[0] for cell in screen['cells'][at:at + cols])
for at in range(0, len(screen['cells']), cols))
def expired(signum, frame):
raise TimeoutError('session deadline exceeded')
def wait_ready(client, ready):
previous = None
while True:
text = screen_text(client)
if ready in text and text == previous:
return text
previous = text
time.sleep(.05)
def run(args, command, status):
temporary = tempfile.TemporaryDirectory(prefix='pardes-agent-session-')
try:
root = Path(temporary.name)
config = root / 'config' / 'pardes'
config.mkdir(parents=True)
(config / 'init').write_text('Shell /bin/sh\n')
script = root / 'command.sh'
script.write_text('exec ' + shlex.join(command) + '\n')
old_handler = signal.signal(signal.SIGALRM, expired)
signal.setitimer(signal.ITIMER_REAL, args.timeout)
try:
context = session(str(args.binary.resolve()), root, 'agent-session', tty=True,
socket_name='agent-session', launch=['--tty', '--9p=agent-session'],
inherited={'PARDES_TEST_GRID': '1'} if args.gui_grid else {})
client, _ = context.__enter__()
try:
try:
while not client.read('/pane/1/body').strip():
time.sleep(.01)
started = time.monotonic()
client.write('/pane/1/pty/data', shlex.join(['/bin/sh', str(script)]).encode() + b'\r')
status['phase'] = 'readiness'
wait_ready(client, args.ready)
if args.prepare_key is not None:
status['phase'] = 'preparation'
client.write('/pane/1/pty/data', args.prepare_key)
wait_ready(client, args.ready)
ready_ms = (time.monotonic() - started) * 1000
status.update(phase='history', required_body_rows=args.min_rows)
while True:
started = time.monotonic()
body = client.read('/pane/1/body')
body_read_ms = (time.monotonic() - started) * 1000
rows = body.count(b'\n') + int(bool(body) and not body.endswith(b'\n'))
status.update(body_bytes=len(body), body_rows=rows)
if rows >= args.min_rows:
break
time.sleep(.05)
status['phase'] = 'settling'
before = wait_ready(client, args.ready)
status['expected_visible_text_sha256'] = hashlib.sha256(before.encode()).hexdigest()
probe = 'p' + uuid.uuid4().hex[:12]
if probe in before:
raise AssertionError('probe is already visible')
status['phase'] = 'probe-visible'
started = time.monotonic()
client.write('/pane/1/pty/data', probe.encode())
while probe not in screen_text(client).replace('\n', ''):
time.sleep(.01)
write_ms = (time.monotonic() - started) * 1000
status['phase'] = 'probe-cleared'
started = time.monotonic()
client.write('/pane/1/pty/data', b'\x15')
while True:
observed = screen_text(client)
status['observed_visible_text_sha256'] = hashlib.sha256(observed.encode()).hexdigest()
if observed == before:
break
time.sleep(.01)
status['metrics'] = {
'ready_ms': ready_ms, 'body_read_ms': body_read_ms,
'ninep_write_to_observation_ms': write_ms,
'ninep_clear_to_observation_ms': (time.monotonic() - started) * 1000,
'visible_text_sha256': hashlib.sha256(before.encode()).hexdigest(),
'body_bytes': len(body), 'body_rows': rows,
'body_sha256': hashlib.sha256(body).hexdigest(),
}
finally:
previous_phase = status['phase']
status['phase'] = 'interrupt-cleanup'
signal.setitimer(signal.ITIMER_REAL, 0)
client.socket.settimeout(.25)
for _ in range(2):
try:
client.write('/pane/1/pty/data', b'\x03')
except (OSError, EOFError):
pass
time.sleep(.05)
status['phase'] = previous_phase
finally:
previous_phase = status['phase']
status['phase'] = 'session-cleanup'
context.__exit__(*sys.exc_info())
status['phase'] = previous_phase
finally:
signal.setitimer(signal.ITIMER_REAL, 0)
signal.signal(signal.SIGALRM, old_handler)
finally:
previous_phase = status['phase']
status['phase'] = 'files-cleanup'
temporary.cleanup()
status['phase'] = previous_phase
return status.pop('metrics')
def main(argv):
parser = argparse.ArgumentParser(description='Check an unsubmitted input probe through an owned Pardes 9P session.')
parser.add_argument('binary', type=Path)
parser.add_argument('--ready', required=True, help='visible application readiness text')
parser.add_argument('--timeout', type=float, default=60, help='interaction deadline in seconds, at most 600')
parser.add_argument('--min-rows', type=int, default=0)
parser.add_argument('--prepare-key', help='one hex control byte before history/probe checks; never CR or LF')
parser.add_argument('--gui-grid', action='store_true', help='use the GUI build headless grid host')
if '--' not in argv:
parser.error('supply the interactive command after --')
split = argv.index('--')
args = parser.parse_args(argv[:split])
command = argv[split + 1:]
if not command or not command[0] or not args.ready.strip():
parser.error('a command and nonempty --ready text are required')
if not 0 < args.timeout <= 600 or args.min_rows < 0:
parser.error('--timeout must be in (0, 600] and --min-rows must be nonnegative')
if args.prepare_key is not None:
try:
args.prepare_key = bytes.fromhex(args.prepare_key)
except ValueError:
parser.error('--prepare-key must be one hex control byte')
if len(args.prepare_key) != 1 or not 0 < args.prepare_key[0] < 32 or args.prepare_key[0] in (10, 13):
parser.error('--prepare-key must be one control byte other than CR or LF')
status = {'phase': 'startup'}
try:
metrics = run(args, command, status)
except (OSError, EOFError, AssertionError, ValueError) as error:
status['error'] = type(error).__name__
if isinstance(error, OSError):
status['errno'] = error.errno
print(json.dumps(status), file=sys.stderr)
return 1
print(json.dumps(metrics))
return 0
if __name__ == '__main__':
sys.exit(main(sys.argv[1:]))
|
