1package app
2
3import (
4 "bytes"
5 "context"
6 "encoding/base64"
7 "encoding/json"
8 "errors"
9 "io"
10 "log/slog"
11 "os"
12 "path/filepath"
13 "strconv"
14 "strings"
15 "syscall"
16 "testing"
17 "time"
18 "unicode"
19
20 "github.com/stretchr/testify/require"
21
22 "github.com/bnema/vev/internal/adapters/brokerconfig"
23 "github.com/bnema/vev/internal/adapters/brokeripc"
24 "github.com/bnema/vev/internal/adapters/daemonmux"
25 "github.com/bnema/vev/internal/adapters/ipc"
26 "github.com/bnema/vev/internal/adapters/quic"
27 "github.com/bnema/vev/internal/adapters/sshstdio"
28 "github.com/bnema/vev/internal/ports"
29 "github.com/bnema/vev/internal/protocol"
30)
31
32// Real-composition coverage for the P3.4 slice E remote mux helpers.
33//
34// The SSH carriage is exercised with an owned test subprocess: a fake `ssh`
35// on PATH re-executes this test binary through TestMain's broker-helper
36// interception, so the real helper code runs with real pipes and a real unix
37// daemonmux fixture, and the exact ssh argv is captured for inspection. The
38// QUIC carriage runs a real loopback authenticated QUIC server, so a successful
39// round trip proves pin, token, nonce, and daemonmux handshake all compose.
40
41const (
42 fakeSSHArgsEnv = "VEV_TEST_SSH_ARGS"
43 fakeSSHReadyEnv = "VEV_TEST_SSH_READY"
44 fakeSSHPIDEnv = "VEV_TEST_SSH_PID"
45 fakeSSHBinEnv = "VEV_TEST_SSH_BIN"
46 fakeSSHModeEnv = "VEV_TEST_SSH_MODE"
47 fakeSSHRuntimeEnv = "VEV_TEST_SSH_REMOTE_RUNTIME"
48 fakeSSHStateEnv = "VEV_TEST_SSH_REMOTE_STATE"
49 fakeSSHConfigEnv = "VEV_TEST_SSH_REMOTE_CONFIG"
50)
51
52// fakeSSHExecHelper is an owned test ssh: it records its exact argv and then
53// re-executes this test binary as the named hidden mux helper against an
54// isolated remote production XDG layout. Its stdin/stdout remain the carriage pipes.
55const fakeSSHExecHelper = `#!/bin/sh
56printf '%s\n' "$@" > "$` + fakeSSHArgsEnv + `"
57export XDG_RUNTIME_DIR="$` + fakeSSHRuntimeEnv + `"
58export XDG_STATE_HOME="$` + fakeSSHStateEnv + `"
59export XDG_CONFIG_HOME="$` + fakeSSHConfigEnv + `"
60exec "$` + fakeSSHBinEnv + `" "$` + fakeSSHModeEnv + `" --production
61`
62
63// fakeSSHEmitReadiness prints one controlled readiness line and exits, so a
64// test can drive the broker-side pinned QUIC dial without a real remote host.
65const fakeSSHEmitReadiness = `#!/bin/sh
66printf '%s\n' "$@" > "$` + fakeSSHArgsEnv + `"
67printf '%s\n' "$` + fakeSSHReadyEnv + `"
68exit 0
69`
70
71// fakeSSHSleep blocks without producing readiness, so a test can prove setup
72// cancellation reaps the bootstrap child.
73const fakeSSHSleep = `#!/bin/sh
74printf '%s\n' "$@" > "$` + fakeSSHArgsEnv + `"
75printf '%s\n' "$$" > "$` + fakeSSHPIDEnv + `"
76sleep 30
77`
78
79// fakeSSHFloodStderr floods stderr with text carrying a terminal escape and a
80// control byte and never prints readiness, so a test can prove the broker-side
81// diagnostic sink is read only after the os/exec copy goroutine has been killed
82// and joined.
83const fakeSSHFloodStderr = `#!/bin/sh
84printf '%s\n' "$@" > "$` + fakeSSHArgsEnv + `"
85printf '%s\n' "$$" > "$` + fakeSSHPIDEnv + `"
86i=0
87while true; do
88 printf 'flood-%s\033[31mred\033[0m\001\002-noise\n' "$i" >&2
89 i=$((i+1))
90 sleep 0.01
91done
92`
93
94// fakeSSHStallNeverReady stalls without printing readiness and without exiting,
95// so a test can prove the independent broker-side setup timeout interrupts a
96// bootstrap that the caller never cancels. It execs the blocking sleep so the
97// reaped process owns the captured stderr pipe and no grandchild lingers.
98const fakeSSHStallNeverReady = `#!/bin/sh
99printf '%s\n' "$@" > "$` + fakeSSHArgsEnv + `"
100printf '%s\n' "$$" > "$` + fakeSSHPIDEnv + `"
101exec sleep 30
102`
103
104// installFakeSSH puts an owned `ssh` executable first on PATH and returns the
105// file its argv is recorded to.
106func installFakeSSH(t *testing.T, script string) string {
107 t.Helper()
108 binDir := t.TempDir()
109 require.NoError(t, os.Chmod(binDir, 0o755))
110 argsFile := filepath.Join(t.TempDir(), "ssh-args")
111 require.NoError(t, os.WriteFile(filepath.Join(binDir, "ssh"), []byte(script), 0o755))
112 t.Setenv("PATH", binDir+string(os.PathListSeparator)+os.Getenv("PATH"))
113 t.Setenv(fakeSSHArgsEnv, argsFile)
114 t.Setenv(fakeSSHBinEnv, os.Args[0])
115 return argsFile
116}
117
118// muxTestEnv isolates production XDG roots and records every detached mux
119// helper so no test leaks a process.
120type muxTestEnv struct {
121 t *testing.T
122 recordDir string
123}
124
125func newMuxTestEnv(t *testing.T) *muxTestEnv {
126 t.Helper()
127 emptyProductionBrokerLayout(t, "")
128 recordDir := filepath.Join(shortTempDir(t, "vevm"), "r")
129 require.NoError(t, os.MkdirAll(recordDir, 0o700))
130 t.Setenv(brokerHelperEnv, "1")
131 t.Setenv(brokerHelperRecordDirEnv, recordDir)
132 env := &muxTestEnv{t: t, recordDir: recordDir}
133 t.Cleanup(env.cleanup)
134 return env
135}
136
137// remoteMuxFixture runs the SSH-side mux helper with a distinct production
138// XDG environment and a fixture listening at that environment's daemon path.
139func remoteMuxFixture(t *testing.T, policy ports.BrokerPolicy) *brokerMuxFixture {
140 t.Helper()
141 localRuntime, localState, localConfig := os.Getenv("XDG_RUNTIME_DIR"), os.Getenv("XDG_STATE_HOME"), os.Getenv("XDG_CONFIG_HOME")
142 remoteRuntime, remoteState, remoteConfig := shortTempDir(t, "vb"), shortTempDir(t, "vb"), shortTempDir(t, "vb")
143 t.Setenv("XDG_RUNTIME_DIR", remoteRuntime)
144 t.Setenv("XDG_STATE_HOME", remoteState)
145 t.Setenv("XDG_CONFIG_HOME", remoteConfig)
146 path := daemonmux.SocketPath(ipc.SocketDir())
147 fixture := newBrokerMuxFixtureAt(t, policy, path)
148 t.Setenv("XDG_RUNTIME_DIR", localRuntime)
149 t.Setenv("XDG_STATE_HOME", localState)
150 t.Setenv("XDG_CONFIG_HOME", localConfig)
151 t.Setenv(fakeSSHRuntimeEnv, remoteRuntime)
152 t.Setenv(fakeSSHStateEnv, remoteState)
153 t.Setenv(fakeSSHConfigEnv, remoteConfig)
154 return fixture
155}
156
157// cleanup kills every recorded mux helper process.
158func (e *muxTestEnv) cleanup() {
159 for _, role := range []string{brokerMuxStdioCommand, brokerMuxQUICBootstrapCommand, brokerMuxQUICProxyCommand} {
160 raw, err := os.ReadFile(filepath.Join(e.recordDir, role+".pids"))
161 if err != nil {
162 continue
163 }
164 for _, field := range strings.Fields(string(raw)) {
165 pid, convErr := strconv.Atoi(field)
166 if convErr != nil {
167 continue
168 }
169 if err := syscall.Kill(pid, syscall.SIGKILL); err != nil && !errors.Is(err, syscall.ESRCH) {
170 continue
171 }
172 _ = waitForProcessExit(pid, 5*time.Second)
173 }
174 }
175}
176
177// writeMuxRoot provisions one pinned host route in this test's production
178// broker.json and returns the production layout for its reader.
179func writeMuxRoot(t *testing.T, route any) brokerconfig.Layout {
180 t.Helper()
181 layout := emptyProductionBrokerLayout(t, "")
182 writeProductionBrokerDocument(t, sandboxRegistrationDocument(route))
183 return layout
184}
185
186// loadTestRoute loads one production broker configuration and returns the route its
187// resolver publishes, so a test can inspect a parsed route without the
188// unexported brokerconfig fields.
189func loadTestRoute(t *testing.T, route any) brokerconfig.Route {
190 t.Helper()
191 _, parsed := loadTestRouteAndConfig(t, route)
192 return parsed
193}
194
195// loadTestRouteAndConfig loads one production broker configuration and returns both the
196// immutable configuration and the route its resolver publishes, so a test can
197// dial the route the way the broker does: with the whole resolved target.
198func loadTestRouteAndConfig(t *testing.T, route any) (*brokerconfig.Config, brokerconfig.Route) {
199 t.Helper()
200 layout := writeMuxRoot(t, route)
201 config, err := brokerconfig.LoadProduction(layout, productionBrokerConfigPath(), "", localDaemonPolicy(), daemonmux.SocketPath(ipc.SocketDir()))
202 require.NoError(t, err)
203 resolved, err := config.Resolver().ResolveDialTarget(context.Background(), ports.BrokerOpenStreamRequest{
204 Purpose: ports.BrokerStreamControl,
205 Stream: 1,
206 Endpoint: brokerTestEndpoint,
207 Registration: brokerTestRegistration(),
208 Policy: brokerTestPolicy(),
209 StartMode: ports.BrokerDaemonStartIfNeeded,
210 })
211 require.NoError(t, err)
212 parsed, ok := config.RouteByAddress(resolved.Address)
213 require.True(t, ok)
214 return config, parsed
215}
216
217// testMuxDialTarget resolves the exact dial target the broker publishes for one
218// provisioned route under one start authorization, so a test dials the route
219// with the same whole target the pool does.
220func testMuxDialTarget(t *testing.T, config *brokerconfig.Config, route brokerconfig.Route, mode ports.BrokerDaemonStartMode) ports.BrokerDialTarget {
221 t.Helper()
222 resolved, err := config.Resolver().ResolveDialTarget(context.Background(), ports.BrokerOpenStreamRequest{
223 Purpose: ports.BrokerStreamControl,
224 Stream: 1,
225 Endpoint: brokerTestEndpoint,
226 Registration: brokerTestRegistration(),
227 Policy: brokerTestPolicy(),
228 StartMode: mode,
229 })
230 require.NoError(t, err)
231 require.Equal(t, route.Address(), resolved.Address, "the test dials the provisioned route")
232 return resolved
233}
234
235// loadTestRouteTarget loads one production broker configuration and returns both the
236// parsed route and the exact resolved dial target the broker would publish for
237// it, so a test can call dialBrokerRoute exactly as the pool does.
238func loadTestRouteTarget(t *testing.T, route any) (brokerconfig.Route, ports.BrokerDialTarget) {
239 t.Helper()
240 config, parsed := loadTestRouteAndConfig(t, route)
241 return parsed, testMuxDialTarget(t, config, parsed, ports.BrokerDaemonStartIfNeeded)
242}
243
244// TestParseBrokerMuxArgs pins the strict hidden-helper argument contract.
245func TestParseBrokerMuxArgs(t *testing.T) {
246 tests := []struct {
247 name string
248 args []string
249 wantErr string
250 want brokerMuxOptions
251 }{
252 {name: "valid", args: []string{"--production"}, want: brokerMuxOptions{startMode: ports.BrokerDaemonExistingOnly}},
253 {name: "explicit existing-only", args: []string{"--production", brokerDaemonStartArg, "existing-only"}, want: brokerMuxOptions{startMode: ports.BrokerDaemonExistingOnly}},
254 {name: "explicit if-needed", args: []string{"--production", brokerDaemonStartArg, "if-needed"}, want: brokerMuxOptions{startMode: ports.BrokerDaemonStartIfNeeded}},
255 {name: "unknown flag", args: []string{"--production", "--idle-grace", "1m"}, wantErr: "unknown flag"},
256 {name: "positional", args: []string{"--production", "extra"}, wantErr: "positional"},
257 {name: "missing daemon-start value", args: []string{"--production", brokerDaemonStartArg}, wantErr: "requires a value"},
258 {name: "duplicate daemon-start", args: []string{"--production", brokerDaemonStartArg, "if-needed", brokerDaemonStartArg, "existing-only"}, wantErr: "duplicate"},
259 {name: "unknown daemon-start", args: []string{"--production", brokerDaemonStartArg, "always"}, wantErr: "unknown daemon start mode"},
260 }
261 for _, tt := range tests {
262 t.Run(tt.name, func(t *testing.T) {
263 command, err := parseBrokerMuxArgs(brokerMuxStdioCommand, kindBrokerMuxStdio, tt.args)
264 if tt.wantErr != "" {
265 require.Error(t, err)
266 require.Contains(t, err.Error(), tt.wantErr)
267 return
268 }
269 require.NoError(t, err)
270 require.Equal(t, tt.want, command.brokerMux)
271 })
272 }
273}
274
275// TestBrokerMuxHelpersHiddenFromPublicHelp proves the three helpers parse but
276// stay out of public help and never alter ordinary commands.
277func TestBrokerMuxHelpersHiddenFromPublicHelp(t *testing.T) {
278 for _, tt := range []struct {
279 name string
280 kind cmdKind
281 }{
282 {brokerMuxStdioCommand, kindBrokerMuxStdio},
283 {brokerMuxQUICBootstrapCommand, kindBrokerMuxQUICBootstrap},
284 {brokerMuxQUICProxyCommand, kindBrokerMuxQUICProxy},
285 } {
286 require.NotContains(t, usageText, tt.name)
287 command, err := parseArgs([]string{tt.name, "--production"})
288 require.NoError(t, err)
289 require.Equal(t, tt.kind, command.kind)
290 }
291 // Ordinary commands are untouched by the new hidden entries.
292 attach, err := parseArgs(nil)
293 require.NoError(t, err)
294 require.Equal(t, kindAttach, attach.kind)
295 list, err := parseArgs([]string{"ls"})
296 require.NoError(t, err)
297 require.Equal(t, kindList, list.kind)
298}
299
300// TestSSHMuxCommandSpecPreservesTrustAndNoPTY proves the ssh argv keeps `-T`,
301// the option terminator, and the quoted remote argv, splices only the explicit
302// trust inputs, and never disables host-key verification.
303func TestSSHMuxCommandSpecPreservesTrustAndNoPTY(t *testing.T) {
304 emptyProductionBrokerLayout(t, "")
305 route := loadTestRoute(t, map[string]any{
306 "kind": "ssh-stdio",
307 "target": "user@host:2222",
308 "argv": []any{"vev", brokerMuxStdioCommand, "--production"},
309 "trust": map[string]any{"knownHostsFile": "/etc/ssh/known_hosts", "connectTimeout": "30s"},
310 })
311 spec, err := sshMuxCommandSpec(route, ports.BrokerDaemonStartIfNeeded)
312 require.NoError(t, err)
313 require.Equal(t, "ssh", spec.Path)
314 require.Equal(t, "-T", spec.Args[0])
315 require.Contains(t, spec.Args, "--")
316 require.Contains(t, spec.Args, "UserKnownHostsFile=/etc/ssh/known_hosts")
317 require.Contains(t, spec.Args, "ConnectTimeout=30")
318 joined := strings.Join(spec.Args, " ")
319 require.Contains(t, joined, "-- user@host:2222 ")
320 require.Contains(t, joined, brokerMuxStdioCommand)
321 require.Contains(t, joined, "'"+brokerDaemonStartArg+"' 'if-needed'")
322 require.NotContains(t, joined, "StrictHostKeyChecking")
323 require.NotContains(t, joined, "BatchMode")
324 require.NotContains(t, joined, "ProxyCommand")
325 // The option terminator protects the target, and the -T option comes first.
326 require.Less(t, indexOf(spec.Args, "-T"), indexOf(spec.Args, "--"))
327 for _, arg := range spec.Args {
328 require.NotEqual(t, "no", strings.ToLower(arg))
329 }
330
331 // A route without trust inputs reuses the mux command construction
332 // unchanged (`-T -- target remote`).
333 plain := loadTestRoute(t, map[string]any{
334 "kind": "ssh-quic",
335 "target": "host",
336 "argv": []any{"vev", brokerMuxQUICBootstrapCommand, "--production"},
337 })
338 plainSpec, err := sshMuxCommandSpec(plain, ports.BrokerDaemonExistingOnly)
339 require.NoError(t, err)
340 require.Equal(t, []string{"-T", "--"}, append([]string(nil), plainSpec.Args[:2]...))
341 require.NotContains(t, plainSpec.Args, "-o")
342 require.Contains(t, strings.Join(plainSpec.Args, " "), "'"+brokerDaemonStartArg+"' 'existing-only'")
343}
344
345// TestSSHMuxCommandSpecRefusesUnknownStartMode proves a route never carries an
346// unreadable start authorization: an unknown mode is refused instead of
347// defaulting the helper to a permissive one.
348func TestSSHMuxCommandSpecRefusesUnknownStartMode(t *testing.T) {
349 emptyProductionBrokerLayout(t, "")
350 route := loadTestRoute(t, map[string]any{
351 "kind": "ssh-stdio",
352 "target": "host",
353 "argv": []any{"vev", brokerMuxStdioCommand, "--production"},
354 })
355 _, err := sshMuxCommandSpec(route, ports.BrokerDaemonStartMode(0))
356 require.Error(t, err)
357}
358
359// TestSSHMuxCommandSpecTargetQuotesSpacesInsideArgv proves a remote argv word
360// with a space stays one quoted word and never becomes a second option.
361func TestSSHMuxCommandSpecTargetQuotesSpacesInsideArgv(t *testing.T) {
362 emptyProductionBrokerLayout(t, "")
363 route := loadTestRoute(t, map[string]any{
364 "kind": "ssh-stdio",
365 "target": "host",
366 "argv": []any{"vev", "_broker-mux-stdio", "--production", "with space"},
367 })
368 spec, err := sshMuxCommandSpec(route, ports.BrokerDaemonExistingOnly)
369 require.NoError(t, err)
370 // The whole remote command stays one quoted argv word, so the space-bearing
371 // path is still one word and never becomes a second option.
372 require.Contains(t, spec.Args[len(spec.Args)-1], `'with space'`)
373 require.Contains(t, spec.Args[len(spec.Args)-1], "'"+brokerDaemonStartArg+"' 'existing-only'")
374}
375
376// TestBrokerQUICPeerAddr pins the host-independent address composition.
377func TestBrokerQUICPeerAddr(t *testing.T) {
378 tests := []struct {
379 target string
380 port int
381 want string
382 wantErr bool
383 }{
384 {target: "user@host", port: 2222, want: "host:2222"},
385 {target: "user@host:22", port: 2222, want: "host:2222"},
386 {target: "host", port: 1, want: "host:1"},
387 {target: "user@[::1]", port: 7, want: "[::1]:7"},
388 {target: "host", port: 0, wantErr: true},
389 {target: "host", port: 65536, wantErr: true},
390 {target: "user@", port: 7, wantErr: true},
391 }
392 for _, tt := range tests {
393 got, err := brokerQUICPeerAddr(tt.target, tt.port)
394 if tt.wantErr {
395 require.Error(t, err)
396 continue
397 }
398 require.NoError(t, err)
399 require.Equal(t, tt.want, got)
400 }
401}
402
403func TestBrokerMuxHelperReturnsTypedNoDaemonExit(t *testing.T) {
404 emptyProductionBrokerLayout(t, "")
405 emptyProductionBrokerLayout(t, "")
406 parsed, err := parseBrokerMuxArgs(brokerMuxStdioCommand, kindBrokerMuxStdio, []string{"--production"})
407 require.NoError(t, err)
408 err = runBrokerMuxStdioCommand(context.Background(), parsed.brokerMux)
409 var coded *exitCoded
410 require.ErrorAs(t, err, &coded)
411 require.Equal(t, sshstdio.MuxExitNoDaemon, coded.code)
412}
413
414// TestBrokerServeSSHStdioRouteRoundTrip drives the full broker path over an SSH
415// stdio route: the broker dials the fake ssh child, the hidden helper bridges
416// its stdio to a real unix daemonmux fixture, and a typed Ping returns a Pong.
417func TestBrokerServeSSHStdioRouteRoundTrip(t *testing.T) {
418 newMuxTestEnv(t)
419 policy := brokerTestPolicy()
420 fixture := remoteMuxFixture(t, policy)
421 argsFile := installFakeSSH(t, fakeSSHExecHelper)
422 t.Setenv(fakeSSHModeEnv, brokerMuxStdioCommand)
423 writeMuxRoot(t, map[string]any{
424 "kind": "ssh-stdio",
425 "target": "user@muxhost:2222",
426 "argv": []any{"vev", brokerMuxStdioCommand, "--production"},
427 })
428
429 socketPath, stop := startMuxBrokerServe(t)
430 defer stop()
431 requireBrokerPing(t, socketPath)
432 require.Positive(t, fixture.streams.Load())
433
434 // The exact ssh argv was built by BuildCommandForMux with OpenSSH trust policy.
435 argv := readFileString(t, argsFile)
436 require.Contains(t, argv, "-T\n")
437 require.Contains(t, argv, "--\nuser@muxhost:2222\n")
438 require.Contains(t, argv, brokerMuxStdioCommand)
439 require.NotContains(t, argv, "UserKnownHostsFile=")
440 require.NotContains(t, argv, "StrictHostKeyChecking")
441 require.NotContains(t, argv, "BatchMode")
442
443 stop()
444}
445
446// TestBrokerServeSSHQUICRouteRoundTrip drives the full broker path over an
447// SSH-authenticated QUIC route with a real loopback QUIC server and a real
448// one-time credential.
449func TestBrokerServeSSHQUICRouteRoundTrip(t *testing.T) {
450 newMuxTestEnv(t)
451 policy := brokerTestPolicy()
452 fixture := remoteMuxFixture(t, policy)
453 installFakeSSH(t, fakeSSHExecHelper)
454 t.Setenv(fakeSSHModeEnv, brokerMuxQUICBootstrapCommand)
455 writeMuxRoot(t, map[string]any{
456 "kind": "ssh-quic",
457 "target": "127.0.0.1",
458 "argv": []any{"vev", brokerMuxQUICBootstrapCommand, "--production"},
459 })
460
461 socketPath, stop := startMuxBrokerServe(t)
462 defer stop()
463 requireBrokerPing(t, socketPath)
464 require.Positive(t, fixture.streams.Load())
465
466 stop()
467}
468
469// TestBrokerMuxQUICRejectsBadPinWithoutLeakingSecret proves the pinned dial
470// fails closed on a wrong certificate pin and that the one-time credential
471// never appears in the returned error.
472func TestBrokerMuxQUICRejectsBadPinWithoutLeakingSecret(t *testing.T) {
473 emptyProductionBrokerLayout(t, "")
474 installFakeSSH(t, fakeSSHEmitReadiness)
475
476 server, readiness, err := quic.NewServer()
477 require.NoError(t, err)
478 defer func() { _ = server.Close() }()
479 // Flip the pin: the TLS handshake must fail before any token is accepted.
480 tampered := readiness
481 tampered.Fingerprint = strings.Repeat("00", 32)
482 t.Setenv(fakeSSHReadyEnv, string(mustJSON(t, tampered)))
483
484 route, target := loadTestRouteTarget(t, map[string]any{
485 "kind": "ssh-quic",
486 "target": "127.0.0.1",
487 "argv": []any{"vev", brokerMuxQUICBootstrapCommand, "--production"},
488 })
489 ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
490 defer cancel()
491 _, err = dialBrokerRoute(ctx, route, target, slog.New(slog.NewTextHandler(io.Discard, nil)))
492 require.Error(t, err)
493 requireBrokerErrorHasNoReadiness(t, err, readiness)
494}
495
496// TestBrokerMuxQUICRejectsExpiredReadiness proves client-side parsing refuses
497// an implausibly old credential before dialing and never echoes its secret.
498func TestBrokerMuxQUICRejectsExpiredReadiness(t *testing.T) {
499 emptyProductionBrokerLayout(t, "")
500 installFakeSSH(t, fakeSSHEmitReadiness)
501
502 token := base64.StdEncoding.EncodeToString([]byte(strings.Repeat("A", 32)))
503 nonce := base64.StdEncoding.EncodeToString([]byte(strings.Repeat("B", 16)))
504 readiness := quic.Readiness{
505 Version: 1,
506 Port: 2222,
507 Token: token,
508 Nonce: nonce,
509 Fingerprint: strings.Repeat("00", 32),
510 ExpiresAt: time.Now().Add(-time.Hour).Unix(),
511 }
512 t.Setenv(fakeSSHReadyEnv, string(mustJSON(t, readiness)))
513
514 route, target := loadTestRouteTarget(t, map[string]any{
515 "kind": "ssh-quic",
516 "target": "127.0.0.1",
517 "argv": []any{"vev", brokerMuxQUICBootstrapCommand, "--production"},
518 })
519 _, err := dialBrokerRoute(context.Background(), route, target, slog.New(slog.NewTextHandler(io.Discard, nil)))
520 require.Error(t, err)
521 require.NotContains(t, err.Error(), token)
522 require.NotContains(t, err.Error(), nonce)
523}
524
525// TestBrokerMuxDialCancellationReapsBootstrap proves one setup context bounds
526// the bootstrap wait, a cancellation returns promptly, and the bootstrap child
527// is killed and reaped rather than leaked.
528func TestBrokerMuxDialCancellationReapsBootstrap(t *testing.T) {
529 emptyProductionBrokerLayout(t, "")
530 installFakeSSH(t, fakeSSHSleep)
531 pidFile := filepath.Join(t.TempDir(), "ssh.pid")
532 t.Setenv(fakeSSHPIDEnv, pidFile)
533
534 route, target := loadTestRouteTarget(t, map[string]any{
535 "kind": "ssh-quic",
536 "target": "127.0.0.1",
537 "argv": []any{"vev", brokerMuxQUICBootstrapCommand, "--production"},
538 })
539 ctx, cancel := context.WithTimeout(context.Background(), 150*time.Millisecond)
540 defer cancel()
541 start := time.Now()
542 _, err := dialBrokerRoute(ctx, route, target, slog.New(slog.NewTextHandler(io.Discard, nil)))
543 require.Error(t, err)
544 require.Less(t, time.Since(start), 3*time.Second, "cancellation must return promptly")
545
546 pid := readPID(t, pidFile)
547 require.NoError(t, waitForProcessExit(pid, 3*time.Second), "the bootstrap child must be reaped")
548}
549
550// TestBrokerMuxDialFloodedStderrIsReapedAndSanitized is the regression for the
551// stderr DiagnosticSink race: a bootstrap that floods stderr and never prints
552// readiness must be killed and reaped before the captured sink is read, and the
553// diagnostic that reaches the log must stay bounded and control-byte-free. Run
554// under -race with -count>1, this fails if the os/exec copy goroutine is ever
555// observed while it is still writing the sink.
556func TestBrokerMuxDialFloodedStderrIsReapedAndSanitized(t *testing.T) {
557 emptyProductionBrokerLayout(t, "")
558 installFakeSSH(t, fakeSSHFloodStderr)
559 pidFile := filepath.Join(t.TempDir(), "ssh.pid")
560 t.Setenv(fakeSSHPIDEnv, pidFile)
561
562 route, target := loadTestRouteTarget(t, map[string]any{
563 "kind": "ssh-quic",
564 "target": "127.0.0.1",
565 "argv": []any{"vev", brokerMuxQUICBootstrapCommand, "--production"},
566 })
567
568 var logs bytes.Buffer
569 logger := slog.New(slog.NewJSONHandler(&logs, nil))
570 ctx, cancel := context.WithTimeout(context.Background(), 400*time.Millisecond)
571 defer cancel()
572 _, err := dialBrokerRoute(ctx, route, target, logger)
573 require.Error(t, err)
574 require.NotContains(t, err.Error(), "flood", "bootstrap stderr must never reach the error")
575
576 // Reaping is observable: the flooding bootstrap is gone, not leaked.
577 pid := readPID(t, pidFile)
578 require.NoError(t, waitForProcessExit(pid, 3*time.Second), "the flooding bootstrap must be reaped")
579
580 record := decodeBootstrapDiagnostic(t, logs.Bytes())
581 // Each remote physical bootstrap logs exactly one broker_remote_dial; the
582 // warm-reuse harness asserts its absence.
583 require.Equal(t, 1, bytes.Count(logs.Bytes(), []byte(`"msg":"broker_remote_dial"`)))
584 require.NotEmpty(t, record.Diagnostic)
585 require.LessOrEqual(t, len(record.Diagnostic), 512, "the captured diagnostic must stay bounded")
586 for _, r := range record.Diagnostic {
587 require.False(t, unicode.IsControl(r) && r != '\t' && r != '\n', "control byte %q leaked into the diagnostic", r)
588 }
589}
590
591// TestBrokerMuxDialStalledBootstrapUsesSetupTimeout proves the independent
592// broker-side setup timeout bounds the readiness wait: with no caller deadline a
593// helper that stalls without printing readiness is still interrupted within the
594// bound, and its child is killed and reaped. It is deterministic because the
595// effective bound is narrowed to a test value instead of the production one.
596func TestBrokerMuxDialStalledBootstrapUsesSetupTimeout(t *testing.T) {
597 emptyProductionBrokerLayout(t, "")
598 installFakeSSH(t, fakeSSHStallNeverReady)
599 pidFile := filepath.Join(t.TempDir(), "ssh.pid")
600 t.Setenv(fakeSSHPIDEnv, pidFile)
601
602 restore := brokerMuxBootstrapReadyTimeout
603 brokerMuxBootstrapReadyTimeout = 250 * time.Millisecond
604 t.Cleanup(func() { brokerMuxBootstrapReadyTimeout = restore })
605
606 route, target := loadTestRouteTarget(t, map[string]any{
607 "kind": "ssh-quic",
608 "target": "127.0.0.1",
609 "argv": []any{"vev", brokerMuxQUICBootstrapCommand, "--production"},
610 })
611 start := time.Now()
612 _, err := dialBrokerRoute(context.Background(), route, target, slog.New(slog.NewTextHandler(io.Discard, nil)))
613 elapsed := time.Since(start)
614 require.Error(t, err)
615 require.Contains(t, err.Error(), "read bootstrap readiness")
616 require.GreaterOrEqual(t, elapsed, 200*time.Millisecond, "the independent bound must not return early")
617 require.Less(t, elapsed, 5*time.Second, "the independent bound must interrupt a stalled bootstrap")
618
619 pid := readPID(t, pidFile)
620 require.NoError(t, waitForProcessExit(pid, 3*time.Second), "the stalled bootstrap must be reaped")
621}
622
623// TestBrokerServeUnixRouteRoundTrip proves the transport selector also handles
624// the explicit unix route variant, not just the legacy path string.
625func TestBrokerServeUnixRouteRoundTrip(t *testing.T) {
626 newMuxTestEnv(t)
627 fixture := newBrokerMuxFixture(t, brokerTestPolicy())
628 writeMuxRoot(t, map[string]any{"kind": "unix", "path": fixture.route})
629
630 socketPath, stop := startMuxBrokerServe(t)
631 defer stop()
632 requireBrokerPing(t, socketPath)
633 require.Positive(t, fixture.streams.Load())
634
635 stop()
636}
637
638// startMuxBrokerServe starts one production broker and returns its
639// socket path plus an idempotent stop.
640func startMuxBrokerServe(t *testing.T) (string, func()) {
641 t.Helper()
642 deps, ready := testBrokerServeDeps(newSandboxClock())
643 ctx, cancel := context.WithCancel(context.Background())
644 done := runSandbox(ctx, brokerServeOptions{}, deps)
645 socketPath := awaitSandboxReady(t, ready, done)
646 var once bool
647 stop := func() {
648 if once {
649 return
650 }
651 once = true
652 awaitSandboxCancel(t, cancel, done)
653 requireSocketRemoved(t, socketPath)
654 }
655 return socketPath, stop
656}
657
658// requireBrokerPing opens one control stream and requires one Pong.
659func requireBrokerPing(t *testing.T, socketPath string) {
660 t.Helper()
661 clientCtx, clientCancel := context.WithTimeout(context.Background(), brokerTestWait)
662 defer clientCancel()
663 service, err := brokeripc.Dial(clientCtx, socketPath, brokeripc.Config{})
664 require.NoError(t, err)
665 defer func() { _ = service.Close() }()
666
667 streamID, err := service.NextStreamID()
668 require.NoError(t, err)
669 stream, err := service.OpenStream(clientCtx, ports.BrokerOpenStreamRequest{
670 Purpose: ports.BrokerStreamControl,
671 Stream: streamID,
672 Endpoint: brokerTestEndpoint,
673 Registration: brokerTestRegistration(),
674 Policy: brokerTestPolicy(),
675 StartMode: ports.BrokerDaemonStartIfNeeded,
676 })
677 require.NoError(t, err)
678 defer func() { _ = stream.Close() }()
679
680 type pingResult struct {
681 message protocol.ServerMessage
682 err error
683 }
684 result := make(chan pingResult, 1)
685 go func() {
686 if sendErr := stream.SendClient(protocol.Ping{}); sendErr != nil {
687 result <- pingResult{err: sendErr}
688 return
689 }
690 message, recvErr := stream.ReceiveServer()
691 result <- pingResult{message: message, err: recvErr}
692 }()
693 select {
694 case outcome := <-result:
695 require.NoError(t, outcome.err)
696 require.Equal(t, protocol.Pong{}, outcome.message)
697 case <-time.After(brokerTestWait):
698 t.Fatal("typed broker -> mux round trip timed out")
699 }
700}
701
702// requireBrokerErrorHasNoReadiness asserts no readiness secret or stderr leaks
703// into a returned error.
704func requireBrokerErrorHasNoReadiness(t *testing.T, err error, readiness quic.Readiness) {
705 t.Helper()
706 message := err.Error()
707 require.NotContains(t, message, readiness.Token)
708 require.NotContains(t, message, readiness.Nonce)
709 require.NotContains(t, message, readiness.Fingerprint)
710 require.NotContains(t, message, "token")
711 require.NotContains(t, message, "nonce")
712}
713
714func mustJSON(t *testing.T, value any) []byte {
715 t.Helper()
716 raw, err := json.Marshal(value)
717 require.NoError(t, err)
718 return raw
719}
720
721func readFileString(t *testing.T, path string) string {
722 t.Helper()
723 raw, err := os.ReadFile(path)
724 require.NoError(t, err)
725 return string(raw)
726}
727
728func readPID(t *testing.T, path string) int {
729 t.Helper()
730 deadline := time.Now().Add(3 * time.Second)
731 for {
732 raw, err := os.ReadFile(path)
733 if err == nil {
734 pid, convErr := strconv.Atoi(strings.TrimSpace(string(raw)))
735 require.NoError(t, convErr)
736 return pid
737 }
738 if !time.Now().Before(deadline) {
739 t.Fatalf("read pid file %s: %v", path, err)
740 }
741 time.Sleep(10 * time.Millisecond)
742 }
743}
744
745func indexOf(values []string, want string) int {
746 for i, value := range values {
747 if value == want {
748 return i
749 }
750 }
751 return -1
752}
753
754// bootstrapDiagnostic is one decoded broker_mux_bootstrap_stderr log record.
755type bootstrapDiagnostic struct {
756 Message string `json:"msg"`
757 Diagnostic string `json:"diagnostic"`
758}
759
760// decodeBootstrapDiagnostic decodes the single sanitized diagnostic the broker
761// logs for a failed bootstrap. Other records (the broker_remote_dial line) are
762// skipped; exactly one diagnostic must be present.
763func decodeBootstrapDiagnostic(t *testing.T, raw []byte) bootstrapDiagnostic {
764 t.Helper()
765 var found []bootstrapDiagnostic
766 for _, line := range bytes.Split(bytes.TrimSpace(raw), []byte("\n")) {
767 var record bootstrapDiagnostic
768 require.NoError(t, json.Unmarshal(line, &record))
769 if record.Message == "broker_mux_bootstrap_stderr" {
770 found = append(found, record)
771 }
772 }
773 require.Len(t, found, 1)
774 return found[0]
775}