diff --git a/core/dnsserver/server_quic.go b/core/dnsserver/server_quic.go index e471a4c0e..07d4334af 100644 --- a/core/dnsserver/server_quic.go +++ b/core/dnsserver/server_quic.go @@ -8,6 +8,7 @@ import ( "fmt" "io" "net" + "os" "time" "github.com/coredns/coredns/plugin/metrics/vars" @@ -225,7 +226,7 @@ func (s *ServerQUIC) serveQUICStream(stream *quic.Stream, conn *quic.Conn) { // server's read timeout (the same deadline used for reading a query on // TCP), so a stalled stream cannot hold a worker forever. A deadline // hit surfaces as a read error handled by the existing error path below, - // which closes the connection and frees the worker. + // which frees the worker by cancelling just this stream. if s.ReadTimeout != 0 { _ = stream.SetReadDeadline(time.Now().Add(s.ReadTimeout)) } @@ -236,6 +237,16 @@ func (s *ServerQUIC) serveQUICStream(stream *quic.Stream, conn *quic.Conn) { // the STREAM FIN indicating that there will be no data to read // anymore from this stream. if err != nil && err != io.EOF { + if isTransientStreamError(err) { + // Abandon just this stream, not the whole connection (RFC 9250 + // §4.3.3). Only RESET_STREAM (CancelWrite): STOP_SENDING is + // client-only (§4.3.1) and would itself force a connection abort. + stream.CancelWrite(quic.StreamErrorCode(DoQCodeInternalError)) + s.countResponse(DoQCodeInternalError) + + return + } + s.closeQUICConn(conn, DoQCodeProtocolError) return @@ -424,6 +435,27 @@ func readDOQMessage(r io.Reader) ([]byte, error) { return buf, err } +// isTransientStreamError reports whether err reflects a condition scoped to +// a single QUIC stream — the server's own read deadline expiring, or the +// peer resetting just that stream — rather than a DoQ message-framing +// violation by the peer. RFC 9250 §4.3.3 requires the latter to abort the +// whole connection; the former must not, since DoQ multiplexes many +// independent queries as separate streams on one connection. +// +// A deadline timeout is identified specifically via os.ErrDeadlineExceeded +// (what stream.SetReadDeadline produces) rather than the broader net.Error +// Timeout() check, because connection-level failures such as +// quic.IdleTimeoutError also report Timeout() == true but must still take +// the existing connection-closing path. +func isTransientStreamError(err error) bool { + if errors.Is(err, os.ErrDeadlineExceeded) { + return true + } + + var streamErr *quic.StreamError + return errors.As(err, &streamErr) +} + // isExpectedErr returns true if err is an expected error, likely related to // the current implementation. func (s *ServerQUIC) isExpectedErr(err error) bool { diff --git a/core/dnsserver/server_quic_test.go b/core/dnsserver/server_quic_test.go index b40e9adcb..959d069e7 100644 --- a/core/dnsserver/server_quic_test.go +++ b/core/dnsserver/server_quic_test.go @@ -9,7 +9,10 @@ import ( "crypto/x509" "crypto/x509/pkix" "errors" + "fmt" + "io" "math/big" + "os" "strings" "testing" "time" @@ -233,6 +236,73 @@ func TestServerQUIC_IsExpectedErr(t *testing.T) { } } +// TestIsTransientStreamError pins down the classification isTransientStreamError +// makes: only conditions local to a single stream (this server's own read +// deadline expiring, or the peer resetting just that stream) may bypass +// tearing down the whole connection. Everything else - including a DoQ +// framing violation, or a connection-level failure that also happens to +// report Timeout() == true - must still be treated as connection-fatal. +func TestIsTransientStreamError(t *testing.T) { + tests := []struct { + name string + err error + expected bool + }{ + { + name: "read deadline exceeded", + err: os.ErrDeadlineExceeded, + expected: true, + }, + { + name: "wrapped read deadline exceeded", + err: fmt.Errorf("stream read: %w", os.ErrDeadlineExceeded), + expected: true, + }, + { + name: "peer reset just this stream", + err: &quic.StreamError{ErrorCode: 2}, + expected: true, + }, + { + name: "connection idle timeout is not stream-scoped", + // quic.IdleTimeoutError.Timeout() is also true, but it signals + // the whole connection died, not a local per-stream condition, + // so it must NOT be classified as transient. + err: &quic.IdleTimeoutError{}, + expected: false, + }, + { + name: "connection-level application error", + err: &quic.ApplicationError{ErrorCode: 2}, + expected: false, + }, + { + name: "premature STREAM FIN mid-message is a framing violation", + err: io.ErrUnexpectedEOF, + expected: false, + }, + { + name: "unsupported DoQ version framing error", + err: fmt.Errorf("message size is 0: probably unsupported DoQ version"), + expected: false, + }, + { + name: "unrelated error", + err: errors.New("some other error"), + expected: false, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + result := isTransientStreamError(tt.err) + if result != tt.expected { + t.Errorf("isTransientStreamError(%v) = %v, want %v", tt.err, result, tt.expected) + } + }) + } +} + func TestValidRequest(t *testing.T) { tests := []struct { name string @@ -746,6 +816,204 @@ func TestServerQUIC_ServeQUIC_StalledStreamDoesNotStarveWorkerPool(t *testing.T) } } +// TestServerQUIC_ServeQUIC_StalledStreamDoesNotKillOtherStreamsOnSameConnection +// is a regression test for https://github.com/coredns/coredns/issues/7087: +// one stream timing out used to close the whole QUIC connection via +// closeQUICConn(conn, DoQCodeProtocolError), silently failing every other +// query multiplexed on that same connection. Since DoQ clients keep one +// connection open and send each subsequent query as a new stream on it, a +// single stalled or malformed stream could make unrelated, well-formed +// queries appear to randomly fail. A per-stream read-deadline timeout must +// only cancel that one stream and leave the connection - and every other +// query on it - alone. +func TestServerQUIC_ServeQUIC_StalledStreamDoesNotKillOtherStreamsOnSameConnection(t *testing.T) { + config := testConfig("quic", echoPlugin{}) + config.TLSConfig = mustMakeQUICServerTLSConfig(t) + + server, err := NewServerQUIC(transport.QUIC+"://127.0.0.1:0", []*Config{config}) + if err != nil { + t.Fatalf("NewServerQUIC() failed: %v", err) + } + // Keep the test fast: the stalled stream must time out quickly. + server.ReadTimeout = 250 * time.Millisecond + + pc, err := server.ListenPacket() + if err != nil { + t.Fatalf("ListenPacket() failed: %v", err) + } + defer pc.Close() + + serveErrCh := make(chan error, 1) + go func() { + serveErrCh <- server.ServeQUIC() + }() + + defer func() { + _ = server.Stop() + select { + case <-serveErrCh: + case <-time.After(2 * time.Second): + } + }() + + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + + conn, err := quic.DialAddr(ctx, pc.LocalAddr().String(), mustMakeQUICClientTLSConfig(), &quic.Config{}) + if err != nil { + t.Fatalf("quic.DialAddr() failed: %v", err) + } + defer conn.CloseWithError(DoQCodeNoError, "") + + // Open a stream and announce a message but never send the body, so the + // server's readDOQMessage blocks on this one stream until the read + // deadline fires. + stallStream, err := conn.OpenStreamSync(ctx) + if err != nil { + t.Fatalf("OpenStreamSync() for stalled stream failed: %v", err) + } + if _, err := stallStream.Write([]byte{0x00, 0x64}); err != nil { + t.Fatalf("stalled stream.Write() failed: %v", err) + } + + // Give the server's read deadline time to fire on the stalled stream + // before sending the well-formed query below, so this test actually + // exercises the connection staying usable afterwards. + time.Sleep(500 * time.Millisecond) + + // A second, well-formed query multiplexed as a new stream on the *same* + // connection must still be answered. + normalStream, err := conn.OpenStreamSync(ctx) + if err != nil { + t.Fatalf("OpenStreamSync() for normal stream failed: %v", err) + } + + q := new(dns.Msg) + q.SetQuestion("example.com.", dns.TypeA) + q.Id = 0 + wire, err := q.Pack() + if err != nil { + t.Fatalf("dns.Msg.Pack() failed: %v", err) + } + if _, err := normalStream.Write(AddPrefix(wire)); err != nil { + t.Fatalf("normal stream.Write() failed: %v", err) + } + if err := normalStream.Close(); err != nil { + t.Fatalf("normal stream.Close() failed: %v", err) + } + + respCh := make(chan error, 1) + go func() { + _, rerr := readDOQMessage(normalStream) + respCh <- rerr + }() + + select { + case rerr := <-respCh: + if rerr != nil { + t.Fatalf("normal query on the same connection was not served: readDOQMessage() error = %v", rerr) + } + case <-time.After(5 * time.Second): + t.Fatal("normal query on the same connection was not served within 5s: the stalled stream killed the connection") + } +} + +// TestServerQUIC_ServeQUIC_PeerResetStreamDoesNotKillOtherStreamsOnSameConnection +// is a regression test covering the other branch of isTransientStreamError: +// a client that abandons one in-flight query (e.g. because it deduplicated +// it, or its own timeout fired) sends a QUIC RESET_STREAM for that one +// stream. That surfaces on the server as a *quic.StreamError while reading +// - a condition local to that one stream, not a DoQ framing violation by +// the peer (RFC 9250 §4.3.3) - so it must not take down the whole +// connection and the other, unrelated queries in flight on it. +func TestServerQUIC_ServeQUIC_PeerResetStreamDoesNotKillOtherStreamsOnSameConnection(t *testing.T) { + config := testConfig("quic", echoPlugin{}) + config.TLSConfig = mustMakeQUICServerTLSConfig(t) + + server, err := NewServerQUIC(transport.QUIC+"://127.0.0.1:0", []*Config{config}) + if err != nil { + t.Fatalf("NewServerQUIC() failed: %v", err) + } + + pc, err := server.ListenPacket() + if err != nil { + t.Fatalf("ListenPacket() failed: %v", err) + } + defer pc.Close() + + serveErrCh := make(chan error, 1) + go func() { + serveErrCh <- server.ServeQUIC() + }() + + defer func() { + _ = server.Stop() + select { + case <-serveErrCh: + case <-time.After(2 * time.Second): + } + }() + + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + + conn, err := quic.DialAddr(ctx, pc.LocalAddr().String(), mustMakeQUICClientTLSConfig(), &quic.Config{}) + if err != nil { + t.Fatalf("quic.DialAddr() failed: %v", err) + } + defer conn.CloseWithError(DoQCodeNoError, "") + + // Open a stream, send only part of the length prefix, then abandon it + // with a RESET_STREAM instead of finishing the query. The server's + // readDOQMessage is mid-read at that point, so it observes the reset + // as a *quic.StreamError. + abandonedStream, err := conn.OpenStreamSync(ctx) + if err != nil { + t.Fatalf("OpenStreamSync() for abandoned stream failed: %v", err) + } + if _, err := abandonedStream.Write([]byte{0x00}); err != nil { + t.Fatalf("abandoned stream.Write() failed: %v", err) + } + abandonedStream.CancelWrite(0) + abandonedStream.CancelRead(0) + + // A well-formed query multiplexed as a new stream on the *same* + // connection must still be answered. + normalStream, err := conn.OpenStreamSync(ctx) + if err != nil { + t.Fatalf("OpenStreamSync() for normal stream failed: %v", err) + } + + q := new(dns.Msg) + q.SetQuestion("example.com.", dns.TypeA) + q.Id = 0 + wire, err := q.Pack() + if err != nil { + t.Fatalf("dns.Msg.Pack() failed: %v", err) + } + if _, err := normalStream.Write(AddPrefix(wire)); err != nil { + t.Fatalf("normal stream.Write() failed: %v", err) + } + if err := normalStream.Close(); err != nil { + t.Fatalf("normal stream.Close() failed: %v", err) + } + + respCh := make(chan error, 1) + go func() { + _, rerr := readDOQMessage(normalStream) + respCh <- rerr + }() + + select { + case rerr := <-respCh: + if rerr != nil { + t.Fatalf("normal query on the same connection was not served: readDOQMessage() error = %v", rerr) + } + case <-time.After(5 * time.Second): + t.Fatal("normal query on the same connection was not served within 5s: the abandoned stream killed the connection") + } +} + func mustMakeQUICServerTLSConfig(t *testing.T) *tls.Config { t.Helper()