1package daemonmux
2
3import (
4 "context"
5 "testing"
6 "time"
7
8 "github.com/stretchr/testify/require"
9 "google.golang.org/protobuf/proto"
10
11 "github.com/bnema/vev/internal/ports"
12 "github.com/bnema/vev/internal/protocol"
13 "github.com/bnema/vev/internal/protocol/wire"
14)
15
16// handshakeEndpoint is the independently resolved endpoint the client verifies
17// the daemon response against.
18func handshakeEndpoint(identity ports.BrokerDaemonIdentity, policy ports.BrokerPolicy) ports.BrokerDialTarget {
19 return ports.BrokerDialTarget{Fence: ports.BrokerEndpointFence{Local: true}, Policy: policy, Address: "quic://127.0.0.1:7777", StartMode: ports.BrokerDaemonExistingOnly, ExpectedIdentity: ports.BrokerExpectedIdentity{Identity: identity, Bound: true}}
20}
21
22func mustServerBinding(t *testing.T) ServerBinding {
23 t.Helper()
24 binding, err := NewServerBinding(testIdentity(), testIncarnation(), testPolicy())
25 require.NoError(t, err)
26 return binding
27}
28
29// mustRemoteServerBinding provisions the same authority with the remote
30// locality, for tests whose Open requests carry a validated remote
31// registration rather than a local endpoint.
32func mustRemoteServerBinding(t *testing.T) ServerBinding {
33 t.Helper()
34 binding, err := NewServerBindings(testIdentity(), testIncarnation(), []ServerPolicyAdmission{remoteAcceptance(testPolicy())})
35 require.NoError(t, err)
36 return binding
37}
38
39// encodedAcceptedResponse builds one valid acceptance and lets the caller
40// corrupt exactly one field before marshaling.
41func encodedAcceptedResponse(t *testing.T, mutate func(*wire.MuxPreambleResponse)) []byte {
42 t.Helper()
43 response, err := EncodePreambleResponse(true, DefaultMuxCeilings(), testPolicy(), testIdentity(), testIncarnation(), 0)
44 require.NoError(t, err)
45 if mutate != nil {
46 mutate(response)
47 }
48 raw, err := proto.Marshal(response)
49 require.NoError(t, err)
50 return raw
51}
52
53// encodedAcceptedPolicy corrupts one field of the accepted policy.
54func encodedAcceptedPolicy(t *testing.T, mutate func(*wire.BrokerWirePolicy)) []byte {
55 t.Helper()
56 return encodedAcceptedResponse(t, func(r *wire.MuxPreambleResponse) { mutate(r.AcceptedPolicy) })
57}
58
59func encodedRefusal(t *testing.T, code uint32) []byte {
60 t.Helper()
61 response, err := EncodePreambleResponse(false, DefaultMuxCeilings(), ports.BrokerPolicy{}, "", ports.BrokerDaemonIncarnation{}, code)
62 require.NoError(t, err)
63 raw, err := proto.Marshal(response)
64 require.NoError(t, err)
65 return raw
66}
67
68// encodedRequest builds one valid request and lets the caller corrupt exactly
69// one field before marshaling.
70func encodedRequest(t *testing.T, ceilings MuxCeilings, policy ports.BrokerPolicy, mutate func(*wire.MuxPreambleRequest)) []byte {
71 t.Helper()
72 request, err := EncodePreambleRequest(ceilings, policy)
73 require.NoError(t, err)
74 if mutate != nil {
75 mutate(request)
76 }
77 raw, err := proto.Marshal(request)
78 require.NoError(t, err)
79 return raw
80}
81
82func awaitSentFrame(t *testing.T, carrier *fakeCarrier) []byte {
83 t.Helper()
84 select {
85 case payload := <-carrier.sent:
86 return payload
87 case <-time.After(time.Second):
88 t.Fatal("handshake sent no frame")
89 return nil
90 }
91}
92
93// TestClientHandshakeAcceptsVerifiedResponse proves a matching acceptance is
94// accepted, yields the server authority, leaves the carrier open, and offers
95// exactly the requested policy and ceilings.
96func TestClientHandshakeAcceptsVerifiedResponse(t *testing.T) {
97 carrier := newFakeCarrier()
98 carrier.inbound <- fakeFrame{payload: encodedAcceptedResponse(t, nil)}
99
100 result, err := RunClientHandshake(context.Background(), carrier, handshakeEndpoint(testIdentity(), testPolicy()), DefaultMuxCeilings())
101 require.NoError(t, err)
102 require.Equal(t, testIdentity(), result.Identity)
103 require.Equal(t, testIncarnation(), result.Incarnation)
104 require.Equal(t, testPolicy(), result.Policy)
105 require.Equal(t, DefaultMuxCeilings(), result.Ceilings)
106 require.Equal(t, 0, carrier.closeCalls())
107
108 request, err := DecodePreambleRequestBytes(awaitSentFrame(t, carrier))
109 require.NoError(t, err)
110 require.Equal(t, DefaultMuxCeilings(), request.Ceilings)
111 require.Equal(t, testPolicy(), request.Policy)
112}
113
114// TestClientHandshakeRejectsUntrustedResponses is the client authority table:
115// wrong identity, every policy field, zero incarnation, wrong role/version/
116// magic/epoch, a typed refusal, an acceptance carrying a rejection field, and
117// excess ceilings are each refused and each close the carrier.
118func TestClientHandshakeRejectsUntrustedResponses(t *testing.T) {
119 smallOffer := MuxCeilings{
120 MaxReceiveEnvelopeBytes: MinMuxEnvelopeBytes,
121 StreamChunkLimit: MinMuxChunkBytes,
122 MaxStreams: MinMuxStreams,
123 MaxAggregateBytes: MinMuxAggregateBytes,
124 }
125
126 tests := []struct {
127 name string
128 offered MuxCeilings
129 response func(t *testing.T) []byte
130 code uint32
131 }{
132 {
133 name: "wrong identity",
134 offered: DefaultMuxCeilings(),
135 response: func(t *testing.T) []byte {
136 return encodedAcceptedResponse(t, func(r *wire.MuxPreambleResponse) { r.DaemonIdentity = "other-identity" })
137 },
138 },
139 {
140 name: "policy protocol version",
141 offered: DefaultMuxCeilings(),
142 response: func(t *testing.T) []byte {
143 return encodedAcceptedPolicy(t, func(p *wire.BrokerWirePolicy) { p.ProtocolVersion++ })
144 },
145 },
146 {
147 name: "policy catalogue version",
148 offered: DefaultMuxCeilings(),
149 response: func(t *testing.T) []byte {
150 return encodedAcceptedPolicy(t, func(p *wire.BrokerWirePolicy) { p.CatalogueVersion++ })
151 },
152 },
153 {
154 name: "policy environment",
155 offered: DefaultMuxCeilings(),
156 response: func(t *testing.T) []byte {
157 return encodedAcceptedPolicy(t, func(p *wire.BrokerWirePolicy) { p.EnvironmentPolicy++ })
158 },
159 },
160 {
161 name: "policy transport",
162 offered: DefaultMuxCeilings(),
163 response: func(t *testing.T) []byte {
164 return encodedAcceptedPolicy(t, func(p *wire.BrokerWirePolicy) { p.Transport = "stdio" })
165 },
166 },
167 {
168 name: "policy trust",
169 offered: DefaultMuxCeilings(),
170 response: func(t *testing.T) []byte {
171 return encodedAcceptedPolicy(t, func(p *wire.BrokerWirePolicy) { p.Trust = "tofu" })
172 },
173 },
174 {
175 name: "policy launch",
176 offered: DefaultMuxCeilings(),
177 response: func(t *testing.T) []byte {
178 return encodedAcceptedPolicy(t, func(p *wire.BrokerWirePolicy) { p.Launch = "manual" })
179 },
180 },
181 {
182 name: "policy isolation",
183 offered: DefaultMuxCeilings(),
184 response: func(t *testing.T) []byte {
185 return encodedAcceptedPolicy(t, func(p *wire.BrokerWirePolicy) { p.Isolation = "system" })
186 },
187 },
188 {
189 name: "zero incarnation",
190 offered: DefaultMuxCeilings(),
191 response: func(t *testing.T) []byte {
192 return encodedAcceptedResponse(t, func(r *wire.MuxPreambleResponse) { r.Incarnation = make([]byte, 16) })
193 },
194 },
195 {
196 name: "wrong role",
197 offered: DefaultMuxCeilings(),
198 response: func(t *testing.T) []byte {
199 return encodedAcceptedResponse(t, func(r *wire.MuxPreambleResponse) { r.Role = &wire.PreambleRole{Role: MuxRoleClient} })
200 },
201 },
202 {
203 name: "wrong version",
204 offered: DefaultMuxCeilings(),
205 response: func(t *testing.T) []byte {
206 return encodedAcceptedResponse(t, func(r *wire.MuxPreambleResponse) { r.Version = uint32(protocol.Version) + 1 })
207 },
208 },
209 {
210 name: "bad magic",
211 offered: DefaultMuxCeilings(),
212 response: func(t *testing.T) []byte {
213 return encodedAcceptedResponse(t, func(r *wire.MuxPreambleResponse) { r.Magic = 0 })
214 },
215 },
216 {
217 name: "epoch mismatch",
218 offered: DefaultMuxCeilings(),
219 response: func(t *testing.T) []byte {
220 return encodedAcceptedResponse(t, func(r *wire.MuxPreambleResponse) { r.Epoch = wire.ProtocolEpoch + 1 })
221 },
222 },
223 {
224 name: "refused response",
225 offered: DefaultMuxCeilings(),
226 response: func(t *testing.T) []byte {
227 return encodedRefusal(t, RejectionVersionMismatch)
228 },
229 code: RejectionVersionMismatch,
230 },
231 {
232 name: "acceptance carrying rejection",
233 offered: DefaultMuxCeilings(),
234 response: func(t *testing.T) []byte {
235 return encodedAcceptedResponse(t, func(r *wire.MuxPreambleResponse) {
236 r.Rejection = &wire.PreambleRejectionCode{Code: RejectionLimitRefused}
237 })
238 },
239 },
240 {
241 name: "excess ceilings",
242 offered: smallOffer,
243 response: func(t *testing.T) []byte {
244 return encodedAcceptedResponse(t, nil)
245 },
246 },
247 }
248
249 for _, test := range tests {
250 t.Run(test.name, func(t *testing.T) {
251 carrier := newFakeCarrier()
252 carrier.inbound <- fakeFrame{payload: test.response(t)}
253
254 _, err := RunClientHandshake(context.Background(), carrier, handshakeEndpoint(testIdentity(), testPolicy()), test.offered)
255 require.ErrorIs(t, err, ErrHandshakeRejected)
256 if test.code != 0 {
257 var refusal *HandshakeRefusal
258 require.ErrorAs(t, err, &refusal)
259 require.Equal(t, test.code, refusal.Code)
260 }
261 require.Equal(t, 1, carrier.closeCalls())
262 })
263 }
264}
265
266// TestServerHandshakeAcceptsAndNegotiates proves the server answers an exact
267// request with the immutable binding and the element-wise minima, and leaves
268// the carrier open.
269func TestServerHandshakeAcceptsAndNegotiates(t *testing.T) {
270 binding := mustServerBinding(t)
271 clientCeilings := MuxCeilings{
272 MaxReceiveEnvelopeBytes: MinMuxEnvelopeBytes,
273 StreamChunkLimit: 8,
274 MaxStreams: 4,
275 MaxAggregateBytes: MinMuxAggregateBytes,
276 }
277 serverCeilings := DefaultMuxCeilings()
278 serverCeilings.MaxStreams = 2
279
280 carrier := newFakeCarrier()
281 carrier.inbound <- fakeFrame{payload: encodedRequest(t, clientCeilings, testPolicy(), nil)}
282
283 result, err := RunServerHandshake(context.Background(), carrier, binding, serverCeilings)
284 require.NoError(t, err)
285 require.Equal(t, binding, result.Binding)
286 require.Equal(t, EffectiveMuxCeilings(clientCeilings, serverCeilings), result.Ceilings)
287 require.Equal(t, testPolicy(), result.Policy, "the accepted policy is the provisioned member")
288 require.Equal(t, ports.SessionOriginLocal, result.Origin, "the accepted origin is the provisioned locality")
289 require.Equal(t, 0, carrier.closeCalls())
290
291 decoded, err := DecodePreambleResponseBytes(awaitSentFrame(t, carrier))
292 require.NoError(t, err)
293 require.True(t, decoded.Accepted)
294 require.Equal(t, testIdentity(), decoded.Identity)
295 require.Equal(t, testIncarnation(), decoded.Incarnation)
296 require.Equal(t, testPolicy(), decoded.Policy)
297 require.Equal(t, result.Ceilings, decoded.Ceilings)
298}
299
300// TestServerHandshakeEnforcesBindingPolicy proves a request differing in one
301// policy field is refused with code 7, no accepted authority is echoed, and
302// the carrier is closed.
303func TestServerHandshakeEnforcesBindingPolicy(t *testing.T) {
304 binding := mustServerBinding(t)
305 tests := []struct {
306 name string
307 mutate func(*ports.BrokerPolicy)
308 }{
309 {name: "protocol version", mutate: func(p *ports.BrokerPolicy) { p.ProtocolVersion++ }},
310 {name: "catalogue schema version", mutate: func(p *ports.BrokerPolicy) { p.CatalogSchemaVersion++ }},
311 {name: "environment policy", mutate: func(p *ports.BrokerPolicy) {
312 p.EnvironmentPolicy = protocol.EnvironmentPolicyClientOwned
313 }},
314 {name: "transport", mutate: func(p *ports.BrokerPolicy) { p.Transport = "stdio" }},
315 {name: "trust", mutate: func(p *ports.BrokerPolicy) { p.Trust = "tofu" }},
316 {name: "launch", mutate: func(p *ports.BrokerPolicy) { p.Launch = "manual" }},
317 {name: "isolation", mutate: func(p *ports.BrokerPolicy) { p.Isolation = "system" }},
318 }
319 for _, test := range tests {
320 t.Run(test.name, func(t *testing.T) {
321 requested := testPolicy()
322 test.mutate(&requested)
323
324 carrier := newFakeCarrier()
325 carrier.inbound <- fakeFrame{payload: encodedRequest(t, DefaultMuxCeilings(), requested, nil)}
326
327 _, err := RunServerHandshake(context.Background(), carrier, binding, DefaultMuxCeilings())
328 var refusal *HandshakeRefusal
329 require.ErrorAs(t, err, &refusal)
330 require.Equal(t, uint32(RejectionLimitRefused), refusal.Code)
331 require.Equal(t, 1, carrier.closeCalls())
332
333 raw := &wire.MuxPreambleResponse{}
334 require.NoError(t, proto.Unmarshal(awaitSentFrame(t, carrier), raw))
335 require.False(t, raw.GetAccepted())
336 require.Nil(t, raw.GetAcceptedPolicy())
337 require.Empty(t, raw.GetDaemonIdentity())
338 require.Empty(t, raw.GetIncarnation())
339 })
340 }
341}
342
343// TestServerHandshakeRefusesMalformedRequest proves the precise refusal code
344// taxonomy for a bad magic, epoch, version, or role, plus oversize and
345// undecodable payloads, and that every refusal closes the carrier.
346func TestServerHandshakeRefusesMalformedRequest(t *testing.T) {
347 binding := mustServerBinding(t)
348 tests := []struct {
349 name string
350 request func(t *testing.T) []byte
351 code uint32
352 }{
353 {
354 name: "bad magic",
355 request: func(t *testing.T) []byte {
356 return encodedRequest(t, DefaultMuxCeilings(), testPolicy(), func(r *wire.MuxPreambleRequest) { r.Magic = 0 })
357 },
358 code: RejectionBadMagic,
359 },
360 {
361 name: "epoch mismatch",
362 request: func(t *testing.T) []byte {
363 return encodedRequest(t, DefaultMuxCeilings(), testPolicy(), func(r *wire.MuxPreambleRequest) { r.Epoch = wire.ProtocolEpoch + 1 })
364 },
365 code: RejectionEpochMismatch,
366 },
367 {
368 name: "wrong version",
369 request: func(t *testing.T) []byte {
370 return encodedRequest(t, DefaultMuxCeilings(), testPolicy(), func(r *wire.MuxPreambleRequest) { r.Version = uint32(protocol.Version) + 1 })
371 },
372 code: RejectionVersionMismatch,
373 },
374 {
375 name: "wrong role",
376 request: func(t *testing.T) []byte {
377 return encodedRequest(t, DefaultMuxCeilings(), testPolicy(), func(r *wire.MuxPreambleRequest) { r.Role = &wire.PreambleRole{Role: MuxRoleServer} })
378 },
379 code: RejectionWrongRole,
380 },
381 {
382 name: "oversize payload",
383 request: func(t *testing.T) []byte {
384 return make([]byte, MuxPreambleLimit+1)
385 },
386 code: RejectionLimitRefused,
387 },
388 {
389 name: "undecodable payload",
390 request: func(t *testing.T) []byte {
391 return []byte{0xff, 0xff, 0xff}
392 },
393 code: RejectionLimitRefused,
394 },
395 }
396 for _, test := range tests {
397 t.Run(test.name, func(t *testing.T) {
398 carrier := newFakeCarrier()
399 carrier.inbound <- fakeFrame{payload: test.request(t)}
400
401 _, err := RunServerHandshake(context.Background(), carrier, binding, DefaultMuxCeilings())
402 var refusal *HandshakeRefusal
403 require.ErrorAs(t, err, &refusal)
404 require.Equal(t, test.code, refusal.Code)
405 require.Equal(t, 1, carrier.closeCalls())
406
407 decoded, err := DecodePreambleResponseBytes(awaitSentFrame(t, carrier))
408 require.ErrorIs(t, err, ErrPreambleRejected)
409 require.Equal(t, test.code, decoded.Code)
410 })
411 }
412}
413
414// TestHandshakeRefusesInvalidConfiguration proves nil carriers, invalid
415// endpoints, invalid bindings, and invalid local ceilings fail closed with the
416// configuration sentinel.
417func TestHandshakeRefusesInvalidConfiguration(t *testing.T) {
418 t.Run("client nil carrier", func(t *testing.T) {
419 _, err := RunClientHandshake(context.Background(), nil, handshakeEndpoint(testIdentity(), testPolicy()), DefaultMuxCeilings())
420 require.ErrorIs(t, err, ErrHandshakeConfig)
421 })
422 t.Run("client invalid endpoint", func(t *testing.T) {
423 carrier := newFakeCarrier()
424 _, err := RunClientHandshake(context.Background(), carrier, handshakeEndpoint("", testPolicy()), DefaultMuxCeilings())
425 require.ErrorIs(t, err, ErrHandshakeConfig)
426 require.Equal(t, 1, carrier.closeCalls())
427 })
428 t.Run("client invalid ceilings", func(t *testing.T) {
429 carrier := newFakeCarrier()
430 _, err := RunClientHandshake(context.Background(), carrier, handshakeEndpoint(testIdentity(), testPolicy()), MuxCeilings{})
431 require.ErrorIs(t, err, ErrHandshakeConfig)
432 require.Equal(t, 1, carrier.closeCalls())
433 })
434 t.Run("server nil carrier", func(t *testing.T) {
435 _, err := RunServerHandshake(context.Background(), nil, mustServerBinding(t), DefaultMuxCeilings())
436 require.ErrorIs(t, err, ErrHandshakeConfig)
437 })
438 t.Run("server invalid binding", func(t *testing.T) {
439 carrier := newFakeCarrier()
440 _, err := RunServerHandshake(context.Background(), carrier, ServerBinding{}, DefaultMuxCeilings())
441 require.ErrorIs(t, err, ErrHandshakeConfig)
442 require.Equal(t, 1, carrier.closeCalls())
443 })
444 t.Run("server invalid ceilings", func(t *testing.T) {
445 carrier := newFakeCarrier()
446 _, err := RunServerHandshake(context.Background(), carrier, mustServerBinding(t), MuxCeilings{})
447 require.ErrorIs(t, err, ErrHandshakeConfig)
448 require.Equal(t, 1, carrier.closeCalls())
449 })
450}
451
452// TestHandshakeHonorsCallerContext proves the caller's context is the single
453// absolute bound on the exchange: a cancelled or already-expired context ends
454// the handshake promptly and closes the carrier.
455func TestHandshakeHonorsCallerContext(t *testing.T) {
456 endpoint := handshakeEndpoint(testIdentity(), testPolicy())
457 binding := mustServerBinding(t)
458
459 tests := []struct {
460 name string
461 ctx func() (context.Context, context.CancelFunc)
462 run func(context.Context, *fakeCarrier) error
463 want error
464 }{
465 {
466 name: "client cancelled",
467 ctx: func() (context.Context, context.CancelFunc) {
468 ctx, cancel := context.WithCancel(context.Background())
469 cancel()
470 return ctx, func() {}
471 },
472 run: func(ctx context.Context, carrier *fakeCarrier) error {
473 _, err := RunClientHandshake(ctx, carrier, endpoint, DefaultMuxCeilings())
474 return err
475 },
476 want: context.Canceled,
477 },
478 {
479 name: "client deadline exceeded",
480 ctx: func() (context.Context, context.CancelFunc) {
481 return context.WithDeadline(context.Background(), time.Now().Add(-time.Second))
482 },
483 run: func(ctx context.Context, carrier *fakeCarrier) error {
484 _, err := RunClientHandshake(ctx, carrier, endpoint, DefaultMuxCeilings())
485 return err
486 },
487 want: context.DeadlineExceeded,
488 },
489 {
490 name: "server cancelled",
491 ctx: func() (context.Context, context.CancelFunc) {
492 ctx, cancel := context.WithCancel(context.Background())
493 cancel()
494 return ctx, func() {}
495 },
496 run: func(ctx context.Context, carrier *fakeCarrier) error {
497 _, err := RunServerHandshake(ctx, carrier, binding, DefaultMuxCeilings())
498 return err
499 },
500 want: context.Canceled,
501 },
502 {
503 name: "server deadline exceeded",
504 ctx: func() (context.Context, context.CancelFunc) {
505 return context.WithDeadline(context.Background(), time.Now().Add(-time.Second))
506 },
507 run: func(ctx context.Context, carrier *fakeCarrier) error {
508 _, err := RunServerHandshake(ctx, carrier, binding, DefaultMuxCeilings())
509 return err
510 },
511 want: context.DeadlineExceeded,
512 },
513 }
514 for _, test := range tests {
515 t.Run(test.name, func(t *testing.T) {
516 ctx, cancel := test.ctx()
517 defer cancel()
518 carrier := newFakeCarrier()
519 err := test.run(ctx, carrier)
520 require.ErrorIs(t, err, test.want)
521 require.Equal(t, 1, carrier.closeCalls())
522 })
523 }
524}
525
526// TestHandshakeEndToEndNegotiates runs the broker and daemon sides over one
527// in-memory link and proves they agree on the element-wise minima and the
528// binding the daemon enforced.
529func TestHandshakeEndToEndNegotiates(t *testing.T) {
530 clientCarrier, serverCarrier := newMemCarrierPair(4)
531 clientCeilings := MuxCeilings{
532 MaxReceiveEnvelopeBytes: MinMuxEnvelopeBytes,
533 StreamChunkLimit: 8,
534 MaxStreams: 4,
535 MaxAggregateBytes: MinMuxAggregateBytes,
536 }
537 serverCeilings := DefaultMuxCeilings()
538 serverCeilings.MaxStreams = 2
539 binding := mustServerBinding(t)
540 endpoint := handshakeEndpoint(testIdentity(), testPolicy())
541
542 type clientOutcome struct {
543 result ClientHandshakeResult
544 err error
545 }
546 type serverOutcome struct {
547 result ServerHandshakeResult
548 err error
549 }
550 clientCh := make(chan clientOutcome, 1)
551 serverCh := make(chan serverOutcome, 1)
552
553 ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
554 defer cancel()
555 go func() {
556 result, err := RunClientHandshake(ctx, clientCarrier, endpoint, clientCeilings)
557 clientCh <- clientOutcome{result, err}
558 }()
559 go func() {
560 result, err := RunServerHandshake(ctx, serverCarrier, binding, serverCeilings)
561 serverCh <- serverOutcome{result, err}
562 }()
563
564 client := <-clientCh
565 server := <-serverCh
566 require.NoError(t, client.err)
567 require.NoError(t, server.err)
568 require.Equal(t, testIdentity(), client.result.Identity)
569 require.Equal(t, testIncarnation(), client.result.Incarnation)
570 require.Equal(t, testPolicy(), client.result.Policy)
571 require.Equal(t, binding, server.result.Binding)
572 require.Equal(t, EffectiveMuxCeilings(clientCeilings, serverCeilings), client.result.Ceilings)
573 require.Equal(t, server.result.Ceilings, client.result.Ceilings)
574}