a73x

8d4f4e41

feat: add native local TCP forwarding

a73x   2026-09-08 17:24

Commit message
feat: add native local TCP forwarding
README.md
Old New
@@ -30,7 +30,7 @@ the default development build and CI gates:
30 make native # Debug development build and no-window unit tests 30 make native # Debug development build and no-window unit tests
31 make native-e2e # One ReleaseSafe GUI user journey; needs python3 31 make native-e2e # One ReleaseSafe GUI user journey; needs python3
32 make native-stress # Linux: raw cat /dev/random, shared and separate daemon neighbours 32 make native-stress # Linux: raw cat /dev/random, shared and separate daemon neighbours
33 ./zig-out/bin/muxg [TARGET] [--session NAME] [--sock PATH] [--via CMD] [--key PATH] [--font-px N] 33 ./zig-out/bin/muxg [TARGET] [--forward LOCAL_PORT:REMOTE_PORT]... [--session NAME] [--sock PATH] [--via CMD] [--key PATH] [--font-px N]
34 ``` 34 ```
35 35
36 For interactive use and performance measurements, build **both** binaries in 36 For interactive use and performance measurements, build **both** binaries in
@@ -146,8 +146,17 @@ Open the saved workspace, or an explicit target in a temporary workspace:
146 ```sh 146 ```sh
147 ./zig-out/bin/muxg 147 ./zig-out/bin/muxg
148 ./zig-out/bin/muxg alpha --session work 148 ./zig-out/bin/muxg alpha --session work
149 ./zig-out/bin/muxg --forward 8080:80 alpha
149 ``` 150 ```
150 151
152 `--forward` binds the local IPv4 loopback address (`127.0.0.1`) and carries
153 TCP to the same loopback address on the target host. It may be repeated; exact
154 duplicates are ignored. Forwarding requires an explicit target for this
155 invocation and is not saved in the workspace. The listeners remain reserved
156 across daemon reconnects, while live streams are reset; removing the final pane
157 for that target releases them. Hostname destinations, wildcard binds, reverse
158 forwarding and UDP are not part of this first slice.
159
151 Press `Ctrl+\`, then **v** for a pane below or **b** for a pane beside it. 160 Press `Ctrl+\`, then **v** for a pane below or **b** for a pane beside it.
152 The new pane uses the focused pane's connection and gets the next free numeric 161 The new pane uses the focused pane's connection and gets the next free numeric
153 session name automatically. It opens in the background; you can keep typing or 162 session name automatically. It opens in the background; you can keep typing or
docs/daemon-contracts.md
Old New
@@ -3,8 +3,10 @@
3 Read when changing these components. Implementation history retained from 3 Read when changing these components. Implementation history retained from
4 CLAUDE.md; the current code and regression tests are authoritative. 4 CLAUDE.md; the current code and regression tests are authoritative.
5 5
6 - **A QUIC client says goodbye.** Every QUIC connection takes one of the 6 - **A QUIC client says goodbye.** Every QUIC connection waits in a bounded,
7 daemon's `max_clients` slots at the handshake, attached or not, and the 7 expiring role table after the handshake, then its first complete frame
8 claims either one of the daemon's `max_clients` terminal slots, a dedicated
9 forwarding slot, or a one-shot observer operation. The
8 wall polls each QUIC host once a second on a connection of its own. 10 wall polls each QUIC host once a second on a connection of its own.
9 `quic.Client.deinit` therefore writes CONNECTION_CLOSE before it closes 11 `quic.Client.deinit` therefore writes CONNECTION_CLOSE before it closes
10 the socket; a teardown that only dropped the socket left the daemon to 12 the socket; a teardown that only dropped the socket left the daemon to
@@ -16,10 +18,12 @@ CLAUDE.md; the current code and regression tests are authoritative.
16 A slot is spent per ATTACH, not per session, so 32 matches `max_sessions` 18 A slot is spent per ATTACH, not per session, so 32 matches `max_sessions`
17 and `wallview.max_tiles` — one full wall of tiles fits exactly, and a 19 and `wallview.max_tiles` — one full wall of tiles fits exactly, and a
18 second wall on the same daemon is refused. The listener's `max_conns` 20 second wall on the same daemon is refused. The listener's `max_conns`
19 stays ABOVE `max_clients` (a connection exists from the handshake and only 21 stays above the admitted terminal and forwarding population, and the
20 then asks for a slot, and a peer that finds no connection is dropped 22 provisional role table can hold every listener connection (a connection
21 silently rather than refused), pinned by a test because a transport file 23 exists from the handshake and only then asks for a role, and a peer that
22 does not read the daemon's tables. And the listener OUTLIVES the slots 24 finds no connection is dropped silently rather than refused), pinned by a
25 test because a transport file does not read the daemon's tables. And the
26 listener OUTLIVES the slots
23 that close through it: a QUIC sink closes its connection via the listener, 27 that close through it: a QUIC sink closes its connection via the listener,
24 so a borrowed listener's `deinit` is registered AFTER the server's. 28 so a borrowed listener's `deinit` is registered AFTER the server's.
25 29
docs/superpowers/plans/2026-09-07-native-port-forwarding-slice-1.md
Old New
@@ -0,0 +1,204 @@
1 # Native local TCP forwarding — slice 1
2
3 ## Behaviour and scope
4
5 - `muxg --forward LOCAL_PORT:REMOTE_PORT TARGET` accepts repeatable, non-zero
6 TCP port pairs. Forwarding requires one explicit entry transport (`TARGET`,
7 `--sock`, or `--via`) and applies only to this invocation.
8 - Each unique rule binds `127.0.0.1:LOCAL_PORT`. Accepted streams connect on
9 the daemon host to `127.0.0.1:REMOTE_PORT`; names, DNS, wildcard binds,
10 reverse forwarding and UDP are deliberately absent.
11 - One process-owned forwarding manager uses one dedicated mux connection for
12 the entry target, independent of pane focus, terminal attachment, session
13 creation and geometry. Duplicate rules are deduplicated. A bind collision
14 or invalid/unsupported peer is reported visibly without damaging terminal
15 attachments.
16 - The manager remains while any pane has the same concrete target identity;
17 focus and session do not matter. Replacing/removing the final matching pane
18 releases its listeners, while another matching pane keeps them alive.
19 - A lost mux connection keeps the local listening sockets reserved, resets
20 live streams, refuses new accepts while unavailable, and redials. The
21 dedicated handoff target never repeats an entry target's start request or
22 narration.
23 - The daemon admits forwarding peers in a small table separate from the 32
24 interactive client slots. A forwarding peer cannot consume a terminal
25 session or displace a full workspace.
26
27 ## Wire and resource contract
28
29 - The existing byte-stream transport remains opaque and single-stream.
30 Forward-role hello/ready frames identify the connection before any channel
31 frame is accepted, with a bounded client handshake wait and explicit
32 unsupported-role failure.
33 - Client-chosen non-zero channel IDs carry open/open-result, data, credit,
34 half-close and reset frames. Data payloads and credit are bounded. Both
35 peers cap channel counts, staged transport bytes, and per-channel queues.
36 - TCP listeners and streams are nonblocking. Connect completion is checked
37 with `SO_ERROR`; accept/read/write and frame work have per-pump budgets so a
38 busy tunnel cannot monopolize the daemon or native event loop.
39 - EOF is propagated as a half-close after queued bytes drain. Invalid role,
40 malformed frames, cap violations and channel-local failures reset a channel
41 or close the forwarding peer without entering terminal frame handling.
42
43 ## Ownership and files
44
45 - Shared wire: `src/engine/protocol.zig`.
46 - Shared client: a forwarding child module under `src/client/`, exported from
47 `src/client/client.zig`; it owns listener/channel state and its worker.
48 - Daemon: a relay child module under `src/server/`, integrated into
49 `src/server/server.zig` observer promotion, admission, polling and teardown.
50 - Native: `src/gui/runtime.zig`, `src/gui/frame.zig`, `src/gui/native_core.zig`
51 and `src/cli/muxg.zig` for process lifetime and option plumbing.
52 - Coordination/docs: `build.zig` only if module wiring needs it, plus
53 `README.md` and focused tests. Terminal-wall files and existing agent
54 forwarding semantics remain untouched.
55
56 ## Acceptance checks
57
58 - Protocol and unit tests cover codecs, malformed/oversized frames, duplicate
59 rules, local collisions, channel caps, credit/backpressure, refusal,
60 half-close, channel isolation and unavailable-peer behaviour.
61 - Daemon/client integration tests use loopback sockets for request/response,
62 streaming and concurrent channels, and assert the dedicated role creates no
63 session and does not alter geometry or interactive admission.
64 - Native-core tests cover manager lifetime independent of focus and session,
65 final-matching-pane removal, replacement and deduplication, and
66 entry-target scoping. Where a real window is available, an isolated native
67 journey checks CLI plumbing and reconnect. Direct QUIC and stdio-proxy
68 forwarding paths are core integration evidence; external SSH and macOS are
69 reported as unavailable where their environments are not present.
70 - Run focused `make daemon-test client-test native-core-test`, `make native`,
71 then required `make check` and `make ci`; use `tools/run-logged.sh` for noisy
72 gates and isolated daemon state for real-binary checks. Run `make
73 native-e2e` when the display environment supports it. Linux results do not
74 claim macOS coverage.
75
76 ## Delegation
77
78 - Implementation and focused validation: delegated worker on `gpt-5.6-sol`
79 at high reasoning effort, owning the files above.
80 - Independent review: separate `gpt-6-astra` worker at high reasoning effort;
81 any review-fix brief is assigned by the
82 coordinating agent; this implementation worker does not self-review as the
83 independent reviewer.
84
85 ## Validation
86
87 Validation snapshot: `HEAD` `03661841b1ef0853f43317067d4894ebe355a937`; the
88 tracked-diff SHA-256 was
89 `57aa4b5573d9d3dba3d9f1b808ae8b2f251654ee6d176a303a850707ec09aa90`.
90 New-file SHA-256 values: `src/client/forward.zig`
91 `e2b2a9c10c3e8f3b65fea6df67abecac552ff9eba238972fe73cb0f6a749b0ec`,
92 `src/server/server_forward.zig`
93 `93dc340496878855aeb87bb2890ee7fdf00fc276c2a629e01cd761361f1519d3`,
94 `test/native_forward.py`
95 `140100e52d2d48773322a23aff4af1e22be41b7c391277bde3755dd9288a6502`, and
96 `test/native_forward_remote.py`
97 `d401d2f1e4efbea0574aa4b0975fc5d8d4f3bc53ba24dc9a5a279b7f7e4ab0c8`.
98 The complete tracked-path list, platform, display environment, and tool
99 availability are in `dist/native-forward-validation/baseline.log`. The host was Linux
100 `charizard` x86_64 with `DISPLAY=:0`, `WAYLAND_DISPLAY=wayland-1`, and
101 `XDG_RUNTIME_DIR=/run/user/1000`. Make resolved the pinned Zig 0.15.2 through
102 mise at `/home/xanderle/.local/share/mise/installs/zig/0.15.2/bin/zig`;
103 `deps/zig/zig` itself was unavailable. Python 3.14.7, ShellCheck, Node,
104 pkg-config, and Make were available.
105
106 All noisy commands used `tools/run-logged.sh`; each status below is the actual
107 wrapper-preserved exit status, and its complete output is retained at the named
108 path. Builds ran serially.
109
110 | Command | Status | Log |
111 | --- | ---: | --- |
112 | `python3 -c` AST parse of `test/native_forward.py` and `test/native_forward_remote.py` | 0 | `dist/native-forward-validation/python-syntax.log` |
113 | `make daemon-test client-test native-core-test` | 0 | `dist/native-forward-validation/focused.log` |
114 | `make native` | 0 | `dist/native-forward-validation/native.log` |
115 | `make install BINDIR="$PWD/dist/native-forward-validation/release/bin" INSTDIR="$PWD/dist/native-forward-validation/release/stage"` | 0 | `dist/native-forward-validation/release-build.log` |
116 | release-binary version check | 0 (`muxg` reported `ReleaseSafe`) | `dist/native-forward-validation/release-versions.log` |
117 | `python3 -B test/native_forward.py .../release/bin/mux .../release/bin/muxg` | 0 (4 checkpoints) | `dist/native-forward-validation/local-forward.log` |
118 | `make check` | 0 | `dist/native-forward-validation/check.log` |
119 | `make ci` | 0 | `dist/native-forward-validation/ci.log` |
120 | `make native-e2e` | 0 (33 checkpoints) | `dist/native-forward-validation/native-e2e.log` |
121
122 `make native` alone produces the ordinary native build, whereas
123 `native_forward.py` explicitly requires ReleaseSafe or ReleaseFast binaries.
124 The isolated `make install` invocation above was therefore additionally needed
125 to produce matching ReleaseSafe `mux` and `muxg` binaries without installing
126 into the user bin directory. The local forwarding script exercised socket,
127 stdio, and QUIC entry routes. Its log also contains a fixture-thread
128 `OSError: [Errno 107] Transport endpoint is not connected`; the driver itself
129 completed its four checkpoints and exited 0, so this is recorded rather than
130 silently treated as a separate passing assertion.
131
132 `test/native_forward_remote.py` was **not executed** and no remote host was
133 contacted. Syntax-only AST parsing exited 0. Independent review still marks
134 that driver unsafe: its `start -d` argument order is invalid; the SSH wrapper
135 cannot handle handoff `PATH` assignment; setup inherits `PATH`; PID
136 publication/ownership is unsafe; and cleanup races GUI reconnect and ignores
137 errors. Consequently real SSH forwarding, including remote QUIC, remains
138 unvalidated. The reviewer also found no black-box admission snapshots. The
139 slow-reader leg is only a responsiveness smoke check, not a backpressure
140 oracle; recovery is whole-daemon rather than forwarding-only; and the stale
141 local comment around line 259 claiming the dedicated connection does not
142 repeat an entry target is not validation evidence.
143
144 Unconfirmed production concerns remain: no connect deadline, hot handshake
145 retry, late QUIC receive cap, and one-shot QUIC admission. A user-reported
146 manual Python `http.server` plus browser success is recorded only as an
147 unverified transport detail, not as acceptance evidence. No macOS result is
148 claimed.
149
150 ### Remote SSH and direct-QUIC forwarding revalidation
151
152 The opt-in check ran against the user-authorized disposable
153 `ubuntu@192.168.0.107` VM (Linux 7.0.0-31-generic x86_64). The existing local
154 ReleaseSafe pair at `dist/native-forward-validation/release/bin/{mux,muxg}`
155 was used without rebuilding: both are x86-64 Linux binaries and `muxg --version`
156 reported `ReleaseSafe` (`remote-binary-platform.log`). The remote fixture
157 copied that `mux` into a private mode-0700 `mktemp` root, generated one
158 fixture-owned key locally and copied it mode 0600, and used private HOME/XDG,
159 Unix socket, services, and a random high (40000--59999) UDP port. It copies
160 and sources `test/os_oracle.sh` remotely.
161
162 | Command | Status | Log/evidence |
163 | --- | ---: | --- |
164 | Python AST syntax parse of both forwarding scripts | 0 | `dist/native-forward-validation/remote-python-syntax.log`, `remote-python-syntax-status.txt` |
165 | `MUX_FORWARD_REMOTE_ENABLE=1 python3 -B test/native_forward_remote.py dist/native-forward-validation/release/bin/mux dist/native-forward-validation/release/bin/muxg` | 0 | `dist/native-forward-validation/remote-forward.log`, `remote-forward-command.txt`, `remote-forward-status.txt` |
166 | ReleaseSafe/version/architecture and remote-platform record | 0 | `dist/native-forward-validation/remote-binary-platform.log` |
167
168 The first route is regression coverage for actual muxg stdio proxying:
169 `muxg --forward --via "ssh ubuntu@192.168.0.107 /absolute/fixture/mux d proxy
170 --sock /absolute/fixture/mux.sock"`; it uses neither `ssh -L` nor a UDP proxy.
171 The second is a direct GUI connection to
172 `quic://192.168.0.107:HIGH_PORT --key FIXTURE_KEY`, with no SSH tunnel in that
173 route. Both routes asserted an exact `HTTP/1.0 200` response and unique body,
174 three concurrent random streams larger than 1 MiB with SHA-256 records in the
175 log, exact bidirectional bytes, TCP half-close EOF, and local listener
176 reservation. The direct-QUIC route additionally stopped and restarted the
177 whole private daemon, confirmed listener reservation while unavailable, then
178 confirmed exact HTTP recovery. This is explicitly whole-daemon recovery, not
179 forwarding-only recovery.
180
181 After each start, the remote OS oracle identifies exactly one daemon PID only
182 when its executable is the copied fixture `mux` and it owns the fixture Unix
183 socket. It separately observes the concrete QUIC UDP endpoint bound via
184 `udp_local_bound`; this does not attribute UDP ownership to that PID. Stop
185 success alone is not accepted: after the restart and at final cleanup the
186 oracle requires the recorded PID gone, Unix socket absent, and UDP listener
187 released. The GUI quits before final stop. The successful log ends
188 `PASS: remote fixture cleanup (daemon PID/socket/UDP released; services and
189 root removed)`; service termination and root deletion are independently
190 verified. There were no retained remote fixture paths or cleanup errors.
191
192 Fresh SHA-256 values are in
193 `dist/native-forward-validation/remote-validation-hashes.txt`: corrected
194 `test/native_forward_remote.py` is
195 `96686da6fc048c08b4df6398a7028326a87263cfdd3057150a553b135c2acc96`, unchanged
196 `test/native_forward.py` is
197 `140100e52d2d48773322a23aff4af1e22be41b7c391277bde3755dd9288a6502`, and the
198 current complete `src/` working-tree manifest is
199 `7d4e114be65dbc0ac501d8a2978df9fefde34ca63fda23ca6ade489d7569e580`.
200 The manifest records the pre-existing dirty production snapshot; this work made
201 no production-file changes. Protocol/platform coverage is Linux x86-64 only:
202 SSH stdio and direct IPv4 QUIC were exercised; macOS was not exercised. No
203 general build or gate was rerun because this delivery changes only the Python
204 harness and validation record.
src/cli/muxg.zig
Old New
@@ -24,13 +24,14 @@ const Color = struct {
24 }; 24 };
25 25
26 const usage = 26 const usage =
27 \\usage: muxg [TARGET] [--session NAME] [--sock PATH] [--via CMD] [--key PATH] [--theme NAME|PATH] [--background HEX] [--foreground HEX] [--cursor-color HEX] [--palette N=HEX] [--font-family FAMILY] [--font-size POINTS] [--font-px N] 27 \\usage: muxg [TARGET] [--forward LOCAL_PORT:REMOTE_PORT]... [--session NAME] [--sock PATH] [--via CMD] [--key PATH] [--theme NAME|PATH] [--background HEX] [--foreground HEX] [--cursor-color HEX] [--palette N=HEX] [--font-family FAMILY] [--font-size POINTS] [--font-px N]
28 \\ 28 \\
29 \\ TARGET HOST (ssh handoff) or quic://HOST[:PORT]; none restores the saved workspace 29 \\ TARGET HOST (ssh handoff) or quic://HOST[:PORT]; none restores the saved workspace
30 \\ --session the session name (default: the daemon's default session) 30 \\ --session the session name (default: the daemon's default session)
31 \\ --sock a local daemon's socket path 31 \\ --sock a local daemon's socket path
32 \\ --via a command whose stdio is the daemon 32 \\ --via a command whose stdio is the daemon
33 \\ --key the QUIC key file (or MUX_KEY_FILE) 33 \\ --key the QUIC key file (or MUX_KEY_FILE)
34 \\ --forward bind 127.0.0.1:LOCAL_PORT and reach 127.0.0.1:REMOTE_PORT on TARGET; repeatable
34 \\ --font-px font pixels at 100% display scale (default 16) 35 \\ --font-px font pixels at 100% display scale (default 16)
35 \\ --font-family ordered font family; repeat for fallbacks (config: font-family) 36 \\ --font-family ordered font family; repeat for fallbacks (config: font-family)
36 \\ --font-size font size in points, 1–192 (config: font-size) 37 \\ --font-size font size in points, 1–192 (config: font-size)
@@ -54,6 +55,7 @@ const Arguments = struct {
54 cursor_color: ?Color = null, 55 cursor_color: ?Color = null,
55 _palette_values: [256]?u32 = [_]?u32{null} ** 256, 56 _palette_values: [256]?u32 = [_]?u32{null} ** 256,
56 _font_families: std.ArrayListUnmanaged([:0]const u8) = .empty, 57 _font_families: std.ArrayListUnmanaged([:0]const u8) = .empty,
58 _forwards: std.ArrayListUnmanaged(client.forward.Rule) = .empty,
57 _argv_alloc: std.mem.Allocator = undefined, 59 _argv_alloc: std.mem.Allocator = undefined,
58 _font_family_oom: bool = false, 60 _font_family_oom: bool = false,
59 _target: ?[]const u8 = null, 61 _target: ?[]const u8 = null,
@@ -79,6 +81,14 @@ const Arguments = struct {
79 self._palette_values[pair.index] = pair.color; 81 self._palette_values[pair.index] = pair.color;
80 return 2; 82 return 2;
81 } 83 }
84 if (std.mem.eql(u8, rest[0], "--forward")) {
85 if (rest.len < 2) return 0;
86 const rule = client.forward.Rule.parse(rest[1]) catch return 0;
87 self._forwards.append(self._argv_alloc, rule) catch {
88 self._font_family_oom = true;
89 };
90 return 2;
91 }
82 return 0; 92 return 0;
83 } 93 }
84 }; 94 };
@@ -99,6 +109,7 @@ pub fn main() !u8 {
99 109
100 var o: Arguments = .{ ._argv_alloc = argv_alloc }; 110 var o: Arguments = .{ ._argv_alloc = argv_alloc };
101 defer o._font_families.deinit(argv_alloc); 111 defer o._font_families.deinit(argv_alloc);
112 defer o._forwards.deinit(argv_alloc);
102 cliflags.parseStrict(Arguments, &o, args[1..]) catch |e| return cliflags.exitFor(e, usage, "muxg", std.fmt.comptimePrint("{s} ({s})", .{ @import("build_options").version, @tagName(@import("builtin").mode) })); 113 cliflags.parseStrict(Arguments, &o, args[1..]) catch |e| return cliflags.exitFor(e, usage, "muxg", std.fmt.comptimePrint("{s} ({s})", .{ @import("build_options").version, @tagName(@import("builtin").mode) }));
103 if (o._font_family_oom) return error.OutOfMemory; 114 if (o._font_family_oom) return error.OutOfMemory;
104 if (o.font_px != null and o.font_size != null) { 115 if (o.font_px != null and o.font_size != null) {
@@ -156,6 +167,10 @@ pub fn main() !u8 {
156 std.debug.print("muxg: name one transport: HOST, --sock, --via or quic://\n{s}", .{usage}); 167 std.debug.print("muxg: name one transport: HOST, --sock, --via or quic://\n{s}", .{usage});
157 return 2; 168 return 2;
158 } 169 }
170 if (o._forwards.items.len != 0 and named == 0) {
171 std.debug.print("muxg: --forward requires an explicit TARGET, --sock, or --via\n", .{});
172 return 2;
173 }
159 const cli_points: ?f64 = if (o.font_size) |points| points.value else null; 174 const cli_points: ?f64 = if (o.font_size) |points| points.value else null;
160 const font_points = if (o.font_px != null) null else cli_points orelse settings.size_points; 175 const font_points = if (o.font_px != null) null else cli_points orelse settings.size_points;
161 const font_px = o.font_px orelse 16; 176 const font_px = o.font_px orelse 16;
@@ -190,6 +205,7 @@ pub fn main() !u8 {
190 .font_points = font_points, 205 .font_points = font_points,
191 .appearance = native.theme.merge(native.theme.legacy, selected, explicit), 206 .appearance = native.theme.merge(native.theme.legacy, selected, explicit),
192 .test_fifo = std.posix.getenv("MUXG_TEST_FIFO"), 207 .test_fifo = std.posix.getenv("MUXG_TEST_FIFO"),
208 .forwards = o._forwards.items,
193 }) catch |err| { 209 }) catch |err| {
194 std.debug.print("muxg: {s}\n", .{if (err == error.WorkspaceAlreadyOpen) "the saved workspace is already open" else @errorName(err)}); 210 std.debug.print("muxg: {s}\n", .{if (err == error.WorkspaceAlreadyOpen) "the saved workspace is already open" else @errorName(err)});
195 return 2; 211 return 2;
src/client/buffered_wire.zig
Old New
@@ -1,6 +1,10 @@
1 const std = @import("std"); 1 const std = @import("std");
2 const client = @import("client.zig"); 2 const client = @import("client.zig");
3 const proto = @import("term").protocol; 3 const proto = @import("term").protocol;
4
5 /// Opaque transport bytes held by a forwarding connection, across fragmented
6 /// frames and bursts. Frame parsing remains forwarding-only in this wrapper.
7 pub const transport_queue_max: usize = 1024 * 1024;
4 // Incremental reads and queued writes keep partial stream frames and a 8 // Incremental reads and queued writes keep partial stream frames and a
5 // peer which stops reading from blocking the mailbox or stop(). QUIC keeps 9 // peer which stops reading from blocking the mailbox or stop(). QUIC keeps
6 // its existing framing and outgoing queue in Link. 10 // its existing framing and outgoing queue in Link.
@@ -32,9 +36,23 @@ pub const Wire = struct {
32 .quic => self.tr.pollFd(), 36 .quic => self.tr.pollFd(),
33 }; 37 };
34 } 38 }
39 /// Stream backlog becomes writable through its output fd. QUIC backlog
40 /// advances only when service processes packets or a transport timer, so
41 /// polling its UDP socket for OUT would spin while stream credit is full.
42 pub fn pendingWriteFd(self: *Wire) ?std.posix.fd_t {
43 if (!self.pending()) return null;
44 return switch (self.tr.link) {
45 .fd => |fd| fd,
46 .pipe => |p| p.w,
47 .quic => null,
48 };
49 }
35 pub fn pending(self: *Wire) bool { 50 pub fn pending(self: *Wire) bool {
36 return if (self.tr.link == .quic) self.tr.link.quic.qout.items.len > 0 else self.output.items.len > 0; 51 return if (self.tr.link == .quic) self.tr.link.quic.qout.items.len > 0 else self.output.items.len > 0;
37 } 52 }
53 pub fn pendingBytes(self: *const Wire) usize {
54 return if (self.tr.link == .quic) self.tr.link.quic.qout.items.len else self.output.items.len;
55 }
38 pub fn send(self: *Wire, kind: proto.MsgType, payload: []const u8) !void { 56 pub fn send(self: *Wire, kind: proto.MsgType, payload: []const u8) !void {
39 if (self.tr.link == .quic) return self.tr.writeFrame(kind, payload); 57 if (self.tr.link == .quic) return self.tr.writeFrame(kind, payload);
40 try proto.appendFrame(&self.output, self.alloc, kind, payload); 58 try proto.appendFrame(&self.output, self.alloc, kind, payload);
@@ -50,20 +68,40 @@ pub const Wire = struct {
50 self.output.replaceRangeAssumeCapacity(0, n, &.{}); 68 self.output.replaceRangeAssumeCapacity(0, n, &.{});
51 } 69 }
52 pub fn read(self: *Wire) !client.Incoming { 70 pub fn read(self: *Wire) !client.Incoming {
53 if (self.tr.link == .quic) return self.tr.readFrame(self.alloc); 71 return self.readLimited(proto.frame_header_len + proto.max_payload);
72 }
73 /// Forwarding peers accept much smaller frames than the terminal wire's
74 /// paste-sized maximum. Bound the partial frame too, rather than letting a
75 /// claimed terminal-sized payload quietly become a forwarding queue.
76 pub fn readLimited(self: *Wire, max_buffered: usize) !client.Incoming {
77 if (self.tr.link == .quic) {
78 if (self.tr.link.quic.cl.in.items.len > max_buffered) return error.FrameTooLarge;
79 return self.tr.readFrame(self.alloc);
80 }
54 var need: usize = proto.frame_header_len; 81 var need: usize = proto.frame_header_len;
55 if (self.input.items.len >= proto.frame_header_len) { 82 if (self.input.items.len >= proto.frame_header_len) {
56 const len = std.mem.readInt(u32, self.input.items[1..5], .little); 83 const len = std.mem.readInt(u32, self.input.items[1..5], .little);
57 if (len > proto.max_payload) return error.FrameTooLarge; 84 if (len > proto.max_payload) return error.FrameTooLarge;
58 need += len; 85 need += len;
59 } 86 }
87 if (need > max_buffered) return error.FrameTooLarge;
88 if (self.input.items.len >= need) {
89 if (try proto.takeFrame(self.alloc, &self.input)) |frame| return .{ .frame = frame };
90 unreachable;
91 }
92 if (self.input.items.len >= max_buffered) return error.FrameTooLarge;
60 var buf: [64 * 1024]u8 = undefined; 93 var buf: [64 * 1024]u8 = undefined;
61 const n = std.posix.read(self.tr.pollFd(), buf[0..@min(buf.len, need - self.input.items.len)]) catch |err| switch (err) { 94 const room = max_buffered - self.input.items.len;
95 const n = std.posix.read(self.tr.pollFd(), buf[0..@min(buf.len, @min(need - self.input.items.len, room))]) catch |err| switch (err) {
62 error.WouldBlock => return .incomplete, 96 error.WouldBlock => return .incomplete,
63 else => return err, 97 else => return err,
64 }; 98 };
65 if (n == 0) return .closed; 99 if (n == 0) return .closed;
66 try self.input.appendSlice(self.alloc, buf[0..n]); 100 try self.input.appendSlice(self.alloc, buf[0..n]);
101 if (self.input.items.len >= proto.frame_header_len) {
102 const staged_len = std.mem.readInt(u32, self.input.items[1..5], .little);
103 if (staged_len > proto.max_payload or proto.frame_header_len + staged_len > max_buffered) return error.FrameTooLarge;
104 }
67 if (try proto.takeFrame(self.alloc, &self.input)) |frame| return .{ .frame = frame }; 105 if (try proto.takeFrame(self.alloc, &self.input)) |frame| return .{ .frame = frame };
68 return .incomplete; 106 return .incomplete;
69 } 107 }
@@ -74,3 +112,65 @@ fn nonblocking(fd: std.posix.fd_t) !void {
74 const bits: u32 = @bitCast(std.posix.O{ .NONBLOCK = true }); 112 const bits: u32 = @bitCast(std.posix.O{ .NONBLOCK = true });
75 _ = try std.posix.fcntl(fd, std.posix.F.SETFL, flags | bits); 113 _ = try std.posix.fcntl(fd, std.posix.F.SETFL, flags | bits);
76 } 114 }
115
116 test "limited wire rejects a claimed forwarding queue before reading its body" {
117 var pair: [2]std.posix.fd_t = undefined;
118 try std.testing.expectEqual(@as(c_int, 0), std.c.socketpair(std.posix.AF.UNIX, std.posix.SOCK.STREAM, 0, &pair));
119 defer std.posix.close(pair[1]);
120 var tr: client.Transport = .{ .link = .{ .fd = pair[0] } };
121 defer tr.close();
122 var wire = try Wire.init(std.testing.allocator, &tr);
123 defer wire.deinit();
124
125 const header = proto.encodeHeader(.forward_data, 1024);
126 try proto.writeAllFd(pair[1], &header);
127 try std.testing.expectError(error.FrameTooLarge, wire.readLimited(512));
128 }
129
130 test "forwarding budget admits a maximum legal forwarding frame" {
131 var pair: [2]std.posix.fd_t = undefined;
132 try std.testing.expectEqual(@as(c_int, 0), std.c.socketpair(std.posix.AF.UNIX, std.posix.SOCK.STREAM, 0, &pair));
133 defer std.posix.close(pair[1]);
134 var tr: client.Transport = .{ .link = .{ .fd = pair[0] } };
135 defer tr.close();
136 var wire = try Wire.init(std.testing.allocator, &tr);
137 defer wire.deinit();
138
139 var payload: [proto.forward_id_len + proto.forward_data_max]u8 = undefined;
140 try proto.writeFrame(pair[1], .forward_data, &payload);
141 const frame = while (true) switch (try wire.readLimited(transport_queue_max)) {
142 .frame => |frame| break frame,
143 .incomplete => continue,
144 .closed => return error.UnexpectedEof,
145 };
146 defer frame.deinit(std.testing.allocator);
147 try std.testing.expectEqual(proto.MsgType.forward_data, frame.type);
148 try std.testing.expectEqual(payload.len, frame.payload.len);
149 try std.testing.expect(proto.frame_header_len + frame.payload.len <= transport_queue_max);
150 }
151
152 test "only stream backlog requests write readiness" {
153 const alloc = std.testing.allocator;
154 const pair = blk: {
155 var fds: [2]std.posix.fd_t = undefined;
156 try std.testing.expectEqual(@as(c_int, 0), std.c.socketpair(std.posix.AF.UNIX, std.posix.SOCK.STREAM, 0, &fds));
157 break :blk fds;
158 };
159 defer std.posix.close(pair[0]);
160 defer std.posix.close(pair[1]);
161
162 var stream_tr: client.Transport = .{ .link = .{ .fd = pair[0] } };
163 var stream_wire: Wire = .{ .alloc = alloc, .tr = &stream_tr };
164 defer stream_wire.deinit();
165 try std.testing.expectEqual(@as(?std.posix.fd_t, null), stream_wire.pendingWriteFd());
166 try stream_wire.output.append(alloc, 1);
167 try std.testing.expectEqual(@as(?std.posix.fd_t, pair[0]), stream_wire.pendingWriteFd());
168
169 var quic_tr: client.Transport = .{ .link = .{ .quic = .{ .cl = undefined, .alloc = alloc } } };
170 defer quic_tr.link.quic.qout.deinit(alloc);
171 var quic_wire: Wire = .{ .alloc = alloc, .tr = &quic_tr };
172 defer quic_wire.deinit();
173 try quic_tr.link.quic.qout.append(alloc, 1);
174 try std.testing.expect(quic_wire.pending());
175 try std.testing.expectEqual(@as(?std.posix.fd_t, null), quic_wire.pendingWriteFd());
176 }
src/client/client.zig
Old New
@@ -36,6 +36,7 @@ pub const discovery = @import("discovery.zig");
36 pub const resolver = @import("resolver.zig"); 36 pub const resolver = @import("resolver.zig");
37 const open_wait = @import("open_wait.zig"); 37 const open_wait = @import("open_wait.zig");
38 pub const session_pump = @import("session_pump.zig"); 38 pub const session_pump = @import("session_pump.zig");
39 pub const forward = @import("forward.zig");
39 40
40 /// A session name held by value. The names a switch travels on are decoded 41 /// A session name held by value. The names a switch travels on are decoded
41 /// out of a frame payload that is freed before the re-dial, so they cannot 42 /// out of a frame payload that is freed before the re-dial, so they cannot
@@ -420,9 +421,13 @@ pub const Transport = struct {
420 budget_ms: u32, 421 budget_ms: u32,
421 carry: ?*std.ArrayList(u8), 422 carry: ?*std.ArrayList(u8),
422 abort_fd: std.posix.fd_t, 423 abort_fd: std.posix.fd_t,
424 inbound_cap: ?usize,
423 ) !Transport { 425 ) !Transport {
424 const cl = try quic.Client.connect(alloc, addr, key, idle_ms); 426 const cl = try quic.Client.connect(alloc, addr, key, idle_ms);
425 errdefer cl.deinit(); 427 errdefer cl.deinit();
428 // The first handshake pump can already deliver stream data. Set a
429 // role's opaque bound before waitReady, not after its first frame.
430 cl.inbound_cap = inbound_cap;
426 try waitReady(cl, budget_ms, alloc, carry, abort_fd); 431 try waitReady(cl, budget_ms, alloc, carry, abort_fd);
427 return .{ .link = .{ .quic = .{ .cl = cl, .alloc = alloc } } }; 432 return .{ .link = .{ .quic = .{ .cl = cl, .alloc = alloc } } };
428 } 433 }
@@ -444,12 +449,22 @@ pub const Transport = struct {
444 /// can be answered against the one dial it belongs to. 449 /// can be answered against the one dial it belongs to.
445 dial: ?*handoff.Dial, 450 dial: ?*handoff.Dial,
446 ) !Transport { 451 ) !Transport {
447 return openUntil(alloc, target, carry, abort_fd, dial, null); 452 return openUntilBounded(alloc, target, carry, abort_fd, dial, null, null);
448 } 453 }
449 454
450 /// An absolute budget shared by DNS, connect, handshake, SSH fallback and 455 /// An absolute budget shared by DNS, connect, handshake, SSH fallback and
451 /// the caller's later request. Null preserves the legacy opening policy. 456 /// the caller's later request. Null preserves the legacy opening policy.
452 pub fn openUntil(alloc: std.mem.Allocator, target: Target, carry: ?*std.ArrayList(u8), abort_fd: std.posix.fd_t, dial: ?*handoff.Dial, deadline: ?i64) !Transport { 457 pub fn openUntil(alloc: std.mem.Allocator, target: Target, carry: ?*std.ArrayList(u8), abort_fd: std.posix.fd_t, dial: ?*handoff.Dial, deadline: ?i64) !Transport {
458 return openUntilBounded(alloc, target, carry, abort_fd, dial, deadline, null);
459 }
460
461 /// `inbound_cap` is an opaque transport receive budget for roles that
462 /// cannot retain terminal-sized traffic. Null preserves terminal policy.
463 pub fn openBounded(alloc: std.mem.Allocator, target: Target, carry: ?*std.ArrayList(u8), abort_fd: std.posix.fd_t, dial: ?*handoff.Dial, inbound_cap: ?usize) !Transport {
464 return openUntilBounded(alloc, target, carry, abort_fd, dial, null, inbound_cap);
465 }
466
467 fn openUntilBounded(alloc: std.mem.Allocator, target: Target, carry: ?*std.ArrayList(u8), abort_fd: std.posix.fd_t, dial: ?*handoff.Dial, deadline: ?i64, inbound_cap: ?usize) !Transport {
453 var wait: open_wait.Wait = .{ .alloc = alloc, .abort_fd = abort_fd, .carry = carry, .deadline = deadline }; 468 var wait: open_wait.Wait = .{ .alloc = alloc, .abort_fd = abort_fd, .carry = carry, .deadline = deadline };
454 // Local first attach has always left stdin to the established client: 469 // Local first attach has always left stdin to the established client:
455 // even an already queued detach must attach before handling its chord. 470 // even an already queued detach must attach before handling its chord.
@@ -459,12 +474,12 @@ pub const Transport = struct {
459 switch (target) { 474 switch (target) {
460 // Delegated whole, because the handoff can end up producing 475 // Delegated whole, because the handoff can end up producing
461 // either of the two links below and owns the choice itself. 476 // either of the two links below and owns the choice itself.
462 .hand => |h| return openHandoffWait(alloc, h, carry, abort_fd, dial, &wait), 477 .hand => |h| return openHandoffWait(alloc, h, carry, abort_fd, dial, &wait, inbound_cap),
463 .quic => |q| { 478 .quic => |q| {
464 const key = try quic.Key.load(q.key_path); 479 const key = try quic.Key.load(q.key_path);
465 const hp = try quic.splitHostPort(q.host_port); 480 const hp = try quic.splitHostPort(q.host_port);
466 const addr = try resolveOpening(alloc, hp.host, hp.port, q.deadline_ms, &wait); 481 const addr = try resolveOpening(alloc, hp.host, hp.port, q.deadline_ms, &wait);
467 return quicTransport(alloc, addr, key, q.idle_ms, try wait.remaining(q.deadline_ms), carry, abort_fd); 482 return quicTransport(alloc, addr, key, q.idle_ms, try wait.remaining(q.deadline_ms), carry, abort_fd, inbound_cap);
468 }, 483 },
469 .via => |cmd| { 484 .via => |cmd| {
470 const argv = try viaArgv(alloc, cmd); 485 const argv = try viaArgv(alloc, cmd);
@@ -492,7 +507,7 @@ pub const Transport = struct {
492 dial: ?*handoff.Dial, 507 dial: ?*handoff.Dial,
493 ) !Transport { 508 ) !Transport {
494 var wait: open_wait.Wait = .{ .alloc = alloc, .abort_fd = abort_fd, .carry = carry }; 509 var wait: open_wait.Wait = .{ .alloc = alloc, .abort_fd = abort_fd, .carry = carry };
495 return openHandoffWait(alloc, h, carry, abort_fd, dial, &wait); 510 return openHandoffWait(alloc, h, carry, abort_fd, dial, &wait, null);
496 } 511 }
497 fn checkHandoffWait(wait: *open_wait.Wait) !void { 512 fn checkHandoffWait(wait: *open_wait.Wait) !void {
498 // First-attach SSH owns cooked stdin for passwords. Only an actual 513 // First-attach SSH owns cooked stdin for passwords. Only an actual
@@ -503,7 +518,7 @@ pub const Transport = struct {
503 try deadline_only.check(); 518 try deadline_only.check();
504 } else try wait.check(); 519 } else try wait.check();
505 } 520 }
506 fn openHandoffWait(alloc: std.mem.Allocator, h: HandoffTarget, carry: ?*std.ArrayList(u8), abort_fd: std.posix.fd_t, dial: ?*handoff.Dial, wait: *open_wait.Wait) !Transport { 521 fn openHandoffWait(alloc: std.mem.Allocator, h: HandoffTarget, carry: ?*std.ArrayList(u8), abort_fd: std.posix.fd_t, dial: ?*handoff.Dial, wait: *open_wait.Wait, inbound_cap: ?usize) !Transport {
507 // The ORDER is `handoff.next`'s; this loop performs the step it is 522 // The ORDER is `handoff.next`'s; this loop performs the step it is
508 // handed and reports what came of it. 523 // handed and reports what came of it.
509 var st: handoff.State = .{ 524 var st: handoff.State = .{
@@ -546,7 +561,7 @@ pub const Transport = struct {
546 const outcome: handoff.Outcome = switch (step) { 561 const outcome: handoff.Outcome = switch (step) {
547 .dial_quic => |ep| blk: { 562 .dial_quic => |ep| blk: {
548 dialed = ep; 563 dialed = ep;
549 if (openQuicEndpointWait(alloc, h, ep, carry, abort_fd, wait)) |t| { 564 if (openQuicEndpointWait(alloc, h, ep, carry, abort_fd, wait, inbound_cap)) |t| {
550 quic_t = t; 565 quic_t = t;
551 break :blk .ok; 566 break :blk .ok;
552 } else |err| { 567 } else |err| {
@@ -658,12 +673,12 @@ pub const Transport = struct {
658 abort_fd: std.posix.fd_t, 673 abort_fd: std.posix.fd_t,
659 ) !Transport { 674 ) !Transport {
660 var wait: open_wait.Wait = .{ .alloc = alloc, .abort_fd = abort_fd, .carry = carry }; 675 var wait: open_wait.Wait = .{ .alloc = alloc, .abort_fd = abort_fd, .carry = carry };
661 return openQuicEndpointWait(alloc, h, ep, carry, abort_fd, &wait); 676 return openQuicEndpointWait(alloc, h, ep, carry, abort_fd, &wait, null);
662 } 677 }
663 fn openQuicEndpointWait(alloc: std.mem.Allocator, h: HandoffTarget, ep: handoff.Endpoint, carry: ?*std.ArrayList(u8), abort_fd: std.posix.fd_t, wait: *open_wait.Wait) !Transport { 678 fn openQuicEndpointWait(alloc: std.mem.Allocator, h: HandoffTarget, ep: handoff.Endpoint, carry: ?*std.ArrayList(u8), abort_fd: std.posix.fd_t, wait: *open_wait.Wait, inbound_cap: ?usize) !Transport {
664 const addr = try resolveOpening(alloc, handoff.dialHost(h.host), ep.port, h.deadline_ms, wait); 679 const addr = try resolveOpening(alloc, handoff.dialHost(h.host), ep.port, h.deadline_ms, wait);
665 const key = quic.Key{ .bytes = ep.key }; 680 const key = quic.Key{ .bytes = ep.key };
666 return quicTransport(alloc, addr, key, h.idle_ms, try wait.remaining(h.deadline_ms), carry, abort_fd); 681 return quicTransport(alloc, addr, key, h.idle_ms, try wait.remaining(h.deadline_ms), carry, abort_fd, inbound_cap);
667 } 682 }
668 fn resolveOpening(alloc: std.mem.Allocator, host: []const u8, port: u16, budget_ms: u32, wait: *open_wait.Wait) !std.net.Address { 683 fn resolveOpening(alloc: std.mem.Allocator, host: []const u8, port: u16, budget_ms: u32, wait: *open_wait.Wait) !std.net.Address {
669 if (wait.abort_fd < 0 and wait.deadline == null) return quic.resolveHost(alloc, host, port); 684 if (wait.abort_fd < 0 and wait.deadline == null) return quic.resolveHost(alloc, host, port);
src/client/forward.zig
Old New
@@ -0,0 +1,943 @@
1 //! Process-local TCP listener manager for mux forwarding-role connections.
2 //! Listeners survive transport reconnects; individual accepted streams do not.
3 const std = @import("std");
4 const client = @import("client.zig");
5 const proto = @import("term").protocol;
6 const Wire = @import("buffered_wire.zig").Wire;
7 const client_os = @import("client_os");
8
9 pub const max_rules: usize = 16;
10 const transport_queue_max = @import("buffered_wire.zig").transport_queue_max;
11 const control_reserve: usize = 64 * 1024;
12 const channel_queue_max: usize = proto.forward_initial_credit;
13 const handshake_ms: i64 = 2000;
14 const reconnect_pause_ms: i32 = 100;
15
16 pub const Rule = struct {
17 local_port: u16,
18 remote_port: u16,
19
20 pub fn parse(text: []const u8) !Rule {
21 const colon = std.mem.indexOfScalar(u8, text, ':') orelse return error.InvalidForward;
22 if (colon == 0 or colon + 1 == text.len or std.mem.indexOfScalarPos(u8, text, colon + 1, ':') != null) return error.InvalidForward;
23 for (text[0..colon]) |byte| if (byte < '0' or byte > '9') return error.InvalidForward;
24 for (text[colon + 1 ..]) |byte| if (byte < '0' or byte > '9') return error.InvalidForward;
25 const local = std.fmt.parseInt(u16, text[0..colon], 10) catch return error.InvalidForward;
26 const remote = std.fmt.parseInt(u16, text[colon + 1 ..], 10) catch return error.InvalidForward;
27 if (local == 0 or remote == 0) return error.InvalidForward;
28 return .{ .local_port = local, .remote_port = remote };
29 }
30
31 pub fn eql(a: Rule, b: Rule) bool {
32 return a.local_port == b.local_port and a.remote_port == b.remote_port;
33 }
34 };
35
36 const Listener = struct { fd: std.posix.fd_t, rule: Rule };
37 const Channel = struct {
38 id: u32,
39 fd: std.posix.fd_t,
40 opening: bool = true,
41 send_credit: u32 = 0,
42 recv_credit: u32 = 0,
43 to_local: std.ArrayList(u8) = .empty,
44 local_eof: bool = false,
45 peer_eof: bool = false,
46 write_shutdown: bool = false,
47 };
48
49 pub const Manager = struct {
50 alloc: std.mem.Allocator,
51 arena: std.heap.ArenaAllocator,
52 target: client.Target,
53 listeners: [max_rules]?Listener = @splat(null),
54 listener_len: usize = 0,
55 channels: [proto.forward_channels_max]?Channel = @splat(null),
56 next_id: u32 = 1,
57 wake_pipe: [2]std.posix.fd_t,
58 closing: std.atomic.Value(bool) = .init(false),
59 thread: ?std.Thread = null,
60
61 /// Binds every unique listener before returning, so an explicit collision
62 /// is a synchronous CLI failure rather than a background warning.
63 pub fn init(alloc: std.mem.Allocator, source: client.Target, rules: []const Rule) !*Manager {
64 if (rules.len == 0) return error.InvalidForward;
65 const self = try alloc.create(Manager);
66 errdefer alloc.destroy(self);
67 var arena = std.heap.ArenaAllocator.init(alloc);
68 errdefer arena.deinit();
69 var target = try client.discovery.cloneTarget(arena.allocator(), source);
70 if (target == .hand) {
71 target.hand.asked = false;
72 target.hand.narrate = false;
73 }
74 const wake = try std.posix.pipe2(.{ .NONBLOCK = true, .CLOEXEC = true });
75 errdefer {
76 std.posix.close(wake[0]);
77 std.posix.close(wake[1]);
78 }
79 self.* = .{ .alloc = alloc, .arena = arena, .target = target, .wake_pipe = wake };
80 errdefer self.closeListeners();
81 for (rules) |rule| {
82 var duplicate = false;
83 for (self.listeners[0..self.listener_len]) |slot| if (slot != null and Rule.eql(slot.?.rule, rule)) {
84 duplicate = true;
85 break;
86 };
87 if (duplicate) continue;
88 if (self.listener_len == max_rules) return error.TooManyForwards;
89 // Two different destinations cannot own the same local port.
90 for (self.listeners[0..self.listener_len]) |slot| if (slot != null and slot.?.rule.local_port == rule.local_port)
91 return error.ForwardConflict;
92 self.listeners[self.listener_len] = .{ .fd = try bindLoopback(rule.local_port), .rule = rule };
93 self.listener_len += 1;
94 }
95 return self;
96 }
97
98 pub fn start(self: *Manager) !void {
99 if (self.thread != null) return;
100 self.thread = try std.Thread.spawn(.{}, entry, .{self});
101 }
102
103 pub fn stop(self: *Manager) void {
104 if (self.closing.swap(true, .acq_rel)) return;
105 _ = std.posix.write(self.wake_pipe[1], &.{1}) catch {};
106 if (self.thread) |thread| thread.join();
107 self.dropChannels(null);
108 self.closeListeners();
109 std.posix.close(self.wake_pipe[0]);
110 std.posix.close(self.wake_pipe[1]);
111 self.arena.deinit();
112 const alloc = self.alloc;
113 alloc.destroy(self);
114 }
115
116 pub fn matches(self: *const Manager, target: client.Target) bool {
117 return targetEqual(self.target, target);
118 }
119
120 fn closeListeners(self: *Manager) void {
121 for (&self.listeners) |*slot| if (slot.*) |listener| {
122 std.posix.close(listener.fd);
123 slot.* = null;
124 };
125 self.listener_len = 0;
126 }
127
128 fn entry(self: *Manager) void {
129 var warned = false;
130 while (!self.closing.load(.acquire)) {
131 var job = DialJob.init(self.alloc, self.target) catch return;
132 defer job.deinit();
133 job.start() catch return;
134 while (!job.done.load(.acquire) and !self.closing.load(.acquire)) self.refusePendingAccepts(50);
135 if (self.closing.load(.acquire)) job.cancel();
136 job.join();
137 var tr = job.take() orelse {
138 if (!self.pauseAfterFailure()) return;
139 continue;
140 };
141 defer tr.close();
142 var wire = Wire.init(self.alloc, &tr) catch {
143 if (!self.pauseAfterFailure()) return;
144 continue;
145 };
146 defer wire.deinit();
147 if (!self.handshake(&wire)) {
148 if (!self.closing.load(.acquire) and !warned) {
149 std.debug.print("muxg: daemon does not support native port forwarding\n", .{});
150 warned = true;
151 }
152 self.dropChannels(null);
153 if (!self.pauseAfterFailure()) return;
154 continue;
155 }
156 warned = false;
157 self.connected(&wire);
158 self.dropChannels(null);
159 if (!self.closing.load(.acquire) and !self.pauseAfterFailure()) return;
160 }
161 }
162
163 fn handshake(self: *Manager, wire: *Wire) bool {
164 const hello = proto.encodeForwardHello();
165 wire.send(.forward_hello, &hello) catch return false;
166 const until = std.time.milliTimestamp() + handshake_ms;
167 while (!self.closing.load(.acquire) and std.time.milliTimestamp() < until) {
168 wire.tr.service();
169 wire.flush() catch return false;
170 switch (wire.readLimited(transport_queue_max) catch return false) {
171 .frame => |frame| {
172 defer frame.deinit(self.alloc);
173 if (frame.type != .forward_ready) return false;
174 const version = proto.decodeForwardHello(frame.payload) catch return false;
175 return version == proto.forward_version;
176 },
177 .closed => return false,
178 .incomplete => {},
179 }
180 var fds = [_]std.posix.pollfd{
181 .{ .fd = wire.tr.pollFd(), .events = std.posix.POLL.IN, .revents = 0 },
182 .{ .fd = wire.pendingWriteFd() orelse -1, .events = std.posix.POLL.OUT, .revents = 0 },
183 .{ .fd = self.wake_pipe[0], .events = std.posix.POLL.IN, .revents = 0 },
184 .{ .fd = wire.tr.errFd() orelse -1, .events = std.posix.POLL.IN, .revents = 0 },
185 };
186 _ = std.posix.poll(&fds, wire.tr.timeoutMs(50)) catch return false;
187 if (fds[3].revents != 0) wire.tr.drainErr();
188 self.refuseReadyAccepts();
189 }
190 return false;
191 }
192
193 fn connected(self: *Manager, wire: *Wire) void {
194 while (!self.closing.load(.acquire)) {
195 wire.tr.service();
196 wire.flush() catch return;
197 if (wire.pendingBytes() > transport_queue_max) return;
198 var fds: [4 + max_rules + proto.forward_channels_max]std.posix.pollfd = undefined;
199 fds[0] = .{ .fd = wire.tr.pollFd(), .events = std.posix.POLL.IN, .revents = 0 };
200 fds[1] = .{ .fd = wire.pendingWriteFd() orelse -1, .events = std.posix.POLL.OUT, .revents = 0 };
201 fds[2] = .{ .fd = self.wake_pipe[0], .events = std.posix.POLL.IN, .revents = 0 };
202 fds[3] = .{ .fd = wire.tr.errFd() orelse -1, .events = std.posix.POLL.IN, .revents = 0 };
203 const listener_base = 4;
204 for (0..max_rules) |i| fds[listener_base + i] = .{ .fd = if (self.listeners[i]) |l| l.fd else -1, .events = std.posix.POLL.IN, .revents = 0 };
205 const chan_base = listener_base + max_rules;
206 for (0..proto.forward_channels_max) |i| {
207 if (self.channels[i]) |ch| {
208 var events: i16 = 0;
209 if (!ch.opening and !ch.local_eof and self.dataReadLimit(wire, ch) != 0) events |= std.posix.POLL.IN;
210 if (ch.to_local.items.len != 0 or (ch.peer_eof and !ch.write_shutdown)) events |= std.posix.POLL.OUT;
211 fds[chan_base + i] = .{ .fd = if (events == 0) -1 else ch.fd, .events = events, .revents = 0 };
212 } else fds[chan_base + i] = .{ .fd = -1, .events = 0, .revents = 0 };
213 }
214 _ = std.posix.poll(&fds, wire.tr.timeoutMs(50)) catch return;
215 if (fds[2].revents != 0) return;
216 if (fds[3].revents != 0) wire.tr.drainErr();
217 if (fds[0].revents != 0 or wire.tr.link == .quic) if (!self.receive(wire)) return;
218 if (fds[1].revents != 0) wire.flush() catch return;
219 for (0..max_rules) |i| if (fds[listener_base + i].revents != 0 and !self.acceptOne(wire, i)) return;
220 for (0..proto.forward_channels_max) |i| {
221 if (self.channels[i] == null or fds[chan_base + i].revents == 0) continue;
222 const revents = fds[chan_base + i].revents;
223 if (revents & std.posix.POLL.OUT != 0 and !self.flushLocal(wire, i)) return;
224 if (self.channels[i] != null and revents & ~@as(i16, std.posix.POLL.OUT) != 0 and !self.readLocal(wire, i)) return;
225 }
226 }
227 }
228
229 fn receive(self: *Manager, wire: *Wire) bool {
230 for (0..64) |_| switch (wire.readLimited(transport_queue_max) catch return false) {
231 .closed => return false,
232 .incomplete => return true,
233 .frame => |frame| {
234 defer frame.deinit(self.alloc);
235 if (!self.handleFrame(wire, frame)) return false;
236 },
237 };
238 return true;
239 }
240
241 fn handleFrame(self: *Manager, wire: *Wire, frame: proto.Frame) bool {
242 switch (frame.type) {
243 .forward_open_result => {
244 const result = proto.decodeForwardOpenResult(frame.payload) catch return false;
245 const ci = self.findChannel(result.id) orelse return true;
246 if (!result.ok) {
247 _ = self.dropChannel(ci, false, wire);
248 return true;
249 }
250 const ch = &self.channels[ci].?;
251 ch.opening = false;
252 ch.recv_credit = proto.forward_initial_credit;
253 const credit = proto.encodeForwardCredit(.{ .id = ch.id, .amount = proto.forward_initial_credit });
254 self.send(wire, .forward_credit, &credit) catch return false;
255 },
256 .forward_credit => {
257 const credit = proto.decodeForwardCredit(frame.payload) catch return false;
258 const ci = self.findChannel(credit.id) orelse return true;
259 const ch = &self.channels[ci].?;
260 if (credit.amount > proto.forward_initial_credit -| ch.send_credit) return false;
261 ch.send_credit += credit.amount;
262 },
263 .forward_data => {
264 if (proto.forwardDataOversize(frame.payload)) return false;
265 const id = proto.decodeForwardId(frame.payload) catch return false;
266 const ci = self.findChannel(id) orelse return true;
267 const ch = &self.channels[ci].?;
268 const data = frame.payload[proto.forward_id_len..];
269 if (ch.peer_eof or data.len > ch.recv_credit or ch.to_local.items.len + data.len > channel_queue_max) {
270 return self.dropChannel(ci, true, wire);
271 }
272 ch.recv_credit -= @intCast(data.len);
273 ch.to_local.appendSlice(self.alloc, data) catch return self.dropChannel(ci, true, wire);
274 if (self.channels[ci] != null and !self.flushLocal(wire, ci)) return false;
275 },
276 .forward_half_close => {
277 const id = proto.decodeForwardId(frame.payload) catch return false;
278 if (frame.payload.len != proto.forward_id_len) return false;
279 const ci = self.findChannel(id) orelse return true;
280 self.channels[ci].?.peer_eof = true;
281 if (!self.flushLocal(wire, ci)) return false;
282 },
283 .forward_reset => {
284 const id = proto.decodeForwardId(frame.payload) catch return false;
285 if (frame.payload.len != proto.forward_id_len) return false;
286 if (self.findChannel(id)) |ci| _ = self.dropChannel(ci, false, wire);
287 },
288 else => return false,
289 }
290 return true;
291 }
292
293 fn acceptOne(self: *Manager, wire: *Wire, li: usize) bool {
294 const listener = self.listeners[li] orelse return true;
295 const fd = std.posix.accept(listener.fd, null, null, std.posix.SOCK.NONBLOCK | std.posix.SOCK.CLOEXEC) catch return true;
296 var ci: ?usize = null;
297 for (self.channels, 0..) |slot, i| if (slot == null) {
298 ci = i;
299 break;
300 };
301 const at = ci orelse {
302 std.posix.close(fd);
303 return true;
304 };
305 const id = self.next_id;
306 if (id == std.math.maxInt(u32)) {
307 std.posix.close(fd);
308 return true;
309 }
310 self.next_id += 1;
311 self.channels[at] = .{ .id = id, .fd = fd };
312 const payload = proto.encodeForwardOpen(.{ .id = id, .port = listener.rule.remote_port });
313 self.send(wire, .forward_open, &payload) catch {
314 _ = self.dropChannel(at, false, wire);
315 return false;
316 };
317 return true;
318 }
319
320 fn readLocal(self: *Manager, wire: *Wire, ci: usize) bool {
321 const ch = &self.channels[ci].?;
322 if (ch.local_eof or ch.send_credit == 0) return true;
323 // Every descriptor in a poll snapshot may have observed the same
324 // shared headroom. Recompute immediately before consuming TCP bytes.
325 const want = self.dataReadLimit(wire, ch.*);
326 if (want == 0) return true;
327 var buf: [proto.forward_data_max]u8 = undefined;
328 const n = std.posix.read(ch.fd, buf[0..want]) catch |err| switch (err) {
329 error.WouldBlock => return true,
330 else => return self.dropChannel(ci, true, wire),
331 };
332 if (n == 0) {
333 ch.local_eof = true;
334 const id = proto.encodeForwardId(ch.id);
335 self.send(wire, .forward_half_close, &id) catch return false;
336 if (self.channels[ci] == null) return true;
337 return self.maybeFinish(ci, wire);
338 }
339 ch.send_credit -= @intCast(n);
340 var payload: [proto.forward_id_len + proto.forward_data_max]u8 = undefined;
341 std.mem.writeInt(u32, payload[0..4], ch.id, .little);
342 @memcpy(payload[4..][0..n], buf[0..n]);
343 self.send(wire, .forward_data, payload[0 .. 4 + n]) catch return false;
344 return true;
345 }
346
347 fn flushLocal(self: *Manager, wire: *Wire, ci: usize) bool {
348 const ch = &self.channels[ci].?;
349 if (ch.to_local.items.len != 0) {
350 const n = client_os.sendNoSigNoWait(ch.fd, ch.to_local.items) catch |err| switch (err) {
351 error.WouldBlock => return true,
352 else => return self.dropChannel(ci, true, wire),
353 };
354 ch.to_local.replaceRangeAssumeCapacity(0, n, &.{});
355 ch.recv_credit += @intCast(n);
356 const credit = proto.encodeForwardCredit(.{ .id = ch.id, .amount = @intCast(n) });
357 self.send(wire, .forward_credit, &credit) catch return false;
358 }
359 if (self.channels[ci] == null) return true;
360 const current = &self.channels[ci].?;
361 if (current.peer_eof and current.to_local.items.len == 0 and !current.write_shutdown) {
362 std.posix.shutdown(current.fd, .send) catch {};
363 current.write_shutdown = true;
364 }
365 return self.maybeFinish(ci, wire);
366 }
367
368 fn maybeFinish(self: *Manager, ci: usize, wire: *Wire) bool {
369 if (self.channels[ci] == null) return true;
370 const ch = &self.channels[ci].?;
371 if (ch.local_eof and ch.peer_eof and ch.to_local.items.len == 0) return self.dropChannel(ci, false, wire);
372 return true;
373 }
374
375 fn findChannel(self: *const Manager, id: u32) ?usize {
376 for (self.channels, 0..) |ch, i| if (ch != null and ch.?.id == id) return i;
377 return null;
378 }
379
380 fn dropChannel(self: *Manager, ci: usize, notify: bool, wire: ?*Wire) bool {
381 if (self.channels[ci]) |*ch| {
382 const id = ch.id;
383 std.posix.close(ch.fd);
384 ch.to_local.deinit(self.alloc);
385 self.channels[ci] = null;
386 if (notify) if (wire) |w| {
387 const payload = proto.encodeForwardId(id);
388 self.send(w, .forward_reset, &payload) catch return false;
389 };
390 }
391 return true;
392 }
393
394 fn dropChannels(self: *Manager, wire: ?*Wire) void {
395 for (0..proto.forward_channels_max) |i| _ = self.dropChannel(i, false, wire);
396 }
397
398 fn send(_: *Manager, wire: *Wire, kind: proto.MsgType, payload: []const u8) !void {
399 if (wire.pendingBytes() + proto.frame_header_len + payload.len > transport_queue_max) return error.ForwardQueueFull;
400 try wire.send(kind, payload);
401 }
402
403 fn dataReadLimit(_: *const Manager, wire: *const Wire, ch: Channel) usize {
404 const overhead = proto.frame_header_len + proto.forward_id_len;
405 const pending = wire.pendingBytes();
406 if (pending >= transport_queue_max - control_reserve - overhead) return 0;
407 const room = transport_queue_max - control_reserve - overhead - pending;
408 return @min(proto.forward_data_max, @min(@as(usize, ch.send_credit), room));
409 }
410
411 /// Every failed setup, including a peer that hangs up immediately after
412 /// dialling, waits through this interruptible pause. This prevents a
413 /// rejected peer from spinning a reconnect loop while listeners continue
414 /// to refuse new local accepts and stop still wakes promptly.
415 fn pauseAfterFailure(self: *Manager) bool {
416 // Listener readability is expected while a transport is down. It may
417 // only refuse bounded work; it must not restart this retry deadline.
418 const deadline = monotonicMs() + reconnect_pause_ms;
419 while (!self.closing.load(.acquire)) {
420 const remaining = deadline - monotonicMs();
421 if (remaining <= 0) return true;
422 var fds: [1 + max_rules]std.posix.pollfd = undefined;
423 fds[0] = .{ .fd = self.wake_pipe[0], .events = std.posix.POLL.IN, .revents = 0 };
424 for (0..max_rules) |i| fds[1 + i] = .{ .fd = if (self.listeners[i]) |l| l.fd else -1, .events = std.posix.POLL.IN, .revents = 0 };
425 _ = std.posix.poll(&fds, @intCast(@min(remaining, @as(i64, std.math.maxInt(i32))))) catch return false;
426 if (fds[0].revents != 0 or self.closing.load(.acquire)) return false;
427 // At most one accept per wake: a flood cannot consume the pause.
428 for (0..max_rules) |i| if (fds[1 + i].revents != 0) {
429 const listener = self.listeners[i] orelse continue;
430 const fd = std.posix.accept(listener.fd, null, null, std.posix.SOCK.NONBLOCK | std.posix.SOCK.CLOEXEC) catch continue;
431 std.posix.close(fd);
432 break;
433 };
434 }
435 return false;
436 }
437
438 fn refusePendingAccepts(self: *Manager, timeout_ms: i32) void {
439 var fds: [1 + max_rules]std.posix.pollfd = undefined;
440 fds[0] = .{ .fd = self.wake_pipe[0], .events = std.posix.POLL.IN, .revents = 0 };
441 for (0..max_rules) |i| fds[1 + i] = .{ .fd = if (self.listeners[i]) |l| l.fd else -1, .events = std.posix.POLL.IN, .revents = 0 };
442 _ = std.posix.poll(&fds, timeout_ms) catch return;
443 for (0..max_rules) |i| if (fds[1 + i].revents != 0) {
444 const listener = self.listeners[i] orelse continue;
445 const fd = std.posix.accept(listener.fd, null, null, std.posix.SOCK.NONBLOCK | std.posix.SOCK.CLOEXEC) catch continue;
446 std.posix.close(fd);
447 };
448 }
449
450 fn refuseReadyAccepts(self: *Manager) void {
451 self.refusePendingAccepts(0);
452 }
453 };
454
455 const DialJob = struct {
456 alloc: std.mem.Allocator,
457 target: client.Target,
458 cancel_pipe: [2]std.posix.fd_t,
459 done: std.atomic.Value(bool) = .init(false),
460 thread: ?std.Thread = null,
461 transport: ?client.Transport = null,
462
463 fn init(alloc: std.mem.Allocator, target: client.Target) !DialJob {
464 return .{ .alloc = alloc, .target = target, .cancel_pipe = try std.posix.pipe2(.{ .NONBLOCK = true, .CLOEXEC = true }) };
465 }
466 fn deinit(self: *DialJob) void {
467 if (self.transport) |*tr| tr.close();
468 std.posix.close(self.cancel_pipe[0]);
469 std.posix.close(self.cancel_pipe[1]);
470 }
471 fn start(self: *DialJob) !void {
472 self.thread = try std.Thread.spawn(.{}, run, .{self});
473 }
474 fn run(self: *DialJob) void {
475 defer self.done.store(true, .release);
476 var dial: client.handoff.Dial = .{};
477 self.transport = client.Transport.openBounded(self.alloc, self.target, null, self.cancel_pipe[0], &dial, transport_queue_max) catch null;
478 }
479 fn cancel(self: *DialJob) void {
480 _ = std.posix.write(self.cancel_pipe[1], &.{client.interrupt.detach_key}) catch {};
481 }
482 fn join(self: *DialJob) void {
483 if (self.thread) |thread| thread.join();
484 self.thread = null;
485 }
486 fn take(self: *DialJob) ?client.Transport {
487 const value = self.transport;
488 self.transport = null;
489 return value;
490 }
491 };
492
493 fn monotonicMs() i64 {
494 const t = std.posix.clock_gettime(.MONOTONIC) catch return std.time.milliTimestamp();
495 return @as(i64, t.sec) * 1000 + @divFloor(t.nsec, 1_000_000);
496 }
497
498 fn bindLoopback(port: u16) !std.posix.fd_t {
499 const fd = try std.posix.socket(std.posix.AF.INET, std.posix.SOCK.STREAM | std.posix.SOCK.NONBLOCK | std.posix.SOCK.CLOEXEC, std.posix.IPPROTO.TCP);
500 errdefer std.posix.close(fd);
501 const addr = try std.net.Address.parseIp4("127.0.0.1", port);
502 try std.posix.bind(fd, &addr.any, addr.getOsSockLen());
503 try std.posix.listen(fd, 128);
504 return fd;
505 }
506
507 pub fn targetEqual(a: client.Target, b: client.Target) bool {
508 if (std.meta.activeTag(a) != std.meta.activeTag(b)) return false;
509 return switch (a) {
510 .sock => |x| std.mem.eql(u8, x, b.sock),
511 .via => |x| std.mem.eql(u8, x, b.via),
512 .quic => |x| std.mem.eql(u8, x.host_port, b.quic.host_port) and std.mem.eql(u8, x.key_path, b.quic.key_path),
513 .hand => |x| std.mem.eql(u8, x.host, b.hand.host) and argvEqual(x.ssh_argv, b.hand.ssh_argv),
514 };
515 }
516
517 fn argvEqual(a: []const []const u8, b: []const []const u8) bool {
518 if (a.len != b.len) return false;
519 for (a, b) |x, y| if (!std.mem.eql(u8, x, y)) return false;
520 return true;
521 }
522
523 test "forward rule accepts only nonzero decimal port pairs" {
524 try std.testing.expectEqual(Rule{ .local_port = 8080, .remote_port = 80 }, try Rule.parse("8080:80"));
525 for ([_][]const u8{ "", "80", "0:1", "1:0", "x:2", "+1:2", "1: 2", "1:2:3", "65536:2" }) |bad|
526 try std.testing.expectError(error.InvalidForward, Rule.parse(bad));
527 }
528
529 fn testUnusedPort() !u16 {
530 const fd = try bindLoopback(0);
531 defer std.posix.close(fd);
532 var actual: std.posix.sockaddr.storage = undefined;
533 var len: std.posix.socklen_t = @sizeOf(@TypeOf(actual));
534 try std.posix.getsockname(fd, @ptrCast(&actual), &len);
535 return std.net.Address.initPosix(@ptrCast(@alignCast(&actual))).getPort();
536 }
537
538 test "manager deduplicates before enforcing its unique listener cap" {
539 const rule: Rule = .{ .local_port = try testUnusedPort(), .remote_port = 80 };
540 const duplicates: [max_rules + 3]Rule = @splat(rule);
541 const manager = try Manager.init(std.testing.allocator, .{ .sock = "/not-dialled" }, &duplicates);
542 defer manager.stop();
543 try std.testing.expectEqual(@as(usize, 1), manager.listener_len);
544 }
545
546 test "manager rejects conflicting destinations without leaking its first bind" {
547 const port = try testUnusedPort();
548 try std.testing.expectError(
549 error.ForwardConflict,
550 Manager.init(std.testing.allocator, .{ .sock = "/not-dialled" }, &.{
551 .{ .local_port = port, .remote_port = 80 },
552 .{ .local_port = port, .remote_port = 81 },
553 }),
554 );
555 const rebound = try bindLoopback(port);
556 std.posix.close(rebound);
557 }
558
559 test "manager keeps the listener reserved and refuses accepts while unavailable" {
560 const port = try testUnusedPort();
561 const manager = try Manager.init(std.testing.allocator, .{ .sock = "/not-dialled" }, &.{.{ .local_port = port, .remote_port = 80 }});
562 defer manager.stop();
563
564 const fd = try std.posix.socket(std.posix.AF.INET, std.posix.SOCK.STREAM | std.posix.SOCK.CLOEXEC, std.posix.IPPROTO.TCP);
565 defer std.posix.close(fd);
566 const addr = try std.net.Address.parseIp4("127.0.0.1", port);
567 try std.posix.connect(fd, &addr.any, addr.getOsSockLen());
568 manager.refusePendingAccepts(100);
569 var pfd = [_]std.posix.pollfd{.{ .fd = fd, .events = std.posix.POLL.IN, .revents = 0 }};
570 try std.testing.expectEqual(@as(usize, 1), try std.posix.poll(&pfd, 1000));
571 try std.testing.expect(pfd[0].revents & (std.posix.POLL.IN | std.posix.POLL.HUP | std.posix.POLL.ERR) != 0);
572 }
573
574 test "target identity scopes forwarding independently of session and handoff narration" {
575 try std.testing.expect(targetEqual(.{ .sock = "/a" }, .{ .sock = "/a" }));
576 try std.testing.expect(!targetEqual(.{ .sock = "/a" }, .{ .sock = "/b" }));
577 try std.testing.expect(!targetEqual(.{ .sock = "/a" }, .{ .via = "/a" }));
578 try std.testing.expect(targetEqual(
579 .{ .quic = .{ .host_port = "host:1", .key_path = "/key", .idle_ms = 1 } },
580 .{ .quic = .{ .host_port = "host:1", .key_path = "/key", .idle_ms = 999 } },
581 ));
582 try std.testing.expect(targetEqual(
583 .{ .hand = .{ .host = "host", .ssh_argv = &.{ "ssh", "host" }, .cache_path = null, .asked = true, .narrate = true } },
584 .{ .hand = .{ .host = "host", .ssh_argv = &.{ "ssh", "host" }, .cache_path = "/elsewhere" } },
585 ));
586 }
587
588 fn testSocketPair() ![2]std.posix.fd_t {
589 var pair: [2]std.posix.fd_t = undefined;
590 if (std.c.socketpair(std.posix.AF.UNIX, std.posix.SOCK.STREAM, 0, &pair) != 0) return error.SocketPairFailed;
591 return pair;
592 }
593
594 fn fillSocket(fd: std.posix.fd_t) !usize {
595 var buf: [4096]u8 = @splat(0xa5);
596 var total: usize = 0;
597 while (true) total += std.posix.write(fd, &buf) catch |err| switch (err) {
598 error.WouldBlock => return total,
599 else => return err,
600 };
601 }
602
603 const NoisyHelper = struct {
604 err_fd: std.posix.fd_t,
605 transport_read_fd: std.posix.fd_t = -1,
606 transport_write_fd: std.posix.fd_t = -1,
607 wake_fd: std.posix.fd_t = -1,
608 completed: *std.atomic.Value(bool),
609
610 const noise_bytes = 128 * 1024;
611
612 fn floodErr(self: NoisyHelper) bool {
613 const flags = std.posix.fcntl(self.err_fd, std.posix.F.GETFL, 0) catch return false;
614 const bits: u32 = @bitCast(std.posix.O{ .NONBLOCK = true });
615 _ = std.posix.fcntl(self.err_fd, std.posix.F.SETFL, flags | bits) catch return false;
616 const until = std.time.milliTimestamp() + handshake_ms;
617 var buf: [4096]u8 = @splat('x');
618 var written: usize = 0;
619 while (written < noise_bytes) {
620 const n = std.posix.write(self.err_fd, buf[0..@min(buf.len, noise_bytes - written)]) catch |err| switch (err) {
621 error.WouldBlock => {
622 if (std.time.milliTimestamp() >= until) return false;
623 var fds = [_]std.posix.pollfd{.{ .fd = self.err_fd, .events = std.posix.POLL.OUT, .revents = 0 }};
624 _ = std.posix.poll(&fds, 10) catch return false;
625 continue;
626 },
627 else => return false,
628 };
629 if (n == 0) return false;
630 written += n;
631 }
632 return true;
633 }
634
635 fn answerHandshake(self: NoisyHelper) void {
636 if (!self.floodErr()) return;
637 var frame = (proto.readFrame(std.testing.allocator, self.transport_read_fd) catch return) orelse return;
638 defer frame.deinit(std.testing.allocator);
639 if (frame.type != .forward_hello) return;
640 const ready = proto.encodeForwardHello();
641 proto.writeFrame(self.transport_write_fd, .forward_ready, &ready) catch return;
642 self.completed.store(true, .release);
643 }
644
645 fn wakeConnected(self: NoisyHelper) void {
646 self.completed.store(self.floodErr(), .release);
647 _ = std.posix.write(self.wake_fd, &.{1}) catch {};
648 }
649 };
650
651 fn testPipe() ![2]std.posix.fd_t {
652 return std.posix.pipe2(.{ .CLOEXEC = true });
653 }
654
655 test "forward handshake drains noisy helper stderr" {
656 const alloc = std.testing.allocator;
657 const port = try testUnusedPort();
658 const manager = try Manager.init(alloc, .{ .sock = "/not-dialled" }, &.{.{ .local_port = port, .remote_port = 80 }});
659 defer manager.stop();
660
661 const incoming = try testPipe();
662 defer std.posix.close(incoming[0]);
663 defer std.posix.close(incoming[1]);
664 const outgoing = try testPipe();
665 defer std.posix.close(outgoing[0]);
666 defer std.posix.close(outgoing[1]);
667 const err_pipe = try testPipe();
668 defer std.posix.close(err_pipe[0]);
669 defer std.posix.close(err_pipe[1]);
670
671 var tr: client.Transport = .{
672 .link = .{ .pipe = .{
673 .child = std.process.Child.init(&.{"unused"}, alloc),
674 .r = incoming[0],
675 .w = outgoing[1],
676 } },
677 .err_fd = err_pipe[0],
678 };
679 var wire = try Wire.init(alloc, &tr);
680 defer wire.deinit();
681
682 var completed: std.atomic.Value(bool) = .init(false);
683 const helper = try std.Thread.spawn(.{}, NoisyHelper.answerHandshake, .{NoisyHelper{
684 .err_fd = err_pipe[1],
685 .transport_read_fd = outgoing[0],
686 .transport_write_fd = incoming[1],
687 .completed = &completed,
688 }});
689
690 const handshake_ok = manager.handshake(&wire);
691 helper.join();
692 try std.testing.expect(handshake_ok);
693 try std.testing.expect(completed.load(.acquire));
694 }
695
696 test "connected forwarding drains noisy helper stderr" {
697 const alloc = std.testing.allocator;
698 const port = try testUnusedPort();
699 const manager = try Manager.init(alloc, .{ .sock = "/not-dialled" }, &.{.{ .local_port = port, .remote_port = 80 }});
700 defer manager.stop();
701
702 const incoming = try testPipe();
703 defer std.posix.close(incoming[0]);
704 defer std.posix.close(incoming[1]);
705 const outgoing = try testPipe();
706 defer std.posix.close(outgoing[0]);
707 defer std.posix.close(outgoing[1]);
708 const err_pipe = try testPipe();
709 defer std.posix.close(err_pipe[0]);
710 defer std.posix.close(err_pipe[1]);
711
712 var tr: client.Transport = .{
713 .link = .{ .pipe = .{
714 .child = std.process.Child.init(&.{"unused"}, alloc),
715 .r = incoming[0],
716 .w = outgoing[1],
717 } },
718 .err_fd = err_pipe[0],
719 };
720 var wire = try Wire.init(alloc, &tr);
721 defer wire.deinit();
722
723 var completed: std.atomic.Value(bool) = .init(false);
724 const helper = try std.Thread.spawn(.{}, NoisyHelper.wakeConnected, .{NoisyHelper{
725 .err_fd = err_pipe[1],
726 .wake_fd = manager.wake_pipe[1],
727 .completed = &completed,
728 }});
729
730 manager.connected(&wire);
731 helper.join();
732 try std.testing.expect(completed.load(.acquire));
733 }
734
735 fn drainQueuedPrefix(wire: *Wire, fd: std.posix.fd_t, count: usize) !void {
736 var left = count;
737 var buf: [64 * 1024]u8 = undefined;
738 while (left != 0) {
739 try wire.flush();
740 const n = try std.posix.read(fd, buf[0..@min(buf.len, left)]);
741 if (n == 0) return error.UnexpectedEof;
742 left -= n;
743 }
744 }
745
746 test "manager stop interrupts an opening dial blocked by a full Unix backlog" {
747 if (@import("builtin").os.tag != .linux) return error.SkipZigTest;
748 const alloc = std.testing.allocator;
749 var tmp = try @import("testtmp").TmpDir.make();
750 defer tmp.cleanup();
751 const path = try std.fmt.allocPrint(alloc, "{s}/full.sock", .{tmp.path()});
752 defer alloc.free(path);
753 const addr = try std.net.Address.initUnix(path);
754 var server = try addr.listen(.{ .kernel_backlog = 1 });
755 defer server.deinit();
756 var queued: std.ArrayList(std.posix.fd_t) = .empty;
757 defer {
758 for (queued.items) |fd| std.posix.close(fd);
759 queued.deinit(alloc);
760 }
761 var full = false;
762 for (0..16) |_| {
763 const fd = try std.posix.socket(std.posix.AF.UNIX, std.posix.SOCK.STREAM | std.posix.SOCK.NONBLOCK | std.posix.SOCK.CLOEXEC, 0);
764 const err = std.posix.errno(std.posix.system.connect(fd, &addr.any, addr.getOsSockLen()));
765 if (err == .AGAIN) {
766 std.posix.close(fd);
767 full = true;
768 break;
769 }
770 if (err != .SUCCESS) {
771 std.posix.close(fd);
772 return error.UnexpectedConnect;
773 }
774 try queued.append(alloc, fd);
775 }
776 try std.testing.expect(full);
777
778 const Releaser = struct {
779 fn run(listener: *std.net.Server) void {
780 std.Thread.sleep(800 * std.time.ns_per_ms);
781 const accepted = listener.accept() catch return;
782 accepted.stream.close();
783 }
784 };
785 const releaser = try std.Thread.spawn(.{}, Releaser.run, .{&server});
786 defer releaser.join();
787
788 const port = try testUnusedPort();
789 const manager = try Manager.init(alloc, .{ .sock = path }, &.{.{ .local_port = port, .remote_port = 80 }});
790 try manager.start();
791 std.Thread.sleep(50 * std.time.ns_per_ms);
792 const started = std.time.milliTimestamp();
793 manager.stop();
794 try std.testing.expect(std.time.milliTimestamp() - started < 500);
795 }
796
797 const RejectingForwardPeer = struct {
798 listener: std.net.Server,
799 ready_then_closed: std.atomic.Value(u32) = .init(0),
800 until_ms: i64,
801
802 fn serve(self: *RejectingForwardPeer) void {
803 while (std.time.milliTimestamp() < self.until_ms) {
804 const conn = self.listener.accept() catch |err| switch (err) {
805 error.WouldBlock => {
806 std.Thread.sleep(std.time.ns_per_ms);
807 continue;
808 },
809 else => return,
810 };
811 const frame = (proto.readFrame(std.testing.allocator, conn.stream.handle) catch {
812 conn.stream.close();
813 continue;
814 }) orelse {
815 conn.stream.close();
816 continue;
817 };
818 if (frame.type == .forward_hello) {
819 const ready = proto.encodeForwardHello();
820 proto.writeFrame(conn.stream.handle, .forward_ready, &ready) catch {
821 frame.deinit(std.testing.allocator);
822 conn.stream.close();
823 continue;
824 };
825 _ = self.ready_then_closed.fetchAdd(1, .acq_rel);
826 }
827 frame.deinit(std.testing.allocator);
828 conn.stream.close();
829 }
830 }
831 };
832
833 const ListenerFlood = struct {
834 port: u16,
835 stop: std.atomic.Value(bool) = .init(false),
836
837 fn run(self: *ListenerFlood) void {
838 const addr = std.net.Address.parseIp4("127.0.0.1", self.port) catch return;
839 while (!self.stop.load(.acquire)) {
840 const fd = std.posix.socket(std.posix.AF.INET, std.posix.SOCK.STREAM | std.posix.SOCK.NONBLOCK | std.posix.SOCK.CLOEXEC, std.posix.IPPROTO.TCP) catch continue;
841 _ = std.posix.connect(fd, &addr.any, addr.getOsSockLen()) catch {};
842 std.posix.close(fd);
843 }
844 }
845 };
846
847 test "manager keeps retry deadline under continuously readable listeners and stop wakes it" {
848 const alloc = std.testing.allocator;
849 var tmp = try @import("testtmp").TmpDir.make();
850 defer tmp.cleanup();
851 const path = try std.fmt.allocPrint(alloc, "{s}/reject.sock", .{tmp.path()});
852 defer alloc.free(path);
853 const addr = try std.net.Address.initUnix(path);
854 var peer = RejectingForwardPeer{
855 .listener = try addr.listen(.{}),
856 .until_ms = std.time.milliTimestamp() + 450,
857 };
858 defer peer.listener.deinit();
859 const flags = try std.posix.fcntl(peer.listener.stream.handle, std.posix.F.GETFL, 0);
860 const nonblock: u32 = @bitCast(std.posix.O{ .NONBLOCK = true });
861 _ = try std.posix.fcntl(peer.listener.stream.handle, std.posix.F.SETFL, flags | nonblock);
862 const peer_thread = try std.Thread.spawn(.{}, RejectingForwardPeer.serve, .{&peer});
863 defer peer_thread.join();
864
865 const port = try testUnusedPort();
866 const manager = try Manager.init(alloc, .{ .sock = path }, &.{.{ .local_port = port, .remote_port = 80 }});
867 try manager.start();
868 var flood = ListenerFlood{ .port = port };
869 const flood_thread = try std.Thread.spawn(.{}, ListenerFlood.run, .{&flood});
870 defer {
871 flood.stop.store(true, .release);
872 flood_thread.join();
873 }
874 std.Thread.sleep(350 * std.time.ns_per_ms);
875 const attempts = peer.ready_then_closed.load(.acquire);
876 // The first attempt plus fixed 100 ms pauses admits several retries, but
877 // not a hot loop even while the local listener is continuously readable.
878 try std.testing.expect(attempts >= 2);
879 try std.testing.expect(attempts <= 5);
880
881 const started = std.time.milliTimestamp();
882 manager.stop();
883 try std.testing.expect(std.time.milliTimestamp() - started < 500);
884 }
885
886 test "concurrent local channels pause at shared transport capacity and resume intact" {
887 const alloc = std.testing.allocator;
888 const port = try testUnusedPort();
889 const manager = try Manager.init(alloc, .{ .sock = "/not-dialled" }, &.{.{ .local_port = port, .remote_port = 80 }});
890 defer manager.stop();
891
892 const transport_pair = try testSocketPair();
893 defer std.posix.close(transport_pair[1]);
894 var tr: client.Transport = .{ .link = .{ .fd = transport_pair[0] } };
895 defer tr.close();
896 var wire = try Wire.init(alloc, &tr);
897 defer wire.deinit();
898 const kernel_fill = try fillSocket(transport_pair[0]);
899
900 const frame_len = proto.frame_header_len + proto.forward_id_len + proto.forward_data_max;
901 const prefix_len = transport_queue_max - control_reserve - frame_len;
902 try wire.output.resize(alloc, prefix_len);
903 @memset(wire.output.items, 0x5a);
904
905 const channel_count = 6;
906 var channel_peers: [channel_count]std.posix.fd_t = undefined;
907 var made: usize = 0;
908 defer for (channel_peers[0..made]) |fd| std.posix.close(fd);
909 var payloads: [channel_count][proto.forward_data_max]u8 = undefined;
910 for (0..channel_count) |i| {
911 const pair = try testSocketPair();
912 channel_peers[i] = pair[1];
913 made += 1;
914 @memset(&payloads[i], @intCast(i + 1));
915 try proto.writeAllFd(pair[1], &payloads[i]);
916 manager.channels[i] = .{ .id = @intCast(i + 1), .fd = pair[0], .opening = false, .send_credit = proto.forward_initial_credit };
917 }
918
919 // Model one poll snapshot that reported every channel readable.
920 for (0..channel_count) |i| try std.testing.expect(manager.readLocal(&wire, i));
921 try std.testing.expectEqual(prefix_len + frame_len, wire.pendingBytes());
922 try std.testing.expect(wire.pendingBytes() <= transport_queue_max - control_reserve);
923 for (0..channel_count) |i| try std.testing.expect(manager.channels[i] != null);
924
925 try drainQueuedPrefix(&wire, transport_pair[1], kernel_fill + prefix_len);
926 try wire.flush();
927 var first = (try proto.readFrame(alloc, transport_pair[1])) orelse return error.UnexpectedEof;
928 defer first.deinit(alloc);
929 try std.testing.expectEqual(proto.MsgType.forward_data, first.type);
930 try std.testing.expectEqual(@as(u32, 1), try proto.decodeForwardId(first.payload));
931 try std.testing.expectEqualSlices(u8, &payloads[0], first.payload[proto.forward_id_len..]);
932
933 for (1..channel_count) |i| {
934 try std.testing.expect(manager.readLocal(&wire, i));
935 try std.testing.expect(wire.pendingBytes() <= transport_queue_max - control_reserve);
936 try wire.flush();
937 var frame = (try proto.readFrame(alloc, transport_pair[1])) orelse return error.UnexpectedEof;
938 defer frame.deinit(alloc);
939 try std.testing.expectEqual(proto.MsgType.forward_data, frame.type);
940 try std.testing.expectEqual(@as(u32, @intCast(i + 1)), try proto.decodeForwardId(frame.payload));
941 try std.testing.expectEqualSlices(u8, &payloads[i], frame.payload[proto.forward_id_len..]);
942 }
943 }
src/engine/protocol.zig
Old New
@@ -24,6 +24,12 @@ pub const MsgType = enum(u8) {
24 upgrade_req = 0x10, // payload: u8 flags (bit0 allow_same_version) ++ version bytes ++ NUL ++ absolute path bytes 24 upgrade_req = 0x10, // payload: u8 flags (bit0 allow_same_version) ++ version bytes ++ NUL ++ absolute path bytes
25 end_req = 0x11, // payload: u8 flags (bit0 force) ++ optional session-name tail (empty = default session) 25 end_req = 0x11, // payload: u8 flags (bit0 force) ++ optional session-name tail (empty = default session)
26 create_req = 0x12, // payload: u16 LE cols, u16 LE rows ++ explicit session name; create only, never attach 26 create_req = 0x12, // payload: u16 LE cols, u16 LE rows ++ explicit session name; create only, never attach
27 forward_hello = 0x20, // payload: u16 LE protocol version; promotes this connection to the forwarding role
28 forward_open = 0x21, // payload: u32 LE client channel id, u16 LE loopback destination port
29 forward_data = 0x22, // payload: u32 LE channel id ++ bounded opaque TCP bytes; BOTH directions
30 forward_credit = 0x23, // payload: u32 LE channel id, u32 LE additional receive credit; BOTH directions
31 forward_half_close = 0x24, // payload: u32 LE channel id; BOTH directions
32 forward_reset = 0x25, // payload: u32 LE channel id; BOTH directions
27 debug_dump = 0x7f, // payload: 1 byte: 0 = plain, 1 = vt ++ optional session-name tail (empty = default session) 33 debug_dump = 0x7f, // payload: 1 byte: 0 = plain, 1 = vt ++ optional session-name tail (empty = default session)
28 // daemon -> client 34 // daemon -> client
29 // The three replay frames were renumbered in 2026-09 when their payloads 35 // The three replay frames were renumbered in 2026-09 when their payloads
@@ -54,6 +60,8 @@ pub const MsgType = enum(u8) {
54 upgrade_reply = 0x93, // payload: u8 status (0 accepted, 1 refused) ++ reason text 60 upgrade_reply = 0x93, // payload: u8 status (0 accepted, 1 refused) ++ reason text
55 end_reply = 0x94, // payload: u8 status (0 accepted, 1 refused) ++ u8 others ++ reason text 61 end_reply = 0x94, // payload: u8 status (0 accepted, 1 refused) ++ u8 others ++ reason text
56 create_reply = 0x98, // payload: CreateStatus byte ++ bounded reason text 62 create_reply = 0x98, // payload: CreateStatus byte ++ bounded reason text
63 forward_ready = 0xa0, // payload: u16 LE protocol version
64 forward_open_result = 0xa1, // payload: u32 LE channel id, u8 status (0 open, 1 refused)
57 dump_reply = 0xff, // payload: requested dump bytes 65 dump_reply = 0xff, // payload: requested dump bytes
58 _, 66 _,
59 }; 67 };
@@ -284,6 +292,123 @@ pub const agent_chans_max = 8;
284 /// alone. 292 /// alone.
285 pub const agent_sock_env = "SSH_AUTH_SOCK"; 293 pub const agent_sock_env = "SSH_AUTH_SOCK";
286 294
295 /// The first forwarding wire contract. A versioned role handshake keeps an
296 /// old daemon's unknown-frame behaviour from looking like an accepted tunnel.
297 pub const forward_version: u16 = 1;
298 pub const forward_id_len = 4;
299 pub const forward_data_max: usize = 16 * 1024;
300 pub const forward_initial_credit: u32 = 64 * 1024;
301 pub const forward_channels_max: usize = 32;
302 pub const forward_hello_len = 2;
303 pub const forward_open_len = 6;
304 pub const forward_credit_len = 8;
305 pub const forward_open_result_len = 5;
306
307 pub const ForwardOpen = struct { id: u32, port: u16 };
308 pub const ForwardCredit = struct { id: u32, amount: u32 };
309 pub const ForwardOpenResult = struct { id: u32, ok: bool };
310
311 pub fn encodeForwardHello() [forward_hello_len]u8 {
312 var out: [forward_hello_len]u8 = undefined;
313 std.mem.writeInt(u16, &out, forward_version, .little);
314 return out;
315 }
316
317 pub fn decodeForwardHello(payload: []const u8) !u16 {
318 if (payload.len != forward_hello_len) return error.BadPayload;
319 return std.mem.readInt(u16, payload[0..2], .little);
320 }
321
322 pub fn encodeForwardOpen(open: ForwardOpen) [forward_open_len]u8 {
323 var out: [forward_open_len]u8 = undefined;
324 std.mem.writeInt(u32, out[0..4], open.id, .little);
325 std.mem.writeInt(u16, out[4..6], open.port, .little);
326 return out;
327 }
328
329 pub fn decodeForwardOpen(payload: []const u8) !ForwardOpen {
330 if (payload.len != forward_open_len) return error.BadPayload;
331 const out: ForwardOpen = .{
332 .id = std.mem.readInt(u32, payload[0..4], .little),
333 .port = std.mem.readInt(u16, payload[4..6], .little),
334 };
335 if (out.id == 0 or out.port == 0) return error.BadPayload;
336 return out;
337 }
338
339 pub fn encodeForwardId(id: u32) [forward_id_len]u8 {
340 var out: [forward_id_len]u8 = undefined;
341 std.mem.writeInt(u32, &out, id, .little);
342 return out;
343 }
344
345 pub fn decodeForwardId(payload: []const u8) !u32 {
346 if (payload.len < forward_id_len) return error.BadPayload;
347 const id = std.mem.readInt(u32, payload[0..4], .little);
348 if (id == 0) return error.BadPayload;
349 return id;
350 }
351
352 pub fn forwardDataOversize(payload: []const u8) bool {
353 return payload.len > forward_id_len + forward_data_max;
354 }
355
356 pub fn encodeForwardCredit(credit: ForwardCredit) [forward_credit_len]u8 {
357 var out: [forward_credit_len]u8 = undefined;
358 std.mem.writeInt(u32, out[0..4], credit.id, .little);
359 std.mem.writeInt(u32, out[4..8], credit.amount, .little);
360 return out;
361 }
362
363 pub fn decodeForwardCredit(payload: []const u8) !ForwardCredit {
364 if (payload.len != forward_credit_len) return error.BadPayload;
365 const out: ForwardCredit = .{
366 .id = std.mem.readInt(u32, payload[0..4], .little),
367 .amount = std.mem.readInt(u32, payload[4..8], .little),
368 };
369 if (out.id == 0 or out.amount == 0) return error.BadPayload;
370 return out;
371 }
372
373 pub fn encodeForwardOpenResult(result: ForwardOpenResult) [forward_open_result_len]u8 {
374 var out: [forward_open_result_len]u8 = undefined;
375 std.mem.writeInt(u32, out[0..4], result.id, .little);
376 out[4] = @intFromBool(!result.ok);
377 return out;
378 }
379
380 pub fn decodeForwardOpenResult(payload: []const u8) !ForwardOpenResult {
381 if (payload.len != forward_open_result_len or payload[4] > 1) return error.BadPayload;
382 const id = std.mem.readInt(u32, payload[0..4], .little);
383 if (id == 0) return error.BadPayload;
384 return .{ .id = id, .ok = payload[4] == 0 };
385 }
386
387 test "forwarding codecs round-trip their bounded wire values" {
388 try std.testing.expectEqual(forward_version, try decodeForwardHello(&encodeForwardHello()));
389 const open: ForwardOpen = .{ .id = 0x12345678, .port = 65535 };
390 try std.testing.expectEqual(open, try decodeForwardOpen(&encodeForwardOpen(open)));
391 try std.testing.expectEqual(open.id, try decodeForwardId(&encodeForwardId(open.id)));
392 const credit: ForwardCredit = .{ .id = open.id, .amount = forward_initial_credit };
393 try std.testing.expectEqual(credit, try decodeForwardCredit(&encodeForwardCredit(credit)));
394 for ([_]bool{ false, true }) |ok| {
395 const result: ForwardOpenResult = .{ .id = open.id, .ok = ok };
396 try std.testing.expectEqual(result, try decodeForwardOpenResult(&encodeForwardOpenResult(result)));
397 }
398 try std.testing.expect(!forwardDataOversize(&([_]u8{0} ** (forward_id_len + forward_data_max))));
399 try std.testing.expect(forwardDataOversize(&([_]u8{0} ** (forward_id_len + forward_data_max + 1))));
400 }
401
402 test "forwarding codecs reject malformed identifiers ports credit and status" {
403 try std.testing.expectError(error.BadPayload, decodeForwardHello(&.{1}));
404 try std.testing.expectError(error.BadPayload, decodeForwardOpen(&.{ 0, 0, 0, 0, 80, 0 }));
405 try std.testing.expectError(error.BadPayload, decodeForwardOpen(&.{ 1, 0, 0, 0, 0, 0 }));
406 try std.testing.expectError(error.BadPayload, decodeForwardId(&.{ 0, 0, 0, 0 }));
407 try std.testing.expectError(error.BadPayload, decodeForwardId(&.{ 1, 0, 0 }));
408 try std.testing.expectError(error.BadPayload, decodeForwardCredit(&.{ 1, 0, 0, 0, 0, 0, 0, 0 }));
409 try std.testing.expectError(error.BadPayload, decodeForwardOpenResult(&.{ 1, 0, 0, 0, 2 }));
410 }
411
287 pub fn encodeAgentId(id: u32) [agent_id_len]u8 { 412 pub fn encodeAgentId(id: u32) [agent_id_len]u8 {
288 var buf: [agent_id_len]u8 = undefined; 413 var buf: [agent_id_len]u8 = undefined;
289 std.mem.writeInt(u32, &buf, id, .little); 414 std.mem.writeInt(u32, &buf, id, .little);
src/gui/frame.zig
Old New
@@ -36,6 +36,7 @@ pub const Options = struct {
36 height: u32 = 600, 36 height: u32 = 600,
37 /// Optional integration FIFO; events use the ordinary window input paths. 37 /// Optional integration FIFO; events use the ordinary window input paths.
38 test_fifo: ?[]const u8 = null, 38 test_fifo: ?[]const u8 = null,
39 forwards: []const client.forward.Rule = &.{},
39 }; 40 };
40 41
41 /// SDL keycode + mods → the input event, or null for a key that types 42 /// SDL keycode + mods → the input event, or null for a key that types
@@ -563,6 +564,11 @@ fn interactionKey(ev: c.SDL_KeyboardEvent) interaction.KeyDown {
563 } 564 }
564 565
565 pub fn run(alloc: std.mem.Allocator, opts: Options) !u8 { 566 pub fn run(alloc: std.mem.Allocator, opts: Options) !u8 {
567 var forward_owner: ?*client.forward.Manager = if (opts.forwards.len != 0)
568 try client.forward.Manager.init(alloc, opts.target orelse return error.ForwardRequiresTarget, opts.forwards)
569 else
570 null;
571 errdefer if (forward_owner) |manager| manager.stop();
566 const trial = opts.test_fifo != null and std.mem.eql(u8, std.posix.getenv("MUXG_TEST_THEME") orelse "", "trial"); 572 const trial = opts.test_fifo != null and std.mem.eql(u8, std.posix.getenv("MUXG_TEST_THEME") orelse "", "trial");
567 const configured_appearance = opts.appearance; 573 const configured_appearance = opts.appearance;
568 const appearance = if (trial) &theme_mod.trial else &configured_appearance; 574 const appearance = if (trial) &theme_mod.trial else &configured_appearance;
@@ -637,6 +643,11 @@ pub fn run(alloc: std.mem.Allocator, opts: Options) !u8 {
637 const layout = rt.workspace.layout(@intCast(@max(fb_w, 0)), @intCast(@max(fb_h, 0)), metrics); 643 const layout = rt.workspace.layout(@intCast(@max(fb_w, 0)), @intCast(@max(fb_h, 0)), metrics);
638 try rt.restore(&layout); 644 try rt.restore(&layout);
639 } else if (opts.target) |target| _ = try rt.add(target, opts.session, @intCast(@max(fb_w, 0)), @intCast(@max(fb_h, 0)), metrics); 645 } else if (opts.target) |target| _ = try rt.add(target, opts.session, @intCast(@max(fb_w, 0)), @intCast(@max(fb_h, 0)), metrics);
646 if (forward_owner) |manager| {
647 try manager.start();
648 rt.setForwarding(manager);
649 forward_owner = null;
650 }
640 defer if (store) |*s| s.save(&rt.workspace) catch |err| { 651 defer if (store) |*s| s.save(&rt.workspace) catch |err| {
641 std.debug.print("muxg: workspace not saved: {s}\n", .{@errorName(err)}); 652 std.debug.print("muxg: workspace not saved: {s}\n", .{@errorName(err)});
642 }; 653 };
src/gui/runtime.zig
Old New
@@ -74,10 +74,13 @@ pub const Runtime = struct {
74 workspace: model.Workspace, 74 workspace: model.Workspace,
75 lives: [model.max_panes]?*Live = @splat(null), 75 lives: [model.max_panes]?*Live = @splat(null),
76 notify: Notify, 76 notify: Notify,
77 forwarding: ?*client.forward.Manager = null,
77 pub fn init(alloc: std.mem.Allocator, notify: Notify) Runtime { 78 pub fn init(alloc: std.mem.Allocator, notify: Notify) Runtime {
78 return .{ .alloc = alloc, .workspace = model.Workspace.init(alloc), .notify = notify }; 79 return .{ .alloc = alloc, .workspace = model.Workspace.init(alloc), .notify = notify };
79 } 80 }
80 pub fn deinit(self: *Runtime) void { 81 pub fn deinit(self: *Runtime) void {
82 if (self.forwarding) |manager| manager.stop();
83 self.forwarding = null;
81 for (self.lives) |p| if (p) |live| { 84 for (self.lives) |p| if (p) |live| {
82 live.pump.say(.quit) catch unreachable; 85 live.pump.say(.quit) catch unreachable;
83 }; 86 };
@@ -86,6 +89,16 @@ pub const Runtime = struct {
86 }; 89 };
87 self.workspace.deinit(); 90 self.workspace.deinit();
88 } 91 }
92 pub fn setForwarding(self: *Runtime, manager: *client.forward.Manager) void {
93 std.debug.assert(self.forwarding == null);
94 self.forwarding = manager;
95 }
96 fn releaseForwardIfUnused(self: *Runtime) void {
97 const manager = self.forwarding orelse return;
98 if (self.workspace.hasTarget(manager.target)) return;
99 self.forwarding = null;
100 manager.stop();
101 }
89 pub fn get(self: *Runtime, id: model.PaneId) ?*Live { 102 pub fn get(self: *Runtime, id: model.PaneId) ?*Live {
90 for (self.lives) |p| if (p) |live| { 103 for (self.lives) |p| if (p) |live| {
91 if (live.key.pane == id) return live; 104 if (live.key.pane == id) return live;
@@ -146,6 +159,7 @@ pub const Runtime = struct {
146 slot.* = live; 159 slot.* = live;
147 break; 160 break;
148 }; 161 };
162 self.releaseForwardIfUnused();
149 } 163 }
150 pub fn retry(self: *Runtime, id: model.PaneId, placement: model.Placement) !void { 164 pub fn retry(self: *Runtime, id: model.PaneId, placement: model.Placement) !void {
151 const pane = self.workspace.pane(id) orelse return error.MissingPane; 165 const pane = self.workspace.pane(id) orelse return error.MissingPane;
@@ -160,6 +174,7 @@ pub const Runtime = struct {
160 } 174 }
161 }; 175 };
162 self.workspace.remove(id); 176 self.workspace.remove(id);
177 self.releaseForwardIfUnused();
163 } 178 }
164 pub fn resize(self: *Runtime, layout: *const model.Layout) !void { 179 pub fn resize(self: *Runtime, layout: *const model.Layout) !void {
165 for (layout.items()) |p| { 180 for (layout.items()) |p| {
@@ -245,6 +260,56 @@ fn waitPhase(live: *Live, phase: client.session_pump.Phase) !void {
245 return error.PhaseTimeout; 260 return error.PhaseTimeout;
246 } 261 }
247 262
263 fn forwardingTestPort() !u16 {
264 const fd = try std.posix.socket(std.posix.AF.INET, std.posix.SOCK.STREAM | std.posix.SOCK.NONBLOCK | std.posix.SOCK.CLOEXEC, std.posix.IPPROTO.TCP);
265 defer std.posix.close(fd);
266 const addr = try std.net.Address.parseIp4("127.0.0.1", 0);
267 try std.posix.bind(fd, &addr.any, addr.getOsSockLen());
268 try std.posix.listen(fd, 1);
269 var actual: std.posix.sockaddr.storage = undefined;
270 var len: std.posix.socklen_t = @sizeOf(@TypeOf(actual));
271 try std.posix.getsockname(fd, @ptrCast(&actual), &len);
272 return std.net.Address.initPosix(@ptrCast(@alignCast(&actual))).getPort();
273 }
274
275 fn forwardingCanBind(port: u16) !void {
276 const fd = try std.posix.socket(std.posix.AF.INET, std.posix.SOCK.STREAM | std.posix.SOCK.NONBLOCK | std.posix.SOCK.CLOEXEC, std.posix.IPPROTO.TCP);
277 defer std.posix.close(fd);
278 const addr = try std.net.Address.parseIp4("127.0.0.1", port);
279 try std.posix.bind(fd, &addr.any, addr.getOsSockLen());
280 }
281
282 test "forwarding lifetime follows target panes rather than focus or session" {
283 const a = std.testing.allocator;
284 var rt = Runtime.init(a, .{});
285 defer rt.deinit();
286 const m: model.Metrics = .{ .cell_w = 10, .cell_h = 20 };
287 const target: client.Target = .{ .via = "false" };
288 const first = try rt.add(target, "one", 800, 600, m);
289 rt.workspace.arm(.beside);
290 const last = try rt.add(target, "two", 800, 600, m);
291 _ = rt.workspace.focus(first);
292
293 const replace_port = try forwardingTestPort();
294 const replace_manager = try client.forward.Manager.init(a, target, &.{.{ .local_port = replace_port, .remote_port = 80 }});
295 rt.setForwarding(replace_manager);
296 rt.remove(first);
297 try std.testing.expect(rt.forwarding == replace_manager);
298 try std.testing.expectEqual(last, rt.workspace.tab().focus.?);
299
300 const placement = rt.workspace.layout(800, 600, m).get(last).?;
301 try rt.replace(last, .{ .via = "true" }, "three", placement, false);
302 try std.testing.expect(rt.forwarding == null);
303 try forwardingCanBind(replace_port);
304
305 const remove_port = try forwardingTestPort();
306 const remove_manager = try client.forward.Manager.init(a, .{ .via = "true" }, &.{.{ .local_port = remove_port, .remote_port = 80 }});
307 rt.setForwarding(remove_manager);
308 rt.remove(last);
309 try std.testing.expect(rt.forwarding == null);
310 try forwardingCanBind(remove_port);
311 }
312
248 test "same-pane replacement is transactional and retry keeps text through an invalid first snapshot" { 313 test "same-pane replacement is transactional and retry keeps text through an invalid first snapshot" {
249 const a = std.testing.allocator; 314 const a = std.testing.allocator;
250 var tmp = std.testing.tmpDir(.{}); 315 var tmp = std.testing.tmpDir(.{});
src/gui/workspace.zig
Old New
@@ -293,6 +293,12 @@ pub const Workspace = struct {
293 }; 293 };
294 return null; 294 return null;
295 } 295 }
296 pub fn hasTarget(self: *Workspace, target: client.Target) bool {
297 for (self.tab().panes) |p| if (p) |pane_value| {
298 if (client.forward.targetEqual(pane_value.identity.target, target)) return true;
299 };
300 return false;
301 }
296 pub fn layout(self: *Workspace, width: u32, height: u32, m: Metrics) Layout { 302 pub fn layout(self: *Workspace, width: u32, height: u32, m: Metrics) Layout {
297 const t = self.tab(); 303 const t = self.tab();
298 const root = if (t.fullscreen and t.focus != null) t.tree.leaf(t.focus.?) else t.tree.root; 304 const root = if (t.fullscreen and t.focus != null) t.tree.leaf(t.focus.?) else t.tree.root;
src/link.zig
Old New
@@ -97,6 +97,16 @@ pub const Link = union(enum) {
97 } 97 }
98 } 98 }
99 99
100 /// Set an opaque inbound-byte budget for a role that cannot retain the
101 /// terminal wire's normal framing allowance. The transport enforces the
102 /// budget before it extends QUIC flow control; it still knows no frames.
103 pub fn setInboundCap(self: *Link, cap: ?usize) void {
104 switch (self.*) {
105 .quic => |*q| q.cl.inbound_cap = cap,
106 .fd, .pipe => {},
107 }
108 }
109
100 /// Queue-and-offer, never blocking: fd and pipe write through; quic 110 /// Queue-and-offer, never blocking: fd and pipe write through; quic
101 /// appends whole and flushes what the ring takes. A caller that must 111 /// appends whole and flushes what the ring takes. A caller that must
102 /// KNOW the bytes left (muxa's verbs) follows with flushWithin. 112 /// KNOW the bytes left (muxa's verbs) follows with flushWithin.
src/os/client_os.zig
Old New
@@ -50,6 +50,12 @@ pub fn sendNoSig(fd: std.posix.socket_t, bytes: []const u8) std.posix.SendError!
50 return impl.sendNoSig(fd, bytes); 50 return impl.sendNoSig(fd, bytes);
51 } 51 }
52 52
53 /// A nonblocking TCP send which cannot raise SIGPIPE, used by the native
54 /// forwarding pump so one stalled local reader never stalls its owner thread.
55 pub fn sendNoSigNoWait(fd: std.posix.socket_t, bytes: []const u8) !usize {
56 return impl.sendNoSigNoWait(fd, bytes);
57 }
58
53 /// The parent of `pid`, or 0 when the OS will not say or `pid` is not 59 /// The parent of `pid`, or 0 when the OS will not say or `pid` is not
54 /// positive. One step of the walk from an askpass helper up to the ssh a 60 /// positive. One step of the walk from an askpass helper up to the ssh a
55 /// dial spawned. 61 /// dial spawned.
src/os/client_os_linux.zig
Old New
@@ -17,6 +17,10 @@ pub fn sendNoSig(fd: std.posix.socket_t, bytes: []const u8) std.posix.SendError!
17 return std.posix.send(fd, bytes, std.posix.MSG.NOSIGNAL); 17 return std.posix.send(fd, bytes, std.posix.MSG.NOSIGNAL);
18 } 18 }
19 19
20 pub fn sendNoSigNoWait(fd: std.posix.socket_t, bytes: []const u8) !usize {
21 return std.posix.send(fd, bytes, std.posix.MSG.NOSIGNAL | std.posix.MSG.DONTWAIT);
22 }
23
20 /// `/proc/<pid>/stat` field 4. Parsed from the LAST ')' rather than by 24 /// `/proc/<pid>/stat` field 4. Parsed from the LAST ')' rather than by
21 /// counting spaces: field 2 is the executable's name, unquoted, and a 25 /// counting spaces: field 2 is the executable's name, unquoted, and a
22 /// program free to call itself `a b) c` is a program free to move every 26 /// program free to call itself `a b) c` is a program free to move every
src/os/client_os_macos.zig
Old New
@@ -92,6 +92,13 @@ pub fn sendNoSig(fd: std.posix.socket_t, bytes: []const u8) std.posix.SendError!
92 }; 92 };
93 } 93 }
94 94
95 pub fn sendNoSigNoWait(fd: std.posix.socket_t, bytes: []const u8) !usize {
96 const flags = try std.posix.fcntl(fd, std.posix.F.GETFL, 0);
97 const nonblock: u32 = @bitCast(std.posix.O{ .NONBLOCK = true });
98 _ = try std.posix.fcntl(fd, std.posix.F.SETFL, flags | nonblock);
99 return sendNoSig(fd, bytes);
100 }
101
95 /// sysctl KERN_PROC_PID: the kernel's own record of the process, the Darwin 102 /// sysctl KERN_PROC_PID: the kernel's own record of the process, the Darwin
96 /// answer to /proc/PID/stat. A pid the kernel no longer has answers 0 with 103 /// answer to /proc/PID/stat. A pid the kernel no longer has answers 0 with
97 /// a length of 0 rather than an error, so the length is read as well as the 104 /// a length of 0 rather than an error, so the length is read as well as the
src/quic.zig
Old New
@@ -899,6 +899,14 @@ fn recvStreamDataCb(
899 ) callconv(.c) c_int { 899 ) callconv(.c) c_int {
900 const self: *Client = @ptrCast(@alignCast(ud.?)); 900 const self: *Client = @ptrCast(@alignCast(ud.?));
901 901
902 // A forwarding wire has a deliberately small opaque-byte budget. Check
903 // before append OR extending flow-control: credit must never be granted
904 // for bytes its owner cannot retain. Terminal clients leave this null.
905 if (self.inbound_cap) |cap| if (datalen > cap -| self.in.items.len) {
906 self.dead = true;
907 return -1;
908 };
909
902 // BOTH windows, and the connection-level one is the half that is easy to 910 // BOTH windows, and the connection-level one is the half that is easy to
903 // forget: extend only the stream and the daemon stops sending a few 911 // forget: extend only the stream and the daemon stops sending a few
904 // hundred kilobytes into a session — a scrollback fetch, a big 912 // hundred kilobytes into a session — a scrollback fetch, a big
@@ -917,6 +925,37 @@ fn recvStreamDataCb(
917 return 0; 925 return 0;
918 } 926 }
919 927
928 test "forwarding receive cap accepts one legal frame exactly and rejects the next fragment" {
929 const alloc = std.testing.allocator;
930 // quic.zig is tested as its own root, so it cannot import the client
931 // protocol module here. A forwarding data frame is its five-byte wire
932 // header, u32 channel id, and the protocol's 16 KiB maximum data body.
933 // The dial installs its larger transport queue cap before waitReady.
934 const cap = 5 + 4 + 16 * 1024;
935 var cl: Client = .{
936 .alloc = alloc,
937 .fd = -1,
938 .out = .{ .buf = &.{} },
939 .inbound_cap = cap,
940 .extend_windows = false,
941 };
942 defer cl.in.deinit(alloc);
943
944 var header: [5]u8 = undefined;
945 var data: [4 + 16 * 1024]u8 = undefined;
946 const first = recvStreamDataCb(null, 0, 0, 0, &header, header.len, &cl, null);
947 const second = recvStreamDataCb(null, 0, 0, header.len, &data, data.len, &cl, null);
948 try std.testing.expectEqual(@as(c_int, 0), first);
949 try std.testing.expectEqual(@as(c_int, 0), second);
950 try std.testing.expectEqual(cap, cl.in.items.len);
951 try std.testing.expect(!cl.dead);
952
953 const extra = [_]u8{0};
954 try std.testing.expectEqual(@as(c_int, -1), recvStreamDataCb(null, 0, 0, cap, &extra, extra.len, &cl, null));
955 try std.testing.expect(cl.dead);
956 try std.testing.expectEqual(cap, cl.in.items.len);
957 }
958
920 fn ackedStreamDataCb( 959 fn ackedStreamDataCb(
921 _: ?*c.ngtcp2_conn, 960 _: ?*c.ngtcp2_conn,
922 _: i64, 961 _: i64,
@@ -958,6 +997,10 @@ pub const Client = struct {
958 out: Egress, 997 out: Egress,
959 /// Stream bytes the caller has not consumed yet. 998 /// Stream bytes the caller has not consumed yet.
960 in: std.ArrayList(u8) = .empty, 999 in: std.ArrayList(u8) = .empty,
1000 /// Optional opaque receive budget selected by a role-specific owner.
1001 /// Null preserves the terminal transport's existing unbounded framing
1002 /// limit; no forwarding frame policy belongs in this transport.
1003 inbound_cap: ?usize = null,
961 /// Whether arriving bytes buy the peer more window. A shipping client 1004 /// Whether arriving bytes buy the peer more window. A shipping client
962 /// always extends (constraint 3 above); the listener's own tests turn 1005 /// always extends (constraint 3 above); the listener's own tests turn
963 /// it off to reach the blocked-stream branch, which is a documented 1006 /// it off to reach the blocked-stream branch, which is a documented
src/server/quic_server.zig
Old New
@@ -52,9 +52,12 @@ pub const Handler = struct {
52 /// 52 ///
53 /// Restated here rather than derived: this file is a transport, and a 53 /// Restated here rather than derived: this file is a transport, and a
54 /// transport does not read the daemon's tables. The restatement is pinned 54 /// transport does not read the daemon's tables. The restatement is pinned
55 /// instead — `server_test_quic` asserts `max_conns > max_clients`, so raising 55 /// instead — `server_test_quic` asserts room for terminal and forwarding
56 /// one without the other fails the build rather than going quiet. 56 /// admissions and pins the provisional role table to this cap, so changing one
57 pub const max_conns = 40; 57 /// without the others fails the build rather than going quiet.
58 // Terminal clients, forwarding-role peers and the small unclassified role
59 // table must all fit before the first authenticated stream frame arrives.
60 pub const max_conns = 48;
58 61
59 /// Source Connection IDs cached per connection. ngtcp2 caps its own pool at 62 /// Source Connection IDs cached per connection. ngtcp2 caps its own pool at
60 /// NGTCP2_MAX_SCID_POOL_SIZE (8), so this is headroom; and because the 63 /// NGTCP2_MAX_SCID_POOL_SIZE (8), so this is headroom; and because the
src/server/server.zig
Old New
@@ -30,6 +30,7 @@ const upgrade_ops = @import("server_upgrade.zig");
30 const SessionTable = @import("server_sessions.zig").SessionTable; 30 const SessionTable = @import("server_sessions.zig").SessionTable;
31 const AgentRelay = agent_mod.AgentRelay; 31 const AgentRelay = agent_mod.AgentRelay;
32 const AgentSock = agent_mod.AgentSock; 32 const AgentSock = agent_mod.AgentSock;
33 const forward_mod = @import("server_forward.zig");
33 pub const max_agent_chans = agent_mod.max_agent_chans; 34 pub const max_agent_chans = agent_mod.max_agent_chans;
34 35
35 /// One slot per ATTACH, not per session: every tile on a wall is its own 36 /// One slot per ATTACH, not per session: every tile on a wall is its own
@@ -41,6 +42,28 @@ pub const max_agent_chans = agent_mod.max_agent_chans;
41 /// daemon is still refused, which is the intended ceiling and not a bug. 42 /// daemon is still refused, which is the intended ceiling and not a bug.
42 pub const max_clients = 32; 43 pub const max_clients = 32;
43 pub const max_observers = 4; 44 pub const max_observers = 4;
45 pub const max_forward_peers = forward_mod.max_peers;
46 /// Every transport-level QUIC connection can briefly be unclassified. Making
47 /// this smaller than the listener silently narrows the existing concurrent
48 /// attach capacity before the first stream frame has a chance to claim a role.
49 pub const max_quic_pending = quic_server.max_conns;
50 /// One legal daemon reply frame: `debug_dump` may use protocol's entire
51 /// 16 MiB payload allowance. This is per provisional peer; together with the
52 /// listener's 256 KiB egress ring it bounds a peer at this plus `egress_cap`.
53 pub const quic_one_shot_pending_cap = proto.max_payload + proto.frame_header_len;
54 pub const quic_one_shot_deadline_ms: i64 = 10_000;
55
56 const QuicPending = struct {
57 id: u64,
58 inbound: std.ArrayList(u8) = .empty,
59 /// A one-shot reply is retained outside the terminal table until the
60 /// QUIC egress ring has accepted and acknowledged its complete frame.
61 /// This keeps large dumps truthful without lending observer verbs a
62 /// terminal slot.
63 pending: std.ArrayList(u8) = .empty,
64 one_shot: bool = false,
65 since_ms: i64,
66 };
44 67
45 /// A connection that has not attached yet: see `Server.observers`. 68 /// A connection that has not attached yet: see `Server.observers`.
46 pub const Observer = struct { 69 pub const Observer = struct {
@@ -536,6 +559,12 @@ pub const Server = struct {
536 /// directory the per-session sockets are bound in. See server_agent.zig 559 /// directory the per-session sockets are bound in. See server_agent.zig
537 /// for why it is handed a `*Server` rather than holding one. 560 /// for why it is handed a `*Server` rather than holding one.
538 agents: AgentRelay = .{}, 561 agents: AgentRelay = .{},
562 /// Forwarding-role connections have their own admission and never enter
563 /// the terminal client/session tables.
564 forwards: forward_mod.Relay,
565 /// Authenticated QUIC connections wait here only until their first frame
566 /// declares terminal/observer traffic or the dedicated forward role.
567 quic_pending: [max_quic_pending]?QuicPending = @splat(null),
539 /// Descriptors the manifest handed over that no session could adopt — an 568 /// Descriptors the manifest handed over that no session could adopt — an
540 /// agent listener whose socket file vanished mid-upgrade. Closing one 569 /// agent listener whose socket file vanished mid-upgrade. Closing one
541 /// where the adopt fails would be wrong: until the last rollback point 570 /// where the adopt fails would be wrong: until the last rollback point
@@ -642,6 +671,7 @@ pub const Server = struct {
642 .shellint_arena = shellint_arena, 671 .shellint_arena = shellint_arena,
643 .shellint_dir = plan.shellint_dir, 672 .shellint_dir = plan.shellint_dir,
644 .agents = .{ .dir = agent_dir }, 673 .agents = .{ .dir = agent_dir },
674 .forwards = forward_mod.Relay.init(alloc),
645 }; 675 };
646 srv.sessions.table[0] = s0; 676 srv.sessions.table[0] = s0;
647 return srv; 677 return srv;
@@ -710,16 +740,18 @@ pub const Server = struct {
710 } 740 }
711 741
712 pub fn deinit(self: *Server) void { 742 pub fn deinit(self: *Server) void {
743 // A listener owns every QUIC connection, including provisional
744 // one-shot peers and forward-role peers. Announce the close before
745 // any of those owners quietly reap their entries, while their sink
746 // references and the borrowed listener are still live.
747 if (self.quicListener()) |listener| listener.closeAll();
748 self.forwards.deinit();
749 for (0..max_quic_pending) |i| self.dropQuicPending(i, true);
713 // Before the client slots, and silent: this loop frees the queues a 750 // Before the client slots, and silent: this loop frees the queues a
714 // notification would append to, and there is nobody left to tell — 751 // notification would append to, and there is nobody left to tell —
715 // the daemon is going. Every channel, not a per-session sweep: what 752 // the daemon is going. Every channel, not a per-session sweep: what
716 // has to happen here is that no descriptor outlives the table. 753 // has to happen here is that no descriptor outlives the table.
717 for (0..max_agent_chans) |s| self.agents.closeChan(self, s, .silent); 754 for (0..max_agent_chans) |s| self.agents.closeChan(self, s, .silent);
718 // Send CONNECTION_CLOSE while the listener and its QUIC connection
719 // objects are still alive. Closing client sinks first used to tear
720 // down the shared connection state without notifying peers, leaving
721 // remote stop requests waiting for their idle timeout.
722 if (self.quicListener()) |listener| listener.closeAll();
723 for (0..self.clients.len) |i| self.teardownClient(i, true); 755 for (0..self.clients.len) |i| self.teardownClient(i, true);
724 for (0..self.observers.len) |i| self.dropObserver(i); 756 for (0..self.observers.len) |i| self.dropObserver(i);
725 // After the client slots, never before: a QUIC sink closes its 757 // After the client slots, never before: a QUIC sink closes its
@@ -795,9 +827,10 @@ pub const Server = struct {
795 const obs_base = client_base + max_clients; 827 const obs_base = client_base + max_clients;
796 const agent_listener_base = obs_base + max_observers; 828 const agent_listener_base = obs_base + max_observers;
797 const agent_chan_base = agent_listener_base + max_sessions; 829 const agent_chan_base = agent_listener_base + max_sessions;
830 const forward_base = agent_chan_base + max_agent_chans;
798 var fds: [ 831 var fds: [
799 max_sessions + 1 + max_clients + max_observers + 832 max_sessions + 1 + max_clients + max_observers +
800 max_sessions + max_agent_chans + 1 833 max_sessions + max_agent_chans + forward_mod.poll_len + 1
801 ]std.posix.pollfd = undefined; 834 ]std.posix.pollfd = undefined;
802 for (&self.sessions.table, 0..) |*slot, si| { 835 for (&self.sessions.table, 0..) |*slot, si| {
803 fds[si] = pollIn(if (slot.*) |*s| s.pty.master else -1); 836 fds[si] = pollIn(if (slot.*) |*s| s.pty.master else -1);
@@ -833,6 +866,7 @@ pub const Server = struct {
833 for (self.agents.chans, 0..) |slot, s| { 866 for (self.agents.chans, 0..) |slot, s| {
834 fds[agent_chan_base + s] = pollIn(if (slot) |ch| ch.fd else -1); 867 fds[agent_chan_base + s] = pollIn(if (slot) |ch| ch.fd else -1);
835 } 868 }
869 self.forwards.fillPoll(fds[forward_base .. forward_base + forward_mod.poll_len]);
836 // One extra descriptor for every QUIC client there will ever be: 870 // One extra descriptor for every QUIC client there will ever be:
837 // they share it, which is the whole reason a client slot cannot be 871 // they share it, which is the whole reason a client slot cannot be
838 // a descriptor. 872 // a descriptor.
@@ -844,6 +878,7 @@ pub const Server = struct {
844 // time without a timerfd or a second loop. 878 // time without a timerfd or a second loop.
845 var wait_ms = timeout_ms; 879 var wait_ms = timeout_ms;
846 if (self.observerBacklog()) wait_ms = 0; 880 if (self.observerBacklog()) wait_ms = 0;
881 if (self.forwards.backlog()) wait_ms = 0;
847 if (self.quicListener()) |q| wait_ms = q.timeoutMs(wait_ms); 882 if (self.quicListener()) |q| wait_ms = q.timeoutMs(wait_ms);
848 _ = try std.posix.poll(&fds, wait_ms); 883 _ = try std.posix.poll(&fds, wait_ms);
849 884
@@ -856,6 +891,7 @@ pub const Server = struct {
856 // QUIC client has no descriptor of its own to go writable: without 891 // QUIC client has no descriptor of its own to go writable: without
857 // this a filled ring waits for the next frame by coincidence. 892 // this a filled ring waits for the next frame by coincidence.
858 self.flushQuicClients(); 893 self.flushQuicClients();
894 self.forwards.flushQuic();
859 } 895 }
860 896
861 for (&self.sessions.table, 0..) |*slot, si| { 897 for (&self.sessions.table, 0..) |*slot, si| {
@@ -916,6 +952,10 @@ pub const Server = struct {
916 } 952 }
917 } 953 }
918 self.reapIdleObservers(monoMs()); 954 self.reapIdleObservers(monoMs());
955 self.reapIdleQuicPending(monoMs());
956 self.flushQuicOneShots();
957
958 self.forwards.servicePoll(fds[forward_base .. forward_base + forward_mod.poll_len]);
919 959
920 // After the client arms, so a channel opened this pump is routed by 960 // After the client arms, so a channel opened this pump is routed by
921 // the activity order those frames left. Servicing BEFORE accepting 961 // the activity order those frames left. Servicing BEFORE accepting
@@ -1290,25 +1330,80 @@ pub const Server = struct {
1290 fn quicOnOpen(ctx: *anyopaque, id: u64) void { 1330 fn quicOnOpen(ctx: *anyopaque, id: u64) void {
1291 const self: *Server = @ptrCast(@alignCast(ctx)); 1331 const self: *Server = @ptrCast(@alignCast(ctx));
1292 const listener = self.quicListener() orelse return; 1332 const listener = self.quicListener() orelse return;
1293 const slot = self.freeClientSlot() orelse { 1333 for (&self.quic_pending) |*slot| if (slot.* == null) {
1294 // Session full: the same answer the socket path gives. A short 1334 slot.* = .{ .id = id, .since_ms = monoMs() };
1295 // accept would truncate the refusal into a corrupt frame — and on
1296 // a just-opened connection that means something is very wrong.
1297 _ = listener.send(id, &refusalFrame()) catch {};
1298 listener.closeConn(id);
1299 return; 1335 return;
1300 }; 1336 };
1301 self.clients[slot] = .{ .sink = .{ .quic = .{ .listener = listener, .id = id } } }; 1337 listener.closeConn(id);
1302 } 1338 }
1303 1339
1304 fn quicOnData(ctx: *anyopaque, id: u64, bytes: []const u8) void { 1340 fn quicOnData(ctx: *anyopaque, id: u64, bytes: []const u8) void {
1305 const self: *Server = @ptrCast(@alignCast(ctx)); 1341 const self: *Server = @ptrCast(@alignCast(ctx));
1306 const i = self.slotForQuic(id) orelse return; 1342 if (self.forwards.pushQuic(id, bytes)) return;
1307 self.pushInbound(i, bytes); 1343 if (self.slotForQuic(id)) |i| return self.pushInbound(i, bytes);
1344 const qi = self.pendingForQuic(id) orelse return;
1345 const q = &self.quic_pending[qi].?;
1346 if (q.one_shot) return;
1347 if (q.inbound.items.len + bytes.len > observer_inbound_max) return self.dropQuicPending(qi, true);
1348 q.inbound.appendSlice(self.alloc, bytes) catch return self.dropQuicPending(qi, true);
1349 const delimited = proto.delimitFrame(q.inbound.items) catch return self.dropQuicPending(qi, true);
1350 const first = delimited orelse return;
1351 if (first.type == .forward_hello) {
1352 if ((proto.decodeForwardHello(first.payload) catch return self.dropQuicPending(qi, true)) != proto.forward_version)
1353 return self.dropQuicPending(qi, true);
1354 q.inbound.replaceRangeAssumeCapacity(0, first.consumed, &.{});
1355 var moved = q.inbound;
1356 q.inbound = .empty;
1357 self.quic_pending[qi] = null;
1358 if (self.forwards.adoptQuic(self.quicListener().?, id, moved) == .no_room) {
1359 moved.deinit(self.alloc);
1360 self.quicListener().?.closeConn(id);
1361 }
1362 return;
1363 }
1364 // Only ATTACH owns an interactive client slot. One-shot replies stay
1365 // in this bounded provisional table until their queued frame drains.
1366 if (first.type != .attach) {
1367 const owned = (proto.takeFrame(self.alloc, &q.inbound) catch {
1368 self.dropQuicPending(qi, true);
1369 return;
1370 }) orelse unreachable;
1371 defer owned.deinit(self.alloc);
1372 // Set this before any reply allocation. replyTo may OOM-drop this
1373 // entry, so no code below may dereference `q` after it replies.
1374 q.one_shot = true;
1375 if (!self.handleDaemonVerb(.{ .quic_one = id }, owned))
1376 self.replyTo(.{ .quic_one = id }, .exit_status, &.{1});
1377 return;
1378 }
1379 // Validate before promotion. A malformed or unresolvable attach must
1380 // not consume a terminal slot merely because its header was complete.
1381 const req = proto.decodeAttach(first.payload) catch return self.dropQuicPending(qi, true);
1382 const slot = self.freeClientSlot() orelse {
1383 // replyTo can free this slot on OOM; mark before calling it.
1384 q.one_shot = true;
1385 self.replyTo(.{ .quic_one = id }, .exit_status, &.{1});
1386 return;
1387 };
1388 const si = self.sessions.resolve(self, req.name, req.cols, req.rows) orelse {
1389 q.one_shot = true;
1390 self.replyTo(.{ .quic_one = id }, .exit_status, &.{1});
1391 return;
1392 };
1393 const owned_attach = (proto.takeFrame(self.alloc, &q.inbound) catch return self.dropQuicPending(qi, true)) orelse unreachable;
1394 defer owned_attach.deinit(self.alloc);
1395 const moved = q.inbound;
1396 q.inbound = .empty;
1397 self.quic_pending[qi] = null;
1398 self.clients[slot] = .{ .sink = .{ .quic = .{ .listener = self.quicListener().?, .id = id } }, .session = si, .inbound = moved };
1399 self.seatClient(slot, si, req);
1400 self.pushInbound(slot, &.{});
1308 } 1401 }
1309 1402
1310 pub fn quicOnClose(ctx: *anyopaque, id: u64) void { 1403 pub fn quicOnClose(ctx: *anyopaque, id: u64) void {
1311 const self: *Server = @ptrCast(@alignCast(ctx)); 1404 const self: *Server = @ptrCast(@alignCast(ctx));
1405 if (self.forwards.closeQuic(id)) return;
1406 if (self.pendingForQuic(id)) |i| return self.dropQuicPending(i, false);
1312 const i = self.slotForQuic(id) orelse return; 1407 const i = self.slotForQuic(id) orelse return;
1313 // The conn is already freed, so there is no sink left to close. 1408 // The conn is already freed, so there is no sink left to close.
1314 self.teardownClient(i, false); 1409 self.teardownClient(i, false);
@@ -1325,6 +1420,69 @@ pub const Server = struct {
1325 return null; 1420 return null;
1326 } 1421 }
1327 1422
1423 fn pendingForQuic(self: *const Server, id: u64) ?usize {
1424 for (self.quic_pending, 0..) |slot, i| if (slot != null and slot.?.id == id) return i;
1425 return null;
1426 }
1427
1428 fn dropQuicPending(self: *Server, i: usize, close_conn: bool) void {
1429 if (self.quic_pending[i]) |*q| {
1430 const id = q.id;
1431 q.inbound.deinit(self.alloc);
1432 q.pending.deinit(self.alloc);
1433 self.quic_pending[i] = null;
1434 if (close_conn) if (self.quicListener()) |listener| listener.closeConn(id);
1435 }
1436 }
1437
1438 pub fn quicOneShotExpired(since_ms: i64, now_ms: i64) bool {
1439 return now_ms - since_ms >= quic_one_shot_deadline_ms;
1440 }
1441
1442 /// The pre-allocation admission check for the one retained response.
1443 /// Keeping it pure makes the 16 MiB boundary and overflow refusal pinable.
1444 pub fn quicOneShotCanQueue(owed: usize, payload_len: usize) bool {
1445 if (payload_len > proto.max_payload) return false;
1446 const frame_len = std.math.add(usize, payload_len, proto.frame_header_len) catch return false;
1447 return owed <= quic_one_shot_pending_cap and frame_len <= quic_one_shot_pending_cap - owed;
1448 }
1449
1450 fn reapIdleQuicPending(self: *Server, now_ms: i64) void {
1451 for (0..max_quic_pending) |i| {
1452 const q = self.quic_pending[i] orelse continue;
1453 // One-shots cannot be exempt forever: an ACK-withholding peer
1454 // otherwise occupies both a listener connection and this role
1455 // slot indefinitely. The explicit helper keeps this deterministic
1456 // under test without a sleep.
1457 if (if (q.one_shot) quicOneShotExpired(q.since_ms, now_ms) else now_ms - q.since_ms > self.observer_idle_ms)
1458 self.dropQuicPending(i, true);
1459 }
1460 }
1461
1462 /// Re-offer queued one-shot frames as ACKs release the listener ring. A
1463 /// close happens only after the whole frame is acknowledged, never after
1464 /// a short enqueue that would leave a valid header with a truncated body.
1465 fn flushQuicOneShots(self: *Server) void {
1466 const listener = self.quicListener() orelse return;
1467 for (0..max_quic_pending) |i| {
1468 const q = if (self.quic_pending[i]) |*p| p else continue;
1469 if (!q.one_shot) continue;
1470 if (q.pending.items.len != 0) {
1471 const n = listener.send(q.id, q.pending.items) catch {
1472 self.dropQuicPending(i, false);
1473 continue;
1474 };
1475 if (n != 0) q.pending.replaceRangeAssumeCapacity(0, n, &.{});
1476 }
1477 // The stop requester is its own shutdown acknowledgement: retain
1478 // its peer until Server.deinit's closeAll sends CONNECTION_CLOSE.
1479 // Quietly reaping it here turns a successful stop into an idle
1480 // timeout at the client.
1481 if (!shutdown_flag.load(.acquire) and q.pending.items.len == 0 and listener.pendingBytes(q.id) == 0)
1482 self.dropQuicPending(i, true);
1483 }
1484 }
1485
1328 pub fn quicHandler(self: *Server) quic_server.Handler { 1486 pub fn quicHandler(self: *Server) quic_server.Handler {
1329 return .{ 1487 return .{
1330 .ctx = self, 1488 .ctx = self,
@@ -1669,6 +1827,8 @@ pub const Server = struct {
1669 const Peer = union(enum) { 1827 const Peer = union(enum) {
1670 client: usize, 1828 client: usize,
1671 observer: usize, 1829 observer: usize,
1830 /// A first-frame QUIC daemon verb. It is never a terminal client.
1831 quic_one: u64,
1672 }; 1832 };
1673 1833
1674 /// The one send for a daemon verb; see handleDaemonVerb for why the two 1834 /// The one send for a daemon verb; see handleDaemonVerb for why the two
@@ -1681,6 +1841,18 @@ pub const Server = struct {
1681 proto.writeFrameBounded(o.fd, t, payload, proto.reply_budget_ms) catch 1841 proto.writeFrameBounded(o.fd, t, payload, proto.reply_budget_ms) catch
1682 self.dropObserver(i); 1842 self.dropObserver(i);
1683 }, 1843 },
1844 .quic_one => |id| {
1845 // A one-shot owns exactly one reply. Check before append so
1846 // an oversized/failed allocation never grows its ArrayList;
1847 // the legal 16 MiB debug dump plus its header still fits.
1848 const qi = self.pendingForQuic(id) orelse return;
1849 const q = &self.quic_pending[qi].?;
1850 if (!quicOneShotCanQueue(q.pending.items.len, payload.len)) {
1851 self.dropQuicPending(qi, true);
1852 return;
1853 }
1854 proto.appendFrame(&q.pending, self.alloc, t, payload) catch self.dropQuicPending(qi, true);
1855 },
1684 } 1856 }
1685 } 1857 }
1686 1858
@@ -1707,6 +1879,7 @@ pub const Server = struct {
1707 switch (p) { 1879 switch (p) {
1708 .client => |i| self.dropClient(i), 1880 .client => |i| self.dropClient(i),
1709 .observer => |i| self.dropObserver(i), 1881 .observer => |i| self.dropObserver(i),
1882 .quic_one => |id| if (self.quicListener()) |listener| listener.closeConn(id),
1710 } 1883 }
1711 } 1884 }
1712 1885
@@ -1754,6 +1927,17 @@ pub const Server = struct {
1754 const payload = proto.encodeEndpointReply(self.endpointPort()); 1927 const payload = proto.encodeEndpointReply(self.endpointPort());
1755 self.replyTo(p, .endpoint_reply, &payload); 1928 self.replyTo(p, .endpoint_reply, &payload);
1756 }, 1929 },
1930 .status_req => switch (p) {
1931 .quic_one => {
1932 const si = self.sessions.find(frame.payload) orelse {
1933 self.replyTo(p, .exit_status, &.{1});
1934 return true;
1935 };
1936 const payload = proto.encodeStatusReply(self.buildStatusReply(si));
1937 self.replyTo(p, .status_reply, &payload);
1938 },
1939 else => return false,
1940 },
1757 .debug_dump => { 1941 .debug_dump => {
1758 // A read against a NAME, not a question about this 1942 // A read against a NAME, not a question about this
1759 // connection's session: an attached client may peek at any 1943 // connection's session: an attached client may peek at any
@@ -1787,7 +1971,7 @@ pub const Server = struct {
1787 // count; an observer holds no session and excludes nobody. 1971 // count; an observer holds no session and excludes nobody.
1788 const v = self.endSession(frame.payload, switch (p) { 1972 const v = self.endSession(frame.payload, switch (p) {
1789 .client => |i| i, 1973 .client => |i| i,
1790 .observer => null, 1974 .observer, .quic_one => null,
1791 }); 1975 });
1792 var buf: [proto.end_reply_max_len]u8 = undefined; 1976 var buf: [proto.end_reply_max_len]u8 = undefined;
1793 self.replyTo(p, .end_reply, proto.encodeEndReply(&buf, v.accepted, v.others, v.reason)); 1977 self.replyTo(p, .end_reply, proto.encodeEndReply(&buf, v.accepted, v.others, v.reason));
@@ -2212,6 +2396,20 @@ pub const Server = struct {
2212 /// One observer frame. Slot `i` is live on entry; the `.attach` arm is 2396 /// One observer frame. Slot `i` is live on entry; the `.attach` arm is
2213 /// the one that ends the slot without dropping it. 2397 /// the one that ends the slot without dropping it.
2214 fn handleObserverFrame(self: *Server, i: usize, frame: proto.Frame) void { 2398 fn handleObserverFrame(self: *Server, i: usize, frame: proto.Frame) void {
2399 if (frame.type == .forward_hello) {
2400 const version = proto.decodeForwardHello(frame.payload) catch return self.dropObserver(i);
2401 if (version != proto.forward_version) return self.dropObserver(i);
2402 const o = &self.observers[i].?;
2403 const fd = o.fd;
2404 var moved = o.inbound;
2405 o.inbound = .empty;
2406 self.observers[i] = null;
2407 if (self.forwards.adoptSocket(fd, moved) == .no_room) {
2408 moved.deinit(self.alloc);
2409 std.posix.close(fd);
2410 }
2411 return;
2412 }
2215 if (self.handleDaemonVerb(.{ .observer = i }, frame)) return; 2413 if (self.handleDaemonVerb(.{ .observer = i }, frame)) return;
2216 const fd = self.observers[i].?.fd; 2414 const fd = self.observers[i].?.fd;
2217 switch (frame.type) { 2415 switch (frame.type) {
@@ -2913,5 +3111,6 @@ test {
2913 _ = @import("server_test_quic.zig"); 3111 _ = @import("server_test_quic.zig");
2914 _ = @import("server_test_session.zig"); 3112 _ = @import("server_test_session.zig");
2915 _ = @import("server_test_upgrade.zig"); 3113 _ = @import("server_test_upgrade.zig");
3114 _ = @import("server_forward.zig");
2916 _ = @import("quic_server.zig"); 3115 _ = @import("quic_server.zig");
2917 } 3116 }
src/server/server_forward.zig
Old New
@@ -0,0 +1,765 @@
1 //! Dedicated forwarding-role peers and their remote-loopback TCP channels.
2 //! This is daemon state, but deliberately has no Session or terminal imports:
3 //! accepting a forwarding role cannot create a shell or claim geometry.
4 const std = @import("std");
5 const proto = @import("term").protocol;
6 const quic_server = @import("quic_server.zig");
7 const server_os = @import("server_os");
8
9 pub const max_peers: usize = 4;
10 pub const poll_len: usize = max_peers * (1 + proto.forward_channels_max);
11 const peer_queue_max: usize = 1024 * 1024;
12 const control_reserve: usize = 64 * 1024;
13 const channel_queue_max: usize = proto.forward_initial_credit;
14 const frames_per_service: usize = 64;
15 const connect_deadline_ms: i64 = 5_000;
16
17 fn monoMs() i64 {
18 const t = std.posix.clock_gettime(.MONOTONIC) catch return std.time.milliTimestamp();
19 return @as(i64, t.sec) * 1000 + @divFloor(t.nsec, 1_000_000);
20 }
21
22 const Sink = union(enum) {
23 socket: std.posix.fd_t,
24 quic: struct { listener: *quic_server.Listener, id: u64 },
25
26 fn pollFd(self: Sink) std.posix.fd_t {
27 return switch (self) {
28 .socket => |fd| fd,
29 .quic => -1,
30 };
31 }
32 fn send(self: Sink, bytes: []const u8) !usize {
33 return switch (self) {
34 .socket => |fd| server_os.sendNoSigNoWait(fd, bytes),
35 .quic => |q| q.listener.send(q.id, bytes),
36 };
37 }
38 fn inFlight(self: Sink) usize {
39 return switch (self) {
40 .socket => 0,
41 .quic => |q| q.listener.pendingBytes(q.id),
42 };
43 }
44 fn close(self: Sink) void {
45 switch (self) {
46 .socket => |fd| std.posix.close(fd),
47 .quic => |q| q.listener.closeConn(q.id),
48 }
49 }
50 };
51
52 const Channel = struct {
53 generation: u64,
54 id: u32,
55 fd: std.posix.fd_t,
56 connecting: bool,
57 /// Monotonic deadline for EINPROGRESS. A descriptor that never wakes must
58 /// not occupy a forwarding channel forever.
59 connect_by_ms: ?i64 = null,
60 send_credit: u32 = 0,
61 recv_credit: u32 = 0,
62 to_tcp: std.ArrayList(u8) = .empty,
63 local_eof: bool = false,
64 peer_eof: bool = false,
65 write_shutdown: bool = false,
66 };
67
68 const Peer = struct {
69 generation: u64,
70 sink: Sink,
71 inbound: std.ArrayList(u8) = .empty,
72 pending: std.ArrayList(u8) = .empty,
73 channels: [proto.forward_channels_max]?Channel = @splat(null),
74 };
75
76 pub const Relay = struct {
77 alloc: std.mem.Allocator,
78 peers: [max_peers]?Peer = @splat(null),
79 next_generation: u64 = 1,
80 poll_peer_generation: [max_peers]u64 = @splat(0),
81 poll_channel_generation: [max_peers * proto.forward_channels_max]u64 = @splat(0),
82
83 pub fn init(alloc: std.mem.Allocator) Relay {
84 return .{ .alloc = alloc };
85 }
86
87 pub fn deinit(self: *Relay) void {
88 for (0..max_peers) |i| self.dropPeer(i, true);
89 }
90
91 pub fn freePeer(self: *const Relay) ?usize {
92 for (self.peers, 0..) |peer, i| if (peer == null) return i;
93 return null;
94 }
95
96 pub const Adopt = enum { adopted, no_room };
97
98 pub fn adoptSocket(self: *Relay, fd: std.posix.fd_t, inbound: std.ArrayList(u8)) Adopt {
99 const i = self.freePeer() orelse return .no_room;
100 self.peers[i] = .{ .generation = self.takeGeneration(), .sink = .{ .socket = fd }, .inbound = inbound };
101 self.ready(i);
102 self.drainFrames(i);
103 return .adopted;
104 }
105
106 pub fn adoptQuic(self: *Relay, listener: *quic_server.Listener, id: u64, inbound: std.ArrayList(u8)) Adopt {
107 const i = self.freePeer() orelse return .no_room;
108 self.peers[i] = .{ .generation = self.takeGeneration(), .sink = .{ .quic = .{ .listener = listener, .id = id } }, .inbound = inbound };
109 self.ready(i);
110 self.drainFrames(i);
111 return .adopted;
112 }
113
114 pub fn pushQuic(self: *Relay, id: u64, bytes: []const u8) bool {
115 for (self.peers, 0..) |peer, i| {
116 const p = peer orelse continue;
117 if (p.sink == .quic and p.sink.quic.id == id) {
118 self.push(i, bytes);
119 return true;
120 }
121 }
122 return false;
123 }
124
125 pub fn closeQuic(self: *Relay, id: u64) bool {
126 for (self.peers, 0..) |peer, i| {
127 const p = peer orelse continue;
128 if (p.sink == .quic and p.sink.quic.id == id) {
129 self.dropPeer(i, false);
130 return true;
131 }
132 }
133 return false;
134 }
135
136 pub fn flushQuic(self: *Relay) void {
137 for (0..max_peers) |i| {
138 const p = if (self.peers[i]) |*peer| peer else continue;
139 if (p.sink == .quic) self.flushPeer(i);
140 }
141 }
142
143 pub fn fillPoll(self: *Relay, fds: []std.posix.pollfd) void {
144 std.debug.assert(fds.len == poll_len);
145 var at: usize = 0;
146 for (self.peers, 0..) |peer, pi| {
147 if (peer) |p| {
148 self.poll_peer_generation[pi] = p.generation;
149 var events: i16 = std.posix.POLL.IN;
150 if (p.pending.items.len != 0) events |= std.posix.POLL.OUT;
151 fds[at] = .{ .fd = p.sink.pollFd(), .events = events, .revents = 0 };
152 at += 1;
153 for (p.channels, 0..) |channel, ci| {
154 if (channel) |ch| {
155 var ce: i16 = 0;
156 if (!ch.local_eof and self.dataReadLimit(p, ch) != 0) ce |= std.posix.POLL.IN;
157 if (ch.connecting or ch.to_tcp.items.len != 0 or (ch.peer_eof and !ch.write_shutdown)) ce |= std.posix.POLL.OUT;
158 fds[at] = .{ .fd = if (ce == 0) -1 else ch.fd, .events = ce, .revents = 0 };
159 self.poll_channel_generation[pi * proto.forward_channels_max + ci] = ch.generation;
160 } else fds[at] = .{ .fd = -1, .events = 0, .revents = 0 };
161 at += 1;
162 }
163 } else {
164 self.poll_peer_generation[pi] = 0;
165 for (0..1 + proto.forward_channels_max) |_| {
166 fds[at] = .{ .fd = -1, .events = 0, .revents = 0 };
167 at += 1;
168 }
169 }
170 }
171 }
172
173 pub fn servicePoll(self: *Relay, fds: []const std.posix.pollfd) void {
174 std.debug.assert(fds.len == poll_len);
175 self.reapConnecting(monoMs());
176 var at: usize = 0;
177 for (0..max_peers) |pi| {
178 const peer_events = fds[at].revents;
179 at += 1;
180 const channel_start = at;
181 at += proto.forward_channels_max;
182 if (self.peers[pi] == null or self.peers[pi].?.generation != self.poll_peer_generation[pi]) continue;
183 // Service the descriptors represented by this poll snapshot before
184 // peer frames can reset and reuse their table slots.
185 for (0..proto.forward_channels_max) |ci| {
186 if (self.peers[pi] == null) break;
187 if (self.peers[pi].?.channels[ci] == null) continue;
188 if (self.peers[pi].?.channels[ci].?.generation != self.poll_channel_generation[pi * proto.forward_channels_max + ci]) continue;
189 const revents = fds[channel_start + ci].revents;
190 if (revents == 0) continue;
191 self.serviceChannel(pi, ci, revents);
192 }
193 if (self.peers[pi] == null or self.peers[pi].?.generation != self.poll_peer_generation[pi]) continue;
194 if (peer_events & std.posix.POLL.OUT != 0) self.flushPeer(pi);
195 if (self.peers[pi] == null) continue;
196 if (peer_events & ~@as(i16, std.posix.POLL.OUT) != 0) self.readPeer(pi);
197 if (self.peers[pi] != null) self.drainFrames(pi);
198 }
199 }
200
201 pub fn backlog(self: *const Relay) bool {
202 for (self.peers) |peer| {
203 const p = peer orelse continue;
204 const framed = proto.delimitFrame(p.inbound.items) catch return true;
205 if (framed != null) return true;
206 }
207 return false;
208 }
209
210 fn ready(self: *Relay, pi: usize) void {
211 const payload = proto.encodeForwardHello();
212 self.queue(pi, .forward_ready, &payload);
213 }
214
215 fn push(self: *Relay, pi: usize, bytes: []const u8) void {
216 const p = if (self.peers[pi]) |*peer| peer else return;
217 if (p.inbound.items.len + bytes.len > peer_queue_max) return self.dropPeer(pi, true);
218 p.inbound.appendSlice(self.alloc, bytes) catch return self.dropPeer(pi, true);
219 self.drainFrames(pi);
220 }
221
222 fn readPeer(self: *Relay, pi: usize) void {
223 const p = if (self.peers[pi]) |*peer| peer else return;
224 if (p.sink != .socket) return;
225 var buf: [64 * 1024]u8 = undefined;
226 const n = std.posix.read(p.sink.socket, &buf) catch |err| switch (err) {
227 error.WouldBlock => return,
228 else => return self.dropPeer(pi, true),
229 };
230 if (n == 0) return self.dropPeer(pi, true);
231 self.push(pi, buf[0..n]);
232 }
233
234 fn drainFrames(self: *Relay, pi: usize) void {
235 for (0..frames_per_service) |_| {
236 const p = if (self.peers[pi]) |*peer| peer else return;
237 const frame = proto.takeFrame(self.alloc, &p.inbound) catch return self.dropPeer(pi, true);
238 if (frame == null) return;
239 const f = frame.?;
240 defer f.deinit(self.alloc);
241 if (!self.handleFrame(pi, f)) return self.dropPeer(pi, true);
242 }
243 }
244
245 fn handleFrame(self: *Relay, pi: usize, frame: proto.Frame) bool {
246 switch (frame.type) {
247 .forward_open => {
248 const request = proto.decodeForwardOpen(frame.payload) catch return false;
249 self.openChannel(pi, request);
250 },
251 .forward_data => {
252 if (proto.forwardDataOversize(frame.payload)) return false;
253 const id = proto.decodeForwardId(frame.payload) catch return false;
254 const ci = self.findChannel(pi, id) orelse return true; // late frame for a reset channel
255 const ch = &self.peers[pi].?.channels[ci].?;
256 const data = frame.payload[proto.forward_id_len..];
257 if (ch.peer_eof or data.len > ch.recv_credit or ch.to_tcp.items.len + data.len > channel_queue_max) {
258 self.resetChannel(pi, ci, true);
259 return true;
260 }
261 ch.recv_credit -= @intCast(data.len);
262 ch.to_tcp.appendSlice(self.alloc, data) catch self.resetChannel(pi, ci, true);
263 if (self.peers[pi] != null and self.peers[pi].?.channels[ci] != null) self.flushTcp(pi, ci);
264 },
265 .forward_credit => {
266 const credit = proto.decodeForwardCredit(frame.payload) catch return false;
267 const ci = self.findChannel(pi, credit.id) orelse return true;
268 const ch = &self.peers[pi].?.channels[ci].?;
269 if (credit.amount > proto.forward_initial_credit -| ch.send_credit) return false;
270 ch.send_credit += credit.amount;
271 },
272 .forward_half_close => {
273 const id = proto.decodeForwardId(frame.payload) catch return false;
274 if (frame.payload.len != proto.forward_id_len) return false;
275 const ci = self.findChannel(pi, id) orelse return true;
276 self.peers[pi].?.channels[ci].?.peer_eof = true;
277 self.flushTcp(pi, ci);
278 },
279 .forward_reset => {
280 const id = proto.decodeForwardId(frame.payload) catch return false;
281 if (frame.payload.len != proto.forward_id_len) return false;
282 const ci = self.findChannel(pi, id) orelse return true;
283 self.resetChannel(pi, ci, false);
284 },
285 else => return false,
286 }
287 return self.peers[pi] != null;
288 }
289
290 fn openChannel(self: *Relay, pi: usize, request: proto.ForwardOpen) void {
291 if (self.findChannel(pi, request.id) != null) return self.refuseOpen(pi, request.id);
292 var ci: ?usize = null;
293 for (self.peers[pi].?.channels, 0..) |ch, i| if (ch == null) {
294 ci = i;
295 break;
296 };
297 const slot = ci orelse return self.refuseOpen(pi, request.id);
298 const fd = std.posix.socket(std.posix.AF.INET, std.posix.SOCK.STREAM | std.posix.SOCK.NONBLOCK | std.posix.SOCK.CLOEXEC, std.posix.IPPROTO.TCP) catch return self.refuseOpen(pi, request.id);
299 const addr = std.net.Address.parseIp4("127.0.0.1", request.port) catch unreachable;
300 var connecting = false;
301 std.posix.connect(fd, &addr.any, addr.getOsSockLen()) catch |err| switch (err) {
302 error.WouldBlock => connecting = true,
303 else => {
304 std.posix.close(fd);
305 return self.refuseOpen(pi, request.id);
306 },
307 };
308 self.peers[pi].?.channels[slot] = .{ .generation = self.takeGeneration(), .id = request.id, .fd = fd, .connecting = connecting, .connect_by_ms = if (connecting) monoMs() + connect_deadline_ms else null };
309 if (!connecting) self.acceptOpen(pi, slot);
310 }
311
312 fn acceptOpen(self: *Relay, pi: usize, ci: usize) void {
313 const id = self.peers[pi].?.channels[ci].?.id;
314 self.peers[pi].?.channels[ci].?.connecting = false;
315 self.peers[pi].?.channels[ci].?.connect_by_ms = null;
316 self.peers[pi].?.channels[ci].?.recv_credit = proto.forward_initial_credit;
317 const result = proto.encodeForwardOpenResult(.{ .id = id, .ok = true });
318 self.queue(pi, .forward_open_result, &result);
319 const credit = proto.encodeForwardCredit(.{ .id = id, .amount = proto.forward_initial_credit });
320 self.queue(pi, .forward_credit, &credit);
321 }
322
323 fn refuseOpen(self: *Relay, pi: usize, id: u32) void {
324 const result = proto.encodeForwardOpenResult(.{ .id = id, .ok = false });
325 self.queue(pi, .forward_open_result, &result);
326 }
327
328 /// Reaped from the regular poll service even if a nonblocking connect
329 /// never produces POLLOUT/POLLERR. Only the expired channel is refused.
330 fn reapConnecting(self: *Relay, now_ms: i64) void {
331 for (0..max_peers) |pi| {
332 for (0..proto.forward_channels_max) |ci| {
333 const ch = self.peers[pi] orelse continue;
334 const channel = ch.channels[ci] orelse continue;
335 const by = channel.connect_by_ms orelse continue;
336 if (channel.connecting and now_ms >= by) {
337 const id = channel.id;
338 self.resetChannel(pi, ci, false);
339 self.refuseOpen(pi, id);
340 }
341 }
342 }
343 }
344
345 fn serviceChannel(self: *Relay, pi: usize, ci: usize, revents: i16) void {
346 var ch = &self.peers[pi].?.channels[ci].?;
347 // Error and hangup bits are reported even when they were not requested.
348 // SO_ERROR is the verdict for every connect wakeup; requiring POLLOUT
349 // can otherwise leave a refused channel permanently "connecting".
350 if (ch.connecting and revents != 0) {
351 std.posix.getsockoptError(ch.fd) catch {
352 const id = ch.id;
353 self.resetChannel(pi, ci, false);
354 return self.refuseOpen(pi, id);
355 };
356 self.acceptOpen(pi, ci);
357 }
358 if (self.peers[pi] == null or self.peers[pi].?.channels[ci] == null) return;
359 ch = &self.peers[pi].?.channels[ci].?;
360 if (!ch.connecting and revents & std.posix.POLL.OUT != 0) self.flushTcp(pi, ci);
361 if (self.peers[pi] == null or self.peers[pi].?.channels[ci] == null) return;
362 ch = &self.peers[pi].?.channels[ci].?;
363 if (!ch.connecting and revents & ~@as(i16, std.posix.POLL.OUT) != 0) self.readTcp(pi, ci);
364 }
365
366 fn flushTcp(self: *Relay, pi: usize, ci: usize) void {
367 const ch = &self.peers[pi].?.channels[ci].?;
368 if (ch.connecting) return;
369 if (ch.to_tcp.items.len != 0) {
370 const n = server_os.sendNoSigNoWait(ch.fd, ch.to_tcp.items) catch |err| switch (err) {
371 error.WouldBlock => return,
372 else => return self.resetChannel(pi, ci, true),
373 };
374 ch.to_tcp.replaceRangeAssumeCapacity(0, n, &.{});
375 ch.recv_credit += @intCast(n);
376 const credit = proto.encodeForwardCredit(.{ .id = ch.id, .amount = @intCast(n) });
377 self.queue(pi, .forward_credit, &credit);
378 }
379 if (self.peers[pi] == null or self.peers[pi].?.channels[ci] == null) return;
380 const current = &self.peers[pi].?.channels[ci].?;
381 if (current.peer_eof and current.to_tcp.items.len == 0 and !current.write_shutdown) {
382 std.posix.shutdown(current.fd, .send) catch {};
383 current.write_shutdown = true;
384 }
385 self.maybeFinish(pi, ci);
386 }
387
388 fn readTcp(self: *Relay, pi: usize, ci: usize) void {
389 const ch = &self.peers[pi].?.channels[ci].?;
390 if (ch.local_eof or ch.send_credit == 0) return;
391 // All readable channels in this poll pass observed one shared queue.
392 // Recheck it for this channel before taking bytes from the TCP socket.
393 const want = self.dataReadLimit(self.peers[pi].?, ch.*);
394 if (want == 0) return;
395 var buf: [proto.forward_data_max]u8 = undefined;
396 const n = std.posix.read(ch.fd, buf[0..want]) catch |err| switch (err) {
397 error.WouldBlock => return,
398 else => return self.resetChannel(pi, ci, true),
399 };
400 if (n == 0) {
401 ch.local_eof = true;
402 const id = proto.encodeForwardId(ch.id);
403 self.queue(pi, .forward_half_close, &id);
404 if (self.peers[pi] == null or self.peers[pi].?.channels[ci] == null) return;
405 return self.maybeFinish(pi, ci);
406 }
407 ch.send_credit -= @intCast(n);
408 var payload: [proto.forward_id_len + proto.forward_data_max]u8 = undefined;
409 std.mem.writeInt(u32, payload[0..4], ch.id, .little);
410 @memcpy(payload[4..][0..n], buf[0..n]);
411 self.queue(pi, .forward_data, payload[0 .. 4 + n]);
412 }
413
414 fn maybeFinish(self: *Relay, pi: usize, ci: usize) void {
415 if (self.peers[pi] == null or self.peers[pi].?.channels[ci] == null) return;
416 const ch = &self.peers[pi].?.channels[ci].?;
417 if (ch.local_eof and ch.peer_eof and ch.to_tcp.items.len == 0) self.resetChannel(pi, ci, false);
418 }
419
420 fn findChannel(self: *const Relay, pi: usize, id: u32) ?usize {
421 const p = self.peers[pi] orelse return null;
422 for (p.channels, 0..) |ch, i| if (ch != null and ch.?.id == id) return i;
423 return null;
424 }
425
426 fn resetChannel(self: *Relay, pi: usize, ci: usize, notify: bool) void {
427 const p = if (self.peers[pi]) |*peer| peer else return;
428 if (p.channels[ci]) |*ch| {
429 const id = ch.id;
430 std.posix.close(ch.fd);
431 ch.to_tcp.deinit(self.alloc);
432 p.channels[ci] = null;
433 if (notify) {
434 const payload = proto.encodeForwardId(id);
435 self.queue(pi, .forward_reset, &payload);
436 }
437 }
438 }
439
440 fn queue(self: *Relay, pi: usize, kind: proto.MsgType, payload: []const u8) void {
441 const p = if (self.peers[pi]) |*peer| peer else return;
442 const frame_len = proto.frame_header_len + payload.len;
443 if (p.pending.items.len + p.sink.inFlight() > peer_queue_max -| frame_len) return self.dropPeer(pi, true);
444 proto.appendFrame(&p.pending, self.alloc, kind, payload) catch return self.dropPeer(pi, true);
445 self.flushPeer(pi);
446 }
447
448 fn flushPeer(self: *Relay, pi: usize) void {
449 const p = if (self.peers[pi]) |*peer| peer else return;
450 if (p.pending.items.len == 0) return;
451 const n = p.sink.send(p.pending.items) catch |err| switch (err) {
452 error.WouldBlock => return,
453 else => return self.dropPeer(pi, true),
454 };
455 p.pending.replaceRangeAssumeCapacity(0, n, &.{});
456 }
457
458 fn dropPeer(self: *Relay, pi: usize, close_sink: bool) void {
459 if (self.peers[pi]) |*p| {
460 for (0..proto.forward_channels_max) |ci| self.resetChannel(pi, ci, false);
461 p.inbound.deinit(self.alloc);
462 p.pending.deinit(self.alloc);
463 if (close_sink) p.sink.close();
464 }
465 self.peers[pi] = null;
466 }
467
468 fn takeGeneration(self: *Relay) u64 {
469 const result = self.next_generation;
470 self.next_generation +%= 1;
471 if (self.next_generation == 0) self.next_generation = 1;
472 return result;
473 }
474
475 fn dataReadLimit(_: *const Relay, peer: Peer, ch: Channel) usize {
476 const overhead = proto.frame_header_len + proto.forward_id_len;
477 const pending = peer.pending.items.len + peer.sink.inFlight();
478 if (pending >= peer_queue_max - control_reserve - overhead) return 0;
479 const room = peer_queue_max - control_reserve - overhead - pending;
480 return @min(proto.forward_data_max, @min(@as(usize, ch.send_credit), room));
481 }
482 };
483
484 fn testSocketPair() ![2]std.posix.fd_t {
485 var pair: [2]std.posix.fd_t = undefined;
486 if (std.c.socketpair(std.posix.AF.UNIX, std.posix.SOCK.STREAM, 0, &pair) != 0) return error.SocketPairFailed;
487 return pair;
488 }
489
490 fn testListener() !struct { fd: std.posix.fd_t, port: u16 } {
491 const fd = try std.posix.socket(std.posix.AF.INET, std.posix.SOCK.STREAM | std.posix.SOCK.NONBLOCK | std.posix.SOCK.CLOEXEC, std.posix.IPPROTO.TCP);
492 errdefer std.posix.close(fd);
493 const addr = try std.net.Address.parseIp4("127.0.0.1", 0);
494 try std.posix.bind(fd, &addr.any, addr.getOsSockLen());
495 try std.posix.listen(fd, 8);
496 var actual: std.posix.sockaddr.storage = undefined;
497 var len: std.posix.socklen_t = @sizeOf(@TypeOf(actual));
498 try std.posix.getsockname(fd, @ptrCast(&actual), &len);
499 return .{ .fd = fd, .port = std.net.Address.initPosix(@ptrCast(@alignCast(&actual))).getPort() };
500 }
501
502 fn testPush(relay: *Relay, kind: proto.MsgType, payload: []const u8) !void {
503 var bytes: std.ArrayList(u8) = .empty;
504 defer bytes.deinit(std.testing.allocator);
505 try proto.appendFrame(&bytes, std.testing.allocator, kind, payload);
506 relay.push(0, bytes.items);
507 }
508
509 fn testPump(relay: *Relay, timeout_ms: i32) !void {
510 var fds: [poll_len]std.posix.pollfd = undefined;
511 relay.fillPoll(&fds);
512 _ = try std.posix.poll(&fds, timeout_ms);
513 relay.servicePoll(&fds);
514 }
515
516 fn testReadFrame(fd: std.posix.fd_t, expected: proto.MsgType) !proto.Frame {
517 const frame = (try proto.readFrame(std.testing.allocator, fd)) orelse return error.UnexpectedEof;
518 errdefer frame.deinit(std.testing.allocator);
519 try std.testing.expectEqual(expected, frame.type);
520 return frame;
521 }
522
523 test "relay carries request response credit and orderly half-closes over loopback" {
524 const pair = try testSocketPair();
525 defer std.posix.close(pair[1]);
526 var relay = Relay.init(std.testing.allocator);
527 defer relay.deinit();
528 try std.testing.expectEqual(Relay.Adopt.adopted, relay.adoptSocket(pair[0], .empty));
529
530 const ready = try testReadFrame(pair[1], .forward_ready);
531 defer ready.deinit(std.testing.allocator);
532 try std.testing.expectEqual(proto.forward_version, try proto.decodeForwardHello(ready.payload));
533
534 const listener = try testListener();
535 defer std.posix.close(listener.fd);
536 const open = proto.encodeForwardOpen(.{ .id = 7, .port = listener.port });
537 try testPush(&relay, .forward_open, &open);
538
539 var remote: ?std.posix.fd_t = null;
540 for (0..20) |_| {
541 try testPump(&relay, 20);
542 remote = std.posix.accept(listener.fd, null, null, std.posix.SOCK.CLOEXEC) catch |err| switch (err) {
543 error.WouldBlock => null,
544 else => return err,
545 };
546 if (remote != null and relay.peers[0].?.channels[0].?.connecting == false) break;
547 }
548 const tcp = remote orelse return error.AcceptTimedOut;
549 defer std.posix.close(tcp);
550 try std.testing.expect(!relay.peers[0].?.channels[0].?.connecting);
551
552 const opened = try testReadFrame(pair[1], .forward_open_result);
553 defer opened.deinit(std.testing.allocator);
554 try std.testing.expect((try proto.decodeForwardOpenResult(opened.payload)).ok);
555 const initial = try testReadFrame(pair[1], .forward_credit);
556 defer initial.deinit(std.testing.allocator);
557 try std.testing.expectEqual(proto.forward_initial_credit, (try proto.decodeForwardCredit(initial.payload)).amount);
558
559 const outbound_credit = proto.encodeForwardCredit(.{ .id = 7, .amount = proto.forward_initial_credit });
560 try testPush(&relay, .forward_credit, &outbound_credit);
561 var request: [proto.forward_id_len + 4]u8 = undefined;
562 std.mem.writeInt(u32, request[0..4], 7, .little);
563 @memcpy(request[4..], "ping");
564 try testPush(&relay, .forward_data, &request);
565 var got_request: [4]u8 = undefined;
566 try std.testing.expectEqual(@as(usize, 4), try std.posix.read(tcp, &got_request));
567 try std.testing.expectEqualStrings("ping", &got_request);
568 const replenished = try testReadFrame(pair[1], .forward_credit);
569 defer replenished.deinit(std.testing.allocator);
570 try std.testing.expectEqual(@as(u32, 4), (try proto.decodeForwardCredit(replenished.payload)).amount);
571
572 try std.testing.expectEqual(@as(usize, 4), try std.posix.write(tcp, "pong"));
573 for (0..20) |_| {
574 try testPump(&relay, 20);
575 if (relay.peers[0].?.channels[0].?.send_credit == proto.forward_initial_credit - 4) break;
576 }
577 const response = try testReadFrame(pair[1], .forward_data);
578 defer response.deinit(std.testing.allocator);
579 try std.testing.expectEqual(@as(u32, 7), try proto.decodeForwardId(response.payload));
580 try std.testing.expectEqualStrings("pong", response.payload[proto.forward_id_len..]);
581
582 const id = proto.encodeForwardId(7);
583 try testPush(&relay, .forward_half_close, &id);
584 var eof_byte: [1]u8 = undefined;
585 try std.testing.expectEqual(@as(usize, 0), try std.posix.read(tcp, &eof_byte));
586 try std.posix.shutdown(tcp, .send);
587 for (0..20) |_| {
588 try testPump(&relay, 20);
589 if (relay.peers[0].?.channels[0] == null) break;
590 }
591 try std.testing.expect(relay.peers[0].?.channels[0] == null);
592 const half = try testReadFrame(pair[1], .forward_half_close);
593 defer half.deinit(std.testing.allocator);
594 try std.testing.expectEqual(@as(u32, 7), try proto.decodeForwardId(half.payload));
595 }
596
597 test "expired nonblocking connects refuse only their channels and release slots" {
598 const pair = try testSocketPair();
599 defer std.posix.close(pair[1]);
600 var relay = Relay.init(std.testing.allocator);
601 defer relay.deinit();
602 try std.testing.expectEqual(Relay.Adopt.adopted, relay.adoptSocket(pair[0], .empty));
603 const ready = try testReadFrame(pair[1], .forward_ready);
604 ready.deinit(std.testing.allocator);
605
606 const first = try testSocketPair();
607 defer std.posix.close(first[1]);
608 const second = try testSocketPair();
609 defer std.posix.close(second[1]);
610 const live = try testSocketPair();
611 defer std.posix.close(live[1]);
612 relay.peers[0].?.channels[0] = .{ .generation = relay.takeGeneration(), .id = 1, .fd = first[0], .connecting = true, .connect_by_ms = 10 };
613 relay.peers[0].?.channels[1] = .{ .generation = relay.takeGeneration(), .id = 2, .fd = second[0], .connecting = true, .connect_by_ms = 10 };
614 relay.peers[0].?.channels[2] = .{ .generation = relay.takeGeneration(), .id = 3, .fd = live[0], .connecting = false };
615
616 // `servicePoll`, rather than a direct helper call, is the production
617 // deadline path: it runs even when every descriptor is quiet. A stuck
618 // EINPROGRESS must therefore be reaped by a zero-event poll snapshot.
619 var fds: [poll_len]std.posix.pollfd = undefined;
620 relay.fillPoll(&fds);
621 relay.servicePoll(&fds);
622 try std.testing.expect(relay.peers[0].?.channels[0] == null);
623 try std.testing.expect(relay.peers[0].?.channels[1] == null);
624 try std.testing.expect(relay.peers[0].?.channels[2] != null);
625 // resetChannel owns the descriptor, not just the table entry. Leaving
626 // this open turns repeated timed-out opens into an fd leak.
627 // std.posix.fcntl deliberately treats EBADF as unreachable, so ask its
628 // raw syscall wrapper for this OS-level ownership assertion.
629 try std.testing.expectEqual(std.posix.E.BADF, std.posix.errno(std.posix.system.fcntl(first[0], std.posix.F.GETFD, @as(c_int, 0))));
630 try std.testing.expectEqual(std.posix.E.BADF, std.posix.errno(std.posix.system.fcntl(second[0], std.posix.F.GETFD, @as(c_int, 0))));
631 for ([_]u32{ 1, 2 }) |id| {
632 const refused = try testReadFrame(pair[1], .forward_open_result);
633 defer refused.deinit(std.testing.allocator);
634 const result = try proto.decodeForwardOpenResult(refused.payload);
635 try std.testing.expectEqual(id, result.id);
636 try std.testing.expect(!result.ok);
637 }
638 const reuse = try testSocketPair();
639 defer std.posix.close(reuse[1]);
640 relay.peers[0].?.channels[0] = .{ .generation = relay.takeGeneration(), .id = 4, .fd = reuse[0], .connecting = false };
641 try std.testing.expect(relay.peers[0].?.channels[0] != null);
642 }
643
644 test "channel flow violation resets only that channel and credit overflow drops the peer" {
645 const pair = try testSocketPair();
646 defer std.posix.close(pair[1]);
647 var relay = Relay.init(std.testing.allocator);
648 defer relay.deinit();
649 try std.testing.expectEqual(Relay.Adopt.adopted, relay.adoptSocket(pair[0], .empty));
650 const ready = try testReadFrame(pair[1], .forward_ready);
651 ready.deinit(std.testing.allocator);
652
653 const first = try testSocketPair();
654 defer std.posix.close(first[1]);
655 const second = try testSocketPair();
656 defer std.posix.close(second[1]);
657 relay.peers[0].?.channels[0] = .{ .generation = relay.takeGeneration(), .id = 1, .fd = first[0], .connecting = false, .recv_credit = 4 };
658 relay.peers[0].?.channels[1] = .{ .generation = relay.takeGeneration(), .id = 2, .fd = second[0], .connecting = false };
659
660 var excess: [proto.forward_id_len + 5]u8 = undefined;
661 std.mem.writeInt(u32, excess[0..4], 1, .little);
662 @memcpy(excess[4..], "12345");
663 try testPush(&relay, .forward_data, &excess);
664 try std.testing.expect(relay.peers[0] != null);
665 try std.testing.expect(relay.peers[0].?.channels[0] == null);
666 try std.testing.expect(relay.peers[0].?.channels[1] != null);
667 const reset = try testReadFrame(pair[1], .forward_reset);
668 defer reset.deinit(std.testing.allocator);
669 try std.testing.expectEqual(@as(u32, 1), try proto.decodeForwardId(reset.payload));
670
671 const overflow = proto.encodeForwardCredit(.{ .id = 2, .amount = proto.forward_initial_credit + 1 });
672 try testPush(&relay, .forward_credit, &overflow);
673 try std.testing.expect(relay.peers[0] == null);
674 }
675
676 fn testNonblocking(fd: std.posix.fd_t) !void {
677 const flags = try std.posix.fcntl(fd, std.posix.F.GETFL, 0);
678 const bits: u32 = @bitCast(std.posix.O{ .NONBLOCK = true });
679 _ = try std.posix.fcntl(fd, std.posix.F.SETFL, flags | bits);
680 }
681
682 fn testFillSocket(fd: std.posix.fd_t) !usize {
683 var buf: [4096]u8 = @splat(0xa5);
684 var total: usize = 0;
685 while (true) total += std.posix.write(fd, &buf) catch |err| switch (err) {
686 error.WouldBlock => return total,
687 else => return err,
688 };
689 }
690
691 fn testDrainPrefix(relay: *Relay, fd: std.posix.fd_t, count: usize) !void {
692 var left = count;
693 var buf: [64 * 1024]u8 = undefined;
694 while (left != 0) {
695 relay.flushPeer(0);
696 const n = try std.posix.read(fd, buf[0..@min(buf.len, left)]);
697 if (n == 0) return error.UnexpectedEof;
698 left -= n;
699 }
700 }
701
702 test "concurrent remote channels pause at shared peer capacity and resume intact" {
703 const alloc = std.testing.allocator;
704 const transport_pair = try testSocketPair();
705 defer std.posix.close(transport_pair[1]);
706 try testNonblocking(transport_pair[0]);
707 var relay = Relay.init(alloc);
708 defer relay.deinit();
709 try std.testing.expectEqual(Relay.Adopt.adopted, relay.adoptSocket(transport_pair[0], .empty));
710 const ready = try testReadFrame(transport_pair[1], .forward_ready);
711 ready.deinit(alloc);
712 const kernel_fill = try testFillSocket(transport_pair[0]);
713
714 const frame_len = proto.frame_header_len + proto.forward_id_len + proto.forward_data_max;
715 const prefix_len = peer_queue_max - control_reserve - frame_len;
716 try relay.peers[0].?.pending.resize(alloc, prefix_len);
717 @memset(relay.peers[0].?.pending.items, 0x5a);
718
719 const channel_count = 6;
720 var channel_peers: [channel_count]std.posix.fd_t = undefined;
721 var made: usize = 0;
722 defer for (channel_peers[0..made]) |fd| std.posix.close(fd);
723 var payloads: [channel_count][proto.forward_data_max]u8 = undefined;
724 for (0..channel_count) |i| {
725 const pair = try testSocketPair();
726 channel_peers[i] = pair[1];
727 made += 1;
728 @memset(&payloads[i], @intCast(i + 1));
729 try proto.writeAllFd(pair[1], &payloads[i]);
730 relay.peers[0].?.channels[i] = .{
731 .generation = relay.takeGeneration(),
732 .id = @intCast(i + 1),
733 .fd = pair[0],
734 .connecting = false,
735 .send_credit = proto.forward_initial_credit,
736 };
737 }
738
739 var fds: [poll_len]std.posix.pollfd = undefined;
740 relay.fillPoll(&fds);
741 _ = try std.posix.poll(&fds, 1000);
742 relay.servicePoll(&fds);
743 try std.testing.expect(relay.peers[0] != null);
744 try std.testing.expectEqual(prefix_len + frame_len, relay.peers[0].?.pending.items.len);
745 try std.testing.expect(relay.peers[0].?.pending.items.len <= peer_queue_max - control_reserve);
746 for (0..channel_count) |i| try std.testing.expect(relay.peers[0].?.channels[i] != null);
747
748 try testDrainPrefix(&relay, transport_pair[1], kernel_fill + prefix_len);
749 relay.flushPeer(0);
750 var first = try testReadFrame(transport_pair[1], .forward_data);
751 defer first.deinit(alloc);
752 try std.testing.expectEqual(@as(u32, 1), try proto.decodeForwardId(first.payload));
753 try std.testing.expectEqualSlices(u8, &payloads[0], first.payload[proto.forward_id_len..]);
754
755 for (1..channel_count) |i| {
756 relay.readTcp(0, i);
757 try std.testing.expect(relay.peers[0] != null);
758 try std.testing.expect(relay.peers[0].?.pending.items.len + relay.peers[0].?.sink.inFlight() <= peer_queue_max - control_reserve);
759 relay.flushPeer(0);
760 var frame = try testReadFrame(transport_pair[1], .forward_data);
761 defer frame.deinit(alloc);
762 try std.testing.expectEqual(@as(u32, @intCast(i + 1)), try proto.decodeForwardId(frame.payload));
763 try std.testing.expectEqualSlices(u8, &payloads[i], frame.payload[proto.forward_id_len..]);
764 }
765 }
src/server/server_test_attach.zig
Old New
@@ -19,6 +19,27 @@ const Lead = h.Lead;
19 const pumpUntil = h.pumpUntil; 19 const pumpUntil = h.pumpUntil;
20 const serverThread = h.serverThread; 20 const serverThread = h.serverThread;
21 21
22 test "forward-role hello leaves terminal admission and the session untouched" {
23 const alloc = std.testing.allocator;
24 var td = try h.TestDaemon.init(alloc, "forward-role", .{ .shell = "/bin/sh" });
25 defer td.deinit();
26 const session = td.srv.sessions.table[0].?.eng;
27 const epoch = td.srv.sessions.table[0].?.epoch;
28
29 const peer = try dial.dial(td.sock_path);
30 defer peer.close();
31 try proto.writeFrame(peer.handle, .forward_hello, &proto.encodeForwardHello());
32 const ready = (try awaitFrame(alloc, &td.srv, peer.handle, .forward_ready, 400)) orelse return error.NoForwardReady;
33 defer ready.deinit(alloc);
34 try std.testing.expectEqual(proto.forward_version, try proto.decodeForwardHello(ready.payload));
35
36 try std.testing.expectEqual(@as(?usize, 1), td.srv.forwards.freePeer());
37 try std.testing.expectEqual(@as(usize, 0), srv_mod.countLive(td.srv.clients));
38 try std.testing.expectEqual(@as(usize, 1), srv_mod.countLive(td.srv.sessions.table));
39 try std.testing.expect(td.srv.sessions.table[0].?.eng == session);
40 try std.testing.expectEqual(epoch, td.srv.sessions.table[0].?.epoch);
41 }
42
22 /// A sink that replays every frame it is handed into a replica and lets the 43 /// A sink that replays every frame it is handed into a replica and lets the
23 /// wait run on. For the waits whose answer is one particular reply while the 44 /// wait run on. For the waits whose answer is one particular reply while the
24 /// session keeps broadcasting underneath it. 45 /// session keeps broadcasting underneath it.
src/server/server_test_quic.zig
Old New
@@ -9,6 +9,7 @@ const srv_mod = @import("server.zig");
9 const Server = srv_mod.Server; 9 const Server = srv_mod.Server;
10 const drainWaitMs = srv_mod.drainWaitMs; 10 const drainWaitMs = srv_mod.drainWaitMs;
11 const stallExhausted = srv_mod.stallExhausted; 11 const stallExhausted = srv_mod.stallExhausted;
12 const shutdown_flag = &srv_mod.shutdown_flag;
12 const findFrame = h.findFrame; 13 const findFrame = h.findFrame;
13 const quicPump = h.quicPump; 14 const quicPump = h.quicPump;
14 const quicTestServer = h.quicTestServer; 15 const quicTestServer = h.quicTestServer;
@@ -19,7 +20,7 @@ fn attachOver(cl: *quic_server.TestPeer, buf: *std.ArrayList(u8), alloc: std.mem
19 cl.drain(); 20 cl.drain();
20 } 21 }
21 22
22 test "the QUIC listener holds more connections than the daemon has client slots" { 23 test "the QUIC listener holds the admitted peers and unclassified role table" {
23 // A QUIC connection exists from the moment its handshake completes and 24 // A QUIC connection exists from the moment its handshake completes and
24 // only THEN asks for a client slot, so the listener needs room the client 25 // only THEN asks for a client slot, so the listener needs room the client
25 // table does not: peers mid-handshake, and peers whose slot request is 26 // table does not: peers mid-handshake, and peers whose slot request is
@@ -34,7 +35,118 @@ test "the QUIC listener holds more connections than the daemon has client slots"
34 // Pinned rather than derived: `quic_server.zig` is a transport and does 35 // Pinned rather than derived: `quic_server.zig` is a transport and does
35 // not read the daemon's tables, so its number is restated there. This is 36 // not read the daemon's tables, so its number is restated there. This is
36 // what catches the restatement going stale. 37 // what catches the restatement going stale.
37 try std.testing.expect(quic_server.max_conns > srv_mod.max_clients); 38 try std.testing.expect(quic_server.max_conns > srv_mod.max_clients + srv_mod.max_forward_peers);
39 try std.testing.expectEqual(quic_server.max_conns, srv_mod.max_quic_pending);
40 }
41
42 test "Server: QUIC forward-role hello uses no terminal client slot" {
43 const alloc = std.testing.allocator;
44 var td = try h.TestDaemon.init(alloc, "qforward", .{ .shell = "/bin/sh" });
45 const key: quic.Key = .{ .bytes = [_]u8{0x46} ** quic.key_len };
46 const q = try quicTestServer(&td.srv, key);
47 defer q.l.deinit();
48 defer td.deinit();
49
50 var cl = try quic_server.TestPeer.init(q.addr, key);
51 defer cl.deinit();
52 var out: std.ArrayList(u8) = .empty;
53 defer out.deinit(alloc);
54 try proto.appendFrame(&out, alloc, .forward_hello, &proto.encodeForwardHello());
55 cl.out = out.items;
56 cl.drain();
57
58 var only = [_]*quic_server.TestPeer{&cl};
59 try quicPump(&td.srv, &only, 8000, &cl, struct {
60 fn f(t: *quic_server.TestPeer) bool {
61 return findFrame(t.cl.in.items, .forward_ready) != null;
62 }
63 }.f);
64 const ready = findFrame(cl.cl.in.items, .forward_ready) orelse return error.NoForwardReady;
65 try std.testing.expectEqual(proto.forward_version, try proto.decodeForwardHello(ready));
66 try std.testing.expectEqual(@as(?usize, 1), td.srv.forwards.freePeer());
67 try std.testing.expectEqual(@as(usize, 0), srv_mod.countLive(td.srv.clients));
68 try std.testing.expectEqual(@as(usize, 1), srv_mod.countLive(td.srv.sessions.table));
69 }
70
71 test "Server: QUIC one-shot stats and status reply without consuming terminal slots" {
72 const alloc = std.testing.allocator;
73 var td = try h.TestDaemon.init(alloc, "qoneshot", .{ .shell = "/bin/cat" });
74 const key: quic.Key = .{ .bytes = [_]u8{0x52} ** quic.key_len };
75 const q = try quicTestServer(&td.srv, key);
76 defer q.l.deinit();
77 defer td.deinit();
78
79 // These are real QUIC peers, deliberately not socket observers. Each
80 // request is its own connection because one-shot requests close only
81 // after their full reply has drained and must never be promoted to a
82 // terminal slot.
83 var stats = try quic_server.TestPeer.init(q.addr, key);
84 defer stats.deinit();
85 var status = try quic_server.TestPeer.init(q.addr, key);
86 defer status.deinit();
87 var stats_out: std.ArrayList(u8) = .empty;
88 defer stats_out.deinit(alloc);
89 var status_out: std.ArrayList(u8) = .empty;
90 defer status_out.deinit(alloc);
91 try proto.appendFrame(&stats_out, alloc, .stats_req, "");
92 try proto.appendFrame(&status_out, alloc, .status_req, "");
93 stats.out = stats_out.items;
94 status.out = status_out.items;
95 stats.drain();
96 status.drain();
97
98 var both = [_]*quic_server.TestPeer{ &stats, &status };
99 try quicPump(&td.srv, &both, 8000, &status, struct {
100 fn f(t: *quic_server.TestPeer) bool {
101 return findFrame(t.cl.in.items, .status_reply) != null;
102 }
103 }.f);
104 const stats_reply = findFrame(stats.cl.in.items, .stats_reply) orelse return error.NoQuicStatsReply;
105 try std.testing.expect(std.mem.indexOf(u8, stats_reply, "clients=0") != null);
106 const status_reply = findFrame(status.cl.in.items, .status_reply) orelse return error.NoQuicStatusReply;
107 const decoded = try proto.decodeStatusReply(status_reply);
108 try std.testing.expectEqual(@as(u16, 80), decoded.cols);
109 try std.testing.expectEqual(@as(u16, 24), decoded.rows);
110 try std.testing.expectEqual(@as(usize, 0), srv_mod.countLive(td.srv.clients));
111
112 // A completed one-shot has no retained provisional entry either. Pumping
113 // until both replies arrived rules out observing a merely queued response.
114 try quicPump(&td.srv, &both, 8000, &td.srv, struct {
115 fn f(s: *Server) bool {
116 for (s.quic_pending) |slot| if (slot != null) return false;
117 return true;
118 }
119 }.f);
120 }
121
122 test "Server: a malformed first QUIC frame releases its provisional slot" {
123 const alloc = std.testing.allocator;
124 var td = try h.TestDaemon.init(alloc, "qbadfirst", .{ .shell = "/bin/cat" });
125 const key: quic.Key = .{ .bytes = [_]u8{0x53} ** quic.key_len };
126 const q = try quicTestServer(&td.srv, key);
127 defer q.l.deinit();
128 defer td.deinit();
129
130 var cl = try quic_server.TestPeer.init(q.addr, key);
131 defer cl.deinit();
132 var out: std.ArrayList(u8) = .empty;
133 defer out.deinit(alloc);
134 // input is neither an attach nor an observer verb. It is a complete,
135 // legal frame in the wrong first-frame position, so it exercises role
136 // classification rather than a parser rejection.
137 try proto.appendFrame(&out, alloc, .input, "x");
138 cl.out = out.items;
139 cl.drain();
140
141 var only = [_]*quic_server.TestPeer{&cl};
142 try quicPump(&td.srv, &only, 8000, &td.srv, struct {
143 fn f(s: *Server) bool {
144 for (s.quic_pending) |slot| if (slot != null) return false;
145 return true;
146 }
147 }.f);
148 try std.testing.expectEqual(@as(usize, 0), srv_mod.countLive(td.srv.clients));
149 try std.testing.expectEqual(@as(?usize, 0), td.srv.forwards.freePeer());
38 } 150 }
39 151
40 test "refusalFrame: the exact refusal wire bytes" { 152 test "refusalFrame: the exact refusal wire bytes" {
@@ -326,6 +438,84 @@ test "Server: one QUIC client leaving does not disturb the other" {
326 try std.testing.expect(b.echoed() > before); 438 try std.testing.expect(b.echoed() > before);
327 } 439 }
328 440
441 test "Server: a QUIC stop requester receives CONNECTION_CLOSE during deinit" {
442 const alloc = std.testing.allocator;
443 var td = try h.TestDaemon.init(alloc, "qstopclose", .{ .shell = "/bin/cat" });
444 const key: quic.Key = .{ .bytes = [_]u8{0x5C} ** quic.key_len };
445 const q = try quicTestServer(&td.srv, key);
446 // The borrowed listener remains live through Server.deinit.
447 defer q.l.deinit();
448 var cl = try quic_server.TestPeer.init(q.addr, key);
449 defer cl.deinit();
450 shutdown_flag.store(false, .release);
451 defer shutdown_flag.store(false, .release);
452
453 var out: std.ArrayList(u8) = .empty;
454 defer out.deinit(alloc);
455 try proto.appendFrame(&out, alloc, .stop_req, "");
456 cl.out = out.items;
457 var only = [_]*quic_server.TestPeer{&cl};
458 try quicPump(&td.srv, &only, 8000, shutdown_flag, struct {
459 fn f(flag: *std.atomic.Value(bool)) bool {
460 return flag.load(.acquire);
461 }
462 }.f);
463 try std.testing.expect(shutdown_flag.load(.acquire));
464
465 td.deinit();
466 // No server pump remains: receiving this packet is solely closeAll's
467 // synchronous CONNECTION_CLOSE, not the listener's idle timeout.
468 var i: usize = 0;
469 while (i < 100 and !cl.cl.dead) : (i += 1) {
470 var pfd = [_]std.posix.pollfd{.{ .fd = cl.cl.fd, .events = std.posix.POLL.IN, .revents = 0 }};
471 _ = std.posix.poll(&pfd, 2) catch 0;
472 cl.drain();
473 }
474 try std.testing.expect(cl.cl.dead);
475 }
476
477 test "Server: a large QUIC dump drains over ACK refill and one-shot bounds expire" {
478 const alloc = std.testing.allocator;
479 // This viewport yields a real dump larger than the 256 KiB transport ring.
480 var td = try h.TestDaemon.init(alloc, "qlargedump", .{ .shell = "/bin/cat", .cols = 1024, .rows = 512 });
481 defer td.deinit();
482 const key: quic.Key = .{ .bytes = [_]u8{0x5B} ** quic.key_len };
483 const q = try quicTestServer(&td.srv, key);
484 defer q.l.deinit();
485 var cl = try quic_server.TestPeer.init(q.addr, key);
486 defer cl.deinit();
487 td.srv.ses(0).eng.feed("x" ** (1024 * 512));
488 const expected = try td.srv.ses(0).eng.dumpPlain(alloc);
489 defer alloc.free(expected);
490 try std.testing.expect(expected.len > quic.egress_cap);
491
492 var out: std.ArrayList(u8) = .empty;
493 defer out.deinit(alloc);
494 try proto.appendFrame(&out, alloc, .debug_dump, &.{0});
495 cl.out = out.items;
496 var only = [_]*quic_server.TestPeer{&cl};
497 try quicPump(&td.srv, &only, 20_000, &cl, struct {
498 fn f(peer: *quic_server.TestPeer) bool {
499 return findFrame(peer.cl.in.items, .dump_reply) != null;
500 }
501 }.f);
502 const dump = findFrame(cl.cl.in.items, .dump_reply) orelse return error.NoDumpReply;
503 try std.testing.expectEqualSlices(u8, expected, dump);
504 try quicPump(&td.srv, &only, 8000, &td.srv, struct {
505 fn f(s: *Server) bool {
506 return srv_mod.countLive(s.quic_pending) == 0;
507 }
508 }.f);
509
510 // This is checked before append/allocation; an impossible reply cannot
511 // expand an ArrayList past the stated per-peer response budget.
512 try std.testing.expect(Server.quicOneShotCanQueue(0, proto.max_payload));
513 try std.testing.expect(!Server.quicOneShotCanQueue(0, proto.max_payload + 1));
514 try std.testing.expect(!Server.quicOneShotCanQueue(srv_mod.quic_one_shot_pending_cap, 0));
515 try std.testing.expect(!Server.quicOneShotExpired(100, 100 + srv_mod.quic_one_shot_deadline_ms - 1));
516 try std.testing.expect(Server.quicOneShotExpired(100, 100 + srv_mod.quic_one_shot_deadline_ms));
517 }
518
329 test "Server: a QUIC client that stops reading is dropped by the cap, not tolerated" { 519 test "Server: a QUIC client that stops reading is dropped by the cap, not tolerated" {
330 const alloc = std.testing.allocator; 520 const alloc = std.testing.allocator;
331 var tmp = try TmpDir.make(); 521 var tmp = try TmpDir.make();
src/server/server_upgrade.zig
Old New
@@ -94,6 +94,7 @@ pub fn initFromManifest(
94 .shellint_arena = shellint_arena, 94 .shellint_arena = shellint_arena,
95 .shellint_dir = injection.dir, 95 .shellint_dir = injection.dir,
96 .agents = .{ .dir = agent_dir }, 96 .agents = .{ .dir = agent_dir },
97 .forwards = .{ .alloc = alloc },
97 }; 98 };
98 srv_mod.logSocket(sock_path, "adopted across upgrade dev={d} ino={d}", .{ srv.bound.path_id.dev, srv.bound.path_id.ino }); 99 srv_mod.logSocket(sock_path, "adopted across upgrade dev={d} ino={d}", .{ srv.bound.path_id.dev, srv.bound.path_id.ino });
99 100
src/tui/interact.zig
Old New
@@ -1663,7 +1663,7 @@ pub const Core = struct {
1663 .agent_open, .agent_data, .agent_close => return .not_mine, 1663 .agent_open, .agent_data, .agent_close => return .not_mine,
1664 // Named rather than swept into the `else`, so giving one a meaning 1664 // Named rather than swept into the `else`, so giving one a meaning
1665 // is an edit here and not a new switch elsewhere. Each belongs to 1665 // is an edit here and not a new switch elsewhere. Each belongs to
1666 // `mux a`'s conversation or travels the other way. 1666 // another client conversation or travels the other way.
1667 .stats_reply, 1667 .stats_reply,
1668 .endpoint_reply, 1668 .endpoint_reply,
1669 .cmd_state, 1669 .cmd_state,
@@ -1683,6 +1683,14 @@ pub const Core = struct {
1683 .selection_req, 1683 .selection_req,
1684 .sessions_req, 1684 .sessions_req,
1685 .agent_offer, 1685 .agent_offer,
1686 .forward_hello,
1687 .forward_open,
1688 .forward_data,
1689 .forward_credit,
1690 .forward_half_close,
1691 .forward_reset,
1692 .forward_ready,
1693 .forward_open_result,
1686 .debug_dump, 1694 .debug_dump,
1687 .upgrade_req, 1695 .upgrade_req,
1688 .upgrade_reply, 1696 .upgrade_reply,
@@ -4774,6 +4782,20 @@ test "interact: only what the driver has an answer of its own for comes back" {
4774 // receiving one means nothing here. 4782 // receiving one means nothing here.
4775 try std.testing.expectEqual(Routed.skip, try core.frame(.agent_offer, "")); 4783 try std.testing.expectEqual(Routed.skip, try core.frame(.agent_offer, ""));
4776 4784
4785 const forwarding = [_]proto.MsgType{
4786 .forward_hello,
4787 .forward_open,
4788 .forward_data,
4789 .forward_credit,
4790 .forward_half_close,
4791 .forward_reset,
4792 .forward_ready,
4793 .forward_open_result,
4794 };
4795 for (forwarding) |t| {
4796 try std.testing.expectEqual(Routed.skip, try core.frame(t, ""));
4797 }
4798
4777 // ...and nothing else is, over every type the enum names. 4799 // ...and nothing else is, over every type the enum names.
4778 inline for (@typeInfo(proto.MsgType).@"enum".fields) |f| { 4800 inline for (@typeInfo(proto.MsgType).@"enum".fields) |f| {
4779 const t: proto.MsgType = @enumFromInt(f.value); 4801 const t: proto.MsgType = @enumFromInt(f.value);
test/native_forward.py
Old New
@@ -0,0 +1,478 @@
1 #!/usr/bin/env python3
2 """Real-binary loopback forwarding acceptance for socket, stdio, and QUIC.
3
4 This is intentionally directly runnable rather than wired into native-e2e: it
5 needs a release mux/muxg pair and opens deliberate TCP listeners. Every daemon,
6 TCP service, GUI, socket, and XDG directory is owned by Rig and is retained on
7 failure. The focused Zig component tests remain the precise cap/admission
8 oracles; this script proves their real-binary plumbing and byte-stream effects.
9 """
10 import hashlib
11 import http.server
12 import os
13 from pathlib import Path
14 import socket
15 import socketserver
16 import subprocess
17 import sys
18 import threading
19 import time
20
21 sys.dont_write_bytecode = True
22 from native_lifecycle import LifecycleRig
23 from native_tiling import eventually, require
24
25
26 TIMEOUT = 8
27 MAX_HTTP_RESPONSE = 1024 * 1024
28
29
30 def free_port():
31 # The listener is released only immediately before its owning fixture binds
32 # it. SO_REUSEADDR makes the reservation probe below reject a live owner
33 # without confusing its own prior TCP connections in TIME_WAIT for one.
34 with socket.socket() as sock:
35 sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
36 sock.bind(('127.0.0.1', 0))
37 return sock.getsockname()[1]
38
39
40 def distinct_ports(count):
41 ports = set()
42 while len(ports) < count:
43 ports.add(free_port())
44 return tuple(ports)
45
46
47 def recv_exact(sock, length):
48 data = bytearray()
49 while len(data) < length:
50 part = sock.recv(length - len(data))
51 require(part, f'unexpected EOF after {len(data)}/{length} bytes')
52 data.extend(part)
53 return bytes(data)
54
55
56 class TCPServices:
57 """Daemon-host loopback endpoints with distinguishable HTTP content."""
58 def __init__(self, root):
59 self.token = hashlib.sha256(str(root).encode()).hexdigest()[:20]
60 self.http_port, self.echo_port, self.slow_port = distinct_ports(3)
61 self.expect_restart_disconnect = threading.Event()
62 token = self.token
63 expect_restart_disconnect = self.expect_restart_disconnect
64
65 class HTTP(http.server.BaseHTTPRequestHandler):
66 def do_GET(self):
67 body = ('native-forward-http-' + token + '\n').encode()
68 self.send_response(200)
69 self.send_header('Content-Length', str(len(body)))
70 self.send_header('Connection', 'close')
71 self.end_headers()
72 self.wfile.write(body)
73 def log_message(self, *_):
74 pass
75
76 class Echo(socketserver.BaseRequestHandler):
77 def handle(inner):
78 while True:
79 data = inner.request.recv(65536)
80 if not data:
81 try:
82 inner.request.shutdown(socket.SHUT_WR)
83 except OSError:
84 # The restart case deliberately tears this peer down.
85 # Other shutdown errors identify a fixture regression.
86 if not expect_restart_disconnect.is_set():
87 raise
88 return
89 inner.request.sendall(data)
90
91 class Slow(socketserver.BaseRequestHandler):
92 def handle(inner):
93 # Fill forwarding queues while the local client deliberately
94 # does not read. It is bounded so fixture cleanup never waits.
95 payload = b'SLOW-' + token.encode() + b'X' * 65536
96 for _ in range(48):
97 inner.request.sendall(payload)
98 inner.request.shutdown(socket.SHUT_WR)
99
100 self.servers = []
101 for port, handler in ((self.http_port, HTTP), (self.echo_port, Echo), (self.slow_port, Slow)):
102 server = socketserver.ThreadingTCPServer(('127.0.0.1', port), handler)
103 server.daemon_threads = True
104 thread = threading.Thread(target=server.serve_forever, daemon=True)
105 thread.start()
106 self.servers.append(server)
107
108 def close(self):
109 for server in self.servers:
110 server.shutdown()
111 server.server_close()
112
113
114 class ForwardRig(LifecycleRig):
115 def launch_forward(self, target_args, rules, label):
116 args = [*target_args]
117 for local, remote in rules:
118 args += ['--forward', f'{local}:{remote}']
119 # A named ephemeral session makes this invocation temporary and avoids
120 # changing the native workspace under the fixture's isolated state.
121 args += ['--session', 'forward']
122 self.launch_gui(args, label)
123
124 def restart_daemon(self, sock, old, label):
125 self.stop_daemon(sock, old)
126 self.daemons.remove((sock, old))
127 # Recovery must use the manager's original concrete socket identity.
128 proc = self.spawn([self.mux, 'd', 'start', '--sock', sock], label)
129 self.daemons.append((sock, proc))
130 eventually(lambda: Path(sock).exists(), label + ' daemon did not restart')
131 return proc
132
133
134 class PreResponseTransportStartupError(ConnectionError):
135 """A reset or EOF before any response byte; listener startup may retry it."""
136
137
138 def _remaining(deadline):
139 remaining = deadline - time.monotonic()
140 require(remaining > 0, 'HTTP response exceeded deadline')
141 return remaining
142
143
144 def recv_http_part(sock, deadline, saw_response):
145 sock.settimeout(_remaining(deadline))
146 try:
147 part = sock.recv(65536)
148 except socket.timeout as error:
149 raise AssertionError('HTTP response exceeded deadline') from error
150 except OSError as error:
151 if not saw_response:
152 raise PreResponseTransportStartupError('HTTP transport reset before response') from error
153 raise AssertionError('HTTP transport failed after response started') from error
154 if not part and not saw_response:
155 raise PreResponseTransportStartupError('HTTP transport EOF before response')
156 return part
157
158
159 def read_http_response(sock, deadline):
160 response = bytearray()
161 while b'\r\n\r\n' not in response:
162 part = recv_http_part(sock, deadline, bool(response))
163 require(part, 'HTTP response ended before headers completed')
164 response.extend(part)
165 require(len(response) <= MAX_HTTP_RESPONSE, 'HTTP response headers exceed bound')
166 raw_headers, body = bytes(response).split(b'\r\n\r\n', 1)
167 lines = raw_headers.split(b'\r\n')
168 status = lines[0].split()
169 require(len(status) == 3 and status[0].startswith(b'HTTP/'), 'malformed HTTP status line')
170 try:
171 code = int(status[1])
172 except ValueError as error:
173 raise AssertionError('malformed HTTP status code') from error
174 lengths = []
175 for line in lines[1:]:
176 require(b':' in line, 'malformed HTTP header')
177 name, value = line.split(b':', 1)
178 if name.lower() == b'transfer-encoding':
179 raise AssertionError('HTTP Transfer-Encoding is unsupported')
180 if name.lower() == b'content-length':
181 lengths.append(value.strip())
182 require(len(lengths) == 1, 'HTTP response requires exactly one Content-Length')
183 require(lengths[0].isdigit(), 'malformed HTTP Content-Length')
184 length = int(lengths[0])
185 require(length <= MAX_HTTP_RESPONSE, 'HTTP Content-Length exceeds bound')
186 while True:
187 require(len(body) <= length, 'HTTP response has surplus body bytes')
188 part = recv_http_part(sock, deadline, True)
189 if not part:
190 break
191 body += part
192 require(len(body) <= MAX_HTTP_RESPONSE, 'HTTP response body exceeds bound')
193 require(len(body) == length, f'HTTP response ended after {len(body)}/{length} body bytes')
194 return code, body
195
196
197 def http_get(port, deadline=None):
198 deadline = time.monotonic() + TIMEOUT if deadline is None else deadline
199 try:
200 sock = socket.create_connection(('127.0.0.1', port), timeout=_remaining(deadline))
201 except OSError as error:
202 raise PreResponseTransportStartupError('HTTP connection failed') from error
203 with sock:
204 try:
205 sock.settimeout(_remaining(deadline))
206 sock.sendall(b'GET / HTTP/1.0\r\nHost: fixture\r\n\r\n')
207 sock.shutdown(socket.SHUT_WR)
208 except OSError as error:
209 raise PreResponseTransportStartupError('HTTP request failed before response') from error
210 return read_http_response(sock, deadline)
211
212
213 def _socketpair_response(parts):
214 reader, writer = socket.socketpair()
215 errors = []
216 def write_response():
217 try:
218 for part in parts:
219 writer.sendall(part)
220 time.sleep(.01)
221 writer.shutdown(socket.SHUT_WR)
222 except OSError as error:
223 errors.append(error)
224 finally:
225 writer.close()
226 thread = threading.Thread(target=write_response)
227 thread.start()
228 result = error = None
229 try:
230 result = read_http_response(reader, time.monotonic() + TIMEOUT)
231 except BaseException as caught:
232 error = caught
233 finally:
234 reader.close()
235 thread.join(TIMEOUT)
236 require(not thread.is_alive(), 'HTTP socketpair fixture thread hung')
237 if error is not None:
238 # Closing after a deliberate parser rejection can break the writer's
239 # shutdown; retain the parser error rather than masking it.
240 raise error
241 require(not errors, 'HTTP socketpair fixture failed: ' + repr(errors))
242 return result
243
244
245 def http_parser_regressions():
246 code, body = _socketpair_response((b'HTTP/1.0 200 OK\r\nContent-', b'Length: 5\r\nConnection: close\r\n\r\nhe', b'llo'))
247 require(code == 200 and body == b'hello', 'fragmented HTTP response was not parsed completely')
248 for response in (b'bad\r\n\r\n',
249 b'HTTP/1.0 200 OK\r\nContent-Length: 3\r\nContent-Length: 3\r\n\r\nabc',
250 b'HTTP/1.0 200 OK\r\nContent-Length: 3\r\nTransfer-Encoding: chunked\r\n\r\nabc',
251 b'HTTP/1.0 200 OK\r\nContent-Length: 5\r\n\r\nabc',
252 b'HTTP/1.0 200 OK\r\nContent-Length: 3\r\n\r\nmore'):
253 try:
254 _socketpair_response((response,))
255 except AssertionError:
256 pass
257 else:
258 raise AssertionError('malformed, duplicate, truncated, or surplus HTTP response was accepted')
259
260
261 def http_retry_classifier_regression():
262 original = http_get
263 try:
264 def parser_failure(_, deadline=None):
265 raise AssertionError('malformed HTTP status line')
266 globals()['http_get'] = parser_failure
267 try:
268 wait_forward(1)
269 except AssertionError as error:
270 require('malformed HTTP status line' in str(error), 'parser failure was changed')
271 else:
272 raise AssertionError('wait_forward swallowed a parser failure')
273 finally:
274 globals()['http_get'] = original
275
276
277 def echo_roundtrip(port, payload, half_close=False):
278 with socket.create_connection(('127.0.0.1', port), timeout=TIMEOUT) as sock:
279 sock.sendall(payload)
280 if half_close:
281 sock.shutdown(socket.SHUT_WR)
282 got = recv_exact(sock, len(payload))
283 if half_close:
284 require(sock.recv(1) == b'', 'remote half-close did not reach local TCP peer')
285 return got
286
287
288 def listener_reserved(port):
289 # SO_REUSEADDR avoids treating a just-closed fixture connection in TIME_WAIT
290 # as listener ownership. Each call is paired with a real forwarded request,
291 # so the failed bind has both kernel and forwarding-owner evidence.
292 with socket.socket() as probe:
293 probe.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
294 try:
295 probe.bind(('127.0.0.1', port))
296 except OSError:
297 return True
298 return False
299
300
301 def wait_forward(port):
302 deadline = time.monotonic() + TIMEOUT
303 while time.monotonic() < deadline:
304 try:
305 return http_get(port, deadline)
306 except PreResponseTransportStartupError:
307 time.sleep(min(.04, max(0, deadline - time.monotonic())))
308 raise AssertionError('forward listener did not become usable')
309
310
311 def route_args(rig, kind, sock):
312 if kind == 'socket':
313 return ['--sock', sock]
314 if kind == 'stdio':
315 return ['--via', rig.mux + ' d proxy --sock ' + sock]
316 if kind == 'quic':
317 target = rig.targets[sock]
318 key = str(Path(rig.env['XDG_CONFIG_HOME']) / 'mux/key')
319 return [target, '--key', key]
320 raise AssertionError('unknown route ' + kind)
321
322
323 def basic_routes(rig, services):
324 """Each entry transport carries plural, concurrent real TCP channels."""
325 for kind in ('socket', 'stdio', 'quic'):
326 sock, _ = rig.daemon('forward-' + kind, quic=kind == 'quic')
327 local_http, local_echo = distinct_ports(2)
328 rig.launch_forward(route_args(rig, kind, sock),
329 [(local_http, services.http_port), (local_echo, services.echo_port)],
330 'forward-' + kind + '-gui')
331 code, body = wait_forward(local_http)
332 require(code == 200, kind + ' forwarding returned HTTP status ' + str(code))
333 require(('native-forward-http-' + services.token).encode() in body,
334 kind + ' forwarding reached wrong HTTP endpoint')
335 payloads = [os.urandom(256 * 1024 + n) for n in range(3)]
336 results, errors = [None] * len(payloads), []
337 error_lock = threading.Lock()
338 def roundtrip_worker(i):
339 try:
340 results[i] = echo_roundtrip(local_echo, payloads[i])
341 except BaseException as error:
342 with error_lock:
343 errors.append(error)
344 workers = [threading.Thread(target=roundtrip_worker, args=(i,)) for i in range(len(payloads))]
345 for worker in workers:
346 worker.start()
347 for worker in workers:
348 worker.join(TIMEOUT)
349 require(not worker.is_alive(), kind + ' concurrent forwarding stream hung')
350 require(not errors, kind + ' concurrent forwarding worker failed: ' + repr(errors))
351 for payload, result in zip(payloads, results):
352 require(result == payload, kind + ' concurrent stream corrupted bytes')
353 require(echo_roundtrip(local_echo, os.urandom(1024 * 1024 + 17), half_close=True) != b'',
354 kind + ' large half-closed stream failed')
355 rig.ok(kind + ' forwarding carries unique HTTP and concurrent large half-closed echo streams')
356 rig.quit()
357
358
359 def local_lifetime_and_recovery(rig, services):
360 sock, daemon = rig.daemon('forward-lifetime')
361 http_local, echo_local, slow_local, refused_local, unavailable_remote = distinct_ports(5)
362 rules = [(http_local, services.http_port), (echo_local, services.echo_port),
363 (slow_local, services.slow_port), (refused_local, unavailable_remote)]
364 rig.launch_forward(['--sock', sock], rules, 'forward-lifetime-gui')
365 code, body = wait_forward(http_local)
366 require(code == 200, 'forward listener returned HTTP status ' + str(code))
367 require(('native-forward-http-' + services.token).encode() in body,
368 'forward listener reached wrong HTTP endpoint')
369 require(listener_reserved(http_local), 'live forward listener was not reserved')
370
371 # The refused destination is channel-local: the adjacent echo rule remains
372 # usable after its open is rejected.
373 with socket.create_connection(('127.0.0.1', refused_local), timeout=TIMEOUT) as refused:
374 refused.settimeout(TIMEOUT)
375 require(refused.recv(1) == b'', 'refused remote destination left a live local stream')
376 require(echo_roundtrip(echo_local, b'channel-isolation') == b'channel-isolation',
377 'refused destination damaged a sibling forwarding rule')
378
379 # This only holds a non-reading peer; it does NOT establish that remote
380 # credit or a forwarding queue actually blocked. The terminal assertion is
381 # therefore an unverified responsiveness smoke check, not backpressure
382 # evidence. Precise queue/admission coverage remains unit-test-only.
383 slow = socket.create_connection(('127.0.0.1', slow_local), timeout=TIMEOUT)
384 try:
385 time.sleep(.15)
386 pane = rig.state()['focus']
387 rig.shell("printf 'FORWARD-TERMINAL-RESPONSIVE\\n'")
388 rig.wait_marker(sock, 'forward', 'FORWARD-TERMINAL-RESPONSIVE')
389 require(rig.gui.poll() is None, 'slow forwarding reader stopped the GUI')
390 finally:
391 slow.close()
392
393 # A matching quick split changes focus. Remove that new sibling and prove
394 # the original entry pane remains before checking manager lifetime.
395 original = rig.state()['focus']
396 rig.chord('b')
397 state = rig.wait_state(lambda s: len(s['panes']) == 2 and s['focus'] != original)
398 sibling = state['focus']
399 rig.chord('d')
400 state = rig.wait_state(lambda s: len(s['panes']) == 1 and
401 original in {pane['id'] for pane in s['panes']} and
402 sibling not in {pane['id'] for pane in s['panes']})
403 require(listener_reserved(http_local), 'removing one matching pane released forward listener')
404
405 # Losing only the dedicated role connection resets old streams, reserves
406 # listeners, and allows a daemon restart at the same socket to recover.
407 old = socket.create_connection(('127.0.0.1', echo_local), timeout=TIMEOUT)
408 old.sendall(b'interrupted')
409 services.expect_restart_disconnect.set()
410 try:
411 daemon = rig.restart_daemon(sock, daemon, 'forward-whole-daemon-restarted')
412 old.settimeout(TIMEOUT)
413 # A reply already queued before the tear is permitted; the old channel
414 # must nevertheless become EOF rather than silently survive/reconnect.
415 while old.recv(65536):
416 pass
417 finally:
418 old.close()
419 services.expect_restart_disconnect.clear()
420 require(listener_reserved(http_local), 'transport interruption released listener reservation')
421 code, body = wait_forward(http_local)
422 require(code == 200, 'recovered forward listener returned HTTP status ' + str(code))
423 require(('native-forward-http-' + services.token).encode() in body,
424 'recovered forward listener reached wrong HTTP endpoint')
425 require(echo_roundtrip(echo_local, b'recovered') == b'recovered', 'forward listener did not recover after daemon restart')
426
427 # A conflicting second invocation fails before it can disturb the active
428 # GUI/listener. No FIFO is inherited because it is not that process's UI.
429 env = rig.env.copy()
430 env.pop('MUXG_TEST_FIFO', None)
431 collision = subprocess.run([rig.muxg, '--sock', sock, '--session', 'collision',
432 '--forward', f'{http_local}:{services.http_port}'],
433 env=env, capture_output=True, text=True, timeout=TIMEOUT)
434 require(collision.returncode != 0, 'local forward collision was accepted')
435 diagnostic = (collision.stdout + collision.stderr).lower()
436 require(any(word in diagnostic for word in ('forward', 'bind', 'address', 'port', 'in use')),
437 'local forward collision lacked a visible bind diagnostic: ' + diagnostic)
438 require(echo_roundtrip(echo_local, b'after-collision') == b'after-collision',
439 'local collision disturbed existing forwarding')
440
441 # Final matching-pane removal is the lifetime boundary. The focused pane
442 # is the only remaining target after the prior detach.
443 rig.chord('d')
444 rig.wait_state(lambda s: not s['panes'])
445 eventually(lambda: not listener_reserved(http_local),
446 'final matching pane did not release forward listener')
447 rig.quit()
448 rig.ok('focus/final-target lifetime, refusal isolation, unverified slow-reader responsiveness smoke, whole-daemon recovery')
449
450
451 def main():
452 require(len(sys.argv) == 3, 'usage: native_forward.py RELEASE_MUX RELEASE_MUXG')
453 rig = ForwardRig(*sys.argv[1:])
454 services = None
455 try:
456 services = TCPServices(rig.root)
457 version = subprocess.check_output([rig.muxg, '--version'], text=True)
458 require('ReleaseSafe' in version or 'ReleaseFast' in version,
459 'native forwarding requires ReleaseSafe or ReleaseFast binaries')
460 http_parser_regressions()
461 http_retry_classifier_regression()
462 basic_routes(rig, services)
463 local_lifetime_and_recovery(rig, services)
464 print(f'PASS: native forwarding ({rig.checkpoints} checkpoints)', flush=True)
465 except BaseException:
466 rig.failure_artifacts()
467 raise
468 finally:
469 # Neither owner may suppress the other's cleanup.
470 try:
471 if services is not None:
472 services.close()
473 finally:
474 rig.close()
475
476
477 if __name__ == '__main__':
478 main()
test/native_forward_remote.py
Old New
@@ -0,0 +1,373 @@
1 #!/usr/bin/env python3
2 """Opt-in real SSH and direct-QUIC forwarding against one private remote fixture.
3
4 Safety latch: this script contacts no host unless MUX_FORWARD_REMOTE_ENABLE=1.
5 The fixture owns every remote path, daemon socket, QUIC key, UDP port, service,
6 and copied binary; it never touches the remote user's mux state.
7 """
8 import contextlib
9 import hashlib
10 import json
11 import os
12 from pathlib import Path
13 import secrets
14 import shlex
15 import shutil
16 import socket
17 import subprocess
18 import sys
19 import tempfile
20 import threading
21 import time
22
23 sys.dont_write_bytecode = True
24 from native_forward import (ForwardRig, PreResponseTransportStartupError, echo_roundtrip,
25 distinct_ports, http_get, listener_reserved)
26 from native_tiling import eventually, require
27
28
29 REMOTE_DEFAULT = 'ubuntu@192.168.0.107'
30 REMOTE_ADDRESS = '192.168.0.107'
31 REMOTE_PATH = '/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin'
32 REMOTE_PYTHON = '/usr/bin/python3'
33
34 REMOTE_SERVICE = r'''import hashlib,http.server,json,os,socket,socketserver,sys,tempfile,threading
35 root,out=sys.argv[1:]
36 token=hashlib.sha256(root.encode()).hexdigest()[:20]
37 def port():
38 s=socket.socket();s.setsockopt(socket.SOL_SOCKET,socket.SO_REUSEADDR,1);s.bind(("127.0.0.1",0));n=s.getsockname()[1];s.close();return n
39 class H(http.server.BaseHTTPRequestHandler):
40 def do_GET(self):
41 b=("native-forward-remote-"+token+"\n").encode();self.send_response(200);self.send_header("Content-Length",str(len(b)));self.send_header("Connection","close");self.end_headers();self.wfile.write(b)
42 def log_message(self,*a):pass
43 class E(socketserver.BaseRequestHandler):
44 def handle(self):
45 while True:
46 b=self.request.recv(65536)
47 if not b:self.request.shutdown(socket.SHUT_WR);return
48 self.request.sendall(b)
49 servers=[]
50 for p,h in ((port(),H),(port(),E)):
51 s=socketserver.ThreadingTCPServer(("127.0.0.1",p),h);s.daemon_threads=True;threading.Thread(target=s.serve_forever,daemon=True).start();servers.append(s)
52 fd,tmp=tempfile.mkstemp(prefix="ports-",dir=os.path.dirname(out));os.write(fd,json.dumps({"http":servers[0].server_address[1],"echo":servers[1].server_address[1],"token":token}).encode());os.fsync(fd);os.close(fd);os.replace(tmp,out)
53 threading.Event().wait()
54 '''
55
56
57 class RemoteFixture:
58 def __init__(self, local_mux):
59 self.local_mux = str(Path(local_mux).resolve())
60 self.host = os.environ.get('MUX_FORWARD_REMOTE', REMOTE_DEFAULT)
61 self.ssh, self.scp = shutil.which('ssh'), shutil.which('scp')
62 require(self.ssh and self.scp, 'remote forwarding requires ssh and scp')
63 self.local = Path(tempfile.mkdtemp(prefix='mux-forward-ssh-'))
64 self.remote = None
65 self.daemon_pid = None
66 self.quic_port = None
67 self.cleanup_errors = []
68 self.marker = secrets.token_hex(24)
69 self._rollback = contextlib.ExitStack()
70 self._rollback.callback(self.close)
71 try:
72 self._make_local_key()
73 self._create_remote_root()
74 self.remote_mux = self.remote + '/bin/mux'
75 self.remote_sock = self.remote + '/mux.sock'
76 self.remote_key = self.remote + '/quic.key'
77 self.remote_oracle = self.remote + '/os_oracle.sh'
78 self.env_words = {
79 'PATH': REMOTE_PATH, 'HOME': self.remote + '/home',
80 'XDG_CONFIG_HOME': self.remote + '/xdg/config',
81 'XDG_STATE_HOME': self.remote + '/xdg/state',
82 'XDG_CACHE_HOME': self.remote + '/xdg/cache',
83 'XDG_RUNTIME_DIR': self.remote + '/xdg/runtime', 'SHELL': '/bin/sh',
84 'MUX_KEY_FILE': self.remote_key, 'MUX_SOCK': self.remote_sock,
85 }
86 self.run('mkdir -p ' + ' '.join(shlex.quote(self.remote + '/' + p) for p in
87 ('bin', 'home', 'xdg/config', 'xdg/state', 'xdg/cache', 'xdg/runtime')))
88 for source, destination in ((self.local_mux, self.remote_mux),
89 (str(self.local_key), self.remote_key),
90 (str(Path('test/os_oracle.sh').resolve()), self.remote_oracle)):
91 subprocess.run([self.scp, '-q', source, self.host + ':' + destination], check=True, timeout=30)
92 self.run('chmod 700 ' + shlex.quote(self.remote_mux) + ' && chmod 600 ' +
93 shlex.quote(self.remote_key) + ' ' + shlex.quote(self.remote_oracle))
94 service = self.local / 'remote-services.py'
95 service.write_text(REMOTE_SERVICE)
96 subprocess.run([self.scp, '-q', str(service), self.host + ':' + self.remote + '/services.py'], check=True, timeout=30)
97 self.run_env(shlex.join([REMOTE_PYTHON, self.remote + '/services.py', self.remote, self.remote + '/ports.json']) +
98 ' > ' + shlex.quote(self.remote + '/services.log') + ' 2>&1 & printf %s "$!" > ' +
99 shlex.quote(self.remote + '/services.pid'))
100 self.ports = eventually(self._read_ports, 'remote loopback services did not publish ports JSON', seconds=15)
101 self.quic_port = self._high_udp_port()
102 self.start_daemon()
103 self._rollback.pop_all()
104 except BaseException:
105 self._rollback.close()
106 raise
107
108 def _make_local_key(self):
109 key_env = os.environ.copy()
110 for key in ('XDG_CONFIG_HOME', 'XDG_STATE_HOME', 'XDG_CACHE_HOME', 'XDG_RUNTIME_DIR', 'HOME'):
111 path = self.local / key
112 path.mkdir(mode=0o700, exist_ok=True)
113 key_env[key] = str(path)
114 subprocess.run([self.local_mux, 'd', 'keygen'], env=key_env, check=True, capture_output=True, text=True, timeout=15)
115 self.local_key = self.local / 'XDG_CONFIG_HOME' / 'mux' / 'key'
116 require(self.local_key.is_file() and self.local_key.stat().st_size > 0, 'fixture keygen did not create a private key')
117
118 def _create_remote_root(self):
119 result = self.run('umask 077; mktemp -d /tmp/mux-forward-%s-XXXXXXXX' % os.getuid())
120 candidate = result.stdout.strip()
121 require(candidate.startswith('/tmp/mux-forward-') and '\n' not in candidate, 'remote mktemp returned an unsafe fixture root')
122 self.remote = candidate
123 validate = ('import os,stat,sys; p,m=sys.argv[1:]; s=os.lstat(p); '
124 'assert stat.S_ISDIR(s.st_mode) and s.st_uid==os.getuid(); os.chmod(p,0o700); '
125 'fd=os.open(p+"/.mux-forward-owner",os.O_WRONLY|os.O_CREAT|os.O_EXCL,0o600); '
126 'os.write(fd,m.encode()); os.close(fd); print("OK")')
127 result = self.run(shlex.join([REMOTE_PYTHON, '-c', validate, candidate, self.marker]), check=False)
128 require(result.returncode == 0 and result.stdout.strip() == 'OK', 'remote fixture root failed owner/mode/marker validation')
129
130 def env_prefix(self):
131 return 'env -u MUX_SESSION ' + ' '.join(k + '=' + shlex.quote(v) for k, v in self.env_words.items())
132
133 def run(self, command, check=True, timeout=30):
134 return subprocess.run([self.ssh, self.host, command], text=True, capture_output=True, check=check, timeout=timeout)
135
136 def run_env(self, command, check=True, timeout=30):
137 return self.run(self.env_prefix() + ' /bin/sh -c ' + shlex.quote(command), check, timeout)
138
139 def run_mux(self, verb, *args, stdout_log=None, check=True):
140 command = shlex.join([self.remote_mux, 'd', verb, *args, '--sock', self.remote_sock])
141 if stdout_log:
142 command += ' > ' + shlex.quote(self.remote + '/' + stdout_log) + ' 2>&1'
143 return self.run_env(command, check)
144
145 def _high_udp_port(self):
146 code = ('import random,socket; '\
147 's=socket.socket(socket.AF_INET,socket.SOCK_DGRAM); '\
148 's.bind(("%s",random.randrange(40000,60000))); print(s.getsockname()[1]); s.close()' % REMOTE_ADDRESS)
149 result = self.run(shlex.join([REMOTE_PYTHON, '-c', code]))
150 port = int(result.stdout.strip())
151 require(40000 <= port < 60000, 'remote fixture did not reserve a high UDP port')
152 return port
153
154 def _udp_hex(self):
155 packed = socket.inet_aton(REMOTE_ADDRESS)
156 return packed[::-1].hex().upper() + ':' + format(self.quic_port, '04X')
157
158 def _oracle(self, body, check=True):
159 return self.run_env('. ' + shlex.quote(self.remote_oracle) + '; ' + body, check=check, timeout=15)
160
161 def _record_daemon_pid(self):
162 # The PID is accepted only when the OS oracle sees both the copied
163 # executable and this concrete Unix listener in its fd table.
164 body = ('want=$(real_path ' + shlex.quote(self.remote_mux) + ') || exit 41; n=0; found=; '
165 'for d in /proc/[0-9]*; do p=${d##*/}; pid_alive "$p" || continue; '
166 'exe=$(pid_exe "$p") || continue; [ "$exe" = "$want" ] || continue; '
167 'pid_holds_unix_sock "$p" ' + shlex.quote(self.remote_sock) + ' || continue; '
168 'n=$((n+1)); found=$p; done; [ "$n" -eq 1 ] || exit 42; printf "%s\\n" "$found"')
169 result = self._oracle(body, check=False)
170 require(result.returncode == 0 and result.stdout.strip().isdigit(),
171 'OS oracle could not uniquely identify private daemon executable/socket owner: ' + result.stderr)
172 self.daemon_pid = int(result.stdout.strip())
173 print('OS-ORACLE: daemon pid=%d owns copied mux executable and %s' %
174 (self.daemon_pid, self.remote_sock), flush=True)
175
176 def _assert_daemon_live(self):
177 self._record_daemon_pid()
178 body = ('pid_alive ' + str(self.daemon_pid) + ' && pid_holds_unix_sock ' +
179 str(self.daemon_pid) + ' ' + shlex.quote(self.remote_sock) + ' && udp_local_bound ' + self._udp_hex())
180 result = self._oracle(body, check=False)
181 require(result.returncode == 0, 'OS oracle did not see private daemon socket and QUIC UDP listener')
182
183 def start_daemon(self):
184 self.run_mux('start', '-d', '--quic', REMOTE_ADDRESS + ':' + str(self.quic_port), '--key', self.remote_key,
185 stdout_log='daemon.log')
186 eventually(lambda: self._daemon_ready(), 'private remote QUIC daemon did not start', seconds=15)
187
188 def _daemon_ready(self):
189 try:
190 self._assert_daemon_live()
191 return True
192 except AssertionError:
193 return False
194
195 def stop_daemon_verified(self):
196 require(self.daemon_pid is not None, 'private daemon PID was never OS-verified')
197 stop = self.run_mux('stop', check=False)
198 body = ('! pid_alive ' + str(self.daemon_pid) + ' && [ ! -e ' + shlex.quote(self.remote_sock) +
199 ' ] && ! udp_local_bound ' + self._udp_hex())
200 result = self._oracle(body, check=False)
201 require(stop.returncode == 0, 'private daemon stop command failed: ' + stop.stderr)
202 require(result.returncode == 0, 'private daemon termination lacked OS evidence (pid/socket/UDP): ' + result.stderr)
203 print('OS-ORACLE: daemon pid=%d gone; Unix socket removed; QUIC UDP listener released' %
204 self.daemon_pid, flush=True)
205 self.daemon_pid = None
206
207 def _read_ports(self):
208 result = self.run(shlex.join([REMOTE_PYTHON, '-c', 'import sys; print(open(sys.argv[1]).read())', self.remote + '/ports.json']), check=False)
209 if result.returncode != 0:
210 return None
211 try:
212 ports = json.loads(result.stdout)
213 return ports if all(isinstance(ports.get(k), int) and 0 < ports[k] < 65536 for k in ('http', 'echo')) and isinstance(ports.get('token'), str) else None
214 except json.JSONDecodeError:
215 return None
216
217 def _stop_owned_service(self):
218 code = '''import os,sys,time
219 root,marker=sys.argv[1:]; pid=int(open(root+'/services.pid').read().strip())
220 cmd=open('/proc/%d/cmdline'%pid,'rb').read(); assert root.encode() in cmd and (root+'/services.py').encode() in cmd and open(root+'/.mux-forward-owner').read()==marker
221 os.kill(pid,15)
222 for _ in range(50):
223 try: os.kill(pid,0)
224 except ProcessLookupError: sys.exit(0)
225 time.sleep(.1)
226 sys.exit(43)'''
227 result = self.run(shlex.join([REMOTE_PYTHON, '-c', code, self.remote, self.marker]), check=False, timeout=10)
228 require(result.returncode == 0, 'remote service identity/stop verification failed: ' + result.stderr)
229
230 def _remove_owned_root(self):
231 code = '''import os,shutil,stat,sys
232 root,marker=sys.argv[1:]; s=os.lstat(root)
233 assert stat.S_ISDIR(s.st_mode) and s.st_uid==os.getuid() and (s.st_mode&0o777)==0o700
234 assert open(root+'/.mux-forward-owner').read()==marker
235 assert not os.path.exists(root+'/mux.sock')
236 shutil.rmtree(root)'''
237 result = self.run(shlex.join([REMOTE_PYTHON, '-c', code, self.remote, self.marker]), check=False, timeout=15)
238 require(result.returncode == 0, 'remote root ownership validation refused cleanup: ' + result.stderr)
239
240 def _copy_log(self, failure_logs, name):
241 result = self.run('cat ' + shlex.quote(self.remote + '/' + name), check=False, timeout=10)
242 (Path(failure_logs) / ('remote-' + name)).write_text(result.stdout + result.stderr)
243
244 def close(self, failure_logs=None):
245 if not self.remote:
246 shutil.rmtree(self.local, ignore_errors=True)
247 return
248 if failure_logs is not None:
249 for name in ('daemon.log', 'services.log'):
250 try: self._copy_log(failure_logs, name)
251 except BaseException as error: self.cleanup_errors.append('copy ' + name + ': ' + repr(error))
252 daemon_stopped = self.daemon_pid is None
253 if not daemon_stopped:
254 try:
255 self.stop_daemon_verified(); daemon_stopped = True
256 except BaseException as error:
257 self.cleanup_errors.append('stop/verify private daemon: ' + repr(error))
258 try: self._stop_owned_service()
259 except BaseException as error: self.cleanup_errors.append('stop verified private service: ' + repr(error))
260 if daemon_stopped:
261 try: self._remove_owned_root()
262 except BaseException as error: self.cleanup_errors.append('remove verified private root: ' + repr(error))
263 else:
264 self.cleanup_errors.append('remote root retained because daemon termination was not OS-verified: ' + self.remote)
265 shutil.rmtree(self.local, ignore_errors=True)
266 if not self.cleanup_errors:
267 print('PASS: remote fixture cleanup (daemon PID/socket/UDP released; services and root removed)', flush=True)
268
269
270 def exact_http(port, expected_body, deadline=None):
271 code, body = http_get(port, deadline)
272 require(code == 200 and body == expected_body,
273 'forwarded HTTP response was not exact HTTP 200 fixture identity: ' +
274 repr((code, body[:160])))
275
276
277 def wait_exact_http(port, body, message, seconds=15):
278 deadline = time.monotonic() + seconds
279 while time.monotonic() < deadline:
280 try:
281 exact_http(port, body, deadline)
282 return
283 except PreResponseTransportStartupError:
284 time.sleep(min(.04, max(0, deadline - time.monotonic())))
285 raise AssertionError(message)
286
287
288 def exact_http_api_regression():
289 original = http_get
290 try:
291 globals()['http_get'] = lambda *_: (200, b'fixture')
292 exact_http(1, b'fixture')
293 for response, message in (((201, b'fixture'), 'non-200'),
294 ((200, b'wrong-body'), 'wrong-body')):
295 globals()['http_get'] = lambda *_, response=response: response
296 try:
297 exact_http(1, b'fixture')
298 except AssertionError:
299 pass
300 else:
301 raise AssertionError('exact_http accepted a ' + message + ' tuple response')
302 finally:
303 globals()['http_get'] = original
304
305
306 def run_mode(rig, remote, label, target_args, recover=False):
307 local_http, local_echo = distinct_ports(2)
308 rules = [(local_http, remote.ports['http']), (local_echo, remote.ports['echo'])]
309 print('COMMAND: ' + shlex.join([rig.muxg, *target_args, '--forward', f'{local_http}:{remote.ports["http"]}', '--forward', f'{local_echo}:{remote.ports["echo"]}', '--session', 'forward']), flush=True)
310 rig.launch_forward(target_args, rules, label + '-gui')
311 body = ('native-forward-remote-' + remote.ports['token'] + '\n').encode()
312 wait_exact_http(local_http, body, label + ' forwarding did not reach exact remote HTTP identity')
313 payloads = [os.urandom(1024 * 1024 + 31 + i) for i in range(3)]
314 replies, errors = [None] * len(payloads), []
315 lock = threading.Lock()
316 def worker(index):
317 try: replies[index] = echo_roundtrip(local_echo, payloads[index], half_close=True)
318 except BaseException as error:
319 with lock: errors.append(error)
320 workers = [threading.Thread(target=worker, args=(i,)) for i in range(3)]
321 for worker in workers: worker.start()
322 for worker in workers:
323 worker.join(20)
324 require(not worker.is_alive(), label + ' concurrent >1MiB stream hung')
325 require(not errors, label + ' concurrent stream failure: ' + repr(errors))
326 require(replies == payloads, label + ' concurrent >1MiB bidirectional half-closed streams corrupted bytes')
327 require(listener_reserved(local_http), label + ' listener was not reserved')
328 checksums = [hashlib.sha256(p).hexdigest() for p in payloads]
329 rig.ok(label + ' exact HTTP 200 and three concurrent >1MiB half-close streams sha256=' + ','.join(checksums))
330 if recover:
331 remote.stop_daemon_verified()
332 eventually(lambda: not _http_or_false(local_http), label + ' old listener did not become unavailable')
333 require(listener_reserved(local_http), label + ' interruption released listener reservation')
334 remote.start_daemon()
335 wait_exact_http(local_http, body, label + ' did not recover after whole-daemon restart')
336 rig.ok(label + ' whole-daemon restart recovery retains listener reservation and reconnects')
337 rig.quit()
338
339
340 def _http_or_false(port):
341 try: return bool(http_get(port))
342 except PreResponseTransportStartupError: return False
343
344
345 def main():
346 require(len(sys.argv) == 3, 'usage: native_forward_remote.py RELEASE_MUX RELEASE_MUXG')
347 exact_http_api_regression()
348 require(os.environ.get('MUX_FORWARD_REMOTE_ENABLE') == '1', 'set MUX_FORWARD_REMOTE_ENABLE=1 to contact the authorized remote fixture host')
349 remote = rig = None
350 try:
351 remote = RemoteFixture(sys.argv[1])
352 rig = ForwardRig(sys.argv[1], sys.argv[2])
353 via = shlex.join([remote.ssh, remote.host, remote.remote_mux, 'd', 'proxy', '--sock', remote.remote_sock])
354 run_mode(rig, remote, 'real SSH stdio', ['--via', via])
355 run_mode(rig, remote, 'direct QUIC', ['quic://' + REMOTE_ADDRESS + ':' + str(remote.quic_port), '--key', str(remote.local_key)], recover=True)
356 print('PASS: remote native forwarding', flush=True)
357 except BaseException:
358 if rig: rig.failure_artifacts()
359 raise
360 finally:
361 failed = sys.exc_info()[0] is not None
362 try:
363 # GUI quit precedes final remote stop, preventing reconnect races.
364 if rig: rig.close()
365 finally:
366 if remote:
367 remote.close(rig.root if rig and failed else None)
368 if remote.cleanup_errors:
369 raise RuntimeError('remote fixture cleanup errors: ' + '; '.join(remote.cleanup_errors))
370
371
372 if __name__ == '__main__':
373 main()