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