123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339 |
- // Copyright (C) 2015 Audrius Butkevicius and Contributors (see the CONTRIBUTORS file).
- package client
- import (
- "crypto/tls"
- "fmt"
- "net"
- "net/url"
- "time"
- "github.com/syncthing/syncthing/lib/dialer"
- syncthingprotocol "github.com/syncthing/syncthing/lib/protocol"
- "github.com/syncthing/syncthing/lib/relay/protocol"
- "github.com/syncthing/syncthing/lib/sync"
- )
- type staticClient struct {
- uri *url.URL
- invitations chan protocol.SessionInvitation
- closeInvitationsOnFinish bool
- config *tls.Config
- messageTimeout time.Duration
- connectTimeout time.Duration
- stop chan struct{}
- stopped chan struct{}
- conn *tls.Conn
- mut sync.RWMutex
- err error
- connected bool
- latency time.Duration
- }
- func newStaticClient(uri *url.URL, certs []tls.Certificate, invitations chan protocol.SessionInvitation, timeout time.Duration) RelayClient {
- closeInvitationsOnFinish := false
- if invitations == nil {
- closeInvitationsOnFinish = true
- invitations = make(chan protocol.SessionInvitation)
- }
- return &staticClient{
- uri: uri,
- invitations: invitations,
- closeInvitationsOnFinish: closeInvitationsOnFinish,
- config: configForCerts(certs),
- messageTimeout: time.Minute * 2,
- connectTimeout: timeout,
- stop: make(chan struct{}),
- stopped: make(chan struct{}),
- mut: sync.NewRWMutex(),
- }
- }
- func (c *staticClient) Serve() {
- defer c.cleanup()
- c.stop = make(chan struct{})
- c.stopped = make(chan struct{})
- defer close(c.stopped)
- if err := c.connect(); err != nil {
- l.Infof("Could not connect to relay %s: %s", c.uri, err)
- c.setError(err)
- return
- }
- l.Debugln(c, "connected", c.conn.RemoteAddr())
- if err := c.join(); err != nil {
- c.conn.Close()
- l.Infof("Could not join relay %s: %s", c.uri, err)
- c.setError(err)
- return
- }
- if err := c.conn.SetDeadline(time.Time{}); err != nil {
- c.conn.Close()
- l.Infoln("Relay set deadline:", err)
- c.setError(err)
- return
- }
- l.Infof("Joined relay %s://%s", c.uri.Scheme, c.uri.Host)
- defer l.Infof("Disconnected from relay %s://%s", c.uri.Scheme, c.uri.Host)
- c.mut.Lock()
- c.connected = true
- c.mut.Unlock()
- messages := make(chan interface{})
- errors := make(chan error, 1)
- go messageReader(c.conn, messages, errors)
- timeout := time.NewTimer(c.messageTimeout)
- for {
- select {
- case message := <-messages:
- timeout.Reset(c.messageTimeout)
- l.Debugf("%s received message %T", c, message)
- switch msg := message.(type) {
- case protocol.Ping:
- if err := protocol.WriteMessage(c.conn, protocol.Pong{}); err != nil {
- l.Infoln("Relay write:", err)
- c.setError(err)
- c.disconnect()
- } else {
- l.Debugln(c, "sent pong")
- }
- case protocol.SessionInvitation:
- ip := net.IP(msg.Address)
- if len(ip) == 0 || ip.IsUnspecified() {
- msg.Address = remoteIPBytes(c.conn)
- }
- c.invitations <- msg
- case protocol.RelayFull:
- l.Infof("Disconnected from relay %s due to it becoming full.", c.uri)
- c.setError(fmt.Errorf("Relay full"))
- c.disconnect()
- default:
- l.Infoln("Relay: protocol error: unexpected message %v", msg)
- c.setError(fmt.Errorf("protocol error: unexpected message %v", msg))
- c.disconnect()
- }
- case <-c.stop:
- l.Debugln(c, "stopping")
- c.setError(nil)
- c.disconnect()
- // We always exit via this branch of the select, to make sure the
- // the reader routine exits.
- case err := <-errors:
- close(errors)
- close(messages)
- c.mut.Lock()
- if c.connected {
- c.conn.Close()
- c.connected = false
- l.Infof("Disconnecting from relay %s due to error: %s", c.uri, err)
- c.err = err
- } else {
- c.err = nil
- }
- c.mut.Unlock()
- return
- case <-timeout.C:
- l.Debugln(c, "timed out")
- c.disconnect()
- c.setError(fmt.Errorf("timed out"))
- }
- }
- }
- func (c *staticClient) Stop() {
- if c.stop == nil {
- return
- }
- close(c.stop)
- <-c.stopped
- }
- func (c *staticClient) StatusOK() bool {
- c.mut.RLock()
- con := c.connected
- c.mut.RUnlock()
- return con
- }
- func (c *staticClient) Latency() time.Duration {
- c.mut.RLock()
- lat := c.latency
- c.mut.RUnlock()
- return lat
- }
- func (c *staticClient) String() string {
- return fmt.Sprintf("StaticClient:%p@%s", c, c.URI())
- }
- func (c *staticClient) URI() *url.URL {
- return c.uri
- }
- func (c *staticClient) Invitations() chan protocol.SessionInvitation {
- c.mut.RLock()
- inv := c.invitations
- c.mut.RUnlock()
- return inv
- }
- func (c *staticClient) cleanup() {
- c.mut.Lock()
- if c.closeInvitationsOnFinish {
- close(c.invitations)
- c.invitations = make(chan protocol.SessionInvitation)
- }
- c.mut.Unlock()
- }
- func (c *staticClient) connect() error {
- if c.uri.Scheme != "relay" {
- return fmt.Errorf("Unsupported relay schema: %v", c.uri.Scheme)
- }
- t0 := time.Now()
- tcpConn, err := dialer.DialTimeout("tcp", c.uri.Host, c.connectTimeout)
- if err != nil {
- return err
- }
- c.mut.Lock()
- c.latency = time.Since(t0)
- c.mut.Unlock()
- conn := tls.Client(tcpConn, c.config)
- if err := conn.SetDeadline(time.Now().Add(c.connectTimeout)); err != nil {
- conn.Close()
- return err
- }
- if err := performHandshakeAndValidation(conn, c.uri); err != nil {
- conn.Close()
- return err
- }
- c.conn = conn
- return nil
- }
- func (c *staticClient) disconnect() {
- l.Debugln(c, "disconnecting")
- c.mut.Lock()
- c.connected = false
- c.mut.Unlock()
- c.conn.Close()
- }
- func (c *staticClient) setError(err error) {
- c.mut.Lock()
- c.err = err
- c.mut.Unlock()
- }
- func (c *staticClient) Error() error {
- c.mut.RLock()
- err := c.err
- c.mut.RUnlock()
- return err
- }
- func (c *staticClient) join() error {
- if err := protocol.WriteMessage(c.conn, protocol.JoinRelayRequest{}); err != nil {
- return err
- }
- message, err := protocol.ReadMessage(c.conn)
- if err != nil {
- return err
- }
- switch msg := message.(type) {
- case protocol.Response:
- if msg.Code != 0 {
- return fmt.Errorf("Incorrect response code %d: %s", msg.Code, msg.Message)
- }
- case protocol.RelayFull:
- return fmt.Errorf("relay full")
- default:
- return fmt.Errorf("protocol error: expecting response got %v", msg)
- }
- return nil
- }
- func performHandshakeAndValidation(conn *tls.Conn, uri *url.URL) error {
- if err := conn.Handshake(); err != nil {
- return err
- }
- cs := conn.ConnectionState()
- if !cs.NegotiatedProtocolIsMutual || cs.NegotiatedProtocol != protocol.ProtocolName {
- return fmt.Errorf("protocol negotiation error")
- }
- q := uri.Query()
- relayIDs := q.Get("id")
- if relayIDs != "" {
- relayID, err := syncthingprotocol.DeviceIDFromString(relayIDs)
- if err != nil {
- return fmt.Errorf("relay address contains invalid verification id: %s", err)
- }
- certs := cs.PeerCertificates
- if cl := len(certs); cl != 1 {
- return fmt.Errorf("unexpected certificate count: %d", cl)
- }
- remoteID := syncthingprotocol.NewDeviceID(certs[0].Raw)
- if remoteID != relayID {
- return fmt.Errorf("relay id does not match. Expected %v got %v", relayID, remoteID)
- }
- }
- return nil
- }
- func messageReader(conn net.Conn, messages chan<- interface{}, errors chan<- error) {
- for {
- msg, err := protocol.ReadMessage(conn)
- if err != nil {
- errors <- err
- return
- }
- messages <- msg
- }
- }
|