mirror of
https://github.com/octoleo/syncthing.git
synced 2024-11-10 07:11:08 +00:00
commit
efa0a06947
@ -14,27 +14,6 @@ import (
|
||||
"github.com/syncthing/relaysrv/protocol"
|
||||
)
|
||||
|
||||
func NewProtocolClient(uri *url.URL, certs []tls.Certificate, invitations chan protocol.SessionInvitation) ProtocolClient {
|
||||
closeInvitationsOnFinish := false
|
||||
if invitations == nil {
|
||||
closeInvitationsOnFinish = true
|
||||
invitations = make(chan protocol.SessionInvitation)
|
||||
}
|
||||
return ProtocolClient{
|
||||
URI: uri,
|
||||
Invitations: invitations,
|
||||
|
||||
closeInvitationsOnFinish: closeInvitationsOnFinish,
|
||||
|
||||
config: configForCerts(certs),
|
||||
|
||||
timeout: time.Minute * 2,
|
||||
|
||||
stop: make(chan struct{}),
|
||||
stopped: make(chan struct{}),
|
||||
}
|
||||
}
|
||||
|
||||
type ProtocolClient struct {
|
||||
URI *url.URL
|
||||
Invitations chan protocol.SessionInvitation
|
||||
@ -51,6 +30,129 @@ type ProtocolClient struct {
|
||||
conn *tls.Conn
|
||||
}
|
||||
|
||||
func NewProtocolClient(uri *url.URL, certs []tls.Certificate, invitations chan protocol.SessionInvitation) *ProtocolClient {
|
||||
closeInvitationsOnFinish := false
|
||||
if invitations == nil {
|
||||
closeInvitationsOnFinish = true
|
||||
invitations = make(chan protocol.SessionInvitation)
|
||||
}
|
||||
|
||||
return &ProtocolClient{
|
||||
URI: uri,
|
||||
Invitations: invitations,
|
||||
|
||||
closeInvitationsOnFinish: closeInvitationsOnFinish,
|
||||
|
||||
config: configForCerts(certs),
|
||||
|
||||
timeout: time.Minute * 2,
|
||||
|
||||
stop: make(chan struct{}),
|
||||
stopped: make(chan struct{}),
|
||||
}
|
||||
}
|
||||
|
||||
func (c *ProtocolClient) Serve() {
|
||||
c.stop = make(chan struct{})
|
||||
c.stopped = make(chan struct{})
|
||||
defer close(c.stopped)
|
||||
|
||||
if err := c.connect(); err != nil {
|
||||
l.Infoln("Relay connect:", err)
|
||||
return
|
||||
}
|
||||
|
||||
if debug {
|
||||
l.Debugln(c, "connected", c.conn.RemoteAddr())
|
||||
}
|
||||
|
||||
if err := c.join(); err != nil {
|
||||
c.conn.Close()
|
||||
l.Infoln("Relay join:", err)
|
||||
return
|
||||
}
|
||||
|
||||
if err := c.conn.SetDeadline(time.Time{}); err != nil {
|
||||
l.Infoln("Relay set deadline:", err)
|
||||
return
|
||||
}
|
||||
|
||||
if debug {
|
||||
l.Debugln(c, "joined", c.conn.RemoteAddr(), "via", c.conn.LocalAddr())
|
||||
}
|
||||
|
||||
defer c.cleanup()
|
||||
|
||||
messages := make(chan interface{})
|
||||
errors := make(chan error, 1)
|
||||
|
||||
go messageReader(c.conn, messages, errors)
|
||||
|
||||
timeout := time.NewTimer(c.timeout)
|
||||
|
||||
for {
|
||||
select {
|
||||
case message := <-messages:
|
||||
timeout.Reset(c.timeout)
|
||||
if debug {
|
||||
log.Printf("%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)
|
||||
return
|
||||
|
||||
}
|
||||
if debug {
|
||||
l.Debugln(c, "sent pong")
|
||||
}
|
||||
|
||||
case protocol.SessionInvitation:
|
||||
ip := net.IP(msg.Address)
|
||||
if len(ip) == 0 || ip.IsUnspecified() {
|
||||
msg.Address = c.conn.RemoteAddr().(*net.TCPAddr).IP[:]
|
||||
}
|
||||
c.Invitations <- msg
|
||||
|
||||
default:
|
||||
l.Infoln("Relay: protocol error: unexpected message %v", msg)
|
||||
return
|
||||
}
|
||||
|
||||
case <-c.stop:
|
||||
if debug {
|
||||
l.Debugln(c, "stopping")
|
||||
}
|
||||
return
|
||||
|
||||
case err := <-errors:
|
||||
l.Infoln("Relay received:", err)
|
||||
return
|
||||
|
||||
case <-timeout.C:
|
||||
if debug {
|
||||
l.Debugln(c, "timed out")
|
||||
}
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (c *ProtocolClient) Stop() {
|
||||
if c.stop == nil {
|
||||
return
|
||||
}
|
||||
|
||||
close(c.stop)
|
||||
<-c.stopped
|
||||
}
|
||||
|
||||
func (c *ProtocolClient) String() string {
|
||||
return fmt.Sprintf("ProtocolClient@%p", c)
|
||||
}
|
||||
|
||||
func (c *ProtocolClient) connect() error {
|
||||
if c.URI.Scheme != "relay" {
|
||||
return fmt.Errorf("Unsupported relay schema:", c.URI.Scheme)
|
||||
@ -61,9 +163,13 @@ func (c *ProtocolClient) connect() error {
|
||||
return err
|
||||
}
|
||||
|
||||
conn.SetDeadline(time.Now().Add(10 * time.Second))
|
||||
if err := conn.SetDeadline(time.Now().Add(10 * time.Second)); err != nil {
|
||||
conn.Close()
|
||||
return err
|
||||
}
|
||||
|
||||
if err := performHandshakeAndValidation(conn, c.URI); err != nil {
|
||||
conn.Close()
|
||||
return err
|
||||
}
|
||||
|
||||
@ -71,101 +177,6 @@ func (c *ProtocolClient) connect() error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *ProtocolClient) Serve() {
|
||||
if err := c.connect(); err != nil {
|
||||
panic(err)
|
||||
}
|
||||
|
||||
if debug {
|
||||
l.Debugln(c, "connected", c.conn.RemoteAddr())
|
||||
}
|
||||
|
||||
if err := c.join(); err != nil {
|
||||
c.conn.Close()
|
||||
panic(err)
|
||||
}
|
||||
|
||||
c.conn.SetDeadline(time.Time{})
|
||||
|
||||
if debug {
|
||||
l.Debugln(c, "joined", c.conn.RemoteAddr(), "via", c.conn.LocalAddr())
|
||||
}
|
||||
|
||||
c.stop = make(chan struct{})
|
||||
c.stopped = make(chan struct{})
|
||||
|
||||
defer c.cleanup()
|
||||
|
||||
messages := make(chan interface{})
|
||||
errors := make(chan error, 1)
|
||||
|
||||
go func(conn net.Conn, message chan<- interface{}, errors chan<- error) {
|
||||
for {
|
||||
msg, err := protocol.ReadMessage(conn)
|
||||
if err != nil {
|
||||
errors <- err
|
||||
return
|
||||
}
|
||||
messages <- msg
|
||||
}
|
||||
}(c.conn, messages, errors)
|
||||
|
||||
timeout := time.NewTimer(c.timeout)
|
||||
for {
|
||||
select {
|
||||
case message := <-messages:
|
||||
timeout.Reset(c.timeout)
|
||||
if debug {
|
||||
log.Printf("%s received message %T", c, message)
|
||||
}
|
||||
switch msg := message.(type) {
|
||||
case protocol.Ping:
|
||||
if err := protocol.WriteMessage(c.conn, protocol.Pong{}); err != nil {
|
||||
panic(err)
|
||||
}
|
||||
if debug {
|
||||
l.Debugln(c, "sent pong")
|
||||
}
|
||||
case protocol.SessionInvitation:
|
||||
ip := net.IP(msg.Address)
|
||||
if len(ip) == 0 || ip.IsUnspecified() {
|
||||
msg.Address = c.conn.RemoteAddr().(*net.TCPAddr).IP[:]
|
||||
}
|
||||
c.Invitations <- msg
|
||||
default:
|
||||
panic(fmt.Errorf("protocol error: unexpected message %v", msg))
|
||||
}
|
||||
case <-c.stop:
|
||||
if debug {
|
||||
l.Debugln(c, "stopping")
|
||||
}
|
||||
break
|
||||
case err := <-errors:
|
||||
panic(err)
|
||||
case <-timeout.C:
|
||||
if debug {
|
||||
l.Debugln(c, "timed out")
|
||||
}
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
c.stopped <- struct{}{}
|
||||
}
|
||||
|
||||
func (c *ProtocolClient) Stop() {
|
||||
if c.stop == nil {
|
||||
return
|
||||
}
|
||||
|
||||
c.stop <- struct{}{}
|
||||
<-c.stopped
|
||||
}
|
||||
|
||||
func (c *ProtocolClient) String() string {
|
||||
return fmt.Sprintf("ProtocolClient@%p", c)
|
||||
}
|
||||
|
||||
func (c *ProtocolClient) cleanup() {
|
||||
if c.closeInvitationsOnFinish {
|
||||
close(c.Invitations)
|
||||
@ -176,24 +187,11 @@ func (c *ProtocolClient) cleanup() {
|
||||
l.Debugln(c, "cleaning up")
|
||||
}
|
||||
|
||||
if c.stop != nil {
|
||||
close(c.stop)
|
||||
c.stop = nil
|
||||
}
|
||||
|
||||
if c.stopped != nil {
|
||||
close(c.stopped)
|
||||
c.stopped = nil
|
||||
}
|
||||
|
||||
if c.conn != nil {
|
||||
c.conn.Close()
|
||||
}
|
||||
c.conn.Close()
|
||||
}
|
||||
|
||||
func (c *ProtocolClient) join() error {
|
||||
err := protocol.WriteMessage(c.conn, protocol.JoinRelayRequest{})
|
||||
if err != nil {
|
||||
if err := protocol.WriteMessage(c.conn, protocol.JoinRelayRequest{}); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
@ -207,6 +205,7 @@ func (c *ProtocolClient) join() error {
|
||||
if msg.Code != 0 {
|
||||
return fmt.Errorf("Incorrect response code %d: %s", msg.Code, msg.Message)
|
||||
}
|
||||
|
||||
default:
|
||||
return fmt.Errorf("protocol error: expecting response got %v", msg)
|
||||
}
|
||||
@ -215,15 +214,12 @@ func (c *ProtocolClient) join() error {
|
||||
}
|
||||
|
||||
func performHandshakeAndValidation(conn *tls.Conn, uri *url.URL) error {
|
||||
err := conn.Handshake()
|
||||
if err != nil {
|
||||
conn.Close()
|
||||
if err := conn.Handshake(); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
cs := conn.ConnectionState()
|
||||
if !cs.NegotiatedProtocolIsMutual || cs.NegotiatedProtocol != protocol.ProtocolName {
|
||||
conn.Close()
|
||||
return fmt.Errorf("protocol negotiation error")
|
||||
}
|
||||
|
||||
@ -232,22 +228,30 @@ func performHandshakeAndValidation(conn *tls.Conn, uri *url.URL) error {
|
||||
if relayIDs != "" {
|
||||
relayID, err := syncthingprotocol.DeviceIDFromString(relayIDs)
|
||||
if err != nil {
|
||||
conn.Close()
|
||||
return fmt.Errorf("relay address contains invalid verification id: %s", err)
|
||||
}
|
||||
|
||||
certs := cs.PeerCertificates
|
||||
if cl := len(certs); cl != 1 {
|
||||
conn.Close()
|
||||
return fmt.Errorf("unexpected certificate count: %d", cl)
|
||||
}
|
||||
|
||||
remoteID := syncthingprotocol.NewDeviceID(certs[0].Raw)
|
||||
if remoteID != relayID {
|
||||
conn.Close()
|
||||
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
|
||||
}
|
||||
}
|
||||
|
@ -27,7 +27,6 @@ func protocolListener(addr string, config *tls.Config) {
|
||||
|
||||
for {
|
||||
conn, err := listener.Accept()
|
||||
setTCPOptions(conn)
|
||||
if err != nil {
|
||||
if debug {
|
||||
log.Println(err)
|
||||
@ -35,6 +34,8 @@ func protocolListener(addr string, config *tls.Config) {
|
||||
continue
|
||||
}
|
||||
|
||||
setTCPOptions(conn)
|
||||
|
||||
if debug {
|
||||
log.Println("Protocol listener accepted connection from", conn.RemoteAddr())
|
||||
}
|
||||
@ -74,16 +75,12 @@ func protocolConnectionHandler(tcpConn net.Conn, config *tls.Config) {
|
||||
errors := make(chan error, 1)
|
||||
outbox := make(chan interface{})
|
||||
|
||||
go func(conn net.Conn, message chan<- interface{}, errors chan<- error) {
|
||||
for {
|
||||
msg, err := protocol.ReadMessage(conn)
|
||||
if err != nil {
|
||||
errors <- err
|
||||
return
|
||||
}
|
||||
messages <- msg
|
||||
}
|
||||
}(conn, messages, errors)
|
||||
// Read messages from the connection and send them on the messages
|
||||
// channel. When there is an error, send it on the error channel and
|
||||
// return. Applies also when the connection gets closed, so the pattern
|
||||
// below is to close the connection on error, then wait for the error
|
||||
// signal from messageReader to exit.
|
||||
go messageReader(conn, messages, errors)
|
||||
|
||||
pingTicker := time.NewTicker(pingInterval)
|
||||
timeoutTicker := time.NewTimer(networkTimeout)
|
||||
@ -96,6 +93,7 @@ func protocolConnectionHandler(tcpConn net.Conn, config *tls.Config) {
|
||||
if debug {
|
||||
log.Printf("Message %T from %s", message, id)
|
||||
}
|
||||
|
||||
switch msg := message.(type) {
|
||||
case protocol.JoinRelayRequest:
|
||||
outboxesMut.RLock()
|
||||
@ -116,6 +114,7 @@ func protocolConnectionHandler(tcpConn net.Conn, config *tls.Config) {
|
||||
joined = true
|
||||
|
||||
protocol.WriteMessage(conn, protocol.ResponseSuccess)
|
||||
|
||||
case protocol.ConnectRequest:
|
||||
requestedPeer := syncthingprotocol.DeviceIDFromBytes(msg.ID)
|
||||
outboxesMut.RLock()
|
||||
@ -151,7 +150,10 @@ func protocolConnectionHandler(tcpConn net.Conn, config *tls.Config) {
|
||||
log.Println("Sent invitation from", id, "to", requestedPeer)
|
||||
}
|
||||
conn.Close()
|
||||
|
||||
case protocol.Pong:
|
||||
// Nothing
|
||||
|
||||
default:
|
||||
if debug {
|
||||
log.Printf("Unknown message %s: %T", id, message)
|
||||
@ -159,21 +161,25 @@ func protocolConnectionHandler(tcpConn net.Conn, config *tls.Config) {
|
||||
protocol.WriteMessage(conn, protocol.ResponseUnexpectedMessage)
|
||||
conn.Close()
|
||||
}
|
||||
|
||||
case err := <-errors:
|
||||
if debug {
|
||||
log.Printf("Closing connection %s: %s", id, err)
|
||||
}
|
||||
// Potentially closing a second time.
|
||||
close(outbox)
|
||||
|
||||
// Potentially closing a second time.
|
||||
conn.Close()
|
||||
// Only delete the outbox if the client join, as it migth be a
|
||||
// lookup request coming from the same client.
|
||||
|
||||
// Only delete the outbox if the client is joined, as it might be
|
||||
// a lookup request coming from the same client.
|
||||
if joined {
|
||||
outboxesMut.Lock()
|
||||
delete(outboxes, id)
|
||||
outboxesMut.Unlock()
|
||||
}
|
||||
return
|
||||
|
||||
case <-pingTicker.C:
|
||||
if !joined {
|
||||
if debug {
|
||||
@ -189,6 +195,7 @@ func protocolConnectionHandler(tcpConn net.Conn, config *tls.Config) {
|
||||
}
|
||||
conn.Close()
|
||||
}
|
||||
|
||||
case <-timeoutTicker.C:
|
||||
// We should receive a error from the reader loop, which will cause
|
||||
// us to quit this loop.
|
||||
@ -196,6 +203,7 @@ func protocolConnectionHandler(tcpConn net.Conn, config *tls.Config) {
|
||||
log.Printf("%s timed out", id)
|
||||
}
|
||||
conn.Close()
|
||||
|
||||
case msg := <-outbox:
|
||||
if debug {
|
||||
log.Printf("Sending message %T to %s", msg, id)
|
||||
@ -209,3 +217,14 @@ func protocolConnectionHandler(tcpConn net.Conn, config *tls.Config) {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
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
|
||||
}
|
||||
}
|
||||
|
@ -18,7 +18,6 @@ func sessionListener(addr string) {
|
||||
|
||||
for {
|
||||
conn, err := listener.Accept()
|
||||
setTCPOptions(conn)
|
||||
if err != nil {
|
||||
if debug {
|
||||
log.Println(err)
|
||||
@ -26,6 +25,8 @@ func sessionListener(addr string) {
|
||||
continue
|
||||
}
|
||||
|
||||
setTCPOptions(conn)
|
||||
|
||||
if debug {
|
||||
log.Println("Session listener accepted connection from", conn.RemoteAddr())
|
||||
}
|
||||
@ -35,10 +36,17 @@ func sessionListener(addr string) {
|
||||
}
|
||||
|
||||
func sessionConnectionHandler(conn net.Conn) {
|
||||
conn.SetDeadline(time.Now().Add(messageTimeout))
|
||||
defer conn.Close()
|
||||
|
||||
if err := conn.SetDeadline(time.Now().Add(messageTimeout)); err != nil {
|
||||
if debug {
|
||||
log.Println("Weird error setting deadline:", err, "on", conn.RemoteAddr())
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
message, err := protocol.ReadMessage(conn)
|
||||
if err != nil {
|
||||
conn.Close()
|
||||
return
|
||||
}
|
||||
|
||||
@ -51,7 +59,6 @@ func sessionConnectionHandler(conn net.Conn) {
|
||||
|
||||
if ses == nil {
|
||||
protocol.WriteMessage(conn, protocol.ResponseNotFound)
|
||||
conn.Close()
|
||||
return
|
||||
}
|
||||
|
||||
@ -60,24 +67,26 @@ func sessionConnectionHandler(conn net.Conn) {
|
||||
log.Println("Failed to add", conn.RemoteAddr(), "to session", ses)
|
||||
}
|
||||
protocol.WriteMessage(conn, protocol.ResponseAlreadyConnected)
|
||||
conn.Close()
|
||||
return
|
||||
}
|
||||
|
||||
err := protocol.WriteMessage(conn, protocol.ResponseSuccess)
|
||||
if err != nil {
|
||||
if err := protocol.WriteMessage(conn, protocol.ResponseSuccess); err != nil {
|
||||
if debug {
|
||||
log.Println("Failed to send session join response to ", conn.RemoteAddr(), "for", ses)
|
||||
}
|
||||
conn.Close()
|
||||
return
|
||||
}
|
||||
conn.SetDeadline(time.Time{})
|
||||
|
||||
if err := conn.SetDeadline(time.Time{}); err != nil {
|
||||
if debug {
|
||||
log.Println("Weird error setting deadline:", err, "on", conn.RemoteAddr())
|
||||
}
|
||||
return
|
||||
}
|
||||
default:
|
||||
if debug {
|
||||
log.Println("Unexpected message from", conn.RemoteAddr(), message)
|
||||
}
|
||||
protocol.WriteMessage(conn, protocol.ResponseUnexpectedMessage)
|
||||
conn.Close()
|
||||
}
|
||||
}
|
||||
|
Loading…
Reference in New Issue
Block a user