Merge pull request #74 from Arceliar/peerauth

Partial support for authenticated peers

src/yggdrasil/admin.go

@@ -3,6 +3,7 @@ package yggdrasil
 import "net"
 import "os"
 import "bytes"
+import "encoding/hex"
 import "errors"
 import "fmt"
 import "net/url"
@@ -104,6 +105,23 @@ func (a *admin) init(c *Core, listenaddr string) {
 			*out = []byte(a.printInfos([]admin_nodeInfo{info}))
 		}
 	})
+	a.addHandler("getAllowedBoxPubs", nil, func(out *[]byte, _ ...string) {
+		*out = []byte(a.getAllowedBoxPubs())
+	})
+	a.addHandler("addAllowedBoxPub", []string{"<boxPubKey>"}, func(out *[]byte, saddr ...string) {
+		if a.addAllowedBoxPub(saddr[0]) == nil {
+			*out = []byte("Adding key: " + saddr[0] + "\n")
+		} else {
+			*out = []byte("Failed to add key: " + saddr[0] + "\n")
+		}
+	})
+	a.addHandler("removeAllowedBoxPub", []string{"<boxPubKey>"}, func(out *[]byte, sport ...string) {
+		if a.removeAllowedBoxPub(sport[0]) == nil {
+			*out = []byte("Removing key: " + sport[0] + "\n")
+		} else {
+			*out = []byte("Failed to remove key: " + sport[0] + "\n")
+		}
+	})
 	go a.listen()
 }
 
@@ -347,6 +365,36 @@ func (a *admin) getData_getSessions() []admin_nodeInfo {
 	return infos
 }
 
+func (a *admin) getAllowedBoxPubs() string {
+	pubs := a.core.peers.getAllowedBoxPubs()
+	var out []string
+	for _, pub := range pubs {
+		out = append(out, hex.EncodeToString(pub[:]))
+	}
+	out = append(out, "")
+	return strings.Join(out, "\n")
+}
+
+func (a *admin) addAllowedBoxPub(bstr string) (err error) {
+	boxBytes, err := hex.DecodeString(bstr)
+	if err == nil {
+		var box boxPubKey
+		copy(box[:], boxBytes)
+		a.core.peers.addAllowedBoxPub(&box)
+	}
+	return
+}
+
+func (a *admin) removeAllowedBoxPub(bstr string) (err error) {
+	boxBytes, err := hex.DecodeString(bstr)
+	if err == nil {
+		var box boxPubKey
+		copy(box[:], boxBytes)
+		a.core.peers.removeAllowedBoxPub(&box)
+	}
+	return
+}
+
 func (a *admin) getResponse_dot() []byte {
 	self := a.getData_getSelf().asMap()
 	myAddr := self["IP"]

src/yggdrasil/config/config.go

@@ -5,6 +5,7 @@ type NodeConfig struct {
 	Listen         string
 	AdminListen    string
 	Peers          []string
+	AllowedBoxPubs []string
 	BoxPub         string
 	BoxPriv        string
 	SigPub         string

src/yggdrasil/core.go

@@ -44,6 +44,7 @@ func (c *Core) init(bpub *boxPubKey,
 	c.log = log.New(ioutil.Discard, "", 0)
 	c.boxPub, c.boxPriv = *bpub, *bpriv
 	c.sigPub, c.sigPriv = *spub, *spriv
+	c.admin.core = c
 	c.sigs.init()
 	c.searches.init(c)
 	c.dht.init(c)

src/yggdrasil/debug.go

@@ -397,6 +397,13 @@ func (c *Core) DEBUG_setIfceExpr(expr *regexp.Regexp) {
 	c.ifceExpr = expr
 }
 
+func (c *Core) DEBUG_addAllowedBoxPub(boxStr string) {
+	err := c.admin.addAllowedBoxPub(boxStr)
+	if err != nil {
+		panic(err)
+	}
+}
+
 ////////////////////////////////////////////////////////////////////////////////
 
 func DEBUG_simLinkPeers(p, q *peer) {

src/yggdrasil/peer.go

@@ -34,6 +34,8 @@ type peers struct {
 	mutex sync.Mutex   // Synchronize writes to atomic
 	ports atomic.Value //map[Port]*peer, use CoW semantics
 	//ports map[Port]*peer
+	authMutex      sync.RWMutex
+	allowedBoxPubs map[boxPubKey]struct{}
 }
 
 func (ps *peers) init(c *Core) {
@@ -41,6 +43,36 @@ func (ps *peers) init(c *Core) {
 	defer ps.mutex.Unlock()
 	ps.putPorts(make(map[switchPort]*peer))
 	ps.core = c
+	ps.allowedBoxPubs = make(map[boxPubKey]struct{})
+}
+
+func (ps *peers) isAllowedBoxPub(box *boxPubKey) bool {
+	ps.authMutex.RLock()
+	defer ps.authMutex.RUnlock()
+	_, isIn := ps.allowedBoxPubs[*box]
+	return isIn || len(ps.allowedBoxPubs) == 0
+}
+
+func (ps *peers) addAllowedBoxPub(box *boxPubKey) {
+	ps.authMutex.Lock()
+	defer ps.authMutex.Unlock()
+	ps.allowedBoxPubs[*box] = struct{}{}
+}
+
+func (ps *peers) removeAllowedBoxPub(box *boxPubKey) {
+	ps.authMutex.Lock()
+	defer ps.authMutex.Unlock()
+	delete(ps.allowedBoxPubs, *box)
+}
+
+func (ps *peers) getAllowedBoxPubs() []boxPubKey {
+	ps.authMutex.RLock()
+	defer ps.authMutex.RUnlock()
+	keys := make([]boxPubKey, 0, len(ps.allowedBoxPubs))
+	for key := range ps.allowedBoxPubs {
+		keys = append(keys, key)
+	}
+	return keys
 }
 
 func (ps *peers) getPorts() map[switchPort]*peer {
@@ -75,6 +107,8 @@ type peer struct {
 	throttle uint8
 	// Called when a peer is removed, to close the underlying connection, or via admin api
 	close func()
+	// To allow the peer to call close if idle for too long
+	lastAnc time.Time
 }
 
 const peer_Throttle = 1
@@ -99,13 +133,10 @@ func (p *peer) updateBandwidth(bytes int, duration time.Duration) {
 
 func (ps *peers) newPeer(box *boxPubKey,
 	sig *sigPubKey) *peer {
-	//in <-chan []byte,
-	//out chan<- []byte) *peer {
 	p := peer{box: *box,
 		sig:     *sig,
 		shared:  *getSharedKey(&ps.core.boxPriv, box),
-		//in: in,
-		//out: out,
+		lastAnc: time.Now(),
 		core:    ps.core}
 	ps.mutex.Lock()
 	defer ps.mutex.Unlock()
@@ -158,7 +189,10 @@ func (p *peer) linkLoop(in <-chan []byte) {
 			}
 			p.handleLinkTraffic(packet)
 		case <-ticker.C:
-			{
+			if time.Since(p.lastAnc) > 16*time.Second && p.close != nil {
+				// Seems to have timed out, try to trigger a close
+				p.close()
+			}
 			p.throttle = 0
 			if p.port == 0 {
 				continue
@@ -184,7 +218,6 @@ func (p *peer) linkLoop(in <-chan []byte) {
 			counter = (counter + 1) % 4
 		}
 	}
-	}
 }
 
 func (p *peer) handlePacket(packet []byte, linkIn chan<- []byte) {
@@ -210,6 +243,10 @@ func (p *peer) handlePacket(packet []byte, linkIn chan<- []byte) {
 }
 
 func (p *peer) handleTraffic(packet []byte, pTypeLen int) {
+	if p.port != 0 && p.msgAnc == nil {
+		// Drop traffic until the peer manages to send us at least one anc
+		return
+	}
 	ttl, ttlLen := wire_decode_uint64(packet[pTypeLen:])
 	ttlBegin := pTypeLen
 	ttlEnd := pTypeLen + ttlLen
@@ -292,6 +329,7 @@ func (p *peer) handleSwitchAnnounce(packet []byte) {
 	}
 	p.msgAnc = &anc
 	p.processSwitchMessage()
+	p.lastAnc = time.Now()
 }
 
 func (p *peer) requestHop(hop uint64) {

src/yggdrasil/tcp.go

@@ -62,7 +62,7 @@ func (iface *tcpInterface) listener() {
 		if err != nil {
 			panic(err)
 		}
-		go iface.handler(sock)
+		go iface.handler(sock, true)
 	}
 }
 
@@ -81,7 +81,7 @@ func (iface *tcpInterface) callWithConn(conn net.Conn) {
 				delete(iface.calls, raddr)
 				iface.mutex.Unlock()
 			}()
-			iface.handler(conn)
+			iface.handler(conn, false)
 		}
 	}()
 }
@@ -106,12 +106,12 @@ func (iface *tcpInterface) call(saddr string) {
 			if err != nil {
 				return
 			}
-			iface.handler(conn)
+			iface.handler(conn, false)
 		}
 	}()
 }
 
-func (iface *tcpInterface) handler(sock net.Conn) {
+func (iface *tcpInterface) handler(sock net.Conn, incoming bool) {
 	defer sock.Close()
 	// Get our keys
 	keys := []byte{}
@@ -150,6 +150,15 @@ func (iface *tcpInterface) handler(sock net.Conn) {
 	if equiv(info.sig[:], iface.core.sigPub[:]) {
 		return
 	}
+	// Check if we're authorized to connect to this key / IP
+	if incoming && !iface.core.peers.isAllowedBoxPub(&info.box) {
+		// Allow unauthorized peers if they're link-local
+		raddrStr, _, _ := net.SplitHostPort(sock.RemoteAddr().String())
+		raddr := net.ParseIP(raddrStr)
+		if !raddr.IsLinkLocalUnicast() {
+			return
+		}
+	}
 	// Check if we already have a connection to this node, close and block if yes
 	info.localAddr, _, _ = net.SplitHostPort(sock.LocalAddr().String())
 	info.remoteAddr, _, _ = net.SplitHostPort(sock.RemoteAddr().String())

src/yggdrasil/udp.go

@@ -204,6 +204,14 @@ func (iface *udpInterface) handleKeys(msg []byte, addr connAddr) {
 	iface.mutex.RUnlock()
 	if !isIn {
 		udpAddr := addr.toUDPAddr()
+		// Check if we're authorized to connect to this key / IP
+		// TODO monitor and always allow outgoing connections
+		if !iface.core.peers.isAllowedBoxPub(&ks.box) {
+			// Allow unauthorized peers if they're link-local
+			if !udpAddr.IP.IsLinkLocalUnicast() {
+				return
+			}
+		}
 		themNodeID := getNodeID(&ks.box)
 		themAddr := address_addrForNodeID(themNodeID)
 		themAddrString := net.IP(themAddr[:]).String()

yggdrasil.go

@@ -66,6 +66,10 @@ func (n *node) init(cfg *nodeConfig, logger *log.Logger) {
 	logger.Println("Starting admin socket...")
 	n.core.DEBUG_setupAndStartAdminInterface(cfg.AdminListen)
 	logger.Println("Started admin socket")
+	for _, pBoxStr := range cfg.AllowedBoxPubs {
+		n.core.DEBUG_addAllowedBoxPub(pBoxStr)
+	}
+
 	go func() {
 		if len(cfg.Peers) == 0 {
 			return
@@ -97,6 +101,7 @@ func generateConfig(isAutoconf bool) *nodeConfig {
 	cfg.SigPub = hex.EncodeToString(spub[:])
 	cfg.SigPriv = hex.EncodeToString(spriv[:])
 	cfg.Peers = []string{}
+	cfg.AllowedBoxPubs = []string{}
 	cfg.Multicast = true
 	cfg.LinkLocal = ""
 	cfg.IfName = core.DEBUG_GetTUNDefaultIfName()