1#!/usr/bin/env python3
2"""Disposable dual-transport fixture and physical Wayland input acceptance."""
3import importlib.util
4import json
5import os
6from pathlib import Path
7import shutil
8import signal
9import subprocess
10import tempfile
11import time
12import uuid
13
14ROOT = Path(__file__).resolve().parents[2]
15
16
17def run(*args, **kwargs):
18 return subprocess.run(args, check=True, timeout=kwargs.pop("timeout", 60), **kwargs)
19
20
21def docker(*args, **kwargs):
22 return run("docker", *args, **kwargs)
23
24
25def main():
26 name = "vev-resilience-" + uuid.uuid4().hex[:10]
27 artifacts = Path(tempfile.mkdtemp(prefix="vev-resilience-artifacts-"))
28 processes = []
29 containers = []
30 network = False
31 def interrupt(signum, frame):
32 raise KeyboardInterrupt("fixture interrupted")
33 signal.signal(signal.SIGTERM, interrupt)
34 with tempfile.TemporaryDirectory(prefix="vev-resilience-private-") as directory:
35 private = Path(directory)
36 runtime = private / "runtime"
37 runtime.mkdir(mode=0o700)
38 try:
39 for tool in ("docker", "foot", "ssh-keygen", "go"):
40 if shutil.which(tool) is None:
41 raise RuntimeError("missing prerequisite: " + tool)
42 run("go", "build", "-o", str(ROOT / "build/vev"), ".", cwd=ROOT, timeout=600)
43 for helper in ("relay", "pixels"):
44 run("go", "build", "-o", str(ROOT / f"build/resilience-{helper}"), f"./scripts/remote-resilience-harness/{helper}", cwd=ROOT, timeout=600)
45 nefer = os.environ.get("NEFERWL_BIN")
46 if not nefer:
47 nefer = str(private / "neferwl")
48 common = run("git", "rev-parse", "--path-format=absolute", "--git-common-dir", cwd=ROOT, capture_output=True, text=True).stdout.strip()
49 source = Path(os.environ.get("NEFERWL_SOURCE", str(Path(common).parent.parent / "neferwl")))
50 run("go", "build", "-o", nefer, "./cmd/neferwl", cwd=source, env={**os.environ, "CGO_ENABLED": "0"}, timeout=600)
51 docker("build", "-f", str(ROOT / "scripts/demo/Dockerfile"), "-t", name, str(ROOT), timeout=600)
52 docker("network", "create", name)
53 network = True
54 for role in ("client", "remote"):
55 container = name + "-" + role
56 docker("run", "-d", "--name", container, "--network", name, "--network-alias", "upstream" if role == "remote" else "remote", "--entrypoint", "sleep", name, "infinity")
57 containers.append(container)
58 client, remote = containers
59 for key in ("client", "host"):
60 run("ssh-keygen", "-q", "-t", "ed25519", "-N", "", "-C", "fixture", "-f", str(private / key))
61 docker("cp", str(private / "host"), remote + ":/etc/ssh/ssh_host_ed25519_key")
62 docker("cp", str(private / "host.pub"), remote + ":/etc/ssh/ssh_host_ed25519_key.pub")
63 docker("cp", str(private / "client.pub"), remote + ":/home/demo/.ssh/authorized_keys")
64 docker("exec", "--user", "root", remote, "sh", "-c", "rm -f /home/demo/.ssh/id_ed25519*; chmod 600 /etc/ssh/ssh_host_ed25519_key /home/demo/.ssh/authorized_keys; chown demo:demo /home/demo/.ssh/authorized_keys; mkdir -p /run/sshd; /usr/sbin/sshd")
65 docker("exec", "--user", "root", client, "sh", "-c", "rm -f /home/demo/.ssh/id_ed25519*")
66 docker("cp", str(private / "client"), client + ":/home/demo/.ssh/id_ed25519")
67 public = (private / "host.pub").read_text().split()
68 (private / "known_hosts").write_text("remote " + " ".join(public[:2]) + "\n")
69 (private / "config").write_text("Host remote\n HostName remote\n Port 2222\n User demo\n IdentityFile ~/.ssh/id_ed25519\n IdentitiesOnly yes\n StrictHostKeyChecking yes\n HostKeyAlias remote\n UserKnownHostsFile ~/.ssh/known_hosts\n")
70 for file in ("known_hosts", "config"):
71 docker("cp", str(private / file), client + ":/home/demo/.ssh/" + file)
72 docker("cp", str(ROOT / "build/resilience-relay"), client + ":/tmp/relay")
73 docker("cp", str(ROOT / "scripts/remote-resilience-harness/ssh.py"), client + ":/tmp/ssh")
74 docker("exec", "--user", "root", client, "sh", "-c", "chown demo:demo /home/demo/.ssh/*; chmod 600 /home/demo/.ssh/*; chmod 755 /tmp/ssh; install -d -m 700 -o demo -g demo /home/demo/relay; mkdir -p /tmp/bin; ln -s /tmp/ssh /tmp/bin/ssh; mv /usr/bin/ssh /usr/bin/ssh-real; ln -s /tmp/ssh /usr/bin/ssh")
75 transport = os.environ.get("VEV_RESILIENCE_TRANSPORT", "quic")
76 if transport not in ("quic", "ssh"):
77 raise ValueError("VEV_RESILIENCE_TRANSPORT must be quic or ssh")
78 relay_args = ["-degraded"] if os.environ.get("VEV_RESILIENCE_DEGRADED") == "1" else []
79 relay = subprocess.Popen(["docker", "exec", client, "/tmp/relay", "-remote", "upstream", "-control", "/home/demo/relay/control.sock", "-tcp", ":2222", "-listen", "0.0.0.0", *relay_args])
80 processes.append(relay)
81 control_id = 0
82 def control(op, **fields):
83 nonlocal control_id
84 control_id += 1
85 request = {"id": control_id, "op": op, **fields}
86 code = "import socket,json; s=socket.socket(socket.AF_UNIX); s.settimeout(3); s.connect('/home/demo/relay/control.sock'); s.sendall(" + repr(json.dumps(request).encode() + b"\n") + "); print(s.makefile().readline())"
87 result = docker("exec", client, "python3", "-c", code, capture_output=True, text=True)
88 reply = json.loads(result.stdout)
89 assert reply["id"] == control_id and "error" not in reply, reply
90 return reply["result"]
91 # Fixture image must explicitly supply Python rather than using host tools.
92 docker("exec", "--user", "root", client, "sh", "-c", "command -v python3", capture_output=True)
93 deadline = time.monotonic() + 5
94 while True:
95 ready = subprocess.run(["docker", "exec", client, "test", "-S", "/home/demo/relay/control.sock"], capture_output=True, timeout=3)
96 if ready.returncode == 0:
97 break
98 if time.monotonic() > deadline:
99 raise RuntimeError("relay control socket did not become ready")
100 time.sleep(0.05)
101 probe = subprocess.run(["docker", "exec", client, "/usr/bin/ssh", "-v", "remote", "true"], capture_output=True, text=True, timeout=10)
102 if probe.returncode:
103 (artifacts / "ssh-probe.log").write_text(probe.stderr)
104 raise RuntimeError("fixture SSH authentication failed")
105 env_args = ["-e", "PATH=/tmp/bin:/usr/local/bin:/usr/bin", "-e", "VEV_LOG=debug", "-e", "SSH_AUTH_SOCK="]
106 spec = importlib.util.spec_from_file_location("acceptance", ROOT / "scripts/remote-picker-harness/acceptance.py")
107 acceptance = importlib.util.module_from_spec(spec)
108 spec.loader.exec_module(acceptance)
109 fixture = acceptance.Driver(remote, ["--session", "resilience"])
110 fixture.close()
111 docker("exec", *env_args, client, "vev", "host", "add", "--transport", transport, "remote")
112 # Wait for the broker publication before resolving direct attach.
113 deadline = time.monotonic() + 30
114 while True:
115 inventory = docker("exec", *env_args, client, "vev", "ls", "remote", capture_output=True, text=True)
116 if "resilience" in inventory.stdout:
117 break
118 (artifacts / "inventory.txt").write_text(inventory.stdout + inventory.stderr)
119 if time.monotonic() > deadline:
120 raise RuntimeError("remote fixture was not published")
121 time.sleep(0.2)
122 shots = artifacts / "shots"
123 shots.mkdir()
124 docker_endpoint = run("docker", "context", "inspect", "--format", "{{.Endpoints.docker.Host}}", capture_output=True, text=True).stdout.strip()
125 env = {**os.environ, "DOCKER_HOST": docker_endpoint, "XDG_RUNTIME_DIR": str(runtime), "XDG_CONFIG_HOME": str(private / "config-home"), "XDG_STATE_HOME": str(private / "state"), "HOME": str(private)}
126 env["TERMINAL"] = "foot --config=/dev/null --override=cursor.blink=no --log-level=info --hold -- docker --host=" + docker_endpoint + " exec -it " + " ".join(env_args) + " " + client + " vev attach demo@remote:resilience --ui-observe --ui-socket /home/demo/visual/ui.sock"
127 log = (artifacts / "compositor.log").open("w")
128 compositor = subprocess.Popen([nefer, "--backend=headless", "--size=1280x720", "--input=-", "--screenshot", str(shots), "--timeout", "180s"], stdin=subprocess.PIPE, stdout=log, stderr=log, text=True, env=env)
129 processes.append(compositor)
130 deadline = time.monotonic() + 15
131 while not list(runtime.glob("wayland-*")):
132 if compositor.poll() is not None or time.monotonic() > deadline:
133 raise RuntimeError("headless compositor did not create socket")
134 time.sleep(0.05)
135 sockets = [p for p in runtime.glob("wayland-*") if not p.name.endswith(".lock")]
136 env["WAYLAND_DISPLAY"] = str(sockets[0])
137 env.pop("WAYLAND_SOCKET", None)
138 env.pop("DISPLAY", None)
139 def capture():
140 bridge = subprocess.Popen(["docker", "exec", "-i", client, "vev", "--ui-driver", "--socket", "/home/demo/visual/ui.sock"], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
141 try:
142 import select
143 if not select.select([bridge.stdout], [], [], 1)[0]:
144 raise RuntimeError("observation discovery timed out")
145 discovery = json.loads(bridge.stdout.readline())
146 request = {"version": 1, "id": 1, "op": "capture", "attachment": discovery["result"]["attachment"]}
147 output, error = bridge.communicate(json.dumps(request) + "\n", timeout=1)
148 envelopes = [json.loads(line) for line in output.splitlines()]
149 if not envelopes or "error" in envelopes[-1]:
150 raise RuntimeError("observation capture failed: " + str(envelopes) + error)
151 return envelopes[-1]["result"]
152 finally:
153 if bridge.poll() is None:
154 bridge.kill()
155 bridge.communicate()
156 compositor.stdin.write("sleep 1s\nkey Return\n")
157 compositor.stdin.flush()
158 deadline = time.monotonic() + 30
159 while True:
160 try:
161 state = capture()
162 if state.get("context", {}).get("status") == "attached" and state.get("context", {}).get("session", {}).get("session_name") == "resilience":
163 break
164 except (subprocess.SubprocessError, ValueError, RuntimeError) as error:
165 (artifacts / "capture-error.txt").write_text(str(error) + "\n" + str(getattr(error, "stderr", "")))
166 if time.monotonic() > deadline:
167 diagnosis = subprocess.run(["docker", "exec", client, "sh", "-c", "ps -ef; ls -l /home/demo/visual/ui.sock"], capture_output=True, text=True, timeout=5)
168 (artifacts / "client-processes.txt").write_text(diagnosis.stdout + diagnosis.stderr)
169 (artifacts / "relay-stats.json").write_text(json.dumps(control("stats")))
170 (artifacts / "last-capture.json").write_text(json.dumps(locals().get("state")))
171 raise RuntimeError("visual client did not attach")
172 time.sleep(0.1)
173 (artifacts / "initial.json").write_text(json.dumps(state))
174 def key(command):
175 compositor.stdin.write(command + "\n")
176 compositor.stdin.flush()
177 def responsive(label):
178 latest = shots / "latest.png"
179 before = artifacts / (label + "-before.png")
180 shutil.copyfile(latest, before)
181 start = time.monotonic()
182 stamp = latest.stat().st_mtime_ns
183 key("key Alt+space")
184 key("type SSP")
185 key("key Return")
186 while time.monotonic() - start < 2:
187 if latest.stat().st_mtime_ns != stamp:
188 after = artifacts / (label + "-after.png")
189 shutil.copyfile(latest, after)
190 changed = subprocess.run([str(ROOT / "build/resilience-pixels"), str(before), str(after)], capture_output=True, text=True)
191 observed = capture()
192 (artifacts / (label + "-observed.json")).write_text(json.dumps(observed))
193 if changed.returncode == 0 and "Sessions" in observed.get("text", "") and "Esc close" in observed.get("text", ""):
194 key("key Escape")
195 return {"elapsed": time.monotonic()-start, "pixels": json.loads(changed.stdout), "context": observed["context"]}
196 time.sleep(0.05)
197 raise RuntimeError(label + ": local picker failed 2s response gate")
198 evidence = {"healthy": responsive("healthy")}
199 time.sleep(0.5)
200 key("key Alt+space")
201 key("type SSP")
202 key("key Return")
203 deadline = time.monotonic() + 5
204 while "Sessions" not in capture().get("text", ""):
205 if time.monotonic() > deadline:
206 raise RuntimeError("picker did not open before blackout")
207 time.sleep(0.1)
208 before = artifacts / "blackout-before.png"
209 shutil.copyfile(shots / "latest.png", before)
210 assert "Sessions" in capture().get("text", ""), "picker was not open before blackout"
211 impairment = os.environ.get("VEV_RESILIENCE_LINK", "both")
212 if impairment not in ("both", "udp", "tcp"):
213 raise ValueError("VEV_RESILIENCE_LINK must be both, udp or tcp")
214 control("blackout", udp=impairment != "tcp", tcp=impairment != "udp", active=True)
215 start = time.monotonic()
216 key("type /resilience")
217 deadline = start + 2
218 while time.monotonic() < deadline:
219 after = artifacts / "blackout-after.png"
220 shutil.copyfile(shots / "latest.png", after)
221 changed = subprocess.run([str(ROOT / "build/resilience-pixels"), str(before), str(after)], capture_output=True, text=True)
222 observed = capture()
223 if changed.returncode == 0 and "1 matches" in observed.get("text", ""):
224 evidence["blackout"] = {"elapsed": time.monotonic()-start, "pixels": json.loads(changed.stdout)}
225 break
226 time.sleep(0.05)
227 else:
228 raise RuntimeError("open picker search froze during blackout")
229 time.sleep(3)
230 control("blackout", udp=True, tcp=True, active=False)
231 key("key Escape")
232 time.sleep(0.2)
233 key("key Escape")
234 deadline = time.monotonic() + 5
235 while "Sessions" in capture().get("text", ""):
236 if time.monotonic() > deadline:
237 raise RuntimeError("picker did not close after restoration")
238 time.sleep(0.1)
239 key("key Ctrl+c")
240 time.sleep(1)
241 initial_session = state["context"]["session"]
242 key("type printf 'RESILIENCE_RECOVERED\\n'")
243 key("key Return")
244 deadline = time.monotonic() + 10
245 while time.monotonic() < deadline:
246 recovered = capture()
247 if "\nRESILIENCE_RECOVERED" in recovered.get("text", "") and recovered["context"]["session"] == initial_session:
248 evidence["recovered"] = recovered["context"]
249 break
250 time.sleep(0.1)
251 else:
252 raise RuntimeError("session did not recover after blackout")
253 # Negative control: compositor can still render while client is
254 # stopped. No client observation is allowed to satisfy the gate.
255 pid_result = docker("exec", client, "pgrep", "-f", "^vev attach", capture_output=True, text=True)
256 pid = pid_result.stdout.strip().splitlines()[0]
257 docker("exec", client, "kill", "-STOP", pid)
258 frozen = artifacts / "frozen.png"
259 shutil.copyfile(shots / "latest.png", frozen)
260 try:
261 frozen_at = (shots / "latest.png").stat().st_mtime_ns
262 key("key Alt+space")
263 # The compositor must keep capturing, or an unchanged frame
264 # proves nothing about the stopped client.
265 deadline = time.monotonic() + 3
266 while (shots / "latest.png").stat().st_mtime_ns == frozen_at:
267 if time.monotonic() > deadline:
268 raise RuntimeError("compositor stopped capturing during the negative control")
269 time.sleep(0.05)
270 time.sleep(0.3)
271 check = subprocess.run([str(ROOT / "build/resilience-pixels"), str(frozen), str(shots / "latest.png")], capture_output=True, text=True)
272 assert check.returncode == 1, "visual gate accepted a stopped client"
273 evidence["negative_control"] = "stopped client rejected"
274 finally:
275 docker("exec", client, "kill", "-CONT", pid)
276 if os.environ.get("VEV_RESILIENCE_LONG") == "1":
277 # The negative control's Alt+space was buffered while the
278 # client was stopped and opens the palette once it resumes.
279 # Close it on observed state only: a blind extra Escape would
280 # reach the shell as a lone ESC, which readline keeps as a
281 # Meta prefix and uses to eat the first replayed key.
282 deadline = time.monotonic() + 5
283 while "Commands" not in capture().get("text", ""):
284 if time.monotonic() > deadline:
285 raise RuntimeError("palette did not open after the stopped client resumed")
286 time.sleep(0.1)
287 key("key Escape")
288 deadline = time.monotonic() + 5
289 while "Commands" in capture().get("text", ""):
290 if time.monotonic() > deadline:
291 raise RuntimeError("palette did not close before the long outage")
292 time.sleep(0.1)
293 time.sleep(0.5)
294 control("blackout", udp=True, tcp=True, active=True)
295 if transport == "ssh":
296 control("disconnect")
297 deadline = time.monotonic() + 90
298 while time.monotonic() < deadline:
299 observed = capture()
300 if observed.get("context", {}).get("status") == "connecting":
301 break
302 time.sleep(0.2)
303 else:
304 raise RuntimeError("long outage did not enter resume")
305 if os.environ.get("VEV_RESILIENCE_RESUME") == "1":
306 key("type echo HELD_ONE; echo HELD_TWO")
307 key("key Return")
308 # Keep the route down until the compositor has delivered
309 # the complete physical input script to the resume owner.
310 time.sleep(2)
311 control("blackout", udp=True, tcp=True, active=False)
312 deadline = time.monotonic() + 30
313 while time.monotonic() < deadline:
314 resumed = capture()
315 text = resumed.get("text", "")
316 if "\nHELD_ONE" in text and "\nHELD_TWO" in text and resumed.get("context", {}).get("session") == initial_session:
317 assert text.index("\nHELD_ONE") < text.index("\nHELD_TWO"), "held input reordered"
318 # The echoed command line proves the first held
319 # byte was not swallowed before the shell saw it.
320 assert "$ echo HELD_ONE; echo HELD_TWO" in text, "held input replayed incompletely"
321 evidence["resumed"] = resumed["context"]
322 break
323 time.sleep(0.1)
324 else:
325 raise RuntimeError("long outage did not replay held input")
326 else:
327 start = time.monotonic()
328 cancel_key = os.environ.get("VEV_RESILIENCE_CANCEL", "Ctrl+c")
329 if cancel_key not in ("Ctrl+c", "Escape"):
330 raise ValueError("cancel must be Ctrl+c or Escape")
331 key("key " + cancel_key)
332 deadline = start + 2
333 while time.monotonic() < deadline:
334 observed = capture()
335 if observed.get("context", {}).get("status") == "picker":
336 evidence["resume_cancel"] = {"elapsed": time.monotonic()-start}
337 break
338 time.sleep(0.05)
339 else:
340 raise RuntimeError(cancel_key + " did not cancel resume")
341 control("blackout", udp=True, tcp=True, active=False)
342 evidence["stats"] = control("stats")
343 if transport == "quic":
344 assert evidence["stats"]["udp"], "QUIC bypassed impairment relay"
345 assert evidence["stats"]["tcp_bytes"] > 0, "SSH bypassed impairment relay"
346 (artifacts / "evidence.json").write_text(json.dumps(evidence, indent=2))
347 control("stop")
348 finally:
349 for process in reversed(processes):
350 if process.poll() is None:
351 process.terminate()
352 try:
353 process.wait(timeout=5)
354 except subprocess.TimeoutExpired:
355 process.kill()
356 process.wait()
357 for container in containers:
358 role = container.rsplit("-", 1)[-1]
359 for filename in ("vev-client.log", "vev-daemon.log", "broker/log/vev-broker.log"):
360 subprocess.run(["docker", "cp", container + ":/home/demo/.local/state/vev/" + filename, str(artifacts / (role + "-" + filename.replace("/", "-")))], timeout=30, capture_output=True)
361 subprocess.run(["docker", "rm", "-f", container], timeout=30, capture_output=True)
362 if network:
363 subprocess.run(["docker", "network", "rm", name], timeout=30, capture_output=True)
364 subprocess.run(["docker", "image", "rm", name], timeout=30, capture_output=True)
365 print("artifacts:", artifacts)
366
367
368if __name__ == "__main__":
369 main()