test/native_ssh.py
Ref: Size: 11.2 KiB History
#!/usr/bin/env python3
"""Optional Linux native acceptance with an owned loopback OpenSSH server.
Run with release mux and muxg paths. Requires sshd, ssh, and ssh-keygen; all keys,
configuration, daemon state, sockets, and processes belong to this fixture.
The SSH adapter supplies fixture options and records attempts, never outcomes.
"""
import os
from pathlib import Path
import pwd
import shlex
import shutil
import signal
import socket
import subprocess
import sys
import time
sys.dont_write_bytecode = True
from native_lifecycle import LifecycleRig, identity
from native_picker import create, end, sessions
from native_tiling import eventually, require
class SSHFixture:
target = 'mux-native-ssh-fixture'
def __init__(self, rig):
self.rig = rig
self.root = rig.root / 'openssh'
self.root.mkdir(mode=0o700)
self.server = None
self.ssh = shutil.which('ssh')
self.sshd = shutil.which('sshd')
require(self.ssh and self.sshd and shutil.which('ssh-keygen'),
'real OpenSSH client, server, and keygen are required')
self.attempt_log = self.root / 'attempts'
self.attempt_log.touch()
self.sock, _ = rig.daemon('ssh-daemon')
for key in ('host', 'good', 'bad'):
subprocess.run(['ssh-keygen', '-q', '-t', 'ed25519', '-N', '',
'-f', str(self.root / key)], check=True)
(self.root / 'authorized_keys').write_bytes((self.root / 'good.pub').read_bytes())
self.credentials(True)
with socket.socket() as reserve:
reserve.bind(('127.0.0.1', 0))
self.port = reserve.getsockname()[1]
user = pwd.getpwuid(os.getuid()).pw_name
config = self.root / 'sshd_config'
config.write_text(f'''ListenAddress 127.0.0.1
Port {self.port}
HostKey {self.root}/host
PidFile {self.root}/sshd.pid
AuthorizedKeysFile {self.root}/authorized_keys
StrictModes no
PasswordAuthentication no
KbdInteractiveAuthentication no
UsePAM no
AllowUsers {user}
LogLevel VERBOSE
''')
subprocess.run([self.sshd, '-t', '-f', str(config)], check=True)
host_key = (self.root / 'host.pub').read_text().split()
(self.root / 'known_hosts').write_text(
f'[127.0.0.1]:{self.port} {host_key[0]} {host_key[1]}\n')
options = ['-F', '/dev/null', '-p', str(self.port), '-i', str(self.root / 'identity'),
'-o', 'IdentitiesOnly=yes', '-o', 'IdentityAgent=none',
'-o', 'BatchMode=yes', '-o', 'ConnectTimeout=2',
'-o', 'StrictHostKeyChecking=yes',
'-o', 'UserKnownHostsFile=' + str(self.root / 'known_hosts'),
'-o', 'GlobalKnownHostsFile=/dev/null']
self.remote_bin = self.root / 'remote-bin'
self.remote_bin.mkdir()
env = {key: str(rig.root / 'ssh-daemon-state' / key) for key in
('XDG_STATE_HOME', 'XDG_RUNTIME_DIR', 'XDG_CONFIG_HOME', 'XDG_CACHE_HOME')}
env.update(SHELL='/bin/sh', MUX_KEY_FILE=str(self.root / 'absent-quic-key'),
MUX_SOCK='', MUX_SESSION='')
mux_wrapper = self.remote_bin / 'mux'
mux_wrapper.write_text('#!/bin/sh\nexec env ' +
shlex.join([key + '=' + value for key, value in env.items()]) +
' ' + shlex.quote(rig.mux) + ' "$@" --sock ' +
shlex.quote(self.sock) + '\n')
mux_wrapper.chmod(0o700)
adapter_bin = self.root / 'client-bin'
adapter_bin.mkdir()
adapter = adapter_bin / 'ssh'
prefix = 'export PATH=' + shlex.quote(str(self.remote_bin)) + ':"$PATH"; '
adapter.write_text('#!' + sys.executable + '\nimport os,sys,time\n' +
'args=sys.argv[1:]\n' +
'i=args.index(' + repr(self.target) + ')\n' +
'with open(' + repr(str(self.attempt_log)) + ',"a") as f:\n' +
' f.write(str(os.getpid())+" "+str(time.monotonic())+"\\n")\n' +
'command=' + repr(prefix) + '+" ".join(args[i+1:])\n' +
'os.execv(' + repr(self.ssh) + ',' + repr([self.ssh, *options]) +
'+args[:i]+[' + repr(user + '@127.0.0.1') + ',command])\n')
adapter.chmod(0o700)
rig.env['PATH'] = str(adapter_bin) + os.pathsep + rig.env.get('PATH', '')
self.start()
def credentials(self, valid):
shutil.copyfile(self.root / ('good' if valid else 'bad'), self.root / 'identity')
(self.root / 'identity').chmod(0o600)
def attempts(self):
return [int(line.split()[0]) for line in self.attempt_log.read_text().splitlines()]
def start(self):
self.server = self.rig.spawn([self.sshd, '-D', '-e', '-f',
str(self.root / 'sshd_config')],
'sshd-' + str(time.time_ns()))
def ready():
require(self.server.poll() is None, 'fixture sshd exited; inspect sshd log')
with socket.socket() as probe:
return probe.connect_ex(('127.0.0.1', self.port)) == 0
eventually(ready, 'fixture sshd did not listen')
def stop(self):
if self.server is not None and self.server.poll() is None:
self.server.terminate()
try:
self.server.wait(timeout=3)
except subprocess.TimeoutExpired:
self.server.kill()
self.server.wait(timeout=3)
def run(r, ssh, show=lambda label: None):
local, _ = r.daemon('neighbour')
create(local, 'neighbour')
create(ssh.sock, 'remote')
r.catalogue(['--sock ' + local, ssh.target])
r.launch_gui([], 'initial', attached=False)
r.choose('--sock ' + local)
r.picker('sessions')
r.choose('neighbour')
r.wait_state(lambda s: len(s['panes']) == 1 and s['panes'][0]['phase'] == 'attached')
r.host(ssh.target, 'beside')
r.choose('remote')
state = r.wait_state(lambda s: len(s['panes']) == 2 and
all(p['phase'] == 'attached' for p in s['panes']))
local_id, remote_id = [p['id'] for p in state['panes']]
r.mark(remote_id, ssh.sock, 'remote', 'REAL-SSH-SESSION')
before = identity(r.state())
r.quit()
ssh.credentials(False)
count = len(ssh.attempts())
r.launch_gui([], 'restore-auth-refused', attached=False)
state = r.wait_state(lambda s: any(p['id'] == remote_id and p['phase'] == 'dial_failed'
for p in s['panes']))
require(identity(state) == before, 'authentication refusal changed saved pane identity')
r.chord('p')
state = r.wait_state(lambda s: s['recovery'] is not None)
require('Permission denied' in state['recovery']['notice'], 'SSH refusal reason missing from recovery')
r.pixels()
show('Authentication refused: pane retained; explicit Retry available')
time.sleep(1.4)
require(len(ssh.attempts()) == count + 1, 'authentication refusal automatically retried on restore')
r.key('escape')
r.mark(local_id, local, 'neighbour', 'NEIGHBOUR-STILL-USABLE')
require('remote' in sessions(ssh.sock), 'authentication failure ended the remote session')
r.ok('real SSH auth refusal stops restore retries, retains identity/reason, and isolates neighbour')
ssh.credentials(True)
count = len(ssh.attempts())
time.sleep(.5)
require(len(ssh.attempts()) == count, 'correcting credentials resumed without explicit Retry')
r.focus(remote_id)
r.recover('Retry')
r.wait_state(lambda s: all(p['phase'] == 'attached' for p in s['panes']))
r.mark(remote_id, ssh.sock, 'remote', 'EXPLICIT-RETRY-RECOVERED')
show('Corrected key + Retry: existing SSH session resumes')
r.ok('explicit Retry with corrected credentials rejoins the existing real SSH session')
ssh.credentials(False)
count = len(ssh.attempts())
attached_ssh = ssh.attempts()[-1]
require(Path(f'/proc/{attached_ssh}/exe').resolve() == Path(ssh.ssh).resolve(),
'recorded attachment PID is not the actual OpenSSH client')
os.kill(attached_ssh, signal.SIGTERM)
r.wait_state(lambda s: any(p['id'] == remote_id and p['phase'] == 'dial_failed'
for p in s['panes']))
time.sleep(1.4)
require(len(ssh.attempts()) == count + 1, 'authentication refusal automatically retried after disconnect')
r.chord('p')
require('Permission denied' in r.wait_state(lambda s: s['recovery'] is not None)['recovery']['notice'],
'post-attachment auth refusal lost reason')
show('Connection lost, key refused: automatic reconnect stops')
r.key('escape')
ssh.credentials(True)
r.recover('Retry')
r.wait_state(lambda s: all(p['phase'] == 'attached' for p in s['panes']))
r.ok('real SSH authentication refusal also stops retries after a previously attached connection')
r.quit()
ssh.stop()
count = len(ssh.attempts())
r.launch_gui([], 'restore-unavailable', attached=False)
r.wait_state(lambda s: any(p['id'] == remote_id and p['phase'] == 'reconnecting'
for p in s['panes']))
eventually(lambda: len(ssh.attempts()) >= count + 2, 'unavailable SSH host did not retry')
require(len(ssh.attempts()) <= count + 5, 'unavailable host retries are unbounded')
times = [float(line.split()[1]) for line in ssh.attempt_log.read_text().splitlines()[count:]]
require(times[1] - times[0] >= .15, 'unavailable host skipped reconnect backoff')
r.mark(local_id, local, 'neighbour', 'OFFLINE-NEIGHBOUR-USABLE')
show('SSH host unavailable: bounded reconnect; neighbour remains usable')
ssh.start()
r.wait_state(lambda s: all(p['phase'] == 'attached' for p in s['panes']), seconds=10)
r.mark(remote_id, ssh.sock, 'remote', 'HOST-RETURNED-AUTOMATICALLY')
show('SSH host returns: session reconnects automatically')
r.ok('unavailable real SSH host retries and recovers automatically')
r.focus(remote_id)
r.quit()
end(ssh.sock, 'remote')
r.launch_gui([], 'missing-session', attached=False)
r.wait_state(lambda s: any(p['id'] == remote_id and p['phase'] == 'refused'
for p in s['panes']))
require('remote' not in sessions(ssh.sock), 'restore recreated a missing session')
show('Missing remote session stays missing: restore never creates it')
r.quit()
ssh.stop()
r.launch_gui([], 'shutdown-offline', attached=False)
r.wait_state(lambda s: any(p['phase'] == 'reconnecting' for p in s['panes']))
start = time.monotonic()
r.quit()
require(time.monotonic() - start < 2, 'shutdown during SSH retries exceeded two seconds')
r.assert_cli_untouched()
r.ok('SSH restoration stays join-only; shutdown during reconnect is bounded; CLI layout untouched')
def main():
require(len(sys.argv) == 3, 'usage: native_ssh.py RELEASE_MUX RELEASE_MUXG')
r = LifecycleRig(*sys.argv[1:])
ssh = None
try:
ssh = SSHFixture(r)
run(r, ssh)
except BaseException:
r.failure_artifacts()
raise
finally:
try:
if ssh:
ssh.stop()
finally:
r.close()
print('Real SSH native checks passed; owned fixtures stopped.', flush=True)
if __name__ == '__main__':
main()