1package app
2
3import (
4 "context"
5 "errors"
6 "fmt"
7 "io"
8 "log/slog"
9 "net"
10 "os"
11 "os/exec"
12 "strconv"
13 "strings"
14 "syscall"
15 "time"
16
17 "github.com/bnema/vev/internal/adapters/brokerconfig"
18 "github.com/bnema/vev/internal/adapters/daemonmux"
19 "github.com/bnema/vev/internal/adapters/ipc"
20 "github.com/bnema/vev/internal/adapters/quic"
21 "github.com/bnema/vev/internal/adapters/sshstdio"
22 "github.com/bnema/vev/internal/domain"
23 "github.com/bnema/vev/internal/ports"
24 "github.com/bnema/vev/internal/protocol/wire"
25 "github.com/bnema/vev/internal/usecase/broker"
26)
27
28// Broker mux remote helper composition.
29//
30// The offline broker reaches a remote daemon's private daemonmux carriage
31// through one of three explicitly provisioned routes: a local Unix mux socket,
32// an SSH stdio child that runs a hidden mux helper, or an SSH-authenticated
33// QUIC bootstrap that pins a freshly minted ephemeral certificate. This file
34// owns both halves.
35//
36// The broker side is a transport-selecting daemonmux.EndpointConnector. In
37// production it rechecks remote targets against committed registry records;
38// sandbox fixtures retain their immutable configured routes and trust inputs. It
39// never authenticates the carriage itself: the daemonmux physical preamble
40// verifies the daemon identity, incarnation, and policy the resolver already
41// pinned. One caller-supplied setup context bounds bootstrap, pinned dial, and
42// the daemonmux handshake together; a successful physical connection detaches
43// from that context and is owned by the pool.
44//
45// The remote side is three hidden helpers, excluded from public help:
46//
47// - `_broker-mux-stdio --offline-root ABS` bridges its own stdio to the
48// single provisioned private Unix daemonmux carriage.
49// - `_broker-mux-quic-bootstrap --offline-root ABS` starts one detached
50// `_broker-mux-quic-proxy`, forwards its single readiness line, and exits.
51// - `_broker-mux-quic-proxy --offline-root ABS` mints one ephemeral QUIC
52// server, admits exactly one authenticated carriage, and bridges it to the
53// same private Unix daemonmux carriage.
54//
55// Each helper also accepts --production instead of --offline-root. That scope
56// derives the private carriage from this user's production runtime. Helpers
57// never start a broker or observer. Only an explicit if-needed authorization
58// permits the shared daemon starter; existing-only observation never spawns.
59// The daemon on the other side owns its identity and incarnation.
60// The one-time QUIC token and nonce are minted per physical connection, travel
61// only inside the SSH channel and the encrypted stream, and never reach an
62// address, a log, or an error. SSH host-key verification, noninteractive
63// authentication, and the absence of a remote PTY are preserved: the helpers
64// only add explicit trust inputs that narrow verification, never disable it.
65
66const (
67 // brokerMuxStdioCommand is the hidden remote SSH stdio mux bridge.
68 brokerMuxStdioCommand = ports.BrokerMuxStdioCommand
69 // brokerMuxQUICBootstrapCommand is the hidden remote QUIC mux bootstrap.
70 brokerMuxQUICBootstrapCommand = ports.BrokerMuxQUICBootstrapCommand
71 // brokerMuxQUICProxyCommand is the hidden detached remote QUIC mux proxy.
72 brokerMuxQUICProxyCommand = "_broker-mux-quic-proxy"
73)
74
75// brokerMuxSetupTimeout bounds each side's bootstrap readiness wait. On the
76// broker side it caps the wait for the local ssh bootstrap child's readiness
77// line together with that child's kill/reap cleanup; on the remote side it caps
78// the bootstrap command's wait for the detached proxy's readiness line. It is an
79// additional backstop only: an earlier caller context deadline always stays
80// authoritative, because context.WithTimeout keeps the sooner deadline, so the
81// independent bound can never extend a shorter caller budget.
82const brokerMuxSetupTimeout = 30 * time.Second
83
84// brokerMuxBootstrapReadyTimeout is the effective broker-side bootstrap
85// readiness budget. Production always uses brokerMuxSetupTimeout; a test
86// narrows it so a stalled helper that never prints readiness is exercised
87// deterministically without waiting out the production bound.
88var brokerMuxBootstrapReadyTimeout = brokerMuxSetupTimeout
89
90// brokerMuxQUICDialTimeout bounds the pinned QUIC dial. The setup context is
91// still authoritative; this narrows the dial itself.
92const brokerMuxQUICDialTimeout = 15 * time.Second
93
94// brokerMuxWaitDelay bounds how long a bootstrap Wait blocks on a captured-
95// stderr copy goroutine after the direct child exits. A remote helper that
96// leaves a grandchild holding the inherited stderr pipe can therefore never pin
97// setup shutdown past this delay; the direct child is still killed and reaped.
98const brokerMuxWaitDelay = 2 * time.Second
99
100// brokerDaemonStartArg is the closed helper flag that carries one daemon-start
101// authorization across the SSH process boundary. It is added by the broker to
102// the provisioned remote argv and parsed strictly by the helper; the value is
103// brokerDaemonStartArgvFlag's exact spelling, so the two sides can never drift.
104const brokerDaemonStartArg = "--daemon-start"
105
106// brokerScopeArg is the scope flag carried by every durable remote route argv
107// (see ports.BrokerRouteForTransport). Stored routes keep it verbatim.
108const brokerScopeArg = ports.BrokerRouteScopeArg
109
110// brokerMuxOptions is the parsed hidden mux helper invocation.
111type brokerMuxOptions struct {
112 // startMode is the daemon-start authorization the broker propagated. An
113 // absent flag is the safe default: the helper observes only what is already
114 // running and never starts a daemon, a broker, or a recursive helper.
115 startMode ports.BrokerDaemonStartMode
116}
117
118// parseBrokerMuxArgs strictly parses `<hidden> [--daemon-start
119// existing-only|if-needed]`. Unknown or duplicate flags, a missing or unknown
120// value, and positionals are refused. An absent --daemon-start defaults to
121// existing-only, and an unrecognized value is never mapped onto a permissive
122// authorization.
123func parseBrokerMuxArgs(name string, kind cmdKind, args []string) (command, error) {
124 options := brokerMuxOptions{startMode: ports.BrokerDaemonExistingOnly}
125 scopeSet, startSet := false, false
126 for index := 0; index < len(args); {
127 switch args[index] {
128 case brokerScopeArg:
129 // The scope flag is part of every durable remote route argv, so it
130 // stays accepted; the helper always serves the per-user scope.
131 if scopeSet {
132 return command{}, usagef("`%s` received duplicate %s", name, brokerScopeArg)
133 }
134 scopeSet = true
135 index++
136 case brokerDaemonStartArg:
137 if startSet {
138 return command{}, usagef("`%s` received duplicate %s", name, brokerDaemonStartArg)
139 }
140 if index+1 >= len(args) {
141 return command{}, usagef("`%s` requires a value", brokerDaemonStartArg)
142 }
143 mode, err := parseBrokerDaemonStartArgvFlag(args[index+1])
144 if err != nil {
145 return command{}, usagef("`%s`: %s", name, err)
146 }
147 options.startMode = mode
148 startSet = true
149 index += 2
150 default:
151 if strings.HasPrefix(args[index], "-") {
152 return command{}, usagef("unknown flag %q for `%s`", args[index], name)
153 }
154 return command{}, usagef("`%s` does not accept positional arguments", name)
155 }
156 }
157 return command{kind: kind, brokerMux: options}, nil
158}
159
160// newBrokerRoutes composes the broker's route authorization policy over the
161// configured local binding. Remote hosts are wired once the registry exists.
162func newBrokerRoutes(config *brokerconfig.Config) *broker.Routes {
163 return &broker.Routes{Local: config.Resolver(), Address: brokerconfig.RouteSpecAddress}
164}
165
166// brokerDynamicMuxConnector dials one authorized target. Remote targets are
167// re-authorized by broker.Routes against committed membership immediately
168// before dialing; the local target must still name the provisioned binding.
169func brokerDynamicMuxConnector(config *brokerconfig.Config, routes *broker.Routes, log *slog.Logger) (*daemonmux.EndpointConnector, error) {
170 return daemonmux.NewEndpointConnector(func(ctx context.Context, target ports.BrokerDialTarget) (daemonmux.RawFramedTransport, error) {
171 if !target.Fence.Local {
172 spec, err := routes.AuthorizeRemoteDial(ctx, target)
173 if err != nil {
174 return nil, err
175 }
176 route, err := brokerconfig.RouteFromSpec(spec)
177 if err != nil {
178 return nil, err
179 }
180 return dialBrokerRoute(ctx, route, target, log)
181 }
182 binding, ok := config.LocalBinding()
183 if !ok || binding.Route.Address() != target.Address || !binding.Policy.Compatible(target.Policy) {
184 return nil, ports.BrokerError{Code: ports.BrokerErrorConflictingPolicy, Cause: errors.New("broker mux: local route authority does not match the resolved target")}
185 }
186 return dialBrokerRoute(ctx, binding.Route, target, log)
187 }, daemonmux.DefaultMuxCeilings())
188}
189
190// dialBrokerRoute dials one provisioned route under one resolved target as an
191// already-authenticated raw framed carriage. Authentication, identity, and
192// policy remain the daemonmux connector's concern; this function selects the
193// transport and forwards the whole target, so a mux stdio/QUIC helper receives
194// the start authorization before it decides anything about the remote daemon.
195//
196// A Unix route is dialed directly by this process. The configuration's own
197// local binding route is the machine daemon the broker may start, so it goes
198// through the broker-owned local daemon start; every other Unix route is a
199// provisioned endpoint carriage that the broker only ever dials.
200func dialBrokerRoute(ctx context.Context, route brokerconfig.Route, target ports.BrokerDialTarget, log *slog.Logger) (daemonmux.RawFramedTransport, error) {
201 if !target.Fence.Local && log != nil {
202 // One line per remote physical bootstrap: warm reuse is verified by
203 // the absence of this event, not by timing.
204 log.Info("broker_remote_dial", "endpoint", target.Fence.Registration.Endpoint, "route", string(route.Kind()))
205 }
206 switch route.Kind() {
207 case brokerconfig.RouteUnix:
208 if target.Fence.Local {
209 // The broker-owned local fence is the machine daemon this process may
210 // start; it goes through the shared daemon start so the authorization
211 // reaches the election. Every other Unix carriage is only dialed.
212 return dialBrokerLocalDaemon(ctx, route.Path(), target)
213 }
214 return ipc.DialMuxContext(ctx, route.Path())
215 case brokerconfig.RouteSSHStdio:
216 spec, err := sshMuxCommandSpec(route, target.StartMode)
217 if err != nil {
218 return nil, err
219 }
220 return sshstdio.DialMuxContext(ctx, spec, log)
221 case brokerconfig.RouteSSHQUIC:
222 return dialBrokerQUICRoute(ctx, route, target.StartMode, log)
223 default:
224 return nil, fmt.Errorf("broker mux: unsupported route kind %q", route.Kind())
225 }
226}
227
228// sshMuxCommandSpec builds the local ssh argv for one provisioned route and one
229// start authorization. It reuses the mux carriage's own command construction
230// (the option terminator, the POSIX-quoted remote argv, and `-T` for no remote
231// PTY), splices in only the explicit trust inputs the operator provisioned, and
232// appends the closed `--daemon-start existing-only|if-needed` word so the remote
233// helper reads the authorization the broker propagated instead of defaulting.
234// It never disables host-key verification, never forces a non-interactive auth
235// bypass, and never requests a PTY.
236func sshMuxCommandSpec(route brokerconfig.Route, mode ports.BrokerDaemonStartMode) (sshstdio.CommandSpec, error) {
237 flag, err := brokerDaemonStartArgvFlag(mode)
238 if err != nil {
239 return sshstdio.CommandSpec{}, err
240 }
241 argv := route.Argv()
242 for _, word := range argv {
243 if word == brokerDaemonStartArg {
244 return sshstdio.CommandSpec{}, fmt.Errorf("broker mux route provisions its own %s", brokerDaemonStartArg)
245 }
246 }
247 argv = append(argv, brokerDaemonStartArg, flag)
248 spec := sshstdio.BuildCommandForMux(route.Target(), argv...)
249 options := sshTrustOptions(route)
250 if len(options) == 0 || len(spec.Args) < 2 {
251 return spec, nil
252 }
253 // BuildCommandForMux returns ["-T", "--", target, remote]; insert options
254 // after `-T` so they stay ssh options and the option terminator still
255 // protects the target.
256 args := make([]string, 0, len(spec.Args)+len(options))
257 args = append(args, spec.Args[0])
258 args = append(args, options...)
259 args = append(args, spec.Args[1:]...)
260 return sshstdio.CommandSpec{Path: spec.Path, Args: args}, nil
261}
262
263// sshTrustOptions renders the provisioned trust inputs as ssh options. A
264// connect timeout only narrows the attempt; a known-hosts file names the
265// verification source. So the option carries a single bounded, whitespace-free
266// value and can never inject a second option.
267func sshTrustOptions(route brokerconfig.Route) []string {
268 var options []string
269 if timeout := route.ConnectTimeout(); timeout > 0 {
270 seconds := int(timeout / time.Second)
271 if seconds < 1 {
272 seconds = 1
273 }
274 options = append(options, "-o", fmt.Sprintf("ConnectTimeout=%d", seconds))
275 }
276 if knownHosts := route.KnownHostsFile(); knownHosts != "" {
277 options = append(options, "-o", "UserKnownHostsFile="+knownHosts)
278 }
279 return options
280}
281
282// dialBrokerQUICRoute runs one SSH-authenticated QUIC bootstrap and returns the
283// pinned, authenticated raw bounded carriage. The local ssh child starts the
284// remote `_broker-mux-quic-bootstrap`, prints exactly one readiness line on
285// stdout, and exits; the detached remote proxy serves one carriage. A refused
286// pin, token, or nonce closes the carriage and hands nothing out. Captured
287// stderr is sanitized and never part of the returned error.
288//
289// The caller's setup context bounds the exchange, and brokerMuxSetupTimeout is
290// an independent backstop: an earlier caller deadline stays authoritative,
291// while a bootstrap that never prints readiness is still killed and reaped
292// within the bound and can never pin pool setup. The captured stderr sink is
293// read only after that kill and reap, and the sink is itself synchronized, so
294// the os/exec copy goroutine can never race the diagnostic.
295func dialBrokerQUICRoute(ctx context.Context, route brokerconfig.Route, mode ports.BrokerDaemonStartMode, log *slog.Logger) (wire.BoundedTransport, error) {
296 spec, err := sshMuxCommandSpec(route, mode)
297 if err != nil {
298 return nil, err
299 }
300 stderr := sshstdio.NewDiagnosticSink(0)
301 cmd := exec.CommandContext(ctx, spec.Path, spec.Args...)
302 cmd.Stderr = stderr
303 // A captured stderr is a pipe with a copy goroutine; WaitDelay ensures a
304 // grandchild that inherits the write end cannot block Wait past the delay
305 // once the direct child is gone.
306 cmd.WaitDelay = brokerMuxWaitDelay
307 stdout, err := cmd.StdoutPipe()
308 if err != nil {
309 return nil, brokerQUICUnavailable("bootstrap stdout", err)
310 }
311 if err := cmd.Start(); err != nil {
312 _ = stdout.Close()
313 return nil, brokerQUICUnavailable("start bootstrap", err)
314 }
315 // reapBootstrap kills the local bootstrap and starts a bounded join, exactly
316 // once. Every error path reaps before it reads the captured stderr sink; the
317 // sink is safe to read whether or not that bounded join has finished, because
318 // it synchronizes its own reads and writes. The remote proxy is detached
319 // in its own session, so killing the local ssh child never ends it.
320 reaped := false
321 reapBootstrap := func() {
322 if reaped {
323 return
324 }
325 reaped = true
326 reapBootstrapProcess(cmd)
327 }
328 defer func() {
329 reapBootstrap()
330 _ = stdout.Close()
331 }()
332
333 // The readiness wait is bounded independently of the caller context, so a
334 // bootstrap that never prints readiness cannot pin pool setup. An earlier
335 // caller deadline still wins: WithTimeout keeps the sooner deadline.
336 readyCtx, cancelReadiness := context.WithTimeout(ctx, brokerMuxBootstrapReadyTimeout)
337 readiness, err := quic.ReadReadiness(readyCtx, stdout)
338 cancelReadiness()
339 if err != nil {
340 reapBootstrap()
341 logBrokerBootstrapStderr(log, stderr)
342 return nil, brokerQUICUnavailable("read bootstrap readiness", classifyBootstrapFailure(stderr, err))
343 }
344 // The bootstrap channel served its purpose: the detached remote proxy owns
345 // the carriage lifetime. Reap the short-lived local child without killing
346 // it, so SIGKILL never races the proxy detach. A bounded I/O drain that the
347 // child's exit outlived (ErrWaitDelay) is not a bootstrap failure.
348 if waitErr := cmd.Wait(); waitErr != nil && !errors.Is(waitErr, exec.ErrWaitDelay) {
349 reaped = true
350 logBrokerBootstrapStderr(log, stderr)
351 return nil, brokerQUICUnavailable("reap bootstrap", classifyBootstrapFailure(stderr, waitErr))
352 }
353 reaped = true
354
355 // Compose the dial address from the configured ssh target host plus the
356 // host-independent readiness port: a remote loopback address is never
357 // dialed. The one-time token and nonce travel only inside the encrypted
358 // stream after the pinned handshake.
359 addr, err := brokerQUICPeerAddr(route.Target(), readiness.Port)
360 if err != nil {
361 return nil, brokerQUICUnavailable("resolve QUIC peer", err)
362 }
363 transport, err := quic.DialMuxContext(ctx, addr, readiness, quic.Config{}, brokerMuxQUICDialTimeout)
364 if err != nil {
365 return nil, brokerQUICUnavailable("dial bootstrap QUIC", err)
366 }
367 return transport, nil
368}
369
370// brokerQUICPeerAddr composes the dial address from the ssh target host and the
371// bootstrap-published port, stripping a user prefix, an explicit port, and
372// bracketed IPv6 literals to the bare host.
373func brokerQUICPeerAddr(target string, port int) (string, error) {
374 if port <= 0 || port > 65535 {
375 return "", fmt.Errorf("invalid bootstrap port %d", port)
376 }
377 host := target
378 if at := strings.LastIndexByte(host, '@'); at >= 0 {
379 host = host[at+1:]
380 }
381 if h, _, err := net.SplitHostPort(host); err == nil {
382 host = h
383 }
384 host = strings.TrimPrefix(strings.TrimSuffix(host, "]"), "[")
385 if host == "" {
386 return "", errors.New("empty bootstrap host")
387 }
388 return net.JoinHostPort(host, strconv.Itoa(port)), nil
389}
390
391// logBrokerBootstrapStderr records a bounded, sanitized bootstrap diagnostic.
392// The readiness secret is never on stderr, and no raw remote text is surfaced.
393// The join performed by the caller is bounded, so the os/exec stderr copy
394// goroutine may still be writing when this reads the sink; the diagnostic is
395// nonetheless consistent because DiagnosticSink synchronizes its own reads and
396// writes and returns a bounded, sanitized snapshot rather than relying on the
397// join having quiesced the sink.
398func logBrokerBootstrapStderr(log *slog.Logger, sink *sshstdio.DiagnosticSink) {
399 if log == nil || sink == nil {
400 return
401 }
402 diagnostic := strings.TrimSpace(sink.String())
403 if diagnostic == "" {
404 return
405 }
406 log.Warn("broker_mux_bootstrap_stderr", "diagnostic", diagnostic)
407}
408
409// classifyBootstrapFailure wraps err in a typed remote failure when the
410// captured ssh stderr names a recognizable cause (authentication, host key,
411// timeout), so the picker explains the failure. The raw text stays in the log.
412func classifyBootstrapFailure(sink *sshstdio.DiagnosticSink, err error) error {
413 if sink == nil {
414 return err
415 }
416 if kind := sshstdio.ClassifyStderr(sink.String()); kind != domain.RemoteFailureNone {
417 return fmt.Errorf("%w: %w", domain.RemoteFailure{Kind: kind}, err)
418 }
419 return err
420}
421
422// reapBootstrapProcess kills one bootstrap child and starts a bounded join of
423// its wait. The join is bounded by brokerMuxSetupTimeout so a helper that
424// leaves a grandchild holding the captured stderr pipe cannot pin cleanup, and
425// exec.Cmd.WaitDelay narrows the wait much further once the direct child is
426// gone. When the bound expires the wait goroutine may still be draining the
427// captured-stderr copy, so a caller that reads the sink afterwards relies on
428// the sink's own synchronization, not on the join having finished. The direct
429// child is always killed, so no bootstrap process is ever leaked.
430func reapBootstrapProcess(cmd *exec.Cmd) {
431 if cmd == nil || cmd.Process == nil {
432 return
433 }
434 _ = cmd.Process.Kill()
435 done := make(chan struct{})
436 go func() {
437 defer close(done)
438 _ = cmd.Wait()
439 }()
440 timer := time.NewTimer(brokerMuxSetupTimeout)
441 defer timer.Stop()
442 select {
443 case <-done:
444 case <-timer.C:
445 }
446}
447
448// brokerQUICUnavailable wraps one bootstrap or pinned-dial failure. The
449// readiness record is deliberately absent: a returned error never carries the
450// one-time token, the nonce, or remote stderr.
451func brokerQUICUnavailable(action string, err error) error {
452 if err == nil {
453 err = errors.New("unavailable")
454 }
455 return fmt.Errorf("broker mux QUIC unavailable: %s: %w", action, err)
456}
457
458// runBrokerMuxStdioCommand bridges the process' own stdio to the single
459// provisioned private Unix daemonmux carriage, under the daemon-start
460// authorization the broker propagated.
461func runBrokerMuxStdioCommand(ctx context.Context, options brokerMuxOptions) error {
462 raw, err := dialBrokerMuxHelper(ctx, options)
463 if err != nil {
464 if daemonCarriageAbsence(err) {
465 return &exitCoded{code: sshstdio.MuxExitNoDaemon, err: err}
466 }
467 return fmt.Errorf("broker mux stdio: dial daemonmux carriage: %w", err)
468 }
469 bounded, ok := raw.(wire.BoundedTransport)
470 if !ok {
471 _ = raw.Close()
472 return errors.New("broker mux stdio: carriage is not a bounded transport")
473 }
474 return bridgeMuxTransports(ctx, sshstdio.NewStdioTransport(), bounded)
475}
476
477// runBrokerMuxQUICBootstrapCommand starts one detached `_broker-mux-quic-proxy`,
478// forwards its single readiness line on this process' stdout, and exits. The
479// proxy owns the carriage lifetime in its own session and survives this short-
480// lived bootstrap, exactly like the ordinary QUIC bootstrap.
481func runBrokerMuxQUICBootstrapCommand(ctx context.Context, options brokerMuxOptions) error {
482 exe, err := os.Executable()
483 if err != nil {
484 return err
485 }
486 reader, writer, err := os.Pipe()
487 if err != nil {
488 return err
489 }
490 defer func() { _ = reader.Close() }()
491 devNull, err := os.OpenFile(os.DevNull, os.O_RDWR, 0)
492 if err != nil {
493 _ = writer.Close()
494 return err
495 }
496 defer func() { _ = devNull.Close() }()
497
498 args := []string{brokerMuxQUICProxyCommand, brokerScopeArg}
499 flag, err := brokerDaemonStartArgvFlag(options.startMode)
500 if err != nil {
501 _ = writer.Close()
502 return err
503 }
504 args = append(args, brokerDaemonStartArg, flag)
505 cmd := exec.CommandContext(ctx, exe, args...)
506 // The proxy writes its readiness through the pipe and its diagnostics
507 // through stderr; stdio is otherwise detached so the SSH bootstrap channel
508 // can close while the detached proxy owns the carriage.
509 cmd.Stdin = devNull
510 cmd.Stdout = writer
511 cmd.Stderr = devNull
512 cmd.SysProcAttr = &syscall.SysProcAttr{Setsid: true}
513 if err := cmd.Start(); err != nil {
514 _ = writer.Close()
515 return err
516 }
517 _ = writer.Close()
518
519 readyCtx, cancel := context.WithTimeout(ctx, brokerMuxSetupTimeout)
520 defer cancel()
521 readiness, err := quic.ReadReadiness(readyCtx, reader)
522 if err != nil {
523 _ = cmd.Process.Kill()
524 _ = cmd.Wait()
525 return fmt.Errorf("broker mux bootstrap readiness: %w", err)
526 }
527 line, err := quic.EncodeReadiness(readiness)
528 if err != nil {
529 _ = cmd.Process.Kill()
530 _ = cmd.Wait()
531 return err
532 }
533 if _, err := fmt.Fprintf(os.Stdout, "%s\n", line); err != nil {
534 _ = cmd.Process.Kill()
535 _ = cmd.Wait()
536 return err
537 }
538 // Release the detached proxy: it is in its own session and outlives this
539 // bootstrap. Reaping it here would block until the carriage ends.
540 return cmd.Process.Release()
541}
542
543// runBrokerMuxQUICProxyCommand mints one ephemeral QUIC server, admits exactly
544// one authenticated carriage, and bridges it to the single provisioned private
545// Unix daemonmux carriage. It exits after its carriage ends, on credential
546// expiry, or on setup failure, so it can never outlive its advertised
547// credential or serve a second physical connection.
548func runBrokerMuxQUICProxyCommand(ctx context.Context, options brokerMuxOptions) error {
549 server, readiness, err := quic.NewServer()
550 if err != nil {
551 return err
552 }
553 defer func() { _ = server.Close() }()
554 line, err := quic.EncodeReadiness(readiness)
555 if err != nil {
556 return err
557 }
558 if _, err := fmt.Fprintf(os.Stdout, "%s\n", line); err != nil {
559 return err
560 }
561 transport, err := server.AcceptMux(ctx)
562 if err != nil {
563 return err
564 }
565 // The proxy owns its process: QUIC's orderly Close only FINs the stream and
566 // defers the connection close, so returning (and exiting) right away would
567 // drop both and leave the broker's preamble read parked until the QUIC idle
568 // timeout. Waiting for the bounded teardown delivers the FIN, so a refused
569 // existing-only dial (no daemon running) fails the broker's attempt at once.
570 defer func() {
571 _ = transport.Close()
572 _ = waitGracefulTeardown(transport)
573 }()
574 raw, err := dialBrokerMuxHelper(ctx, options)
575 if err != nil {
576 if daemonCarriageAbsence(err) {
577 return &exitCoded{code: sshstdio.MuxExitNoDaemon, err: err}
578 }
579 return fmt.Errorf("broker mux quic proxy: dial daemonmux carriage: %w", err)
580 }
581 bounded, ok := raw.(wire.BoundedTransport)
582 if !ok {
583 _ = raw.Close()
584 return errors.New("broker mux quic proxy: carriage is not a bounded transport")
585 }
586 return bridgeMuxTransports(ctx, transport, bounded)
587}
588
589// dialBrokerMuxHelper is a remote-side broker carriage helper, not a client
590// daemon dial. Production authority is derived locally, never from SSH argv.
591func dialBrokerMuxHelper(ctx context.Context, options brokerMuxOptions) (daemonmux.RawFramedTransport, error) {
592 route, err := brokerconfig.RouteFromSpec(ports.BrokerRouteSpec{Kind: ports.BrokerRouteUnix, Path: daemonmux.SocketPath(ipc.SocketDir())})
593 if err != nil {
594 return nil, err
595 }
596 target := ports.BrokerDialTarget{Fence: ports.BrokerEndpointFence{Local: true}, Policy: localDaemonPolicy(), Address: route.Address(), StartMode: options.startMode}
597 if err := target.Validate(); err != nil {
598 return nil, err
599 }
600 return dialBrokerLocalDaemon(ctx, route.Path(), target)
601}
602
603// bridgeMuxTransports copies raw frames both ways until either carriage ends or
604// ctx is done, then closes both. An orderly end of either direction (EOF) is a
605// normal exit for a bridge owner that merely forwards bytes.
606func bridgeMuxTransports(ctx context.Context, a, b wire.BoundedTransport) error {
607 ctx, cancel := context.WithCancel(ctx)
608 defer cancel()
609 defer func() { _ = a.Close() }()
610 defer func() { _ = b.Close() }()
611 errCh := make(chan error, 2)
612 go func() { errCh <- relayMux(ctx, a, b) }()
613 go func() { errCh <- relayMux(ctx, b, a) }()
614 select {
615 case err := <-errCh:
616 return orderlyRelay(err)
617 case <-ctx.Done():
618 return ctx.Err()
619 }
620}
621
622// relayMux copies one direction of raw bounded frames.
623func relayMux(ctx context.Context, src, dst wire.BoundedTransport) error {
624 for {
625 envelope, err := src.RecvBounded(wire.AbsoluteEnvelopeLimit)
626 if err != nil {
627 return err
628 }
629 if err := dst.Send(envelope); err != nil {
630 return err
631 }
632 if err := ctx.Err(); err != nil {
633 return err
634 }
635 }
636}
637
638// orderlyRelay treats EOF as a clean bridge end and passes every other outcome
639// through unchanged.
640func orderlyRelay(err error) error {
641 if err == nil || errors.Is(err, io.EOF) {
642 return nil
643 }
644 return err
645}
646
647// gracefulTeardownWaiter is implemented by carriages (QUIC) whose Close defers
648// a bounded connection close past its return. The broker mux QUIC proxy owns its
649// process, so it must wait for that close before returning: exiting first
650// would discard the final synchronous envelope Close already handed to the
651// stream.
652type gracefulTeardownWaiter interface {
653 WaitGracefulTeardown(ctx context.Context) error
654}
655
656// proxyTeardownTimeout bounds the proxy's wait for a deferred graceful close,
657// leaving headroom beyond the adapter's own graceful window. A carriage that
658// does not defer its close is not waited on.
659const proxyTeardownTimeout = 2 * time.Second
660
661func waitGracefulTeardown(transport wire.Transport) error {
662 waiter, ok := transport.(gracefulTeardownWaiter)
663 if !ok {
664 return nil
665 }
666 ctx, cancel := context.WithTimeout(context.Background(), proxyTeardownTimeout)
667 defer cancel()
668 return waiter.WaitGracefulTeardown(ctx)
669}