| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277 |
- // Copyright (C) 2015 The Syncthing Authors.
- //
- // This Source Code Form is subject to the terms of the Mozilla Public
- // License, v. 2.0. If a copy of the MPL was not distributed with this file,
- // You can obtain one at http://mozilla.org/MPL/2.0/.
- package relay
- import (
- "crypto/tls"
- "encoding/json"
- "net"
- "net/http"
- "net/url"
- "time"
- "github.com/syncthing/relaysrv/client"
- "github.com/syncthing/relaysrv/protocol"
- "github.com/syncthing/syncthing/lib/config"
- "github.com/syncthing/syncthing/lib/discover"
- "github.com/syncthing/syncthing/lib/model"
- "github.com/syncthing/syncthing/lib/osutil"
- "github.com/syncthing/syncthing/lib/sync"
- "github.com/thejerf/suture"
- )
- func NewSvc(cfg *config.Wrapper, tlsCfg *tls.Config) *Svc {
- conns := make(chan model.IntermediateConnection)
- svc := &Svc{
- Supervisor: suture.New("Svc", suture.Spec{
- Log: func(log string) {
- if debug {
- l.Debugln(log)
- }
- },
- FailureBackoff: 5 * time.Minute,
- FailureDecay: float64((10 * time.Minute) / time.Second),
- FailureThreshold: 5,
- }),
- cfg: cfg,
- tlsCfg: tlsCfg,
- tokens: make(map[string]suture.ServiceToken),
- clients: make(map[string]*client.ProtocolClient),
- mut: sync.NewRWMutex(),
- invitations: make(chan protocol.SessionInvitation),
- conns: conns,
- }
- rcfg := cfg.Raw()
- svc.CommitConfiguration(rcfg, rcfg)
- cfg.Subscribe(svc)
- receiver := &invitationReceiver{
- tlsCfg: tlsCfg,
- conns: conns,
- invitations: svc.invitations,
- stop: make(chan struct{}),
- }
- svc.Add(receiver)
- return svc
- }
- type Svc struct {
- *suture.Supervisor
- cfg *config.Wrapper
- tlsCfg *tls.Config
- tokens map[string]suture.ServiceToken
- clients map[string]*client.ProtocolClient
- mut sync.RWMutex
- invitations chan protocol.SessionInvitation
- conns chan model.IntermediateConnection
- }
- func (s *Svc) VerifyConfiguration(from, to config.Configuration) error {
- for _, addr := range to.Options.RelayServers {
- _, err := url.Parse(addr)
- if err != nil {
- return err
- }
- }
- return nil
- }
- func (s *Svc) CommitConfiguration(from, to config.Configuration) bool {
- existing := make(map[string]*url.URL, len(to.Options.RelayServers))
- for _, addr := range to.Options.RelayServers {
- uri, err := url.Parse(addr)
- if err != nil {
- if debug {
- l.Debugln("Failed to parse relay address", addr, err)
- }
- continue
- }
- existing[uri.String()] = uri
- }
- // Query dynamic addresses, and pick the closest relay from the ones they provide.
- for key, uri := range existing {
- if uri.Scheme != "dynamic+http" && uri.Scheme != "dynamic+https" {
- continue
- }
- delete(existing, key)
- // Trim off the `dynamic+` prefix
- uri.Scheme = uri.Scheme[8:]
- if debug {
- l.Debugln("Looking up dynamic relays from", uri)
- }
- data, err := http.Get(uri.String())
- if err != nil {
- if debug {
- l.Debugln("Failed to lookup dynamic relays", err)
- }
- continue
- }
- var ann dynamicAnnouncement
- err = json.NewDecoder(data.Body).Decode(&ann)
- data.Body.Close()
- if err != nil {
- if debug {
- l.Debugln("Failed to lookup dynamic relays", err)
- }
- continue
- }
- dynRelays := make([]discover.Relay, 0, len(ann.Relays))
- for _, relayAnn := range ann.Relays {
- ruri, err := url.Parse(relayAnn.URL)
- if err != nil {
- if debug {
- l.Debugln("Failed to parse dynamic relay address", relayAnn.URL, err)
- }
- continue
- }
- if debug {
- l.Debugln("Found", ruri, "via", uri)
- }
- dynRelays = append(dynRelays, discover.Relay{
- Address: ruri.String(),
- })
- }
- dynRelayAddrs := discover.RelayAddressesSortedByLatency(dynRelays)
- if len(dynRelayAddrs) > 0 {
- closestRelay := dynRelayAddrs[0]
- if debug {
- l.Debugln("Picking", closestRelay, "as closest dynamic relay from", uri)
- }
- ruri, _ := url.Parse(closestRelay)
- existing[closestRelay] = ruri
- } else if debug {
- l.Debugln("No dynamic relay found on", uri)
- }
- }
- s.mut.Lock()
- for key, uri := range existing {
- _, ok := s.tokens[key]
- if !ok {
- if debug {
- l.Debugln("Connecting to relay", uri)
- }
- c := client.NewProtocolClient(uri, s.tlsCfg.Certificates, s.invitations)
- s.tokens[key] = s.Add(c)
- s.clients[key] = c
- }
- }
- for key, token := range s.tokens {
- _, ok := existing[key]
- if !ok {
- err := s.Remove(token)
- delete(s.tokens, key)
- delete(s.clients, key)
- if debug {
- l.Debugln("Disconnecting from relay", key, err)
- }
- }
- }
- s.mut.Unlock()
- return true
- }
- func (s *Svc) ClientStatus() map[string]bool {
- if s == nil {
- // A nil client does not have a status, really. Yet we may be called
- // this way, for raisins...
- return nil
- }
- s.mut.RLock()
- status := make(map[string]bool, len(s.clients))
- for uri, client := range s.clients {
- status[uri] = client.StatusOK()
- }
- s.mut.RUnlock()
- return status
- }
- func (s *Svc) Accept() model.IntermediateConnection {
- return <-s.conns
- }
- type invitationReceiver struct {
- invitations chan protocol.SessionInvitation
- tlsCfg *tls.Config
- conns chan<- model.IntermediateConnection
- stop chan struct{}
- }
- func (r *invitationReceiver) Serve() {
- for {
- select {
- case inv := <-r.invitations:
- if debug {
- l.Debugln("Received relay invitation", inv)
- }
- conn, err := client.JoinSession(inv)
- if err != nil {
- if debug {
- l.Debugf("Failed to join relay session %s: %v", inv, err)
- }
- continue
- }
- err = osutil.SetTCPOptions(conn.(*net.TCPConn))
- if err != nil {
- l.Infoln(err)
- }
- var tc *tls.Conn
- if inv.ServerSocket {
- tc = tls.Server(conn, r.tlsCfg)
- } else {
- tc = tls.Client(conn, r.tlsCfg)
- }
- err = tc.Handshake()
- if err != nil {
- l.Infof("TLS handshake (BEP/relay %s): %v", inv, err)
- tc.Close()
- continue
- }
- r.conns <- model.IntermediateConnection{
- tc, model.ConnectionTypeRelayAccept,
- }
- case <-r.stop:
- return
- }
- }
- }
- func (r *invitationReceiver) Stop() {
- close(r.stop)
- }
- // This is the announcement recieved from the relay server;
- // {"relays": [{"url": "relay://10.20.30.40:5060"}, ...]}
- type dynamicAnnouncement struct {
- Relays []struct {
- URL string
- }
- }
|