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 }
364 require.Len(t, messages, 13)
365 for _, tc := range messages {
366 t.Run(tc.name, func(t *testing.T) {
367 raw := mustEncodeClient(t, tc.message)
368 number, wireType, consumed := protowire.ConsumeTag(raw)
369 require.Greater(t, consumed, 0)
370 require.Equal(t, tc.tag, number)
371 require.Equal(t, protowire.BytesType, wireType)
372 decoded, err := DecodeClient(raw, testEnvelopeCeiling, testChunkCeiling)
373 require.NoError(t, err)
374 require.Equal(t, tc.message, decoded)
375 })
376 }
377}
378
379// TestBrokerOpenStreamAdmissionVariants pins the attachment-admission
380// contract: exact, create-named, and create-ephemeral round-trip, control
381// carries none, and wrong admission/purpose/name/target combinations are
382// refused on decode exactly like ports.BrokerOpenStreamRequest.Validate.
383func TestBrokerOpenStreamAdmissionVariants(t *testing.T) {
384 connection := testConnectionID(0x11)
385 base := func() OpenStream {
386 return OpenStream{
387 Epoch: 7, Connection: connection, Stream: 3, Purpose: ports.BrokerStreamAttachment,
388 Endpoint: "dev@host:22", Registration: testRegistration(), Policy: testPolicy(),
389 // An attachment or creation always carries the explicit
390 // start-if-needed authorization; the zero mode is refused.
391 StartMode: ports.BrokerDaemonStartIfNeeded,
392 }
393 }
394 roundTrips := []struct {
395 name string
396 message OpenStream
397 }{
398 {"exact", func() OpenStream {
399 message := base()
400 message.Admission = ports.BrokerAdmissionExact
401 message.Target = testTarget()
402 return message
403 }()},
404 {"create named", func() OpenStream {
405 message := base()
406 message.Admission = ports.BrokerAdmissionCreateNamed
407 message.Name = "work"
408 return message
409 }()},
410 {"create ephemeral", func() OpenStream {
411 message := base()
412 message.Admission = ports.BrokerAdmissionCreateEphemeral
413 return message
414 }()},
415 {"attach named", func() OpenStream {
416 message := base()
417 message.Admission = ports.BrokerAdmissionAttachNamed
418 message.Name = "work"
419 return message
420 }()},
421 {"control carries none", OpenStream{
422 Epoch: 7, Connection: connection, Stream: 3, Purpose: ports.BrokerStreamControl,
423 Local: true, Policy: testPolicy(), StartMode: ports.BrokerDaemonStartIfNeeded,
424 }},
425 }
426 for _, tc := range roundTrips {
427 t.Run(tc.name, func(t *testing.T) {
428 raw := mustEncodeClient(t, tc.message)
429 decoded, err := DecodeClient(raw, testEnvelopeCeiling, testChunkCeiling)
430 require.NoError(t, err)
431 require.Equal(t, tc.message, decoded)
432 })
433 }
434
435 t.Run("wrong combinations refused", func(t *testing.T) {
436 exact := func() OpenStream {
437 message := base()
438 message.Admission = ports.BrokerAdmissionExact
439 message.Target = testTarget()
440 return message
441 }
442 cases := map[string]OpenStream{
443 "attachment without admission": func() OpenStream { m := base(); return m }(),
444 "exact with name": func() OpenStream {
445 m := exact()
446 m.Name = "work"
447 return m
448 }(),
449 "named with target": func() OpenStream {
450 m := base()
451 m.Admission = ports.BrokerAdmissionCreateNamed
452 m.Name = "work"
453 m.Target = testTarget()
454 return m
455 }(),
456 "named without name": func() OpenStream {
457 m := base()
458 m.Admission = ports.BrokerAdmissionCreateNamed
459 return m
460 }(),
461 "attach named with target": func() OpenStream {
462 m := base()
463 m.Admission = ports.BrokerAdmissionAttachNamed
464 m.Name = "work"
465 m.Target = testTarget()
466 return m
467 }(),
468 "attach named without name": func() OpenStream {
469 m := base()
470 m.Admission = ports.BrokerAdmissionAttachNamed
471 return m
472 }(),
473 "ephemeral with name": func() OpenStream {
474 m := base()
475 m.Admission = ports.BrokerAdmissionCreateEphemeral
476 m.Name = "work"
477 return m
478 }(),
479 "control with admission": OpenStream{
480 Epoch: 7, Connection: connection, Stream: 3, Purpose: ports.BrokerStreamControl,
481 Local: true, Policy: testPolicy(), Admission: ports.BrokerAdmissionExact, StartMode: ports.BrokerDaemonStartIfNeeded,
482 },
483 "attachment with zero start mode": func() OpenStream { m := exact(); m.StartMode = 0; return m }(),
484 "attachment with existing-only start mode": func() OpenStream {
485 m := exact()
486 m.StartMode = ports.BrokerDaemonExistingOnly
487 return m
488 }(),
489 "observation with start-if-needed start mode": OpenStream{
490 Epoch: 7, Connection: connection, Stream: 3, Purpose: ports.BrokerStreamObservation,
491 Local: true, Policy: testPolicy(), StartMode: ports.BrokerDaemonStartIfNeeded,
492 },
493 }
494 for name, message := range cases {
495 t.Run(name, func(t *testing.T) {
496 // Start mode is a closed wire enum and is refused even by the encoder.
497 if message.StartMode == 0 {
498 _, err := EncodeClient(message, testEnvelopeCeiling, testChunkCeiling)
499 require.Error(t, err)
500 return
501 }
502 raw, err := EncodeClient(message, testEnvelopeCeiling, testChunkCeiling)
503 require.NoError(t, err)
504 _, err = DecodeClient(raw, testEnvelopeCeiling, testChunkCeiling)
505 if name == "named with target" || name == "attach named with target" {
506 // The encoder omits a target on creation; test the untrusted wire directly.
507 envelope := &wire.BrokerClientEnvelope{}
508 require.NoError(t, proto.Unmarshal(raw, envelope))
509 envelope.GetOpenStream().Target = exactTargetToWire(testTarget())
510 raw, err = proto.Marshal(envelope)
511 require.NoError(t, err)
512 _, err = DecodeClient(raw, testEnvelopeCeiling, testChunkCeiling)
513 }
514 require.ErrorIs(t, err, ErrInvalidMessage)
515 })
516 }
517 })
518
519 t.Run("unknown admission code refused on decode", func(t *testing.T) {
520 message := base()
521 message.Admission = ports.BrokerAdmissionExact
522 message.Target = testTarget()
523 raw := mustEncodeClient(t, message)
524 envelope := &wire.BrokerClientEnvelope{}
525 require.NoError(t, wire.ScanEnvelope(envelope, raw))
526 require.NoError(t, proto.Unmarshal(raw, envelope))
527 envelope.GetOpenStream().Admission = 9
528 mutated, err := proto.Marshal(envelope)
529 require.NoError(t, err)
530 _, err = DecodeClient(mutated, testEnvelopeCeiling, testChunkCeiling)
531 require.ErrorIs(t, err, ErrInvalidMessage)
532 })
533
534 t.Run("start mode round-trips and is never defaulted", func(t *testing.T) {
535 // The mode round-trips with the purpose it is legal for: an existing-only
536 // authorization belongs to control/observation, a start-if-needed one to
537 // attachment/control.
538 control := func(mode ports.BrokerDaemonStartMode) OpenStream {
539 return OpenStream{Epoch: 7, Connection: connection, Stream: 3, Purpose: ports.BrokerStreamControl, Local: true, Policy: testPolicy(), StartMode: mode}
540 }
541 for _, message := range []OpenStream{
542 control(ports.BrokerDaemonExistingOnly),
543 control(ports.BrokerDaemonStartIfNeeded),
544 func() OpenStream {
545 m := base()
546 m.Admission = ports.BrokerAdmissionExact
547 m.Target = testTarget()
548 m.StartMode = ports.BrokerDaemonStartIfNeeded
549 return m
550 }(),
551 } {
552 decoded, err := DecodeClient(mustEncodeClient(t, message), testEnvelopeCeiling, testChunkCeiling)
553 require.NoError(t, err)
554 admitting, ok := decoded.(OpenStream)
555 require.True(t, ok)
556 require.Equal(t, message.StartMode, admitting.StartMode)
557 }
558 // An absent wire mode is refused, never widened into a spawn-capable
559 // default, and an unknown code is refused too.
560 message := base()
561 message.Admission = ports.BrokerAdmissionExact
562 message.Target = testTarget()
563 raw := mustEncodeClient(t, message)
564 envelope := &wire.BrokerClientEnvelope{}
565 require.NoError(t, proto.Unmarshal(raw, envelope))
566 envelope.GetOpenStream().StartMode = 0
567 mutated, err := proto.Marshal(envelope)
568 require.NoError(t, err)
569 _, err = DecodeClient(mutated, testEnvelopeCeiling, testChunkCeiling)
570 require.ErrorIs(t, err, ErrInvalidMessage)
571 envelope.GetOpenStream().StartMode = 3
572 mutated, err = proto.Marshal(envelope)
573 require.NoError(t, err)
574 _, err = DecodeClient(mutated, testEnvelopeCeiling, testChunkCeiling)
575 require.ErrorIs(t, err, ErrInvalidMessage)
576 // An observation that authorized a spawn is refused rather than narrowed.
577 envelope.GetOpenStream().Purpose = 3
578 envelope.GetOpenStream().StartMode = 2
579 envelope.GetOpenStream().Target = nil
580 envelope.GetOpenStream().Admission = 0
581 mutated, err = proto.Marshal(envelope)
582 require.NoError(t, err)
583 _, err = DecodeClient(mutated, testEnvelopeCeiling, testChunkCeiling)
584 require.ErrorIs(t, err, ErrInvalidMessage)
585 })
586}
587
588// TestBrokerSnapshotDaemonByteForByte proves the daemon and session parts are
589// byte-for-byte stable across an encode/decode/re-encode cycle, including an
590// observed identity/incarnation/version/capability set, the local daemon, and a
591// locally-bound session.
592func TestBrokerSnapshotDaemonByteForByte(t *testing.T) {
593 connection := testConnectionID(0x11)
594 messages := []struct {
595 name string
596 message SnapshotPart
597 }{
598 {"observed remote daemon", SnapshotPart{Epoch: 7, Connection: connection, Generation: 2, Revision: 5, Index: 1, Part: SnapshotDaemonPart{HostIndex: 1, Daemon: testDaemonObservation(), SessionCount: 1}}},
599 {"reachable remote without daemon", SnapshotPart{Epoch: 7, Connection: connection, Generation: 2, Revision: 5, Index: 2, Part: SnapshotDaemonPart{HostIndex: 1, Daemon: func() ports.BrokerDaemonObservation {
600 daemon := testDaemonObservation()
601 daemon.Identity = ""
602 daemon.Incarnation = ports.BrokerDaemonIncarnation{}
603 daemon.ProtocolVersion = 0
604 daemon.Capabilities = 0
605 daemon.Availability = domain.RemoteAvailabilityNoDaemon
606 daemon.LastFailure = domain.RemoteFailure{}
607 daemon.LastSuccess = time.Time{}
608 daemon.InventoryKnown = false
609 return daemon
610 }(), SessionCount: 0}}},
611 {"local daemon", SnapshotPart{Epoch: 7, Connection: connection, Generation: 2, Revision: 5, Index: 0, Part: SnapshotDaemonPart{HostIndex: 0, Daemon: testLocalDaemonObservation(), SessionCount: 1}}},
612 {"local session", SnapshotPart{Epoch: 7, Connection: connection, Generation: 2, Revision: 5, Index: 2, Part: SnapshotSessionPart{HostIndex: 0, SessionIndex: 0, Local: true, Session: testCatalogSession()}}},
613 }
614 for _, tc := range messages {
615 t.Run(tc.name, func(t *testing.T) {
616 raw := mustEncodeServer(t, tc.message)
617 decoded, err := DecodeServer(raw, testEnvelopeCeiling, testChunkCeiling)
618 require.NoError(t, err)
619 require.Equal(t, tc.message, decoded)
620 again, err := EncodeServer(decoded, testEnvelopeCeiling, testChunkCeiling)
621 require.NoError(t, err)
622 require.Equal(t, raw, again, "re-encode must be byte-for-byte identical")
623 })
624 }
625}
626
627// TestBrokerServerVariants inventories every server variant tag.
628func TestBrokerServerVariants(t *testing.T) {
629 connection := testConnectionID(0x11)
630 operation := testOperationID(0x77)
631 messages := []struct {
632 name string
633 message ServerMessage
634 tag protowire.Number
635 }{
636 {"registered", Registered{Epoch: 7, Connection: connection}, 201},
637 {"snapshot_begin", SnapshotPart{Epoch: 7, Connection: connection, Generation: 2, Revision: 5, Index: 0, Part: SnapshotBegin{HostCount: 2, SessionCount: 2, TombstoneCount: 0, LocalPresent: true}}, 202},
638 {"snapshot_daemon", SnapshotPart{Epoch: 7, Connection: connection, Generation: 2, Revision: 5, Index: 1, Part: SnapshotDaemonPart{HostIndex: 0, Daemon: testLocalDaemonObservation(), SessionCount: 0}}, 202},
639 {"snapshot_daemon_remote", SnapshotPart{Epoch: 7, Connection: connection, Generation: 2, Revision: 5, Index: 2, Part: SnapshotDaemonPart{HostIndex: 1, Daemon: testDaemonObservation(), SessionCount: 1}}, 202},
640 {"snapshot_session", SnapshotPart{Epoch: 7, Connection: connection, Generation: 2, Revision: 5, Index: 3, Part: SnapshotSessionPart{HostIndex: 1, SessionIndex: 0, Session: testCatalogSession()}}, 202},
641 {"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},
642 {"snapshot_tombstone", SnapshotPart{Epoch: 7, Connection: connection, Generation: 2, Revision: 5, Index: 5, Part: SnapshotTombstonePart{TombstoneIndex: 0, Registration: testRegistration(), RetiredRevision: 4}}, 202},
643 {"snapshot_end", SnapshotPart{Epoch: 7, Connection: connection, Generation: 2, Revision: 5, Index: 6, Part: SnapshotEnd{}}, 202},
644 {"operation_result", OperationResult{Epoch: 7, Connection: connection, Operation: operation, Outcome: ports.BrokerOutcomeFailed, Error: testErrorDetail(), HasError: true}, 203},
645 {"operation_result_add", OperationResult{Epoch: 7, Connection: connection, Operation: operation, Outcome: ports.BrokerOutcomeOK, Registration: testRegistration()}, 203},
646 {"operation_result_remove", OperationResult{Epoch: 7, Connection: connection, Operation: operation, Outcome: ports.BrokerOutcomeOK, Removed: true}, 203},
647 {"stream_opened", StreamOpened{Epoch: 7, Connection: connection, Stream: 3}, 204},
648 {"server_stream_data", ServerStreamData{Epoch: 7, Connection: connection, Stream: 3, Data: []byte("frame")}, 205},
649 {"stream_closed", StreamClosed{Epoch: 7, Connection: connection, Stream: 3, Error: testErrorDetail(), HasError: true}, 206},
650 {"stream_closed_orderly", StreamClosed{Epoch: 7, Connection: connection, Stream: 4}, 206},
651 {"progress", Progress{Epoch: 7, Connection: connection, Stream: 3, Phase: BrokerProgressProbing, Text: "probing"}, 207},
652 {"broker_error", BrokerErrorMessage{Epoch: 7, Connection: connection, Error: testErrorDetail()}, 208},
653 {"shutdown", Shutdown{Epoch: 7, Connection: connection, Reason: ShutdownTerminating, Text: "stopping"}, 209},
654 {"preview_publication", testPreviewPublication(connection), 210},
655 {"preview_publication_error", PreviewPublication{Epoch: 7, Connection: connection, Generation: 5, Error: testErrorDetail(), HasError: true}, 210},
656 }
657 for _, tc := range messages {
658 t.Run(tc.name, func(t *testing.T) {
659 raw := mustEncodeServer(t, tc.message)
660 number, wireType, consumed := protowire.ConsumeTag(raw)
661 require.Greater(t, consumed, 0)
662 require.Equal(t, tc.tag, number)
663 require.Equal(t, protowire.BytesType, wireType)
664 decoded, err := DecodeServer(raw, testEnvelopeCeiling, testChunkCeiling)
665 require.NoError(t, err)
666 require.Equal(t, tc.message, decoded)
667 })
668 }
669}
670
671// TestBrokerDownSessionEmptyTabsRoundTrip proves a down session with an
672// empty-present tab list survives a full encode/decode round trip: proto3
673// repeated fields carry no presence, so the decoder must map an absent
674// field back to an empty-present list rather than refusing it.
675func TestBrokerDownSessionEmptyTabsRoundTrip(t *testing.T) {
676 connection := testConnectionID(0x11)
677 session := catalogue.RemoteCatalogSession{
678 LifecycleID: domain.SessionLifecycleID{1, 2, 3},
679 Name: "down",
680 State: catalogue.RemoteCatalogSessionDown,
681 Tabs: []catalogue.RemoteCatalogTab{},
682 }
683 message := SnapshotPart{
684 Epoch: 7, Connection: connection, Generation: 2, Revision: 5, Index: 2,
685 Part: SnapshotSessionPart{HostIndex: 0, SessionIndex: 0, Session: session},
686 }
687 raw := mustEncodeServer(t, message)
688 decoded, err := DecodeServer(raw, testEnvelopeCeiling, testChunkCeiling)
689 require.NoError(t, err)
690 require.Equal(t, message, decoded)
691
692 part, ok := decoded.(SnapshotPart)
693 require.True(t, ok)
694 sessionPart, ok := part.Part.(SnapshotSessionPart)
695 require.True(t, ok)
696 require.NotNil(t, sessionPart.Session.Tabs, "empty tabs decode as empty-present, never nil")
697 require.Empty(t, sessionPart.Session.Tabs)
698}
699
700// TestBrokerTabIndexBound proves a tab index at the per-session maximum is a
701// range refusal, not a semantic one: valid indexes are 0..max-1.
702func TestBrokerTabIndexBound(t *testing.T) {
703 connection := testConnectionID(0x11)
704 envelope := &wire.BrokerServerEnvelope{Payload: &wire.BrokerServerEnvelope_SnapshotPart{SnapshotPart: &wire.SnapshotPart{
705 Scope: &wire.BrokerScope{BrokerEpoch: 7, ConnectionId: connection[:]},
706 Generation: 2,
707 Revision: 5,
708 Index: 2,
709 Part: &wire.SnapshotPart_Session{Session: &wire.SnapshotSession{
710 HostIndex: 0, SessionIndex: 0,
711 Session: &wire.BrokerCatalogSession{
712 LifecycleId: []byte{1, 2, 3},
713 Name: "down",
714 State: 2,
715 Tabs: []*wire.BrokerCatalogTab{{Id: "tab", Index: catalogue.RemoteCatalogMaxTabsPerSess, Name: "tab"}},
716 },
717 }},
718 }}}
719 raw, err := proto.Marshal(envelope)
720 require.NoError(t, err)
721 _, err = DecodeServer(raw, testEnvelopeCeiling, testChunkCeiling)
722 require.ErrorIs(t, err, errConvertRange)
723}
724
725// mutateDaemonIndex re-encodes one daemon snapshot part envelope with its
726// daemon index overwritten, so decode can be fed a value encode would never
727// emit.
728func mutateDaemonIndex(t *testing.T, part SnapshotPart, hostIndex uint32) []byte {
729 t.Helper()
730 raw := mustEncodeServer(t, part)
731 envelope := &wire.BrokerServerEnvelope{}
732 require.NoError(t, wire.ScanEnvelope(envelope, raw))
733 require.NoError(t, proto.Unmarshal(raw, envelope))
734 envelope.GetSnapshotPart().GetDaemon().HostIndex = hostIndex
735 mutated, err := proto.Marshal(envelope)
736 require.NoError(t, err)
737 return mutated
738}
739
740// mutateSessionIndexes re-encodes one session snapshot part envelope with its
741// host and session indexes overwritten.
742func mutateSessionIndexes(t *testing.T, part SnapshotPart, hostIndex, sessionIndex uint32) []byte {
743 t.Helper()
744 raw := mustEncodeServer(t, part)
745 envelope := &wire.BrokerServerEnvelope{}
746 require.NoError(t, wire.ScanEnvelope(envelope, raw))
747 require.NoError(t, proto.Unmarshal(raw, envelope))
748 envelope.GetSnapshotPart().GetSession().HostIndex = hostIndex
749 envelope.GetSnapshotPart().GetSession().SessionIndex = sessionIndex
750 mutated, err := proto.Marshal(envelope)
751 require.NoError(t, err)
752 return mutated
753}
754
755// TestBrokerSnapshotIndexBounds proves snapshot part indexes are strict
756// ranges: max-1 is accepted on both encode and decode, and max is refused as a
757// range refusal on both paths. Daemon indexes are bounded by
758// BrokerMaxDaemonsPerSnapshot and session indexes by BrokerMaxSessionsPerHost.
759func TestBrokerSnapshotIndexBounds(t *testing.T) {
760 connection := testConnectionID(0x11)
761 daemonPart := func(hostIndex uint32) SnapshotPart {
762 return SnapshotPart{Epoch: 7, Connection: connection, Generation: 2, Revision: 5, Index: 1, Part: SnapshotDaemonPart{HostIndex: hostIndex, Daemon: testDaemonObservation()}}
763 }
764 sessionPart := func(hostIndex, sessionIndex uint32) SnapshotPart {
765 return SnapshotPart{Epoch: 7, Connection: connection, Generation: 2, Revision: 5, Index: 2, Part: SnapshotSessionPart{HostIndex: hostIndex, SessionIndex: sessionIndex, Session: testCatalogSession()}}
766 }
767
768 t.Run("daemon part at max-1 accepted", func(t *testing.T) {
769 message := daemonPart(ports.BrokerMaxDaemonsPerSnapshot - 1)
770 raw := mustEncodeServer(t, message)
771 decoded, err := DecodeServer(raw, testEnvelopeCeiling, testChunkCeiling)
772 require.NoError(t, err)
773 require.Equal(t, message, decoded)
774 decoded, err = DecodeServer(mutateDaemonIndex(t, message, ports.BrokerMaxDaemonsPerSnapshot-1), testEnvelopeCeiling, testChunkCeiling)
775 require.NoError(t, err)
776 require.Equal(t, message, decoded)
777 })
778
779 t.Run("daemon part at max refused", func(t *testing.T) {
780 _, err := EncodeServer(daemonPart(ports.BrokerMaxDaemonsPerSnapshot), testEnvelopeCeiling, testChunkCeiling)
781 require.ErrorIs(t, err, ErrTooLarge)
782 _, err = DecodeServer(mutateDaemonIndex(t, daemonPart(0), ports.BrokerMaxDaemonsPerSnapshot), testEnvelopeCeiling, testChunkCeiling)
783 require.ErrorIs(t, err, ErrTooLarge)
784 })
785
786 t.Run("daemon part at uint32 maximum refused", func(t *testing.T) {
787 _, err := EncodeServer(daemonPart(^uint32(0)), testEnvelopeCeiling, testChunkCeiling)
788 require.ErrorIs(t, err, ErrTooLarge)
789 _, err = DecodeServer(mutateDaemonIndex(t, daemonPart(0), ^uint32(0)), testEnvelopeCeiling, testChunkCeiling)
790 require.ErrorIs(t, err, ErrTooLarge)
791 })
792
793 t.Run("session part at max-1 accepted", func(t *testing.T) {
794 message := sessionPart(ports.BrokerMaxDaemonsPerSnapshot-1, ports.BrokerMaxSessionsPerHost-1)
795 raw := mustEncodeServer(t, message)
796 decoded, err := DecodeServer(raw, testEnvelopeCeiling, testChunkCeiling)
797 require.NoError(t, err)
798 require.Equal(t, message, decoded)
799 decoded, err = DecodeServer(mutateSessionIndexes(t, message, ports.BrokerMaxDaemonsPerSnapshot-1, ports.BrokerMaxSessionsPerHost-1), testEnvelopeCeiling, testChunkCeiling)
800 require.NoError(t, err)
801 require.Equal(t, message, decoded)
802 })
803
804 t.Run("session part host index at max refused", func(t *testing.T) {
805 _, err := EncodeServer(sessionPart(ports.BrokerMaxDaemonsPerSnapshot, 0), testEnvelopeCeiling, testChunkCeiling)
806 require.ErrorIs(t, err, ErrTooLarge)
807 _, err = DecodeServer(mutateSessionIndexes(t, sessionPart(0, 0), ports.BrokerMaxDaemonsPerSnapshot, 0), testEnvelopeCeiling, testChunkCeiling)
808 require.ErrorIs(t, err, ErrTooLarge)
809 })
810
811 t.Run("session part session index at max refused", func(t *testing.T) {
812 _, err := EncodeServer(sessionPart(0, ports.BrokerMaxSessionsPerHost), testEnvelopeCeiling, testChunkCeiling)
813 require.ErrorIs(t, err, ErrTooLarge)
814 _, err = DecodeServer(mutateSessionIndexes(t, sessionPart(0, 0), 0, ports.BrokerMaxSessionsPerHost), testEnvelopeCeiling, testChunkCeiling)
815 require.ErrorIs(t, err, ErrTooLarge)
816 })
817}
818
819// TestBrokerGoldenVectors pins byte-for-byte wire encodings.
820func TestBrokerGoldenVectors(t *testing.T) {
821 connection := testConnectionID(0x11)
822 raw := mustEncodeClient(t, Register{})
823 decoded, err := DecodeClient(raw, testEnvelopeCeiling, testChunkCeiling)
824 require.NoError(t, err)
825 require.Equal(t, Register{}, decoded)
826 // Re-encode is deterministic: the same message marshals identically.
827 again, err := EncodeClient(Register{}, testEnvelopeCeiling, testChunkCeiling)
828 require.NoError(t, err)
829 require.Equal(t, raw, again)
830
831 sub := Subscribe{Epoch: 7, Connection: connection, Generation: 1 << 40}
832 rawSub := mustEncodeClient(t, sub)
833 envelope := &wire.BrokerClientEnvelope{}
834 require.NoError(t, wire.ScanEnvelope(envelope, rawSub))
835 require.NoError(t, proto.Unmarshal(rawSub, envelope))
836 require.Equal(t, uint64(7), envelope.GetSubscribe().GetScope().GetBrokerEpoch())
837 require.Equal(t, uint64(1<<40), envelope.GetSubscribe().GetGeneration())
838 back, err := DecodeClient(rawSub, testEnvelopeCeiling, testChunkCeiling)
839 require.NoError(t, err)
840 require.Equal(t, sub, back)
841
842 // The passive reader flag survives the round trip, and an active
843 // subscription encodes without it (proto3 omits false).
844 passive := sub
845 passive.Passive = true
846 rawPassive := mustEncodeClient(t, passive)
847 back, err = DecodeClient(rawPassive, testEnvelopeCeiling, testChunkCeiling)
848 require.NoError(t, err)
849 require.Equal(t, passive, back)
850 require.Greater(t, len(rawPassive), len(rawSub))
851}
852
853// TestBrokerBounds proves stateless bound refusals.
854func TestBrokerBounds(t *testing.T) {
855 connection := testConnectionID(0x11)
856 t.Run("envelope ceiling", func(t *testing.T) {
857 raw := mustEncodeClient(t, Register{})
858 _, err := DecodeClient(raw, uint64(len(raw)-1), testChunkCeiling)
859 require.ErrorIs(t, err, wire.ErrScanLength)
860 _, err = DecodeClient(make([]byte, testEnvelopeCeiling+1), testEnvelopeCeiling, testChunkCeiling)
861 require.ErrorIs(t, err, wire.ErrScanLength)
862 })
863 t.Run("chunk bound", func(t *testing.T) {
864 big := bytes.Repeat([]byte{0x66}, int(MaxStreamChunkBytes)+1)
865 _, err := EncodeClient(ClientStreamData{Epoch: 7, Connection: connection, Stream: 3, Data: big}, testEnvelopeCeiling, testEnvelopeCeiling)
866 require.ErrorIs(t, err, ErrTooLarge)
867 _, err = EncodeServer(ServerStreamData{Epoch: 7, Connection: connection, Stream: 3, Data: big}, testEnvelopeCeiling, testEnvelopeCeiling)
868 require.ErrorIs(t, err, ErrTooLarge)
869 // A negotiated chunk below the absolute max also refuses.
870 raw := mustEncodeClient(t, ClientStreamData{Epoch: 7, Connection: connection, Stream: 3, Data: []byte("12345678")})
871 _, err = DecodeClient(raw, testEnvelopeCeiling, 4)
872 require.ErrorIs(t, err, ErrTooLarge)
873 // Empty chunk is invalid, not too large.
874 _, err = EncodeClient(ClientStreamData{Epoch: 7, Connection: connection, Stream: 3}, testEnvelopeCeiling, testChunkCeiling)
875 require.ErrorIs(t, err, ErrInvalidMessage)
876 })
877 t.Run("env entries", func(t *testing.T) {
878 many := make([]string, ports.BrokerMaxEnvEntries+1)
879 for i := range many {
880 many[i] = "A=B"
881 }
882 _, 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)
883 require.ErrorIs(t, err, ErrTooLarge)
884 oversize := []string{"K=" + strings.Repeat("x", ports.BrokerMaxEnvEntryBytes)}
885 _, 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)
886 require.ErrorIs(t, err, ErrTooLarge)
887 })
888 t.Run("error text", func(t *testing.T) {
889 oversize := testErrorDetail()
890 oversize.Text = strings.Repeat("e", ports.BrokerMaxErrorBytes+1)
891 _, err := EncodeServer(BrokerErrorMessage{Epoch: 7, Connection: connection, Error: oversize}, testEnvelopeCeiling, testChunkCeiling)
892 require.ErrorIs(t, err, ErrTooLarge)
893 })
894 t.Run("snapshot counts", func(t *testing.T) {
895 _, err := EncodeServer(SnapshotPart{Epoch: 7, Connection: connection, Generation: 1, Revision: 1, Index: 0, Part: SnapshotBegin{HostCount: ports.BrokerMaxDaemonsPerSnapshot + 1}}, testEnvelopeCeiling, testChunkCeiling)
896 require.ErrorIs(t, err, ErrTooLarge)
897 })
898 t.Run("control env carries no attachment state", func(t *testing.T) {
899 message := OpenStream{Epoch: 7, Connection: connection, Stream: 3, Purpose: ports.BrokerStreamControl, Local: true, Policy: testPolicy(), StartMode: ports.BrokerDaemonStartIfNeeded}
900 raw := mustEncodeClient(t, message)
901 envelope := &wire.BrokerClientEnvelope{}
902 require.NoError(t, proto.Unmarshal(raw, envelope))
903 envelope.GetOpenStream().Target = exactTargetToWire(testTarget())
904 raw, err := proto.Marshal(envelope)
905 require.NoError(t, err)
906 _, err = DecodeClient(raw, testEnvelopeCeiling, testChunkCeiling)
907 require.ErrorIs(t, err, ErrInvalidMessage)
908 })
909}
910
911// TestBrokerValidationNegatives is the table of semantic refusals: IDs,
912// UTF-8/control/bidi text, taxonomy, times, env/policy, snapshot parts.
913func TestBrokerValidationNegatives(t *testing.T) {
914 connection := testConnectionID(0x11)
915 t.Run("zero scope", func(t *testing.T) {
916 _, err := EncodeClient(Subscribe{Epoch: 0, Connection: connection}, testEnvelopeCeiling, testChunkCeiling)
917 require.ErrorIs(t, err, ErrInvalidMessage)
918 _, err = EncodeClient(Subscribe{Epoch: 7}, testEnvelopeCeiling, testChunkCeiling)
919 require.ErrorIs(t, err, ErrInvalidMessage)
920 _, err = EncodeServer(Registered{Epoch: 7}, testEnvelopeCeiling, testChunkCeiling)
921 require.ErrorIs(t, err, ErrInvalidMessage)
922 })
923 t.Run("zero stream", func(t *testing.T) {
924 _, err := EncodeClient(CloseStream{Epoch: 7, Connection: connection}, testEnvelopeCeiling, testChunkCeiling)
925 require.ErrorIs(t, err, ErrInvalidMessage)
926 })
927 t.Run("error detail never travels without HasError", func(t *testing.T) {
928 detail := testErrorDetail()
929 // HasError=false is exactly the absent detail, whatever the outcome
930 // and whatever field of the detail is nonzero.
931 for name, stray := range map[string]ErrorDetail{
932 "full detail": detail,
933 "code only": {Code: ports.BrokerErrorTimeout},
934 "text only": {Text: "late"},
935 "admission only": {AdmissionCode: 2},
936 "failure kind only": {FailureKind: domain.RemoteFailureTransport},
937 } {
938 t.Run(name, func(t *testing.T) {
939 for _, outcome := range []ports.BrokerMutationOutcome{ports.BrokerOutcomeOK, ports.BrokerOutcomeFailed, ports.BrokerOutcomeUnknown} {
940 _, err := EncodeServer(OperationResult{Epoch: 7, Connection: connection, Operation: testOperationID(1), Outcome: outcome, Error: stray}, testEnvelopeCeiling, testChunkCeiling)
941 require.ErrorIs(t, err, ErrInvalidMessage)
942 }
943 _, err := EncodeServer(StreamClosed{Epoch: 7, Connection: connection, Stream: 3, Error: stray}, testEnvelopeCeiling, testChunkCeiling)
944 require.ErrorIs(t, err, ErrInvalidMessage)
945 })
946 }
947 // The zero detail is the canonical absent value and still encodes; a
948 // declared detail still round-trips exactly.
949 _, err := EncodeServer(StreamClosed{Epoch: 7, Connection: connection, Stream: 3}, testEnvelopeCeiling, testChunkCeiling)
950 require.NoError(t, err)
951 _, err = EncodeServer(OperationResult{Epoch: 7, Connection: connection, Operation: testOperationID(1), Outcome: ports.BrokerOutcomeOK, Registration: testRegistration()}, testEnvelopeCeiling, testChunkCeiling)
952 require.NoError(t, err)
953 declared := OperationResult{Epoch: 7, Connection: connection, Operation: testOperationID(1), Outcome: ports.BrokerOutcomeFailed, Error: detail, HasError: true}
954 decoded, err := DecodeServer(mustEncodeServer(t, declared), testEnvelopeCeiling, testChunkCeiling)
955 require.NoError(t, err)
956 require.Equal(t, declared, decoded)
957 })
958 t.Run("invalid error detail refused on every encode path", func(t *testing.T) {
959 badAdmission := testErrorDetail()
960 badAdmission.AdmissionCode = MaxBrokerAdmissionCode + 1
961 badKind := testErrorDetail()
962 badKind.FailureKind = MaxBrokerFailureKind + 1
963 badCode := testErrorDetail()
964 badCode.Code = 0
965 for name, detail := range map[string]ErrorDetail{
966 "admission over maximum": badAdmission,
967 "failure kind over bound": badKind,
968 "invalid error code": badCode,
969 } {
970 t.Run(name, func(t *testing.T) {
971 _, err := EncodeServer(BrokerErrorMessage{Epoch: 7, Connection: connection, Error: detail}, testEnvelopeCeiling, testChunkCeiling)
972 require.ErrorIs(t, err, ErrInvalidMessage)
973 _, err = EncodeServer(OperationResult{Epoch: 7, Connection: connection, Operation: testOperationID(1), Outcome: ports.BrokerOutcomeFailed, Error: detail, HasError: true}, testEnvelopeCeiling, testChunkCeiling)
974 require.ErrorIs(t, err, ErrInvalidMessage)
975 _, err = EncodeServer(StreamClosed{Epoch: 7, Connection: connection, Stream: 3, Error: detail, HasError: true}, testEnvelopeCeiling, testChunkCeiling)
976 require.ErrorIs(t, err, ErrInvalidMessage)
977 })
978 }
979 // The closed taxonomy boundaries are accepted.
980 boundary := testErrorDetail()
981 boundary.AdmissionCode = MaxBrokerAdmissionCode
982 boundary.FailureKind = MaxBrokerFailureKind
983 _, err := EncodeServer(BrokerErrorMessage{Epoch: 7, Connection: connection, Error: boundary}, testEnvelopeCeiling, testChunkCeiling)
984 require.NoError(t, err)
985 })
986 t.Run("membership request authority", func(t *testing.T) {
987 base := AddHost{Epoch: 7, Connection: connection, Operation: testOperationID(1), Endpoint: "dev@host:22", Policy: testPolicy()}
988 // An absent policy is not authority: it is refused exactly like an
989 // invalid one rather than inherited from a default.
990 missing := base
991 missing.Policy = ports.BrokerPolicy{}
992 _, err := EncodeClient(missing, testEnvelopeCeiling, testChunkCeiling)
993 require.ErrorIs(t, err, ErrInvalidMessage)
994
995 // A partial registration is not exact authority either.
996 _, err = EncodeClient(RemoveHost{Epoch: 7, Connection: connection, Operation: testOperationID(1)}, testEnvelopeCeiling, testChunkCeiling)
997 require.ErrorIs(t, err, ErrInvalidMessage)
998 partial := testRegistration()
999 partial.Incarnation = [16]byte{}
1000 _, err = EncodeClient(RemoveHost{Epoch: 7, Connection: connection, Operation: testOperationID(1), Registration: partial}, testEnvelopeCeiling, testChunkCeiling)
1001 require.ErrorIs(t, err, ErrInvalidMessage)
1002 partial = testRegistration()
1003 partial.Generation = 0
1004 _, err = EncodeClient(RemoveHost{Epoch: 7, Connection: connection, Operation: testOperationID(1), Registration: partial}, testEnvelopeCeiling, testChunkCeiling)
1005 require.ErrorIs(t, err, ErrInvalidMessage)
1006
1007 update := UpdateHostPolicy{Epoch: 7, Connection: connection, Operation: testOperationID(1), Registration: testRegistration(), Policy: testPolicy()}
1008 missingPolicy := update
1009 missingPolicy.Policy = ports.BrokerPolicy{}
1010 _, err = EncodeClient(missingPolicy, testEnvelopeCeiling, testChunkCeiling)
1011 require.ErrorIs(t, err, ErrInvalidMessage)
1012 missingRegistration := update
1013 missingRegistration.Registration = domain.RemoteRegistration{}
1014 _, err = EncodeClient(missingRegistration, testEnvelopeCeiling, testChunkCeiling)
1015 require.ErrorIs(t, err, ErrInvalidMessage)
1016 })
1017 t.Run("membership result presence", func(t *testing.T) {
1018 // A result may never carry both removal and registration, and a
1019 // failed or outcome-unknown result may carry neither authority and
1020 // must carry its typed error.
1021 both := OperationResult{Epoch: 7, Connection: connection, Operation: testOperationID(1), Outcome: ports.BrokerOutcomeOK, Removed: true, Registration: testRegistration()}
1022 _, err := EncodeServer(both, testEnvelopeCeiling, testChunkCeiling)
1023 require.ErrorIs(t, err, ErrInvalidMessage)
1024 for _, outcome := range []ports.BrokerMutationOutcome{ports.BrokerOutcomeFailed, ports.BrokerOutcomeUnknown} {
1025 _, err = EncodeServer(OperationResult{Epoch: 7, Connection: connection, Operation: testOperationID(1), Outcome: outcome, Registration: testRegistration(), Error: testErrorDetail(), HasError: true}, testEnvelopeCeiling, testChunkCeiling)
1026 require.ErrorIs(t, err, ErrInvalidMessage)
1027 _, err = EncodeServer(OperationResult{Epoch: 7, Connection: connection, Operation: testOperationID(1), Outcome: outcome, Removed: true, Error: testErrorDetail(), HasError: true}, testEnvelopeCeiling, testChunkCeiling)
1028 require.ErrorIs(t, err, ErrInvalidMessage)
1029 // Neither authority without an error is refused too: a failure
1030 // with no detail would read as an unexplained refusal.
1031 _, err = EncodeServer(OperationResult{Epoch: 7, Connection: connection, Operation: testOperationID(1), Outcome: outcome}, testEnvelopeCeiling, testChunkCeiling)
1032 require.ErrorIs(t, err, ErrInvalidMessage)
1033 }
1034 // A success carries no error at all.
1035 _, err = EncodeServer(OperationResult{Epoch: 7, Connection: connection, Operation: testOperationID(1), Outcome: ports.BrokerOutcomeOK, Registration: testRegistration(), Error: testErrorDetail(), HasError: true}, testEnvelopeCeiling, testChunkCeiling)
1036 require.ErrorIs(t, err, ErrInvalidMessage)
1037 // A partial registration is refused on the result path too.
1038 partial := testRegistration()
1039 partial.Incarnation = [16]byte{}
1040 _, err = EncodeServer(OperationResult{Epoch: 7, Connection: connection, Operation: testOperationID(1), Outcome: ports.BrokerOutcomeOK, Registration: partial}, testEnvelopeCeiling, testChunkCeiling)
1041 require.ErrorIs(t, err, ErrInvalidMessage)
1042 })
1043 t.Run("membership result authority by mutation kind", func(t *testing.T) {
1044 registration := testRegistration()
1045 add := OperationResult{Epoch: 7, Connection: connection, Operation: testOperationID(1), Outcome: ports.BrokerOutcomeOK, Registration: registration}
1046 require.NoError(t, add.ValidateForMutation(MutationKindAddHost))
1047 require.ErrorIs(t, add.ValidateForMutation(MutationKindRemoveHost), ErrInvalidMessage)
1048 require.NoError(t, add.ValidateForMutation(MutationKindUpdateHostPolicy))
1049
1050 remove := OperationResult{Epoch: 7, Connection: connection, Operation: testOperationID(1), Outcome: ports.BrokerOutcomeOK, Removed: true}
1051 require.NoError(t, remove.ValidateForMutation(MutationKindRemoveHost))
1052 require.ErrorIs(t, remove.ValidateForMutation(MutationKindAddHost), ErrInvalidMessage)
1053 require.ErrorIs(t, remove.ValidateForMutation(MutationKindUpdateHostPolicy), ErrInvalidMessage)
1054
1055 failed := OperationResult{Epoch: 7, Connection: connection, Operation: testOperationID(1), Outcome: ports.BrokerOutcomeFailed, Error: testErrorDetail(), HasError: true}
1056 for _, kind := range []RegisterMutationKind{MutationKindAddHost, MutationKindRemoveHost, MutationKindUpdateHostPolicy} {
1057 require.NoError(t, failed.ValidateForMutation(kind))
1058 }
1059 // A kind outside the closed taxonomy is refused, never guessed.
1060 require.Error(t, add.ValidateForMutation(RegisterMutationKind(9)))
1061 // The request-side classifier names exactly the three mutations.
1062 for message, want := range map[ClientMessage]RegisterMutationKind{
1063 AddHost{}: MutationKindAddHost,
1064 RemoveHost{}: MutationKindRemoveHost,
1065 UpdateHostPolicy{}: MutationKindUpdateHostPolicy,
1066 Subscribe{}: 0,
1067 } {
1068 got, ok := MutationKindOf(message)
1069 if want == 0 {
1070 require.False(t, ok)
1071 continue
1072 }
1073 require.True(t, ok)
1074 require.Equal(t, want, got)
1075 }
1076 require.Equal(t, "add_host", MutationKindAddHost.String())
1077 require.Equal(t, "remove_host", MutationKindRemoveHost.String())
1078 require.Equal(t, "update_host_policy", MutationKindUpdateHostPolicy.String())
1079 require.Equal(t, "unknown", RegisterMutationKind(9).String())
1080 })
1081 t.Run("bad endpoint", func(t *testing.T) {
1082 _, err := EncodeClient(AddHost{Epoch: 7, Connection: connection, Operation: testOperationID(1), Endpoint: "not a host", Policy: testPolicy()}, testEnvelopeCeiling, testChunkCeiling)
1083 require.ErrorIs(t, err, ErrInvalidMessage)
1084 })
1085 t.Run("taxonomy switches", func(t *testing.T) {
1086 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})
1087 envelope := &wire.BrokerClientEnvelope{}
1088 require.NoError(t, wire.ScanEnvelope(envelope, raw))
1089 require.NoError(t, proto.Unmarshal(raw, envelope))
1090 envelope.GetOpenStream().Purpose = 9
1091 mutated, err := proto.Marshal(envelope)
1092 require.NoError(t, err)
1093 _, err = DecodeClient(mutated, testEnvelopeCeiling, testChunkCeiling)
1094 require.Error(t, err)
1095
1096 rawErr := mustEncodeServer(t, BrokerErrorMessage{Epoch: 7, Connection: connection, Error: testErrorDetail()})
1097 server := &wire.BrokerServerEnvelope{}
1098 require.NoError(t, wire.ScanEnvelope(server, rawErr))
1099 require.NoError(t, proto.Unmarshal(rawErr, server))
1100 server.GetBrokerErrorMessage().GetError().Code = 99
1101 mutated, err = proto.Marshal(server)
1102 require.NoError(t, err)
1103 _, err = DecodeServer(mutated, testEnvelopeCeiling, testChunkCeiling)
1104 require.Error(t, err)
1105 })
1106 t.Run("bidi and control text refused", func(t *testing.T) {
1107 evil := testErrorDetail()
1108 evil.Text = "timeout\u202erorrE"
1109 _, err := EncodeServer(BrokerErrorMessage{Epoch: 7, Connection: connection, Error: evil}, testEnvelopeCeiling, testChunkCeiling)
1110 require.ErrorIs(t, err, ErrInvalidMessage)
1111 evil.Text = "bad\x01text"
1112 _, err = EncodeServer(BrokerErrorMessage{Epoch: 7, Connection: connection, Error: evil}, testEnvelopeCeiling, testChunkCeiling)
1113 require.ErrorIs(t, err, ErrInvalidMessage)
1114 _, err = EncodeServer(Shutdown{Epoch: 7, Connection: connection, Reason: ShutdownIdleExit, Text: "x\x1b"}, testEnvelopeCeiling, testChunkCeiling)
1115 require.ErrorIs(t, err, ErrInvalidMessage)
1116 })
1117 t.Run("env without equals refused", func(t *testing.T) {
1118 _, 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)
1119 require.ErrorIs(t, err, ErrInvalidMessage)
1120 })
1121 t.Run("invalid policy refused", func(t *testing.T) {
1122 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)
1123 require.NoError(t, err)
1124 _, err = DecodeClient(raw, testEnvelopeCeiling, testChunkCeiling)
1125 require.ErrorIs(t, err, ErrInvalidMessage)
1126 })
1127 t.Run("snapshot revision zero refused", func(t *testing.T) {
1128 _, err := EncodeServer(SnapshotPart{Epoch: 7, Connection: connection, Generation: 1, Index: 0, Part: SnapshotEnd{}}, testEnvelopeCeiling, testChunkCeiling)
1129 require.ErrorIs(t, err, ErrInvalidMessage)
1130 })
1131 t.Run("missing snapshot part refused", func(t *testing.T) {
1132 _, err := EncodeServer(SnapshotPart{Epoch: 7, Connection: connection, Generation: 1, Revision: 1}, testEnvelopeCeiling, testChunkCeiling)
1133 require.ErrorIs(t, err, ErrInvalidMessage)
1134 })
1135 t.Run("local open with registration refused", func(t *testing.T) {
1136 message := OpenStream{Epoch: 7, Connection: connection, Stream: 3, Purpose: ports.BrokerStreamAttachment, Admission: ports.BrokerAdmissionExact, Local: true, Target: testTarget(), Policy: testPolicy(), StartMode: ports.BrokerDaemonStartIfNeeded}
1137 raw := mustEncodeClient(t, message)
1138 envelope := &wire.BrokerClientEnvelope{}
1139 require.NoError(t, proto.Unmarshal(raw, envelope))
1140 envelope.GetOpenStream().Registration = registrationToWire(testRegistration())
1141 raw, err := proto.Marshal(envelope)
1142 require.NoError(t, err)
1143 _, err = DecodeClient(raw, testEnvelopeCeiling, testChunkCeiling)
1144 require.ErrorIs(t, err, ErrInvalidMessage)
1145 })
1146 t.Run("bad shutdown reason refused", func(t *testing.T) {
1147 _, err := EncodeServer(Shutdown{Epoch: 7, Connection: connection, Reason: ShutdownReason(9)}, testEnvelopeCeiling, testChunkCeiling)
1148 require.ErrorIs(t, err, ErrInvalidMessage)
1149 })
1150 t.Run("bad progress phase refused", func(t *testing.T) {
1151 _, err := EncodeServer(Progress{Epoch: 7, Connection: connection, Stream: 3, Phase: BrokerProgressPhase(9)}, testEnvelopeCeiling, testChunkCeiling)
1152 require.ErrorIs(t, err, ErrInvalidMessage)
1153 })
1154 t.Run("nil messages refused", func(t *testing.T) {
1155 var clientNil *Register
1156 _, err := EncodeClient(clientNil, testEnvelopeCeiling, testChunkCeiling)
1157 require.ErrorIs(t, err, ErrInvalidMessage)
1158 var serverNil *Registered
1159 _, err = EncodeServer(serverNil, testEnvelopeCeiling, testChunkCeiling)
1160 require.ErrorIs(t, err, ErrInvalidMessage)
1161 })
1162}
1163
1164// TestBrokerTruncatedAndTrailing proves truncated prefixes and trailing
1165// garbage are rejected on both directions.
1166func TestBrokerTruncatedAndTrailing(t *testing.T) {
1167 raw := mustEncodeClient(t, Subscribe{Epoch: 7, Connection: testConnectionID(1), Generation: 3})
1168 for size := range len(raw) {
1169 _, err := DecodeClient(raw[:size], testEnvelopeCeiling, testChunkCeiling)
1170 require.Error(t, err, "client prefix[:%d] accepted", size)
1171 }
1172 require.Error(t, func() error {
1173 _, err := DecodeClient(append(append([]byte(nil), raw...), 0xFF), testEnvelopeCeiling, testChunkCeiling)
1174 return err
1175 }())
1176
1177 serverRaw := mustEncodeServer(t, Registered{Epoch: 7, Connection: testConnectionID(1)})
1178 for size := range len(serverRaw) {
1179 _, err := DecodeServer(serverRaw[:size], testEnvelopeCeiling, testChunkCeiling)
1180 require.Error(t, err, "server prefix[:%d] accepted", size)
1181 }
1182 concatenated := append(append([]byte(nil), serverRaw...), serverRaw...)
1183 _, err := DecodeServer(concatenated, testEnvelopeCeiling, testChunkCeiling)
1184 require.Error(t, err)
1185}
1186
1187// TestBrokerWrongDirection proves wrong-direction decode is rejected:
1188// client bytes never decode as server output and vice versa.
1189func TestBrokerWrongDirection(t *testing.T) {
1190 clientRaw := mustEncodeClient(t, Register{})
1191 _, err := DecodeServer(clientRaw, testEnvelopeCeiling, testChunkCeiling)
1192 require.Error(t, err)
1193
1194 serverRaw := mustEncodeServer(t, Registered{Epoch: 7, Connection: testConnectionID(1)})
1195 _, err = DecodeClient(serverRaw, testEnvelopeCeiling, testChunkCeiling)
1196 require.Error(t, err)
1197
1198 // Go-typed misuse fails with ErrWrongDirection at encode time: unknown
1199 // concrete types presented to either directional codec fail closed.
1200 _, err = encodeClientEnvelope(unknownClientMessage{}, testChunkCeiling)
1201 require.ErrorIs(t, err, ErrWrongDirection)
1202 _, err = encodeServerEnvelope(unknownServerMessage{}, testChunkCeiling)
1203 require.ErrorIs(t, err, ErrWrongDirection)
1204
1205 // Nil-payload envelopes decode as wrong-direction.
1206 _, err = decodeClientEnvelope(&wire.BrokerClientEnvelope{}, testChunkCeiling)
1207 require.ErrorIs(t, err, ErrWrongDirection)
1208 _, err = decodeServerEnvelope(&wire.BrokerServerEnvelope{}, testChunkCeiling)
1209 require.ErrorIs(t, err, ErrWrongDirection)
1210}
1211
1212// TestBrokerCorruptInputSmoke feeds representative corruptions through both
1213// decoders as a smoke test: bit flips, truncation, unknown fields, and
1214// direction swaps must never panic and must never be accepted silently.
1215func TestBrokerCorruptInputSmoke(t *testing.T) {
1216 seeds := [][]byte{
1217 mustEncodeClient(t, Register{}),
1218 mustEncodeClient(t, AddHost{Epoch: 7, Connection: testConnectionID(2), Operation: testOperationID(0x21), Endpoint: "dev@host:22", Policy: testPolicy()}),
1219 mustEncodeClient(t, RemoveHost{Epoch: 7, Connection: testConnectionID(2), Operation: testOperationID(0x22), Registration: testRegistration()}),
1220 mustEncodeClient(t, UpdateHostPolicy{Epoch: 7, Connection: testConnectionID(2), Operation: testOperationID(0x23), Registration: testRegistration(), Policy: testPolicy()}),
1221 mustEncodeClient(t, OpenStream{Epoch: 7, Connection: testConnectionID(2), Stream: 3, Purpose: ports.BrokerStreamControl, Local: true, Policy: testPolicy(), StartMode: ports.BrokerDaemonStartIfNeeded}),
1222 mustEncodeServer(t, SnapshotPart{Epoch: 7, Connection: testConnectionID(2), Generation: 1, Revision: 1, Index: 0, Part: SnapshotEnd{}}),
1223 mustEncodeServer(t, OperationResult{Epoch: 7, Connection: testConnectionID(2), Operation: testOperationID(0x24), Outcome: ports.BrokerOutcomeFailed, Error: testErrorDetail(), HasError: true}),
1224 mustEncodeServer(t, Shutdown{Epoch: 7, Connection: testConnectionID(2), Reason: ShutdownIdleExit}),
1225 }
1226 for _, seed := range seeds {
1227 mutated := append([]byte(nil), seed...)
1228 if len(mutated) > 2 {
1229 mutated[len(mutated)-1] ^= 0xFF
1230 mutated[len(mutated)-2] ^= 0x01
1231 }
1232 _, _ = DecodeClient(mutated, testEnvelopeCeiling, testChunkCeiling)
1233 _, _ = DecodeServer(mutated, testEnvelopeCeiling, testChunkCeiling)
1234 unknown := protowire.AppendTag(append([]byte(nil), seed...), 199, protowire.VarintType)
1235 unknown = protowire.AppendVarint(unknown, 1)
1236 _, _ = DecodeClient(unknown, testEnvelopeCeiling, testChunkCeiling)
1237 _, _ = DecodeServer(unknown, testEnvelopeCeiling, testChunkCeiling)
1238 }
1239 require.NotEmpty(t, seeds)
1240}
1241
1242// fuzzClientSeeds and fuzzServerSeeds are the shared strict-scan seed corpus:
1243// valid brokered messages plus vectors that exercise ScanEnvelope rejection
1244// (unknown tag, wrong wire type, truncation, empty, and trailing bytes).
1245func fuzzClientSeeds(t testing.TB) [][]byte {
1246 t.Helper()
1247 seeds := [][]byte{}
1248 for _, message := range []ClientMessage{
1249 Register{},
1250 Subscribe{Epoch: 7, Connection: testConnectionID(0x11), Generation: 1},
1251 AddHost{Epoch: 7, Connection: testConnectionID(0x11), Operation: testOperationID(0x11), Endpoint: "dev@host:22", Policy: testPolicy()},
1252 RemoveHost{Epoch: 7, Connection: testConnectionID(0x11), Operation: testOperationID(0x12), Registration: testRegistration()},
1253 UpdateHostPolicy{Epoch: 7, Connection: testConnectionID(0x11), Operation: testOperationID(0x13), Registration: testRegistration(), Policy: testPolicy()},
1254 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},
1255 ClientStreamData{Epoch: 7, Connection: testConnectionID(0x11), Stream: 3, Data: []byte("frame")},
1256 CloseStream{Epoch: 7, Connection: testConnectionID(0x11), Stream: 3},
1257 } {
1258 raw, err := EncodeClient(message, testEnvelopeCeiling, testChunkCeiling)
1259 require.NoError(t, err)
1260 seeds = append(seeds, raw)
1261 }
1262 return appendStrictScanSeeds(seeds)
1263}
1264
1265func fuzzServerSeeds(t testing.TB) [][]byte {
1266 t.Helper()
1267 seeds := [][]byte{}
1268 for _, message := range []ServerMessage{
1269 Registered{Epoch: 7, Connection: testConnectionID(0x11)},
1270 SnapshotPart{Epoch: 7, Connection: testConnectionID(0x11), Generation: 2, Revision: 5, Index: 0, Part: SnapshotEnd{}},
1271 SnapshotPart{Epoch: 7, Connection: testConnectionID(0x11), Generation: 2, Revision: 5, Index: 2, Part: SnapshotSessionPart{HostIndex: 0, SessionIndex: 0, Session: testCatalogSession()}},
1272 OperationResult{Epoch: 7, Connection: testConnectionID(0x11), Operation: testOperationID(0x14), Outcome: ports.BrokerOutcomeOK, Registration: testRegistration()},
1273 BrokerErrorMessage{Epoch: 7, Connection: testConnectionID(0x11), Error: testErrorDetail()},
1274 Shutdown{Epoch: 7, Connection: testConnectionID(0x11), Reason: ShutdownTerminating},
1275 } {
1276 raw, err := EncodeServer(message, testEnvelopeCeiling, testChunkCeiling)
1277 require.NoError(t, err)
1278 seeds = append(seeds, raw)
1279 }
1280 return appendStrictScanSeeds(seeds)
1281}
1282
1283// appendStrictScanSeeds adds raw vectors that must be refused by the strict
1284// scanner rather than reaching generated unmarshal.
1285func appendStrictScanSeeds(seeds [][]byte) [][]byte {
1286 unknown := protowire.AppendTag(nil, 199, protowire.VarintType)
1287 unknown = protowire.AppendVarint(unknown, 1)
1288 wrongType := protowire.AppendTag(nil, 101, protowire.VarintType)
1289 wrongType = protowire.AppendVarint(wrongType, 1)
1290 return append(seeds, unknown, wrongType, []byte{0xFF}, []byte{0x80}, nil)
1291}
1292
1293// FuzzDecodeClient proves DecodeClient never panics on arbitrary input.
1294func FuzzDecodeClient(f *testing.F) {
1295 for _, seed := range fuzzClientSeeds(f) {
1296 f.Add(seed)
1297 }
1298 f.Fuzz(func(t *testing.T, payload []byte) {
1299 _, _ = DecodeClient(payload, testEnvelopeCeiling, testChunkCeiling)
1300 })
1301}
1302
1303// FuzzDecodeServer proves DecodeServer never panics on arbitrary input.
1304func FuzzDecodeServer(f *testing.F) {
1305 for _, seed := range fuzzServerSeeds(f) {
1306 f.Add(seed)
1307 }
1308 f.Fuzz(func(t *testing.T, payload []byte) {
1309 _, _ = DecodeServer(payload, testEnvelopeCeiling, testChunkCeiling)
1310 })
1311}
1312
1313func TestBuildRegistrationCodec(t *testing.T) {
1314 for _, tc := range []struct {
1315 name string
1316 build string
1317 retiring bool
1318 }{
1319 {"empty", "", false}, {"matching", "release+abc", false}, {"retiring", "release+def", true},
1320 } {
1321 t.Run(tc.name, func(t *testing.T) {
1322 request := Register{Build: tc.build}
1323 raw := mustEncodeClient(t, request)
1324 decoded, err := DecodeClient(raw, testEnvelopeCeiling, testChunkCeiling)
1325 require.NoError(t, err)
1326 require.Equal(t, request, decoded)
1327 want := []byte{0xaa, 0x06, 0x00}
1328 if tc.build != "" {
1329 want = append([]byte{0xaa, 0x06, byte(len(tc.build) + 2), 0x0a, byte(len(tc.build))}, []byte(tc.build)...)
1330 }
1331 require.Equal(t, want, raw)
1332 response := Registered{Epoch: 7, Connection: testConnectionID(1), Build: tc.build, Retiring: tc.retiring}
1333 raw = mustEncodeServer(t, response)
1334 decodedServer, err := DecodeServer(raw, testEnvelopeCeiling, testChunkCeiling)
1335 require.NoError(t, err)
1336 require.Equal(t, response, decodedServer)
1337 envelope := &wire.BrokerServerEnvelope{}
1338 require.NoError(t, proto.Unmarshal(raw, envelope))
1339 require.Equal(t, tc.build, envelope.GetRegistered().GetBuild())
1340 require.Equal(t, tc.retiring, envelope.GetRegistered().GetRetiring())
1341 })
1342 }
1343 for _, tc := range []struct {
1344 name string
1345 server bool
1346 }{{"client", false}, {"server", true}} {
1347 t.Run("oversize-"+tc.name, func(t *testing.T) {
1348 build := strings.Repeat("x", 129)
1349 if tc.server {
1350 _, err := EncodeServer(Registered{Epoch: 7, Connection: testConnectionID(1), Build: build}, testEnvelopeCeiling, testChunkCeiling)
1351 require.ErrorIs(t, err, ErrInvalidMessage)
1352 raw, err := proto.Marshal(&wire.BrokerServerEnvelope{Payload: &wire.BrokerServerEnvelope_Registered{Registered: &wire.Registered{Scope: scopeToWire(7, testConnectionID(1)), Build: build}}})
1353 require.NoError(t, err)
1354 _, err = DecodeServer(raw, testEnvelopeCeiling, testChunkCeiling)
1355 require.ErrorIs(t, err, ErrInvalidMessage)
1356 } else {
1357 _, err := EncodeClient(Register{Build: build}, testEnvelopeCeiling, testChunkCeiling)
1358 require.ErrorIs(t, err, ErrInvalidMessage)
1359 raw, err := proto.Marshal(&wire.BrokerClientEnvelope{Payload: &wire.BrokerClientEnvelope_Register{Register: &wire.Register{Build: build}}})
1360 require.NoError(t, err)
1361 _, err = DecodeClient(raw, testEnvelopeCeiling, testChunkCeiling)
1362 require.ErrorIs(t, err, ErrInvalidMessage)
1363 }
1364 })
1365 }
1366}