test/native_forward.py
Ref: Size: 20.3 KiB History
#!/usr/bin/env python3
"""Real-binary loopback forwarding acceptance for socket, stdio, and QUIC.
This is intentionally directly runnable rather than wired into native-e2e: it
needs a release mux/muxg pair and opens deliberate TCP listeners. Every daemon,
TCP service, GUI, socket, and XDG directory is owned by Rig and is retained on
failure. The focused Zig component tests remain the precise cap/admission
oracles; this script proves their real-binary plumbing and byte-stream effects.
"""
import hashlib
import http.server
import os
from pathlib import Path
import socket
import socketserver
import subprocess
import sys
import threading
import time
sys.dont_write_bytecode = True
from native_lifecycle import LifecycleRig
from native_tiling import eventually, require
TIMEOUT = 8
MAX_HTTP_RESPONSE = 1024 * 1024
def free_port():
# The listener is released only immediately before its owning fixture binds
# it. SO_REUSEADDR makes the reservation probe below reject a live owner
# without confusing its own prior TCP connections in TIME_WAIT for one.
with socket.socket() as sock:
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.bind(('127.0.0.1', 0))
return sock.getsockname()[1]
def distinct_ports(count):
ports = set()
while len(ports) < count:
ports.add(free_port())
return tuple(ports)
def recv_exact(sock, length):
data = bytearray()
while len(data) < length:
part = sock.recv(length - len(data))
require(part, f'unexpected EOF after {len(data)}/{length} bytes')
data.extend(part)
return bytes(data)
class TCPServices:
"""Daemon-host loopback endpoints with distinguishable HTTP content."""
def __init__(self, root):
self.token = hashlib.sha256(str(root).encode()).hexdigest()[:20]
self.http_port, self.echo_port, self.slow_port = distinct_ports(3)
self.expect_restart_disconnect = threading.Event()
token = self.token
expect_restart_disconnect = self.expect_restart_disconnect
class HTTP(http.server.BaseHTTPRequestHandler):
def do_GET(self):
body = ('native-forward-http-' + token + '\n').encode()
self.send_response(200)
self.send_header('Content-Length', str(len(body)))
self.send_header('Connection', 'close')
self.end_headers()
self.wfile.write(body)
def log_message(self, *_):
pass
class Echo(socketserver.BaseRequestHandler):
def handle(inner):
while True:
data = inner.request.recv(65536)
if not data:
try:
inner.request.shutdown(socket.SHUT_WR)
except OSError:
# The restart case deliberately tears this peer down.
# Other shutdown errors identify a fixture regression.
if not expect_restart_disconnect.is_set():
raise
return
inner.request.sendall(data)
class Slow(socketserver.BaseRequestHandler):
def handle(inner):
# Fill forwarding queues while the local client deliberately
# does not read. It is bounded so fixture cleanup never waits.
payload = b'SLOW-' + token.encode() + b'X' * 65536
for _ in range(48):
inner.request.sendall(payload)
inner.request.shutdown(socket.SHUT_WR)
self.servers = []
for port, handler in ((self.http_port, HTTP), (self.echo_port, Echo), (self.slow_port, Slow)):
server = socketserver.ThreadingTCPServer(('127.0.0.1', port), handler)
server.daemon_threads = True
thread = threading.Thread(target=server.serve_forever, daemon=True)
thread.start()
self.servers.append(server)
def close(self):
for server in self.servers:
server.shutdown()
server.server_close()
class ForwardRig(LifecycleRig):
def launch_forward(self, target_args, rules, label):
args = [*target_args]
for local, remote in rules:
args += ['--forward', f'{local}:{remote}']
# A named ephemeral session makes this invocation temporary and avoids
# changing the native workspace under the fixture's isolated state.
args += ['--session', 'forward']
self.launch_gui(args, label)
def restart_daemon(self, sock, old, label):
self.stop_daemon(sock, old)
self.daemons.remove((sock, old))
# Recovery must use the manager's original concrete socket identity.
proc = self.spawn([self.mux, 'd', 'start', '--sock', sock], label)
self.daemons.append((sock, proc))
eventually(lambda: Path(sock).exists(), label + ' daemon did not restart')
return proc
class PreResponseTransportStartupError(ConnectionError):
"""A reset or EOF before any response byte; listener startup may retry it."""
def _remaining(deadline):
remaining = deadline - time.monotonic()
require(remaining > 0, 'HTTP response exceeded deadline')
return remaining
def recv_http_part(sock, deadline, saw_response):
sock.settimeout(_remaining(deadline))
try:
part = sock.recv(65536)
except socket.timeout as error:
raise AssertionError('HTTP response exceeded deadline') from error
except OSError as error:
if not saw_response:
raise PreResponseTransportStartupError('HTTP transport reset before response') from error
raise AssertionError('HTTP transport failed after response started') from error
if not part and not saw_response:
raise PreResponseTransportStartupError('HTTP transport EOF before response')
return part
def read_http_response(sock, deadline):
response = bytearray()
while b'\r\n\r\n' not in response:
part = recv_http_part(sock, deadline, bool(response))
require(part, 'HTTP response ended before headers completed')
response.extend(part)
require(len(response) <= MAX_HTTP_RESPONSE, 'HTTP response headers exceed bound')
raw_headers, body = bytes(response).split(b'\r\n\r\n', 1)
lines = raw_headers.split(b'\r\n')
status = lines[0].split()
require(len(status) == 3 and status[0].startswith(b'HTTP/'), 'malformed HTTP status line')
try:
code = int(status[1])
except ValueError as error:
raise AssertionError('malformed HTTP status code') from error
lengths = []
for line in lines[1:]:
require(b':' in line, 'malformed HTTP header')
name, value = line.split(b':', 1)
if name.lower() == b'transfer-encoding':
raise AssertionError('HTTP Transfer-Encoding is unsupported')
if name.lower() == b'content-length':
lengths.append(value.strip())
require(len(lengths) == 1, 'HTTP response requires exactly one Content-Length')
require(lengths[0].isdigit(), 'malformed HTTP Content-Length')
length = int(lengths[0])
require(length <= MAX_HTTP_RESPONSE, 'HTTP Content-Length exceeds bound')
while True:
require(len(body) <= length, 'HTTP response has surplus body bytes')
part = recv_http_part(sock, deadline, True)
if not part:
break
body += part
require(len(body) <= MAX_HTTP_RESPONSE, 'HTTP response body exceeds bound')
require(len(body) == length, f'HTTP response ended after {len(body)}/{length} body bytes')
return code, body
def http_get(port, deadline=None):
deadline = time.monotonic() + TIMEOUT if deadline is None else deadline
try:
sock = socket.create_connection(('127.0.0.1', port), timeout=_remaining(deadline))
except OSError as error:
raise PreResponseTransportStartupError('HTTP connection failed') from error
with sock:
try:
sock.settimeout(_remaining(deadline))
sock.sendall(b'GET / HTTP/1.0\r\nHost: fixture\r\n\r\n')
sock.shutdown(socket.SHUT_WR)
except OSError as error:
raise PreResponseTransportStartupError('HTTP request failed before response') from error
return read_http_response(sock, deadline)
def _socketpair_response(parts):
reader, writer = socket.socketpair()
errors = []
def write_response():
try:
for part in parts:
writer.sendall(part)
time.sleep(.01)
writer.shutdown(socket.SHUT_WR)
except OSError as error:
errors.append(error)
finally:
writer.close()
thread = threading.Thread(target=write_response)
thread.start()
result = error = None
try:
result = read_http_response(reader, time.monotonic() + TIMEOUT)
except BaseException as caught:
error = caught
finally:
reader.close()
thread.join(TIMEOUT)
require(not thread.is_alive(), 'HTTP socketpair fixture thread hung')
if error is not None:
# Closing after a deliberate parser rejection can break the writer's
# shutdown; retain the parser error rather than masking it.
raise error
require(not errors, 'HTTP socketpair fixture failed: ' + repr(errors))
return result
def http_parser_regressions():
code, body = _socketpair_response((b'HTTP/1.0 200 OK\r\nContent-', b'Length: 5\r\nConnection: close\r\n\r\nhe', b'llo'))
require(code == 200 and body == b'hello', 'fragmented HTTP response was not parsed completely')
for response in (b'bad\r\n\r\n',
b'HTTP/1.0 200 OK\r\nContent-Length: 3\r\nContent-Length: 3\r\n\r\nabc',
b'HTTP/1.0 200 OK\r\nContent-Length: 3\r\nTransfer-Encoding: chunked\r\n\r\nabc',
b'HTTP/1.0 200 OK\r\nContent-Length: 5\r\n\r\nabc',
b'HTTP/1.0 200 OK\r\nContent-Length: 3\r\n\r\nmore'):
try:
_socketpair_response((response,))
except AssertionError:
pass
else:
raise AssertionError('malformed, duplicate, truncated, or surplus HTTP response was accepted')
def http_retry_classifier_regression():
original = http_get
try:
def parser_failure(_, deadline=None):
raise AssertionError('malformed HTTP status line')
globals()['http_get'] = parser_failure
try:
wait_forward(1)
except AssertionError as error:
require('malformed HTTP status line' in str(error), 'parser failure was changed')
else:
raise AssertionError('wait_forward swallowed a parser failure')
finally:
globals()['http_get'] = original
def echo_roundtrip(port, payload, half_close=False):
with socket.create_connection(('127.0.0.1', port), timeout=TIMEOUT) as sock:
sock.sendall(payload)
if half_close:
sock.shutdown(socket.SHUT_WR)
got = recv_exact(sock, len(payload))
if half_close:
require(sock.recv(1) == b'', 'remote half-close did not reach local TCP peer')
return got
def listener_reserved(port):
# SO_REUSEADDR avoids treating a just-closed fixture connection in TIME_WAIT
# as listener ownership. Each call is paired with a real forwarded request,
# so the failed bind has both kernel and forwarding-owner evidence.
with socket.socket() as probe:
probe.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
try:
probe.bind(('127.0.0.1', port))
except OSError:
return True
return False
def wait_forward(port):
deadline = time.monotonic() + TIMEOUT
while time.monotonic() < deadline:
try:
return http_get(port, deadline)
except PreResponseTransportStartupError:
time.sleep(min(.04, max(0, deadline - time.monotonic())))
raise AssertionError('forward listener did not become usable')
def route_args(rig, kind, sock):
if kind == 'socket':
return ['--sock', sock]
if kind == 'stdio':
return ['--via', rig.mux + ' d proxy --sock ' + sock]
if kind == 'quic':
target = rig.targets[sock]
key = str(Path(rig.env['XDG_CONFIG_HOME']) / 'mux/key')
return [target, '--key', key]
raise AssertionError('unknown route ' + kind)
def basic_routes(rig, services):
"""Each entry transport carries plural, concurrent real TCP channels."""
for kind in ('socket', 'stdio', 'quic'):
sock, _ = rig.daemon('forward-' + kind, quic=kind == 'quic')
local_http, local_echo = distinct_ports(2)
rig.launch_forward(route_args(rig, kind, sock),
[(local_http, services.http_port), (local_echo, services.echo_port)],
'forward-' + kind + '-gui')
code, body = wait_forward(local_http)
require(code == 200, kind + ' forwarding returned HTTP status ' + str(code))
require(('native-forward-http-' + services.token).encode() in body,
kind + ' forwarding reached wrong HTTP endpoint')
payloads = [os.urandom(256 * 1024 + n) for n in range(3)]
results, errors = [None] * len(payloads), []
error_lock = threading.Lock()
def roundtrip_worker(i):
try:
results[i] = echo_roundtrip(local_echo, payloads[i])
except BaseException as error:
with error_lock:
errors.append(error)
workers = [threading.Thread(target=roundtrip_worker, args=(i,)) for i in range(len(payloads))]
for worker in workers:
worker.start()
for worker in workers:
worker.join(TIMEOUT)
require(not worker.is_alive(), kind + ' concurrent forwarding stream hung')
require(not errors, kind + ' concurrent forwarding worker failed: ' + repr(errors))
for payload, result in zip(payloads, results):
require(result == payload, kind + ' concurrent stream corrupted bytes')
require(echo_roundtrip(local_echo, os.urandom(1024 * 1024 + 17), half_close=True) != b'',
kind + ' large half-closed stream failed')
rig.ok(kind + ' forwarding carries unique HTTP and concurrent large half-closed echo streams')
rig.quit()
def local_lifetime_and_recovery(rig, services):
sock, daemon = rig.daemon('forward-lifetime')
http_local, echo_local, slow_local, refused_local, unavailable_remote = distinct_ports(5)
rules = [(http_local, services.http_port), (echo_local, services.echo_port),
(slow_local, services.slow_port), (refused_local, unavailable_remote)]
rig.launch_forward(['--sock', sock], rules, 'forward-lifetime-gui')
code, body = wait_forward(http_local)
require(code == 200, 'forward listener returned HTTP status ' + str(code))
require(('native-forward-http-' + services.token).encode() in body,
'forward listener reached wrong HTTP endpoint')
require(listener_reserved(http_local), 'live forward listener was not reserved')
# The refused destination is channel-local: the adjacent echo rule remains
# usable after its open is rejected.
with socket.create_connection(('127.0.0.1', refused_local), timeout=TIMEOUT) as refused:
refused.settimeout(TIMEOUT)
require(refused.recv(1) == b'', 'refused remote destination left a live local stream')
require(echo_roundtrip(echo_local, b'channel-isolation') == b'channel-isolation',
'refused destination damaged a sibling forwarding rule')
# This only holds a non-reading peer; it does NOT establish that remote
# credit or a forwarding queue actually blocked. The terminal assertion is
# therefore an unverified responsiveness smoke check, not backpressure
# evidence. Precise queue/admission coverage remains unit-test-only.
slow = socket.create_connection(('127.0.0.1', slow_local), timeout=TIMEOUT)
try:
time.sleep(.15)
pane = rig.state()['focus']
rig.shell("printf 'FORWARD-TERMINAL-RESPONSIVE\\n'")
rig.wait_marker(sock, 'forward', 'FORWARD-TERMINAL-RESPONSIVE')
require(rig.gui.poll() is None, 'slow forwarding reader stopped the GUI')
finally:
slow.close()
# A matching quick split changes focus. Remove that new sibling and prove
# the original entry pane remains before checking manager lifetime.
original = rig.state()['focus']
rig.chord('b')
state = rig.wait_state(lambda s: len(s['panes']) == 2 and s['focus'] != original)
sibling = state['focus']
rig.chord('d')
state = rig.wait_state(lambda s: len(s['panes']) == 1 and
original in {pane['id'] for pane in s['panes']} and
sibling not in {pane['id'] for pane in s['panes']})
require(listener_reserved(http_local), 'removing one matching pane released forward listener')
# Losing only the dedicated role connection resets old streams, reserves
# listeners, and allows a daemon restart at the same socket to recover.
old = socket.create_connection(('127.0.0.1', echo_local), timeout=TIMEOUT)
old.sendall(b'interrupted')
services.expect_restart_disconnect.set()
try:
daemon = rig.restart_daemon(sock, daemon, 'forward-whole-daemon-restarted')
old.settimeout(TIMEOUT)
# A reply already queued before the tear is permitted; the old channel
# must nevertheless become EOF rather than silently survive/reconnect.
while old.recv(65536):
pass
finally:
old.close()
services.expect_restart_disconnect.clear()
require(listener_reserved(http_local), 'transport interruption released listener reservation')
code, body = wait_forward(http_local)
require(code == 200, 'recovered forward listener returned HTTP status ' + str(code))
require(('native-forward-http-' + services.token).encode() in body,
'recovered forward listener reached wrong HTTP endpoint')
require(echo_roundtrip(echo_local, b'recovered') == b'recovered', 'forward listener did not recover after daemon restart')
# A conflicting second invocation fails before it can disturb the active
# GUI/listener. No FIFO is inherited because it is not that process's UI.
env = rig.env.copy()
env.pop('MUXG_TEST_FIFO', None)
collision = subprocess.run([rig.muxg, '--sock', sock, '--session', 'collision',
'--forward', f'{http_local}:{services.http_port}'],
env=env, capture_output=True, text=True, timeout=TIMEOUT)
require(collision.returncode != 0, 'local forward collision was accepted')
diagnostic = (collision.stdout + collision.stderr).lower()
require(any(word in diagnostic for word in ('forward', 'bind', 'address', 'port', 'in use')),
'local forward collision lacked a visible bind diagnostic: ' + diagnostic)
require(echo_roundtrip(echo_local, b'after-collision') == b'after-collision',
'local collision disturbed existing forwarding')
# Final matching-pane removal is the lifetime boundary. The focused pane
# is the only remaining target after the prior detach.
rig.chord('d')
rig.wait_state(lambda s: not s['panes'])
eventually(lambda: not listener_reserved(http_local),
'final matching pane did not release forward listener')
rig.quit()
rig.ok('focus/final-target lifetime, refusal isolation, unverified slow-reader responsiveness smoke, whole-daemon recovery')
def main():
require(len(sys.argv) == 3, 'usage: native_forward.py RELEASE_MUX RELEASE_MUXG')
rig = ForwardRig(*sys.argv[1:])
services = None
try:
services = TCPServices(rig.root)
version = subprocess.check_output([rig.muxg, '--version'], text=True)
require('ReleaseSafe' in version or 'ReleaseFast' in version,
'native forwarding requires ReleaseSafe or ReleaseFast binaries')
http_parser_regressions()
http_retry_classifier_regression()
basic_routes(rig, services)
local_lifetime_and_recovery(rig, services)
print(f'PASS: native forwarding ({rig.checkpoints} checkpoints)', flush=True)
except BaseException:
rig.failure_artifacts()
raise
finally:
# Neither owner may suppress the other's cleanup.
try:
if services is not None:
services.close()
finally:
rig.close()
if __name__ == '__main__':
main()