// 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 https://mozilla.org/MPL/2.0/. package connections import ( "bytes" "context" "crypto/tls" "errors" "fmt" "io" "math/rand" "net" "net/url" "strings" "testing" "time" "github.com/thejerf/suture/v4" "github.com/syncthing/syncthing/lib/config" "github.com/syncthing/syncthing/lib/connections/registry" "github.com/syncthing/syncthing/lib/dialer" "github.com/syncthing/syncthing/lib/events" "github.com/syncthing/syncthing/lib/nat" "github.com/syncthing/syncthing/lib/protocol" "github.com/syncthing/syncthing/lib/sync" "github.com/syncthing/syncthing/lib/tlsutil" ) func TestFixupPort(t *testing.T) { cases := [][2]string{ {"tcp://1.2.3.4:5", "tcp://1.2.3.4:5"}, {"tcp://1.2.3.4:", "tcp://1.2.3.4:22000"}, {"tcp://1.2.3.4", "tcp://1.2.3.4:22000"}, {"tcp://[fe80::1]", "tcp://[fe80::1]:22000"}, {"tcp://[fe80::1]:", "tcp://[fe80::1]:22000"}, {"tcp://[fe80::1]:22000", "tcp://[fe80::1]:22000"}, {"tcp://[fe80::1]:22000", "tcp://[fe80::1]:22000"}, {"tcp://[fe80::1%25abc]", "tcp://[fe80::1%25abc]:22000"}, {"tcp://[fe80::1%25abc]:", "tcp://[fe80::1%25abc]:22000"}, {"tcp://[fe80::1%25abc]:22000", "tcp://[fe80::1%25abc]:22000"}, {"tcp://[fe80::1%25abc]:22000", "tcp://[fe80::1%25abc]:22000"}, } for _, tc := range cases { u0, err := url.Parse(tc[0]) if err != nil { t.Fatal(err) } u1 := fixupPort(u0, 22000).String() if u1 != tc[1] { t.Errorf("fixupPort(%q, 22000) => %q, expected %q", tc[0], u1, tc[1]) } } } func TestAllowedNetworks(t *testing.T) { cases := []struct { host string allowed []string ok bool }{ { "192.168.0.1", nil, false, }, { "192.168.0.1", []string{}, false, }, { "fe80::1", nil, false, }, { "fe80::1", []string{}, false, }, { "192.168.0.1", []string{"fe80::/48", "192.168.0.0/24"}, true, }, { "fe80::1", []string{"192.168.0.0/24", "fe80::/48"}, true, }, { "192.168.0.1", []string{"192.168.1.0/24", "fe80::/48"}, false, }, { "fe80::1", []string{"fe82::/48", "192.168.1.0/24"}, false, }, { "192.168.0.1:4242", []string{"fe80::/48", "192.168.0.0/24"}, true, }, { "[fe80::1]:4242", []string{"192.168.0.0/24", "fe80::/48"}, true, }, { "10.20.30.40", []string{"!10.20.30.0/24", "10.0.0.0/8"}, false, }, { "10.20.30.40", []string{"10.0.0.0/8", "!10.20.30.0/24"}, true, }, { "[fe80::1]:4242", []string{"192.168.0.0/24", "!fe00::/8", "fe80::/48"}, false, }, } for _, tc := range cases { res := IsAllowedNetwork(tc.host, tc.allowed) if res != tc.ok { t.Errorf("allowedNetwork(%q, %q) == %v, want %v", tc.host, tc.allowed, res, tc.ok) } } } 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) } } } func TestConnectionStatus(t *testing.T) { s := newConnectionStatusHandler() addr := "testAddr" testErr := errors.New("testErr") if stats := s.ConnectionStatus(); len(stats) != 0 { t.Fatal("newly created connectionStatusHandler isn't empty:", len(stats)) } check := func(in, out error) { t.Helper() s.setConnectionStatus(addr, in) switch stat, ok := s.ConnectionStatus()[addr]; { case !ok: t.Fatal("entry missing") case out == nil: if stat.Error != nil { t.Fatal("expected nil error, got", stat.Error) } case *stat.Error != out.Error(): t.Fatalf("expected %v error, got %v", out.Error(), *stat.Error) } } check(nil, nil) check(context.Canceled, nil) check(testErr, testErr) check(context.Canceled, testErr) check(nil, nil) } func TestNextDialRegistryCleanup(t *testing.T) { now := time.Now() firsts := []time.Time{ now.Add(-dialCoolDownInterval + time.Second), now.Add(-dialCoolDownDelay + time.Second), now.Add(-2 * dialCoolDownDelay), } r := make(nextDialRegistry) // Cases where the device should be cleaned up r[protocol.LocalDeviceID] = nextDialDevice{} r.sleepDurationAndCleanup(now) if l := len(r); l > 0 { t.Errorf("Expected empty to be cleaned up, got length %v", l) } for _, dev := range []nextDialDevice{ // attempts below threshold, outside of interval { attempts: 1, coolDownIntervalStart: firsts[1], }, { attempts: 1, coolDownIntervalStart: firsts[2], }, // Threshold reached, but outside of cooldown delay { attempts: dialCoolDownMaxAttemps, coolDownIntervalStart: firsts[2], }, } { r[protocol.LocalDeviceID] = dev r.sleepDurationAndCleanup(now) if l := len(r); l > 0 { t.Errorf("attempts: %v, start: %v: Expected all cleaned up, got length %v", dev.attempts, dev.coolDownIntervalStart, l) } } // Cases where the device should stay monitored for _, dev := range []nextDialDevice{ // attempts below threshold, inside of interval { attempts: 1, coolDownIntervalStart: firsts[0], }, // attempts at threshold, inside delay { attempts: dialCoolDownMaxAttemps, coolDownIntervalStart: firsts[0], }, { attempts: dialCoolDownMaxAttemps, coolDownIntervalStart: firsts[1], }, } { r[protocol.LocalDeviceID] = dev r.sleepDurationAndCleanup(now) if l := len(r); l != 1 { t.Errorf("attempts: %v, start: %v: Expected device still tracked, got length %v", dev.attempts, dev.coolDownIntervalStart, l) } } } func BenchmarkConnections(b *testing.B) { addrs := []string{ "tcp://127.0.0.1:0", "quic://127.0.0.1:0", "relay://127.0.0.1:22067", } sizes := []int{ 1 << 10, 1 << 15, 1 << 20, 1 << 22, } haveRelay := false // Check if we have a relay running locally conn, err := net.DialTimeout("tcp", "127.0.0.1:22067", 100*time.Millisecond) if err == nil { haveRelay = true _ = conn.Close() } for _, addr := range addrs { for _, sz := range sizes { data := make([]byte, sz) if _, err := rand.Read(data); err != nil { b.Fatal(err) } for _, direction := range []string{"cs", "sc"} { proto := strings.SplitN(addr, ":", 2)[0] b.Run(fmt.Sprintf("%s_%d_%s", proto, sz, direction), func(b *testing.B) { if proto == "relay" && !haveRelay { b.Skip("could not connect to relay") } withConnectionPair(b, addr, func(client, server internalConn) { if direction == "sc" { server, client = client, server } total := 0 b.ResetTimer() for i := 0; i < b.N; i++ { wg := sync.NewWaitGroup() wg.Add(2) errC := make(chan error, 2) go func() { wg.Wait() close(errC) }() go func() { if _, err := client.Write(data); err != nil { errC <- err return } wg.Done() }() go func() { if _, err := io.ReadFull(server, data); err != nil { errC <- err return } total += sz wg.Done() }() err := <-errC if err != nil { b.Fatal(err) } } b.ReportAllocs() b.SetBytes(int64(total / b.N)) }) }) } } } } func TestConnectionEstablishment(t *testing.T) { addrs := []string{ "tcp://127.0.0.1:0", "quic://127.0.0.1:0", } send := make([]byte, 128<<10) if _, err := rand.Read(send); err != nil { t.Fatal(err) } for _, addr := range addrs { proto := strings.SplitN(addr, ":", 2)[0] t.Run(proto, func(t *testing.T) { withConnectionPair(t, addr, func(client, server internalConn) { if _, err := client.Write(send); err != nil { t.Fatal(err) } recv := make([]byte, len(send)) if _, err := io.ReadFull(server, recv); err != nil { t.Fatal(err) } if !bytes.Equal(recv, send) { t.Fatal("data mismatch") } }) }) } } func withConnectionPair(b interface{ Fatal(...interface{}) }, connUri string, h func(client, server internalConn)) { // We're connecting to ourselves over localhost. Reuseport doesn't work // well in this scenario since we can't typically have a connection with // the same source and destination port... dialer.SupportsReusePort = false // Root of the service tree. supervisor := suture.New("main", suture.Spec{ PassThroughPanics: true, }) cert := mustGetCert(b) deviceId := protocol.NewDeviceID(cert.Certificate[0]) tlsCfg := tlsutil.SecureDefaultTLS13() tlsCfg.Certificates = []tls.Certificate{cert} tlsCfg.NextProtos = []string{"bench"} tlsCfg.ClientAuth = tls.RequestClientCert tlsCfg.SessionTicketsDisabled = true tlsCfg.InsecureSkipVerify = true ctx, cancel := context.WithCancel(context.Background()) defer cancel() supervisor.ServeBackground(ctx) cfg := config.Configuration{ Options: config.OptionsConfiguration{ RelaysEnabled: true, }, } wcfg := config.Wrap("", cfg, deviceId, events.NoopLogger) uri, err := url.Parse(connUri) if err != nil { b.Fatal(err) } lf, err := getListenerFactory(cfg, uri) if err != nil { b.Fatal(err) } natSvc := nat.NewService(deviceId, wcfg) conns := make(chan internalConn, 1) fmt.Println("listening", uri) listenSvc := lf.New(uri, wcfg, tlsCfg, conns, natSvc, registry.New()) supervisor.Add(listenSvc) var addr *url.URL for { addrs := listenSvc.LANAddresses() if len(addrs) > 0 { if !strings.HasSuffix(addrs[0].Host, ":0") { addr = addrs[0] break } } time.Sleep(time.Millisecond) } df, err := getDialerFactory(cfg, addr) if err != nil { b.Fatal(err) } // Purposely using a different registry: Don't want to reuse port between dialer and listener on the same device dialer := df.New(cfg.Options, tlsCfg, registry.New()) // Relays might take some time to register the device, so dial multiple times fmt.Println("dialing", addr) clientConn, err := dialer.Dial(ctx, deviceId, addr) if err != nil { fmt.Println("error", err) for i := 0; i < 10 && err != nil; i++ { clientConn, err = dialer.Dial(ctx, deviceId, addr) time.Sleep(100 * time.Millisecond) } if err != nil { b.Fatal(err) } } fmt.Println("clientConn", clientConn.LocalAddr(), "->", clientConn.RemoteAddr()) // Quic does not start a stream until some data is sent through, so send something for the AcceptStream // to fire on the other side. send := []byte("hello") if _, err := clientConn.Write(send); err != nil { b.Fatal(err) } serverConn := <-conns fmt.Println("serverConn", serverConn.LocalAddr(), "->", serverConn.RemoteAddr()) recv := make([]byte, len(send)) if _, err := io.ReadFull(serverConn, recv); err != nil { b.Fatal(err) } if !bytes.Equal(recv, send) { b.Fatal("data mismatch") } h(clientConn, serverConn) _ = clientConn.Close() _ = serverConn.Close() } func mustGetCert(b interface{ Fatal(...interface{}) }) tls.Certificate { cert, err := tlsutil.NewCertificateInMemory("bench", 10) if err != nil { b.Fatal(err) } return cert }