2016-05-04 19:38:12 +00:00
// Copyright (C) 2015 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,
2017-02-09 06:52:18 +00:00
// You can obtain one at https://mozilla.org/MPL/2.0/.
2016-05-04 19:38:12 +00:00
2021-05-08 10:52:06 +00:00
//go:generate -command counterfeiter go run github.com/maxbrunsfeld/counterfeiter/v6
2021-03-03 07:53:50 +00:00
//go:generate counterfeiter -o mocks/service.go --fake-name Service . Service
2016-05-04 19:38:12 +00:00
package connections
import (
2019-11-21 07:41:15 +00:00
"context"
2023-09-06 10:52:01 +00:00
"crypto/rand"
2016-05-04 19:38:12 +00:00
"crypto/tls"
2022-04-07 15:35:33 +00:00
"crypto/x509"
2023-09-06 10:52:01 +00:00
"encoding/base32"
"encoding/binary"
2022-08-16 08:01:49 +00:00
"errors"
2016-05-04 19:38:12 +00:00
"fmt"
2023-09-06 10:52:01 +00:00
"io"
2020-12-21 15:40:13 +00:00
"math"
2016-05-04 19:38:12 +00:00
"net"
"net/url"
2024-02-10 20:02:42 +00:00
"slices"
2017-11-15 09:36:33 +00:00
"sort"
2017-04-13 17:43:29 +00:00
"strings"
2017-12-07 07:08:24 +00:00
stdsync "sync"
2016-05-04 19:38:12 +00:00
"time"
2023-09-06 10:52:01 +00:00
"github.com/syncthing/syncthing/lib/build"
2016-05-04 19:38:12 +00:00
"github.com/syncthing/syncthing/lib/config"
2022-04-09 14:04:56 +00:00
"github.com/syncthing/syncthing/lib/connections/registry"
2016-05-04 19:38:12 +00:00
"github.com/syncthing/syncthing/lib/discover"
"github.com/syncthing/syncthing/lib/events"
"github.com/syncthing/syncthing/lib/nat"
2017-11-06 14:05:29 +00:00
"github.com/syncthing/syncthing/lib/osutil"
2016-05-04 19:38:12 +00:00
"github.com/syncthing/syncthing/lib/protocol"
2023-08-21 17:44:33 +00:00
"github.com/syncthing/syncthing/lib/semaphore"
2023-09-06 10:52:01 +00:00
"github.com/syncthing/syncthing/lib/sliceutil"
2023-08-21 17:44:33 +00:00
"github.com/syncthing/syncthing/lib/stringutil"
2020-12-22 19:17:14 +00:00
"github.com/syncthing/syncthing/lib/svcutil"
2016-05-04 19:38:12 +00:00
"github.com/syncthing/syncthing/lib/sync"
// Registers NAT service providers
_ "github.com/syncthing/syncthing/lib/pmp"
_ "github.com/syncthing/syncthing/lib/upnp"
2020-11-17 12:19:04 +00:00
"github.com/thejerf/suture/v4"
2017-01-02 11:29:20 +00:00
"golang.org/x/time/rate"
2016-05-04 19:38:12 +00:00
)
var (
2019-02-02 11:09:07 +00:00
dialers = make ( map [ string ] dialerFactory )
listeners = make ( map [ string ] listenerFactory )
2016-05-04 19:38:12 +00:00
)
2018-02-09 10:40:57 +00:00
var (
2020-12-09 18:23:50 +00:00
// Dialers and listeners return errUnsupported (or a wrapped variant)
// when they are intentionally out of service due to configuration,
// build, etc. This is not logged loudly.
errUnsupported = errors . New ( "unsupported protocol" )
// These are specific explanations for errUnsupported.
errDisabled = fmt . Errorf ( "%w: disabled by configuration" , errUnsupported )
errDeprecated = fmt . Errorf ( "%w: deprecated" , errUnsupported )
2022-04-07 15:35:33 +00:00
// Various reasons to reject a connection
errNetworkNotAllowed = errors . New ( "network not allowed" )
errDeviceAlreadyConnected = errors . New ( "already connected to this device" )
errDeviceIgnored = errors . New ( "device is ignored" )
errConnLimitReached = errors . New ( "connection limit reached" )
errDevicePaused = errors . New ( "device is paused" )
2023-09-06 10:52:01 +00:00
// A connection is being closed to make space for better ones
errReplacingConnection = errors . New ( "replacing connection" )
2018-02-09 10:40:57 +00:00
)
2016-07-02 20:33:31 +00:00
const (
2021-01-11 14:14:44 +00:00
perDeviceWarningIntv = 15 * time . Minute
tlsHandshakeTimeout = 10 * time . Second
minConnectionLoopSleep = 5 * time . Second
stdConnectionLoopSleep = time . Minute
worstDialerPriority = math . MaxInt32
recentlySeenCutoff = 7 * 24 * time . Hour
shortLivedConnectionThreshold = 5 * time . Second
2021-06-25 09:38:04 +00:00
dialMaxParallel = 64
dialMaxParallelPerDevice = 8
2023-09-06 10:52:01 +00:00
maxNumConnections = 128 // the maximum number of connections we maintain to any given device
2016-07-02 20:33:31 +00:00
)
2016-06-09 12:30:35 +00:00
2016-12-18 21:07:44 +00:00
// From go/src/crypto/tls/cipher_suites.go
var tlsCipherSuiteNames = map [ uint16 ] string {
2019-02-26 10:49:02 +00:00
// TLS 1.2
2016-12-18 21:07:44 +00:00
0x0005 : "TLS_RSA_WITH_RC4_128_SHA" ,
0x000a : "TLS_RSA_WITH_3DES_EDE_CBC_SHA" ,
0x002f : "TLS_RSA_WITH_AES_128_CBC_SHA" ,
0x0035 : "TLS_RSA_WITH_AES_256_CBC_SHA" ,
0x003c : "TLS_RSA_WITH_AES_128_CBC_SHA256" ,
0x009c : "TLS_RSA_WITH_AES_128_GCM_SHA256" ,
0x009d : "TLS_RSA_WITH_AES_256_GCM_SHA384" ,
0xc007 : "TLS_ECDHE_ECDSA_WITH_RC4_128_SHA" ,
0xc009 : "TLS_ECDHE_ECDSA_WITH_AES_128_CBC_SHA" ,
0xc00a : "TLS_ECDHE_ECDSA_WITH_AES_256_CBC_SHA" ,
0xc011 : "TLS_ECDHE_RSA_WITH_RC4_128_SHA" ,
0xc012 : "TLS_ECDHE_RSA_WITH_3DES_EDE_CBC_SHA" ,
0xc013 : "TLS_ECDHE_RSA_WITH_AES_128_CBC_SHA" ,
0xc014 : "TLS_ECDHE_RSA_WITH_AES_256_CBC_SHA" ,
0xc023 : "TLS_ECDHE_ECDSA_WITH_AES_128_CBC_SHA256" ,
0xc027 : "TLS_ECDHE_RSA_WITH_AES_128_CBC_SHA256" ,
0xc02f : "TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256" ,
0xc02b : "TLS_ECDHE_ECDSA_WITH_AES_128_GCM_SHA256" ,
0xc030 : "TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384" ,
0xc02c : "TLS_ECDHE_ECDSA_WITH_AES_256_GCM_SHA384" ,
0xcca8 : "TLS_ECDHE_RSA_WITH_CHACHA20_POLY1305" ,
0xcca9 : "TLS_ECDHE_ECDSA_WITH_CHACHA20_POLY1305" ,
2019-02-26 10:49:02 +00:00
// TLS 1.3
0x1301 : "TLS_AES_128_GCM_SHA256" ,
0x1302 : "TLS_AES_256_GCM_SHA384" ,
0x1303 : "TLS_CHACHA20_POLY1305_SHA256" ,
}
var tlsVersionNames = map [ uint16 ] string {
tls . VersionTLS12 : "TLS1.2" ,
2019-11-10 08:47:17 +00:00
tls . VersionTLS13 : "TLS1.3" ,
2016-12-18 21:07:44 +00:00
}
2016-05-04 19:38:12 +00:00
// Service listens and dials all configured unconnected devices, via supported
// dialers. Successful connections are handed to the model.
2019-02-26 08:09:25 +00:00
type Service interface {
suture . Service
2019-08-13 07:33:33 +00:00
discover . AddressLister
2019-05-16 21:11:46 +00:00
ListenerStatus ( ) map [ string ] ListenerStatusEntry
ConnectionStatus ( ) map [ string ] ConnectionStatusEntry
2019-02-26 08:09:25 +00:00
NATType ( ) string
}
2019-05-16 21:11:46 +00:00
type ListenerStatusEntry struct {
Error * string ` json:"error" `
LANAddresses [ ] string ` json:"lanAddresses" `
WANAddresses [ ] string ` json:"wanAddresses" `
}
type ConnectionStatusEntry struct {
When time . Time ` json:"when" `
Error * string ` json:"error" `
}
2022-04-07 15:35:33 +00:00
type connWithHello struct {
c internalConn
hello protocol . Hello
err error
remoteID protocol . DeviceID
remoteCert * x509 . Certificate
}
2019-02-26 08:09:25 +00:00
type service struct {
2016-05-04 19:38:12 +00:00
* suture . Supervisor
2020-02-25 20:18:31 +00:00
connectionStatusHandler
2023-09-06 10:52:01 +00:00
deviceConnectionTracker
2020-02-25 20:18:31 +00:00
2019-02-26 08:09:25 +00:00
cfg config . Wrapper
2016-05-04 19:38:12 +00:00
myID protocol . DeviceID
model Model
tlsCfg * tls . Config
discoverer discover . Finder
2016-11-30 07:54:20 +00:00
conns chan internalConn
2022-04-07 15:35:33 +00:00
hellos chan * connWithHello
2016-05-04 19:38:12 +00:00
bepProtocolName string
tlsDefaultCommonName string
2017-01-02 11:29:20 +00:00
limiter * limiter
2016-05-04 19:38:12 +00:00
natService * nat . Service
2019-08-15 14:29:37 +00:00
evLogger events . Logger
2022-04-09 14:04:56 +00:00
registry * registry . Registry
2023-03-12 19:06:59 +00:00
keyGen * protocol . KeyGenerator
2023-04-16 12:54:28 +00:00
lanChecker * lanChecker
2016-05-04 19:38:12 +00:00
2021-06-17 11:57:44 +00:00
dialNow chan struct { }
dialNowDevices map [ protocol . DeviceID ] struct { }
dialNowDevicesMut sync . Mutex
listenersMut sync . RWMutex
listeners map [ string ] genericListener
listenerTokens map [ string ] suture . ServiceToken
2016-05-04 19:38:12 +00:00
}
2023-03-12 19:06:59 +00:00
func NewService ( cfg config . Wrapper , myID protocol . DeviceID , mdl Model , tlsCfg * tls . Config , discoverer discover . Finder , bepProtocolName string , tlsDefaultCommonName string , evLogger events . Logger , registry * registry . Registry , keyGen * protocol . KeyGenerator ) Service {
2020-12-22 19:17:14 +00:00
spec := svcutil . SpecWithInfoLogger ( l )
2019-02-26 08:09:25 +00:00
service := & service {
2020-11-17 12:19:04 +00:00
Supervisor : suture . New ( "connections.Service" , spec ) ,
2020-02-25 20:18:31 +00:00
connectionStatusHandler : newConnectionStatusHandler ( ) ,
2016-05-04 19:38:12 +00:00
cfg : cfg ,
myID : myID ,
model : mdl ,
tlsCfg : tlsCfg ,
discoverer : discoverer ,
2016-11-30 07:54:20 +00:00
conns : make ( chan internalConn ) ,
2022-04-11 13:32:22 +00:00
hellos : make ( chan * connWithHello ) ,
2016-05-04 19:38:12 +00:00
bepProtocolName : bepProtocolName ,
tlsDefaultCommonName : tlsDefaultCommonName ,
2020-08-25 06:11:14 +00:00
limiter : newLimiter ( myID , cfg ) ,
2016-05-04 19:38:12 +00:00
natService : nat . NewService ( myID , cfg ) ,
2019-08-15 14:29:37 +00:00
evLogger : evLogger ,
2022-04-09 14:04:56 +00:00
registry : registry ,
2023-03-12 19:06:59 +00:00
keyGen : keyGen ,
2023-04-16 12:54:28 +00:00
lanChecker : & lanChecker { cfg } ,
2016-05-04 19:38:12 +00:00
2021-06-17 11:57:44 +00:00
dialNowDevicesMut : sync . NewMutex ( ) ,
dialNow : make ( chan struct { } , 1 ) ,
dialNowDevices : make ( map [ protocol . DeviceID ] struct { } ) ,
listenersMut : sync . NewRWMutex ( ) ,
listeners : make ( map [ string ] genericListener ) ,
listenerTokens : make ( map [ string ] suture . ServiceToken ) ,
2016-05-04 19:38:12 +00:00
}
cfg . Subscribe ( service )
2017-10-22 12:36:36 +00:00
raw := cfg . RawCopy ( )
// Actually starts the listeners and NAT service
// Need to start this before service.connect so that any dials that
// try punch through already have a listener to cling on.
service . CommitConfiguration ( raw , raw )
2016-05-04 19:38:12 +00:00
// There are several moving parts here; one routine per listening address
// (handled in configuration changing) to handle incoming connections,
// one routine to periodically attempt outgoing connections, one routine to
2017-02-25 08:12:13 +00:00
// the common handling regardless of whether the connection was
2016-05-04 19:38:12 +00:00
// incoming or outgoing.
2020-12-22 19:17:14 +00:00
service . Add ( svcutil . AsService ( service . connect , fmt . Sprintf ( "%s/connect" , service ) ) )
2022-04-07 15:35:33 +00:00
service . Add ( svcutil . AsService ( service . handleConns , fmt . Sprintf ( "%s/handleConns" , service ) ) )
service . Add ( svcutil . AsService ( service . handleHellos , fmt . Sprintf ( "%s/handleHellos" , service ) ) )
2020-06-07 18:29:53 +00:00
service . Add ( service . natService )
2016-05-04 19:38:12 +00:00
2020-12-22 19:17:14 +00:00
svcutil . OnSupervisorDone ( service . Supervisor , func ( ) {
2020-11-17 12:19:04 +00:00
service . cfg . Unsubscribe ( service . limiter )
service . cfg . Unsubscribe ( service )
} )
2016-05-04 19:38:12 +00:00
2020-11-17 12:19:04 +00:00
return service
2019-12-04 06:15:00 +00:00
}
2022-04-07 15:35:33 +00:00
func ( s * service ) handleConns ( ctx context . Context ) error {
2019-07-09 09:40:30 +00:00
for {
2022-05-04 16:16:36 +00:00
var c internalConn
2019-07-09 09:40:30 +00:00
select {
2019-11-21 07:41:15 +00:00
case <- ctx . Done ( ) :
2020-11-17 12:19:04 +00:00
return ctx . Err ( )
2019-07-09 09:40:30 +00:00
case c = <- s . conns :
}
2016-05-04 19:38:12 +00:00
cs := c . ConnectionState ( )
// We should have negotiated the next level protocol "bep/1.0" as part
// of the TLS handshake. Unfortunately this can't be a hard error,
// because there are implementations out there that don't support
// protocol negotiation (iOS for one...).
2022-07-28 15:14:49 +00:00
if cs . NegotiatedProtocol != s . bepProtocolName {
2018-01-12 11:27:55 +00:00
l . Infof ( "Peer at %s did not negotiate bep/1.0" , c )
2016-05-04 19:38:12 +00:00
}
// We should have received exactly one certificate from the other
// side. If we didn't, they don't have a device ID and we drop the
// connection.
certs := cs . PeerCertificates
if cl := len ( certs ) ; cl != 1 {
2018-01-12 11:27:55 +00:00
l . Infof ( "Got peer certificate list of length %d != 1 from peer at %s; protocol error" , cl , c )
2016-05-04 19:38:12 +00:00
c . Close ( )
continue
}
remoteCert := certs [ 0 ]
remoteID := protocol . NewDeviceID ( remoteCert . Raw )
// The device ID should not be that of ourselves. It can happen
// though, especially in the presence of NAT hairpinning, multiple
// clients between the same NAT gateway, and global discovery.
if remoteID == s . myID {
2021-03-17 22:53:20 +00:00
l . Debugf ( "Connected to myself (%s) at %s" , remoteID , c )
2016-05-04 19:38:12 +00:00
c . Close ( )
continue
}
2022-04-07 15:35:33 +00:00
if err := s . connectionCheckEarly ( remoteID , c ) ; err != nil {
2024-09-21 07:19:58 +00:00
if errors . Is ( err , errDeviceAlreadyConnected ) {
l . Debugf ( "Connection from %s at %s (%s) rejected: %v" , remoteID , c . RemoteAddr ( ) , c . Type ( ) , err )
} else {
l . Infof ( "Connection from %s at %s (%s) rejected: %v" , remoteID , c . RemoteAddr ( ) , c . Type ( ) , err )
}
2022-04-07 15:35:33 +00:00
c . Close ( )
continue
}
2019-08-09 11:31:42 +00:00
_ = c . SetDeadline ( time . Now ( ) . Add ( 20 * time . Second ) )
2022-04-07 15:35:33 +00:00
go func ( ) {
2023-09-06 10:52:01 +00:00
// Exchange Hello messages with the peer.
outgoing := s . helloForDevice ( remoteID )
incoming , err := protocol . ExchangeHello ( c , outgoing )
// The timestamps are used to create the connection ID.
c . connectionID = newConnectionID ( outgoing . Timestamp , incoming . Timestamp )
2022-04-07 15:35:33 +00:00
select {
2023-09-06 10:52:01 +00:00
case s . hellos <- & connWithHello { c , incoming , err , remoteID , remoteCert } :
2022-04-07 15:35:33 +00:00
case <- ctx . Done ( ) :
}
} ( )
}
}
2023-09-06 10:52:01 +00:00
func ( s * service ) helloForDevice ( remoteID protocol . DeviceID ) protocol . Hello {
hello := protocol . Hello {
ClientName : "syncthing" ,
ClientVersion : build . Version ,
Timestamp : time . Now ( ) . UnixNano ( ) ,
}
if cfg , ok := s . cfg . Device ( remoteID ) ; ok {
hello . NumConnections = cfg . NumConnections ( )
// Set our name (from the config of our device ID) only if we
// already know about the other side device ID.
if myCfg , ok := s . cfg . Device ( s . myID ) ; ok {
hello . DeviceName = myCfg . Name
}
}
return hello
}
2022-04-07 15:35:33 +00:00
func ( s * service ) connectionCheckEarly ( remoteID protocol . DeviceID , c internalConn ) error {
if s . cfg . IgnoredDevice ( remoteID ) {
return errDeviceIgnored
}
2023-09-06 10:52:01 +00:00
if max := s . cfg . Options ( ) . ConnectionLimitMax ; max > 0 && s . numConnectedDevices ( ) >= max {
2022-04-07 15:35:33 +00:00
// We're not allowed to accept any more connections.
return errConnLimitReached
}
cfg , ok := s . cfg . Device ( remoteID )
if ! ok {
// We do go ahead exchanging hello messages to get information about the device.
return nil
}
if cfg . Paused {
return errDevicePaused
}
if len ( cfg . AllowedNetworks ) > 0 && ! IsAllowedNetwork ( c . RemoteAddr ( ) . String ( ) , cfg . AllowedNetworks ) {
// The connection is not from an allowed network.
return errNetworkNotAllowed
}
2023-09-06 10:52:01 +00:00
currentConns := s . numConnectionsForDevice ( cfg . DeviceID )
desiredConns := s . desiredConnectionsToDevice ( cfg . DeviceID )
worstPrio := s . worstConnectionPriority ( remoteID )
ourUpgradeThreshold := c . priority + s . cfg . Options ( ) . ConnectionPriorityUpgradeThreshold
if currentConns >= desiredConns && ourUpgradeThreshold >= worstPrio {
l . Debugf ( "Not accepting connection to %s at %s: already have %d connections, desire %d" , remoteID , c , currentConns , desiredConns )
return errDeviceAlreadyConnected
2022-04-07 15:35:33 +00:00
}
return nil
}
func ( s * service ) handleHellos ( ctx context . Context ) error {
for {
2023-09-06 10:52:01 +00:00
var c internalConn
var hello protocol . Hello
var err error
var remoteID protocol . DeviceID
var remoteCert * x509 . Certificate
2022-04-07 15:35:33 +00:00
select {
case <- ctx . Done ( ) :
return ctx . Err ( )
case withHello := <- s . hellos :
c = withHello . c
hello = withHello . hello
err = withHello . err
remoteID = withHello . remoteID
remoteCert = withHello . remoteCert
}
2016-05-04 19:38:12 +00:00
if err != nil {
2016-06-09 10:50:14 +00:00
if protocol . IsVersionMismatch ( err ) {
// The error will be a relatively user friendly description
2016-07-04 10:40:29 +00:00
// of what's wrong with the version compatibility. By
// default identify the other side by device ID and IP.
remote := fmt . Sprintf ( "%v (%v)" , remoteID , c . RemoteAddr ( ) )
if hello . DeviceName != "" {
// If the name was set in the hello return, use that to
// give the user more info about which device is the
// affected one. It probably says more than the remote
// IP.
remote = fmt . Sprintf ( "%q (%s %s, %v)" , hello . DeviceName , hello . ClientName , hello . ClientVersion , remoteID )
}
msg := fmt . Sprintf ( "Connecting to %s: %s" , remote , err )
2016-06-09 12:30:35 +00:00
warningFor ( remoteID , msg )
2016-06-09 10:50:14 +00:00
} else {
// It's something else - connection reset or whatever
2018-01-12 11:27:55 +00:00
l . Infof ( "Failed to exchange Hello messages with %s at %s: %s" , remoteID , c , err )
2016-06-09 10:50:14 +00:00
}
2016-05-04 19:38:12 +00:00
c . Close ( )
continue
}
2019-08-09 11:31:42 +00:00
_ = c . SetDeadline ( time . Time { } )
2016-05-04 19:38:12 +00:00
2016-08-05 09:29:49 +00:00
// The Model will return an error for devices that we don't want to
// have a connection with for whatever reason, for example unknown devices.
if err := s . model . OnHello ( remoteID , c . RemoteAddr ( ) , hello ) ; err != nil {
2016-12-02 11:56:14 +00:00
l . Infof ( "Connection from %s at %s (%s) rejected: %v" , remoteID , c . RemoteAddr ( ) , c . Type ( ) , err )
2016-08-05 09:29:49 +00:00
c . Close ( )
continue
}
2016-05-04 19:38:12 +00:00
2016-08-05 09:29:49 +00:00
deviceCfg , ok := s . cfg . Device ( remoteID )
if ! ok {
2018-10-30 09:34:19 +00:00
l . Infof ( "Device %s removed from config during connection attempt at %s" , remoteID , c )
c . Close ( )
continue
2016-08-05 09:29:49 +00:00
}
2016-05-04 19:38:12 +00:00
2016-08-05 09:29:49 +00:00
// Verify the name on the certificate. By default we set it to
// "syncthing" when generating, but the user may have replaced
// the certificate and used another name.
certName := deviceCfg . CertName
if certName == "" {
certName = s . tlsDefaultCommonName
}
2020-07-30 11:36:11 +00:00
if remoteCert . Subject . CommonName == certName {
// All good. We do this check because our old style certificates
// have "syncthing" in the CommonName field and no SANs, which
// is not accepted by VerifyHostname() any more as of Go 1.15.
} else if err := remoteCert . VerifyHostname ( certName ) ; err != nil {
2016-08-05 09:29:49 +00:00
// Incorrect certificate name is something the user most
// likely wants to know about, since it's an advanced
// config. Warn instead of Info.
2018-01-12 11:27:55 +00:00
l . Warnf ( "Bad certificate from %s at %s: %v" , remoteID , c , err )
2016-08-05 09:29:49 +00:00
c . Close ( )
2019-07-09 09:40:30 +00:00
continue
2016-08-05 09:29:49 +00:00
}
2016-05-04 19:38:12 +00:00
2017-01-02 11:29:20 +00:00
// Wrap the connection in rate limiters. The limiter itself will
// keep up with config changes to the rate and whether or not LAN
// connections are limited.
2022-11-28 08:28:33 +00:00
rd , wr := s . limiter . getLimiters ( remoteID , c , c . IsLocal ( ) )
2016-05-04 19:38:12 +00:00
2023-03-12 19:06:59 +00:00
protoConn := protocol . NewConnection ( remoteID , rd , wr , c , s . model , c , deviceCfg . Compression , s . cfg . FolderPasswords ( remoteID ) , s . keyGen )
2023-09-06 10:52:01 +00:00
s . accountAddedConnection ( protoConn , hello , s . cfg . Options ( ) . ConnectionPriorityUpgradeThreshold )
2021-06-17 11:57:44 +00:00
go func ( ) {
<- protoConn . Closed ( )
2023-09-06 10:52:01 +00:00
s . accountRemovedConnection ( protoConn )
2021-06-17 11:57:44 +00:00
s . dialNowDevicesMut . Lock ( )
s . dialNowDevices [ remoteID ] = struct { } { }
s . scheduleDialNow ( )
s . dialNowDevicesMut . Unlock ( )
} ( )
2016-05-04 19:38:12 +00:00
2023-09-06 10:52:01 +00:00
l . Infof ( "Established secure connection to %s at %s" , remoteID . Short ( ) , c )
2016-05-04 19:38:12 +00:00
2020-12-21 10:40:51 +00:00
s . model . AddConnection ( protoConn , hello )
2019-07-09 09:40:30 +00:00
continue
2016-05-04 19:38:12 +00:00
}
}
2023-03-12 19:06:59 +00:00
2020-11-17 12:19:04 +00:00
func ( s * service ) connect ( ctx context . Context ) error {
2020-12-21 15:40:13 +00:00
// Map of when to earliest dial each given device + address again
2021-06-17 11:57:44 +00:00
nextDialAt := make ( nextDialRegistry )
2016-05-09 15:03:12 +00:00
2020-12-21 15:40:13 +00:00
// Used as delay for the first few connection attempts (adjusted up to
// minConnectionLoopSleep), increased exponentially until it reaches
// stdConnectionLoopSleep, at which time the normal sleep mechanism
// kicks in.
2016-05-17 00:05:38 +00:00
initialRampup := time . Second
2016-05-04 19:38:12 +00:00
for {
2016-11-12 09:34:18 +00:00
cfg := s . cfg . RawCopy ( )
2020-12-21 15:40:13 +00:00
bestDialerPriority := s . bestDialerPriority ( cfg )
isInitialRampup := initialRampup < stdConnectionLoopSleep
2016-05-17 00:05:38 +00:00
2020-12-21 15:40:13 +00:00
l . Debugln ( "Connection loop" )
if isInitialRampup {
l . Debugln ( "Connection loop in initial rampup" )
2016-05-17 00:05:38 +00:00
}
2020-12-21 15:40:13 +00:00
// Used for consistency throughout this loop run, as time passes
// while we try connections etc.
2016-05-04 19:38:12 +00:00
now := time . Now ( )
2020-12-21 15:40:13 +00:00
// Attempt to dial all devices that are unconnected or can be connection-upgraded
s . dialDevices ( ctx , now , cfg , bestDialerPriority , nextDialAt , isInitialRampup )
2020-02-13 13:43:00 +00:00
2020-12-21 15:40:13 +00:00
var sleep time . Duration
if isInitialRampup {
// We are in the initial rampup time, so we slowly, statically
// increase the sleep time.
sleep = initialRampup
initialRampup *= 2
} else {
// The sleep time is until the next dial scheduled in nextDialAt,
// clamped by stdConnectionLoopSleep as we don't want to sleep too
// long (config changes might happen).
2021-06-17 11:57:44 +00:00
sleep = nextDialAt . sleepDurationAndCleanup ( now )
2020-12-21 15:40:13 +00:00
}
2016-05-04 19:38:12 +00:00
2020-12-21 15:40:13 +00:00
// ... while making sure not to loop too quickly either.
if sleep < minConnectionLoopSleep {
sleep = minConnectionLoopSleep
}
2016-05-09 15:03:12 +00:00
2020-12-21 15:40:13 +00:00
l . Debugln ( "Next connection loop in" , sleep )
2016-05-09 15:33:25 +00:00
2021-06-17 11:57:44 +00:00
timeout := time . NewTimer ( sleep )
2020-12-21 15:40:13 +00:00
select {
2021-06-17 11:57:44 +00:00
case <- s . dialNow :
// Remove affected devices from nextDialAt to dial immediately,
// regardless of when we last dialed it (there's cool down in the
// registry for too many repeat dials).
s . dialNowDevicesMut . Lock ( )
for device := range s . dialNowDevices {
nextDialAt . redialDevice ( device , now )
}
s . dialNowDevices = make ( map [ protocol . DeviceID ] struct { } )
s . dialNowDevicesMut . Unlock ( )
timeout . Stop ( )
case <- timeout . C :
2020-12-21 15:40:13 +00:00
case <- ctx . Done ( ) :
return ctx . Err ( )
}
}
}
2016-05-04 19:38:12 +00:00
2023-04-16 12:54:28 +00:00
func ( s * service ) bestDialerPriority ( cfg config . Configuration ) int {
2020-12-21 15:40:13 +00:00
bestDialerPriority := worstDialerPriority
for _ , df := range dialers {
if df . Valid ( cfg ) != nil {
continue
}
2023-04-16 12:54:28 +00:00
prio := df . New ( cfg . Options , s . tlsCfg , s . registry , s . lanChecker ) . Priority ( "127.0.0.1" )
if prio < bestDialerPriority {
2020-12-21 15:40:13 +00:00
bestDialerPriority = prio
}
}
return bestDialerPriority
}
2017-11-22 07:05:49 +00:00
2021-06-17 11:57:44 +00:00
func ( s * service ) dialDevices ( ctx context . Context , now time . Time , cfg config . Configuration , bestDialerPriority int , nextDialAt nextDialRegistry , initial bool ) {
2021-01-11 14:14:44 +00:00
// Figure out current connection limits up front to see if there's any
// point in resolving devices and such at all.
allowAdditional := 0 // no limit
connectionLimit := cfg . Options . LowestConnectionLimit ( )
if connectionLimit > 0 {
2023-09-06 10:52:01 +00:00
current := s . numConnectedDevices ( )
2021-01-11 14:14:44 +00:00
allowAdditional = connectionLimit - current
if allowAdditional <= 0 {
l . Debugf ( "Skipping dial because we've reached the connection limit, current %d >= limit %d" , current , connectionLimit )
return
}
}
// Get device statistics for the last seen time of each device. This
// isn't critical, so ignore the potential error.
stats , _ := s . model . DeviceStatistics ( )
queue := make ( dialQueue , 0 , len ( cfg . Devices ) )
2020-12-21 15:40:13 +00:00
for _ , deviceCfg := range cfg . Devices {
// Don't attempt to connect to ourselves...
if deviceCfg . DeviceID == s . myID {
continue
}
2017-11-21 07:25:38 +00:00
2020-12-21 15:40:13 +00:00
// Don't attempt to connect to paused devices...
if deviceCfg . Paused {
continue
}
2017-11-15 09:36:33 +00:00
2020-12-21 15:40:13 +00:00
// See if we are already connected and, if so, what our cutoff is
// for dialer priority.
priorityCutoff := worstDialerPriority
2023-09-06 10:52:01 +00:00
if currentConns := s . numConnectionsForDevice ( deviceCfg . DeviceID ) ; currentConns > 0 {
2023-04-16 12:54:28 +00:00
// Set the priority cutoff to the current connection's priority,
// so that we don't attempt any dialers with worse priority.
2023-09-06 10:52:01 +00:00
priorityCutoff = s . worstConnectionPriority ( deviceCfg . DeviceID )
2023-04-16 12:54:28 +00:00
// Reduce the priority cutoff by the upgrade threshold, so that
// we don't attempt dialers that aren't considered a worthy upgrade.
priorityCutoff -= cfg . Options . ConnectionPriorityUpgradeThreshold
2023-09-06 10:52:01 +00:00
if bestDialerPriority >= priorityCutoff && currentConns >= s . desiredConnectionsToDevice ( deviceCfg . DeviceID ) {
2020-12-21 15:40:13 +00:00
// Our best dialer is not any better than what we already
2023-09-06 10:52:01 +00:00
// have, and we already have the desired number of
// connections to this device,so nothing to do here.
l . Debugf ( "Skipping dial to %s because we already have %d connections and our best dialer is not better than %d" , deviceCfg . DeviceID . Short ( ) , currentConns , priorityCutoff )
2020-12-21 15:40:13 +00:00
continue
}
}
2016-05-04 19:38:12 +00:00
2020-12-21 15:40:13 +00:00
dialTargets := s . resolveDialTargets ( ctx , now , cfg , deviceCfg , nextDialAt , initial , priorityCutoff )
2021-01-11 14:14:44 +00:00
if len ( dialTargets ) > 0 {
queue = append ( queue , dialQueueEntry {
id : deviceCfg . DeviceID ,
lastSeen : stats [ deviceCfg . DeviceID ] . LastSeen ,
shortLived : stats [ deviceCfg . DeviceID ] . LastConnectionDurationS < shortLivedConnectionThreshold . Seconds ( ) ,
targets : dialTargets ,
} )
}
}
// Sort the queue in an order we think will be useful (most recent
2023-04-16 12:54:28 +00:00
// first, deprioritising unstable devices, randomizing those we haven't
2021-01-11 14:14:44 +00:00
// seen in a long while). If we don't do connection limiting the sorting
// doesn't have much effect, but it may result in getting up and running
// quicker if only a subset of configured devices are actually reachable
// (by prioritizing those that were reachable recently).
2021-03-17 22:12:26 +00:00
queue . Sort ( )
2021-01-11 14:14:44 +00:00
// Perform dials according to the queue, stopping when we've reached the
// allowed additional number of connections (if limited).
numConns := 0
2021-06-25 09:38:04 +00:00
var numConnsMut stdsync . Mutex
2023-08-21 17:44:33 +00:00
dialSemaphore := semaphore . New ( dialMaxParallel )
2021-06-25 09:38:04 +00:00
dialWG := new ( stdsync . WaitGroup )
dialCtx , dialCancel := context . WithCancel ( ctx )
defer func ( ) {
dialWG . Wait ( )
dialCancel ( )
} ( )
for i := range queue {
select {
case <- dialCtx . Done ( ) :
return
default :
2020-12-21 15:40:13 +00:00
}
2021-06-25 09:38:04 +00:00
dialWG . Add ( 1 )
go func ( entry dialQueueEntry ) {
defer dialWG . Done ( )
conn , ok := s . dialParallel ( dialCtx , entry . id , entry . targets , dialSemaphore )
if ! ok {
return
}
numConnsMut . Lock ( )
if allowAdditional == 0 || numConns < allowAdditional {
select {
case s . conns <- conn :
numConns ++
if allowAdditional > 0 && numConns >= allowAdditional {
dialCancel ( )
}
case <- dialCtx . Done ( ) :
}
}
numConnsMut . Unlock ( )
} ( queue [ i ] )
2020-12-21 15:40:13 +00:00
}
}
2017-04-01 09:52:31 +00:00
2021-06-17 11:57:44 +00:00
func ( s * service ) resolveDialTargets ( ctx context . Context , now time . Time , cfg config . Configuration , deviceCfg config . DeviceConfiguration , nextDialAt nextDialRegistry , initial bool , priorityCutoff int ) [ ] dialTarget {
2020-12-21 15:40:13 +00:00
deviceID := deviceCfg . DeviceID
2016-05-04 19:38:12 +00:00
2020-12-21 15:40:13 +00:00
addrs := s . resolveDeviceAddrs ( ctx , deviceCfg )
2023-09-06 10:52:01 +00:00
l . Debugln ( "Resolved device" , deviceID . Short ( ) , "addresses:" , addrs )
2017-11-15 09:36:33 +00:00
2020-12-21 15:40:13 +00:00
dialTargets := make ( [ ] dialTarget , 0 , len ( addrs ) )
for _ , addr := range addrs {
2021-06-17 11:57:44 +00:00
// Use both device and address, as you might have two devices connected
// to the same relay
if ! initial && nextDialAt . get ( deviceID , addr ) . After ( now ) {
2023-09-06 10:52:01 +00:00
l . Debugf ( "Not dialing %s via %v as it's not time yet" , deviceID . Short ( ) , addr )
2020-12-21 15:40:13 +00:00
continue
}
2016-05-04 19:38:12 +00:00
2020-12-21 15:40:13 +00:00
// If we fail at any step before actually getting the dialer
// retry in a minute
2021-06-17 11:57:44 +00:00
nextDialAt . set ( deviceID , addr , now . Add ( time . Minute ) )
2016-05-17 00:05:38 +00:00
2020-12-21 15:40:13 +00:00
uri , err := url . Parse ( addr )
if err != nil {
s . setConnectionStatus ( addr , err )
l . Infof ( "Parsing dialer address %s: %v" , addr , err )
continue
}
2017-11-22 07:05:49 +00:00
2020-12-21 15:40:13 +00:00
if len ( deviceCfg . AllowedNetworks ) > 0 {
if ! IsAllowedNetwork ( uri . Host , deviceCfg . AllowedNetworks ) {
s . setConnectionStatus ( addr , errors . New ( "network disallowed" ) )
l . Debugln ( "Network for" , uri , "is disallowed" )
continue
2017-11-15 09:36:33 +00:00
}
2020-12-21 15:40:13 +00:00
}
2016-05-04 19:38:12 +00:00
2020-12-21 15:40:13 +00:00
dialerFactory , err := getDialerFactory ( cfg , uri )
if err != nil {
s . setConnectionStatus ( addr , err )
}
if errors . Is ( err , errUnsupported ) {
l . Debugf ( "Dialer for %v: %v" , uri , err )
continue
} else if err != nil {
l . Infof ( "Dialer for %v: %v" , uri , err )
continue
2016-05-04 19:38:12 +00:00
}
2023-04-16 12:54:28 +00:00
dialer := dialerFactory . New ( s . cfg . Options ( ) , s . tlsCfg , s . registry , s . lanChecker )
priority := dialer . Priority ( uri . Host )
2023-09-06 10:52:01 +00:00
currentConns := s . numConnectionsForDevice ( deviceCfg . DeviceID )
if priority > priorityCutoff {
l . Debugf ( "Not dialing %s at %s using %s as priority is worse than current connection (%d > %d)" , deviceID . Short ( ) , addr , dialerFactory , priority , priorityCutoff )
continue
}
if currentConns > 0 && ! dialer . AllowsMultiConns ( ) {
l . Debugf ( "Not dialing %s at %s using %s as it does not allow multiple connections and we already have a connection" , deviceID . Short ( ) , addr , dialerFactory )
continue
}
if currentConns >= s . desiredConnectionsToDevice ( deviceCfg . DeviceID ) && priority == priorityCutoff {
l . Debugf ( "Not dialing %s at %s using %s as priority is equal and we already have %d/%d connections" , deviceID . Short ( ) , addr , dialerFactory , currentConns , deviceCfg . NumConnections )
2020-12-21 15:40:13 +00:00
continue
}
2016-05-04 19:38:12 +00:00
2021-06-17 11:57:44 +00:00
nextDialAt . set ( deviceID , addr , now . Add ( dialer . RedialFrequency ( ) ) )
2020-12-21 15:40:13 +00:00
dialTargets = append ( dialTargets , dialTarget {
addr : addr ,
dialer : dialer ,
priority : priority ,
deviceID : deviceID ,
uri : uri ,
} )
}
return dialTargets
}
func ( s * service ) resolveDeviceAddrs ( ctx context . Context , cfg config . DeviceConfiguration ) [ ] string {
var addrs [ ] string
for _ , addr := range cfg . Addresses {
if addr == "dynamic" {
if s . discoverer != nil {
if t , err := s . discoverer . Lookup ( ctx , cfg . DeviceID ) ; err == nil {
addrs = append ( addrs , t ... )
}
}
} else {
addrs = append ( addrs , addr )
2016-05-04 19:38:12 +00:00
}
}
2023-08-21 17:44:33 +00:00
return stringutil . UniqueTrimmedStrings ( addrs )
2016-05-04 19:38:12 +00:00
}
2023-04-16 12:54:28 +00:00
type lanChecker struct {
cfg config . Wrapper
}
func ( s * lanChecker ) isLANHost ( host string ) bool {
2017-11-22 08:00:07 +00:00
// Probably we are called with an ip:port combo which we can resolve as
// a TCP address.
if addr , err := net . ResolveTCPAddr ( "tcp" , host ) ; err == nil {
return s . isLAN ( addr )
2017-11-22 07:05:49 +00:00
}
2017-11-22 08:00:07 +00:00
// ... but this function looks general enough that someone might try
// with just an IP as well in the future so lets allow that.
if addr , err := net . ResolveIPAddr ( "ip" , host ) ; err == nil {
return s . isLAN ( addr )
2017-11-22 07:05:49 +00:00
}
2017-11-22 08:00:07 +00:00
return false
2017-11-22 07:05:49 +00:00
}
2023-04-16 12:54:28 +00:00
func ( s * lanChecker ) isLAN ( addr net . Addr ) bool {
2017-11-22 08:00:07 +00:00
var ip net . IP
switch addr := addr . ( type ) {
case * net . IPAddr :
ip = addr . IP
case * net . TCPAddr :
ip = addr . IP
case * net . UDPAddr :
ip = addr . IP
default :
// From the standard library, just Unix sockets.
// If you invent your own, handle it.
2017-01-02 11:29:20 +00:00
return false
2016-05-04 19:38:12 +00:00
}
2017-11-06 14:05:29 +00:00
2017-11-22 08:00:07 +00:00
if ip . IsLoopback ( ) {
2017-11-06 14:05:29 +00:00
return true
}
2023-09-12 12:34:30 +00:00
if ip . IsLinkLocalUnicast ( ) {
return true
}
2017-11-06 14:05:29 +00:00
for _ , lan := range s . cfg . Options ( ) . AlwaysLocalNets {
_ , ipnet , err := net . ParseCIDR ( lan )
if err != nil {
l . Debugln ( "Network" , lan , "is malformed:" , err )
continue
}
2017-11-22 08:00:07 +00:00
if ipnet . Contains ( ip ) {
2017-11-06 14:05:29 +00:00
return true
}
}
2024-09-23 12:32:19 +00:00
lans , err := osutil . GetInterfaceAddrs ( false )
2023-09-12 12:34:30 +00:00
if err != nil {
l . Debugln ( "Failed to retrieve interface IPs:" , err )
priv := ip . IsPrivate ( )
l . Debugf ( "Assuming isLAN=%v for IP %v" , priv , ip )
return priv
}
2017-11-06 14:05:29 +00:00
for _ , lan := range lans {
2017-11-22 08:00:07 +00:00
if lan . Contains ( ip ) {
2017-01-02 11:29:20 +00:00
return true
2016-05-04 19:38:12 +00:00
}
}
2017-11-06 14:05:29 +00:00
return false
2016-05-04 19:38:12 +00:00
}
2019-02-26 08:09:25 +00:00
func ( s * service ) createListener ( factory listenerFactory , uri * url . URL ) bool {
2016-05-09 15:03:12 +00:00
// must be called with listenerMut held
2016-05-04 19:38:12 +00:00
2016-05-17 00:05:38 +00:00
l . Debugln ( "Starting listener" , uri )
2016-05-04 19:38:12 +00:00
2023-04-16 12:54:28 +00:00
listener := factory . New ( uri , s . cfg , s . tlsCfg , s . conns , s . natService , s . registry , s . lanChecker )
2016-05-04 19:38:12 +00:00
listener . OnAddressesChanged ( s . logListenAddressesChangedEvent )
2021-05-10 20:26:51 +00:00
// Retrying a listener many times in rapid succession is unlikely to help,
// thus back off quickly. A listener may soon be functional again, e.g. due
// to a network interface coming back online - retry every minute.
spec := svcutil . SpecWithInfoLogger ( l )
spec . FailureThreshold = 2
spec . FailureBackoff = time . Minute
sup := suture . New ( fmt . Sprintf ( "listenerSupervisor@%v" , listener ) , spec )
sup . Add ( listener )
2016-05-17 00:05:38 +00:00
s . listeners [ uri . String ( ) ] = listener
2021-05-10 20:26:51 +00:00
s . listenerTokens [ uri . String ( ) ] = s . Add ( sup )
2016-05-17 00:05:38 +00:00
return true
2016-05-04 19:38:12 +00:00
}
2020-05-11 13:02:22 +00:00
func ( s * service ) logListenAddressesChangedEvent ( l ListenerAddresses ) {
2019-08-15 14:29:37 +00:00
s . evLogger . Log ( events . ListenAddressesChanged , map [ string ] interface { } {
2020-05-11 13:02:22 +00:00
"address" : l . URI ,
"lan" : l . LANAddresses ,
"wan" : l . WANAddresses ,
2016-05-04 19:38:12 +00:00
} )
}
2019-02-26 08:09:25 +00:00
func ( s * service ) CommitConfiguration ( from , to config . Configuration ) bool {
2016-05-04 19:38:12 +00:00
newDevices := make ( map [ protocol . DeviceID ] bool , len ( to . Devices ) )
for _ , dev := range to . Devices {
newDevices [ dev . DeviceID ] = true
2024-05-04 20:31:37 +00:00
registerDeviceMetrics ( dev . DeviceID . String ( ) )
2016-05-04 19:38:12 +00:00
}
for _ , dev := range from . Devices {
if ! newDevices [ dev . DeviceID ] {
2016-08-07 16:21:59 +00:00
warningLimitersMut . Lock ( )
delete ( warningLimiters , dev . DeviceID )
warningLimitersMut . Unlock ( )
2024-05-04 20:31:37 +00:00
metricDeviceActiveConnections . DeleteLabelValues ( dev . DeviceID . String ( ) )
2016-05-04 19:38:12 +00:00
}
}
2021-04-26 20:13:59 +00:00
s . checkAndSignalConnectLoopOnUpdatedDevices ( from , to )
2016-05-09 15:03:12 +00:00
s . listenersMut . Lock ( )
2016-05-04 19:38:12 +00:00
seen := make ( map [ string ] struct { } )
2019-11-26 16:07:25 +00:00
for _ , addr := range to . Options . ListenAddresses ( ) {
2017-11-17 09:12:35 +00:00
if addr == "" {
// We can get an empty address if there is an empty listener
// element in the config, indicating no listeners should be
// used. This is not an error.
continue
}
2020-06-02 09:19:51 +00:00
uri , err := url . Parse ( addr )
if err != nil {
l . Warnf ( "Skipping malformed listener URL %q: %v" , addr , err )
2016-05-17 00:05:38 +00:00
continue
2016-05-04 19:38:12 +00:00
}
2016-05-17 00:05:38 +00:00
2020-06-02 09:19:51 +00:00
// Make sure we always have the canonical representation of the URL.
// This is for consistency as we use it as a map key, but also to
// avoid misunderstandings. We do not just use the canonicalized
// version, because an URL that looks very similar to a human might
// mean something entirely different to the computer (e.g.,
// tcp:/127.0.0.1:22000 in fact being equivalent to tcp://:22000).
if canonical := uri . String ( ) ; canonical != addr {
l . Warnf ( "Skipping malformed listener URL %q (not canonical)" , addr )
continue
}
if _ , ok := s . listeners [ addr ] ; ok {
seen [ addr ] = struct { } { }
2016-05-17 00:05:38 +00:00
continue
}
2018-02-09 10:40:57 +00:00
factory , err := getListenerFactory ( to , uri )
2020-12-09 18:23:50 +00:00
if errors . Is ( err , errUnsupported ) {
l . Debugf ( "Listener for %v: %v" , uri , err )
2018-02-09 10:40:57 +00:00
continue
2020-12-09 18:23:50 +00:00
} else if err != nil {
2018-02-09 10:40:57 +00:00
l . Infof ( "Listener for %v: %v" , uri , err )
2016-05-17 00:05:38 +00:00
continue
}
s . createListener ( factory , uri )
2016-05-04 19:38:12 +00:00
seen [ addr ] = struct { } { }
}
2016-05-17 00:05:38 +00:00
for addr , listener := range s . listeners {
2018-02-09 10:40:57 +00:00
if _ , ok := seen [ addr ] ; ! ok || listener . Factory ( ) . Valid ( to ) != nil {
2016-05-04 19:38:12 +00:00
l . Debugln ( "Stopping listener" , addr )
2021-05-11 12:35:13 +00:00
s . Remove ( s . listenerTokens [ addr ] )
2016-05-04 19:38:12 +00:00
delete ( s . listenerTokens , addr )
delete ( s . listeners , addr )
}
}
2016-05-09 15:03:12 +00:00
s . listenersMut . Unlock ( )
2016-05-04 19:38:12 +00:00
2016-08-07 16:21:59 +00:00
return true
2016-05-04 19:38:12 +00:00
}
2021-04-26 20:13:59 +00:00
func ( s * service ) checkAndSignalConnectLoopOnUpdatedDevices ( from , to config . Configuration ) {
2021-06-17 11:57:44 +00:00
oldDevices := from . DeviceMap ( )
2021-07-03 16:26:55 +00:00
dial := false
s . dialNowDevicesMut . Lock ( )
2021-04-26 20:13:59 +00:00
for _ , dev := range to . Devices {
2021-07-03 16:26:55 +00:00
if dev . Paused {
continue
2021-04-26 20:13:59 +00:00
}
2021-07-03 16:26:55 +00:00
if oldDev , ok := oldDevices [ dev . DeviceID ] ; ! ok || oldDev . Paused {
s . dialNowDevices [ dev . DeviceID ] = struct { } { }
dial = true
2023-08-21 17:44:33 +00:00
} else if ! slices . Equal ( oldDev . Addresses , dev . Addresses ) {
2021-07-03 16:26:55 +00:00
dial = true
}
}
if dial {
s . scheduleDialNow ( )
2021-04-26 20:13:59 +00:00
}
2021-07-03 16:26:55 +00:00
s . dialNowDevicesMut . Unlock ( )
2021-04-26 20:13:59 +00:00
}
2021-06-17 11:57:44 +00:00
func ( s * service ) scheduleDialNow ( ) {
select {
case s . dialNow <- struct { } { } :
default :
// channel is blocked - a config update is already pending for the connection loop.
}
}
2019-02-26 08:09:25 +00:00
func ( s * service ) AllAddresses ( ) [ ] string {
2016-05-09 15:03:12 +00:00
s . listenersMut . RLock ( )
2016-05-04 19:38:12 +00:00
var addrs [ ] string
for _ , listener := range s . listeners {
for _ , lanAddr := range listener . LANAddresses ( ) {
addrs = append ( addrs , lanAddr . String ( ) )
}
for _ , wanAddr := range listener . WANAddresses ( ) {
addrs = append ( addrs , wanAddr . String ( ) )
}
}
2016-05-09 15:03:12 +00:00
s . listenersMut . RUnlock ( )
2023-08-21 17:44:33 +00:00
return stringutil . UniqueTrimmedStrings ( addrs )
2016-05-04 19:38:12 +00:00
}
2019-02-26 08:09:25 +00:00
func ( s * service ) ExternalAddresses ( ) [ ] string {
2020-08-25 09:48:14 +00:00
if s . cfg . Options ( ) . AnnounceLANAddresses {
return s . AllAddresses ( )
}
2016-05-09 15:03:12 +00:00
s . listenersMut . RLock ( )
2016-05-04 19:38:12 +00:00
var addrs [ ] string
for _ , listener := range s . listeners {
for _ , wanAddr := range listener . WANAddresses ( ) {
addrs = append ( addrs , wanAddr . String ( ) )
}
}
2016-05-09 15:03:12 +00:00
s . listenersMut . RUnlock ( )
2023-08-21 17:44:33 +00:00
return stringutil . UniqueTrimmedStrings ( addrs )
2016-05-04 19:38:12 +00:00
}
2019-05-16 21:11:46 +00:00
func ( s * service ) ListenerStatus ( ) map [ string ] ListenerStatusEntry {
result := make ( map [ string ] ListenerStatusEntry )
2016-05-09 15:03:12 +00:00
s . listenersMut . RLock ( )
2016-05-04 19:38:12 +00:00
for addr , listener := range s . listeners {
2019-05-16 21:11:46 +00:00
var status ListenerStatusEntry
2016-05-04 19:38:12 +00:00
2019-05-16 21:11:46 +00:00
if err := listener . Error ( ) ; err != nil {
errStr := err . Error ( )
status . Error = & errStr
2016-05-04 19:38:12 +00:00
}
2019-05-16 21:11:46 +00:00
status . LANAddresses = urlsToStrings ( listener . LANAddresses ( ) )
status . WANAddresses = urlsToStrings ( listener . WANAddresses ( ) )
2016-05-04 19:38:12 +00:00
result [ addr ] = status
}
2016-05-09 15:03:12 +00:00
s . listenersMut . RUnlock ( )
2016-05-04 19:38:12 +00:00
return result
}
2020-02-25 20:18:31 +00:00
type connectionStatusHandler struct {
connectionStatusMut sync . RWMutex
connectionStatus map [ string ] ConnectionStatusEntry // address -> latest error/status
}
func newConnectionStatusHandler ( ) connectionStatusHandler {
return connectionStatusHandler {
connectionStatusMut : sync . NewRWMutex ( ) ,
connectionStatus : make ( map [ string ] ConnectionStatusEntry ) ,
}
}
func ( s * connectionStatusHandler ) ConnectionStatus ( ) map [ string ] ConnectionStatusEntry {
2019-05-16 21:11:46 +00:00
result := make ( map [ string ] ConnectionStatusEntry )
s . connectionStatusMut . RLock ( )
for k , v := range s . connectionStatus {
result [ k ] = v
}
s . connectionStatusMut . RUnlock ( )
return result
}
2020-02-25 20:18:31 +00:00
func ( s * connectionStatusHandler ) setConnectionStatus ( address string , err error ) {
2022-03-26 10:05:57 +00:00
if errors . Is ( err , context . Canceled ) {
2019-11-26 07:39:51 +00:00
return
}
2019-05-16 21:11:46 +00:00
status := ConnectionStatusEntry { When : time . Now ( ) . UTC ( ) . Truncate ( time . Second ) }
if err != nil {
errStr := err . Error ( )
status . Error = & errStr
}
s . connectionStatusMut . Lock ( )
s . connectionStatus [ address ] = status
s . connectionStatusMut . Unlock ( )
}
2019-02-26 08:09:25 +00:00
func ( s * service ) NATType ( ) string {
2017-10-12 06:16:46 +00:00
s . listenersMut . RLock ( )
defer s . listenersMut . RUnlock ( )
for _ , listener := range s . listeners {
natType := listener . NATType ( )
if natType != "unknown" {
return natType
}
}
return "unknown"
}
2018-02-09 10:40:57 +00:00
func getDialerFactory ( cfg config . Configuration , uri * url . URL ) ( dialerFactory , error ) {
2016-05-17 00:05:38 +00:00
dialerFactory , ok := dialers [ uri . Scheme ]
if ! ok {
return nil , fmt . Errorf ( "unknown address scheme %q" , uri . Scheme )
}
2018-02-09 10:40:57 +00:00
if err := dialerFactory . Valid ( cfg ) ; err != nil {
return nil , err
2016-05-17 00:05:38 +00:00
}
return dialerFactory , nil
}
2018-02-09 10:40:57 +00:00
func getListenerFactory ( cfg config . Configuration , uri * url . URL ) ( listenerFactory , error ) {
2016-05-17 00:05:38 +00:00
listenerFactory , ok := listeners [ uri . Scheme ]
if ! ok {
return nil , fmt . Errorf ( "unknown address scheme %q" , uri . Scheme )
}
2018-02-09 10:40:57 +00:00
if err := listenerFactory . Valid ( cfg ) ; err != nil {
return nil , err
2016-05-17 00:05:38 +00:00
}
return listenerFactory , nil
}
2016-05-04 19:38:12 +00:00
func urlsToStrings ( urls [ ] * url . URL ) [ ] string {
strings := make ( [ ] string , len ( urls ) )
for i , url := range urls {
strings [ i ] = url . String ( )
}
return strings
}
2016-06-09 12:30:35 +00:00
2023-03-12 19:06:59 +00:00
var (
warningLimiters = make ( map [ protocol . DeviceID ] * rate . Limiter )
warningLimitersMut = sync . NewMutex ( )
)
2016-06-09 12:30:35 +00:00
func warningFor ( dev protocol . DeviceID , msg string ) {
warningLimitersMut . Lock ( )
defer warningLimitersMut . Unlock ( )
lim , ok := warningLimiters [ dev ]
if ! ok {
2017-01-02 11:29:20 +00:00
lim = rate . NewLimiter ( rate . Every ( perDeviceWarningIntv ) , 1 )
2016-06-09 12:30:35 +00:00
warningLimiters [ dev ] = lim
}
2017-01-02 11:29:20 +00:00
if lim . Allow ( ) {
2016-06-09 12:30:35 +00:00
l . Warnln ( msg )
}
}
2016-07-02 20:33:31 +00:00
func tlsTimedHandshake ( tc * tls . Conn ) error {
2019-02-02 11:16:27 +00:00
tc . SetDeadline ( time . Now ( ) . Add ( tlsHandshakeTimeout ) )
defer tc . SetDeadline ( time . Time { } )
2016-07-02 20:33:31 +00:00
return tc . Handshake ( )
}
2017-04-01 09:52:31 +00:00
// IsAllowedNetwork returns true if the given host (IP or resolvable
// hostname) is in the set of allowed networks (CIDR format only).
func IsAllowedNetwork ( host string , allowed [ ] string ) bool {
if hostNoPort , _ , err := net . SplitHostPort ( host ) ; err == nil {
host = hostNoPort
}
addr , err := net . ResolveIPAddr ( "ip" , host )
if err != nil {
return false
}
for _ , n := range allowed {
2017-04-13 17:43:29 +00:00
result := true
if strings . HasPrefix ( n , "!" ) {
result = false
n = n [ 1 : ]
}
2017-04-01 09:52:31 +00:00
_ , cidr , err := net . ParseCIDR ( n )
if err != nil {
continue
}
if cidr . Contains ( addr . IP ) {
2017-04-13 17:43:29 +00:00
return result
2017-04-01 09:52:31 +00:00
}
}
return false
}
2017-11-15 09:36:33 +00:00
2023-08-21 17:44:33 +00:00
func ( s * service ) dialParallel ( ctx context . Context , deviceID protocol . DeviceID , dialTargets [ ] dialTarget , parentSema * semaphore . Semaphore ) ( internalConn , bool ) {
2017-11-15 09:36:33 +00:00
// Group targets into buckets by priority
dialTargetBuckets := make ( map [ int ] [ ] dialTarget , len ( dialTargets ) )
for _ , tgt := range dialTargets {
dialTargetBuckets [ tgt . priority ] = append ( dialTargetBuckets [ tgt . priority ] , tgt )
}
// Get all available priorities
priorities := make ( [ ] int , 0 , len ( dialTargetBuckets ) )
for prio := range dialTargetBuckets {
priorities = append ( priorities , prio )
}
// Sort the priorities so that we dial lowest first (which means highest...)
sort . Ints ( priorities )
2023-08-21 17:44:33 +00:00
sema := semaphore . MultiSemaphore { semaphore . New ( dialMaxParallelPerDevice ) , parentSema }
2017-11-15 09:36:33 +00:00
for _ , prio := range priorities {
tgts := dialTargetBuckets [ prio ]
res := make ( chan internalConn , len ( tgts ) )
2017-12-07 07:08:24 +00:00
wg := stdsync . WaitGroup { }
2017-11-15 09:36:33 +00:00
for _ , tgt := range tgts {
2021-06-25 09:38:04 +00:00
sema . Take ( 1 )
2017-11-15 09:36:33 +00:00
wg . Add ( 1 )
2017-11-19 17:38:13 +00:00
go func ( tgt dialTarget ) {
2021-06-25 09:38:04 +00:00
defer func ( ) {
wg . Done ( )
sema . Give ( 1 )
} ( )
2019-11-26 07:39:51 +00:00
conn , err := tgt . Dial ( ctx )
2017-11-15 09:36:33 +00:00
if err == nil {
2019-08-09 11:31:42 +00:00
// Closes the connection on error
err = s . validateIdentity ( conn , deviceID )
}
s . setConnectionStatus ( tgt . addr , err )
if err != nil {
l . Debugln ( "dialing" , deviceID , tgt . uri , "error:" , err )
} else {
l . Debugln ( "dialing" , deviceID , tgt . uri , "success:" , conn )
2017-11-15 09:36:33 +00:00
res <- conn
}
2017-11-19 17:38:13 +00:00
} ( tgt )
2017-11-15 09:36:33 +00:00
}
// Spawn a routine which will unblock main routine in case we fail
// to connect to anyone.
go func ( ) {
wg . Wait ( )
close ( res )
} ( )
// Wait for the first connection, or for channel closure.
2018-01-12 11:27:55 +00:00
if conn , ok := <- res ; ok {
// Got a connection, means more might come back, hence spawn a
// routine that will do the discarding.
2017-11-15 09:36:33 +00:00
l . Debugln ( "connected to" , deviceID , prio , "using" , conn , conn . priority )
go func ( deviceID protocol . DeviceID , prio int ) {
wg . Wait ( )
l . Debugln ( "discarding" , len ( res ) , "connections while connecting to" , deviceID , prio )
for conn := range res {
conn . Close ( )
}
} ( deviceID , prio )
2017-11-21 14:58:18 +00:00
return conn , ok
2017-11-15 09:36:33 +00:00
}
2018-01-12 11:27:55 +00:00
// Failed to connect, report that fact.
l . Debugln ( "failed to connect to" , deviceID , prio )
2017-11-15 09:36:33 +00:00
}
return internalConn { } , false
}
2019-08-09 11:31:42 +00:00
func ( s * service ) validateIdentity ( c internalConn , expectedID protocol . DeviceID ) error {
cs := c . ConnectionState ( )
// We should have received exactly one certificate from the other
// side. If we didn't, they don't have a device ID and we drop the
// connection.
certs := cs . PeerCertificates
if cl := len ( certs ) ; cl != 1 {
l . Infof ( "Got peer certificate list of length %d != 1 from peer at %s; protocol error" , cl , c )
c . Close ( )
return fmt . Errorf ( "expected 1 certificate, got %d" , cl )
}
remoteCert := certs [ 0 ]
remoteID := protocol . NewDeviceID ( remoteCert . Raw )
// The device ID should not be that of ourselves. It can happen
// though, especially in the presence of NAT hairpinning, multiple
// clients between the same NAT gateway, and global discovery.
if remoteID == s . myID {
2021-03-17 22:53:20 +00:00
l . Debugf ( "Connected to myself (%s) at %s" , remoteID , c )
2019-08-09 11:31:42 +00:00
c . Close ( )
2020-03-03 21:40:00 +00:00
return errors . New ( "connected to self" )
2019-08-09 11:31:42 +00:00
}
// We should see the expected device ID
if ! remoteID . Equals ( expectedID ) {
c . Close ( )
return fmt . Errorf ( "unexpected device id, expected %s got %s" , expectedID , remoteID )
}
return nil
}
2021-06-17 11:57:44 +00:00
type nextDialRegistry map [ protocol . DeviceID ] nextDialDevice
type nextDialDevice struct {
nextDial map [ string ] time . Time
coolDownIntervalStart time . Time
attempts int
}
func ( r nextDialRegistry ) get ( device protocol . DeviceID , addr string ) time . Time {
return r [ device ] . nextDial [ addr ]
}
const (
2023-05-09 06:54:02 +00:00
dialCoolDownInterval = 2 * time . Minute
dialCoolDownDelay = 5 * time . Minute
dialCoolDownMaxAttempts = 3
2021-06-17 11:57:44 +00:00
)
// redialDevice marks the device for immediate redial, unless the remote keeps
// dropping established connections. Thus we keep track of when the first forced
// re-dial happened, and how many attempts happen in the dialCoolDownInterval
// after that. If it's more than dialCoolDownMaxAttempts, don't force-redial
// that device for dialCoolDownDelay (regular dials still happen).
func ( r nextDialRegistry ) redialDevice ( device protocol . DeviceID , now time . Time ) {
dev , ok := r [ device ]
if ! ok {
r [ device ] = nextDialDevice {
2021-06-17 17:13:57 +00:00
nextDial : make ( map [ string ] time . Time ) ,
2021-06-17 11:57:44 +00:00
coolDownIntervalStart : now ,
attempts : 1 ,
}
return
}
if dev . attempts == 0 || now . Before ( dev . coolDownIntervalStart . Add ( dialCoolDownInterval ) ) {
2023-05-09 06:54:02 +00:00
if dev . attempts >= dialCoolDownMaxAttempts {
2021-06-17 11:57:44 +00:00
// Device has been force redialed too often - let it cool down.
return
}
if dev . attempts == 0 {
dev . coolDownIntervalStart = now
}
dev . attempts ++
dev . nextDial = make ( map [ string ] time . Time )
return
}
2023-05-09 06:54:02 +00:00
if dev . attempts >= dialCoolDownMaxAttempts && now . Before ( dev . coolDownIntervalStart . Add ( dialCoolDownDelay ) ) {
2021-06-17 11:57:44 +00:00
return // Still cooling down
}
delete ( r , device )
}
func ( r nextDialRegistry ) set ( device protocol . DeviceID , addr string , next time . Time ) {
if _ , ok := r [ device ] ; ! ok {
r [ device ] = nextDialDevice { nextDial : make ( map [ string ] time . Time ) }
}
r [ device ] . nextDial [ addr ] = next
}
func ( r nextDialRegistry ) sleepDurationAndCleanup ( now time . Time ) time . Duration {
sleep := stdConnectionLoopSleep
for id , dev := range r {
for address , next := range dev . nextDial {
if next . Before ( now ) {
// Expired entry, address was not seen in last pass(es)
delete ( dev . nextDial , address )
continue
}
if cur := next . Sub ( now ) ; cur < sleep {
sleep = cur
}
}
if dev . attempts > 0 {
interval := dialCoolDownInterval
2023-05-09 06:54:02 +00:00
if dev . attempts >= dialCoolDownMaxAttempts {
2021-06-17 11:57:44 +00:00
interval = dialCoolDownDelay
}
if now . After ( dev . coolDownIntervalStart . Add ( interval ) ) {
dev . attempts = 0
}
}
if len ( dev . nextDial ) == 0 && dev . attempts == 0 {
delete ( r , id )
}
}
return sleep
}
2023-09-06 10:52:01 +00:00
func ( s * service ) desiredConnectionsToDevice ( deviceID protocol . DeviceID ) int {
cfg , ok := s . cfg . Device ( deviceID )
if ! ok {
// We want no connections to an unknown device.
return 0
}
otherSide := s . wantConnectionsForDevice ( deviceID )
thisSide := cfg . NumConnections ( )
switch {
case otherSide <= 0 :
// The other side doesn't support multiple connections, or we
// haven't yet connected to them so we don't know what they support
// or not. Use a single connection until we know better.
return 1
case otherSide == 1 :
// The other side supports multiple connections, but only wants
// one. We should honour that.
return 1
case thisSide == 1 :
// We want only one connection, so we should honour that.
return 1
// Finally, we allow negotiation and use the higher of the two values,
// while keeping at or below the max allowed value.
default :
return min ( max ( thisSide , otherSide ) , maxNumConnections )
}
}
// The deviceConnectionTracker keeps track of how many devices we are
// connected to and how many connections we have to each device. It also
// tracks how many connections they are willing to use.
type deviceConnectionTracker struct {
connectionsMut stdsync . Mutex
connections map [ protocol . DeviceID ] [ ] protocol . Connection // current connections
wantConnections map [ protocol . DeviceID ] int // number of connections they want
}
func ( c * deviceConnectionTracker ) accountAddedConnection ( conn protocol . Connection , h protocol . Hello , upgradeThreshold int ) {
c . connectionsMut . Lock ( )
defer c . connectionsMut . Unlock ( )
// Lazily initialize the maps
if c . connections == nil {
c . connections = make ( map [ protocol . DeviceID ] [ ] protocol . Connection )
c . wantConnections = make ( map [ protocol . DeviceID ] int )
}
// Add the connection to the list of current connections and remember
// how many total connections they want
d := conn . DeviceID ( )
c . connections [ d ] = append ( c . connections [ d ] , conn )
c . wantConnections [ d ] = int ( h . NumConnections )
l . Debugf ( "Added connection for %s (now %d), they want %d connections" , d . Short ( ) , len ( c . connections [ d ] ) , h . NumConnections )
2024-05-04 20:31:37 +00:00
// Update active connections metric
metricDeviceActiveConnections . WithLabelValues ( d . String ( ) ) . Inc ( )
2023-09-06 10:52:01 +00:00
// Close any connections we no longer want to retain.
c . closeWorsePriorityConnectionsLocked ( d , conn . Priority ( ) - upgradeThreshold )
}
func ( c * deviceConnectionTracker ) accountRemovedConnection ( conn protocol . Connection ) {
c . connectionsMut . Lock ( )
defer c . connectionsMut . Unlock ( )
d := conn . DeviceID ( )
cid := conn . ConnectionID ( )
// Remove the connection from the list of current connections
for i , conn := range c . connections [ d ] {
if conn . ConnectionID ( ) == cid {
c . connections [ d ] = sliceutil . RemoveAndZero ( c . connections [ d ] , i )
break
}
}
// Clean up if required
if len ( c . connections [ d ] ) == 0 {
delete ( c . connections , d )
delete ( c . wantConnections , d )
}
2024-05-04 20:31:37 +00:00
// Update active connections metric
metricDeviceActiveConnections . WithLabelValues ( d . String ( ) ) . Dec ( )
2023-09-06 10:52:01 +00:00
l . Debugf ( "Removed connection for %s (now %d)" , d . Short ( ) , c . connections [ d ] )
}
func ( c * deviceConnectionTracker ) numConnectionsForDevice ( d protocol . DeviceID ) int {
c . connectionsMut . Lock ( )
defer c . connectionsMut . Unlock ( )
return len ( c . connections [ d ] )
}
func ( c * deviceConnectionTracker ) wantConnectionsForDevice ( d protocol . DeviceID ) int {
c . connectionsMut . Lock ( )
defer c . connectionsMut . Unlock ( )
return c . wantConnections [ d ]
}
func ( c * deviceConnectionTracker ) numConnectedDevices ( ) int {
c . connectionsMut . Lock ( )
defer c . connectionsMut . Unlock ( )
return len ( c . connections )
}
func ( c * deviceConnectionTracker ) worstConnectionPriority ( d protocol . DeviceID ) int {
c . connectionsMut . Lock ( )
defer c . connectionsMut . Unlock ( )
if len ( c . connections [ d ] ) == 0 {
return math . MaxInt // worst possible priority
}
worstPriority := c . connections [ d ] [ 0 ] . Priority ( )
for _ , conn := range c . connections [ d ] [ 1 : ] {
if p := conn . Priority ( ) ; p > worstPriority {
worstPriority = p
}
}
return worstPriority
}
// closeWorsePriorityConnectionsLocked closes all connections to the given
// device that are worse than the cutoff priority. Must be called with the
// lock held.
func ( c * deviceConnectionTracker ) closeWorsePriorityConnectionsLocked ( d protocol . DeviceID , cutoff int ) {
for _ , conn := range c . connections [ d ] {
if p := conn . Priority ( ) ; p > cutoff {
l . Debugf ( "Closing connection %s to %s with priority %d (cutoff %d)" , conn , d . Short ( ) , p , cutoff )
go conn . Close ( errReplacingConnection )
}
}
}
// newConnectionID generates a connection ID. The connection ID is designed
// to be unique for each connection and chronologically sortable. It is
// based on the sum of two timestamps: when we think the connection was
// started, and when the other side thinks the connection was started. We
// then add some random data for good measure. This way, even if the other
// side does some funny business with the timestamp, we will get no worse
// than random connection IDs.
func newConnectionID ( t0 , t1 int64 ) string {
var buf [ 16 ] byte // 8 bytes timestamp, 8 bytes random
binary . BigEndian . PutUint64 ( buf [ : ] , uint64 ( t0 + t1 ) )
_ , _ = io . ReadFull ( rand . Reader , buf [ 8 : ] )
enc := base32 . HexEncoding . WithPadding ( base32 . NoPadding )
// We encode the two parts separately and concatenate the results. The
// reason for this is that the timestamp (64 bits) doesn't precisely
// align to the base32 encoding (5 bits per character), so we'd get a
// character in the middle that is a mix of bits from the timestamp and
// from the random. We want the timestamp part deterministic.
return enc . EncodeToString ( buf [ : 8 ] ) + enc . EncodeToString ( buf [ 8 : ] )
}