1// Private daemonmux QUIC carriage seam coverage (P3.2e).
2//
3// These tests exercise the mux-specific raw QUIC carriage over a real loopback
4// QUIC connection: quic.DialMuxContext pins the ephemeral certificate and sends
5// the single bounded token/nonce auth record, quic.Server.AcceptMux consumes it
6// exactly once, and both hand daemonmux a raw bounded transport over the one
7// bidirectional stream. The daemonmux-level parity tests live in
8// internal/adapters/daemonmux/quic_carriage_test.go.
9package quic
10
11import (
12 "bytes"
13 "context"
14 "crypto/rand"
15 "crypto/sha256"
16 "encoding/base64"
17 "encoding/hex"
18 "testing"
19 "time"
20
21 "github.com/stretchr/testify/require"
22
23 "github.com/bnema/vev/internal/ports"
24 "github.com/bnema/vev/internal/protocol/wire"
25)
26
27// muxTestDeadline bounds every real loopback QUIC mux carriage wait.
28const muxTestDeadline = 10 * time.Second
29
30// muxTestServer mints one real loopback bootstrap endpoint and returns it with
31// its readiness record.
32func muxTestServer(t *testing.T) (*Server, Readiness) {
33 t.Helper()
34 server, readiness, err := NewServer()
35 require.NoError(t, err)
36 t.Cleanup(func() { _ = server.Close() })
37 return server, readiness
38}
39
40// newMuxBootstrapServer mints one bootstrap server with an explicit QUIC
41// config, so a test can make a stream-contract violation admissible at the QUIC
42// layer and observe the lifetime guard itself.
43func newMuxBootstrapServer(t *testing.T, config Config) (*Server, Readiness) {
44 t.Helper()
45 cert, fingerprint, err := GenerateEphemeralCert()
46 require.NoError(t, err)
47 listener, err := ListenConfig("127.0.0.1:0", cert, config, bootstrapMaxPending)
48 require.NoError(t, err)
49 token := make([]byte, bootstrapTokenBytes)
50 _, err = rand.Read(token)
51 require.NoError(t, err)
52 nonce := make([]byte, bootstrapNonceBytes)
53 _, err = rand.Read(nonce)
54 require.NoError(t, err)
55 tokenVerifier := sha256.Sum256(token)
56 nonceVerifier := sha256.Sum256(nonce)
57 expires := time.Now().Add(bootstrapExpiry)
58 server := &Server{
59 listener: listener,
60 tokenVerifier: tokenVerifier[:],
61 nonceVerifier: nonceVerifier[:],
62 expires: expires,
63 fingerprint: fingerprint,
64 }
65 t.Cleanup(func() { _ = server.Close() })
66 return server, Readiness{
67 Version: bootstrapSchemaVersion,
68 Port: listenerPort(listener),
69 Token: base64.StdEncoding.EncodeToString(token),
70 Nonce: base64.StdEncoding.EncodeToString(nonce),
71 Fingerprint: hex.EncodeToString(fingerprint),
72 ExpiresAt: expires.Unix(),
73 }
74}
75
76// muxAcceptResult is one AcceptMux outcome delivered from a worker goroutine.
77type muxAcceptResult struct {
78 carriage wire.BoundedTransport
79 err error
80}
81
82// acceptMuxAsync accepts one carriage from server and delivers the outcome.
83func acceptMuxAsync(server *Server, ctx context.Context) <-chan muxAcceptResult {
84 out := make(chan muxAcceptResult, 1)
85 go func() {
86 carriage, err := server.AcceptMux(ctx)
87 out <- muxAcceptResult{carriage: carriage, err: err}
88 }()
89 return out
90}
91
92// awaitMuxAccept waits for one successful authenticated carriage.
93func awaitMuxAccept(t *testing.T, out <-chan muxAcceptResult) wire.BoundedTransport {
94 t.Helper()
95 select {
96 case result := <-out:
97 require.NoError(t, result.err)
98 require.NotNil(t, result.carriage)
99 t.Cleanup(func() { _ = result.carriage.Close() })
100 return result.carriage
101 case <-time.After(muxTestDeadline):
102 t.Fatal("AcceptMux did not return")
103 return nil
104 }
105}
106
107// muxTestPair establishes one authenticated mux carriage over a real loopback
108// QUIC connection and returns its broker and daemon ends.
109func muxTestPair(t *testing.T) (wire.BoundedTransport, wire.BoundedTransport) {
110 t.Helper()
111 server, readiness := muxTestServer(t)
112 ctx, cancel := context.WithTimeout(context.Background(), muxTestDeadline)
113 t.Cleanup(cancel)
114 accepted := acceptMuxAsync(server, ctx)
115 client, err := DialMuxContext(ctx, dialTestAddr(t, readiness), readiness, Config{}, muxTestDeadline)
116 require.NoError(t, err)
117 t.Cleanup(func() { _ = client.Close() })
118 return client, awaitMuxAccept(t, accepted)
119}
120
121// awaitBoundedReadError proves a carriage closed its bounded read: the refused
122// or lost carriage must return an error instead of yielding data.
123func awaitBoundedReadError(t *testing.T, carriage wire.BoundedTransport) error {
124 t.Helper()
125 result := make(chan error, 1)
126 go func() {
127 _, err := carriage.RecvBounded(bootstrapMaxAuthRecord)
128 result <- err
129 }()
130 select {
131 case err := <-result:
132 require.Error(t, err)
133 return err
134 case <-time.After(muxTestDeadline):
135 t.Fatal("a closed mux carriage did not fail its bounded read")
136 return nil
137 }
138}
139
140// TestMuxDialAcceptHandsOutOneAuthenticatedBoundedCarriage proves the seam
141// hands daemonmux a bounded carriage that carries traffic both ways over
142// exactly one client-initiated bidirectional QUIC stream.
143func TestMuxDialAcceptHandsOutOneAuthenticatedBoundedCarriage(t *testing.T) {
144 client, daemon := muxTestPair(t)
145
146 payload := []byte("mux carriage stays exact")
147 require.NoError(t, client.Send(wire.Envelope{Payload: payload}))
148 got, err := daemon.RecvBounded(uint64(len(payload)))
149 require.NoError(t, err)
150 require.Equal(t, payload, got.Payload)
151
152 reply := []byte("mux reply stays exact")
153 require.NoError(t, daemon.Send(wire.Envelope{Payload: reply}))
154 echo, err := client.RecvBounded(uint64(len(reply)))
155 require.NoError(t, err)
156 require.Equal(t, reply, echo.Payload)
157
158 // One physical connection carries one bidirectional stream: the first
159 // client-initiated stream, multiplexed by daemonmux, never a native stream
160 // per logical attachment.
161 brokerStream := client.(*Transport)
162 daemonStream := daemon.(*Transport)
163 require.Equal(t, brokerStream.stream.StreamID(), daemonStream.stream.StreamID())
164 require.Zero(t, uint64(brokerStream.stream.StreamID()), "the carriage is the first client-initiated bidirectional stream")
165}
166
167// TestMuxCarriageRecvBoundedNeverWidensTheFramingCeiling proves the raw
168// bounded transport daemonmux consumes really enforces the caller's ceiling
169// before allocating or reading a body.
170func TestMuxCarriageRecvBoundedNeverWidensTheFramingCeiling(t *testing.T) {
171 client, daemon := muxTestPair(t)
172
173 require.NoError(t, client.Send(wire.Envelope{Payload: []byte("larger than the caller ceiling")}))
174 _, err := daemon.RecvBounded(4)
175 require.ErrorIs(t, err, errTooLarge)
176}
177
178// TestMuxDialRejectsBadPinWithoutConsumingTheCredential proves an unpinned
179// peer is refused before authentication and that no bootstrap secret reaches
180// the error, while the one-time credential survives for the legitimate peer.
181func TestMuxDialRejectsBadPinWithoutConsumingTheCredential(t *testing.T) {
182 server, readiness := muxTestServer(t)
183 ctx, cancel := context.WithTimeout(context.Background(), muxTestDeadline)
184 defer cancel()
185
186 wrong := readiness
187 wrong.Fingerprint = hex.EncodeToString(make([]byte, sha256.Size))
188 carriage, err := DialMuxContext(ctx, dialTestAddr(t, readiness), wrong, Config{}, 5*time.Second)
189 require.Error(t, err)
190 require.Nil(t, carriage)
191 require.NotContains(t, err.Error(), readiness.Token)
192 require.NotContains(t, err.Error(), readiness.Nonce)
193 require.NotContains(t, err.Error(), readiness.Fingerprint)
194
195 server.consumedMu.Lock()
196 require.False(t, server.consumed, "a refused pin must not consume the one-time credential")
197 server.consumedMu.Unlock()
198
199 // No carriage was admitted: an accept bound to the test deadline expires
200 // without handing anything out.
201 refusedCtx, stopRefused := context.WithTimeout(context.Background(), 300*time.Millisecond)
202 defer stopRefused()
203 result := <-acceptMuxAsync(server, refusedCtx)
204 require.Nil(t, result.carriage)
205 require.Error(t, result.err)
206
207 // The legitimate pin still authenticates exactly one carriage.
208 accepted := acceptMuxAsync(server, ctx)
209 good, err := DialMuxContext(ctx, dialTestAddr(t, readiness), readiness, Config{}, 5*time.Second)
210 require.NoError(t, err)
211 t.Cleanup(func() { _ = good.Close() })
212 daemon := awaitMuxAccept(t, accepted)
213 payload := []byte("the legitimate carriage admits")
214 require.NoError(t, good.Send(wire.Envelope{Payload: payload}))
215 got, err := daemon.RecvBounded(uint64(len(payload)))
216 require.NoError(t, err)
217 require.Equal(t, payload, got.Payload)
218}
219
220// TestMuxFailedAuthClosesOnlyThatTransport proves a refused token or nonce
221// closes its own carriage, hands nothing out, and leaves the one-time
222// credential available for the legitimate peer.
223func TestMuxFailedAuthClosesOnlyThatTransport(t *testing.T) {
224 tests := []struct {
225 name string
226 mutate func(*Readiness)
227 }{
228 {
229 name: "wrong token",
230 mutate: func(r *Readiness) {
231 r.Token = base64.StdEncoding.EncodeToString(bytes.Repeat([]byte{0xA5}, bootstrapTokenBytes))
232 },
233 },
234 {
235 name: "wrong nonce",
236 mutate: func(r *Readiness) {
237 r.Nonce = base64.StdEncoding.EncodeToString(bytes.Repeat([]byte{0x5A}, bootstrapNonceBytes))
238 },
239 },
240 }
241 for _, tt := range tests {
242 t.Run(tt.name, func(t *testing.T) {
243 server, readiness := muxTestServer(t)
244 ctx, cancel := context.WithTimeout(context.Background(), muxTestDeadline)
245 defer cancel()
246
247 accepted := acceptMuxAsync(server, ctx)
248 bad := readiness
249 tt.mutate(&bad)
250 refused, err := DialMuxContext(ctx, dialTestAddr(t, readiness), bad, Config{}, 5*time.Second)
251 require.NoError(t, err, "auth is one-way; the peer learns the refusal from the closed carriage")
252 t.Cleanup(func() { _ = refused.Close() })
253 awaitBoundedReadError(t, refused)
254
255 // The credential was not consumed by the failed attempt.
256 good, err := DialMuxContext(ctx, dialTestAddr(t, readiness), readiness, Config{}, 5*time.Second)
257 require.NoError(t, err)
258 t.Cleanup(func() { _ = good.Close() })
259 daemon := awaitMuxAccept(t, accepted)
260 payload := []byte("the legitimate carriage admits")
261 require.NoError(t, good.Send(wire.Envelope{Payload: payload}))
262 got, err := daemon.RecvBounded(uint64(len(payload)))
263 require.NoError(t, err)
264 require.Equal(t, payload, got.Payload)
265 })
266 }
267}
268
269// TestMuxReplayedReadinessYieldsNoSecondCarriage proves the one-time credential
270// admits exactly one physical carriage: a replay fails closed and never
271// disturbs the carriage the legitimate peer already owns.
272func TestMuxReplayedReadinessYieldsNoSecondCarriage(t *testing.T) {
273 server, readiness := muxTestServer(t)
274 ctx, cancel := context.WithTimeout(context.Background(), muxTestDeadline)
275 defer cancel()
276
277 accepted := acceptMuxAsync(server, ctx)
278 client, err := DialMuxContext(ctx, dialTestAddr(t, readiness), readiness, Config{}, 5*time.Second)
279 require.NoError(t, err)
280 t.Cleanup(func() { _ = client.Close() })
281 daemon := awaitMuxAccept(t, accepted)
282
283 replayCtx, stopReplay := context.WithCancel(context.Background())
284 defer stopReplay()
285 replayedAccept := acceptMuxAsync(server, replayCtx)
286 replayed, err := DialMuxContext(ctx, dialTestAddr(t, readiness), readiness, Config{}, 5*time.Second)
287 require.NoError(t, err)
288 t.Cleanup(func() { _ = replayed.Close() })
289 // The server refuses and closes the replayed carriage; only then does the
290 // still-waiting accept end.
291 awaitBoundedReadError(t, replayed)
292 stopReplay()
293
294 result := <-replayedAccept
295 require.Nil(t, result.carriage, "a replayed credential must admit no second carriage")
296 require.Error(t, result.err)
297
298 // The first carriage is untouched by the replay.
299 payload := []byte("the original carriage survives a replay")
300 require.NoError(t, client.Send(wire.Envelope{Payload: payload}))
301 got, err := daemon.RecvBounded(uint64(len(payload)))
302 require.NoError(t, err)
303 require.Equal(t, payload, got.Payload)
304}
305
306// TestMuxAcceptMuxRefusesExpiredBootstrap proves an expired credential admits
307// no carriage even though the listener is still bound.
308func TestMuxAcceptMuxRefusesExpiredBootstrap(t *testing.T) {
309 server, _ := newBootstrapTestServer(t, bootstrapMaxPending, -time.Minute)
310 defer func() { _ = server.Close() }()
311 ctx, cancel := context.WithTimeout(context.Background(), muxTestDeadline)
312 defer cancel()
313
314 carriage, err := server.AcceptMux(ctx)
315 require.Nil(t, carriage)
316 require.ErrorIs(t, err, errBootstrapExpired)
317}
318
319// TestMuxCarriageRejectsExtraQuicStream proves the one-stream guard stays in
320// force for the mux carriage lifetime: a second native QUIC stream never becomes
321// a logical daemonmux attachment, it fails the whole physical carriage. The
322// listener admits two streams at the QUIC layer so the guard, not flow control,
323// is what enforces the contract.
324func TestMuxCarriageRejectsExtraQuicStream(t *testing.T) {
325 server, readiness := newMuxBootstrapServer(t, Config{
326 HandshakeIdleTimeout: bootstrapAuthDeadline,
327 MaxIncomingStreams: 2,
328 })
329 ctx, cancel := context.WithTimeout(context.Background(), muxTestDeadline)
330 defer cancel()
331
332 accepted := acceptMuxAsync(server, ctx)
333 client, err := DialMuxContext(ctx, dialTestAddr(t, readiness), readiness, Config{}, 5*time.Second)
334 require.NoError(t, err)
335 t.Cleanup(func() { _ = client.Close() })
336 daemon := awaitMuxAccept(t, accepted)
337
338 payload := []byte("the single stream stays exact")
339 require.NoError(t, client.Send(wire.Envelope{Payload: payload}))
340 got, err := daemon.RecvBounded(uint64(len(payload)))
341 require.NoError(t, err)
342 require.Equal(t, payload, got.Payload)
343
344 extra, err := client.(*Transport).conn.OpenStreamSync(ctx)
345 require.NoError(t, err, "the second stream is admissible under the explicit limit")
346 defer func() { _ = extra.Close() }()
347 _, err = extra.Write([]byte("extra stream violation"))
348 require.NoError(t, err)
349
350 awaitBoundedReadError(t, daemon)
351 require.Eventually(t, func() bool {
352 return client.(*Transport).LinkState() == ports.LinkStateOffline &&
353 daemon.(*Transport).LinkState() == ports.LinkStateOffline
354 }, muxTestDeadline, 10*time.Millisecond, "the extra stream must fail the whole carriage on both ends")
355}
356
357// TestMuxNilServerFailsClosed proves a nil bootstrap server hands out no
358// carriage.
359func TestMuxNilServerFailsClosed(t *testing.T) {
360 var server *Server
361 carriage, err := server.AcceptMux(context.Background())
362 require.ErrorIs(t, err, ErrMuxCarriage)
363 require.Nil(t, carriage)
364}
365
366// TestMuxSetupCancellationAndCarriageClose proves clean cancellation and
367// teardown: a canceled setup context hands out no carriage on either role, and
368// closing one end unblocks a blocked bounded read on the other.
369func TestMuxSetupCancellationAndCarriageClose(t *testing.T) {
370 t.Run("canceled dial", func(t *testing.T) {
371 _, readiness := muxTestServer(t)
372 ctx, cancel := context.WithCancel(context.Background())
373 cancel()
374 carriage, err := DialMuxContext(ctx, dialTestAddr(t, readiness), readiness, Config{}, 5*time.Second)
375 require.ErrorIs(t, err, context.Canceled)
376 require.Nil(t, carriage)
377 })
378
379 t.Run("canceled accept", func(t *testing.T) {
380 server, _ := muxTestServer(t)
381 ctx, cancel := context.WithCancel(context.Background())
382 cancel()
383 carriage, err := server.AcceptMux(ctx)
384 require.ErrorIs(t, err, context.Canceled)
385 require.Nil(t, carriage)
386 })
387
388 t.Run("peer close unblocks the bounded read", func(t *testing.T) {
389 client, daemon := muxTestPair(t)
390 result := make(chan error, 1)
391 go func() {
392 _, err := daemon.RecvBounded(bootstrapMaxAuthRecord)
393 result <- err
394 }()
395 require.NoError(t, client.Close())
396 select {
397 case err := <-result:
398 require.Error(t, err)
399 case <-time.After(muxTestDeadline):
400 t.Fatal("peer close did not unblock the bounded read")
401 }
402 })
403}
404
405// TestMuxCarriageErrorsCarryNoSecret proves the sealed-secret invariant for the
406// seam's error surface: no error text contains the token or nonce, whether the
407// refusal is observed by the dial or by the refused carriage's own read.
408func TestMuxCarriageErrorsCarryNoSecret(t *testing.T) {
409 server, readiness := muxTestServer(t)
410 ctx, cancel := context.WithTimeout(context.Background(), muxTestDeadline)
411 defer cancel()
412
413 wrongPin := readiness
414 wrongPin.Fingerprint = hex.EncodeToString(bytes.Repeat([]byte{0x11}, sha256.Size))
415 carriage, err := DialMuxContext(ctx, dialTestAddr(t, readiness), wrongPin, Config{}, 2*time.Second)
416 require.Nil(t, carriage)
417 require.Error(t, err)
418 require.NotContains(t, err.Error(), readiness.Token)
419 require.NotContains(t, err.Error(), readiness.Nonce)
420 require.NotContains(t, err.Error(), readiness.Fingerprint)
421
422 // A refused credential is observed by the carriage's own closed read, whose
423 // error must stay equally free of authorization material.
424 acceptCtx, stopAccept := context.WithCancel(context.Background())
425 defer stopAccept()
426 accepted := acceptMuxAsync(server, acceptCtx)
427 wrongToken := readiness
428 wrongToken.Token = base64.StdEncoding.EncodeToString(bytes.Repeat([]byte{0x22}, bootstrapTokenBytes))
429 refused, err := DialMuxContext(ctx, dialTestAddr(t, readiness), wrongToken, Config{}, 2*time.Second)
430 require.NoError(t, err)
431 t.Cleanup(func() { _ = refused.Close() })
432 readErr := awaitBoundedReadError(t, refused)
433 require.NotContains(t, readErr.Error(), readiness.Token)
434 require.NotContains(t, readErr.Error(), readiness.Nonce)
435 stopAccept()
436
437 result := <-accepted
438 require.Nil(t, result.carriage, "a refused credential admits no carriage")
439 require.Error(t, result.err)
440}