1package brokerwire
2
3// Directional envelope adaptation.
4//
5// EncodeClient wraps one adapter-local client value in its directional
6// BrokerClientEnvelope; DecodeClient unwraps and validates. EncodeServer
7// and DecodeServer do the same for the server direction. Every convert
8// path uses an explicit taxonomy switch: unknown variants and
9// wrong-direction payloads fail with ErrWrongDirection before any
10// mutation. Decode paths run strict wire.ScanEnvelope before generated
11// unmarshal, then validate IDs, UTF-8/control/bidi text, bounds, times,
12// environment/policy, chunks, and snapshot parts.
13//
14// The codecs are stateless: no connection state, multipart assembler,
15// operation tracker, stream lifecycle, socket, framing pump, or
16// transport lives here. Envelope sizes ride under the caller's negotiated
17// ceiling; stream chunks ride under the negotiated chunk ceiling.
18
19import (
20 "math"
21
22 "google.golang.org/protobuf/proto"
23
24 "github.com/bnema/vev/internal/adapters/protoconv"
25 "github.com/bnema/vev/internal/domain"
26 "github.com/bnema/vev/internal/ports"
27 "github.com/bnema/vev/internal/protocol"
28 "github.com/bnema/vev/internal/protocol/catalogue"
29 "github.com/bnema/vev/internal/protocol/wire"
30)
31
32// EncodeClient wraps one client message in its serialized directional
33// envelope. The payload must fit maxEnvelopeBytes; stream chunks must fit
34// 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 := checkBrokerEnvelopeCeiling(raw, maxEnvelopeBytes); err != nil {
45 return nil, err
46 }
47 return raw, nil
48}
49
50// DecodeClient unwraps one serialized client 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 := checkBrokerEnvelopeCeiling(payload, maxEnvelopeBytes); err != nil {
55 return nil, err
56 }
57 envelope := &wire.BrokerClientEnvelope{}
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 server message in its serialized directional
68// envelope.
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 := checkBrokerEnvelopeCeiling(raw, maxEnvelopeBytes); err != nil {
79 return nil, err
80 }
81 return raw, nil
82}
83
84// DecodeServer unwraps one serialized server envelope into its
85// adapter-local value.
86func DecodeServer(payload []byte, maxEnvelopeBytes, maxChunkBytes uint64) (ServerMessage, error) {
87 if err := checkBrokerEnvelopeCeiling(payload, maxEnvelopeBytes); err != nil {
88 return nil, err
89 }
90 envelope := &wire.BrokerServerEnvelope{}
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.BrokerClientEnvelope, error) {
101 if message == nil {
102 return nil, ErrInvalidMessage
103 }
104 switch m := message.(type) {
105 case Register:
106 if len(m.Build) > 128 {
107 return nil, ErrInvalidMessage
108 }
109 return &wire.BrokerClientEnvelope{Payload: &wire.BrokerClientEnvelope_Register{Register: &wire.Register{Build: m.Build}}}, nil
110 case *Register:
111 if m == nil {
112 return nil, ErrInvalidMessage
113 }
114 return encodeClientEnvelope(*m, maxChunkBytes)
115 case Subscribe:
116 if m.Epoch == 0 {
117 return nil, ErrInvalidMessage
118 }
119 if err := m.Connection.Validate(); err != nil {
120 return nil, ErrInvalidMessage
121 }
122 return &wire.BrokerClientEnvelope{Payload: &wire.BrokerClientEnvelope_Subscribe{Subscribe: &wire.Subscribe{
123 Scope: scopeToWire(m.Epoch, m.Connection), Generation: m.Generation, Passive: m.Passive,
124 }}}, nil
125 case *Subscribe:
126 if m == nil {
127 return nil, ErrInvalidMessage
128 }
129 return encodeClientEnvelope(*m, maxChunkBytes)
130 case Resync:
131 if m.Epoch == 0 {
132 return nil, ErrInvalidMessage
133 }
134 if err := m.Connection.Validate(); err != nil {
135 return nil, ErrInvalidMessage
136 }
137 return &wire.BrokerClientEnvelope{Payload: &wire.BrokerClientEnvelope_Resync{Resync: &wire.Resync{
138 Scope: scopeToWire(m.Epoch, m.Connection), Generation: m.Generation,
139 }}}, nil
140 case *Resync:
141 if m == nil {
142 return nil, ErrInvalidMessage
143 }
144 return encodeClientEnvelope(*m, maxChunkBytes)
145 case Unsubscribe:
146 if m.Epoch == 0 {
147 return nil, ErrInvalidMessage
148 }
149 if err := m.Connection.Validate(); err != nil {
150 return nil, ErrInvalidMessage
151 }
152 return &wire.BrokerClientEnvelope{Payload: &wire.BrokerClientEnvelope_Unsubscribe{Unsubscribe: &wire.Unsubscribe{
153 Scope: scopeToWire(m.Epoch, m.Connection), Generation: m.Generation,
154 }}}, nil
155 case *Unsubscribe:
156 if m == nil {
157 return nil, ErrInvalidMessage
158 }
159 return encodeClientEnvelope(*m, maxChunkBytes)
160 case AddHost:
161 converted, err := addHostToWire(m)
162 if err != nil {
163 return nil, err
164 }
165 return &wire.BrokerClientEnvelope{Payload: &wire.BrokerClientEnvelope_AddHost{AddHost: converted}}, nil
166 case *AddHost:
167 if m == nil {
168 return nil, ErrInvalidMessage
169 }
170 return encodeClientEnvelope(*m, maxChunkBytes)
171 case RemoveHost:
172 converted, err := removeHostToWire(m)
173 if err != nil {
174 return nil, err
175 }
176 return &wire.BrokerClientEnvelope{Payload: &wire.BrokerClientEnvelope_RemoveHost{RemoveHost: converted}}, nil
177 case *RemoveHost:
178 if m == nil {
179 return nil, ErrInvalidMessage
180 }
181 return encodeClientEnvelope(*m, maxChunkBytes)
182 case UpdateHostPolicy:
183 converted, err := updateHostPolicyToWire(m)
184 if err != nil {
185 return nil, err
186 }
187 return &wire.BrokerClientEnvelope{Payload: &wire.BrokerClientEnvelope_UpdateHostPolicy{UpdateHostPolicy: converted}}, nil
188 case *UpdateHostPolicy:
189 if m == nil {
190 return nil, ErrInvalidMessage
191 }
192 return encodeClientEnvelope(*m, maxChunkBytes)
193 case Reconcile:
194 converted, err := reconcileToWire(m)
195 if err != nil {
196 return nil, err
197 }
198 return &wire.BrokerClientEnvelope{Payload: &wire.BrokerClientEnvelope_Reconcile{Reconcile: converted}}, nil
199 case *Reconcile:
200 if m == nil {
201 return nil, ErrInvalidMessage
202 }
203 return encodeClientEnvelope(*m, maxChunkBytes)
204 case OpenStream:
205 converted, err := openStreamToWire(m)
206 if err != nil {
207 return nil, err
208 }
209 return &wire.BrokerClientEnvelope{Payload: &wire.BrokerClientEnvelope_OpenStream{OpenStream: converted}}, nil
210 case *OpenStream:
211 if m == nil {
212 return nil, ErrInvalidMessage
213 }
214 return encodeClientEnvelope(*m, maxChunkBytes)
215 case ClientStreamData:
216 converted, err := clientStreamDataToWire(m, maxChunkBytes)
217 if err != nil {
218 return nil, err
219 }
220 return &wire.BrokerClientEnvelope{Payload: &wire.BrokerClientEnvelope_ClientStreamData{ClientStreamData: converted}}, nil
221 case *ClientStreamData:
222 if m == nil {
223 return nil, ErrInvalidMessage
224 }
225 return encodeClientEnvelope(*m, maxChunkBytes)
226 case CloseStream:
227 if m.Epoch == 0 {
228 return nil, ErrInvalidMessage
229 }
230 if err := m.Connection.Validate(); err != nil {
231 return nil, ErrInvalidMessage
232 }
233 if err := m.Stream.Validate(); err != nil {
234 return nil, ErrInvalidMessage
235 }
236 return &wire.BrokerClientEnvelope{Payload: &wire.BrokerClientEnvelope_CloseStream{CloseStream: &wire.CloseStream{
237 Ref: refToWire(m.Epoch, m.Connection, m.Stream),
238 }}}, nil
239 case *CloseStream:
240 if m == nil {
241 return nil, ErrInvalidMessage
242 }
243 return encodeClientEnvelope(*m, maxChunkBytes)
244 case StreamWindowUpdate:
245 converted, err := streamWindowUpdateToWire(m)
246 if err != nil {
247 return nil, err
248 }
249 return &wire.BrokerClientEnvelope{Payload: &wire.BrokerClientEnvelope_StreamWindowUpdate{StreamWindowUpdate: converted}}, nil
250 case *StreamWindowUpdate:
251 if m == nil {
252 return nil, ErrInvalidMessage
253 }
254 return encodeClientEnvelope(*m, maxChunkBytes)
255 case StartPreview:
256 converted, err := startPreviewToWire(m)
257 if err != nil {
258 return nil, err
259 }
260 return &wire.BrokerClientEnvelope{Payload: &wire.BrokerClientEnvelope_StartPreview{StartPreview: converted}}, nil
261 case *StartPreview:
262 if m == nil {
263 return nil, ErrInvalidMessage
264 }
265 return encodeClientEnvelope(*m, maxChunkBytes)
266 case CancelPreview:
267 converted, err := cancelPreviewToWire(m)
268 if err != nil {
269 return nil, err
270 }
271 return &wire.BrokerClientEnvelope{Payload: &wire.BrokerClientEnvelope_CancelPreview{CancelPreview: converted}}, nil
272 case *CancelPreview:
273 if m == nil {
274 return nil, ErrInvalidMessage
275 }
276 return encodeClientEnvelope(*m, maxChunkBytes)
277 default:
278 return nil, ErrWrongDirection
279 }
280}
281
282func decodeClientEnvelope(envelope *wire.BrokerClientEnvelope, maxChunkBytes uint64) (ClientMessage, error) {
283 if envelope == nil {
284 return nil, ErrInvalidMessage
285 }
286 switch payload := envelope.Payload.(type) {
287 case *wire.BrokerClientEnvelope_Register:
288 if payload.Register == nil {
289 return nil, ErrInvalidMessage
290 }
291 if len(payload.Register.GetBuild()) > 128 {
292 return nil, ErrInvalidMessage
293 }
294 return Register{Build: payload.Register.GetBuild()}, nil
295 case *wire.BrokerClientEnvelope_Subscribe:
296 return subscribeFromWire(payload.Subscribe)
297 case *wire.BrokerClientEnvelope_Resync:
298 return resyncFromWire(payload.Resync)
299 case *wire.BrokerClientEnvelope_Unsubscribe:
300 return unsubscribeFromWire(payload.Unsubscribe)
301 case *wire.BrokerClientEnvelope_AddHost:
302 return addHostFromWire(payload.AddHost)
303 case *wire.BrokerClientEnvelope_RemoveHost:
304 return removeHostFromWire(payload.RemoveHost)
305 case *wire.BrokerClientEnvelope_UpdateHostPolicy:
306 return updateHostPolicyFromWire(payload.UpdateHostPolicy)
307 case *wire.BrokerClientEnvelope_Reconcile:
308 return reconcileFromWire(payload.Reconcile)
309 case *wire.BrokerClientEnvelope_OpenStream:
310 return openStreamFromWire(payload.OpenStream)
311 case *wire.BrokerClientEnvelope_ClientStreamData:
312 return clientStreamDataFromWire(payload.ClientStreamData, maxChunkBytes)
313 case *wire.BrokerClientEnvelope_CloseStream:
314 if payload.CloseStream == nil {
315 return nil, ErrInvalidMessage
316 }
317 epoch, connection, stream, err := refFromWire(payload.CloseStream.GetRef())
318 if err != nil {
319 return nil, ErrInvalidMessage
320 }
321 return CloseStream{Epoch: epoch, Connection: connection, Stream: stream}, nil
322 case *wire.BrokerClientEnvelope_StreamWindowUpdate:
323 return streamWindowUpdateFromWire(payload.StreamWindowUpdate)
324 case *wire.BrokerClientEnvelope_StartPreview:
325 return startPreviewFromWire(payload.StartPreview)
326 case *wire.BrokerClientEnvelope_CancelPreview:
327 return cancelPreviewFromWire(payload.CancelPreview)
328 default:
329 return nil, ErrWrongDirection
330 }
331}
332
333func addHostToWire(m AddHost) (*wire.AddHost, error) {
334 if m.Epoch == 0 {
335 return nil, ErrInvalidMessage
336 }
337 if err := m.Connection.Validate(); err != nil {
338 return nil, ErrInvalidMessage
339 }
340 if err := m.Operation.Validate(); err != nil {
341 return nil, ErrInvalidMessage
342 }
343 if err := validateBrokerEndpoint(m.Endpoint); err != nil {
344 return nil, err
345 }
346 // Policy is required authority: a zero or invalid policy is refused
347 // here rather than travelling as an absent message the peer must
348 // interpret.
349 if err := m.Policy.Validate(); err != nil {
350 return nil, ErrInvalidMessage
351 }
352 return &wire.AddHost{
353 Scope: scopeToWire(m.Epoch, m.Connection),
354 OperationId: operationToWire(m.Operation),
355 Endpoint: m.Endpoint,
356 Policy: policyToWire(m.Policy),
357 }, nil
358}
359
360func addHostFromWire(message *wire.AddHost) (AddHost, error) {
361 var out AddHost
362 if message == nil {
363 return out, ErrInvalidMessage
364 }
365 epoch, connection, err := scopeFromWire(message.GetScope())
366 if err != nil {
367 return AddHost{}, ErrInvalidMessage
368 }
369 operation, err := operationFromWire(message.GetOperationId())
370 if err != nil {
371 return AddHost{}, ErrInvalidMessage
372 }
373 if err := validateBrokerEndpoint(message.GetEndpoint()); err != nil {
374 return AddHost{}, err
375 }
376 policy, err := policyFromWire(message.GetPolicy())
377 if err != nil {
378 return AddHost{}, ErrInvalidMessage
379 }
380 return AddHost{Epoch: epoch, Connection: connection, Operation: operation, Endpoint: message.GetEndpoint(), Policy: policy}, nil
381}
382
383func removeHostToWire(m RemoveHost) (*wire.RemoveHost, error) {
384 if m.Epoch == 0 {
385 return nil, ErrInvalidMessage
386 }
387 if err := m.Connection.Validate(); err != nil {
388 return nil, ErrInvalidMessage
389 }
390 if err := m.Operation.Validate(); err != nil {
391 return nil, ErrInvalidMessage
392 }
393 // Removal carries exact authority, never a bare endpoint: a zero or
394 // partial registration could not fence a remove/re-add race.
395 if err := m.Registration.Validate(); err != nil {
396 return nil, ErrInvalidMessage
397 }
398 return &wire.RemoveHost{
399 Scope: scopeToWire(m.Epoch, m.Connection),
400 OperationId: operationToWire(m.Operation),
401 Registration: registrationToWire(m.Registration),
402 }, nil
403}
404
405func removeHostFromWire(message *wire.RemoveHost) (RemoveHost, error) {
406 var out RemoveHost
407 if message == nil {
408 return out, ErrInvalidMessage
409 }
410 epoch, connection, err := scopeFromWire(message.GetScope())
411 if err != nil {
412 return RemoveHost{}, ErrInvalidMessage
413 }
414 operation, err := operationFromWire(message.GetOperationId())
415 if err != nil {
416 return RemoveHost{}, ErrInvalidMessage
417 }
418 registration, err := registrationFromWire(message.GetRegistration())
419 if err != nil {
420 return RemoveHost{}, ErrInvalidMessage
421 }
422 return RemoveHost{Epoch: epoch, Connection: connection, Operation: operation, Registration: registration}, nil
423}
424
425func updateHostPolicyToWire(m UpdateHostPolicy) (*wire.UpdateHostPolicy, error) {
426 if m.Epoch == 0 {
427 return nil, ErrInvalidMessage
428 }
429 if err := m.Connection.Validate(); err != nil {
430 return nil, ErrInvalidMessage
431 }
432 if err := m.Operation.Validate(); err != nil {
433 return nil, ErrInvalidMessage
434 }
435 if err := m.Registration.Validate(); err != nil {
436 return nil, ErrInvalidMessage
437 }
438 if err := m.Policy.Validate(); err != nil {
439 return nil, ErrInvalidMessage
440 }
441 return &wire.UpdateHostPolicy{
442 Scope: scopeToWire(m.Epoch, m.Connection),
443 OperationId: operationToWire(m.Operation),
444 Registration: registrationToWire(m.Registration),
445 Policy: policyToWire(m.Policy),
446 }, nil
447}
448
449func updateHostPolicyFromWire(message *wire.UpdateHostPolicy) (UpdateHostPolicy, error) {
450 var out UpdateHostPolicy
451 if message == nil {
452 return out, ErrInvalidMessage
453 }
454 epoch, connection, err := scopeFromWire(message.GetScope())
455 if err != nil {
456 return UpdateHostPolicy{}, ErrInvalidMessage
457 }
458 operation, err := operationFromWire(message.GetOperationId())
459 if err != nil {
460 return UpdateHostPolicy{}, ErrInvalidMessage
461 }
462 registration, err := registrationFromWire(message.GetRegistration())
463 if err != nil {
464 return UpdateHostPolicy{}, ErrInvalidMessage
465 }
466 policy, err := policyFromWire(message.GetPolicy())
467 if err != nil {
468 return UpdateHostPolicy{}, ErrInvalidMessage
469 }
470 return UpdateHostPolicy{Epoch: epoch, Connection: connection, Operation: operation, Registration: registration, Policy: policy}, nil
471}
472
473func reconcileToWire(m Reconcile) (*wire.Reconcile, error) {
474 if m.Epoch == 0 {
475 return nil, ErrInvalidMessage
476 }
477 if err := m.Connection.Validate(); err != nil {
478 return nil, ErrInvalidMessage
479 }
480 if err := m.Registration.Validate(); err != nil {
481 return nil, ErrInvalidMessage
482 }
483 return &wire.Reconcile{
484 Scope: scopeToWire(m.Epoch, m.Connection),
485 Registration: registrationToWire(m.Registration),
486 }, nil
487}
488
489func reconcileFromWire(message *wire.Reconcile) (Reconcile, error) {
490 var out Reconcile
491 if message == nil {
492 return out, ErrInvalidMessage
493 }
494 epoch, connection, err := scopeFromWire(message.GetScope())
495 if err != nil {
496 return Reconcile{}, ErrInvalidMessage
497 }
498 registration, err := registrationFromWire(message.GetRegistration())
499 if err != nil {
500 return Reconcile{}, ErrInvalidMessage
501 }
502 return Reconcile{Epoch: epoch, Connection: connection, Registration: registration}, nil
503}
504
505func subscribeFromWire(message *wire.Subscribe) (Subscribe, error) {
506 var out Subscribe
507 if message == nil {
508 return out, ErrInvalidMessage
509 }
510 epoch, connection, err := scopeFromWire(message.GetScope())
511 if err != nil {
512 return Subscribe{}, ErrInvalidMessage
513 }
514 return Subscribe{Epoch: epoch, Connection: connection, Generation: message.GetGeneration(), Passive: message.GetPassive()}, nil
515}
516
517func resyncFromWire(message *wire.Resync) (Resync, error) {
518 var out Resync
519 if message == nil {
520 return out, ErrInvalidMessage
521 }
522 epoch, connection, err := scopeFromWire(message.GetScope())
523 if err != nil {
524 return Resync{}, ErrInvalidMessage
525 }
526 return Resync{Epoch: epoch, Connection: connection, Generation: message.GetGeneration()}, nil
527}
528
529func unsubscribeFromWire(message *wire.Unsubscribe) (Unsubscribe, error) {
530 var out Unsubscribe
531 if message == nil {
532 return out, ErrInvalidMessage
533 }
534 epoch, connection, err := scopeFromWire(message.GetScope())
535 if err != nil {
536 return Unsubscribe{}, ErrInvalidMessage
537 }
538 return Unsubscribe{Epoch: epoch, Connection: connection, Generation: message.GetGeneration()}, nil
539}
540
541func openStreamToWire(m OpenStream) (*wire.OpenStream, error) {
542 if err := checkEnvEntries(m.Env); err != nil {
543 return nil, err
544 }
545 var purpose uint32
546 switch m.Purpose {
547 case ports.BrokerStreamAttachment:
548 purpose = 1
549 case ports.BrokerStreamControl:
550 purpose = 2
551 case ports.BrokerStreamObservation:
552 purpose = 3
553 default:
554 return nil, ErrInvalidMessage
555 }
556 admission, err := admissionToWire(m.Admission)
557 if err != nil {
558 return nil, err
559 }
560 startMode, err := startModeToWire(m.StartMode)
561 if err != nil {
562 return nil, err
563 }
564 out := &wire.OpenStream{
565 Ref: refToWire(m.Epoch, m.Connection, m.Stream),
566 Purpose: purpose, Local: m.Local, Endpoint: m.Endpoint,
567 Env: append([]string(nil), m.Env...),
568 Policy: policyToWire(m.Policy),
569 // The admission taxonomy mirrors the port values (0 none, 1 exact,
570 // 2 create named, 3 create ephemeral, 4 attach named); the name
571 // travels only for the two named variants.
572 Admission: admission, Name: m.Name, StartMode: startMode,
573 }
574 if !m.Local {
575 out.Registration = registrationToWire(m.Registration)
576 }
577 // Only exact attach/resume carries a session target; a creation
578 // admission carries a zero target that must never be encoded as a
579 // present message.
580 if m.Purpose == ports.BrokerStreamAttachment && m.Admission == ports.BrokerAdmissionExact {
581 out.Target = exactTargetToWire(m.Target)
582 }
583 return out, nil
584}
585
586// admissionToWire maps the closed admission taxonomy onto its wire code.
587func admissionToWire(admission ports.BrokerStreamAdmission) (uint32, error) {
588 switch admission {
589 case 0:
590 return 0, nil
591 case ports.BrokerAdmissionExact:
592 return 1, nil
593 case ports.BrokerAdmissionCreateNamed:
594 return 2, nil
595 case ports.BrokerAdmissionCreateEphemeral:
596 return 3, nil
597 case ports.BrokerAdmissionAttachNamed:
598 return 4, nil
599 default:
600 return 0, errConvertRange
601 }
602}
603
604// admissionFromWire maps a wire admission code onto the closed taxonomy.
605func admissionFromWire(value uint32) (ports.BrokerStreamAdmission, error) {
606 switch value {
607 case 0:
608 return 0, nil
609 case 1:
610 return ports.BrokerAdmissionExact, nil
611 case 2:
612 return ports.BrokerAdmissionCreateNamed, nil
613 case 3:
614 return ports.BrokerAdmissionCreateEphemeral, nil
615 case 4:
616 return ports.BrokerAdmissionAttachNamed, nil
617 default:
618 return 0, errConvertRange
619 }
620}
621
622// Start-mode conversion retains the brokerwire range sentinel at the boundary.
623func startModeToWire(mode ports.BrokerDaemonStartMode) (uint32, error) {
624 value, err := protoconv.BrokerStartModeToWire(mode)
625 if err != nil {
626 return 0, errConvertRange
627 }
628 return value, nil
629}
630
631func startModeFromWire(value uint32) (ports.BrokerDaemonStartMode, error) {
632 mode, err := protoconv.BrokerStartModeFromWire(value)
633 if err != nil {
634 return 0, errConvertRange
635 }
636 return mode, nil
637}
638
639func openStreamFromWire(message *wire.OpenStream) (OpenStream, error) {
640 var out OpenStream
641 if message == nil {
642 return out, ErrInvalidMessage
643 }
644 epoch, connection, stream, err := refFromWire(message.GetRef())
645 if err != nil {
646 return OpenStream{}, ErrInvalidMessage
647 }
648 var purpose ports.BrokerStreamPurpose
649 switch message.GetPurpose() {
650 case 1:
651 purpose = ports.BrokerStreamAttachment
652 case 2:
653 purpose = ports.BrokerStreamControl
654 case 3:
655 purpose = ports.BrokerStreamObservation
656 default:
657 return OpenStream{}, ErrInvalidMessage
658 }
659 admission, err := admissionFromWire(message.GetAdmission())
660 if err != nil {
661 return OpenStream{}, ErrInvalidMessage
662 }
663 startMode, err := startModeFromWire(message.GetStartMode())
664 if err != nil {
665 return OpenStream{}, ErrInvalidMessage
666 }
667 local := message.GetLocal()
668 var registration domain.RemoteRegistration
669 if !local {
670 registration, err = registrationFromWire(message.GetRegistration())
671 if err != nil {
672 return OpenStream{}, ErrInvalidMessage
673 }
674 } else if message.GetRegistration() != nil {
675 return OpenStream{}, ErrInvalidMessage
676 }
677 // Only exact attach/resume carries a session target. A present target on
678 // any other admission or purpose is malformed.
679 var target protocol.ExactSessionTarget
680 if purpose == ports.BrokerStreamAttachment && admission == ports.BrokerAdmissionExact {
681 target, err = exactTargetFromWire(message.GetTarget())
682 if err != nil {
683 return OpenStream{}, ErrInvalidMessage
684 }
685 } else if message.GetTarget() != nil {
686 return OpenStream{}, ErrInvalidMessage
687 }
688 env := append([]string(nil), message.GetEnv()...)
689 if err := checkEnvEntries(env); err != nil {
690 return OpenStream{}, err
691 }
692 if purpose != ports.BrokerStreamAttachment && len(env) != 0 {
693 return OpenStream{}, ErrInvalidMessage
694 }
695 policy, err := policyFromWire(message.GetPolicy())
696 if err != nil {
697 return OpenStream{}, ErrInvalidMessage
698 }
699 endpoint := message.GetEndpoint()
700 candidate := OpenStream{
701 Epoch: epoch, Connection: connection, Stream: stream,
702 Purpose: purpose, Admission: admission, Name: message.GetName(),
703 Local: local,
704 Endpoint: endpoint,
705 Registration: registration, Target: target, Env: env, Policy: policy,
706 StartMode: startMode,
707 }
708 request := ports.BrokerOpenStreamRequest{
709 Epoch: candidate.Epoch, Purpose: candidate.Purpose,
710 Admission: candidate.Admission, Name: candidate.Name, Local: candidate.Local,
711 Connection: candidate.Connection, Stream: candidate.Stream,
712 Endpoint: candidate.Endpoint, Registration: candidate.Registration,
713 Target: candidate.Target, Env: candidate.Env, Policy: candidate.Policy,
714 StartMode: candidate.StartMode,
715 }
716 // Decode owns validation of untrusted open requests before they enter the broker.
717 if err := request.Validate(); err != nil {
718 return OpenStream{}, ErrInvalidMessage
719 }
720 return candidate, nil
721}
722
723// streamWindowUpdateToWire refuses a zero or above-window grant: credit is
724// returned only for consumed data, so a valid grant never exceeds one window.
725func streamWindowUpdateToWire(m StreamWindowUpdate) (*wire.StreamWindowUpdate, error) {
726 if m.Epoch == 0 || m.Connection.Validate() != nil || m.Stream.Validate() != nil {
727 return nil, ErrInvalidMessage
728 }
729 if m.Credit == 0 || m.Credit > StreamWindowBytes {
730 return nil, ErrInvalidMessage
731 }
732 return &wire.StreamWindowUpdate{Ref: refToWire(m.Epoch, m.Connection, m.Stream), Credit: m.Credit}, nil
733}
734
735func streamWindowUpdateFromWire(message *wire.StreamWindowUpdate) (StreamWindowUpdate, error) {
736 if message == nil {
737 return StreamWindowUpdate{}, ErrInvalidMessage
738 }
739 epoch, connection, stream, err := refFromWire(message.GetRef())
740 if err != nil {
741 return StreamWindowUpdate{}, ErrInvalidMessage
742 }
743 credit := message.GetCredit()
744 if credit == 0 || credit > StreamWindowBytes {
745 return StreamWindowUpdate{}, ErrInvalidMessage
746 }
747 return StreamWindowUpdate{Epoch: epoch, Connection: connection, Stream: stream, Credit: credit}, nil
748}
749
750func clientStreamDataToWire(m ClientStreamData, maxChunkBytes uint64) (*wire.ClientStreamData, error) {
751 if m.Epoch == 0 {
752 return nil, ErrInvalidMessage
753 }
754 if err := m.Connection.Validate(); err != nil {
755 return nil, ErrInvalidMessage
756 }
757 if err := m.Stream.Validate(); err != nil {
758 return nil, ErrInvalidMessage
759 }
760 if uint64(len(m.Data)) > maxChunkBytes {
761 return nil, ErrTooLarge
762 }
763 if uint64(len(m.Data)) > MaxStreamChunkBytes {
764 return nil, ErrTooLarge
765 }
766 if len(m.Data) == 0 {
767 return nil, ErrInvalidMessage
768 }
769 return &wire.ClientStreamData{
770 Ref: refToWire(m.Epoch, m.Connection, m.Stream),
771 Data: append([]byte(nil), m.Data...),
772 }, nil
773}
774
775func clientStreamDataFromWire(message *wire.ClientStreamData, maxChunkBytes uint64) (ClientStreamData, error) {
776 var out ClientStreamData
777 if message == nil {
778 return out, ErrInvalidMessage
779 }
780 epoch, connection, stream, err := refFromWire(message.GetRef())
781 if err != nil {
782 return ClientStreamData{}, ErrInvalidMessage
783 }
784 data := message.GetData()
785 if uint64(len(data)) > maxChunkBytes {
786 return ClientStreamData{}, ErrTooLarge
787 }
788 if uint64(len(data)) > MaxStreamChunkBytes {
789 return ClientStreamData{}, ErrTooLarge
790 }
791 if len(data) == 0 {
792 return ClientStreamData{}, ErrInvalidMessage
793 }
794 // proto.Unmarshal copies bytes fields out of the payload, so the caller
795 // owns data without another copy.
796 return ClientStreamData{Epoch: epoch, Connection: connection, Stream: stream, Data: data}, nil
797}
798
799func encodeServerEnvelope(message ServerMessage, maxChunkBytes uint64) (*wire.BrokerServerEnvelope, error) {
800 if message == nil {
801 return nil, ErrInvalidMessage
802 }
803 switch m := message.(type) {
804 case Registered:
805 if len(m.Build) > 128 {
806 return nil, ErrInvalidMessage
807 }
808 if m.Epoch == 0 {
809 return nil, ErrInvalidMessage
810 }
811 if err := m.Connection.Validate(); err != nil {
812 return nil, ErrInvalidMessage
813 }
814 return &wire.BrokerServerEnvelope{Payload: &wire.BrokerServerEnvelope_Registered{Registered: &wire.Registered{
815 Scope: scopeToWire(m.Epoch, m.Connection), Build: m.Build, Retiring: m.Retiring,
816 }}}, nil
817 case *Registered:
818 if m == nil {
819 return nil, ErrInvalidMessage
820 }
821 return encodeServerEnvelope(*m, maxChunkBytes)
822 case SnapshotPart:
823 converted, err := snapshotPartToWire(m)
824 if err != nil {
825 return nil, err
826 }
827 return &wire.BrokerServerEnvelope{Payload: &wire.BrokerServerEnvelope_SnapshotPart{SnapshotPart: converted}}, nil
828 case *SnapshotPart:
829 if m == nil {
830 return nil, ErrInvalidMessage
831 }
832 return encodeServerEnvelope(*m, maxChunkBytes)
833 case OperationResult:
834 converted, err := operationResultToWire(m)
835 if err != nil {
836 return nil, err
837 }
838 return &wire.BrokerServerEnvelope{Payload: &wire.BrokerServerEnvelope_OperationResult{OperationResult: converted}}, nil
839 case *OperationResult:
840 if m == nil {
841 return nil, ErrInvalidMessage
842 }
843 return encodeServerEnvelope(*m, maxChunkBytes)
844 case StreamOpened:
845 if m.Epoch == 0 {
846 return nil, ErrInvalidMessage
847 }
848 if err := m.Connection.Validate(); err != nil {
849 return nil, ErrInvalidMessage
850 }
851 if err := m.Stream.Validate(); err != nil {
852 return nil, ErrInvalidMessage
853 }
854 return &wire.BrokerServerEnvelope{Payload: &wire.BrokerServerEnvelope_StreamOpened{StreamOpened: &wire.StreamOpened{
855 Ref: refToWire(m.Epoch, m.Connection, m.Stream),
856 }}}, nil
857 case *StreamOpened:
858 if m == nil {
859 return nil, ErrInvalidMessage
860 }
861 return encodeServerEnvelope(*m, maxChunkBytes)
862 case ServerStreamData:
863 converted, err := serverStreamDataToWire(m, maxChunkBytes)
864 if err != nil {
865 return nil, err
866 }
867 return &wire.BrokerServerEnvelope{Payload: &wire.BrokerServerEnvelope_ServerStreamData{ServerStreamData: converted}}, nil
868 case *ServerStreamData:
869 if m == nil {
870 return nil, ErrInvalidMessage
871 }
872 return encodeServerEnvelope(*m, maxChunkBytes)
873 case StreamWindowUpdate:
874 converted, err := streamWindowUpdateToWire(m)
875 if err != nil {
876 return nil, err
877 }
878 return &wire.BrokerServerEnvelope{Payload: &wire.BrokerServerEnvelope_StreamWindowUpdate{StreamWindowUpdate: converted}}, nil
879 case *StreamWindowUpdate:
880 if m == nil {
881 return nil, ErrInvalidMessage
882 }
883 return encodeServerEnvelope(*m, maxChunkBytes)
884 case StreamClosed:
885 converted, err := streamClosedToWire(m)
886 if err != nil {
887 return nil, err
888 }
889 return &wire.BrokerServerEnvelope{Payload: &wire.BrokerServerEnvelope_StreamClosed{StreamClosed: converted}}, nil
890 case *StreamClosed:
891 if m == nil {
892 return nil, ErrInvalidMessage
893 }
894 return encodeServerEnvelope(*m, maxChunkBytes)
895 case Progress:
896 converted, err := progressToWire(m)
897 if err != nil {
898 return nil, err
899 }
900 return &wire.BrokerServerEnvelope{Payload: &wire.BrokerServerEnvelope_Progress{Progress: converted}}, nil
901 case *Progress:
902 if m == nil {
903 return nil, ErrInvalidMessage
904 }
905 return encodeServerEnvelope(*m, maxChunkBytes)
906 case BrokerErrorMessage:
907 if m.Epoch == 0 {
908 return nil, ErrInvalidMessage
909 }
910 if err := m.Connection.Validate(); err != nil {
911 return nil, ErrInvalidMessage
912 }
913 if err := m.Error.validate(); err != nil {
914 return nil, err
915 }
916 return &wire.BrokerServerEnvelope{Payload: &wire.BrokerServerEnvelope_BrokerErrorMessage{BrokerErrorMessage: &wire.BrokerErrorMessage{
917 Scope: scopeToWire(m.Epoch, m.Connection), Error: errorDetailToWire(m.Error),
918 }}}, nil
919 case *BrokerErrorMessage:
920 if m == nil {
921 return nil, ErrInvalidMessage
922 }
923 return encodeServerEnvelope(*m, maxChunkBytes)
924 case Shutdown:
925 converted, err := shutdownToWire(m)
926 if err != nil {
927 return nil, err
928 }
929 return &wire.BrokerServerEnvelope{Payload: &wire.BrokerServerEnvelope_Shutdown{Shutdown: converted}}, nil
930 case *Shutdown:
931 if m == nil {
932 return nil, ErrInvalidMessage
933 }
934 return encodeServerEnvelope(*m, maxChunkBytes)
935 case PreviewPublication:
936 converted, err := previewPublicationToWire(m)
937 if err != nil {
938 return nil, err
939 }
940 return &wire.BrokerServerEnvelope{Payload: &wire.BrokerServerEnvelope_PreviewPublication{PreviewPublication: converted}}, nil
941 case *PreviewPublication:
942 if m == nil {
943 return nil, ErrInvalidMessage
944 }
945 return encodeServerEnvelope(*m, maxChunkBytes)
946 default:
947 return nil, ErrWrongDirection
948 }
949}
950
951func decodeServerEnvelope(envelope *wire.BrokerServerEnvelope, maxChunkBytes uint64) (ServerMessage, error) {
952 if envelope == nil {
953 return nil, ErrInvalidMessage
954 }
955 switch payload := envelope.Payload.(type) {
956 case *wire.BrokerServerEnvelope_Registered:
957 if payload.Registered == nil {
958 return nil, ErrInvalidMessage
959 }
960 epoch, connection, err := scopeFromWire(payload.Registered.GetScope())
961 if err != nil {
962 return nil, ErrInvalidMessage
963 }
964 if len(payload.Registered.GetBuild()) > 128 {
965 return nil, ErrInvalidMessage
966 }
967 return Registered{Epoch: epoch, Connection: connection, Build: payload.Registered.GetBuild(), Retiring: payload.Registered.GetRetiring()}, nil
968 case *wire.BrokerServerEnvelope_SnapshotPart:
969 return snapshotPartFromWire(payload.SnapshotPart)
970 case *wire.BrokerServerEnvelope_OperationResult:
971 return operationResultFromWire(payload.OperationResult)
972 case *wire.BrokerServerEnvelope_StreamOpened:
973 if payload.StreamOpened == nil {
974 return nil, ErrInvalidMessage
975 }
976 epoch, connection, stream, err := refFromWire(payload.StreamOpened.GetRef())
977 if err != nil {
978 return nil, ErrInvalidMessage
979 }
980 return StreamOpened{Epoch: epoch, Connection: connection, Stream: stream}, nil
981 case *wire.BrokerServerEnvelope_ServerStreamData:
982 return serverStreamDataFromWire(payload.ServerStreamData, maxChunkBytes)
983 case *wire.BrokerServerEnvelope_StreamClosed:
984 return streamClosedFromWire(payload.StreamClosed)
985 case *wire.BrokerServerEnvelope_StreamWindowUpdate:
986 return streamWindowUpdateFromWire(payload.StreamWindowUpdate)
987 case *wire.BrokerServerEnvelope_Progress:
988 return progressFromWire(payload.Progress)
989 case *wire.BrokerServerEnvelope_BrokerErrorMessage:
990 if payload.BrokerErrorMessage == nil {
991 return nil, ErrInvalidMessage
992 }
993 epoch, connection, err := scopeFromWire(payload.BrokerErrorMessage.GetScope())
994 if err != nil {
995 return nil, ErrInvalidMessage
996 }
997 detail, err := errorDetailFromWire(payload.BrokerErrorMessage.GetError())
998 if err != nil {
999 return nil, ErrInvalidMessage
1000 }
1001 return BrokerErrorMessage{Epoch: epoch, Connection: connection, Error: detail}, nil
1002 case *wire.BrokerServerEnvelope_Shutdown:
1003 return shutdownFromWire(payload.Shutdown)
1004 case *wire.BrokerServerEnvelope_PreviewPublication:
1005 return previewPublicationFromWire(payload.PreviewPublication)
1006 default:
1007 return nil, ErrWrongDirection
1008 }
1009}
1010
1011func snapshotPartToWire(m SnapshotPart) (*wire.SnapshotPart, error) {
1012 if m.Epoch == 0 {
1013 return nil, ErrInvalidMessage
1014 }
1015 if err := m.Connection.Validate(); err != nil {
1016 return nil, ErrInvalidMessage
1017 }
1018 if m.Revision == 0 {
1019 return nil, ErrInvalidMessage
1020 }
1021 if m.Part == nil {
1022 return nil, ErrInvalidMessage
1023 }
1024 out := &wire.SnapshotPart{
1025 Scope: scopeToWire(m.Epoch, m.Connection),
1026 Generation: m.Generation, Revision: uint64(m.Revision), Index: m.Index,
1027 }
1028 switch part := m.Part.(type) {
1029 case SnapshotBegin:
1030 if part.HostCount > ports.BrokerMaxDaemonsPerSnapshot ||
1031 part.SessionCount > ports.BrokerMaxDaemonsPerSnapshot*ports.BrokerMaxSessionsPerHost ||
1032 part.TombstoneCount > ports.BrokerMaxTombstones {
1033 return nil, ErrTooLarge
1034 }
1035 if part.LocalPresent && part.HostCount == 0 {
1036 return nil, ErrInvalidMessage
1037 }
1038 out.Part = &wire.SnapshotPart_Begin{Begin: &wire.SnapshotBegin{
1039 HostCount: part.HostCount, SessionCount: part.SessionCount, TombstoneCount: part.TombstoneCount,
1040 LocalPresent: part.LocalPresent,
1041 }}
1042 case *SnapshotBegin:
1043 if part == nil {
1044 return nil, ErrInvalidMessage
1045 }
1046 return snapshotPartToWire(SnapshotPart{Epoch: m.Epoch, Connection: m.Connection, Generation: m.Generation, Revision: m.Revision, Index: m.Index, Part: *part})
1047 case SnapshotDaemonPart:
1048 converted, err := snapshotDaemonToWire(part)
1049 if err != nil {
1050 return nil, err
1051 }
1052 out.Part = &wire.SnapshotPart_Daemon{Daemon: converted}
1053 case *SnapshotDaemonPart:
1054 if part == nil {
1055 return nil, ErrInvalidMessage
1056 }
1057 return snapshotPartToWire(SnapshotPart{Epoch: m.Epoch, Connection: m.Connection, Generation: m.Generation, Revision: m.Revision, Index: m.Index, Part: *part})
1058 case SnapshotSessionPart:
1059 converted, err := snapshotSessionToWire(part)
1060 if err != nil {
1061 return nil, err
1062 }
1063 out.Part = &wire.SnapshotPart_Session{Session: converted}
1064 case *SnapshotSessionPart:
1065 if part == nil {
1066 return nil, ErrInvalidMessage
1067 }
1068 return snapshotPartToWire(SnapshotPart{Epoch: m.Epoch, Connection: m.Connection, Generation: m.Generation, Revision: m.Revision, Index: m.Index, Part: *part})
1069 case SnapshotTombstonePart:
1070 if part.RetiredRevision == 0 {
1071 return nil, ErrInvalidMessage
1072 }
1073 if err := part.Registration.Validate(); err != nil {
1074 return nil, ErrInvalidMessage
1075 }
1076 out.Part = &wire.SnapshotPart_Tombstone{Tombstone: &wire.SnapshotTombstone{
1077 TombstoneIndex: part.TombstoneIndex,
1078 Registration: registrationToWire(part.Registration),
1079 RetiredRevision: uint64(part.RetiredRevision),
1080 }}
1081 case *SnapshotTombstonePart:
1082 if part == nil {
1083 return nil, ErrInvalidMessage
1084 }
1085 return snapshotPartToWire(SnapshotPart{Epoch: m.Epoch, Connection: m.Connection, Generation: m.Generation, Revision: m.Revision, Index: m.Index, Part: *part})
1086 case SnapshotEnd:
1087 out.Part = &wire.SnapshotPart_End{End: &wire.SnapshotEnd{}}
1088 case *SnapshotEnd:
1089 if part == nil {
1090 return nil, ErrInvalidMessage
1091 }
1092 return snapshotPartToWire(SnapshotPart{Epoch: m.Epoch, Connection: m.Connection, Generation: m.Generation, Revision: m.Revision, Index: m.Index, Part: *part})
1093 default:
1094 return nil, ErrWrongDirection
1095 }
1096 return out, nil
1097}
1098
1099func snapshotPartFromWire(message *wire.SnapshotPart) (SnapshotPart, error) {
1100 var out SnapshotPart
1101 if message == nil {
1102 return out, ErrInvalidMessage
1103 }
1104 epoch, connection, err := scopeFromWire(message.GetScope())
1105 if err != nil {
1106 return SnapshotPart{}, ErrInvalidMessage
1107 }
1108 if message.GetRevision() == 0 {
1109 return SnapshotPart{}, ErrInvalidMessage
1110 }
1111 part := SnapshotPart{
1112 Epoch: epoch, Connection: connection,
1113 Generation: message.GetGeneration(), Revision: ports.BrokerRevision(message.GetRevision()),
1114 Index: message.GetIndex(),
1115 }
1116 switch payload := message.GetPart().(type) {
1117 case *wire.SnapshotPart_Begin:
1118 if payload.Begin == nil {
1119 return SnapshotPart{}, ErrInvalidMessage
1120 }
1121 if payload.Begin.GetHostCount() > ports.BrokerMaxDaemonsPerSnapshot ||
1122 payload.Begin.GetSessionCount() > ports.BrokerMaxDaemonsPerSnapshot*ports.BrokerMaxSessionsPerHost ||
1123 payload.Begin.GetTombstoneCount() > ports.BrokerMaxTombstones {
1124 return SnapshotPart{}, ErrTooLarge
1125 }
1126 if payload.Begin.GetLocalPresent() && payload.Begin.GetHostCount() == 0 {
1127 return SnapshotPart{}, ErrInvalidMessage
1128 }
1129 part.Part = SnapshotBegin{
1130 HostCount: payload.Begin.GetHostCount(),
1131 SessionCount: payload.Begin.GetSessionCount(),
1132 TombstoneCount: payload.Begin.GetTombstoneCount(),
1133 LocalPresent: payload.Begin.GetLocalPresent(),
1134 }
1135 case *wire.SnapshotPart_Daemon:
1136 if payload.Daemon == nil {
1137 return SnapshotPart{}, ErrInvalidMessage
1138 }
1139 daemon, index, sessionCount, err := snapshotDaemonFromWire(payload.Daemon)
1140 if err != nil {
1141 return SnapshotPart{}, err
1142 }
1143 part.Part = SnapshotDaemonPart{HostIndex: index, Daemon: daemon, SessionCount: sessionCount}
1144 case *wire.SnapshotPart_Session:
1145 if payload.Session == nil {
1146 return SnapshotPart{}, ErrInvalidMessage
1147 }
1148 session, hostIndex, sessionIndex, local, err := snapshotSessionFromWire(payload.Session)
1149 if err != nil {
1150 return SnapshotPart{}, err
1151 }
1152 part.Part = SnapshotSessionPart{HostIndex: hostIndex, SessionIndex: sessionIndex, Local: local, Session: session}
1153 case *wire.SnapshotPart_Tombstone:
1154 if payload.Tombstone == nil {
1155 return SnapshotPart{}, ErrInvalidMessage
1156 }
1157 registration, err := registrationFromWire(payload.Tombstone.GetRegistration())
1158 if err != nil {
1159 return SnapshotPart{}, ErrInvalidMessage
1160 }
1161 if payload.Tombstone.GetRetiredRevision() == 0 {
1162 return SnapshotPart{}, ErrInvalidMessage
1163 }
1164 part.Part = SnapshotTombstonePart{
1165 TombstoneIndex: payload.Tombstone.GetTombstoneIndex(),
1166 Registration: registration,
1167 RetiredRevision: ports.BrokerRevision(payload.Tombstone.GetRetiredRevision()),
1168 }
1169 case *wire.SnapshotPart_End:
1170 if payload.End == nil {
1171 return SnapshotPart{}, ErrInvalidMessage
1172 }
1173 part.Part = SnapshotEnd{}
1174 default:
1175 return SnapshotPart{}, ErrWrongDirection
1176 }
1177 return part, nil
1178}
1179
1180// snapshotDaemonToWire converts one daemon part losslessly. It refuses a
1181// malformed local/remote authority pairing, a partial identity, an
1182// out-of-range enum or rank, a session inventory that could not be durable,
1183// inline sessions (they travel separately), and any index or count above
1184// the snapshot bounds.
1185func snapshotDaemonToWire(part SnapshotDaemonPart) (*wire.SnapshotDaemon, error) {
1186 if part.HostIndex >= ports.BrokerMaxDaemonsPerSnapshot {
1187 return nil, ErrTooLarge
1188 }
1189 daemon := part.Daemon
1190 if err := validateSnapshotDaemonFields(daemon); err != nil {
1191 return nil, err
1192 }
1193 if err := daemon.Validate(); err != nil {
1194 return nil, ErrInvalidMessage
1195 }
1196 if !daemon.Local {
1197 if err := ports.ValidateDurableHostProjection(daemon); err != nil {
1198 return nil, ErrInvalidMessage
1199 }
1200 }
1201 if len(daemon.Sessions) != 0 {
1202 return nil, ErrInvalidMessage
1203 }
1204 if part.SessionCount > ports.BrokerMaxSessionsPerHost {
1205 return nil, ErrTooLarge
1206 }
1207 lastAttemptSeconds, lastAttemptNanos := brokerTimeToWire(daemon.LastAttempt)
1208 lastSuccessSeconds, lastSuccessNanos := brokerTimeToWire(daemon.LastSuccess)
1209 nextDueSeconds, nextDueNanos := brokerTimeToWire(daemon.NextDue)
1210 var registration *wire.RemoteRegistration
1211 if !daemon.Local {
1212 registration = registrationToWire(daemon.Registration)
1213 }
1214 var incarnation []byte
1215 if !daemon.Incarnation.IsZero() {
1216 incarnation = append([]byte(nil), daemon.Incarnation[:]...)
1217 }
1218 return &wire.SnapshotDaemon{
1219 HostIndex: part.HostIndex, Local: daemon.Local,
1220 Endpoint: daemon.Endpoint, DisplayOrigin: daemon.DisplayOrigin,
1221 Rank: uint32(daemon.Rank), Registration: registration,
1222 Policy: policyToWire(daemon.Policy),
1223 DaemonIdentity: string(daemon.Identity), DaemonIncarnation: incarnation,
1224 ProtocolVersion: uint32(daemon.ProtocolVersion), Capabilities: daemon.Capabilities,
1225 Availability: uint32(daemon.Availability), Checking: daemon.Checking,
1226 LastAttempt: &wire.BrokerTimestamp{Seconds: lastAttemptSeconds, Nanos: lastAttemptNanos},
1227 LastSuccess: &wire.BrokerTimestamp{Seconds: lastSuccessSeconds, Nanos: lastSuccessNanos},
1228 NextDue: &wire.BrokerTimestamp{Seconds: nextDueSeconds, Nanos: nextDueNanos},
1229 ConsecutiveFailures: uint64(daemon.ConsecutiveFailures), FailureEpisode: daemon.FailureEpisode,
1230 FailureKind: uint32(daemon.LastFailure.Kind),
1231 InventoryKnown: daemon.InventoryKnown, SessionCount: part.SessionCount,
1232 }, nil
1233}
1234
1235func snapshotDaemonFromWire(message *wire.SnapshotDaemon) (ports.BrokerDaemonObservation, uint32, uint32, error) {
1236 var daemon ports.BrokerDaemonObservation
1237 if message == nil {
1238 return daemon, 0, 0, ErrInvalidMessage
1239 }
1240 if message.GetHostIndex() >= ports.BrokerMaxDaemonsPerSnapshot {
1241 return daemon, 0, 0, ErrTooLarge
1242 }
1243 local := message.GetLocal()
1244 endpoint := message.GetEndpoint()
1245 origin := message.GetDisplayOrigin()
1246 if err := validateBrokerOrigin(origin); err != nil {
1247 return daemon, 0, 0, err
1248 }
1249 var registration domain.RemoteRegistration
1250 if local {
1251 if message.GetRegistration() != nil || endpoint != "" {
1252 return daemon, 0, 0, ErrInvalidMessage
1253 }
1254 } else {
1255 if err := validateBrokerEndpoint(endpoint); err != nil {
1256 return daemon, 0, 0, err
1257 }
1258 converted, err := registrationFromWire(message.GetRegistration())
1259 if err != nil {
1260 return daemon, 0, 0, ErrInvalidMessage
1261 }
1262 if endpoint != converted.Endpoint {
1263 return daemon, 0, 0, ErrInvalidMessage
1264 }
1265 registration = converted
1266 }
1267 policy, err := policyFromWire(message.GetPolicy())
1268 if err != nil {
1269 return daemon, 0, 0, ErrInvalidMessage
1270 }
1271 raw := message.GetDaemonIncarnation()
1272 if len(raw) != 0 && len(raw) != len(daemon.Incarnation) {
1273 return daemon, 0, 0, errConvertRange
1274 }
1275 if len(raw) == len(daemon.Incarnation) {
1276 copy(daemon.Incarnation[:], raw)
1277 }
1278 protocolVersion, err := brokerEnum16[uint16](message.GetProtocolVersion())
1279 if err != nil {
1280 return daemon, 0, 0, ErrInvalidMessage
1281 }
1282 availability, err := brokerEnum8[domain.RemoteAvailability](message.GetAvailability())
1283 if err != nil {
1284 return daemon, 0, 0, ErrInvalidMessage
1285 }
1286 if availability < domain.RemoteAvailabilityUnknown || availability > domain.RemoteAvailabilityNoDaemon {
1287 return daemon, 0, 0, ErrInvalidMessage
1288 }
1289 failureKind, err := brokerEnum8[domain.RemoteFailureKind](message.GetFailureKind())
1290 if err != nil {
1291 return daemon, 0, 0, ErrInvalidMessage
1292 }
1293 if failureKind > domain.RemoteFailureInvalidResponse {
1294 return daemon, 0, 0, ErrInvalidMessage
1295 }
1296 lastAttempt, err := brokerTimeFromWire(message.GetLastAttempt().GetSeconds(), message.GetLastAttempt().GetNanos())
1297 if err != nil {
1298 return daemon, 0, 0, ErrInvalidMessage
1299 }
1300 lastSuccess, err := brokerTimeFromWire(message.GetLastSuccess().GetSeconds(), message.GetLastSuccess().GetNanos())
1301 if err != nil {
1302 return daemon, 0, 0, ErrInvalidMessage
1303 }
1304 nextDue, err := brokerTimeFromWire(message.GetNextDue().GetSeconds(), message.GetNextDue().GetNanos())
1305 if err != nil {
1306 return daemon, 0, 0, ErrInvalidMessage
1307 }
1308 if message.GetConsecutiveFailures() > math.MaxUint32 {
1309 return daemon, 0, 0, errConvertRange
1310 }
1311 if message.GetSessionCount() > ports.BrokerMaxSessionsPerHost {
1312 return daemon, 0, 0, ErrTooLarge
1313 }
1314 daemon = ports.BrokerDaemonObservation{
1315 Local: local, Endpoint: endpoint, DisplayOrigin: origin, Rank: int(message.GetRank()),
1316 Registration: registration, Policy: policy,
1317 Identity: ports.BrokerDaemonIdentity(message.GetDaemonIdentity()), Incarnation: daemon.Incarnation,
1318 ProtocolVersion: protocolVersion, Capabilities: message.GetCapabilities(),
1319 Availability: availability, Checking: message.GetChecking(),
1320 LastAttempt: lastAttempt, LastSuccess: lastSuccess, NextDue: nextDue,
1321 ConsecutiveFailures: uint(message.GetConsecutiveFailures()), FailureEpisode: message.GetFailureEpisode(),
1322 LastFailure: domain.RemoteFailure{Kind: failureKind},
1323 InventoryKnown: message.GetInventoryKnown(),
1324 Sessions: []catalogue.RemoteCatalogSession{},
1325 }
1326 if err := validateSnapshotDaemonFields(daemon); err != nil {
1327 return daemon, 0, 0, err
1328 }
1329 if err := daemon.Validate(); err != nil {
1330 return daemon, 0, 0, ErrInvalidMessage
1331 }
1332 return daemon, message.GetHostIndex(), message.GetSessionCount(), nil
1333}
1334
1335func snapshotSessionToWire(part SnapshotSessionPart) (*wire.SnapshotSession, error) {
1336 if part.Local {
1337 if part.HostIndex != 0 {
1338 return nil, ErrInvalidMessage
1339 }
1340 } else if part.HostIndex >= ports.BrokerMaxDaemonsPerSnapshot {
1341 return nil, ErrTooLarge
1342 }
1343 if part.SessionIndex >= ports.BrokerMaxSessionsPerHost {
1344 return nil, ErrTooLarge
1345 }
1346 if err := validateCatalogSession(part.Session); err != nil {
1347 return nil, err
1348 }
1349 return &wire.SnapshotSession{
1350 HostIndex: part.HostIndex, SessionIndex: part.SessionIndex,
1351 Local: part.Local,
1352 Session: catalogSessionToWire(part.Session),
1353 }, nil
1354}
1355
1356func snapshotSessionFromWire(message *wire.SnapshotSession) (catalogue.RemoteCatalogSession, uint32, uint32, bool, error) {
1357 var session catalogue.RemoteCatalogSession
1358 if message == nil {
1359 return session, 0, 0, false, ErrInvalidMessage
1360 }
1361 local := message.GetLocal()
1362 if local {
1363 if message.GetHostIndex() != 0 {
1364 return session, 0, 0, false, ErrInvalidMessage
1365 }
1366 } else if message.GetHostIndex() >= ports.BrokerMaxDaemonsPerSnapshot {
1367 return session, 0, 0, false, ErrTooLarge
1368 }
1369 if message.GetSessionIndex() >= ports.BrokerMaxSessionsPerHost {
1370 return session, 0, 0, false, ErrTooLarge
1371 }
1372 converted, err := catalogSessionFromWire(message.GetSession())
1373 if err != nil {
1374 return session, 0, 0, false, err
1375 }
1376 return converted, message.GetHostIndex(), message.GetSessionIndex(), local, nil
1377}
1378
1379func operationResultToWire(m OperationResult) (*wire.OperationResult, error) {
1380 if m.Epoch == 0 {
1381 return nil, ErrInvalidMessage
1382 }
1383 if err := m.Connection.Validate(); err != nil {
1384 return nil, ErrInvalidMessage
1385 }
1386 if err := m.Operation.Validate(); err != nil {
1387 return nil, ErrInvalidMessage
1388 }
1389 if err := m.Outcome.Validate(); err != nil {
1390 return nil, ErrInvalidMessage
1391 }
1392 var outcome uint32
1393 switch m.Outcome {
1394 case ports.BrokerOutcomeOK:
1395 outcome = 1
1396 case ports.BrokerOutcomeFailed:
1397 outcome = 2
1398 case ports.BrokerOutcomeUnknown:
1399 outcome = 3
1400 default:
1401 return nil, ErrInvalidMessage
1402 }
1403 result := ports.BrokerOperationResult{Operation: m.Operation, Outcome: m.Outcome}
1404 if m.HasError {
1405 if err := m.Error.validate(); err != nil {
1406 return nil, err
1407 }
1408 result.Text = m.Error.Text
1409 } else if m.Error != (ErrorDetail{}) {
1410 // HasError=false is exactly the absent detail: a nonzero
1411 // ErrorDetail would otherwise be silently dropped instead of
1412 // travelling, so the whole result fails closed.
1413 return nil, ErrInvalidMessage
1414 }
1415 if err := result.Validate(); err != nil {
1416 return nil, ErrInvalidMessage
1417 }
1418 hasRegistration := m.Registration != (domain.RemoteRegistration{})
1419 // The presence rules that hold without knowing the request kind: removal
1420 // and registration never travel together, a failed or outcome-unknown
1421 // result carries neither authority and must carry its typed error, and a
1422 // successful result carries no error. The per-kind rule (an
1423 // AddHost/UpdateHostPolicy success needs a registration, a RemoveHost
1424 // success must not) is enforced where the pending request is known; the
1425 // stateless codec has no such state.
1426 if err := validateResultAuthority(m.Outcome, m.Removed, hasRegistration, m.HasError); err != nil {
1427 return nil, err
1428 }
1429 var registration *wire.RemoteRegistration
1430 if hasRegistration {
1431 if err := m.Registration.Validate(); err != nil {
1432 return nil, ErrInvalidMessage
1433 }
1434 registration = registrationToWire(m.Registration)
1435 }
1436 out := &wire.OperationResult{
1437 Scope: scopeToWire(m.Epoch, m.Connection),
1438 OperationId: operationToWire(m.Operation),
1439 Outcome: outcome, Removed: m.Removed,
1440 Registration: registration,
1441 }
1442 if m.HasError {
1443 out.Error = errorDetailToWire(m.Error)
1444 }
1445 return out, nil
1446}
1447
1448func operationResultFromWire(message *wire.OperationResult) (OperationResult, error) {
1449 var out OperationResult
1450 if message == nil {
1451 return out, ErrInvalidMessage
1452 }
1453 epoch, connection, err := scopeFromWire(message.GetScope())
1454 if err != nil {
1455 return OperationResult{}, ErrInvalidMessage
1456 }
1457 operation, err := operationFromWire(message.GetOperationId())
1458 if err != nil {
1459 return OperationResult{}, ErrInvalidMessage
1460 }
1461 var outcome ports.BrokerMutationOutcome
1462 switch message.GetOutcome() {
1463 case 1:
1464 outcome = ports.BrokerOutcomeOK
1465 case 2:
1466 outcome = ports.BrokerOutcomeFailed
1467 case 3:
1468 outcome = ports.BrokerOutcomeUnknown
1469 default:
1470 return OperationResult{}, ErrInvalidMessage
1471 }
1472 var registration domain.RemoteRegistration
1473 if message.GetRegistration() != nil {
1474 converted, err := registrationFromWire(message.GetRegistration())
1475 if err != nil {
1476 return OperationResult{}, ErrInvalidMessage
1477 }
1478 registration = converted
1479 }
1480 result := OperationResult{
1481 Epoch: epoch, Connection: connection, Operation: operation,
1482 Outcome: outcome, Removed: message.GetRemoved(),
1483 Registration: registration,
1484 }
1485 if message.GetError() != nil {
1486 detail, err := errorDetailFromWire(message.GetError())
1487 if err != nil {
1488 return OperationResult{}, ErrInvalidMessage
1489 }
1490 result.Error = detail
1491 result.HasError = true
1492 }
1493 if err := validateResultAuthority(result.Outcome, result.Removed, registration != (domain.RemoteRegistration{}), result.HasError); err != nil {
1494 return OperationResult{}, err
1495 }
1496 semantic := ports.BrokerOperationResult{Operation: result.Operation, Outcome: result.Outcome}
1497 if result.HasError {
1498 semantic.Text = result.Error.Text
1499 }
1500 if err := semantic.Validate(); err != nil {
1501 return OperationResult{}, ErrInvalidMessage
1502 }
1503 return result, nil
1504}
1505
1506func serverStreamDataToWire(m ServerStreamData, maxChunkBytes uint64) (*wire.ServerStreamData, error) {
1507 if m.Epoch == 0 {
1508 return nil, ErrInvalidMessage
1509 }
1510 if err := m.Connection.Validate(); err != nil {
1511 return nil, ErrInvalidMessage
1512 }
1513 if err := m.Stream.Validate(); err != nil {
1514 return nil, ErrInvalidMessage
1515 }
1516 if uint64(len(m.Data)) > maxChunkBytes {
1517 return nil, ErrTooLarge
1518 }
1519 if uint64(len(m.Data)) > MaxStreamChunkBytes {
1520 return nil, ErrTooLarge
1521 }
1522 if len(m.Data) == 0 {
1523 return nil, ErrInvalidMessage
1524 }
1525 return &wire.ServerStreamData{
1526 Ref: refToWire(m.Epoch, m.Connection, m.Stream),
1527 Data: append([]byte(nil), m.Data...),
1528 }, nil
1529}
1530
1531func serverStreamDataFromWire(message *wire.ServerStreamData, maxChunkBytes uint64) (ServerStreamData, error) {
1532 var out ServerStreamData
1533 if message == nil {
1534 return out, ErrInvalidMessage
1535 }
1536 epoch, connection, stream, err := refFromWire(message.GetRef())
1537 if err != nil {
1538 return ServerStreamData{}, ErrInvalidMessage
1539 }
1540 data := message.GetData()
1541 if uint64(len(data)) > maxChunkBytes {
1542 return ServerStreamData{}, ErrTooLarge
1543 }
1544 if uint64(len(data)) > MaxStreamChunkBytes {
1545 return ServerStreamData{}, ErrTooLarge
1546 }
1547 if len(data) == 0 {
1548 return ServerStreamData{}, ErrInvalidMessage
1549 }
1550 // proto.Unmarshal copies bytes fields out of the payload, so the caller
1551 // owns data without another copy.
1552 return ServerStreamData{Epoch: epoch, Connection: connection, Stream: stream, Data: data}, nil
1553}
1554
1555func streamClosedToWire(m StreamClosed) (*wire.StreamClosed, error) {
1556 if m.Epoch == 0 {
1557 return nil, ErrInvalidMessage
1558 }
1559 if err := m.Connection.Validate(); err != nil {
1560 return nil, ErrInvalidMessage
1561 }
1562 if err := m.Stream.Validate(); err != nil {
1563 return nil, ErrInvalidMessage
1564 }
1565 if m.HasError {
1566 if err := m.Error.validate(); err != nil {
1567 return nil, err
1568 }
1569 } else if m.Error != (ErrorDetail{}) {
1570 // An orderly close carries no detail at all; a nonzero one under
1571 // HasError=false is a caller bug that must fail closed.
1572 return nil, ErrInvalidMessage
1573 }
1574 out := &wire.StreamClosed{Ref: refToWire(m.Epoch, m.Connection, m.Stream)}
1575 if m.HasError {
1576 out.Error = errorDetailToWire(m.Error)
1577 }
1578 return out, nil
1579}
1580
1581func streamClosedFromWire(message *wire.StreamClosed) (StreamClosed, error) {
1582 var out StreamClosed
1583 if message == nil {
1584 return out, ErrInvalidMessage
1585 }
1586 epoch, connection, stream, err := refFromWire(message.GetRef())
1587 if err != nil {
1588 return StreamClosed{}, ErrInvalidMessage
1589 }
1590 result := StreamClosed{Epoch: epoch, Connection: connection, Stream: stream}
1591 if message.GetError() != nil {
1592 detail, err := errorDetailFromWire(message.GetError())
1593 if err != nil {
1594 return StreamClosed{}, ErrInvalidMessage
1595 }
1596 result.Error = detail
1597 result.HasError = true
1598 }
1599 return result, nil
1600}
1601
1602func progressToWire(m Progress) (*wire.Progress, error) {
1603 if m.Epoch == 0 {
1604 return nil, ErrInvalidMessage
1605 }
1606 if err := m.Connection.Validate(); err != nil {
1607 return nil, ErrInvalidMessage
1608 }
1609 if err := m.Stream.Validate(); err != nil {
1610 return nil, ErrInvalidMessage
1611 }
1612 if err := m.Phase.validate(); err != nil {
1613 return nil, ErrInvalidMessage
1614 }
1615 if err := validateBrokerDisplayText(m.Text, ports.BrokerMaxErrorBytes); err != nil {
1616 return nil, err
1617 }
1618 return &wire.Progress{
1619 Ref: refToWire(m.Epoch, m.Connection, m.Stream),
1620 Phase: uint32(m.Phase), Text: m.Text,
1621 }, nil
1622}
1623
1624func progressFromWire(message *wire.Progress) (Progress, error) {
1625 var out Progress
1626 if message == nil {
1627 return out, ErrInvalidMessage
1628 }
1629 epoch, connection, stream, err := refFromWire(message.GetRef())
1630 if err != nil {
1631 return Progress{}, ErrInvalidMessage
1632 }
1633 phase := BrokerProgressPhase(message.GetPhase())
1634 if err := phase.validate(); err != nil {
1635 return Progress{}, ErrInvalidMessage
1636 }
1637 if err := validateBrokerDisplayText(message.GetText(), ports.BrokerMaxErrorBytes); err != nil {
1638 return Progress{}, err
1639 }
1640 return Progress{Epoch: epoch, Connection: connection, Stream: stream, Phase: phase, Text: message.GetText()}, nil
1641}
1642
1643func shutdownToWire(m Shutdown) (*wire.Shutdown, error) {
1644 if m.Epoch == 0 {
1645 return nil, ErrInvalidMessage
1646 }
1647 if err := m.Connection.Validate(); err != nil {
1648 return nil, ErrInvalidMessage
1649 }
1650 if err := m.Reason.validate(); err != nil {
1651 return nil, ErrInvalidMessage
1652 }
1653 if err := validateBrokerDisplayText(m.Text, ports.BrokerMaxErrorBytes); err != nil {
1654 return nil, err
1655 }
1656 var reason uint32
1657 switch m.Reason {
1658 case ShutdownIdleExit:
1659 reason = 1
1660 case ShutdownTerminating:
1661 reason = 2
1662 }
1663 return &wire.Shutdown{
1664 Scope: scopeToWire(m.Epoch, m.Connection),
1665 Reason: reason, Text: m.Text,
1666 }, nil
1667}
1668
1669func shutdownFromWire(message *wire.Shutdown) (Shutdown, error) {
1670 var out Shutdown
1671 if message == nil {
1672 return out, ErrInvalidMessage
1673 }
1674 epoch, connection, err := scopeFromWire(message.GetScope())
1675 if err != nil {
1676 return Shutdown{}, ErrInvalidMessage
1677 }
1678 var reason ShutdownReason
1679 switch message.GetReason() {
1680 case 1:
1681 reason = ShutdownIdleExit
1682 case 2:
1683 reason = ShutdownTerminating
1684 default:
1685 return Shutdown{}, ErrInvalidMessage
1686 }
1687 if err := validateBrokerDisplayText(message.GetText(), ports.BrokerMaxErrorBytes); err != nil {
1688 return Shutdown{}, err
1689 }
1690 return Shutdown{Epoch: epoch, Connection: connection, Reason: reason, Text: message.GetText()}, nil
1691}