service.go 46 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455145614571458145914601461146214631464146514661467146814691470147114721473147414751476147714781479148014811482
  1. // Copyright (C) 2015 The Syncthing Authors.
  2. //
  3. // This Source Code Form is subject to the terms of the Mozilla Public
  4. // License, v. 2.0. If a copy of the MPL was not distributed with this file,
  5. // You can obtain one at https://mozilla.org/MPL/2.0/.
  6. //go:generate -command counterfeiter go run github.com/maxbrunsfeld/counterfeiter/v6
  7. //go:generate counterfeiter -o mocks/service.go --fake-name Service . Service
  8. package connections
  9. import (
  10. "context"
  11. "crypto/rand"
  12. "crypto/tls"
  13. "crypto/x509"
  14. "encoding/base32"
  15. "encoding/binary"
  16. "errors"
  17. "fmt"
  18. "io"
  19. "math"
  20. "net"
  21. "net/url"
  22. "slices"
  23. "sort"
  24. "strings"
  25. stdsync "sync"
  26. "time"
  27. "github.com/thejerf/suture/v4"
  28. "golang.org/x/time/rate"
  29. "github.com/syncthing/syncthing/lib/build"
  30. "github.com/syncthing/syncthing/lib/config"
  31. "github.com/syncthing/syncthing/lib/connections/registry"
  32. "github.com/syncthing/syncthing/lib/discover"
  33. "github.com/syncthing/syncthing/lib/events"
  34. "github.com/syncthing/syncthing/lib/nat"
  35. "github.com/syncthing/syncthing/lib/osutil"
  36. "github.com/syncthing/syncthing/lib/protocol"
  37. "github.com/syncthing/syncthing/lib/semaphore"
  38. "github.com/syncthing/syncthing/lib/sliceutil"
  39. "github.com/syncthing/syncthing/lib/stringutil"
  40. "github.com/syncthing/syncthing/lib/svcutil"
  41. "github.com/syncthing/syncthing/lib/sync"
  42. // Registers NAT service providers
  43. _ "github.com/syncthing/syncthing/lib/pmp"
  44. _ "github.com/syncthing/syncthing/lib/upnp"
  45. )
  46. var (
  47. dialers = make(map[string]dialerFactory)
  48. listeners = make(map[string]listenerFactory)
  49. )
  50. var (
  51. // Dialers and listeners return errUnsupported (or a wrapped variant)
  52. // when they are intentionally out of service due to configuration,
  53. // build, etc. This is not logged loudly.
  54. errUnsupported = errors.New("unsupported protocol")
  55. // These are specific explanations for errUnsupported.
  56. errDisabled = fmt.Errorf("%w: disabled by configuration", errUnsupported)
  57. errDeprecated = fmt.Errorf("%w: deprecated", errUnsupported)
  58. // Various reasons to reject a connection
  59. errNetworkNotAllowed = errors.New("network not allowed")
  60. errDeviceAlreadyConnected = errors.New("already connected to this device")
  61. errDeviceIgnored = errors.New("device is ignored")
  62. errConnLimitReached = errors.New("connection limit reached")
  63. errDevicePaused = errors.New("device is paused")
  64. // A connection is being closed to make space for better ones
  65. errReplacingConnection = errors.New("replacing connection")
  66. )
  67. const (
  68. perDeviceWarningIntv = 15 * time.Minute
  69. tlsHandshakeTimeout = 10 * time.Second
  70. minConnectionLoopSleep = 5 * time.Second
  71. stdConnectionLoopSleep = time.Minute
  72. worstDialerPriority = math.MaxInt32
  73. recentlySeenCutoff = 7 * 24 * time.Hour
  74. shortLivedConnectionThreshold = 5 * time.Second
  75. dialMaxParallel = 64
  76. dialMaxParallelPerDevice = 8
  77. maxNumConnections = 128 // the maximum number of connections we maintain to any given device
  78. )
  79. // From go/src/crypto/tls/cipher_suites.go
  80. var tlsCipherSuiteNames = map[uint16]string{
  81. // TLS 1.2
  82. 0x0005: "TLS_RSA_WITH_RC4_128_SHA",
  83. 0x000a: "TLS_RSA_WITH_3DES_EDE_CBC_SHA",
  84. 0x002f: "TLS_RSA_WITH_AES_128_CBC_SHA",
  85. 0x0035: "TLS_RSA_WITH_AES_256_CBC_SHA",
  86. 0x003c: "TLS_RSA_WITH_AES_128_CBC_SHA256",
  87. 0x009c: "TLS_RSA_WITH_AES_128_GCM_SHA256",
  88. 0x009d: "TLS_RSA_WITH_AES_256_GCM_SHA384",
  89. 0xc007: "TLS_ECDHE_ECDSA_WITH_RC4_128_SHA",
  90. 0xc009: "TLS_ECDHE_ECDSA_WITH_AES_128_CBC_SHA",
  91. 0xc00a: "TLS_ECDHE_ECDSA_WITH_AES_256_CBC_SHA",
  92. 0xc011: "TLS_ECDHE_RSA_WITH_RC4_128_SHA",
  93. 0xc012: "TLS_ECDHE_RSA_WITH_3DES_EDE_CBC_SHA",
  94. 0xc013: "TLS_ECDHE_RSA_WITH_AES_128_CBC_SHA",
  95. 0xc014: "TLS_ECDHE_RSA_WITH_AES_256_CBC_SHA",
  96. 0xc023: "TLS_ECDHE_ECDSA_WITH_AES_128_CBC_SHA256",
  97. 0xc027: "TLS_ECDHE_RSA_WITH_AES_128_CBC_SHA256",
  98. 0xc02f: "TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256",
  99. 0xc02b: "TLS_ECDHE_ECDSA_WITH_AES_128_GCM_SHA256",
  100. 0xc030: "TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384",
  101. 0xc02c: "TLS_ECDHE_ECDSA_WITH_AES_256_GCM_SHA384",
  102. 0xcca8: "TLS_ECDHE_RSA_WITH_CHACHA20_POLY1305",
  103. 0xcca9: "TLS_ECDHE_ECDSA_WITH_CHACHA20_POLY1305",
  104. // TLS 1.3
  105. 0x1301: "TLS_AES_128_GCM_SHA256",
  106. 0x1302: "TLS_AES_256_GCM_SHA384",
  107. 0x1303: "TLS_CHACHA20_POLY1305_SHA256",
  108. }
  109. var tlsVersionNames = map[uint16]string{
  110. tls.VersionTLS12: "TLS1.2",
  111. tls.VersionTLS13: "TLS1.3",
  112. }
  113. // Service listens and dials all configured unconnected devices, via supported
  114. // dialers. Successful connections are handed to the model.
  115. type Service interface {
  116. suture.Service
  117. discover.AddressLister
  118. ListenerStatus() map[string]ListenerStatusEntry
  119. ConnectionStatus() map[string]ConnectionStatusEntry
  120. NATType() string
  121. }
  122. type ListenerStatusEntry struct {
  123. Error *string `json:"error"`
  124. LANAddresses []string `json:"lanAddresses"`
  125. WANAddresses []string `json:"wanAddresses"`
  126. }
  127. type ConnectionStatusEntry struct {
  128. When time.Time `json:"when"`
  129. Error *string `json:"error"`
  130. }
  131. type connWithHello struct {
  132. c internalConn
  133. hello protocol.Hello
  134. err error
  135. remoteID protocol.DeviceID
  136. remoteCert *x509.Certificate
  137. }
  138. type service struct {
  139. *suture.Supervisor
  140. connectionStatusHandler
  141. deviceConnectionTracker
  142. cfg config.Wrapper
  143. myID protocol.DeviceID
  144. model Model
  145. tlsCfg *tls.Config
  146. discoverer discover.Finder
  147. conns chan internalConn
  148. hellos chan *connWithHello
  149. bepProtocolName string
  150. tlsDefaultCommonName string
  151. limiter *limiter
  152. natService *nat.Service
  153. evLogger events.Logger
  154. registry *registry.Registry
  155. keyGen *protocol.KeyGenerator
  156. lanChecker *lanChecker
  157. dialNow chan struct{}
  158. dialNowDevices map[protocol.DeviceID]struct{}
  159. dialNowDevicesMut sync.Mutex
  160. listenersMut sync.RWMutex
  161. listeners map[string]genericListener
  162. listenerTokens map[string]suture.ServiceToken
  163. }
  164. 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 {
  165. spec := svcutil.SpecWithInfoLogger(l)
  166. service := &service{
  167. Supervisor: suture.New("connections.Service", spec),
  168. connectionStatusHandler: newConnectionStatusHandler(),
  169. cfg: cfg,
  170. myID: myID,
  171. model: mdl,
  172. tlsCfg: tlsCfg,
  173. discoverer: discoverer,
  174. conns: make(chan internalConn),
  175. hellos: make(chan *connWithHello),
  176. bepProtocolName: bepProtocolName,
  177. tlsDefaultCommonName: tlsDefaultCommonName,
  178. limiter: newLimiter(myID, cfg),
  179. natService: nat.NewService(myID, cfg),
  180. evLogger: evLogger,
  181. registry: registry,
  182. keyGen: keyGen,
  183. lanChecker: &lanChecker{cfg},
  184. dialNowDevicesMut: sync.NewMutex(),
  185. dialNow: make(chan struct{}, 1),
  186. dialNowDevices: make(map[protocol.DeviceID]struct{}),
  187. listenersMut: sync.NewRWMutex(),
  188. listeners: make(map[string]genericListener),
  189. listenerTokens: make(map[string]suture.ServiceToken),
  190. }
  191. cfg.Subscribe(service)
  192. raw := cfg.RawCopy()
  193. // Actually starts the listeners and NAT service
  194. // Need to start this before service.connect so that any dials that
  195. // try punch through already have a listener to cling on.
  196. service.CommitConfiguration(raw, raw)
  197. // There are several moving parts here; one routine per listening address
  198. // (handled in configuration changing) to handle incoming connections,
  199. // one routine to periodically attempt outgoing connections, one routine to
  200. // the common handling regardless of whether the connection was
  201. // incoming or outgoing.
  202. service.Add(svcutil.AsService(service.connect, fmt.Sprintf("%s/connect", service)))
  203. service.Add(svcutil.AsService(service.handleConns, fmt.Sprintf("%s/handleConns", service)))
  204. service.Add(svcutil.AsService(service.handleHellos, fmt.Sprintf("%s/handleHellos", service)))
  205. service.Add(service.natService)
  206. svcutil.OnSupervisorDone(service.Supervisor, func() {
  207. service.cfg.Unsubscribe(service.limiter)
  208. service.cfg.Unsubscribe(service)
  209. })
  210. return service
  211. }
  212. func (s *service) handleConns(ctx context.Context) error {
  213. for {
  214. var c internalConn
  215. select {
  216. case <-ctx.Done():
  217. return ctx.Err()
  218. case c = <-s.conns:
  219. }
  220. cs := c.ConnectionState()
  221. // We should have negotiated the next level protocol "bep/1.0" as part
  222. // of the TLS handshake. Unfortunately this can't be a hard error,
  223. // because there are implementations out there that don't support
  224. // protocol negotiation (iOS for one...).
  225. if cs.NegotiatedProtocol != s.bepProtocolName {
  226. l.Infof("Peer at %s did not negotiate bep/1.0", c)
  227. }
  228. // We should have received exactly one certificate from the other
  229. // side. If we didn't, they don't have a device ID and we drop the
  230. // connection.
  231. certs := cs.PeerCertificates
  232. if cl := len(certs); cl != 1 {
  233. l.Infof("Got peer certificate list of length %d != 1 from peer at %s; protocol error", cl, c)
  234. c.Close()
  235. continue
  236. }
  237. remoteCert := certs[0]
  238. remoteID := protocol.NewDeviceID(remoteCert.Raw)
  239. // The device ID should not be that of ourselves. It can happen
  240. // though, especially in the presence of NAT hairpinning, multiple
  241. // clients between the same NAT gateway, and global discovery.
  242. if remoteID == s.myID {
  243. l.Debugf("Connected to myself (%s) at %s", remoteID, c)
  244. c.Close()
  245. continue
  246. }
  247. if err := s.connectionCheckEarly(remoteID, c); err != nil {
  248. if errors.Is(err, errDeviceAlreadyConnected) {
  249. l.Debugf("Connection from %s at %s (%s) rejected: %v", remoteID, c.RemoteAddr(), c.Type(), err)
  250. } else {
  251. l.Infof("Connection from %s at %s (%s) rejected: %v", remoteID, c.RemoteAddr(), c.Type(), err)
  252. }
  253. c.Close()
  254. continue
  255. }
  256. _ = c.SetDeadline(time.Now().Add(20 * time.Second))
  257. go func() {
  258. // Exchange Hello messages with the peer.
  259. outgoing := s.helloForDevice(remoteID)
  260. incoming, err := protocol.ExchangeHello(c, outgoing)
  261. // The timestamps are used to create the connection ID.
  262. c.connectionID = newConnectionID(outgoing.Timestamp, incoming.Timestamp)
  263. select {
  264. case s.hellos <- &connWithHello{c, incoming, err, remoteID, remoteCert}:
  265. case <-ctx.Done():
  266. }
  267. }()
  268. }
  269. }
  270. func (s *service) helloForDevice(remoteID protocol.DeviceID) protocol.Hello {
  271. hello := protocol.Hello{
  272. ClientName: "syncthing",
  273. ClientVersion: build.Version,
  274. Timestamp: time.Now().UnixNano(),
  275. }
  276. if cfg, ok := s.cfg.Device(remoteID); ok {
  277. hello.NumConnections = cfg.NumConnections()
  278. // Set our name (from the config of our device ID) only if we
  279. // already know about the other side device ID.
  280. if myCfg, ok := s.cfg.Device(s.myID); ok {
  281. hello.DeviceName = myCfg.Name
  282. }
  283. }
  284. return hello
  285. }
  286. func (s *service) connectionCheckEarly(remoteID protocol.DeviceID, c internalConn) error {
  287. if s.cfg.IgnoredDevice(remoteID) {
  288. return errDeviceIgnored
  289. }
  290. if max := s.cfg.Options().ConnectionLimitMax; max > 0 && s.numConnectedDevices() >= max {
  291. // We're not allowed to accept any more connections.
  292. return errConnLimitReached
  293. }
  294. cfg, ok := s.cfg.Device(remoteID)
  295. if !ok {
  296. // We do go ahead exchanging hello messages to get information about the device.
  297. return nil
  298. }
  299. if cfg.Paused {
  300. return errDevicePaused
  301. }
  302. if len(cfg.AllowedNetworks) > 0 && !IsAllowedNetwork(c.RemoteAddr().String(), cfg.AllowedNetworks) {
  303. // The connection is not from an allowed network.
  304. return errNetworkNotAllowed
  305. }
  306. currentConns := s.numConnectionsForDevice(cfg.DeviceID)
  307. desiredConns := s.desiredConnectionsToDevice(cfg.DeviceID)
  308. worstPrio := s.worstConnectionPriority(remoteID)
  309. ourUpgradeThreshold := c.priority + s.cfg.Options().ConnectionPriorityUpgradeThreshold
  310. if currentConns >= desiredConns && ourUpgradeThreshold >= worstPrio {
  311. l.Debugf("Not accepting connection to %s at %s: already have %d connections, desire %d", remoteID, c, currentConns, desiredConns)
  312. return errDeviceAlreadyConnected
  313. }
  314. return nil
  315. }
  316. func (s *service) handleHellos(ctx context.Context) error {
  317. for {
  318. var c internalConn
  319. var hello protocol.Hello
  320. var err error
  321. var remoteID protocol.DeviceID
  322. var remoteCert *x509.Certificate
  323. select {
  324. case <-ctx.Done():
  325. return ctx.Err()
  326. case withHello := <-s.hellos:
  327. c = withHello.c
  328. hello = withHello.hello
  329. err = withHello.err
  330. remoteID = withHello.remoteID
  331. remoteCert = withHello.remoteCert
  332. }
  333. if err != nil {
  334. if protocol.IsVersionMismatch(err) {
  335. // The error will be a relatively user friendly description
  336. // of what's wrong with the version compatibility. By
  337. // default identify the other side by device ID and IP.
  338. remote := fmt.Sprintf("%v (%v)", remoteID, c.RemoteAddr())
  339. if hello.DeviceName != "" {
  340. // If the name was set in the hello return, use that to
  341. // give the user more info about which device is the
  342. // affected one. It probably says more than the remote
  343. // IP.
  344. remote = fmt.Sprintf("%q (%s %s, %v)", hello.DeviceName, hello.ClientName, hello.ClientVersion, remoteID)
  345. }
  346. msg := fmt.Sprintf("Connecting to %s: %s", remote, err)
  347. warningFor(remoteID, msg)
  348. } else {
  349. // It's something else - connection reset or whatever
  350. l.Infof("Failed to exchange Hello messages with %s at %s: %s", remoteID, c, err)
  351. }
  352. c.Close()
  353. continue
  354. }
  355. _ = c.SetDeadline(time.Time{})
  356. // The Model will return an error for devices that we don't want to
  357. // have a connection with for whatever reason, for example unknown devices.
  358. if err := s.model.OnHello(remoteID, c.RemoteAddr(), hello); err != nil {
  359. l.Infof("Connection from %s at %s (%s) rejected: %v", remoteID, c.RemoteAddr(), c.Type(), err)
  360. c.Close()
  361. continue
  362. }
  363. deviceCfg, ok := s.cfg.Device(remoteID)
  364. if !ok {
  365. l.Infof("Device %s removed from config during connection attempt at %s", remoteID, c)
  366. c.Close()
  367. continue
  368. }
  369. // Verify the name on the certificate. By default we set it to
  370. // "syncthing" when generating, but the user may have replaced
  371. // the certificate and used another name.
  372. certName := deviceCfg.CertName
  373. if certName == "" {
  374. certName = s.tlsDefaultCommonName
  375. }
  376. if remoteCert.Subject.CommonName == certName {
  377. // All good. We do this check because our old style certificates
  378. // have "syncthing" in the CommonName field and no SANs, which
  379. // is not accepted by VerifyHostname() any more as of Go 1.15.
  380. } else if err := remoteCert.VerifyHostname(certName); err != nil {
  381. // Incorrect certificate name is something the user most
  382. // likely wants to know about, since it's an advanced
  383. // config. Warn instead of Info.
  384. l.Warnf("Bad certificate from %s at %s: %v", remoteID, c, err)
  385. c.Close()
  386. continue
  387. }
  388. // Wrap the connection in rate limiters. The limiter itself will
  389. // keep up with config changes to the rate and whether or not LAN
  390. // connections are limited.
  391. rd, wr := s.limiter.getLimiters(remoteID, c, c.IsLocal())
  392. protoConn := protocol.NewConnection(remoteID, rd, wr, c, s.model, c, deviceCfg.Compression.ToProtocol(), s.cfg.FolderPasswords(remoteID), s.keyGen)
  393. s.accountAddedConnection(protoConn, hello, s.cfg.Options().ConnectionPriorityUpgradeThreshold)
  394. go func() {
  395. <-protoConn.Closed()
  396. s.accountRemovedConnection(protoConn)
  397. s.dialNowDevicesMut.Lock()
  398. s.dialNowDevices[remoteID] = struct{}{}
  399. s.scheduleDialNow()
  400. s.dialNowDevicesMut.Unlock()
  401. }()
  402. l.Infof("Established secure connection to %s at %s", remoteID.Short(), c)
  403. s.model.AddConnection(protoConn, hello)
  404. continue
  405. }
  406. }
  407. func (s *service) connect(ctx context.Context) error {
  408. // Map of when to earliest dial each given device + address again
  409. nextDialAt := make(nextDialRegistry)
  410. // Used as delay for the first few connection attempts (adjusted up to
  411. // minConnectionLoopSleep), increased exponentially until it reaches
  412. // stdConnectionLoopSleep, at which time the normal sleep mechanism
  413. // kicks in.
  414. initialRampup := time.Second
  415. for {
  416. cfg := s.cfg.RawCopy()
  417. bestDialerPriority := s.bestDialerPriority(cfg)
  418. isInitialRampup := initialRampup < stdConnectionLoopSleep
  419. l.Debugln("Connection loop")
  420. if isInitialRampup {
  421. l.Debugln("Connection loop in initial rampup")
  422. }
  423. // Used for consistency throughout this loop run, as time passes
  424. // while we try connections etc.
  425. now := time.Now()
  426. // Attempt to dial all devices that are unconnected or can be connection-upgraded
  427. s.dialDevices(ctx, now, cfg, bestDialerPriority, nextDialAt, isInitialRampup)
  428. var sleep time.Duration
  429. if isInitialRampup {
  430. // We are in the initial rampup time, so we slowly, statically
  431. // increase the sleep time.
  432. sleep = initialRampup
  433. initialRampup *= 2
  434. } else {
  435. // The sleep time is until the next dial scheduled in nextDialAt,
  436. // clamped by stdConnectionLoopSleep as we don't want to sleep too
  437. // long (config changes might happen).
  438. sleep = nextDialAt.sleepDurationAndCleanup(now)
  439. }
  440. // ... while making sure not to loop too quickly either.
  441. if sleep < minConnectionLoopSleep {
  442. sleep = minConnectionLoopSleep
  443. }
  444. l.Debugln("Next connection loop in", sleep)
  445. timeout := time.NewTimer(sleep)
  446. select {
  447. case <-s.dialNow:
  448. // Remove affected devices from nextDialAt to dial immediately,
  449. // regardless of when we last dialed it (there's cool down in the
  450. // registry for too many repeat dials).
  451. s.dialNowDevicesMut.Lock()
  452. for device := range s.dialNowDevices {
  453. nextDialAt.redialDevice(device, now)
  454. }
  455. s.dialNowDevices = make(map[protocol.DeviceID]struct{})
  456. s.dialNowDevicesMut.Unlock()
  457. timeout.Stop()
  458. case <-timeout.C:
  459. case <-ctx.Done():
  460. return ctx.Err()
  461. }
  462. }
  463. }
  464. func (s *service) bestDialerPriority(cfg config.Configuration) int {
  465. bestDialerPriority := worstDialerPriority
  466. for _, df := range dialers {
  467. if df.Valid(cfg) != nil {
  468. continue
  469. }
  470. prio := df.New(cfg.Options, s.tlsCfg, s.registry, s.lanChecker).Priority("127.0.0.1")
  471. if prio < bestDialerPriority {
  472. bestDialerPriority = prio
  473. }
  474. }
  475. return bestDialerPriority
  476. }
  477. func (s *service) dialDevices(ctx context.Context, now time.Time, cfg config.Configuration, bestDialerPriority int, nextDialAt nextDialRegistry, initial bool) {
  478. // Figure out current connection limits up front to see if there's any
  479. // point in resolving devices and such at all.
  480. allowAdditional := 0 // no limit
  481. connectionLimit := cfg.Options.LowestConnectionLimit()
  482. if connectionLimit > 0 {
  483. current := s.numConnectedDevices()
  484. allowAdditional = connectionLimit - current
  485. if allowAdditional <= 0 {
  486. l.Debugf("Skipping dial because we've reached the connection limit, current %d >= limit %d", current, connectionLimit)
  487. return
  488. }
  489. }
  490. // Get device statistics for the last seen time of each device. This
  491. // isn't critical, so ignore the potential error.
  492. stats, _ := s.model.DeviceStatistics()
  493. queue := make(dialQueue, 0, len(cfg.Devices))
  494. for _, deviceCfg := range cfg.Devices {
  495. // Don't attempt to connect to ourselves...
  496. if deviceCfg.DeviceID == s.myID {
  497. continue
  498. }
  499. // Don't attempt to connect to paused devices...
  500. if deviceCfg.Paused {
  501. continue
  502. }
  503. // See if we are already connected and, if so, what our cutoff is
  504. // for dialer priority.
  505. priorityCutoff := worstDialerPriority
  506. if currentConns := s.numConnectionsForDevice(deviceCfg.DeviceID); currentConns > 0 {
  507. // Set the priority cutoff to the current connection's priority,
  508. // so that we don't attempt any dialers with worse priority.
  509. priorityCutoff = s.worstConnectionPriority(deviceCfg.DeviceID)
  510. // Reduce the priority cutoff by the upgrade threshold, so that
  511. // we don't attempt dialers that aren't considered a worthy upgrade.
  512. priorityCutoff -= cfg.Options.ConnectionPriorityUpgradeThreshold
  513. if bestDialerPriority >= priorityCutoff && currentConns >= s.desiredConnectionsToDevice(deviceCfg.DeviceID) {
  514. // Our best dialer is not any better than what we already
  515. // have, and we already have the desired number of
  516. // connections to this device,so nothing to do here.
  517. 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)
  518. continue
  519. }
  520. }
  521. dialTargets := s.resolveDialTargets(ctx, now, cfg, deviceCfg, nextDialAt, initial, priorityCutoff)
  522. if len(dialTargets) > 0 {
  523. queue = append(queue, dialQueueEntry{
  524. id: deviceCfg.DeviceID,
  525. lastSeen: stats[deviceCfg.DeviceID].LastSeen,
  526. shortLived: stats[deviceCfg.DeviceID].LastConnectionDurationS < shortLivedConnectionThreshold.Seconds(),
  527. targets: dialTargets,
  528. })
  529. }
  530. }
  531. // Sort the queue in an order we think will be useful (most recent
  532. // first, deprioritising unstable devices, randomizing those we haven't
  533. // seen in a long while). If we don't do connection limiting the sorting
  534. // doesn't have much effect, but it may result in getting up and running
  535. // quicker if only a subset of configured devices are actually reachable
  536. // (by prioritizing those that were reachable recently).
  537. queue.Sort()
  538. // Perform dials according to the queue, stopping when we've reached the
  539. // allowed additional number of connections (if limited).
  540. numConns := 0
  541. var numConnsMut stdsync.Mutex
  542. dialSemaphore := semaphore.New(dialMaxParallel)
  543. dialWG := new(stdsync.WaitGroup)
  544. dialCtx, dialCancel := context.WithCancel(ctx)
  545. defer func() {
  546. dialWG.Wait()
  547. dialCancel()
  548. }()
  549. for i := range queue {
  550. select {
  551. case <-dialCtx.Done():
  552. return
  553. default:
  554. }
  555. dialWG.Add(1)
  556. go func(entry dialQueueEntry) {
  557. defer dialWG.Done()
  558. conn, ok := s.dialParallel(dialCtx, entry.id, entry.targets, dialSemaphore)
  559. if !ok {
  560. return
  561. }
  562. numConnsMut.Lock()
  563. if allowAdditional == 0 || numConns < allowAdditional {
  564. select {
  565. case s.conns <- conn:
  566. numConns++
  567. if allowAdditional > 0 && numConns >= allowAdditional {
  568. dialCancel()
  569. }
  570. case <-dialCtx.Done():
  571. }
  572. }
  573. numConnsMut.Unlock()
  574. }(queue[i])
  575. }
  576. }
  577. func (s *service) resolveDialTargets(ctx context.Context, now time.Time, cfg config.Configuration, deviceCfg config.DeviceConfiguration, nextDialAt nextDialRegistry, initial bool, priorityCutoff int) []dialTarget {
  578. deviceID := deviceCfg.DeviceID
  579. addrs := s.resolveDeviceAddrs(ctx, deviceCfg)
  580. l.Debugln("Resolved device", deviceID.Short(), "addresses:", addrs)
  581. dialTargets := make([]dialTarget, 0, len(addrs))
  582. for _, addr := range addrs {
  583. // Use both device and address, as you might have two devices connected
  584. // to the same relay
  585. if !initial && nextDialAt.get(deviceID, addr).After(now) {
  586. l.Debugf("Not dialing %s via %v as it's not time yet", deviceID.Short(), addr)
  587. continue
  588. }
  589. // If we fail at any step before actually getting the dialer
  590. // retry in a minute
  591. nextDialAt.set(deviceID, addr, now.Add(time.Minute))
  592. uri, err := url.Parse(addr)
  593. if err != nil {
  594. s.setConnectionStatus(addr, err)
  595. l.Infof("Parsing dialer address %s: %v", addr, err)
  596. continue
  597. }
  598. if len(deviceCfg.AllowedNetworks) > 0 {
  599. if !IsAllowedNetwork(uri.Host, deviceCfg.AllowedNetworks) {
  600. s.setConnectionStatus(addr, errors.New("network disallowed"))
  601. l.Debugln("Network for", uri, "is disallowed")
  602. continue
  603. }
  604. }
  605. dialerFactory, err := getDialerFactory(cfg, uri)
  606. if err != nil {
  607. s.setConnectionStatus(addr, err)
  608. }
  609. if errors.Is(err, errUnsupported) {
  610. l.Debugf("Dialer for %v: %v", uri, err)
  611. continue
  612. } else if err != nil {
  613. l.Infof("Dialer for %v: %v", uri, err)
  614. continue
  615. }
  616. dialer := dialerFactory.New(s.cfg.Options(), s.tlsCfg, s.registry, s.lanChecker)
  617. priority := dialer.Priority(uri.Host)
  618. currentConns := s.numConnectionsForDevice(deviceCfg.DeviceID)
  619. if priority > priorityCutoff {
  620. l.Debugf("Not dialing %s at %s using %s as priority is worse than current connection (%d > %d)", deviceID.Short(), addr, dialerFactory, priority, priorityCutoff)
  621. continue
  622. }
  623. if currentConns > 0 && !dialer.AllowsMultiConns() {
  624. 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)
  625. continue
  626. }
  627. if currentConns >= s.desiredConnectionsToDevice(deviceCfg.DeviceID) && priority == priorityCutoff {
  628. 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)
  629. continue
  630. }
  631. nextDialAt.set(deviceID, addr, now.Add(dialer.RedialFrequency()))
  632. dialTargets = append(dialTargets, dialTarget{
  633. addr: addr,
  634. dialer: dialer,
  635. priority: priority,
  636. deviceID: deviceID,
  637. uri: uri,
  638. })
  639. }
  640. return dialTargets
  641. }
  642. func (s *service) resolveDeviceAddrs(ctx context.Context, cfg config.DeviceConfiguration) []string {
  643. var addrs []string
  644. for _, addr := range cfg.Addresses {
  645. if addr == "dynamic" {
  646. if s.discoverer != nil {
  647. if t, err := s.discoverer.Lookup(ctx, cfg.DeviceID); err == nil {
  648. addrs = append(addrs, t...)
  649. }
  650. }
  651. } else {
  652. addrs = append(addrs, addr)
  653. }
  654. }
  655. return stringutil.UniqueTrimmedStrings(addrs)
  656. }
  657. type lanChecker struct {
  658. cfg config.Wrapper
  659. }
  660. func (s *lanChecker) isLANHost(host string) bool {
  661. // Probably we are called with an ip:port combo which we can resolve as
  662. // a TCP address.
  663. if addr, err := net.ResolveTCPAddr("tcp", host); err == nil {
  664. return s.isLAN(addr)
  665. }
  666. // ... but this function looks general enough that someone might try
  667. // with just an IP as well in the future so lets allow that.
  668. if addr, err := net.ResolveIPAddr("ip", host); err == nil {
  669. return s.isLAN(addr)
  670. }
  671. return false
  672. }
  673. func (s *lanChecker) isLAN(addr net.Addr) bool {
  674. var ip net.IP
  675. switch addr := addr.(type) {
  676. case *net.IPAddr:
  677. ip = addr.IP
  678. case *net.TCPAddr:
  679. ip = addr.IP
  680. case *net.UDPAddr:
  681. ip = addr.IP
  682. default:
  683. // From the standard library, just Unix sockets.
  684. // If you invent your own, handle it.
  685. return false
  686. }
  687. if ip.IsLoopback() {
  688. return true
  689. }
  690. if ip.IsLinkLocalUnicast() {
  691. return true
  692. }
  693. for _, lan := range s.cfg.Options().AlwaysLocalNets {
  694. _, ipnet, err := net.ParseCIDR(lan)
  695. if err != nil {
  696. l.Debugln("Network", lan, "is malformed:", err)
  697. continue
  698. }
  699. if ipnet.Contains(ip) {
  700. return true
  701. }
  702. }
  703. lans, err := osutil.GetInterfaceAddrs(false)
  704. if err != nil {
  705. l.Debugln("Failed to retrieve interface IPs:", err)
  706. priv := ip.IsPrivate()
  707. l.Debugf("Assuming isLAN=%v for IP %v", priv, ip)
  708. return priv
  709. }
  710. for _, lan := range lans {
  711. if lan.Contains(ip) {
  712. return true
  713. }
  714. }
  715. return false
  716. }
  717. func (s *service) createListener(factory listenerFactory, uri *url.URL) bool {
  718. // must be called with listenerMut held
  719. l.Debugln("Starting listener", uri)
  720. listener := factory.New(uri, s.cfg, s.tlsCfg, s.conns, s.natService, s.registry, s.lanChecker)
  721. listener.OnAddressesChanged(s.logListenAddressesChangedEvent)
  722. // Retrying a listener many times in rapid succession is unlikely to help,
  723. // thus back off quickly. A listener may soon be functional again, e.g. due
  724. // to a network interface coming back online - retry every minute.
  725. spec := svcutil.SpecWithInfoLogger(l)
  726. spec.FailureThreshold = 2
  727. spec.FailureBackoff = time.Minute
  728. sup := suture.New(fmt.Sprintf("listenerSupervisor@%v", listener), spec)
  729. sup.Add(listener)
  730. s.listeners[uri.String()] = listener
  731. s.listenerTokens[uri.String()] = s.Add(sup)
  732. return true
  733. }
  734. func (s *service) logListenAddressesChangedEvent(l ListenerAddresses) {
  735. s.evLogger.Log(events.ListenAddressesChanged, map[string]interface{}{
  736. "address": l.URI,
  737. "lan": l.LANAddresses,
  738. "wan": l.WANAddresses,
  739. })
  740. }
  741. func (s *service) CommitConfiguration(from, to config.Configuration) bool {
  742. newDevices := make(map[protocol.DeviceID]bool, len(to.Devices))
  743. for _, dev := range to.Devices {
  744. newDevices[dev.DeviceID] = true
  745. registerDeviceMetrics(dev.DeviceID.String())
  746. }
  747. for _, dev := range from.Devices {
  748. if !newDevices[dev.DeviceID] {
  749. warningLimitersMut.Lock()
  750. delete(warningLimiters, dev.DeviceID)
  751. warningLimitersMut.Unlock()
  752. metricDeviceActiveConnections.DeleteLabelValues(dev.DeviceID.String())
  753. }
  754. }
  755. s.checkAndSignalConnectLoopOnUpdatedDevices(from, to)
  756. s.listenersMut.Lock()
  757. seen := make(map[string]struct{})
  758. for _, addr := range to.Options.ListenAddresses() {
  759. if addr == "" {
  760. // We can get an empty address if there is an empty listener
  761. // element in the config, indicating no listeners should be
  762. // used. This is not an error.
  763. continue
  764. }
  765. uri, err := url.Parse(addr)
  766. if err != nil {
  767. l.Warnf("Skipping malformed listener URL %q: %v", addr, err)
  768. continue
  769. }
  770. // Make sure we always have the canonical representation of the URL.
  771. // This is for consistency as we use it as a map key, but also to
  772. // avoid misunderstandings. We do not just use the canonicalized
  773. // version, because an URL that looks very similar to a human might
  774. // mean something entirely different to the computer (e.g.,
  775. // tcp:/127.0.0.1:22000 in fact being equivalent to tcp://:22000).
  776. if canonical := uri.String(); canonical != addr {
  777. l.Warnf("Skipping malformed listener URL %q (not canonical)", addr)
  778. continue
  779. }
  780. if _, ok := s.listeners[addr]; ok {
  781. seen[addr] = struct{}{}
  782. continue
  783. }
  784. factory, err := getListenerFactory(to, uri)
  785. if errors.Is(err, errUnsupported) {
  786. l.Debugf("Listener for %v: %v", uri, err)
  787. continue
  788. } else if err != nil {
  789. l.Infof("Listener for %v: %v", uri, err)
  790. continue
  791. }
  792. s.createListener(factory, uri)
  793. seen[addr] = struct{}{}
  794. }
  795. for addr, listener := range s.listeners {
  796. if _, ok := seen[addr]; !ok || listener.Factory().Valid(to) != nil {
  797. l.Debugln("Stopping listener", addr)
  798. s.Remove(s.listenerTokens[addr])
  799. delete(s.listenerTokens, addr)
  800. delete(s.listeners, addr)
  801. }
  802. }
  803. s.listenersMut.Unlock()
  804. return true
  805. }
  806. func (s *service) checkAndSignalConnectLoopOnUpdatedDevices(from, to config.Configuration) {
  807. oldDevices := from.DeviceMap()
  808. dial := false
  809. s.dialNowDevicesMut.Lock()
  810. for _, dev := range to.Devices {
  811. if dev.Paused {
  812. continue
  813. }
  814. if oldDev, ok := oldDevices[dev.DeviceID]; !ok || oldDev.Paused {
  815. s.dialNowDevices[dev.DeviceID] = struct{}{}
  816. dial = true
  817. } else if !slices.Equal(oldDev.Addresses, dev.Addresses) {
  818. dial = true
  819. }
  820. }
  821. if dial {
  822. s.scheduleDialNow()
  823. }
  824. s.dialNowDevicesMut.Unlock()
  825. }
  826. func (s *service) scheduleDialNow() {
  827. select {
  828. case s.dialNow <- struct{}{}:
  829. default:
  830. // channel is blocked - a config update is already pending for the connection loop.
  831. }
  832. }
  833. func (s *service) AllAddresses() []string {
  834. s.listenersMut.RLock()
  835. var addrs []string
  836. for _, listener := range s.listeners {
  837. for _, lanAddr := range listener.LANAddresses() {
  838. addrs = append(addrs, lanAddr.String())
  839. }
  840. for _, wanAddr := range listener.WANAddresses() {
  841. addrs = append(addrs, wanAddr.String())
  842. }
  843. }
  844. s.listenersMut.RUnlock()
  845. return stringutil.UniqueTrimmedStrings(addrs)
  846. }
  847. func (s *service) ExternalAddresses() []string {
  848. if s.cfg.Options().AnnounceLANAddresses {
  849. return s.AllAddresses()
  850. }
  851. s.listenersMut.RLock()
  852. var addrs []string
  853. for _, listener := range s.listeners {
  854. for _, wanAddr := range listener.WANAddresses() {
  855. addrs = append(addrs, wanAddr.String())
  856. }
  857. }
  858. s.listenersMut.RUnlock()
  859. return stringutil.UniqueTrimmedStrings(addrs)
  860. }
  861. func (s *service) ListenerStatus() map[string]ListenerStatusEntry {
  862. result := make(map[string]ListenerStatusEntry)
  863. s.listenersMut.RLock()
  864. for addr, listener := range s.listeners {
  865. var status ListenerStatusEntry
  866. if err := listener.Error(); err != nil {
  867. errStr := err.Error()
  868. status.Error = &errStr
  869. }
  870. status.LANAddresses = urlsToStrings(listener.LANAddresses())
  871. status.WANAddresses = urlsToStrings(listener.WANAddresses())
  872. result[addr] = status
  873. }
  874. s.listenersMut.RUnlock()
  875. return result
  876. }
  877. type connectionStatusHandler struct {
  878. connectionStatusMut sync.RWMutex
  879. connectionStatus map[string]ConnectionStatusEntry // address -> latest error/status
  880. }
  881. func newConnectionStatusHandler() connectionStatusHandler {
  882. return connectionStatusHandler{
  883. connectionStatusMut: sync.NewRWMutex(),
  884. connectionStatus: make(map[string]ConnectionStatusEntry),
  885. }
  886. }
  887. func (s *connectionStatusHandler) ConnectionStatus() map[string]ConnectionStatusEntry {
  888. result := make(map[string]ConnectionStatusEntry)
  889. s.connectionStatusMut.RLock()
  890. for k, v := range s.connectionStatus {
  891. result[k] = v
  892. }
  893. s.connectionStatusMut.RUnlock()
  894. return result
  895. }
  896. func (s *connectionStatusHandler) setConnectionStatus(address string, err error) {
  897. if errors.Is(err, context.Canceled) {
  898. return
  899. }
  900. status := ConnectionStatusEntry{When: time.Now().UTC().Truncate(time.Second)}
  901. if err != nil {
  902. errStr := err.Error()
  903. status.Error = &errStr
  904. }
  905. s.connectionStatusMut.Lock()
  906. s.connectionStatus[address] = status
  907. s.connectionStatusMut.Unlock()
  908. }
  909. func (s *service) NATType() string {
  910. s.listenersMut.RLock()
  911. defer s.listenersMut.RUnlock()
  912. for _, listener := range s.listeners {
  913. natType := listener.NATType()
  914. if natType != "unknown" {
  915. return natType
  916. }
  917. }
  918. return "unknown"
  919. }
  920. func getDialerFactory(cfg config.Configuration, uri *url.URL) (dialerFactory, error) {
  921. dialerFactory, ok := dialers[uri.Scheme]
  922. if !ok {
  923. return nil, fmt.Errorf("unknown address scheme %q", uri.Scheme)
  924. }
  925. if err := dialerFactory.Valid(cfg); err != nil {
  926. return nil, err
  927. }
  928. return dialerFactory, nil
  929. }
  930. func getListenerFactory(cfg config.Configuration, uri *url.URL) (listenerFactory, error) {
  931. listenerFactory, ok := listeners[uri.Scheme]
  932. if !ok {
  933. return nil, fmt.Errorf("unknown address scheme %q", uri.Scheme)
  934. }
  935. if err := listenerFactory.Valid(cfg); err != nil {
  936. return nil, err
  937. }
  938. return listenerFactory, nil
  939. }
  940. func urlsToStrings(urls []*url.URL) []string {
  941. strings := make([]string, len(urls))
  942. for i, url := range urls {
  943. strings[i] = url.String()
  944. }
  945. return strings
  946. }
  947. var (
  948. warningLimiters = make(map[protocol.DeviceID]*rate.Limiter)
  949. warningLimitersMut = sync.NewMutex()
  950. )
  951. func warningFor(dev protocol.DeviceID, msg string) {
  952. warningLimitersMut.Lock()
  953. defer warningLimitersMut.Unlock()
  954. lim, ok := warningLimiters[dev]
  955. if !ok {
  956. lim = rate.NewLimiter(rate.Every(perDeviceWarningIntv), 1)
  957. warningLimiters[dev] = lim
  958. }
  959. if lim.Allow() {
  960. l.Warnln(msg)
  961. }
  962. }
  963. func tlsTimedHandshake(tc *tls.Conn) error {
  964. tc.SetDeadline(time.Now().Add(tlsHandshakeTimeout))
  965. defer tc.SetDeadline(time.Time{})
  966. return tc.Handshake()
  967. }
  968. // IsAllowedNetwork returns true if the given host (IP or resolvable
  969. // hostname) is in the set of allowed networks (CIDR format only).
  970. func IsAllowedNetwork(host string, allowed []string) bool {
  971. if hostNoPort, _, err := net.SplitHostPort(host); err == nil {
  972. host = hostNoPort
  973. }
  974. addr, err := net.ResolveIPAddr("ip", host)
  975. if err != nil {
  976. return false
  977. }
  978. for _, n := range allowed {
  979. result := true
  980. if strings.HasPrefix(n, "!") {
  981. result = false
  982. n = n[1:]
  983. }
  984. _, cidr, err := net.ParseCIDR(n)
  985. if err != nil {
  986. continue
  987. }
  988. if cidr.Contains(addr.IP) {
  989. return result
  990. }
  991. }
  992. return false
  993. }
  994. func (s *service) dialParallel(ctx context.Context, deviceID protocol.DeviceID, dialTargets []dialTarget, parentSema *semaphore.Semaphore) (internalConn, bool) {
  995. // Group targets into buckets by priority
  996. dialTargetBuckets := make(map[int][]dialTarget, len(dialTargets))
  997. for _, tgt := range dialTargets {
  998. dialTargetBuckets[tgt.priority] = append(dialTargetBuckets[tgt.priority], tgt)
  999. }
  1000. // Get all available priorities
  1001. priorities := make([]int, 0, len(dialTargetBuckets))
  1002. for prio := range dialTargetBuckets {
  1003. priorities = append(priorities, prio)
  1004. }
  1005. // Sort the priorities so that we dial lowest first (which means highest...)
  1006. sort.Ints(priorities)
  1007. sema := semaphore.MultiSemaphore{semaphore.New(dialMaxParallelPerDevice), parentSema}
  1008. for _, prio := range priorities {
  1009. tgts := dialTargetBuckets[prio]
  1010. res := make(chan internalConn, len(tgts))
  1011. wg := stdsync.WaitGroup{}
  1012. for _, tgt := range tgts {
  1013. sema.Take(1)
  1014. wg.Add(1)
  1015. go func(tgt dialTarget) {
  1016. defer func() {
  1017. wg.Done()
  1018. sema.Give(1)
  1019. }()
  1020. conn, err := tgt.Dial(ctx)
  1021. if err == nil {
  1022. // Closes the connection on error
  1023. err = s.validateIdentity(conn, deviceID)
  1024. }
  1025. s.setConnectionStatus(tgt.addr, err)
  1026. if err != nil {
  1027. l.Debugln("dialing", deviceID, tgt.uri, "error:", err)
  1028. } else {
  1029. l.Debugln("dialing", deviceID, tgt.uri, "success:", conn)
  1030. res <- conn
  1031. }
  1032. }(tgt)
  1033. }
  1034. // Spawn a routine which will unblock main routine in case we fail
  1035. // to connect to anyone.
  1036. go func() {
  1037. wg.Wait()
  1038. close(res)
  1039. }()
  1040. // Wait for the first connection, or for channel closure.
  1041. if conn, ok := <-res; ok {
  1042. // Got a connection, means more might come back, hence spawn a
  1043. // routine that will do the discarding.
  1044. l.Debugln("connected to", deviceID, prio, "using", conn, conn.priority)
  1045. go func(deviceID protocol.DeviceID, prio int) {
  1046. wg.Wait()
  1047. l.Debugln("discarding", len(res), "connections while connecting to", deviceID, prio)
  1048. for conn := range res {
  1049. conn.Close()
  1050. }
  1051. }(deviceID, prio)
  1052. return conn, ok
  1053. }
  1054. // Failed to connect, report that fact.
  1055. l.Debugln("failed to connect to", deviceID, prio)
  1056. }
  1057. return internalConn{}, false
  1058. }
  1059. func (s *service) validateIdentity(c internalConn, expectedID protocol.DeviceID) error {
  1060. cs := c.ConnectionState()
  1061. // We should have received exactly one certificate from the other
  1062. // side. If we didn't, they don't have a device ID and we drop the
  1063. // connection.
  1064. certs := cs.PeerCertificates
  1065. if cl := len(certs); cl != 1 {
  1066. l.Infof("Got peer certificate list of length %d != 1 from peer at %s; protocol error", cl, c)
  1067. c.Close()
  1068. return fmt.Errorf("expected 1 certificate, got %d", cl)
  1069. }
  1070. remoteCert := certs[0]
  1071. remoteID := protocol.NewDeviceID(remoteCert.Raw)
  1072. // The device ID should not be that of ourselves. It can happen
  1073. // though, especially in the presence of NAT hairpinning, multiple
  1074. // clients between the same NAT gateway, and global discovery.
  1075. if remoteID == s.myID {
  1076. l.Debugf("Connected to myself (%s) at %s", remoteID, c)
  1077. c.Close()
  1078. return errors.New("connected to self")
  1079. }
  1080. // We should see the expected device ID
  1081. if !remoteID.Equals(expectedID) {
  1082. c.Close()
  1083. return fmt.Errorf("unexpected device id, expected %s got %s", expectedID, remoteID)
  1084. }
  1085. return nil
  1086. }
  1087. type nextDialRegistry map[protocol.DeviceID]nextDialDevice
  1088. type nextDialDevice struct {
  1089. nextDial map[string]time.Time
  1090. coolDownIntervalStart time.Time
  1091. attempts int
  1092. }
  1093. func (r nextDialRegistry) get(device protocol.DeviceID, addr string) time.Time {
  1094. return r[device].nextDial[addr]
  1095. }
  1096. const (
  1097. dialCoolDownInterval = 2 * time.Minute
  1098. dialCoolDownDelay = 5 * time.Minute
  1099. dialCoolDownMaxAttempts = 3
  1100. )
  1101. // redialDevice marks the device for immediate redial, unless the remote keeps
  1102. // dropping established connections. Thus we keep track of when the first forced
  1103. // re-dial happened, and how many attempts happen in the dialCoolDownInterval
  1104. // after that. If it's more than dialCoolDownMaxAttempts, don't force-redial
  1105. // that device for dialCoolDownDelay (regular dials still happen).
  1106. func (r nextDialRegistry) redialDevice(device protocol.DeviceID, now time.Time) {
  1107. dev, ok := r[device]
  1108. if !ok {
  1109. r[device] = nextDialDevice{
  1110. nextDial: make(map[string]time.Time),
  1111. coolDownIntervalStart: now,
  1112. attempts: 1,
  1113. }
  1114. return
  1115. }
  1116. if dev.attempts == 0 || now.Before(dev.coolDownIntervalStart.Add(dialCoolDownInterval)) {
  1117. if dev.attempts >= dialCoolDownMaxAttempts {
  1118. // Device has been force redialed too often - let it cool down.
  1119. return
  1120. }
  1121. if dev.attempts == 0 {
  1122. dev.coolDownIntervalStart = now
  1123. }
  1124. dev.attempts++
  1125. dev.nextDial = make(map[string]time.Time)
  1126. r[device] = dev
  1127. return
  1128. }
  1129. if dev.attempts >= dialCoolDownMaxAttempts && now.Before(dev.coolDownIntervalStart.Add(dialCoolDownDelay)) {
  1130. return // Still cooling down
  1131. }
  1132. delete(r, device)
  1133. }
  1134. func (r nextDialRegistry) set(device protocol.DeviceID, addr string, next time.Time) {
  1135. if _, ok := r[device]; !ok {
  1136. r[device] = nextDialDevice{nextDial: make(map[string]time.Time)}
  1137. }
  1138. r[device].nextDial[addr] = next
  1139. }
  1140. func (r nextDialRegistry) sleepDurationAndCleanup(now time.Time) time.Duration {
  1141. sleep := stdConnectionLoopSleep
  1142. for id, dev := range r {
  1143. for address, next := range dev.nextDial {
  1144. if next.Before(now) {
  1145. // Expired entry, address was not seen in last pass(es)
  1146. delete(dev.nextDial, address)
  1147. continue
  1148. }
  1149. if cur := next.Sub(now); cur < sleep {
  1150. sleep = cur
  1151. }
  1152. }
  1153. if dev.attempts > 0 {
  1154. interval := dialCoolDownInterval
  1155. if dev.attempts >= dialCoolDownMaxAttempts {
  1156. interval = dialCoolDownDelay
  1157. }
  1158. if now.After(dev.coolDownIntervalStart.Add(interval)) {
  1159. dev.attempts = 0
  1160. }
  1161. }
  1162. if len(dev.nextDial) == 0 && dev.attempts == 0 {
  1163. delete(r, id)
  1164. }
  1165. }
  1166. return sleep
  1167. }
  1168. func (s *service) desiredConnectionsToDevice(deviceID protocol.DeviceID) int {
  1169. cfg, ok := s.cfg.Device(deviceID)
  1170. if !ok {
  1171. // We want no connections to an unknown device.
  1172. return 0
  1173. }
  1174. otherSide := s.wantConnectionsForDevice(deviceID)
  1175. thisSide := cfg.NumConnections()
  1176. switch {
  1177. case otherSide <= 0:
  1178. // The other side doesn't support multiple connections, or we
  1179. // haven't yet connected to them so we don't know what they support
  1180. // or not. Use a single connection until we know better.
  1181. return 1
  1182. case otherSide == 1:
  1183. // The other side supports multiple connections, but only wants
  1184. // one. We should honour that.
  1185. return 1
  1186. case thisSide == 1:
  1187. // We want only one connection, so we should honour that.
  1188. return 1
  1189. // Finally, we allow negotiation and use the higher of the two values,
  1190. // while keeping at or below the max allowed value.
  1191. default:
  1192. return min(max(thisSide, otherSide), maxNumConnections)
  1193. }
  1194. }
  1195. // The deviceConnectionTracker keeps track of how many devices we are
  1196. // connected to and how many connections we have to each device. It also
  1197. // tracks how many connections they are willing to use.
  1198. type deviceConnectionTracker struct {
  1199. connectionsMut stdsync.Mutex
  1200. connections map[protocol.DeviceID][]protocol.Connection // current connections
  1201. wantConnections map[protocol.DeviceID]int // number of connections they want
  1202. }
  1203. func (c *deviceConnectionTracker) accountAddedConnection(conn protocol.Connection, h protocol.Hello, upgradeThreshold int) {
  1204. c.connectionsMut.Lock()
  1205. defer c.connectionsMut.Unlock()
  1206. // Lazily initialize the maps
  1207. if c.connections == nil {
  1208. c.connections = make(map[protocol.DeviceID][]protocol.Connection)
  1209. c.wantConnections = make(map[protocol.DeviceID]int)
  1210. }
  1211. // Add the connection to the list of current connections and remember
  1212. // how many total connections they want
  1213. d := conn.DeviceID()
  1214. c.connections[d] = append(c.connections[d], conn)
  1215. c.wantConnections[d] = int(h.NumConnections)
  1216. l.Debugf("Added connection for %s (now %d), they want %d connections", d.Short(), len(c.connections[d]), h.NumConnections)
  1217. // Update active connections metric
  1218. metricDeviceActiveConnections.WithLabelValues(d.String()).Inc()
  1219. // Close any connections we no longer want to retain.
  1220. c.closeWorsePriorityConnectionsLocked(d, conn.Priority()-upgradeThreshold)
  1221. }
  1222. func (c *deviceConnectionTracker) accountRemovedConnection(conn protocol.Connection) {
  1223. c.connectionsMut.Lock()
  1224. defer c.connectionsMut.Unlock()
  1225. d := conn.DeviceID()
  1226. cid := conn.ConnectionID()
  1227. // Remove the connection from the list of current connections
  1228. for i, conn := range c.connections[d] {
  1229. if conn.ConnectionID() == cid {
  1230. c.connections[d] = sliceutil.RemoveAndZero(c.connections[d], i)
  1231. break
  1232. }
  1233. }
  1234. // Clean up if required
  1235. if len(c.connections[d]) == 0 {
  1236. delete(c.connections, d)
  1237. delete(c.wantConnections, d)
  1238. }
  1239. // Update active connections metric
  1240. metricDeviceActiveConnections.WithLabelValues(d.String()).Dec()
  1241. l.Debugf("Removed connection for %s (now %d)", d.Short(), c.connections[d])
  1242. }
  1243. func (c *deviceConnectionTracker) numConnectionsForDevice(d protocol.DeviceID) int {
  1244. c.connectionsMut.Lock()
  1245. defer c.connectionsMut.Unlock()
  1246. return len(c.connections[d])
  1247. }
  1248. func (c *deviceConnectionTracker) wantConnectionsForDevice(d protocol.DeviceID) int {
  1249. c.connectionsMut.Lock()
  1250. defer c.connectionsMut.Unlock()
  1251. return c.wantConnections[d]
  1252. }
  1253. func (c *deviceConnectionTracker) numConnectedDevices() int {
  1254. c.connectionsMut.Lock()
  1255. defer c.connectionsMut.Unlock()
  1256. return len(c.connections)
  1257. }
  1258. func (c *deviceConnectionTracker) worstConnectionPriority(d protocol.DeviceID) int {
  1259. c.connectionsMut.Lock()
  1260. defer c.connectionsMut.Unlock()
  1261. if len(c.connections[d]) == 0 {
  1262. return math.MaxInt // worst possible priority
  1263. }
  1264. worstPriority := c.connections[d][0].Priority()
  1265. for _, conn := range c.connections[d][1:] {
  1266. if p := conn.Priority(); p > worstPriority {
  1267. worstPriority = p
  1268. }
  1269. }
  1270. return worstPriority
  1271. }
  1272. // closeWorsePriorityConnectionsLocked closes all connections to the given
  1273. // device that are worse than the cutoff priority. Must be called with the
  1274. // lock held.
  1275. func (c *deviceConnectionTracker) closeWorsePriorityConnectionsLocked(d protocol.DeviceID, cutoff int) {
  1276. for _, conn := range c.connections[d] {
  1277. if p := conn.Priority(); p > cutoff {
  1278. l.Debugf("Closing connection %s to %s with priority %d (cutoff %d)", conn, d.Short(), p, cutoff)
  1279. go conn.Close(errReplacingConnection)
  1280. }
  1281. }
  1282. }
  1283. // newConnectionID generates a connection ID. The connection ID is designed
  1284. // to be unique for each connection and chronologically sortable. It is
  1285. // based on the sum of two timestamps: when we think the connection was
  1286. // started, and when the other side thinks the connection was started. We
  1287. // then add some random data for good measure. This way, even if the other
  1288. // side does some funny business with the timestamp, we will get no worse
  1289. // than random connection IDs.
  1290. func newConnectionID(t0, t1 int64) string {
  1291. var buf [16]byte // 8 bytes timestamp, 8 bytes random
  1292. binary.BigEndian.PutUint64(buf[:], uint64(t0+t1))
  1293. _, _ = io.ReadFull(rand.Reader, buf[8:])
  1294. enc := base32.HexEncoding.WithPadding(base32.NoPadding)
  1295. // We encode the two parts separately and concatenate the results. The
  1296. // reason for this is that the timestamp (64 bits) doesn't precisely
  1297. // align to the base32 encoding (5 bits per character), so we'd get a
  1298. // character in the middle that is a mix of bits from the timestamp and
  1299. // from the random. We want the timestamp part deterministic.
  1300. return enc.EncodeToString(buf[:8]) + enc.EncodeToString(buf[8:])
  1301. }