mirrors/uquic
1
0
Fork 0
mirror of https://github.com/refraction-networking/uquic.git synced 2025-04-03 20:27:35 +03:00
Code Issues Projects Releases Packages Wiki Activity

Merge pull request #3074 from lucas-clemente/fix-0rtt-queue-race

Browse source 
fix race condition when receiving 0-RTT packets
This commit is contained in:
Marten Seemann 2021-03-11 21:36:04 +08:00 • committed by GitHub
commit 3bce408c8d
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
10 changed files with 448 additions and 439 deletions
Download patch file Download diff file Expand all files Collapse all files

4
fuzzing/header/fuzz.go
View file

@ -29,6 +29,7 @@ func Fuzz(data []byte) int {
if err != nil {
return 0
}
is0RTTPacket := wire.Is0RTTPacket(data)
hdr, _, _, err := wire.ParsePacket(data, connIDLen)
if err != nil {
return 0
@ -36,6 +37,9 @@ func Fuzz(data []byte) int {
if !hdr.DestConnectionID.Equal(connID) {
panic(fmt.Sprintf("Expected connection IDs to match: %s vs %s", hdr.DestConnectionID, connID))
}
if (hdr.Type == protocol.PacketType0RTT) != is0RTTPacket {
panic("inconsistent 0-RTT packet detection")
}
var extHdr *wire.ExtendedHeader
// Parse the extended header, if this is not a Retry packet.

287
integrationtests/self/zero_rtt_test.go
View file

@ -7,19 +7,40 @@ import (
"io/ioutil"
mrand "math/rand"
"net"
"sort"
"sync"
"sync/atomic"
"time"
quic "github.com/lucas-clemente/quic-go"
"github.com/lucas-clemente/quic-go"
quicproxy "github.com/lucas-clemente/quic-go/integrationtests/tools/proxy"
"github.com/lucas-clemente/quic-go/internal/protocol"
"github.com/lucas-clemente/quic-go/internal/wire"
"github.com/lucas-clemente/quic-go/logging"
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
)
type rcvdPacketTracer struct {
connTracer
closed chan struct{}
rcvdPackets []*logging.ExtendedHeader
}
func newRcvdPacketTracer() *rcvdPacketTracer {
return &rcvdPacketTracer{closed: make(chan struct{})}
}
func (t *rcvdPacketTracer) ReceivedPacket(hdr *logging.ExtendedHeader, _ logging.ByteCount, _ []logging.Frame) {
t.rcvdPackets = append(t.rcvdPackets, hdr)
}
func (t *rcvdPacketTracer) Close() { close(t.closed) }
func (t *rcvdPacketTracer) getRcvdPackets() []*logging.ExtendedHeader {
<-t.closed
return t.rcvdPackets
}
var _ = Describe("0-RTT", func() {
rtt := scaleDuration(5 * time.Millisecond)
@ -27,16 +48,6 @@ var _ = Describe("0-RTT", func() {
version := v
Context(fmt.Sprintf("with QUIC version %s", version), func() {
runDelayProxy := func(serverPort int) *quicproxy.QuicProxy {
proxy, err := quicproxy.NewQuicProxy("localhost:0", &quicproxy.Opts{
RemoteAddr: fmt.Sprintf("localhost:%d", serverPort),
DelayPacket: func(_ quicproxy.Direction, data []byte) time.Duration { return rtt / 2 },
})
Expect(err).ToNot(HaveOccurred())
return proxy
}
runCountingProxy := func(serverPort int) (*quicproxy.QuicProxy, *uint32) {
var num0RTTPackets uint32 // to be used as an atomic
proxy, err := quicproxy.NewQuicProxy("localhost:0", &quicproxy.Opts{
@ -55,12 +66,37 @@ var _ = Describe("0-RTT", func() {
return proxy, &num0RTTPackets
}
dialAndReceiveSessionTicket := func(ln quic.EarlyListener, proxyPort int) *tls.Config {
dialAndReceiveSessionTicket := func(serverConf *quic.Config) (*tls.Config, *tls.Config) {
tlsConf := getTLSConfig()
if serverConf == nil {
serverConf = getQuicConfig(&quic.Config{
AcceptToken: func(_ net.Addr, _ *quic.Token) bool { return true },
})
serverConf.Versions = []protocol.VersionNumber{version}
}
ln, err := quic.ListenAddrEarly(
"localhost:0",
tlsConf,
serverConf,
)
Expect(err).ToNot(HaveOccurred())
defer ln.Close()
proxy, err := quicproxy.NewQuicProxy("localhost:0", &quicproxy.Opts{
RemoteAddr: fmt.Sprintf("localhost:%d", ln.Addr().(*net.UDPAddr).Port),
DelayPacket: func(_ quicproxy.Direction, data []byte) time.Duration { return rtt / 2 },
})
Expect(err).ToNot(HaveOccurred())
defer proxy.Close()
// dial the first session in order to receive a session ticket
done := make(chan struct{})
go func() {
defer GinkgoRecover()
_, err := ln.Accept(context.Background())
defer close(done)
sess, err := ln.Accept(context.Background())
Expect(err).ToNot(HaveOccurred())
<-sess.Context().Done()
}()
clientConf := getTLSClientConfig()
@ -68,7 +104,7 @@ var _ = Describe("0-RTT", func() {
puts := make(chan string, 100)
clientConf.ClientSessionCache = newClientSessionCache(gets, puts)
sess, err := quic.DialAddr(
fmt.Sprintf("localhost:%d", proxyPort),
fmt.Sprintf("localhost:%d", proxy.LocalPort()),
clientConf,
getQuicConfig(&quic.Config{Versions: []protocol.VersionNumber{version}}),
)
@ -76,7 +112,8 @@ var _ = Describe("0-RTT", func() {
Eventually(puts).Should(Receive())
// received the session ticket. We're done here.
Expect(sess.CloseWithError(0, "")).To(Succeed())
return clientConf
Eventually(done).Should(BeClosed())
return tlsConf, clientConf
}
transfer0RTTData := func(
@ -98,6 +135,7 @@ var _ = Describe("0-RTT", func() {
Expect(err).ToNot(HaveOccurred())
Expect(data).To(Equal(testdata))
Expect(sess.ConnectionState().TLS.Used0RTT).To(Equal(expect0RTT))
Expect(sess.CloseWithError(0, "")).To(Succeed())
close(done)
}()
@ -107,6 +145,7 @@ var _ = Describe("0-RTT", func() {
getQuicConfig(&quic.Config{Versions: []protocol.VersionNumber{version}}),
)
Expect(err).ToNot(HaveOccurred())
defer sess.CloseWithError(0, "")
str, err := sess.OpenUniStream()
Expect(err).ToNot(HaveOccurred())
_, err = str.Write(testdata)
@ -114,6 +153,7 @@ var _ = Describe("0-RTT", func() {
Expect(str.Close()).To(Succeed())
Expect(sess.ConnectionState().TLS.Used0RTT).To(Equal(expect0RTT))
Eventually(done).Should(BeClosed())
Eventually(sess.Context().Done()).Should(BeClosed())
}
check0RTTRejected := func(
@ -142,15 +182,32 @@ var _ = Describe("0-RTT", func() {
Expect(serverSess.ConnectionState().TLS.Used0RTT).To(BeFalse())
_, err = serverSess.AcceptUniStream(ctx)
Expect(err).To(Equal(context.DeadlineExceeded))
Expect(serverSess.CloseWithError(0, "")).To(Succeed())
Eventually(sess.Context().Done()).Should(BeClosed())
}
// can be used to extract 0-RTT from a rcvdPacketTracer
get0RTTPackets := func(hdrs []*logging.ExtendedHeader) []protocol.PacketNumber {
var zeroRTTPackets []protocol.PacketNumber
for _, hdr := range hdrs {
if hdr.Type == protocol.PacketType0RTT {
zeroRTTPackets = append(zeroRTTPackets, hdr.PacketNumber)
}
}
return zeroRTTPackets
}
It("transfers 0-RTT data", func() {
tlsConf, clientConf := dialAndReceiveSessionTicket(nil)
tracer := newRcvdPacketTracer()
ln, err := quic.ListenAddrEarly(
"localhost:0",
getTLSConfig(),
tlsConf,
getQuicConfig(&quic.Config{
Versions: []protocol.VersionNumber{version},
AcceptToken: func(_ net.Addr, _ *quic.Token) bool { return true },
Tracer: newTracer(func() logging.ConnectionTracer { return tracer }),
}),
)
Expect(err).ToNot(HaveOccurred())
@ -159,37 +216,38 @@ var _ = Describe("0-RTT", func() {
proxy, num0RTTPackets := runCountingProxy(ln.Addr().(*net.UDPAddr).Port)
defer proxy.Close()
clientConf := dialAndReceiveSessionTicket(ln, proxy.LocalPort())
transfer0RTTData(ln, proxy.LocalPort(), clientConf, PRData, true)
num0RTT := atomic.LoadUint32(num0RTTPackets)
fmt.Fprintf(GinkgoWriter, "Sent %d 0-RTT packets.", num0RTT)
Expect(num0RTT).ToNot(BeZero())
zeroRTTPackets := get0RTTPackets(tracer.getRcvdPackets())
Expect(len(zeroRTTPackets)).To(BeNumerically(">", 10))
sort.Slice(zeroRTTPackets, func(i, j int) bool { return zeroRTTPackets[i] < zeroRTTPackets[j] })
Expect(zeroRTTPackets[0]).To(Equal(protocol.PacketNumber(0)))
})
// Test that data intended to be sent with 1-RTT protection is not sent in 0-RTT packets.
It("waits until a session until the handshake is done", func() {
ln, err := quic.ListenAddrEarly(
"localhost:0",
getTLSConfig(),
&quic.Config{
Versions: []protocol.VersionNumber{version},
AcceptToken: func(_ net.Addr, _ *quic.Token) bool { return true },
},
)
Expect(err).ToNot(HaveOccurred())
defer ln.Close()
proxy, num0RTTPackets := runCountingProxy(ln.Addr().(*net.UDPAddr).Port)
defer proxy.Close()
clientConf := dialAndReceiveSessionTicket(ln, proxy.LocalPort())
tlsConf, clientConf := dialAndReceiveSessionTicket(nil)
zeroRTTData := GeneratePRData(2 * 1100) // 2 packets
oneRTTData := PRData
tracer := newRcvdPacketTracer()
ln, err := quic.ListenAddrEarly(
"localhost:0",
tlsConf,
getQuicConfig(&quic.Config{
Versions: []protocol.VersionNumber{version},
AcceptToken: func(_ net.Addr, _ *quic.Token) bool { return true },
Tracer: newTracer(func() logging.ConnectionTracer { return tracer }),
}),
)
Expect(err).ToNot(HaveOccurred())
defer ln.Close()
// now dial the second session, and use 0-RTT to send some data
done := make(chan struct{})
go func() {
defer GinkgoRecover()
sess, err := ln.Accept(context.Background())
@ -204,9 +262,12 @@ var _ = Describe("0-RTT", func() {
data, err = ioutil.ReadAll(str)
Expect(err).ToNot(HaveOccurred())
Expect(data).To(Equal(oneRTTData))
close(done)
Expect(sess.CloseWithError(0, "")).To(Succeed())
}()
proxy, num0RTTPackets := runCountingProxy(ln.Addr().(*net.UDPAddr).Port)
defer proxy.Close()
sess, err := quic.DialAddrEarly(
fmt.Sprintf("localhost:%d", proxy.LocalPort()),
clientConf,
@ -232,12 +293,12 @@ var _ = Describe("0-RTT", func() {
_, err = str.Write(PRData)
Expect(err).ToNot(HaveOccurred())
Expect(str.Close()).To(Succeed())
Eventually(done).Should(BeClosed())
<-sess.Context().Done()
num0RTT := atomic.LoadUint32(num0RTTPackets)
fmt.Fprintf(GinkgoWriter, "Sent %d 0-RTT packets.", num0RTT)
Expect(num0RTT).To(Or(BeEquivalentTo(2), BeEquivalentTo(3))) // the FIN might be sent in a separate packet
Expect(get0RTTPackets(tracer.getRcvdPackets())).To(HaveLen(int(num0RTT)))
})
It("transfers 0-RTT data, when 0-RTT packets are lost", func() {
@ -246,20 +307,22 @@ var _ = Describe("0-RTT", func() {
num0RTTDropped uint32
)
tlsConf, clientConf := dialAndReceiveSessionTicket(nil)
tracer := newRcvdPacketTracer()
ln, err := quic.ListenAddrEarly(
"localhost:0",
getTLSConfig(),
tlsConf,
getQuicConfig(&quic.Config{
Versions: []protocol.VersionNumber{version},
AcceptToken: func(_ net.Addr, _ *quic.Token) bool { return true },
Versions: []protocol.VersionNumber{version},
Tracer: newTracer(func() logging.ConnectionTracer { return tracer }),
}),
)
Expect(err).ToNot(HaveOccurred())
defer ln.Close()
serverPort := ln.Addr().(*net.UDPAddr).Port
proxy, err := quicproxy.NewQuicProxy("localhost:0", &quicproxy.Opts{
RemoteAddr: fmt.Sprintf("localhost:%d", serverPort),
RemoteAddr: fmt.Sprintf("localhost:%d", ln.Addr().(*net.UDPAddr).Port),
DelayPacket: func(_ quicproxy.Direction, data []byte) time.Duration {
hdr, _, _, err := wire.ParsePacket(data, 0)
Expect(err).ToNot(HaveOccurred())
@ -285,7 +348,6 @@ var _ = Describe("0-RTT", func() {
Expect(err).ToNot(HaveOccurred())
defer proxy.Close()
clientConf := dialAndReceiveSessionTicket(ln, proxy.LocalPort())
transfer0RTTData(ln, proxy.LocalPort(), clientConf, PRData, true)
num0RTT := atomic.LoadUint32(&num0RTTPackets)
@ -293,6 +355,7 @@ var _ = Describe("0-RTT", func() {
fmt.Fprintf(GinkgoWriter, "Sent %d 0-RTT packets. Dropped %d of those.", num0RTT, numDropped)
Expect(numDropped).ToNot(BeZero())
Expect(num0RTT).ToNot(BeZero())
Expect(get0RTTPackets(tracer.getRcvdPackets())).ToNot(BeEmpty())
})
It("retransmits all 0-RTT data when the server performs a Retry", func() {
@ -300,14 +363,7 @@ var _ = Describe("0-RTT", func() {
var firstConnID, secondConnID protocol.ConnectionID
var firstCounter, secondCounter protocol.ByteCount
ln, err := quic.ListenAddrEarly(
"localhost:0",
getTLSConfig(),
getQuicConfig(&quic.Config{Versions: []protocol.VersionNumber{version}}),
)
Expect(err).ToNot(HaveOccurred())
defer ln.Close()
serverPort := ln.Addr().(*net.UDPAddr).Port
tlsConf, clientConf := dialAndReceiveSessionTicket(nil)
countZeroRTTBytes := func(data []byte) (n protocol.ByteCount) {
for len(data) > 0 {
@ -323,8 +379,20 @@ var _ = Describe("0-RTT", func() {
return
}
tracer := newRcvdPacketTracer()
ln, err := quic.ListenAddrEarly(
"localhost:0",
tlsConf,
getQuicConfig(&quic.Config{
Versions: []protocol.VersionNumber{version},
Tracer: newTracer(func() logging.ConnectionTracer { return tracer }),
}),
)
Expect(err).ToNot(HaveOccurred())
defer ln.Close()
proxy, err := quicproxy.NewQuicProxy("localhost:0", &quicproxy.Opts{
RemoteAddr: fmt.Sprintf("localhost:%d", serverPort),
RemoteAddr: fmt.Sprintf("localhost:%d", ln.Addr().(*net.UDPAddr).Port),
DelayPacket: func(dir quicproxy.Direction, data []byte) time.Duration {
connID, err := wire.ParseConnectionID(data, 0)
Expect(err).ToNot(HaveOccurred())
@ -354,43 +422,37 @@ var _ = Describe("0-RTT", func() {
Expect(err).ToNot(HaveOccurred())
defer proxy.Close()
clientConf := dialAndReceiveSessionTicket(ln, proxy.LocalPort())
transfer0RTTData(ln, proxy.LocalPort(), clientConf, GeneratePRData(5000), true) // ~5 packets
mutex.Lock()
defer mutex.Unlock()
Expect(firstCounter).To(BeNumerically("~", 5000+100 /* framing overhead */, 100)) // the FIN bit might be sent extra
Expect(secondCounter).To(BeNumerically("~", firstCounter, 20))
zeroRTTPackets := get0RTTPackets(tracer.getRcvdPackets())
Expect(len(zeroRTTPackets)).To(BeNumerically(">=", 5))
Expect(zeroRTTPackets[0]).To(BeNumerically(">=", protocol.PacketNumber(5)))
})
It("rejects 0-RTT when the server's transport parameters changed", func() {
const maxStreams = 42
tlsConf := getTLSConfig()
tlsConf, clientConf := dialAndReceiveSessionTicket(getQuicConfig(&quic.Config{
MaxIncomingStreams: maxStreams,
AcceptToken: func(_ net.Addr, _ *quic.Token) bool { return true },
}))
tracer := newRcvdPacketTracer()
ln, err := quic.ListenAddrEarly(
"localhost:0",
tlsConf,
getQuicConfig(&quic.Config{
Versions: []protocol.VersionNumber{version},
AcceptToken: func(_ net.Addr, _ *quic.Token) bool { return true },
MaxIncomingStreams: maxStreams,
}),
)
Expect(err).ToNot(HaveOccurred())
clientConf := dialAndReceiveSessionTicket(ln, ln.Addr().(*net.UDPAddr).Port)
// now close the listener and restart it with a different config
Expect(ln.Close()).To(Succeed())
ln, err = quic.ListenAddrEarly(
"localhost:0",
tlsConf,
getQuicConfig(&quic.Config{
Versions: []protocol.VersionNumber{version},
AcceptToken: func(_ net.Addr, _ *quic.Token) bool { return true },
MaxIncomingStreams: maxStreams + 1,
Tracer: newTracer(func() logging.ConnectionTracer { return tracer }),
}),
)
Expect(err).ToNot(HaveOccurred())
defer ln.Close()
proxy, num0RTTPackets := runCountingProxy(ln.Addr().(*net.UDPAddr).Port)
defer proxy.Close()
check0RTTRejected(ln, proxy.LocalPort(), clientConf)
@ -399,37 +461,27 @@ var _ = Describe("0-RTT", func() {
num0RTT := atomic.LoadUint32(num0RTTPackets)
fmt.Fprintf(GinkgoWriter, "Sent %d 0-RTT packets.", num0RTT)
Expect(num0RTT).ToNot(BeZero())
Expect(get0RTTPackets(tracer.getRcvdPackets())).To(BeEmpty())
})
It("rejects 0-RTT when the ALPN changed", func() {
tlsConf := getTLSConfig()
tlsConf, clientConf := dialAndReceiveSessionTicket(nil)
// now close the listener and dial new connection with a different ALPN
clientConf.NextProtos = []string{"new-alpn"}
tlsConf.NextProtos = []string{"new-alpn"}
tracer := newRcvdPacketTracer()
ln, err := quic.ListenAddrEarly(
"localhost:0",
tlsConf,
getQuicConfig(&quic.Config{
Versions: []protocol.VersionNumber{version},
AcceptToken: func(_ net.Addr, _ *quic.Token) bool { return true },
Tracer: newTracer(func() logging.ConnectionTracer { return tracer }),
}),
)
Expect(err).ToNot(HaveOccurred())
delayProxy := runDelayProxy(ln.Addr().(*net.UDPAddr).Port)
defer delayProxy.Close()
clientConf := dialAndReceiveSessionTicket(ln, delayProxy.LocalPort())
// now close the listener and dial new connection with a different ALPN
Expect(ln.Close()).To(Succeed())
clientConf.NextProtos = []string{"new-alpn"}
tlsConf.NextProtos = []string{"new-alpn"}
ln, err = quic.ListenAddrEarly(
"localhost:0",
tlsConf,
getQuicConfig(&quic.Config{
Versions: []protocol.VersionNumber{version},
AcceptToken: func(_ net.Addr, _ *quic.Token) bool { return true },
}),
)
Expect(err).ToNot(HaveOccurred())
defer ln.Close()
proxy, num0RTTPackets := runCountingProxy(ln.Addr().(*net.UDPAddr).Port)
defer proxy.Close()
@ -439,35 +491,24 @@ var _ = Describe("0-RTT", func() {
num0RTT := atomic.LoadUint32(num0RTTPackets)
fmt.Fprintf(GinkgoWriter, "Sent %d 0-RTT packets.", num0RTT)
Expect(num0RTT).ToNot(BeZero())
Expect(get0RTTPackets(tracer.getRcvdPackets())).To(BeEmpty())
})
It("correctly deals with 0-RTT rejections", func() {
tlsConf := getTLSConfig()
tlsConf, clientConf := dialAndReceiveSessionTicket(nil)
// now dial new connection with different transport parameters
tracer := newRcvdPacketTracer()
ln, err := quic.ListenAddrEarly(
"localhost:0",
tlsConf,
getQuicConfig(&quic.Config{
Versions: []protocol.VersionNumber{version},
MaxIncomingUniStreams: 2,
AcceptToken: func(_ net.Addr, _ *quic.Token) bool { return true },
}),
)
Expect(err).ToNot(HaveOccurred())
delayProxy := runDelayProxy(ln.Addr().(*net.UDPAddr).Port)
defer delayProxy.Close()
clientConf := dialAndReceiveSessionTicket(ln, delayProxy.LocalPort())
// now close the listener and dial new connection with different transport parameters
Expect(ln.Close()).To(Succeed())
ln, err = quic.ListenAddrEarly(
"localhost:0",
tlsConf,
getQuicConfig(&quic.Config{
Versions: []protocol.VersionNumber{version},
MaxIncomingUniStreams: 1,
Tracer: newTracer(func() logging.ConnectionTracer { return tracer }),
}),
)
Expect(err).ToNot(HaveOccurred())
defer ln.Close()
proxy, num0RTTPackets := runCountingProxy(ln.Addr().(*net.UDPAddr).Port)
defer proxy.Close()
@ -482,6 +523,7 @@ var _ = Describe("0-RTT", func() {
data, err := ioutil.ReadAll(str)
Expect(err).ToNot(HaveOccurred())
Expect(string(data)).To(Equal("second flight"))
Expect(sess.CloseWithError(0, "")).To(Succeed())
}()
sess, err := quic.DialAddrEarly(
@ -517,11 +559,48 @@ var _ = Describe("0-RTT", func() {
Expect(str.Close()).To(Succeed())
Eventually(done).Should(BeClosed())
Eventually(sess.Context().Done()).Should(BeClosed())
// The client should send 0-RTT packets, but the server doesn't process them.
num0RTT := atomic.LoadUint32(num0RTTPackets)
fmt.Fprintf(GinkgoWriter, "Sent %d 0-RTT packets.", num0RTT)
Expect(num0RTT).ToNot(BeZero())
Expect(get0RTTPackets(tracer.getRcvdPackets())).To(BeEmpty())
})
It("queues 0-RTT packets, if the Initial is delayed", func() {
tlsConf, clientConf := dialAndReceiveSessionTicket(nil)
tracer := newRcvdPacketTracer()
ln, err := quic.ListenAddrEarly(
"localhost:0",
tlsConf,
getQuicConfig(&quic.Config{
Versions: []protocol.VersionNumber{version},
AcceptToken: func(_ net.Addr, _ *quic.Token) bool { return true },
Tracer: newTracer(func() logging.ConnectionTracer { return tracer }),
}),
)
Expect(err).ToNot(HaveOccurred())
defer ln.Close()
proxy, err := quicproxy.NewQuicProxy("localhost:0", &quicproxy.Opts{
RemoteAddr: ln.Addr().String(),
DelayPacket: func(dir quicproxy.Direction, data []byte) time.Duration {
if dir == quicproxy.DirectionIncoming && data[0]&0x80 > 0 && data[0]&0x30>>4 == 0 { // Initial packet from client
return rtt/2 + rtt
}
return rtt / 2
},
})
Expect(err).ToNot(HaveOccurred())
defer proxy.Close()
transfer0RTTData(ln, proxy.LocalPort(), clientConf, PRData, true)
Expect(tracer.rcvdPackets[0].Type).To(Equal(protocol.PacketTypeInitial))
zeroRTTPackets := get0RTTPackets(tracer.getRcvdPackets())
Expect(len(zeroRTTPackets)).To(BeNumerically(">", 10))
Expect(zeroRTTPackets[0]).To(Equal(protocol.PacketNumber(0)))
})
})
}

16
internal/wire/header.go
View file

@ -2,6 +2,7 @@ package wire
import (
"bytes"
"encoding/binary"
"errors"
"fmt"
"io"
@ -43,6 +44,21 @@ func IsVersionNegotiationPacket(b []byte) bool {
return b[0]&0x80 > 0 && b[1] == 0 && b[2] == 0 && b[3] == 0 && b[4] == 0
}
// Is0RTTPacket says if this is a 0-RTT packet.
// A packet sent with a version we don't understand can never be a 0-RTT packet.
func Is0RTTPacket(b []byte) bool {
if len(b) < 5 {
return false
}
if b[0]&0x80 == 0 {
return false
}
if !protocol.IsSupportedVersion(protocol.SupportedVersions, protocol.VersionNumber(binary.BigEndian.Uint32(b[1:5]))) {
return false
}
return b[0]&0x30>>4 == 0x1
}
var ErrUnsupportedVersion = errors.New("unsupported version")
// The Header is the version independent part of the header

18
internal/wire/header_test.go
View file

@ -93,6 +93,24 @@ var _ = Describe("Header Parsing", func() {
})
})
Context("identifying 0-RTT packets", func() {
var zeroRTTHeader []byte
BeforeEach(func() {
zeroRTTHeader = make([]byte, 5)
zeroRTTHeader[0] = 0x80 | 0x1<<4
binary.BigEndian.PutUint32(zeroRTTHeader[1:], uint32(versionIETFFrames))
})
It("recognizes 0-RTT packets", func() {
Expect(Is0RTTPacket(zeroRTTHeader[:4])).To(BeFalse()) // too short
Expect(Is0RTTPacket([]byte{zeroRTTHeader[0], 1, 2, 3, 4})).To(BeFalse()) // unknown version
Expect(Is0RTTPacket([]byte{zeroRTTHeader[0] | 0x80, 1, 2, 3, 4})).To(BeFalse()) // short header
Expect(Is0RTTPacket(zeroRTTHeader)).To(BeTrue())
Expect(Is0RTTPacket(append(zeroRTTHeader, []byte("foobar")...))).To(BeTrue())
})
})
Context("Identifying Version Negotiation Packets", func() {
It("identifies version negotiation packets", func() {
Expect(IsVersionNegotiationPacket([]byte{0x80 | 0x56, 0, 0, 0, 0})).To(BeTrue())

123
packet_handler_map.go
View file

@ -26,6 +26,38 @@ func (e statelessResetErr) Error() string {
return fmt.Sprintf("received a stateless reset with token %x", e.token)
}
type zeroRTTQueue struct {
queue []*receivedPacket
retireTimer *time.Timer
}
var _ packetHandler = &zeroRTTQueue{}
func (h *zeroRTTQueue) handlePacket(p *receivedPacket) {
if len(h.queue) < protocol.Max0RTTQueueLen {
h.queue = append(h.queue, p)
}
}
func (h *zeroRTTQueue) shutdown() {}
func (h *zeroRTTQueue) destroy(error) {}
func (h *zeroRTTQueue) getPerspective() protocol.Perspective { return protocol.PerspectiveClient }
func (h *zeroRTTQueue) EnqueueAll(sess packetHandler) {
for _, p := range h.queue {
sess.handlePacket(p)
}
}
func (h *zeroRTTQueue) Clear() {
for _, p := range h.queue {
p.buffer.Release()
}
}
type packetHandlerMapEntry struct {
packetHandler packetHandler
is0RTTQueue bool
}
// The packetHandlerMap stores packetHandlers, identified by connection ID.
// It is used:
// * by the server to store sessions
@ -36,14 +68,16 @@ type packetHandlerMap struct {
conn connection
connIDLen int
handlers map[string] /* string(ConnectionID)*/ packetHandler
resetTokens map[protocol.StatelessResetToken] /* stateless reset token */ packetHandler
server unknownPacketHandler
handlers map[string] /* string(ConnectionID)*/ packetHandlerMapEntry
resetTokens map[protocol.StatelessResetToken] /* stateless reset token */ packetHandler
server unknownPacketHandler
numZeroRTTEntries int
listening chan struct{} // is closed when listen returns
closed bool
deleteRetiredSessionsAfter time.Duration
zeroRTTQueueDuration time.Duration
statelessResetEnabled bool
statelessResetMutex sync.Mutex
@ -107,9 +141,10 @@ func newPacketHandlerMap(
conn: conn,
connIDLen: connIDLen,
listening: make(chan struct{}),
handlers: make(map[string]packetHandler),
handlers: make(map[string]packetHandlerMapEntry),
resetTokens: make(map[protocol.StatelessResetToken]packetHandler),
deleteRetiredSessionsAfter: protocol.RetiredConnectionIDDeleteTimeout,
zeroRTTQueueDuration: protocol.Max0RTTQueueingDuration,
statelessResetEnabled: len(statelessResetKey) > 0,
statelessResetHasher: hmac.New(sha256.New, statelessResetKey),
tracer: tracer,
@ -157,7 +192,7 @@ func (h *packetHandlerMap) Add(id protocol.ConnectionID, handler packetHandler)
h.logger.Debugf("Not adding connection ID %s, as it already exists.", id)
return false
}
h.handlers[string(id)] = handler
h.handlers[string(id)] = packetHandlerMapEntry{packetHandler: handler}
h.logger.Debugf("Adding connection ID %s.", id)
return true
}
@ -166,14 +201,25 @@ func (h *packetHandlerMap) AddWithConnID(clientDestConnID, newConnID protocol.Co
h.mutex.Lock()
defer h.mutex.Unlock()
if _, ok := h.handlers[string(clientDestConnID)]; ok {
h.logger.Debugf("Not adding connection ID %s for a new session, as it already exists.", clientDestConnID)
return false
var q *zeroRTTQueue
if entry, ok := h.handlers[string(clientDestConnID)]; ok {
if !entry.is0RTTQueue {
h.logger.Debugf("Not adding connection ID %s for a new session, as it already exists.", clientDestConnID)
return false
}
q = entry.packetHandler.(*zeroRTTQueue)
q.retireTimer.Stop()
h.numZeroRTTEntries--
if h.numZeroRTTEntries < 0 {
panic("number of 0-RTT queues < 0")
}
}
sess := fn()
h.handlers[string(clientDestConnID)] = sess
h.handlers[string(newConnID)] = sess
if q != nil {
q.EnqueueAll(sess)
}
h.handlers[string(clientDestConnID)] = packetHandlerMapEntry{packetHandler: sess}
h.handlers[string(newConnID)] = packetHandlerMapEntry{packetHandler: sess}
h.logger.Debugf("Adding connection IDs %s and %s for a new session.", clientDestConnID, newConnID)
return true
}
@ -197,7 +243,7 @@ func (h *packetHandlerMap) Retire(id protocol.ConnectionID) {
func (h *packetHandlerMap) ReplaceWithClosed(id protocol.ConnectionID, handler packetHandler) {
h.mutex.Lock()
h.handlers[string(id)] = handler
h.handlers[string(id)] = packetHandlerMapEntry{packetHandler: handler}
h.mutex.Unlock()
h.logger.Debugf("Replacing session for connection ID %s with a closed session.", id)
@ -236,14 +282,14 @@ func (h *packetHandlerMap) CloseServer() {
}
h.server = nil
var wg sync.WaitGroup
for _, handler := range h.handlers {
if handler.getPerspective() == protocol.PerspectiveServer {
for _, entry := range h.handlers {
if entry.packetHandler.getPerspective() == protocol.PerspectiveServer {
wg.Add(1)
go func(handler packetHandler) {
// blocks until the CONNECTION_CLOSE has been sent and the run-loop has stopped
handler.shutdown()
wg.Done()
}(handler)
}(entry.packetHandler)
}
}
h.mutex.Unlock()
@ -268,12 +314,12 @@ func (h *packetHandlerMap) close(e error) error {
}
var wg sync.WaitGroup
for _, handler := range h.handlers {
for _, entry := range h.handlers {
wg.Add(1)
go func(handler packetHandler) {
handler.destroy(e)
wg.Done()
}(handler)
}(entry.packetHandler)
}
if h.server != nil {
@ -319,9 +365,16 @@ func (h *packetHandlerMap) handlePacket(p *receivedPacket) {
return
}
if handler, ok := h.handlers[string(connID)]; ok { // existing session
handler.handlePacket(p)
return
if entry, ok := h.handlers[string(connID)]; ok {
if entry.is0RTTQueue { // only enqueue 0-RTT packets in the 0-RTT queue
if wire.Is0RTTPacket(p.data) {
entry.packetHandler.handlePacket(p)
return
}
} else { // existing session
entry.packetHandler.handlePacket(p)
return
}
}
if p.data[0]&0x80 == 0 {
go h.maybeSendStatelessReset(p, connID)
@ -331,6 +384,36 @@ func (h *packetHandlerMap) handlePacket(p *receivedPacket) {
h.logger.Debugf("received a packet with an unexpected connection ID %s", connID)
return
}
if wire.Is0RTTPacket(p.data) {
if h.numZeroRTTEntries >= protocol.Max0RTTQueues {
return
}
h.numZeroRTTEntries++
queue := &zeroRTTQueue{queue: make([]*receivedPacket, 0, 8)}
h.handlers[string(connID)] = packetHandlerMapEntry{
packetHandler: queue,
is0RTTQueue: true,
}
queue.retireTimer = time.AfterFunc(h.zeroRTTQueueDuration, func() {
h.mutex.Lock()
defer h.mutex.Unlock()
// The entry might have been replaced by an actual session.
// Only delete it if it's still a 0-RTT queue.
if entry, ok := h.handlers[string(connID)]; ok && entry.is0RTTQueue {
delete(h.handlers, string(connID))
h.numZeroRTTEntries--
if h.numZeroRTTEntries < 0 {
panic("number of 0-RTT queues < 0")
}
entry.packetHandler.(*zeroRTTQueue).Clear()
if h.logger.Debug() {
h.logger.Debugf("Removing 0-RTT queue for %s.", connID)
}
}
})
queue.handlePacket(p)
return
}
h.server.handlePacket(p)
}

88
packet_handler_map_test.go
View file

@ -36,12 +36,12 @@ var _ = Describe("Packet Handler Map", func() {
statelessResetKey []byte
)
getPacketWithLength := func(connID protocol.ConnectionID, length protocol.ByteCount) []byte {
getPacketWithPacketType := func(connID protocol.ConnectionID, t protocol.PacketType, length protocol.ByteCount) []byte {
buf := &bytes.Buffer{}
Expect((&wire.ExtendedHeader{
Header: wire.Header{
IsLongHeader: true,
Type: protocol.PacketTypeHandshake,
Type: t,
DestConnectionID: connID,
Length: length,
Version: protocol.VersionTLS,
@ -52,7 +52,7 @@ var _ = Describe("Packet Handler Map", func() {
}
getPacket := func(connID protocol.ConnectionID) []byte {
return getPacketWithLength(connID, 2)
return getPacketWithPacketType(connID, protocol.PacketTypeHandshake, 2)
}
BeforeEach(func() {
@ -274,6 +274,88 @@ var _ = Describe("Packet Handler Map", func() {
})
})
Context("0-RTT", func() {
JustBeforeEach(func() {
handler.zeroRTTQueueDuration = time.Hour
server := NewMockUnknownPacketHandler(mockCtrl)
// we don't expect any calls to server.handlePacket
handler.SetServer(server)
})
It("queues 0-RTT packets", func() {
server := NewMockUnknownPacketHandler(mockCtrl)
// don't EXPECT any calls to server.handlePacket
handler.SetServer(server)
connID := protocol.ConnectionID{0x11, 0x22, 0x33, 0x44, 0x55, 0x66, 0x77, 0x88}
p1 := &receivedPacket{data: getPacketWithPacketType(connID, protocol.PacketType0RTT, 1)}
p2 := &receivedPacket{data: getPacketWithPacketType(connID, protocol.PacketType0RTT, 2)}
p3 := &receivedPacket{data: getPacketWithPacketType(connID, protocol.PacketType0RTT, 3)}
handler.handlePacket(p1)
handler.handlePacket(p2)
handler.handlePacket(p3)
sess := NewMockPacketHandler(mockCtrl)
done := make(chan struct{})
gomock.InOrder(
sess.EXPECT().handlePacket(p1),
sess.EXPECT().handlePacket(p2),
sess.EXPECT().handlePacket(p3).Do(func(packet *receivedPacket) { close(done) }),
)
handler.AddWithConnID(connID, protocol.ConnectionID{1, 2, 3, 4}, func() packetHandler { return sess })
Eventually(done).Should(BeClosed())
})
It("directs 0-RTT packets to existing sessions", func() {
connID := protocol.ConnectionID{0x11, 0x22, 0x33, 0x44, 0x55, 0x66, 0x77, 0x88}
sess := NewMockPacketHandler(mockCtrl)
handler.AddWithConnID(connID, protocol.ConnectionID{1, 2, 3, 4}, func() packetHandler { return sess })
p1 := &receivedPacket{data: getPacketWithPacketType(connID, protocol.PacketType0RTT, 1)}
sess.EXPECT().handlePacket(p1)
handler.handlePacket(p1)
})
It("limits the number of 0-RTT queues", func() {
for i := 0; i < protocol.Max0RTTQueues; i++ {
connID := make(protocol.ConnectionID, 8)
rand.Read(connID)
p := &receivedPacket{data: getPacketWithPacketType(connID, protocol.PacketType0RTT, 1)}
handler.handlePacket(p)
}
// We're already storing the maximum number of queues. This packet will be dropped.
connID := protocol.ConnectionID{1, 2, 3, 4, 5, 6, 7, 8, 9}
handler.handlePacket(&receivedPacket{data: getPacketWithPacketType(connID, protocol.PacketType0RTT, 1)})
// Don't EXPECT any handlePacket() calls.
sess := NewMockPacketHandler(mockCtrl)
handler.AddWithConnID(connID, protocol.ConnectionID{1, 2, 3, 4}, func() packetHandler { return sess })
time.Sleep(20 * time.Millisecond)
})
It("deletes queues if no session is created for this connection ID", func() {
queueDuration := scaleDuration(10 * time.Millisecond)
handler.zeroRTTQueueDuration = queueDuration
server := NewMockUnknownPacketHandler(mockCtrl)
// don't EXPECT any calls to server.handlePacket
handler.SetServer(server)
connID := protocol.ConnectionID{0x11, 0x22, 0x33, 0x44, 0x55, 0x66, 0x77, 0x88}
p1 := &receivedPacket{
data: getPacketWithPacketType(connID, protocol.PacketType0RTT, 1),
buffer: getPacketBuffer(),
}
p2 := &receivedPacket{
data: getPacketWithPacketType(connID, protocol.PacketType0RTT, 2),
buffer: getPacketBuffer(),
}
handler.handlePacket(p1)
handler.handlePacket(p2)
// wait a bit. The queue should now already be deleted.
time.Sleep(queueDuration * 3)
// Don't EXPECT any handlePacket() calls.
sess := NewMockPacketHandler(mockCtrl)
handler.AddWithConnID(connID, protocol.ConnectionID{1, 2, 3, 4}, func() packetHandler { return sess })
time.Sleep(20 * time.Millisecond)
})
})
Context("stateless resets", func() {
BeforeEach(func() {
connIDLen = 5

83
server.go
View file

@ -68,7 +68,6 @@ type baseServer struct {
tokenGenerator *handshake.TokenGenerator
zeroRTTQueue *zeroRTTQueue
sessionHandler packetHandlerManager
receivedPackets chan *receivedPacket
@ -200,7 +199,6 @@ func listen(conn net.PacketConn, tlsConf *tls.Config, config *Config, acceptEarl
config: config,
tokenGenerator: tokenGenerator,
sessionHandler: sessionHandler,
zeroRTTQueue: newZeroRTTQueue(),
sessionQueue: make(chan quicSession),
errorChan: make(chan struct{}),
running: make(chan struct{}),
@ -365,20 +363,15 @@ func (s *baseServer) handlePacketImpl(p *receivedPacket) bool /* is the buffer s
go s.sendVersionNegotiationPacket(p, hdr)
return false
}
if hdr.IsLongHeader {
if hdr.Type == protocol.PacketType0RTT {
s.zeroRTTQueue.Enqueue(hdr.DestConnectionID, p)
return true
} else if hdr.Type != protocol.PacketTypeInitial {
// Drop long header packets.
// There's little point in sending a Stateless Reset, since the client
// might not have received the token yet.
s.logger.Debugf("Dropping long header packet of type %s (%d bytes)", hdr.Type, len(p.data))
if s.config.Tracer != nil {
s.config.Tracer.DroppedPacket(p.remoteAddr, logging.PacketTypeFromHeader(hdr), p.Size(), logging.PacketDropUnexpectedPacket)
}
return false
if hdr.IsLongHeader && hdr.Type != protocol.PacketTypeInitial {
// Drop long header packets.
// There's little point in sending a Stateless Reset, since the client
// might not have received the token yet.
s.logger.Debugf("Dropping long header packet of type %s (%d bytes)", hdr.Type, len(p.data))
if s.config.Tracer != nil {
s.config.Tracer.DroppedPacket(p.remoteAddr, logging.PacketTypeFromHeader(hdr), p.Size(), logging.PacketDropUnexpectedPacket)
}
return false
}
s.logger.Debugf("<- Received Initial packet.")
@ -401,10 +394,10 @@ func (s *baseServer) handleInitialImpl(p *receivedPacket, hdr *wire.Header) erro
}
var (
token *Token
retrySrcConnectionID *protocol.ConnectionID
token *Token
retrySrcConnID *protocol.ConnectionID
)
origDestConnectionID := hdr.DestConnectionID
origDestConnID := hdr.DestConnectionID
if len(hdr.Token) > 0 {
c, err := s.tokenGenerator.DecodeToken(hdr.Token)
if err == nil {
@ -414,8 +407,8 @@ func (s *baseServer) handleInitialImpl(p *receivedPacket, hdr *wire.Header) erro
SentTime: c.SentTime,
}
if token.IsRetryToken {
origDestConnectionID = c.OriginalDestConnectionID
retrySrcConnectionID = &c.RetrySrcConnectionID
origDestConnID = c.OriginalDestConnectionID
retrySrcConnID = &c.RetrySrcConnectionID
}
}
}
@ -451,68 +444,46 @@ func (s *baseServer) handleInitialImpl(p *receivedPacket, hdr *wire.Header) erro
return err
}
s.logger.Debugf("Changing connection ID to %s.", connID)
sess := s.createNewSession(
p.remoteAddr,
origDestConnectionID,
retrySrcConnectionID,
hdr.DestConnectionID,
hdr.SrcConnectionID,
connID,
hdr.Version,
)
if sess == nil {
p.buffer.Release()
return nil
}
sess.handlePacket(p)
s.zeroRTTQueue.DequeueToSession(hdr.DestConnectionID, sess)
return nil
}
func (s *baseServer) createNewSession(
remoteAddr net.Addr,
origDestConnID protocol.ConnectionID,
retrySrcConnID *protocol.ConnectionID,
clientDestConnID protocol.ConnectionID,
destConnID protocol.ConnectionID,
srcConnID protocol.ConnectionID,
version protocol.VersionNumber,
) quicSession {
var sess quicSession
if added := s.sessionHandler.AddWithConnID(clientDestConnID, srcConnID, func() packetHandler {
if added := s.sessionHandler.AddWithConnID(hdr.DestConnectionID, connID, func() packetHandler {
var tracer logging.ConnectionTracer
if s.config.Tracer != nil {
// Use the same connection ID that is passed to the client's GetLogWriter callback.
connID := clientDestConnID
connID := hdr.DestConnectionID
if origDestConnID.Len() > 0 {
connID = origDestConnID
}
tracer = s.config.Tracer.TracerForConnection(protocol.PerspectiveServer, connID)
}
sess = s.newSession(
newSendConn(s.conn, remoteAddr),
newSendConn(s.conn, p.remoteAddr),
s.sessionHandler,
origDestConnID,
retrySrcConnID,
clientDestConnID,
destConnID,
srcConnID,
s.sessionHandler.GetStatelessResetToken(srcConnID),
hdr.DestConnectionID,
hdr.SrcConnectionID,
connID,
s.sessionHandler.GetStatelessResetToken(connID),
s.config,
s.tlsConf,
s.tokenGenerator,
s.acceptEarlySessions,
tracer,
s.logger,
version,
hdr.Version,
)
sess.handlePacket(p)
return sess
}); !added {
return nil
}
go sess.run()
go s.handleNewSession(sess)
return sess
if sess == nil {
p.buffer.Release()
return nil
}
return nil
}
func (s *baseServer) handleNewSession(sess quicSession) {

68
server_test.go
View file

@ -629,60 +629,6 @@ var _ = Describe("Server", func() {
Eventually(done).Should(BeClosed())
})
It("passes queued 0-RTT packets to the session", func() {
serv.config.AcceptToken = func(_ net.Addr, _ *Token) bool { return true }
var createdSession bool
sess := NewMockQuicSession(mockCtrl)
connID := protocol.ConnectionID{1, 2, 3, 4, 5, 6, 7, 8, 9}
initialPacket := getInitial(connID)
zeroRTTPacket := getPacket(&wire.Header{
IsLongHeader: true,
Type: protocol.PacketType0RTT,
SrcConnectionID: protocol.ConnectionID{5, 4, 3, 2, 1},
DestConnectionID: connID,
Version: protocol.VersionTLS,
}, []byte("foobar"))
sess.EXPECT().Context().Return(context.Background()).MaxTimes(1)
sess.EXPECT().HandshakeComplete().Return(context.Background()).MaxTimes(1)
sess.EXPECT().run().MaxTimes(1)
gomock.InOrder(
sess.EXPECT().handlePacket(initialPacket),
sess.EXPECT().handlePacket(zeroRTTPacket),
)
serv.newSession = func(
_ sendConn,
runner sessionRunner,
_ protocol.ConnectionID,
_ *protocol.ConnectionID,
_ protocol.ConnectionID,
_ protocol.ConnectionID,
_ protocol.ConnectionID,
_ protocol.StatelessResetToken,
_ *Config,
_ *tls.Config,
_ *handshake.TokenGenerator,
_ bool,
_ logging.ConnectionTracer,
_ utils.Logger,
_ protocol.VersionNumber,
) quicSession {
createdSession = true
return sess
}
// Receive the 0-RTT packet first.
Expect(serv.handlePacketImpl(zeroRTTPacket)).To(BeTrue())
// Then receive the Initial packet.
phm.EXPECT().GetStatelessResetToken(gomock.Any())
phm.EXPECT().AddWithConnID(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(func(_, _ protocol.ConnectionID, fn func() packetHandler) bool {
fn()
return true
})
tracer.EXPECT().TracerForConnection(protocol.PerspectiveServer, gomock.Any())
Expect(serv.handlePacketImpl(initialPacket)).To(BeTrue())
Expect(createdSession).To(BeTrue())
})
It("drops packets if the receive queue is full", func() {
phm.EXPECT().AddWithConnID(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(func(_, _ protocol.ConnectionID, fn func() packetHandler) bool {
phm.EXPECT().GetStatelessResetToken(gomock.Any())
@ -958,6 +904,7 @@ var _ = Describe("Server", func() {
}()
ctx, cancel := context.WithCancel(context.Background()) // handshake context
serv.config.AcceptToken = func(_ net.Addr, _ *Token) bool { return true }
serv.newSession = func(
_ sendConn,
runner sessionRunner,
@ -975,6 +922,7 @@ var _ = Describe("Server", func() {
_ utils.Logger,
_ protocol.VersionNumber,
) quicSession {
sess.EXPECT().handlePacket(gomock.Any())
sess.EXPECT().HandshakeComplete().Return(ctx)
sess.EXPECT().run().Do(func() {})
sess.EXPECT().Context().Return(context.Background())
@ -986,7 +934,10 @@ var _ = Describe("Server", func() {
return true
})
tracer.EXPECT().TracerForConnection(protocol.PerspectiveServer, gomock.Any())
serv.createNewSession(&net.UDPAddr{}, nil, nil, nil, nil, nil, protocol.VersionWhatever)
serv.handleInitialImpl(
&receivedPacket{buffer: getPacketBuffer()},
&wire.Header{DestConnectionID: protocol.ConnectionID{1, 2, 3, 4, 5, 6, 7, 8}},
)
Consistently(done).ShouldNot(BeClosed())
cancel() // complete the handshake
Eventually(done).Should(BeClosed())
@ -1026,6 +977,7 @@ var _ = Describe("Server", func() {
}()
ready := make(chan struct{})
serv.config.AcceptToken = func(_ net.Addr, _ *Token) bool { return true }
serv.newSession = func(
_ sendConn,
runner sessionRunner,
@ -1044,6 +996,7 @@ var _ = Describe("Server", func() {
_ protocol.VersionNumber,
) quicSession {
Expect(enable0RTT).To(BeTrue())
sess.EXPECT().handlePacket(gomock.Any())
sess.EXPECT().run().Do(func() {})
sess.EXPECT().earlySessionReady().Return(ready)
sess.EXPECT().Context().Return(context.Background())
@ -1054,7 +1007,10 @@ var _ = Describe("Server", func() {
fn()
return true
})
serv.createNewSession(&net.UDPAddr{}, nil, nil, nil, nil, nil, protocol.VersionWhatever)
serv.handleInitialImpl(
&receivedPacket{buffer: getPacketBuffer()},
&wire.Header{DestConnectionID: protocol.ConnectionID{1, 2, 3, 4, 5, 6, 7, 8}},
)
Consistently(done).ShouldNot(BeClosed())
close(ready)
Eventually(done).Should(BeClosed())

79
zero_rtt_queue.go
View file

@ -1,79 +0,0 @@
package quic
import (
"sync"
"time"
"github.com/lucas-clemente/quic-go/internal/protocol"
)
type zeroRTTQueueEntry struct {
timer *time.Timer
packets []*receivedPacket
}
type zeroRTTQueue struct {
mutex sync.Mutex
queue map[string]*zeroRTTQueueEntry
queueDuration time.Duration // so we can set it in tests
}
func newZeroRTTQueue() *zeroRTTQueue {
return &zeroRTTQueue{
queue: make(map[string]*zeroRTTQueueEntry),
queueDuration: protocol.Max0RTTQueueingDuration,
}
}
func (h *zeroRTTQueue) Enqueue(connID protocol.ConnectionID, p *receivedPacket) {
h.mutex.Lock()
defer h.mutex.Unlock()
cid := string(connID)
if _, ok := h.queue[cid]; !ok {
if len(h.queue) >= protocol.Max0RTTQueues {
return
}
h.queue[cid] = &zeroRTTQueueEntry{timer: time.AfterFunc(h.queueDuration, func() {
h.deleteQueue(connID)
})}
}
entry := h.queue[cid]
if len(entry.packets) >= protocol.Max0RTTQueueLen {
return
}
entry.packets = append(entry.packets, p)
}
func (h *zeroRTTQueue) DequeueToSession(connID protocol.ConnectionID, sess packetHandler) {
h.mutex.Lock()
defer h.mutex.Unlock()
h.dequeueToSession(connID, sess)
}
func (h *zeroRTTQueue) dequeueToSession(connID protocol.ConnectionID, sess packetHandler) {
entry, ok := h.queue[string(connID)]
if !ok {
return
}
entry.timer.Stop()
for _, p := range entry.packets {
sess.handlePacket(p)
}
delete(h.queue, string(connID))
}
func (h *zeroRTTQueue) deleteQueue(connID protocol.ConnectionID) {
h.mutex.Lock()
defer h.mutex.Unlock()
entry, ok := h.queue[string(connID)]
if !ok {
return
}
for _, p := range entry.packets {
p.buffer.Release()
}
delete(h.queue, string(connID))
}

121
zero_rtt_queue_test.go
View file

@ -1,121 +0,0 @@
package quic
import (
"encoding/binary"
"time"
"github.com/golang/mock/gomock"
"github.com/lucas-clemente/quic-go/internal/protocol"
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
)
var _ = Describe("0-RTT queue", func() {
var q *zeroRTTQueue
queueDuration := scaleDuration(50 * time.Millisecond)
getQueuedPackets := func(connID protocol.ConnectionID) []*receivedPacket {
var packets []*receivedPacket
sess := NewMockPacketHandler(mockCtrl)
sess.EXPECT().handlePacket(gomock.Any()).Do(func(p *receivedPacket) {
packets = append(packets, p)
}).AnyTimes()
q.DequeueToSession(connID, sess)
return packets
}
BeforeEach(func() {
q = newZeroRTTQueue()
q.queueDuration = queueDuration
})
AfterEach(func() {
// dequeue all packets to make sure the timers are stopped
q.mutex.Lock()
for connID := range q.queue {
sess := NewMockPacketHandler(mockCtrl)
sess.EXPECT().handlePacket(gomock.Any()).AnyTimes()
q.dequeueToSession(protocol.ConnectionID(connID), sess)
}
q.mutex.Unlock()
})
It("stores a 0-RTT packet", func() {
connID := protocol.ConnectionID{0xde, 0xad, 0xbe, 0xef}
p := &receivedPacket{data: []byte("foobar")}
q.Enqueue(connID, p)
queuedPackets := getQueuedPackets(connID)
Expect(queuedPackets).To(Equal([]*receivedPacket{p}))
})
It("doesn't dequeue for unknown connection IDs", func() {
Expect(getQueuedPackets(protocol.ConnectionID{0x42})).To(BeEmpty())
})
It("only stores packets for Max0RTTQueues connection", func() {
getConnID := func(i int) protocol.ConnectionID {
connID := make([]byte, 4)
binary.BigEndian.PutUint32(connID, uint32(i))
return connID
}
// fill up the queues
for i := 0; i < protocol.Max0RTTQueues; i++ {
connID := getConnID(i)
q.Enqueue(connID, &receivedPacket{data: []byte(connID)})
}
// now try to enqueue a packet for another connection ID
connID := protocol.ConnectionID{0xde, 0xad, 0xbe, 0xef}
q.Enqueue(connID, &receivedPacket{data: []byte("foobar")})
// check that the other queues were all saved
for i := 0; i < protocol.Max0RTTQueues; i++ {
queuedPackets := getQueuedPackets(getConnID(i))
Expect(queuedPackets).To(HaveLen(1))
Expect(binary.BigEndian.Uint32(queuedPackets[0].data)).To(BeEquivalentTo(i))
}
})
It("removes queues when packets are dequeued", func() {
// fill up the queues
for i := 0; i < protocol.Max0RTTQueues; i++ {
data := make([]byte, 4)
binary.BigEndian.PutUint32(data, uint32(i))
q.Enqueue(protocol.ConnectionID(data), &receivedPacket{data: data})
}
// now try to enqueue a packet for another connection ID
connID := protocol.ConnectionID{0xde, 0xad, 0xbe, 0xef}
q.Enqueue(connID, &receivedPacket{data: []byte("foobar")})
Expect(getQueuedPackets(connID)).To(BeEmpty())
// dequeue the packet from the first queue
Expect(getQueuedPackets(protocol.ConnectionID{0, 0, 0, 0})).ToNot(BeNil())
// now it should be possible to queue another packet
q.Enqueue(connID, &receivedPacket{data: []byte("foobar")})
Expect(getQueuedPackets(connID)).ToNot(BeNil())
})
It("limits the number of packets it stores for one connection", func() {
connID := protocol.ConnectionID{0xde, 0xad, 0xbe, 0xef}
// fill up the queue
for i := 0; i < protocol.Max0RTTQueueLen; i++ {
data := make([]byte, 4)
binary.BigEndian.PutUint32(data, uint32(i))
q.Enqueue(connID, &receivedPacket{data: data})
}
// The queue is full now. This packet will be dropped.
q.Enqueue(connID, &receivedPacket{data: []byte("foobar")})
queuedPackets := getQueuedPackets(connID)
Expect(queuedPackets).To(HaveLen(protocol.Max0RTTQueueLen))
for i, p := range queuedPackets {
Expect(binary.BigEndian.Uint32(p.data)).To(BeEquivalentTo(i))
}
})
It("deletes packets if they aren't dequeued after a short while", func() {
connID := protocol.ConnectionID{0xde, 0xad, 0xbe, 0xef}
p := &receivedPacket{data: []byte("foobar"), buffer: getPacketBuffer()}
q.Enqueue(connID, p)
time.Sleep(queueDuration * 3 / 2)
Expect(getQueuedPackets(connID)).To(BeNil())
})
})