1// Package quic carries complete serialized Protobuf envelopes over one
2// bidirectional QUIC stream per connection. The stream is framed with the
3// shared streamframe protocol (4-byte big-endian length, no type byte);
4// typed wrapping stays in sessionwire. QUIC library types never leave
5// this package: composition sees only wire.Transport, wire.Dialer, and
6// wire.Listener.
7package quic
8
9import (
10 "context"
11 "crypto/sha256"
12 "crypto/tls"
13 "crypto/x509"
14 "errors"
15 "io"
16 "net"
17 "sync"
18 "time"
19
20 quicgo "github.com/quic-go/quic-go"
21
22 "github.com/bnema/vev/internal/adapters/streamframe"
23 "github.com/bnema/vev/internal/ports"
24 "github.com/bnema/vev/internal/protocol/wire"
25)
26
27// alpn binds the QUIC handshake to the protocol epoch: only matched peers
28// complete the TLS handshake. The value is fixed per epoch; bump the epoch
29// for an intentional clean break.
30const alpn = "vev/1"
31
32// maxSkippedProbes bounds consecutive zero-length dial probes skipped by
33// Recv. One probe is sent per Dial; anything beyond a small burst is a
34// stalling peer, not a handshake.
35const maxSkippedProbes = 4
36
37// Close codes for explicit close mapping.
38const (
39 codeClosed quicgo.ApplicationErrorCode = 0
40 codeReset quicgo.ApplicationErrorCode = 1
41)
42
43// Stream reset codes mirror the close taxonomy on the single stream.
44const streamCodeClosed quicgo.StreamErrorCode = 0
45
46var (
47 errNoStream = errors.New("quic: peer opened no stream")
48 errNotConnected = errors.New("quic: connection is closed")
49 errFingerprint = errors.New("quic: certificate fingerprint mismatch")
50)
51
52// Terminal framing errors reuse the streamframe taxonomy so callers
53// match one error set across IPC, SSH, and QUIC carriages.
54var (
55 errZeroLength = streamframe.ErrZeroLength
56 errTooLarge = streamframe.ErrTooLarge
57)
58
59// Config tunes one QUIC endpoint. Zero values select the normative
60// defaults: TLS 1.3, one peer-initiated bidirectional stream on the
61// listener and none on the dialer, no unidirectional streams, keepalive /
62// idle limits, bounded admission, no 0-RTT.
63type Config struct {
64 // KeepAlivePeriod defaults to 2s; MaxIdleTimeout defaults to 60s;
65 // HandshakeIdleTimeout defaults to 5s. The short keepalive keeps mobile
66 // CGNAT bindings open and probes the path often enough that loss
67 // recovery restarts promptly once the link returns. It does not shorten
68 // failure detection: a dead path is still declared only after
69 // MaxIdleTimeout, which stays long so one connection rides out short
70 // outages and delivers queued input without reconnecting. Both roles use
71 // it; a 2s PING is a few dozen bytes per connection.
72 KeepAlivePeriod time.Duration
73 MaxIdleTimeout time.Duration
74 HandshakeIdleTimeout time.Duration
75 // MaxIncomingStreams defaults to 1 on the listener (the dialer's
76 // single stream) and to 0 on the dialer (the server never opens
77 // streams): a zero public value always means "the role default".
78 MaxIncomingStreams int64
79 // MaxIncomingUniStreams defaults to 0 (no unidirectional streams on
80 // either role). quic-go treats 0 as "library default" and only a
81 // negative value as "none", so the zero value is translated to an
82 // explicit negative before reaching the library.
83 MaxIncomingUniStreams int64
84}
85
86// noIncomingStreams is quic-go's sentinel for "this direction is truly
87// disabled": a zero quic-go limit would instead select its 100-stream
88// library default.
89const noIncomingStreams int64 = -1
90
91// baseQUICConfig applies the transport tuning shared by both roles.
92func baseQUICConfig(config Config) *quicgo.Config {
93 keepAlive := config.KeepAlivePeriod
94 if keepAlive == 0 {
95 keepAlive = 2 * time.Second
96 }
97 maxIdle := config.MaxIdleTimeout
98 if maxIdle == 0 {
99 maxIdle = 60 * time.Second
100 }
101 handshakeIdle := config.HandshakeIdleTimeout
102 if handshakeIdle == 0 {
103 handshakeIdle = 5 * time.Second
104 }
105 return &quicgo.Config{
106 // 1200-byte QUIC packets fit IPv6's minimum 1280-byte link MTU after
107 // UDP/IP headers. This matters for VPN interfaces such as NetBird's
108 // wt0, where quic-go's larger default initial packet fails with
109 // EMSGSIZE before path MTU discovery can run.
110 InitialPacketSize: 1200,
111 KeepAlivePeriod: keepAlive,
112 MaxIdleTimeout: maxIdle,
113 HandshakeIdleTimeout: handshakeIdle,
114 MaxIncomingStreams: config.MaxIncomingStreams,
115 MaxIncomingUniStreams: config.MaxIncomingUniStreams,
116 // 0-RTT stays disabled: only DialAddr/ListenAddr (never *Early)
117 // are used, and resumption is observable via Used0RTT.
118 }
119}
120
121// serverQUICConfig is the listener role: exactly one peer-initiated
122// bidirectional stream (the dialer's) and no unidirectional streams.
123func serverQUICConfig(config Config) *quicgo.Config {
124 conf := baseQUICConfig(config)
125 if config.MaxIncomingStreams == 0 {
126 conf.MaxIncomingStreams = 1
127 }
128 if config.MaxIncomingUniStreams == 0 {
129 conf.MaxIncomingUniStreams = noIncomingStreams
130 }
131 return conf
132}
133
134// clientQUICConfig is the dialer role: the server never opens streams, so
135// both incoming directions are truly disabled by default. The dialer's own
136// outgoing stream is unaffected by the incoming limits.
137func clientQUICConfig(config Config) *quicgo.Config {
138 conf := baseQUICConfig(config)
139 if config.MaxIncomingStreams == 0 {
140 conf.MaxIncomingStreams = noIncomingStreams
141 }
142 if config.MaxIncomingUniStreams == 0 {
143 conf.MaxIncomingUniStreams = noIncomingStreams
144 }
145 return conf
146}
147
148// serverTLSConfig presents cert and requires TLS 1.3 with the epoch ALPN.
149// Client certificates are not used: authentication is the SHA-256 pin
150// checked by the dialer plus the bootstrap token consumed during authentication.
151func serverTLSConfig(cert tls.Certificate) *tls.Config {
152 return &tls.Config{
153 Certificates: []tls.Certificate{cert},
154 MinVersion: tls.VersionTLS13,
155 NextProtos: []string{alpn},
156 }
157}
158
159// clientTLSConfig skips hostname verification: the dialer pins the exact
160// SHA-256 of the server certificate instead. TLS 1.3 and the epoch ALPN
161// are still enforced by the handshake.
162func clientTLSConfig(serverName string, fingerprint []byte) *tls.Config {
163 return &tls.Config{
164 ServerName: serverName,
165 MinVersion: tls.VersionTLS13,
166 NextProtos: []string{alpn},
167 InsecureSkipVerify: true,
168 VerifyPeerCertificate: func(rawCerts [][]byte, _ [][]*x509.Certificate) error {
169 if len(rawCerts) == 0 {
170 return errFingerprint
171 }
172 sum := sha256.Sum256(rawCerts[0])
173 if string(sum[:]) != string(fingerprint) {
174 return errFingerprint
175 }
176 return nil
177 },
178 }
179}
180
181// Transport is one QUIC connection's single bidirectional stream,
182// framed as wire.Envelopes. Close is safe concurrently with Send/Recv
183// and unblocks both.
184type Transport struct {
185 conn *quicgo.Conn
186 stream *quicgo.Stream
187 framer *streamframe.Framer
188 // packetConn is the dialer-owned local socket behind conn; the listener
189 // owns its own socket and passes nil.
190 packetConn net.PacketConn
191
192 observer ports.SerializedRuntimeObserver
193
194 // Observation bookkeeping mirrors the IPC transport: an accepted
195 // operation always closes its mark pair, and Close stops accepting new
196 // marks and then waits for in-flight ones to finish.
197 operationMu sync.Mutex
198 operationCount int
199 observedClosing bool
200 operationsDone chan struct{}
201
202 linkMu sync.Mutex
203 linkState ports.LinkState
204 events chan ports.LinkEvent
205
206 closeOnce sync.Once
207 terminalOnce sync.Once
208 // teardownDone closes when the deferred hard close after an orderly
209 // Close has run.
210 teardownDone chan struct{}
211 closeErr error
212}
213
214var (
215 _ wire.Transport = (*Transport)(nil)
216 _ wire.AsyncTransport = (*Transport)(nil)
217 _ wire.BoundedTransport = (*Transport)(nil)
218 _ ports.LinkStateReporter = (*Transport)(nil)
219)
220
221// newTransport owns one connected stream. The caller must have enforced
222// the single-stream contract already. observer may be nil; when set it
223// receives process-local adapter marks for send and receive.
224func newTransport(conn *quicgo.Conn, stream *quicgo.Stream, packetConn net.PacketConn, observer ports.SerializedRuntimeObserver) *Transport {
225 t := &Transport{
226 conn: conn,
227 stream: stream,
228 packetConn: packetConn,
229 observer: observer,
230 linkState: ports.LinkStateConnected,
231 events: make(chan ports.LinkEvent, 4),
232 teardownDone: make(chan struct{}),
233 }
234 t.framer = streamframe.NewFramer(stream, stream, t.closeStream)
235 // The single-stream contract is enforced for the whole connection
236 // lifetime by quic-go's role limits (no unidirectional streams, no
237 // server-initiated streams) plus the guard watcher. There is no
238 // startup grace window: it would tax every connection while the guard
239 // already fails a violation whenever it lands.
240 guardStreamContract(conn)
241 // Link-state honesty: a single QUIC stream cannot derive a reliable
242 // degraded/probing transition, so the adapter reports only the
243 // terminal Offline transition and closes the event stream when the
244 // connection dies on its own.
245 go func() {
246 <-conn.Context().Done()
247 if t.packetConn != nil {
248 _ = t.packetConn.Close()
249 }
250 t.signalTerminal(linkEventError(context.Cause(conn.Context())))
251 }()
252 return t
253}
254
255// guardStreamContract closes the connection, for the connection lifetime,
256// the moment the peer opens anything beyond the single client-initiated
257// bidirectional stream: a second bidirectional stream or any
258// unidirectional stream. quic-go's role limits should prevent both
259// entirely; the guard is defense in depth against an explicit limit.
260//
261// The acceptors are deliberately never canceled: quic-go's AcceptStream may
262// hand a queued stream to an already-canceled waiter, so an abandoned
263// acceptor would silently swallow a delayed violation.
264func guardStreamContract(conn *quicgo.Conn) {
265 go func() {
266 bidi := make(chan error, 1)
267 go func() {
268 _, err := conn.AcceptStream(context.Background())
269 bidi <- err
270 }()
271 uni := make(chan error, 1)
272 go func() {
273 _, err := conn.AcceptUniStream(context.Background())
274 uni <- err
275 }()
276 select {
277 case err := <-bidi:
278 if err == nil {
279 _ = conn.CloseWithError(codeReset, "extra bidirectional stream")
280 }
281 case err := <-uni:
282 if err == nil {
283 _ = conn.CloseWithError(codeReset, "forbidden unidirectional stream")
284 }
285 }
286 }()
287}
288
289// Send queues the envelope for the sole writer and waits for its wire attempt.
290func (t *Transport) Send(envelope wire.Envelope) error {
291 end := t.observe(ports.RuntimeAdapterSendStart, uint64(len(envelope.Payload)))
292 err := t.framer.Send(envelope.Payload)
293 end(err == nil)
294 if err != nil {
295 return mapFramerError(err)
296 }
297 return nil
298}
299
300// SendAsync accepts the envelope for ordered background transmission.
301func (t *Transport) SendAsync(envelope wire.Envelope) error {
302 end := t.observe(ports.RuntimeAdapterSendStart, uint64(len(envelope.Payload)))
303 err := t.framer.SendAsync(envelope.Payload)
304 end(err == nil)
305 if err != nil {
306 return mapFramerError(err)
307 }
308 return nil
309}
310
311// observe opens one process-local adapter mark pair, or returns a no-op end
312// when observation is disabled or Close has already begun. An accepted
313// operation always emits its end mark, so Close can safely wait for it.
314func (t *Transport) observe(start ports.RuntimeMarkKind, bytes uint64) func(bool) {
315 if t.observer == nil {
316 return func(bool) {}
317 }
318 if !t.beginObservedOperation() {
319 return func(bool) {}
320 }
321 correlation := ports.NewRuntimeCorrelation()
322 t.observer.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("quic", correlation, start, bytes, true))
323 end := ports.RuntimeAdapterSendEnd
324 if start == ports.RuntimeAdapterReceiveStart {
325 end = ports.RuntimeAdapterReceiveEnd
326 }
327 return func(valid bool) {
328 defer t.finishObservedOperation()
329 t.observer.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("quic", correlation, end, bytes, valid))
330 }
331}
332
333func (t *Transport) beginObservedOperation() bool {
334 t.operationMu.Lock()
335 defer t.operationMu.Unlock()
336 if t.observedClosing {
337 return false
338 }
339 if t.operationCount == 0 {
340 t.operationsDone = make(chan struct{})
341 }
342 t.operationCount++
343 return true
344}
345
346func (t *Transport) finishObservedOperation() {
347 t.operationMu.Lock()
348 defer t.operationMu.Unlock()
349 t.operationCount--
350 if t.operationCount == 0 {
351 close(t.operationsDone)
352 }
353}
354
355// beginShutdown stops accepting new observed operations and returns a channel
356// closed when the in-flight ones finish, or nil when none were running.
357func (t *Transport) beginShutdown() <-chan struct{} {
358 t.operationMu.Lock()
359 defer t.operationMu.Unlock()
360 t.observedClosing = true
361 if t.operationCount == 0 {
362 return nil
363 }
364 return t.operationsDone
365}
366
367// Recv reads one complete envelope, blocking until arrival or close.
368// The dial probe (a framed zero-length envelope) is skipped at most
369// maxSkippedProbes times: it proves stream liveness without reaching
370// application decoders, but an unbounded skip would let a peer stall
371// Recv forever without progress.
372func (t *Transport) Recv() (wire.Envelope, error) {
373 return t.recvSkippingProbes(t.framer.Recv)
374}
375
376// RecvBounded reads exactly one framed envelope without allocating a
377// payload larger than limit: the four-byte length prefix is validated
378// before any payload buffer exists. It backs the bootstrap auth record so
379// an unauthenticated peer cannot force a large pre-auth allocation by
380// advertising a huge frame and stalling. Zero-length dial probes are
381// skipped exactly like Recv, and the shared streamframe framer enforces the
382// same framing as every other carriage. It reads through the transport's one
383// framer, so it must only be used before regular Send/Recv traffic begins.
384func (t *Transport) RecvBounded(limit uint64) (wire.Envelope, error) {
385 return t.recvSkippingProbes(func() ([]byte, error) { return t.framer.RecvBounded(limit) })
386}
387
388// recvSkippingProbes observes one receive operation and runs read for exactly
389// one envelope, skipping the framed zero-length dial probe at most
390// maxSkippedProbes times. Both Recv and RecvBounded share it, so the bounded
391// path adds only the per-call ceiling and never a second framing copy.
392func (t *Transport) recvSkippingProbes(read func() ([]byte, error)) (wire.Envelope, error) {
393 end := t.observe(ports.RuntimeAdapterReceiveStart, 0)
394 for skipped := 0; ; skipped++ {
395 payload, err := read()
396 if errors.Is(err, streamframe.ErrZeroLength) {
397 if skipped >= maxSkippedProbes {
398 end(false)
399 return wire.Envelope{}, errZeroLength
400 }
401 continue
402 }
403 if err != nil {
404 end(false)
405 return wire.Envelope{}, mapFramerError(err)
406 }
407 end(true)
408 return wire.Envelope{Payload: payload}, nil
409 }
410}
411
412// zeroBytes best-effort erases a mutable secret buffer.
413func zeroBytes(buf []byte) {
414 for i := range buf {
415 buf[i] = 0
416 }
417}
418
419// LinkState reports transport connectivity only, never semantic
420// handshake acceptance.
421func (t *Transport) LinkState() ports.LinkState {
422 t.linkMu.Lock()
423 defer t.linkMu.Unlock()
424 return t.linkState
425}
426
427// LinkEvents returns the best-effort link event stream. The adapter derives
428// one terminal Offline event on connection failure or Close and then closes
429// the channel; no degraded/probing transitions are invented.
430func (t *Transport) LinkEvents() <-chan ports.LinkEvent { return t.events }
431
432// gracefulCloseWindow bounds how long an orderly close keeps the connection
433// alive after its FIN so an already-written final envelope can be transmitted
434// before CONNECTION_CLOSE. Close itself never waits on it.
435const gracefulCloseWindow = time.Second
436
437// Close performs an orderly teardown: it stops the writer, FINs the send side
438// instead of RESET_STREAMing it, and defers the hard connection close so an
439// envelope already handed to the stream (the final envelope of an orderly
440// close) is transmitted rather than discarded. It publishes the terminal
441// Offline link event and closes the event stream.
442func (t *Transport) Close() error {
443 t.closeOnce.Do(func() {
444 // Stop accepting observed operation marks and wait for in-flight
445 // ones to close their pairs before returning.
446 operationsDone := t.beginShutdown()
447 // Publish the graceful terminal event before closing the
448 // connection so a local Close reports a nil cause instead of
449 // racing the peer-failure watcher.
450 t.signalTerminal(nil)
451 // Start the bounded hard close first: if the writer is blocked on a
452 // peer that stopped reading, it is what unblocks the FIN below.
453 go t.deferCloseConnection()
454 // framer.Close stops new sends, unblocks a blocked Recv through the
455 // injected closer, and waits for the writer to finish. It must run
456 // before the FIN so an in-flight write is not cut short.
457 t.closeErr = t.framer.Close()
458 _ = t.stream.Close()
459 if operationsDone != nil {
460 <-operationsDone
461 }
462 })
463 return t.closeErr
464}
465
466// WaitGracefulTeardown blocks until the deferred hard connection close that
467// follows an orderly Close has run: the graceful window elapsed or the peer
468// closed first. A process-owning caller (the broker mux QUIC proxy) waits after Close so
469// the final synchronous envelope is not discarded by process exit; ctx bounds
470// the wait. It returns ctx.Err() when the wait is cut short, and blocks until
471// ctx expires when Close was never called.
472func (t *Transport) WaitGracefulTeardown(ctx context.Context) error {
473 select {
474 case <-t.teardownDone:
475 return nil
476 case <-ctx.Done():
477 return ctx.Err()
478 }
479}
480
481// deferCloseConnection keeps the connection alive long enough to transmit a
482// FIN and any stream data queued behind it, then hard-closes unless the peer
483// already did. Closing the connection first would send CONNECTION_CLOSE ahead
484// of the buffered stream data and discard it.
485func (t *Transport) deferCloseConnection() {
486 defer close(t.teardownDone)
487 timer := time.NewTimer(gracefulCloseWindow)
488 defer timer.Stop()
489 select {
490 case <-t.conn.Context().Done():
491 case <-timer.C:
492 _ = t.conn.CloseWithError(codeClosed, "closed")
493 }
494}
495
496// signalTerminal publishes the terminal Offline event and closes the event
497// stream exactly once, whether the transport ended via Close or via the
498// peer. Publishing never blocks transport progress: when the consumer is
499// slow the event is dropped and the closed channel plus LinkState still
500// report Offline.
501func (t *Transport) signalTerminal(err error) {
502 t.terminalOnce.Do(func() {
503 t.linkMu.Lock()
504 t.linkState = ports.LinkStateOffline
505 t.linkMu.Unlock()
506 select {
507 case t.events <- ports.LinkEvent{State: ports.LinkStateOffline, At: time.Now(), Err: err}:
508 default:
509 }
510 close(t.events)
511 })
512}
513
514// linkEventError keeps LinkEvent.Err boundary-safe: QUIC library types never
515// leave this package.
516func linkEventError(err error) error {
517 if err == nil {
518 return nil
519 }
520 var appErr *quicgo.ApplicationError
521 var streamErr *quicgo.StreamError
522 var idleErr *quicgo.IdleTimeoutError
523 var transportErr *quicgo.TransportError
524 if errors.As(err, &appErr) || errors.As(err, &streamErr) ||
525 errors.As(err, &idleErr) || errors.As(err, &transportErr) {
526 return io.EOF
527 }
528 return err
529}
530
531// closeStream is the framer's injected closer: it unblocks a blocked Recv
532// without touching the send side, which Close() FINs only after the writer
533// has drained.
534func (t *Transport) closeStream() error {
535 t.stream.CancelRead(streamCodeClosed)
536 return nil
537}
538
539func mapFramerError(err error) error {
540 switch {
541 case err == nil:
542 return nil
543 case errors.Is(err, streamframe.ErrZeroLength):
544 return errZeroLength
545 case errors.Is(err, streamframe.ErrTooLarge):
546 return errTooLarge
547 case errors.Is(err, streamframe.ErrClosed):
548 return errNotConnected
549 default:
550 var appErr *quicgo.ApplicationError
551 if errors.As(err, &appErr) {
552 // QUIC library types never leave this package: a remote
553 // or local application close reads as a terminal EOF so
554 // callers classify graceful shutdown like IPC/SSH.
555 return io.EOF
556 }
557 var streamErr *quicgo.StreamError
558 if errors.As(err, &streamErr) {
559 return io.EOF
560 }
561 return err
562 }
563}