mirror of
https://github.com/octoleo/syncthing.git
synced 2024-12-23 03:18:59 +00:00
ea54525a33
GitHub-Pull-Request: https://github.com/syncthing/syncthing/pull/2991
639 lines
19 KiB
Go
639 lines
19 KiB
Go
// 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 connections
|
|
|
|
import (
|
|
"crypto/tls"
|
|
"encoding/binary"
|
|
"fmt"
|
|
"io"
|
|
"net"
|
|
"net/url"
|
|
"sync"
|
|
"time"
|
|
|
|
"github.com/juju/ratelimit"
|
|
"github.com/syncthing/syncthing/lib/config"
|
|
"github.com/syncthing/syncthing/lib/discover"
|
|
"github.com/syncthing/syncthing/lib/events"
|
|
"github.com/syncthing/syncthing/lib/model"
|
|
"github.com/syncthing/syncthing/lib/nat"
|
|
"github.com/syncthing/syncthing/lib/protocol"
|
|
"github.com/syncthing/syncthing/lib/relay"
|
|
"github.com/syncthing/syncthing/lib/relay/client"
|
|
"github.com/syncthing/syncthing/lib/util"
|
|
|
|
"github.com/thejerf/suture"
|
|
)
|
|
|
|
type DialerFactory func(*url.URL, *tls.Config) (*tls.Conn, error)
|
|
type ListenerFactory func(*url.URL, *tls.Config, chan<- model.IntermediateConnection)
|
|
|
|
var (
|
|
dialers = make(map[string]DialerFactory, 0)
|
|
listeners = make(map[string]ListenerFactory, 0)
|
|
)
|
|
|
|
type Model interface {
|
|
protocol.Model
|
|
AddConnection(conn model.Connection, hello protocol.HelloMessage)
|
|
ConnectedTo(remoteID protocol.DeviceID) bool
|
|
IsPaused(remoteID protocol.DeviceID) bool
|
|
OnHello(protocol.DeviceID, net.Addr, protocol.HelloMessage)
|
|
GetHello(protocol.DeviceID) protocol.HelloMessage
|
|
}
|
|
|
|
// Service listens on TLS and dials configured unconnected devices. Successful
|
|
// connections are handed to the model.
|
|
type Service struct {
|
|
*suture.Supervisor
|
|
cfg *config.Wrapper
|
|
myID protocol.DeviceID
|
|
model Model
|
|
tlsCfg *tls.Config
|
|
discoverer discover.Finder
|
|
conns chan model.IntermediateConnection
|
|
mappings []*nat.Mapping
|
|
relayService relay.Service
|
|
bepProtocolName string
|
|
tlsDefaultCommonName string
|
|
lans []*net.IPNet
|
|
writeRateLimit *ratelimit.Bucket
|
|
readRateLimit *ratelimit.Bucket
|
|
|
|
mut sync.RWMutex
|
|
connType map[protocol.DeviceID]model.ConnectionType
|
|
}
|
|
|
|
func NewConnectionService(cfg *config.Wrapper, myID protocol.DeviceID, mdl Model, tlsCfg *tls.Config, discoverer discover.Finder, mappings []*nat.Mapping,
|
|
relayService relay.Service, bepProtocolName string, tlsDefaultCommonName string, lans []*net.IPNet) *Service {
|
|
service := &Service{
|
|
Supervisor: suture.NewSimple("connections.Service"),
|
|
cfg: cfg,
|
|
myID: myID,
|
|
model: mdl,
|
|
tlsCfg: tlsCfg,
|
|
discoverer: discoverer,
|
|
mappings: mappings,
|
|
relayService: relayService,
|
|
conns: make(chan model.IntermediateConnection),
|
|
bepProtocolName: bepProtocolName,
|
|
tlsDefaultCommonName: tlsDefaultCommonName,
|
|
lans: lans,
|
|
|
|
connType: make(map[protocol.DeviceID]model.ConnectionType),
|
|
}
|
|
cfg.Subscribe(service)
|
|
|
|
// The rate variables are in KiB/s in the UI (despite the camel casing
|
|
// of the name). We multiply by 1024 here to get B/s.
|
|
if service.cfg.Options().MaxSendKbps > 0 {
|
|
service.writeRateLimit = ratelimit.NewBucketWithRate(float64(1024*service.cfg.Options().MaxSendKbps), int64(5*1024*service.cfg.Options().MaxSendKbps))
|
|
}
|
|
if service.cfg.Options().MaxRecvKbps > 0 {
|
|
service.readRateLimit = ratelimit.NewBucketWithRate(float64(1024*service.cfg.Options().MaxRecvKbps), int64(5*1024*service.cfg.Options().MaxRecvKbps))
|
|
}
|
|
|
|
// There are several moving parts here; one routine per listening address
|
|
// to handle incoming connections, one routine to periodically attempt
|
|
// outgoing connections, one routine to the the common handling
|
|
// regardless of whether the connection was incoming or outgoing.
|
|
// Furthermore, a relay service which handles incoming requests to connect
|
|
// via the relays.
|
|
//
|
|
// TODO: Clean shutdown, and/or handling config changes on the fly. We
|
|
// partly do this now - new devices and addresses will be picked up, but
|
|
// not new listen addresses and we don't support disconnecting devices
|
|
// that are removed and so on...
|
|
|
|
service.Add(serviceFunc(service.connect))
|
|
for _, addr := range service.cfg.Options().ListenAddress {
|
|
uri, err := url.Parse(addr)
|
|
if err != nil {
|
|
l.Infoln("Failed to parse listen address:", addr, err)
|
|
continue
|
|
}
|
|
|
|
listener, ok := listeners[uri.Scheme]
|
|
if !ok {
|
|
l.Infoln("Unknown listen address scheme:", uri.String())
|
|
continue
|
|
}
|
|
|
|
l.Debugln("listening on", uri)
|
|
|
|
service.Add(serviceFunc(func() {
|
|
listener(uri, service.tlsCfg, service.conns)
|
|
}))
|
|
}
|
|
service.Add(serviceFunc(service.handle))
|
|
|
|
if service.relayService != nil {
|
|
service.Add(serviceFunc(service.acceptRelayConns))
|
|
}
|
|
|
|
for _, mapping := range mappings {
|
|
mapping.OnChanged(func(m *nat.Mapping, added, removed []nat.Address) {
|
|
events.Default.Log(events.ExternalPortMappingChanged, map[string]interface{}{
|
|
"protocol": m.Protocol(),
|
|
"local": m.Address().String(),
|
|
"added": added,
|
|
"removed": removed,
|
|
})
|
|
})
|
|
}
|
|
|
|
return service
|
|
}
|
|
|
|
func (s *Service) handle() {
|
|
next:
|
|
for c := range s.conns {
|
|
cs := c.ConnectionState()
|
|
|
|
// We should have negotiated the next level protocol "bep/1.0" as part
|
|
// of the TLS handshake. Unfortunately this can't be a hard error,
|
|
// because there are implementations out there that don't support
|
|
// protocol negotiation (iOS for one...).
|
|
if !cs.NegotiatedProtocolIsMutual || cs.NegotiatedProtocol != s.bepProtocolName {
|
|
l.Infof("Peer %s did not negotiate bep/1.0", c.RemoteAddr())
|
|
}
|
|
|
|
// We should have received exactly one certificate from the other
|
|
// side. If we didn't, they don't have a device ID and we drop the
|
|
// connection.
|
|
certs := cs.PeerCertificates
|
|
if cl := len(certs); cl != 1 {
|
|
l.Infof("Got peer certificate list of length %d != 1 from %s; protocol error", cl, c.RemoteAddr())
|
|
c.Close()
|
|
continue
|
|
}
|
|
remoteCert := certs[0]
|
|
remoteID := protocol.NewDeviceID(remoteCert.Raw)
|
|
|
|
// The device ID should not be that of ourselves. It can happen
|
|
// though, especially in the presence of NAT hairpinning, multiple
|
|
// clients between the same NAT gateway, and global discovery.
|
|
if remoteID == s.myID {
|
|
l.Infof("Connected to myself (%s) - should not happen", remoteID)
|
|
c.Close()
|
|
continue
|
|
}
|
|
|
|
hello, err := exchangeHello(c, s.model.GetHello(remoteID))
|
|
if err != nil {
|
|
l.Infof("Failed to exchange Hello messages with %s (%s): %s", remoteID, c.RemoteAddr(), err)
|
|
c.Close()
|
|
continue next
|
|
}
|
|
|
|
s.model.OnHello(remoteID, c.RemoteAddr(), hello)
|
|
|
|
// If we have a relay connection, and the new incoming connection is
|
|
// not a relay connection, we should drop that, and prefer the this one.
|
|
s.mut.RLock()
|
|
skip := false
|
|
ct, ok := s.connType[remoteID]
|
|
if ok && !ct.IsDirect() && c.Type.IsDirect() {
|
|
l.Debugln("Switching connections", remoteID)
|
|
s.model.Close(remoteID, fmt.Errorf("switching connections"))
|
|
} else if s.model.ConnectedTo(remoteID) {
|
|
// We should not already be connected to the other party. TODO: This
|
|
// could use some better handling. If the old connection is dead but
|
|
// hasn't timed out yet we may want to drop *that* connection and keep
|
|
// this one. But in case we are two devices connecting to each other
|
|
// in parallel we don't want to do that or we end up with no
|
|
// connections still established...
|
|
l.Infof("Connected to already connected device (%s)", remoteID)
|
|
c.Close()
|
|
skip = true
|
|
} else if s.model.IsPaused(remoteID) {
|
|
l.Infof("Connection from paused device (%s)", remoteID)
|
|
c.Close()
|
|
skip = true
|
|
}
|
|
s.mut.RUnlock()
|
|
if skip {
|
|
continue
|
|
}
|
|
|
|
for deviceID, deviceCfg := range s.cfg.Devices() {
|
|
if deviceID == remoteID {
|
|
// Verify the name on the certificate. By default we set it to
|
|
// "syncthing" when generating, but the user may have replaced
|
|
// the certificate and used another name.
|
|
certName := deviceCfg.CertName
|
|
if certName == "" {
|
|
certName = s.tlsDefaultCommonName
|
|
}
|
|
err := remoteCert.VerifyHostname(certName)
|
|
if err != nil {
|
|
// Incorrect certificate name is something the user most
|
|
// likely wants to know about, since it's an advanced
|
|
// config. Warn instead of Info.
|
|
l.Warnf("Bad certificate from %s (%v): %v", remoteID, c.RemoteAddr(), err)
|
|
c.Close()
|
|
continue next
|
|
}
|
|
|
|
// If rate limiting is set, and based on the address we should
|
|
// limit the connection, then we wrap it in a limiter.
|
|
|
|
limit := s.shouldLimit(c.RemoteAddr())
|
|
|
|
wr := io.Writer(c.Conn)
|
|
if limit && s.writeRateLimit != nil {
|
|
wr = NewWriteLimiter(c.Conn, s.writeRateLimit)
|
|
}
|
|
|
|
rd := io.Reader(c.Conn)
|
|
if limit && s.readRateLimit != nil {
|
|
rd = NewReadLimiter(c.Conn, s.readRateLimit)
|
|
}
|
|
|
|
name := fmt.Sprintf("%s-%s (%s)", c.LocalAddr(), c.RemoteAddr(), c.Type)
|
|
protoConn := protocol.NewConnection(remoteID, rd, wr, s.model, name, deviceCfg.Compression)
|
|
|
|
l.Infof("Established secure connection to %s at %s", remoteID, name)
|
|
l.Debugf("cipher suite: %04X in lan: %t", c.ConnectionState().CipherSuite, !limit)
|
|
|
|
s.mut.Lock()
|
|
s.model.AddConnection(model.Connection{
|
|
Conn: c,
|
|
Connection: protoConn,
|
|
Type: c.Type,
|
|
}, hello)
|
|
s.connType[remoteID] = c.Type
|
|
s.mut.Unlock()
|
|
continue next
|
|
}
|
|
}
|
|
|
|
l.Infof("Connection from %s (%s) with ignored device ID %s", c.RemoteAddr(), c.Type, remoteID)
|
|
c.Close()
|
|
}
|
|
}
|
|
|
|
func (s *Service) connect() {
|
|
lastRelayCheck := make(map[protocol.DeviceID]time.Time)
|
|
|
|
delay := time.Second
|
|
for {
|
|
l.Debugln("Reconnect loop")
|
|
relaysEnabled := s.cfg.Options().RelaysEnabled
|
|
nextDevice:
|
|
for deviceID, deviceCfg := range s.cfg.Devices() {
|
|
if deviceID == s.myID {
|
|
continue
|
|
}
|
|
|
|
l.Debugln("Reconnect loop for", deviceID)
|
|
|
|
s.mut.RLock()
|
|
paused := s.model.IsPaused(deviceID)
|
|
connected := s.model.ConnectedTo(deviceID)
|
|
ct, ok := s.connType[deviceID]
|
|
s.mut.RUnlock()
|
|
|
|
if paused {
|
|
continue
|
|
}
|
|
if connected && ok && ct.IsDirect() {
|
|
l.Debugln("Already connected to", deviceID, "via", ct.String())
|
|
continue
|
|
}
|
|
|
|
addrs, relays := s.resolveAddresses(deviceID, deviceCfg.Addresses)
|
|
|
|
for _, addr := range addrs {
|
|
if conn := s.connectDirect(deviceID, addr); conn != nil {
|
|
l.Debugln("Connecting to", deviceID, "via", addr, "succeeded")
|
|
if connected {
|
|
s.model.Close(deviceID, fmt.Errorf("switching connections"))
|
|
}
|
|
s.conns <- model.IntermediateConnection{
|
|
Conn: conn,
|
|
Type: model.ConnectionTypeDirectDial,
|
|
}
|
|
continue nextDevice
|
|
}
|
|
l.Debugln("Connecting to", deviceID, "via", addr, "failed")
|
|
}
|
|
|
|
// Only connect via relays if not already connected
|
|
// Also, do not set lastRelayCheck time if we have no relays,
|
|
// as otherwise when we do discover relays, we might have to
|
|
// wait up to RelayReconnectIntervalM to connect again.
|
|
// Also, do not try relays if we are explicitly told not to.
|
|
if connected || len(relays) == 0 || !relaysEnabled {
|
|
l.Debugln("Not connecting via relay", connected, len(relays) == 0, !relaysEnabled)
|
|
continue nextDevice
|
|
}
|
|
|
|
reconIntv := time.Duration(s.cfg.Options().RelayReconnectIntervalM) * time.Minute
|
|
if last, ok := lastRelayCheck[deviceID]; ok && time.Since(last) < reconIntv {
|
|
l.Debugln("Skipping connecting via relay to", deviceID, "last checked at", last)
|
|
continue nextDevice
|
|
}
|
|
|
|
l.Debugln("Trying relay connections to", deviceID, relays)
|
|
|
|
lastRelayCheck[deviceID] = time.Now()
|
|
|
|
for _, addr := range relays {
|
|
if conn := s.connectViaRelay(deviceID, addr); conn != nil {
|
|
l.Debugln("Connecting to", deviceID, "via", addr, "succeeded")
|
|
s.conns <- model.IntermediateConnection{
|
|
Conn: conn,
|
|
Type: model.ConnectionTypeRelayDial,
|
|
}
|
|
continue nextDevice
|
|
}
|
|
l.Debugln("Connecting to", deviceID, "via", addr, "failed")
|
|
}
|
|
}
|
|
|
|
time.Sleep(delay)
|
|
delay *= 2
|
|
if maxD := time.Duration(s.cfg.Options().ReconnectIntervalS) * time.Second; delay > maxD {
|
|
delay = maxD
|
|
}
|
|
}
|
|
}
|
|
|
|
func (s *Service) resolveAddresses(deviceID protocol.DeviceID, inAddrs []string) (addrs []string, relays []discover.Relay) {
|
|
for _, addr := range inAddrs {
|
|
if addr == "dynamic" {
|
|
if s.discoverer != nil {
|
|
if t, r, err := s.discoverer.Lookup(deviceID); err == nil {
|
|
addrs = append(addrs, t...)
|
|
relays = append(relays, r...)
|
|
}
|
|
}
|
|
} else {
|
|
addrs = append(addrs, addr)
|
|
}
|
|
}
|
|
return
|
|
}
|
|
|
|
func (s *Service) connectDirect(deviceID protocol.DeviceID, addr string) *tls.Conn {
|
|
uri, err := url.Parse(addr)
|
|
if err != nil {
|
|
l.Infoln("Failed to parse connection url:", addr, err)
|
|
return nil
|
|
}
|
|
|
|
dialer, ok := dialers[uri.Scheme]
|
|
if !ok {
|
|
l.Infoln("Unknown address schema", uri)
|
|
return nil
|
|
}
|
|
|
|
l.Debugln("dial", deviceID, uri)
|
|
conn, err := dialer(uri, s.tlsCfg)
|
|
if err != nil {
|
|
l.Debugln("dial failed", deviceID, uri, err)
|
|
return nil
|
|
}
|
|
|
|
return conn
|
|
}
|
|
|
|
func (s *Service) connectViaRelay(deviceID protocol.DeviceID, addr discover.Relay) *tls.Conn {
|
|
uri, err := url.Parse(addr.URL)
|
|
if err != nil {
|
|
l.Infoln("Failed to parse relay connection url:", addr, err)
|
|
return nil
|
|
}
|
|
|
|
inv, err := client.GetInvitationFromRelay(uri, deviceID, s.tlsCfg.Certificates, 10*time.Second)
|
|
if err != nil {
|
|
l.Debugf("Failed to get invitation for %s from %s: %v", deviceID, uri, err)
|
|
return nil
|
|
}
|
|
l.Debugln("Succesfully retrieved relay invitation", inv, "from", uri)
|
|
|
|
conn, err := client.JoinSession(inv)
|
|
if err != nil {
|
|
l.Debugf("Failed to join relay session %s: %v", inv, err)
|
|
return nil
|
|
}
|
|
l.Debugln("Successfully joined relay session", inv)
|
|
|
|
var tc *tls.Conn
|
|
|
|
if inv.ServerSocket {
|
|
tc = tls.Server(conn, s.tlsCfg)
|
|
} else {
|
|
tc = tls.Client(conn, s.tlsCfg)
|
|
}
|
|
|
|
err = tc.Handshake()
|
|
if err != nil {
|
|
l.Infof("TLS handshake (BEP/relay %s): %v", inv, err)
|
|
tc.Close()
|
|
return nil
|
|
}
|
|
|
|
return tc
|
|
}
|
|
|
|
func (s *Service) acceptRelayConns() {
|
|
for {
|
|
conn := s.relayService.Accept()
|
|
s.conns <- model.IntermediateConnection{
|
|
Conn: conn,
|
|
Type: model.ConnectionTypeRelayAccept,
|
|
}
|
|
}
|
|
}
|
|
|
|
func (s *Service) shouldLimit(addr net.Addr) bool {
|
|
if s.cfg.Options().LimitBandwidthInLan {
|
|
return true
|
|
}
|
|
|
|
tcpaddr, ok := addr.(*net.TCPAddr)
|
|
if !ok {
|
|
return true
|
|
}
|
|
for _, lan := range s.lans {
|
|
if lan.Contains(tcpaddr.IP) {
|
|
return false
|
|
}
|
|
}
|
|
return !tcpaddr.IP.IsLoopback()
|
|
}
|
|
|
|
func (s *Service) VerifyConfiguration(from, to config.Configuration) error {
|
|
return nil
|
|
}
|
|
|
|
func (s *Service) CommitConfiguration(from, to config.Configuration) bool {
|
|
// We require a restart if a device as been removed.
|
|
|
|
newDevices := make(map[protocol.DeviceID]bool, len(to.Devices))
|
|
for _, dev := range to.Devices {
|
|
newDevices[dev.DeviceID] = true
|
|
}
|
|
|
|
for _, dev := range from.Devices {
|
|
if !newDevices[dev.DeviceID] {
|
|
return false
|
|
}
|
|
}
|
|
|
|
return true
|
|
}
|
|
|
|
// ExternalAddresses returns a list of addresses that are our best guess for
|
|
// where we are reachable from the outside. As a special case, we may return
|
|
// one or more addresses with an empty IP address (0.0.0.0 or ::) and just
|
|
// port number - this means that the outside address of a NAT gateway should
|
|
// be substituted.
|
|
func (s *Service) ExternalAddresses() []string {
|
|
return s.addresses(false)
|
|
}
|
|
|
|
// AllAddresses returns a list of addresses that are our best guess for where
|
|
// we are reachable from the local network. Same conditions as
|
|
// ExternalAddresses, but private IPv4 addresses are included.
|
|
func (s *Service) AllAddresses() []string {
|
|
return s.addresses(true)
|
|
}
|
|
|
|
func (s *Service) addresses(includePrivateIPV4 bool) []string {
|
|
var addrs []string
|
|
|
|
// Grab our listen addresses from the config. Unspecified ones are passed
|
|
// on verbatim (to be interpreted by a global discovery server or local
|
|
// discovery peer). Public addresses are passed on verbatim. Private
|
|
// addresses are filtered.
|
|
for _, addrStr := range s.cfg.Options().ListenAddress {
|
|
addrURL, err := url.Parse(addrStr)
|
|
if err != nil {
|
|
l.Infoln("Listen address", addrStr, "is invalid:", err)
|
|
continue
|
|
}
|
|
addr, err := net.ResolveTCPAddr(addrURL.Scheme, addrURL.Host)
|
|
if err != nil {
|
|
l.Infoln("Listen address", addrStr, "is invalid:", err)
|
|
continue
|
|
}
|
|
|
|
if addr.IP == nil || addr.IP.IsUnspecified() {
|
|
// Address like 0.0.0.0:22000 or [::]:22000 or :22000; include as is.
|
|
addrs = append(addrs, util.Address(addrURL.Scheme, addr.String()))
|
|
} else if isPublicIPv4(addr.IP) || isPublicIPv6(addr.IP) {
|
|
// A public address; include as is.
|
|
addrs = append(addrs, util.Address(addrURL.Scheme, addr.String()))
|
|
} else if includePrivateIPV4 && addr.IP.To4().IsGlobalUnicast() {
|
|
// A private IPv4 address.
|
|
addrs = append(addrs, util.Address(addrURL.Scheme, addr.String()))
|
|
}
|
|
}
|
|
|
|
// Add addresses provided by the mappings from the NAT service.
|
|
for _, mapping := range s.mappings {
|
|
for _, addr := range mapping.ExternalAddresses() {
|
|
addrs = append(addrs, fmt.Sprintf("tcp://%s", addr))
|
|
}
|
|
}
|
|
|
|
return addrs
|
|
}
|
|
|
|
func isPublicIPv4(ip net.IP) bool {
|
|
ip = ip.To4()
|
|
if ip == nil {
|
|
// Not an IPv4 address (IPv6)
|
|
return false
|
|
}
|
|
|
|
// IsGlobalUnicast below only checks that it's not link local or
|
|
// multicast, and we want to exclude private (NAT:ed) addresses as well.
|
|
rfc1918 := []net.IPNet{
|
|
{IP: net.IP{10, 0, 0, 0}, Mask: net.IPMask{255, 0, 0, 0}},
|
|
{IP: net.IP{172, 16, 0, 0}, Mask: net.IPMask{255, 240, 0, 0}},
|
|
{IP: net.IP{192, 168, 0, 0}, Mask: net.IPMask{255, 255, 0, 0}},
|
|
}
|
|
for _, n := range rfc1918 {
|
|
if n.Contains(ip) {
|
|
return false
|
|
}
|
|
}
|
|
|
|
return ip.IsGlobalUnicast()
|
|
}
|
|
|
|
func isPublicIPv6(ip net.IP) bool {
|
|
if ip.To4() != nil {
|
|
// Not an IPv6 address (IPv4)
|
|
// (To16() returns a v6 mapped v4 address so can't be used to check
|
|
// that it's an actual v6 address)
|
|
return false
|
|
}
|
|
|
|
return ip.IsGlobalUnicast()
|
|
}
|
|
|
|
func exchangeHello(c net.Conn, h protocol.HelloMessage) (protocol.HelloMessage, error) {
|
|
if err := c.SetDeadline(time.Now().Add(2 * time.Second)); err != nil {
|
|
return protocol.HelloMessage{}, err
|
|
}
|
|
defer c.SetDeadline(time.Time{})
|
|
|
|
header := make([]byte, 8)
|
|
msg := h.MustMarshalXDR()
|
|
|
|
binary.BigEndian.PutUint32(header[:4], protocol.HelloMessageMagic)
|
|
binary.BigEndian.PutUint32(header[4:], uint32(len(msg)))
|
|
|
|
if _, err := c.Write(header); err != nil {
|
|
return protocol.HelloMessage{}, err
|
|
}
|
|
|
|
if _, err := c.Write(msg); err != nil {
|
|
return protocol.HelloMessage{}, err
|
|
}
|
|
|
|
if _, err := io.ReadFull(c, header); err != nil {
|
|
return protocol.HelloMessage{}, err
|
|
}
|
|
|
|
if binary.BigEndian.Uint32(header[:4]) != protocol.HelloMessageMagic {
|
|
return protocol.HelloMessage{}, fmt.Errorf("incorrect magic")
|
|
}
|
|
|
|
msgSize := binary.BigEndian.Uint32(header[4:])
|
|
if msgSize > 1024 {
|
|
return protocol.HelloMessage{}, fmt.Errorf("hello message too big")
|
|
}
|
|
|
|
buf := make([]byte, msgSize)
|
|
|
|
var hello protocol.HelloMessage
|
|
|
|
if _, err := io.ReadFull(c, buf); err != nil {
|
|
return protocol.HelloMessage{}, err
|
|
}
|
|
|
|
if err := hello.UnmarshalXDR(buf); err != nil {
|
|
return protocol.HelloMessage{}, err
|
|
}
|
|
|
|
return hello, nil
|
|
}
|
|
|
|
// serviceFunc wraps a function to create a suture.Service without stop
|
|
// functionality.
|
|
type serviceFunc func()
|
|
|
|
func (f serviceFunc) Serve() { f() }
|
|
func (f serviceFunc) Stop() {}
|