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