Merge branch 'develop' into circlecirpm

cmd/yggdrasil/main.go

@@ -2,6 +2,7 @@ package main
 
 import (
 	"bytes"
+	"encoding/hex"
 	"encoding/json"
 	"flag"
 	"fmt"
@@ -14,28 +15,28 @@ import (
 	"golang.org/x/text/encoding/unicode"
 
 	"github.com/gologme/log"
+	gsyslog "github.com/hashicorp/go-syslog"
 	"github.com/hjson/hjson-go"
 	"github.com/kardianos/minwinsvc"
 	"github.com/mitchellh/mapstructure"
 
 	"github.com/yggdrasil-network/yggdrasil-go/src/admin"
 	"github.com/yggdrasil-network/yggdrasil-go/src/config"
+	"github.com/yggdrasil-network/yggdrasil-go/src/crypto"
 	"github.com/yggdrasil-network/yggdrasil-go/src/multicast"
 	"github.com/yggdrasil-network/yggdrasil-go/src/tuntap"
 	"github.com/yggdrasil-network/yggdrasil-go/src/yggdrasil"
 )
 
-type nodeConfig = config.NodeConfig
-type Core = yggdrasil.Core
-
 type node struct {
-	core      Core
+	core      yggdrasil.Core
+	state     *config.NodeState
 	tuntap    tuntap.TunAdapter
 	multicast multicast.Multicast
 	admin     admin.AdminSocket
 }
 
-func readConfig(useconf *bool, useconffile *string, normaliseconf *bool) *nodeConfig {
+func readConfig(useconf *bool, useconffile *string, normaliseconf *bool) *config.NodeConfig {
 	// Use a configuration file. If -useconf, the configuration will be read
 	// from stdin. If -useconffile, the configuration will be read from the
 	// filesystem.
@@ -114,9 +115,10 @@ func main() {
 	autoconf := flag.Bool("autoconf", false, "automatic mode (dynamic IP, peer with IPv6 neighbors)")
 	version := flag.Bool("version", false, "prints the version of this build")
 	logging := flag.String("logging", "info,warn,error", "comma-separated list of logging levels to enable")
+	logto := flag.String("logto", "stdout", "file path to log to, \"syslog\" or \"stdout\"")
 	flag.Parse()
 
-	var cfg *nodeConfig
+	var cfg *config.NodeConfig
 	var err error
 	switch {
 	case *version:
@@ -161,7 +163,23 @@ func main() {
 		return
 	}
 	// Create a new logger that logs output to stdout.
-	logger := log.New(os.Stdout, "", log.Flags())
+	var logger *log.Logger
+	switch *logto {
+	case "stdout":
+		logger = log.New(os.Stdout, "", log.Flags())
+	case "syslog":
+		if syslogger, err := gsyslog.NewLogger(gsyslog.LOG_NOTICE, "DAEMON", yggdrasil.BuildName()); err == nil {
+			logger = log.New(syslogger, "", log.Flags())
+		}
+	default:
+		if logfd, err := os.OpenFile(*logto, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644); err == nil {
+			logger = log.New(logfd, "", log.Flags())
+		}
+	}
+	if logger == nil {
+		logger = log.New(os.Stdout, "", log.Flags())
+		logger.Warnln("Logging defaulting to stdout")
+	}
 	//logger.EnableLevel("error")
 	//logger.EnableLevel("warn")
 	//logger.EnableLevel("info")
@@ -181,18 +199,20 @@ func main() {
 	n := node{}
 	// Now start Yggdrasil - this starts the DHT, router, switch and other core
 	// components needed for Yggdrasil to operate
-	state, err := n.core.Start(cfg, logger)
+	n.state, err = n.core.Start(cfg, logger)
 	if err != nil {
 		logger.Errorln("An error occurred during startup")
 		panic(err)
 	}
+	// Register the session firewall gatekeeper function
+	n.core.SetSessionGatekeeper(n.sessionFirewall)
 	// Start the admin socket
-	n.admin.Init(&n.core, state, logger, nil)
+	n.admin.Init(&n.core, n.state, logger, nil)
 	if err := n.admin.Start(); err != nil {
 		logger.Errorln("An error occurred starting admin socket:", err)
 	}
 	// Start the multicast interface
-	n.multicast.Init(&n.core, state, logger, nil)
+	n.multicast.Init(&n.core, n.state, logger, nil)
 	if err := n.multicast.Start(); err != nil {
 		logger.Errorln("An error occurred starting multicast:", err)
 	}
@@ -200,7 +220,7 @@ func main() {
 	// Start the TUN/TAP interface
 	if listener, err := n.core.ConnListen(); err == nil {
 		if dialer, err := n.core.ConnDialer(); err == nil {
-			n.tuntap.Init(state, logger, listener, dialer)
+			n.tuntap.Init(n.state, logger, listener, dialer)
 			if err := n.tuntap.Start(); err != nil {
 				logger.Errorln("An error occurred starting TUN/TAP:", err)
 			}
@@ -211,11 +231,6 @@ func main() {
 	} else {
 		logger.Errorln("Unable to get Listener:", err)
 	}
-	// The Stop function ensures that the TUN/TAP adapter is correctly shut down
-	// before the program exits.
-	defer func() {
-		n.core.Stop()
-	}()
 	// Make some nice output that tells us what our IPv6 address and subnet are.
 	// This is just logged to stdout for the user.
 	address := n.core.Address()
@@ -227,15 +242,15 @@ func main() {
 	r := make(chan os.Signal, 1)
 	signal.Notify(c, os.Interrupt, syscall.SIGTERM)
 	signal.Notify(r, os.Interrupt, syscall.SIGHUP)
-	// Create a function to capture the service being stopped on Windows.
+	// Capture the service being stopped on Windows.
-	winTerminate := func() {
+	minwinsvc.SetOnExit(n.shutdown)
-		c <- os.Interrupt
+	defer n.shutdown()
-	}
-	minwinsvc.SetOnExit(winTerminate)
 	// Wait for the terminate/interrupt signal. Once a signal is received, the
 	// deferred Stop function above will run which will shut down TUN/TAP.
 	for {
 		select {
+		case _ = <-c:
+			goto exit
 		case _ = <-r:
 			if *useconffile != "" {
 				cfg = readConfig(useconf, useconffile, normaliseconf)
@@ -245,9 +260,78 @@ func main() {
 			} else {
 				logger.Errorln("Reloading config at runtime is only possible with -useconffile")
 			}
-		case _ = <-c:
-			goto exit
 		}
 	}
 exit:
 }
+
+func (n *node) shutdown() {
+	n.core.Stop()
+	n.admin.Stop()
+	n.multicast.Stop()
+	n.tuntap.Stop()
+	os.Exit(0)
+}
+
+func (n *node) sessionFirewall(pubkey *crypto.BoxPubKey, initiator bool) bool {
+	n.state.Mutex.RLock()
+	defer n.state.Mutex.RUnlock()
+
+	// Allow by default if the session firewall is disabled
+	if !n.state.Current.SessionFirewall.Enable {
+		return true
+	}
+
+	// Prepare for checking whitelist/blacklist
+	var box crypto.BoxPubKey
+	// Reject blacklisted nodes
+	for _, b := range n.state.Current.SessionFirewall.BlacklistEncryptionPublicKeys {
+		key, err := hex.DecodeString(b)
+		if err == nil {
+			copy(box[:crypto.BoxPubKeyLen], key)
+			if box == *pubkey {
+				return false
+			}
+		}
+	}
+
+	// Allow whitelisted nodes
+	for _, b := range n.state.Current.SessionFirewall.WhitelistEncryptionPublicKeys {
+		key, err := hex.DecodeString(b)
+		if err == nil {
+			copy(box[:crypto.BoxPubKeyLen], key)
+			if box == *pubkey {
+				return true
+			}
+		}
+	}
+
+	// Allow outbound sessions if appropriate
+	if n.state.Current.SessionFirewall.AlwaysAllowOutbound {
+		if initiator {
+			return true
+		}
+	}
+
+	// Look and see if the pubkey is that of a direct peer
+	var isDirectPeer bool
+	for _, peer := range n.core.GetPeers() {
+		if peer.PublicKey == *pubkey {
+			isDirectPeer = true
+			break
+		}
+	}
+
+	// Allow direct peers if appropriate
+	if n.state.Current.SessionFirewall.AllowFromDirect && isDirectPeer {
+		return true
+	}
+
+	// Allow remote nodes if appropriate
+	if n.state.Current.SessionFirewall.AllowFromRemote && !isDirectPeer {
+		return true
+	}
+
+	// Finally, default-deny if not matching any of the above rules
+	return false
+}

doc/Whitepaper.md

@@ -1,4 +1,4 @@
-# Yggdasil
+# Yggdrasil
 
 Note: This is a very rough early draft.
 

go.mod

@@ -3,6 +3,7 @@ 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/hashicorp/go-syslog v1.0.0
 	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

go.sum

@@ -2,6 +2,8 @@ github.com/docker/libcontainer v2.2.1+incompatible h1:++SbbkCw+X8vAd4j2gOCzZ2Nn7
 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/hashicorp/go-syslog v1.0.0 h1:KaodqZuhUoZereWVIYmpUgZysurB1kBLX2j0MwMrUAE=
+github.com/hashicorp/go-syslog v1.0.0/go.mod h1:qPfqrKkXGihmCqbJM2mZgkZGvKG1dFdvsLplgctolz4=
 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=

src/admin/admin.go

@@ -381,11 +381,11 @@ func (a *AdminSocket) handleRequest(conn net.Conn) {
 		if r != nil {
 			send = Info{
 				"status": "error",
-				"error":  "Unrecoverable error, possibly as a result of invalid input types or malformed syntax",
+				"error":  "Check your syntax and input types",
 			}
-			a.log.Errorln("Admin socket error:", r)
+			a.log.Debugln("Admin socket error:", r)
 			if err := encoder.Encode(&send); err != nil {
-				a.log.Errorln("Admin socket JSON encode error:", err)
+				a.log.Debugln("Admin socket JSON encode error:", err)
 			}
 			conn.Close()
 		}
@@ -407,13 +407,14 @@ func (a *AdminSocket) handleRequest(conn net.Conn) {
 		send["request"] = recv
 		send["status"] = "error"
 
+		n := strings.ToLower(recv["request"].(string))
+
 		if _, ok := recv["request"]; !ok {
 			send["error"] = "No request sent"
-			break
+			goto respond
 		}
 
-		n := strings.ToLower(recv["request"].(string))
+		if h, ok := a.handlers[n]; ok {
-		if h, ok := a.handlers[strings.ToLower(n)]; ok {
 			// Check that we have all the required arguments
 			for _, arg := range h.args {
 				// An argument in [square brackets] is optional and not required,
@@ -428,7 +429,7 @@ func (a *AdminSocket) handleRequest(conn net.Conn) {
 						"error":     "Expected field missing: " + arg,
 						"expecting": arg,
 					}
-					break
+					goto respond
 				}
 			}
 
@@ -439,16 +440,28 @@ func (a *AdminSocket) handleRequest(conn net.Conn) {
 				send["error"] = err.Error()
 				if response != nil {
 					send["response"] = response
+					goto respond
 				}
 			} else {
 				send["status"] = "success"
 				if response != nil {
 					send["response"] = response
+					goto respond
 				}
 			}
+		} else {
+			// Start with a clean response on each request, which defaults to an error
+			// state. If a handler is found below then this will be overwritten
+			send = Info{
+				"request": recv,
+				"status":  "error",
+				"error":   fmt.Sprintf("Unknown action '%s', try 'list' for help", recv["request"].(string)),
+			}
+			goto respond
 		}
 
 		// Send the response back
+	respond:
 		if err := encoder.Encode(&send); err != nil {
 			return
 		}

src/config/config.go

@@ -102,7 +102,7 @@ func GenerateConfig() *NodeConfig {
 	cfg.Peers = []string{}
 	cfg.InterfacePeers = map[string][]string{}
 	cfg.AllowedEncryptionPublicKeys = []string{}
-	cfg.MulticastInterfaces = []string{".*"}
+	cfg.MulticastInterfaces = defaults.GetDefaults().DefaultMulticastInterfaces
 	cfg.IfName = defaults.GetDefaults().DefaultIfName
 	cfg.IfMTU = defaults.GetDefaults().DefaultIfMTU
 	cfg.IfTAPMode = defaults.GetDefaults().DefaultIfTAPMode

src/defaults/defaults.go

@@ -10,6 +10,9 @@ type platformDefaultParameters struct {
 	// Configuration (used for yggdrasilctl)
 	DefaultConfigFile string
 
+	// Multicast interfaces
+	DefaultMulticastInterfaces []string
+
 	// TUN/TAP
 	MaximumIfMTU     int
 	DefaultIfMTU     int

src/defaults/defaults_darwin.go

@@ -12,6 +12,12 @@ func GetDefaults() platformDefaultParameters {
 		// Configuration (used for yggdrasilctl)
 		DefaultConfigFile: "/etc/yggdrasil.conf",
 
+		// Multicast interfaces
+		DefaultMulticastInterfaces: []string{
+			"en.*",
+			"bridge.*",
+		},
+
 		// TUN/TAP
 		MaximumIfMTU:     65535,
 		DefaultIfMTU:     65535,

src/defaults/defaults_freebsd.go

@@ -12,6 +12,11 @@ func GetDefaults() platformDefaultParameters {
 		// Configuration (used for yggdrasilctl)
 		DefaultConfigFile: "/etc/yggdrasil.conf",
 
+		// Multicast interfaces
+		DefaultMulticastInterfaces: []string{
+			".*",
+		},
+
 		// TUN/TAP
 		MaximumIfMTU:     32767,
 		DefaultIfMTU:     32767,

src/defaults/defaults_linux.go

@@ -12,6 +12,11 @@ func GetDefaults() platformDefaultParameters {
 		// Configuration (used for yggdrasilctl)
 		DefaultConfigFile: "/etc/yggdrasil.conf",
 
+		// Multicast interfaces
+		DefaultMulticastInterfaces: []string{
+			".*",
+		},
+
 		// TUN/TAP
 		MaximumIfMTU:     65535,
 		DefaultIfMTU:     65535,

src/defaults/defaults_netbsd.go

@@ -12,6 +12,11 @@ func GetDefaults() platformDefaultParameters {
 		// Configuration (used for yggdrasilctl)
 		DefaultConfigFile: "/etc/yggdrasil.conf",
 
+		// Multicast interfaces
+		DefaultMulticastInterfaces: []string{
+			".*",
+		},
+
 		// TUN/TAP
 		MaximumIfMTU:     9000,
 		DefaultIfMTU:     9000,

src/defaults/defaults_openbsd.go

@@ -12,6 +12,11 @@ func GetDefaults() platformDefaultParameters {
 		// Configuration (used for yggdrasilctl)
 		DefaultConfigFile: "/etc/yggdrasil.conf",
 
+		// Multicast interfaces
+		DefaultMulticastInterfaces: []string{
+			".*",
+		},
+
 		// TUN/TAP
 		MaximumIfMTU:     16384,
 		DefaultIfMTU:     16384,

src/defaults/defaults_other.go

@@ -12,6 +12,11 @@ func GetDefaults() platformDefaultParameters {
 		// Configuration (used for yggdrasilctl)
 		DefaultConfigFile: "/etc/yggdrasil.conf",
 
+		// Multicast interfaces
+		DefaultMulticastInterfaces: []string{
+			".*",
+		},
+
 		// TUN/TAP
 		MaximumIfMTU:     65535,
 		DefaultIfMTU:     65535,

src/defaults/defaults_windows.go

@@ -12,6 +12,11 @@ func GetDefaults() platformDefaultParameters {
 		// Configuration (used for yggdrasilctl)
 		DefaultConfigFile: "C:\\Program Files\\Yggdrasil\\yggdrasil.conf",
 
+		// Multicast interfaces
+		DefaultMulticastInterfaces: []string{
+			".*",
+		},
+
 		// TUN/TAP
 		MaximumIfMTU:     65535,
 		DefaultIfMTU:     65535,

src/multicast/admin.go

@@ -5,7 +5,7 @@ import "github.com/yggdrasil-network/yggdrasil-go/src/admin"
 func (m *Multicast) SetupAdminHandlers(a *admin.AdminSocket) {
 	a.AddHandler("getMulticastInterfaces", []string{}, func(in admin.Info) (admin.Info, error) {
 		var intfs []string
-		for _, v := range m.interfaces() {
+		for _, v := range m.Interfaces() {
 			intfs = append(intfs, v.Name)
 		}
 		return admin.Info{"multicast_interfaces": intfs}, nil

src/multicast/multicast.go

@@ -22,11 +22,11 @@ type Multicast struct {
 	core       *yggdrasil.Core
 	config     *config.NodeState
 	log        *log.Logger
-	reconfigure chan chan error
 	sock       *ipv6.PacketConn
 	groupAddr  string
 	listeners  map[string]*yggdrasil.TcpListener
 	listenPort uint16
+	isOpen     bool
 }
 
 // Init prepares the multicast interface for use.
@@ -34,25 +34,10 @@ func (m *Multicast) Init(core *yggdrasil.Core, state *config.NodeState, log *log
 	m.core = core
 	m.config = state
 	m.log = log
-	m.reconfigure = make(chan chan error, 1)
 	m.listeners = make(map[string]*yggdrasil.TcpListener)
 	current, _ := m.config.Get()
 	m.listenPort = current.LinkLocalTCPPort
-	go func() {
-		for {
-			e := <-m.reconfigure
-			// There's nothing particularly to do here because the multicast module
-			// already consults the config.NodeState when enumerating multicast
-			// interfaces on each pass. We just need to return nil so that the
-			// reconfiguration doesn't block indefinitely
-			e <- nil
-		}
-	}()
 	m.groupAddr = "[ff02::114]:9001"
-	// Check if we've been given any expressions
-	if count := len(m.interfaces()); count != 0 {
-		m.log.Infoln("Found", count, "multicast interface(s)")
-	}
 	return nil
 }
 
@@ -60,11 +45,6 @@ func (m *Multicast) Init(core *yggdrasil.Core, state *config.NodeState, log *log
 // listen for multicast beacons from other hosts and will advertise multicast
 // beacons out to the network.
 func (m *Multicast) Start() error {
-	current, _ := m.config.Get()
-	if len(current.MulticastInterfaces) == 0 {
-		m.log.Infoln("Multicast discovery is disabled")
-	} else {
-		m.log.Infoln("Multicast discovery is enabled")
 	addr, err := net.ResolveUDPAddr("udp", m.groupAddr)
 	if err != nil {
 		return err
@@ -82,15 +62,18 @@ func (m *Multicast) Start() error {
 		// Windows can't set this flag, so we need to handle it in other ways
 	}
 
+	m.isOpen = true
 	go m.multicastStarted()
 	go m.listen()
 	go m.announce()
-	}
+
 	return nil
 }
 
 // Stop is not implemented for multicast yet.
 func (m *Multicast) Stop() error {
+	m.isOpen = false
+	m.sock.Close()
 	return nil
 }
 
@@ -99,37 +82,19 @@ func (m *Multicast) Stop() error {
 // needed.
 func (m *Multicast) UpdateConfig(config *config.NodeConfig) {
 	m.log.Debugln("Reloading multicast configuration...")
-
 	m.config.Replace(*config)
-
-	errors := 0
-
-	components := []chan chan error{
-		m.reconfigure,
-	}
-
-	for _, component := range components {
-		response := make(chan error)
-		component <- response
-		if err := <-response; err != nil {
-			m.log.Errorln(err)
-			errors++
-		}
-	}
-
-	if errors > 0 {
-		m.log.Warnln(errors, "multicast module(s) reported errors during configuration reload")
-	} else {
 	m.log.Infoln("Multicast configuration reloaded successfully")
 }
-}
 
-func (m *Multicast) interfaces() map[string]net.Interface {
+// GetInterfaces returns the currently known/enabled multicast interfaces. It is
+// expected that UpdateInterfaces has been called at least once before calling
+// this method.
+func (m *Multicast) Interfaces() map[string]net.Interface {
+	interfaces := make(map[string]net.Interface)
 	// Get interface expressions from config
 	current, _ := m.config.Get()
 	exprs := current.MulticastInterfaces
 	// Ask the system for network interfaces
-	interfaces := make(map[string]net.Interface)
 	allifaces, err := net.Interfaces()
 	if err != nil {
 		panic(err)
@@ -173,7 +138,7 @@ func (m *Multicast) announce() {
 		panic(err)
 	}
 	for {
-		interfaces := m.interfaces()
+		interfaces := m.Interfaces()
 		// There might be interfaces that we configured listeners for but are no
 		// longer up - if that's the case then we should stop the listeners
 		for name, listener := range m.listeners {
@@ -285,6 +250,9 @@ func (m *Multicast) listen() {
 	for {
 		nBytes, rcm, fromAddr, err := m.sock.ReadFrom(bs)
 		if err != nil {
+			if !m.isOpen {
+				return
+			}
 			panic(err)
 		}
 		if rcm != nil {
@@ -307,9 +275,11 @@ func (m *Multicast) listen() {
 		if addr.IP.String() != from.IP.String() {
 			continue
 		}
+		if _, ok := m.Interfaces()[from.Zone]; ok {
 			addr.Zone = ""
 			if err := m.core.CallPeer("tcp://"+addr.String(), from.Zone); err != nil {
 				m.log.Debugln("Call from multicast failed:", err)
 			}
 		}
 	}
+}

src/multicast/multicast_darwin.go

@@ -35,12 +35,12 @@ func (m *Multicast) multicastStarted() {
 	if awdlGoroutineStarted {
 		return
 	}
-	m.log.Infoln("Multicast discovery will wake up AWDL if required")
 	awdlGoroutineStarted = true
 	for {
 		C.StopAWDLBrowsing()
-		for _, intf := range m.interfaces() {
+		for intf := range m.Interfaces() {
-			if intf.Name == "awdl0" {
+			if intf == "awdl0" {
+				m.log.Infoln("Multicast discovery is using AWDL discovery")
 				C.StartAWDLBrowsing()
 				break
 			}

src/tuntap/ckr.go

@@ -48,19 +48,21 @@ func (c *cryptokey) init(tun *TunAdapter) {
 		}
 	}()
 
+	c.tun.log.Debugln("Configuring CKR...")
 	if err := c.configure(); err != nil {
 		c.tun.log.Errorln("CKR configuration failed:", err)
+	} else {
+		c.tun.log.Debugln("CKR configured")
 	}
 }
 
 // Configure the CKR routes - this must only ever be called from the router
 // goroutine, e.g. through router.doAdmin
 func (c *cryptokey) configure() error {
-	c.tun.config.Mutex.RLock()
+	current, _ := c.tun.config.Get()
-	defer c.tun.config.Mutex.RUnlock()
 
 	// Set enabled/disabled state
-	c.setEnabled(c.tun.config.Current.TunnelRouting.Enable)
+	c.setEnabled(current.TunnelRouting.Enable)
 
 	// Clear out existing routes
 	c.mutexroutes.Lock()
@@ -69,14 +71,14 @@ func (c *cryptokey) configure() error {
 	c.mutexroutes.Unlock()
 
 	// Add IPv6 routes
-	for ipv6, pubkey := range c.tun.config.Current.TunnelRouting.IPv6Destinations {
+	for ipv6, pubkey := range current.TunnelRouting.IPv6Destinations {
 		if err := c.addRoute(ipv6, pubkey); err != nil {
 			return err
 		}
 	}
 
 	// Add IPv4 routes
-	for ipv4, pubkey := range c.tun.config.Current.TunnelRouting.IPv4Destinations {
+	for ipv4, pubkey := range current.TunnelRouting.IPv4Destinations {
 		if err := c.addRoute(ipv4, pubkey); err != nil {
 			return err
 		}
@@ -90,7 +92,7 @@ func (c *cryptokey) configure() error {
 
 	// Add IPv6 sources
 	c.ipv6sources = make([]net.IPNet, 0)
-	for _, source := range c.tun.config.Current.TunnelRouting.IPv6Sources {
+	for _, source := range current.TunnelRouting.IPv6Sources {
 		if err := c.addSourceSubnet(source); err != nil {
 			return err
 		}
@@ -98,7 +100,7 @@ func (c *cryptokey) configure() error {
 
 	// Add IPv4 sources
 	c.ipv4sources = make([]net.IPNet, 0)
-	for _, source := range c.tun.config.Current.TunnelRouting.IPv4Sources {
+	for _, source := range current.TunnelRouting.IPv4Sources {
 		if err := c.addSourceSubnet(source); err != nil {
 			return err
 		}

src/tuntap/icmpv6.go

@@ -13,6 +13,7 @@ import (
 	"encoding/binary"
 	"errors"
 	"net"
+	"sync"
 	"time"
 
 	"golang.org/x/net/icmp"
@@ -21,19 +22,18 @@ import (
 	"github.com/yggdrasil-network/yggdrasil-go/src/address"
 )
 
-type macAddress [6]byte
-
 const len_ETHER = 14
 
 type ICMPv6 struct {
 	tun           *TunAdapter
 	mylladdr      net.IP
-	mymac    macAddress
+	mymac         net.HardwareAddr
 	peermacs      map[address.Address]neighbor
+	peermacsmutex sync.RWMutex
 }
 
 type neighbor struct {
-	mac               macAddress
+	mac               net.HardwareAddr
 	learned           bool
 	lastadvertisement time.Time
 	lastsolicitation  time.Time
@@ -61,10 +61,12 @@ func ipv6Header_Marshal(h *ipv6.Header) ([]byte, error) {
 // addresses.
 func (i *ICMPv6) Init(t *TunAdapter) {
 	i.tun = t
+	i.peermacsmutex.Lock()
 	i.peermacs = make(map[address.Address]neighbor)
+	i.peermacsmutex.Unlock()
 
 	// Our MAC address and link-local address
-	i.mymac = macAddress{
+	i.mymac = net.HardwareAddr{
 		0x02, 0x00, 0x00, 0x00, 0x00, 0x02}
 	i.mylladdr = net.IP{
 		0xFE, 0x80, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
@@ -181,16 +183,30 @@ func (i *ICMPv6) UnmarshalPacket(datain []byte, datamac *[]byte) ([]byte, error)
 		if datamac != nil {
 			var addr address.Address
 			var target address.Address
-			var mac macAddress
+			mac := net.HardwareAddr{0x00, 0x00, 0x00, 0x00, 0x00, 0x00}
 			copy(addr[:], ipv6Header.Src[:])
 			copy(target[:], datain[48:64])
 			copy(mac[:], (*datamac)[:])
-			// fmt.Printf("Learning peer MAC %x for %x\n", mac, target)
+			i.peermacsmutex.Lock()
 			neighbor := i.peermacs[target]
 			neighbor.mac = mac
 			neighbor.learned = true
 			neighbor.lastadvertisement = time.Now()
 			i.peermacs[target] = neighbor
+			i.peermacsmutex.Unlock()
+			i.tun.log.Debugln("Learned peer MAC", mac.String(), "for", net.IP(target[:]).String())
+			/*
+				i.tun.log.Debugln("Peer MAC table:")
+				i.peermacsmutex.RLock()
+				for t, n := range i.peermacs {
+					if n.learned {
+						i.tun.log.Debugln("- Target", net.IP(t[:]).String(), "has MAC", n.mac.String())
+					} else {
+						i.tun.log.Debugln("- Target", net.IP(t[:]).String(), "is not learned yet")
+					}
+				}
+				i.peermacsmutex.RUnlock()
+			*/
 		}
 		return nil, errors.New("No response needed")
 	}
@@ -201,7 +217,7 @@ func (i *ICMPv6) UnmarshalPacket(datain []byte, datamac *[]byte) ([]byte, error)
 // Creates an ICMPv6 packet based on the given icmp.MessageBody and other
 // parameters, complete with ethernet and IP headers, which can be written
 // directly to a TAP adapter.
-func (i *ICMPv6) CreateICMPv6L2(dstmac macAddress, dst net.IP, src net.IP, mtype ipv6.ICMPType, mcode int, mbody icmp.MessageBody) ([]byte, error) {
+func (i *ICMPv6) CreateICMPv6L2(dstmac net.HardwareAddr, dst net.IP, src net.IP, mtype ipv6.ICMPType, mcode int, mbody icmp.MessageBody) ([]byte, error) {
 	// Pass through to CreateICMPv6
 	ipv6packet, err := CreateICMPv6(dst, src, mtype, mcode, mbody)
 	if err != nil {
@@ -264,13 +280,54 @@ func CreateICMPv6(dst net.IP, src net.IP, mtype ipv6.ICMPType, mcode int, mbody
 	return responsePacket, nil
 }
 
-func (i *ICMPv6) CreateNDPL2(dst address.Address) ([]byte, error) {
+func (i *ICMPv6) Solicit(addr address.Address) {
+	retries := 5
+	for retries > 0 {
+		retries--
+		i.peermacsmutex.RLock()
+		if n, ok := i.peermacs[addr]; ok && n.learned {
+			i.tun.log.Debugln("MAC learned for", net.IP(addr[:]).String())
+			i.peermacsmutex.RUnlock()
+			return
+		}
+		i.peermacsmutex.RUnlock()
+		i.tun.log.Debugln("Sending neighbor solicitation for", net.IP(addr[:]).String())
+		i.peermacsmutex.Lock()
+		if n, ok := i.peermacs[addr]; !ok {
+			i.peermacs[addr] = neighbor{
+				lastsolicitation: time.Now(),
+			}
+		} else {
+			n.lastsolicitation = time.Now()
+		}
+		i.peermacsmutex.Unlock()
+		request, err := i.createNDPL2(addr)
+		if err != nil {
+			panic(err)
+		}
+		if _, err := i.tun.iface.Write(request); err != nil {
+			panic(err)
+		}
+		i.tun.log.Debugln("Sent neighbor solicitation for", net.IP(addr[:]).String())
+		time.Sleep(time.Second)
+	}
+}
+
+func (i *ICMPv6) getNeighbor(addr address.Address) (neighbor, bool) {
+	i.peermacsmutex.RLock()
+	defer i.peermacsmutex.RUnlock()
+
+	n, ok := i.peermacs[addr]
+	return n, ok
+}
+
+func (i *ICMPv6) createNDPL2(dst address.Address) ([]byte, error) {
 	// Create the ND payload
 	var payload [28]byte
-	copy(payload[:4], []byte{0x00, 0x00, 0x00, 0x00})
+	copy(payload[:4], []byte{0x00, 0x00, 0x00, 0x00}) // Flags
-	copy(payload[4:20], dst[:])
+	copy(payload[4:20], dst[:])                       // Destination
-	copy(payload[20:22], []byte{0x01, 0x01})
+	copy(payload[20:22], []byte{0x01, 0x01})          // Type & length
-	copy(payload[22:28], i.mymac[:6])
+	copy(payload[22:28], i.mymac[:6])                 // Link layer address
 
 	// Create the ICMPv6 solicited-node address
 	var dstaddr address.Address
@@ -281,7 +338,7 @@ func (i *ICMPv6) CreateNDPL2(dst address.Address) ([]byte, error) {
 	copy(dstaddr[13:], dst[13:16])
 
 	// Create the multicast MAC
-	var dstmac macAddress
+	dstmac := net.HardwareAddr{0x00, 0x00, 0x00, 0x00, 0x00, 0x00}
 	copy(dstmac[:2], []byte{0x33, 0x33})
 	copy(dstmac[2:6], dstaddr[12:16])
 
@@ -293,9 +350,6 @@ func (i *ICMPv6) CreateNDPL2(dst address.Address) ([]byte, error) {
 	if err != nil {
 		return nil, err
 	}
-	neighbor := i.peermacs[dstaddr]
-	neighbor.lastsolicitation = time.Now()
-	i.peermacs[dstaddr] = neighbor
 
 	return requestPacket, nil
 }
@@ -319,10 +373,10 @@ func (i *ICMPv6) HandleNDP(in []byte) ([]byte, error) {
 
 	// Create our NDP message body response
 	body := make([]byte, 28)
-	binary.BigEndian.PutUint32(body[:4], uint32(0x20000000))
+	binary.BigEndian.PutUint32(body[:4], uint32(0x40000000)) // Flags
 	copy(body[4:20], in[8:24])                               // Target address
-	body[20] = uint8(2)
+	body[20] = uint8(2)                                      // Type: Target link-layer address
-	body[21] = uint8(1)
+	body[21] = uint8(1)                                      // Length: 1x address (8 bytes)
 	copy(body[22:28], i.mymac[:6])
 
 	// Send it back

src/tuntap/iface.go

@@ -3,6 +3,7 @@ package tuntap
 import (
 	"bytes"
 	"errors"
+	"net"
 	"time"
 
 	"github.com/songgao/packets/ethernet"
@@ -40,22 +41,13 @@ func (tun *TunAdapter) writer() error {
 				return errors.New("Invalid address family")
 			}
 			sendndp := func(dstAddr address.Address) {
-				neigh, known := tun.icmpv6.peermacs[dstAddr]
+				neigh, known := tun.icmpv6.getNeighbor(dstAddr)
 				known = known && (time.Since(neigh.lastsolicitation).Seconds() < 30)
 				if !known {
-					request, err := tun.icmpv6.CreateNDPL2(dstAddr)
+					tun.icmpv6.Solicit(dstAddr)
-					if err != nil {
-						panic(err)
-					}
-					if _, err := tun.iface.Write(request); err != nil {
-						panic(err)
-					}
-					tun.icmpv6.peermacs[dstAddr] = neighbor{
-						lastsolicitation: time.Now(),
 				}
 			}
-			}
+			peermac := net.HardwareAddr{0x00, 0x00, 0x00, 0x00, 0x00, 0x00}
-			var peermac macAddress
 			var peerknown bool
 			if b[0]&0xf0 == 0x40 {
 				dstAddr = tun.addr
@@ -64,15 +56,20 @@ func (tun *TunAdapter) writer() error {
 					dstAddr = tun.addr
 				}
 			}
-			if neighbor, ok := tun.icmpv6.peermacs[dstAddr]; ok && neighbor.learned {
+			if neighbor, ok := tun.icmpv6.getNeighbor(dstAddr); ok && neighbor.learned {
+				// If we've learned the MAC of a 300::/7 address, for example, or a CKR
+				// address, use the MAC address of that
 				peermac = neighbor.mac
 				peerknown = true
-			} else if neighbor, ok := tun.icmpv6.peermacs[tun.addr]; ok && neighbor.learned {
+			} else if neighbor, ok := tun.icmpv6.getNeighbor(tun.addr); ok && neighbor.learned {
+				// Otherwise send directly to the MAC address of the host if that's
+				// known instead
 				peermac = neighbor.mac
 				peerknown = true
-				sendndp(dstAddr)
 			} else {
+				// Nothing has been discovered, try to discover the destination
 				sendndp(tun.addr)
+
 			}
 			if peerknown {
 				var proto ethernet.Ethertype
@@ -92,12 +89,17 @@ func (tun *TunAdapter) writer() error {
 				copy(frame[tun_ETHER_HEADER_LENGTH:], b[:n])
 				n += tun_ETHER_HEADER_LENGTH
 				w, err = tun.iface.Write(frame[:n])
+			} else {
+				tun.log.Errorln("TUN/TAP iface write error: no peer MAC known for", net.IP(dstAddr[:]).String(), "- dropping packet")
 			}
 		} else {
 			w, err = tun.iface.Write(b[:n])
 			util.PutBytes(b)
 		}
 		if err != nil {
+			if !tun.isOpen {
+				return err
+			}
 			tun.log.Errorln("TUN/TAP iface write error:", err)
 			continue
 		}
@@ -114,6 +116,9 @@ func (tun *TunAdapter) reader() error {
 		// Wait for a packet to be delivered to us through the TUN/TAP adapter
 		n, err := tun.iface.Read(bs)
 		if err != nil {
+			if !tun.isOpen {
+				return err
+			}
 			panic(err)
 		}
 		if n == 0 {
@@ -134,7 +139,7 @@ func (tun *TunAdapter) reader() error {
 			}
 			// Then offset the buffer so that we can now just treat it as an IP
 			// packet from now on
-			bs = bs[offset:]
+			bs = bs[offset:] // FIXME this breaks bs for the next read and means n is the wrong value
 		}
 		// From the IP header, work out what our source and destination addresses
 		// and node IDs are. We will need these in order to work out where to send
@@ -178,7 +183,7 @@ func (tun *TunAdapter) reader() error {
 			// Unknown address length or protocol, so drop the packet and ignore it
 			continue
 		}
-		if !tun.ckr.isValidSource(srcAddr, addrlen) {
+		if tun.ckr.isEnabled() && !tun.ckr.isValidSource(srcAddr, addrlen) {
 			// The packet had a source address that doesn't belong to us or our
 			// configured crypto-key routing source subnets
 			continue
@@ -225,22 +230,47 @@ func (tun *TunAdapter) reader() error {
 				panic("Given empty dstNodeID and dstNodeIDMask - this shouldn't happen")
 			}
 			// Dial to the remote node
+			packet := append(util.GetBytes(), bs[:n]...)
+			go func() {
+				// FIXME just spitting out a goroutine to do this is kind of ugly and means we drop packets until the dial finishes
+				tun.mutex.Lock()
+				_, known := tun.dials[*dstNodeID]
+				tun.dials[*dstNodeID] = append(tun.dials[*dstNodeID], packet)
+				for len(tun.dials[*dstNodeID]) > 32 {
+					util.PutBytes(tun.dials[*dstNodeID][0])
+					tun.dials[*dstNodeID] = tun.dials[*dstNodeID][1:]
+				}
+				tun.mutex.Unlock()
+				if known {
+					return
+				}
+				var tc *tunConn
 				if conn, err := tun.dialer.DialByNodeIDandMask(dstNodeID, dstNodeIDMask); err == nil {
 					// We've been given a connection so prepare the session wrapper
-				if s, err := tun.wrap(conn); err != nil {
+					if tc, err = tun.wrap(conn); err != nil {
 						// Something went wrong when storing the connection, typically that
 						// something already exists for this address or subnet
 						tun.log.Debugln("TUN/TAP iface wrap:", err)
-				} else {
-					// Update our reference to the connection
-					session, isIn = s, true
 					}
-			} else {
+				}
-				// We weren't able to dial for some reason so there's no point in
+				tun.mutex.Lock()
+				packets := tun.dials[*dstNodeID]
+				delete(tun.dials, *dstNodeID)
+				tun.mutex.Unlock()
+				if tc != nil {
+					for _, packet := range packets {
+						select {
+						case tc.send <- packet:
+						default:
+							util.PutBytes(packet)
+						}
+					}
+				}
+			}()
+			// While the dial is going on we can't do much else
 			// continuing this iteration - skip to the next one
 			continue
 		}
-		}
 		// If we have a connection now, try writing to it
 		if isIn && session != nil {
 			packet := append(util.GetBytes(), bs[:n]...)

src/tuntap/tun.go

@@ -14,7 +14,6 @@ import (
 	"fmt"
 	"net"
 	"sync"
-	"time"
 
 	"github.com/gologme/log"
 	"github.com/yggdrasil-network/water"
@@ -49,6 +48,7 @@ type TunAdapter struct {
 	mutex        sync.RWMutex // Protects the below
 	addrToConn   map[address.Address]*tunConn
 	subnetToConn map[address.Subnet]*tunConn
+	dials        map[crypto.NodeID][][]byte // Buffer of packets to send after dialing finishes
 	isOpen       bool
 }
 
@@ -113,18 +113,18 @@ func (tun *TunAdapter) Init(config *config.NodeState, log *log.Logger, listener
 	tun.dialer = dialer
 	tun.addrToConn = make(map[address.Address]*tunConn)
 	tun.subnetToConn = make(map[address.Subnet]*tunConn)
+	tun.dials = make(map[crypto.NodeID][][]byte)
 }
 
 // Start the setup process for the TUN/TAP adapter. If successful, starts the
 // read/write goroutines to handle packets on that interface.
 func (tun *TunAdapter) Start() error {
-	tun.config.Mutex.RLock()
+	current, _ := tun.config.Get()
-	defer tun.config.Mutex.RUnlock()
 	if tun.config == nil || tun.listener == nil || tun.dialer == nil {
 		return errors.New("No configuration available to TUN/TAP")
 	}
 	var boxPub crypto.BoxPubKey
-	boxPubHex, err := hex.DecodeString(tun.config.Current.EncryptionPublicKey)
+	boxPubHex, err := hex.DecodeString(current.EncryptionPublicKey)
 	if err != nil {
 		return err
 	}
@@ -132,9 +132,9 @@ func (tun *TunAdapter) Start() error {
 	nodeID := crypto.GetNodeID(&boxPub)
 	tun.addr = *address.AddrForNodeID(nodeID)
 	tun.subnet = *address.SubnetForNodeID(nodeID)
-	tun.mtu = tun.config.Current.IfMTU
+	tun.mtu = current.IfMTU
-	ifname := tun.config.Current.IfName
+	ifname := current.IfName
-	iftapmode := tun.config.Current.IfTAPMode
+	iftapmode := current.IfTAPMode
 	addr := fmt.Sprintf("%s/%d", net.IP(tun.addr[:]).String(), 8*len(address.GetPrefix())-1)
 	if ifname != "none" {
 		if err := tun.setup(ifname, iftapmode, addr, tun.mtu); err != nil {
@@ -150,21 +150,6 @@ func (tun *TunAdapter) Start() error {
 	tun.send = make(chan []byte, 32) // TODO: is this a sensible value?
 	tun.reconfigure = make(chan chan error)
 	tun.mutex.Unlock()
-	if iftapmode {
-		go func() {
-			for {
-				if _, ok := tun.icmpv6.peermacs[tun.addr]; ok {
-					break
-				}
-				request, err := tun.icmpv6.CreateNDPL2(tun.addr)
-				if err != nil {
-					panic(err)
-				}
-				tun.send <- request
-				time.Sleep(time.Second)
-			}
-		}()
-	}
 	go func() {
 		for {
 			e := <-tun.reconfigure
@@ -175,10 +160,23 @@ func (tun *TunAdapter) Start() error {
 	go tun.reader()
 	go tun.writer()
 	tun.icmpv6.Init(tun)
+	if iftapmode {
+		go tun.icmpv6.Solicit(tun.addr)
+	}
 	tun.ckr.init(tun)
 	return nil
 }
 
+// Start the setup process for the TUN/TAP adapter. If successful, starts the
+// read/write goroutines to handle packets on that interface.
+func (tun *TunAdapter) Stop() error {
+	tun.isOpen = false
+	// TODO: we have nothing that cleanly stops all the various goroutines opened
+	// by TUN/TAP, e.g. readers/writers, sessions
+	tun.iface.Close()
+	return nil
+}
+
 // UpdateConfig updates the TUN/TAP module with the provided config.NodeConfig
 // and then signals the various module goroutines to reconfigure themselves if
 // needed.
@@ -235,6 +233,7 @@ func (tun *TunAdapter) wrap(conn *yggdrasil.Conn) (c *tunConn, err error) {
 		stop:  make(chan struct{}),
 		alive: make(chan struct{}, 1),
 	}
+	c = &s
 	// Get the remote address and subnet of the other side
 	remoteNodeID := conn.RemoteAddr()
 	s.addr = *address.AddrForNodeID(&remoteNodeID)

src/tuntap/tun_windows.go

@@ -31,18 +31,18 @@ func (tun *TunAdapter) setup(ifname string, iftapmode bool, addr string, mtu int
 	}
 	// Disable/enable the interface to resets its configuration (invalidating iface)
 	cmd := exec.Command("netsh", "interface", "set", "interface", iface.Name(), "admin=DISABLED")
-	tun.log.Printf("netsh command: %v", strings.Join(cmd.Args, " "))
+	tun.log.Debugln("netsh command:", strings.Join(cmd.Args, " "))
 	output, err := cmd.CombinedOutput()
 	if err != nil {
-		tun.log.Errorf("Windows netsh failed: %v.", err)
+		tun.log.Errorln("Windows netsh failed:", err)
 		tun.log.Traceln(string(output))
 		return err
 	}
 	cmd = exec.Command("netsh", "interface", "set", "interface", iface.Name(), "admin=ENABLED")
-	tun.log.Printf("netsh command: %v", strings.Join(cmd.Args, " "))
+	tun.log.Debugln("netsh command:", strings.Join(cmd.Args, " "))
 	output, err = cmd.CombinedOutput()
 	if err != nil {
-		tun.log.Errorf("Windows netsh failed: %v.", err)
+		tun.log.Errorln("Windows netsh failed:", err)
 		tun.log.Traceln(string(output))
 		return err
 	}
@@ -71,10 +71,10 @@ func (tun *TunAdapter) setupMTU(mtu int) error {
 		fmt.Sprintf("interface=%s", tun.iface.Name()),
 		fmt.Sprintf("mtu=%d", mtu),
 		"store=active")
-	tun.log.Debugln("netsh command: %v", strings.Join(cmd.Args, " "))
+	tun.log.Debugln("netsh command:", strings.Join(cmd.Args, " "))
 	output, err := cmd.CombinedOutput()
 	if err != nil {
-		tun.log.Errorf("Windows netsh failed: %v.", err)
+		tun.log.Errorln("Windows netsh failed:", err)
 		tun.log.Traceln(string(output))
 		return err
 	}
@@ -88,10 +88,10 @@ func (tun *TunAdapter) setupAddress(addr string) error {
 		fmt.Sprintf("interface=%s", tun.iface.Name()),
 		fmt.Sprintf("addr=%s", addr),
 		"store=active")
-	tun.log.Debugln("netsh command: %v", strings.Join(cmd.Args, " "))
+	tun.log.Debugln("netsh command:", strings.Join(cmd.Args, " "))
 	output, err := cmd.CombinedOutput()
 	if err != nil {
-		tun.log.Errorf("Windows netsh failed: %v.", err)
+		tun.log.Errorln("Windows netsh failed:", err)
 		tun.log.Traceln(string(output))
 		return err
 	}

src/yggdrasil/api.go

@@ -211,8 +211,9 @@ func (c *Core) GetSessions() []Session {
 	var sessions []Session
 	getSessions := func() {
 		for _, sinfo := range c.sessions.sinfos {
-			// TODO? skipped known but timed out sessions?
+			var session Session
-			session := Session{
+			workerFunc := func() {
+				session = Session{
 					Coords:      append([]byte{}, sinfo.coords...),
 					MTU:         sinfo.getMTU(),
 					BytesSent:   sinfo.bytesSent,
@@ -221,6 +222,20 @@ func (c *Core) GetSessions() []Session {
 					WasMTUFixed: sinfo.wasMTUFixed,
 				}
 				copy(session.PublicKey[:], sinfo.theirPermPub[:])
+			}
+			var skip bool
+			func() {
+				defer func() {
+					if recover() != nil {
+						skip = true
+					}
+				}()
+				sinfo.doWorker(workerFunc)
+			}()
+			if skip {
+				continue
+			}
+			// TODO? skipped known but timed out sessions?
 			sessions = append(sessions, session)
 		}
 	}
@@ -232,7 +247,7 @@ func (c *Core) GetSessions() []Session {
 // from git, or returns "unknown" otherwise.
 func BuildName() string {
 	if buildName == "" {
-		return "unknown"
+		return "yggdrasil"
 	}
 	return buildName
 }
@@ -395,6 +410,19 @@ func (c *Core) GetNodeInfo(keyString, coordString string, nocache bool) (NodeInf
 	return NodeInfoPayload{}, errors.New(fmt.Sprintf("getNodeInfo timeout: %s", keyString))
 }
 
+// SetSessionGatekeeper allows you to configure a handler function for deciding
+// whether a session should be allowed or not. The default session firewall is
+// implemented in this way. The function receives the public key of the remote
+// side and a boolean which is true if we initiated the session or false if we
+// received an incoming session request. The function should return true to
+// allow the session or false to reject it.
+func (c *Core) SetSessionGatekeeper(f func(pubkey *crypto.BoxPubKey, initiator bool) bool) {
+	c.sessions.isAllowedMutex.Lock()
+	defer c.sessions.isAllowedMutex.Unlock()
+
+	c.sessions.isAllowedHandler = f
+}
+
 // SetLogger sets the output logger of the Yggdrasil node after startup. This
 // may be useful if you want to redirect the output later.
 func (c *Core) SetLogger(log *log.Logger) {
@@ -504,21 +532,14 @@ func (c *Core) DHTPing(keyString, coordString, targetString string) (DHTRes, err
 	rq := dhtReqKey{info.key, target}
 	sendPing := func() {
 		c.dht.addCallback(&rq, func(res *dhtRes) {
-			defer func() { recover() }()
+			resCh <- res
-			select {
-			case resCh <- res:
-			default:
-			}
 		})
 		c.dht.ping(&info, &target)
 	}
 	c.router.doAdmin(sendPing)
-	go func() {
-		time.Sleep(6 * time.Second)
-		close(resCh)
-	}()
 	// TODO: do something better than the below...
-	for res := range resCh {
+	res := <-resCh
+	if res != nil {
 		r := DHTRes{
 			Coords: append([]byte{}, res.Coords...),
 		}

src/yggdrasil/conn.go

@@ -47,9 +47,6 @@ type Conn struct {
 	session       *sessionInfo
 	readDeadline  atomic.Value // time.Time // TODO timer
 	writeDeadline atomic.Value // time.Time // TODO timer
-	searching     atomic.Value  // bool
-	searchwait    chan struct{} // Never reset this, it's only used for the initial search
-	writebuf      [][]byte      // Packets to be sent if/when the search finishes
 }
 
 // TODO func NewConn() that initializes additional fields as needed
@@ -59,9 +56,7 @@ func newConn(core *Core, nodeID *crypto.NodeID, nodeMask *crypto.NodeID, session
 		nodeID:   nodeID,
 		nodeMask: nodeMask,
 		session:  session,
-		searchwait: make(chan struct{}),
 	}
-	conn.searching.Store(false)
 	return &conn
 }
 
@@ -69,91 +64,44 @@ func (c *Conn) String() string {
 	return fmt.Sprintf("conn=%p", c)
 }
 
-// This method should only be called from the router goroutine
+// This should never be called from the router goroutine
-func (c *Conn) startSearch() {
+func (c *Conn) search() error {
-	// The searchCompleted callback is given to the search
+	var sinfo *searchInfo
-	searchCompleted := func(sinfo *sessionInfo, err error) {
+	var isIn bool
-		defer c.searching.Store(false)
+	c.core.router.doAdmin(func() { sinfo, isIn = c.core.searches.searches[*c.nodeID] })
-		// If the search failed for some reason, e.g. it hit a dead end or timed
+	if !isIn {
-		// out, then do nothing
+		done := make(chan struct{}, 1)
-		if err != nil {
+		var sess *sessionInfo
-			c.core.log.Debugln(c.String(), "DHT search failed:", err)
+		var err error
-			return
+		searchCompleted := func(sinfo *sessionInfo, e error) {
+			sess = sinfo
+			err = e
+			// FIXME close can be called multiple times, do a non-blocking send instead
+			select {
+			case done <- struct{}{}:
+			default:
 			}
-		// Take the connection mutex
+		}
-		c.mutex.Lock()
+		c.core.router.doAdmin(func() {
-		defer c.mutex.Unlock()
+			sinfo = c.core.searches.newIterSearch(c.nodeID, c.nodeMask, searchCompleted)
-		// Were we successfully given a sessionInfo pointer?
+			sinfo.continueSearch()
-		if sinfo != nil {
+		})
-			// Store it, and update the nodeID and nodeMask (which may have been
+		<-done
-			// wildcarded before now) with their complete counterparts
+		c.session = sess
-			c.core.log.Debugln(c.String(), "DHT search completed")
+		if c.session == nil && err == nil {
-			c.session = sinfo
+			panic("search failed but returned no error")
-			c.nodeID = crypto.GetNodeID(&sinfo.theirPermPub)
+		}
+		if c.session != nil {
+			c.nodeID = crypto.GetNodeID(&c.session.theirPermPub)
 			for i := range c.nodeMask {
 				c.nodeMask[i] = 0xFF
 			}
-			// Make sure that any blocks on read/write operations are lifted
+		}
-			defer func() { recover() }() // So duplicate searches don't panic
+		return err
-			close(c.searchwait)
 	} else {
-			// No session was returned - this shouldn't really happen because we
+		return errors.New("search already exists")
-			// should always return an error reason if we don't return a session
-			panic("DHT search didn't return an error or a sessionInfo")
-		}
-		if c.closed {
-			// Things were closed before the search returned
-			// Go ahead and close it again to make sure the session is cleaned up
-			go c.Close()
-		} else {
-			// Send any messages we may have buffered
-			var msgs [][]byte
-			msgs, c.writebuf = c.writebuf, nil
-			go func() {
-				for _, msg := range msgs {
-					c.Write(msg)
-					util.PutBytes(msg)
-				}
-			}()
-		}
-	}
-	// doSearch will be called below in response to one or more conditions
-	doSearch := func() {
-		c.searching.Store(true)
-		// Check to see if there is a search already matching the destination
-		sinfo, isIn := c.core.searches.searches[*c.nodeID]
-		if !isIn {
-			// Nothing was found, so create a new search
-			sinfo = c.core.searches.newIterSearch(c.nodeID, c.nodeMask, searchCompleted)
-			c.core.log.Debugf("%s DHT search started: %p", c.String(), sinfo)
-		}
-		// Continue the search
-		c.core.searches.continueSearch(sinfo)
-	}
-	// Take a copy of the session object, in case it changes later
-	c.mutex.RLock()
-	sinfo := c.session
-	c.mutex.RUnlock()
-	if c.session == nil {
-		// No session object is present so previous searches, if we ran any, have
-		// not yielded a useful result (dead end, remote host not found)
-		doSearch()
-	} else {
-		sinfo.worker <- func() {
-			switch {
-			case !sinfo.init:
-				doSearch()
-			case time.Since(sinfo.time) > 6*time.Second:
-				if sinfo.time.Before(sinfo.pingTime) && time.Since(sinfo.pingTime) > 6*time.Second {
-					// TODO double check that the above condition is correct
-					doSearch()
-				} else {
-					c.core.sessions.ping(sinfo)
-				}
-			default: // Don't do anything, to keep traffic throttled
-			}
-		}
 	}
+	return nil
 }
 
 func getDeadlineTimer(value *atomic.Value) *time.Timer {
@@ -167,30 +115,9 @@ func getDeadlineTimer(value *atomic.Value) *time.Timer {
 
 func (c *Conn) Read(b []byte) (int, error) {
 	// Take a copy of the session object
-	c.mutex.RLock()
 	sinfo := c.session
-	c.mutex.RUnlock()
 	timer := getDeadlineTimer(&c.readDeadline)
 	defer util.TimerStop(timer)
-	// If there is a search in progress then wait for the result
-	if sinfo == nil {
-		// Wait for the search to complete
-		select {
-		case <-c.searchwait:
-		case <-timer.C:
-			return 0, ConnError{errors.New("timeout"), true, false, 0}
-		}
-		// Retrieve our session info again
-		c.mutex.RLock()
-		sinfo = c.session
-		c.mutex.RUnlock()
-		// If sinfo is still nil at this point then the search failed and the
-		// searchwait channel has been recreated, so might as well give up and
-		// return an error code
-		if sinfo == nil {
-			return 0, errors.New("search failed")
-		}
-	}
 	for {
 		// Wait for some traffic to come through from the session
 		select {
@@ -232,6 +159,12 @@ func (c *Conn) Read(b []byte) (int, error) {
 				sinfo.bytesRecvd += uint64(len(b))
 			}
 			// Hand over to the session worker
+			defer func() {
+				if recover() != nil {
+					err = errors.New("read failed, session already closed")
+					close(done)
+				}
+			}() // In case we're racing with a close
 			select { // Send to worker
 			case sinfo.worker <- workerFunc:
 			case <-timer.C:
@@ -253,32 +186,7 @@ func (c *Conn) Read(b []byte) (int, error) {
 }
 
 func (c *Conn) Write(b []byte) (bytesWritten int, err error) {
-	c.mutex.RLock()
 	sinfo := c.session
-	c.mutex.RUnlock()
-	// If the session doesn't exist, or isn't initialised (which probably means
-	// that the search didn't complete successfully) then we may need to wait for
-	// the search to complete or start the search again
-	if sinfo == nil || !sinfo.init {
-		// Is a search already taking place?
-		if searching, sok := c.searching.Load().(bool); !sok || (sok && !searching) {
-			// No search was already taking place so start a new one
-			c.core.router.doAdmin(c.startSearch)
-		}
-		// Buffer the packet to be sent if/when the search is finished
-		c.mutex.Lock()
-		defer c.mutex.Unlock()
-		c.writebuf = append(c.writebuf, append(util.GetBytes(), b...))
-		for len(c.writebuf) > 32 {
-			util.PutBytes(c.writebuf[0])
-			c.writebuf = c.writebuf[1:]
-		}
-		return len(b), nil
-	} else {
-		// This triggers some session keepalive traffic
-		// FIXME this desparately needs to be refactored, since the ping case needlessly goes through the router goroutine just to have it pass a function to the session worker when it determines that a session already exists.
-		c.core.router.doAdmin(c.startSearch)
-	}
 	var packet []byte
 	done := make(chan struct{})
 	written := len(b)
@@ -301,11 +209,45 @@ func (c *Conn) Write(b []byte) (bytesWritten int, err error) {
 		}
 		packet = p.encode()
 		sinfo.bytesSent += uint64(len(b))
+		// The rest of this work is session keep-alive traffic
+		doSearch := func() {
+			routerWork := func() {
+				// Check to see if there is a search already matching the destination
+				sinfo, isIn := c.core.searches.searches[*c.nodeID]
+				if !isIn {
+					// Nothing was found, so create a new search
+					searchCompleted := func(sinfo *sessionInfo, e error) {}
+					sinfo = c.core.searches.newIterSearch(c.nodeID, c.nodeMask, searchCompleted)
+					c.core.log.Debugf("%s DHT search started: %p", c.String(), sinfo)
+				}
+				// Continue the search
+				sinfo.continueSearch()
+			}
+			go func() { c.core.router.admin <- routerWork }()
+		}
+		switch {
+		case time.Since(sinfo.time) > 6*time.Second:
+			if sinfo.time.Before(sinfo.pingTime) && time.Since(sinfo.pingTime) > 6*time.Second {
+				// TODO double check that the above condition is correct
+				doSearch()
+			} else {
+				sinfo.core.sessions.ping(sinfo)
+			}
+		case sinfo.reset && sinfo.pingTime.Before(sinfo.time):
+			sinfo.core.sessions.ping(sinfo)
+		default: // Don't do anything, to keep traffic throttled
+		}
 	}
 	// Set up a timer so this doesn't block forever
 	timer := getDeadlineTimer(&c.writeDeadline)
 	defer util.TimerStop(timer)
 	// Hand over to the session worker
+	defer func() {
+		if recover() != nil {
+			err = errors.New("write failed, session already closed")
+			close(done)
+		}
+	}() // In case we're racing with a close
 	select { // Send to worker
 	case sinfo.worker <- workerFunc:
 	case <-timer.C:
@@ -326,8 +268,7 @@ func (c *Conn) Close() error {
 	defer c.mutex.Unlock()
 	if c.session != nil {
 		// Close the session, if it hasn't been closed already
-		c.session.close()
+		c.core.router.doAdmin(c.session.close)
-		c.session = nil
 	}
 	// This can't fail yet - TODO?
 	c.closed = true

src/yggdrasil/core.go

@@ -88,14 +88,14 @@ func (c *Core) addPeerLoop() {
 
 		// Add peers from the Peers section
 		for _, peer := range current.Peers {
-			c.AddPeer(peer, "")
+			go c.AddPeer(peer, "")
 			time.Sleep(time.Second)
 		}
 
 		// Add peers from the InterfacePeers section
 		for intf, intfpeers := range current.InterfacePeers {
 			for _, peer := range intfpeers {
-				c.AddPeer(peer, intf)
+				go c.AddPeer(peer, intf)
 				time.Sleep(time.Second)
 			}
 		}

src/yggdrasil/dht.go

@@ -70,7 +70,7 @@ type dht struct {
 	nodeID      crypto.NodeID
 	peers       chan *dhtInfo                    // other goroutines put incoming dht updates here
 	reqs        map[dhtReqKey]time.Time          // Keeps track of recent outstanding requests
-	callbacks   map[dhtReqKey]dht_callbackInfo // Search and admin lookup callbacks
+	callbacks   map[dhtReqKey][]dht_callbackInfo // Search and admin lookup callbacks
 	// These next two could be replaced by a single linked list or similar...
 	table map[crypto.NodeID]*dhtInfo
 	imp   []*dhtInfo
@@ -88,7 +88,7 @@ func (t *dht) init(c *Core) {
 	}()
 	t.nodeID = *t.core.NodeID()
 	t.peers = make(chan *dhtInfo, 1024)
-	t.callbacks = make(map[dhtReqKey]dht_callbackInfo)
+	t.callbacks = make(map[dhtReqKey][]dht_callbackInfo)
 	t.reset()
 }
 
@@ -244,15 +244,17 @@ type dht_callbackInfo struct {
 // Adds a callback and removes it after some timeout.
 func (t *dht) addCallback(rq *dhtReqKey, callback func(*dhtRes)) {
 	info := dht_callbackInfo{callback, time.Now().Add(6 * time.Second)}
-	t.callbacks[*rq] = info
+	t.callbacks[*rq] = append(t.callbacks[*rq], info)
 }
 
 // Reads a lookup response, checks that we had sent a matching request, and processes the response info.
 // This mainly consists of updating the node we asked in our DHT (they responded, so we know they're still alive), and deciding if we want to do anything with their responses
 func (t *dht) handleRes(res *dhtRes) {
 	rq := dhtReqKey{res.Key, res.Dest}
-	if callback, isIn := t.callbacks[rq]; isIn {
+	if callbacks, isIn := t.callbacks[rq]; isIn {
+		for _, callback := range callbacks {
 			callback.f(res)
+		}
 		delete(t.callbacks, rq)
 	}
 	_, isIn := t.reqs[rq]
@@ -326,10 +328,15 @@ func (t *dht) doMaintenance() {
 		}
 	}
 	t.reqs = newReqs
-	newCallbacks := make(map[dhtReqKey]dht_callbackInfo, len(t.callbacks))
+	newCallbacks := make(map[dhtReqKey][]dht_callbackInfo, len(t.callbacks))
-	for key, callback := range t.callbacks {
+	for key, cs := range t.callbacks {
-		if now.Before(callback.time) {
+		for _, c := range cs {
-			newCallbacks[key] = callback
+			if now.Before(c.time) {
+				newCallbacks[key] = append(newCallbacks[key], c)
+			} else {
+				// Signal failure
+				c.f(nil)
+			}
 		}
 	}
 	t.callbacks = newCallbacks

src/yggdrasil/dialer.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"strconv"
 	"strings"
+	"time"
 
 	"github.com/yggdrasil-network/yggdrasil-go/src/crypto"
 )
@@ -14,6 +15,8 @@ type Dialer struct {
 	core *Core
 }
 
+// TODO DialContext that allows timeouts/cancellation, Dial should just call this with no timeout set in the context
+
 // Dial opens a session to the given node. The first paramter should be "nodeid"
 // and the second parameter should contain a hexadecimal representation of the
 // target node ID.
@@ -58,5 +61,17 @@ func (d *Dialer) Dial(network, address string) (*Conn, error) {
 // NodeID parameters.
 func (d *Dialer) DialByNodeIDandMask(nodeID, nodeMask *crypto.NodeID) (*Conn, error) {
 	conn := newConn(d.core, nodeID, nodeMask, nil)
-	return conn, nil
+	if err := conn.search(); err != nil {
+		conn.Close()
+		return nil, err
+	}
+	t := time.NewTimer(6 * time.Second) // TODO use a context instead
+	defer t.Stop()
+	select {
+	case <-conn.session.init:
+		return conn, nil
+	case <-t.C:
+		conn.Close()
+		return nil, errors.New("session handshake timeout")
+	}
 }

src/yggdrasil/router.go

@@ -119,7 +119,7 @@ func (r *router) mainLoop() {
 		case info := <-r.core.dht.peers:
 			r.core.dht.insertPeer(info)
 		case <-r.reset:
-			r.core.sessions.resetInits()
+			r.core.sessions.reset()
 			r.core.dht.reset()
 		case <-ticker.C:
 			{

src/yggdrasil/search.go

@@ -33,13 +33,14 @@ const search_RETRY_TIME = time.Second
 // Information about an ongoing search.
 // Includes the target NodeID, the bitmask to match it to an IP, and the list of nodes to visit / already visited.
 type searchInfo struct {
+	core     *Core
 	dest     crypto.NodeID
 	mask     crypto.NodeID
 	time     time.Time
-	packet   []byte
 	toVisit  []*dhtInfo
 	visited  map[crypto.NodeID]bool
 	callback func(*sessionInfo, error)
+	// TODO context.Context for timeout and cancellation
 }
 
 // This stores a map of active searches.
@@ -49,7 +50,7 @@ type searches struct {
 	searches    map[crypto.NodeID]*searchInfo
 }
 
-// Intializes the searches struct.
+// Initializes the searches struct.
 func (s *searches) init(core *Core) {
 	s.core = core
 	s.reconfigure = make(chan chan error, 1)
@@ -65,12 +66,13 @@ func (s *searches) init(core *Core) {
 // Creates a new search info, adds it to the searches struct, and returns a pointer to the info.
 func (s *searches) createSearch(dest *crypto.NodeID, mask *crypto.NodeID, callback func(*sessionInfo, error)) *searchInfo {
 	now := time.Now()
-	for dest, sinfo := range s.searches {
+	//for dest, sinfo := range s.searches {
-		if now.Sub(sinfo.time) > time.Minute {
+	//	if now.Sub(sinfo.time) > time.Minute {
-			delete(s.searches, dest)
+	//		delete(s.searches, dest)
-		}
+	//	}
-	}
+	//}
 	info := searchInfo{
+		core:     s.core,
 		dest:     *dest,
 		mask:     *mask,
 		time:     now.Add(-time.Second),
@@ -82,30 +84,29 @@ func (s *searches) createSearch(dest *crypto.NodeID, mask *crypto.NodeID, callba
 
 ////////////////////////////////////////////////////////////////////////////////
 
-// Checks if there's an ongoing search relaed to a dhtRes.
+// Checks if there's an ongoing search related to a dhtRes.
 // If there is, it adds the response info to the search and triggers a new search step.
 // If there's no ongoing search, or we if the dhtRes finished the search (it was from the target node), then don't do anything more.
-func (s *searches) handleDHTRes(res *dhtRes) {
+func (sinfo *searchInfo) handleDHTRes(res *dhtRes) {
-	sinfo, isIn := s.searches[res.Dest]
+	if res == nil || sinfo.checkDHTRes(res) {
-	if !isIn || s.checkDHTRes(sinfo, res) {
 		// Either we don't recognize this search, or we just finished it
 		return
 	}
 	// Add to the search and continue
-	s.addToSearch(sinfo, res)
+	sinfo.addToSearch(res)
-	s.doSearchStep(sinfo)
+	sinfo.doSearchStep()
 }
 
 // Adds the information from a dhtRes to an ongoing search.
 // Info about a node that has already been visited is not re-added to the search.
 // Duplicate information about nodes toVisit is deduplicated (the newest information is kept).
 // The toVisit list is sorted in ascending order of keyspace distance from the destination.
-func (s *searches) addToSearch(sinfo *searchInfo, res *dhtRes) {
+func (sinfo *searchInfo) addToSearch(res *dhtRes) {
 	// Add responses to toVisit if closer to dest than the res node
 	from := dhtInfo{key: res.Key, coords: res.Coords}
 	sinfo.visited[*from.getNodeID()] = true
 	for _, info := range res.Infos {
-		if *info.getNodeID() == s.core.dht.nodeID || sinfo.visited[*info.getNodeID()] {
+		if *info.getNodeID() == sinfo.core.dht.nodeID || sinfo.visited[*info.getNodeID()] {
 			continue
 		}
 		if dht_ordered(&sinfo.dest, info.getNodeID(), from.getNodeID()) {
@@ -135,42 +136,43 @@ func (s *searches) addToSearch(sinfo *searchInfo, res *dhtRes) {
 
 // If there are no nodes left toVisit, then this cleans up the search.
 // Otherwise, it pops the closest node to the destination (in keyspace) off of the toVisit list and sends a dht ping.
-func (s *searches) doSearchStep(sinfo *searchInfo) {
+func (sinfo *searchInfo) doSearchStep() {
 	if len(sinfo.toVisit) == 0 {
 		// Dead end, do cleanup
-		delete(s.searches, sinfo.dest)
+		delete(sinfo.core.searches.searches, sinfo.dest)
-		go sinfo.callback(nil, errors.New("search reached dead end"))
+		sinfo.callback(nil, errors.New("search reached dead end"))
 		return
 	}
 	// Send to the next search target
 	var next *dhtInfo
 	next, sinfo.toVisit = sinfo.toVisit[0], sinfo.toVisit[1:]
 	rq := dhtReqKey{next.key, sinfo.dest}
-	s.core.dht.addCallback(&rq, s.handleDHTRes)
+	sinfo.core.dht.addCallback(&rq, sinfo.handleDHTRes)
-	s.core.dht.ping(next, &sinfo.dest)
+	sinfo.core.dht.ping(next, &sinfo.dest)
 }
 
 // If we've recenty sent a ping for this search, do nothing.
 // Otherwise, doSearchStep and schedule another continueSearch to happen after search_RETRY_TIME.
-func (s *searches) continueSearch(sinfo *searchInfo) {
+func (sinfo *searchInfo) continueSearch() {
 	if time.Since(sinfo.time) < search_RETRY_TIME {
 		return
 	}
 	sinfo.time = time.Now()
-	s.doSearchStep(sinfo)
+	sinfo.doSearchStep()
 	// In case the search dies, try to spawn another thread later
 	// Note that this will spawn multiple parallel searches as time passes
 	// Any that die aren't restarted, but a new one will start later
 	retryLater := func() {
-		newSearchInfo := s.searches[sinfo.dest]
+		// FIXME this keeps the search alive forever if not for the searches map, fix that
+		newSearchInfo := sinfo.core.searches.searches[sinfo.dest]
 		if newSearchInfo != sinfo {
 			return
 		}
-		s.continueSearch(sinfo)
+		sinfo.continueSearch()
 	}
 	go func() {
 		time.Sleep(search_RETRY_TIME)
-		s.core.router.admin <- retryLater
+		sinfo.core.router.admin <- retryLater
 	}()
 }
 
@@ -185,37 +187,36 @@ func (s *searches) newIterSearch(dest *crypto.NodeID, mask *crypto.NodeID, callb
 // Checks if a dhtRes is good (called by handleDHTRes).
 // If the response is from the target, get/create a session, trigger a session ping, and return true.
 // Otherwise return false.
-func (s *searches) checkDHTRes(info *searchInfo, res *dhtRes) bool {
+func (sinfo *searchInfo) checkDHTRes(res *dhtRes) bool {
 	them := crypto.GetNodeID(&res.Key)
 	var destMasked crypto.NodeID
 	var themMasked crypto.NodeID
 	for idx := 0; idx < crypto.NodeIDLen; idx++ {
-		destMasked[idx] = info.dest[idx] & info.mask[idx]
+		destMasked[idx] = sinfo.dest[idx] & sinfo.mask[idx]
-		themMasked[idx] = them[idx] & info.mask[idx]
+		themMasked[idx] = them[idx] & sinfo.mask[idx]
 	}
 	if themMasked != destMasked {
 		return false
 	}
 	// They match, so create a session and send a sessionRequest
-	sinfo, isIn := s.core.sessions.getByTheirPerm(&res.Key)
+	sess, isIn := sinfo.core.sessions.getByTheirPerm(&res.Key)
 	if !isIn {
-		sinfo = s.core.sessions.createSession(&res.Key)
+		sess = sinfo.core.sessions.createSession(&res.Key)
-		if sinfo == nil {
+		if sess == nil {
 			// nil if the DHT search finished but the session wasn't allowed
-			go info.callback(nil, errors.New("session not allowed"))
+			sinfo.callback(nil, errors.New("session not allowed"))
 			return true
 		}
-		_, isIn := s.core.sessions.getByTheirPerm(&res.Key)
+		_, isIn := sinfo.core.sessions.getByTheirPerm(&res.Key)
 		if !isIn {
 			panic("This should never happen")
 		}
 	}
 	// FIXME (!) replay attacks could mess with coords? Give it a handle (tstamp)?
-	sinfo.coords = res.Coords
+	sess.coords = res.Coords
-	sinfo.packet = info.packet
+	sinfo.core.sessions.ping(sess)
-	s.core.sessions.ping(sinfo)
+	sinfo.callback(sess, nil)
-	go info.callback(sinfo, nil)
 	// Cleanup
-	delete(s.searches, res.Dest)
+	delete(sinfo.core.searches.searches, res.Dest)
 	return true
 }

src/yggdrasil/session.go

@@ -6,7 +6,6 @@ package yggdrasil
 
 import (
 	"bytes"
-	"encoding/hex"
 	"sync"
 	"time"
 
@@ -40,13 +39,13 @@ type sessionInfo struct {
 	pingTime       time.Time                // time the first ping was sent since the last received packet
 	pingSend       time.Time                // time the last ping was sent
 	coords         []byte                   // coords of destination
-	packet         []byte                   // a buffered packet, sent immediately on ping/pong
+	reset          bool                     // reset if coords change
-	init           bool                     // Reset if coords change
 	tstamp         int64                    // ATOMIC - tstamp from their last session ping, replay attack mitigation
 	bytesSent      uint64                   // Bytes of real traffic sent in this session
 	bytesRecvd     uint64                   // Bytes of real traffic received in this session
 	worker         chan func()              // Channel to send work to the session worker
 	recv           chan *wire_trafficPacket // Received packets go here, picked up by the associated Conn
+	init           chan struct{}            // Closed when the first session pong arrives, used to signal that the session is ready for initial use
 }
 
 func (sinfo *sessionInfo) doWorker(f func()) {
@@ -103,7 +102,14 @@ func (s *sessionInfo) update(p *sessionPing) bool {
 	}
 	s.time = time.Now()
 	s.tstamp = p.Tstamp
-	s.init = true
+	s.reset = false
+	defer func() { recover() }() // Recover if the below panics
+	select {
+	case <-s.init:
+	default:
+		// Unblock anything waiting for the session to initialize
+		close(s.init)
+	}
 	return true
 }
 
@@ -116,13 +122,11 @@ type sessions struct {
 	listenerMutex    sync.Mutex
 	reconfigure      chan chan error
 	lastCleanup      time.Time
+	isAllowedHandler func(pubkey *crypto.BoxPubKey, initiator bool) bool // Returns true or false if session setup is allowed
+	isAllowedMutex   sync.RWMutex                                        // Protects the above
 	permShared       map[crypto.BoxPubKey]*crypto.BoxSharedKey           // Maps known permanent keys to their shared key, used by DHT a lot
-	sinfos        map[crypto.Handle]*sessionInfo            // Maps (secret) handle onto session info
+	sinfos           map[crypto.Handle]*sessionInfo                      // Maps handle onto session info
-	conns         map[crypto.Handle]*Conn                   // Maps (secret) handle onto connections
-	byMySes       map[crypto.BoxPubKey]*crypto.Handle       // Maps mySesPub onto handle
 	byTheirPerm      map[crypto.BoxPubKey]*crypto.Handle                 // Maps theirPermPub onto handle
-	addrToPerm    map[address.Address]*crypto.BoxPubKey
-	subnetToPerm  map[address.Subnet]*crypto.BoxPubKey
 }
 
 // Initializes the session struct.
@@ -148,77 +152,21 @@ func (ss *sessions) init(core *Core) {
 	}()
 	ss.permShared = make(map[crypto.BoxPubKey]*crypto.BoxSharedKey)
 	ss.sinfos = make(map[crypto.Handle]*sessionInfo)
-	ss.byMySes = make(map[crypto.BoxPubKey]*crypto.Handle)
 	ss.byTheirPerm = make(map[crypto.BoxPubKey]*crypto.Handle)
-	ss.addrToPerm = make(map[address.Address]*crypto.BoxPubKey)
-	ss.subnetToPerm = make(map[address.Subnet]*crypto.BoxPubKey)
 	ss.lastCleanup = time.Now()
 }
 
-// Determines whether the session firewall is enabled.
-func (ss *sessions) isSessionFirewallEnabled() bool {
-	ss.core.config.Mutex.RLock()
-	defer ss.core.config.Mutex.RUnlock()
-
-	return ss.core.config.Current.SessionFirewall.Enable
-}
-
 // Determines whether the session with a given publickey is allowed based on
 // session firewall rules.
 func (ss *sessions) isSessionAllowed(pubkey *crypto.BoxPubKey, initiator bool) bool {
-	ss.core.config.Mutex.RLock()
+	ss.isAllowedMutex.RLock()
-	defer ss.core.config.Mutex.RUnlock()
+	defer ss.isAllowedMutex.RUnlock()
 
-	// Allow by default if the session firewall is disabled
+	if ss.isAllowedHandler == nil {
-	if !ss.isSessionFirewallEnabled() {
 		return true
 	}
-	// Prepare for checking whitelist/blacklist
+
-	var box crypto.BoxPubKey
+	return ss.isAllowedHandler(pubkey, initiator)
-	// Reject blacklisted nodes
-	for _, b := range ss.core.config.Current.SessionFirewall.BlacklistEncryptionPublicKeys {
-		key, err := hex.DecodeString(b)
-		if err == nil {
-			copy(box[:crypto.BoxPubKeyLen], key)
-			if box == *pubkey {
-				return false
-			}
-		}
-	}
-	// Allow whitelisted nodes
-	for _, b := range ss.core.config.Current.SessionFirewall.WhitelistEncryptionPublicKeys {
-		key, err := hex.DecodeString(b)
-		if err == nil {
-			copy(box[:crypto.BoxPubKeyLen], key)
-			if box == *pubkey {
-				return true
-			}
-		}
-	}
-	// Allow outbound sessions if appropriate
-	if ss.core.config.Current.SessionFirewall.AlwaysAllowOutbound {
-		if initiator {
-			return true
-		}
-	}
-	// Look and see if the pubkey is that of a direct peer
-	var isDirectPeer bool
-	for _, peer := range ss.core.peers.ports.Load().(map[switchPort]*peer) {
-		if peer.box == *pubkey {
-			isDirectPeer = true
-			break
-		}
-	}
-	// Allow direct peers if appropriate
-	if ss.core.config.Current.SessionFirewall.AllowFromDirect && isDirectPeer {
-		return true
-	}
-	// Allow remote nodes if appropriate
-	if ss.core.config.Current.SessionFirewall.AllowFromRemote && !isDirectPeer {
-		return true
-	}
-	// Finally, default-deny if not matching any of the above rules
-	return false
 }
 
 // Gets the session corresponding to a given handle.
@@ -227,16 +175,6 @@ func (ss *sessions) getSessionForHandle(handle *crypto.Handle) (*sessionInfo, bo
 	return sinfo, isIn
 }
 
-// Gets a session corresponding to an ephemeral session key used by this node.
-func (ss *sessions) getByMySes(key *crypto.BoxPubKey) (*sessionInfo, bool) {
-	h, isIn := ss.byMySes[*key]
-	if !isIn {
-		return nil, false
-	}
-	sinfo, isIn := ss.getSessionForHandle(h)
-	return sinfo, isIn
-}
-
 // Gets a session corresponding to a permanent key used by the remote node.
 func (ss *sessions) getByTheirPerm(key *crypto.BoxPubKey) (*sessionInfo, bool) {
 	h, isIn := ss.byTheirPerm[*key]
@@ -247,30 +185,11 @@ func (ss *sessions) getByTheirPerm(key *crypto.BoxPubKey) (*sessionInfo, bool) {
 	return sinfo, isIn
 }
 
-// Gets a session corresponding to an IPv6 address used by the remote node.
-func (ss *sessions) getByTheirAddr(addr *address.Address) (*sessionInfo, bool) {
-	p, isIn := ss.addrToPerm[*addr]
-	if !isIn {
-		return nil, false
-	}
-	sinfo, isIn := ss.getByTheirPerm(p)
-	return sinfo, isIn
-}
-
-// Gets a session corresponding to an IPv6 /64 subnet used by the remote node/network.
-func (ss *sessions) getByTheirSubnet(snet *address.Subnet) (*sessionInfo, bool) {
-	p, isIn := ss.subnetToPerm[*snet]
-	if !isIn {
-		return nil, false
-	}
-	sinfo, isIn := ss.getByTheirPerm(p)
-	return sinfo, isIn
-}
-
 // Creates a new session and lazily cleans up old existing sessions. This
 // includse initializing session info to sane defaults (e.g. lowest supported
 // MTU).
 func (ss *sessions) createSession(theirPermKey *crypto.BoxPubKey) *sessionInfo {
+	// TODO: this check definitely needs to be moved
 	if !ss.isSessionAllowed(theirPermKey, true) {
 		return nil
 	}
@@ -292,6 +211,7 @@ func (ss *sessions) createSession(theirPermKey *crypto.BoxPubKey) *sessionInfo {
 	sinfo.mtuTime = now
 	sinfo.pingTime = now
 	sinfo.pingSend = now
+	sinfo.init = make(chan struct{})
 	higher := false
 	for idx := range ss.core.boxPub {
 		if ss.core.boxPub[idx] > sinfo.theirPermPub[idx] {
@@ -314,10 +234,7 @@ func (ss *sessions) createSession(theirPermKey *crypto.BoxPubKey) *sessionInfo {
 	sinfo.worker = make(chan func(), 1)
 	sinfo.recv = make(chan *wire_trafficPacket, 32)
 	ss.sinfos[sinfo.myHandle] = &sinfo
-	ss.byMySes[sinfo.mySesPub] = &sinfo.myHandle
 	ss.byTheirPerm[sinfo.theirPermPub] = &sinfo.myHandle
-	ss.addrToPerm[sinfo.theirAddr] = &sinfo.theirPermPub
-	ss.subnetToPerm[sinfo.theirSubnet] = &sinfo.theirPermPub
 	go sinfo.workerMain()
 	return &sinfo
 }
@@ -342,36 +259,21 @@ func (ss *sessions) cleanup() {
 		sinfos[k] = v
 	}
 	ss.sinfos = sinfos
-	byMySes := make(map[crypto.BoxPubKey]*crypto.Handle, len(ss.byMySes))
-	for k, v := range ss.byMySes {
-		byMySes[k] = v
-	}
-	ss.byMySes = byMySes
 	byTheirPerm := make(map[crypto.BoxPubKey]*crypto.Handle, len(ss.byTheirPerm))
 	for k, v := range ss.byTheirPerm {
 		byTheirPerm[k] = v
 	}
 	ss.byTheirPerm = byTheirPerm
-	addrToPerm := make(map[address.Address]*crypto.BoxPubKey, len(ss.addrToPerm))
-	for k, v := range ss.addrToPerm {
-		addrToPerm[k] = v
-	}
-	ss.addrToPerm = addrToPerm
-	subnetToPerm := make(map[address.Subnet]*crypto.BoxPubKey, len(ss.subnetToPerm))
-	for k, v := range ss.subnetToPerm {
-		subnetToPerm[k] = v
-	}
-	ss.subnetToPerm = subnetToPerm
 	ss.lastCleanup = time.Now()
 }
 
 // Closes a session, removing it from sessions maps and killing the worker goroutine.
 func (sinfo *sessionInfo) close() {
+	if s := sinfo.core.sessions.sinfos[sinfo.myHandle]; s == sinfo {
 		delete(sinfo.core.sessions.sinfos, sinfo.myHandle)
-	delete(sinfo.core.sessions.byMySes, sinfo.mySesPub)
 		delete(sinfo.core.sessions.byTheirPerm, sinfo.theirPermPub)
-	delete(sinfo.core.sessions.addrToPerm, sinfo.theirAddr)
+	}
-	delete(sinfo.core.sessions.subnetToPerm, sinfo.theirSubnet)
+	defer func() { recover() }()
 	close(sinfo.worker)
 }
 
@@ -396,6 +298,8 @@ func (ss *sessions) getPing(sinfo *sessionInfo) sessionPing {
 // This comes up with dht req/res and session ping/pong traffic.
 func (ss *sessions) getSharedKey(myPriv *crypto.BoxPrivKey,
 	theirPub *crypto.BoxPubKey) *crypto.BoxSharedKey {
+	return crypto.GetSharedKey(myPriv, theirPub)
+	// FIXME concurrency issues with the below, so for now we just burn the CPU every time
 	if skey, isIn := ss.permShared[*theirPub]; isIn {
 		return skey
 	}
@@ -434,8 +338,8 @@ func (ss *sessions) sendPingPong(sinfo *sessionInfo, isPong bool) {
 	}
 	packet := p.encode()
 	ss.core.router.out(packet)
-	if !isPong {
+	if sinfo.pingTime.Before(sinfo.time) {
-		sinfo.pingSend = time.Now()
+		sinfo.pingTime = time.Now()
 	}
 }
 
@@ -444,12 +348,12 @@ func (ss *sessions) sendPingPong(sinfo *sessionInfo, isPong bool) {
 func (ss *sessions) handlePing(ping *sessionPing) {
 	// Get the corresponding session (or create a new session)
 	sinfo, isIn := ss.getByTheirPerm(&ping.SendPermPub)
-	// Check the session firewall
+	// Check if the session is allowed
-	if !isIn && ss.isSessionFirewallEnabled() {
+	// TODO: this check may need to be moved
-		if !ss.isSessionAllowed(&ping.SendPermPub, false) {
+	if !isIn && !ss.isSessionAllowed(&ping.SendPermPub, false) {
 		return
 	}
-	}
+	// Create the session if it doesn't already exist
 	if !isIn {
 		ss.createSession(&ping.SendPermPub)
 		sinfo, isIn = ss.getByTheirPerm(&ping.SendPermPub)
@@ -476,15 +380,6 @@ func (ss *sessions) handlePing(ping *sessionPing) {
 		if !ping.IsPong {
 			ss.sendPingPong(sinfo, true)
 		}
-		if sinfo.packet != nil {
-			/* FIXME this needs to live in the net.Conn or something, needs work in Write
-			// send
-			var bs []byte
-			bs, sinfo.packet = sinfo.packet, nil
-			ss.core.router.sendPacket(bs) // FIXME this needs to live in the net.Conn or something, needs work in Write
-			*/
-			sinfo.packet = nil
-		}
 	})
 }
 
@@ -529,10 +424,10 @@ func (sinfo *sessionInfo) updateNonce(theirNonce *crypto.BoxNonce) {
 
 // Resets all sessions to an uninitialized state.
 // Called after coord changes, so attemtps to use a session will trigger a new ping and notify the remote end of the coord change.
-func (ss *sessions) resetInits() {
+func (ss *sessions) reset() {
 	for _, sinfo := range ss.sinfos {
 		sinfo.doWorker(func() {
-			sinfo.init = false
+			sinfo.reset = true
 		})
 	}
 }