1package brokerwire
2
3import (
4 "bytes"
5 "strings"
6 "testing"
7 "time"
8
9 "google.golang.org/protobuf/encoding/protowire"
10 "google.golang.org/protobuf/proto"
11
12 "github.com/bnema/vev/internal/domain"
13 "github.com/bnema/vev/internal/ports"
14 "github.com/bnema/vev/internal/protocol"
15 "github.com/bnema/vev/internal/protocol/catalogue"
16 "github.com/bnema/vev/internal/protocol/wire"
17 "github.com/stretchr/testify/require"
18)
19
20func testConnectionID(seed byte) ports.BrokerConnectionID {
21 var id ports.BrokerConnectionID
22 for i := range id {
23 id[i] = seed + byte(i)
24 }
25 return id
26}
27
28func testOperationID(seed byte) ports.BrokerOperationID {
29 var id ports.BrokerOperationID
30 for i := range id {
31 id[i] = seed + byte(i)
32 }
33 return id
34}
35
36func testRegistration() domain.RemoteRegistration {
37 return domain.RemoteRegistration{
38 Endpoint: "dev@host:22",
39 Incarnation: [16]byte{1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16},
40 Generation: 9,
41 }
42}
43
44func testPolicy() ports.BrokerPolicy {
45 return ports.BrokerPolicy{
46 ProtocolVersion: protocol.Version,
47 CatalogSchemaVersion: catalogue.RemoteCatalogSchemaVersion,
48 EnvironmentPolicy: protocol.EnvironmentPolicyDaemonOwned,
49 Transport: "quic",
50 Trust: "pinned",
51 Launch: "agent",
52 Isolation: "user",
53 }
54}
55
56func testTarget() protocol.ExactSessionTarget {
57 return protocol.ExactSessionTarget{
58 LifecycleID: domain.SessionLifecycleID{1, 2, 3},
59 SessionName: "work",
60 }
61}
62
63func testCatalogSession() catalogue.RemoteCatalogSession {
64 return catalogue.RemoteCatalogSession{
65 LifecycleID: domain.SessionLifecycleID{9, 8, 7},
66 Name: "work",
67 State: catalogue.RemoteCatalogSessionUp,
68 Ephemeral: true,
69 Tabs: []catalogue.RemoteCatalogTab{{ID: "tab-1", Index: 0, Name: "shell", Detail: "1 pane", Attention: true}},
70 Attached: true,
71 LastUsedSeq: 4,
72 ActiveTabID: "tab-1",
73 Reason: domain.RemoteReasonRefreshing,
74 }
75}
76
77// testDaemonObservation builds one observed, catalogue-valid remote daemon
78// projection. Sessions are empty: they travel as separate parts.
79func testDaemonObservation() ports.BrokerDaemonObservation {
80 return ports.BrokerDaemonObservation{
81 Endpoint: "dev@host:22",
82 DisplayOrigin: "dev@host",
83 Rank: 2,
84 Registration: testRegistration(),
85 Policy: testPolicy(),
86 Identity: ports.BrokerDaemonIdentity("authed-daemon"),
87 Incarnation: ports.BrokerDaemonIncarnation{1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16},
88 ProtocolVersion: protocol.Version,
89 Capabilities: protocol.CapabilityResume | protocol.CapabilityUDP,
90 Availability: domain.RemoteAvailabilityReachable,
91 Checking: true,
92 LastAttempt: time.Unix(1700000000, 1).UTC(),
93 LastSuccess: time.Unix(1700000001, 2).UTC(),
94 NextDue: time.Unix(1700000002, 3).UTC(),
95 FailureEpisode: 6,
96 LastFailure: domain.RemoteFailure{Kind: domain.RemoteFailureTransport},
97 InventoryKnown: true,
98 Sessions: []catalogue.RemoteCatalogSession{},
99 }
100}
101
102// testLocalDaemonObservation builds the local daemon projection: no endpoint,
103// no registration, always a valid policy, and an explicit availability.
104func testLocalDaemonObservation() ports.BrokerDaemonObservation {
105 return ports.BrokerDaemonObservation{
106 Local: true,
107 DisplayOrigin: "local",
108 Policy: testPolicy(),
109 Availability: domain.RemoteAvailabilityReachable,
110 InventoryKnown: true,
111 Sessions: []catalogue.RemoteCatalogSession{},
112 }
113}
114
115func testErrorDetail() ErrorDetail {
116 return ErrorDetail{Code: ports.BrokerErrorUnavailable, Text: "dial refused", AdmissionCode: 1, FailureKind: domain.RemoteFailureTransport}
117}
118
119// unknownClientMessage and unknownServerMessage prove the directional
120// codecs fail closed on concrete types outside their closed union.
121type unknownClientMessage struct{}
122
123func (unknownClientMessage) brokerClientMessage() {}
124
125type unknownServerMessage struct{}
126
127func (unknownServerMessage) brokerServerMessage() {}
128
129const (
130 testEnvelopeCeiling = wire.AbsoluteEnvelopeLimit
131 testChunkCeiling = MaxStreamChunkBytes
132)
133
134func mustEncodeClient(t *testing.T, message ClientMessage) []byte {
135 t.Helper()
136 raw, err := EncodeClient(message, testEnvelopeCeiling, testChunkCeiling)
137 require.NoError(t, err)
138 return raw
139}
140
141func mustEncodeServer(t *testing.T, message ServerMessage) []byte {
142 t.Helper()
143 raw, err := EncodeServer(message, testEnvelopeCeiling, testChunkCeiling)
144 require.NoError(t, err)
145 return raw
146}
147
148// TestBrokerPreambleByteLengthHelper proves the exposed 4 KiB byte-length
149// helper accepts at the boundary and refuses one byte over, for both encoded
150// and received preambles.
151func TestBrokerPreambleByteLengthHelper(t *testing.T) {
152 require.Equal(t, wire.PreambleLimit, BrokerPreambleLimit)
153 require.Equal(t, 4<<10, BrokerPreambleLimit)
154 require.NoError(t, CheckPreambleSize(nil))
155 require.NoError(t, CheckPreambleSize(make([]byte, BrokerPreambleLimit)))
156 require.ErrorIs(t, CheckPreambleSize(make([]byte, BrokerPreambleLimit+1)), ErrPreambleRejected)
157
158 encoded, err := proto.Marshal(EncodePreambleRequest(defaultBrokerCeilings()))
159 require.NoError(t, err)
160 require.NoError(t, CheckPreambleSize(encoded))
161 response, err := proto.Marshal(EncodePreambleResponse(true, defaultBrokerCeilings(), 0))
162 require.NoError(t, err)
163 require.NoError(t, CheckPreambleSize(response))
164}
165
166// TestBrokerPreambleRoles proves the broker preamble uses roles 3/4, the
167// existing magic/epoch/exact version, zero capability bits, the 4 KiB
168// preamble bound, and precise refusal mapping.
169// TestBrokerCeilingsExportedSurface proves the exported ceiling value the
170// connection dispatcher consumes stays equivalent to the preamble helpers'
171// internal view: defaults are negotiable, minima are element-wise, and a
172// non-negotiable advertisement is refused.
173func TestBrokerCeilingsExportedSurface(t *testing.T) {
174 defaults := DefaultCeilings()
175 require.NoError(t, defaults.Validate())
176 require.Equal(t, defaultBrokerCeilings(), defaults)
177
178 smaller := Ceilings{MaxReceiveEnvelopeBytes: MinBrokerEnvelopeBytes, StreamChunkLimit: 1024}
179 require.NoError(t, smaller.Validate())
180 require.Equal(t, smaller, EffectiveCeilings(defaults, smaller))
181 require.Equal(t, smaller, EffectiveCeilings(smaller, defaults))
182 require.Equal(t, defaults, EffectiveCeilings(defaults, defaults))
183
184 for _, invalid := range []Ceilings{
185 {},
186 {MaxReceiveEnvelopeBytes: MinBrokerEnvelopeBytes - 1, StreamChunkLimit: MaxStreamChunkBytes},
187 {MaxReceiveEnvelopeBytes: MaxBrokerEnvelopeBytes + 1, StreamChunkLimit: MaxStreamChunkBytes},
188 {MaxReceiveEnvelopeBytes: MaxBrokerEnvelopeBytes, StreamChunkLimit: 0},
189 {MaxReceiveEnvelopeBytes: MaxBrokerEnvelopeBytes, StreamChunkLimit: MaxStreamChunkBytes + 1},
190 } {
191 require.ErrorIs(t, invalid.Validate(), ErrPreambleRejected)
192 }
193
194 // The decoded preamble's exported ceilings drive the codec ceilings.
195 request := EncodePreambleRequest(smaller)
196 decoded, err := DecodePreambleRequest(request)
197 require.NoError(t, err)
198 require.Equal(t, smaller, decoded.Ceilings)
199 require.NoError(t, decoded.Ceilings.Validate())
200}
201
202func TestBrokerPreambleRoles(t *testing.T) {
203 t.Run("client role is 3 and server role is 4", func(t *testing.T) {
204 request := EncodePreambleRequest(defaultBrokerCeilings())
205 require.Equal(t, BrokerRoleClient, request.GetRole().GetRole())
206 require.Equal(t, uint32(3), request.GetRole().GetRole())
207 response := EncodePreambleResponse(true, defaultBrokerCeilings(), 0)
208 require.Equal(t, BrokerRoleServer, response.GetRole().GetRole())
209 require.Equal(t, uint32(4), response.GetRole().GetRole())
210 })
211 t.Run("magic epoch version match session contract", func(t *testing.T) {
212 request := EncodePreambleRequest(defaultBrokerCeilings())
213 require.Equal(t, wire.PreambleMagic, request.GetMagic())
214 require.Equal(t, wire.ProtocolEpoch, request.GetEpoch())
215 require.Equal(t, uint32(protocol.Version), request.GetVersion())
216 require.Zero(t, request.GetCapabilityBits())
217 response := EncodePreambleResponse(true, defaultBrokerCeilings(), 0)
218 require.Equal(t, wire.PreambleMagic, response.GetMagic())
219 require.Equal(t, wire.ProtocolEpoch, response.GetEpoch())
220 require.Equal(t, uint32(protocol.Version), response.GetVersion())
221 require.Zero(t, response.GetCapabilityBits())
222 })
223 t.Run("preamble fits 4KiB", func(t *testing.T) {
224 raw, err := proto.Marshal(EncodePreambleRequest(defaultBrokerCeilings()))
225 require.NoError(t, err)
226 require.LessOrEqual(t, len(raw), BrokerPreambleLimit)
227 require.Equal(t, 4<<10, BrokerPreambleLimit)
228 raw, err = proto.Marshal(EncodePreambleResponse(true, defaultBrokerCeilings(), 0))
229 require.NoError(t, err)
230 require.LessOrEqual(t, len(raw), BrokerPreambleLimit)
231 })
232 t.Run("session roles rejected", func(t *testing.T) {
233 request := EncodePreambleRequest(defaultBrokerCeilings())
234 request.Role.Role = 1
235 _, err := DecodePreambleRequest(request)
236 require.ErrorIs(t, err, ErrPreambleRejected)
237 require.Equal(t, RejectionWrongRole, RejectionCodeFor(request, err))
238 request.Role.Role = 2
239 _, err = DecodePreambleRequest(request)
240 require.ErrorIs(t, err, ErrPreambleRejected)
241 response := EncodePreambleResponse(true, defaultBrokerCeilings(), 0)
242 response.Role.Role = 2
243 _, err = DecodePreambleResponse(response)
244 require.ErrorIs(t, err, ErrPreambleRejected)
245 })
246 t.Run("round trip negotiates minima", func(t *testing.T) {
247 local := defaultBrokerCeilings()
248 remote := brokerCeilings{MaxReceiveEnvelopeBytes: testEnvelopeCeiling / 2, StreamChunkLimit: 1024}
249 decoded, err := DecodePreambleRequest(EncodePreambleRequest(remote))
250 require.NoError(t, err)
251 require.Equal(t, remote, decoded.Ceilings)
252 require.Equal(t, remote, effectiveBrokerCeilings(local, decoded.Ceilings))
253 resp, err := DecodePreambleResponse(EncodePreambleResponse(true, remote, 0))
254 require.NoError(t, err)
255 require.True(t, resp.Accepted)
256 require.Equal(t, remote, resp.Ceilings)
257 })
258 t.Run("advertised envelope window 1MiB..16MiB", func(t *testing.T) {
259 require.Equal(t, uint64(1<<20), uint64(MinBrokerEnvelopeBytes))
260 require.Equal(t, uint64(16<<20), uint64(MaxBrokerEnvelopeBytes))
261 for _, advertised := range []uint64{0, MinBrokerEnvelopeBytes - 1, MaxBrokerEnvelopeBytes + 1} {
262 request := EncodePreambleRequest(defaultBrokerCeilings())
263 request.MaxReceiveEnvelopeBytes = advertised
264 _, err := DecodePreambleRequest(request)
265 require.ErrorIs(t, err, ErrPreambleRejected)
266 require.Equal(t, RejectionLimitRefused, RejectionCodeFor(request, err))
267 }
268 for _, advertised := range []uint64{MinBrokerEnvelopeBytes, MaxBrokerEnvelopeBytes} {
269 request := EncodePreambleRequest(defaultBrokerCeilings())
270 request.MaxReceiveEnvelopeBytes = advertised
271 _, err := DecodePreambleRequest(request)
272 require.NoError(t, err)
273 }
274 })
275 t.Run("stream chunk window 1..64KiB", func(t *testing.T) {
276 require.Equal(t, uint64(64<<10), uint64(MaxStreamChunkBytes))
277 for _, advertised := range []uint64{0, MaxStreamChunkBytes + 1} {
278 request := EncodePreambleRequest(defaultBrokerCeilings())
279 request.OutputDataLimit = advertised
280 _, err := DecodePreambleRequest(request)
281 require.ErrorIs(t, err, ErrPreambleRejected)
282 require.Equal(t, RejectionLimitRefused, RejectionCodeFor(request, err))
283 }
284 for _, advertised := range []uint64{MinStreamChunkBytes, MaxStreamChunkBytes} {
285 request := EncodePreambleRequest(defaultBrokerCeilings())
286 request.OutputDataLimit = advertised
287 _, err := DecodePreambleRequest(request)
288 require.NoError(t, err)
289 }
290 })
291 t.Run("caps zero", func(t *testing.T) {
292 request := EncodePreambleRequest(defaultBrokerCeilings())
293 request.CapabilityBits = 1
294 _, err := DecodePreambleRequest(request)
295 require.ErrorIs(t, err, ErrPreambleRejected)
296 require.Equal(t, RejectionLimitRefused, RejectionCodeFor(request, err))
297 response := EncodePreambleResponse(true, defaultBrokerCeilings(), 0)
298 response.CapabilityBits = 7
299 _, err = DecodePreambleResponse(response)
300 require.ErrorIs(t, err, ErrPreambleRejected)
301 })
302 t.Run("refusal mapping is precise", func(t *testing.T) {
303 cases := []struct {
304 name string
305 mutate func(*wire.PreambleRequest)
306 want uint32
307 }{
308 {"bad magic", func(r *wire.PreambleRequest) { r.Magic = 0xDEAD }, RejectionBadMagic},
309 {"epoch", func(r *wire.PreambleRequest) { r.Epoch++ }, RejectionEpochMismatch},
310 {"version", func(r *wire.PreambleRequest) { r.Version++ }, RejectionVersionMismatch},
311 {"role", func(r *wire.PreambleRequest) { r.Role.Role = 9 }, RejectionWrongRole},
312 {"caps", func(r *wire.PreambleRequest) { r.CapabilityBits = 1 }, RejectionLimitRefused},
313 {"envelope floor", func(r *wire.PreambleRequest) { r.MaxReceiveEnvelopeBytes = 1 }, RejectionLimitRefused},
314 {"chunk", func(r *wire.PreambleRequest) { r.OutputDataLimit = 0 }, RejectionLimitRefused},
315 }
316 for _, tc := range cases {
317 t.Run(tc.name, func(t *testing.T) {
318 request := EncodePreambleRequest(defaultBrokerCeilings())
319 tc.mutate(request)
320 _, err := DecodePreambleRequest(request)
321 require.ErrorIs(t, err, ErrPreambleRejected)
322 require.Equal(t, tc.want, RejectionCodeFor(request, err))
323 })
324 }
325 // Refusal responses decode as rejected with their code.
326 response := EncodePreambleResponse(false, defaultBrokerCeilings(), RejectionVersionMismatch)
327 require.False(t, response.GetAccepted())
328 require.Equal(t, RejectionVersionMismatch, response.GetRejection().GetCode())
329 decoded, err := DecodePreambleResponse(response)
330 require.ErrorIs(t, err, ErrPreambleRejected)
331 require.Equal(t, RejectionVersionMismatch, decoded.Code)
332 })
333 t.Run("nil preambles rejected", func(t *testing.T) {
334 _, err := DecodePreambleRequest(nil)
335 require.ErrorIs(t, err, ErrPreambleRejected)
336 _, err = DecodePreambleResponse(nil)
337 require.ErrorIs(t, err, ErrPreambleRejected)
338 })
339}
340
341// TestBrokerClientVariants inventories every client variant tag.
342func TestBrokerClientVariants(t *testing.T) {
343 connection := testConnectionID(0x11)
344 operation := testOperationID(0x44)
345 messages := []struct {
346 name string
347 message ClientMessage
348 tag protowire.Number
349 }{
350 {"register", Register{}, 101},
351 {"subscribe", Subscribe{Epoch: 7, Connection: connection, Generation: 1}, 102},
352 {"resync", Resync{Epoch: 7, Connection: connection, Generation: 2}, 103},
353 {"unsubscribe", Unsubscribe{Epoch: 7, Connection: connection, Generation: 3}, 104},
354 {"add_host", AddHost{Epoch: 7, Connection: connection, Operation: operation, Endpoint: "dev@host:22", Policy: testPolicy()}, 105},
355 {"remove_host", RemoveHost{Epoch: 7, Connection: connection, Operation: operation, Registration: testRegistration()}, 106},
356 {"reconcile", Reconcile{Epoch: 7, Connection: connection, Registration: testRegistration()}, 107},
357 {"open_stream", OpenStream{Epoch: 7, Connection: connection, Stream: 3, Purpose: ports.BrokerStreamAttachment, Admission: ports.BrokerAdmissionExact, Endpoint: "dev@host:22", Registration: testRegistration(), Target: testTarget(), Env: []string{"TERM=xterm"}, Policy: testPolicy(), StartMode: ports.BrokerDaemonStartIfNeeded}, 108},
358 {"client_stream_data", ClientStreamData{Epoch: 7, Connection: connection, Stream: 3, Data: []byte("frame")}, 109},
359 {"close_stream", CloseStream{Epoch: 7, Connection: connection, Stream: 3}, 110},
360 {"update_host_policy", UpdateHostPolicy{Epoch: 7, Connection: connection, Operation: operation, Registration: testRegistration(), Policy: testPolicy()}, 111},
361 {"start_preview", testStartPreview(connection), 112},
362 {"cancel_preview", CancelPreview{Epoch: 7, Connection: connection, Generation: 3}, 113},
363 {"stream_window_update", StreamWindowUpdate{Epoch: 7, Connection: connection, Stream: 3, Credit: 4096}, 114},
364 }
365 require.Len(t, messages, 14)
366 for _, tc := range messages {
367 t.Run(tc.name, func(t *testing.T) {
368 raw := mustEncodeClient(t, tc.message)
369 number, wireType, consumed := protowire.ConsumeTag(raw)
370 require.Greater(t, consumed, 0)
371 require.Equal(t, tc.tag, number)
372 require.Equal(t, protowire.BytesType, wireType)
373 decoded, err := DecodeClient(raw, testEnvelopeCeiling, testChunkCeiling)
374 require.NoError(t, err)
375 require.Equal(t, tc.message, decoded)
376 })
377 }
378}
379
380// TestBrokerStreamDataOwnsItsBytes proves decoded stream data never aliases the
381// payload: receivers reuse their buffers, and brokeripc queues Data as is.
382func TestBrokerStreamDataOwnsItsBytes(t *testing.T) {
383 connection := testConnectionID(0x11)
384 client := mustEncodeClient(t, ClientStreamData{Epoch: 7, Connection: connection, Stream: 3, Data: []byte("frame")})
385 server := mustEncodeServer(t, ServerStreamData{Epoch: 7, Connection: connection, Stream: 3, Data: []byte("frame")})
386 decodedClient, err := DecodeClient(client, testEnvelopeCeiling, testChunkCeiling)
387 require.NoError(t, err)
388 decodedServer, err := DecodeServer(server, testEnvelopeCeiling, testChunkCeiling)
389 require.NoError(t, err)
390 clear(client)
391 clear(server)
392 require.Equal(t, []byte("frame"), decodedClient.(ClientStreamData).Data)
393 require.Equal(t, []byte("frame"), decodedServer.(ServerStreamData).Data)
394}
395
396// TestBrokerOpenStreamAdmissionVariants pins the attachment-admission
397// contract: exact, create-named, and create-ephemeral round-trip, control
398// carries none, and wrong admission/purpose/name/target combinations are
399// refused on decode exactly like ports.BrokerOpenStreamRequest.Validate.
400func TestBrokerOpenStreamAdmissionVariants(t *testing.T) {
401 connection := testConnectionID(0x11)
402 base := func() OpenStream {
403 return OpenStream{
404 Epoch: 7, Connection: connection, Stream: 3, Purpose: ports.BrokerStreamAttachment,
405 Endpoint: "dev@host:22", Registration: testRegistration(), Policy: testPolicy(),
406 // An attachment or creation always carries the explicit
407 // start-if-needed authorization; the zero mode is refused.
408 StartMode: ports.BrokerDaemonStartIfNeeded,
409 }
410 }
411 roundTrips := []struct {
412 name string
413 message OpenStream
414 }{
415 {"exact", func() OpenStream {
416 message := base()
417 message.Admission = ports.BrokerAdmissionExact
418 message.Target = testTarget()
419 return message
420 }()},
421 {"create named", func() OpenStream {
422 message := base()
423 message.Admission = ports.BrokerAdmissionCreateNamed
424 message.Name = "work"
425 return message
426 }()},
427 {"create ephemeral", func() OpenStream {
428 message := base()
429 message.Admission = ports.BrokerAdmissionCreateEphemeral
430 return message
431 }()},
432 {"attach named", func() OpenStream {
433 message := base()
434 message.Admission = ports.BrokerAdmissionAttachNamed
435 message.Name = "work"
436 return message
437 }()},
438 {"control carries none", OpenStream{
439 Epoch: 7, Connection: connection, Stream: 3, Purpose: ports.BrokerStreamControl,
440 Local: true, Policy: testPolicy(), StartMode: ports.BrokerDaemonStartIfNeeded,
441 }},
442 }
443 for _, tc := range roundTrips {
444 t.Run(tc.name, func(t *testing.T) {
445 raw := mustEncodeClient(t, tc.message)
446 decoded, err := DecodeClient(raw, testEnvelopeCeiling, testChunkCeiling)
447 require.NoError(t, err)
448 require.Equal(t, tc.message, decoded)
449 })
450 }
451
452 t.Run("wrong combinations refused", func(t *testing.T) {
453 exact := func() OpenStream {
454 message := base()
455 message.Admission = ports.BrokerAdmissionExact
456 message.Target = testTarget()
457 return message
458 }
459 cases := map[string]OpenStream{
460 "attachment without admission": func() OpenStream { m := base(); return m }(),
461 "exact with name": func() OpenStream {
462 m := exact()
463 m.Name = "work"
464 return m
465 }(),
466 "named with target": func() OpenStream {
467 m := base()
468 m.Admission = ports.BrokerAdmissionCreateNamed
469 m.Name = "work"
470 m.Target = testTarget()
471 return m
472 }(),
473 "named without name": func() OpenStream {
474 m := base()
475 m.Admission = ports.BrokerAdmissionCreateNamed
476 return m
477 }(),
478 "attach named with target": func() OpenStream {
479 m := base()
480 m.Admission = ports.BrokerAdmissionAttachNamed
481 m.Name = "work"
482 m.Target = testTarget()
483 return m
484 }(),
485 "attach named without name": func() OpenStream {
486 m := base()
487 m.Admission = ports.BrokerAdmissionAttachNamed
488 return m
489 }(),
490 "ephemeral with name": func() OpenStream {
491 m := base()
492 m.Admission = ports.BrokerAdmissionCreateEphemeral
493 m.Name = "work"
494 return m
495 }(),
496 "control with admission": OpenStream{
497 Epoch: 7, Connection: connection, Stream: 3, Purpose: ports.BrokerStreamControl,
498 Local: true, Policy: testPolicy(), Admission: ports.BrokerAdmissionExact, StartMode: ports.BrokerDaemonStartIfNeeded,
499 },
500 "attachment with zero start mode": func() OpenStream { m := exact(); m.StartMode = 0; return m }(),
501 "attachment with existing-only start mode": func() OpenStream {
502 m := exact()
503 m.StartMode = ports.BrokerDaemonExistingOnly
504 return m
505 }(),
506 "observation with start-if-needed start mode": OpenStream{
507 Epoch: 7, Connection: connection, Stream: 3, Purpose: ports.BrokerStreamObservation,
508 Local: true, Policy: testPolicy(), StartMode: ports.BrokerDaemonStartIfNeeded,
509 },
510 }
511 for name, message := range cases {
512 t.Run(name, func(t *testing.T) {
513 // Start mode is a closed wire enum and is refused even by the encoder.
514 if message.StartMode == 0 {
515 _, err := EncodeClient(message, testEnvelopeCeiling, testChunkCeiling)
516 require.Error(t, err)
517 return
518 }
519 raw, err := EncodeClient(message, testEnvelopeCeiling, testChunkCeiling)
520 require.NoError(t, err)
521 _, err = DecodeClient(raw, testEnvelopeCeiling, testChunkCeiling)
522 if name == "named with target" || name == "attach named with target" {
523 // The encoder omits a target on creation; test the untrusted wire directly.
524 envelope := &wire.BrokerClientEnvelope{}
525 require.NoError(t, proto.Unmarshal(raw, envelope))
526 envelope.GetOpenStream().Target = exactTargetToWire(testTarget())
527 raw, err = proto.Marshal(envelope)
528 require.NoError(t, err)
529 _, err = DecodeClient(raw, testEnvelopeCeiling, testChunkCeiling)
530 }
531 require.ErrorIs(t, err, ErrInvalidMessage)
532 })
533 }
534 })
535
536 t.Run("unknown admission code refused on decode", func(t *testing.T) {
537 message := base()
538 message.Admission = ports.BrokerAdmissionExact
539 message.Target = testTarget()
540 raw := mustEncodeClient(t, message)
541 envelope := &wire.BrokerClientEnvelope{}
542 require.NoError(t, wire.ScanEnvelope(envelope, raw))
543 require.NoError(t, proto.Unmarshal(raw, envelope))
544 envelope.GetOpenStream().Admission = 9
545 mutated, err := proto.Marshal(envelope)
546 require.NoError(t, err)
547 _, err = DecodeClient(mutated, testEnvelopeCeiling, testChunkCeiling)
548 require.ErrorIs(t, err, ErrInvalidMessage)
549 })
550
551 t.Run("start mode round-trips and is never defaulted", func(t *testing.T) {
552 // The mode round-trips with the purpose it is legal for: an existing-only
553 // authorization belongs to control/observation, a start-if-needed one to
554 // attachment/control.
555 control := func(mode ports.BrokerDaemonStartMode) OpenStream {
556 return OpenStream{Epoch: 7, Connection: connection, Stream: 3, Purpose: ports.BrokerStreamControl, Local: true, Policy: testPolicy(), StartMode: mode}
557 }
558 for _, message := range []OpenStream{
559 control(ports.BrokerDaemonExistingOnly),
560 control(ports.BrokerDaemonStartIfNeeded),
561 func() OpenStream {
562 m := base()
563 m.Admission = ports.BrokerAdmissionExact
564 m.Target = testTarget()
565 m.StartMode = ports.BrokerDaemonStartIfNeeded
566 return m
567 }(),
568 } {
569 decoded, err := DecodeClient(mustEncodeClient(t, message), testEnvelopeCeiling, testChunkCeiling)
570 require.NoError(t, err)
571 admitting, ok := decoded.(OpenStream)
572 require.True(t, ok)
573 require.Equal(t, message.StartMode, admitting.StartMode)
574 }
575 // An absent wire mode is refused, never widened into a spawn-capable
576 // default, and an unknown code is refused too.
577 message := base()
578 message.Admission = ports.BrokerAdmissionExact
579 message.Target = testTarget()
580 raw := mustEncodeClient(t, message)
581 envelope := &wire.BrokerClientEnvelope{}
582 require.NoError(t, proto.Unmarshal(raw, envelope))
583 envelope.GetOpenStream().StartMode = 0
584 mutated, err := proto.Marshal(envelope)
585 require.NoError(t, err)
586 _, err = DecodeClient(mutated, testEnvelopeCeiling, testChunkCeiling)
587 require.ErrorIs(t, err, ErrInvalidMessage)
588 envelope.GetOpenStream().StartMode = 3
589 mutated, err = proto.Marshal(envelope)
590 require.NoError(t, err)
591 _, err = DecodeClient(mutated, testEnvelopeCeiling, testChunkCeiling)
592 require.ErrorIs(t, err, ErrInvalidMessage)
593 // An observation that authorized a spawn is refused rather than narrowed.
594 envelope.GetOpenStream().Purpose = 3
595 envelope.GetOpenStream().StartMode = 2
596 envelope.GetOpenStream().Target = nil
597 envelope.GetOpenStream().Admission = 0
598 mutated, err = proto.Marshal(envelope)
599 require.NoError(t, err)
600 _, err = DecodeClient(mutated, testEnvelopeCeiling, testChunkCeiling)
601 require.ErrorIs(t, err, ErrInvalidMessage)
602 })
603}
604
605// TestBrokerSnapshotDaemonByteForByte proves the daemon and session parts are
606// byte-for-byte stable across an encode/decode/re-encode cycle, including an
607// observed identity/incarnation/version/capability set, the local daemon, and a
608// locally-bound session.
609func TestBrokerSnapshotDaemonByteForByte(t *testing.T) {
610 connection := testConnectionID(0x11)
611 messages := []struct {
612 name string
613 message SnapshotPart
614 }{
615 {"observed remote daemon", SnapshotPart{Epoch: 7, Connection: connection, Generation: 2, Revision: 5, Index: 1, Part: SnapshotDaemonPart{HostIndex: 1, Daemon: testDaemonObservation(), SessionCount: 1}}},
616 {"reachable remote without daemon", SnapshotPart{Epoch: 7, Connection: connection, Generation: 2, Revision: 5, Index: 2, Part: SnapshotDaemonPart{HostIndex: 1, Daemon: func() ports.BrokerDaemonObservation {
617 daemon := testDaemonObservation()
618 daemon.Identity = ""
619 daemon.Incarnation = ports.BrokerDaemonIncarnation{}
620 daemon.ProtocolVersion = 0
621 daemon.Capabilities = 0
622 daemon.Availability = domain.RemoteAvailabilityNoDaemon
623 daemon.LastFailure = domain.RemoteFailure{}
624 daemon.LastSuccess = time.Time{}
625 daemon.InventoryKnown = false
626 return daemon
627 }(), SessionCount: 0}}},
628 {"local daemon", SnapshotPart{Epoch: 7, Connection: connection, Generation: 2, Revision: 5, Index: 0, Part: SnapshotDaemonPart{HostIndex: 0, Daemon: testLocalDaemonObservation(), SessionCount: 1}}},
629 {"local session", SnapshotPart{Epoch: 7, Connection: connection, Generation: 2, Revision: 5, Index: 2, Part: SnapshotSessionPart{HostIndex: 0, SessionIndex: 0, Local: true, Session: testCatalogSession()}}},
630 }
631 for _, tc := range messages {
632 t.Run(tc.name, func(t *testing.T) {
633 raw := mustEncodeServer(t, tc.message)
634 decoded, err := DecodeServer(raw, testEnvelopeCeiling, testChunkCeiling)
635 require.NoError(t, err)
636 require.Equal(t, tc.message, decoded)
637 again, err := EncodeServer(decoded, testEnvelopeCeiling, testChunkCeiling)
638 require.NoError(t, err)
639 require.Equal(t, raw, again, "re-encode must be byte-for-byte identical")
640 })
641 }
642}
643
644// TestBrokerServerVariants inventories every server variant tag.
645func TestBrokerServerVariants(t *testing.T) {
646 connection := testConnectionID(0x11)
647 operation := testOperationID(0x77)
648 messages := []struct {
649 name string
650 message ServerMessage
651 tag protowire.Number
652 }{
653 {"registered", Registered{Epoch: 7, Connection: connection}, 201},
654 {"snapshot_begin", SnapshotPart{Epoch: 7, Connection: connection, Generation: 2, Revision: 5, Index: 0, Part: SnapshotBegin{HostCount: 2, SessionCount: 2, TombstoneCount: 0, LocalPresent: true}}, 202},
655 {"snapshot_daemon", SnapshotPart{Epoch: 7, Connection: connection, Generation: 2, Revision: 5, Index: 1, Part: SnapshotDaemonPart{HostIndex: 0, Daemon: testLocalDaemonObservation(), SessionCount: 0}}, 202},
656 {"snapshot_daemon_remote", SnapshotPart{Epoch: 7, Connection: connection, Generation: 2, Revision: 5, Index: 2, Part: SnapshotDaemonPart{HostIndex: 1, Daemon: testDaemonObservation(), SessionCount: 1}}, 202},
657 {"snapshot_session", SnapshotPart{Epoch: 7, Connection: connection, Generation: 2, Revision: 5, Index: 3, Part: SnapshotSessionPart{HostIndex: 1, SessionIndex: 0, Session: testCatalogSession()}}, 202},
658 {"snapshot_session_local", SnapshotPart{Epoch: 7, Connection: connection, Generation: 2, Revision: 5, Index: 4, Part: SnapshotSessionPart{HostIndex: 0, SessionIndex: 0, Local: true, Session: testCatalogSession()}}, 202},
659 {"snapshot_tombstone", SnapshotPart{Epoch: 7, Connection: connection, Generation: 2, Revision: 5, Index: 5, Part: SnapshotTombstonePart{TombstoneIndex: 0, Registration: testRegistration(), RetiredRevision: 4}}, 202},
660 {"snapshot_end", SnapshotPart{Epoch: 7, Connection: connection, Generation: 2, Revision: 5, Index: 6, Part: SnapshotEnd{}}, 202},
661 {"operation_result", OperationResult{Epoch: 7, Connection: connection, Operation: operation, Outcome: ports.BrokerOutcomeFailed, Error: testErrorDetail(), HasError: true}, 203},
662 {"operation_result_add", OperationResult{Epoch: 7, Connection: connection, Operation: operation, Outcome: ports.BrokerOutcomeOK, Registration: testRegistration()}, 203},
663 {"operation_result_remove", OperationResult{Epoch: 7, Connection: connection, Operation: operation, Outcome: ports.BrokerOutcomeOK, Removed: true}, 203},
664 {"stream_opened", StreamOpened{Epoch: 7, Connection: connection, Stream: 3}, 204},
665 {"server_stream_data", ServerStreamData{Epoch: 7, Connection: connection, Stream: 3, Data: []byte("frame")}, 205},
666 {"stream_closed", StreamClosed{Epoch: 7, Connection: connection, Stream: 3, Error: testErrorDetail(), HasError: true}, 206},
667 {"stream_closed_orderly", StreamClosed{Epoch: 7, Connection: connection, Stream: 4}, 206},
668 {"progress", Progress{Epoch: 7, Connection: connection, Stream: 3, Phase: BrokerProgressProbing, Text: "probing"}, 207},
669 {"broker_error", BrokerErrorMessage{Epoch: 7, Connection: connection, Error: testErrorDetail()}, 208},
670 {"shutdown", Shutdown{Epoch: 7, Connection: connection, Reason: ShutdownTerminating, Text: "stopping"}, 209},
671 {"preview_publication", testPreviewPublication(connection), 210},
672 {"preview_publication_error", PreviewPublication{Epoch: 7, Connection: connection, Generation: 5, Error: testErrorDetail(), HasError: true}, 210},
673 {"stream_window_update", StreamWindowUpdate{Epoch: 7, Connection: connection, Stream: 3, Credit: StreamWindowBytes}, 211},
674 }
675 for _, tc := range messages {
676 t.Run(tc.name, func(t *testing.T) {
677 raw := mustEncodeServer(t, tc.message)
678 number, wireType, consumed := protowire.ConsumeTag(raw)
679 require.Greater(t, consumed, 0)
680 require.Equal(t, tc.tag, number)
681 require.Equal(t, protowire.BytesType, wireType)
682 decoded, err := DecodeServer(raw, testEnvelopeCeiling, testChunkCeiling)
683 require.NoError(t, err)
684 require.Equal(t, tc.message, decoded)
685 })
686 }
687}
688
689// TestBrokerDownSessionEmptyTabsRoundTrip proves a down session with an
690// empty-present tab list survives a full encode/decode round trip: proto3
691// repeated fields carry no presence, so the decoder must map an absent
692// field back to an empty-present list rather than refusing it.
693func TestBrokerDownSessionEmptyTabsRoundTrip(t *testing.T) {
694 connection := testConnectionID(0x11)
695 session := catalogue.RemoteCatalogSession{
696 LifecycleID: domain.SessionLifecycleID{1, 2, 3},
697 Name: "down",
698 State: catalogue.RemoteCatalogSessionDown,
699 Tabs: []catalogue.RemoteCatalogTab{},
700 }
701 message := SnapshotPart{
702 Epoch: 7, Connection: connection, Generation: 2, Revision: 5, Index: 2,
703 Part: SnapshotSessionPart{HostIndex: 0, SessionIndex: 0, Session: session},
704 }
705 raw := mustEncodeServer(t, message)
706 decoded, err := DecodeServer(raw, testEnvelopeCeiling, testChunkCeiling)
707 require.NoError(t, err)
708 require.Equal(t, message, decoded)
709
710 part, ok := decoded.(SnapshotPart)
711 require.True(t, ok)
712 sessionPart, ok := part.Part.(SnapshotSessionPart)
713 require.True(t, ok)
714 require.NotNil(t, sessionPart.Session.Tabs, "empty tabs decode as empty-present, never nil")
715 require.Empty(t, sessionPart.Session.Tabs)
716}
717
718// TestBrokerTabIndexBound proves a tab index at the per-session maximum is a
719// range refusal, not a semantic one: valid indexes are 0..max-1.
720func TestBrokerTabIndexBound(t *testing.T) {
721 connection := testConnectionID(0x11)
722 envelope := &wire.BrokerServerEnvelope{Payload: &wire.BrokerServerEnvelope_SnapshotPart{SnapshotPart: &wire.SnapshotPart{
723 Scope: &wire.BrokerScope{BrokerEpoch: 7, ConnectionId: connection[:]},
724 Generation: 2,
725 Revision: 5,
726 Index: 2,
727 Part: &wire.SnapshotPart_Session{Session: &wire.SnapshotSession{
728 HostIndex: 0, SessionIndex: 0,
729 Session: &wire.BrokerCatalogSession{
730 LifecycleId: []byte{1, 2, 3},
731 Name: "down",
732 State: 2,
733 Tabs: []*wire.BrokerCatalogTab{{Id: "tab", Index: catalogue.RemoteCatalogMaxTabsPerSess, Name: "tab"}},
734 },
735 }},
736 }}}
737 raw, err := proto.Marshal(envelope)
738 require.NoError(t, err)
739 _, err = DecodeServer(raw, testEnvelopeCeiling, testChunkCeiling)
740 require.ErrorIs(t, err, errConvertRange)
741}
742
743// mutateDaemonIndex re-encodes one daemon snapshot part envelope with its
744// daemon index overwritten, so decode can be fed a value encode would never
745// emit.
746func mutateDaemonIndex(t *testing.T, part SnapshotPart, hostIndex uint32) []byte {
747 t.Helper()
748 raw := mustEncodeServer(t, part)
749 envelope := &wire.BrokerServerEnvelope{}
750 require.NoError(t, wire.ScanEnvelope(envelope, raw))
751 require.NoError(t, proto.Unmarshal(raw, envelope))
752 envelope.GetSnapshotPart().GetDaemon().HostIndex = hostIndex
753 mutated, err := proto.Marshal(envelope)
754 require.NoError(t, err)
755 return mutated
756}
757
758// mutateSessionIndexes re-encodes one session snapshot part envelope with its
759// host and session indexes overwritten.
760func mutateSessionIndexes(t *testing.T, part SnapshotPart, hostIndex, sessionIndex uint32) []byte {
761 t.Helper()
762 raw := mustEncodeServer(t, part)
763 envelope := &wire.BrokerServerEnvelope{}
764 require.NoError(t, wire.ScanEnvelope(envelope, raw))
765 require.NoError(t, proto.Unmarshal(raw, envelope))
766 envelope.GetSnapshotPart().GetSession().HostIndex = hostIndex
767 envelope.GetSnapshotPart().GetSession().SessionIndex = sessionIndex
768 mutated, err := proto.Marshal(envelope)
769 require.NoError(t, err)
770 return mutated
771}
772
773// TestBrokerSnapshotIndexBounds proves snapshot part indexes are strict
774// ranges: max-1 is accepted on both encode and decode, and max is refused as a
775// range refusal on both paths. Daemon indexes are bounded by
776// BrokerMaxDaemonsPerSnapshot and session indexes by BrokerMaxSessionsPerHost.
777func TestBrokerSnapshotIndexBounds(t *testing.T) {
778 connection := testConnectionID(0x11)
779 daemonPart := func(hostIndex uint32) SnapshotPart {
780 return SnapshotPart{Epoch: 7, Connection: connection, Generation: 2, Revision: 5, Index: 1, Part: SnapshotDaemonPart{HostIndex: hostIndex, Daemon: testDaemonObservation()}}
781 }
782 sessionPart := func(hostIndex, sessionIndex uint32) SnapshotPart {
783 return SnapshotPart{Epoch: 7, Connection: connection, Generation: 2, Revision: 5, Index: 2, Part: SnapshotSessionPart{HostIndex: hostIndex, SessionIndex: sessionIndex, Session: testCatalogSession()}}
784 }
785
786 t.Run("daemon part at max-1 accepted", func(t *testing.T) {
787 message := daemonPart(ports.BrokerMaxDaemonsPerSnapshot - 1)
788 raw := mustEncodeServer(t, message)
789 decoded, err := DecodeServer(raw, testEnvelopeCeiling, testChunkCeiling)
790 require.NoError(t, err)
791 require.Equal(t, message, decoded)
792 decoded, err = DecodeServer(mutateDaemonIndex(t, message, ports.BrokerMaxDaemonsPerSnapshot-1), testEnvelopeCeiling, testChunkCeiling)
793 require.NoError(t, err)
794 require.Equal(t, message, decoded)
795 })
796
797 t.Run("daemon part at max refused", func(t *testing.T) {
798 _, err := EncodeServer(daemonPart(ports.BrokerMaxDaemonsPerSnapshot), testEnvelopeCeiling, testChunkCeiling)
799 require.ErrorIs(t, err, ErrTooLarge)
800 _, err = DecodeServer(mutateDaemonIndex(t, daemonPart(0), ports.BrokerMaxDaemonsPerSnapshot), testEnvelopeCeiling, testChunkCeiling)
801 require.ErrorIs(t, err, ErrTooLarge)
802 })
803
804 t.Run("daemon part at uint32 maximum refused", func(t *testing.T) {
805 _, err := EncodeServer(daemonPart(^uint32(0)), testEnvelopeCeiling, testChunkCeiling)
806 require.ErrorIs(t, err, ErrTooLarge)
807 _, err = DecodeServer(mutateDaemonIndex(t, daemonPart(0), ^uint32(0)), testEnvelopeCeiling, testChunkCeiling)
808 require.ErrorIs(t, err, ErrTooLarge)
809 })
810
811 t.Run("session part at max-1 accepted", func(t *testing.T) {
812 message := sessionPart(ports.BrokerMaxDaemonsPerSnapshot-1, ports.BrokerMaxSessionsPerHost-1)
813 raw := mustEncodeServer(t, message)
814 decoded, err := DecodeServer(raw, testEnvelopeCeiling, testChunkCeiling)
815 require.NoError(t, err)
816 require.Equal(t, message, decoded)
817 decoded, err = DecodeServer(mutateSessionIndexes(t, message, ports.BrokerMaxDaemonsPerSnapshot-1, ports.BrokerMaxSessionsPerHost-1), testEnvelopeCeiling, testChunkCeiling)
818 require.NoError(t, err)
819 require.Equal(t, message, decoded)
820 })
821
822 t.Run("session part host index at max refused", func(t *testing.T) {
823 _, err := EncodeServer(sessionPart(ports.BrokerMaxDaemonsPerSnapshot, 0), testEnvelopeCeiling, testChunkCeiling)
824 require.ErrorIs(t, err, ErrTooLarge)
825 _, err = DecodeServer(mutateSessionIndexes(t, sessionPart(0, 0), ports.BrokerMaxDaemonsPerSnapshot, 0), testEnvelopeCeiling, testChunkCeiling)
826 require.ErrorIs(t, err, ErrTooLarge)
827 })
828
829 t.Run("session part session index at max refused", func(t *testing.T) {
830 _, err := EncodeServer(sessionPart(0, ports.BrokerMaxSessionsPerHost), testEnvelopeCeiling, testChunkCeiling)
831 require.ErrorIs(t, err, ErrTooLarge)
832 _, err = DecodeServer(mutateSessionIndexes(t, sessionPart(0, 0), 0, ports.BrokerMaxSessionsPerHost), testEnvelopeCeiling, testChunkCeiling)
833 require.ErrorIs(t, err, ErrTooLarge)
834 })
835}
836
837// TestBrokerGoldenVectors pins byte-for-byte wire encodings.
838func TestBrokerGoldenVectors(t *testing.T) {
839 connection := testConnectionID(0x11)
840 raw := mustEncodeClient(t, Register{})
841 decoded, err := DecodeClient(raw, testEnvelopeCeiling, testChunkCeiling)
842 require.NoError(t, err)
843 require.Equal(t, Register{}, decoded)
844 // Re-encode is deterministic: the same message marshals identically.
845 again, err := EncodeClient(Register{}, testEnvelopeCeiling, testChunkCeiling)
846 require.NoError(t, err)
847 require.Equal(t, raw, again)
848
849 sub := Subscribe{Epoch: 7, Connection: connection, Generation: 1 << 40}
850 rawSub := mustEncodeClient(t, sub)
851 envelope := &wire.BrokerClientEnvelope{}
852 require.NoError(t, wire.ScanEnvelope(envelope, rawSub))
853 require.NoError(t, proto.Unmarshal(rawSub, envelope))
854 require.Equal(t, uint64(7), envelope.GetSubscribe().GetScope().GetBrokerEpoch())
855 require.Equal(t, uint64(1<<40), envelope.GetSubscribe().GetGeneration())
856 back, err := DecodeClient(rawSub, testEnvelopeCeiling, testChunkCeiling)
857 require.NoError(t, err)
858 require.Equal(t, sub, back)
859
860 // The passive reader flag survives the round trip, and an active
861 // subscription encodes without it (proto3 omits false).
862 passive := sub
863 passive.Passive = true
864 rawPassive := mustEncodeClient(t, passive)
865 back, err = DecodeClient(rawPassive, testEnvelopeCeiling, testChunkCeiling)
866 require.NoError(t, err)
867 require.Equal(t, passive, back)
868 require.Greater(t, len(rawPassive), len(rawSub))
869}
870
871// TestBrokerBounds proves stateless bound refusals.
872func TestBrokerBounds(t *testing.T) {
873 connection := testConnectionID(0x11)
874 t.Run("envelope ceiling", func(t *testing.T) {
875 raw := mustEncodeClient(t, Register{})
876 _, err := DecodeClient(raw, uint64(len(raw)-1), testChunkCeiling)
877 require.ErrorIs(t, err, wire.ErrScanLength)
878 _, err = DecodeClient(make([]byte, testEnvelopeCeiling+1), testEnvelopeCeiling, testChunkCeiling)
879 require.ErrorIs(t, err, wire.ErrScanLength)
880 })
881 t.Run("chunk bound", func(t *testing.T) {
882 big := bytes.Repeat([]byte{0x66}, int(MaxStreamChunkBytes)+1)
883 _, err := EncodeClient(ClientStreamData{Epoch: 7, Connection: connection, Stream: 3, Data: big}, testEnvelopeCeiling, testEnvelopeCeiling)
884 require.ErrorIs(t, err, ErrTooLarge)
885 _, err = EncodeServer(ServerStreamData{Epoch: 7, Connection: connection, Stream: 3, Data: big}, testEnvelopeCeiling, testEnvelopeCeiling)
886 require.ErrorIs(t, err, ErrTooLarge)
887 // A negotiated chunk below the absolute max also refuses.
888 raw := mustEncodeClient(t, ClientStreamData{Epoch: 7, Connection: connection, Stream: 3, Data: []byte("12345678")})
889 _, err = DecodeClient(raw, testEnvelopeCeiling, 4)
890 require.ErrorIs(t, err, ErrTooLarge)
891 // Empty chunk is invalid, not too large.
892 _, err = EncodeClient(ClientStreamData{Epoch: 7, Connection: connection, Stream: 3}, testEnvelopeCeiling, testChunkCeiling)
893 require.ErrorIs(t, err, ErrInvalidMessage)
894 })
895 t.Run("env entries", func(t *testing.T) {
896 many := make([]string, ports.BrokerMaxEnvEntries+1)
897 for i := range many {
898 many[i] = "A=B"
899 }
900 _, err := EncodeClient(OpenStream{Epoch: 7, Connection: connection, Stream: 3, Purpose: ports.BrokerStreamAttachment, Admission: ports.BrokerAdmissionExact, Endpoint: "dev@host:22", Registration: testRegistration(), Target: testTarget(), Env: many, Policy: testPolicy()}, testEnvelopeCeiling, testChunkCeiling)
901 require.ErrorIs(t, err, ErrTooLarge)
902 oversize := []string{"K=" + strings.Repeat("x", ports.BrokerMaxEnvEntryBytes)}
903 _, err = EncodeClient(OpenStream{Epoch: 7, Connection: connection, Stream: 3, Purpose: ports.BrokerStreamAttachment, Admission: ports.BrokerAdmissionExact, Endpoint: "dev@host:22", Registration: testRegistration(), Target: testTarget(), Env: oversize, Policy: testPolicy()}, testEnvelopeCeiling, testChunkCeiling)
904 require.ErrorIs(t, err, ErrTooLarge)
905 })
906 t.Run("error text", func(t *testing.T) {
907 oversize := testErrorDetail()
908 oversize.Text = strings.Repeat("e", ports.BrokerMaxErrorBytes+1)
909 _, err := EncodeServer(BrokerErrorMessage{Epoch: 7, Connection: connection, Error: oversize}, testEnvelopeCeiling, testChunkCeiling)
910 require.ErrorIs(t, err, ErrTooLarge)
911 })
912 t.Run("snapshot counts", func(t *testing.T) {
913 _, err := EncodeServer(SnapshotPart{Epoch: 7, Connection: connection, Generation: 1, Revision: 1, Index: 0, Part: SnapshotBegin{HostCount: ports.BrokerMaxDaemonsPerSnapshot + 1}}, testEnvelopeCeiling, testChunkCeiling)
914 require.ErrorIs(t, err, ErrTooLarge)
915 })
916 t.Run("control env carries no attachment state", func(t *testing.T) {
917 message := OpenStream{Epoch: 7, Connection: connection, Stream: 3, Purpose: ports.BrokerStreamControl, Local: true, Policy: testPolicy(), StartMode: ports.BrokerDaemonStartIfNeeded}
918 raw := mustEncodeClient(t, message)
919 envelope := &wire.BrokerClientEnvelope{}
920 require.NoError(t, proto.Unmarshal(raw, envelope))
921 envelope.GetOpenStream().Target = exactTargetToWire(testTarget())
922 raw, err := proto.Marshal(envelope)
923 require.NoError(t, err)
924 _, err = DecodeClient(raw, testEnvelopeCeiling, testChunkCeiling)
925 require.ErrorIs(t, err, ErrInvalidMessage)
926 })
927}
928
929// TestBrokerValidationNegatives is the table of semantic refusals: IDs,
930// UTF-8/control/bidi text, taxonomy, times, env/policy, snapshot parts.
931func TestBrokerValidationNegatives(t *testing.T) {
932 connection := testConnectionID(0x11)
933 t.Run("zero scope", func(t *testing.T) {
934 _, err := EncodeClient(Subscribe{Epoch: 0, Connection: connection}, testEnvelopeCeiling, testChunkCeiling)
935 require.ErrorIs(t, err, ErrInvalidMessage)
936 _, err = EncodeClient(Subscribe{Epoch: 7}, testEnvelopeCeiling, testChunkCeiling)
937 require.ErrorIs(t, err, ErrInvalidMessage)
938 _, err = EncodeServer(Registered{Epoch: 7}, testEnvelopeCeiling, testChunkCeiling)
939 require.ErrorIs(t, err, ErrInvalidMessage)
940 })
941 t.Run("zero stream", func(t *testing.T) {
942 _, err := EncodeClient(CloseStream{Epoch: 7, Connection: connection}, testEnvelopeCeiling, testChunkCeiling)
943 require.ErrorIs(t, err, ErrInvalidMessage)
944 })
945 t.Run("stream window update bounds", func(t *testing.T) {
946 valid := StreamWindowUpdate{Epoch: 7, Connection: connection, Stream: 3, Credit: 1}
947 cases := map[string]func(*StreamWindowUpdate){
948 "zero credit": func(m *StreamWindowUpdate) { m.Credit = 0 },
949 "credit above window": func(m *StreamWindowUpdate) { m.Credit = StreamWindowBytes + 1 },
950 "zero epoch": func(m *StreamWindowUpdate) { m.Epoch = 0 },
951 "zero stream": func(m *StreamWindowUpdate) { m.Stream = 0 },
952 }
953 for name, mutate := range cases {
954 t.Run(name, func(t *testing.T) {
955 bad := valid
956 mutate(&bad)
957 _, err := EncodeClient(bad, testEnvelopeCeiling, testChunkCeiling)
958 require.ErrorIs(t, err, ErrInvalidMessage)
959 _, err = EncodeServer(bad, testEnvelopeCeiling, testChunkCeiling)
960 require.ErrorIs(t, err, ErrInvalidMessage)
961 })
962 }
963 // The decoder refuses the same out-of-range credit a peer could put
964 // on the wire directly.
965 for name, credit := range map[string]uint64{"zero credit": 0, "credit above window": StreamWindowBytes + 1} {
966 t.Run("decode "+name, func(t *testing.T) {
967 message := &wire.StreamWindowUpdate{Ref: refToWire(7, connection, 3), Credit: credit}
968 client, err := proto.Marshal(&wire.BrokerClientEnvelope{Payload: &wire.BrokerClientEnvelope_StreamWindowUpdate{StreamWindowUpdate: message}})
969 require.NoError(t, err)
970 _, err = DecodeClient(client, testEnvelopeCeiling, testChunkCeiling)
971 require.ErrorIs(t, err, ErrInvalidMessage)
972 server, err := proto.Marshal(&wire.BrokerServerEnvelope{Payload: &wire.BrokerServerEnvelope_StreamWindowUpdate{StreamWindowUpdate: message}})
973 require.NoError(t, err)
974 _, err = DecodeServer(server, testEnvelopeCeiling, testChunkCeiling)
975 require.ErrorIs(t, err, ErrInvalidMessage)
976 })
977 }
978 })
979 t.Run("error detail never travels without HasError", func(t *testing.T) {
980 detail := testErrorDetail()
981 // HasError=false is exactly the absent detail, whatever the outcome
982 // and whatever field of the detail is nonzero.
983 for name, stray := range map[string]ErrorDetail{
984 "full detail": detail,
985 "code only": {Code: ports.BrokerErrorTimeout},
986 "text only": {Text: "late"},
987 "admission only": {AdmissionCode: 2},
988 "failure kind only": {FailureKind: domain.RemoteFailureTransport},
989 } {
990 t.Run(name, func(t *testing.T) {
991 for _, outcome := range []ports.BrokerMutationOutcome{ports.BrokerOutcomeOK, ports.BrokerOutcomeFailed, ports.BrokerOutcomeUnknown} {
992 _, err := EncodeServer(OperationResult{Epoch: 7, Connection: connection, Operation: testOperationID(1), Outcome: outcome, Error: stray}, testEnvelopeCeiling, testChunkCeiling)
993 require.ErrorIs(t, err, ErrInvalidMessage)
994 }
995 _, err := EncodeServer(StreamClosed{Epoch: 7, Connection: connection, Stream: 3, Error: stray}, testEnvelopeCeiling, testChunkCeiling)
996 require.ErrorIs(t, err, ErrInvalidMessage)
997 })
998 }
999 // The zero detail is the canonical absent value and still encodes; a
1000 // declared detail still round-trips exactly.
1001 _, err := EncodeServer(StreamClosed{Epoch: 7, Connection: connection, Stream: 3}, testEnvelopeCeiling, testChunkCeiling)
1002 require.NoError(t, err)
1003 _, err = EncodeServer(OperationResult{Epoch: 7, Connection: connection, Operation: testOperationID(1), Outcome: ports.BrokerOutcomeOK, Registration: testRegistration()}, testEnvelopeCeiling, testChunkCeiling)
1004 require.NoError(t, err)
1005 declared := OperationResult{Epoch: 7, Connection: connection, Operation: testOperationID(1), Outcome: ports.BrokerOutcomeFailed, Error: detail, HasError: true}
1006 decoded, err := DecodeServer(mustEncodeServer(t, declared), testEnvelopeCeiling, testChunkCeiling)
1007 require.NoError(t, err)
1008 require.Equal(t, declared, decoded)
1009 })
1010 t.Run("invalid error detail refused on every encode path", func(t *testing.T) {
1011 badAdmission := testErrorDetail()
1012 badAdmission.AdmissionCode = MaxBrokerAdmissionCode + 1
1013 badKind := testErrorDetail()
1014 badKind.FailureKind = MaxBrokerFailureKind + 1
1015 badCode := testErrorDetail()
1016 badCode.Code = 0
1017 for name, detail := range map[string]ErrorDetail{
1018 "admission over maximum": badAdmission,
1019 "failure kind over bound": badKind,
1020 "invalid error code": badCode,
1021 } {
1022 t.Run(name, func(t *testing.T) {
1023 _, err := EncodeServer(BrokerErrorMessage{Epoch: 7, Connection: connection, Error: detail}, testEnvelopeCeiling, testChunkCeiling)
1024 require.ErrorIs(t, err, ErrInvalidMessage)
1025 _, err = EncodeServer(OperationResult{Epoch: 7, Connection: connection, Operation: testOperationID(1), Outcome: ports.BrokerOutcomeFailed, Error: detail, HasError: true}, testEnvelopeCeiling, testChunkCeiling)
1026 require.ErrorIs(t, err, ErrInvalidMessage)
1027 _, err = EncodeServer(StreamClosed{Epoch: 7, Connection: connection, Stream: 3, Error: detail, HasError: true}, testEnvelopeCeiling, testChunkCeiling)
1028 require.ErrorIs(t, err, ErrInvalidMessage)
1029 })
1030 }
1031 // The closed taxonomy boundaries are accepted.
1032 boundary := testErrorDetail()
1033 boundary.AdmissionCode = MaxBrokerAdmissionCode
1034 boundary.FailureKind = MaxBrokerFailureKind
1035 _, err := EncodeServer(BrokerErrorMessage{Epoch: 7, Connection: connection, Error: boundary}, testEnvelopeCeiling, testChunkCeiling)
1036 require.NoError(t, err)
1037 })
1038 t.Run("membership request authority", func(t *testing.T) {
1039 base := AddHost{Epoch: 7, Connection: connection, Operation: testOperationID(1), Endpoint: "dev@host:22", Policy: testPolicy()}
1040 // An absent policy is not authority: it is refused exactly like an
1041 // invalid one rather than inherited from a default.
1042 missing := base
1043 missing.Policy = ports.BrokerPolicy{}
1044 _, err := EncodeClient(missing, testEnvelopeCeiling, testChunkCeiling)
1045 require.ErrorIs(t, err, ErrInvalidMessage)
1046
1047 // A partial registration is not exact authority either.
1048 _, err = EncodeClient(RemoveHost{Epoch: 7, Connection: connection, Operation: testOperationID(1)}, testEnvelopeCeiling, testChunkCeiling)
1049 require.ErrorIs(t, err, ErrInvalidMessage)
1050 partial := testRegistration()
1051 partial.Incarnation = [16]byte{}
1052 _, err = EncodeClient(RemoveHost{Epoch: 7, Connection: connection, Operation: testOperationID(1), Registration: partial}, testEnvelopeCeiling, testChunkCeiling)
1053 require.ErrorIs(t, err, ErrInvalidMessage)
1054 partial = testRegistration()
1055 partial.Generation = 0
1056 _, err = EncodeClient(RemoveHost{Epoch: 7, Connection: connection, Operation: testOperationID(1), Registration: partial}, testEnvelopeCeiling, testChunkCeiling)
1057 require.ErrorIs(t, err, ErrInvalidMessage)
1058
1059 update := UpdateHostPolicy{Epoch: 7, Connection: connection, Operation: testOperationID(1), Registration: testRegistration(), Policy: testPolicy()}
1060 missingPolicy := update
1061 missingPolicy.Policy = ports.BrokerPolicy{}
1062 _, err = EncodeClient(missingPolicy, testEnvelopeCeiling, testChunkCeiling)
1063 require.ErrorIs(t, err, ErrInvalidMessage)
1064 missingRegistration := update
1065 missingRegistration.Registration = domain.RemoteRegistration{}
1066 _, err = EncodeClient(missingRegistration, testEnvelopeCeiling, testChunkCeiling)
1067 require.ErrorIs(t, err, ErrInvalidMessage)
1068 })
1069 t.Run("membership result presence", func(t *testing.T) {
1070 // A result may never carry both removal and registration, and a
1071 // failed or outcome-unknown result may carry neither authority and
1072 // must carry its typed error.
1073 both := OperationResult{Epoch: 7, Connection: connection, Operation: testOperationID(1), Outcome: ports.BrokerOutcomeOK, Removed: true, Registration: testRegistration()}
1074 _, err := EncodeServer(both, testEnvelopeCeiling, testChunkCeiling)
1075 require.ErrorIs(t, err, ErrInvalidMessage)
1076 for _, outcome := range []ports.BrokerMutationOutcome{ports.BrokerOutcomeFailed, ports.BrokerOutcomeUnknown} {
1077 _, err = EncodeServer(OperationResult{Epoch: 7, Connection: connection, Operation: testOperationID(1), Outcome: outcome, Registration: testRegistration(), Error: testErrorDetail(), HasError: true}, testEnvelopeCeiling, testChunkCeiling)
1078 require.ErrorIs(t, err, ErrInvalidMessage)
1079 _, err = EncodeServer(OperationResult{Epoch: 7, Connection: connection, Operation: testOperationID(1), Outcome: outcome, Removed: true, Error: testErrorDetail(), HasError: true}, testEnvelopeCeiling, testChunkCeiling)
1080 require.ErrorIs(t, err, ErrInvalidMessage)
1081 // Neither authority without an error is refused too: a failure
1082 // with no detail would read as an unexplained refusal.
1083 _, err = EncodeServer(OperationResult{Epoch: 7, Connection: connection, Operation: testOperationID(1), Outcome: outcome}, testEnvelopeCeiling, testChunkCeiling)
1084 require.ErrorIs(t, err, ErrInvalidMessage)
1085 }
1086 // A success carries no error at all.
1087 _, err = EncodeServer(OperationResult{Epoch: 7, Connection: connection, Operation: testOperationID(1), Outcome: ports.BrokerOutcomeOK, Registration: testRegistration(), Error: testErrorDetail(), HasError: true}, testEnvelopeCeiling, testChunkCeiling)
1088 require.ErrorIs(t, err, ErrInvalidMessage)
1089 // A partial registration is refused on the result path too.
1090 partial := testRegistration()
1091 partial.Incarnation = [16]byte{}
1092 _, err = EncodeServer(OperationResult{Epoch: 7, Connection: connection, Operation: testOperationID(1), Outcome: ports.BrokerOutcomeOK, Registration: partial}, testEnvelopeCeiling, testChunkCeiling)
1093 require.ErrorIs(t, err, ErrInvalidMessage)
1094 })
1095 t.Run("membership result authority by mutation kind", func(t *testing.T) {
1096 registration := testRegistration()
1097 add := OperationResult{Epoch: 7, Connection: connection, Operation: testOperationID(1), Outcome: ports.BrokerOutcomeOK, Registration: registration}
1098 require.NoError(t, add.ValidateForMutation(MutationKindAddHost))
1099 require.ErrorIs(t, add.ValidateForMutation(MutationKindRemoveHost), ErrInvalidMessage)
1100 require.NoError(t, add.ValidateForMutation(MutationKindUpdateHostPolicy))
1101
1102 remove := OperationResult{Epoch: 7, Connection: connection, Operation: testOperationID(1), Outcome: ports.BrokerOutcomeOK, Removed: true}
1103 require.NoError(t, remove.ValidateForMutation(MutationKindRemoveHost))
1104 require.ErrorIs(t, remove.ValidateForMutation(MutationKindAddHost), ErrInvalidMessage)
1105 require.ErrorIs(t, remove.ValidateForMutation(MutationKindUpdateHostPolicy), ErrInvalidMessage)
1106
1107 failed := OperationResult{Epoch: 7, Connection: connection, Operation: testOperationID(1), Outcome: ports.BrokerOutcomeFailed, Error: testErrorDetail(), HasError: true}
1108 for _, kind := range []RegisterMutationKind{MutationKindAddHost, MutationKindRemoveHost, MutationKindUpdateHostPolicy} {
1109 require.NoError(t, failed.ValidateForMutation(kind))
1110 }
1111 // A kind outside the closed taxonomy is refused, never guessed.
1112 require.Error(t, add.ValidateForMutation(RegisterMutationKind(9)))
1113 // The request-side classifier names exactly the three mutations.
1114 for message, want := range map[ClientMessage]RegisterMutationKind{
1115 AddHost{}: MutationKindAddHost,
1116 RemoveHost{}: MutationKindRemoveHost,
1117 UpdateHostPolicy{}: MutationKindUpdateHostPolicy,
1118 Subscribe{}: 0,
1119 } {
1120 got, ok := MutationKindOf(message)
1121 if want == 0 {
1122 require.False(t, ok)
1123 continue
1124 }
1125 require.True(t, ok)
1126 require.Equal(t, want, got)
1127 }
1128 require.Equal(t, "add_host", MutationKindAddHost.String())
1129 require.Equal(t, "remove_host", MutationKindRemoveHost.String())
1130 require.Equal(t, "update_host_policy", MutationKindUpdateHostPolicy.String())
1131 require.Equal(t, "unknown", RegisterMutationKind(9).String())
1132 })
1133 t.Run("bad endpoint", func(t *testing.T) {
1134 _, err := EncodeClient(AddHost{Epoch: 7, Connection: connection, Operation: testOperationID(1), Endpoint: "not a host", Policy: testPolicy()}, testEnvelopeCeiling, testChunkCeiling)
1135 require.ErrorIs(t, err, ErrInvalidMessage)
1136 })
1137 t.Run("taxonomy switches", func(t *testing.T) {
1138 raw := mustEncodeClient(t, OpenStream{Epoch: 7, Connection: connection, Stream: 3, Purpose: ports.BrokerStreamAttachment, Admission: ports.BrokerAdmissionExact, Endpoint: "dev@host:22", Registration: testRegistration(), Target: testTarget(), Policy: testPolicy(), StartMode: ports.BrokerDaemonStartIfNeeded})
1139 envelope := &wire.BrokerClientEnvelope{}
1140 require.NoError(t, wire.ScanEnvelope(envelope, raw))
1141 require.NoError(t, proto.Unmarshal(raw, envelope))
1142 envelope.GetOpenStream().Purpose = 9
1143 mutated, err := proto.Marshal(envelope)
1144 require.NoError(t, err)
1145 _, err = DecodeClient(mutated, testEnvelopeCeiling, testChunkCeiling)
1146 require.Error(t, err)
1147
1148 rawErr := mustEncodeServer(t, BrokerErrorMessage{Epoch: 7, Connection: connection, Error: testErrorDetail()})
1149 server := &wire.BrokerServerEnvelope{}
1150 require.NoError(t, wire.ScanEnvelope(server, rawErr))
1151 require.NoError(t, proto.Unmarshal(rawErr, server))
1152 server.GetBrokerErrorMessage().GetError().Code = 99
1153 mutated, err = proto.Marshal(server)
1154 require.NoError(t, err)
1155 _, err = DecodeServer(mutated, testEnvelopeCeiling, testChunkCeiling)
1156 require.Error(t, err)
1157 })
1158 t.Run("bidi and control text refused", func(t *testing.T) {
1159 evil := testErrorDetail()
1160 evil.Text = "timeout\u202erorrE"
1161 _, err := EncodeServer(BrokerErrorMessage{Epoch: 7, Connection: connection, Error: evil}, testEnvelopeCeiling, testChunkCeiling)
1162 require.ErrorIs(t, err, ErrInvalidMessage)
1163 evil.Text = "bad\x01text"
1164 _, err = EncodeServer(BrokerErrorMessage{Epoch: 7, Connection: connection, Error: evil}, testEnvelopeCeiling, testChunkCeiling)
1165 require.ErrorIs(t, err, ErrInvalidMessage)
1166 _, err = EncodeServer(Shutdown{Epoch: 7, Connection: connection, Reason: ShutdownIdleExit, Text: "x\x1b"}, testEnvelopeCeiling, testChunkCeiling)
1167 require.ErrorIs(t, err, ErrInvalidMessage)
1168 })
1169 t.Run("env without equals refused", func(t *testing.T) {
1170 _, err := EncodeClient(OpenStream{Epoch: 7, Connection: connection, Stream: 3, Purpose: ports.BrokerStreamAttachment, Admission: ports.BrokerAdmissionExact, Endpoint: "dev@host:22", Registration: testRegistration(), Target: testTarget(), Env: []string{"NOEQUALS"}, Policy: testPolicy(), StartMode: ports.BrokerDaemonStartIfNeeded}, testEnvelopeCeiling, testChunkCeiling)
1171 require.ErrorIs(t, err, ErrInvalidMessage)
1172 })
1173 t.Run("invalid policy refused", func(t *testing.T) {
1174 raw, err := EncodeClient(OpenStream{Epoch: 7, Connection: connection, Stream: 3, Purpose: ports.BrokerStreamAttachment, Admission: ports.BrokerAdmissionExact, Endpoint: "dev@host:22", Registration: testRegistration(), Target: testTarget(), Policy: ports.BrokerPolicy{}, StartMode: ports.BrokerDaemonStartIfNeeded}, testEnvelopeCeiling, testChunkCeiling)
1175 require.NoError(t, err)
1176 _, err = DecodeClient(raw, testEnvelopeCeiling, testChunkCeiling)
1177 require.ErrorIs(t, err, ErrInvalidMessage)
1178 })
1179 t.Run("snapshot revision zero refused", func(t *testing.T) {
1180 _, err := EncodeServer(SnapshotPart{Epoch: 7, Connection: connection, Generation: 1, Index: 0, Part: SnapshotEnd{}}, testEnvelopeCeiling, testChunkCeiling)
1181 require.ErrorIs(t, err, ErrInvalidMessage)
1182 })
1183 t.Run("missing snapshot part refused", func(t *testing.T) {
1184 _, err := EncodeServer(SnapshotPart{Epoch: 7, Connection: connection, Generation: 1, Revision: 1}, testEnvelopeCeiling, testChunkCeiling)
1185 require.ErrorIs(t, err, ErrInvalidMessage)
1186 })
1187 t.Run("local open with registration refused", func(t *testing.T) {
1188 message := OpenStream{Epoch: 7, Connection: connection, Stream: 3, Purpose: ports.BrokerStreamAttachment, Admission: ports.BrokerAdmissionExact, Local: true, Target: testTarget(), Policy: testPolicy(), StartMode: ports.BrokerDaemonStartIfNeeded}
1189 raw := mustEncodeClient(t, message)
1190 envelope := &wire.BrokerClientEnvelope{}
1191 require.NoError(t, proto.Unmarshal(raw, envelope))
1192 envelope.GetOpenStream().Registration = registrationToWire(testRegistration())
1193 raw, err := proto.Marshal(envelope)
1194 require.NoError(t, err)
1195 _, err = DecodeClient(raw, testEnvelopeCeiling, testChunkCeiling)
1196 require.ErrorIs(t, err, ErrInvalidMessage)
1197 })
1198 t.Run("bad shutdown reason refused", func(t *testing.T) {
1199 _, err := EncodeServer(Shutdown{Epoch: 7, Connection: connection, Reason: ShutdownReason(9)}, testEnvelopeCeiling, testChunkCeiling)
1200 require.ErrorIs(t, err, ErrInvalidMessage)
1201 })
1202 t.Run("bad progress phase refused", func(t *testing.T) {
1203 _, err := EncodeServer(Progress{Epoch: 7, Connection: connection, Stream: 3, Phase: BrokerProgressPhase(9)}, testEnvelopeCeiling, testChunkCeiling)
1204 require.ErrorIs(t, err, ErrInvalidMessage)
1205 })
1206 t.Run("nil messages refused", func(t *testing.T) {
1207 var clientNil *Register
1208 _, err := EncodeClient(clientNil, testEnvelopeCeiling, testChunkCeiling)
1209 require.ErrorIs(t, err, ErrInvalidMessage)
1210 var serverNil *Registered
1211 _, err = EncodeServer(serverNil, testEnvelopeCeiling, testChunkCeiling)
1212 require.ErrorIs(t, err, ErrInvalidMessage)
1213 })
1214}
1215
1216// TestBrokerTruncatedAndTrailing proves truncated prefixes and trailing
1217// garbage are rejected on both directions.
1218func TestBrokerTruncatedAndTrailing(t *testing.T) {
1219 raw := mustEncodeClient(t, Subscribe{Epoch: 7, Connection: testConnectionID(1), Generation: 3})
1220 for size := range len(raw) {
1221 _, err := DecodeClient(raw[:size], testEnvelopeCeiling, testChunkCeiling)
1222 require.Error(t, err, "client prefix[:%d] accepted", size)
1223 }
1224 require.Error(t, func() error {
1225 _, err := DecodeClient(append(append([]byte(nil), raw...), 0xFF), testEnvelopeCeiling, testChunkCeiling)
1226 return err
1227 }())
1228
1229 serverRaw := mustEncodeServer(t, Registered{Epoch: 7, Connection: testConnectionID(1)})
1230 for size := range len(serverRaw) {
1231 _, err := DecodeServer(serverRaw[:size], testEnvelopeCeiling, testChunkCeiling)
1232 require.Error(t, err, "server prefix[:%d] accepted", size)
1233 }
1234 concatenated := append(append([]byte(nil), serverRaw...), serverRaw...)
1235 _, err := DecodeServer(concatenated, testEnvelopeCeiling, testChunkCeiling)
1236 require.Error(t, err)
1237}
1238
1239// TestBrokerStreamWindowUpdateByteForByte pins the window update layout in
1240// both directions, then rejects every truncated prefix and trailing garbage.
1241func TestBrokerStreamWindowUpdateByteForByte(t *testing.T) {
1242 connection := testConnectionID(0x11)
1243 message := StreamWindowUpdate{Epoch: 7, Connection: connection, Stream: 3, Credit: 4096}
1244 scope := append([]byte{0x08, 0x07, 0x12, 0x10}, connection[:]...)
1245 ref := append(append([]byte{0x0a, byte(len(scope))}, scope...), 0x10, 0x03)
1246 body := append(append([]byte{0x0a, byte(len(ref))}, ref...), 0x10, 0x80, 0x20)
1247 for _, tc := range []struct {
1248 name string
1249 key []byte // field 114 (client) or 211 (server), wire type 2
1250 encode func(*testing.T, any) []byte
1251 decode func([]byte) (any, error)
1252 }{
1253 {"client", []byte{0x92, 0x07}, func(t *testing.T, m any) []byte { return mustEncodeClient(t, m.(ClientMessage)) }, func(raw []byte) (any, error) {
1254 return DecodeClient(raw, testEnvelopeCeiling, testChunkCeiling)
1255 }},
1256 {"server", []byte{0x9a, 0x0d}, func(t *testing.T, m any) []byte { return mustEncodeServer(t, m.(ServerMessage)) }, func(raw []byte) (any, error) {
1257 return DecodeServer(raw, testEnvelopeCeiling, testChunkCeiling)
1258 }},
1259 } {
1260 t.Run(tc.name, func(t *testing.T) {
1261 want := append(append(append([]byte(nil), tc.key...), byte(len(body))), body...)
1262 raw := tc.encode(t, message)
1263 require.Equal(t, want, raw)
1264 decoded, err := tc.decode(raw)
1265 require.NoError(t, err)
1266 require.Equal(t, message, decoded)
1267 for size := range len(raw) {
1268 _, err := tc.decode(raw[:size])
1269 require.Error(t, err, "prefix[:%d] accepted", size)
1270 }
1271 _, err = tc.decode(append(append([]byte(nil), raw...), 0xFF))
1272 require.Error(t, err, "trailing garbage accepted")
1273 })
1274 }
1275}
1276
1277// TestBrokerWrongDirection proves wrong-direction decode is rejected:
1278// client bytes never decode as server output and vice versa.
1279func TestBrokerWrongDirection(t *testing.T) {
1280 clientRaw := mustEncodeClient(t, Register{})
1281 _, err := DecodeServer(clientRaw, testEnvelopeCeiling, testChunkCeiling)
1282 require.Error(t, err)
1283
1284 serverRaw := mustEncodeServer(t, Registered{Epoch: 7, Connection: testConnectionID(1)})
1285 _, err = DecodeClient(serverRaw, testEnvelopeCeiling, testChunkCeiling)
1286 require.Error(t, err)
1287
1288 // Go-typed misuse fails with ErrWrongDirection at encode time: unknown
1289 // concrete types presented to either directional codec fail closed.
1290 _, err = encodeClientEnvelope(unknownClientMessage{}, testChunkCeiling)
1291 require.ErrorIs(t, err, ErrWrongDirection)
1292 _, err = encodeServerEnvelope(unknownServerMessage{}, testChunkCeiling)
1293 require.ErrorIs(t, err, ErrWrongDirection)
1294
1295 // Nil-payload envelopes decode as wrong-direction.
1296 _, err = decodeClientEnvelope(&wire.BrokerClientEnvelope{}, testChunkCeiling)
1297 require.ErrorIs(t, err, ErrWrongDirection)
1298 _, err = decodeServerEnvelope(&wire.BrokerServerEnvelope{}, testChunkCeiling)
1299 require.ErrorIs(t, err, ErrWrongDirection)
1300}
1301
1302// TestBrokerCorruptInputSmoke feeds representative corruptions through both
1303// decoders as a smoke test: bit flips, truncation, unknown fields, and
1304// direction swaps must never panic and must never be accepted silently.
1305func TestBrokerCorruptInputSmoke(t *testing.T) {
1306 seeds := [][]byte{
1307 mustEncodeClient(t, Register{}),
1308 mustEncodeClient(t, AddHost{Epoch: 7, Connection: testConnectionID(2), Operation: testOperationID(0x21), Endpoint: "dev@host:22", Policy: testPolicy()}),
1309 mustEncodeClient(t, RemoveHost{Epoch: 7, Connection: testConnectionID(2), Operation: testOperationID(0x22), Registration: testRegistration()}),
1310 mustEncodeClient(t, UpdateHostPolicy{Epoch: 7, Connection: testConnectionID(2), Operation: testOperationID(0x23), Registration: testRegistration(), Policy: testPolicy()}),
1311 mustEncodeClient(t, OpenStream{Epoch: 7, Connection: testConnectionID(2), Stream: 3, Purpose: ports.BrokerStreamControl, Local: true, Policy: testPolicy(), StartMode: ports.BrokerDaemonStartIfNeeded}),
1312 mustEncodeServer(t, SnapshotPart{Epoch: 7, Connection: testConnectionID(2), Generation: 1, Revision: 1, Index: 0, Part: SnapshotEnd{}}),
1313 mustEncodeServer(t, OperationResult{Epoch: 7, Connection: testConnectionID(2), Operation: testOperationID(0x24), Outcome: ports.BrokerOutcomeFailed, Error: testErrorDetail(), HasError: true}),
1314 mustEncodeServer(t, Shutdown{Epoch: 7, Connection: testConnectionID(2), Reason: ShutdownIdleExit}),
1315 }
1316 for _, seed := range seeds {
1317 mutated := append([]byte(nil), seed...)
1318 if len(mutated) > 2 {
1319 mutated[len(mutated)-1] ^= 0xFF
1320 mutated[len(mutated)-2] ^= 0x01
1321 }
1322 _, _ = DecodeClient(mutated, testEnvelopeCeiling, testChunkCeiling)
1323 _, _ = DecodeServer(mutated, testEnvelopeCeiling, testChunkCeiling)
1324 unknown := protowire.AppendTag(append([]byte(nil), seed...), 199, protowire.VarintType)
1325 unknown = protowire.AppendVarint(unknown, 1)
1326 _, _ = DecodeClient(unknown, testEnvelopeCeiling, testChunkCeiling)
1327 _, _ = DecodeServer(unknown, testEnvelopeCeiling, testChunkCeiling)
1328 }
1329 require.NotEmpty(t, seeds)
1330}
1331
1332// fuzzClientSeeds and fuzzServerSeeds are the shared strict-scan seed corpus:
1333// valid brokered messages plus vectors that exercise ScanEnvelope rejection
1334// (unknown tag, wrong wire type, truncation, empty, and trailing bytes).
1335func fuzzClientSeeds(t testing.TB) [][]byte {
1336 t.Helper()
1337 seeds := [][]byte{}
1338 for _, message := range []ClientMessage{
1339 Register{},
1340 Subscribe{Epoch: 7, Connection: testConnectionID(0x11), Generation: 1},
1341 AddHost{Epoch: 7, Connection: testConnectionID(0x11), Operation: testOperationID(0x11), Endpoint: "dev@host:22", Policy: testPolicy()},
1342 RemoveHost{Epoch: 7, Connection: testConnectionID(0x11), Operation: testOperationID(0x12), Registration: testRegistration()},
1343 UpdateHostPolicy{Epoch: 7, Connection: testConnectionID(0x11), Operation: testOperationID(0x13), Registration: testRegistration(), Policy: testPolicy()},
1344 OpenStream{Epoch: 7, Connection: testConnectionID(0x11), Stream: 3, Purpose: ports.BrokerStreamAttachment, Admission: ports.BrokerAdmissionExact, Endpoint: "dev@host:22", Registration: testRegistration(), Target: testTarget(), Policy: testPolicy(), StartMode: ports.BrokerDaemonStartIfNeeded},
1345 ClientStreamData{Epoch: 7, Connection: testConnectionID(0x11), Stream: 3, Data: []byte("frame")},
1346 CloseStream{Epoch: 7, Connection: testConnectionID(0x11), Stream: 3},
1347 StreamWindowUpdate{Epoch: 7, Connection: testConnectionID(0x11), Stream: 3, Credit: 4096},
1348 } {
1349 raw, err := EncodeClient(message, testEnvelopeCeiling, testChunkCeiling)
1350 require.NoError(t, err)
1351 seeds = append(seeds, raw)
1352 }
1353 return appendStrictScanSeeds(seeds)
1354}
1355
1356func fuzzServerSeeds(t testing.TB) [][]byte {
1357 t.Helper()
1358 seeds := [][]byte{}
1359 for _, message := range []ServerMessage{
1360 Registered{Epoch: 7, Connection: testConnectionID(0x11)},
1361 SnapshotPart{Epoch: 7, Connection: testConnectionID(0x11), Generation: 2, Revision: 5, Index: 0, Part: SnapshotEnd{}},
1362 SnapshotPart{Epoch: 7, Connection: testConnectionID(0x11), Generation: 2, Revision: 5, Index: 2, Part: SnapshotSessionPart{HostIndex: 0, SessionIndex: 0, Session: testCatalogSession()}},
1363 OperationResult{Epoch: 7, Connection: testConnectionID(0x11), Operation: testOperationID(0x14), Outcome: ports.BrokerOutcomeOK, Registration: testRegistration()},
1364 BrokerErrorMessage{Epoch: 7, Connection: testConnectionID(0x11), Error: testErrorDetail()},
1365 Shutdown{Epoch: 7, Connection: testConnectionID(0x11), Reason: ShutdownTerminating},
1366 StreamWindowUpdate{Epoch: 7, Connection: testConnectionID(0x11), Stream: 3, Credit: 4096},
1367 } {
1368 raw, err := EncodeServer(message, testEnvelopeCeiling, testChunkCeiling)
1369 require.NoError(t, err)
1370 seeds = append(seeds, raw)
1371 }
1372 return appendStrictScanSeeds(seeds)
1373}
1374
1375// appendStrictScanSeeds adds raw vectors that must be refused by the strict
1376// scanner rather than reaching generated unmarshal.
1377func appendStrictScanSeeds(seeds [][]byte) [][]byte {
1378 unknown := protowire.AppendTag(nil, 199, protowire.VarintType)
1379 unknown = protowire.AppendVarint(unknown, 1)
1380 wrongType := protowire.AppendTag(nil, 101, protowire.VarintType)
1381 wrongType = protowire.AppendVarint(wrongType, 1)
1382 return append(seeds, unknown, wrongType, []byte{0xFF}, []byte{0x80}, nil)
1383}
1384
1385// FuzzDecodeClient proves DecodeClient never panics on arbitrary input.
1386func FuzzDecodeClient(f *testing.F) {
1387 for _, seed := range fuzzClientSeeds(f) {
1388 f.Add(seed)
1389 }
1390 f.Fuzz(func(t *testing.T, payload []byte) {
1391 _, _ = DecodeClient(payload, testEnvelopeCeiling, testChunkCeiling)
1392 })
1393}
1394
1395// FuzzDecodeServer proves DecodeServer never panics on arbitrary input.
1396func FuzzDecodeServer(f *testing.F) {
1397 for _, seed := range fuzzServerSeeds(f) {
1398 f.Add(seed)
1399 }
1400 f.Fuzz(func(t *testing.T, payload []byte) {
1401 _, _ = DecodeServer(payload, testEnvelopeCeiling, testChunkCeiling)
1402 })
1403}
1404
1405func TestBuildRegistrationCodec(t *testing.T) {
1406 for _, tc := range []struct {
1407 name string
1408 build string
1409 retiring bool
1410 }{
1411 {"empty", "", false}, {"matching", "release+abc", false}, {"retiring", "release+def", true},
1412 } {
1413 t.Run(tc.name, func(t *testing.T) {
1414 request := Register{Build: tc.build}
1415 raw := mustEncodeClient(t, request)
1416 decoded, err := DecodeClient(raw, testEnvelopeCeiling, testChunkCeiling)
1417 require.NoError(t, err)
1418 require.Equal(t, request, decoded)
1419 want := []byte{0xaa, 0x06, 0x00}
1420 if tc.build != "" {
1421 want = append([]byte{0xaa, 0x06, byte(len(tc.build) + 2), 0x0a, byte(len(tc.build))}, []byte(tc.build)...)
1422 }
1423 require.Equal(t, want, raw)
1424 response := Registered{Epoch: 7, Connection: testConnectionID(1), Build: tc.build, Retiring: tc.retiring}
1425 raw = mustEncodeServer(t, response)
1426 decodedServer, err := DecodeServer(raw, testEnvelopeCeiling, testChunkCeiling)
1427 require.NoError(t, err)
1428 require.Equal(t, response, decodedServer)
1429 envelope := &wire.BrokerServerEnvelope{}
1430 require.NoError(t, proto.Unmarshal(raw, envelope))
1431 require.Equal(t, tc.build, envelope.GetRegistered().GetBuild())
1432 require.Equal(t, tc.retiring, envelope.GetRegistered().GetRetiring())
1433 })
1434 }
1435 for _, tc := range []struct {
1436 name string
1437 server bool
1438 }{{"client", false}, {"server", true}} {
1439 t.Run("oversize-"+tc.name, func(t *testing.T) {
1440 build := strings.Repeat("x", 129)
1441 if tc.server {
1442 _, err := EncodeServer(Registered{Epoch: 7, Connection: testConnectionID(1), Build: build}, testEnvelopeCeiling, testChunkCeiling)
1443 require.ErrorIs(t, err, ErrInvalidMessage)
1444 raw, err := proto.Marshal(&wire.BrokerServerEnvelope{Payload: &wire.BrokerServerEnvelope_Registered{Registered: &wire.Registered{Scope: scopeToWire(7, testConnectionID(1)), Build: build}}})
1445 require.NoError(t, err)
1446 _, err = DecodeServer(raw, testEnvelopeCeiling, testChunkCeiling)
1447 require.ErrorIs(t, err, ErrInvalidMessage)
1448 } else {
1449 _, err := EncodeClient(Register{Build: build}, testEnvelopeCeiling, testChunkCeiling)
1450 require.ErrorIs(t, err, ErrInvalidMessage)
1451 raw, err := proto.Marshal(&wire.BrokerClientEnvelope{Payload: &wire.BrokerClientEnvelope_Register{Register: &wire.Register{Build: build}}})
1452 require.NoError(t, err)
1453 _, err = DecodeClient(raw, testEnvelopeCeiling, testChunkCeiling)
1454 require.ErrorIs(t, err, ErrInvalidMessage)
1455 }
1456 })
1457 }
1458}