perso/syncthing-arm
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Remove KCP (fixes #4737) (#4741)

Browse Source
This commit is contained in:
Jakob Borg
2018-02-09 11:40:57 +01:00
committed by GitHub
parent 97068c10f3
commit b97d5bcca8
54 changed files with 268 additions and 6169 deletions
Download Patch File Download Diff File Expand all files Collapse all files

50
lib/config/config.go
View File

@@ -39,8 +39,6 @@ const (
var (
// DefaultTCPPort defines default TCP port used if the URI does not specify one, for example tcp://0.0.0.0
DefaultTCPPort = 22000
// DefaultKCPPort defines default KCP (UDP) port used if the URI does not specify one, for example kcp://0.0.0.0
DefaultKCPPort = 22020
// DefaultListenAddresses should be substituted when the configuration
// contains <listenAddress>default</listenAddress>. This is done by the
// "consumer" of the configuration as we don't want these saved to the
@@ -48,7 +46,6 @@ var (
DefaultListenAddresses = []string{
util.Address("tcp", net.JoinHostPort("0.0.0.0", strconv.Itoa(DefaultTCPPort))),
"dynamic+https://relays.syncthing.net/endpoint",
util.Address("kcp", net.JoinHostPort("0.0.0.0", strconv.Itoa(DefaultKCPPort))),
}
// DefaultDiscoveryServersV4 should be substituted when the configuration
// contains <globalAnnounceServer>default-v4</globalAnnounceServer>.
@@ -65,25 +62,6 @@ var (
// DefaultDiscoveryServers should be substituted when the configuration
// contains <globalAnnounceServer>default</globalAnnounceServer>.
DefaultDiscoveryServers = append(DefaultDiscoveryServersV4, DefaultDiscoveryServersV6...)
// DefaultStunServers should be substituted when the configuration
// contains <stunServer>default</stunServer>.
DefaultStunServers = []string{
"stun.callwithus.com:3478",
"stun.counterpath.com:3478",
"stun.counterpath.net:3478",
"stun.ekiga.net:3478",
"stun.ideasip.com:3478",
"stun.internetcalls.com:3478",
"stun.schlund.de:3478",
"stun.sipgate.net:10000",
"stun.sipgate.net:3478",
"stun.voip.aebc.com:3478",
"stun.voiparound.com:3478",
"stun.voipbuster.com:3478",
"stun.voipstunt.com:3478",
"stun.voxgratia.org:3478",
"stun.xten.com:3478",
}
// DefaultTheme is the default and fallback theme for the web UI.
DefaultTheme = "default"
)
@@ -380,6 +358,16 @@ func (cfg *Configuration) clean() error {
}
cfg.IgnoredDevices = newIgnoredDevices
// Deprecated protocols are removed from the list of listeners and
// device addresses. So far just kcp*.
for _, prefix := range []string{"kcp"} {
cfg.Options.ListenAddresses = filterURLSchemePrefix(cfg.Options.ListenAddresses, prefix)
for i := range cfg.Devices {
dev := &cfg.Devices[i]
dev.Addresses = filterURLSchemePrefix(dev.Addresses, prefix)
}
}
return nil
}
@@ -768,3 +756,21 @@ func cleanSymlinks(filesystem fs.Filesystem, dir string) {
return nil
})
}
// filterURLSchemePrefix returns the list of addresses after removing all
// entries whose URL scheme matches the given prefix.
func filterURLSchemePrefix(addrs []string, prefix string) []string {
for i := 0; i < len(addrs); i++ {
uri, err := url.Parse(addrs[i])
if err != nil {
continue
}
if strings.HasPrefix(uri.Scheme, prefix) {
// Remove this entry
copy(addrs[i:], addrs[i+1:])
addrs = addrs[:len(addrs)-1]
i--
}
}
return addrs
}

40
lib/config/config_test.go
View File

@@ -68,13 +68,6 @@ func TestDefaultValues(t *testing.T) {
TempIndexMinBlocks: 10,
UnackedNotificationIDs: []string{},
WeakHashSelectionMethod: WeakHashAuto,
StunKeepaliveS: 24,
StunServers: []string{"default"},
KCPCongestionControl: true,
KCPReceiveWindowSize: 128,
KCPSendWindowSize: 128,
KCPUpdateIntervalMs: 25,
KCPFastResend: false,
DefaultFolderPath: "~",
SetLowPriority: true,
}
@@ -217,13 +210,6 @@ func TestOverriddenValues(t *testing.T) {
"channelNotification", // added in 17->18 migration
},
WeakHashSelectionMethod: WeakHashNever,
StunKeepaliveS: 10,
StunServers: []string{"a.stun.com", "b.stun.com"},
KCPCongestionControl: false,
KCPReceiveWindowSize: 1280,
KCPSendWindowSize: 1280,
KCPUpdateIntervalMs: 1000,
KCPFastResend: true,
DefaultFolderPath: "/media/syncthing",
SetLowPriority: false,
}
@@ -953,6 +939,32 @@ func TestInvalidFolderIDRejected(t *testing.T) {
}
}
func TestFilterURLSchemePrefix(t *testing.T) {
cases := []struct {
before []string
prefix string
after []string
}{
{[]string{}, "kcp", []string{}},
{[]string{"tcp://foo"}, "kcp", []string{"tcp://foo"}},
{[]string{"kcp://foo"}, "kcp", []string{}},
{[]string{"tcp6://foo", "kcp6://foo"}, "kcp", []string{"tcp6://foo"}},
{[]string{"kcp6://foo", "tcp6://foo"}, "kcp", []string{"tcp6://foo"}},
{
[]string{"tcp://foo", "tcp4://foo", "kcp://foo", "kcp4://foo", "banana://foo", "banana4://foo", "banananas!"},
"kcp",
[]string{"tcp://foo", "tcp4://foo", "banana://foo", "banana4://foo", "banananas!"},
},
}
for _, tc := range cases {
res := filterURLSchemePrefix(tc.before, tc.prefix)
if !reflect.DeepEqual(res, tc.after) {
t.Errorf("filterURLSchemePrefix => %q, expected %q", res, tc.after)
}
}
}
// defaultConfigAsMap returns a valid default config as a JSON-decoded
// map[string]interface{}. This is useful to override random elements and
// re-encode into JSON.

8
lib/config/optionsconfiguration.go
View File

@@ -134,14 +134,6 @@ type OptionsConfiguration struct {
UnackedNotificationIDs []string `xml:"unackedNotificationID" json:"unackedNotificationIDs"`
TrafficClass int `xml:"trafficClass" json:"trafficClass"`
WeakHashSelectionMethod WeakHashSelectionMethod `xml:"weakHashSelectionMethod" json:"weakHashSelectionMethod" restart:"true"`
StunServers []string `xml:"stunServer" json:"stunServers" default:"default"`
StunKeepaliveS int `xml:"stunKeepaliveSeconds" json:"stunKeepaliveSeconds" default:"24"`
KCPNoDelay bool `xml:"kcpNoDelay" json:"kcpNoDelay" default:"false"`
KCPUpdateIntervalMs int `xml:"kcpUpdateIntervalMs" json:"kcpUpdateIntervalMs" default:"25"`
KCPFastResend bool `xml:"kcpFastResend" json:"kcpFastResend" default:"false"`
KCPCongestionControl bool `xml:"kcpCongestionControl" json:"kcpCongestionControl" default:"true"`
KCPSendWindowSize int `xml:"kcpSendWindowSize" json:"kcpSendWindowSize" default:"128"`
KCPReceiveWindowSize int `xml:"kcpReceiveWindowSize" json:"kcpReceiveWindowSize" default:"128"`
DefaultFolderPath string `xml:"defaultFolderPath" json:"defaultFolderPath" default:"~"`
SetLowPriority bool `xml:"setLowPriority" json:"setLowPriority" default:"true"`

8
lib/config/testdata/overridenvalues.xml vendored
View File

@@ -35,14 +35,6 @@
<overwriteRemoteDeviceNamesOnConnect>true</overwriteRemoteDeviceNamesOnConnect>
<tempIndexMinBlocks>100</tempIndexMinBlocks>
<weakHashSelectionMethod>never</weakHashSelectionMethod>
<stunKeepaliveSeconds>10</stunKeepaliveSeconds>
<stunServer>a.stun.com</stunServer>
<stunServer>b.stun.com</stunServer>
<kcpCongestionControl>false</kcpCongestionControl>
<kcpReceiveWindowSize>1280</kcpReceiveWindowSize>
<kcpSendWindowSize>1280</kcpSendWindowSize>
<kcpUpdateIntervalMs>1000</kcpUpdateIntervalMs>
<kcpFastResend>true</kcpFastResend>
<defaultFolderPath>/media/syncthing</defaultFolderPath>
<setLowPriority>false</setLowPriority>
</options>

24
lib/config/wrapper.go
View File

@@ -15,7 +15,6 @@ import (
"github.com/syncthing/syncthing/lib/fs"
"github.com/syncthing/syncthing/lib/osutil"
"github.com/syncthing/syncthing/lib/protocol"
"github.com/syncthing/syncthing/lib/rand"
"github.com/syncthing/syncthing/lib/sync"
"github.com/syncthing/syncthing/lib/util"
)
@@ -433,29 +432,6 @@ func (w *Wrapper) setRequiresRestart() {
atomic.StoreUint32(&w.requiresRestart, 1)
}
func (w *Wrapper) StunServers() []string {
var addresses []string
for _, addr := range w.cfg.Options.StunServers {
switch addr {
case "default":
addresses = append(addresses, DefaultStunServers...)
default:
addresses = append(addresses, addr)
}
}
addresses = util.UniqueStrings(addresses)
// Shuffle
l := len(addresses)
for i := range addresses {
r := rand.Intn(l)
addresses[i], addresses[r] = addresses[r], addresses[i]
}
return addresses
}
func (w *Wrapper) MyName() string {
w.mut.Lock()
myID := w.cfg.MyID

21
lib/connections/config.go
View File

@@ -6,28 +6,7 @@
package connections
import (
"time"
"github.com/xtaci/smux"
)
const (
tcpPriority = 10
kcpPriority = 50
relayPriority = 200
// KCP filter priorities
kcpNoFilterPriority = 100
kcpConversationFilterPriority = 20
kcpStunFilterPriority = 10
)
var (
smuxConfig = &smux.Config{
KeepAliveInterval: 10 * time.Second,
KeepAliveTimeout: 30 * time.Second,
MaxFrameSize: 4096,
MaxReceiveBuffer: 4 * 1024 * 1024,
}
)

66
lib/connections/connections_test.go
View File

@@ -6,8 +6,13 @@
package connections
import "testing"
import "net/url"
import (
"net/url"
"testing"
"github.com/syncthing/syncthing/lib/config"
"github.com/syncthing/syncthing/lib/protocol"
)
func TestFixupPort(t *testing.T) {
cases := [][2]string{
@@ -105,3 +110,60 @@ func TestAllowedNetworks(t *testing.T) {
}
}
}
func TestGetDialer(t *testing.T) {
mustParseURI := func(v string) *url.URL {
uri, err := url.Parse(v)
if err != nil {
panic(err)
}
return uri
}
cases := []struct {
uri *url.URL
ok bool
disabled bool
deprecated bool
}{
{mustParseURI("tcp://1.2.3.4:5678"), true, false, false}, // ok
{mustParseURI("tcp4://1.2.3.4:5678"), true, false, false}, // ok
{mustParseURI("kcp://1.2.3.4:5678"), false, false, true}, // deprecated
{mustParseURI("relay://1.2.3.4:5678"), false, true, false}, // disabled
{mustParseURI("http://1.2.3.4:5678"), false, false, false}, // generally bad
{mustParseURI("bananas!"), false, false, false}, // wat
}
cfg := config.New(protocol.LocalDeviceID)
cfg.Options.RelaysEnabled = false
for _, tc := range cases {
df, err := getDialerFactory(cfg, tc.uri)
if tc.ok && err != nil {
t.Errorf("getDialerFactory(%q) => %v, expected nil err", tc.uri, err)
}
if tc.ok && df == nil {
t.Errorf("getDialerFactory(%q) => nil factory, expected non-nil", tc.uri)
}
if tc.deprecated && err != errDeprecated {
t.Errorf("getDialerFactory(%q) => %v, expected %v", tc.uri, err, errDeprecated)
}
if tc.disabled && err != errDisabled {
t.Errorf("getDialerFactory(%q) => %v, expected %v", tc.uri, err, errDisabled)
}
lf, err := getListenerFactory(cfg, tc.uri)
if tc.ok && err != nil {
t.Errorf("getListenerFactory(%q) => %v, expected nil err", tc.uri, err)
}
if tc.ok && lf == nil {
t.Errorf("getListenerFactory(%q) => nil factory, expected non-nil", tc.uri)
}
if tc.deprecated && err != errDeprecated {
t.Errorf("getListenerFactory(%q) => %v, expected %v", tc.uri, err, errDeprecated)
}
if tc.disabled && err != errDisabled {
t.Errorf("getListenerFactory(%q) => %v, expected %v", tc.uri, err, errDisabled)
}
}
}

36
lib/connections/deprecated.go Normal file
View File

@@ -0,0 +1,36 @@
// Copyright (C) 2018 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 connections
import "github.com/syncthing/syncthing/lib/config"
// deprecatedListener is never valid
type deprecatedListener struct {
listenerFactory
}
func (deprecatedListener) Valid(_ config.Configuration) error {
return errDeprecated
}
// deprecatedDialer is never valid
type deprecatedDialer struct {
dialerFactory
}
func (deprecatedDialer) Valid(_ config.Configuration) error {
return errDeprecated
}
func init() {
listeners["kcp"] = deprecatedListener{}
listeners["kcp4"] = deprecatedListener{}
listeners["kcp6"] = deprecatedListener{}
dialers["kcp"] = deprecatedDialer{}
dialers["kcp4"] = deprecatedDialer{}
dialers["kcp6"] = deprecatedDialer{}
}

112
lib/connections/kcp_dial.go
View File

@@ -1,112 +0,0 @@
// 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/syncthing/syncthing/lib/config"
"github.com/syncthing/syncthing/lib/protocol"
"github.com/xtaci/kcp-go"
"github.com/xtaci/smux"
)
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 {
return internalConn{}, err
}
opts := d.cfg.Options()
conn.SetStreamMode(true)
conn.SetACKNoDelay(false)
conn.SetWindowSize(opts.KCPSendWindowSize, opts.KCPReceiveWindowSize)
conn.SetNoDelay(boolInt(opts.KCPNoDelay), opts.KCPUpdateIntervalMs, boolInt(opts.KCPFastResend), boolInt(!opts.KCPCongestionControl))
ses, err := smux.Client(conn, smuxConfig)
if err != nil {
conn.Close()
return internalConn{}, err
}
ses.SetDeadline(time.Now().Add(10 * time.Second))
stream, err := ses.OpenStream()
if err != nil {
ses.Close()
return internalConn{}, err
}
ses.SetDeadline(time.Time{})
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) AlwaysWAN() bool {
return false
}
func (kcpDialerFactory) Enabled(cfg config.Configuration) bool {
return true
}
func (kcpDialerFactory) String() string {
return "KCP Dialer"
}

326
lib/connections/kcp_listen.go
View File

@@ -1,326 +0,0 @@
// 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"
"sync/atomic"
"time"
"github.com/AudriusButkevicius/pfilter"
"github.com/ccding/go-stun/stun"
"github.com/xtaci/kcp-go"
"github.com/xtaci/smux"
"github.com/syncthing/syncthing/lib/config"
"github.com/syncthing/syncthing/lib/nat"
)
const stunRetryInterval = 5 * time.Minute
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
nat atomic.Value
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("Listen (BEP/kcp): Accepting connection:", err)
}
continue
}
opts := t.cfg.Options()
conn.SetStreamMode(true)
conn.SetACKNoDelay(false)
conn.SetWindowSize(opts.KCPSendWindowSize, opts.KCPReceiveWindowSize)
conn.SetNoDelay(boolInt(opts.KCPNoDelay), opts.KCPUpdateIntervalMs, boolInt(opts.KCPFastResend), boolInt(!opts.KCPCongestionControl))
l.Debugln("connect from", conn.RemoteAddr())
ses, err := smux.Server(conn, smuxConfig)
if err != nil {
l.Debugln("smux server:", err)
conn.Close()
continue
}
ses.SetDeadline(time.Now().Add(10 * time.Second))
stream, err := ses.AcceptStream()
if err != nil {
l.Debugln("smux accept:", err)
ses.Close()
continue
}
ses.SetDeadline(time.Time{})
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) NATType() string {
v := t.nat.Load().(stun.NATType)
if v == stun.NATUnknown || v == stun.NATError {
return "unknown"
}
return v.String()
}
func (t *kcpListener) stunRenewal(listener net.PacketConn) {
client := stun.NewClientWithConnection(listener)
client.SetSoftwareName("syncthing")
var natType stun.NATType
var extAddr *stun.Host
var udpAddr *net.UDPAddr
var err error
oldType := stun.NATUnknown
for {
disabled:
if t.cfg.Options().StunKeepaliveS < 1 {
time.Sleep(time.Second)
oldType = stun.NATUnknown
t.nat.Store(stun.NATUnknown)
t.mut.Lock()
t.address = nil
t.mut.Unlock()
continue
}
for _, addr := range t.cfg.StunServers() {
// Resolve the address, so that in case the server advertises two
// IPs, we always hit the same one, as otherwise, the mapping might
// expire as we hit the other address, and cause us to flip flop
// between servers/external addresses, as a result flooding discovery
// servers.
udpAddr, err = net.ResolveUDPAddr("udp", addr)
if err != nil {
l.Debugf("%s stun addr resolution on %s: %s", t.uri, addr, err)
continue
}
client.SetServerAddr(udpAddr.String())
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)
t.nat.Store(natType)
oldType = natType
}
// We can't punch through this one, so no point doing keepalives
// and such, just try again in a minute and hope that the NAT type changes.
if !isPunchable(natType) {
break
}
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
}
}
}
// We failed to contact all provided stun servers or the nat is not punchable.
// Chillout for a while.
time.Sleep(stunRetryInterval)
}
}
type kcpListenerFactory struct{}
func (f *kcpListenerFactory) New(uri *url.URL, cfg *config.Wrapper, tlsCfg *tls.Config, conns chan internalConn, natService *nat.Service) genericListener {
l := &kcpListener{
uri: fixupPort(uri, config.DefaultKCPPort),
cfg: cfg,
tlsCfg: tlsCfg,
conns: conns,
stop: make(chan struct{}),
factory: f,
}
l.nat.Store(stun.NATUnknown)
return l
}
func (kcpListenerFactory) Enabled(cfg config.Configuration) bool {
return true
}
func isPunchable(natType stun.NATType) bool {
return natType == stun.NATNone || natType == stun.NATPortRestricted || natType == stun.NATRestricted || natType == stun.NATFull
}

194
lib/connections/kcp_misc.go
View File

@@ -1,194 +0,0 @@
// 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/xtaci/kcp-go"
"github.com/xtaci/smux"
)
var (
mut sync.Mutex
filters filterList
)
func init() {
kcp.BlacklistDuration = 10 * time.Minute
}
type filterList []*pfilter.PacketFilter
// Sort connections by whether they are unspecified or not, as connections
// listening 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 at least 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 at least 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 {
*smux.Stream
session *smux.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
}
func boolInt(b bool) int {
if b {
return 1
}
return 0
}

7
lib/connections/relay_dial.go
View File

@@ -85,8 +85,11 @@ func (relayDialerFactory) AlwaysWAN() bool {
return true
}
func (relayDialerFactory) Enabled(cfg config.Configuration) bool {
return cfg.Options.RelaysEnabled
func (relayDialerFactory) Valid(cfg config.Configuration) error {
if !cfg.Options.RelaysEnabled {
return errDisabled
}
return nil
}
func (relayDialerFactory) String() string {

7
lib/connections/relay_listen.go
View File

@@ -190,6 +190,9 @@ func (f *relayListenerFactory) New(uri *url.URL, cfg *config.Wrapper, tlsCfg *tl
}
}
func (relayListenerFactory) Enabled(cfg config.Configuration) bool {
return cfg.Options.RelaysEnabled
func (relayListenerFactory) Valid(cfg config.Configuration) error {
if !cfg.Options.RelaysEnabled {
return errDisabled
}
return nil
}

59
lib/connections/service.go
View File

@@ -39,6 +39,11 @@ var (
listeners = make(map[string]listenerFactory, 0)
)
var (
errDisabled = errors.New("disabled by configuration")
errDeprecated = errors.New("deprecated protocol")
)
const (
perDeviceWarningIntv = 15 * time.Minute
tlsHandshakeTimeout = 10 * time.Second
@@ -149,10 +154,6 @@ func NewService(cfg *config.Wrapper, myID protocol.DeviceID, mdl Model, tlsCfg *
return service
}
var (
errDisabled = errors.New("disabled by configuration")
)
func (s *Service) handle() {
next:
for c := range s.conns {
@@ -293,7 +294,7 @@ func (s *Service) connect() {
bestDialerPrio := 1<<31 - 1 // worse prio won't build on 32 bit
for _, df := range dialers {
if !df.Enabled(cfg) {
if df.Valid(cfg) != nil {
continue
}
if prio := df.Priority(); prio < bestDialerPrio {
@@ -367,13 +368,18 @@ func (s *Service) connect() {
}
}
dialerFactory, err := s.getDialerFactory(cfg, uri)
if err == errDisabled {
l.Debugln(dialerFactory, "for", uri, "is disabled")
dialerFactory, err := getDialerFactory(cfg, uri)
switch err {
case nil:
// all good
case errDisabled:
l.Debugln("Dialer for", uri, "is disabled")
continue
}
if err != nil {
l.Infof("%v for %v: %v", dialerFactory, uri, err)
case errDeprecated:
l.Debugln("Dialer for", uri, "is deprecated")
continue
default:
l.Infof("Dialer for %v: %v", uri, err)
continue
}
@@ -537,13 +543,18 @@ func (s *Service) CommitConfiguration(from, to config.Configuration) bool {
continue
}
factory, err := s.getListenerFactory(to, uri)
if err == errDisabled {
factory, err := getListenerFactory(to, uri)
switch err {
case nil:
// all good
case errDisabled:
l.Debugln("Listener for", uri, "is disabled")
continue
}
if err != nil {
l.Infof("Getting listener factory for %v: %v", uri, err)
case errDeprecated:
l.Debugln("Listener for", uri, "is deprecated")
continue
default:
l.Infof("Listener for %v: %v", uri, err)
continue
}
@@ -552,7 +563,7 @@ func (s *Service) CommitConfiguration(from, to config.Configuration) bool {
}
for addr, listener := range s.listeners {
if _, ok := seen[addr]; !ok || !listener.Factory().Enabled(to) {
if _, ok := seen[addr]; !ok || listener.Factory().Valid(to) != nil {
l.Debugln("Stopping listener", addr)
s.listenerSupervisor.Remove(s.listenerTokens[addr])
delete(s.listenerTokens, addr)
@@ -633,27 +644,25 @@ func (s *Service) NATType() string {
return "unknown"
}
func (s *Service) getDialerFactory(cfg config.Configuration, uri *url.URL) (dialerFactory, error) {
func getDialerFactory(cfg config.Configuration, uri *url.URL) (dialerFactory, error) {
dialerFactory, ok := dialers[uri.Scheme]
if !ok {
return nil, fmt.Errorf("unknown address scheme %q", uri.Scheme)
}
if !dialerFactory.Enabled(cfg) {
return nil, errDisabled
if err := dialerFactory.Valid(cfg); err != nil {
return nil, err
}
return dialerFactory, nil
}
func (s *Service) getListenerFactory(cfg config.Configuration, uri *url.URL) (listenerFactory, error) {
func getListenerFactory(cfg config.Configuration, uri *url.URL) (listenerFactory, error) {
listenerFactory, ok := listeners[uri.Scheme]
if !ok {
return nil, fmt.Errorf("unknown address scheme %q", uri.Scheme)
}
if !listenerFactory.Enabled(cfg) {
return nil, errDisabled
if err := listenerFactory.Valid(cfg); err != nil {
return nil, err
}
return listenerFactory, nil

12
lib/connections/structs.go
View File

@@ -54,8 +54,6 @@ const (
connTypeRelayServer
connTypeTCPClient
connTypeTCPServer
connTypeKCPClient
connTypeKCPServer
)
func (t connType) String() string {
@@ -68,10 +66,6 @@ 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"
}
@@ -83,8 +77,6 @@ func (t connType) Transport() string {
return "relay"
case connTypeTCPClient, connTypeTCPServer:
return "tcp"
case connTypeKCPClient, connTypeKCPServer:
return "kcp"
default:
return "unknown"
}
@@ -122,7 +114,7 @@ type dialerFactory interface {
New(*config.Wrapper, *tls.Config) genericDialer
Priority() int
AlwaysWAN() bool
Enabled(config.Configuration) bool
Valid(config.Configuration) error
String() string
}
@@ -133,7 +125,7 @@ type genericDialer interface {
type listenerFactory interface {
New(*url.URL, *config.Wrapper, *tls.Config, chan internalConn, *nat.Service) genericListener
Enabled(config.Configuration) bool
Valid(config.Configuration) error
}
type genericListener interface {

5
lib/connections/tcp_dial.go
View File

@@ -77,8 +77,9 @@ func (tcpDialerFactory) AlwaysWAN() bool {
return false
}
func (tcpDialerFactory) Enabled(cfg config.Configuration) bool {
return true
func (tcpDialerFactory) Valid(_ config.Configuration) error {
// Always valid
return nil
}
func (tcpDialerFactory) String() string {

5
lib/connections/tcp_listen.go
View File

@@ -193,6 +193,7 @@ func (f *tcpListenerFactory) New(uri *url.URL, cfg *config.Wrapper, tlsCfg *tls.
}
}
func (tcpListenerFactory) Enabled(cfg config.Configuration) bool {
return true
func (tcpListenerFactory) Valid(_ config.Configuration) error {
// Always valid
return nil
}

61
lib/protocol/benchmark_test.go
View File

@@ -9,7 +9,6 @@ import (
"testing"
"github.com/syncthing/syncthing/lib/dialer"
"github.com/xtaci/kcp-go"
)
func BenchmarkRequestsRawTCP(b *testing.B) {
@@ -29,25 +28,6 @@ func BenchmarkRequestsRawTCP(b *testing.B) {
benchmarkRequestsConnPair(b, conn0, conn1)
}
func BenchmarkRequestsRawKCP(b *testing.B) {
b.Skip("KCP broken")
// Benchmarks the rate at which we can serve requests over a single,
// unencrypted KCP channel over the loopback interface.
// Get a connected KCP pair
conn0, conn1, err := getKCPConnectionPair()
if err != nil {
b.Fatal(err)
}
defer conn0.Close()
defer conn1.Close()
// Bench it
benchmarkRequestsConnPair(b, conn0, conn1)
}
func BenchmarkRequestsTLSoTCP(b *testing.B) {
conn0, conn1, err := getTCPConnectionPair()
if err != nil {
@@ -58,18 +38,6 @@ func BenchmarkRequestsTLSoTCP(b *testing.B) {
benchmarkRequestsTLS(b, conn0, conn1)
}
func BenchmarkRequestsTLSoKCP(b *testing.B) {
b.Skip("KCP broken")
conn0, conn1, err := getKCPConnectionPair()
if err != nil {
b.Fatal(err)
}
defer conn0.Close()
defer conn1.Close()
benchmarkRequestsTLS(b, conn0, conn1)
}
func benchmarkRequestsTLS(b *testing.B, conn0, conn1 net.Conn) {
// Benchmarks the rate at which we can serve requests over a single,
// TLS encrypted channel over the loopback interface.
@@ -170,35 +138,6 @@ func getTCPConnectionPair() (net.Conn, net.Conn, error) {
return conn0, conn1, nil
}
func getKCPConnectionPair() (net.Conn, net.Conn, error) {
lst, err := kcp.Listen("127.0.0.1:0")
if err != nil {
return nil, nil, err
}
var conn0 net.Conn
var err0 error
done := make(chan struct{})
go func() {
conn0, err0 = lst.Accept()
close(done)
}()
// Dial the connection
conn1, err := kcp.Dial(lst.Addr().String())
if err != nil {
return nil, nil, err
}
// Check any error from accept
<-done
if err0 != nil {
return nil, nil, err0
}
return conn0, conn1, nil
}
func negotiateTLS(cert tls.Certificate, conn0, conn1 net.Conn) (net.Conn, net.Conn) {
cfg := &tls.Config{
Certificates: []tls.Certificate{cert},