1package daemonmux
2
3import (
4 "testing"
5
6 "google.golang.org/protobuf/encoding/protowire"
7 "google.golang.org/protobuf/proto"
8
9 "github.com/bnema/vev/internal/ports"
10 "github.com/bnema/vev/internal/protocol"
11 "github.com/bnema/vev/internal/protocol/wire"
12 "github.com/stretchr/testify/require"
13)
14
15func testIdentity() ports.BrokerDaemonIdentity {
16 return ports.BrokerDaemonIdentity("daemon-identity-1")
17}
18
19func testIncarnation() ports.BrokerDaemonIncarnation {
20 return ports.BrokerDaemonIncarnation{1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16}
21}
22
23func mustEncodePreambleRequest(t *testing.T, ceilings MuxCeilings, policy ports.BrokerPolicy) *wire.MuxPreambleRequest {
24 t.Helper()
25 request, err := EncodePreambleRequest(ceilings, policy)
26 require.NoError(t, err)
27 return request
28}
29
30func mustEncodePreambleResponse(t *testing.T, accepted bool, ceilings MuxCeilings, policy ports.BrokerPolicy, identity ports.BrokerDaemonIdentity, incarnation ports.BrokerDaemonIncarnation, code uint32) *wire.MuxPreambleResponse {
31 t.Helper()
32 response, err := EncodePreambleResponse(accepted, ceilings, policy, identity, incarnation, code)
33 require.NoError(t, err)
34 return response
35}
36
37// TestMuxPreambleByteLengthHelper proves the exposed 4 KiB byte-length helper
38// accepts at the boundary and refuses one byte over, for both encoded and
39// received preambles.
40func TestMuxPreambleByteLengthHelper(t *testing.T) {
41 require.Equal(t, wire.PreambleLimit, MuxPreambleLimit)
42 require.Equal(t, 4<<10, MuxPreambleLimit)
43 require.NoError(t, CheckPreambleSize(nil))
44 require.NoError(t, CheckPreambleSize(make([]byte, MuxPreambleLimit)))
45 require.ErrorIs(t, CheckPreambleSize(make([]byte, MuxPreambleLimit+1)), ErrPreambleRejected)
46
47 request, err := proto.Marshal(mustEncodePreambleRequest(t, DefaultMuxCeilings(), testPolicy()))
48 require.NoError(t, err)
49 require.NoError(t, CheckPreambleSize(request))
50 response, err := proto.Marshal(mustEncodePreambleResponse(t, true, DefaultMuxCeilings(), testPolicy(), testIdentity(), testIncarnation(), 0))
51 require.NoError(t, err)
52 require.NoError(t, CheckPreambleSize(response))
53}
54
55// TestMuxPreambleRoles proves the daemonmux preamble uses roles 5/6, the
56// shared magic/epoch/exact version, zero capability bits, the 4 KiB preamble
57// bound, and precise refusal mapping.
58func TestMuxPreambleRoles(t *testing.T) {
59 t.Run("client role is 5 and server role is 6", func(t *testing.T) {
60 request := mustEncodePreambleRequest(t, DefaultMuxCeilings(), testPolicy())
61 require.Equal(t, MuxRoleClient, request.GetRole().GetRole())
62 require.Equal(t, uint32(5), request.GetRole().GetRole())
63 response := mustEncodePreambleResponse(t, true, DefaultMuxCeilings(), testPolicy(), testIdentity(), testIncarnation(), 0)
64 require.Equal(t, MuxRoleServer, response.GetRole().GetRole())
65 require.Equal(t, uint32(6), response.GetRole().GetRole())
66 // The daemonmux roles are disjoint from the session (1/2) and broker
67 // (3/4) roles.
68 require.NotEqual(t, MuxRoleClient, MuxRoleServer)
69 require.NotContains(t, []uint32{1, 2, 3, 4}, MuxRoleClient)
70 require.NotContains(t, []uint32{1, 2, 3, 4}, MuxRoleServer)
71 })
72 t.Run("magic epoch version match the shared contract", func(t *testing.T) {
73 request := mustEncodePreambleRequest(t, DefaultMuxCeilings(), testPolicy())
74 require.Equal(t, wire.PreambleMagic, request.GetMagic())
75 require.Equal(t, wire.ProtocolEpoch, request.GetEpoch())
76 require.Equal(t, uint32(protocol.Version), request.GetVersion())
77 require.Zero(t, request.GetCapabilityBits())
78 response := mustEncodePreambleResponse(t, true, DefaultMuxCeilings(), testPolicy(), testIdentity(), testIncarnation(), 0)
79 require.Equal(t, wire.PreambleMagic, response.GetMagic())
80 require.Equal(t, wire.ProtocolEpoch, response.GetEpoch())
81 require.Equal(t, uint32(protocol.Version), response.GetVersion())
82 require.Zero(t, response.GetCapabilityBits())
83 })
84 t.Run("preamble fits 4KiB with a maximum identity", func(t *testing.T) {
85 identity := ports.BrokerDaemonIdentity(string(make([]byte, 0, ports.BrokerMaxIdentityBytes)))
86 for range ports.BrokerMaxIdentityBytes {
87 identity += "a"
88 }
89 require.NoError(t, identity.Validate())
90 raw, err := proto.Marshal(mustEncodePreambleResponse(t, true, DefaultMuxCeilings(), testPolicy(), identity, testIncarnation(), 0))
91 require.NoError(t, err)
92 require.LessOrEqual(t, len(raw), MuxPreambleLimit)
93 require.Equal(t, 4<<10, MuxPreambleLimit)
94 raw, err = proto.Marshal(mustEncodePreambleRequest(t, DefaultMuxCeilings(), testPolicy()))
95 require.NoError(t, err)
96 require.LessOrEqual(t, len(raw), MuxPreambleLimit)
97 })
98 t.Run("foreign roles rejected", func(t *testing.T) {
99 request := mustEncodePreambleRequest(t, DefaultMuxCeilings(), testPolicy())
100 for _, role := range []uint32{1, 2, 3, 4, 6, 9} {
101 request.Role.Role = role
102 _, err := DecodePreambleRequest(request)
103 require.ErrorIs(t, err, ErrPreambleRejected)
104 require.Equal(t, RejectionWrongRole, RejectionCodeFor(request, err))
105 }
106 response := mustEncodePreambleResponse(t, true, DefaultMuxCeilings(), testPolicy(), testIdentity(), testIncarnation(), 0)
107 for _, role := range []uint32{1, 2, 3, 4, 5, 9} {
108 response.Role.Role = role
109 _, err := DecodePreambleResponse(response)
110 require.ErrorIs(t, err, ErrPreambleRejected)
111 }
112 })
113 t.Run("caps zero", func(t *testing.T) {
114 request := mustEncodePreambleRequest(t, DefaultMuxCeilings(), testPolicy())
115 request.CapabilityBits = 1
116 _, err := DecodePreambleRequest(request)
117 require.ErrorIs(t, err, ErrPreambleRejected)
118 require.Equal(t, RejectionLimitRefused, RejectionCodeFor(request, err))
119 response := mustEncodePreambleResponse(t, true, DefaultMuxCeilings(), testPolicy(), testIdentity(), testIncarnation(), 0)
120 response.CapabilityBits = 7
121 _, err = DecodePreambleResponse(response)
122 require.ErrorIs(t, err, ErrPreambleRejected)
123 })
124 t.Run("nil preambles rejected", func(t *testing.T) {
125 _, err := DecodePreambleRequest(nil)
126 require.ErrorIs(t, err, ErrPreambleRejected)
127 _, err = DecodePreambleResponse(nil)
128 require.ErrorIs(t, err, ErrPreambleRejected)
129 })
130}
131
132// TestMuxPreambleCeilingWindows proves every advertised window is refused at
133// max-1 outside the floor, accepted at the window edges, and refused one byte
134// above the ceiling.
135func TestMuxPreambleCeilingWindows(t *testing.T) {
136 mutate := func(name string, mutate func(*wire.MuxPreambleRequest)) {
137 t.Run(name, func(t *testing.T) {
138 request := mustEncodePreambleRequest(t, DefaultMuxCeilings(), testPolicy())
139 mutate(request)
140 _, err := DecodePreambleRequest(request)
141 require.ErrorIs(t, err, ErrPreambleRejected)
142 require.Equal(t, RejectionLimitRefused, RejectionCodeFor(request, err))
143 })
144 }
145 accept := func(name string, mutate func(*wire.MuxPreambleRequest), want MuxCeilings) {
146 t.Run(name, func(t *testing.T) {
147 request := mustEncodePreambleRequest(t, DefaultMuxCeilings(), testPolicy())
148 mutate(request)
149 decoded, err := DecodePreambleRequest(request)
150 require.NoError(t, err)
151 require.Equal(t, want, decoded.Ceilings)
152 })
153 }
154 defaults := DefaultMuxCeilings()
155
156 t.Run("envelope window 1MiB..16MiB", func(t *testing.T) {
157 require.Equal(t, uint64(1<<20), MinMuxEnvelopeBytes)
158 require.Equal(t, uint64(16<<20), MaxMuxEnvelopeBytes)
159 mutate("below floor", func(r *wire.MuxPreambleRequest) { r.MaxReceiveEnvelopeBytes = MinMuxEnvelopeBytes - 1 })
160 mutate("above ceiling", func(r *wire.MuxPreambleRequest) { r.MaxReceiveEnvelopeBytes = MaxMuxEnvelopeBytes + 1 })
161 mutate("zero", func(r *wire.MuxPreambleRequest) { r.MaxReceiveEnvelopeBytes = 0 })
162 accept("at floor", func(r *wire.MuxPreambleRequest) { r.MaxReceiveEnvelopeBytes = MinMuxEnvelopeBytes }, MuxCeilings{
163 MaxReceiveEnvelopeBytes: MinMuxEnvelopeBytes,
164 StreamChunkLimit: defaults.StreamChunkLimit,
165 MaxStreams: defaults.MaxStreams,
166 MaxAggregateBytes: defaults.MaxAggregateBytes,
167 })
168 accept("floor+1", func(r *wire.MuxPreambleRequest) { r.MaxReceiveEnvelopeBytes = MinMuxEnvelopeBytes + 1 }, MuxCeilings{
169 MaxReceiveEnvelopeBytes: MinMuxEnvelopeBytes + 1,
170 StreamChunkLimit: defaults.StreamChunkLimit,
171 MaxStreams: defaults.MaxStreams,
172 MaxAggregateBytes: defaults.MaxAggregateBytes,
173 })
174 accept("at ceiling", func(r *wire.MuxPreambleRequest) { r.MaxReceiveEnvelopeBytes = MaxMuxEnvelopeBytes }, defaults)
175 accept("ceiling-1", func(r *wire.MuxPreambleRequest) { r.MaxReceiveEnvelopeBytes = MaxMuxEnvelopeBytes - 1 }, MuxCeilings{
176 MaxReceiveEnvelopeBytes: MaxMuxEnvelopeBytes - 1,
177 StreamChunkLimit: defaults.StreamChunkLimit,
178 MaxStreams: defaults.MaxStreams,
179 MaxAggregateBytes: defaults.MaxAggregateBytes,
180 })
181 })
182 t.Run("chunk window 1..64KiB", func(t *testing.T) {
183 require.Equal(t, uint64(1), MinMuxChunkBytes)
184 require.Equal(t, uint64(64<<10), MaxMuxChunkBytes)
185 mutate("zero", func(r *wire.MuxPreambleRequest) { r.StreamChunkLimit = 0 })
186 mutate("above ceiling", func(r *wire.MuxPreambleRequest) { r.StreamChunkLimit = MaxMuxChunkBytes + 1 })
187 accept("at floor", func(r *wire.MuxPreambleRequest) { r.StreamChunkLimit = MinMuxChunkBytes }, MuxCeilings{
188 MaxReceiveEnvelopeBytes: defaults.MaxReceiveEnvelopeBytes,
189 StreamChunkLimit: MinMuxChunkBytes,
190 MaxStreams: defaults.MaxStreams,
191 MaxAggregateBytes: defaults.MaxAggregateBytes,
192 })
193 accept("floor+1", func(r *wire.MuxPreambleRequest) { r.StreamChunkLimit = MinMuxChunkBytes + 1 }, MuxCeilings{
194 MaxReceiveEnvelopeBytes: defaults.MaxReceiveEnvelopeBytes,
195 StreamChunkLimit: MinMuxChunkBytes + 1,
196 MaxStreams: defaults.MaxStreams,
197 MaxAggregateBytes: defaults.MaxAggregateBytes,
198 })
199 accept("at ceiling", func(r *wire.MuxPreambleRequest) { r.StreamChunkLimit = MaxMuxChunkBytes }, defaults)
200 accept("ceiling-1", func(r *wire.MuxPreambleRequest) { r.StreamChunkLimit = MaxMuxChunkBytes - 1 }, MuxCeilings{
201 MaxReceiveEnvelopeBytes: defaults.MaxReceiveEnvelopeBytes,
202 StreamChunkLimit: MaxMuxChunkBytes - 1,
203 MaxStreams: defaults.MaxStreams,
204 MaxAggregateBytes: defaults.MaxAggregateBytes,
205 })
206 })
207 t.Run("stream window 1..128", func(t *testing.T) {
208 require.Equal(t, uint64(1), MinMuxStreams)
209 require.Equal(t, uint64(128), MaxMuxStreams)
210 mutate("zero", func(r *wire.MuxPreambleRequest) { r.MaxStreams = 0 })
211 mutate("above ceiling", func(r *wire.MuxPreambleRequest) { r.MaxStreams = MaxMuxStreams + 1 })
212 accept("at floor", func(r *wire.MuxPreambleRequest) { r.MaxStreams = MinMuxStreams }, MuxCeilings{
213 MaxReceiveEnvelopeBytes: defaults.MaxReceiveEnvelopeBytes,
214 StreamChunkLimit: defaults.StreamChunkLimit,
215 MaxStreams: MinMuxStreams,
216 MaxAggregateBytes: defaults.MaxAggregateBytes,
217 })
218 accept("floor+1", func(r *wire.MuxPreambleRequest) { r.MaxStreams = MinMuxStreams + 1 }, MuxCeilings{
219 MaxReceiveEnvelopeBytes: defaults.MaxReceiveEnvelopeBytes,
220 StreamChunkLimit: defaults.StreamChunkLimit,
221 MaxStreams: MinMuxStreams + 1,
222 MaxAggregateBytes: defaults.MaxAggregateBytes,
223 })
224 accept("at ceiling", func(r *wire.MuxPreambleRequest) { r.MaxStreams = MaxMuxStreams }, defaults)
225 accept("ceiling-1", func(r *wire.MuxPreambleRequest) { r.MaxStreams = MaxMuxStreams - 1 }, MuxCeilings{
226 MaxReceiveEnvelopeBytes: defaults.MaxReceiveEnvelopeBytes,
227 StreamChunkLimit: defaults.StreamChunkLimit,
228 MaxStreams: MaxMuxStreams - 1,
229 MaxAggregateBytes: defaults.MaxAggregateBytes,
230 })
231 })
232 t.Run("aggregate window floor..64MiB", func(t *testing.T) {
233 require.Equal(t, MaxMuxChunkBytes+MuxEnvelopeOverheadBytes, MinMuxAggregateBytes)
234 require.Equal(t, uint64(64<<20), MaxMuxAggregateBytes)
235 mutate("zero", func(r *wire.MuxPreambleRequest) { r.MaxAggregateBytes = 0 })
236 mutate("above ceiling", func(r *wire.MuxPreambleRequest) { r.MaxAggregateBytes = MaxMuxAggregateBytes + 1 })
237 accept("at floor", func(r *wire.MuxPreambleRequest) { r.MaxAggregateBytes = MinMuxAggregateBytes }, MuxCeilings{
238 MaxReceiveEnvelopeBytes: defaults.MaxReceiveEnvelopeBytes,
239 StreamChunkLimit: defaults.StreamChunkLimit,
240 MaxStreams: defaults.MaxStreams,
241 MaxAggregateBytes: MinMuxAggregateBytes,
242 })
243 accept("at ceiling", func(r *wire.MuxPreambleRequest) { r.MaxAggregateBytes = MaxMuxAggregateBytes }, defaults)
244 accept("ceiling-1", func(r *wire.MuxPreambleRequest) { r.MaxAggregateBytes = MaxMuxAggregateBytes - 1 }, MuxCeilings{
245 MaxReceiveEnvelopeBytes: defaults.MaxReceiveEnvelopeBytes,
246 StreamChunkLimit: defaults.StreamChunkLimit,
247 MaxStreams: defaults.MaxStreams,
248 MaxAggregateBytes: MaxMuxAggregateBytes - 1,
249 })
250 })
251}
252
253// TestMuxPreambleNegotiatesMinima proves the effective ceilings are the
254// element-wise minima of both advertisements, round-tripped through the wire.
255func TestMuxPreambleNegotiatesMinima(t *testing.T) {
256 local := DefaultMuxCeilings()
257 remote := MuxCeilings{
258 MaxReceiveEnvelopeBytes: MinMuxEnvelopeBytes,
259 StreamChunkLimit: 1024,
260 MaxStreams: 8,
261 MaxAggregateBytes: 1 << 20,
262 }
263 decoded, err := DecodePreambleRequest(mustEncodePreambleRequest(t, remote, testPolicy()))
264 require.NoError(t, err)
265 require.Equal(t, remote, decoded.Ceilings)
266 require.Equal(t, remote, EffectiveMuxCeilings(local, decoded.Ceilings))
267
268 // A response advertising the same minima round-trips them.
269 response := mustEncodePreambleResponse(t, true, remote, testPolicy(), testIdentity(), testIncarnation(), 0)
270 decodedResponse, err := DecodePreambleResponse(response)
271 require.NoError(t, err)
272 require.True(t, decodedResponse.Accepted)
273 require.Equal(t, remote, decodedResponse.Ceilings)
274
275 // The local default is never above the absolute ceilings.
276 require.Equal(t, MaxMuxEnvelopeBytes, local.MaxReceiveEnvelopeBytes)
277 require.Equal(t, MaxMuxChunkBytes, local.StreamChunkLimit)
278 require.Equal(t, MaxMuxStreams, local.MaxStreams)
279 require.Equal(t, MaxMuxAggregateBytes, local.MaxAggregateBytes)
280}
281
282// TestEffectiveMuxCeilings proves the effective ceilings are the element-wise
283// minima of both advertisements: a lower remote offer lowers only that field,
284// and the result is symmetric and idempotent.
285func TestEffectiveMuxCeilings(t *testing.T) {
286 local := DefaultMuxCeilings()
287 remote := MuxCeilings{
288 MaxReceiveEnvelopeBytes: local.MaxReceiveEnvelopeBytes - 1,
289 StreamChunkLimit: local.StreamChunkLimit,
290 MaxStreams: 4,
291 MaxAggregateBytes: local.MaxAggregateBytes,
292 }
293 effective := EffectiveMuxCeilings(local, remote)
294 require.Equal(t, remote.MaxReceiveEnvelopeBytes, effective.MaxReceiveEnvelopeBytes)
295 require.Equal(t, local.StreamChunkLimit, effective.StreamChunkLimit)
296 require.Equal(t, uint64(4), effective.MaxStreams)
297 require.Equal(t, local.MaxAggregateBytes, effective.MaxAggregateBytes)
298
299 require.Equal(t, effective, EffectiveMuxCeilings(remote, local))
300 require.Equal(t, effective, EffectiveMuxCeilings(effective, effective))
301}
302
303// TestEffectiveMuxCeilingsWindowsFitAggregate proves the negotiated stream
304// ceiling never lets the sum of every stream's full window exceed the
305// aggregate budget, so a peer that respects its credit can never trip the
306// aggregate bound and be reset for it.
307func TestEffectiveMuxCeilingsWindowsFitAggregate(t *testing.T) {
308 cases := []struct {
309 name string
310 chunk uint64
311 streams uint64
312 aggregate uint64
313 wantStreams uint64
314 }{
315 {"defaults unchanged", MaxMuxChunkBytes, MaxMuxStreams, MaxMuxAggregateBytes, MaxMuxStreams},
316 {"aggregate floor admits one stream", MaxMuxChunkBytes, MaxMuxStreams, MinMuxAggregateBytes, 1},
317 {"1 MiB aggregate with max chunks", MaxMuxChunkBytes, MaxMuxStreams, 1 << 20, (1 << 20) / chunkCredit(int(MaxMuxChunkBytes))},
318 {"small chunks keep every stream", 1024, MaxMuxStreams, 1 << 20, MaxMuxStreams},
319 }
320 for _, tc := range cases {
321 t.Run(tc.name, func(t *testing.T) {
322 offer := MuxCeilings{
323 MaxReceiveEnvelopeBytes: MinMuxEnvelopeBytes,
324 StreamChunkLimit: tc.chunk,
325 MaxStreams: tc.streams,
326 MaxAggregateBytes: tc.aggregate,
327 }
328 effective := EffectiveMuxCeilings(DefaultMuxCeilings(), offer)
329 require.NoError(t, effective.Validate())
330 require.Equal(t, tc.wantStreams, effective.MaxStreams)
331 require.LessOrEqual(t, effective.MaxStreams*effective.StreamWindow(), effective.MaxAggregateBytes)
332 require.Equal(t, effective, EffectiveMuxCeilings(effective, effective), "idempotent, so both peers agree")
333 require.NoError(t, ValidatePreambleResponseAgainstOffer(MuxPreambleResponse{Accepted: true, Ceilings: effective}, offer))
334 })
335 }
336}
337
338// TestValidatePreambleResponseAgainstOffer proves an accepted response is
339// refused when any effective ceiling exceeds the local offer, and accepted
340// when every ceiling is at or below it. A refusal, or an invalid offer, is a
341// protocol violation.
342func TestValidatePreambleResponseAgainstOffer(t *testing.T) {
343 offered := MuxCeilings{
344 MaxReceiveEnvelopeBytes: MinMuxEnvelopeBytes + 1,
345 StreamChunkLimit: MinMuxChunkBytes + 1,
346 MaxStreams: MinMuxStreams + 1,
347 MaxAggregateBytes: MinMuxAggregateBytes + 1,
348 }
349 accepted := func(mutate func(*MuxCeilings)) MuxPreambleResponse {
350 ceilings := offered
351 if mutate != nil {
352 mutate(&ceilings)
353 }
354 return MuxPreambleResponse{Accepted: true, Ceilings: ceilings}
355 }
356
357 cases := []struct {
358 name string
359 response MuxPreambleResponse
360 wantErr bool
361 }{
362 {"at offer", accepted(nil), false},
363 {"below offer", accepted(func(c *MuxCeilings) {
364 c.MaxReceiveEnvelopeBytes = MinMuxEnvelopeBytes
365 c.StreamChunkLimit = MinMuxChunkBytes
366 c.MaxStreams = MinMuxStreams
367 c.MaxAggregateBytes = MinMuxAggregateBytes
368 }), false},
369 {"envelope above offer", accepted(func(c *MuxCeilings) { c.MaxReceiveEnvelopeBytes++ }), true},
370 {"chunk above offer", accepted(func(c *MuxCeilings) { c.StreamChunkLimit++ }), true},
371 {"streams above offer", accepted(func(c *MuxCeilings) { c.MaxStreams++ }), true},
372 {"aggregate above offer", accepted(func(c *MuxCeilings) { c.MaxAggregateBytes++ }), true},
373 }
374 for _, tc := range cases {
375 t.Run(tc.name, func(t *testing.T) {
376 err := ValidatePreambleResponseAgainstOffer(tc.response, offered)
377 if tc.wantErr {
378 require.ErrorIs(t, err, ErrPreambleRejected)
379 return
380 }
381 require.NoError(t, err)
382 })
383 }
384
385 t.Run("refusal", func(t *testing.T) {
386 response := MuxPreambleResponse{Accepted: false, Code: RejectionLimitRefused}
387 require.ErrorIs(t, ValidatePreambleResponseAgainstOffer(response, offered), ErrPreambleRejected)
388 })
389
390 t.Run("invalid offer", func(t *testing.T) {
391 require.ErrorIs(t, ValidatePreambleResponseAgainstOffer(accepted(nil), MuxCeilings{}), ErrPreambleRejected)
392 })
393}
394
395// TestMuxPreambleAcceptCarriesAuthority proves an acceptance carries the
396// authenticated daemon identity, the 16-byte incarnation, and the accepted
397// policy losslessly, and that a request carries the requested policy.
398func TestMuxPreambleAcceptCarriesAuthority(t *testing.T) {
399 request, err := proto.Marshal(mustEncodePreambleRequest(t, DefaultMuxCeilings(), testPolicy()))
400 require.NoError(t, err)
401 decodedRequest := &wire.MuxPreambleRequest{}
402 require.NoError(t, proto.Unmarshal(request, decodedRequest))
403 parsedRequest, err := DecodePreambleRequest(decodedRequest)
404 require.NoError(t, err)
405 require.Equal(t, testPolicy(), parsedRequest.Policy)
406
407 response, err := proto.Marshal(mustEncodePreambleResponse(t, true, DefaultMuxCeilings(), testPolicy(), testIdentity(), testIncarnation(), 0))
408 require.NoError(t, err)
409 decodedResponse := &wire.MuxPreambleResponse{}
410 require.NoError(t, proto.Unmarshal(response, decodedResponse))
411 parsedResponse, err := DecodePreambleResponse(decodedResponse)
412 require.NoError(t, err)
413 require.True(t, parsedResponse.Accepted)
414 require.Equal(t, testPolicy(), parsedResponse.Policy)
415 require.Equal(t, testIdentity(), parsedResponse.Identity)
416 require.Equal(t, testIncarnation(), parsedResponse.Incarnation)
417 require.Zero(t, parsedResponse.Code)
418
419 // The accepted policy is exactly the requested policy when they match;
420 // a different accepted policy stays visible to the caller.
421 other := testPolicy()
422 other.Trust = "tofu"
423 acceptedOther := mustEncodePreambleResponse(t, true, DefaultMuxCeilings(), other, testIdentity(), testIncarnation(), 0)
424 parsedOther, err := DecodePreambleResponse(acceptedOther)
425 require.NoError(t, err)
426 require.Equal(t, other, parsedOther.Policy)
427 require.NotEqual(t, parsedRequest.Policy, parsedOther.Policy)
428 require.False(t, parsedRequest.Policy.Compatible(parsedOther.Policy))
429}
430
431// TestMuxPreambleIdentityChecks proves the accepted identity and incarnation
432// are mandatory, exact, and validated: they authorize the physical
433// connection, so a malformed one is never accepted silently.
434func TestMuxPreambleIdentityChecks(t *testing.T) {
435 t.Run("encode refuses a missing or invalid authority", func(t *testing.T) {
436 _, err := EncodePreambleResponse(true, DefaultMuxCeilings(), testPolicy(), "", testIncarnation(), 0)
437 require.ErrorIs(t, err, ErrPreambleRejected)
438 _, err = EncodePreambleResponse(true, DefaultMuxCeilings(), testPolicy(), ports.BrokerDaemonIdentity("has space"), testIncarnation(), 0)
439 require.ErrorIs(t, err, ErrPreambleRejected)
440 _, err = EncodePreambleResponse(true, DefaultMuxCeilings(), testPolicy(), testIdentity(), ports.BrokerDaemonIncarnation{}, 0)
441 require.ErrorIs(t, err, ErrPreambleRejected)
442 })
443 t.Run("decode refuses mutated authority", func(t *testing.T) {
444 cases := map[string]func(*wire.MuxPreambleResponse){
445 "empty identity": func(r *wire.MuxPreambleResponse) { r.DaemonIdentity = "" },
446 "blank identity": func(r *wire.MuxPreambleResponse) { r.DaemonIdentity = " " },
447 "control identity": func(r *wire.MuxPreambleResponse) { r.DaemonIdentity = "id\x01" },
448 "short incarnation": func(r *wire.MuxPreambleResponse) { r.Incarnation = r.Incarnation[:15] },
449 "long incarnation": func(r *wire.MuxPreambleResponse) { r.Incarnation = append(r.Incarnation, 0) },
450 "zero incarnation": func(r *wire.MuxPreambleResponse) { r.Incarnation = make([]byte, 16) },
451 "missing policy": func(r *wire.MuxPreambleResponse) { r.AcceptedPolicy = nil },
452 "invalid policy token": func(r *wire.MuxPreambleResponse) { r.AcceptedPolicy.Trust = "" },
453 }
454 for name, mutate := range cases {
455 t.Run(name, func(t *testing.T) {
456 response := mustEncodePreambleResponse(t, true, DefaultMuxCeilings(), testPolicy(), testIdentity(), testIncarnation(), 0)
457 mutate(response)
458 _, err := DecodePreambleResponse(response)
459 require.ErrorIs(t, err, ErrPreambleRejected)
460 })
461 }
462 })
463 t.Run("oversize authority is refused", func(t *testing.T) {
464 identity := ports.BrokerDaemonIdentity(string(make([]byte, 0, ports.BrokerMaxIdentityBytes+1)))
465 for range ports.BrokerMaxIdentityBytes + 1 {
466 identity += "a"
467 }
468 _, err := EncodePreambleResponse(true, DefaultMuxCeilings(), testPolicy(), identity, testIncarnation(), 0)
469 require.ErrorIs(t, err, ErrPreambleRejected)
470 })
471}
472
473// TestMuxPreambleRefusal proves refusals carry the precise code, that a
474// code-less refusal is refused locally, and that a request with an
475// unnegotiable policy is a refusal, not a decode error class of its own.
476func TestMuxPreambleRefusal(t *testing.T) {
477 for name, code := range map[string]uint32{
478 "bad magic": RejectionBadMagic,
479 "epoch": RejectionEpochMismatch,
480 "version": RejectionVersionMismatch,
481 "wrong role": RejectionWrongRole,
482 "duplicate": RejectionDuplicate,
483 "out of order": RejectionOutOfOrder,
484 "limit refused": RejectionLimitRefused,
485 "invalid policy": RejectionLimitRefused,
486 "capability bits": RejectionLimitRefused,
487 "envelope floor": RejectionLimitRefused,
488 "chunk limit": RejectionLimitRefused,
489 "stream limit": RejectionLimitRefused,
490 "aggregate limit": RejectionLimitRefused,
491 "zero refusal code": RejectionLimitRefused,
492 } {
493 t.Run("response/"+name, func(t *testing.T) {
494 response := mustEncodePreambleResponse(t, false, DefaultMuxCeilings(), testPolicy(), testIdentity(), testIncarnation(), code)
495 require.False(t, response.GetAccepted())
496 require.Equal(t, code, response.GetRejection().GetCode())
497 decoded, err := DecodePreambleResponse(response)
498 require.ErrorIs(t, err, ErrPreambleRejected)
499 require.Equal(t, code, decoded.Code)
500 require.False(t, decoded.Accepted)
501 // A refusal never claims accepted authority.
502 require.Nil(t, response.GetAcceptedPolicy())
503 require.Empty(t, response.GetDaemonIdentity())
504 })
505 }
506 t.Run("code-less refusal is refused locally", func(t *testing.T) {
507 _, err := EncodePreambleResponse(false, DefaultMuxCeilings(), testPolicy(), testIdentity(), testIncarnation(), 0)
508 require.ErrorIs(t, err, ErrPreambleRejected)
509 })
510 t.Run("accepted-flag refusal without a code is refused", func(t *testing.T) {
511 response := mustEncodePreambleResponse(t, false, DefaultMuxCeilings(), testPolicy(), testIdentity(), testIncarnation(), RejectionVersionMismatch)
512 response.Rejection = nil
513 _, err := DecodePreambleResponse(response)
514 require.ErrorIs(t, err, ErrPreambleRejected)
515 })
516 t.Run("request refusals map precisely", func(t *testing.T) {
517 for name, tc := range map[string]struct {
518 mutate func(*wire.MuxPreambleRequest)
519 want uint32
520 }{
521 "bad magic": {func(r *wire.MuxPreambleRequest) { r.Magic = 0xDEAD }, RejectionBadMagic},
522 "epoch": {func(r *wire.MuxPreambleRequest) { r.Epoch++ }, RejectionEpochMismatch},
523 "version": {func(r *wire.MuxPreambleRequest) { r.Version++ }, RejectionVersionMismatch},
524 "role": {func(r *wire.MuxPreambleRequest) { r.Role.Role = 9 }, RejectionWrongRole},
525 "caps": {func(r *wire.MuxPreambleRequest) { r.CapabilityBits = 1 }, RejectionLimitRefused},
526 "envelope floor": {func(r *wire.MuxPreambleRequest) { r.MaxReceiveEnvelopeBytes = 1 }, RejectionLimitRefused},
527 "chunk limit": {func(r *wire.MuxPreambleRequest) { r.StreamChunkLimit = 0 }, RejectionLimitRefused},
528 "stream limit": {func(r *wire.MuxPreambleRequest) { r.MaxStreams = 0 }, RejectionLimitRefused},
529 "aggregate limit": {func(r *wire.MuxPreambleRequest) { r.MaxAggregateBytes = 0 }, RejectionLimitRefused},
530 "missing policy": {func(r *wire.MuxPreambleRequest) { r.Policy = nil }, RejectionLimitRefused},
531 "invalid policy": {func(r *wire.MuxPreambleRequest) { r.Policy.Transport = "" }, RejectionLimitRefused},
532 } {
533 t.Run(name, func(t *testing.T) {
534 request := mustEncodePreambleRequest(t, DefaultMuxCeilings(), testPolicy())
535 tc.mutate(request)
536 _, err := DecodePreambleRequest(request)
537 require.ErrorIs(t, err, ErrPreambleRejected)
538 require.Equal(t, tc.want, RejectionCodeFor(request, err))
539 })
540 }
541 })
542 t.Run("nil request maps to limit refusal", func(t *testing.T) {
543 require.Equal(t, RejectionLimitRefused, RejectionCodeFor(nil, ErrPreambleRejected))
544 require.Zero(t, RejectionCodeFor(nil, nil))
545 })
546}
547
548// TestMuxPreambleEncodeFailClosed proves the local encoder refuses an
549// out-of-window advertisement or an unacceptably empty policy instead of
550// emitting a preamble the peer must reject.
551func TestMuxPreambleEncodeFailClosed(t *testing.T) {
552 bad := DefaultMuxCeilings()
553 bad.MaxStreams = MaxMuxStreams + 1
554 _, err := EncodePreambleRequest(bad, testPolicy())
555 require.ErrorIs(t, err, ErrPreambleRejected)
556
557 _, err = EncodePreambleRequest(DefaultMuxCeilings(), ports.BrokerPolicy{})
558 require.ErrorIs(t, err, ErrPreambleRejected)
559
560 _, err = EncodePreambleResponse(true, bad, testPolicy(), testIdentity(), testIncarnation(), 0)
561 require.ErrorIs(t, err, ErrPreambleRejected)
562
563 _, err = EncodePreambleResponse(true, DefaultMuxCeilings(), ports.BrokerPolicy{}, testIdentity(), testIncarnation(), 0)
564 require.ErrorIs(t, err, ErrPreambleRejected)
565
566 // A refusal is legal without any accepted authority, but it still
567 // carries the shared magic, epoch, version, and server role.
568 refusal := mustEncodePreambleResponse(t, false, MuxCeilings{}, ports.BrokerPolicy{}, "", ports.BrokerDaemonIncarnation{}, RejectionLimitRefused)
569 require.Equal(t, wire.PreambleMagic, refusal.GetMagic())
570 require.Equal(t, MuxRoleServer, refusal.GetRole().GetRole())
571 require.Equal(t, RejectionLimitRefused, refusal.GetRejection().GetCode())
572}
573
574// TestMuxPreambleRefusalCodeRange proves a refusal code is exactly 1..7 on
575// both encode and decode: 0 and anything above 7 fail closed, 6 is accepted as
576// a well-formed wire code even though this codec never emits it.
577func TestMuxPreambleRefusalCodeRange(t *testing.T) {
578 for code := RejectionBadMagic; code <= RejectionLimitRefused; code++ {
579 response, err := EncodePreambleResponse(false, DefaultMuxCeilings(), testPolicy(), testIdentity(), testIncarnation(), code)
580 require.NoError(t, err, "code %d", code)
581 require.False(t, response.GetAccepted())
582 require.Equal(t, code, response.GetRejection().GetCode())
583
584 decoded, err := DecodePreambleResponse(response)
585 require.ErrorIs(t, err, ErrPreambleRejected)
586 require.Equal(t, code, decoded.Code)
587 }
588
589 for _, code := range []uint32{0, 8, 99, 0xFFFFFFFF} {
590 _, err := EncodePreambleResponse(false, DefaultMuxCeilings(), testPolicy(), testIdentity(), testIncarnation(), code)
591 require.ErrorIs(t, err, ErrPreambleRejected, "encode code %d", code)
592 }
593
594 // A well-formed refusal frame with an out-of-range code still fails the
595 // semantic decode.
596 base := mustEncodePreambleResponse(t, false, DefaultMuxCeilings(), testPolicy(), testIdentity(), testIncarnation(), RejectionLimitRefused)
597 for _, code := range []uint32{0, 8, 99} {
598 mutated := proto.Clone(base).(*wire.MuxPreambleResponse)
599 mutated.Rejection = &wire.PreambleRejectionCode{Code: code}
600 _, err := DecodePreambleResponse(mutated)
601 require.ErrorIs(t, err, ErrPreambleRejected, "decode code %d", code)
602 }
603
604 // An acceptance never carries a rejection code.
605 _, err := EncodePreambleResponse(true, DefaultMuxCeilings(), testPolicy(), testIdentity(), testIncarnation(), RejectionDuplicate)
606 require.ErrorIs(t, err, ErrPreambleRejected)
607}
608
609// TestMuxPreambleResponseExclusivity proves an answer is an acceptance or a
610// refusal but never a blend: an acceptance carrying a rejection field and a
611// refusal carrying accepted-only authority are both rejected.
612func TestMuxPreambleResponseExclusivity(t *testing.T) {
613 t.Run("acceptance rejects a rejection field", func(t *testing.T) {
614 response := mustEncodePreambleResponse(t, true, DefaultMuxCeilings(), testPolicy(), testIdentity(), testIncarnation(), 0)
615 require.NotNil(t, response.GetAcceptedPolicy())
616 response.Rejection = &wire.PreambleRejectionCode{Code: RejectionDuplicate}
617 _, err := DecodePreambleResponse(response)
618 require.ErrorIs(t, err, ErrPreambleRejected)
619 })
620
621 t.Run("refusal rejects accepted-only authority", func(t *testing.T) {
622 incarnation := testIncarnation()
623 mutations := map[string]func(*wire.MuxPreambleResponse){
624 "accepted policy": func(r *wire.MuxPreambleResponse) { r.AcceptedPolicy = policyToWire(testPolicy()) },
625 "daemon identity": func(r *wire.MuxPreambleResponse) {
626 r.DaemonIdentity = string(testIdentity())
627 },
628 "incarnation": func(r *wire.MuxPreambleResponse) {
629 r.Incarnation = append([]byte(nil), incarnation[:]...)
630 },
631 }
632 for name, mutate := range mutations {
633 t.Run(name, func(t *testing.T) {
634 response := mustEncodePreambleResponse(t, false, DefaultMuxCeilings(), testPolicy(), testIdentity(), testIncarnation(), RejectionLimitRefused)
635 mutate(response)
636 _, err := DecodePreambleResponse(response)
637 require.ErrorIs(t, err, ErrPreambleRejected)
638 })
639 }
640 })
641}
642
643// TestMuxPreambleStrictBytes proves the byte helpers enforce the 4 KiB bound,
644// the strict scanner, generated unmarshal, and semantic decode for both
645// directions: oversize, empty, unknown, duplicate, trailing/concatenated,
646// truncated, wrong-wire-type, and semantic failures are all refused.
647func TestMuxPreambleStrictBytes(t *testing.T) {
648 requestRaw, err := proto.Marshal(mustEncodePreambleRequest(t, DefaultMuxCeilings(), testPolicy()))
649 require.NoError(t, err)
650 responseRaw, err := proto.Marshal(mustEncodePreambleResponse(t, true, DefaultMuxCeilings(), testPolicy(), testIdentity(), testIncarnation(), 0))
651 require.NoError(t, err)
652
653 t.Run("valid request round-trips", func(t *testing.T) {
654 decoded, err := DecodePreambleRequestBytes(requestRaw)
655 require.NoError(t, err)
656 require.Equal(t, DefaultMuxCeilings(), decoded.Ceilings)
657 require.Equal(t, testPolicy(), decoded.Policy)
658 })
659 t.Run("valid response round-trips", func(t *testing.T) {
660 decoded, err := DecodePreambleResponseBytes(responseRaw)
661 require.NoError(t, err)
662 require.True(t, decoded.Accepted)
663 require.Equal(t, testIdentity(), decoded.Identity)
664 require.Equal(t, testIncarnation(), decoded.Incarnation)
665 })
666 t.Run("refusal frame surfaces its code", func(t *testing.T) {
667 refusal, err := proto.Marshal(mustEncodePreambleResponse(t, false, DefaultMuxCeilings(), testPolicy(), testIdentity(), testIncarnation(), RejectionLimitRefused))
668 require.NoError(t, err)
669 decoded, err := DecodePreambleResponseBytes(refusal)
670 require.ErrorIs(t, err, ErrPreambleRejected)
671 require.Equal(t, RejectionLimitRefused, decoded.Code)
672 })
673
674 corruptions := []struct {
675 name string
676 mutate func(raw []byte) []byte
677 want error
678 }{
679 {"oversize", func([]byte) []byte { return make([]byte, MuxPreambleLimit+1) }, ErrPreambleRejected},
680 {"empty", func([]byte) []byte { return nil }, wire.ErrScanEmpty},
681 {"unknown field", func(raw []byte) []byte {
682 out := protowire.AppendTag(append([]byte(nil), raw...), 399, protowire.VarintType)
683 return protowire.AppendVarint(out, 1)
684 }, wire.ErrScanUnknown},
685 {"duplicate singular field", func(raw []byte) []byte {
686 out := protowire.AppendTag(append([]byte(nil), raw...), 1, protowire.VarintType)
687 return protowire.AppendVarint(out, uint64(wire.PreambleMagic))
688 }, wire.ErrScanDuplicate},
689 {"concatenated frame", func(raw []byte) []byte {
690 return append(append([]byte(nil), raw...), raw...)
691 }, wire.ErrScanDuplicate},
692 {"wrong wire type", func(raw []byte) []byte {
693 out := protowire.AppendTag(append([]byte(nil), raw...), 1, protowire.BytesType)
694 return protowire.AppendString(out, "x")
695 }, wire.ErrScanWireType},
696 {"truncated varint", func(raw []byte) []byte {
697 return append(append([]byte(nil), raw...), 0x80)
698 }, wire.ErrScanTruncated},
699 }
700
701 directions := []struct {
702 name string
703 raw []byte
704 decode func([]byte) error
705 }{
706 {"request", requestRaw, func(payload []byte) error { _, err := DecodePreambleRequestBytes(payload); return err }},
707 {"response", responseRaw, func(payload []byte) error { _, err := DecodePreambleResponseBytes(payload); return err }},
708 }
709 for _, direction := range directions {
710 for _, corruption := range corruptions {
711 t.Run(direction.name+"/"+corruption.name, func(t *testing.T) {
712 err := direction.decode(corruption.mutate(direction.raw))
713 require.Error(t, err)
714 require.ErrorIs(t, err, corruption.want)
715 })
716 }
717 }
718
719 t.Run("truncated prefixes", func(t *testing.T) {
720 for size := range len(requestRaw) {
721 _, err := DecodePreambleRequestBytes(requestRaw[:size])
722 require.Error(t, err, "request prefix[:%d] accepted", size)
723 }
724 _, err := DecodePreambleResponseBytes(requestRaw)
725 require.Error(t, err, "client preamble accepted by the response helper")
726 for size := range len(responseRaw) {
727 _, err := DecodePreambleResponseBytes(responseRaw[:size])
728 require.Error(t, err, "response prefix[:%d] accepted", size)
729 }
730 })
731
732 t.Run("semantic failure from valid carriage", func(t *testing.T) {
733 bad := proto.Clone(mustEncodePreambleRequest(t, DefaultMuxCeilings(), testPolicy())).(*wire.MuxPreambleRequest)
734 bad.Magic = 0xDEADBEEF
735 raw, err := proto.Marshal(bad)
736 require.NoError(t, err)
737 _, err = DecodePreambleRequestBytes(raw)
738 require.ErrorIs(t, err, ErrPreambleRejected)
739 })
740}
741
742// FuzzDecodePreambleRequestBytes proves the strict byte helper never panics on
743// arbitrary input.
744func FuzzDecodePreambleRequestBytes(f *testing.F) {
745 request, err := EncodePreambleRequest(DefaultMuxCeilings(), testPolicy())
746 require.NoError(f, err)
747 raw, err := proto.Marshal(request)
748 require.NoError(f, err)
749 for _, seed := range appendStrictScanSeeds([][]byte{raw}) {
750 f.Add(seed)
751 }
752 f.Fuzz(func(t *testing.T, payload []byte) {
753 _, _ = DecodePreambleRequestBytes(payload)
754 })
755}
756
757// FuzzDecodePreambleResponseBytes proves the strict byte helper never panics on
758// arbitrary input.
759func FuzzDecodePreambleResponseBytes(f *testing.F) {
760 response, err := EncodePreambleResponse(true, DefaultMuxCeilings(), testPolicy(), testIdentity(), testIncarnation(), 0)
761 require.NoError(f, err)
762 raw, err := proto.Marshal(response)
763 require.NoError(f, err)
764 for _, seed := range appendStrictScanSeeds([][]byte{raw}) {
765 f.Add(seed)
766 }
767 f.Fuzz(func(t *testing.T, payload []byte) {
768 _, _ = DecodePreambleResponseBytes(payload)
769 })
770}