lib/connections: Add KCP support (fixes #804)

GitHub-Pull-Request: https://github.com/syncthing/syncthing/pull/3489

lib/connections/config.go

@@ -0,0 +1,53 @@
+// 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
+
+import (
+	"io/ioutil"
+	"time"
+
+	"github.com/hashicorp/yamux"
+)
+
+const (
+	tcpPriority   = 10
+	kcpPriority   = 50
+	relayPriority = 200
+
+	// KCP filter priorities
+	kcpNoFilterPriority           = 100
+	kcpConversationFilterPriority = 20
+	kcpStunFilterPriority         = 10
+
+	// KCP SetNoDelay options
+
+	// 0 - disabled (default)
+	// 1 - enabled
+	kcpNoDelay        = 0
+	kcpUpdateInterval = 100 // ms (default)
+	// 0 - disable (default)
+	// 1 - enabled
+	kcpFastResend = 0
+	// 0 - enabled (default)
+	// 1 - disabled
+	kcpCongestionControl = 0
+
+	// KCP window sizes
+	kcpSendWindowSize    = 128
+	kcpReceiveWindowSize = 128
+)
+
+var (
+	yamuxConfig = &yamux.Config{
+		AcceptBacklog:          256,
+		EnableKeepAlive:        true,
+		KeepAliveInterval:      30 * time.Second,
+		ConnectionWriteTimeout: 10 * time.Second,
+		MaxStreamWindowSize:    256 * 1024,
+		LogOutput:              ioutil.Discard,
+	}
+)

lib/connections/connections_test.go

@@ -18,9 +18,9 @@ func TestFixupPort(t *testing.T) {
 
 	for _, tc := range cases {
 		u0, _ := url.Parse(tc[0])
-		u1 := fixupPort(u0).String()
+		u1 := fixupPort(u0, 22000).String()
 		if u1 != tc[1] {
-			t.Errorf("fixupPort(%q) => %q, expected %q", tc[0], u1, tc[1])
+			t.Errorf("fixupPort(%q, 22000) => %q, expected %q", tc[0], u1, tc[1])
 		}
 	}
 }

lib/connections/kcp_dial.go

@@ -0,0 +1,106 @@
+// Copyright (C) 2016 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
+
+import (
+	"crypto/tls"
+	"net/url"
+	"time"
+
+	"github.com/hashicorp/yamux"
+	"github.com/syncthing/syncthing/lib/config"
+	"github.com/syncthing/syncthing/lib/protocol"
+	"github.com/xtaci/kcp-go"
+)
+
+func init() {
+	factory := &kcpDialerFactory{}
+	for _, scheme := range []string{"kcp", "kcp4", "kcp6"} {
+		dialers[scheme] = factory
+	}
+}
+
+type kcpDialer struct {
+	cfg    *config.Wrapper
+	tlsCfg *tls.Config
+}
+
+func (d *kcpDialer) Dial(id protocol.DeviceID, uri *url.URL) (internalConn, error) {
+	uri = fixupPort(uri, config.DefaultKCPPort)
+
+	var conn *kcp.UDPSession
+	var err error
+
+	// Try to dial via an existing listening connection
+	// giving better changes punching through NAT.
+	if f := getDialingFilter(); f != nil {
+		conn, err = kcp.NewConn(uri.Host, nil, 0, 0, f.NewConn(kcpConversationFilterPriority, &kcpConversationFilter{}))
+		l.Debugf("dial %s using existing conn on %s", uri.String(), conn.LocalAddr())
+	} else {
+		conn, err = kcp.DialWithOptions(uri.Host, nil, 0, 0)
+	}
+	if err != nil {
+		l.Debugln(err)
+		conn.Close()
+		return internalConn{}, err
+	}
+
+	conn.SetKeepAlive(0) // yamux and stun service does keep-alives.
+	conn.SetStreamMode(true)
+	conn.SetACKNoDelay(false)
+	conn.SetWindowSize(kcpSendWindowSize, kcpReceiveWindowSize)
+	conn.SetNoDelay(kcpNoDelay, kcpUpdateInterval, kcpFastResend, kcpCongestionControl)
+
+	ses, err := yamux.Client(conn, yamuxConfig)
+	if err != nil {
+		conn.Close()
+		return internalConn{}, err
+	}
+	stream, err := ses.OpenStream()
+	if err != nil {
+		ses.Close()
+		return internalConn{}, err
+	}
+
+	tc := tls.Client(&sessionClosingStream{stream, ses}, d.tlsCfg)
+	tc.SetDeadline(time.Now().Add(time.Second * 10))
+	err = tc.Handshake()
+	if err != nil {
+		tc.Close()
+		return internalConn{}, err
+	}
+	tc.SetDeadline(time.Time{})
+
+	return internalConn{tc, connTypeKCPClient, kcpPriority}, nil
+}
+
+func (d *kcpDialer) RedialFrequency() time.Duration {
+	// For restricted NATs, the UDP mapping will potentially only be open for 20-30 seconds
+	// hence try dialing just as often.
+	return time.Duration(d.cfg.Options().StunKeepaliveS) * time.Second
+}
+
+type kcpDialerFactory struct{}
+
+func (kcpDialerFactory) New(cfg *config.Wrapper, tlsCfg *tls.Config) genericDialer {
+	return &kcpDialer{
+		cfg:    cfg,
+		tlsCfg: tlsCfg,
+	}
+}
+
+func (kcpDialerFactory) Priority() int {
+	return kcpPriority
+}
+
+func (kcpDialerFactory) Enabled(cfg config.Configuration) bool {
+	return true
+}
+
+func (kcpDialerFactory) String() string {
+	return "KCP Dialer"
+}

lib/connections/kcp_listen.go

@@ -0,0 +1,285 @@
+// Copyright (C) 2016 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
+
+import (
+	"crypto/tls"
+	"net"
+	"net/url"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/AudriusButkevicius/pfilter"
+	"github.com/ccding/go-stun/stun"
+	"github.com/hashicorp/yamux"
+	"github.com/syncthing/syncthing/lib/config"
+	"github.com/syncthing/syncthing/lib/nat"
+	"github.com/xtaci/kcp-go"
+)
+
+func init() {
+	factory := &kcpListenerFactory{}
+	for _, scheme := range []string{"kcp", "kcp4", "kcp6"} {
+		listeners[scheme] = factory
+	}
+}
+
+type kcpListener struct {
+	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.RWMutex
+}
+
+func (t *kcpListener) Serve() {
+	t.mut.Lock()
+	t.err = nil
+	t.mut.Unlock()
+
+	network := strings.Replace(t.uri.Scheme, "kcp", "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/kcp):", err)
+		return
+	}
+	filterConn := pfilter.NewPacketFilter(packetConn)
+	kcpConn := filterConn.NewConn(kcpNoFilterPriority, nil)
+	stunConn := filterConn.NewConn(kcpStunFilterPriority, &stunFilter{
+		ids: make(map[string]time.Time),
+	})
+
+	filterConn.Start()
+	registerFilter(filterConn)
+
+	listener, err := kcp.ServeConn(nil, 0, 0, kcpConn)
+	if err != nil {
+		t.mut.Lock()
+		t.err = err
+		t.mut.Unlock()
+		l.Infoln("listen (BEP/kcp):", err)
+		return
+	}
+
+	defer listener.Close()
+	defer stunConn.Close()
+	defer kcpConn.Close()
+	defer deregisterFilter(filterConn)
+	defer packetConn.Close()
+
+	l.Infof("KCP listener (%v) starting", kcpConn.LocalAddr())
+	defer l.Infof("KCP listener (%v) shutting down", kcpConn.LocalAddr())
+
+	go t.stunRenewal(stunConn)
+
+	for {
+		listener.SetDeadline(time.Now().Add(time.Second))
+		conn, err := listener.AcceptKCP()
+
+		select {
+		case <-t.stop:
+			if err == nil {
+				conn.Close()
+			}
+			return
+		default:
+		}
+		if err != nil {
+			if err, ok := err.(net.Error); !ok || !err.Timeout() {
+				l.Warnln("Accepting connection (BEP/kcp):", err)
+			}
+			continue
+		}
+
+		conn.SetStreamMode(true)
+		conn.SetACKNoDelay(false)
+		conn.SetWindowSize(kcpSendWindowSize, kcpReceiveWindowSize)
+		conn.SetNoDelay(kcpNoDelay, kcpUpdateInterval, kcpFastResend, kcpCongestionControl)
+
+		conn.SetKeepAlive(0) // yamux and stun service does keep-alives.
+
+		l.Debugln("connect from", conn.RemoteAddr())
+
+		ses, err := yamux.Server(conn, yamuxConfig)
+		if err != nil {
+			l.Debugln("yamux server:", err)
+			conn.Close()
+			continue
+		}
+
+		stream, err := ses.AcceptStream()
+		if err != nil {
+			l.Debugln("yamux accept:", err)
+			ses.Close()
+			continue
+		}
+
+		tc := tls.Server(&sessionClosingStream{stream, ses}, t.tlsCfg)
+		tc.SetDeadline(time.Now().Add(time.Second * 10))
+		err = tc.Handshake()
+		if err != nil {
+			l.Debugln("TLS handshake (BEP/kcp):", err)
+			tc.Close()
+			continue
+		}
+		tc.SetDeadline(time.Time{})
+
+		t.conns <- internalConn{tc, connTypeKCPServer, kcpPriority}
+	}
+}
+
+func (t *kcpListener) Stop() {
+	close(t.stop)
+}
+
+func (t *kcpListener) URI() *url.URL {
+	return t.uri
+}
+
+func (t *kcpListener) WANAddresses() []*url.URL {
+	uris := t.LANAddresses()
+	t.mut.RLock()
+	if t.address != nil {
+		uris = append(uris, t.address)
+	}
+	t.mut.RUnlock()
+	return uris
+}
+
+func (t *kcpListener) LANAddresses() []*url.URL {
+	return []*url.URL{t.uri}
+}
+
+func (t *kcpListener) Error() error {
+	t.mut.RLock()
+	err := t.err
+	t.mut.RUnlock()
+	return err
+}
+
+func (t *kcpListener) String() string {
+	return t.uri.String()
+}
+
+func (t *kcpListener) Factory() listenerFactory {
+	return t.factory
+}
+
+func (t *kcpListener) stunRenewal(listener net.PacketConn) {
+	client := stun.NewClientWithConnection(listener)
+	client.SetSoftwareName("syncthing")
+
+	var uri url.URL
+	var natType stun.NATType
+	var extAddr *stun.Host
+	var err error
+
+	oldType := stun.NATUnknown
+
+	for {
+
+	disabled:
+		if t.cfg.Options().StunKeepaliveS < 1 {
+			time.Sleep(time.Second)
+			oldType = stun.NATUnknown
+			t.mut.Lock()
+			t.address = nil
+			t.mut.Unlock()
+			continue
+		}
+
+		for _, addr := range t.cfg.StunServers() {
+			client.SetServerAddr(addr)
+
+			natType, extAddr, err = client.Discover()
+			if err != nil || extAddr == nil {
+				l.Debugf("%s stun discovery on %s: %s", t.uri, addr, err)
+				continue
+			}
+
+			// The stun server is most likely borked, try another one.
+			if natType == stun.NATError || natType == stun.NATUnknown || natType == stun.NATBlocked {
+				l.Debugf("%s stun discovery on %s resolved to %s", t.uri, addr, natType)
+				continue
+			}
+
+			if oldType != natType {
+				l.Infof("%s detected NAT type: %s", t.uri, natType)
+			}
+
+			for {
+				changed := false
+				uri = *t.uri
+				uri.Host = extAddr.TransportAddr()
+
+				t.mut.Lock()
+				if t.address == nil || t.address.String() != uri.String() {
+					l.Infof("%s resolved external address %s (via %s)", t.uri, uri.String(), addr)
+					t.address = &uri
+					changed = true
+				}
+				t.mut.Unlock()
+
+				// This will most likely result in a call to WANAddresses() which tries to
+				// get t.mut, so notify while unlocked.
+				if changed {
+					t.notifyAddressesChanged(t)
+				}
+
+				select {
+				case <-time.After(time.Duration(t.cfg.Options().StunKeepaliveS) * time.Second):
+				case <-t.stop:
+					return
+				}
+
+				if t.cfg.Options().StunKeepaliveS < 1 {
+					goto disabled
+				}
+
+				extAddr, err = client.Keepalive()
+				if err != nil {
+					l.Debugf("%s stun keepalive on %s: %s (%v)", t.uri, addr, err, extAddr)
+					break
+				}
+			}
+
+			oldType = natType
+		}
+
+		// We failed to contact all provided stun servers, chillout for a while.
+		time.Sleep(time.Minute)
+	}
+}
+
+type kcpListenerFactory struct{}
+
+func (f *kcpListenerFactory) New(uri *url.URL, cfg *config.Wrapper, tlsCfg *tls.Config, conns chan internalConn, natService *nat.Service) genericListener {
+	return &kcpListener{
+		uri:     fixupPort(uri, config.DefaultKCPPort),
+		cfg:     cfg,
+		tlsCfg:  tlsCfg,
+		conns:   conns,
+		stop:    make(chan struct{}),
+		factory: f,
+	}
+}
+
+func (kcpListenerFactory) Enabled(cfg config.Configuration) bool {
+	return true
+}

lib/connections/kcp_misc.go

@@ -0,0 +1,182 @@
+// Copyright (C) 2016 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
+
+import (
+	"bytes"
+	"encoding/binary"
+	"net"
+	"sort"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"github.com/AudriusButkevicius/pfilter"
+	"github.com/hashicorp/yamux"
+)
+
+var (
+	mut     sync.Mutex
+	filters filterList
+)
+
+type filterList []*pfilter.PacketFilter
+
+// Sort connections by wether the are unspecified or not, as connections
+// listenin on all addresses are more useful.
+func (f filterList) Len() int      { return len(f) }
+func (f filterList) Swap(i, j int) { f[i], f[j] = f[j], f[i] }
+func (f filterList) Less(i, j int) bool {
+	iIsUnspecified := false
+	jIsUnspecified := false
+	if host, _, err := net.SplitHostPort(f[i].LocalAddr().String()); err == nil {
+		iIsUnspecified = net.ParseIP(host).IsUnspecified()
+	}
+	if host, _, err := net.SplitHostPort(f[j].LocalAddr().String()); err == nil {
+		jIsUnspecified = net.ParseIP(host).IsUnspecified()
+	}
+	return (iIsUnspecified && !jIsUnspecified) || (iIsUnspecified && jIsUnspecified)
+}
+
+// As we open listen KCP connections, we register them here, so that Dial calls through
+// KCP could reuse them. This way we will hopefully work around restricted NATs by
+// dialing via the same connection we are listening on, creating a mapping on our NAT
+// to that IP, and hoping that the other end will try to dial our listen address and
+// using the mapping we've established when we dialed.
+func getDialingFilter() *pfilter.PacketFilter {
+	mut.Lock()
+	defer mut.Unlock()
+	if len(filters) == 0 {
+		return nil
+	}
+	return filters[0]
+}
+
+func registerFilter(filter *pfilter.PacketFilter) {
+	mut.Lock()
+	defer mut.Unlock()
+	filters = append(filters, filter)
+	sort.Sort(filterList(filters))
+}
+
+func deregisterFilter(filter *pfilter.PacketFilter) {
+	mut.Lock()
+	defer mut.Unlock()
+
+	for i, f := range filters {
+		if f == filter {
+			copy(filters[i:], filters[i+1:])
+			filters[len(filters)-1] = nil
+			filters = filters[:len(filters)-1]
+			break
+		}
+	}
+	sort.Sort(filterList(filters))
+}
+
+// Filters
+
+type kcpConversationFilter struct {
+	convID uint32
+}
+
+func (f *kcpConversationFilter) Outgoing(out []byte, addr net.Addr) {
+	if !f.isKCPConv(out) {
+		panic("not a kcp conversation")
+	}
+	atomic.StoreUint32(&f.convID, binary.LittleEndian.Uint32(out[:4]))
+}
+
+func (kcpConversationFilter) isKCPConv(data []byte) bool {
+	// Need atleast 5 bytes
+	if len(data) < 5 {
+		return false
+	}
+
+	// First 4 bytes convID
+	// 5th byte is cmd
+	// IKCP_CMD_PUSH    = 81 // cmd: push data
+	// IKCP_CMD_ACK     = 82 // cmd: ack
+	// IKCP_CMD_WASK    = 83 // cmd: window probe (ask)
+	// IKCP_CMD_WINS    = 84 // cmd: window size (tell)
+	return 80 < data[4] && data[4] < 85
+}
+
+func (f *kcpConversationFilter) ClaimIncoming(in []byte, addr net.Addr) bool {
+	if f.isKCPConv(in) {
+		convID := atomic.LoadUint32(&f.convID)
+		return convID != 0 && binary.LittleEndian.Uint32(in[:4]) == convID
+	}
+	return false
+}
+
+type stunFilter struct {
+	ids map[string]time.Time
+	mut sync.Mutex
+}
+
+func (f *stunFilter) Outgoing(out []byte, addr net.Addr) {
+	if !f.isStunPayload(out) {
+		panic("not a stun payload")
+	}
+	id := string(out[8:20])
+	f.mut.Lock()
+	f.ids[id] = time.Now().Add(time.Minute)
+	f.reap()
+	f.mut.Unlock()
+}
+
+func (f *stunFilter) ClaimIncoming(in []byte, addr net.Addr) bool {
+	if f.isStunPayload(in) {
+		id := string(in[8:20])
+		f.mut.Lock()
+		_, ok := f.ids[id]
+		f.reap()
+		f.mut.Unlock()
+		return ok
+	}
+	return false
+}
+
+func (f *stunFilter) isStunPayload(data []byte) bool {
+	// Need atleast 20 bytes
+	if len(data) < 20 {
+		return false
+	}
+
+	// First two bits always unset, and should always send magic cookie.
+	return data[0]&0xc0 == 0 && bytes.Equal(data[4:8], []byte{0x21, 0x12, 0xA4, 0x42})
+}
+
+func (f *stunFilter) reap() {
+	now := time.Now()
+	for id, timeout := range f.ids {
+		if timeout.Before(now) {
+			delete(f.ids, id)
+		}
+	}
+}
+
+type sessionClosingStream struct {
+	*yamux.Stream
+	session *yamux.Session
+}
+
+func (w *sessionClosingStream) Close() error {
+	err1 := w.Stream.Close()
+
+	deadline := time.Now().Add(5 * time.Second)
+	for w.session.NumStreams() > 0 && time.Now().Before(deadline) {
+		time.Sleep(200 * time.Millisecond)
+	}
+
+	err2 := w.session.Close()
+	if err1 != nil {
+		return err1
+	}
+	return err2
+}

lib/connections/relay_dial.go

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

lib/connections/structs.go

@@ -51,6 +51,8 @@ const (
 	connTypeRelayServer
 	connTypeTCPClient
 	connTypeTCPServer
+	connTypeKCPClient
+	connTypeKCPServer
 )
 
 func (t connType) String() string {
@@ -63,6 +65,10 @@ func (t connType) String() string {
 		return "tcp-client"
 	case connTypeTCPServer:
 		return "tcp-server"
+	case connTypeKCPClient:
+		return "kcp-client"
+	case connTypeKCPServer:
+		return "kcp-server"
 	default:
 		return "unknown-type"
 	}

lib/connections/tcp_dial.go

@@ -16,8 +16,6 @@ import (
 	"github.com/syncthing/syncthing/lib/protocol"
 )
 
-const tcpPriority = 10
-
 func init() {
 	factory := &tcpDialerFactory{}
 	for _, scheme := range []string{"tcp", "tcp4", "tcp6"} {
@@ -31,7 +29,7 @@ type tcpDialer struct {
 }
 
 func (d *tcpDialer) Dial(id protocol.DeviceID, uri *url.URL) (internalConn, error) {
-	uri = fixupPort(uri)
+	uri = fixupPort(uri, config.DefaultTCPPort)
 
 	conn, err := dialer.DialTimeout(uri.Scheme, uri.Host, 10*time.Second)
 	if err != nil {

lib/connections/tcp_listen.go

@@ -10,7 +10,6 @@ import (
 	"crypto/tls"
 	"net"
 	"net/url"
-	"strings"
 	"sync"
 	"time"
 
@@ -181,7 +180,7 @@ type tcpListenerFactory struct{}
 
 func (f *tcpListenerFactory) New(uri *url.URL, cfg *config.Wrapper, tlsCfg *tls.Config, conns chan internalConn, natService *nat.Service) genericListener {
 	return &tcpListener{
-		uri:        fixupPort(uri),
+		uri:        fixupPort(uri, config.DefaultTCPPort),
 		cfg:        cfg,
 		tlsCfg:     tlsCfg,
 		conns:      conns,
@@ -194,18 +193,3 @@ func (f *tcpListenerFactory) New(uri *url.URL, cfg *config.Wrapper, tlsCfg *tls.
 func (tcpListenerFactory) Enabled(cfg config.Configuration) bool {
 	return true
 }
-
-func fixupPort(uri *url.URL) *url.URL {
-	copyURI := *uri
-
-	host, port, err := net.SplitHostPort(uri.Host)
-	if err != nil && strings.Contains(err.Error(), "missing port") {
-		// addr is on the form "1.2.3.4"
-		copyURI.Host = net.JoinHostPort(uri.Host, "22000")
-	} else if err == nil && port == "" {
-		// addr is on the form "1.2.3.4:"
-		copyURI.Host = net.JoinHostPort(host, "22000")
-	}
-
-	return &copyURI
-}

lib/connections/util.go

@@ -0,0 +1,29 @@
+// Copyright (C) 2016 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
+
+import (
+	"net"
+	"net/url"
+	"strconv"
+	"strings"
+)
+
+func fixupPort(uri *url.URL, defaultPort int) *url.URL {
+	copyURI := *uri
+
+	host, port, err := net.SplitHostPort(uri.Host)
+	if err != nil && strings.Contains(err.Error(), "missing port") {
+		// addr is on the form "1.2.3.4"
+		copyURI.Host = net.JoinHostPort(uri.Host, strconv.Itoa(defaultPort))
+	} else if err == nil && port == "" {
+		// addr is on the form "1.2.3.4:"
+		copyURI.Host = net.JoinHostPort(host, strconv.Itoa(defaultPort))
+	}
+
+	return &copyURI
+}