From 79bcfbf17578c594f53d1ba260cc4c29d5abca9b Mon Sep 17 00:00:00 2001 From: Neil Alexander Date: Sun, 21 Apr 2019 11:50:41 +0100 Subject: [PATCH] Change some mutexes to atomics, change conns map to pointers, sort of works but seems to deadlock very easily --- src/tuntap/tun.go | 12 ++--- src/yggdrasil/conn.go | 41 ++++++--------- src/yggdrasil/dialer.go | 12 ++--- src/yggdrasil/router.go | 22 ++++---- src/yggdrasil/search.go | 2 - src/yggdrasil/session.go | 107 ++++++++++++++++----------------------- 6 files changed, 81 insertions(+), 115 deletions(-) diff --git a/src/tuntap/tun.go b/src/tuntap/tun.go index 4b72b465..9e46124a 100644 --- a/src/tuntap/tun.go +++ b/src/tuntap/tun.go @@ -30,8 +30,7 @@ type TunAdapter struct { config *config.NodeState log *log.Logger reconfigure chan chan error - conns map[crypto.NodeID]yggdrasil.Conn - connsMutex sync.RWMutex + conns map[crypto.NodeID]*yggdrasil.Conn listener *yggdrasil.Listener dialer *yggdrasil.Dialer addr address.Address @@ -102,7 +101,7 @@ func (tun *TunAdapter) Init(config *config.NodeState, log *log.Logger, listener tun.log = log tun.listener = listener tun.dialer = dialer - tun.conns = make(map[crypto.NodeID]yggdrasil.Conn) + tun.conns = make(map[crypto.NodeID]*yggdrasil.Conn) } // Start the setup process for the TUN/TAP adapter. If successful, starts the @@ -180,6 +179,7 @@ func (tun *TunAdapter) handler() error { } func (tun *TunAdapter) connReader(conn *yggdrasil.Conn) error { + tun.conns[conn.RemoteAddr()] = conn b := make([]byte, 65535) for { n, err := conn.Read(b) @@ -203,7 +203,6 @@ func (tun *TunAdapter) connReader(conn *yggdrasil.Conn) error { } func (tun *TunAdapter) ifaceReader() error { - tun.log.Println("Start TUN reader") bs := make([]byte, 65535) for { n, err := tun.iface.Read(bs) @@ -244,6 +243,7 @@ func (tun *TunAdapter) ifaceReader() error { dstNodeID, dstNodeIDMask = dstAddr.GetNodeIDandMask() // Do we have an active connection for this node ID? if conn, isIn := tun.conns[*dstNodeID]; isIn { + tun.log.Println("Got", &conn) w, err := conn.Write(bs) if err != nil { tun.log.Println("Unable to write to remote:", err) @@ -254,14 +254,12 @@ func (tun *TunAdapter) ifaceReader() error { } } else { tun.log.Println("Opening connection for", *dstNodeID) - tun.connsMutex.Lock() if conn, err := tun.dialer.DialByNodeIDandMask(dstNodeID, dstNodeIDMask); err == nil { - tun.conns[*dstNodeID] = conn + tun.conns[*dstNodeID] = &conn go tun.connReader(&conn) } else { tun.log.Println("Error dialing:", err) } - tun.connsMutex.Unlock() } /*if !r.cryptokey.isValidSource(srcAddr, addrlen) { diff --git a/src/yggdrasil/conn.go b/src/yggdrasil/conn.go index 9566cae3..daba2987 100644 --- a/src/yggdrasil/conn.go +++ b/src/yggdrasil/conn.go @@ -1,8 +1,8 @@ package yggdrasil import ( - "encoding/hex" "errors" + "fmt" "sync" "sync/atomic" "time" @@ -42,27 +42,27 @@ func (c *Conn) startSearch() { doSearch := func() { sinfo, isIn := c.core.searches.searches[*c.nodeID] if !isIn { - c.core.log.Debugln("Starting search for", hex.EncodeToString(c.nodeID[:])) sinfo = c.core.searches.newIterSearch(c.nodeID, c.nodeMask, searchCompleted) } c.core.searches.continueSearch(sinfo) } - var sinfo *sessionInfo - var isIn bool switch { - case !isIn || !sinfo.init: + case c.session == nil || !c.session.init.Load().(bool): doSearch() - case time.Since(sinfo.time) > 6*time.Second: - if sinfo.time.Before(sinfo.pingTime) && time.Since(sinfo.pingTime) > 6*time.Second { + case time.Since(c.session.time.Load().(time.Time)) > 6*time.Second: + sTime := c.session.time.Load().(time.Time) + pingTime := c.session.pingTime.Load().(time.Time) + if sTime.Before(pingTime) && time.Since(pingTime) > 6*time.Second { doSearch() } else { + pingSend := c.session.pingSend.Load().(time.Time) now := time.Now() - if !sinfo.time.Before(sinfo.pingTime) { - sinfo.pingTime = now + if !sTime.Before(pingTime) { + c.session.pingTime.Store(now) } - if time.Since(sinfo.pingSend) > time.Second { - sinfo.pingSend = now - c.core.sessions.sendPingPong(sinfo, false) + if time.Since(pingSend) > time.Second { + c.session.pingSend.Store(now) + c.core.sessions.sendPingPong(c.session, false) } } } @@ -77,12 +77,9 @@ func (c *Conn) Read(b []byte) (int, error) { if c.session == nil { return 0, errors.New("searching for remote side") } - c.session.initMutex.RLock() - if !c.session.init { - c.session.initMutex.RUnlock() + if !c.session.init.Load().(bool) { return 0, errors.New("waiting for remote side to accept") } - c.session.initMutex.RUnlock() select { case p, ok := <-c.session.recv: if !ok { @@ -106,9 +103,7 @@ func (c *Conn) Read(b []byte) (int, error) { b = b[:len(bs)] } c.session.updateNonce(&p.Nonce) - c.session.timeMutex.Lock() - c.session.time = time.Now() - c.session.timeMutex.Unlock() + c.session.time.Store(time.Now()) return nil }() if err != nil { @@ -129,22 +124,18 @@ func (c *Conn) Write(b []byte) (bytesWritten int, err error) { return 0, errors.New("session is closed") } if c.session == nil { + fmt.Println("No session found, starting search for", &c) c.core.router.doAdmin(func() { c.startSearch() }) return 0, errors.New("searching for remote side") } defer util.PutBytes(b) - c.session.initMutex.RLock() - if !c.session.init { - c.session.initMutex.RUnlock() + if !c.session.init.Load().(bool) { return 0, errors.New("waiting for remote side to accept") } - c.session.initMutex.RUnlock() // code isn't multithreaded so appending to this is safe - c.session.coordsMutex.RLock() coords := c.session.coords - c.session.coordsMutex.RUnlock() // Prepare the payload c.session.myNonceMutex.Lock() payload, nonce := crypto.BoxSeal(&c.session.sharedSesKey, b, &c.session.myNonce) diff --git a/src/yggdrasil/dialer.go b/src/yggdrasil/dialer.go index fee355e4..4a3d8167 100644 --- a/src/yggdrasil/dialer.go +++ b/src/yggdrasil/dialer.go @@ -35,7 +35,7 @@ func (d *Dialer) Dial(network, address string) (Conn, error) { return Conn{}, err } copy(nodeID[:], dest) - for idx := 0; idx < len; idx++ { + for idx := 0; idx <= len; idx++ { nodeMask[idx/8] |= 0x80 >> byte(idx%8) } } else { @@ -59,15 +59,13 @@ func (d *Dialer) Dial(network, address string) (Conn, error) { // NodeID parameters. func (d *Dialer) DialByNodeIDandMask(nodeID, nodeMask *crypto.NodeID) (Conn, error) { conn := Conn{ - mutex: &sync.RWMutex{}, + core: d.core, + mutex: &sync.RWMutex{}, + nodeID: nodeID, + nodeMask: nodeMask, } - conn.core = d.core - conn.nodeID = nodeID - conn.nodeMask = nodeMask conn.core.router.doAdmin(func() { conn.startSearch() }) - conn.mutex.Lock() - defer conn.mutex.Unlock() return conn, nil } diff --git a/src/yggdrasil/router.go b/src/yggdrasil/router.go index 19b62f91..7da5162f 100644 --- a/src/yggdrasil/router.go +++ b/src/yggdrasil/router.go @@ -287,16 +287,15 @@ func (r *router) sendPacket(bs []byte) { if destSnet.IsValid() { sinfo, isIn = r.core.sessions.getByTheirSubnet(&destSnet) } - sinfo.timeMutex.Lock() - sinfo.initMutex.RLock() - defer sinfo.timeMutex.Unlock() - defer sinfo.initMutex.RUnlock() + sTime := sinfo.time.Load().(time.Time) + pingTime := sinfo.pingTime.Load().(time.Time) + pingSend := sinfo.pingSend.Load().(time.Time) switch { - case !isIn || !sinfo.init: + case !isIn || !sinfo.init.Load().(bool): // No or unintiialized session, so we need to search first doSearch(bs) - case time.Since(sinfo.time) > 6*time.Second: - if sinfo.time.Before(sinfo.pingTime) && time.Since(sinfo.pingTime) > 6*time.Second { + case time.Since(sTime) > 6*time.Second: + if sTime.Before(pingTime) && time.Since(pingTime) > 6*time.Second { // We haven't heard from the dest in a while // We tried pinging but didn't get a response // They may have changed coords @@ -307,16 +306,15 @@ func (r *router) sendPacket(bs []byte) { // We haven't heard about the dest in a while now := time.Now() - if !sinfo.time.Before(sinfo.pingTime) { + if !sTime.Before(pingTime) { // Update pingTime to start the clock for searches (above) - sinfo.pingTime = now + sinfo.pingTime.Store(now) } - if time.Since(sinfo.pingSend) > time.Second { + if time.Since(pingSend) > time.Second { // Send at most 1 ping per second - sinfo.pingSend = now + sinfo.pingSend.Store(now) r.core.sessions.sendPingPong(sinfo, false) } - sinfo.timeMutex.Unlock() } fallthrough // Also send the packet default: diff --git a/src/yggdrasil/search.go b/src/yggdrasil/search.go index dcf0c81f..e81a9723 100644 --- a/src/yggdrasil/search.go +++ b/src/yggdrasil/search.go @@ -212,9 +212,7 @@ func (s *searches) checkDHTRes(info *searchInfo, res *dhtRes) bool { } } // FIXME (!) replay attacks could mess with coords? Give it a handle (tstamp)? - sinfo.coordsMutex.Lock() sinfo.coords = res.Coords - sinfo.coordsMutex.Unlock() sinfo.packet = info.packet s.core.sessions.ping(sinfo) info.callback(sinfo, nil) diff --git a/src/yggdrasil/session.go b/src/yggdrasil/session.go index 3bf69a8c..b0bba2d0 100644 --- a/src/yggdrasil/session.go +++ b/src/yggdrasil/session.go @@ -18,41 +18,38 @@ import ( // 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. type sessionInfo struct { - core *Core - reconfigure chan chan error - theirAddr address.Address - theirSubnet address.Subnet - theirPermPub crypto.BoxPubKey - theirSesPub crypto.BoxPubKey - mySesPub crypto.BoxPubKey - mySesPriv crypto.BoxPrivKey - sharedSesKey crypto.BoxSharedKey // derived from session keys - theirHandle crypto.Handle - myHandle crypto.Handle - theirNonce crypto.BoxNonce - theirNonceMask uint64 - theirNonceMutex sync.Mutex // protects the above - myNonce crypto.BoxNonce - myNonceMutex sync.Mutex // protects the above - theirMTU uint16 - myMTU uint16 - wasMTUFixed bool // Was the MTU fixed by a receive error? - time time.Time // Time we last received a packet - mtuTime time.Time // time myMTU was last changed - pingTime time.Time // time the first ping was sent since the last received packet - pingSend time.Time // time the last ping was sent - timeMutex sync.RWMutex // protects all time fields above - coords []byte // coords of destination - coordsMutex sync.RWMutex // protects the above - packet []byte // a buffered packet, sent immediately on ping/pong - init bool // Reset if coords change - initMutex sync.RWMutex - send chan []byte - recv chan *wire_trafficPacket - closed chan interface{} - 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 + core *Core // + reconfigure chan chan error // + theirAddr address.Address // + theirSubnet address.Subnet // + theirPermPub crypto.BoxPubKey // + theirSesPub crypto.BoxPubKey // + mySesPub crypto.BoxPubKey // + mySesPriv crypto.BoxPrivKey // + sharedSesKey crypto.BoxSharedKey // derived from session keys + theirHandle crypto.Handle // + myHandle crypto.Handle // + theirNonce crypto.BoxNonce // + theirNonceMask uint64 // + theirNonceMutex sync.Mutex // protects the above + myNonce crypto.BoxNonce // + myNonceMutex sync.Mutex // protects the above + theirMTU uint16 // + myMTU uint16 // + wasMTUFixed bool // Was the MTU fixed by a receive error? + time atomic.Value // time.Time // Time we last received a packet + mtuTime atomic.Value // time.Time // time myMTU was last changed + pingTime atomic.Value // time.Time // time the first ping was sent since the last received packet + pingSend atomic.Value // time.Time // time the last ping was sent + coords []byte // coords of destination + packet []byte // a buffered packet, sent immediately on ping/pong + init atomic.Value // bool // Reset if coords change + send chan []byte // + recv chan *wire_trafficPacket // + closed chan interface{} // + 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 } // Represents a session ping/pong packet, andincludes information like public keys, a session handle, coords, a timestamp to prevent replays, and the tun/tap MTU. @@ -60,10 +57,10 @@ type sessionPing struct { SendPermPub crypto.BoxPubKey // Sender's permanent key Handle crypto.Handle // Random number to ID session SendSesPub crypto.BoxPubKey // Session key to use - Coords []byte - Tstamp int64 // unix time, but the only real requirement is that it increases - IsPong bool - MTU uint16 + Coords []byte // + Tstamp int64 // unix time, but the only real requirement is that it increases + IsPong bool // + MTU uint16 // } // Updates session info in response to a ping, after checking that the ping is OK. @@ -93,21 +90,15 @@ func (s *sessionInfo) update(p *sessionPing) bool { s.coords = append(make([]byte, 0, len(p.Coords)+11), p.Coords...) } now := time.Now() - s.timeMutex.Lock() - s.time = now - s.timeMutex.Unlock() + s.time.Store(now) atomic.StoreInt64(&s.tstamp, p.Tstamp) - s.initMutex.Lock() - s.init = true - s.initMutex.Unlock() + s.init.Store(true) return true } // Returns true if the session has been idle for longer than the allowed timeout. func (s *sessionInfo) timedout() bool { - s.timeMutex.RLock() - defer s.timeMutex.RUnlock() - return time.Since(s.time) > time.Minute + return time.Since(s.time.Load().(time.Time)) > time.Minute } // Struct of all active sessions. @@ -291,12 +282,10 @@ func (ss *sessions) createSession(theirPermKey *crypto.BoxPubKey) *sessionInfo { sinfo.theirMTU = 1280 sinfo.myMTU = 1280 now := time.Now() - sinfo.timeMutex.Lock() - sinfo.time = now - sinfo.mtuTime = now - sinfo.pingTime = now - sinfo.pingSend = now - sinfo.timeMutex.Unlock() + sinfo.time.Store(now) + sinfo.mtuTime.Store(now) + sinfo.pingTime.Store(now) + sinfo.pingSend.Store(now) higher := false for idx := range ss.core.boxPub { if ss.core.boxPub[idx] > sinfo.theirPermPub[idx] { @@ -437,7 +426,6 @@ func (ss *sessions) sendPingPong(sinfo *sessionInfo, isPong bool) { bs := ping.encode() shared := ss.getSharedKey(&ss.core.boxPriv, &sinfo.theirPermPub) payload, nonce := crypto.BoxSeal(shared, bs, nil) - sinfo.coordsMutex.RLock() p := wire_protoTrafficPacket{ Coords: sinfo.coords, ToKey: sinfo.theirPermPub, @@ -445,13 +433,10 @@ func (ss *sessions) sendPingPong(sinfo *sessionInfo, isPong bool) { Nonce: *nonce, Payload: payload, } - sinfo.coordsMutex.RUnlock() packet := p.encode() ss.core.router.out(packet) if !isPong { - sinfo.timeMutex.Lock() - sinfo.pingSend = time.Now() - sinfo.timeMutex.Unlock() + sinfo.pingSend.Store(time.Now()) } } @@ -551,8 +536,6 @@ func (sinfo *sessionInfo) updateNonce(theirNonce *crypto.BoxNonce) { // 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() { for _, sinfo := range ss.sinfos { - sinfo.initMutex.Lock() - sinfo.init = false - sinfo.initMutex.Unlock() + sinfo.init.Store(false) } }