1// Package sessionwire adapts typed session messages to Protobuf envelopes.
2package sessionwire
3
4import (
5 "context"
6 "errors"
7 "sync"
8 "time"
9
10 "google.golang.org/protobuf/encoding/protowire"
11 "google.golang.org/protobuf/proto"
12
13 "github.com/bnema/vev/internal/ports"
14 "github.com/bnema/vev/internal/protocol"
15 "github.com/bnema/vev/internal/protocol/wire"
16)
17
18var (
19 ErrWrongDirection = errors.New("sessionwire: wrong message direction")
20 ErrInvalidMessage = errors.New("sessionwire: invalid message")
21 ErrUnsupportedSend = errors.New("sessionwire: send mode is unsupported")
22)
23
24type serverConnection struct {
25 raw wire.Transport
26 ceilings protoCeilings
27 // ceilingsMu guards ceilings. The lazy preamble publishes the negotiated
28 // values from whichever goroutine first uses the connection, while another
29 // goroutine may already be asking for capabilities, so the once-guarded write
30 // is not by itself a happens-before edge for that reader.
31 ceilingsMu sync.Mutex
32
33 preambleOnce sync.Once
34 preambleErr error
35 deadline time.Time
36
37 // admission is the provisioned admission metadata the accepting side
38 // stamped, or the zero value with hasAdmission=false for a connection built
39 // by a legacy constructor that carried no admission. It is never mutated
40 // after construction and is exposed only through SessionAdmission, which
41 // returns a deep copy.
42 admission ports.SessionAdmission
43 hasAdmission bool
44
45 // preambleDone closes exactly once when the handshake has run, whether it
46 // succeeded or failed. It backs the handshake completion seam a mux listener
47 // exposes to the daemon; the absolute deadline is never restarted.
48 preambleDone chan struct{}
49
50 hooksMu sync.Mutex
51 handshakeDone bool
52 hooksRun bool
53 hooks []func()
54}
55
56var (
57 _ ports.ServerConnection = (*serverConnection)(nil)
58 _ ports.SessionAdmissionProvider = (*serverConnection)(nil)
59)
60
61// NewServerConnection wraps one raw connection incarnation exactly once. The
62// server preamble runs on the first ReceiveClient, bounded by the handshake
63// deadline started here; the daemon's transport watcher closes the link on
64// timeout, which unblocks the preamble read. Prefer
65// NewServerConnectionWithDeadline when the caller already owns the absolute
66// deadline, as the daemon-side mux listener does at Open admission. The
67// returned connection carries no admission metadata.
68func NewServerConnection(raw wire.Transport) ports.ServerConnection {
69 return NewServerConnectionWithDeadline(raw, time.Now().Add(protocol.HandshakeTimeout))
70}
71
72// NewServerConnectionWithDeadline wraps one raw connection incarnation exactly
73// once with an absolute local handshake deadline its caller already fixed. A
74// daemon-side mux listener passes the deadline it accepted at Open admission,
75// so a stream delayed in an accept queue spends that one budget instead of
76// starting a second one. The preamble still runs lazily on the first
77// ReceiveClient; a deadline already in the past fails it immediately. A zero
78// deadline falls back to a fresh protocol.HandshakeTimeout from now. The
79// returned connection carries no admission metadata.
80func NewServerConnectionWithDeadline(raw wire.Transport, deadline time.Time) ports.ServerConnection {
81 return newServerConnection(raw, deadline, ports.SessionAdmission{}, false)
82}
83
84// NewServerConnectionWithAdmission wraps one raw connection incarnation exactly
85// once with an absolute local handshake deadline and the provisioned admission
86// metadata its accepting side stamped (the exact accepted policy and origin,
87// plus the peer's declared purpose, admission variant, name, target, and
88// bounded environment). The admission is stored as a deep copy and exposed only
89// through SessionAdmission, which returns a fresh copy so a consumer can never
90// mutate admitted authority. An entirely zero admission is reported as ok=false,
91// as absent, so a caller cannot accidentally
92// publish an unvalidated admission.
93func NewServerConnectionWithAdmission(raw wire.Transport, deadline time.Time, admission ports.SessionAdmission) ports.ServerConnection {
94 return newServerConnection(raw, deadline, admission, !admissionAbsent(admission))
95}
96
97// admissionAbsent reports whether an admission is entirely unset. It is the
98// presence test for NewServerConnectionWithAdmission: an admission that names no
99// origin, policy, purpose, or payload is indistinguishable from absent metadata and is reported as absent rather than trusted.
100func admissionAbsent(admission ports.SessionAdmission) bool {
101 return admission.Origin == 0 && admission.Policy == (ports.BrokerPolicy{}) &&
102 admission.Purpose == 0 && admission.Admission == 0 && admission.Name == "" &&
103 admission.Target == (protocol.ExactSessionTarget{}) && len(admission.Env) == 0
104}
105
106// newServerConnection is the shared constructor: it clones the admission and
107// records whether it is present, so every exported constructor funnels through
108// one place that fixes the deadline and allocates the completion signal.
109func newServerConnection(raw wire.Transport, deadline time.Time, admission ports.SessionAdmission, hasAdmission bool) ports.ServerConnection {
110 if raw == nil {
111 return nil
112 }
113 if deadline.IsZero() {
114 deadline = time.Now().Add(protocol.HandshakeTimeout)
115 }
116 return &serverConnection{
117 raw: raw,
118 ceilings: defaultProtoCeilings(),
119 deadline: deadline,
120 admission: admission.Clone(),
121 hasAdmission: hasAdmission,
122 preambleDone: make(chan struct{}),
123 }
124}
125
126// SessionAdmission returns a deep copy of the provisioned admission metadata
127// stamped on this connection, implementing ports.SessionAdmissionProvider. It
128// reports ok=false for a connection built by a constructor that carried no
129// admission, so a consumer distinguishes an absent admission from an admitted
130// one instead of trusting a zero value.
131func (c *serverConnection) SessionAdmission() (ports.SessionAdmission, bool) {
132 if c == nil || !c.hasAdmission {
133 return ports.SessionAdmission{}, false
134 }
135 return c.admission.Clone(), true
136}
137
138// HandshakeDeadline returns the absolute local deadline of the session
139// handshake this connection accepted. It is fixed for the connection's
140// lifetime and is never restarted: the preamble, Hello/Welcome, and the first
141// committed publication all share it.
142func (c *serverConnection) HandshakeDeadline() time.Time { return c.deadline }
143
144// HandshakeDone returns a channel that is closed exactly once when the
145// handshake has run, whether it succeeded or failed.
146func (c *serverConnection) HandshakeDone() <-chan struct{} { return c.preambleDone }
147
148// OnHandshakeComplete registers fn to run exactly once when the handshake has
149// run. Registering after completion runs fn immediately; a nil fn is ignored.
150func (c *serverConnection) OnHandshakeComplete(fn func()) {
151 if fn == nil {
152 return
153 }
154 c.hooksMu.Lock()
155 if c.handshakeDone {
156 c.hooksMu.Unlock()
157 fn()
158 return
159 }
160 c.hooks = append(c.hooks, fn)
161 c.hooksMu.Unlock()
162}
163
164// finishPreamble publishes handshake completion from inside the preamble's
165// sync.Once: it closes the done channel and marks the handshake complete so a
166// later registration runs immediately. It deliberately runs no registered hook,
167// because a hook that calls back into the connection would re-enter sync.Once
168// and deadlock; runHandshakeHooks runs the hooks after the Once body returns.
169// It tolerates a zero-value connection that never allocated the done channel
170// (test literals and seam-only construction), so closing is nil-safe.
171func (c *serverConnection) finishPreamble() {
172 if c.preambleDone != nil {
173 close(c.preambleDone)
174 }
175 c.hooksMu.Lock()
176 c.handshakeDone = true
177 c.hooksMu.Unlock()
178}
179
180// runHandshakeHooks runs the completion hooks registered before the handshake
181// finished exactly once, after the preamble's sync.Once body has returned.
182// Running them outside the Once is what lets a hook call back into the
183// connection - for example to receive a message - without re-entering
184// sync.Once and deadlocking. Every ensurePreamble caller invokes it; only the
185// first drains the hooks, and a registrant that arrives after completion
186// observes the completed flag and runs itself.
187func (c *serverConnection) runHandshakeHooks() {
188 c.hooksMu.Lock()
189 if c.hooksRun {
190 c.hooksMu.Unlock()
191 return
192 }
193 c.hooksRun = true
194 hooks := c.hooks
195 c.hooks = nil
196 c.hooksMu.Unlock()
197 for _, fn := range hooks {
198 fn()
199 }
200}
201
202func (c *serverConnection) ensurePreamble() error {
203 c.preambleOnce.Do(func() {
204 ctx, cancel := context.WithDeadline(context.Background(), c.deadline)
205 defer cancel()
206 next, err := runProtoServerPreamble(ctx, c.raw, c.limits())
207 c.publishLimits(next)
208 c.preambleErr = err
209 if err != nil {
210 _ = c.raw.Close()
211 }
212 c.finishPreamble()
213 })
214 c.runHandshakeHooks()
215 return c.preambleErr
216}
217
218// limits returns the negotiated ceilings under the lock the preamble publishes
219// them through. Send paths may read them without the lock once ensurePreamble
220// has returned, because that call orders the publish before the read.
221func (c *serverConnection) limits() protoCeilings {
222 c.ceilingsMu.Lock()
223 defer c.ceilingsMu.Unlock()
224 return c.ceilings
225}
226
227// publishLimits records the ceilings one preamble negotiation produced.
228func (c *serverConnection) publishLimits(next protoCeilings) {
229 c.ceilingsMu.Lock()
230 c.ceilings = next
231 c.ceilingsMu.Unlock()
232}
233
234func (c *serverConnection) ReceiveClient() (protocol.ClientMessage, error) {
235 if err := c.ensurePreamble(); err != nil {
236 return nil, c.preambleDecodeFailure(err)
237 }
238 envelope, err := c.raw.Recv()
239 if err != nil {
240 return nil, err
241 }
242 message, decodeErr := decodeClientEnvelopeWithin(envelope.Payload, c.ceilings.maxReceiveEnvelopeBytes)
243 if decodeErr == nil {
244 return message, nil
245 }
246 return nil, decodeErr
247}
248
249// preambleDecodeFailure maps a preamble refusal to the typed compatibility
250// responses the daemon sends before any mutation. Only Hello carries a
251// version the daemon can answer; other first-frame kinds are handled after
252// a successful preamble by the normal decode path.
253func (c *serverConnection) preambleDecodeFailure(err error) *protocol.DecodeFailure {
254 return &protocol.DecodeFailure{
255 Category: protocol.DecodeMalformed,
256 Kind: protocol.DecodeMessageHello,
257 Version: uint16(protocol.Version),
258 Err: err,
259 }
260}
261
262func (c *serverConnection) SendServer(message protocol.ServerMessage) error {
263 envelope, err := encodeProtoServer(message)
264 if err != nil {
265 return err
266 }
267 return c.sendEnvelope(envelope)
268}
269
270func (c *serverConnection) SendServerAsync(message protocol.ServerMessage) error {
271 transport, ok := c.raw.(wire.AsyncTransport)
272 if !ok {
273 return ErrUnsupportedSend
274 }
275 envelope, err := encodeProtoServer(message)
276 if err != nil {
277 return err
278 }
279 raw, err := sendEnvelopeWithin(envelope, c.ceilings)
280 if err != nil {
281 return err
282 }
283 return transport.SendAsync(raw)
284}
285
286func (c *serverConnection) SendServerSynchronous(message protocol.ServerMessage) error {
287 transport, ok := c.raw.(wire.OwnedSynchronousTransport)
288 if !ok {
289 return ErrUnsupportedSend
290 }
291 envelope, err := encodeProtoServer(message)
292 if err != nil {
293 return err
294 }
295 raw, err := sendEnvelopeWithin(envelope, c.ceilings)
296 if err != nil {
297 return err
298 }
299 return transport.SendSynchronous(raw)
300}
301
302func (c *serverConnection) SendOutput(output protocol.Output) error {
303 envelope, err := encodeProtoServer(output)
304 if err != nil {
305 return err
306 }
307 return c.sendEnvelope(envelope)
308}
309
310func (c *serverConnection) SendOutputAsync(output protocol.Output) error {
311 transport, ok := c.raw.(wire.AsyncTransport)
312 if !ok {
313 return ErrUnsupportedSend
314 }
315 envelope, err := encodeProtoServer(output)
316 if err != nil {
317 return err
318 }
319 raw, err := sendEnvelopeWithin(envelope, c.ceilings)
320 if err != nil {
321 return err
322 }
323 return transport.SendAsync(raw)
324}
325
326func (c *serverConnection) SendOutputSynchronous(output protocol.Output) error {
327 transport, ok := c.raw.(wire.OwnedSynchronousTransport)
328 if !ok {
329 return ErrUnsupportedSend
330 }
331 envelope, err := encodeProtoServer(output)
332 if err != nil {
333 return err
334 }
335 raw, err := sendEnvelopeWithin(envelope, c.ceilings)
336 if err != nil {
337 return err
338 }
339 return transport.SendSynchronous(raw)
340}
341
342func (c *serverConnection) sendEnvelope(envelope *wire.ServerEnvelope) error {
343 raw, err := sendEnvelopeWithin(envelope, c.ceilings)
344 if err != nil {
345 return err
346 }
347 return c.raw.Send(raw)
348}
349
350func marshalServerEnvelope(envelope *wire.ServerEnvelope) (wire.Envelope, error) {
351 raw, err := (proto.Marshal(envelope))
352 if err != nil {
353 return wire.Envelope{}, err
354 }
355 return wire.Envelope{Payload: raw}, nil
356}
357
358// DecodeClientEnvelope unwraps one scanned client envelope payload into its
359// semantic message. App composition uses it for hidden one-shot carriages
360// (remote preview) that validate a request before dialing the daemon.
361func DecodeClientEnvelope(payload []byte) (protocol.ClientMessage, error) {
362 return decodeClientEnvelope(payload)
363}
364
365// EncodePreambleResponseForTest marshals the canned server preamble
366// acceptance exchanged before application traffic. Tests own their
367// preamble fixtures; production always negotiates through the handshake.
368func EncodePreambleResponseForTest() ([]byte, error) {
369 return proto.Marshal(preambleResponseToWire(true, defaultProtoCeilings(), 0))
370}
371
372// EncodePreambleRequestForTest marshals the canned client preamble request
373// exchanged before application traffic.
374func EncodePreambleRequestForTest() ([]byte, error) {
375 return proto.Marshal(preambleRequestToWire(defaultProtoCeilings()))
376}
377
378// DecodePreambleAcceptanceForTest reports whether a raw preamble response
379// payload is a server acceptance.
380func DecodePreambleAcceptanceForTest(payload []byte) bool {
381 var response wire.PreambleResponse
382 if serr := wire.ScanEnvelope(&response, payload); serr != nil {
383 return false
384 }
385 if uerr := (proto.UnmarshalOptions{DiscardUnknown: false}.Unmarshal(payload, &response)); uerr != nil {
386 return false
387 }
388 _, err := checkPreambleResponse(&response)
389 return err == nil
390}
391
392// EncodeClientMessage wraps one semantic client message in its serialized
393// directional envelope for hidden one-shot carriages.
394func EncodeClientMessage(message protocol.ClientMessage) ([]byte, error) {
395 envelope, err := encodeProtoClient(message)
396 if err != nil {
397 return nil, err
398 }
399 return proto.Marshal(envelope)
400}
401
402// EncodeServerMessage wraps one semantic server message in its serialized
403// directional envelope for hidden one-shot carriages.
404func EncodeServerMessage(message protocol.ServerMessage) ([]byte, error) {
405 envelope, err := encodeProtoServer(message)
406 if err != nil {
407 return nil, err
408 }
409 return proto.Marshal(envelope)
410}
411
412func decodeClientEnvelope(payload []byte) (protocol.ClientMessage, error) {
413 return decodeClientEnvelopeWithin(payload, wire.AbsoluteEnvelopeLimit)
414}
415
416// decodeClientEnvelopeWithin unwraps one client envelope under the
417// negotiated receive envelope ceiling. The one-shot exported decoder uses
418// the absolute ceiling because no preamble was negotiated there.
419func decodeClientEnvelopeWithin(payload []byte, envelopeLimit uint64) (protocol.ClientMessage, error) {
420 envelope := &wire.ClientEnvelope{}
421 if err := checkCategoryCeiling(payload, clientEnvelopeCategory(payload), envelopeLimit); err != nil {
422 return nil, &protocol.DecodeFailure{Category: protocol.DecodeMalformed, Err: err, Kind: clientVariantKind(payload)}
423 }
424 if err := wire.ScanEnvelope(envelope, payload); err != nil {
425 failure := &protocol.DecodeFailure{Category: protocol.DecodeMalformed, Err: err}
426 if errors.Is(err, wire.ErrScanUnknown) {
427 failure.Category = protocol.DecodeUnknownType
428 }
429 // Scan-level malformation (truncation, trailing data) still
430 // carries the envelope variant in its leading tag. Recovering
431 // the kind preserves first-frame compatibility routing: the
432 // daemon answers malformed Command/Kill/Hello frames with typed
433 // refusals instead of a generic hello error.
434 failure.Kind = clientVariantKind(payload)
435 return nil, failure
436 }
437 if err := (proto.UnmarshalOptions{DiscardUnknown: false}.Unmarshal(payload, envelope)); err != nil {
438 return nil, &protocol.DecodeFailure{Category: protocol.DecodeMalformed, Err: err}
439 }
440 message, err := decodeProtoClient(envelope)
441 if err != nil {
442 return nil, clientFailureFor(envelope, err)
443 }
444 return message, nil
445}
446
447// clientVariantKind recovers the envelope variant from the leading tag of
448// a payload that failed strict scanning. Only the kinds with typed
449// first-frame refusals (Hello/Command/Kill/RemotePreview/Navigation
450// Inventory) are reported; anything else stays Unknown so the daemon
451// answers with its generic hello error.
452// ClientVariantKindForTest exposes scan-failure kind recovery to
453// integration probes. Production uses it through decodeClientEnvelope.
454func ClientVariantKindForTest(payload []byte) protocol.DecodeMessageKind {
455 return clientVariantKind(payload)
456}
457
458func clientVariantKind(payload []byte) protocol.DecodeMessageKind {
459 number, wireType, length := protowire.ConsumeTag(payload)
460 if wireType != protowire.BytesType {
461 return protocol.DecodeMessageUnknown
462 }
463 // ConsumeTag reports errCodeFieldNumber (-1..-4) for bad tags;
464 // only accept a positively consumed tag.
465 if length <= 0 {
466 return protocol.DecodeMessageUnknown
467 }
468 switch number {
469 case 1:
470 return protocol.DecodeMessageHello
471 case 7:
472 return protocol.DecodeMessageKill
473 case 12:
474 return protocol.DecodeMessageCommand
475 case 32:
476 return protocol.DecodeMessageRemotePreview
477 case 21:
478 return protocol.DecodeMessageNavigationInventory
479 default:
480 return protocol.DecodeMessageUnknown
481 }
482}
483
484// clientFailureFor preserves the daemon's first-frame compatibility routing:
485// Hello/Command/Kill/RemotePreview/NavigationInventory/PickerControl kinds
486// select the typed refusal a malformed first frame receives.
487func clientFailureFor(envelope *wire.ClientEnvelope, err error) *protocol.DecodeFailure {
488 failure := &protocol.DecodeFailure{Category: protocol.DecodeMalformed, Err: err}
489 if errors.Is(err, ErrWrongDirection) {
490 failure.Category = protocol.DecodeWrongDirection
491 return failure
492 }
493 switch envelope.Payload.(type) {
494 case *wire.ClientEnvelope_Hello:
495 failure.Kind = protocol.DecodeMessageHello
496 if payload, ok := envelope.Payload.(*wire.ClientEnvelope_Hello); ok && payload.Hello != nil {
497 failure.Version = uint16(payload.Hello.GetVersion())
498 }
499 case *wire.ClientEnvelope_CommandRequest:
500 failure.Kind = protocol.DecodeMessageCommand
501 if payload, ok := envelope.Payload.(*wire.ClientEnvelope_CommandRequest); ok && payload.CommandRequest != nil {
502 failure.Version = uint16(payload.CommandRequest.GetVersion())
503 failure.RequestID = payload.CommandRequest.GetRequestId()
504 failure.HasRequestID = true
505 }
506 case *wire.ClientEnvelope_Kill:
507 failure.Kind = protocol.DecodeMessageKill
508 case *wire.ClientEnvelope_RemotePreviewWatch:
509 failure.Kind = protocol.DecodeMessageRemotePreview
510 case *wire.ClientEnvelope_NavigationInventoryRequest:
511 failure.Kind = protocol.DecodeMessageNavigationInventory
512 if payload, ok := envelope.Payload.(*wire.ClientEnvelope_NavigationInventoryRequest); ok && payload.NavigationInventoryRequest != nil {
513 failure.Version = uint16(payload.NavigationInventoryRequest.GetVersion())
514 failure.RequestID = payload.NavigationInventoryRequest.GetRequestId()
515 failure.HasRequestID = true
516 }
517 }
518 return failure
519}
520
521func (c *serverConnection) Capabilities() protocol.ConnectionCapabilities {
522 return rawCapabilities(c.raw, c.limits().outputDataLimit)
523}
524
525func (c *serverConnection) LinkState() ports.LinkState { return rawLinkState(c.raw) }
526func (c *serverConnection) LinkEvents() <-chan ports.LinkEvent { return rawLinkEvents(c.raw) }
527
528func rawCapabilities(raw wire.Transport, outputDataLimit uint64) protocol.ConnectionCapabilities {
529 _, datagram := raw.(wire.DatagramTransport)
530 _, async := raw.(wire.AsyncTransport)
531 _, synchronous := raw.(wire.OwnedSynchronousTransport)
532 _, linkState := raw.(ports.LinkStateReporter)
533 window := uint8(protocol.MaxOutputWindow)
534 if datagram {
535 window = 1
536 }
537 return protocol.ConnectionCapabilities{
538 OutputDataLimit: int(outputDataLimit),
539 PreferredOutputWindow: window,
540 AsyncSend: async,
541 OwnedSynchronousSend: synchronous,
542 LinkState: linkState,
543 }
544}
545
546func rawLinkState(raw wire.Transport) ports.LinkState {
547 if reporter, ok := raw.(ports.LinkStateReporter); ok {
548 return reporter.LinkState()
549 }
550 return ports.LinkStateConnected
551}
552
553func rawLinkEvents(raw wire.Transport) <-chan ports.LinkEvent {
554 if reporter, ok := raw.(ports.LinkStateReporter); ok {
555 return reporter.LinkEvents()
556 }
557 return nil
558}
559
560func (c *serverConnection) Close() error { return c.raw.Close() }
561
562type serverListener struct{ raw wire.Listener }
563
564var _ ports.ServerListener = (*serverListener)(nil)
565
566// NewServerListener wraps every accepted raw connection in a stable typed adapter.
567func NewServerListener(raw wire.Listener) ports.ServerListener {
568 if raw == nil {
569 return nil
570 }
571 return &serverListener{raw: raw}
572}
573
574func (l *serverListener) Accept() (ports.ServerConnection, error) {
575 raw, err := l.raw.Accept()
576 if err != nil {
577 return nil, err
578 }
579 return NewServerConnection(raw), nil
580}
581
582func (l *serverListener) Close() error { return l.raw.Close() }
583func (l *serverListener) Addr() string { return l.raw.Addr() }