Merge pull request #175 from yggdrasil-network/develop

Version 0.2.6

.circleci/config.yml

@@ -31,7 +31,8 @@ jobs:
               PKGARCH=i386 sh contrib/deb/generate.sh && mv yggdrasil /tmp/upload/$CINAME-$CIVERSION-linux-i386 && mv yggdrasilctl /tmp/upload/$CINAME-$CIVERSION-yggdrasilctl-linux-i386;
               PKGARCH=mipsel sh contrib/deb/generate.sh && mv yggdrasil /tmp/upload/$CINAME-$CIVERSION-linux-mipsel && mv yggdrasilctl /tmp/upload/$CINAME-$CIVERSION-yggdrasilctl-linux-mipsel;
               PKGARCH=mips sh contrib/deb/generate.sh && mv yggdrasil /tmp/upload/$CINAME-$CIVERSION-linux-mips && mv yggdrasilctl /tmp/upload/$CINAME-$CIVERSION-yggdrasilctl-linux-mips;
-              PKGARCH=armhf sh contrib/deb/generate.sh && mv yggdrasil /tmp/upload/$CINAME-$CIVERSION-linux-armh && mv yggdrasilctl /tmp/upload/$CINAME-$CIVERSION-yggdrasilctl-linux-armhf;
+              PKGARCH=armhf sh contrib/deb/generate.sh && mv yggdrasil /tmp/upload/$CINAME-$CIVERSION-linux-armhf && mv yggdrasilctl /tmp/upload/$CINAME-$CIVERSION-yggdrasilctl-linux-armhf;
+              PKGARCH=arm64 sh contrib/deb/generate.sh && mv yggdrasil /tmp/upload/$CINAME-$CIVERSION-linux-arm64 && mv yggdrasilctl /tmp/upload/$CINAME-$CIVERSION-yggdrasilctl-linux-arm64;
               sudo alien --to-rpm yggdrasil*.deb --scripts --keep-version && mv *.rpm /tmp/upload/;
               mv *.deb /tmp/upload/
 

CHANGELOG.md

@@ -25,6 +25,15 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
 - in case of vulnerabilities.
 -->
 
+## [0.2.6] - 2018-07-31
+### Added
+- Configurable TCP timeouts to assist in peering over Tor/I2P
+- Prefer IPv6 flow label when extending coordinates to sort backpressure queues
+- `arm64` builds through CircleCI
+
+### Changed
+- Sort dot graph links by integer value
+
 ## [0.2.5] - 2018-07-19
 ### Changed
 - Make `yggdrasilctl` less case sensitive

contrib/deb/generate.sh

@@ -26,8 +26,9 @@ elif [ $PKGARCH = "i386" ]; then GOARCH=386 GOOS=linux ./build
 elif [ $PKGARCH = "mipsel" ]; then GOARCH=mipsle GOOS=linux ./build
 elif [ $PKGARCH = "mips" ]; then GOARCH=mips64 GOOS=linux ./build
 elif [ $PKGARCH = "armhf" ]; then GOARCH=arm GOOS=linux GOARM=7 ./build
+elif [ $PKGARCH = "arm64" ]; then GOARCH=arm64 GOOS=linux ./build
 else
-  echo "Specify PKGARCH=amd64,i386,mips,mipsel,armhf"
+  echo "Specify PKGARCH=amd64,i386,mips,mipsel,armhf,arm64"
   exit -1
 fi
 

src/yggdrasil/admin.go

@@ -605,9 +605,16 @@ func (a *admin) getResponse_dot() []byte {
 		name    string
 		key     string
 		parent  string
+		port    switchPort
 		options string
 	}
 	infos := make(map[string]nodeInfo)
+	// Get coords as a slice of strings, FIXME? this looks very fragile
+	coordSlice := func(coords string) []string {
+		tmp := strings.Replace(coords, "[", "", -1)
+		tmp = strings.Replace(tmp, "]", "", -1)
+		return strings.Split(tmp, " ")
+	}
 	// First fill the tree with all known nodes, no parents
 	addInfo := func(nodes []admin_nodeInfo, options string, tag string) {
 		for _, node := range nodes {
@@ -621,6 +628,14 @@ func (a *admin) getResponse_dot() []byte {
 			} else {
 				info.name = n["ip"].(string)
 			}
+			coordsSplit := coordSlice(info.key)
+			if len(coordsSplit) != 0 {
+				portStr := coordsSplit[len(coordsSplit)-1]
+				portUint, err := strconv.ParseUint(portStr, 10, 64)
+				if err == nil {
+					info.port = switchPort(portUint)
+				}
+			}
 			infos[info.key] = info
 		}
 	}
@@ -628,12 +643,6 @@ func (a *admin) getResponse_dot() []byte {
 	addInfo(sessions, "fillcolor=\"#acf3fd\" style=filled fontname=\"sans serif\"", "Open session")                          // blue
 	addInfo(peers, "fillcolor=\"#ffffb5\" style=filled fontname=\"sans serif\"", "Connected peer")                           // yellow
 	addInfo(append([]admin_nodeInfo(nil), *self), "fillcolor=\"#a5ff8a\" style=filled fontname=\"sans serif\"", "This node") // green
-	// Get coords as a slice of strings, FIXME? this looks very fragile
-	coordSlice := func(coords string) []string {
-		tmp := strings.Replace(coords, "[", "", -1)
-		tmp = strings.Replace(tmp, "]", "", -1)
-		return strings.Split(tmp, " ")
-	}
 	// Now go through and create placeholders for any missing nodes
 	for _, info := range infos {
 		// This is ugly string manipulation
@@ -665,10 +674,12 @@ func (a *admin) getResponse_dot() []byte {
 		keys = append(keys, info.key)
 	}
 	// sort
-	less := func(i, j int) bool {
+	sort.SliceStable(keys, func(i, j int) bool {
 		return keys[i] < keys[j]
-	}
+	})
-	sort.Slice(keys, less)
+	sort.SliceStable(keys, func(i, j int) bool {
+		return infos[keys[i]].port < infos[keys[j]].port
+	})
 	// Now print it all out
 	var out []byte
 	put := func(s string) {
@@ -686,11 +697,7 @@ func (a *admin) getResponse_dot() []byte {
 		if info.key == info.parent {
 			continue
 		} // happens for the root, skip it
-		coordsSplit := coordSlice(key)
+		port := fmt.Sprint(info.port)
-		if len(coordsSplit) == 0 {
-			continue
-		}
-		port := coordsSplit[len(coordsSplit)-1]
 		style := "fontname=\"sans serif\""
 		if infos[info.parent].name == "?" || infos[info.key].name == "?" {
 			style = "fontname=\"sans serif\" style=dashed color=\"#999999\" fontcolor=\"#999999\""

src/yggdrasil/config/config.go

@@ -5,6 +5,7 @@ type NodeConfig struct {
 	Listen                      string   `comment:"Listen address for peer connections. Default is to listen for all\nTCP connections over IPv4 and IPv6 with a random port."`
 	AdminListen                 string   `comment:"Listen address for admin connections Default is to listen for local\nconnections either on TCP/9001 or a UNIX socket depending on your\nplatform. Use this value for yggdrasilctl -endpoint=X."`
 	Peers                       []string `comment:"List of connection strings for static peers in URI format, i.e.\ntcp://a.b.c.d:e or socks://a.b.c.d:e/f.g.h.i:j"`
+	ReadTimeout                 int32    `comment:"Read timeout for connections, specified in milliseconds. If less than 6000 and not negative, 6000 (the default) is used. If negative, reads won't time out."`
 	AllowedEncryptionPublicKeys []string `comment:"List of peer encryption public keys to allow or incoming TCP\nconnections from. If left empty/undefined then all connections\nwill be allowed by default."`
 	EncryptionPublicKey         string   `comment:"Your public encryption key. Your peers may ask you for this to put\ninto their AllowedEncryptionPublicKeys configuration."`
 	EncryptionPrivateKey        string   `comment:"Your private encryption key. DO NOT share this with anyone!"`

src/yggdrasil/core.go

@@ -97,7 +97,7 @@ func (c *Core) Start(nc *config.NodeConfig, log *log.Logger) error {
 	c.init(&boxPub, &boxPriv, &sigPub, &sigPriv)
 	c.admin.init(c, nc.AdminListen)
 
-	if err := c.tcp.init(c, nc.Listen); err != nil {
+	if err := c.tcp.init(c, nc.Listen, nc.ReadTimeout); err != nil {
 		c.log.Println("Failed to start TCP interface")
 		return err
 	}

src/yggdrasil/session.go

@@ -4,7 +4,10 @@ package yggdrasil
 // It's responsible for keeping track of open sessions to other nodes
 // The session information consists of crypto keys and coords
 
-import "time"
+import (
+	"bytes"
+	"time"
+)
 
 // All the information we know about an active session.
 // This includes coords, permanent and ephemeral keys, handles and nonces, various sorts of timing information for timeout and maintenance, and some metadata for the admin API.
@@ -72,7 +75,10 @@ func (s *sessionInfo) update(p *sessionPing) bool {
 	if p.MTU >= 1280 || p.MTU == 0 {
 		s.theirMTU = p.MTU
 	}
-	s.coords = append([]byte{}, p.Coords...)
+	if !bytes.Equal(s.coords, p.Coords) {
+		// allocate enough space for additional coords
+		s.coords = append(make([]byte, 0, len(p.Coords)+11), p.Coords...)
+	}
 	now := time.Now()
 	s.time = now
 	s.tstamp = p.Tstamp
@@ -423,12 +429,42 @@ func (sinfo *sessionInfo) doWorker() {
 func (sinfo *sessionInfo) doSend(bs []byte) {
 	defer util_putBytes(bs)
 	if !sinfo.init {
+		// To prevent using empty session keys
 		return
-	} // To prevent using empty session keys
+	}
+	// code isn't multithreaded so appending to this is safe
+	coords := sinfo.coords
+	// Read IPv6 flowlabel field (20 bits).
+	// Assumes packet at least contains IPv6 header.
+	flowkey := uint64(bs[1]&0x0f)<<16 | uint64(bs[2])<<8 | uint64(bs[3])
+	// Check if the flowlabel was specified
+	if flowkey == 0 {
+		// Does the packet meet the minimum UDP packet size? (others are bigger)
+		if len(bs) >= 48 {
+			// Is the protocol TCP, UDP, SCTP?
+			if bs[6] == 0x06 || bs[6] == 0x11 || bs[6] == 0x84 {
+				// if flowlabel was unspecified (0), try to use known protocols' ports
+				// protokey: proto | sport | dport
+				flowkey = uint64(bs[6])<<32 /* proto */ |
+					uint64(bs[40])<<24 | uint64(bs[41])<<16 /* sport */ |
+					uint64(bs[42])<<8 | uint64(bs[43]) /* dport */
+			}
+		}
+	}
+	// If we have a flowkey, either through the IPv6 flowlabel field or through
+	// known TCP/UDP/SCTP proto-sport-dport triplet, then append it to the coords.
+	// Appending extra coords after a 0 ensures that we still target the local router
+	// but lets us send extra data (which is otherwise ignored) to help separate
+	// traffic streams into independent queues
+	if flowkey != 0 {
+		coords = append(coords, 0)                // First target the local switchport
+		coords = wire_put_uint64(flowkey, coords) // Then variable-length encoded flowkey
+	}
+	// Prepare the payload
 	payload, nonce := boxSeal(&sinfo.sharedSesKey, bs, &sinfo.myNonce)
 	defer util_putBytes(payload)
 	p := wire_trafficPacket{
-		Coords:  sinfo.coords,
+		Coords:  coords,
 		Handle:  sinfo.theirHandle,
 		Nonce:   *nonce,
 		Payload: payload,

src/yggdrasil/switch.go

@@ -529,24 +529,13 @@ func switch_getPacketCoords(packet []byte) []byte {
 }
 
 // Returns a unique string for each stream of traffic
-// Equal to type+coords+handle for traffic packets
+// Equal to coords
-// Equal to type+coords+toKey+fromKey for protocol traffic packets
+// The sender may append arbitrary info to the end of coords (as long as it's begins with a 0x00) to designate separate traffic streams
+// Currently, it's the IPv6 next header type and the first 2 uint16 of the next header
+// This is equivalent to the TCP/UDP protocol numbers and the source / dest ports
+// TODO figure out if something else would make more sense (other transport protocols?)
 func switch_getPacketStreamID(packet []byte) string {
-	pType, pTypeLen := wire_decode_uint64(packet)
+	return string(switch_getPacketCoords(packet))
-	_, coordLen := wire_decode_coords(packet[pTypeLen:])
-	end := pTypeLen + coordLen
-	switch {
-	case pType == wire_Traffic:
-		end += handleLen // handle
-	case pType == wire_ProtocolTraffic:
-		end += 2 * boxPubKeyLen
-	default:
-		end = 0
-	}
-	if end > len(packet) {
-		end = len(packet)
-	}
-	return string(packet[:end])
 }
 
 // Handle an incoming packet

src/yggdrasil/tcp.go

@@ -28,7 +28,8 @@ import (
 )
 
 const tcp_msgSize = 2048 + 65535 // TODO figure out what makes sense
-const tcp_timeout = 6 * time.Second
+const default_tcp_timeout = 6 * time.Second
+const tcp_ping_interval = (default_tcp_timeout * 2 / 3)
 
 // Wrapper function for non tcp/ip connections.
 func setNoDelay(c net.Conn, delay bool) {
@@ -42,6 +43,7 @@ func setNoDelay(c net.Conn, delay bool) {
 type tcpInterface struct {
 	core        *Core
 	serv        net.Listener
+	tcp_timeout time.Duration
 	mutex       sync.Mutex // Protecting the below
 	calls       map[string]struct{}
 	conns       map[tcpInfo](chan struct{})
@@ -72,9 +74,14 @@ func (iface *tcpInterface) connectSOCKS(socksaddr, peeraddr string) {
 }
 
 // Initializes the struct.
-func (iface *tcpInterface) init(core *Core, addr string) (err error) {
+func (iface *tcpInterface) init(core *Core, addr string, readTimeout int32) (err error) {
 	iface.core = core
 
+	iface.tcp_timeout = time.Duration(readTimeout) * time.Millisecond
+	if iface.tcp_timeout >= 0 && iface.tcp_timeout < default_tcp_timeout {
+		iface.tcp_timeout = default_tcp_timeout
+	}
+
 	iface.serv, err = net.Listen("tcp", addr)
 	if err == nil {
 		iface.calls = make(map[string]struct{})
@@ -113,7 +120,7 @@ func (iface *tcpInterface) call(saddr string, socksaddr *string) {
 			iface.calls[saddr] = struct{}{}
 			defer func() {
 				// Block new calls for a little while, to mitigate livelock scenarios
-				time.Sleep(tcp_timeout)
+				time.Sleep(default_tcp_timeout)
 				time.Sleep(time.Duration(rand.Intn(1000)) * time.Millisecond)
 				iface.mutex.Lock()
 				delete(iface.calls, saddr)
@@ -168,8 +175,9 @@ func (iface *tcpInterface) handler(sock net.Conn, incoming bool) {
 	if err != nil {
 		return
 	}
-	timeout := time.Now().Add(tcp_timeout)
+	if iface.tcp_timeout > 0 {
-	sock.SetReadDeadline(timeout)
+		sock.SetReadDeadline(time.Now().Add(iface.tcp_timeout))
+	}
 	_, err = sock.Read(metaBytes)
 	if err != nil {
 		return
@@ -254,7 +262,7 @@ func (iface *tcpInterface) handler(sock net.Conn, incoming bool) {
 			atomic.AddUint64(&p.bytesSent, uint64(len(tcp_msg)+len(msgLen)+len(msg)))
 			util_putBytes(msg)
 		}
-		timerInterval := tcp_timeout * 2 / 3
+		timerInterval := tcp_ping_interval
 		timer := time.NewTimer(timerInterval)
 		defer timer.Stop()
 		for {
@@ -321,8 +329,9 @@ func (iface *tcpInterface) reader(sock net.Conn, in func([]byte)) error {
 	bs := make([]byte, 2*tcp_msgSize)
 	frag := bs[:0]
 	for {
-		timeout := time.Now().Add(tcp_timeout)
+		if iface.tcp_timeout > 0 {
-		sock.SetReadDeadline(timeout)
+			sock.SetReadDeadline(time.Now().Add(iface.tcp_timeout))
+		}
 		n, err := sock.Read(bs[len(frag):])
 		if n > 0 {
 			frag = bs[:len(frag)+n]

src/yggdrasil/wire.go

@@ -25,19 +25,15 @@ func wire_encode_uint64(elem uint64) []byte {
 
 // Encode uint64 using a variable length scheme.
 // Similar to binary.Uvarint, but big-endian.
-func wire_put_uint64(elem uint64, out []byte) []byte {
+func wire_put_uint64(e uint64, out []byte) []byte {
-	bs := make([]byte, 0, 10)
+	var b [10]byte
-	bs = append(bs, byte(elem&0x7f))
+	i := len(b) - 1
-	for e := elem >> 7; e > 0; e >>= 7 {
+	b[i] = byte(e & 0x7f)
-		bs = append(bs, byte(e|0x80))
+	for e >>= 7; e != 0; e >>= 7 {
+		i--
+		b[i] = byte(e | 0x80)
 	}
-	// Now reverse bytes, because we set them in the wrong order
+	return append(out, b[i:]...)
-	// TODO just put them in the right place the first time...
-	last := len(bs) - 1
-	for idx := 0; idx < len(bs)/2; idx++ {
-		bs[idx], bs[last-idx] = bs[last-idx], bs[idx]
-	}
-	return append(out, bs...)
 }
 
 // Returns the length of a wire encoded uint64 of this value.