syncthing/lib/relay/client/static.go

345 lines
7.2 KiB
Go
Raw Normal View History

// Copyright (C) 2015 Audrius Butkevicius and Contributors (see the CONTRIBUTORS file).
package client
import (
"crypto/tls"
"fmt"
"net"
"net/url"
"time"
2015-11-26 23:41:11 +00:00
"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
2015-11-23 21:14:46 +00:00
messageTimeout time.Duration
connectTimeout time.Duration
stop chan struct{}
stopped chan struct{}
stopMut sync.RWMutex
conn *tls.Conn
mut sync.RWMutex
err error
connected bool
latency time.Duration
}
2015-11-23 21:14:46 +00:00
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)
}
stopped := make(chan struct{})
close(stopped) // not yet started, don't block on Stop()
return &staticClient{
uri: uri,
invitations: invitations,
closeInvitationsOnFinish: closeInvitationsOnFinish,
config: configForCerts(certs),
2015-11-23 21:14:46 +00:00
messageTimeout: time.Minute * 2,
connectTimeout: timeout,
stop: make(chan struct{}),
stopped: stopped,
stopMut: sync.NewRWMutex(),
mut: sync.NewRWMutex(),
}
}
func (c *staticClient) Serve() {
defer c.cleanup()
c.stopMut.Lock()
c.stop = make(chan struct{})
c.stopped = make(chan struct{})
c.stopMut.Unlock()
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)
2015-11-23 21:14:46 +00:00
timeout := time.NewTimer(c.messageTimeout)
c.stopMut.RLock()
defer c.stopMut.RUnlock()
for {
select {
case message := <-messages:
2015-11-23 21:14:46 +00:00
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
2015-11-20 23:42:49 +00:00
case protocol.RelayFull:
l.Infof("Disconnected from relay %s due to it becoming full.", c.uri)
c.setError(fmt.Errorf("Relay full"))
c.disconnect()
2015-11-20 23:42:49 +00:00
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() {
c.stopMut.RLock()
close(c.stop)
<-c.stopped
c.stopMut.RUnlock()
}
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()
2015-11-26 23:41:11 +00:00
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)
2015-11-23 21:14:46 +00:00
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)
}
2015-11-20 23:42:49 +00:00
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
}
}