1package wire
2
3import (
4 "bytes"
5 "errors"
6 "testing"
7
8 "google.golang.org/protobuf/encoding/protowire"
9 "google.golang.org/protobuf/proto"
10 "google.golang.org/protobuf/reflect/protoreflect"
11)
12
13// The broker conversation is separate from the session conversation: its
14// tags are disjoint from every session tag, both directions are closed
15// oneof unions, and the strict scanner treats them exactly like the
16// session envelopes.
17
18// brokerClientTags is the frozen client tag inventory.
19var brokerClientTags = map[string]protoreflect.FieldNumber{
20 "register": 101, "subscribe": 102, "resync": 103, "unsubscribe": 104,
21 "add_host": 105, "remove_host": 106, "reconcile": 107, "open_stream": 108,
22 "client_stream_data": 109, "close_stream": 110, "update_host_policy": 111,
23 "start_preview": 112, "cancel_preview": 113,
24}
25
26// brokerServerTags is the frozen server tag inventory.
27var brokerServerTags = map[string]protoreflect.FieldNumber{
28 "registered": 201, "snapshot_part": 202, "operation_result": 203,
29 "stream_opened": 204, "server_stream_data": 205, "stream_closed": 206,
30 "progress": 207, "broker_error_message": 208, "shutdown": 209,
31 "preview_publication": 210,
32}
33
34func oneofTags(t *testing.T, message proto.Message) map[string]protoreflect.FieldNumber {
35 t.Helper()
36 descriptor := message.ProtoReflect().Descriptor()
37 if descriptor.Oneofs().Len() != 1 {
38 t.Fatalf("%v has %d oneofs, want exactly the payload oneof", descriptor.FullName(), descriptor.Oneofs().Len())
39 }
40 oneof := descriptor.Oneofs().Get(0)
41 if string(oneof.Name()) != "payload" {
42 t.Fatalf("%v oneof %q is not the payload oneof", descriptor.FullName(), oneof.Name())
43 }
44 tags := make(map[string]protoreflect.FieldNumber, oneof.Fields().Len())
45 for i := range oneof.Fields().Len() {
46 field := oneof.Fields().Get(i)
47 tags[string(field.Name())] = field.Number()
48 }
49 return tags
50}
51
52// TestBrokerEnvelopeInventory proves both broker envelopes carry exactly
53// the frozen tag inventory and share no tag with the session envelopes or
54// with each other.
55func TestBrokerEnvelopeInventory(t *testing.T) {
56 client := oneofTags(t, &BrokerClientEnvelope{})
57 server := oneofTags(t, &BrokerServerEnvelope{})
58
59 if len(client) != len(brokerClientTags) {
60 t.Fatalf("broker client envelope has %d variants, want %d (%v)", len(client), len(brokerClientTags), client)
61 }
62 for name, number := range brokerClientTags {
63 if got, ok := client[name]; !ok || got != number {
64 t.Fatalf("broker client variant %q = %d (present %t), want %d", name, got, ok, number)
65 }
66 }
67 if len(server) != len(brokerServerTags) {
68 t.Fatalf("broker server envelope has %d variants, want %d (%v)", len(server), len(brokerServerTags), server)
69 }
70 for name, number := range brokerServerTags {
71 if got, ok := server[name]; !ok || got != number {
72 t.Fatalf("broker server variant %q = %d (present %t), want %d", name, got, ok, number)
73 }
74 }
75
76 sessionTags := map[protoreflect.FieldNumber]string{}
77 for _, envelope := range []proto.Message{&ClientEnvelope{}, &ServerEnvelope{}} {
78 for name, number := range oneofTags(t, envelope) {
79 sessionTags[number] = name
80 }
81 }
82 for name, number := range brokerClientTags {
83 if session, taken := sessionTags[number]; taken {
84 t.Fatalf("broker client tag %d (%s) collides with session variant %s", number, name, session)
85 }
86 }
87 clientByTag := map[protoreflect.FieldNumber]string{}
88 for name, number := range brokerClientTags {
89 clientByTag[number] = name
90 }
91 for name, number := range brokerServerTags {
92 if session, taken := sessionTags[number]; taken {
93 t.Fatalf("broker server tag %d (%s) collides with session variant %s", number, name, session)
94 }
95 if peer, taken := clientByTag[number]; taken {
96 t.Fatalf("broker server tag %d (%s) collides with broker client variant %s", number, name, peer)
97 }
98 }
99
100 for _, envelope := range []proto.Message{&BrokerClientEnvelope{}, &BrokerServerEnvelope{}} {
101 if !isEnvelopeDescriptor(envelope.ProtoReflect().Descriptor()) {
102 t.Fatalf("%v is not recognized as a directional envelope", envelope.ProtoReflect().Descriptor().FullName())
103 }
104 }
105 // Payload messages are never envelopes: only the two directions require
106 // exactly one variant.
107 for _, payload := range []proto.Message{&Subscribe{}, &SnapshotPart{}, &Registered{}} {
108 if isEnvelopeDescriptor(payload.ProtoReflect().Descriptor()) {
109 t.Fatalf("payload %v is wrongly recognized as a directional envelope", payload.ProtoReflect().Descriptor().FullName())
110 }
111 }
112}
113
114func brokerScope() *BrokerScope {
115 return &BrokerScope{BrokerEpoch: 7, ConnectionId: bytes.Repeat([]byte{0x11}, 16)}
116}
117
118func brokerRef() *BrokerStreamRef {
119 return &BrokerStreamRef{Scope: brokerScope(), StreamId: 3}
120}
121
122func brokerRegistration() *RemoteRegistration {
123 return &RemoteRegistration{Endpoint: "dev@host:22", Incarnation: bytes.Repeat([]byte{0x22}, 16), Generation: 9}
124}
125
126func brokerPolicy() *BrokerWirePolicy {
127 return &BrokerWirePolicy{
128 ProtocolVersion: 53, CatalogueVersion: 3, EnvironmentPolicy: 1,
129 Transport: "quic", Trust: "pinned", Launch: "agent", Isolation: "user",
130 }
131}
132
133func brokerCatalogTab() *BrokerCatalogTab {
134 return &BrokerCatalogTab{Id: "tab-1", Index: 0, Name: "work", Detail: "1 pane", Attention: true}
135}
136
137func brokerCatalogSession() *BrokerCatalogSession {
138 return &BrokerCatalogSession{
139 LifecycleId: bytes.Repeat([]byte{0x33}, 16), Name: "work", State: 1,
140 Ephemeral: true, Tabs: []*BrokerCatalogTab{brokerCatalogTab()}, Attached: true,
141 LastUsedSeq: 4, ActiveTabId: "tab-1", Reason: "refreshing",
142 }
143}
144
145// brokerClientSamples returns one populated envelope per client variant.
146// Every sample's final serialized field is deliberately multi-byte so a
147// one-byte truncation always lands inside a field.
148func brokerClientSamples() map[string]*BrokerClientEnvelope {
149 return map[string]*BrokerClientEnvelope{
150 "register": {Payload: &BrokerClientEnvelope_Register{Register: &Register{}}},
151 "subscribe": {Payload: &BrokerClientEnvelope_Subscribe{Subscribe: &Subscribe{Scope: brokerScope(), Generation: 1 << 40}}},
152 "resync": {Payload: &BrokerClientEnvelope_Resync{Resync: &Resync{Scope: brokerScope(), Generation: 1 << 41}}},
153 "unsubscribe": {Payload: &BrokerClientEnvelope_Unsubscribe{Unsubscribe: &Unsubscribe{Scope: brokerScope(), Generation: 1 << 42}}},
154 "add_host": {Payload: &BrokerClientEnvelope_AddHost{AddHost: &AddHost{
155 Scope: brokerScope(), OperationId: bytes.Repeat([]byte{0x44}, 16), Endpoint: "dev@host:22",
156 Policy: brokerPolicy(),
157 }}},
158 "remove_host": {Payload: &BrokerClientEnvelope_RemoveHost{RemoveHost: &RemoveHost{
159 Scope: brokerScope(), OperationId: bytes.Repeat([]byte{0x45}, 16), Registration: brokerRegistration(),
160 }}},
161 "update_host_policy": {Payload: &BrokerClientEnvelope_UpdateHostPolicy{UpdateHostPolicy: &UpdateHostPolicy{
162 Scope: brokerScope(), OperationId: bytes.Repeat([]byte{0x46}, 16),
163 Registration: brokerRegistration(), Policy: brokerPolicy(),
164 }}},
165 "reconcile": {Payload: &BrokerClientEnvelope_Reconcile{Reconcile: &Reconcile{
166 Scope: brokerScope(), Registration: brokerRegistration(),
167 }}},
168 "open_stream": {Payload: &BrokerClientEnvelope_OpenStream{OpenStream: &OpenStream{
169 Ref: brokerRef(), Purpose: 1, Local: false, Endpoint: "dev@host:22",
170 Registration: brokerRegistration(),
171 Target: &ExactTarget{LifecycleId: &LifecycleID{Value: bytes.Repeat([]byte{0x55}, 16)}, SessionName: "work"},
172 Env: []string{"TERM=xterm-256color", "LANG=C.UTF-8"},
173 Policy: brokerPolicy(),
174 }}},
175 "client_stream_data": {Payload: &BrokerClientEnvelope_ClientStreamData{ClientStreamData: &ClientStreamData{
176 Ref: brokerRef(), Data: bytes.Repeat([]byte{0x66}, 24),
177 }}},
178 "close_stream": {Payload: &BrokerClientEnvelope_CloseStream{CloseStream: &CloseStream{Ref: brokerRef()}}},
179 "start_preview": {Payload: &BrokerClientEnvelope_StartPreview{StartPreview: &StartPreview{
180 Scope: brokerScope(), Generation: 3,
181 Route: &PreviewRoute{
182 Endpoint: "dev@host:22", Registration: brokerRegistration(), Policy: brokerPolicy(),
183 },
184 Preview: &RemotePreviewRequest{
185 Version: 1,
186 Target: &RemoteTarget{
187 Endpoint: "dev@host:22", DisplayOrigin: "dev@host",
188 LifecycleId: &LifecycleID{Value: bytes.Repeat([]byte{0x55}, 16)},
189 SessionName: "work", LiveTabId: "tab-1",
190 },
191 Width: 80, Height: 24,
192 },
193 }}},
194 "cancel_preview": {Payload: &BrokerClientEnvelope_CancelPreview{CancelPreview: &CancelPreview{
195 Scope: brokerScope(), Generation: 3,
196 }}},
197 }
198}
199
200// brokerServerSamples returns one populated envelope per server variant.
201func brokerServerSamples() map[string]*BrokerServerEnvelope {
202 return map[string]*BrokerServerEnvelope{
203 "registered": {Payload: &BrokerServerEnvelope_Registered{Registered: &Registered{Scope: brokerScope()}}},
204 "snapshot_part": {Payload: &BrokerServerEnvelope_SnapshotPart{SnapshotPart: &SnapshotPart{
205 Scope: brokerScope(), Generation: 2, Revision: 5, Index: 9,
206 Part: &SnapshotPart_Daemon{Daemon: &SnapshotDaemon{
207 HostIndex: 1, Endpoint: "dev@host:22", DisplayOrigin: "dev@host",
208 Rank: 2, Registration: brokerRegistration(), Policy: &BrokerWirePolicy{ProtocolVersion: 56},
209 DaemonIdentity: "authed-daemon", DaemonIncarnation: bytes.Repeat([]byte{0x9}, 16),
210 ProtocolVersion: 56, Capabilities: 1,
211 Availability: 1,
212 Checking: true,
213 LastAttempt: &BrokerTimestamp{Seconds: 1700000000, Nanos: 1},
214 LastSuccess: &BrokerTimestamp{Seconds: 1700000001, Nanos: 2},
215 NextDue: &BrokerTimestamp{Seconds: 1700000002, Nanos: 3},
216 ConsecutiveFailures: 4, FailureEpisode: 6, FailureKind: 1,
217 InventoryKnown: true, SessionCount: 1,
218 }},
219 }}},
220 "operation_result": {Payload: &BrokerServerEnvelope_OperationResult{OperationResult: &OperationResult{
221 Scope: brokerScope(), OperationId: bytes.Repeat([]byte{0x77}, 16), Outcome: 2,
222 Error: &BrokerErrorDetail{Code: 2, Text: "incompatible", AdmissionCode: 1, FailureKind: 3},
223 }}},
224 "stream_opened": {Payload: &BrokerServerEnvelope_StreamOpened{StreamOpened: &StreamOpened{Ref: brokerRef()}}},
225 "server_stream_data": {Payload: &BrokerServerEnvelope_ServerStreamData{ServerStreamData: &ServerStreamData{
226 Ref: brokerRef(), Data: bytes.Repeat([]byte{0x88}, 24),
227 }}},
228 "stream_closed": {Payload: &BrokerServerEnvelope_StreamClosed{StreamClosed: &StreamClosed{
229 Ref: brokerRef(), Error: &BrokerErrorDetail{Code: 6, Text: "attachment_lost"},
230 }}},
231 "progress": {Payload: &BrokerServerEnvelope_Progress{Progress: &Progress{
232 Ref: brokerRef(), Phase: 1, Text: "probing",
233 }}},
234 "broker_error_message": {Payload: &BrokerServerEnvelope_BrokerErrorMessage{BrokerErrorMessage: &BrokerErrorMessage{
235 Scope: brokerScope(), Error: &BrokerErrorDetail{Code: 1, Text: "unavailable"},
236 }}},
237 "shutdown": {Payload: &BrokerServerEnvelope_Shutdown{Shutdown: &Shutdown{
238 Scope: brokerScope(), Reason: 1, Text: "idle exit",
239 }}},
240 "preview_publication": {Payload: &BrokerServerEnvelope_PreviewPublication{PreviewPublication: &PreviewPublication{
241 Scope: brokerScope(), Generation: 3,
242 Result: &PreviewPublication_Error{Error: &BrokerErrorDetail{Code: 1, Text: "unavailable"}},
243 }}},
244 }
245}
246
247func mustMarshalBroker(t *testing.T, message proto.Message) []byte {
248 t.Helper()
249 raw, err := proto.MarshalOptions{Deterministic: true}.Marshal(message)
250 if err != nil {
251 t.Fatal(err)
252 }
253 return raw
254}
255
256// TestBrokerEnvelopeRoundTrip proves every variant survives strict scan,
257// generated unmarshal, and equality, including nested snapshot payloads.
258func TestBrokerEnvelopeRoundTrip(t *testing.T) {
259 for name, sample := range brokerClientSamples() {
260 t.Run("client/"+name, func(t *testing.T) {
261 raw := mustMarshalBroker(t, sample)
262 decoded := &BrokerClientEnvelope{}
263 if err := validatedUnmarshal(decoded, raw); err != nil {
264 t.Fatalf("validatedUnmarshal() = %v", err)
265 }
266 if !proto.Equal(sample, decoded) {
267 t.Fatalf("roundtrip mismatch:\n got %v\nwant %v", decoded, sample)
268 }
269 })
270 }
271 for name, sample := range brokerServerSamples() {
272 t.Run("server/"+name, func(t *testing.T) {
273 raw := mustMarshalBroker(t, sample)
274 decoded := &BrokerServerEnvelope{}
275 if err := validatedUnmarshal(decoded, raw); err != nil {
276 t.Fatalf("validatedUnmarshal() = %v", err)
277 }
278 if !proto.Equal(sample, decoded) {
279 t.Fatalf("roundtrip mismatch:\n got %v\nwant %v", decoded, sample)
280 }
281 })
282 }
283
284 // The snapshot nested unions round-trip their non-daemon parts too, so
285 // every part tag is exercised.
286 for name, part := range map[string]*SnapshotPart{
287 "begin": {Part: &SnapshotPart_Begin{Begin: &SnapshotBegin{HostCount: 1, SessionCount: 2, TombstoneCount: 3, LocalPresent: true}}},
288 "session": {Part: &SnapshotPart_Session{Session: &SnapshotSession{HostIndex: 1, SessionIndex: 2, Session: brokerCatalogSession(), Local: true}}},
289 "tombstone": {Part: &SnapshotPart_Tombstone{Tombstone: &SnapshotTombstone{TombstoneIndex: 3, Registration: brokerRegistration(), RetiredRevision: 4}}},
290 "end": {Part: &SnapshotPart_End{End: &SnapshotEnd{}}},
291 } {
292 t.Run("snapshot/"+name, func(t *testing.T) {
293 raw := mustMarshalBroker(t, &BrokerServerEnvelope{Payload: &BrokerServerEnvelope_SnapshotPart{
294 SnapshotPart: &SnapshotPart{
295 Scope: brokerScope(), Generation: 1, Revision: 2, Index: 3, Part: part.Part,
296 },
297 }})
298 decoded := &BrokerServerEnvelope{}
299 if err := validatedUnmarshal(decoded, raw); err != nil {
300 t.Fatalf("validatedUnmarshal() = %v", err)
301 }
302 got := decoded.GetSnapshotPart()
303 if got == nil || got.GetIndex() != 3 || got.Part == nil {
304 t.Fatalf("snapshot part did not round-trip: %v", decoded)
305 }
306 })
307 }
308}
309
310// TestBrokerEnvelopeRequiredVariant proves both broker envelopes demand
311// exactly one payload variant, like the session envelopes.
312func TestBrokerEnvelopeRequiredVariant(t *testing.T) {
313 for _, message := range []proto.Message{&BrokerClientEnvelope{}, &BrokerServerEnvelope{}} {
314 if err := ScanEnvelope(message, mustMarshalBroker(t, message)); !errors.Is(err, ErrScanEmpty) {
315 t.Fatalf("ScanEnvelope(%T empty) = %v, want ErrScanEmpty", message, err)
316 }
317 }
318}
319
320// TestBrokerEnvelopeWrongDirection proves a payload serialized in one
321// direction is rejected when scanned as the other direction.
322func TestBrokerEnvelopeWrongDirection(t *testing.T) {
323 client := mustMarshalBroker(t, brokerClientSamples()["subscribe"])
324 if err := ScanEnvelope(&BrokerServerEnvelope{}, client); !errors.Is(err, ErrScanUnknown) {
325 t.Fatalf("client payload as server envelope = %v, want ErrScanUnknown", err)
326 }
327 server := mustMarshalBroker(t, brokerServerSamples()["progress"])
328 if err := ScanEnvelope(&BrokerClientEnvelope{}, server); !errors.Is(err, ErrScanUnknown) {
329 t.Fatalf("server payload as client envelope = %v, want ErrScanUnknown", err)
330 }
331 // A session payload is unknown on the broker conversation too.
332 session := mustMarshalBroker(t, &ClientEnvelope{Payload: &ClientEnvelope_Ping{Ping: &Ping{}}})
333 if err := ScanEnvelope(&BrokerClientEnvelope{}, session); !errors.Is(err, ErrScanUnknown) {
334 t.Fatalf("session payload as broker envelope = %v, want ErrScanUnknown", err)
335 }
336}
337
338// TestBrokerEnvelopeDuplicateAlternatives proves a second oneof occurrence
339// is rejected whether it repeats the same variant or switches alternatives.
340func TestBrokerEnvelopeDuplicateAlternatives(t *testing.T) {
341 register := mustMarshalBroker(t, brokerClientSamples()["register"])
342 subscribe := mustMarshalBroker(t, brokerClientSamples()["subscribe"])
343 concatenated := append(append([]byte(nil), register...), subscribe...)
344 if err := ScanEnvelope(&BrokerClientEnvelope{}, concatenated); !errors.Is(err, ErrScanDuplicate) {
345 t.Fatalf("alternate alternatives = %v, want ErrScanDuplicate", err)
346 }
347 repeated := append(append([]byte(nil), subscribe...), subscribe...)
348 if err := ScanEnvelope(&BrokerClientEnvelope{}, repeated); !errors.Is(err, ErrScanDuplicate) {
349 t.Fatalf("duplicate variant = %v, want ErrScanDuplicate", err)
350 }
351
352 // The nested snapshot part union is exclusive as well.
353 daemon := mustMarshalBroker(t, &SnapshotPart{Part: &SnapshotPart_Daemon{Daemon: &SnapshotDaemon{Endpoint: "dev@host:22"}}})
354 end := mustMarshalBroker(t, &SnapshotPart{Part: &SnapshotPart_End{End: &SnapshotEnd{}}})
355 if err := ScanEnvelope(&SnapshotPart{}, append(append([]byte(nil), daemon...), end...)); !errors.Is(err, ErrScanDuplicate) {
356 t.Fatalf("duplicate snapshot part = %v, want ErrScanDuplicate", err)
357 }
358}
359
360// TestBrokerEnvelopeRejectsUnknownAndWrongTypes proves unknown and
361// mistyped fields are rejected recursively, at the envelope level and
362// several levels into a nested snapshot payload.
363func TestBrokerEnvelopeRejectsUnknownAndWrongTypes(t *testing.T) {
364 // Unknown field on the envelope itself.
365 raw := mustMarshalBroker(t, brokerClientSamples()["close_stream"])
366 raw = protowire.AppendTag(raw, 199, protowire.VarintType)
367 raw = protowire.AppendVarint(raw, 1)
368 if err := ScanEnvelope(&BrokerClientEnvelope{}, raw); !errors.Is(err, ErrScanUnknown) {
369 t.Fatalf("unknown envelope field = %v, want ErrScanUnknown", err)
370 }
371
372 // Unknown field two levels down: BrokerClientEnvelope → OpenStream → BrokerStreamRef.
373 ref := protowire.AppendTag(nil, 1, protowire.BytesType) // BrokerStreamRef.scope
374 ref = protowire.AppendBytes(ref, mustMarshalBroker(t, brokerScope()))
375 ref = protowire.AppendTag(ref, 2, protowire.VarintType) // BrokerStreamRef.stream_id
376 ref = protowire.AppendVarint(ref, 3)
377 ref = protowire.AppendTag(ref, 9, protowire.VarintType) // unknown in BrokerStreamRef
378 ref = protowire.AppendVarint(ref, 1)
379 open := protowire.AppendTag(nil, 1, protowire.BytesType) // OpenStream.ref
380 open = protowire.AppendBytes(open, ref)
381 outer := protowire.AppendTag(nil, 108, protowire.BytesType) // BrokerClientEnvelope.open_stream
382 outer = protowire.AppendBytes(outer, open)
383 if err := ScanEnvelope(&BrokerClientEnvelope{}, outer); !errors.Is(err, ErrScanUnknown) {
384 t.Fatalf("nested unknown field = %v, want ErrScanUnknown", err)
385 }
386
387 // Unknown field inside a nested snapshot part message.
388 part := protowire.AppendTag(nil, 15, protowire.BytesType) // SnapshotPart.daemon
389 daemon := protowire.AppendTag(nil, 3, protowire.BytesType) // SnapshotDaemon.endpoint
390 daemon = protowire.AppendBytes(daemon, []byte("dev@host:22"))
391 daemon = protowire.AppendTag(daemon, 99, protowire.VarintType)
392 daemon = protowire.AppendVarint(daemon, 1)
393 part = protowire.AppendBytes(part, daemon)
394 raw = protowire.AppendTag(nil, 202, protowire.BytesType) // BrokerServerEnvelope.snapshot_part
395 raw = protowire.AppendBytes(raw, part)
396 if err := ScanEnvelope(&BrokerServerEnvelope{}, raw); !errors.Is(err, ErrScanUnknown) {
397 t.Fatalf("unknown nested snapshot field = %v, want ErrScanUnknown", err)
398 }
399
400 // Wrong wire type: BrokerClientEnvelope.close_stream (bytes) as a varint.
401 wrongType := protowire.AppendTag(nil, 110, protowire.VarintType)
402 wrongType = protowire.AppendVarint(wrongType, 1)
403 if err := ScanEnvelope(&BrokerClientEnvelope{}, wrongType); !errors.Is(err, ErrScanWireType) {
404 t.Fatalf("wrong wire type = %v, want ErrScanWireType", err)
405 }
406
407 // A raw varint wider than the declared uint32 is rejected before
408 // generated unmarshal can truncate its high bits.
409 open = protowire.AppendTag(nil, 2, protowire.VarintType) // OpenStream.purpose
410 open = protowire.AppendVarint(open, 1<<33)
411 outer = protowire.AppendTag(nil, 108, protowire.BytesType)
412 outer = protowire.AppendBytes(outer, open)
413 if err := ScanEnvelope(&BrokerClientEnvelope{}, outer); !errors.Is(err, ErrScanVarintRange) {
414 t.Fatalf("uint32 overflow = %v, want ErrScanVarintRange", err)
415 }
416}
417
418// TestBrokerEnvelopeTruncation proves truncated tags, varints, lengths, and
419// every strict prefix of a valid envelope are rejected.
420func TestBrokerEnvelopeTruncation(t *testing.T) {
421 if err := ScanEnvelope(&BrokerClientEnvelope{}, []byte{0xFF}); !errors.Is(err, ErrScanTruncated) {
422 t.Fatalf("truncated tag = %v, want ErrScanTruncated", err)
423 }
424 overlong := protowire.AppendTag(nil, 101, protowire.BytesType)
425 overlong = protowire.AppendBytes(overlong, make([]byte, 8))
426 overlong = overlong[:len(overlong)-4]
427 if err := ScanEnvelope(&BrokerClientEnvelope{}, overlong); !errors.Is(err, ErrScanTruncated) {
428 t.Fatalf("overlong length = %v, want ErrScanTruncated", err)
429 }
430 partialVarint := protowire.AppendTag(nil, 102, protowire.BytesType) // BrokerClientEnvelope.subscribe
431 subscribe := protowire.AppendTag(nil, 2, protowire.VarintType) // Subscribe.generation
432 partialVarint = protowire.AppendBytes(partialVarint, append(subscribe, 0x80))
433 if err := ScanEnvelope(&BrokerClientEnvelope{}, partialVarint); !errors.Is(err, ErrScanTruncated) {
434 t.Fatalf("truncated varint = %v, want ErrScanTruncated", err)
435 }
436
437 // Every strict prefix of a valid envelope must be refused: a top-level
438 // envelope requires its complete variant.
439 for name, sample := range brokerClientSamples() {
440 raw := mustMarshalBroker(t, sample)
441 for size := range len(raw) {
442 if err := ScanEnvelope(&BrokerClientEnvelope{}, raw[:size]); err == nil {
443 t.Fatalf("client prefix %s[:%d] accepted", name, size)
444 }
445 }
446 }
447 for name, sample := range brokerServerSamples() {
448 raw := mustMarshalBroker(t, sample)
449 for size := range len(raw) {
450 if err := ScanEnvelope(&BrokerServerEnvelope{}, raw[:size]); err == nil {
451 t.Fatalf("server prefix %s[:%d] accepted", name, size)
452 }
453 }
454 }
455}
456
457// TestBrokerEnvelopeTrailingGarbage proves trailing bytes and a second
458// concatenated envelope are rejected.
459func TestBrokerEnvelopeTrailingGarbage(t *testing.T) {
460 raw := mustMarshalBroker(t, brokerServerSamples()["stream_opened"])
461 for _, tail := range [][]byte{
462 {0x00},
463 protowire.AppendTag(nil, 209, protowire.VarintType),
464 mustMarshalBroker(t, brokerServerSamples()["shutdown"]),
465 } {
466 payload := append(append([]byte(nil), raw...), tail...)
467 if err := ScanEnvelope(&BrokerServerEnvelope{}, payload); err == nil {
468 t.Fatalf("trailing %x accepted", tail)
469 }
470 }
471}
472
473// TestBrokerMembershipSchemaShape pins the Phase C1 membership layout: the new
474// client variant, the required policy on AddHost/UpdateHostPolicy, the exact
475// registration on RemoveHost/UpdateHostPolicy, and the registration member of
476// OperationResult. A field that moves would silently change the meaning of a
477// mutating request without changing its tag.
478func TestBrokerMembershipSchemaShape(t *testing.T) {
479 fields := func(message proto.Message) map[string]protoreflect.FieldNumber {
480 t.Helper()
481 descriptor := message.ProtoReflect().Descriptor()
482 out := make(map[string]protoreflect.FieldNumber, descriptor.Fields().Len())
483 for i := range descriptor.Fields().Len() {
484 field := descriptor.Fields().Get(i)
485 out[string(field.Name())] = field.Number()
486 }
487 return out
488 }
489
490 if got, want := oneofTags(t, &BrokerClientEnvelope{})["update_host_policy"], protoreflect.FieldNumber(111); got != want {
491 t.Fatalf("BrokerClientEnvelope.update_host_policy = %d, want %d", got, want)
492 }
493
494 addHost := fields(&AddHost{})
495 if addHost["policy"] != 4 {
496 t.Fatalf("AddHost.policy = %d, want 4 (required authority)", addHost["policy"])
497 }
498 removeHost := fields(&RemoveHost{})
499 if removeHost["registration"] != 3 {
500 t.Fatalf("RemoveHost.registration = %d, want 3", removeHost["registration"])
501 }
502 if _, present := removeHost["endpoint"]; present {
503 t.Fatal("RemoveHost still carries an endpoint field; removal is exact-registration authority")
504 }
505 update := fields(&UpdateHostPolicy{})
506 if update["registration"] != 3 || update["policy"] != 4 {
507 t.Fatalf("UpdateHostPolicy fields = %v, want registration 3 and policy 4", update)
508 }
509 result := fields(&OperationResult{})
510 if result["registration"] != 6 {
511 t.Fatalf("OperationResult.registration = %d, want 6", result["registration"])
512 }
513
514 // Every membership sample survives strict scan, unmarshal, and equality,
515 // and every strict prefix of it is refused.
516 for name, sample := range brokerClientSamples() {
517 switch name {
518 case "add_host", "remove_host", "update_host_policy":
519 default:
520 continue
521 }
522 raw := mustMarshalBroker(t, sample)
523 decoded := &BrokerClientEnvelope{}
524 if err := validatedUnmarshal(decoded, raw); err != nil {
525 t.Fatalf("%s: validatedUnmarshal() = %v", name, err)
526 }
527 if !proto.Equal(sample, decoded) {
528 t.Fatalf("%s: roundtrip mismatch:\n got %v\nwant %v", name, decoded, sample)
529 }
530 for size := range len(raw) {
531 if err := ScanEnvelope(&BrokerClientEnvelope{}, raw[:size]); err == nil {
532 t.Fatalf("%s prefix[:%d] accepted", name, size)
533 }
534 }
535 }
536}