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