1package daemonmux
2
3// Directional envelope adaptation.
4//
5// EncodeClient wraps one adapter-local client value in its directional
6// MuxClientEnvelope; DecodeClient unwraps and validates. EncodeServer and
7// DecodeServer do the same for the daemon direction. Every convert path uses
8// an explicit taxonomy switch: unknown variants and wrong-direction payloads
9// fail with ErrWrongDirection before any mutation. Decode paths run strict
10// wire.ScanEnvelope before generated unmarshal, then validate stream
11// identity, UTF-8/control/bidi text, bounds, environment/policy, purpose,
12// chunks, and error details.
13//
14// The codecs are stateless: no stream table, admission decision, scheduler,
15// queue, pump, socket, or framing engine lives here. Envelope sizes ride
16// under the caller's negotiated ceiling, every application envelope is
17// additionally capped at the fixed 1 MiB control-class bound
18// (MaxMuxApplicationBytes), and opaque stream frames ride under the
19// negotiated chunk ceiling. Data, Close, and Reset are shared payloads: the
20// same committed value converts into either direction, and direction is fixed
21// by the closed envelope oneof.
22
23import (
24 "google.golang.org/protobuf/proto"
25
26 "github.com/bnema/vev/internal/domain"
27 "github.com/bnema/vev/internal/ports"
28 "github.com/bnema/vev/internal/protocol"
29 "github.com/bnema/vev/internal/protocol/wire"
30)
31
32// EncodeClient wraps one broker-to-daemon message in its serialized
33// directional envelope. The payload must fit maxEnvelopeBytes and the fixed
34// 1 MiB application ceiling; stream chunks must fit maxChunkBytes.
35func EncodeClient(message ClientMessage, maxEnvelopeBytes, maxChunkBytes uint64) ([]byte, error) {
36 envelope, err := encodeClientEnvelope(message, maxChunkBytes)
37 if err != nil {
38 return nil, err
39 }
40 raw, err := proto.Marshal(envelope)
41 if err != nil {
42 return nil, err
43 }
44 if err := checkMuxEnvelopeCeiling(raw, maxEnvelopeBytes); err != nil {
45 return nil, err
46 }
47 return raw, nil
48}
49
50// DecodeClient unwraps one serialized broker-to-daemon envelope into its
51// adapter-local value. Strict scanning runs before generated unmarshal;
52// wrong-direction payloads fail with ErrWrongDirection.
53func DecodeClient(payload []byte, maxEnvelopeBytes, maxChunkBytes uint64) (ClientMessage, error) {
54 if err := checkMuxEnvelopeCeiling(payload, maxEnvelopeBytes); err != nil {
55 return nil, err
56 }
57 envelope := &wire.MuxClientEnvelope{}
58 if err := wire.ScanEnvelope(envelope, payload); err != nil {
59 return nil, err
60 }
61 if err := (proto.UnmarshalOptions{DiscardUnknown: false}.Unmarshal(payload, envelope)); err != nil {
62 return nil, err
63 }
64 return decodeClientEnvelope(envelope, maxChunkBytes)
65}
66
67// EncodeServer wraps one daemon-to-broker message in its serialized
68// directional envelope under the same envelope and chunk bounds.
69func EncodeServer(message ServerMessage, maxEnvelopeBytes, maxChunkBytes uint64) ([]byte, error) {
70 envelope, err := encodeServerEnvelope(message, maxChunkBytes)
71 if err != nil {
72 return nil, err
73 }
74 raw, err := proto.Marshal(envelope)
75 if err != nil {
76 return nil, err
77 }
78 if err := checkMuxEnvelopeCeiling(raw, maxEnvelopeBytes); err != nil {
79 return nil, err
80 }
81 return raw, nil
82}
83
84// DecodeServer unwraps one serialized daemon-to-broker envelope into its
85// adapter-local value.
86func DecodeServer(payload []byte, maxEnvelopeBytes, maxChunkBytes uint64) (ServerMessage, error) {
87 if err := checkMuxEnvelopeCeiling(payload, maxEnvelopeBytes); err != nil {
88 return nil, err
89 }
90 envelope := &wire.MuxServerEnvelope{}
91 if err := wire.ScanEnvelope(envelope, payload); err != nil {
92 return nil, err
93 }
94 if err := (proto.UnmarshalOptions{DiscardUnknown: false}.Unmarshal(payload, envelope)); err != nil {
95 return nil, err
96 }
97 return decodeServerEnvelope(envelope, maxChunkBytes)
98}
99
100func encodeClientEnvelope(message ClientMessage, maxChunkBytes uint64) (*wire.MuxClientEnvelope, error) {
101 if message == nil {
102 return nil, ErrInvalidMessage
103 }
104 switch m := message.(type) {
105 case Open:
106 converted, err := openToWire(m)
107 if err != nil {
108 return nil, err
109 }
110 return &wire.MuxClientEnvelope{Payload: &wire.MuxClientEnvelope_Open{Open: converted}}, nil
111 case *Open:
112 if m == nil {
113 return nil, ErrInvalidMessage
114 }
115 return encodeClientEnvelope(*m, maxChunkBytes)
116 case Data:
117 converted, err := dataToWire(m, maxChunkBytes)
118 if err != nil {
119 return nil, err
120 }
121 return &wire.MuxClientEnvelope{Payload: &wire.MuxClientEnvelope_Data{Data: converted}}, nil
122 case *Data:
123 if m == nil {
124 return nil, ErrInvalidMessage
125 }
126 return encodeClientEnvelope(*m, maxChunkBytes)
127 case Close:
128 converted, err := closeToWire(m)
129 if err != nil {
130 return nil, err
131 }
132 return &wire.MuxClientEnvelope{Payload: &wire.MuxClientEnvelope_Close{Close: converted}}, nil
133 case *Close:
134 if m == nil {
135 return nil, ErrInvalidMessage
136 }
137 return encodeClientEnvelope(*m, maxChunkBytes)
138 case Reset:
139 converted, err := resetToWire(m)
140 if err != nil {
141 return nil, err
142 }
143 return &wire.MuxClientEnvelope{Payload: &wire.MuxClientEnvelope_Reset_{Reset_: converted}}, nil
144 case *Reset:
145 if m == nil {
146 return nil, ErrInvalidMessage
147 }
148 return encodeClientEnvelope(*m, maxChunkBytes)
149 case WindowUpdate:
150 converted, err := windowUpdateToWire(m)
151 if err != nil {
152 return nil, err
153 }
154 return &wire.MuxClientEnvelope{Payload: &wire.MuxClientEnvelope_WindowUpdate{WindowUpdate: converted}}, nil
155 case *WindowUpdate:
156 if m == nil {
157 return nil, ErrInvalidMessage
158 }
159 return encodeClientEnvelope(*m, maxChunkBytes)
160 default:
161 return nil, ErrWrongDirection
162 }
163}
164
165func decodeClientEnvelope(envelope *wire.MuxClientEnvelope, maxChunkBytes uint64) (ClientMessage, error) {
166 if envelope == nil {
167 return nil, ErrInvalidMessage
168 }
169 switch payload := envelope.Payload.(type) {
170 case *wire.MuxClientEnvelope_Open:
171 if payload.Open == nil {
172 return nil, ErrInvalidMessage
173 }
174 return openFromWire(payload.Open)
175 case *wire.MuxClientEnvelope_Data:
176 return dataFromWire(payload.Data, maxChunkBytes)
177 case *wire.MuxClientEnvelope_Close:
178 return closeFromWire(payload.Close)
179 case *wire.MuxClientEnvelope_Reset_:
180 return resetFromWire(payload.Reset_)
181 case *wire.MuxClientEnvelope_WindowUpdate:
182 return windowUpdateFromWire(payload.WindowUpdate)
183 default:
184 return nil, ErrWrongDirection
185 }
186}
187
188func encodeServerEnvelope(message ServerMessage, maxChunkBytes uint64) (*wire.MuxServerEnvelope, error) {
189 if message == nil {
190 return nil, ErrInvalidMessage
191 }
192 switch m := message.(type) {
193 case Opened:
194 if err := m.Ref.Validate(); err != nil {
195 return nil, ErrInvalidMessage
196 }
197 return &wire.MuxServerEnvelope{Payload: &wire.MuxServerEnvelope_Opened{Opened: &wire.MuxOpened{
198 Ref: refToWire(m.Ref),
199 }}}, nil
200 case *Opened:
201 if m == nil {
202 return nil, ErrInvalidMessage
203 }
204 return encodeServerEnvelope(*m, maxChunkBytes)
205 case Refused:
206 converted, err := refusedToWire(m)
207 if err != nil {
208 return nil, err
209 }
210 return &wire.MuxServerEnvelope{Payload: &wire.MuxServerEnvelope_Refused{Refused: converted}}, nil
211 case *Refused:
212 if m == nil {
213 return nil, ErrInvalidMessage
214 }
215 return encodeServerEnvelope(*m, maxChunkBytes)
216 case Data:
217 converted, err := dataToWire(m, maxChunkBytes)
218 if err != nil {
219 return nil, err
220 }
221 return &wire.MuxServerEnvelope{Payload: &wire.MuxServerEnvelope_Data{Data: converted}}, nil
222 case *Data:
223 if m == nil {
224 return nil, ErrInvalidMessage
225 }
226 return encodeServerEnvelope(*m, maxChunkBytes)
227 case Close:
228 converted, err := closeToWire(m)
229 if err != nil {
230 return nil, err
231 }
232 return &wire.MuxServerEnvelope{Payload: &wire.MuxServerEnvelope_Close{Close: converted}}, nil
233 case *Close:
234 if m == nil {
235 return nil, ErrInvalidMessage
236 }
237 return encodeServerEnvelope(*m, maxChunkBytes)
238 case Reset:
239 converted, err := resetToWire(m)
240 if err != nil {
241 return nil, err
242 }
243 return &wire.MuxServerEnvelope{Payload: &wire.MuxServerEnvelope_Reset_{Reset_: converted}}, nil
244 case *Reset:
245 if m == nil {
246 return nil, ErrInvalidMessage
247 }
248 return encodeServerEnvelope(*m, maxChunkBytes)
249 case WindowUpdate:
250 converted, err := windowUpdateToWire(m)
251 if err != nil {
252 return nil, err
253 }
254 return &wire.MuxServerEnvelope{Payload: &wire.MuxServerEnvelope_WindowUpdate{WindowUpdate: converted}}, nil
255 case *WindowUpdate:
256 if m == nil {
257 return nil, ErrInvalidMessage
258 }
259 return encodeServerEnvelope(*m, maxChunkBytes)
260 default:
261 return nil, ErrWrongDirection
262 }
263}
264
265func decodeServerEnvelope(envelope *wire.MuxServerEnvelope, maxChunkBytes uint64) (ServerMessage, error) {
266 if envelope == nil {
267 return nil, ErrInvalidMessage
268 }
269 switch payload := envelope.Payload.(type) {
270 case *wire.MuxServerEnvelope_Opened:
271 if payload.Opened == nil {
272 return nil, ErrInvalidMessage
273 }
274 ref, err := refFromWire(payload.Opened.GetRef())
275 if err != nil {
276 return nil, ErrInvalidMessage
277 }
278 return Opened{Ref: ref}, nil
279 case *wire.MuxServerEnvelope_Refused:
280 return refusedFromWire(payload.Refused)
281 case *wire.MuxServerEnvelope_Data:
282 return dataFromWire(payload.Data, maxChunkBytes)
283 case *wire.MuxServerEnvelope_Close:
284 return closeFromWire(payload.Close)
285 case *wire.MuxServerEnvelope_Reset_:
286 return resetFromWire(payload.Reset_)
287 case *wire.MuxServerEnvelope_WindowUpdate:
288 return windowUpdateFromWire(payload.WindowUpdate)
289 default:
290 return nil, ErrWrongDirection
291 }
292}
293
294func openToWire(m Open) (*wire.MuxOpen, error) {
295 if err := m.Ref.Validate(); err != nil {
296 return nil, ErrInvalidMessage
297 }
298 if err := checkEnvEntries(m.Env); err != nil {
299 return nil, err
300 }
301 purpose, err := purposeToWire(m.Purpose)
302 if err != nil {
303 return nil, err
304 }
305 startMode, err := startModeToWire(m.StartMode)
306 if err != nil {
307 return nil, err
308 }
309 out := &wire.MuxOpen{
310 Ref: refToWire(m.Ref),
311 Purpose: purpose, Local: m.Local, Endpoint: m.Endpoint,
312 Env: append([]string(nil), m.Env...),
313 Policy: policyToWire(m.Policy),
314 // The daemon-start authorization always travels, including for
315 // observation and control streams, so the daemon-side transport can
316 // never infer a spawn from an absent field.
317 StartMode: startMode,
318 }
319 if !m.Local {
320 out.Registration = registrationToWire(m.Registration)
321 }
322 if m.Purpose == ports.BrokerStreamAttachment {
323 out.Admission = uint32(m.Admission)
324 out.Name = m.Name
325 if m.Admission == ports.BrokerAdmissionExact {
326 out.Target = exactTargetToWire(m.Target)
327 }
328 }
329 return out, nil
330}
331
332func openFromWire(message *wire.MuxOpen) (Open, error) {
333 var out Open
334 if message == nil {
335 return out, ErrInvalidMessage
336 }
337 ref, err := refFromWire(message.GetRef())
338 if err != nil {
339 return Open{}, ErrInvalidMessage
340 }
341 purpose, err := purposeFromWire(message.GetPurpose())
342 if err != nil {
343 return Open{}, err
344 }
345 // Narrow the wire uint32 before the cast: a value whose high bits
346 // truncate onto a valid admission code would otherwise alias silently.
347 admission, err := muxEnum8[ports.BrokerStreamAdmission](message.GetAdmission())
348 if err != nil {
349 return Open{}, ErrInvalidMessage
350 }
351 name := message.GetName()
352 if purpose != ports.BrokerStreamAttachment && (admission != 0 || name != "") {
353 return Open{}, ErrInvalidMessage
354 }
355 startMode, err := startModeFromWire(message.GetStartMode())
356 if err != nil {
357 return Open{}, ErrInvalidMessage
358 }
359 local := message.GetLocal()
360 var registration domain.RemoteRegistration
361 if !local {
362 registration, err = registrationFromWire(message.GetRegistration())
363 if err != nil {
364 return Open{}, ErrInvalidMessage
365 }
366 } else if message.GetRegistration() != nil {
367 return Open{}, ErrInvalidMessage
368 }
369 var target protocol.ExactSessionTarget
370 if purpose == ports.BrokerStreamAttachment {
371 if admission == ports.BrokerAdmissionExact {
372 target, err = exactTargetFromWire(message.GetTarget())
373 if err != nil {
374 return Open{}, ErrInvalidMessage
375 }
376 } else if message.GetTarget() != nil {
377 return Open{}, ErrInvalidMessage
378 }
379 } else if message.GetTarget() != nil {
380 return Open{}, ErrInvalidMessage
381 }
382 env := append([]string(nil), message.GetEnv()...)
383 if err := checkEnvEntries(env); err != nil {
384 return Open{}, err
385 }
386 if purpose != ports.BrokerStreamAttachment && len(env) != 0 {
387 return Open{}, ErrInvalidMessage
388 }
389 policy, err := policyFromWire(message.GetPolicy())
390 if err != nil {
391 return Open{}, ErrInvalidMessage
392 }
393 endpoint := message.GetEndpoint()
394 candidate := Open{
395 Ref: ref, Purpose: purpose, Admission: admission, Name: name, Local: local, Endpoint: endpoint,
396 Registration: registration, Target: target, Env: env, Policy: policy,
397 StartMode: startMode,
398 }
399 request := ports.BrokerOpenStreamRequest{
400 Epoch: candidate.Ref.Epoch, Purpose: candidate.Purpose, Admission: candidate.Admission, Name: candidate.Name, Local: candidate.Local,
401 Connection: candidate.Ref.Connection, Stream: candidate.Ref.Client,
402 Endpoint: candidate.Endpoint, Registration: candidate.Registration,
403 Target: candidate.Target, Env: candidate.Env, Policy: candidate.Policy,
404 StartMode: candidate.StartMode,
405 }
406 // Decode owns validation of untrusted mux opens before daemon admission.
407 if err := request.Validate(); err != nil {
408 return Open{}, ErrInvalidMessage
409 }
410 return candidate, nil
411}
412
413func refusedToWire(m Refused) (*wire.MuxRefused, error) {
414 if err := m.Ref.Validate(); err != nil {
415 return nil, ErrInvalidMessage
416 }
417 if err := m.Error.validate(); err != nil {
418 return nil, err
419 }
420 return &wire.MuxRefused{Ref: refToWire(m.Ref), Error: errorDetailToWire(m.Error)}, nil
421}
422
423func refusedFromWire(message *wire.MuxRefused) (Refused, error) {
424 var out Refused
425 if message == nil {
426 return out, ErrInvalidMessage
427 }
428 ref, err := refFromWire(message.GetRef())
429 if err != nil {
430 return Refused{}, ErrInvalidMessage
431 }
432 detail, err := errorDetailFromWire(message.GetError())
433 if err != nil {
434 return Refused{}, ErrInvalidMessage
435 }
436 return Refused{Ref: ref, Error: detail}, nil
437}
438
439func dataToWire(m Data, maxChunkBytes uint64) (*wire.MuxData, error) {
440 if err := m.Physical.Validate(); err != nil {
441 return nil, ErrInvalidMessage
442 }
443 if uint64(len(m.Data)) > maxChunkBytes {
444 return nil, ErrTooLarge
445 }
446 if uint64(len(m.Data)) > MaxMuxChunkBytes {
447 return nil, ErrTooLarge
448 }
449 if len(m.Data) == 0 {
450 return nil, ErrInvalidMessage
451 }
452 return &wire.MuxData{
453 PhysicalStreamId: physicalToWire(m.Physical),
454 Data: append([]byte(nil), m.Data...),
455 }, nil
456}
457
458func dataFromWire(message *wire.MuxData, maxChunkBytes uint64) (Data, error) {
459 var out Data
460 if message == nil {
461 return out, ErrInvalidMessage
462 }
463 physical, err := physicalFromWire(message.GetPhysicalStreamId())
464 if err != nil {
465 return Data{}, ErrInvalidMessage
466 }
467 data := message.GetData()
468 if uint64(len(data)) > maxChunkBytes {
469 return Data{}, ErrTooLarge
470 }
471 if uint64(len(data)) > MaxMuxChunkBytes {
472 return Data{}, ErrTooLarge
473 }
474 if len(data) == 0 {
475 return Data{}, ErrInvalidMessage
476 }
477 return Data{Physical: physical, Data: append([]byte(nil), data...)}, nil
478}
479
480func closeToWire(m Close) (*wire.MuxClose, error) {
481 if err := m.Physical.Validate(); err != nil {
482 return nil, ErrInvalidMessage
483 }
484 return &wire.MuxClose{PhysicalStreamId: physicalToWire(m.Physical)}, nil
485}
486
487func closeFromWire(message *wire.MuxClose) (Close, error) {
488 var out Close
489 if message == nil {
490 return out, ErrInvalidMessage
491 }
492 physical, err := physicalFromWire(message.GetPhysicalStreamId())
493 if err != nil {
494 return Close{}, ErrInvalidMessage
495 }
496 return Close{Physical: physical}, nil
497}
498
499func windowUpdateToWire(m WindowUpdate) (*wire.MuxWindowUpdate, error) {
500 if err := m.Physical.Validate(); err != nil {
501 return nil, ErrInvalidMessage
502 }
503 if m.Credit == 0 || m.Credit > MaxMuxStreamWindowBytes {
504 return nil, ErrInvalidMessage
505 }
506 return &wire.MuxWindowUpdate{PhysicalStreamId: physicalToWire(m.Physical), CreditBytes: m.Credit}, nil
507}
508
509func windowUpdateFromWire(message *wire.MuxWindowUpdate) (WindowUpdate, error) {
510 if message == nil {
511 return WindowUpdate{}, ErrInvalidMessage
512 }
513 physical, err := physicalFromWire(message.GetPhysicalStreamId())
514 if err != nil {
515 return WindowUpdate{}, ErrInvalidMessage
516 }
517 credit := message.GetCreditBytes()
518 if credit == 0 || credit > MaxMuxStreamWindowBytes {
519 return WindowUpdate{}, ErrInvalidMessage
520 }
521 return WindowUpdate{Physical: physical, Credit: credit}, nil
522}
523
524func resetToWire(m Reset) (*wire.MuxReset, error) {
525 if err := m.Physical.Validate(); err != nil {
526 return nil, ErrInvalidMessage
527 }
528 if m.HasError {
529 if err := m.Error.validate(); err != nil {
530 return nil, err
531 }
532 } else if m.Error != (ErrorDetail{}) {
533 // A reset without a declared detail carries no detail at all; a
534 // nonzero one under HasError=false is a caller bug that must fail
535 // closed rather than be silently dropped.
536 return nil, ErrInvalidMessage
537 }
538 out := &wire.MuxReset{PhysicalStreamId: physicalToWire(m.Physical)}
539 if m.HasError {
540 out.Error = errorDetailToWire(m.Error)
541 }
542 return out, nil
543}
544
545func resetFromWire(message *wire.MuxReset) (Reset, error) {
546 var out Reset
547 if message == nil {
548 return out, ErrInvalidMessage
549 }
550 physical, err := physicalFromWire(message.GetPhysicalStreamId())
551 if err != nil {
552 return Reset{}, ErrInvalidMessage
553 }
554 result := Reset{Physical: physical}
555 if message.GetError() != nil {
556 detail, err := errorDetailFromWire(message.GetError())
557 if err != nil {
558 return Reset{}, ErrInvalidMessage
559 }
560 result.Error = detail
561 result.HasError = true
562 }
563 return result, nil
564}
565
566// purposeToWire maps the closed ports stream-purpose taxonomy onto the wire
567// taxonomy. The switch is explicit: an unknown purpose never defaults to a
568// wire value.
569func purposeToWire(purpose ports.BrokerStreamPurpose) (uint32, error) {
570 switch purpose {
571 case ports.BrokerStreamAttachment:
572 return 1, nil
573 case ports.BrokerStreamControl:
574 return 2, nil
575 case ports.BrokerStreamObservation:
576 return 3, nil
577 default:
578 return 0, ErrInvalidMessage
579 }
580}
581
582func purposeFromWire(value uint32) (ports.BrokerStreamPurpose, error) {
583 switch value {
584 case 1:
585 return ports.BrokerStreamAttachment, nil
586 case 2:
587 return ports.BrokerStreamControl, nil
588 case 3:
589 return ports.BrokerStreamObservation, nil
590 default:
591 return 0, ErrInvalidMessage
592 }
593}