Merge pull request yggdrasil-network#318 from yggdrasil-network/link

Implement link interface

contrib/ansible/genkeys.go

@@ -35,22 +35,22 @@ func main() {
 	}
 
 	var encryptionKeys []keySet
-	for i := 0; i < *numHosts + 1; i++ {
+	for i := 0; i < *numHosts+1; i++ {
 		encryptionKeys = append(encryptionKeys, newBoxKey())
 	}
 	encryptionKeys = sortKeySetArray(encryptionKeys)
-	for i := 0; i < *keyTries - *numHosts - 1; i++ {
-		encryptionKeys[0] = newBoxKey();
+	for i := 0; i < *keyTries-*numHosts-1; i++ {
+		encryptionKeys[0] = newBoxKey()
 		encryptionKeys = bubbleUpTo(encryptionKeys, 0)
 	}
 
 	var signatureKeys []keySet
-	for i := 0; i < *numHosts + 1; i++ {
+	for i := 0; i < *numHosts+1; i++ {
 		signatureKeys = append(signatureKeys, newSigKey())
 	}
 	signatureKeys = sortKeySetArray(signatureKeys)
-	for i := 0; i < *keyTries - *numHosts - 1; i++ {
-		signatureKeys[0] = newSigKey();
+	for i := 0; i < *keyTries-*numHosts-1; i++ {
+		signatureKeys[0] = newSigKey()
 		signatureKeys = bubbleUpTo(signatureKeys, 0)
 	}
 
@@ -112,11 +112,11 @@ func sortKeySetArray(sets []keySet) []keySet {
 }
 
 func bubbleUpTo(sets []keySet, num int) []keySet {
-	for i := 0; i < len(sets) - num - 1; i++ {
-		if isBetter(sets[i + 1].id, sets[i].id) {
+	for i := 0; i < len(sets)-num-1; i++ {
+		if isBetter(sets[i+1].id, sets[i].id) {
 			var tmp = sets[i]
-			sets[i] = sets[i + 1]
-			sets[i + 1] = tmp
+			sets[i] = sets[i+1]
+			sets[i+1] = tmp
 		} else {
 			break
 		}

go.mod

@@ -2,10 +2,10 @@ module github.com/yggdrasil-network/yggdrasil-go
 
 require (
 	github.com/docker/libcontainer v2.2.1+incompatible
+	github.com/gologme/log v0.0.0-20181207131047-4e5d8ccb38e8
 	github.com/hjson/hjson-go v0.0.0-20181010104306-a25ecf6bd222
 	github.com/kardianos/minwinsvc v0.0.0-20151122163309-cad6b2b879b0
 	github.com/mitchellh/mapstructure v1.1.2
-	github.com/gologme/log v0.0.0-20181207131047-4e5d8ccb38e8
 	github.com/songgao/packets v0.0.0-20160404182456-549a10cd4091
 	github.com/yggdrasil-network/water v0.0.0-20180615095340-f732c88f34ae
 	golang.org/x/crypto v0.0.0-20181203042331-505ab145d0a9

go.sum

@@ -1,5 +1,7 @@
 github.com/docker/libcontainer v2.2.1+incompatible h1:++SbbkCw+X8vAd4j2gOCzZ2Nn7s2xFALTf7LZKmM1/0=
 github.com/docker/libcontainer v2.2.1+incompatible/go.mod h1:osvj61pYsqhNCMLGX31xr7klUBhHb/ZBuXS0o1Fvwbw=
+github.com/gologme/log v0.0.0-20181207131047-4e5d8ccb38e8 h1:WD8iJ37bRNwvETMfVTusVSAi0WdXTpfNVGY2aHycNKY=
+github.com/gologme/log v0.0.0-20181207131047-4e5d8ccb38e8/go.mod h1:gq31gQ8wEHkR+WekdWsqDuf8pXTUZA9BnnzTuPz1Y9U=
 github.com/hjson/hjson-go v0.0.0-20181010104306-a25ecf6bd222 h1:xmvkbxXDeN1ffWq8kvrhyqVYAO2aXuRBsbpxVTR+JyU=
 github.com/hjson/hjson-go v0.0.0-20181010104306-a25ecf6bd222/go.mod h1:qsetwF8NlsTsOTwZTApNlTCerV+b2GjYRRcIk4JMFio=
 github.com/kardianos/minwinsvc v0.0.0-20151122163309-cad6b2b879b0 h1:YnZmFjg0Nvk8851WTVWlqMC1ecJH07Ctz+Ezxx4u54g=
@@ -18,5 +20,3 @@ golang.org/x/sys v0.0.0-20181206074257-70b957f3b65e h1:njOxP/wVblhCLIUhjHXf6X+dz
 golang.org/x/sys v0.0.0-20181206074257-70b957f3b65e/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/text v0.3.0 h1:g61tztE5qeGQ89tm6NTjjM9VPIm088od1l6aSorWRWg=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
-github.com/gologme/log v0.0.0-20181207131047-4e5d8ccb38e8 h1:WD8iJ37bRNwvETMfVTusVSAi0WdXTpfNVGY2aHycNKY=
-github.com/gologme/log v0.0.0-20181207131047-4e5d8ccb38e8/go.mod h1:gq31gQ8wEHkR+WekdWsqDuf8pXTUZA9BnnzTuPz1Y9U=

src/util/util.go

@@ -3,6 +3,7 @@ package util
 // These are misc. utility functions that didn't really fit anywhere else
 
 import "runtime"
+import "time"
 
 // A wrapper around runtime.Gosched() so it doesn't need to be imported elsewhere.
 func Yield() {
@@ -44,3 +45,14 @@ func PutBytes(bs []byte) {
 	default:
 	}
 }
+
+// This is a workaround to go's broken timer implementation
+func TimerStop(t *time.Timer) bool {
+	if !t.Stop() {
+		select {
+		case <-t.C:
+		default:
+		}
+	}
+	return true
+}

src/yggdrasil/awdl.go

@@ -2,147 +2,97 @@ package yggdrasil
 
 import (
 	"errors"
-	"fmt"
+	"io"
 	"sync"
-	"sync/atomic"
-	"time"
-
-	"github.com/yggdrasil-network/yggdrasil-go/src/crypto"
-	"github.com/yggdrasil-network/yggdrasil-go/src/util"
 )
 
 type awdl struct {
-	core       *Core
+	link       *link
 	mutex      sync.RWMutex // protects interfaces below
 	interfaces map[string]*awdlInterface
 }
 
 type awdlInterface struct {
-	awdl     *awdl
+	linkif *linkInterface
+	rwc    awdlReadWriteCloser
+	peer   *peer
+	stream stream
+}
+
+type awdlReadWriteCloser struct {
 	fromAWDL chan []byte
 	toAWDL   chan []byte
-	shutdown chan bool
-	peer     *peer
 }
 
-func (l *awdl) init(c *Core) error {
-	l.core = c
-	l.mutex.Lock()
-	l.interfaces = make(map[string]*awdlInterface)
-	l.mutex.Unlock()
+func (c awdlReadWriteCloser) Read(p []byte) (n int, err error) {
+	if packet, ok := <-c.fromAWDL; ok {
+		n = copy(p, packet)
+		return n, nil
+	}
+	return 0, io.EOF
+}
 
+func (c awdlReadWriteCloser) Write(p []byte) (n int, err error) {
+	var pc []byte
+	pc = append(pc, p...)
+	c.toAWDL <- pc
+	return len(pc), nil
+}
+
+func (c awdlReadWriteCloser) Close() error {
+	close(c.fromAWDL)
+	close(c.toAWDL)
 	return nil
 }
 
-func (l *awdl) create(fromAWDL chan []byte, toAWDL chan []byte /*boxPubKey *crypto.BoxPubKey, sigPubKey *crypto.SigPubKey*/, name string) (*awdlInterface, error) {
-	intf := awdlInterface{
-		awdl:     l,
-		fromAWDL: fromAWDL,
-		toAWDL:   toAWDL,
-		shutdown: make(chan bool),
-	}
-	l.mutex.Lock()
-	l.interfaces[name] = &intf
-	l.mutex.Unlock()
-	myLinkPub, myLinkPriv := crypto.NewBoxKeys()
-	meta := version_getBaseMetadata()
-	meta.box = l.core.boxPub
-	meta.sig = l.core.sigPub
-	meta.link = *myLinkPub
-	metaBytes := meta.encode()
-	l.core.log.Traceln("toAWDL <- metaBytes")
-	toAWDL <- metaBytes
-	l.core.log.Traceln("metaBytes = <-fromAWDL")
-	metaBytes = <-fromAWDL
-	l.core.log.Traceln("version_metadata{}")
-	meta = version_metadata{}
-	if !meta.decode(metaBytes) || !meta.check() {
-		return nil, errors.New("Metadata decode failure")
+func (a *awdl) init(l *link) error {
+	a.link = l
+	a.mutex.Lock()
+	a.interfaces = make(map[string]*awdlInterface)
+	a.mutex.Unlock()
+
+	return nil
+}
+
+func (a *awdl) create(name, local, remote string, incoming bool) (*awdlInterface, error) {
+	rwc := awdlReadWriteCloser{
+		fromAWDL: make(chan []byte, 1),
+		toAWDL:   make(chan []byte, 1),
 	}
-	l.core.log.Traceln("version_getBaseMetadata{}")
-	base := version_getBaseMetadata()
-	if meta.ver > base.ver || meta.ver == base.ver && meta.minorVer > base.minorVer {
-		return nil, errors.New("Failed to connect to node: " + name + " version: " + fmt.Sprintf("%d.%d", meta.ver, meta.minorVer))
+	s := stream{}
+	s.init(rwc)
+	linkif, err := a.link.create(&s, name, "awdl", local, remote, incoming, true)
+	if err != nil {
+		return nil, err
 	}
-	l.core.log.Traceln("crypto.GetSharedKey")
-	shared := crypto.GetSharedKey(myLinkPriv, &meta.link)
-	//shared := crypto.GetSharedKey(&l.core.boxPriv, boxPubKey)
-	l.core.log.Traceln("l.core.peers.newPeer")
-	intf.peer = l.core.peers.newPeer(&meta.box, &meta.sig, shared, name)
-	if intf.peer != nil {
-		intf.peer.linkOut = make(chan []byte, 1) // protocol traffic
-		intf.peer.out = func(msg []byte) {
-			defer func() { recover() }()
-			intf.toAWDL <- msg
-		} // called by peer.sendPacket()
-		l.core.switchTable.idleIn <- intf.peer.port // notify switch that we're idle
-		intf.peer.close = func() {
-			close(intf.fromAWDL)
-			close(intf.toAWDL)
-		}
-		go intf.handler()
-		go intf.peer.linkLoop()
-		return &intf, nil
+	intf := awdlInterface{
+		linkif: linkif,
+		rwc:    rwc,
 	}
-	delete(l.interfaces, name)
-	return nil, errors.New("l.core.peers.newPeer failed")
+	a.mutex.Lock()
+	a.interfaces[name] = &intf
+	a.mutex.Unlock()
+	go intf.linkif.handler()
+	return &intf, nil
 }
 
-func (l *awdl) getInterface(identity string) *awdlInterface {
-	l.mutex.RLock()
-	defer l.mutex.RUnlock()
-	if intf, ok := l.interfaces[identity]; ok {
+func (a *awdl) getInterface(identity string) *awdlInterface {
+	a.mutex.RLock()
+	defer a.mutex.RUnlock()
+	if intf, ok := a.interfaces[identity]; ok {
 		return intf
 	}
 	return nil
 }
 
-func (l *awdl) shutdown(identity string) error {
-	if intf, ok := l.interfaces[identity]; ok {
-		intf.shutdown <- true
-		l.core.peers.removePeer(intf.peer.port)
-		l.mutex.Lock()
-		delete(l.interfaces, identity)
-		l.mutex.Unlock()
+func (a *awdl) shutdown(identity string) error {
+	if intf, ok := a.interfaces[identity]; ok {
+		close(intf.linkif.closed)
+		intf.rwc.Close()
+		a.mutex.Lock()
+		delete(a.interfaces, identity)
+		a.mutex.Unlock()
 		return nil
-	} else {
-		return errors.New(fmt.Sprintf("Interface '%s' doesn't exist or already shutdown", identity))
-	}
-}
-
-func (ai *awdlInterface) handler() {
-	send := func(msg []byte) {
-		ai.toAWDL <- msg
-		atomic.AddUint64(&ai.peer.bytesSent, uint64(len(msg)))
-		util.PutBytes(msg)
-	}
-	for {
-		timerInterval := tcp_ping_interval
-		timer := time.NewTimer(timerInterval)
-		defer timer.Stop()
-		select {
-		case p := <-ai.peer.linkOut:
-			send(p)
-			continue
-		default:
-		}
-		timer.Stop()
-		select {
-		case <-timer.C:
-		default:
-		}
-		timer.Reset(timerInterval)
-		select {
-		case _ = <-timer.C:
-			send([]byte{})
-		case p := <-ai.peer.linkOut:
-			send(p)
-			continue
-		case r := <-ai.fromAWDL:
-			ai.peer.handlePacket(r)
-			ai.awdl.core.switchTable.idleIn <- ai.peer.port
-		case <-ai.shutdown:
-			return
-		}
 	}
+	return errors.New("Interface not found or already closed")
 }

src/yggdrasil/core.go

@@ -45,7 +45,7 @@ type Core struct {
 	searches    searches
 	multicast   multicast
 	tcp         tcpInterface
-	awdl        awdl
+	link        link
 	log         *log.Logger
 }
 
@@ -199,8 +199,8 @@ func (c *Core) Start(nc *config.NodeConfig, log *log.Logger) error {
 		return err
 	}
 
-	if err := c.awdl.init(c); err != nil {
-		c.log.Errorln("Failed to start AWDL interface")
+	if err := c.link.init(c); err != nil {
+		c.log.Errorln("Failed to start link interfaces")
 		return err
 	}
 

src/yggdrasil/debug.go

@@ -97,9 +97,7 @@ func (c *Core) DEBUG_getPeers() *peers {
 }
 
 func (ps *peers) DEBUG_newPeer(box crypto.BoxPubKey, sig crypto.SigPubKey, link crypto.BoxSharedKey) *peer {
-	//in <-chan []byte,
-	//out chan<- []byte) *peer {
-	return ps.newPeer(&box, &sig, &link, "(simulator)") //, in, out)
+	return ps.newPeer(&box, &sig, &link, "(simulator)", nil)
 }
 
 /*