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