From 9778f5d2b86336189d259b9bb272dcb9e470f0e4 Mon Sep 17 00:00:00 2001 From: Neil Alexander Date: Mon, 22 Apr 2019 15:00:19 +0100 Subject: [PATCH] Fix search behaviour on closed Conns, various other fixes --- src/tuntap/tun.go | 66 +++++++++++++++++-------------- src/yggdrasil/conn.go | 84 +++++++++++++++++++++++++++++++--------- src/yggdrasil/dialer.go | 3 -- src/yggdrasil/search.go | 7 ++-- src/yggdrasil/session.go | 1 + 5 files changed, 107 insertions(+), 54 deletions(-) diff --git a/src/tuntap/tun.go b/src/tuntap/tun.go index 3bdbde2..07264da 100644 --- a/src/tuntap/tun.go +++ b/src/tuntap/tun.go @@ -193,12 +193,11 @@ func (tun *TunAdapter) connReader(conn *yggdrasil.Conn) error { delete(tun.conns, remoteNodeID) tun.mutex.Unlock() }() - tun.log.Debugln("Start connection reader for", conn.String()) b := make([]byte, 65535) for { n, err := conn.Read(b) if err != nil { - tun.log.Errorln("TUN/TAP conn read error:", err) + tun.log.Errorln(conn.String(), "TUN/TAP conn read error:", err) continue } if n == 0 { @@ -206,11 +205,11 @@ func (tun *TunAdapter) connReader(conn *yggdrasil.Conn) error { } w, err := tun.iface.Write(b[:n]) if err != nil { - tun.log.Errorln("TUN/TAP iface write error:", err) + tun.log.Errorln(conn.String(), "TUN/TAP iface write error:", err) continue } if w != n { - tun.log.Errorln("TUN/TAP iface write mismatch:", w, "bytes written vs", n, "bytes given") + tun.log.Errorln(conn.String(), "TUN/TAP iface write mismatch:", w, "bytes written vs", n, "bytes given") continue } } @@ -219,20 +218,24 @@ func (tun *TunAdapter) connReader(conn *yggdrasil.Conn) error { func (tun *TunAdapter) ifaceReader() error { bs := make([]byte, 65535) for { + // Wait for a packet to be delivered to us through the TUN/TAP adapter n, err := tun.iface.Read(bs) if err != nil { continue } - // Look up if the dstination address is somewhere we already have an - // open connection to + // 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 + // the packet var srcAddr address.Address var dstAddr address.Address var dstNodeID *crypto.NodeID var dstNodeIDMask *crypto.NodeID var dstSnet address.Subnet var addrlen int + // Check the IP protocol - if it doesn't match then we drop the packet and + // do nothing with it if bs[0]&0xf0 == 0x60 { - // Check if we have a fully-sized header + // Check if we have a fully-sized IPv6 header if len(bs) < 40 { continue } @@ -242,7 +245,7 @@ func (tun *TunAdapter) ifaceReader() error { copy(dstAddr[:addrlen], bs[24:]) copy(dstSnet[:addrlen/2], bs[24:]) } else if bs[0]&0xf0 == 0x40 { - // Check if we have a fully-sized header + // Check if we have a fully-sized IPv4 header if len(bs) < 20 { continue } @@ -251,7 +254,7 @@ func (tun *TunAdapter) ifaceReader() error { copy(srcAddr[:addrlen], bs[12:]) copy(dstAddr[:addrlen], bs[16:]) } else { - // Unknown address length or protocol + // Unknown address length or protocol, so drop the packet and ignore it continue } if !dstAddr.IsValid() && !dstSnet.IsValid() { @@ -261,32 +264,39 @@ func (tun *TunAdapter) ifaceReader() error { dstNodeID, dstNodeIDMask = dstAddr.GetNodeIDandMask() // Do we have an active connection for this node ID? tun.mutex.RLock() - if conn, isIn := tun.conns[*dstNodeID]; isIn { - tun.mutex.RUnlock() + conn, isIn := tun.conns[*dstNodeID] + tun.mutex.RUnlock() + // If we don't have a connection then we should open one + if !isIn { + // Dial to the remote node + if c, err := tun.dialer.DialByNodeIDandMask(dstNodeID, dstNodeIDMask); err == nil { + // We've been given a connection, so save it in our connections so we + // can refer to it the next time we send a packet to this destination + tun.mutex.Lock() + tun.conns[*dstNodeID] = &c + tun.mutex.Unlock() + // Start the connection reader goroutine + go tun.connReader(&c) + // Then update our reference to the connection + conn, isIn = &c, true + } else { + // We weren't able to dial for some reason so there's no point in + // continuing this iteration - skip to the next one + continue + } + } + // If we have an open connection, either because we already had one or + // because we opened one above, try writing the packet to it + if isIn && conn != nil { w, err := conn.Write(bs[:n]) if err != nil { - tun.log.Errorln("TUN/TAP conn write error:", err) + tun.log.Errorln(conn.String(), "TUN/TAP conn write error:", err) continue } if w != n { - tun.log.Errorln("TUN/TAP conn write mismatch:", w, "bytes written vs", n, "bytes given") + tun.log.Errorln(conn.String(), "TUN/TAP conn write mismatch:", w, "bytes written vs", n, "bytes given") continue } - } else { - tun.mutex.RUnlock() - func() { - tun.mutex.Lock() - defer tun.mutex.Unlock() - if conn, err := tun.dialer.DialByNodeIDandMask(dstNodeID, dstNodeIDMask); err == nil { - tun.log.Debugln("Opening new session connection") - tun.log.Debugln("Node:", dstNodeID) - tun.log.Debugln("Mask:", dstNodeIDMask) - tun.conns[*dstNodeID] = &conn - go tun.connReader(&conn) - } else { - tun.log.Errorln("TUN/TAP dial error:", err) - } - }() } /*if !r.cryptokey.isValidSource(srcAddr, addrlen) { diff --git a/src/yggdrasil/conn.go b/src/yggdrasil/conn.go index 9308d34..0accf16 100644 --- a/src/yggdrasil/conn.go +++ b/src/yggdrasil/conn.go @@ -21,41 +21,50 @@ type Conn struct { readDeadline atomic.Value // time.Time // TODO timer writeDeadline atomic.Value // time.Time // TODO timer expired atomic.Value // bool + searching atomic.Value // bool } func (c *Conn) String() string { - return fmt.Sprintf("c=%p", c) + return fmt.Sprintf("conn=%p", c) } // This method should only be called from the router goroutine func (c *Conn) startSearch() { searchCompleted := func(sinfo *sessionInfo, err error) { + c.searching.Store(false) + c.mutex.Lock() + defer c.mutex.Unlock() if err != nil { - c.core.log.Debugln("DHT search failed:", err) - c.mutex.Lock() + c.core.log.Debugln(c.String(), "DHT search failed:", err) c.expired.Store(true) return } if sinfo != nil { - c.mutex.Lock() + c.core.log.Debugln(c.String(), "DHT search completed") c.session = sinfo c.nodeID, c.nodeMask = sinfo.theirAddr.GetNodeIDandMask() - c.mutex.Unlock() + c.expired.Store(false) + } else { + c.core.log.Debugln(c.String(), "DHT search failed: no session returned") + c.expired.Store(true) + return } } doSearch := func() { + c.searching.Store(true) sinfo, isIn := c.core.searches.searches[*c.nodeID] if !isIn { sinfo = c.core.searches.newIterSearch(c.nodeID, c.nodeMask, searchCompleted) + c.core.log.Debugf("%s DHT search started: %p", c.String(), sinfo) } c.core.searches.continueSearch(sinfo) } c.mutex.RLock() - defer c.mutex.RUnlock() + sinfo := c.session + c.mutex.RUnlock() if c.session == nil { doSearch() } else { - sinfo := c.session // In case c.session is somehow changed meanwhile sinfo.worker <- func() { switch { case !sinfo.init: @@ -74,43 +83,66 @@ func (c *Conn) startSearch() { } func (c *Conn) Read(b []byte) (int, error) { + // If the session is marked as expired then do nothing at this point if e, ok := c.expired.Load().(bool); ok && e { return 0, errors.New("session is closed") } + // Take a copy of the session object c.mutex.RLock() sinfo := c.session c.mutex.RUnlock() + // If the session is not initialised, do nothing. Currently in this instance + // in a write, we would trigger a new session, but it doesn't make sense for + // us to block forever here if the session will not reopen. + // TODO: should this return an error or just a zero-length buffer? + if !sinfo.init { + return 0, errors.New("session is closed") + } + // Wait for some traffic to come through from the session select { // TODO... case p, ok := <-c.recv: + // If the channel was closed then mark the connection as expired, this will + // mean that the next write will start a new search and reopen the session if !ok { c.expired.Store(true) return 0, errors.New("session is closed") } defer util.PutBytes(p.Payload) var err error + // Hand over to the session worker sinfo.doWorker(func() { + // If the nonce is bad then drop the packet and return an error if !sinfo.nonceIsOK(&p.Nonce) { err = errors.New("packet dropped due to invalid nonce") return } + // Decrypt the packet bs, isOK := crypto.BoxOpen(&sinfo.sharedSesKey, p.Payload, &p.Nonce) + // Check if we were unable to decrypt the packet for some reason and + // return an error if we couldn't if !isOK { util.PutBytes(bs) err = errors.New("packet dropped due to decryption failure") return } + // Return the newly decrypted buffer back to the slice we were given copy(b, bs) + // Trim the slice down to size based on the data we received if len(bs) < len(b) { b = b[:len(bs)] } + // Update the session sinfo.updateNonce(&p.Nonce) sinfo.time = time.Now() sinfo.bytesRecvd += uint64(len(b)) }) + // Something went wrong in the session worker so abort if err != nil { return 0, err } + // If we've reached this point then everything went to plan, return the + // number of bytes we populated back into the given slice return len(b), nil //case <-c.recvTimeout: //case <-c.session.closed: @@ -120,27 +152,35 @@ func (c *Conn) Read(b []byte) (int, error) { } func (c *Conn) Write(b []byte) (bytesWritten int, err error) { - if e, ok := c.expired.Load().(bool); ok && e { - return 0, errors.New("session is closed") - } c.mutex.RLock() sinfo := c.session c.mutex.RUnlock() - if sinfo == nil { - c.core.router.doAdmin(func() { - c.startSearch() - }) - return 0, errors.New("searching for remote side") + // Check whether the connection is expired, if it is we can start a new + // search to revive it + expired, eok := c.expired.Load().(bool) + // If the session doesn't exist, or isn't initialised (which probably means + // that the session was never set up or it closed by timeout), or the conn + // is marked as expired, then see if we can start a new search + if sinfo == nil || !sinfo.init || (eok && expired) { + // 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(func() { + c.startSearch() + }) + return 0, errors.New("starting search") + } + // A search is already taking place so wait for it to finish + return 0, errors.New("waiting for search to complete") } //defer util.PutBytes(b) var packet []byte + // Hand over to the session worker sinfo.doWorker(func() { - if !sinfo.init { - err = errors.New("waiting for remote side to accept " + c.String()) - return - } + // Encrypt the packet payload, nonce := crypto.BoxSeal(&sinfo.sharedSesKey, b, &sinfo.myNonce) defer util.PutBytes(payload) + // Construct the wire packet to send to the router p := wire_trafficPacket{ Coords: sinfo.coords, Handle: sinfo.theirHandle, @@ -150,13 +190,19 @@ func (c *Conn) Write(b []byte) (bytesWritten int, err error) { packet = p.encode() sinfo.bytesSent += uint64(len(b)) }) + // Give the packet to the router sinfo.core.router.out(packet) + // Finally return the number of bytes we wrote return len(b), nil } func (c *Conn) Close() error { + // Mark the connection as expired, so that a future read attempt will fail + // and a future write attempt will start a new search c.expired.Store(true) + // Close the session, if it hasn't been closed already c.session.close() + // This can't fail yet - TODO? return nil } diff --git a/src/yggdrasil/dialer.go b/src/yggdrasil/dialer.go index bd4b87e..8901610 100644 --- a/src/yggdrasil/dialer.go +++ b/src/yggdrasil/dialer.go @@ -65,8 +65,5 @@ func (d *Dialer) DialByNodeIDandMask(nodeID, nodeMask *crypto.NodeID) (Conn, err nodeMask: nodeMask, recv: make(chan *wire_trafficPacket, 32), } - conn.core.router.admin <- func() { - conn.startSearch() - } return conn, nil } diff --git a/src/yggdrasil/search.go b/src/yggdrasil/search.go index 3460fd4..0a64336 100644 --- a/src/yggdrasil/search.go +++ b/src/yggdrasil/search.go @@ -90,11 +90,10 @@ func (s *searches) handleDHTRes(res *dhtRes) { if !isIn || s.checkDHTRes(sinfo, res) { // Either we don't recognize this search, or we just finished it return - } else { - // Add to the search and continue - s.addToSearch(sinfo, res) - s.doSearchStep(sinfo) } + // Add to the search and continue + s.addToSearch(sinfo, res) + s.doSearchStep(sinfo) } // Adds the information from a dhtRes to an ongoing search. diff --git a/src/yggdrasil/session.go b/src/yggdrasil/session.go index 55d4a0c..e356f61 100644 --- a/src/yggdrasil/session.go +++ b/src/yggdrasil/session.go @@ -387,6 +387,7 @@ func (sinfo *sessionInfo) close() { delete(sinfo.core.sessions.addrToPerm, sinfo.theirAddr) delete(sinfo.core.sessions.subnetToPerm, sinfo.theirSubnet) close(sinfo.worker) + sinfo.init = false } // Returns a session ping appropriate for the given session info.