Commit 917534e3 authored by Marten Seemann's avatar Marten Seemann Committed by Matt Holt

vendor: update quic-go to v0.10.1 (#2431)

parent 8ab447e6
...@@ -25,12 +25,9 @@ type client struct { ...@@ -25,12 +25,9 @@ type client struct {
// If it is started with Dial, we take a packet conn as a parameter. // If it is started with Dial, we take a packet conn as a parameter.
createdPacketConn bool createdPacketConn bool
hostname string
packetHandlers packetHandlerManager packetHandlers packetHandlerManager
token []byte token []byte
numRetries int
versionNegotiated bool // has the server accepted our version versionNegotiated bool // has the server accepted our version
receivedVersionNegotiationPacket bool receivedVersionNegotiationPacket bool
...@@ -159,13 +156,12 @@ func newClient( ...@@ -159,13 +156,12 @@ func newClient(
closeCallback func(protocol.ConnectionID), closeCallback func(protocol.ConnectionID),
createdPacketConn bool, createdPacketConn bool,
) (*client, error) { ) (*client, error) {
var hostname string if tlsConf == nil {
if tlsConf != nil { tlsConf = &tls.Config{}
hostname = tlsConf.ServerName
} }
if hostname == "" { if tlsConf.ServerName == "" {
var err error var err error
hostname, _, err = net.SplitHostPort(host) tlsConf.ServerName, _, err = net.SplitHostPort(host)
if err != nil { if err != nil {
return nil, err return nil, err
} }
...@@ -186,7 +182,6 @@ func newClient( ...@@ -186,7 +182,6 @@ func newClient(
c := &client{ c := &client{
conn: &conn{pconn: pconn, currentAddr: remoteAddr}, conn: &conn{pconn: pconn, currentAddr: remoteAddr},
createdPacketConn: createdPacketConn, createdPacketConn: createdPacketConn,
hostname: hostname,
tlsConf: tlsConf, tlsConf: tlsConf,
config: config, config: config,
version: config.Versions[0], version: config.Versions[0],
...@@ -286,7 +281,7 @@ func (c *client) generateConnectionIDs() error { ...@@ -286,7 +281,7 @@ func (c *client) generateConnectionIDs() error {
} }
func (c *client) dial(ctx context.Context) error { func (c *client) dial(ctx context.Context) error {
c.logger.Infof("Starting new connection to %s (%s -> %s), source connection ID %s, destination connection ID %s, version %s", c.hostname, c.conn.LocalAddr(), c.conn.RemoteAddr(), c.srcConnID, c.destConnID, c.version) c.logger.Infof("Starting new connection to %s (%s -> %s), source connection ID %s, destination connection ID %s, version %s", c.tlsConf.ServerName, c.conn.LocalAddr(), c.conn.RemoteAddr(), c.srcConnID, c.destConnID, c.version)
var err error var err error
if c.version.UsesTLS() { if c.version.UsesTLS() {
...@@ -324,7 +319,6 @@ func (c *client) dialTLS(ctx context.Context) error { ...@@ -324,7 +319,6 @@ func (c *client) dialTLS(ctx context.Context) error {
return err return err
} }
mintConf.ExtensionHandler = extHandler mintConf.ExtensionHandler = extHandler
mintConf.ServerName = c.hostname
c.mintConf = mintConf c.mintConf = mintConf
if err := c.createNewTLSSession(extHandler.GetPeerParams(), c.version); err != nil { if err := c.createNewTLSSession(extHandler.GetPeerParams(), c.version); err != nil {
...@@ -483,19 +477,18 @@ func (c *client) handleVersionNegotiationPacket(hdr *wire.Header) error { ...@@ -483,19 +477,18 @@ func (c *client) handleVersionNegotiationPacket(hdr *wire.Header) error {
func (c *client) handleRetryPacket(hdr *wire.Header) { func (c *client) handleRetryPacket(hdr *wire.Header) {
c.logger.Debugf("<- Received Retry") c.logger.Debugf("<- Received Retry")
hdr.Log(c.logger) hdr.Log(c.logger)
// A server that performs multiple retries must use a source connection ID of at least 8 bytes. if !hdr.OrigDestConnectionID.Equal(c.destConnID) {
// Only a server that won't send additional Retries can use shorter connection IDs. c.logger.Debugf("Ignoring spoofed Retry. Original Destination Connection ID: %s, expected: %s", hdr.OrigDestConnectionID, c.destConnID)
if hdr.OrigDestConnectionID.Len() < protocol.MinConnectionIDLenInitial {
c.logger.Debugf("Received a Retry with a too short Original Destination Connection ID: %d bytes, must have at least %d bytes.", hdr.OrigDestConnectionID.Len(), protocol.MinConnectionIDLenInitial)
return return
} }
if !hdr.OrigDestConnectionID.Equal(c.destConnID) { if hdr.SrcConnectionID.Equal(c.destConnID) {
c.logger.Debugf("Received spoofed Retry. Original Destination Connection ID: %s, expected: %s", hdr.OrigDestConnectionID, c.destConnID) c.logger.Debugf("Ignoring Retry, since the server didn't change the Source Connection ID.")
return return
} }
c.numRetries++ // If a token is already set, this means that we already received a Retry from the server.
if c.numRetries > protocol.MaxRetries { // Ignore this Retry packet.
c.session.destroy(qerr.CryptoTooManyRejects) if len(c.token) > 0 {
c.logger.Debugf("Ignoring Retry, since a Retry was already received.")
return return
} }
c.destConnID = hdr.SrcConnectionID c.destConnID = hdr.SrcConnectionID
...@@ -513,7 +506,6 @@ func (c *client) createNewGQUICSession() error { ...@@ -513,7 +506,6 @@ func (c *client) createNewGQUICSession() error {
sess, err := newClientSession( sess, err := newClientSession(
c.conn, c.conn,
runner, runner,
c.hostname,
c.version, c.version,
c.destConnID, c.destConnID,
c.srcConnID, c.srcConnID,
......
...@@ -13,6 +13,7 @@ type cryptoStream interface { ...@@ -13,6 +13,7 @@ type cryptoStream interface {
io.Reader io.Reader
io.Writer io.Writer
handleStreamFrame(*wire.StreamFrame) error handleStreamFrame(*wire.StreamFrame) error
hasData() bool
popStreamFrame(protocol.ByteCount) (*wire.StreamFrame, bool) popStreamFrame(protocol.ByteCount) (*wire.StreamFrame, bool)
closeForShutdown(error) closeForShutdown(error)
setReadOffset(protocol.ByteCount) setReadOffset(protocol.ByteCount)
......
...@@ -16,6 +16,7 @@ import ( ...@@ -16,6 +16,7 @@ import (
func main() { func main() {
verbose := flag.Bool("v", false, "verbose") verbose := flag.Bool("v", false, "verbose")
tls := flag.Bool("tls", false, "activate support for IETF QUIC (work in progress)") tls := flag.Bool("tls", false, "activate support for IETF QUIC (work in progress)")
quiet := flag.Bool("q", false, "don't print the data")
flag.Parse() flag.Parse()
urls := flag.Args() urls := flag.Args()
...@@ -57,8 +58,12 @@ func main() { ...@@ -57,8 +58,12 @@ func main() {
if err != nil { if err != nil {
panic(err) panic(err)
} }
logger.Infof("Request Body:") if *quiet {
logger.Infof("%s", body.Bytes()) logger.Infof("Request Body: %d bytes", body.Len())
} else {
logger.Infof("Request Body:")
logger.Infof("%s", body.Bytes())
}
wg.Done() wg.Done()
}(addr) }(addr)
} }
......
...@@ -7,23 +7,25 @@ import ( ...@@ -7,23 +7,25 @@ import (
"github.com/lucas-clemente/quic-go/internal/wire" "github.com/lucas-clemente/quic-go/internal/wire"
) )
type streamFramer struct { type framer struct {
streamGetter streamGetter streamGetter streamGetter
cryptoStream cryptoStream cryptoStream cryptoStream
version protocol.VersionNumber version protocol.VersionNumber
streamQueueMutex sync.Mutex streamQueueMutex sync.Mutex
activeStreams map[protocol.StreamID]struct{} activeStreams map[protocol.StreamID]struct{}
streamQueue []protocol.StreamID streamQueue []protocol.StreamID
hasCryptoStreamData bool
controlFrameMutex sync.Mutex
controlFrames []wire.Frame
} }
func newStreamFramer( func newFramer(
cryptoStream cryptoStream, cryptoStream cryptoStream,
streamGetter streamGetter, streamGetter streamGetter,
v protocol.VersionNumber, v protocol.VersionNumber,
) *streamFramer { ) *framer {
return &streamFramer{ return &framer{
streamGetter: streamGetter, streamGetter: streamGetter,
cryptoStream: cryptoStream, cryptoStream: cryptoStream,
activeStreams: make(map[protocol.StreamID]struct{}), activeStreams: make(map[protocol.StreamID]struct{}),
...@@ -31,13 +33,32 @@ func newStreamFramer( ...@@ -31,13 +33,32 @@ func newStreamFramer(
} }
} }
func (f *streamFramer) AddActiveStream(id protocol.StreamID) { func (f *framer) QueueControlFrame(frame wire.Frame) {
if id == f.version.CryptoStreamID() { // the crypto stream is handled separately f.controlFrameMutex.Lock()
f.streamQueueMutex.Lock() f.controlFrames = append(f.controlFrames, frame)
f.hasCryptoStreamData = true f.controlFrameMutex.Unlock()
f.streamQueueMutex.Unlock() }
return
func (f *framer) AppendControlFrames(frames []wire.Frame, maxLen protocol.ByteCount) ([]wire.Frame, protocol.ByteCount) {
var length protocol.ByteCount
f.controlFrameMutex.Lock()
for len(f.controlFrames) > 0 {
frame := f.controlFrames[len(f.controlFrames)-1]
frameLen := frame.Length(f.version)
if length+frameLen > maxLen {
break
}
frames = append(frames, frame)
length += frameLen
f.controlFrames = f.controlFrames[:len(f.controlFrames)-1]
} }
f.controlFrameMutex.Unlock()
return frames, length
}
// AddActiveStream adds a stream that has data to write.
// It should not be used for the crypto stream.
func (f *framer) AddActiveStream(id protocol.StreamID) {
f.streamQueueMutex.Lock() f.streamQueueMutex.Lock()
if _, ok := f.activeStreams[id]; !ok { if _, ok := f.activeStreams[id]; !ok {
f.streamQueue = append(f.streamQueue, id) f.streamQueue = append(f.streamQueue, id)
...@@ -46,29 +67,13 @@ func (f *streamFramer) AddActiveStream(id protocol.StreamID) { ...@@ -46,29 +67,13 @@ func (f *streamFramer) AddActiveStream(id protocol.StreamID) {
f.streamQueueMutex.Unlock() f.streamQueueMutex.Unlock()
} }
func (f *streamFramer) HasCryptoStreamData() bool { func (f *framer) AppendStreamFrames(frames []wire.Frame, maxLen protocol.ByteCount) []wire.Frame {
f.streamQueueMutex.Lock() var length protocol.ByteCount
hasCryptoStreamData := f.hasCryptoStreamData
f.streamQueueMutex.Unlock()
return hasCryptoStreamData
}
func (f *streamFramer) PopCryptoStreamFrame(maxLen protocol.ByteCount) *wire.StreamFrame {
f.streamQueueMutex.Lock()
frame, hasMoreData := f.cryptoStream.popStreamFrame(maxLen)
f.hasCryptoStreamData = hasMoreData
f.streamQueueMutex.Unlock()
return frame
}
func (f *streamFramer) PopStreamFrames(maxTotalLen protocol.ByteCount) []*wire.StreamFrame {
var currentLen protocol.ByteCount
var frames []*wire.StreamFrame
f.streamQueueMutex.Lock() f.streamQueueMutex.Lock()
// pop STREAM frames, until less than MinStreamFrameSize bytes are left in the packet // pop STREAM frames, until less than MinStreamFrameSize bytes are left in the packet
numActiveStreams := len(f.streamQueue) numActiveStreams := len(f.streamQueue)
for i := 0; i < numActiveStreams; i++ { for i := 0; i < numActiveStreams; i++ {
if maxTotalLen-currentLen < protocol.MinStreamFrameSize { if maxLen-length < protocol.MinStreamFrameSize {
break break
} }
id := f.streamQueue[0] id := f.streamQueue[0]
...@@ -81,7 +86,7 @@ func (f *streamFramer) PopStreamFrames(maxTotalLen protocol.ByteCount) []*wire.S ...@@ -81,7 +86,7 @@ func (f *streamFramer) PopStreamFrames(maxTotalLen protocol.ByteCount) []*wire.S
delete(f.activeStreams, id) delete(f.activeStreams, id)
continue continue
} }
frame, hasMoreData := str.popStreamFrame(maxTotalLen - currentLen) frame, hasMoreData := str.popStreamFrame(maxLen - length)
if hasMoreData { // put the stream back in the queue (at the end) if hasMoreData { // put the stream back in the queue (at the end)
f.streamQueue = append(f.streamQueue, id) f.streamQueue = append(f.streamQueue, id)
} else { // no more data to send. Stream is not active any more } else { // no more data to send. Stream is not active any more
...@@ -91,7 +96,7 @@ func (f *streamFramer) PopStreamFrames(maxTotalLen protocol.ByteCount) []*wire.S ...@@ -91,7 +96,7 @@ func (f *streamFramer) PopStreamFrames(maxTotalLen protocol.ByteCount) []*wire.S
continue continue
} }
frames = append(frames, frame) frames = append(frames, frame)
currentLen += frame.Length(f.version) length += frame.Length(f.version)
} }
f.streamQueueMutex.Unlock() f.streamQueueMutex.Unlock()
return frames return frames
......
...@@ -141,7 +141,7 @@ func (s *Server) handleHeaderStream(session streamCreator) { ...@@ -141,7 +141,7 @@ func (s *Server) handleHeaderStream(session streamCreator) {
// In this case, the session has already logged the error, so we don't // In this case, the session has already logged the error, so we don't
// need to log it again. // need to log it again.
errorCode := qerr.InternalError errorCode := qerr.InternalError
if qerr, ok := err.(*qerr.QuicError); !ok { if qerr, ok := err.(*qerr.QuicError); ok {
errorCode = qerr.ErrorCode errorCode = qerr.ErrorCode
s.logger.Errorf("error handling h2 request: %s", err.Error()) s.logger.Errorf("error handling h2 request: %s", err.Error())
} }
......
...@@ -23,8 +23,6 @@ const ( ...@@ -23,8 +23,6 @@ const (
VersionGQUIC43 = protocol.Version43 VersionGQUIC43 = protocol.Version43
// VersionGQUIC44 is gQUIC version 44. // VersionGQUIC44 is gQUIC version 44.
VersionGQUIC44 = protocol.Version44 VersionGQUIC44 = protocol.Version44
// VersionMilestone0_10_0 uses TLS
VersionMilestone0_10_0 = protocol.VersionMilestone0_10_0
) )
// A Cookie can be used to verify the ownership of the client address. // A Cookie can be used to verify the ownership of the client address.
......
...@@ -148,6 +148,7 @@ func (h *sentPacketHandler) SentPacketsAsRetransmission(packets []*Packet, retra ...@@ -148,6 +148,7 @@ func (h *sentPacketHandler) SentPacketsAsRetransmission(packets []*Packet, retra
func (h *sentPacketHandler) sentPacketImpl(packet *Packet) bool /* isRetransmittable */ { func (h *sentPacketHandler) sentPacketImpl(packet *Packet) bool /* isRetransmittable */ {
for p := h.lastSentPacketNumber + 1; p < packet.PacketNumber; p++ { for p := h.lastSentPacketNumber + 1; p < packet.PacketNumber; p++ {
h.logger.Debugf("Skipping packet number %#x", p)
h.skippedPackets = append(h.skippedPackets, p) h.skippedPackets = append(h.skippedPackets, p)
if len(h.skippedPackets) > protocol.MaxTrackedSkippedPackets { if len(h.skippedPackets) > protocol.MaxTrackedSkippedPackets {
h.skippedPackets = h.skippedPackets[1:] h.skippedPackets = h.skippedPackets[1:]
...@@ -633,7 +634,7 @@ func (h *sentPacketHandler) computeRTOTimeout() time.Duration { ...@@ -633,7 +634,7 @@ func (h *sentPacketHandler) computeRTOTimeout() time.Duration {
} }
rto = utils.MaxDuration(rto, minRTOTimeout) rto = utils.MaxDuration(rto, minRTOTimeout)
// Exponential backoff // Exponential backoff
rto = rto << h.rtoCount rto <<= h.rtoCount
return utils.MinDuration(rto, maxRTOTimeout) return utils.MinDuration(rto, maxRTOTimeout)
} }
......
...@@ -193,7 +193,7 @@ func (c *cubicSender) OnPacketLost( ...@@ -193,7 +193,7 @@ func (c *cubicSender) OnPacketLost(
if c.congestionWindow >= 2*c.initialCongestionWindow { if c.congestionWindow >= 2*c.initialCongestionWindow {
c.minSlowStartExitWindow = c.congestionWindow / 2 c.minSlowStartExitWindow = c.congestionWindow / 2
} }
c.congestionWindow = c.congestionWindow - protocol.DefaultTCPMSS c.congestionWindow -= protocol.DefaultTCPMSS
} else if c.reno { } else if c.reno {
c.congestionWindow = protocol.ByteCount(float32(c.congestionWindow) * c.RenoBeta()) c.congestionWindow = protocol.ByteCount(float32(c.congestionWindow) * c.RenoBeta())
} else { } else {
......
...@@ -56,8 +56,7 @@ func (c *certChain) GetLeafCert(sni string) ([]byte, error) { ...@@ -56,8 +56,7 @@ func (c *certChain) GetLeafCert(sni string) ([]byte, error) {
} }
func (c *certChain) getCertForSNI(sni string) (*tls.Certificate, error) { func (c *certChain) getCertForSNI(sni string) (*tls.Certificate, error) {
conf := c.config conf, err := maybeGetConfigForClient(c.config, sni)
conf, err := maybeGetConfigForClient(conf, sni)
if err != nil { if err != nil {
return nil, err return nil, err
} }
...@@ -107,7 +106,13 @@ func maybeGetConfigForClient(c *tls.Config, sni string) (*tls.Config, error) { ...@@ -107,7 +106,13 @@ func maybeGetConfigForClient(c *tls.Config, sni string) (*tls.Config, error) {
if c.GetConfigForClient == nil { if c.GetConfigForClient == nil {
return c, nil return c, nil
} }
return c.GetConfigForClient(&tls.ClientHelloInfo{ confForClient, err := c.GetConfigForClient(&tls.ClientHelloInfo{ServerName: sni})
ServerName: sni, if err != nil {
}) return nil, err
}
// if GetConfigForClient returns nil, use the original config
if confForClient == nil {
return c, nil
}
return confForClient, nil
} }
package crypto
import (
"crypto"
"crypto/hmac"
"encoding/binary"
)
// copied from https://github.com/cloudflare/tls-tris/blob/master/hkdf.go
func hkdfExtract(hash crypto.Hash, secret, salt []byte) []byte {
if salt == nil {
salt = make([]byte, hash.Size())
}
if secret == nil {
secret = make([]byte, hash.Size())
}
extractor := hmac.New(hash.New, salt)
extractor.Write(secret)
return extractor.Sum(nil)
}
// copied from https://github.com/cloudflare/tls-tris/blob/master/hkdf.go
func hkdfExpand(hash crypto.Hash, prk, info []byte, l int) []byte {
var (
expander = hmac.New(hash.New, prk)
res = make([]byte, l)
counter = byte(1)
prev []byte
)
if l > 255*expander.Size() {
panic("hkdf: requested too much output")
}
p := res
for len(p) > 0 {
expander.Reset()
expander.Write(prev)
expander.Write(info)
expander.Write([]byte{counter})
prev = expander.Sum(prev[:0])
counter++
n := copy(p, prev)
p = p[n:]
}
return res
}
func qhkdfExpand(secret []byte, label string, length int) []byte {
qlabel := make([]byte, 2+1+5+len(label))
binary.BigEndian.PutUint16(qlabel[0:2], uint16(length))
qlabel[2] = uint8(5 + len(label))
copy(qlabel[3:], []byte("QUIC "+label))
return hkdfExpand(crypto.SHA256, secret, qlabel, length)
}
package crypto package crypto
import ( import (
"crypto"
"encoding/binary"
"github.com/bifurcation/mint" "github.com/bifurcation/mint"
"github.com/lucas-clemente/quic-go/internal/protocol" "github.com/lucas-clemente/quic-go/internal/protocol"
) )
...@@ -19,14 +16,6 @@ type TLSExporter interface { ...@@ -19,14 +16,6 @@ type TLSExporter interface {
ComputeExporter(label string, context []byte, keyLength int) ([]byte, error) ComputeExporter(label string, context []byte, keyLength int) ([]byte, error)
} }
func qhkdfExpand(secret []byte, label string, length int) []byte {
qlabel := make([]byte, 2+1+5+len(label))
binary.BigEndian.PutUint16(qlabel[0:2], uint16(length))
qlabel[2] = uint8(5 + len(label))
copy(qlabel[3:], []byte("QUIC "+label))
return mint.HkdfExpand(crypto.SHA256, secret, qlabel, length)
}
// DeriveAESKeys derives the AES keys and creates a matching AES-GCM AEAD instance // DeriveAESKeys derives the AES keys and creates a matching AES-GCM AEAD instance
func DeriveAESKeys(tls TLSExporter, pers protocol.Perspective) (AEAD, error) { func DeriveAESKeys(tls TLSExporter, pers protocol.Perspective) (AEAD, error) {
var myLabel, otherLabel string var myLabel, otherLabel string
......
...@@ -3,7 +3,6 @@ package crypto ...@@ -3,7 +3,6 @@ package crypto
import ( import (
"crypto" "crypto"
"github.com/bifurcation/mint"
"github.com/lucas-clemente/quic-go/internal/protocol" "github.com/lucas-clemente/quic-go/internal/protocol"
) )
...@@ -28,7 +27,7 @@ func newNullAEADAESGCM(connectionID protocol.ConnectionID, pers protocol.Perspec ...@@ -28,7 +27,7 @@ func newNullAEADAESGCM(connectionID protocol.ConnectionID, pers protocol.Perspec
} }
func computeSecrets(connID protocol.ConnectionID) (clientSecret, serverSecret []byte) { func computeSecrets(connID protocol.ConnectionID) (clientSecret, serverSecret []byte) {
handshakeSecret := mint.HkdfExtract(crypto.SHA256, quicVersion1Salt, connID) handshakeSecret := hkdfExtract(crypto.SHA256, connID, quicVersion1Salt)
clientSecret = qhkdfExpand(handshakeSecret, "client hs", crypto.SHA256.Size()) clientSecret = qhkdfExpand(handshakeSecret, "client hs", crypto.SHA256.Size())
serverSecret = qhkdfExpand(handshakeSecret, "server hs", crypto.SHA256.Size()) serverSecret = qhkdfExpand(handshakeSecret, "server hs", crypto.SHA256.Size())
return return
......
...@@ -69,10 +69,9 @@ var ( ...@@ -69,10 +69,9 @@ var (
// NewCryptoSetupClient creates a new CryptoSetup instance for a client // NewCryptoSetupClient creates a new CryptoSetup instance for a client
func NewCryptoSetupClient( func NewCryptoSetupClient(
cryptoStream io.ReadWriter, cryptoStream io.ReadWriter,
hostname string,
connID protocol.ConnectionID, connID protocol.ConnectionID,
version protocol.VersionNumber, version protocol.VersionNumber,
tlsConfig *tls.Config, tlsConf *tls.Config,
params *TransportParameters, params *TransportParameters,
paramsChan chan<- TransportParameters, paramsChan chan<- TransportParameters,
handshakeEvent chan<- struct{}, handshakeEvent chan<- struct{},
...@@ -87,10 +86,10 @@ func NewCryptoSetupClient( ...@@ -87,10 +86,10 @@ func NewCryptoSetupClient(
divNonceChan := make(chan struct{}) divNonceChan := make(chan struct{})
cs := &cryptoSetupClient{ cs := &cryptoSetupClient{
cryptoStream: cryptoStream, cryptoStream: cryptoStream,
hostname: hostname, hostname: tlsConf.ServerName,
connID: connID, connID: connID,
version: version, version: version,
certManager: crypto.NewCertManager(tlsConfig), certManager: crypto.NewCertManager(tlsConf),
params: params, params: params,
keyDerivation: crypto.DeriveQuicCryptoAESKeys, keyDerivation: crypto.DeriveQuicCryptoAESKeys,
nullAEAD: nullAEAD, nullAEAD: nullAEAD,
......
...@@ -3,8 +3,8 @@ package handshake ...@@ -3,8 +3,8 @@ package handshake
import ( import (
"bytes" "bytes"
"encoding/binary" "encoding/binary"
"errors"
"fmt" "fmt"
"sort"
"time" "time"
"github.com/lucas-clemente/quic-go/internal/protocol" "github.com/lucas-clemente/quic-go/internal/protocol"
...@@ -97,16 +97,18 @@ func (p *TransportParameters) getHelloMap() map[Tag][]byte { ...@@ -97,16 +97,18 @@ func (p *TransportParameters) getHelloMap() map[Tag][]byte {
} }
func (p *TransportParameters) unmarshal(data []byte) error { func (p *TransportParameters) unmarshal(data []byte) error {
var foundIdleTimeout bool // needed to check that every parameter is only sent at most once
var parameterIDs []transportParameterID
for len(data) >= 4 { for len(data) >= 4 {
paramID := binary.BigEndian.Uint16(data[:2]) paramID := transportParameterID(binary.BigEndian.Uint16(data[:2]))
paramLen := int(binary.BigEndian.Uint16(data[2:4])) paramLen := int(binary.BigEndian.Uint16(data[2:4]))
data = data[4:] data = data[4:]
if len(data) < paramLen { if len(data) < paramLen {
return fmt.Errorf("remaining length (%d) smaller than parameter length (%d)", len(data), paramLen) return fmt.Errorf("remaining length (%d) smaller than parameter length (%d)", len(data), paramLen)
} }
switch transportParameterID(paramID) { parameterIDs = append(parameterIDs, paramID)
switch paramID {
case initialMaxStreamDataParameterID: case initialMaxStreamDataParameterID:
if paramLen != 4 { if paramLen != 4 {
return fmt.Errorf("wrong length for initial_max_stream_data: %d (expected 4)", paramLen) return fmt.Errorf("wrong length for initial_max_stream_data: %d (expected 4)", paramLen)
...@@ -128,7 +130,6 @@ func (p *TransportParameters) unmarshal(data []byte) error { ...@@ -128,7 +130,6 @@ func (p *TransportParameters) unmarshal(data []byte) error {
} }
p.MaxUniStreams = binary.BigEndian.Uint16(data[:2]) p.MaxUniStreams = binary.BigEndian.Uint16(data[:2])
case idleTimeoutParameterID: case idleTimeoutParameterID:
foundIdleTimeout = true
if paramLen != 2 { if paramLen != 2 {
return fmt.Errorf("wrong length for idle_timeout: %d (expected 2)", paramLen) return fmt.Errorf("wrong length for idle_timeout: %d (expected 2)", paramLen)
} }
...@@ -156,12 +157,17 @@ func (p *TransportParameters) unmarshal(data []byte) error { ...@@ -156,12 +157,17 @@ func (p *TransportParameters) unmarshal(data []byte) error {
data = data[paramLen:] data = data[paramLen:]
} }
// check that every transport parameter was sent at most once
sort.Slice(parameterIDs, func(i, j int) bool { return parameterIDs[i] < parameterIDs[j] })
for i := 0; i < len(parameterIDs)-1; i++ {
if parameterIDs[i] == parameterIDs[i+1] {
return fmt.Errorf("received duplicate transport parameter %#x", parameterIDs[i])
}
}
if len(data) != 0 { if len(data) != 0 {
return fmt.Errorf("should have read all data. Still have %d bytes", len(data)) return fmt.Errorf("should have read all data. Still have %d bytes", len(data))
} }
if !foundIdleTimeout {
return errors.New("missing parameter")
}
return nil return nil
} }
......
package mocks package mocks
//go:generate sh -c "../mockgen_internal.sh mocks tls_extension_handler.go github.com/lucas-clemente/quic-go/internal/handshake TLSExtensionHandler" //go:generate sh -c "../mockgen_internal.sh mocks tls_extension_handler.go github.com/lucas-clemente/quic-go/internal/handshake TLSExtensionHandler"
//go:generate sh -c "../mockgen_internal.sh mocks sealer.go github.com/lucas-clemente/quic-go/internal/handshake Sealer"
//go:generate sh -c "../mockgen_internal.sh mocks stream_flow_controller.go github.com/lucas-clemente/quic-go/internal/flowcontrol StreamFlowController" //go:generate sh -c "../mockgen_internal.sh mocks stream_flow_controller.go github.com/lucas-clemente/quic-go/internal/flowcontrol StreamFlowController"
//go:generate sh -c "../mockgen_internal.sh mockackhandler ackhandler/sent_packet_handler.go github.com/lucas-clemente/quic-go/internal/ackhandler SentPacketHandler" //go:generate sh -c "../mockgen_internal.sh mockackhandler ackhandler/sent_packet_handler.go github.com/lucas-clemente/quic-go/internal/ackhandler SentPacketHandler"
//go:generate sh -c "../mockgen_internal.sh mockackhandler ackhandler/received_packet_handler.go github.com/lucas-clemente/quic-go/internal/ackhandler ReceivedPacketHandler" //go:generate sh -c "../mockgen_internal.sh mockackhandler ackhandler/received_packet_handler.go github.com/lucas-clemente/quic-go/internal/ackhandler ReceivedPacketHandler"
......
// Code generated by MockGen. DO NOT EDIT.
// Source: github.com/lucas-clemente/quic-go/internal/handshake (interfaces: Sealer)
// Package mocks is a generated GoMock package.
package mocks
import (
reflect "reflect"
gomock "github.com/golang/mock/gomock"
protocol "github.com/lucas-clemente/quic-go/internal/protocol"
)
// MockSealer is a mock of Sealer interface
type MockSealer struct {
ctrl *gomock.Controller
recorder *MockSealerMockRecorder
}
// MockSealerMockRecorder is the mock recorder for MockSealer
type MockSealerMockRecorder struct {
mock *MockSealer
}
// NewMockSealer creates a new mock instance
func NewMockSealer(ctrl *gomock.Controller) *MockSealer {
mock := &MockSealer{ctrl: ctrl}
mock.recorder = &MockSealerMockRecorder{mock}
return mock
}
// EXPECT returns an object that allows the caller to indicate expected use
func (m *MockSealer) EXPECT() *MockSealerMockRecorder {
return m.recorder
}
// Overhead mocks base method
func (m *MockSealer) Overhead() int {
ret := m.ctrl.Call(m, "Overhead")
ret0, _ := ret[0].(int)
return ret0
}
// Overhead indicates an expected call of Overhead
func (mr *MockSealerMockRecorder) Overhead() *gomock.Call {
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Overhead", reflect.TypeOf((*MockSealer)(nil).Overhead))
}
// Seal mocks base method
func (m *MockSealer) Seal(arg0, arg1 []byte, arg2 protocol.PacketNumber, arg3 []byte) []byte {
ret := m.ctrl.Call(m, "Seal", arg0, arg1, arg2, arg3)
ret0, _ := ret[0].([]byte)
return ret0
}
// Seal indicates an expected call of Seal
func (mr *MockSealerMockRecorder) Seal(arg0, arg1, arg2, arg3 interface{}) *gomock.Call {
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Seal", reflect.TypeOf((*MockSealer)(nil).Seal), arg0, arg1, arg2, arg3)
}
...@@ -149,6 +149,3 @@ const MinPacingDelay time.Duration = 100 * time.Microsecond ...@@ -149,6 +149,3 @@ const MinPacingDelay time.Duration = 100 * time.Microsecond
// DefaultConnectionIDLength is the connection ID length that is used for multiplexed connections // DefaultConnectionIDLength is the connection ID length that is used for multiplexed connections
// if no other value is configured. // if no other value is configured.
const DefaultConnectionIDLength = 4 const DefaultConnectionIDLength = 4
// MaxRetries is the maximum number of Retries a client will do before failing the connection.
const MaxRetries = 3
...@@ -24,8 +24,6 @@ const ( ...@@ -24,8 +24,6 @@ const (
VersionTLS VersionNumber = 101 VersionTLS VersionNumber = 101
VersionWhatever VersionNumber = 0 // for when the version doesn't matter VersionWhatever VersionNumber = 0 // for when the version doesn't matter
VersionUnknown VersionNumber = math.MaxUint32 VersionUnknown VersionNumber = math.MaxUint32
VersionMilestone0_10_0 VersionNumber = 0x51474f02
) )
// SupportedVersions lists the versions that the server supports // SupportedVersions lists the versions that the server supports
...@@ -38,7 +36,7 @@ var SupportedVersions = []VersionNumber{ ...@@ -38,7 +36,7 @@ var SupportedVersions = []VersionNumber{
// IsValidVersion says if the version is known to quic-go // IsValidVersion says if the version is known to quic-go
func IsValidVersion(v VersionNumber) bool { func IsValidVersion(v VersionNumber) bool {
return v == VersionTLS || v == VersionMilestone0_10_0 || IsSupportedVersion(SupportedVersions, v) return v == VersionTLS || IsSupportedVersion(SupportedVersions, v)
} }
// UsesTLS says if this QUIC version uses TLS 1.3 for the handshake // UsesTLS says if this QUIC version uses TLS 1.3 for the handshake
...@@ -52,8 +50,6 @@ func (vn VersionNumber) String() string { ...@@ -52,8 +50,6 @@ func (vn VersionNumber) String() string {
return "whatever" return "whatever"
case VersionUnknown: case VersionUnknown:
return "unknown" return "unknown"
case VersionMilestone0_10_0:
return "quic-go Milestone 0.10.0"
case VersionTLS: case VersionTLS:
return "TLS dev version (WIP)" return "TLS dev version (WIP)"
default: default:
......
package utils package utils
import "time" import (
"math"
"time"
)
// A Timer wrapper that behaves correctly when resetting // A Timer wrapper that behaves correctly when resetting
type Timer struct { type Timer struct {
...@@ -11,7 +14,7 @@ type Timer struct { ...@@ -11,7 +14,7 @@ type Timer struct {
// NewTimer creates a new timer that is not set // NewTimer creates a new timer that is not set
func NewTimer() *Timer { func NewTimer() *Timer {
return &Timer{t: time.NewTimer(0)} return &Timer{t: time.NewTimer(time.Duration(math.MaxInt64))}
} }
// Chan returns the channel of the wrapped timer // Chan returns the channel of the wrapped timer
...@@ -31,7 +34,9 @@ func (t *Timer) Reset(deadline time.Time) { ...@@ -31,7 +34,9 @@ func (t *Timer) Reset(deadline time.Time) {
if !t.t.Stop() && !t.read { if !t.t.Stop() && !t.read {
<-t.t.C <-t.t.C
} }
t.t.Reset(time.Until(deadline)) if !deadline.IsZero() {
t.t.Reset(time.Until(deadline))
}
t.read = false t.read = false
t.deadline = deadline t.deadline = deadline
......
...@@ -19,23 +19,17 @@ type AckFrame struct { ...@@ -19,23 +19,17 @@ type AckFrame struct {
DelayTime time.Duration DelayTime time.Duration
} }
func parseAckFrame(r *bytes.Reader, version protocol.VersionNumber) (*AckFrame, error) {
return parseAckOrAckEcnFrame(r, false, version)
}
func parseAckEcnFrame(r *bytes.Reader, version protocol.VersionNumber) (*AckFrame, error) {
return parseAckOrAckEcnFrame(r, true, version)
}
// parseAckFrame reads an ACK frame // parseAckFrame reads an ACK frame
func parseAckOrAckEcnFrame(r *bytes.Reader, ecn bool, version protocol.VersionNumber) (*AckFrame, error) { func parseAckFrame(r *bytes.Reader, version protocol.VersionNumber) (*AckFrame, error) {
if !version.UsesIETFFrameFormat() { if !version.UsesIETFFrameFormat() {
return parseAckFrameLegacy(r, version) return parseAckFrameLegacy(r, version)
} }
if _, err := r.ReadByte(); err != nil { typeByte, err := r.ReadByte()
if err != nil {
return nil, err return nil, err
} }
ecn := typeByte&0x1 > 0
frame := &AckFrame{} frame := &AckFrame{}
...@@ -50,14 +44,6 @@ func parseAckOrAckEcnFrame(r *bytes.Reader, ecn bool, version protocol.VersionNu ...@@ -50,14 +44,6 @@ func parseAckOrAckEcnFrame(r *bytes.Reader, ecn bool, version protocol.VersionNu
} }
frame.DelayTime = time.Duration(delay*1<<ackDelayExponent) * time.Microsecond frame.DelayTime = time.Duration(delay*1<<ackDelayExponent) * time.Microsecond
if ecn {
for i := 0; i < 3; i++ {
if _, err := utils.ReadVarInt(r); err != nil {
return nil, err
}
}
}
numBlocks, err := utils.ReadVarInt(r) numBlocks, err := utils.ReadVarInt(r)
if err != nil { if err != nil {
return nil, err return nil, err
...@@ -103,6 +89,16 @@ func parseAckOrAckEcnFrame(r *bytes.Reader, ecn bool, version protocol.VersionNu ...@@ -103,6 +89,16 @@ func parseAckOrAckEcnFrame(r *bytes.Reader, ecn bool, version protocol.VersionNu
if !frame.validateAckRanges() { if !frame.validateAckRanges() {
return nil, errInvalidAckRanges return nil, errInvalidAckRanges
} }
// parse (and skip) the ECN section
if ecn {
for i := 0; i < 3; i++ {
if _, err := utils.ReadVarInt(r); err != nil {
return nil, err
}
}
}
return frame, nil return frame, nil
} }
...@@ -112,7 +108,7 @@ func (f *AckFrame) Write(b *bytes.Buffer, version protocol.VersionNumber) error ...@@ -112,7 +108,7 @@ func (f *AckFrame) Write(b *bytes.Buffer, version protocol.VersionNumber) error
return f.writeLegacy(b, version) return f.writeLegacy(b, version)
} }
b.WriteByte(0x0d) b.WriteByte(0x1a)
utils.WriteVarInt(b, uint64(f.LargestAcked())) utils.WriteVarInt(b, uint64(f.LargestAcked()))
utils.WriteVarInt(b, encodeAckDelay(f.DelayTime)) utils.WriteVarInt(b, encodeAckDelay(f.DelayTime))
......
...@@ -86,11 +86,6 @@ func parseIETFFrame(r *bytes.Reader, typeByte byte, v protocol.VersionNumber) (F ...@@ -86,11 +86,6 @@ func parseIETFFrame(r *bytes.Reader, typeByte byte, v protocol.VersionNumber) (F
if err != nil { if err != nil {
err = qerr.Error(qerr.InvalidFrameData, err.Error()) err = qerr.Error(qerr.InvalidFrameData, err.Error())
} }
case 0xd:
frame, err = parseAckFrame(r, v)
if err != nil {
err = qerr.Error(qerr.InvalidAckData, err.Error())
}
case 0xe: case 0xe:
frame, err = parsePathChallengeFrame(r, v) frame, err = parsePathChallengeFrame(r, v)
if err != nil { if err != nil {
...@@ -101,8 +96,8 @@ func parseIETFFrame(r *bytes.Reader, typeByte byte, v protocol.VersionNumber) (F ...@@ -101,8 +96,8 @@ func parseIETFFrame(r *bytes.Reader, typeByte byte, v protocol.VersionNumber) (F
if err != nil { if err != nil {
err = qerr.Error(qerr.InvalidFrameData, err.Error()) err = qerr.Error(qerr.InvalidFrameData, err.Error())
} }
case 0x1a: case 0x1a, 0x1b:
frame, err = parseAckEcnFrame(r, v) frame, err = parseAckFrame(r, v)
if err != nil { if err != nil {
err = qerr.Error(qerr.InvalidAckData, err.Error()) err = qerr.Error(qerr.InvalidAckData, err.Error())
} }
......
...@@ -5,10 +5,13 @@ package quic ...@@ -5,10 +5,13 @@ package quic
//go:generate sh -c "./mockgen_private.sh quic mock_send_stream_internal_test.go github.com/lucas-clemente/quic-go sendStreamI" //go:generate sh -c "./mockgen_private.sh quic mock_send_stream_internal_test.go github.com/lucas-clemente/quic-go sendStreamI"
//go:generate sh -c "./mockgen_private.sh quic mock_stream_sender_test.go github.com/lucas-clemente/quic-go streamSender" //go:generate sh -c "./mockgen_private.sh quic mock_stream_sender_test.go github.com/lucas-clemente/quic-go streamSender"
//go:generate sh -c "./mockgen_private.sh quic mock_stream_getter_test.go github.com/lucas-clemente/quic-go streamGetter" //go:generate sh -c "./mockgen_private.sh quic mock_stream_getter_test.go github.com/lucas-clemente/quic-go streamGetter"
//go:generate sh -c "./mockgen_private.sh quic mock_stream_frame_source_test.go github.com/lucas-clemente/quic-go streamFrameSource" //go:generate sh -c "./mockgen_private.sh quic mock_frame_source_test.go github.com/lucas-clemente/quic-go frameSource"
//go:generate sh -c "./mockgen_private.sh quic mock_ack_frame_source_test.go github.com/lucas-clemente/quic-go ackFrameSource"
//go:generate sh -c "./mockgen_private.sh quic mock_crypto_stream_test.go github.com/lucas-clemente/quic-go cryptoStream" //go:generate sh -c "./mockgen_private.sh quic mock_crypto_stream_test.go github.com/lucas-clemente/quic-go cryptoStream"
//go:generate sh -c "./mockgen_private.sh quic mock_stream_manager_test.go github.com/lucas-clemente/quic-go streamManager" //go:generate sh -c "./mockgen_private.sh quic mock_stream_manager_test.go github.com/lucas-clemente/quic-go streamManager"
//go:generate sh -c "./mockgen_private.sh quic mock_sealing_manager_test.go github.com/lucas-clemente/quic-go sealingManager"
//go:generate sh -c "./mockgen_private.sh quic mock_unpacker_test.go github.com/lucas-clemente/quic-go unpacker" //go:generate sh -c "./mockgen_private.sh quic mock_unpacker_test.go github.com/lucas-clemente/quic-go unpacker"
//go:generate sh -c "./mockgen_private.sh quic mock_packer_test.go github.com/lucas-clemente/quic-go packer"
//go:generate sh -c "./mockgen_private.sh quic mock_quic_aead_test.go github.com/lucas-clemente/quic-go quicAEAD" //go:generate sh -c "./mockgen_private.sh quic mock_quic_aead_test.go github.com/lucas-clemente/quic-go quicAEAD"
//go:generate sh -c "./mockgen_private.sh quic mock_gquic_aead_test.go github.com/lucas-clemente/quic-go gQUICAEAD" //go:generate sh -c "./mockgen_private.sh quic mock_gquic_aead_test.go github.com/lucas-clemente/quic-go gQUICAEAD"
//go:generate sh -c "./mockgen_private.sh quic mock_session_runner_test.go github.com/lucas-clemente/quic-go sessionRunner" //go:generate sh -c "./mockgen_private.sh quic mock_session_runner_test.go github.com/lucas-clemente/quic-go sessionRunner"
......
...@@ -18,10 +18,12 @@ type packetNumberGenerator struct { ...@@ -18,10 +18,12 @@ type packetNumberGenerator struct {
} }
func newPacketNumberGenerator(initial, averagePeriod protocol.PacketNumber) *packetNumberGenerator { func newPacketNumberGenerator(initial, averagePeriod protocol.PacketNumber) *packetNumberGenerator {
return &packetNumberGenerator{ g := &packetNumberGenerator{
next: initial, next: initial,
averagePeriod: averagePeriod, averagePeriod: averagePeriod,
} }
g.generateNewSkip()
return g
} }
func (p *packetNumberGenerator) Peek() protocol.PacketNumber { func (p *packetNumberGenerator) Peek() protocol.PacketNumber {
...@@ -42,28 +44,19 @@ func (p *packetNumberGenerator) Pop() protocol.PacketNumber { ...@@ -42,28 +44,19 @@ func (p *packetNumberGenerator) Pop() protocol.PacketNumber {
return next return next
} }
func (p *packetNumberGenerator) generateNewSkip() error { func (p *packetNumberGenerator) generateNewSkip() {
num, err := p.getRandomNumber() num := p.getRandomNumber()
if err != nil {
return err
}
skip := protocol.PacketNumber(num) * (p.averagePeriod - 1) / (math.MaxUint16 / 2) skip := protocol.PacketNumber(num) * (p.averagePeriod - 1) / (math.MaxUint16 / 2)
// make sure that there are never two consecutive packet numbers that are skipped // make sure that there are never two consecutive packet numbers that are skipped
p.nextToSkip = p.next + 2 + skip p.nextToSkip = p.next + 2 + skip
return nil
} }
// getRandomNumber() generates a cryptographically secure random number between 0 and MaxUint16 (= 65535) // getRandomNumber() generates a cryptographically secure random number between 0 and MaxUint16 (= 65535)
// The expectation value is 65535/2 // The expectation value is 65535/2
func (p *packetNumberGenerator) getRandomNumber() (uint16, error) { func (p *packetNumberGenerator) getRandomNumber() uint16 {
b := make([]byte, 2) b := make([]byte, 2)
_, err := rand.Read(b) rand.Read(b) // ignore the error here
if err != nil {
return 0, err
}
num := uint16(b[0])<<8 + uint16(b[1]) num := uint16(b[0])<<8 + uint16(b[1])
return num, nil return num
} }
...@@ -8,6 +8,7 @@ import ( ...@@ -8,6 +8,7 @@ import (
"github.com/lucas-clemente/quic-go/internal/flowcontrol" "github.com/lucas-clemente/quic-go/internal/flowcontrol"
"github.com/lucas-clemente/quic-go/internal/protocol" "github.com/lucas-clemente/quic-go/internal/protocol"
"github.com/lucas-clemente/quic-go/internal/utils"
"github.com/lucas-clemente/quic-go/internal/wire" "github.com/lucas-clemente/quic-go/internal/wire"
) )
...@@ -43,8 +44,8 @@ type receiveStream struct { ...@@ -43,8 +44,8 @@ type receiveStream struct {
canceledRead bool // set when CancelRead() is called canceledRead bool // set when CancelRead() is called
resetRemotely bool // set when HandleRstStreamFrame() is called resetRemotely bool // set when HandleRstStreamFrame() is called
readChan chan struct{} readChan chan struct{}
readDeadline time.Time deadline time.Time
flowController flowcontrol.StreamFlowController flowController flowcontrol.StreamFlowController
version protocol.VersionNumber version protocol.VersionNumber
...@@ -108,6 +109,7 @@ func (s *receiveStream) readImpl(p []byte) (bool /*stream completed */, int, err ...@@ -108,6 +109,7 @@ func (s *receiveStream) readImpl(p []byte) (bool /*stream completed */, int, err
return false, bytesRead, s.closeForShutdownErr return false, bytesRead, s.closeForShutdownErr
} }
var deadlineTimer *utils.Timer
for { for {
// Stop waiting on errors // Stop waiting on errors
if s.closedForShutdown { if s.closedForShutdown {
...@@ -120,9 +122,15 @@ func (s *receiveStream) readImpl(p []byte) (bool /*stream completed */, int, err ...@@ -120,9 +122,15 @@ func (s *receiveStream) readImpl(p []byte) (bool /*stream completed */, int, err
return false, bytesRead, s.resetRemotelyErr return false, bytesRead, s.resetRemotelyErr
} }
deadline := s.readDeadline deadline := s.deadline
if !deadline.IsZero() && !time.Now().Before(deadline) { if !deadline.IsZero() {
return false, bytesRead, errDeadline if !time.Now().Before(deadline) {
return false, bytesRead, errDeadline
}
if deadlineTimer == nil {
deadlineTimer = utils.NewTimer()
}
deadlineTimer.Reset(deadline)
} }
if s.currentFrame != nil || s.currentFrameIsLast { if s.currentFrame != nil || s.currentFrameIsLast {
...@@ -135,7 +143,8 @@ func (s *receiveStream) readImpl(p []byte) (bool /*stream completed */, int, err ...@@ -135,7 +143,8 @@ func (s *receiveStream) readImpl(p []byte) (bool /*stream completed */, int, err
} else { } else {
select { select {
case <-s.readChan: case <-s.readChan:
case <-time.After(time.Until(deadline)): case <-deadlineTimer.Chan():
deadlineTimer.SetRead()
} }
} }
s.mutex.Lock() s.mutex.Lock()
...@@ -164,7 +173,9 @@ func (s *receiveStream) readImpl(p []byte) (bool /*stream completed */, int, err ...@@ -164,7 +173,9 @@ func (s *receiveStream) readImpl(p []byte) (bool /*stream completed */, int, err
s.flowController.AddBytesRead(protocol.ByteCount(m)) s.flowController.AddBytesRead(protocol.ByteCount(m))
} }
// increase the flow control window, if necessary // increase the flow control window, if necessary
s.flowController.MaybeQueueWindowUpdate() if s.streamID != s.version.CryptoStreamID() {
s.flowController.MaybeQueueWindowUpdate()
}
if s.readPosInFrame >= len(s.currentFrame) && s.currentFrameIsLast { if s.readPosInFrame >= len(s.currentFrame) && s.currentFrameIsLast {
s.finRead = true s.finRead = true
...@@ -270,13 +281,9 @@ func (s *receiveStream) onClose(offset protocol.ByteCount) { ...@@ -270,13 +281,9 @@ func (s *receiveStream) onClose(offset protocol.ByteCount) {
func (s *receiveStream) SetReadDeadline(t time.Time) error { func (s *receiveStream) SetReadDeadline(t time.Time) error {
s.mutex.Lock() s.mutex.Lock()
oldDeadline := s.readDeadline s.deadline = t
s.readDeadline = t
s.mutex.Unlock() s.mutex.Unlock()
// if the new deadline is before the currently set deadline, wake up Read() s.signalRead()
if t.Before(oldDeadline) {
s.signalRead()
}
return nil return nil
} }
......
...@@ -15,6 +15,7 @@ import ( ...@@ -15,6 +15,7 @@ import (
type sendStreamI interface { type sendStreamI interface {
SendStream SendStream
handleStopSendingFrame(*wire.StopSendingFrame) handleStopSendingFrame(*wire.StopSendingFrame)
hasData() bool
popStreamFrame(maxBytes protocol.ByteCount) (*wire.StreamFrame, bool) popStreamFrame(maxBytes protocol.ByteCount) (*wire.StreamFrame, bool)
closeForShutdown(error) closeForShutdown(error)
handleMaxStreamDataFrame(*wire.MaxStreamDataFrame) handleMaxStreamDataFrame(*wire.MaxStreamDataFrame)
...@@ -40,8 +41,9 @@ type sendStream struct { ...@@ -40,8 +41,9 @@ type sendStream struct {
finSent bool // set when a STREAM_FRAME with FIN bit has b finSent bool // set when a STREAM_FRAME with FIN bit has b
dataForWriting []byte dataForWriting []byte
writeChan chan struct{}
writeDeadline time.Time writeChan chan struct{}
deadline time.Time
flowController flowcontrol.StreamFlowController flowController flowcontrol.StreamFlowController
...@@ -85,49 +87,60 @@ func (s *sendStream) Write(p []byte) (int, error) { ...@@ -85,49 +87,60 @@ func (s *sendStream) Write(p []byte) (int, error) {
if s.closeForShutdownErr != nil { if s.closeForShutdownErr != nil {
return 0, s.closeForShutdownErr return 0, s.closeForShutdownErr
} }
if !s.writeDeadline.IsZero() && !time.Now().Before(s.writeDeadline) { if !s.deadline.IsZero() && !time.Now().Before(s.deadline) {
return 0, errDeadline return 0, errDeadline
} }
if len(p) == 0 { if len(p) == 0 {
return 0, nil return 0, nil
} }
s.dataForWriting = make([]byte, len(p)) s.dataForWriting = p
copy(s.dataForWriting, p)
s.sender.onHasStreamData(s.streamID)
var bytesWritten int var (
var err error deadlineTimer *utils.Timer
bytesWritten int
notifiedSender bool
)
for { for {
bytesWritten = len(p) - len(s.dataForWriting) bytesWritten = len(p) - len(s.dataForWriting)
deadline := s.writeDeadline deadline := s.deadline
if !deadline.IsZero() && !time.Now().Before(deadline) { if !deadline.IsZero() {
s.dataForWriting = nil if !time.Now().Before(deadline) {
err = errDeadline s.dataForWriting = nil
break return bytesWritten, errDeadline
}
if deadlineTimer == nil {
deadlineTimer = utils.NewTimer()
}
deadlineTimer.Reset(deadline)
} }
if s.dataForWriting == nil || s.canceledWrite || s.closedForShutdown { if s.dataForWriting == nil || s.canceledWrite || s.closedForShutdown {
break break
} }
s.mutex.Unlock() s.mutex.Unlock()
if !notifiedSender {
s.sender.onHasStreamData(s.streamID) // must be called without holding the mutex
notifiedSender = true
}
if deadline.IsZero() { if deadline.IsZero() {
<-s.writeChan <-s.writeChan
} else { } else {
select { select {
case <-s.writeChan: case <-s.writeChan:
case <-time.After(time.Until(deadline)): case <-deadlineTimer.Chan():
deadlineTimer.SetRead()
} }
} }
s.mutex.Lock() s.mutex.Lock()
} }
if s.closeForShutdownErr != nil { if s.closeForShutdownErr != nil {
err = s.closeForShutdownErr return bytesWritten, s.closeForShutdownErr
} else if s.cancelWriteErr != nil { } else if s.cancelWriteErr != nil {
err = s.cancelWriteErr return bytesWritten, s.cancelWriteErr
} }
return bytesWritten, err return bytesWritten, nil
} }
// popStreamFrame returns the next STREAM frame that is supposed to be sent on this stream // popStreamFrame returns the next STREAM frame that is supposed to be sent on this stream
...@@ -181,12 +194,18 @@ func (s *sendStream) popStreamFrameImpl(maxBytes protocol.ByteCount) (bool /* co ...@@ -181,12 +194,18 @@ func (s *sendStream) popStreamFrameImpl(maxBytes protocol.ByteCount) (bool /* co
return frame.FinBit, frame, s.dataForWriting != nil return frame.FinBit, frame, s.dataForWriting != nil
} }
func (s *sendStream) hasData() bool {
s.mutex.Lock()
hasData := len(s.dataForWriting) > 0
s.mutex.Unlock()
return hasData
}
func (s *sendStream) getDataForWriting(maxBytes protocol.ByteCount) ([]byte, bool /* should send FIN */) { func (s *sendStream) getDataForWriting(maxBytes protocol.ByteCount) ([]byte, bool /* should send FIN */) {
if s.dataForWriting == nil { if s.dataForWriting == nil {
return nil, s.finishedWriting && !s.finSent return nil, s.finishedWriting && !s.finSent
} }
// TODO(#657): Flow control for the crypto stream
if s.streamID != s.version.CryptoStreamID() { if s.streamID != s.version.CryptoStreamID() {
maxBytes = utils.MinByteCount(maxBytes, s.flowController.SendWindowSize()) maxBytes = utils.MinByteCount(maxBytes, s.flowController.SendWindowSize())
} }
...@@ -196,10 +215,12 @@ func (s *sendStream) getDataForWriting(maxBytes protocol.ByteCount) ([]byte, boo ...@@ -196,10 +215,12 @@ func (s *sendStream) getDataForWriting(maxBytes protocol.ByteCount) ([]byte, boo
var ret []byte var ret []byte
if protocol.ByteCount(len(s.dataForWriting)) > maxBytes { if protocol.ByteCount(len(s.dataForWriting)) > maxBytes {
ret = s.dataForWriting[:maxBytes] ret = make([]byte, int(maxBytes))
copy(ret, s.dataForWriting[:maxBytes])
s.dataForWriting = s.dataForWriting[maxBytes:] s.dataForWriting = s.dataForWriting[maxBytes:]
} else { } else {
ret = s.dataForWriting ret = make([]byte, len(s.dataForWriting))
copy(ret, s.dataForWriting)
s.dataForWriting = nil s.dataForWriting = nil
s.signalWrite() s.signalWrite()
} }
...@@ -210,13 +231,14 @@ func (s *sendStream) getDataForWriting(maxBytes protocol.ByteCount) ([]byte, boo ...@@ -210,13 +231,14 @@ func (s *sendStream) getDataForWriting(maxBytes protocol.ByteCount) ([]byte, boo
func (s *sendStream) Close() error { func (s *sendStream) Close() error {
s.mutex.Lock() s.mutex.Lock()
defer s.mutex.Unlock()
if s.canceledWrite { if s.canceledWrite {
s.mutex.Unlock()
return fmt.Errorf("Close called for canceled stream %d", s.streamID) return fmt.Errorf("Close called for canceled stream %d", s.streamID)
} }
s.finishedWriting = true s.finishedWriting = true
s.sender.onHasStreamData(s.streamID) // need to send the FIN s.mutex.Unlock()
s.sender.onHasStreamData(s.streamID) // need to send the FIN, must be called without holding the mutex
s.ctxCancel() s.ctxCancel()
return nil return nil
} }
...@@ -227,7 +249,7 @@ func (s *sendStream) CancelWrite(errorCode protocol.ApplicationErrorCode) error ...@@ -227,7 +249,7 @@ func (s *sendStream) CancelWrite(errorCode protocol.ApplicationErrorCode) error
s.mutex.Unlock() s.mutex.Unlock()
if completed { if completed {
s.sender.onStreamCompleted(s.streamID) s.sender.onStreamCompleted(s.streamID) // must be called without holding the mutex
} }
return err return err
} }
...@@ -260,12 +282,13 @@ func (s *sendStream) handleStopSendingFrame(frame *wire.StopSendingFrame) { ...@@ -260,12 +282,13 @@ func (s *sendStream) handleStopSendingFrame(frame *wire.StopSendingFrame) {
} }
func (s *sendStream) handleMaxStreamDataFrame(frame *wire.MaxStreamDataFrame) { func (s *sendStream) handleMaxStreamDataFrame(frame *wire.MaxStreamDataFrame) {
s.flowController.UpdateSendWindow(frame.ByteOffset)
s.mutex.Lock() s.mutex.Lock()
if s.dataForWriting != nil { hasStreamData := s.dataForWriting != nil
s.mutex.Unlock()
s.flowController.UpdateSendWindow(frame.ByteOffset)
if hasStreamData {
s.sender.onHasStreamData(s.streamID) s.sender.onHasStreamData(s.streamID)
} }
s.mutex.Unlock()
} }
// must be called after locking the mutex // must be called after locking the mutex
...@@ -291,12 +314,9 @@ func (s *sendStream) Context() context.Context { ...@@ -291,12 +314,9 @@ func (s *sendStream) Context() context.Context {
func (s *sendStream) SetWriteDeadline(t time.Time) error { func (s *sendStream) SetWriteDeadline(t time.Time) error {
s.mutex.Lock() s.mutex.Lock()
oldDeadline := s.writeDeadline s.deadline = t
s.writeDeadline = t
s.mutex.Unlock() s.mutex.Unlock()
if t.Before(oldDeadline) { s.signalWrite()
s.signalWrite()
}
return nil return nil
} }
......
...@@ -123,6 +123,9 @@ func Listen(conn net.PacketConn, tlsConf *tls.Config, config *Config) (Listener, ...@@ -123,6 +123,9 @@ func Listen(conn net.PacketConn, tlsConf *tls.Config, config *Config) (Listener,
} }
func listen(conn net.PacketConn, tlsConf *tls.Config, config *Config) (*server, error) { func listen(conn net.PacketConn, tlsConf *tls.Config, config *Config) (*server, error) {
if tlsConf == nil || (len(tlsConf.Certificates) == 0 && tlsConf.GetCertificate == nil) {
return nil, errors.New("quic: neither Certificates nor GetCertificate set in tls.Config")
}
certChain := crypto.NewCertChain(tlsConf) certChain := crypto.NewCertChain(tlsConf)
kex, err := crypto.NewCurve25519KEX() kex, err := crypto.NewCurve25519KEX()
if err != nil { if err != nil {
......
...@@ -118,9 +118,6 @@ func (s *serverTLS) handleInitialImpl(p *receivedPacket) (quicSession, protocol. ...@@ -118,9 +118,6 @@ func (s *serverTLS) handleInitialImpl(p *receivedPacket) (quicSession, protocol.
mconf := s.mintConf.Clone() mconf := s.mintConf.Clone()
mconf.ExtensionHandler = extHandler mconf.ExtensionHandler = extHandler
// A server is allowed to perform multiple Retries.
// It doesn't make much sense, but it's something that our API allows.
// In that case it must use a source connection ID of at least 8 bytes.
connID, err := protocol.GenerateConnectionID(s.config.ConnectionIDLength) connID, err := protocol.GenerateConnectionID(s.config.ConnectionIDLength)
if err != nil { if err != nil {
return nil, nil, err return nil, nil, err
...@@ -152,7 +149,7 @@ func (s *serverTLS) sendRetry(remoteAddr net.Addr, hdr *wire.Header) error { ...@@ -152,7 +149,7 @@ func (s *serverTLS) sendRetry(remoteAddr net.Addr, hdr *wire.Header) error {
if err != nil { if err != nil {
return err return err
} }
connID, err := protocol.GenerateConnectionIDForInitial() connID, err := protocol.GenerateConnectionID(s.config.ConnectionIDLength)
if err != nil { if err != nil {
return err return err
} }
......
...@@ -52,6 +52,7 @@ type streamI interface { ...@@ -52,6 +52,7 @@ type streamI interface {
handleRstStreamFrame(*wire.RstStreamFrame) error handleRstStreamFrame(*wire.RstStreamFrame) error
getWindowUpdate() protocol.ByteCount getWindowUpdate() protocol.ByteCount
// for sending // for sending
hasData() bool
handleStopSendingFrame(*wire.StopSendingFrame) handleStopSendingFrame(*wire.StopSendingFrame)
popStreamFrame(maxBytes protocol.ByteCount) (*wire.StreamFrame, bool) popStreamFrame(maxBytes protocol.ByteCount) (*wire.StreamFrame, bool)
handleMaxStreamDataFrame(*wire.MaxStreamDataFrame) handleMaxStreamDataFrame(*wire.MaxStreamDataFrame)
......
...@@ -14,7 +14,6 @@ type windowUpdateQueue struct { ...@@ -14,7 +14,6 @@ type windowUpdateQueue struct {
queue map[protocol.StreamID]bool // used as a set queue map[protocol.StreamID]bool // used as a set
queuedConn bool // connection-level window update queuedConn bool // connection-level window update
cryptoStream cryptoStream
streamGetter streamGetter streamGetter streamGetter
connFlowController flowcontrol.ConnectionFlowController connFlowController flowcontrol.ConnectionFlowController
callback func(wire.Frame) callback func(wire.Frame)
...@@ -22,14 +21,12 @@ type windowUpdateQueue struct { ...@@ -22,14 +21,12 @@ type windowUpdateQueue struct {
func newWindowUpdateQueue( func newWindowUpdateQueue(
streamGetter streamGetter, streamGetter streamGetter,
cryptoStream cryptoStream,
connFC flowcontrol.ConnectionFlowController, connFC flowcontrol.ConnectionFlowController,
cb func(wire.Frame), cb func(wire.Frame),
) *windowUpdateQueue { ) *windowUpdateQueue {
return &windowUpdateQueue{ return &windowUpdateQueue{
queue: make(map[protocol.StreamID]bool), queue: make(map[protocol.StreamID]bool),
streamGetter: streamGetter, streamGetter: streamGetter,
cryptoStream: cryptoStream,
connFlowController: connFC, connFlowController: connFC,
callback: cb, callback: cb,
} }
...@@ -55,17 +52,12 @@ func (q *windowUpdateQueue) QueueAll() { ...@@ -55,17 +52,12 @@ func (q *windowUpdateQueue) QueueAll() {
q.queuedConn = false q.queuedConn = false
} }
// queue all stream-level window updates // queue all stream-level window updates
var offset protocol.ByteCount
for id := range q.queue { for id := range q.queue {
if id == q.cryptoStream.StreamID() { str, err := q.streamGetter.GetOrOpenReceiveStream(id)
offset = q.cryptoStream.getWindowUpdate() if err != nil || str == nil { // the stream can be nil if it was completed before dequeing the window update
} else { continue
str, err := q.streamGetter.GetOrOpenReceiveStream(id)
if err != nil || str == nil { // the stream can be nil if it was completed before dequeing the window update
continue
}
offset = str.getWindowUpdate()
} }
offset := str.getWindowUpdate()
if offset == 0 { // can happen if we received a final offset, right after queueing the window update if offset == 0 { // can happen if we received a final offset, right after queueing the window update
continue continue
} }
......
...@@ -130,7 +130,7 @@ ...@@ -130,7 +130,7 @@
"importpath": "github.com/lucas-clemente/quic-go", "importpath": "github.com/lucas-clemente/quic-go",
"repository": "https://github.com/lucas-clemente/quic-go", "repository": "https://github.com/lucas-clemente/quic-go",
"vcs": "git", "vcs": "git",
"revision": "71635f6961ad00ca5c088be625624e4a2cd1c066", "revision": "f90751eabaa39364e3861ee5a8b179f140847d7e",
"branch": "HEAD", "branch": "HEAD",
"notests": true "notests": true
}, },
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment