1package quic
2
3import (
4 "context"
5 "crypto/tls"
6 "errors"
7 "fmt"
8 "net"
9 "net/netip"
10 "strconv"
11 "sync"
12 "time"
13
14 quicgo "github.com/quic-go/quic-go"
15
16 "github.com/bnema/vev/internal/ports"
17 "github.com/bnema/vev/internal/protocol/wire"
18)
19
20// Dialer establishes one QUIC connection and opens its single bidirectional
21// stream. Authentication is the exact SHA-256 certificate pin; TLS 1.3 and
22// the epoch ALPN are enforced by the handshake. 0-RTT is never used.
23//
24// The dialer admits no peer-initiated stream at all: the server role never
25// opens streams, so both incoming directions are truly disabled with the
26// quic-go negative-limit semantics. Any violation that still arrives is
27// also enforced for the connection lifetime by the stream-contract guard.
28type Dialer struct {
29 addr string
30 serverName string
31 fingerprint []byte
32 config *quicgo.Config
33 timeout time.Duration
34 observer ports.SerializedRuntimeObserver
35 // resolve is the address-resolution seam; nil selects the context-aware
36 // default resolver. Multi-address fallback is asserted through it without
37 // depending on real DNS answers or black-holed routes.
38 resolve func(ctx context.Context, addr string) ([]*net.UDPAddr, error)
39}
40
41var _ wire.Dialer = (*Dialer)(nil)
42
43// Option tunes a dial.
44type Option func(*dialOptions)
45
46type dialOptions struct {
47 observer ports.SerializedRuntimeObserver
48}
49
50// DialConfig carries the dial parameters. Fingerprint is the exact
51// SHA-256 of the server certificate (32 bytes). Timeout bounds the whole
52// dial: address resolution, every candidate connection, stream open, and
53// the liveness probe share one deadline; zero selects 15s, negative is
54// rejected at Dial time.
55func DialConfig(addr, serverName string, fingerprint []byte, config Config, timeout time.Duration, opts ...Option) Dialer {
56 if timeout == 0 {
57 timeout = 15 * time.Second
58 }
59 var options dialOptions
60 for _, opt := range opts {
61 if opt != nil {
62 opt(&options)
63 }
64 }
65 return Dialer{
66 addr: addr,
67 serverName: serverName,
68 fingerprint: append([]byte(nil), fingerprint...),
69 config: clientQUICConfig(config),
70 timeout: timeout,
71 observer: options.observer,
72 }
73}
74
75// Dial connects, rejects 0-RTT resumption, and opens exactly one
76// bidirectional stream. The stream liveness probe makes the stream visible
77// to the peer's AcceptStream; the dialer does not pay a startup grace window
78// after it. Extra or unidirectional streams fail closed: quic-go's role
79// limits refuse them before the handshake and the transport's lifetime guard
80// kills the connection on any violation that still lands.
81func (d Dialer) Dial(ctx context.Context) (wire.Transport, error) {
82 if len(d.fingerprint) != 32 {
83 return nil, errFingerprint
84 }
85 if d.addr == "" {
86 return nil, errors.New("quic: empty dial address")
87 }
88 if d.timeout < 0 {
89 return nil, errors.New("quic: negative dial timeout")
90 }
91 // One shared deadline bounds resolution, every candidate address, stream
92 // open, and the probe, so the documented timeout is truly
93 // whole-operation rather than per-step.
94 ctx, cancel := context.WithTimeout(ctx, d.timeout)
95 defer cancel()
96 tlsConf := clientTLSConfig(d.serverName, d.fingerprint)
97
98 addrs, err := d.resolveAddrs(ctx)
99 if err != nil {
100 return nil, err
101 }
102
103 var lastErr error
104 for i, addr := range addrs {
105 if err := ctx.Err(); err != nil {
106 if lastErr != nil {
107 return nil, lastErr
108 }
109 return nil, err
110 }
111 // Split the shared deadline across the remaining candidates so a
112 // black-holed address cannot starve the rest of the resolved set.
113 budget := time.Duration(0)
114 if remaining := len(addrs) - i; remaining > 1 {
115 if deadline, ok := ctx.Deadline(); ok {
116 if slice := time.Until(deadline) / time.Duration(remaining); slice > 0 {
117 budget = slice
118 }
119 }
120 }
121 conn, packetConn, err := d.dialCandidate(ctx, addr, tlsConf, budget)
122 if err != nil {
123 lastErr = err
124 continue
125 }
126 transport, err := d.finishDial(ctx, conn, packetConn)
127 if err != nil {
128 lastErr = err
129 _ = conn.CloseWithError(codeReset, "dial failed")
130 _ = packetConn.Close()
131 continue
132 }
133 return transport, nil
134 }
135 if lastErr != nil {
136 return nil, lastErr
137 }
138 return nil, ctx.Err()
139}
140
141// dialCandidate opens one QUIC connection to a single resolved address. A
142// positive budget carves a slice from the shared deadline so a black-holed
143// candidate cannot consume the whole window. The returned socket is owned by
144// the caller on success and closed here on failure.
145func (d Dialer) dialCandidate(ctx context.Context, addr *net.UDPAddr, tlsConf *tls.Config, budget time.Duration) (*quicgo.Conn, net.PacketConn, error) {
146 if budget > 0 {
147 attemptCtx, cancel := context.WithTimeout(ctx, budget)
148 defer cancel()
149 ctx = attemptCtx
150 }
151 packetConn, err := listenUDPFor(addr)
152 if err != nil {
153 return nil, nil, err
154 }
155 conn, err := quicgo.Dial(ctx, packetConn, addr, tlsConf, d.config)
156 if err != nil {
157 _ = packetConn.Close()
158 return nil, nil, err
159 }
160 return conn, packetConn, nil
161}
162
163// finishDial rejects 0-RTT resumption, opens the single stream under the
164// shared deadline, and sends the liveness probe. packetConn is the
165// caller-created local socket backing conn; the returned transport owns it.
166// On error the caller owns closing conn and packetConn.
167func (d Dialer) finishDial(ctx context.Context, conn *quicgo.Conn, packetConn net.PacketConn) (wire.Transport, error) {
168 if conn.ConnectionState().Used0RTT {
169 return nil, errors.New("quic: 0-RTT resumption refused")
170 }
171 stream, err := conn.OpenStreamSync(ctx)
172 if err != nil {
173 return nil, err
174 }
175 // The server observes the connection once this stream carries data:
176 // quic-go only surfaces the accepted connection after the peer's
177 // AcceptStream fires, which requires stream activity. The probe is a
178 // framed zero-length envelope: the receiver's scanner reports
179 // ErrZeroLength for it, and both Recv paths skip that error without
180 // reaching application decoders, so the stream stays usable for
181 // application envelopes afterwards.
182 if _, err := stream.Write([]byte{0, 0, 0, 0}); err != nil {
183 return nil, err
184 }
185 return newTransport(conn, stream, packetConn, d.observer), nil
186}
187
188// listenUDPFor binds a local socket matching the candidate's address family.
189// quic-go's Dial takes ownership semantics from the caller (unlike DialAddr),
190// so the returned socket must be closed with the connection.
191func listenUDPFor(remote *net.UDPAddr) (*net.UDPConn, error) {
192 local := &net.UDPAddr{IP: net.IPv6unspecified, Port: 0}
193 if remote.IP.To4() != nil {
194 local = &net.UDPAddr{IP: net.IPv4zero, Port: 0}
195 }
196 return net.ListenUDP("udp", local)
197}
198
199// resolveAddrs turns the configured dial address into one or more concrete
200// UDP addresses within ctx.
201func (d Dialer) resolveAddrs(ctx context.Context) ([]*net.UDPAddr, error) {
202 if d.resolve != nil {
203 return d.resolve(ctx, d.addr)
204 }
205 return resolveQUICAddrs(ctx, d.addr)
206}
207
208// resolveQUICAddrs resolves addr with the context-aware default resolver.
209// A literal IP yields exactly one address; a hostname yields every resolved
210// address (deduplicated) so the dialer can fail over between them. The
211// lookup is bounded by ctx, unlike quic-go's own unbounded DialAddr
212// resolution.
213func resolveQUICAddrs(ctx context.Context, addr string) ([]*net.UDPAddr, error) {
214 host, port, err := net.SplitHostPort(addr)
215 if err != nil {
216 return nil, fmt.Errorf("quic: invalid dial address %q: %w", addr, err)
217 }
218 portNum, err := strconv.Atoi(port)
219 if err != nil || portNum <= 0 || portNum > 65535 {
220 return nil, fmt.Errorf("quic: invalid dial port %q", port)
221 }
222 if ip := net.ParseIP(host); ip != nil {
223 return []*net.UDPAddr{{IP: ip, Port: portNum}}, nil
224 }
225 ips, err := net.DefaultResolver.LookupNetIP(ctx, "ip", host)
226 if err != nil {
227 return nil, err
228 }
229 addrs := make([]*net.UDPAddr, 0, len(ips))
230 seen := make(map[netip.Addr]struct{}, len(ips))
231 for _, ip := range ips {
232 ip = ip.Unmap()
233 if _, ok := seen[ip]; ok {
234 continue
235 }
236 seen[ip] = struct{}{}
237 addrs = append(addrs, &net.UDPAddr{IP: net.IP(ip.AsSlice()), Port: portNum})
238 }
239 if len(addrs) == 0 {
240 return nil, fmt.Errorf("quic: no addresses for host %q", host)
241 }
242 return addrs, nil
243}
244
245// Listener accepts QUIC connections with bounded admission: at most
246// maxPending unauthenticated connections wait; overflow fails fast.
247type Listener struct {
248 listener *quicgo.Listener
249 config Config
250 cert tls.Certificate
251 addr string
252 observer ports.SerializedRuntimeObserver
253 pendingMu sync.Mutex
254 pending int
255 maxPending int
256
257 closeOnce sync.Once
258 closeErr error
259}
260
261var _ wire.Listener = (*Listener)(nil)
262
263// ListenerOption tunes a listener.
264type ListenerOption func(*Listener)
265
266// WithListenerRuntimeObserver enables process-local adapter marks on every
267// accepted transport, mirroring the dial-side WithRuntimeObserver. A nil
268// observer leaves observation disabled.
269func WithListenerRuntimeObserver(observer ports.SerializedRuntimeObserver) ListenerOption {
270 return func(l *Listener) { l.observer = observer }
271}
272
273// ListenConfig bounds admission and validates its inputs. Empty addr
274// selects loopback (never the wildcard); negative durations and stream
275// limits are rejected. MaxPendingUnauthenticated defaults to 4 per the
276// normative table; the generic accept budget defaults to 5s. The bootstrap
277// server passes its own 3s auth deadline as that budget and sets the
278// matching handshake idle timeout, so both bounds agree there. A zero stream
279// limit means "no streams", translated to quic-go's negative semantics
280// inside the role-specific config.
281func ListenConfig(addr string, cert tls.Certificate, config Config, maxPending int, opts ...ListenerOption) (*Listener, error) {
282 if addr == "" {
283 addr = "127.0.0.1:0"
284 }
285 if config.KeepAlivePeriod < 0 || config.MaxIdleTimeout < 0 || config.HandshakeIdleTimeout < 0 {
286 return nil, errors.New("quic: negative timeout")
287 }
288 if config.MaxIncomingStreams < 0 || config.MaxIncomingUniStreams < 0 {
289 return nil, errors.New("quic: negative stream limit")
290 }
291 if maxPending < 0 {
292 return nil, errors.New("quic: negative admission bound")
293 }
294 if maxPending == 0 {
295 maxPending = 4
296 }
297 listener, err := quicgo.ListenAddr(addr, serverTLSConfig(cert), serverQUICConfig(config))
298 if err != nil {
299 return nil, err
300 }
301 l := &Listener{
302 listener: listener,
303 config: config,
304 cert: cert,
305 addr: listener.Addr().String(),
306 maxPending: maxPending,
307 }
308 for _, opt := range opts {
309 if opt != nil {
310 opt(l)
311 }
312 }
313 return l, nil
314}
315
316// admitted is one connection whose admission slot is still held. deadline
317// bounds stream open and authentication together; release MUST be called
318// exactly once when the caller finishes authenticating (success or
319// failure) so the unauthenticated admission bound is never released while
320// a peer still holds an unauthenticated connection.
321type admitted struct {
322 transport *Transport
323 release func()
324 deadline time.Time
325}
326
327// Accept admits one connection, enforces exactly one peer-initiated
328// bidirectional stream, and rejects extra or unidirectional streams. The
329// admission slot is released once the stream contract is established:
330// this generic listener performs no authentication of its own. Callers
331// that authenticate after accepting (the bootstrap server) use admitOne
332// and hold the slot across authentication.
333func (l *Listener) Accept() (wire.Transport, error) {
334 admittedConn, err := l.admitOne(context.Background(), acceptStreamBudget)
335 if err != nil {
336 return nil, err
337 }
338 admittedConn.release()
339 return admittedConn.transport, nil
340}
341
342// acceptStreamBudget bounds generic Accept's wait for the peer's stream.
343const acceptStreamBudget = 5 * time.Second
344
345// admitOne admits one connection and holds its admission slot until
346// release is called. The per-connection budget starts when the connection
347// is admitted and covers stream open and authentication together, so a
348// peer cannot occupy an unauthenticated slot past that deadline. When the
349// stream contract is violated the slot is released and the next connection
350// is admitted instead.
351func (l *Listener) admitOne(ctx context.Context, budget time.Duration) (admitted, error) {
352 if budget <= 0 {
353 budget = acceptStreamBudget
354 }
355 for {
356 if err := ctx.Err(); err != nil {
357 return admitted{}, err
358 }
359 conn, err := l.listener.Accept(ctx)
360 if err != nil {
361 return admitted{}, err
362 }
363 if !l.admit() {
364 _ = conn.CloseWithError(codeReset, "admission full")
365 continue
366 }
367 deadline := time.Now().Add(budget)
368 streamCtx, cancel := context.WithDeadline(ctx, deadline)
369 transport, err := acceptSingleStream(streamCtx, conn, l.observer)
370 cancel()
371 if err != nil {
372 l.release()
373 _ = conn.CloseWithError(codeReset, "stream contract")
374 continue
375 }
376 var once sync.Once
377 return admitted{
378 transport: transport,
379 deadline: deadline,
380 release: func() { once.Do(l.release) },
381 }, nil
382 }
383}
384
385func (l *Listener) admit() bool {
386 l.pendingMu.Lock()
387 defer l.pendingMu.Unlock()
388 if l.pending >= l.maxPending {
389 return false
390 }
391 l.pending++
392 return true
393}
394
395func (l *Listener) release() {
396 l.pendingMu.Lock()
397 defer l.pendingMu.Unlock()
398 if l.pending > 0 {
399 l.pending--
400 }
401}
402
403// acceptSingleStream takes the peer's single bidirectional stream and
404// fails when the peer opens none or the caller's deadline expires. The
405// transport's lifetime guard enforces the stream contract on every later
406// stream.
407func acceptSingleStream(ctx context.Context, conn *quicgo.Conn, observer ports.SerializedRuntimeObserver) (*Transport, error) {
408 stream, err := conn.AcceptStream(ctx)
409 if err != nil {
410 if ctx.Err() != nil {
411 return nil, errNoStream
412 }
413 return nil, err
414 }
415 if stream == nil {
416 return nil, errNoStream
417 }
418 return newTransport(conn, stream, nil, observer), nil
419}
420
421// Close stops the listener; established transports are unaffected.
422func (l *Listener) Close() error {
423 l.closeOnce.Do(func() {
424 l.closeErr = l.listener.Close()
425 })
426 return l.closeErr
427}
428
429// Addr returns the bound address.
430func (l *Listener) Addr() string { return l.addr }