service.go 26 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938
  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. package connections
  7. import (
  8. "context"
  9. "crypto/tls"
  10. "fmt"
  11. "net"
  12. "net/url"
  13. "sort"
  14. "strings"
  15. stdsync "sync"
  16. "time"
  17. "github.com/syncthing/syncthing/lib/config"
  18. "github.com/syncthing/syncthing/lib/discover"
  19. "github.com/syncthing/syncthing/lib/events"
  20. "github.com/syncthing/syncthing/lib/nat"
  21. "github.com/syncthing/syncthing/lib/osutil"
  22. "github.com/syncthing/syncthing/lib/protocol"
  23. "github.com/syncthing/syncthing/lib/sync"
  24. "github.com/syncthing/syncthing/lib/util"
  25. // Registers NAT service providers
  26. _ "github.com/syncthing/syncthing/lib/pmp"
  27. _ "github.com/syncthing/syncthing/lib/upnp"
  28. "github.com/pkg/errors"
  29. "github.com/thejerf/suture"
  30. "golang.org/x/time/rate"
  31. )
  32. var (
  33. dialers = make(map[string]dialerFactory)
  34. listeners = make(map[string]listenerFactory)
  35. )
  36. var (
  37. errDisabled = errors.New("disabled by configuration")
  38. errDeprecated = errors.New("deprecated protocol")
  39. )
  40. const (
  41. perDeviceWarningIntv = 15 * time.Minute
  42. tlsHandshakeTimeout = 10 * time.Second
  43. )
  44. // From go/src/crypto/tls/cipher_suites.go
  45. var tlsCipherSuiteNames = map[uint16]string{
  46. // TLS 1.2
  47. 0x0005: "TLS_RSA_WITH_RC4_128_SHA",
  48. 0x000a: "TLS_RSA_WITH_3DES_EDE_CBC_SHA",
  49. 0x002f: "TLS_RSA_WITH_AES_128_CBC_SHA",
  50. 0x0035: "TLS_RSA_WITH_AES_256_CBC_SHA",
  51. 0x003c: "TLS_RSA_WITH_AES_128_CBC_SHA256",
  52. 0x009c: "TLS_RSA_WITH_AES_128_GCM_SHA256",
  53. 0x009d: "TLS_RSA_WITH_AES_256_GCM_SHA384",
  54. 0xc007: "TLS_ECDHE_ECDSA_WITH_RC4_128_SHA",
  55. 0xc009: "TLS_ECDHE_ECDSA_WITH_AES_128_CBC_SHA",
  56. 0xc00a: "TLS_ECDHE_ECDSA_WITH_AES_256_CBC_SHA",
  57. 0xc011: "TLS_ECDHE_RSA_WITH_RC4_128_SHA",
  58. 0xc012: "TLS_ECDHE_RSA_WITH_3DES_EDE_CBC_SHA",
  59. 0xc013: "TLS_ECDHE_RSA_WITH_AES_128_CBC_SHA",
  60. 0xc014: "TLS_ECDHE_RSA_WITH_AES_256_CBC_SHA",
  61. 0xc023: "TLS_ECDHE_ECDSA_WITH_AES_128_CBC_SHA256",
  62. 0xc027: "TLS_ECDHE_RSA_WITH_AES_128_CBC_SHA256",
  63. 0xc02f: "TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256",
  64. 0xc02b: "TLS_ECDHE_ECDSA_WITH_AES_128_GCM_SHA256",
  65. 0xc030: "TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384",
  66. 0xc02c: "TLS_ECDHE_ECDSA_WITH_AES_256_GCM_SHA384",
  67. 0xcca8: "TLS_ECDHE_RSA_WITH_CHACHA20_POLY1305",
  68. 0xcca9: "TLS_ECDHE_ECDSA_WITH_CHACHA20_POLY1305",
  69. // TLS 1.3
  70. 0x1301: "TLS_AES_128_GCM_SHA256",
  71. 0x1302: "TLS_AES_256_GCM_SHA384",
  72. 0x1303: "TLS_CHACHA20_POLY1305_SHA256",
  73. }
  74. var tlsVersionNames = map[uint16]string{
  75. tls.VersionTLS12: "TLS1.2",
  76. tls.VersionTLS13: "TLS1.3",
  77. }
  78. // Service listens and dials all configured unconnected devices, via supported
  79. // dialers. Successful connections are handed to the model.
  80. type Service interface {
  81. suture.Service
  82. discover.AddressLister
  83. ListenerStatus() map[string]ListenerStatusEntry
  84. ConnectionStatus() map[string]ConnectionStatusEntry
  85. NATType() string
  86. }
  87. type ListenerStatusEntry struct {
  88. Error *string `json:"error"`
  89. LANAddresses []string `json:"lanAddresses"`
  90. WANAddresses []string `json:"wanAddresses"`
  91. }
  92. type ConnectionStatusEntry struct {
  93. When time.Time `json:"when"`
  94. Error *string `json:"error"`
  95. }
  96. type service struct {
  97. *suture.Supervisor
  98. cfg config.Wrapper
  99. myID protocol.DeviceID
  100. model Model
  101. tlsCfg *tls.Config
  102. discoverer discover.Finder
  103. conns chan internalConn
  104. bepProtocolName string
  105. tlsDefaultCommonName string
  106. limiter *limiter
  107. natService *nat.Service
  108. natServiceToken *suture.ServiceToken
  109. evLogger events.Logger
  110. listenersMut sync.RWMutex
  111. listeners map[string]genericListener
  112. listenerTokens map[string]suture.ServiceToken
  113. listenerSupervisor *suture.Supervisor
  114. connectionStatusMut sync.RWMutex
  115. connectionStatus map[string]ConnectionStatusEntry // address -> latest error/status
  116. }
  117. func NewService(cfg config.Wrapper, myID protocol.DeviceID, mdl Model, tlsCfg *tls.Config, discoverer discover.Finder, bepProtocolName string, tlsDefaultCommonName string, evLogger events.Logger) Service {
  118. service := &service{
  119. Supervisor: suture.New("connections.Service", suture.Spec{
  120. Log: func(line string) {
  121. l.Infoln(line)
  122. },
  123. PassThroughPanics: true,
  124. }),
  125. cfg: cfg,
  126. myID: myID,
  127. model: mdl,
  128. tlsCfg: tlsCfg,
  129. discoverer: discoverer,
  130. conns: make(chan internalConn),
  131. bepProtocolName: bepProtocolName,
  132. tlsDefaultCommonName: tlsDefaultCommonName,
  133. limiter: newLimiter(cfg),
  134. natService: nat.NewService(myID, cfg),
  135. evLogger: evLogger,
  136. listenersMut: sync.NewRWMutex(),
  137. listeners: make(map[string]genericListener),
  138. listenerTokens: make(map[string]suture.ServiceToken),
  139. // A listener can fail twice, rapidly. Any more than that and it
  140. // will be put on suspension for ten minutes. Restarts and changes
  141. // due to config are done by removing and adding services, so are
  142. // not subject to these limitations.
  143. listenerSupervisor: suture.New("c.S.listenerSupervisor", suture.Spec{
  144. Log: func(line string) {
  145. l.Infoln(line)
  146. },
  147. FailureThreshold: 2,
  148. FailureBackoff: 600 * time.Second,
  149. PassThroughPanics: true,
  150. }),
  151. connectionStatusMut: sync.NewRWMutex(),
  152. connectionStatus: make(map[string]ConnectionStatusEntry),
  153. }
  154. cfg.Subscribe(service)
  155. raw := cfg.RawCopy()
  156. // Actually starts the listeners and NAT service
  157. // Need to start this before service.connect so that any dials that
  158. // try punch through already have a listener to cling on.
  159. service.CommitConfiguration(raw, raw)
  160. // There are several moving parts here; one routine per listening address
  161. // (handled in configuration changing) to handle incoming connections,
  162. // one routine to periodically attempt outgoing connections, one routine to
  163. // the common handling regardless of whether the connection was
  164. // incoming or outgoing.
  165. service.Add(util.AsService(service.connect, fmt.Sprintf("%s/connect", service)))
  166. service.Add(util.AsService(service.handle, fmt.Sprintf("%s/handle", service)))
  167. service.Add(service.listenerSupervisor)
  168. return service
  169. }
  170. func (s *service) Stop() {
  171. s.cfg.Unsubscribe(s.limiter)
  172. s.cfg.Unsubscribe(s)
  173. s.Supervisor.Stop()
  174. }
  175. func (s *service) handle(ctx context.Context) {
  176. var c internalConn
  177. for {
  178. select {
  179. case <-ctx.Done():
  180. return
  181. case c = <-s.conns:
  182. }
  183. cs := c.ConnectionState()
  184. // We should have negotiated the next level protocol "bep/1.0" as part
  185. // of the TLS handshake. Unfortunately this can't be a hard error,
  186. // because there are implementations out there that don't support
  187. // protocol negotiation (iOS for one...).
  188. if !cs.NegotiatedProtocolIsMutual || cs.NegotiatedProtocol != s.bepProtocolName {
  189. l.Infof("Peer at %s did not negotiate bep/1.0", c)
  190. }
  191. // We should have received exactly one certificate from the other
  192. // side. If we didn't, they don't have a device ID and we drop the
  193. // connection.
  194. certs := cs.PeerCertificates
  195. if cl := len(certs); cl != 1 {
  196. l.Infof("Got peer certificate list of length %d != 1 from peer at %s; protocol error", cl, c)
  197. c.Close()
  198. continue
  199. }
  200. remoteCert := certs[0]
  201. remoteID := protocol.NewDeviceID(remoteCert.Raw)
  202. // The device ID should not be that of ourselves. It can happen
  203. // though, especially in the presence of NAT hairpinning, multiple
  204. // clients between the same NAT gateway, and global discovery.
  205. if remoteID == s.myID {
  206. l.Infof("Connected to myself (%s) at %s - should not happen", remoteID, c)
  207. c.Close()
  208. continue
  209. }
  210. _ = c.SetDeadline(time.Now().Add(20 * time.Second))
  211. hello, err := protocol.ExchangeHello(c, s.model.GetHello(remoteID))
  212. if err != nil {
  213. if protocol.IsVersionMismatch(err) {
  214. // The error will be a relatively user friendly description
  215. // of what's wrong with the version compatibility. By
  216. // default identify the other side by device ID and IP.
  217. remote := fmt.Sprintf("%v (%v)", remoteID, c.RemoteAddr())
  218. if hello.DeviceName != "" {
  219. // If the name was set in the hello return, use that to
  220. // give the user more info about which device is the
  221. // affected one. It probably says more than the remote
  222. // IP.
  223. remote = fmt.Sprintf("%q (%s %s, %v)", hello.DeviceName, hello.ClientName, hello.ClientVersion, remoteID)
  224. }
  225. msg := fmt.Sprintf("Connecting to %s: %s", remote, err)
  226. warningFor(remoteID, msg)
  227. } else {
  228. // It's something else - connection reset or whatever
  229. l.Infof("Failed to exchange Hello messages with %s at %s: %s", remoteID, c, err)
  230. }
  231. c.Close()
  232. continue
  233. }
  234. _ = c.SetDeadline(time.Time{})
  235. // The Model will return an error for devices that we don't want to
  236. // have a connection with for whatever reason, for example unknown devices.
  237. if err := s.model.OnHello(remoteID, c.RemoteAddr(), hello); err != nil {
  238. l.Infof("Connection from %s at %s (%s) rejected: %v", remoteID, c.RemoteAddr(), c.Type(), err)
  239. c.Close()
  240. continue
  241. }
  242. // If we have a relay connection, and the new incoming connection is
  243. // not a relay connection, we should drop that, and prefer this one.
  244. ct, connected := s.model.Connection(remoteID)
  245. // Lower priority is better, just like nice etc.
  246. if connected && ct.Priority() > c.priority {
  247. l.Debugf("Switching connections %s (existing: %s new: %s)", remoteID, ct, c)
  248. } else if connected {
  249. // We should not already be connected to the other party. TODO: This
  250. // could use some better handling. If the old connection is dead but
  251. // hasn't timed out yet we may want to drop *that* connection and keep
  252. // this one. But in case we are two devices connecting to each other
  253. // in parallel we don't want to do that or we end up with no
  254. // connections still established...
  255. l.Infof("Connected to already connected device %s (existing: %s new: %s)", remoteID, ct, c)
  256. c.Close()
  257. continue
  258. }
  259. deviceCfg, ok := s.cfg.Device(remoteID)
  260. if !ok {
  261. l.Infof("Device %s removed from config during connection attempt at %s", remoteID, c)
  262. c.Close()
  263. continue
  264. }
  265. // Verify the name on the certificate. By default we set it to
  266. // "syncthing" when generating, but the user may have replaced
  267. // the certificate and used another name.
  268. certName := deviceCfg.CertName
  269. if certName == "" {
  270. certName = s.tlsDefaultCommonName
  271. }
  272. if err := remoteCert.VerifyHostname(certName); err != nil {
  273. // Incorrect certificate name is something the user most
  274. // likely wants to know about, since it's an advanced
  275. // config. Warn instead of Info.
  276. l.Warnf("Bad certificate from %s at %s: %v", remoteID, c, err)
  277. c.Close()
  278. continue
  279. }
  280. // Wrap the connection in rate limiters. The limiter itself will
  281. // keep up with config changes to the rate and whether or not LAN
  282. // connections are limited.
  283. isLAN := s.isLAN(c.RemoteAddr())
  284. rd, wr := s.limiter.getLimiters(remoteID, c, isLAN)
  285. protoConn := protocol.NewConnection(remoteID, rd, wr, s.model, c.String(), deviceCfg.Compression)
  286. modelConn := completeConn{c, protoConn}
  287. l.Infof("Established secure connection to %s at %s", remoteID, c)
  288. s.model.AddConnection(modelConn, hello)
  289. continue
  290. }
  291. }
  292. func (s *service) connect(ctx context.Context) {
  293. nextDial := make(map[string]time.Time)
  294. // Used as delay for the first few connection attempts, increases
  295. // exponentially
  296. initialRampup := time.Second
  297. // Calculated from actual dialers reconnectInterval
  298. var sleep time.Duration
  299. for {
  300. cfg := s.cfg.RawCopy()
  301. bestDialerPrio := 1<<31 - 1 // worse prio won't build on 32 bit
  302. for _, df := range dialers {
  303. if df.Valid(cfg) != nil {
  304. continue
  305. }
  306. if prio := df.Priority(); prio < bestDialerPrio {
  307. bestDialerPrio = prio
  308. }
  309. }
  310. l.Debugln("Reconnect loop")
  311. now := time.Now()
  312. var seen []string
  313. for _, deviceCfg := range cfg.Devices {
  314. deviceID := deviceCfg.DeviceID
  315. if deviceID == s.myID {
  316. continue
  317. }
  318. if deviceCfg.Paused {
  319. continue
  320. }
  321. ct, connected := s.model.Connection(deviceID)
  322. if connected && ct.Priority() == bestDialerPrio {
  323. // Things are already as good as they can get.
  324. continue
  325. }
  326. var addrs []string
  327. for _, addr := range deviceCfg.Addresses {
  328. if addr == "dynamic" {
  329. if s.discoverer != nil {
  330. if t, err := s.discoverer.Lookup(deviceID); err == nil {
  331. addrs = append(addrs, t...)
  332. }
  333. }
  334. } else {
  335. addrs = append(addrs, addr)
  336. }
  337. }
  338. addrs = util.UniqueTrimmedStrings(addrs)
  339. l.Debugln("Reconnect loop for", deviceID, addrs)
  340. dialTargets := make([]dialTarget, 0)
  341. for _, addr := range addrs {
  342. // Use a special key that is more than just the address, as you might have two devices connected to the same relay
  343. nextDialKey := deviceID.String() + "/" + addr
  344. seen = append(seen, nextDialKey)
  345. nextDialAt, ok := nextDial[nextDialKey]
  346. if ok && initialRampup >= sleep && nextDialAt.After(now) {
  347. l.Debugf("Not dialing %s via %v as sleep is %v, next dial is at %s and current time is %s", deviceID, addr, sleep, nextDialAt, now)
  348. continue
  349. }
  350. // If we fail at any step before actually getting the dialer
  351. // retry in a minute
  352. nextDial[nextDialKey] = now.Add(time.Minute)
  353. uri, err := url.Parse(addr)
  354. if err != nil {
  355. s.setConnectionStatus(addr, err)
  356. l.Infof("Parsing dialer address %s: %v", addr, err)
  357. continue
  358. }
  359. if len(deviceCfg.AllowedNetworks) > 0 {
  360. if !IsAllowedNetwork(uri.Host, deviceCfg.AllowedNetworks) {
  361. s.setConnectionStatus(addr, errors.New("network disallowed"))
  362. l.Debugln("Network for", uri, "is disallowed")
  363. continue
  364. }
  365. }
  366. dialerFactory, err := getDialerFactory(cfg, uri)
  367. if err != nil {
  368. s.setConnectionStatus(addr, err)
  369. }
  370. switch err {
  371. case nil:
  372. // all good
  373. case errDisabled:
  374. l.Debugln("Dialer for", uri, "is disabled")
  375. continue
  376. case errDeprecated:
  377. l.Debugln("Dialer for", uri, "is deprecated")
  378. continue
  379. default:
  380. l.Infof("Dialer for %v: %v", uri, err)
  381. continue
  382. }
  383. priority := dialerFactory.Priority()
  384. if connected && priority >= ct.Priority() {
  385. l.Debugf("Not dialing using %s as priority is less than current connection (%d >= %d)", dialerFactory, dialerFactory.Priority(), ct.Priority())
  386. continue
  387. }
  388. dialer := dialerFactory.New(s.cfg.Options(), s.tlsCfg)
  389. nextDial[nextDialKey] = now.Add(dialer.RedialFrequency())
  390. // For LAN addresses, increase the priority so that we
  391. // try these first.
  392. switch {
  393. case dialerFactory.AlwaysWAN():
  394. // Do nothing.
  395. case s.isLANHost(uri.Host):
  396. priority -= 1
  397. }
  398. dialTargets = append(dialTargets, dialTarget{
  399. addr: addr,
  400. dialer: dialer,
  401. priority: priority,
  402. deviceID: deviceID,
  403. uri: uri,
  404. })
  405. }
  406. conn, ok := s.dialParallel(ctx, deviceCfg.DeviceID, dialTargets)
  407. if ok {
  408. s.conns <- conn
  409. }
  410. }
  411. nextDial, sleep = filterAndFindSleepDuration(nextDial, seen, now)
  412. if initialRampup < sleep {
  413. l.Debugln("initial rampup; sleep", initialRampup, "and update to", initialRampup*2)
  414. sleep = initialRampup
  415. initialRampup *= 2
  416. } else {
  417. l.Debugln("sleep until next dial", sleep)
  418. }
  419. select {
  420. case <-time.After(sleep):
  421. case <-ctx.Done():
  422. return
  423. }
  424. }
  425. }
  426. func (s *service) isLANHost(host string) bool {
  427. // Probably we are called with an ip:port combo which we can resolve as
  428. // a TCP address.
  429. if addr, err := net.ResolveTCPAddr("tcp", host); err == nil {
  430. return s.isLAN(addr)
  431. }
  432. // ... but this function looks general enough that someone might try
  433. // with just an IP as well in the future so lets allow that.
  434. if addr, err := net.ResolveIPAddr("ip", host); err == nil {
  435. return s.isLAN(addr)
  436. }
  437. return false
  438. }
  439. func (s *service) isLAN(addr net.Addr) bool {
  440. var ip net.IP
  441. switch addr := addr.(type) {
  442. case *net.IPAddr:
  443. ip = addr.IP
  444. case *net.TCPAddr:
  445. ip = addr.IP
  446. case *net.UDPAddr:
  447. ip = addr.IP
  448. default:
  449. // From the standard library, just Unix sockets.
  450. // If you invent your own, handle it.
  451. return false
  452. }
  453. if ip.IsLoopback() {
  454. return true
  455. }
  456. for _, lan := range s.cfg.Options().AlwaysLocalNets {
  457. _, ipnet, err := net.ParseCIDR(lan)
  458. if err != nil {
  459. l.Debugln("Network", lan, "is malformed:", err)
  460. continue
  461. }
  462. if ipnet.Contains(ip) {
  463. return true
  464. }
  465. }
  466. lans, _ := osutil.GetLans()
  467. for _, lan := range lans {
  468. if lan.Contains(ip) {
  469. return true
  470. }
  471. }
  472. return false
  473. }
  474. func (s *service) createListener(factory listenerFactory, uri *url.URL) bool {
  475. // must be called with listenerMut held
  476. l.Debugln("Starting listener", uri)
  477. listener := factory.New(uri, s.cfg, s.tlsCfg, s.conns, s.natService)
  478. listener.OnAddressesChanged(s.logListenAddressesChangedEvent)
  479. s.listeners[uri.String()] = listener
  480. s.listenerTokens[uri.String()] = s.listenerSupervisor.Add(listener)
  481. return true
  482. }
  483. func (s *service) logListenAddressesChangedEvent(l genericListener) {
  484. s.evLogger.Log(events.ListenAddressesChanged, map[string]interface{}{
  485. "address": l.URI(),
  486. "lan": l.LANAddresses(),
  487. "wan": l.WANAddresses(),
  488. })
  489. }
  490. func (s *service) VerifyConfiguration(from, to config.Configuration) error {
  491. return nil
  492. }
  493. func (s *service) CommitConfiguration(from, to config.Configuration) bool {
  494. newDevices := make(map[protocol.DeviceID]bool, len(to.Devices))
  495. for _, dev := range to.Devices {
  496. newDevices[dev.DeviceID] = true
  497. }
  498. for _, dev := range from.Devices {
  499. if !newDevices[dev.DeviceID] {
  500. warningLimitersMut.Lock()
  501. delete(warningLimiters, dev.DeviceID)
  502. warningLimitersMut.Unlock()
  503. }
  504. }
  505. s.listenersMut.Lock()
  506. seen := make(map[string]struct{})
  507. for _, addr := range to.Options.ListenAddresses() {
  508. if addr == "" {
  509. // We can get an empty address if there is an empty listener
  510. // element in the config, indicating no listeners should be
  511. // used. This is not an error.
  512. continue
  513. }
  514. if _, ok := s.listeners[addr]; ok {
  515. seen[addr] = struct{}{}
  516. continue
  517. }
  518. uri, err := url.Parse(addr)
  519. if err != nil {
  520. l.Infof("Parsing listener address %s: %v", addr, err)
  521. continue
  522. }
  523. factory, err := getListenerFactory(to, uri)
  524. switch err {
  525. case nil:
  526. // all good
  527. case errDisabled:
  528. l.Debugln("Listener for", uri, "is disabled")
  529. continue
  530. case errDeprecated:
  531. l.Debugln("Listener for", uri, "is deprecated")
  532. continue
  533. default:
  534. l.Infof("Listener for %v: %v", uri, err)
  535. continue
  536. }
  537. s.createListener(factory, uri)
  538. seen[addr] = struct{}{}
  539. }
  540. for addr, listener := range s.listeners {
  541. if _, ok := seen[addr]; !ok || listener.Factory().Valid(to) != nil {
  542. l.Debugln("Stopping listener", addr)
  543. s.listenerSupervisor.Remove(s.listenerTokens[addr])
  544. delete(s.listenerTokens, addr)
  545. delete(s.listeners, addr)
  546. }
  547. }
  548. s.listenersMut.Unlock()
  549. if to.Options.NATEnabled && s.natServiceToken == nil {
  550. l.Debugln("Starting NAT service")
  551. token := s.Add(s.natService)
  552. s.natServiceToken = &token
  553. } else if !to.Options.NATEnabled && s.natServiceToken != nil {
  554. l.Debugln("Stopping NAT service")
  555. s.Remove(*s.natServiceToken)
  556. s.natServiceToken = nil
  557. }
  558. return true
  559. }
  560. func (s *service) AllAddresses() []string {
  561. s.listenersMut.RLock()
  562. var addrs []string
  563. for _, listener := range s.listeners {
  564. for _, lanAddr := range listener.LANAddresses() {
  565. addrs = append(addrs, lanAddr.String())
  566. }
  567. for _, wanAddr := range listener.WANAddresses() {
  568. addrs = append(addrs, wanAddr.String())
  569. }
  570. }
  571. s.listenersMut.RUnlock()
  572. return util.UniqueTrimmedStrings(addrs)
  573. }
  574. func (s *service) ExternalAddresses() []string {
  575. s.listenersMut.RLock()
  576. var addrs []string
  577. for _, listener := range s.listeners {
  578. for _, wanAddr := range listener.WANAddresses() {
  579. addrs = append(addrs, wanAddr.String())
  580. }
  581. }
  582. s.listenersMut.RUnlock()
  583. return util.UniqueTrimmedStrings(addrs)
  584. }
  585. func (s *service) ListenerStatus() map[string]ListenerStatusEntry {
  586. result := make(map[string]ListenerStatusEntry)
  587. s.listenersMut.RLock()
  588. for addr, listener := range s.listeners {
  589. var status ListenerStatusEntry
  590. if err := listener.Error(); err != nil {
  591. errStr := err.Error()
  592. status.Error = &errStr
  593. }
  594. status.LANAddresses = urlsToStrings(listener.LANAddresses())
  595. status.WANAddresses = urlsToStrings(listener.WANAddresses())
  596. result[addr] = status
  597. }
  598. s.listenersMut.RUnlock()
  599. return result
  600. }
  601. func (s *service) ConnectionStatus() map[string]ConnectionStatusEntry {
  602. result := make(map[string]ConnectionStatusEntry)
  603. s.connectionStatusMut.RLock()
  604. for k, v := range s.connectionStatus {
  605. result[k] = v
  606. }
  607. s.connectionStatusMut.RUnlock()
  608. return result
  609. }
  610. func (s *service) setConnectionStatus(address string, err error) {
  611. if errors.Cause(err) != context.Canceled {
  612. return
  613. }
  614. status := ConnectionStatusEntry{When: time.Now().UTC().Truncate(time.Second)}
  615. if err != nil {
  616. errStr := err.Error()
  617. status.Error = &errStr
  618. }
  619. s.connectionStatusMut.Lock()
  620. s.connectionStatus[address] = status
  621. s.connectionStatusMut.Unlock()
  622. }
  623. func (s *service) NATType() string {
  624. s.listenersMut.RLock()
  625. defer s.listenersMut.RUnlock()
  626. for _, listener := range s.listeners {
  627. natType := listener.NATType()
  628. if natType != "unknown" {
  629. return natType
  630. }
  631. }
  632. return "unknown"
  633. }
  634. func getDialerFactory(cfg config.Configuration, uri *url.URL) (dialerFactory, error) {
  635. dialerFactory, ok := dialers[uri.Scheme]
  636. if !ok {
  637. return nil, fmt.Errorf("unknown address scheme %q", uri.Scheme)
  638. }
  639. if err := dialerFactory.Valid(cfg); err != nil {
  640. return nil, err
  641. }
  642. return dialerFactory, nil
  643. }
  644. func getListenerFactory(cfg config.Configuration, uri *url.URL) (listenerFactory, error) {
  645. listenerFactory, ok := listeners[uri.Scheme]
  646. if !ok {
  647. return nil, fmt.Errorf("unknown address scheme %q", uri.Scheme)
  648. }
  649. if err := listenerFactory.Valid(cfg); err != nil {
  650. return nil, err
  651. }
  652. return listenerFactory, nil
  653. }
  654. func filterAndFindSleepDuration(nextDial map[string]time.Time, seen []string, now time.Time) (map[string]time.Time, time.Duration) {
  655. newNextDial := make(map[string]time.Time)
  656. for _, addr := range seen {
  657. nextDialAt, ok := nextDial[addr]
  658. if ok {
  659. newNextDial[addr] = nextDialAt
  660. }
  661. }
  662. min := time.Minute
  663. for _, next := range newNextDial {
  664. cur := next.Sub(now)
  665. if cur < min {
  666. min = cur
  667. }
  668. }
  669. return newNextDial, min
  670. }
  671. func urlsToStrings(urls []*url.URL) []string {
  672. strings := make([]string, len(urls))
  673. for i, url := range urls {
  674. strings[i] = url.String()
  675. }
  676. return strings
  677. }
  678. var warningLimiters = make(map[protocol.DeviceID]*rate.Limiter)
  679. var warningLimitersMut = sync.NewMutex()
  680. func warningFor(dev protocol.DeviceID, msg string) {
  681. warningLimitersMut.Lock()
  682. defer warningLimitersMut.Unlock()
  683. lim, ok := warningLimiters[dev]
  684. if !ok {
  685. lim = rate.NewLimiter(rate.Every(perDeviceWarningIntv), 1)
  686. warningLimiters[dev] = lim
  687. }
  688. if lim.Allow() {
  689. l.Warnln(msg)
  690. }
  691. }
  692. func tlsTimedHandshake(tc *tls.Conn) error {
  693. tc.SetDeadline(time.Now().Add(tlsHandshakeTimeout))
  694. defer tc.SetDeadline(time.Time{})
  695. return tc.Handshake()
  696. }
  697. // IsAllowedNetwork returns true if the given host (IP or resolvable
  698. // hostname) is in the set of allowed networks (CIDR format only).
  699. func IsAllowedNetwork(host string, allowed []string) bool {
  700. if hostNoPort, _, err := net.SplitHostPort(host); err == nil {
  701. host = hostNoPort
  702. }
  703. addr, err := net.ResolveIPAddr("ip", host)
  704. if err != nil {
  705. return false
  706. }
  707. for _, n := range allowed {
  708. result := true
  709. if strings.HasPrefix(n, "!") {
  710. result = false
  711. n = n[1:]
  712. }
  713. _, cidr, err := net.ParseCIDR(n)
  714. if err != nil {
  715. continue
  716. }
  717. if cidr.Contains(addr.IP) {
  718. return result
  719. }
  720. }
  721. return false
  722. }
  723. func (s *service) dialParallel(ctx context.Context, deviceID protocol.DeviceID, dialTargets []dialTarget) (internalConn, bool) {
  724. // Group targets into buckets by priority
  725. dialTargetBuckets := make(map[int][]dialTarget, len(dialTargets))
  726. for _, tgt := range dialTargets {
  727. dialTargetBuckets[tgt.priority] = append(dialTargetBuckets[tgt.priority], tgt)
  728. }
  729. // Get all available priorities
  730. priorities := make([]int, 0, len(dialTargetBuckets))
  731. for prio := range dialTargetBuckets {
  732. priorities = append(priorities, prio)
  733. }
  734. // Sort the priorities so that we dial lowest first (which means highest...)
  735. sort.Ints(priorities)
  736. for _, prio := range priorities {
  737. tgts := dialTargetBuckets[prio]
  738. res := make(chan internalConn, len(tgts))
  739. wg := stdsync.WaitGroup{}
  740. for _, tgt := range tgts {
  741. wg.Add(1)
  742. go func(tgt dialTarget) {
  743. conn, err := tgt.Dial(ctx)
  744. if err == nil {
  745. // Closes the connection on error
  746. err = s.validateIdentity(conn, deviceID)
  747. }
  748. s.setConnectionStatus(tgt.addr, err)
  749. if err != nil {
  750. l.Debugln("dialing", deviceID, tgt.uri, "error:", err)
  751. } else {
  752. l.Debugln("dialing", deviceID, tgt.uri, "success:", conn)
  753. res <- conn
  754. }
  755. wg.Done()
  756. }(tgt)
  757. }
  758. // Spawn a routine which will unblock main routine in case we fail
  759. // to connect to anyone.
  760. go func() {
  761. wg.Wait()
  762. close(res)
  763. }()
  764. // Wait for the first connection, or for channel closure.
  765. if conn, ok := <-res; ok {
  766. // Got a connection, means more might come back, hence spawn a
  767. // routine that will do the discarding.
  768. l.Debugln("connected to", deviceID, prio, "using", conn, conn.priority)
  769. go func(deviceID protocol.DeviceID, prio int) {
  770. wg.Wait()
  771. l.Debugln("discarding", len(res), "connections while connecting to", deviceID, prio)
  772. for conn := range res {
  773. conn.Close()
  774. }
  775. }(deviceID, prio)
  776. return conn, ok
  777. }
  778. // Failed to connect, report that fact.
  779. l.Debugln("failed to connect to", deviceID, prio)
  780. }
  781. return internalConn{}, false
  782. }
  783. func (s *service) validateIdentity(c internalConn, expectedID protocol.DeviceID) error {
  784. cs := c.ConnectionState()
  785. // We should have received exactly one certificate from the other
  786. // side. If we didn't, they don't have a device ID and we drop the
  787. // connection.
  788. certs := cs.PeerCertificates
  789. if cl := len(certs); cl != 1 {
  790. l.Infof("Got peer certificate list of length %d != 1 from peer at %s; protocol error", cl, c)
  791. c.Close()
  792. return fmt.Errorf("expected 1 certificate, got %d", cl)
  793. }
  794. remoteCert := certs[0]
  795. remoteID := protocol.NewDeviceID(remoteCert.Raw)
  796. // The device ID should not be that of ourselves. It can happen
  797. // though, especially in the presence of NAT hairpinning, multiple
  798. // clients between the same NAT gateway, and global discovery.
  799. if remoteID == s.myID {
  800. l.Infof("Connected to myself (%s) at %s - should not happen", remoteID, c)
  801. c.Close()
  802. return fmt.Errorf("connected to self")
  803. }
  804. // We should see the expected device ID
  805. if !remoteID.Equals(expectedID) {
  806. c.Close()
  807. return fmt.Errorf("unexpected device id, expected %s got %s", expectedID, remoteID)
  808. }
  809. return nil
  810. }