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, "stream_window_update": 114,
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, "stream_window_update": 211,
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 "stream_window_update": {Payload: &BrokerClientEnvelope_StreamWindowUpdate{StreamWindowUpdate: &StreamWindowUpdate{
198 Ref: brokerRef(), Credit: 4096,
199 }}},
200 }
201}
202
203// brokerServerSamples returns one populated envelope per server variant.
204func brokerServerSamples() map[string]*BrokerServerEnvelope {
205 return map[string]*BrokerServerEnvelope{
206 "registered": {Payload: &BrokerServerEnvelope_Registered{Registered: &Registered{Scope: brokerScope()}}},
207 "snapshot_part": {Payload: &BrokerServerEnvelope_SnapshotPart{SnapshotPart: &SnapshotPart{
208 Scope: brokerScope(), Generation: 2, Revision: 5, Index: 9,
209 Part: &SnapshotPart_Daemon{Daemon: &SnapshotDaemon{
210 HostIndex: 1, Endpoint: "dev@host:22", DisplayOrigin: "dev@host",
211 Rank: 2, Registration: brokerRegistration(), Policy: &BrokerWirePolicy{ProtocolVersion: 56},
212 DaemonIdentity: "authed-daemon", DaemonIncarnation: bytes.Repeat([]byte{0x9}, 16),
213 ProtocolVersion: 56, Capabilities: 1,
214 Availability: 1,
215 Checking: true,
216 LastAttempt: &BrokerTimestamp{Seconds: 1700000000, Nanos: 1},
217 LastSuccess: &BrokerTimestamp{Seconds: 1700000001, Nanos: 2},
218 NextDue: &BrokerTimestamp{Seconds: 1700000002, Nanos: 3},
219 ConsecutiveFailures: 4, FailureEpisode: 6, FailureKind: 1,
220 InventoryKnown: true, SessionCount: 1,
221 }},
222 }}},
223 "operation_result": {Payload: &BrokerServerEnvelope_OperationResult{OperationResult: &OperationResult{
224 Scope: brokerScope(), OperationId: bytes.Repeat([]byte{0x77}, 16), Outcome: 2,
225 Error: &BrokerErrorDetail{Code: 2, Text: "incompatible", AdmissionCode: 1, FailureKind: 3},
226 }}},
227 "stream_opened": {Payload: &BrokerServerEnvelope_StreamOpened{StreamOpened: &StreamOpened{Ref: brokerRef()}}},
228 "server_stream_data": {Payload: &BrokerServerEnvelope_ServerStreamData{ServerStreamData: &ServerStreamData{
229 Ref: brokerRef(), Data: bytes.Repeat([]byte{0x88}, 24),
230 }}},
231 "stream_closed": {Payload: &BrokerServerEnvelope_StreamClosed{StreamClosed: &StreamClosed{
232 Ref: brokerRef(), Error: &BrokerErrorDetail{Code: 6, Text: "attachment_lost"},
233 }}},
234 "progress": {Payload: &BrokerServerEnvelope_Progress{Progress: &Progress{
235 Ref: brokerRef(), Phase: 1, Text: "probing",
236 }}},
237 "broker_error_message": {Payload: &BrokerServerEnvelope_BrokerErrorMessage{BrokerErrorMessage: &BrokerErrorMessage{
238 Scope: brokerScope(), Error: &BrokerErrorDetail{Code: 1, Text: "unavailable"},
239 }}},
240 "shutdown": {Payload: &BrokerServerEnvelope_Shutdown{Shutdown: &Shutdown{
241 Scope: brokerScope(), Reason: 1, Text: "idle exit",
242 }}},
243 "preview_publication": {Payload: &BrokerServerEnvelope_PreviewPublication{PreviewPublication: &PreviewPublication{
244 Scope: brokerScope(), Generation: 3,
245 Result: &PreviewPublication_Error{Error: &BrokerErrorDetail{Code: 1, Text: "unavailable"}},
246 }}},
247 "stream_window_update": {Payload: &BrokerServerEnvelope_StreamWindowUpdate{StreamWindowUpdate: &StreamWindowUpdate{
248 Ref: brokerRef(), Credit: 4096,
249 }}},
250 }
251}
252
253func mustMarshalBroker(t *testing.T, message proto.Message) []byte {
254 t.Helper()
255 raw, err := proto.MarshalOptions{Deterministic: true}.Marshal(message)
256 if err != nil {
257 t.Fatal(err)
258 }
259 return raw
260}
261
262// TestBrokerEnvelopeRoundTrip proves every variant survives strict scan,
263// generated unmarshal, and equality, including nested snapshot payloads.
264func TestBrokerEnvelopeRoundTrip(t *testing.T) {
265 for name, sample := range brokerClientSamples() {
266 t.Run("client/"+name, func(t *testing.T) {
267 raw := mustMarshalBroker(t, sample)
268 decoded := &BrokerClientEnvelope{}
269 if err := validatedUnmarshal(decoded, raw); err != nil {
270 t.Fatalf("validatedUnmarshal() = %v", err)
271 }
272 if !proto.Equal(sample, decoded) {
273 t.Fatalf("roundtrip mismatch:\n got %v\nwant %v", decoded, sample)
274 }
275 })
276 }
277 for name, sample := range brokerServerSamples() {
278 t.Run("server/"+name, func(t *testing.T) {
279 raw := mustMarshalBroker(t, sample)
280 decoded := &BrokerServerEnvelope{}
281 if err := validatedUnmarshal(decoded, raw); err != nil {
282 t.Fatalf("validatedUnmarshal() = %v", err)
283 }
284 if !proto.Equal(sample, decoded) {
285 t.Fatalf("roundtrip mismatch:\n got %v\nwant %v", decoded, sample)
286 }
287 })
288 }
289
290 // The snapshot nested unions round-trip their non-daemon parts too, so
291 // every part tag is exercised.
292 for name, part := range map[string]*SnapshotPart{
293 "begin": {Part: &SnapshotPart_Begin{Begin: &SnapshotBegin{HostCount: 1, SessionCount: 2, TombstoneCount: 3, LocalPresent: true}}},
294 "session": {Part: &SnapshotPart_Session{Session: &SnapshotSession{HostIndex: 1, SessionIndex: 2, Session: brokerCatalogSession(), Local: true}}},
295 "tombstone": {Part: &SnapshotPart_Tombstone{Tombstone: &SnapshotTombstone{TombstoneIndex: 3, Registration: brokerRegistration(), RetiredRevision: 4}}},
296 "end": {Part: &SnapshotPart_End{End: &SnapshotEnd{}}},
297 } {
298 t.Run("snapshot/"+name, func(t *testing.T) {
299 raw := mustMarshalBroker(t, &BrokerServerEnvelope{Payload: &BrokerServerEnvelope_SnapshotPart{
300 SnapshotPart: &SnapshotPart{
301 Scope: brokerScope(), Generation: 1, Revision: 2, Index: 3, Part: part.Part,
302 },
303 }})
304 decoded := &BrokerServerEnvelope{}
305 if err := validatedUnmarshal(decoded, raw); err != nil {
306 t.Fatalf("validatedUnmarshal() = %v", err)
307 }
308 got := decoded.GetSnapshotPart()
309 if got == nil || got.GetIndex() != 3 || got.Part == nil {
310 t.Fatalf("snapshot part did not round-trip: %v", decoded)
311 }
312 })
313 }
314}
315
316// TestBrokerEnvelopeRequiredVariant proves both broker envelopes demand
317// exactly one payload variant, like the session envelopes.
318func TestBrokerEnvelopeRequiredVariant(t *testing.T) {
319 for _, message := range []proto.Message{&BrokerClientEnvelope{}, &BrokerServerEnvelope{}} {
320 if err := ScanEnvelope(message, mustMarshalBroker(t, message)); !errors.Is(err, ErrScanEmpty) {
321 t.Fatalf("ScanEnvelope(%T empty) = %v, want ErrScanEmpty", message, err)
322 }
323 }
324}
325
326// TestBrokerEnvelopeWrongDirection proves a payload serialized in one
327// direction is rejected when scanned as the other direction.
328func TestBrokerEnvelopeWrongDirection(t *testing.T) {
329 client := mustMarshalBroker(t, brokerClientSamples()["subscribe"])
330 if err := ScanEnvelope(&BrokerServerEnvelope{}, client); !errors.Is(err, ErrScanUnknown) {
331 t.Fatalf("client payload as server envelope = %v, want ErrScanUnknown", err)
332 }
333 server := mustMarshalBroker(t, brokerServerSamples()["progress"])
334 if err := ScanEnvelope(&BrokerClientEnvelope{}, server); !errors.Is(err, ErrScanUnknown) {
335 t.Fatalf("server payload as client envelope = %v, want ErrScanUnknown", err)
336 }
337 // A session payload is unknown on the broker conversation too.
338 session := mustMarshalBroker(t, &ClientEnvelope{Payload: &ClientEnvelope_Ping{Ping: &Ping{}}})
339 if err := ScanEnvelope(&BrokerClientEnvelope{}, session); !errors.Is(err, ErrScanUnknown) {
340 t.Fatalf("session payload as broker envelope = %v, want ErrScanUnknown", err)
341 }
342}
343
344// TestBrokerEnvelopeDuplicateAlternatives proves a second oneof occurrence
345// is rejected whether it repeats the same variant or switches alternatives.
346func TestBrokerEnvelopeDuplicateAlternatives(t *testing.T) {
347 register := mustMarshalBroker(t, brokerClientSamples()["register"])
348 subscribe := mustMarshalBroker(t, brokerClientSamples()["subscribe"])
349 concatenated := append(append([]byte(nil), register...), subscribe...)
350 if err := ScanEnvelope(&BrokerClientEnvelope{}, concatenated); !errors.Is(err, ErrScanDuplicate) {
351 t.Fatalf("alternate alternatives = %v, want ErrScanDuplicate", err)
352 }
353 repeated := append(append([]byte(nil), subscribe...), subscribe...)
354 if err := ScanEnvelope(&BrokerClientEnvelope{}, repeated); !errors.Is(err, ErrScanDuplicate) {
355 t.Fatalf("duplicate variant = %v, want ErrScanDuplicate", err)
356 }
357
358 // The nested snapshot part union is exclusive as well.
359 daemon := mustMarshalBroker(t, &SnapshotPart{Part: &SnapshotPart_Daemon{Daemon: &SnapshotDaemon{Endpoint: "dev@host:22"}}})
360 end := mustMarshalBroker(t, &SnapshotPart{Part: &SnapshotPart_End{End: &SnapshotEnd{}}})
361 if err := ScanEnvelope(&SnapshotPart{}, append(append([]byte(nil), daemon...), end...)); !errors.Is(err, ErrScanDuplicate) {
362 t.Fatalf("duplicate snapshot part = %v, want ErrScanDuplicate", err)
363 }
364}
365
366// TestBrokerEnvelopeRejectsUnknownAndWrongTypes proves unknown and
367// mistyped fields are rejected recursively, at the envelope level and
368// several levels into a nested snapshot payload.
369func TestBrokerEnvelopeRejectsUnknownAndWrongTypes(t *testing.T) {
370 // Unknown field on the envelope itself.
371 raw := mustMarshalBroker(t, brokerClientSamples()["close_stream"])
372 raw = protowire.AppendTag(raw, 199, protowire.VarintType)
373 raw = protowire.AppendVarint(raw, 1)
374 if err := ScanEnvelope(&BrokerClientEnvelope{}, raw); !errors.Is(err, ErrScanUnknown) {
375 t.Fatalf("unknown envelope field = %v, want ErrScanUnknown", err)
376 }
377
378 // Unknown field two levels down: BrokerClientEnvelope → OpenStream → BrokerStreamRef.
379 ref := protowire.AppendTag(nil, 1, protowire.BytesType) // BrokerStreamRef.scope
380 ref = protowire.AppendBytes(ref, mustMarshalBroker(t, brokerScope()))
381 ref = protowire.AppendTag(ref, 2, protowire.VarintType) // BrokerStreamRef.stream_id
382 ref = protowire.AppendVarint(ref, 3)
383 ref = protowire.AppendTag(ref, 9, protowire.VarintType) // unknown in BrokerStreamRef
384 ref = protowire.AppendVarint(ref, 1)
385 open := protowire.AppendTag(nil, 1, protowire.BytesType) // OpenStream.ref
386 open = protowire.AppendBytes(open, ref)
387 outer := protowire.AppendTag(nil, 108, protowire.BytesType) // BrokerClientEnvelope.open_stream
388 outer = protowire.AppendBytes(outer, open)
389 if err := ScanEnvelope(&BrokerClientEnvelope{}, outer); !errors.Is(err, ErrScanUnknown) {
390 t.Fatalf("nested unknown field = %v, want ErrScanUnknown", err)
391 }
392
393 // Unknown field inside a nested snapshot part message.
394 part := protowire.AppendTag(nil, 15, protowire.BytesType) // SnapshotPart.daemon
395 daemon := protowire.AppendTag(nil, 3, protowire.BytesType) // SnapshotDaemon.endpoint
396 daemon = protowire.AppendBytes(daemon, []byte("dev@host:22"))
397 daemon = protowire.AppendTag(daemon, 99, protowire.VarintType)
398 daemon = protowire.AppendVarint(daemon, 1)
399 part = protowire.AppendBytes(part, daemon)
400 raw = protowire.AppendTag(nil, 202, protowire.BytesType) // BrokerServerEnvelope.snapshot_part
401 raw = protowire.AppendBytes(raw, part)
402 if err := ScanEnvelope(&BrokerServerEnvelope{}, raw); !errors.Is(err, ErrScanUnknown) {
403 t.Fatalf("unknown nested snapshot field = %v, want ErrScanUnknown", err)
404 }
405
406 // Wrong wire type: BrokerClientEnvelope.close_stream (bytes) as a varint.
407 wrongType := protowire.AppendTag(nil, 110, protowire.VarintType)
408 wrongType = protowire.AppendVarint(wrongType, 1)
409 if err := ScanEnvelope(&BrokerClientEnvelope{}, wrongType); !errors.Is(err, ErrScanWireType) {
410 t.Fatalf("wrong wire type = %v, want ErrScanWireType", err)
411 }
412
413 // A raw varint wider than the declared uint32 is rejected before
414 // generated unmarshal can truncate its high bits.
415 open = protowire.AppendTag(nil, 2, protowire.VarintType) // OpenStream.purpose
416 open = protowire.AppendVarint(open, 1<<33)
417 outer = protowire.AppendTag(nil, 108, protowire.BytesType)
418 outer = protowire.AppendBytes(outer, open)
419 if err := ScanEnvelope(&BrokerClientEnvelope{}, outer); !errors.Is(err, ErrScanVarintRange) {
420 t.Fatalf("uint32 overflow = %v, want ErrScanVarintRange", err)
421 }
422}
423
424// TestBrokerEnvelopeTruncation proves truncated tags, varints, lengths, and
425// every strict prefix of a valid envelope are rejected.
426func TestBrokerEnvelopeTruncation(t *testing.T) {
427 if err := ScanEnvelope(&BrokerClientEnvelope{}, []byte{0xFF}); !errors.Is(err, ErrScanTruncated) {
428 t.Fatalf("truncated tag = %v, want ErrScanTruncated", err)
429 }
430 overlong := protowire.AppendTag(nil, 101, protowire.BytesType)
431 overlong = protowire.AppendBytes(overlong, make([]byte, 8))
432 overlong = overlong[:len(overlong)-4]
433 if err := ScanEnvelope(&BrokerClientEnvelope{}, overlong); !errors.Is(err, ErrScanTruncated) {
434 t.Fatalf("overlong length = %v, want ErrScanTruncated", err)
435 }
436 partialVarint := protowire.AppendTag(nil, 102, protowire.BytesType) // BrokerClientEnvelope.subscribe
437 subscribe := protowire.AppendTag(nil, 2, protowire.VarintType) // Subscribe.generation
438 partialVarint = protowire.AppendBytes(partialVarint, append(subscribe, 0x80))
439 if err := ScanEnvelope(&BrokerClientEnvelope{}, partialVarint); !errors.Is(err, ErrScanTruncated) {
440 t.Fatalf("truncated varint = %v, want ErrScanTruncated", err)
441 }
442
443 // Every strict prefix of a valid envelope must be refused: a top-level
444 // envelope requires its complete variant.
445 for name, sample := range brokerClientSamples() {
446 raw := mustMarshalBroker(t, sample)
447 for size := range len(raw) {
448 if err := ScanEnvelope(&BrokerClientEnvelope{}, raw[:size]); err == nil {
449 t.Fatalf("client prefix %s[:%d] accepted", name, size)
450 }
451 }
452 }
453 for name, sample := range brokerServerSamples() {
454 raw := mustMarshalBroker(t, sample)
455 for size := range len(raw) {
456 if err := ScanEnvelope(&BrokerServerEnvelope{}, raw[:size]); err == nil {
457 t.Fatalf("server prefix %s[:%d] accepted", name, size)
458 }
459 }
460 }
461}
462
463// TestBrokerEnvelopeTrailingGarbage proves trailing bytes and a second
464// concatenated envelope are rejected.
465func TestBrokerEnvelopeTrailingGarbage(t *testing.T) {
466 raw := mustMarshalBroker(t, brokerServerSamples()["stream_opened"])
467 for _, tail := range [][]byte{
468 {0x00},
469 protowire.AppendTag(nil, 209, protowire.VarintType),
470 mustMarshalBroker(t, brokerServerSamples()["shutdown"]),
471 } {
472 payload := append(append([]byte(nil), raw...), tail...)
473 if err := ScanEnvelope(&BrokerServerEnvelope{}, payload); err == nil {
474 t.Fatalf("trailing %x accepted", tail)
475 }
476 }
477}
478
479// TestBrokerMembershipSchemaShape pins the Phase C1 membership layout: the new
480// client variant, the required policy on AddHost/UpdateHostPolicy, the exact
481// registration on RemoveHost/UpdateHostPolicy, and the registration member of
482// OperationResult. A field that moves would silently change the meaning of a
483// mutating request without changing its tag.
484func TestBrokerMembershipSchemaShape(t *testing.T) {
485 fields := func(message proto.Message) map[string]protoreflect.FieldNumber {
486 t.Helper()
487 descriptor := message.ProtoReflect().Descriptor()
488 out := make(map[string]protoreflect.FieldNumber, descriptor.Fields().Len())
489 for i := range descriptor.Fields().Len() {
490 field := descriptor.Fields().Get(i)
491 out[string(field.Name())] = field.Number()
492 }
493 return out
494 }
495
496 if got, want := oneofTags(t, &BrokerClientEnvelope{})["update_host_policy"], protoreflect.FieldNumber(111); got != want {
497 t.Fatalf("BrokerClientEnvelope.update_host_policy = %d, want %d", got, want)
498 }
499
500 addHost := fields(&AddHost{})
501 if addHost["policy"] != 4 {
502 t.Fatalf("AddHost.policy = %d, want 4 (required authority)", addHost["policy"])
503 }
504 removeHost := fields(&RemoveHost{})
505 if removeHost["registration"] != 3 {
506 t.Fatalf("RemoveHost.registration = %d, want 3", removeHost["registration"])
507 }
508 if _, present := removeHost["endpoint"]; present {
509 t.Fatal("RemoveHost still carries an endpoint field; removal is exact-registration authority")
510 }
511 update := fields(&UpdateHostPolicy{})
512 if update["registration"] != 3 || update["policy"] != 4 {
513 t.Fatalf("UpdateHostPolicy fields = %v, want registration 3 and policy 4", update)
514 }
515 result := fields(&OperationResult{})
516 if result["registration"] != 6 {
517 t.Fatalf("OperationResult.registration = %d, want 6", result["registration"])
518 }
519
520 // Every membership sample survives strict scan, unmarshal, and equality,
521 // and every strict prefix of it is refused.
522 for name, sample := range brokerClientSamples() {
523 switch name {
524 case "add_host", "remove_host", "update_host_policy":
525 default:
526 continue
527 }
528 raw := mustMarshalBroker(t, sample)
529 decoded := &BrokerClientEnvelope{}
530 if err := validatedUnmarshal(decoded, raw); err != nil {
531 t.Fatalf("%s: validatedUnmarshal() = %v", name, err)
532 }
533 if !proto.Equal(sample, decoded) {
534 t.Fatalf("%s: roundtrip mismatch:\n got %v\nwant %v", name, decoded, sample)
535 }
536 for size := range len(raw) {
537 if err := ScanEnvelope(&BrokerClientEnvelope{}, raw[:size]); err == nil {
538 t.Fatalf("%s prefix[:%d] accepted", name, size)
539 }
540 }
541 }
542}