lib/connections: Add QUIC protocol support (fixes #5377) (#5737)

lib/connections/config.go

@@ -1,12 +0,0 @@
-// Copyright (C) 2017 The Syncthing Authors.
-//
-// This Source Code Form is subject to the terms of the Mozilla Public
-// License, v. 2.0. If a copy of the MPL was not distributed with this file,
-// You can obtain one at http://mozilla.org/MPL/2.0/.
-
-package connections
-
-const (
-	tcpPriority   = 10
-	relayPriority = 200
-)

lib/connections/quic_dial.go

@@ -0,0 +1,128 @@
+// Copyright (C) 2019 The Syncthing Authors.
+//
+// This Source Code Form is subject to the terms of the Mozilla Public
+// License, v. 2.0. If a copy of the MPL was not distributed with this file,
+// You can obtain one at https://mozilla.org/MPL/2.0/.
+
+// +build go1.12
+
+package connections
+
+import (
+	"context"
+	"crypto/tls"
+	"net"
+	"net/url"
+	"time"
+
+	"github.com/lucas-clemente/quic-go"
+
+	"github.com/syncthing/syncthing/lib/config"
+	"github.com/syncthing/syncthing/lib/connections/registry"
+	"github.com/syncthing/syncthing/lib/protocol"
+)
+
+const quicPriority = 100
+
+func init() {
+	factory := &quicDialerFactory{}
+	for _, scheme := range []string{"quic", "quic4", "quic6"} {
+		dialers[scheme] = factory
+	}
+}
+
+type quicDialer struct {
+	cfg    config.Wrapper
+	tlsCfg *tls.Config
+}
+
+func (d *quicDialer) Dial(id protocol.DeviceID, uri *url.URL) (internalConn, error) {
+	uri = fixupPort(uri, config.DefaultQUICPort)
+
+	addr, err := net.ResolveUDPAddr("udp", uri.Host)
+	if err != nil {
+		return internalConn{}, err
+	}
+
+	var conn net.PacketConn
+	closeConn := false
+	if listenConn := registry.Get(uri.Scheme, packetConnLess); listenConn != nil {
+		conn = listenConn.(net.PacketConn)
+	} else {
+		if packetConn, err := net.ListenPacket("udp", ":0"); err != nil {
+			return internalConn{}, err
+		} else {
+			closeConn = true
+			conn = packetConn
+		}
+	}
+
+	ctx, _ := context.WithTimeout(context.Background(), 10*time.Second)
+	session, err := quic.DialContext(ctx, conn, addr, uri.Host, d.tlsCfg, quicConfig)
+	if err != nil {
+		if closeConn {
+			_ = conn.Close()
+		}
+		return internalConn{}, err
+	}
+
+	// OpenStreamSync is blocks, but we want to make sure the connection is usable
+	// before we start killing off other connections, so do the dance.
+	ok := make(chan struct{})
+	go func() {
+		select {
+		case <-ok:
+			return
+		case <-time.After(10 * time.Second):
+			l.Debugln("timed out waiting for OpenStream on", session.RemoteAddr())
+			// This will unblock OpenStreamSync
+			_ = session.Close()
+		}
+	}()
+
+	stream, err := session.OpenStreamSync()
+	close(ok)
+	if err != nil {
+		// It's ok to close these, this does not close the underlying packetConn.
+		_ = session.Close()
+		if closeConn {
+			_ = conn.Close()
+		}
+		return internalConn{}, err
+	}
+
+	return internalConn{&quicTlsConn{session, stream}, connTypeQUICClient, quicPriority}, nil
+}
+
+func (d *quicDialer) RedialFrequency() time.Duration {
+	return time.Duration(d.cfg.Options().ReconnectIntervalS) * time.Second
+}
+
+type quicDialerFactory struct {
+	cfg    config.Wrapper
+	tlsCfg *tls.Config
+}
+
+func (quicDialerFactory) New(cfg config.Wrapper, tlsCfg *tls.Config) genericDialer {
+	return &quicDialer{
+		cfg:    cfg,
+		tlsCfg: tlsCfg,
+	}
+}
+
+func (quicDialerFactory) Priority() int {
+	return quicPriority
+}
+
+func (quicDialerFactory) AlwaysWAN() bool {
+	return false
+}
+
+func (quicDialerFactory) Valid(_ config.Configuration) error {
+	// Always valid
+	return nil
+}
+
+func (quicDialerFactory) String() string {
+	return "QUIC Dialer"
+}

lib/connections/quic_listen.go

@@ -0,0 +1,238 @@
+// Copyright (C) 2019 The Syncthing Authors.
+//
+// This Source Code Form is subject to the terms of the Mozilla Public
+// License, v. 2.0. If a copy of the MPL was not distributed with this file,
+// You can obtain one at http://mozilla.org/MPL/2.0/.
+
+// +build go1.12
+
+package connections
+
+import (
+	"crypto/tls"
+	"net"
+	"net/url"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"github.com/lucas-clemente/quic-go"
+
+	"github.com/syncthing/syncthing/lib/config"
+	"github.com/syncthing/syncthing/lib/connections/registry"
+	"github.com/syncthing/syncthing/lib/nat"
+	"github.com/syncthing/syncthing/lib/stun"
+)
+
+func init() {
+	factory := &quicListenerFactory{}
+	for _, scheme := range []string{"quic", "quic4", "quic6"} {
+		listeners[scheme] = factory
+	}
+}
+
+type quicListener struct {
+	nat atomic.Value
+
+	onAddressesChangedNotifier
+
+	uri     *url.URL
+	cfg     config.Wrapper
+	tlsCfg  *tls.Config
+	stop    chan struct{}
+	conns   chan internalConn
+	factory listenerFactory
+
+	address *url.URL
+	err     error
+	mut     sync.Mutex
+}
+
+func (t *quicListener) OnNATTypeChanged(natType stun.NATType) {
+	if natType != stun.NATUnknown {
+		l.Infof("%s detected NAT type: %s", t.uri, natType)
+	}
+	t.nat.Store(natType)
+}
+
+func (t *quicListener) OnExternalAddressChanged(address *stun.Host, via string) {
+	var uri *url.URL
+	if address != nil {
+		uri = &(*t.uri)
+		uri.Host = address.TransportAddr()
+	}
+
+	t.mut.Lock()
+	existingAddress := t.address
+	t.address = uri
+	t.mut.Unlock()
+
+	if uri != nil && (existingAddress == nil || existingAddress.String() != uri.String()) {
+		l.Infof("%s resolved external address %s (via %s)", t.uri, uri.String(), via)
+		t.notifyAddressesChanged(t)
+	} else if uri == nil && existingAddress != nil {
+		t.notifyAddressesChanged(t)
+	}
+}
+
+func (t *quicListener) Serve() {
+	t.mut.Lock()
+	t.err = nil
+	t.mut.Unlock()
+
+	network := strings.Replace(t.uri.Scheme, "quic", "udp", -1)
+
+	packetConn, err := net.ListenPacket(network, t.uri.Host)
+	if err != nil {
+		t.mut.Lock()
+		t.err = err
+		t.mut.Unlock()
+		l.Infoln("Listen (BEP/quic):", err)
+		return
+	}
+	defer func() { _ = packetConn.Close() }()
+
+	svc, conn := stun.New(t.cfg, t, packetConn)
+	defer func() { _ = conn.Close() }()
+
+	go svc.Serve()
+	defer svc.Stop()
+
+	registry.Register(t.uri.Scheme, conn)
+	defer registry.Unregister(t.uri.Scheme, conn)
+
+	listener, err := quic.Listen(conn, t.tlsCfg, quicConfig)
+	if err != nil {
+		t.mut.Lock()
+		t.err = err
+		t.mut.Unlock()
+		l.Infoln("Listen (BEP/quic):", err)
+		return
+	}
+
+	l.Infof("QUIC listener (%v) starting", packetConn.LocalAddr())
+	defer l.Infof("QUIC listener (%v) shutting down", packetConn.LocalAddr())
+
+	// Accept is forever, so handle stops externally.
+	go func() {
+		select {
+		case <-t.stop:
+			_ = listener.Close()
+		}
+	}()
+
+	for {
+		// Blocks forever, see https://github.com/lucas-clemente/quic-go/issues/1915
+		session, err := listener.Accept()
+
+		select {
+		case <-t.stop:
+			if err == nil {
+				_ = session.Close()
+			}
+			return
+		default:
+		}
+		if err != nil {
+			if err, ok := err.(net.Error); !ok || !err.Timeout() {
+				l.Warnln("Listen (BEP/quic): Accepting connection:", err)
+			}
+			continue
+		}
+
+		l.Debugln("connect from", session.RemoteAddr())
+
+		// Accept blocks forever, give it 10s to do it's thing.
+		ok := make(chan struct{})
+		go func() {
+			select {
+			case <-ok:
+				return
+			case <-t.stop:
+				_ = session.Close()
+			case <-time.After(10 * time.Second):
+				l.Debugln("timed out waiting for AcceptStream on", session.RemoteAddr())
+				_ = session.Close()
+			}
+		}()
+
+		stream, err := session.AcceptStream()
+		close(ok)
+		if err != nil {
+			l.Debugln("failed to accept stream from", session.RemoteAddr(), err.Error())
+			_ = session.Close()
+			continue
+		}
+
+		t.conns <- internalConn{&quicTlsConn{session, stream}, connTypeQUICServer, quicPriority}
+	}
+}
+
+func (t *quicListener) Stop() {
+	close(t.stop)
+}
+
+func (t *quicListener) URI() *url.URL {
+	return t.uri
+}
+
+func (t *quicListener) WANAddresses() []*url.URL {
+	uris := t.LANAddresses()
+	t.mut.Lock()
+	if t.address != nil {
+		uris = append(uris, t.address)
+	}
+	t.mut.Unlock()
+	return uris
+}
+
+func (t *quicListener) LANAddresses() []*url.URL {
+	return []*url.URL{t.uri}
+}
+
+func (t *quicListener) Error() error {
+	t.mut.Lock()
+	err := t.err
+	t.mut.Unlock()
+	return err
+}
+
+func (t *quicListener) String() string {
+	return t.uri.String()
+}
+
+func (t *quicListener) Factory() listenerFactory {
+	return t.factory
+}
+
+func (t *quicListener) NATType() string {
+	v := t.nat.Load().(stun.NATType)
+	if v == stun.NATUnknown || v == stun.NATError {
+		return "unknown"
+	}
+	return v.String()
+}
+
+type quicListenerFactory struct{}
+
+func (f *quicListenerFactory) Valid(config.Configuration) error {
+	return nil
+}
+
+func (f *quicListenerFactory) New(uri *url.URL, cfg config.Wrapper, tlsCfg *tls.Config, conns chan internalConn, natService *nat.Service) genericListener {
+	l := &quicListener{
+		uri:     fixupPort(uri, config.DefaultQUICPort),
+		cfg:     cfg,
+		tlsCfg:  tlsCfg,
+		conns:   conns,
+		stop:    make(chan struct{}),
+		factory: f,
+	}
+	l.nat.Store(stun.NATUnknown)
+	return l
+}
+
+func (quicListenerFactory) Enabled(cfg config.Configuration) bool {
+	return true
+}

lib/connections/quic_misc.go

@@ -0,0 +1,57 @@
+// Copyright (C) 2019 The Syncthing Authors.
+//
+// This Source Code Form is subject to the terms of the Mozilla Public
+// License, v. 2.0. If a copy of the MPL was not distributed with this file,
+// You can obtain one at http://mozilla.org/MPL/2.0/.
+
+// +build go1.12
+
+package connections
+
+import (
+	"net"
+
+	"github.com/lucas-clemente/quic-go"
+)
+
+var (
+	quicConfig = &quic.Config{
+		ConnectionIDLength: 4,
+		KeepAlive:          true,
+	}
+)
+
+type quicTlsConn struct {
+	quic.Session
+	quic.Stream
+}
+
+func (q *quicTlsConn) Close() error {
+	sterr := q.Stream.Close()
+	seerr := q.Session.Close()
+	if sterr != nil {
+		return sterr
+	}
+	return seerr
+}
+
+// Sort available packet connections by ip address, preferring unspecified local address.
+func packetConnLess(i interface{}, j interface{}) bool {
+	iIsUnspecified := false
+	jIsUnspecified := false
+	iLocalAddr := i.(net.PacketConn).LocalAddr()
+	jLocalAddr := j.(net.PacketConn).LocalAddr()
+
+	if host, _, err := net.SplitHostPort(iLocalAddr.String()); err == nil {
+		iIsUnspecified = host == "" || net.ParseIP(host).IsUnspecified()
+	}
+	if host, _, err := net.SplitHostPort(jLocalAddr.String()); err == nil {
+		jIsUnspecified = host == "" || net.ParseIP(host).IsUnspecified()
+	}
+
+	if jIsUnspecified == iIsUnspecified {
+		return len(iLocalAddr.Network()) < len(jLocalAddr.Network())
+	}
+
+	return iIsUnspecified
+}

lib/connections/quic_misc_test.go

@@ -0,0 +1,92 @@
+// Copyright (C) 2019 The Syncthing Authors.
+//
+// This Source Code Form is subject to the terms of the Mozilla Public
+// License, v. 2.0. If a copy of the MPL was not distributed with this file,
+// You can obtain one at http://mozilla.org/MPL/2.0/.
+
+// +build go1.12
+
+package connections
+
+import (
+	"net"
+	"testing"
+	"time"
+)
+
+type mockPacketConn struct {
+	addr mockedAddr
+}
+
+func (mockPacketConn) ReadFrom(p []byte) (n int, addr net.Addr, err error) {
+	panic("implement me")
+}
+
+func (mockPacketConn) WriteTo(p []byte, addr net.Addr) (n int, err error) {
+	panic("implement me")
+}
+
+func (mockPacketConn) Close() error {
+	panic("implement me")
+}
+
+func (c *mockPacketConn) LocalAddr() net.Addr {
+	return c.addr
+}
+
+func (mockPacketConn) SetDeadline(t time.Time) error {
+	panic("implement me")
+}
+
+func (mockPacketConn) SetReadDeadline(t time.Time) error {
+	panic("implement me")
+}
+
+func (mockPacketConn) SetWriteDeadline(t time.Time) error {
+	panic("implement me")
+}
+
+type mockedAddr struct {
+	network string
+	addr    string
+}
+
+func (a mockedAddr) Network() string {
+	return a.network
+}
+
+func (a mockedAddr) String() string {
+	return a.addr
+}
+
+func TestPacketConnLess(t *testing.T) {
+	cases := []struct {
+		netA  string
+		addrA string
+		netB  string
+		addrB string
+	}{
+		// B is assumed the winner.
+		{"tcp", "127.0.0.1:1234", "tcp", ":1235"},
+		{"tcp", "127.0.0.1:1234", "tcp", "0.0.0.0:1235"},
+		{"tcp4", "0.0.0.0:1234", "tcp", "0.0.0.0:1235"}, // tcp4 on the first one
+	}
+
+	for i, testCase := range cases {
+
+		conns := []*mockPacketConn{
+			{mockedAddr{testCase.netA, testCase.addrA}},
+			{mockedAddr{testCase.netB, testCase.addrB}},
+		}
+
+		if packetConnLess(conns[0], conns[1]) {
+			t.Error(i, "unexpected")
+		}
+		if !packetConnLess(conns[1], conns[0]) {
+			t.Error(i, "unexpected")
+		}
+		if packetConnLess(conns[0], conns[0]) || packetConnLess(conns[1], conns[1]) {
+			t.Error(i, "unexpected")
+		}
+	}
+}

lib/connections/registry/registry.go

@@ -0,0 +1,89 @@
+// Copyright (C) 2019 The Syncthing Authors.
+//
+// This Source Code Form is subject to the terms of the Mozilla Public
+// License, v. 2.0. If a copy of the MPL was not distributed with this file,
+// You can obtain one at https://mozilla.org/MPL/2.0/.
+
+// Registry tracks connections/addresses on which we are listening on, to allow us to pick a connection/address that
+// has a NAT port mapping. This also makes our outgoing port stable and same as incoming port which should allow
+// better probability of punching through.
+package registry
+
+import (
+	"sort"
+	"strings"
+
+	"github.com/syncthing/syncthing/lib/sync"
+)
+
+var (
+	Default = New()
+)
+
+type Registry struct {
+	mut       sync.Mutex
+	available map[string][]interface{}
+}
+
+func New() *Registry {
+	return &Registry{
+		mut:       sync.NewMutex(),
+		available: make(map[string][]interface{}),
+	}
+}
+
+func (r *Registry) Register(scheme string, item interface{}) {
+	r.mut.Lock()
+	defer r.mut.Unlock()
+
+	r.available[scheme] = append(r.available[scheme], item)
+}
+
+func (r *Registry) Unregister(scheme string, item interface{}) {
+	r.mut.Lock()
+	defer r.mut.Unlock()
+
+	candidates := r.available[scheme]
+	for i, existingItem := range candidates {
+		if existingItem == item {
+			copy(candidates[i:], candidates[i+1:])
+			candidates[len(candidates)-1] = nil
+			r.available[scheme] = candidates[:len(candidates)-1]
+			break
+		}
+	}
+}
+
+func (r *Registry) Get(scheme string, less func(i, j interface{}) bool) interface{} {
+	r.mut.Lock()
+	defer r.mut.Unlock()
+
+	candidates := make([]interface{}, 0)
+	for availableScheme, items := range r.available {
+		// quic:// should be considered ok for both quic4:// and quic6://
+		if strings.HasPrefix(scheme, availableScheme) {
+			candidates = append(candidates, items...)
+		}
+	}
+
+	if len(candidates) == 0 {
+		return nil
+	}
+
+	sort.Slice(candidates, func(i, j int) bool {
+		return less(candidates[i], candidates[j])
+	})
+	return candidates[0]
+}
+
+func Register(scheme string, item interface{}) {
+	Default.Register(scheme, item)
+}
+
+func Unregister(scheme string, item interface{}) {
+	Default.Unregister(scheme, item)
+}
+
+func Get(scheme string, less func(i, j interface{}) bool) interface{} {
+	return Default.Get(scheme, less)
+}

lib/connections/registry/registry_test.go

@@ -0,0 +1,71 @@
+// Copyright (C) 2019 The Syncthing Authors.
+//
+// This Source Code Form is subject to the terms of the Mozilla Public
+// License, v. 2.0. If a copy of the MPL was not distributed with this file,
+// You can obtain one at https://mozilla.org/MPL/2.0/.
+
+package registry
+
+import (
+	"testing"
+)
+
+func TestRegistry(t *testing.T) {
+	r := New()
+
+	if res := r.Get("int", intLess); res != nil {
+		t.Error("unexpected")
+	}
+
+	r.Register("int", 1)
+	r.Register("int", 11)
+	r.Register("int4", 4)
+	r.Register("int4", 44)
+	r.Register("int6", 6)
+	r.Register("int6", 66)
+
+	if res := r.Get("int", intLess).(int); res != 1 {
+		t.Error("unexpected", res)
+	}
+
+	// int is prefix of int4, so returns 1
+	if res := r.Get("int4", intLess).(int); res != 1 {
+		t.Error("unexpected", res)
+	}
+
+	r.Unregister("int", 1)
+
+	// Check that falls through to 11
+	if res := r.Get("int", intLess).(int); res != 11 {
+		t.Error("unexpected", res)
+	}
+
+	// 6 is smaller than 11 available in int.
+	if res := r.Get("int6", intLess).(int); res != 6 {
+		t.Error("unexpected", res)
+	}
+
+	// Unregister 11, int should be impossible to find
+	r.Unregister("int", 11)
+	if res := r.Get("int", intLess); res != nil {
+		t.Error("unexpected")
+	}
+
+	// Unregister a second time does nothing.
+	r.Unregister("int", 1)
+
+	// Can have multiple of the same
+	r.Register("int", 1)
+	r.Register("int", 1)
+	r.Unregister("int", 1)
+
+	if res := r.Get("int4", intLess).(int); res != 1 {
+		t.Error("unexpected", res)
+	}
+}
+
+func intLess(i, j interface{}) bool {
+	iInt := i.(int)
+	jInt := j.(int)
+	return iInt < jInt
+}

lib/connections/relay_dial.go

@@ -17,6 +17,8 @@ import (
 	"github.com/syncthing/syncthing/lib/relay/client"
 )
 
+const relayPriority = 200
+
 func init() {
 	dialers["relay"] = relayDialerFactory{}
 }

lib/connections/service.go

@@ -375,7 +375,7 @@ func (s *service) connect() {
 				}
 			}
 
-			addrs = util.UniqueStrings(addrs)
+			addrs = util.UniqueTrimmedStrings(addrs)
 
 			l.Debugln("Reconnect loop for", deviceID, addrs)
 
@@ -642,7 +642,7 @@ func (s *service) AllAddresses() []string {
 		}
 	}
 	s.listenersMut.RUnlock()
-	return util.UniqueStrings(addrs)
+	return util.UniqueTrimmedStrings(addrs)
 }
 
 func (s *service) ExternalAddresses() []string {
@@ -654,7 +654,7 @@ func (s *service) ExternalAddresses() []string {
 		}
 	}
 	s.listenersMut.RUnlock()
-	return util.UniqueStrings(addrs)
+	return util.UniqueTrimmedStrings(addrs)
 }
 
 func (s *service) ListenerStatus() map[string]ListenerStatusEntry {

lib/connections/structs.go

@@ -9,6 +9,7 @@ package connections
 import (
 	"crypto/tls"
 	"fmt"
+	"io"
 	"net"
 	"net/url"
 	"time"
@@ -43,10 +44,19 @@ func (c completeConn) Close(err error) {
 	c.internalConn.Close()
 }
 
+type tlsConn interface {
+	io.ReadWriteCloser
+	ConnectionState() tls.ConnectionState
+	RemoteAddr() net.Addr
+	SetDeadline(time.Time) error
+	SetWriteDeadline(time.Time) error
+	LocalAddr() net.Addr
+}
+
 // internalConn is the raw TLS connection plus some metadata on where it
 // came from (type, priority).
 type internalConn struct {
-	*tls.Conn
+	tlsConn
 	connType connType
 	priority int
 }
@@ -58,6 +68,8 @@ const (
 	connTypeRelayServer
 	connTypeTCPClient
 	connTypeTCPServer
+	connTypeQUICClient
+	connTypeQUICServer
 )
 
 func (t connType) String() string {
@@ -70,6 +82,10 @@ func (t connType) String() string {
 		return "tcp-client"
 	case connTypeTCPServer:
 		return "tcp-server"
+	case connTypeQUICClient:
+		return "quic-client"
+	case connTypeQUICServer:
+		return "quic-server"
 	default:
 		return "unknown-type"
 	}
@@ -81,6 +97,8 @@ func (t connType) Transport() string {
 		return "relay"
 	case connTypeTCPClient, connTypeTCPServer:
 		return "tcp"
+	case connTypeQUICClient, connTypeQUICServer:
+		return "quic"
 	default:
 		return "unknown"
 	}
@@ -90,8 +108,8 @@ func (c internalConn) Close() {
 	// *tls.Conn.Close() does more than it says on the tin. Specifically, it
 	// sends a TLS alert message, which might block forever if the
 	// connection is dead and we don't have a deadline set.
-	c.SetWriteDeadline(time.Now().Add(250 * time.Millisecond))
-	c.Conn.Close()
+	_ = c.SetWriteDeadline(time.Now().Add(250 * time.Millisecond))
+	_ = c.tlsConn.Close()
 }
 
 func (c internalConn) Type() string {

lib/connections/tcp_dial.go

@@ -16,6 +16,8 @@ import (
 	"github.com/syncthing/syncthing/lib/protocol"
 )
 
+const tcpPriority = 10
+
 func init() {
 	factory := &tcpDialerFactory{}
 	for _, scheme := range []string{"tcp", "tcp4", "tcp6"} {