1// Broker endpoint connector and physical connection.
2//
3// EndpointConnector is the broker-side bridge between the transport-neutral
4// ports.BrokerEndpointConnector contract and the daemonmux multiplexer. It owns
5// the setup half of one pooled physical connection: dial one already
6// authenticated raw framed carriage for the resolved endpoint, run the
7// daemonmux physical preamble over the FramedCarrierBridge, negotiate the
8// effective ceilings, and hand the live pump and logical connector to a
9// PhysicalConnection.
10//
11// The dial function is injected, not chosen here: the connector never knows
12// whether the carriage is IPC, QUIC, SSH stdio, or a test pipe, and it never
13// authenticates it. The carriage must already be authenticated before this
14// connector sees it; the resolved endpoint's identity and policy are the
15// authoritative values the handshake verifies the daemon's response against.
16//
17// One caller-supplied setup context (or its deadline) bounds dial and handshake
18// as a single unit: the same context goes to the dial function and to
19// RunClientHandshake, and no timer is created here. A successful connection is
20// deliberately detached from that setup context: the pump is started on a
21// context that preserves the setup values but not its cancellation, so a caller
22// that cancels or lets its setup deadline elapse after Connect returned never
23// tears down a pooled physical connection. The connection's lifetime is owned
24// by PhysicalConnection.Close.
25//
26// PhysicalConnection is the broker-side ports.BrokerPhysicalConnection over
27// that live pump. Its Identity, Incarnation, and Policy are immutable values
28// captured from the verified handshake; Done, Err, and FailureKind delegate to
29// the pump's publish-once terminal authority, so a physical loss closes Done
30// before every logical stream is terminalized; OpenStream exact-checks the
31// request policy against the accepted physical policy and then defers to the
32// LogicalConnector; Close tears the pump (and therefore the carrier) down
33// exactly once.
34package daemonmux
35
36import (
37 "context"
38 "errors"
39 "fmt"
40 "os/exec"
41 "sync"
42
43 "github.com/bnema/vev/internal/adapters/sshstdio"
44 "github.com/bnema/vev/internal/domain"
45 "github.com/bnema/vev/internal/ports"
46)
47
48// ErrConnectorConfig reports an endpoint connector constructed without a dial
49// function or with an invalid ceiling advertisement, or a Connect call with an
50// invalid resolved endpoint. It is a caller bug, not a peer refusal.
51var ErrConnectorConfig = errors.New("daemonmux: invalid endpoint connector configuration")
52
53// RawCarrierDialer establishes one already authenticated raw framed carriage
54// for one resolved dial target. It returns a carriage the daemonmux
55// FramedCarrierBridge can adapt; it returns promptly once ctx is done. The
56// carriage's authentication and transport selection live entirely outside this
57// package, so the connector carries no dial policy of its own.
58//
59// The whole target travels, not only its opaque address: the resolved start
60// authorization, policy, and endpoint fence are exactly what a carriage owner
61// needs to decide whether the transport may start the target daemon. They are
62// passed as one value so a dial can never observe an address that belongs to a
63// different authorization than the one the connector validated and will verify
64// in the physical preamble.
65type RawCarrierDialer func(ctx context.Context, target ports.BrokerDialTarget) (RawFramedTransport, error)
66
67// EndpointConnector implements ports.BrokerEndpointConnector over daemonmux. It
68// owns no carriage state itself: every Connect produces one independent
69// PhysicalConnection that owns its own pump, carrier, and logical connector.
70// It is safe for concurrent use.
71type EndpointConnector struct {
72 dial RawCarrierDialer
73 ceilings MuxCeilings
74}
75
76var _ ports.BrokerEndpointConnector = (*EndpointConnector)(nil)
77
78// NewEndpointConnector returns a connector that dials carriage through dial and
79// advertises ceilings in the daemonmux physical preamble. A nil dial or an
80// invalid ceiling advertisement is refused with ErrConnectorConfig; the
81// ceilings are validated and copied, so a later change to the caller's value
82// never changes what a connection negotiates.
83func NewEndpointConnector(dial RawCarrierDialer, ceilings MuxCeilings) (*EndpointConnector, error) {
84 if dial == nil {
85 return nil, ErrConnectorConfig
86 }
87 if err := ceilings.Validate(); err != nil {
88 return nil, fmt.Errorf("%w: %w", ErrConnectorConfig, err)
89 }
90 return &EndpointConnector{dial: dial, ceilings: ceilings}, nil
91}
92
93// Connect dials the resolved endpoint, completes the daemonmux physical
94// preamble against the endpoint's authoritative identity and policy, and
95// returns a live ports.BrokerPhysicalConnection. The resolved target is handed
96// to the dial function whole, after validation, so the carriage owner sees the
97// exact start authorization and fence the connector is about to verify. The
98// caller's ctx (or its deadline) bounds dial and handshake together; every
99// failure closes the carriage, so no half-open physical connection survives. A
100// successful connection is detached from ctx: cancelling or expiring ctx
101// afterwards never stops the pooled physical connection, which Close alone
102// owns.
103func (c *EndpointConnector) Connect(ctx context.Context, endpoint ports.BrokerDialTarget) (ports.BrokerPhysicalConnection, error) {
104 if c == nil || c.dial == nil {
105 return nil, ErrConnectorConfig
106 }
107 if ctx == nil {
108 ctx = context.Background()
109 }
110 if err := endpoint.Validate(); err != nil {
111 return nil, ports.BrokerError{Code: ports.BrokerErrorIncompatible, Cause: err}
112 }
113
114 raw, err := c.dial(ctx, endpoint)
115 if err != nil {
116 if raw != nil {
117 _ = raw.Close()
118 }
119 return nil, dialError(err)
120 }
121 if raw == nil {
122 return nil, ports.BrokerError{Code: ports.BrokerErrorIncompatible, Cause: ErrCarrierState}
123 }
124
125 bridge, err := NewPreambleCarrier(raw)
126 if err != nil {
127 _ = raw.Close()
128 return nil, ports.BrokerError{Code: ports.BrokerErrorIncompatible, Cause: err}
129 }
130
131 result, err := RunClientHandshake(ctx, bridge, endpoint, c.ceilings)
132 if err != nil {
133 // The handshake already closed a refused carriage; Close is idempotent.
134 _ = bridge.Close()
135 return nil, handshakeError(err)
136 }
137 // The endpoint stays authoritative: re-verify the accepted authority even
138 // though the handshake already checked it, so a future handshake change can
139 // never weaken what the connector publishes.
140 if (endpoint.ExpectedIdentity.Bound && result.Identity != endpoint.ExpectedIdentity.Identity) || !result.Policy.Compatible(endpoint.Policy) || result.Incarnation.Validate() != nil {
141 _ = bridge.Close()
142 return nil, ports.BrokerError{Code: ports.BrokerErrorIncompatible}
143 }
144 if err := bridge.Negotiate(result.Ceilings); err != nil {
145 _ = bridge.Close()
146 return nil, ports.BrokerError{Code: ports.BrokerErrorIncompatible, Cause: err}
147 }
148
149 pump, err := NewPump(bridge, DirectionServer, result.Ceilings)
150 if err != nil {
151 _ = bridge.Close()
152 return nil, ports.BrokerError{Code: ports.BrokerErrorIncompatible, Cause: err}
153 }
154 logical, err := NewLogicalConnector(pump)
155 if err != nil {
156 _ = pump.Close()
157 return nil, ports.BrokerError{Code: ports.BrokerErrorIncompatible, Cause: err}
158 }
159
160 physical := &PhysicalConnection{
161 pump: pump,
162 logical: logical,
163 identity: result.Identity,
164 incarnation: result.Incarnation,
165 policy: result.Policy,
166 }
167 // The setup context ends here. WithoutCancel keeps the setup values while
168 // removing its cancellation and deadline, so Close - not the caller's setup
169 // context - ends the pooled connection.
170 pump.Start(context.WithoutCancel(ctx))
171 return physical, nil
172}
173
174// dialError classifies one dial failure: cancellation and deadline stay
175// recognizable for the pool's precedence, a validated broker error passes
176// through, and anything else becomes an unavailable adapter failure with its
177// cause retained locally.
178func dialError(err error) error {
179 if err == nil {
180 return nil
181 }
182 if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
183 return err
184 }
185 if missingRemoteDaemon(err) {
186 return ports.BrokerError{Code: ports.BrokerErrorNoDaemon, Cause: err}
187 }
188 var typed ports.BrokerError
189 if errors.As(err, &typed) && typed.Validate() == nil {
190 return err
191 }
192 return ports.BrokerError{Code: ports.BrokerErrorUnavailable, Cause: err}
193}
194
195// missingRemoteDaemon only accepts the reserved helper exit after SSH has
196// started; SSH's own failures (including exit 255) remain unavailable.
197func missingRemoteDaemon(err error) bool {
198 if !errors.Is(err, sshstdio.ErrMuxSSHExit) {
199 return false
200 }
201 var exit *exec.ExitError
202 return errors.As(err, &exit) && exit.ExitCode() == sshstdio.MuxExitNoDaemon
203}
204
205// handshakeError classifies one handshake failure: cancellation and deadline
206// stay recognizable, a peer refusal is a policy conflict, a local authority
207// mismatch is an incompatibility, and anything else is unavailable.
208func handshakeError(err error) error {
209 if err == nil {
210 return nil
211 }
212 if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
213 return err
214 }
215 if missingRemoteDaemon(err) {
216 return ports.BrokerError{Code: ports.BrokerErrorNoDaemon, Cause: err}
217 }
218 var refusal *HandshakeRefusal
219 if errors.As(err, &refusal) {
220 return ports.BrokerError{Code: ports.BrokerErrorConflictingPolicy, Cause: err}
221 }
222 if errors.Is(err, ErrHandshakeRejected) {
223 return ports.BrokerError{Code: ports.BrokerErrorIncompatible, Cause: err}
224 }
225 if errors.Is(err, ErrHandshakeConfig) {
226 return ports.BrokerError{Code: ports.BrokerErrorIncompatible, Cause: err}
227 }
228 return ports.BrokerError{Code: ports.BrokerErrorUnavailable, Cause: err}
229}
230
231// PhysicalConnection is one pooled physical daemonmux connection: the live
232// pump, its logical connector, and the immutable daemon authority the physical
233// preamble verified. It implements ports.BrokerPhysicalConnection and is safe
234// for concurrent use.
235type PhysicalConnection struct {
236 pump *Pump
237 logical *LogicalConnector
238
239 identity ports.BrokerDaemonIdentity
240 incarnation ports.BrokerDaemonIncarnation
241 policy ports.BrokerPolicy
242
243 closeOnce sync.Once
244 closeErr error
245}
246
247var _ ports.BrokerPhysicalConnection = (*PhysicalConnection)(nil)
248
249// Done returns the pump's terminal channel: closed exactly once, before every
250// logical stream of the connection is terminalized.
251func (c *PhysicalConnection) Done() <-chan struct{} {
252 if c == nil || c.pump == nil {
253 return nil
254 }
255 return c.pump.Done()
256}
257
258// Err returns the pump's terminal cause: nil while open and for an orderly
259// local Close.
260func (c *PhysicalConnection) Err() error {
261 if c == nil || c.pump == nil {
262 return nil
263 }
264 return c.pump.Err()
265}
266
267// FailureKind returns the pump's terminal classification: RemoteFailureNone
268// while open and for an orderly local Close.
269func (c *PhysicalConnection) FailureKind() domain.RemoteFailureKind {
270 if c == nil || c.pump == nil {
271 return domain.RemoteFailureNone
272 }
273 return c.pump.FailureKind()
274}
275
276// Identity returns the authenticated daemon identity the handshake verified
277// against the resolved endpoint. It is immutable for the connection's lifetime.
278func (c *PhysicalConnection) Identity() ports.BrokerDaemonIdentity {
279 if c == nil {
280 return ""
281 }
282 return c.identity
283}
284
285// Incarnation returns the daemon process incarnation captured at handshake: a
286// stable identity across restarts carries a fresh, nonzero incarnation.
287func (c *PhysicalConnection) Incarnation() ports.BrokerDaemonIncarnation {
288 if c == nil {
289 return ports.BrokerDaemonIncarnation{}
290 }
291 return c.incarnation
292}
293
294// Policy returns the exact accepted connection policy this physical connection
295// was pooled under. It is immutable for the connection's lifetime.
296func (c *PhysicalConnection) Policy() ports.BrokerPolicy {
297 if c == nil {
298 return ports.BrokerPolicy{}
299 }
300 return c.policy
301}
302
303// OpenStream exact-checks one open request against the accepted physical
304// policy and then opens one independently cancellable logical stream through
305// the LogicalConnector. A policy that differs in any field is refused as a
306// conflicting policy without touching the carriage, so a physical connection
307// can never multiplex a stream under a policy it did not negotiate.
308func (c *PhysicalConnection) OpenStream(ctx context.Context, request ports.BrokerOpenStreamRequest) (ports.BrokerEnvelopeStream, error) {
309 if c == nil || c.logical == nil {
310 return nil, ports.BrokerError{Code: ports.BrokerErrorIncompatible, Cause: ErrLogicalConfig}
311 }
312 if !request.Policy.Compatible(c.policy) {
313 return nil, ports.BrokerError{Code: ports.BrokerErrorConflictingPolicy}
314 }
315 // Never hand back a typed nil: Open returns a concrete *LogicalConnection, so
316 // returning it directly on the error path would wrap a nil pointer in a
317 // non-nil interface, and a caller that checks the interface against nil would
318 // then call a method on it.
319 connection, err := c.logical.Open(ctx, request)
320 if err != nil {
321 return nil, err
322 }
323 if connection == nil {
324 return nil, ports.BrokerError{Code: ports.BrokerErrorIncompatible, Cause: ports.BrokerAdmissionInvalid}
325 }
326 return connection, nil
327}
328
329// Close tears the physical connection down exactly once: it closes the pump,
330// which cancels both carriage goroutines and closes the carrier through the
331// adapter prompt-close contract. Every caller observes the same carrier close
332// error. Close is concurrent-safe and prompt.
333func (c *PhysicalConnection) Close() error {
334 if c == nil {
335 return nil
336 }
337 c.closeOnce.Do(func() {
338 if c.pump != nil {
339 c.closeErr = c.pump.Close()
340 }
341 })
342 return c.closeErr
343}