wireguard-go/device/peer.go

272 lines
6.5 KiB
Go
Raw Normal View History

2019-01-02 00:55:51 +00:00
/* SPDX-License-Identifier: MIT
*
* Copyright (C) 2017-2021 WireGuard LLC. All Rights Reserved.
*/
2019-03-03 03:04:41 +00:00
package device
import (
"encoding/base64"
2017-06-26 11:14:02 +00:00
"errors"
"fmt"
"sync"
"sync/atomic"
"time"
"golang.zx2c4.com/wireguard/conn"
)
type Peer struct {
isRunning AtomicBool
sync.RWMutex // Mostly protects endpoint, but is generally taken whenever we modify peer
2018-05-13 16:23:40 +00:00
keypairs Keypairs
2017-06-24 13:34:17 +00:00
handshake Handshake
device *Device
endpoint conn.Endpoint
persistentKeepaliveInterval uint32 // accessed atomically
firstTrieEntry *trieEntry
stopping sync.WaitGroup // routines pending stop
// These fields are accessed with atomic operations, which must be
// 64-bit aligned even on 32-bit platforms. Go guarantees that an
// allocated struct will be 64-bit aligned. So we place
// atomically-accessed fields up front, so that they can share in
// this alignment before smaller fields throw it off.
stats struct {
txBytes uint64 // bytes send to peer (endpoint)
rxBytes uint64 // bytes received from peer
lastHandshakeNano int64 // nano seconds since epoch
}
disableRoaming bool
timers struct {
retransmitHandshake *Timer
sendKeepalive *Timer
newHandshake *Timer
zeroKeyMaterial *Timer
persistentKeepalive *Timer
2018-05-20 04:50:07 +00:00
handshakeAttempts uint32
needAnotherKeepalive AtomicBool
sentLastMinuteHandshake AtomicBool
}
device: remove mutex from Peer send/receive The immediate motivation for this change is an observed deadlock. 1. A goroutine calls peer.Stop. That calls peer.queue.Lock(). 2. Another goroutine is in RoutineSequentialReceiver. It receives an elem from peer.queue.inbound. 3. The peer.Stop goroutine calls close(peer.queue.inbound), close(peer.queue.outbound), and peer.stopping.Wait(). It blocks waiting for RoutineSequentialReceiver and RoutineSequentialSender to exit. 4. The RoutineSequentialReceiver goroutine calls peer.SendStagedPackets(). SendStagedPackets attempts peer.queue.RLock(). That blocks forever because the peer.Stop goroutine holds a write lock on that mutex. A background motivation for this change is that it can be expensive to have a mutex in the hot code path of RoutineSequential*. The mutex was necessary to avoid attempting to send elems on a closed channel. This commit removes that danger by never closing the channel. Instead, we send a sentinel nil value on the channel to indicate to the receiver that it should exit. The only problem with this is that if the receiver exits, we could write an elem into the channel which would never get received. If it never gets received, it cannot get returned to the device pools. To work around this, we use a finalizer. When the channel can be GC'd, the finalizer drains any remaining elements from the channel and restores them to the device pool. After that change, peer.queue.RWMutex no longer makes sense where it is. It is only used to prevent concurrent calls to Start and Stop. Move it to a more sensible location and make it a plain sync.Mutex. Signed-off-by: Josh Bleecher Snyder <josh@tailscale.com>
2021-02-08 21:02:52 +00:00
state struct {
sync.Mutex // protects against concurrent Start/Stop
device: remove mutex from Peer send/receive The immediate motivation for this change is an observed deadlock. 1. A goroutine calls peer.Stop. That calls peer.queue.Lock(). 2. Another goroutine is in RoutineSequentialReceiver. It receives an elem from peer.queue.inbound. 3. The peer.Stop goroutine calls close(peer.queue.inbound), close(peer.queue.outbound), and peer.stopping.Wait(). It blocks waiting for RoutineSequentialReceiver and RoutineSequentialSender to exit. 4. The RoutineSequentialReceiver goroutine calls peer.SendStagedPackets(). SendStagedPackets attempts peer.queue.RLock(). That blocks forever because the peer.Stop goroutine holds a write lock on that mutex. A background motivation for this change is that it can be expensive to have a mutex in the hot code path of RoutineSequential*. The mutex was necessary to avoid attempting to send elems on a closed channel. This commit removes that danger by never closing the channel. Instead, we send a sentinel nil value on the channel to indicate to the receiver that it should exit. The only problem with this is that if the receiver exits, we could write an elem into the channel which would never get received. If it never gets received, it cannot get returned to the device pools. To work around this, we use a finalizer. When the channel can be GC'd, the finalizer drains any remaining elements from the channel and restores them to the device pool. After that change, peer.queue.RWMutex no longer makes sense where it is. It is only used to prevent concurrent calls to Start and Stop. Move it to a more sensible location and make it a plain sync.Mutex. Signed-off-by: Josh Bleecher Snyder <josh@tailscale.com>
2021-02-08 21:02:52 +00:00
}
queue struct {
staged chan *QueueOutboundElement // staged packets before a handshake is available
outbound *autodrainingOutboundQueue // sequential ordering of udp transmission
inbound *autodrainingInboundQueue // sequential ordering of tun writing
}
2018-05-13 21:14:43 +00:00
cookieGenerator CookieGenerator
2017-06-24 13:34:17 +00:00
}
func (device *Device) NewPeer(pk NoisePublicKey) (*Peer, error) {
if device.isClosed() {
2018-05-13 17:33:41 +00:00
return nil, errors.New("device closed")
}
// lock resources
device.staticIdentity.RLock()
defer device.staticIdentity.RUnlock()
device.peers.Lock()
defer device.peers.Unlock()
// check if over limit
if len(device.peers.keyMap) >= MaxPeers {
2018-05-13 17:33:41 +00:00
return nil, errors.New("too many peers")
}
2017-06-26 11:14:02 +00:00
// create peer
peer := new(Peer)
peer.Lock()
defer peer.Unlock()
2017-07-01 21:29:22 +00:00
2018-05-13 21:14:43 +00:00
peer.cookieGenerator.Init(pk)
2017-07-01 21:29:22 +00:00
peer.device = device
2017-07-08 21:51:26 +00:00
// map public key
_, ok := device.peers.keyMap[pk]
2017-06-26 11:14:02 +00:00
if ok {
2018-05-13 17:33:41 +00:00
return nil, errors.New("adding existing peer")
2017-06-26 11:14:02 +00:00
}
2017-06-24 13:34:17 +00:00
// pre-compute DH
2017-06-26 11:14:02 +00:00
handshake := &peer.handshake
handshake.mutex.Lock()
2018-05-13 21:14:43 +00:00
handshake.precomputedStaticStatic = device.staticIdentity.privateKey.sharedSecret(pk)
2019-08-05 14:57:41 +00:00
handshake.remoteStatic = pk
2017-06-26 11:14:02 +00:00
handshake.mutex.Unlock()
2017-06-24 13:34:17 +00:00
// reset endpoint
peer.endpoint = nil
2020-03-18 05:06:56 +00:00
// add
device.peers.keyMap[pk] = peer
device.peers.empty.Set(false)
2019-08-05 14:57:41 +00:00
// start peer
peer.timersInit()
if peer.device.isUp() {
peer.Start()
}
return peer, nil
}
func (peer *Peer) SendBuffer(buffer []byte) error {
peer.device.net.RLock()
defer peer.device.net.RUnlock()
2017-12-29 16:42:09 +00:00
2018-02-02 19:45:25 +00:00
if peer.device.net.bind == nil {
// Packets can leak through to SendBuffer while the device is closing.
// When that happens, drop them silently to avoid spurious errors.
if peer.device.isClosed() {
return nil
}
2018-05-13 17:33:41 +00:00
return errors.New("no bind")
2018-02-02 19:45:25 +00:00
}
peer.RLock()
defer peer.RUnlock()
2017-12-29 16:42:09 +00:00
if peer.endpoint == nil {
2018-05-13 17:33:41 +00:00
return errors.New("no known endpoint for peer")
}
2017-12-29 16:42:09 +00:00
err := peer.device.net.bind.Send(buffer, peer.endpoint)
if err == nil {
atomic.AddUint64(&peer.stats.txBytes, uint64(len(buffer)))
}
return err
}
func (peer *Peer) String() string {
2018-05-13 17:33:41 +00:00
base64Key := base64.StdEncoding.EncodeToString(peer.handshake.remoteStatic[:])
abbreviatedKey := "invalid"
if len(base64Key) == 44 {
2018-05-13 21:27:28 +00:00
abbreviatedKey = base64Key[0:4] + "…" + base64Key[39:43]
2018-05-13 17:33:41 +00:00
}
return fmt.Sprintf("peer(%s)", abbreviatedKey)
}
func (peer *Peer) Start() {
// should never start a peer on a closed device
if peer.device.isClosed() {
return
}
// prevent simultaneous start/stop operations
peer.state.Lock()
defer peer.state.Unlock()
2018-02-02 19:45:25 +00:00
if peer.isRunning.Get() {
return
}
2018-02-04 18:18:44 +00:00
device := peer.device
device.log.Verbosef("%v - Starting...", peer)
// reset routine state
peer.stopping.Wait()
peer.stopping.Add(2)
2017-12-29 16:42:09 +00:00
peer.handshake.mutex.Lock()
peer.handshake.lastSentHandshake = time.Now().Add(-(RekeyTimeout + time.Second))
peer.handshake.mutex.Unlock()
// prepare queues (once)
if peer.queue.outbound == nil {
peer.queue.outbound = newAutodrainingOutboundQueue(device)
peer.queue.inbound = newAutodrainingInboundQueue(device)
peer.queue.staged = make(chan *QueueOutboundElement, QueueStagedSize)
}
peer.device.queue.encryption.wg.Add(1) // keep encryption queue open for our writes
peer.timersStart()
2018-02-02 19:45:25 +00:00
2017-12-29 16:42:09 +00:00
go peer.RoutineSequentialSender()
go peer.RoutineSequentialReceiver()
peer.isRunning.Set(true)
2017-12-29 16:42:09 +00:00
}
2018-05-13 21:14:43 +00:00
func (peer *Peer) ZeroAndFlushAll() {
device := peer.device
// clear key pairs
keypairs := &peer.keypairs
keypairs.Lock()
2018-05-13 21:14:43 +00:00
device.DeleteKeypair(keypairs.previous)
device.DeleteKeypair(keypairs.current)
device.DeleteKeypair(keypairs.loadNext())
2018-05-13 21:14:43 +00:00
keypairs.previous = nil
keypairs.current = nil
keypairs.storeNext(nil)
keypairs.Unlock()
2018-05-13 21:14:43 +00:00
// clear handshake state
handshake := &peer.handshake
handshake.mutex.Lock()
device.indexTable.Delete(handshake.localIndex)
handshake.Clear()
handshake.mutex.Unlock()
peer.FlushStagedPackets()
2018-05-13 21:14:43 +00:00
}
func (peer *Peer) ExpireCurrentKeypairs() {
handshake := &peer.handshake
handshake.mutex.Lock()
peer.device.indexTable.Delete(handshake.localIndex)
handshake.Clear()
peer.handshake.lastSentHandshake = time.Now().Add(-(RekeyTimeout + time.Second))
handshake.mutex.Unlock()
keypairs := &peer.keypairs
keypairs.Lock()
if keypairs.current != nil {
atomic.StoreUint64(&keypairs.current.sendNonce, RejectAfterMessages)
}
if keypairs.next != nil {
next := keypairs.loadNext()
atomic.StoreUint64(&next.sendNonce, RejectAfterMessages)
}
keypairs.Unlock()
}
2017-12-29 16:42:09 +00:00
func (peer *Peer) Stop() {
peer.state.Lock()
defer peer.state.Unlock()
2018-02-02 19:45:25 +00:00
if !peer.isRunning.Swap(false) {
return
}
peer.device.log.Verbosef("%v - Stopping...", peer)
peer.timersStop()
device: remove mutex from Peer send/receive The immediate motivation for this change is an observed deadlock. 1. A goroutine calls peer.Stop. That calls peer.queue.Lock(). 2. Another goroutine is in RoutineSequentialReceiver. It receives an elem from peer.queue.inbound. 3. The peer.Stop goroutine calls close(peer.queue.inbound), close(peer.queue.outbound), and peer.stopping.Wait(). It blocks waiting for RoutineSequentialReceiver and RoutineSequentialSender to exit. 4. The RoutineSequentialReceiver goroutine calls peer.SendStagedPackets(). SendStagedPackets attempts peer.queue.RLock(). That blocks forever because the peer.Stop goroutine holds a write lock on that mutex. A background motivation for this change is that it can be expensive to have a mutex in the hot code path of RoutineSequential*. The mutex was necessary to avoid attempting to send elems on a closed channel. This commit removes that danger by never closing the channel. Instead, we send a sentinel nil value on the channel to indicate to the receiver that it should exit. The only problem with this is that if the receiver exits, we could write an elem into the channel which would never get received. If it never gets received, it cannot get returned to the device pools. To work around this, we use a finalizer. When the channel can be GC'd, the finalizer drains any remaining elements from the channel and restores them to the device pool. After that change, peer.queue.RWMutex no longer makes sense where it is. It is only used to prevent concurrent calls to Start and Stop. Move it to a more sensible location and make it a plain sync.Mutex. Signed-off-by: Josh Bleecher Snyder <josh@tailscale.com>
2021-02-08 21:02:52 +00:00
// Signal that RoutineSequentialSender and RoutineSequentialReceiver should exit.
peer.queue.inbound.c <- nil
peer.queue.outbound.c <- nil
peer.stopping.Wait()
peer.device.queue.encryption.wg.Done() // no more writes to encryption queue from us
2018-05-13 21:14:43 +00:00
peer.ZeroAndFlushAll()
}
2018-05-26 00:59:26 +00:00
func (peer *Peer) SetEndpointFromPacket(endpoint conn.Endpoint) {
if peer.disableRoaming {
2018-05-26 00:59:26 +00:00
return
}
peer.Lock()
2018-05-26 00:59:26 +00:00
peer.endpoint = endpoint
peer.Unlock()
2018-05-26 00:59:26 +00:00
}