relay.go 6.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277
  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 http://mozilla.org/MPL/2.0/.
  6. package relay
  7. import (
  8. "crypto/tls"
  9. "encoding/json"
  10. "net"
  11. "net/http"
  12. "net/url"
  13. "time"
  14. "github.com/syncthing/relaysrv/client"
  15. "github.com/syncthing/relaysrv/protocol"
  16. "github.com/syncthing/syncthing/lib/config"
  17. "github.com/syncthing/syncthing/lib/discover"
  18. "github.com/syncthing/syncthing/lib/model"
  19. "github.com/syncthing/syncthing/lib/osutil"
  20. "github.com/syncthing/syncthing/lib/sync"
  21. "github.com/thejerf/suture"
  22. )
  23. func NewSvc(cfg *config.Wrapper, tlsCfg *tls.Config) *Svc {
  24. conns := make(chan model.IntermediateConnection)
  25. svc := &Svc{
  26. Supervisor: suture.New("Svc", suture.Spec{
  27. Log: func(log string) {
  28. if debug {
  29. l.Debugln(log)
  30. }
  31. },
  32. FailureBackoff: 5 * time.Minute,
  33. FailureDecay: float64((10 * time.Minute) / time.Second),
  34. FailureThreshold: 5,
  35. }),
  36. cfg: cfg,
  37. tlsCfg: tlsCfg,
  38. tokens: make(map[string]suture.ServiceToken),
  39. clients: make(map[string]*client.ProtocolClient),
  40. mut: sync.NewRWMutex(),
  41. invitations: make(chan protocol.SessionInvitation),
  42. conns: conns,
  43. }
  44. rcfg := cfg.Raw()
  45. svc.CommitConfiguration(rcfg, rcfg)
  46. cfg.Subscribe(svc)
  47. receiver := &invitationReceiver{
  48. tlsCfg: tlsCfg,
  49. conns: conns,
  50. invitations: svc.invitations,
  51. stop: make(chan struct{}),
  52. }
  53. svc.Add(receiver)
  54. return svc
  55. }
  56. type Svc struct {
  57. *suture.Supervisor
  58. cfg *config.Wrapper
  59. tlsCfg *tls.Config
  60. tokens map[string]suture.ServiceToken
  61. clients map[string]*client.ProtocolClient
  62. mut sync.RWMutex
  63. invitations chan protocol.SessionInvitation
  64. conns chan model.IntermediateConnection
  65. }
  66. func (s *Svc) VerifyConfiguration(from, to config.Configuration) error {
  67. for _, addr := range to.Options.RelayServers {
  68. _, err := url.Parse(addr)
  69. if err != nil {
  70. return err
  71. }
  72. }
  73. return nil
  74. }
  75. func (s *Svc) CommitConfiguration(from, to config.Configuration) bool {
  76. existing := make(map[string]*url.URL, len(to.Options.RelayServers))
  77. for _, addr := range to.Options.RelayServers {
  78. uri, err := url.Parse(addr)
  79. if err != nil {
  80. if debug {
  81. l.Debugln("Failed to parse relay address", addr, err)
  82. }
  83. continue
  84. }
  85. existing[uri.String()] = uri
  86. }
  87. // Query dynamic addresses, and pick the closest relay from the ones they provide.
  88. for key, uri := range existing {
  89. if uri.Scheme != "dynamic+http" && uri.Scheme != "dynamic+https" {
  90. continue
  91. }
  92. delete(existing, key)
  93. // Trim off the `dynamic+` prefix
  94. uri.Scheme = uri.Scheme[8:]
  95. if debug {
  96. l.Debugln("Looking up dynamic relays from", uri)
  97. }
  98. data, err := http.Get(uri.String())
  99. if err != nil {
  100. if debug {
  101. l.Debugln("Failed to lookup dynamic relays", err)
  102. }
  103. continue
  104. }
  105. var ann dynamicAnnouncement
  106. err = json.NewDecoder(data.Body).Decode(&ann)
  107. data.Body.Close()
  108. if err != nil {
  109. if debug {
  110. l.Debugln("Failed to lookup dynamic relays", err)
  111. }
  112. continue
  113. }
  114. dynRelays := make([]discover.Relay, 0, len(ann.Relays))
  115. for _, relayAnn := range ann.Relays {
  116. ruri, err := url.Parse(relayAnn.URL)
  117. if err != nil {
  118. if debug {
  119. l.Debugln("Failed to parse dynamic relay address", relayAnn.URL, err)
  120. }
  121. continue
  122. }
  123. if debug {
  124. l.Debugln("Found", ruri, "via", uri)
  125. }
  126. dynRelays = append(dynRelays, discover.Relay{
  127. Address: ruri.String(),
  128. })
  129. }
  130. dynRelayAddrs := discover.RelayAddressesSortedByLatency(dynRelays)
  131. if len(dynRelayAddrs) > 0 {
  132. closestRelay := dynRelayAddrs[0]
  133. if debug {
  134. l.Debugln("Picking", closestRelay, "as closest dynamic relay from", uri)
  135. }
  136. ruri, _ := url.Parse(closestRelay)
  137. existing[closestRelay] = ruri
  138. } else if debug {
  139. l.Debugln("No dynamic relay found on", uri)
  140. }
  141. }
  142. s.mut.Lock()
  143. for key, uri := range existing {
  144. _, ok := s.tokens[key]
  145. if !ok {
  146. if debug {
  147. l.Debugln("Connecting to relay", uri)
  148. }
  149. c := client.NewProtocolClient(uri, s.tlsCfg.Certificates, s.invitations)
  150. s.tokens[key] = s.Add(c)
  151. s.clients[key] = c
  152. }
  153. }
  154. for key, token := range s.tokens {
  155. _, ok := existing[key]
  156. if !ok {
  157. err := s.Remove(token)
  158. delete(s.tokens, key)
  159. delete(s.clients, key)
  160. if debug {
  161. l.Debugln("Disconnecting from relay", key, err)
  162. }
  163. }
  164. }
  165. s.mut.Unlock()
  166. return true
  167. }
  168. func (s *Svc) ClientStatus() map[string]bool {
  169. if s == nil {
  170. // A nil client does not have a status, really. Yet we may be called
  171. // this way, for raisins...
  172. return nil
  173. }
  174. s.mut.RLock()
  175. status := make(map[string]bool, len(s.clients))
  176. for uri, client := range s.clients {
  177. status[uri] = client.StatusOK()
  178. }
  179. s.mut.RUnlock()
  180. return status
  181. }
  182. func (s *Svc) Accept() model.IntermediateConnection {
  183. return <-s.conns
  184. }
  185. type invitationReceiver struct {
  186. invitations chan protocol.SessionInvitation
  187. tlsCfg *tls.Config
  188. conns chan<- model.IntermediateConnection
  189. stop chan struct{}
  190. }
  191. func (r *invitationReceiver) Serve() {
  192. for {
  193. select {
  194. case inv := <-r.invitations:
  195. if debug {
  196. l.Debugln("Received relay invitation", inv)
  197. }
  198. conn, err := client.JoinSession(inv)
  199. if err != nil {
  200. if debug {
  201. l.Debugf("Failed to join relay session %s: %v", inv, err)
  202. }
  203. continue
  204. }
  205. err = osutil.SetTCPOptions(conn.(*net.TCPConn))
  206. if err != nil {
  207. l.Infoln(err)
  208. }
  209. var tc *tls.Conn
  210. if inv.ServerSocket {
  211. tc = tls.Server(conn, r.tlsCfg)
  212. } else {
  213. tc = tls.Client(conn, r.tlsCfg)
  214. }
  215. err = tc.Handshake()
  216. if err != nil {
  217. l.Infof("TLS handshake (BEP/relay %s): %v", inv, err)
  218. tc.Close()
  219. continue
  220. }
  221. r.conns <- model.IntermediateConnection{
  222. tc, model.ConnectionTypeRelayAccept,
  223. }
  224. case <-r.stop:
  225. return
  226. }
  227. }
  228. }
  229. func (r *invitationReceiver) Stop() {
  230. close(r.stop)
  231. }
  232. // This is the announcement recieved from the relay server;
  233. // {"relays": [{"url": "relay://10.20.30.40:5060"}, ...]}
  234. type dynamicAnnouncement struct {
  235. Relays []struct {
  236. URL string
  237. }
  238. }