1package broker
2
3import (
4 "context"
5 "crypto/rand"
6 "encoding/binary"
7 "encoding/json"
8 "errors"
9 "fmt"
10 "sync/atomic"
11
12 "github.com/bnema/vev/internal/domain"
13 "github.com/bnema/vev/internal/ports"
14 "github.com/bnema/vev/internal/protocol"
15 "github.com/bnema/vev/internal/protocol/catalogue"
16)
17
18// Daemon observation.
19//
20// RemoteProbe and LocalDaemonProbe observe one daemon over the broker-owned
21// multiplexed carriage. Both open an observation stream, never an attachment,
22// and ask the daemon for its own catalogue through the typed command protocol.
23// Authenticated identity and incarnation come from the physical handshake,
24// never from a configured hostname. Both share observeDaemonCatalogue, so the
25// two producers cannot drift apart, and both borrow the pool's transport
26// before dialing their own, so neither becomes a second transport owner.
27
28// pooledPhysicals lets a probe borrow the broker pool's physical transport
29// instead of dialing (and, for a remote, bootstrapping SSH/QUIC) again. The
30// pool is published after composition because it binds identities through the
31// registry that runs the probes. A probe dials only when nothing is pooled, or
32// when the borrowed transport retired under it.
33type pooledPhysicals struct{ pool atomic.Pointer[Pool] }
34
35func (s *pooledPhysicals) adopt(physical ports.BrokerPhysicalConnection, policy ports.BrokerPolicy) bool {
36 pool := s.pool.Load()
37 return pool != nil && pool.AdoptPhysical(physical, policy)
38}
39
40// observe reports handled=false when the probe must dial its own transport.
41func (s *pooledPhysicals) observe(ctx context.Context, identity ports.BrokerDaemonIdentity, policy ports.BrokerPolicy, request ports.BrokerOpenStreamRequest, codec ports.SessionCodec) (ports.BrokerDaemonObservation, bool, error) {
42 pool := s.pool.Load()
43 if pool == nil || identity == "" {
44 return ports.BrokerDaemonObservation{}, false, nil
45 }
46 physical, ok := pool.SharedPhysical(identity, policy)
47 if !ok {
48 return ports.BrokerDaemonObservation{}, false, nil
49 }
50 observation, err := observeDaemonCatalogue(ctx, physical, request, codec)
51 if err == nil || ctx.Err() != nil {
52 return observation, true, err
53 }
54 select {
55 case <-physical.Done():
56 return ports.BrokerDaemonObservation{}, false, nil
57 default:
58 return observation, true, err
59 }
60}
61
62// RemoteProbe observes one configured remote daemon. Routes, Binder, and Hosts
63// may be assigned after the registry that runs the probe is built, but before
64// the registry starts; composition is cyclic because the registry is both the
65// host authority and the probe's caller.
66type RemoteProbe struct {
67 Epoch ports.BrokerEpoch
68 Routes ports.BrokerRouteAuthority
69 Binder ports.BrokerIdentityBinder
70 Connector ports.BrokerEndpointConnector
71 Hosts ports.BrokerHostAuthorityReader
72 Codec ports.SessionCodec
73
74 shared pooledPhysicals
75}
76
77var _ ports.BrokerHostProbe = (*RemoteProbe)(nil)
78
79// SharePool lets later probes borrow the pool's attached or warm transports.
80func (p *RemoteProbe) SharePool(pool *Pool) { p.shared.pool.Store(pool) }
81
82func (p *RemoteProbe) Probe(ctx context.Context, registration domain.RemoteRegistration) (ports.BrokerDaemonObservation, error) {
83 if p == nil || p.Epoch == 0 || p.Routes == nil || p.Binder == nil || p.Connector == nil {
84 return ports.BrokerDaemonObservation{}, errors.New("incomplete broker remote probe")
85 }
86 request, err := p.request(ctx, registration)
87 if err != nil {
88 return ports.BrokerDaemonObservation{}, err
89 }
90 endpoint, err := p.Routes.ResolveDialTarget(ctx, request)
91 if err != nil {
92 return ports.BrokerDaemonObservation{}, err
93 }
94 // Borrow the pooled transport of an attached or warm daemon instead of
95 // bootstrapping SSH/QUIC again. A borrowed transport that retires mid-probe
96 // falls back to a fresh dial.
97 if endpoint.ExpectedIdentity.Bound {
98 if observation, handled, err := p.shared.observe(ctx, endpoint.ExpectedIdentity.Identity, endpoint.Policy, request, p.Codec); handled {
99 return observation, err
100 }
101 }
102 physical, err := p.Connector.Connect(ctx, endpoint)
103 if err != nil {
104 var brokerErr ports.BrokerError
105 if errors.As(err, &brokerErr) && brokerErr.Code == ports.BrokerErrorNoDaemon {
106 // This typed dial-time result means the SSH helper ran on the remote
107 // machine but its local daemon carriage was absent. Do not infer this
108 // from a later stream loss on an already authenticated connection.
109 return ports.BrokerDaemonObservation{Availability: domain.RemoteAvailabilityNoDaemon}, nil
110 }
111 return ports.BrokerDaemonObservation{}, err
112 }
113 adopted := false
114 defer func() {
115 if !adopted {
116 _ = physical.Close()
117 }
118 }()
119 if (endpoint.ExpectedIdentity.Bound && physical.Identity() != endpoint.ExpectedIdentity.Identity) || physical.Policy() != endpoint.Policy {
120 return ports.BrokerDaemonObservation{}, errors.New("remote probe authenticated binding mismatch")
121 }
122 identity, err := p.Binder.BindAuthenticatedIdentity(ctx, ports.BrokerIdentityBindingRequest{Fence: endpoint.Fence, Policy: endpoint.Policy, Identity: physical.Identity()})
123 if err != nil || identity != physical.Identity() {
124 return ports.BrokerDaemonObservation{}, errors.New("remote probe could not commit authenticated binding")
125 }
126 observation, err := observeDaemonCatalogue(ctx, physical, request, p.Codec)
127 if err == nil && ctx.Err() == nil {
128 adopted = p.shared.adopt(physical, endpoint.Policy)
129 }
130 return observation, err
131}
132
133func (p *RemoteProbe) request(ctx context.Context, registration domain.RemoteRegistration) (ports.BrokerOpenStreamRequest, error) {
134 if p.Hosts == nil {
135 return ports.BrokerOpenStreamRequest{}, errors.New("remote probe has no policy authority")
136 }
137 record, ok, err := p.Hosts.LookupHost(ctx, registration.Endpoint)
138 if err != nil {
139 return ports.BrokerOpenStreamRequest{}, err
140 }
141 if !ok || !record.Registration.Equal(registration) {
142 return ports.BrokerOpenStreamRequest{}, ports.BrokerError{Code: ports.BrokerErrorUnavailable, Text: "unknown or stale probe registration"}
143 }
144 request, err := newObservationRequest(p.Epoch, record.Policy)
145 if err != nil {
146 return ports.BrokerOpenStreamRequest{}, err
147 }
148 request.Endpoint = registration.Endpoint
149 request.Registration = registration
150 return request, nil
151}
152
153// LocalIdentityLoader returns the local daemon's persisted identity, or ""
154// when no daemon has ever published one.
155type LocalIdentityLoader func() (ports.BrokerDaemonIdentity, error)
156
157// LocalDaemonProbe observes the broker's own machine daemon over one local
158// address. It commits no durable identity binding: the local daemon owns no
159// configured membership, and the identity the physical preamble verified is
160// the fence.
161type LocalDaemonProbe struct {
162 epoch ports.BrokerEpoch
163 address string
164 policy ports.BrokerPolicy
165 loadIdentity LocalIdentityLoader
166 connector ports.BrokerEndpointConnector
167 codec ports.SessionCodec
168 shared pooledPhysicals
169}
170
171var _ LocalProbe = (*LocalDaemonProbe)(nil)
172
173// NewLocalDaemonProbe validates one local observation route. The caller must
174// have proven the address is a local socket this process dials directly:
175// probing must never spawn a process.
176func NewLocalDaemonProbe(epoch ports.BrokerEpoch, address string, policy ports.BrokerPolicy, connector ports.BrokerEndpointConnector, codec ports.SessionCodec, loadIdentity LocalIdentityLoader) (*LocalDaemonProbe, error) {
177 if epoch == 0 {
178 return nil, errors.New("broker local probe requires a broker epoch")
179 }
180 if nilDependency(connector) {
181 return nil, errors.New("broker local probe requires a connector")
182 }
183 if nilDependency(codec) {
184 return nil, errors.New("broker local probe requires a codec")
185 }
186 if loadIdentity == nil {
187 return nil, errors.New("broker local probe requires an identity loader")
188 }
189 if address == "" {
190 return nil, errors.New("broker local probe requires an address")
191 }
192 if err := policy.Validate(); err != nil {
193 return nil, fmt.Errorf("broker local probe: %w", err)
194 }
195 return &LocalDaemonProbe{epoch: epoch, address: address, policy: policy, loadIdentity: loadIdentity, connector: connector, codec: codec}, nil
196}
197
198// SharePool lets later probes borrow the pool's attached or warm transports.
199func (p *LocalDaemonProbe) SharePool(pool *Pool) { p.shared.pool.Store(pool) }
200
201// ProbeLocal authenticates the daemonmux physical preamble and then opens
202// exactly one observation stream that asks the daemon for its own catalogue.
203// Capabilities are not carried by the catalogue command, so they stay zero.
204// Every failure reports an explicit availability with zero observed identity:
205// a probe never manufactures identity, version, or inventory it did not
206// observe, and it never creates an attachment, a session, or a process.
207func (p *LocalDaemonProbe) ProbeLocal(ctx context.Context) (ports.BrokerDaemonObservation, error) {
208 unreachable := ports.BrokerDaemonObservation{Availability: domain.RemoteAvailabilityUnreachable}
209 if p == nil || p.connector == nil {
210 return unreachable, errors.New("broker local probe is nil")
211 }
212 identity, err := p.loadIdentity()
213 if err != nil {
214 return unreachable, err
215 }
216 if identity == "" {
217 return ports.BrokerDaemonObservation{Availability: domain.RemoteAvailabilityUnknown}, nil
218 }
219 endpoint := ports.BrokerDialTarget{
220 Fence: ports.BrokerEndpointFence{Local: true}, Policy: p.policy, Address: p.address,
221 StartMode: ports.BrokerDaemonExistingOnly,
222 ExpectedIdentity: ports.BrokerExpectedIdentity{Identity: identity, Bound: true},
223 }
224 if err := endpoint.Validate(); err != nil {
225 return unreachable, fmt.Errorf("broker local probe: %w", err)
226 }
227 request, err := p.request()
228 if err != nil {
229 return unreachable, err
230 }
231 if observation, handled, err := p.shared.observe(ctx, identity, p.policy, request, p.codec); handled {
232 if err != nil {
233 return unreachable, err
234 }
235 return observation, nil
236 }
237 physical, err := p.connector.Connect(ctx, endpoint)
238 if err != nil {
239 return unreachable, err
240 }
241 defer func() { _ = physical.Close() }()
242 // The endpoint stays authoritative: re-verify the accepted authority even
243 // though the connector already checked it, so a future connector change can
244 // never weaken what the probe publishes.
245 if physical.Identity() != endpoint.ExpectedIdentity.Identity || !physical.Policy().Compatible(endpoint.Policy) {
246 return unreachable, errors.New("broker local probe authenticated binding mismatch")
247 }
248 observation, err := observeDaemonCatalogue(ctx, physical, request, p.codec)
249 if err != nil {
250 return unreachable, err
251 }
252 return observation, nil
253}
254
255// request builds the one local observation open this probe performs. It
256// carries no endpoint, registration, admission, name, target, or environment.
257func (p *LocalDaemonProbe) request() (ports.BrokerOpenStreamRequest, error) {
258 request, err := newObservationRequest(p.epoch, p.policy)
259 if err != nil {
260 return ports.BrokerOpenStreamRequest{}, err
261 }
262 request.Local = true
263 return request, nil
264}
265
266// newObservationRequest builds an observation open with fresh connection and
267// stream identities, so a replayed or duplicated open is refused by the peer's
268// anti-replay window. Observation never starts the target: it always carries
269// the existing-only authorization, and a start-if-needed one is refused by
270// validation rather than silently widened into a spawn.
271func newObservationRequest(epoch ports.BrokerEpoch, policy ports.BrokerPolicy) (ports.BrokerOpenStreamRequest, error) {
272 connection, err := randomConnectionID()
273 if err != nil {
274 return ports.BrokerOpenStreamRequest{}, err
275 }
276 stream, err := randomNonzeroUint64()
277 if err != nil {
278 return ports.BrokerOpenStreamRequest{}, err
279 }
280 return ports.BrokerOpenStreamRequest{
281 Epoch: epoch, Purpose: ports.BrokerStreamObservation,
282 Connection: connection, Stream: ports.BrokerStreamID(stream),
283 Policy: policy, StartMode: ports.BrokerDaemonExistingOnly,
284 }, nil
285}
286
287// observeDaemonCatalogue observes one already authenticated daemon by opening
288// exactly one observation logical stream and asking the daemon for its own
289// catalogue. It sends no Hello, claims no geometry, requests no admission, and
290// creates no session or process.
291//
292// A refused open, an unmatched or invalid result, a failed outcome, a malformed
293// catalogue, and a catalogue outside the exact current schema all fail closed:
294// the caller receives an error and no observation it could publish.
295func observeDaemonCatalogue(ctx context.Context, physical ports.BrokerPhysicalConnection, request ports.BrokerOpenStreamRequest, codec ports.SessionCodec) (ports.BrokerDaemonObservation, error) {
296 if physical == nil || codec == nil || request.Purpose != ports.BrokerStreamObservation {
297 return ports.BrokerDaemonObservation{}, errors.New("invalid daemon observation request")
298 }
299 logical, err := physical.OpenStream(ctx, request)
300 if err != nil {
301 // A connection that completed the authenticated physical handshake had
302 // a daemon. Losing its logical stream is a transport failure, not proof
303 // that the daemon is absent; preserve the failure so pooled callers can
304 // discard a dead connection and retry with a fresh dial.
305 return ports.BrokerDaemonObservation{}, err
306 }
307 if logical == nil {
308 return ports.BrokerDaemonObservation{}, errors.New("daemon observation opened no logical stream")
309 }
310 defer func() { _ = logical.Close() }()
311 commandID, err := randomNonzeroUint64()
312 if err != nil {
313 return ports.BrokerDaemonObservation{}, err
314 }
315 reply, err := exchangeObservationCommand(ctx, codec.Client(logical), logical, protocol.CommandRequest{RequestID: commandID, Version: protocol.Version, Slug: "remote-catalog", JSON: true})
316 if err != nil {
317 return ports.BrokerDaemonObservation{}, err
318 }
319 result, ok := reply.(protocol.CommandResult)
320 if !ok || result.RequestID != commandID || !result.Valid() {
321 return ports.BrokerDaemonObservation{}, errors.New("invalid daemon observation result")
322 }
323 if result.Outcome != protocol.CommandSucceeded {
324 return ports.BrokerDaemonObservation{}, fmt.Errorf("daemon observation failed: %s", result.Text)
325 }
326 var catalog catalogue.RemoteCatalog
327 if err := json.Unmarshal([]byte(result.Output), &catalog); err != nil {
328 return ports.BrokerDaemonObservation{}, fmt.Errorf("decode daemon observation: %w", err)
329 }
330 if err := catalogue.ValidateRemoteCatalog(catalog); err != nil {
331 return ports.BrokerDaemonObservation{}, err
332 }
333 return ports.BrokerDaemonObservation{
334 Identity: physical.Identity(),
335 Incarnation: physical.Incarnation(),
336 ProtocolVersion: catalog.ProtocolVersion,
337 Availability: domain.RemoteAvailabilityReachable,
338 InventoryKnown: true,
339 Sessions: catalog.Sessions,
340 }, nil
341}
342
343// exchangeObservationCommand sends one typed command and waits for one reply
344// under the caller's context. A cancelled or expired context closes exactly
345// this logical stream, which unblocks the pending read, and the worker is
346// joined before returning, so no observation goroutine outlives its attempt.
347func exchangeObservationCommand(ctx context.Context, typed ports.ClientConnection, logical ports.BrokerEnvelopeStream, command protocol.CommandRequest) (protocol.ServerMessage, error) {
348 if ctx == nil {
349 ctx = context.Background()
350 }
351 type outcome struct {
352 message protocol.ServerMessage
353 err error
354 }
355 done := make(chan outcome, 1)
356 go func() {
357 if err := typed.SendClient(command); err != nil {
358 done <- outcome{err: err}
359 return
360 }
361 message, err := typed.ReceiveServer()
362 done <- outcome{message: message, err: err}
363 }()
364 select {
365 case result := <-done:
366 if err := ctx.Err(); err != nil {
367 return nil, err
368 }
369 return result.message, result.err
370 case <-ctx.Done():
371 _ = logical.Close()
372 <-done
373 return nil, ctx.Err()
374 }
375}
376
377func randomConnectionID() (ports.BrokerConnectionID, error) {
378 var id ports.BrokerConnectionID
379 if _, err := rand.Read(id[:]); err != nil {
380 return id, err
381 }
382 if err := id.Validate(); err != nil {
383 return randomConnectionID()
384 }
385 return id, nil
386}
387
388func randomNonzeroUint64() (uint64, error) {
389 var raw [8]byte
390 if _, err := rand.Read(raw[:]); err != nil {
391 return 0, err
392 }
393 value := binary.BigEndian.Uint64(raw[:])
394 if value == 0 {
395 return randomNonzeroUint64()
396 }
397 return value, nil
398}