1package daemonmux
2
3import (
4 "bytes"
5 "strings"
6 "testing"
7
8 "google.golang.org/protobuf/encoding/protowire"
9 "google.golang.org/protobuf/proto"
10
11 "github.com/bnema/vev/internal/domain"
12 "github.com/bnema/vev/internal/ports"
13 "github.com/bnema/vev/internal/protocol"
14 "github.com/bnema/vev/internal/protocol/wire"
15 "github.com/stretchr/testify/require"
16)
17
18func testConnectionID(seed byte) ports.BrokerConnectionID {
19 var id ports.BrokerConnectionID
20 for i := range id {
21 id[i] = seed + byte(i)
22 }
23 return id
24}
25
26func testRegistration() domain.RemoteRegistration {
27 return domain.RemoteRegistration{
28 Endpoint: "dev@host:22",
29 Incarnation: [16]byte{1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16},
30 Generation: 9,
31 }
32}
33
34func testPolicy() ports.BrokerPolicy {
35 return ports.BrokerPolicy{
36 ProtocolVersion: protocol.Version,
37 CatalogSchemaVersion: 3,
38 EnvironmentPolicy: protocol.EnvironmentPolicyDaemonOwned,
39 Transport: "quic",
40 Trust: "pinned",
41 Launch: "agent",
42 Isolation: "user",
43 }
44}
45
46// localAcceptance provisions one local carriage entry under policy, matching
47// the local Open requests the shared fixtures send.
48func localAcceptance(policy ports.BrokerPolicy) ServerPolicyAdmission {
49 return ServerPolicyAdmission{Policy: policy, Origin: ports.SessionOriginLocal}
50}
51
52// remoteAcceptance provisions one remote carriage entry under policy, matching
53// a remote Open carrying a validated registration.
54func remoteAcceptance(policy ports.BrokerPolicy) ServerPolicyAdmission {
55 return ServerPolicyAdmission{Policy: policy, Origin: ports.SessionOriginRemote}
56}
57
58func testTarget() protocol.ExactSessionTarget {
59 return protocol.ExactSessionTarget{
60 LifecycleID: domain.SessionLifecycleID{1, 2, 3},
61 SessionName: "work",
62 }
63}
64
65func testRef(physical PhysicalStreamID) StreamRef {
66 return StreamRef{
67 Physical: physical,
68 Epoch: 7,
69 Connection: testConnectionID(0x11),
70 Client: 3,
71 }
72}
73
74func testErrorDetail() ErrorDetail {
75 return ErrorDetail{Code: ports.BrokerErrorUnavailable, Text: "dial refused", AdmissionCode: 1, FailureKind: domain.RemoteFailureTransport}
76}
77
78// unknownClientMessage and unknownServerMessage prove the directional codecs
79// fail closed on concrete types outside their closed union.
80type unknownClientMessage struct{}
81
82func (unknownClientMessage) muxClientMessage() {}
83
84type unknownServerMessage struct{}
85
86func (unknownServerMessage) muxServerMessage() {}
87
88const (
89 testEnvelopeCeiling = wire.AbsoluteEnvelopeLimit
90 testChunkCeiling = MaxMuxChunkBytes
91)
92
93func mustEncodeClient(t *testing.T, message ClientMessage) []byte {
94 t.Helper()
95 raw, err := EncodeClient(message, testEnvelopeCeiling, testChunkCeiling)
96 require.NoError(t, err)
97 return raw
98}
99
100func mustEncodeServer(t *testing.T, message ServerMessage) []byte {
101 t.Helper()
102 raw, err := EncodeServer(message, testEnvelopeCeiling, testChunkCeiling)
103 require.NoError(t, err)
104 return raw
105}
106
107func testOpen() Open {
108 return Open{
109 Ref: testRef(5), Purpose: ports.BrokerStreamAttachment, Admission: ports.BrokerAdmissionExact, Local: false,
110 Endpoint: "dev@host:22", Registration: testRegistration(), Target: testTarget(),
111 Env: []string{"TERM=xterm-256color"}, Policy: testPolicy(),
112 StartMode: ports.BrokerDaemonStartIfNeeded,
113 }
114}
115
116// TestMuxClientVariants inventories every client variant tag and proves each
117// one round-trips through encode and decode.
118func TestMuxClientVariants(t *testing.T) {
119 messages := []struct {
120 name string
121 message ClientMessage
122 tag protowire.Number
123 }{
124 {"open", testOpen(), 301},
125 {"data", Data{Physical: 5, Data: []byte("frame")}, 302},
126 {"close", Close{Physical: 5}, 303},
127 {"reset", Reset{Physical: 5, Error: testErrorDetail(), HasError: true}, 304},
128 {"reset_orderly", Reset{Physical: 5}, 304},
129 {"window_update", WindowUpdate{Physical: 5, Credit: 4096}, 305},
130 }
131 for _, tc := range messages {
132 t.Run(tc.name, func(t *testing.T) {
133 raw := mustEncodeClient(t, tc.message)
134 number, wireType, consumed := protowire.ConsumeTag(raw)
135 require.Greater(t, consumed, 0)
136 require.Equal(t, tc.tag, number)
137 require.Equal(t, protowire.BytesType, wireType)
138 decoded, err := DecodeClient(raw, testEnvelopeCeiling, testChunkCeiling)
139 require.NoError(t, err)
140 require.Equal(t, tc.message, decoded)
141 })
142 }
143}
144
145// TestMuxServerVariants inventories every server variant tag and proves each
146// one round-trips through encode and decode.
147func TestMuxServerVariants(t *testing.T) {
148 messages := []struct {
149 name string
150 message ServerMessage
151 tag protowire.Number
152 }{
153 {"opened", Opened{Ref: testRef(5)}, 401},
154 {"refused", Refused{Ref: testRef(5), Error: testErrorDetail()}, 402},
155 {"data", Data{Physical: 5, Data: []byte("frame")}, 403},
156 {"close", Close{Physical: 5}, 404},
157 {"reset", Reset{Physical: 5, Error: testErrorDetail(), HasError: true}, 405},
158 {"reset_orderly", Reset{Physical: 5}, 405},
159 {"window_update", WindowUpdate{Physical: 5, Credit: 4096}, 406},
160 }
161 for _, tc := range messages {
162 t.Run(tc.name, func(t *testing.T) {
163 raw := mustEncodeServer(t, tc.message)
164 number, wireType, consumed := protowire.ConsumeTag(raw)
165 require.Greater(t, consumed, 0)
166 require.Equal(t, tc.tag, number)
167 require.Equal(t, protowire.BytesType, wireType)
168 decoded, err := DecodeServer(raw, testEnvelopeCeiling, testChunkCeiling)
169 require.NoError(t, err)
170 require.Equal(t, tc.message, decoded)
171 })
172 }
173}
174
175// TestMuxSharedPayloadDirections proves the shared Data/Close/Reset payloads
176// convert into both directions: the same value produces the direction's own
177// tag and decodes back identically on either side.
178func TestMuxSharedPayloadDirections(t *testing.T) {
179 pairs := []struct {
180 name string
181 client ClientMessage
182 server ServerMessage
183 clientTag protowire.Number
184 serverTag protowire.Number
185 expectedSame any
186 }{
187 {"data", Data{Physical: 5, Data: []byte("frame")}, Data{Physical: 5, Data: []byte("frame")}, 302, 403, Data{Physical: 5, Data: []byte("frame")}},
188 {"close", Close{Physical: 5}, Close{Physical: 5}, 303, 404, Close{Physical: 5}},
189 {"reset", Reset{Physical: 5}, Reset{Physical: 5}, 304, 405, Reset{Physical: 5}},
190 }
191 for _, tc := range pairs {
192 t.Run(tc.name, func(t *testing.T) {
193 clientRaw := mustEncodeClient(t, tc.client)
194 clientTag, _, _ := protowire.ConsumeTag(clientRaw)
195 require.Equal(t, tc.clientTag, clientTag)
196 clientDecoded, err := DecodeClient(clientRaw, testEnvelopeCeiling, testChunkCeiling)
197 require.NoError(t, err)
198 require.Equal(t, tc.expectedSame, clientDecoded)
199
200 serverRaw := mustEncodeServer(t, tc.server)
201 serverTag, _, _ := protowire.ConsumeTag(serverRaw)
202 require.Equal(t, tc.serverTag, serverTag)
203 serverDecoded, err := DecodeServer(serverRaw, testEnvelopeCeiling, testChunkCeiling)
204 require.NoError(t, err)
205 require.Equal(t, tc.expectedSame, serverDecoded)
206 })
207 }
208}
209
210// TestMuxGoldenVectors pins literal byte encodings where the wire layout is
211// stable: tag, length, and one varint field.
212func TestMuxGoldenVectors(t *testing.T) {
213 vectors := []struct {
214 name string
215 message ClientMessage
216 want []byte
217 }{
218 {"close", Close{Physical: 1}, []byte{0xFA, 0x12, 0x02, 0x08, 0x01}},
219 {"data", Data{Physical: 1, Data: []byte("hi")}, []byte{0xF2, 0x12, 0x06, 0x08, 0x01, 0x12, 0x02, 0x68, 0x69}},
220 {"reset_orderly", Reset{Physical: 1}, []byte{0x82, 0x13, 0x02, 0x08, 0x01}},
221 }
222 for _, tc := range vectors {
223 t.Run(tc.name, func(t *testing.T) {
224 raw := mustEncodeClient(t, tc.message)
225 require.Equal(t, tc.want, raw)
226 // Re-encode is deterministic: the same message marshals
227 // identically, so a receiver can rely on a stable byte layout.
228 again := mustEncodeClient(t, tc.message)
229 require.Equal(t, tc.want, again)
230 decoded, err := DecodeClient(tc.want, testEnvelopeCeiling, testChunkCeiling)
231 require.NoError(t, err)
232 require.Equal(t, tc.message, decoded)
233 })
234 }
235
236 // The server direction has its own tags: the same value never produces
237 // the client bytes.
238 serverRaw := mustEncodeServer(t, Close{Physical: 1})
239 require.Equal(t, []byte{0xA2, 0x19, 0x02, 0x08, 0x01}, serverRaw)
240
241 // A populated ref is stable too: MuxOpened carries the four-field ref in
242 // numeric order.
243 ref := StreamRef{Physical: 1, Epoch: 1, Connection: testConnectionID(1), Client: 1}
244 opened := mustEncodeServer(t, Opened{Ref: ref})
245 back, err := DecodeServer(opened, testEnvelopeCeiling, testChunkCeiling)
246 require.NoError(t, err)
247 require.Equal(t, Opened{Ref: ref}, back)
248}
249
250// TestMuxBounds proves stateless bound refusals at max-1, max, and max+1.
251func TestMuxBounds(t *testing.T) {
252 t.Run("envelope ceiling", func(t *testing.T) {
253 raw := mustEncodeClient(t, Close{Physical: 1})
254 _, err := DecodeClient(raw, testEnvelopeCeiling, testChunkCeiling)
255 require.NoError(t, err)
256 // One byte below the exact serialized length refuses the envelope.
257 _, err = DecodeClient(raw, uint64(len(raw)-1), testChunkCeiling)
258 require.ErrorIs(t, err, wire.ErrScanLength)
259 // At the ceiling the envelope decodes; one byte above the ceiling is
260 // refused before scanning.
261 _, err = DecodeClient(raw, uint64(len(raw)), testChunkCeiling)
262 require.NoError(t, err)
263 _, err = DecodeClient(make([]byte, MaxMuxEnvelopeBytes+1), MaxMuxEnvelopeBytes, testChunkCeiling)
264 require.ErrorIs(t, err, wire.ErrScanLength)
265 // Encoding above the local ceiling fails closed.
266 _, err = EncodeClient(Close{Physical: 1}, uint64(len(raw)-1), testChunkCeiling)
267 require.ErrorIs(t, err, wire.ErrScanLength)
268 _, err = EncodeClient(Close{Physical: 1}, uint64(len(raw)), testChunkCeiling)
269 require.NoError(t, err)
270 })
271 t.Run("application envelope ceiling is 1 MiB", func(t *testing.T) {
272 require.Equal(t, uint64(1<<20), MaxMuxApplicationBytes)
273 require.Equal(t, uint64(1<<20), MinMuxEnvelopeBytes)
274 // The fixed application ceiling applies at max and refuses at max+1,
275 // whatever the negotiated receive ceiling advertises.
276 require.NoError(t, checkMuxEnvelopeCeiling(make([]byte, MaxMuxApplicationBytes), testEnvelopeCeiling))
277 require.ErrorIs(t, checkMuxEnvelopeCeiling(make([]byte, MaxMuxApplicationBytes+1), testEnvelopeCeiling), wire.ErrScanLength)
278 require.ErrorIs(t, checkMuxEnvelopeCeiling(make([]byte, MaxMuxApplicationBytes+1), MaxMuxEnvelopeBytes), wire.ErrScanLength)
279 // A hostile open envelope above the application ceiling is refused
280 // before scan even though the negotiated ceiling admits 16 MiB.
281 env := make([]string, 400)
282 for i := range env {
283 env[i] = "K=" + strings.Repeat("x", ports.BrokerMaxEnvEntryBytes-2)
284 }
285 hostile := &wire.MuxClientEnvelope{Payload: &wire.MuxClientEnvelope_Open{Open: &wire.MuxOpen{
286 Ref: &wire.MuxStreamRef{PhysicalStreamId: 1, BrokerEpoch: 1, ConnectionId: make([]byte, 16), ClientStreamId: 1},
287 Purpose: 1, Local: true, Env: env,
288 }}}
289 payload, err := proto.Marshal(hostile)
290 require.NoError(t, err)
291 require.Greater(t, len(payload), int(MaxMuxApplicationBytes))
292 _, err = DecodeClient(payload, testEnvelopeCeiling, testChunkCeiling)
293 require.ErrorIs(t, err, wire.ErrScanLength)
294 })
295 t.Run("chunk window 1..64KiB", func(t *testing.T) {
296 require.Equal(t, uint64(64<<10), MaxMuxChunkBytes)
297 atMax := bytes.Repeat([]byte{0x66}, int(MaxMuxChunkBytes))
298 belowMax := bytes.Repeat([]byte{0x66}, int(MaxMuxChunkBytes)-1)
299 aboveMax := bytes.Repeat([]byte{0x66}, int(MaxMuxChunkBytes)+1)
300 for _, data := range [][]byte{belowMax, atMax} {
301 clientRaw, err := EncodeClient(Data{Physical: 5, Data: data}, testEnvelopeCeiling, testChunkCeiling)
302 require.NoError(t, err)
303 clientDecoded, err := DecodeClient(clientRaw, testEnvelopeCeiling, testChunkCeiling)
304 require.NoError(t, err)
305 require.Equal(t, data, clientDecoded.(Data).Data)
306
307 serverRaw, err := EncodeServer(Data{Physical: 5, Data: data}, testEnvelopeCeiling, testChunkCeiling)
308 require.NoError(t, err)
309 serverDecoded, err := DecodeServer(serverRaw, testEnvelopeCeiling, testChunkCeiling)
310 require.NoError(t, err)
311 require.Equal(t, data, serverDecoded.(Data).Data)
312 }
313 _, err := EncodeClient(Data{Physical: 5, Data: aboveMax}, testEnvelopeCeiling, testChunkCeiling)
314 require.ErrorIs(t, err, ErrTooLarge)
315 _, err = EncodeServer(Data{Physical: 5, Data: aboveMax}, testEnvelopeCeiling, testChunkCeiling)
316 require.ErrorIs(t, err, ErrTooLarge)
317 // A negotiated chunk below the absolute maximum also refuses.
318 raw := mustEncodeClient(t, Data{Physical: 5, Data: []byte("12345678")})
319 _, err = DecodeClient(raw, testEnvelopeCeiling, 7)
320 require.ErrorIs(t, err, ErrTooLarge)
321 _, err = DecodeClient(raw, testEnvelopeCeiling, 8)
322 require.NoError(t, err)
323 // Empty chunk is invalid, not too large.
324 _, err = EncodeClient(Data{Physical: 5}, testEnvelopeCeiling, testChunkCeiling)
325 require.ErrorIs(t, err, ErrInvalidMessage)
326 })
327 t.Run("environment entries", func(t *testing.T) {
328 atMax := "K=" + strings.Repeat("x", ports.BrokerMaxEnvEntryBytes-2)
329 require.Len(t, atMax, ports.BrokerMaxEnvEntryBytes)
330 aboveMax := atMax + "x"
331 message := testOpen()
332 message.Env = []string{atMax}
333 _, err := EncodeClient(message, testEnvelopeCeiling, testChunkCeiling)
334 require.NoError(t, err)
335 message.Env = []string{aboveMax}
336 _, err = EncodeClient(message, testEnvelopeCeiling, testChunkCeiling)
337 require.ErrorIs(t, err, ErrTooLarge)
338
339 many := make([]string, ports.BrokerMaxEnvEntries)
340 for i := range many {
341 many[i] = "A=B"
342 }
343 message = testOpen()
344 message.Env = many
345 _, err = EncodeClient(message, testEnvelopeCeiling, testChunkCeiling)
346 require.NoError(t, err)
347 message.Env = append(append([]string(nil), many...), "C=D")
348 _, err = EncodeClient(message, testEnvelopeCeiling, testChunkCeiling)
349 require.ErrorIs(t, err, ErrTooLarge)
350 })
351 t.Run("error text", func(t *testing.T) {
352 atMax := testErrorDetail()
353 atMax.Text = strings.Repeat("e", ports.BrokerMaxErrorBytes)
354 _, err := EncodeServer(Refused{Ref: testRef(5), Error: atMax}, testEnvelopeCeiling, testChunkCeiling)
355 require.NoError(t, err)
356 over := testErrorDetail()
357 over.Text = strings.Repeat("e", ports.BrokerMaxErrorBytes+1)
358 _, err = EncodeServer(Refused{Ref: testRef(5), Error: over}, testEnvelopeCeiling, testChunkCeiling)
359 require.ErrorIs(t, err, ErrTooLarge)
360 })
361}
362
363// TestMuxValidationNegatives is the table of semantic refusals: stream
364// identity, taxonomy, text, environment, policy, and error details.
365func TestMuxValidationNegatives(t *testing.T) {
366 t.Run("zero stream identities refused", func(t *testing.T) {
367 _, err := EncodeClient(Close{}, testEnvelopeCeiling, testChunkCeiling)
368 require.ErrorIs(t, err, ErrInvalidMessage)
369 _, err = EncodeClient(Data{Physical: 0, Data: []byte("x")}, testEnvelopeCeiling, testChunkCeiling)
370 require.ErrorIs(t, err, ErrInvalidMessage)
371 _, err = EncodeServer(Reset{Physical: 0}, testEnvelopeCeiling, testChunkCeiling)
372 require.ErrorIs(t, err, ErrInvalidMessage)
373
374 zeroEpoch := testRef(5)
375 zeroEpoch.Epoch = 0
376 _, err = EncodeServer(Opened{Ref: zeroEpoch}, testEnvelopeCeiling, testChunkCeiling)
377 require.ErrorIs(t, err, ErrInvalidMessage)
378
379 zeroConnection := testRef(5)
380 zeroConnection.Connection = ports.BrokerConnectionID{}
381 _, err = EncodeServer(Opened{Ref: zeroConnection}, testEnvelopeCeiling, testChunkCeiling)
382 require.ErrorIs(t, err, ErrInvalidMessage)
383
384 zeroClient := testRef(5)
385 zeroClient.Client = 0
386 _, err = EncodeServer(Opened{Ref: zeroClient}, testEnvelopeCeiling, testChunkCeiling)
387 require.ErrorIs(t, err, ErrInvalidMessage)
388
389 _, err = EncodeClient(Open{Ref: zeroClient, Purpose: ports.BrokerStreamControl, Local: true, Policy: testPolicy(), StartMode: ports.BrokerDaemonStartIfNeeded}, testEnvelopeCeiling, testChunkCeiling)
390 require.ErrorIs(t, err, ErrInvalidMessage)
391 })
392 t.Run("open taxonomy and shape", func(t *testing.T) {
393 badPurpose := testOpen()
394 badPurpose.Purpose = ports.BrokerStreamPurpose(9)
395 _, err := EncodeClient(badPurpose, testEnvelopeCeiling, testChunkCeiling)
396 require.ErrorIs(t, err, ErrInvalidMessage)
397
398 badEndpoint := testOpen()
399 badEndpoint.Endpoint = "not a host"
400 raw, err := EncodeClient(badEndpoint, testEnvelopeCeiling, testChunkCeiling)
401 require.NoError(t, err)
402 _, err = DecodeClient(raw, testEnvelopeCeiling, testChunkCeiling)
403 require.ErrorIs(t, err, ErrInvalidMessage)
404
405 local := testOpen()
406 local.Local = true
407 raw, err = EncodeClient(local, testEnvelopeCeiling, testChunkCeiling)
408 require.NoError(t, err)
409 _, err = DecodeClient(raw, testEnvelopeCeiling, testChunkCeiling)
410 require.ErrorIs(t, err, ErrInvalidMessage)
411
412 controlWithTarget := testOpen()
413 controlWithTarget.Purpose = ports.BrokerStreamControl
414 raw, err = EncodeClient(controlWithTarget, testEnvelopeCeiling, testChunkCeiling)
415 require.NoError(t, err)
416 envelope := &wire.MuxClientEnvelope{}
417 require.NoError(t, proto.Unmarshal(raw, envelope))
418 envelope.GetOpen().Target = exactTargetToWire(testOpen().Target)
419 raw, err = proto.Marshal(envelope)
420 require.NoError(t, err)
421 _, err = DecodeClient(raw, testEnvelopeCeiling, testChunkCeiling)
422 require.ErrorIs(t, err, ErrInvalidMessage)
423
424 controlWithEnv := testOpen()
425 controlWithEnv.Purpose = ports.BrokerStreamControl
426 controlWithEnv.Target = protocol.ExactSessionTarget{}
427 raw, err = EncodeClient(controlWithEnv, testEnvelopeCeiling, testChunkCeiling)
428 require.NoError(t, err)
429 _, err = DecodeClient(raw, testEnvelopeCeiling, testChunkCeiling)
430 require.ErrorIs(t, err, ErrInvalidMessage)
431
432 noEquals := testOpen()
433 noEquals.Env = []string{"NOEQUALS"}
434 _, err = EncodeClient(noEquals, testEnvelopeCeiling, testChunkCeiling)
435 require.ErrorIs(t, err, ErrInvalidMessage)
436
437 emptyPolicy := testOpen()
438 emptyPolicy.Policy = ports.BrokerPolicy{}
439 raw, err = EncodeClient(emptyPolicy, testEnvelopeCeiling, testChunkCeiling)
440 require.NoError(t, err)
441 _, err = DecodeClient(raw, testEnvelopeCeiling, testChunkCeiling)
442 require.ErrorIs(t, err, ErrInvalidMessage)
443
444 // A local control open with no attachment state is legal.
445 valid := Open{Ref: testRef(5), Purpose: ports.BrokerStreamControl, Local: true, Policy: testPolicy(), StartMode: ports.BrokerDaemonStartIfNeeded}
446 raw, err = EncodeClient(valid, testEnvelopeCeiling, testChunkCeiling)
447 require.NoError(t, err)
448 decoded, err := DecodeClient(raw, testEnvelopeCeiling, testChunkCeiling)
449 require.NoError(t, err)
450 require.Equal(t, valid, decoded)
451 })
452 t.Run("taxonomy switches refuse mutated wire values", func(t *testing.T) {
453 raw := mustEncodeClient(t, testOpen())
454 envelope := &wire.MuxClientEnvelope{}
455 require.NoError(t, wire.ScanEnvelope(envelope, raw))
456 require.NoError(t, proto.Unmarshal(raw, envelope))
457 envelope.GetOpen().Purpose = 9
458 mutated, err := proto.Marshal(envelope)
459 require.NoError(t, err)
460 _, err = DecodeClient(mutated, testEnvelopeCeiling, testChunkCeiling)
461 require.Error(t, err)
462
463 rawErr := mustEncodeServer(t, Refused{Ref: testRef(5), Error: testErrorDetail()})
464 server := &wire.MuxServerEnvelope{}
465 require.NoError(t, wire.ScanEnvelope(server, rawErr))
466 require.NoError(t, proto.Unmarshal(rawErr, server))
467 server.GetRefused().GetError().Code = 99
468 mutated, err = proto.Marshal(server)
469 require.NoError(t, err)
470 _, err = DecodeServer(mutated, testEnvelopeCeiling, testChunkCeiling)
471 require.Error(t, err)
472 })
473 t.Run("error detail is mandatory on refusal", func(t *testing.T) {
474 _, err := EncodeServer(Refused{Ref: testRef(5)}, testEnvelopeCeiling, testChunkCeiling)
475 require.ErrorIs(t, err, ErrInvalidMessage)
476 // The zero detail has no valid code, so it can never travel.
477 _, err = EncodeServer(Refused{Ref: testRef(5), Error: ErrorDetail{AdmissionCode: 1}}, testEnvelopeCeiling, testChunkCeiling)
478 require.ErrorIs(t, err, ErrInvalidMessage)
479 })
480 t.Run("reset never drops a declared detail", func(t *testing.T) {
481 for name, stray := range map[string]ErrorDetail{
482 "full detail": testErrorDetail(),
483 "code only": {Code: ports.BrokerErrorTimeout},
484 "text only": {Text: "late"},
485 "admission only": {AdmissionCode: 2},
486 "failure kind only": {FailureKind: domain.RemoteFailureTransport},
487 } {
488 t.Run(name, func(t *testing.T) {
489 _, err := EncodeClient(Reset{Physical: 5, Error: stray}, testEnvelopeCeiling, testChunkCeiling)
490 require.ErrorIs(t, err, ErrInvalidMessage)
491 _, err = EncodeServer(Reset{Physical: 5, Error: stray}, testEnvelopeCeiling, testChunkCeiling)
492 require.ErrorIs(t, err, ErrInvalidMessage)
493 })
494 }
495 // The zero detail is the canonical absent value and still encodes; a
496 // declared detail still round-trips exactly.
497 _, err := EncodeServer(Reset{Physical: 5}, testEnvelopeCeiling, testChunkCeiling)
498 require.NoError(t, err)
499 declared := Reset{Physical: 5, Error: testErrorDetail(), HasError: true}
500 decoded, err := DecodeServer(mustEncodeServer(t, declared), testEnvelopeCeiling, testChunkCeiling)
501 require.NoError(t, err)
502 require.Equal(t, declared, decoded)
503 })
504 t.Run("invalid error detail refused on every path", func(t *testing.T) {
505 badAdmission := testErrorDetail()
506 badAdmission.AdmissionCode = MaxAdmissionCode + 1
507 badKind := testErrorDetail()
508 badKind.FailureKind = MaxFailureKind + 1
509 badCode := testErrorDetail()
510 badCode.Code = 0
511 for name, detail := range map[string]ErrorDetail{
512 "admission over maximum": badAdmission,
513 "failure kind over bound": badKind,
514 "invalid error code": badCode,
515 } {
516 t.Run(name, func(t *testing.T) {
517 _, err := EncodeServer(Refused{Ref: testRef(5), Error: detail}, testEnvelopeCeiling, testChunkCeiling)
518 require.ErrorIs(t, err, ErrInvalidMessage)
519 _, err = EncodeServer(Reset{Physical: 5, Error: detail, HasError: true}, testEnvelopeCeiling, testChunkCeiling)
520 require.ErrorIs(t, err, ErrInvalidMessage)
521 _, err = EncodeClient(Reset{Physical: 5, Error: detail, HasError: true}, testEnvelopeCeiling, testChunkCeiling)
522 require.ErrorIs(t, err, ErrInvalidMessage)
523 })
524 }
525 // The closed taxonomy boundaries are accepted.
526 boundary := testErrorDetail()
527 boundary.AdmissionCode = MaxAdmissionCode
528 boundary.FailureKind = MaxFailureKind
529 _, err := EncodeServer(Refused{Ref: testRef(5), Error: boundary}, testEnvelopeCeiling, testChunkCeiling)
530 require.NoError(t, err)
531 })
532 t.Run("display control text and env NUL refused", func(t *testing.T) {
533 evil := testErrorDetail()
534 evil.Text = "timeout\u202erorrE"
535 _, err := EncodeServer(Refused{Ref: testRef(5), Error: evil}, testEnvelopeCeiling, testChunkCeiling)
536 require.ErrorIs(t, err, ErrInvalidMessage)
537 evil.Text = "bad\x01text"
538 _, err = EncodeServer(Refused{Ref: testRef(5), Error: evil}, testEnvelopeCeiling, testChunkCeiling)
539 require.ErrorIs(t, err, ErrInvalidMessage)
540
541 badEnv := testOpen()
542 badEnv.Env = []string{"K=bad\x00value"}
543 _, err = EncodeClient(badEnv, testEnvelopeCeiling, testChunkCeiling)
544 require.ErrorIs(t, err, ErrInvalidMessage)
545 })
546 t.Run("nil messages refused", func(t *testing.T) {
547 var openNil *Open
548 _, err := EncodeClient(openNil, testEnvelopeCeiling, testChunkCeiling)
549 require.ErrorIs(t, err, ErrInvalidMessage)
550 var dataNil *Data
551 _, err = EncodeServer(dataNil, testEnvelopeCeiling, testChunkCeiling)
552 require.ErrorIs(t, err, ErrInvalidMessage)
553 var refusedNil *Refused
554 _, err = EncodeServer(refusedNil, testEnvelopeCeiling, testChunkCeiling)
555 require.ErrorIs(t, err, ErrInvalidMessage)
556 })
557}
558
559// TestMuxTruncatedAndTrailing proves truncated prefixes and trailing garbage
560// are rejected on both directions.
561func TestMuxTruncatedAndTrailing(t *testing.T) {
562 raw := mustEncodeClient(t, testOpen())
563 for size := range len(raw) {
564 _, err := DecodeClient(raw[:size], testEnvelopeCeiling, testChunkCeiling)
565 require.Error(t, err, "client prefix[:%d] accepted", size)
566 }
567 _, err := DecodeClient(append(append([]byte(nil), raw...), 0xFF), testEnvelopeCeiling, testChunkCeiling)
568 require.Error(t, err)
569
570 serverRaw := mustEncodeServer(t, Opened{Ref: testRef(5)})
571 for size := range len(serverRaw) {
572 _, err := DecodeServer(serverRaw[:size], testEnvelopeCeiling, testChunkCeiling)
573 require.Error(t, err, "server prefix[:%d] accepted", size)
574 }
575 concatenated := append(append([]byte(nil), serverRaw...), serverRaw...)
576 _, err = DecodeServer(concatenated, testEnvelopeCeiling, testChunkCeiling)
577 require.Error(t, err)
578}
579
580// TestMuxWrongDirection proves wrong-direction decode is rejected: client
581// bytes never decode as daemon output and vice versa.
582func TestMuxWrongDirection(t *testing.T) {
583 clientRaw := mustEncodeClient(t, Close{Physical: 1})
584 _, err := DecodeServer(clientRaw, testEnvelopeCeiling, testChunkCeiling)
585 require.Error(t, err)
586
587 serverRaw := mustEncodeServer(t, Close{Physical: 1})
588 _, err = DecodeClient(serverRaw, testEnvelopeCeiling, testChunkCeiling)
589 require.Error(t, err)
590
591 // Go-typed misuse fails with ErrWrongDirection at encode time: unknown
592 // concrete types presented to either directional codec fail closed.
593 _, err = encodeClientEnvelope(unknownClientMessage{}, testChunkCeiling)
594 require.ErrorIs(t, err, ErrWrongDirection)
595 _, err = encodeServerEnvelope(unknownServerMessage{}, testChunkCeiling)
596 require.ErrorIs(t, err, ErrWrongDirection)
597
598 // Nil-payload envelopes decode as wrong-direction.
599 _, err = decodeClientEnvelope(&wire.MuxClientEnvelope{}, testChunkCeiling)
600 require.ErrorIs(t, err, ErrWrongDirection)
601 _, err = decodeServerEnvelope(&wire.MuxServerEnvelope{}, testChunkCeiling)
602 require.ErrorIs(t, err, ErrWrongDirection)
603}
604
605// TestMuxCorruptInputSmoke feeds representative corruptions through both
606// decoders as a smoke test: bit flips, truncation, unknown fields, and
607// direction swaps must never panic and must never be accepted silently.
608func TestMuxCorruptInputSmoke(t *testing.T) {
609 seeds := [][]byte{
610 mustEncodeClient(t, testOpen()),
611 mustEncodeClient(t, Data{Physical: 5, Data: []byte("frame")}),
612 mustEncodeClient(t, Close{Physical: 5}),
613 mustEncodeServer(t, Opened{Ref: testRef(5)}),
614 mustEncodeServer(t, Refused{Ref: testRef(5), Error: testErrorDetail()}),
615 mustEncodeServer(t, Reset{Physical: 5, Error: testErrorDetail(), HasError: true}),
616 }
617 for _, seed := range seeds {
618 mutated := append([]byte(nil), seed...)
619 if len(mutated) > 2 {
620 mutated[len(mutated)-1] ^= 0xFF
621 mutated[len(mutated)-2] ^= 0x01
622 }
623 _, _ = DecodeClient(mutated, testEnvelopeCeiling, testChunkCeiling)
624 _, _ = DecodeServer(mutated, testEnvelopeCeiling, testChunkCeiling)
625 unknown := protowire.AppendTag(append([]byte(nil), seed...), 399, protowire.VarintType)
626 unknown = protowire.AppendVarint(unknown, 1)
627 _, _ = DecodeClient(unknown, testEnvelopeCeiling, testChunkCeiling)
628 _, _ = DecodeServer(unknown, testEnvelopeCeiling, testChunkCeiling)
629 }
630 require.NotEmpty(t, seeds)
631}
632
633// appendStrictScanSeeds adds raw vectors that must be refused by the strict
634// scanner rather than reaching generated unmarshal.
635func appendStrictScanSeeds(seeds [][]byte) [][]byte {
636 unknown := protowire.AppendTag(nil, 399, protowire.VarintType)
637 unknown = protowire.AppendVarint(unknown, 1)
638 wrongType := protowire.AppendTag(nil, 301, protowire.VarintType)
639 wrongType = protowire.AppendVarint(wrongType, 1)
640 return append(seeds, unknown, wrongType, []byte{0xFF}, []byte{0x80}, nil)
641}
642
643func fuzzClientSeeds(t testing.TB) [][]byte {
644 t.Helper()
645 seeds := [][]byte{}
646 for _, message := range []ClientMessage{
647 testOpen(),
648 Data{Physical: 5, Data: []byte("frame")},
649 Close{Physical: 5},
650 Reset{Physical: 5, Error: testErrorDetail(), HasError: true},
651 } {
652 raw, err := EncodeClient(message, testEnvelopeCeiling, testChunkCeiling)
653 require.NoError(t, err)
654 seeds = append(seeds, raw)
655 }
656 return appendStrictScanSeeds(seeds)
657}
658
659func fuzzServerSeeds(t testing.TB) [][]byte {
660 t.Helper()
661 seeds := [][]byte{}
662 for _, message := range []ServerMessage{
663 Opened{Ref: testRef(5)},
664 Refused{Ref: testRef(5), Error: testErrorDetail()},
665 Data{Physical: 5, Data: []byte("frame")},
666 Close{Physical: 5},
667 Reset{Physical: 5, Error: testErrorDetail(), HasError: true},
668 } {
669 raw, err := EncodeServer(message, testEnvelopeCeiling, testChunkCeiling)
670 require.NoError(t, err)
671 seeds = append(seeds, raw)
672 }
673 return appendStrictScanSeeds(seeds)
674}
675
676// FuzzDecodeClient proves DecodeClient never panics on arbitrary input.
677func FuzzDecodeClient(f *testing.F) {
678 for _, seed := range fuzzClientSeeds(f) {
679 f.Add(seed)
680 }
681 f.Fuzz(func(t *testing.T, payload []byte) {
682 _, _ = DecodeClient(payload, testEnvelopeCeiling, testChunkCeiling)
683 })
684}
685
686// FuzzDecodeServer proves DecodeServer never panics on arbitrary input.
687func FuzzDecodeServer(f *testing.F) {
688 for _, seed := range fuzzServerSeeds(f) {
689 f.Add(seed)
690 }
691 f.Fuzz(func(t *testing.T, payload []byte) {
692 _, _ = DecodeServer(payload, testEnvelopeCeiling, testChunkCeiling)
693 })
694}
695
696// TestMuxAdmissionVariants proves the attachment-admission contract travels
697// the mux wire losslessly and rejects every incoherent combination at the
698// same boundary the ports contract rejects it.
699func TestMuxAdmissionVariants(t *testing.T) {
700 roundTrip := func(t *testing.T, open Open) Open {
701 t.Helper()
702 raw, err := EncodeClient(open, testEnvelopeCeiling, testChunkCeiling)
703 require.NoError(t, err)
704 decoded, err := DecodeClient(raw, testEnvelopeCeiling, testChunkCeiling)
705 require.NoError(t, err)
706 open, ok := decoded.(Open)
707 require.True(t, ok)
708 return open
709 }
710 t.Run("exact attach round-trips", func(t *testing.T) {
711 require.Equal(t, testOpen(), roundTrip(t, testOpen()))
712 })
713 t.Run("named creation round-trips", func(t *testing.T) {
714 named := testOpen()
715 named.Admission = ports.BrokerAdmissionCreateNamed
716 named.Name = "work"
717 named.Target = protocol.ExactSessionTarget{}
718 require.Equal(t, named, roundTrip(t, named))
719 })
720 t.Run("named attach round-trips", func(t *testing.T) {
721 named := testOpen()
722 named.Admission = ports.BrokerAdmissionAttachNamed
723 named.Name = "work"
724 named.Target = protocol.ExactSessionTarget{}
725 require.Equal(t, named, roundTrip(t, named))
726 })
727 t.Run("ephemeral creation round-trips", func(t *testing.T) {
728 ephemeral := testOpen()
729 ephemeral.Admission = ports.BrokerAdmissionCreateEphemeral
730 ephemeral.Target = protocol.ExactSessionTarget{}
731 require.Equal(t, ephemeral, roundTrip(t, ephemeral))
732 })
733 t.Run("control never carries admission", func(t *testing.T) {
734 control := Open{Ref: testRef(6), Purpose: ports.BrokerStreamControl, Local: true, Policy: testPolicy(), StartMode: ports.BrokerDaemonStartIfNeeded}
735 require.Equal(t, control, roundTrip(t, control))
736 mutated := mustEncodeClient(t, control)
737 envelope := &wire.MuxClientEnvelope{}
738 require.NoError(t, proto.Unmarshal(mutated, envelope))
739 envelope.GetOpen().Admission = 2
740 envelope.GetOpen().Name = "work"
741 mutated, err := proto.Marshal(envelope)
742 require.NoError(t, err)
743 _, err = DecodeClient(mutated, testEnvelopeCeiling, testChunkCeiling)
744 require.ErrorIs(t, err, ErrInvalidMessage)
745 })
746 t.Run("exact with a creation name is refused", func(t *testing.T) {
747 open := testOpen()
748 open.Name = "work"
749 raw, err := EncodeClient(open, testEnvelopeCeiling, testChunkCeiling)
750 require.NoError(t, err)
751 _, err = DecodeClient(raw, testEnvelopeCeiling, testChunkCeiling)
752 require.ErrorIs(t, err, ErrInvalidMessage)
753 })
754 t.Run("named creation with a target is refused", func(t *testing.T) {
755 named := testOpen()
756 named.Admission = ports.BrokerAdmissionCreateNamed
757 named.Name = "work"
758 raw, err := EncodeClient(named, testEnvelopeCeiling, testChunkCeiling)
759 require.NoError(t, err)
760 envelope := &wire.MuxClientEnvelope{}
761 require.NoError(t, proto.Unmarshal(raw, envelope))
762 envelope.GetOpen().Target = exactTargetToWire(testOpen().Target)
763 raw, err = proto.Marshal(envelope)
764 require.NoError(t, err)
765 _, err = DecodeClient(raw, testEnvelopeCeiling, testChunkCeiling)
766 require.ErrorIs(t, err, ErrInvalidMessage)
767 })
768 t.Run("unknown admission code is refused", func(t *testing.T) {
769 raw := mustEncodeClient(t, testOpen())
770 envelope := &wire.MuxClientEnvelope{}
771 require.NoError(t, proto.Unmarshal(raw, envelope))
772 envelope.GetOpen().Admission = 9
773 mutated, err := proto.Marshal(envelope)
774 require.NoError(t, err)
775 _, err = DecodeClient(mutated, testEnvelopeCeiling, testChunkCeiling)
776 require.ErrorIs(t, err, ErrInvalidMessage)
777 })
778}
779
780// TestMuxStartModeRoundTrips proves the daemon-start authorization travels the
781// mux wire losslessly for every stream purpose, that the daemon-side decoder
782// never widens an existing-only authorization, and that the zero and unknown
783// wire codes are refused rather than defaulted.
784func TestMuxStartModeRoundTrips(t *testing.T) {
785 for _, tc := range []struct {
786 name string
787 mode ports.BrokerDaemonStartMode
788 purpose ports.BrokerStreamPurpose
789 admission ports.BrokerStreamAdmission
790 }{
791 {"existing only control", ports.BrokerDaemonExistingOnly, ports.BrokerStreamControl, 0},
792 {"start if needed control", ports.BrokerDaemonStartIfNeeded, ports.BrokerStreamControl, 0},
793 {"existing only observation", ports.BrokerDaemonExistingOnly, ports.BrokerStreamObservation, 0},
794 {"start if needed exact attach", ports.BrokerDaemonStartIfNeeded, ports.BrokerStreamAttachment, ports.BrokerAdmissionExact},
795 } {
796 t.Run(tc.name, func(t *testing.T) {
797 open := Open{Ref: testRef(5), Purpose: tc.purpose, Admission: tc.admission, Local: true, Policy: testPolicy(), StartMode: tc.mode}
798 if tc.admission == ports.BrokerAdmissionExact {
799 open.Target = testTarget()
800 }
801 raw, err := EncodeClient(open, testEnvelopeCeiling, testChunkCeiling)
802 require.NoError(t, err)
803 decoded, err := DecodeClient(raw, testEnvelopeCeiling, testChunkCeiling)
804 require.NoError(t, err)
805 admitting, ok := decoded.(Open)
806 require.True(t, ok)
807 require.Equal(t, tc.mode, admitting.StartMode)
808 })
809 }
810
811 t.Run("zero wire code is refused", func(t *testing.T) {
812 raw := mustEncodeClient(t, Open{Ref: testRef(6), Purpose: ports.BrokerStreamControl, Local: true, Policy: testPolicy(), StartMode: ports.BrokerDaemonExistingOnly})
813 envelope := &wire.MuxClientEnvelope{}
814 require.NoError(t, proto.Unmarshal(raw, envelope))
815 envelope.GetOpen().StartMode = 0
816 mutated, err := proto.Marshal(envelope)
817 require.NoError(t, err)
818 _, err = DecodeClient(mutated, testEnvelopeCeiling, testChunkCeiling)
819 require.ErrorIs(t, err, ErrInvalidMessage, "an absent start mode is never defaulted")
820 })
821
822 t.Run("unknown wire code is refused", func(t *testing.T) {
823 raw := mustEncodeClient(t, Open{Ref: testRef(6), Purpose: ports.BrokerStreamControl, Local: true, Policy: testPolicy(), StartMode: ports.BrokerDaemonExistingOnly})
824 envelope := &wire.MuxClientEnvelope{}
825 require.NoError(t, proto.Unmarshal(raw, envelope))
826 envelope.GetOpen().StartMode = 3
827 mutated, err := proto.Marshal(envelope)
828 require.NoError(t, err)
829 _, err = DecodeClient(mutated, testEnvelopeCeiling, testChunkCeiling)
830 require.ErrorIs(t, err, ErrInvalidMessage)
831 })
832
833 t.Run("encoding a zero start mode is refused", func(t *testing.T) {
834 _, err := EncodeClient(Open{Ref: testRef(6), Purpose: ports.BrokerStreamControl, Local: true, Policy: testPolicy()}, testEnvelopeCeiling, testChunkCeiling)
835 require.Error(t, err) // Closed wire enum; decode refusal is tested above.
836 })
837}
838
839// TestMuxAdmissionWireRange proves the MuxOpen decoder narrows the wire
840// admission field before any cast: the accepted semantic set (no admission, or
841// one of the three closed admission codes) is unchanged, while every wire value
842// whose high bits would alias a valid code is refused instead of silently
843// truncating onto that code.
844func TestMuxAdmissionWireRange(t *testing.T) {
845 control := Open{Ref: testRef(6), Purpose: ports.BrokerStreamControl, Local: true, Policy: testPolicy(), StartMode: ports.BrokerDaemonStartIfNeeded}
846 named := testOpen()
847 named.Admission = ports.BrokerAdmissionCreateNamed
848 named.Name = "work"
849 named.Target = protocol.ExactSessionTarget{}
850 ephemeral := testOpen()
851 ephemeral.Admission = ports.BrokerAdmissionCreateEphemeral
852 ephemeral.Target = protocol.ExactSessionTarget{}
853 attachNamed := testOpen()
854 attachNamed.Admission = ports.BrokerAdmissionAttachNamed
855 attachNamed.Name = "work"
856 attachNamed.Target = protocol.ExactSessionTarget{}
857
858 tests := []struct {
859 name string
860 base Open
861 admission uint32
862 wantErr bool
863 }{
864 {"no admission on a control stream", control, 0, false},
865 {"exact admission", testOpen(), uint32(ports.BrokerAdmissionExact), false},
866 {"create-named admission", named, uint32(ports.BrokerAdmissionCreateNamed), false},
867 {"create-ephemeral admission", ephemeral, uint32(ports.BrokerAdmissionCreateEphemeral), false},
868 {"attach-named admission", attachNamed, uint32(ports.BrokerAdmissionAttachNamed), false},
869 {"wire value 5 past the closed set", attachNamed, 5, true},
870 {"wire value 260 aliasing attach-named", attachNamed, 260, true},
871 {"wire value 257 aliasing exact", testOpen(), 257, true},
872 {"wire value 258 aliasing create-named", named, 258, true},
873 {"wire value 1024 aliasing no admission", control, 1024, true},
874 {"wire value max uint32", control, 1<<32 - 1, true},
875 }
876 for _, tc := range tests {
877 t.Run(tc.name, func(t *testing.T) {
878 raw := mustEncodeClient(t, tc.base)
879 envelope := &wire.MuxClientEnvelope{}
880 require.NoError(t, proto.Unmarshal(raw, envelope))
881 envelope.GetOpen().Admission = tc.admission
882 mutated, err := proto.Marshal(envelope)
883 require.NoError(t, err)
884 decoded, err := DecodeClient(mutated, testEnvelopeCeiling, testChunkCeiling)
885 if tc.wantErr {
886 require.ErrorIs(t, err, ErrInvalidMessage)
887 require.Zero(t, decoded, "an out-of-range wire value is never accepted as an alias")
888 return
889 }
890 require.NoError(t, err)
891 require.Equal(t, tc.base, decoded)
892 })
893 }
894}