perso/syncthing-arm
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Expose connection type and relay status in the UI

Browse Source
This commit is contained in:
Audrius Butkevicius
2015-07-17 21:22:07 +01:00
parent 2c0f8dc546
commit 8f2db99c86
15 changed files with 183 additions and 123 deletions
Download Patch File Download Diff File Expand all files Collapse all files

84
cmd/syncthing/connections.go
View File

@@ -16,17 +16,17 @@ import (
"time"
"github.com/syncthing/protocol"
"github.com/syncthing/relaysrv/client"
"github.com/syncthing/syncthing/lib/config"
"github.com/syncthing/syncthing/lib/events"
"github.com/syncthing/syncthing/lib/model"
"github.com/syncthing/syncthing/lib/osutil"
"github.com/thejerf/suture"
)
type DialerFactory func(*url.URL, *tls.Config) (*tls.Conn, error)
type ListenerFactory func(*url.URL, *tls.Config, chan<- intermediateConnection)
type ListenerFactory func(*url.URL, *tls.Config, chan<- model.IntermediateConnection)
var (
dialers = make(map[string]DialerFactory, 0)
@@ -41,7 +41,7 @@ type connectionSvc struct {
myID protocol.DeviceID
model *model.Model
tlsCfg *tls.Config
conns chan intermediateConnection
conns chan model.IntermediateConnection
lastRelayCheck map[protocol.DeviceID]time.Time
@@ -49,11 +49,6 @@ type connectionSvc struct {
connType map[protocol.DeviceID]model.ConnectionType
}
type intermediateConnection struct {
conn *tls.Conn
connType model.ConnectionType
}
func newConnectionSvc(cfg *config.Wrapper, myID protocol.DeviceID, mdl *model.Model, tlsCfg *tls.Config) *connectionSvc {
svc := &connectionSvc{
Supervisor: suture.NewSimple("connectionSvc"),
@@ -61,7 +56,7 @@ func newConnectionSvc(cfg *config.Wrapper, myID protocol.DeviceID, mdl *model.Mo
myID: myID,
model: mdl,
tlsCfg: tlsCfg,
conns: make(chan intermediateConnection),
conns: make(chan model.IntermediateConnection),
connType: make(map[protocol.DeviceID]model.ConnectionType),
lastRelayCheck: make(map[protocol.DeviceID]time.Time),
@@ -110,14 +105,14 @@ func newConnectionSvc(cfg *config.Wrapper, myID protocol.DeviceID, mdl *model.Mo
func (s *connectionSvc) handle() {
next:
for c := range s.conns {
cs := c.conn.ConnectionState()
cs := c.Conn.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 != bepProtocolName {
l.Infof("Peer %s did not negotiate bep/1.0", c.conn.RemoteAddr())
l.Infof("Peer %s did not negotiate bep/1.0", c.Conn.RemoteAddr())
}
// We should have received exactly one certificate from the other
@@ -125,8 +120,8 @@ next:
// 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.conn.RemoteAddr())
c.conn.Close()
l.Infof("Got peer certificate list of length %d != 1 from %s; protocol error", cl, c.Conn.RemoteAddr())
c.Conn.Close()
continue
}
remoteCert := certs[0]
@@ -137,7 +132,7 @@ next:
// clients between the same NAT gateway, and global discovery.
if remoteID == myID {
l.Infof("Connected to myself (%s) - should not happen", remoteID)
c.conn.Close()
c.Conn.Close()
continue
}
@@ -146,7 +141,7 @@ next:
s.mut.RLock()
ct, ok := s.connType[remoteID]
s.mut.RUnlock()
if ok && !ct.IsDirect() && c.connType.IsDirect() {
if ok && !ct.IsDirect() && c.ConnType.IsDirect() {
if debugNet {
l.Debugln("Switching connections", remoteID)
}
@@ -159,7 +154,7 @@ next:
// 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.conn.Close()
c.Conn.Close()
continue
}
@@ -177,41 +172,41 @@ next:
// 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.conn.RemoteAddr(), err)
c.conn.Close()
l.Warnf("Bad certificate from %s (%v): %v", remoteID, c.Conn.RemoteAddr(), err)
c.Conn.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.conn.RemoteAddr())
limit := s.shouldLimit(c.Conn.RemoteAddr())
wr := io.Writer(c.conn)
wr := io.Writer(c.Conn)
if limit && writeRateLimit != nil {
wr = &limitedWriter{c.conn, writeRateLimit}
wr = &limitedWriter{c.Conn, writeRateLimit}
}
rd := io.Reader(c.conn)
rd := io.Reader(c.Conn)
if limit && readRateLimit != nil {
rd = &limitedReader{c.conn, readRateLimit}
rd = &limitedReader{c.Conn, readRateLimit}
}
name := fmt.Sprintf("%s-%s (%s)", c.conn.LocalAddr(), c.conn.RemoteAddr(), c.connType)
name := fmt.Sprintf("%s-%s (%s)", c.Conn.LocalAddr(), c.Conn.RemoteAddr(), c.ConnType)
protoConn := protocol.NewConnection(remoteID, rd, wr, s.model, name, deviceCfg.Compression)
l.Infof("Established secure connection to %s at %s", remoteID, name)
if debugNet {
l.Debugf("cipher suite: %04X in lan: %t", c.conn.ConnectionState().CipherSuite, !limit)
l.Debugf("cipher suite: %04X in lan: %t", c.Conn.ConnectionState().CipherSuite, !limit)
}
s.model.AddConnection(model.Connection{
c.conn,
c.Conn,
protoConn,
c.connType,
c.ConnType,
})
s.mut.Lock()
s.connType[remoteID] = c.connType
s.connType[remoteID] = c.ConnType
s.mut.Unlock()
continue next
}
@@ -220,14 +215,14 @@ next:
if !s.cfg.IgnoredDevice(remoteID) {
events.Default.Log(events.DeviceRejected, map[string]string{
"device": remoteID.String(),
"address": c.conn.RemoteAddr().String(),
"address": c.Conn.RemoteAddr().String(),
})
l.Infof("Connection from %s (%s) with unknown device ID %s", c.conn.RemoteAddr(), c.connType, remoteID)
l.Infof("Connection from %s (%s) with unknown device ID %s", c.Conn.RemoteAddr(), c.ConnType, remoteID)
} else {
l.Infof("Connection from %s (%s) with ignored device ID %s", c.conn.RemoteAddr(), c.connType, remoteID)
l.Infof("Connection from %s (%s) with ignored device ID %s", c.Conn.RemoteAddr(), c.ConnType, remoteID)
}
c.conn.Close()
c.Conn.Close()
}
}
@@ -294,7 +289,7 @@ func (s *connectionSvc) connect() {
s.model.Close(deviceID, fmt.Errorf("switching connections"))
}
s.conns <- intermediateConnection{
s.conns <- model.IntermediateConnection{
conn, model.ConnectionTypeBasicDial,
}
continue nextDevice
@@ -347,7 +342,10 @@ func (s *connectionSvc) connect() {
l.Debugln("Sucessfully joined relay session", inv)
}
setTCPOptions(conn.(*net.TCPConn))
err = osutil.SetTCPOptions(conn.(*net.TCPConn))
if err != nil {
l.Infoln(err)
}
var tc *tls.Conn
@@ -362,7 +360,7 @@ func (s *connectionSvc) connect() {
tc.Close()
continue
}
s.conns <- intermediateConnection{
s.conns <- model.IntermediateConnection{
tc, model.ConnectionTypeRelayDial,
}
continue nextDevice
@@ -414,19 +412,3 @@ func (s *connectionSvc) CommitConfiguration(from, to config.Configuration) bool
return true
}
func setTCPOptions(conn *net.TCPConn) {
var err error
if err = conn.SetLinger(0); err != nil {
l.Infoln(err)
}
if err = conn.SetNoDelay(false); err != nil {
l.Infoln(err)
}
if err = conn.SetKeepAlivePeriod(60 * time.Second); err != nil {
l.Infoln(err)
}
if err = conn.SetKeepAlive(true); err != nil {
l.Infoln(err)
}
}

16
cmd/syncthing/connections_tcp.go
View File

@@ -13,6 +13,7 @@ import (
"strings"
"github.com/syncthing/syncthing/lib/model"
"github.com/syncthing/syncthing/lib/osutil"
)
func init() {
@@ -46,7 +47,10 @@ func tcpDialer(uri *url.URL, tlsCfg *tls.Config) (*tls.Conn, error) {
return nil, err
}
setTCPOptions(conn)
err = osutil.SetTCPOptions(conn)
if err != nil {
l.Infoln(err)
}
tc := tls.Client(conn, tlsCfg)
err = tc.Handshake()
@@ -58,7 +62,7 @@ func tcpDialer(uri *url.URL, tlsCfg *tls.Config) (*tls.Conn, error) {
return tc, nil
}
func tcpListener(uri *url.URL, tlsCfg *tls.Config, conns chan<- intermediateConnection) {
func tcpListener(uri *url.URL, tlsCfg *tls.Config, conns chan<- model.IntermediateConnection) {
tcaddr, err := net.ResolveTCPAddr("tcp", uri.Host)
if err != nil {
l.Fatalln("listen (BEP/tcp):", err)
@@ -81,8 +85,10 @@ func tcpListener(uri *url.URL, tlsCfg *tls.Config, conns chan<- intermediateConn
l.Debugln("connect from", conn.RemoteAddr())
}
tcpConn := conn.(*net.TCPConn)
setTCPOptions(tcpConn)
err = osutil.SetTCPOptions(conn.(*net.TCPConn))
if err != nil {
l.Infoln(err)
}
tc := tls.Server(conn, tlsCfg)
err = tc.Handshake()
@@ -92,7 +98,7 @@ func tcpListener(uri *url.URL, tlsCfg *tls.Config, conns chan<- intermediateConn
continue
}
conns <- intermediateConnection{
conns <- model.IntermediateConnection{
tc, model.ConnectionTypeBasicAccept,
}
}

3
cmd/syncthing/gui.go
View File

@@ -628,6 +628,9 @@ func (s *apiSvc) getSystemStatus(w http.ResponseWriter, r *http.Request) {
if cfg.Options().GlobalAnnEnabled && discoverer != nil {
res["extAnnounceOK"] = discoverer.ExtAnnounceOK()
}
if relaySvc != nil {
res["relayClientStatus"] = relaySvc.ClientStatus()
}
cpuUsageLock.RLock()
var cpusum float64
for _, p := range cpuUsagePercent {

12
cmd/syncthing/main.go
View File

@@ -34,8 +34,10 @@ import (
"github.com/syncthing/syncthing/lib/events"
"github.com/syncthing/syncthing/lib/model"
"github.com/syncthing/syncthing/lib/osutil"
"github.com/syncthing/syncthing/lib/relay"
"github.com/syncthing/syncthing/lib/symlinks"
"github.com/syncthing/syncthing/lib/upgrade"
"github.com/syndtr/goleveldb/leveldb"
"github.com/syndtr/goleveldb/leveldb/errors"
"github.com/syndtr/goleveldb/leveldb/opt"
@@ -110,6 +112,7 @@ var (
readRateLimit *ratelimit.Bucket
stop = make(chan int)
discoverer *discover.Discoverer
relaySvc *relay.Svc
cert tls.Certificate
lans []*net.IPNet
)
@@ -671,14 +674,14 @@ func syncthingMain() {
// Start the relevant services
connectionSvc := newConnectionSvc(cfg, myID, m, tlsCfg)
relaySvc := newRelaySvc(cfg, tlsCfg, connectionSvc.conns)
relaySvc = relay.NewSvc(cfg, tlsCfg, connectionSvc.conns)
connectionSvc.Add(relaySvc)
mainSvc.Add(connectionSvc)
// Start discovery
localPort := addr.Port
discoverer = discovery(localPort)
discoverer = discovery(localPort, relaySvc)
// Start UPnP. The UPnP service will restart global discovery if the
// external port changes.
@@ -908,10 +911,9 @@ func shutdown() {
stop <- exitSuccess
}
func discovery(extPort int) *discover.Discoverer {
func discovery(extPort int, relaySvc *relay.Svc) *discover.Discoverer {
opts := cfg.Options()
disc := discover.NewDiscoverer(myID, opts.ListenAddress, opts.RelayServers)
disc := discover.NewDiscoverer(myID, opts.ListenAddress, relaySvc)
if opts.LocalAnnEnabled {
l.Infoln("Starting local discovery announcements")
disc.StartLocal(opts.LocalAnnPort, opts.LocalAnnMCAddr)

193
cmd/syncthing/relays.go
View File

@@ -1,193 +0,0 @@
// 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 main
import (
"crypto/tls"
"net"
"net/url"
"time"
"github.com/syncthing/relaysrv/client"
"github.com/syncthing/relaysrv/protocol"
"github.com/syncthing/syncthing/lib/config"
"github.com/syncthing/syncthing/lib/model"
"github.com/syncthing/syncthing/lib/sync"
"github.com/thejerf/suture"
)
func newRelaySvc(cfg *config.Wrapper, tlsCfg *tls.Config, conns chan<- intermediateConnection) *relaySvc {
svc := &relaySvc{
Supervisor: suture.New("relaySvc", suture.Spec{
Log: func(log string) {
if debugNet {
l.Infoln(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),
}
rcfg := cfg.Raw()
svc.CommitConfiguration(rcfg, rcfg)
cfg.Subscribe(svc)
receiver := &invitationReceiver{
tlsCfg: tlsCfg,
conns: conns,
invitations: svc.invitations,
}
svc.receiverToken = svc.Add(receiver)
return svc
}
type relaySvc struct {
*suture.Supervisor
cfg *config.Wrapper
tlsCfg *tls.Config
receiverToken suture.ServiceToken
tokens map[string]suture.ServiceToken
clients map[string]*client.ProtocolClient
mut sync.RWMutex
invitations chan protocol.SessionInvitation
}
func (s *relaySvc) 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 *relaySvc) CommitConfiguration(from, to config.Configuration) bool {
existing := make(map[string]struct{}, len(to.Options.RelayServers))
for _, addr := range to.Options.RelayServers {
uri, err := url.Parse(addr)
if err != nil {
if debugNet {
l.Debugln("Failed to parse relay address", addr, err)
}
continue
}
existing[uri.String()] = struct{}{}
_, ok := s.tokens[uri.String()]
if !ok {
if debugNet {
l.Debugln("Connecting to relay", uri)
}
c := client.NewProtocolClient(uri, s.tlsCfg.Certificates, s.invitations)
s.tokens[uri.String()] = s.Add(c)
s.mut.Lock()
s.clients[uri.String()] = c
s.mut.Unlock()
}
}
for uri, token := range s.tokens {
_, ok := existing[uri]
if !ok {
err := s.Remove(token)
delete(s.tokens, uri)
s.mut.Lock()
delete(s.clients, uri)
s.mut.Unlock()
if debugNet {
l.Debugln("Disconnecting from relay", uri, err)
}
}
}
return true
}
func (s *relaySvc) ClientStatus() map[string]bool {
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
}
type invitationReceiver struct {
invitations chan protocol.SessionInvitation
tlsCfg *tls.Config
conns chan<- intermediateConnection
stop chan struct{}
}
func (r *invitationReceiver) Serve() {
if r.stop != nil {
return
}
r.stop = make(chan struct{})
for {
select {
case inv := <-r.invitations:
if debugNet {
l.Debugln("Received relay invitation", inv)
}
conn, err := client.JoinSession(inv)
if err != nil {
if debugNet {
l.Debugf("Failed to join relay session %s: %v", inv, err)
}
continue
}
setTCPOptions(conn.(*net.TCPConn))
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 <- intermediateConnection{
tc, model.ConnectionTypeRelayAccept,
}
case <-r.stop:
return
}
}
}
func (r *invitationReceiver) Stop() {
if r.stop == nil {
return
}
r.stop <- struct{}{}
r.stop = nil
}

2
cmd/syncthing/verbose.go
View File

@@ -74,7 +74,7 @@ func (s *verboseSvc) formatEvent(ev events.Event) string {
return fmt.Sprintf("Discovered device %v at %v", data["device"], data["addrs"])
case events.DeviceConnected:
data := ev.Data.(map[string]string)
return fmt.Sprintf("Connected to device %v at %v", data["id"], data["addr"])
return fmt.Sprintf("Connected to device %v at %v (type %s)", data["id"], data["addr"], data["type"])
case events.DeviceDisconnected:
data := ev.Data.(map[string]string)
return fmt.Sprintf("Disconnected from device %v", data["id"])