syncthing/discover/discover.go

package discover

import (
	"encoding/hex"
	"errors"
	"fmt"
	"log"
	"net"
	"strings"
	"sync"
	"time"

	"github.com/calmh/syncthing/buffers"
)

const (
	AnnouncementPort = 21025
)

type Discoverer struct {
	MyID             string
	ListenPort       int
	BroadcastIntv    time.Duration
	ExtBroadcastIntv time.Duration

	conn         *net.UDPConn
	registry     map[string][]string
	registryLock sync.RWMutex
	extServer    string

	localBroadcastTick  <-chan time.Time
	forcedBroadcastTick chan time.Time
}

var (
	ErrIncorrectMagic = errors.New("incorrect magic number")
)

// We tolerate a certain amount of errors because we might be running on
// laptops that sleep and wake, have intermittent network connectivity, etc.
// When we hit this many errors in succession, we stop.
const maxErrors = 30

func NewDiscoverer(id string, port int, extServer string) (*Discoverer, error) {
	local := &net.UDPAddr{IP: nil, Port: AnnouncementPort}
	conn, err := net.ListenUDP("udp", local)
	if err != nil {
		return nil, err
	}

	disc := &Discoverer{
		MyID:             id,
		ListenPort:       port,
		BroadcastIntv:    30 * time.Second,
		ExtBroadcastIntv: 1800 * time.Second,

		conn:      conn,
		registry:  make(map[string][]string),
		extServer: extServer,
	}

	go disc.recvAnnouncements()

	if disc.ListenPort > 0 {
		disc.localBroadcastTick = time.Tick(disc.BroadcastIntv)
		disc.forcedBroadcastTick = make(chan time.Time)
		go disc.sendAnnouncements()
	}
	if len(disc.extServer) > 0 {
		go disc.sendExtAnnouncements()
	}

	return disc, nil
}

func (d *Discoverer) sendAnnouncements() {
	var pkt = AnnounceV2{AnnouncementMagicV2, d.MyID, []Address{{nil, 22000}}}
	var buf = pkt.MarshalXDR()
	var errCounter = 0
	var err error

	remote := &net.UDPAddr{
		IP:   net.IP{255, 255, 255, 255},
		Port: AnnouncementPort,
	}

	for errCounter < maxErrors {
		intfs, err := net.Interfaces()
		if err != nil {
			log.Printf("discover/listInterfaces: %v; no local announcements", err)
			return
		}

		for _, intf := range intfs {
			if intf.Flags&(net.FlagBroadcast|net.FlagLoopback) == net.FlagBroadcast {
				addrs, err := intf.Addrs()
				if err != nil {
					log.Println("discover/listAddrs: warning:", err)
					errCounter++
					continue
				}

				var srcAddr string
				for _, addr := range addrs {
					if strings.Contains(addr.String(), ".") {
						// Found an IPv4 adress
						parts := strings.Split(addr.String(), "/")
						srcAddr = parts[0]
						break
					}
				}
				if len(srcAddr) == 0 {
					if debug {
						dlog.Println("no source address found on interface", intf.Name)
					}
					continue
				}

				iaddr, err := net.ResolveUDPAddr("udp4", srcAddr+":0")
				if err != nil {
					log.Println("discover/resolve: warning:", err)
					errCounter++
					continue
				}

				conn, err := net.ListenUDP("udp4", iaddr)
				if err != nil {
					log.Println("discover/listen: warning:", err)
					errCounter++
					continue
				}

				if debug {
					dlog.Println("send announcement from", conn.LocalAddr(), "to", remote, "on", intf.Name)
				}

				_, err = conn.WriteTo(buf, remote)
				if err != nil {
					// Some interfaces don't seem to support broadcast even though the flags claims they do, i.e. vmnet
					conn.Close()

					if debug {
						log.Println(err)
					}

					errCounter++
					continue
				}

				conn.Close()
				errCounter = 0
			}
		}

		select {
		case <-d.localBroadcastTick:
		case <-d.forcedBroadcastTick:
		}
	}
	log.Println("discover/write: local: stopping due to too many errors:", err)
}

func (d *Discoverer) sendExtAnnouncements() {
	remote, err := net.ResolveUDPAddr("udp", d.extServer)
	if err != nil {
		log.Printf("discover/external: %v; no external announcements", err)
		return
	}

	var pkt = AnnounceV2{AnnouncementMagicV2, d.MyID, []Address{{nil, 22000}}}
	var buf = pkt.MarshalXDR()
	var errCounter = 0

	for errCounter < maxErrors {
		if debug {
			dlog.Println("send announcement -> ", remote)
		}
		_, err = d.conn.WriteTo(buf, remote)
		if err != nil {
			log.Println("discover/write: warning:", err)
			errCounter++
		} else {
			errCounter = 0
		}
		time.Sleep(d.ExtBroadcastIntv)
	}
	log.Printf("discover/write: %v: stopping due to too many errors: %v", remote, err)
}

func (d *Discoverer) recvAnnouncements() {
	var buf = make([]byte, 1024)
	var errCounter = 0
	var err error
	for errCounter < maxErrors {
		n, addr, err := d.conn.ReadFromUDP(buf)
		if err != nil {
			errCounter++
			time.Sleep(time.Second)
			continue
		}

		if debug {
			dlog.Printf("read announcement:\n%s", hex.Dump(buf[:n]))
		}

		var pkt AnnounceV2
		err = pkt.UnmarshalXDR(buf[:n])
		if err != nil {
			errCounter++
			time.Sleep(time.Second)
			continue
		}

		if debug {
			dlog.Printf("parsed announcement: %#v", pkt)
		}

		errCounter = 0

		if pkt.NodeID != d.MyID {
			var addrs []string
			for _, a := range pkt.Addresses {
				var nodeAddr string
				if len(a.IP) > 0 {
					nodeAddr = fmt.Sprintf("%s:%d", ipStr(a.IP), a.Port)
				} else {
					nodeAddr = fmt.Sprintf("%s:%d", addr.IP.String(), a.Port)
				}
				addrs = append(addrs, nodeAddr)
			}
			if debug {
				dlog.Printf("register: %#v", addrs)
			}
			d.registryLock.Lock()
			_, seen := d.registry[pkt.NodeID]
			if !seen {
				select {
				case d.forcedBroadcastTick <- time.Now():
				}
			}
			d.registry[pkt.NodeID] = addrs
			d.registryLock.Unlock()
		}
	}
	log.Println("discover/read: stopping due to too many errors:", err)
}

func (d *Discoverer) externalLookup(node string) []string {
	extIP, err := net.ResolveUDPAddr("udp", d.extServer)
	if err != nil {
		log.Printf("discover/external: %v; no external lookup", err)
		return nil
	}

	conn, err := net.DialUDP("udp", nil, extIP)
	if err != nil {
		log.Printf("discover/external: %v; no external lookup", err)
		return nil
	}
	defer conn.Close()

	err = conn.SetDeadline(time.Now().Add(5 * time.Second))
	if err != nil {
		log.Printf("discover/external: %v; no external lookup", err)
		return nil
	}

	buf := QueryV2{QueryMagicV2, node}.MarshalXDR()
	_, err = conn.Write(buf)
	if err != nil {
		log.Printf("discover/external: %v; no external lookup", err)
		return nil
	}
	buffers.Put(buf)

	buf = buffers.Get(256)
	defer buffers.Put(buf)

	n, err := conn.Read(buf)
	if err != nil {
		if err, ok := err.(net.Error); ok && err.Timeout() {
			// Expected if the server doesn't know about requested node ID
			return nil
		}
		log.Printf("discover/external/read: %v; no external lookup", err)
		return nil
	}

	if debug {
		dlog.Printf("read external:\n%s", hex.Dump(buf[:n]))
	}

	var pkt AnnounceV2
	err = pkt.UnmarshalXDR(buf[:n])
	if err != nil {
		log.Println("discover/external/decode:", err)
		return nil
	}

	if debug {
		dlog.Printf("parsed external: %#v", pkt)
	}

	var addrs []string
	for _, a := range pkt.Addresses {
		var nodeAddr string
		if len(a.IP) > 0 {
			nodeAddr = fmt.Sprintf("%s:%d", ipStr(a.IP), a.Port)
		}
		addrs = append(addrs, nodeAddr)
	}
	return addrs
}

func (d *Discoverer) Lookup(node string) []string {
	d.registryLock.Lock()
	addr, ok := d.registry[node]
	d.registryLock.Unlock()

	if ok {
		return addr
	} else if len(d.extServer) != 0 {
		// We might want to cache this, but not permanently so it needs some intelligence
		return d.externalLookup(node)
	}
	return nil
}

func ipStr(ip []byte) string {
	var f = "%d"
	var s = "."
	if len(ip) > 4 {
		f = "%x"
		s = ":"
	}
	var ss = make([]string, len(ip))
	for i := range ip {
		ss[i] = fmt.Sprintf(f, ip[i])
	}
	return strings.Join(ss, s)
}