Hesham/wireguard-go
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

device: add write queue mutex for peer

Browse Source 
fix panic: send on closed channel when remove peer

Signed-off-by: Haichao Liu <liuhaichao@bytedance.com>
Signed-off-by: Jason A. Donenfeld <Jason@zx2c4.com>
This commit is contained in:
Haichao Liu 2020-11-18 20:53:22 +08:00 committed by Jason A. Donenfeld
parent 60b3766b89
commit 913f68ce38
3 changed files with 11 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
device/peer.go
View File

@ -58,6 +58,7 @@ type Peer struct {
} }
queue struct { queue struct {
sync.RWMutex
nonce chan *QueueOutboundElement // nonce / pre-handshake queue nonce chan *QueueOutboundElement // nonce / pre-handshake queue
outbound chan *QueueOutboundElement // sequential ordering of work outbound chan *QueueOutboundElement // sequential ordering of work
inbound chan *QueueInboundElement // sequential ordering of work inbound chan *QueueInboundElement // sequential ordering of work
@ -195,10 +196,11 @@ func (peer *Peer) Start() {
peer.routines.stopping.Add(PeerRoutineNumber) peer.routines.stopping.Add(PeerRoutineNumber)
// prepare queues // prepare queues
peer.queue.Lock()
peer.queue.nonce = make(chan *QueueOutboundElement, QueueOutboundSize) peer.queue.nonce = make(chan *QueueOutboundElement, QueueOutboundSize)
peer.queue.outbound = make(chan *QueueOutboundElement, QueueOutboundSize) peer.queue.outbound = make(chan *QueueOutboundElement, QueueOutboundSize)
peer.queue.inbound = make(chan *QueueInboundElement, QueueInboundSize) peer.queue.inbound = make(chan *QueueInboundElement, QueueInboundSize)
peer.queue.Unlock()
peer.timersInit() peer.timersInit()
peer.handshake.lastSentHandshake = time.Now().Add(-(RekeyTimeout + time.Second)) peer.handshake.lastSentHandshake = time.Now().Add(-(RekeyTimeout + time.Second))
@ -284,9 +286,11 @@ func (peer *Peer) Stop() {
// close queues // close queues
peer.queue.Lock()
close(peer.queue.nonce) close(peer.queue.nonce)
close(peer.queue.outbound) close(peer.queue.outbound)
close(peer.queue.inbound) close(peer.queue.inbound)
peer.queue.Unlock()
peer.ZeroAndFlushAll() peer.ZeroAndFlushAll()
} }

2
device/receive.go
View File

@ -184,11 +184,13 @@ func (device *Device) RoutineReceiveIncoming(IP int, bind conn.Bind) {
// add to decryption queues // add to decryption queues
peer.queue.RLock()
if peer.isRunning.Get() { if peer.isRunning.Get() {
if device.addToInboundAndDecryptionQueues(peer.queue.inbound, device.queue.decryption, elem) { if device.addToInboundAndDecryptionQueues(peer.queue.inbound, device.queue.decryption, elem) {
buffer = device.GetMessageBuffer() buffer = device.GetMessageBuffer()
} }
} }
peer.queue.RUnlock()
continue continue

4
device/send.go
View File

@ -107,6 +107,8 @@ func addToOutboundAndEncryptionQueues(outboundQueue chan *QueueOutboundElement,
/* Queues a keepalive if no packets are queued for peer /* Queues a keepalive if no packets are queued for peer
*/ */
func (peer *Peer) SendKeepalive() bool { func (peer *Peer) SendKeepalive() bool {
peer.queue.RLock()
defer peer.queue.RUnlock()
if len(peer.queue.nonce) != 0 || peer.queue.packetInNonceQueueIsAwaitingKey.Get() || !peer.isRunning.Get() { if len(peer.queue.nonce) != 0 || peer.queue.packetInNonceQueueIsAwaitingKey.Get() || !peer.isRunning.Get() {
return false return false
} }
@ -310,6 +312,7 @@ func (device *Device) RoutineReadFromTUN() {
// insert into nonce/pre-handshake queue // insert into nonce/pre-handshake queue
peer.queue.RLock()
if peer.isRunning.Get() { if peer.isRunning.Get() {
if peer.queue.packetInNonceQueueIsAwaitingKey.Get() { if peer.queue.packetInNonceQueueIsAwaitingKey.Get() {
peer.SendHandshakeInitiation(false) peer.SendHandshakeInitiation(false)
@ -317,6 +320,7 @@ func (device *Device) RoutineReadFromTUN() {
addToNonceQueue(peer.queue.nonce, elem, device) addToNonceQueue(peer.queue.nonce, elem, device)
elem = nil elem = nil
} }
peer.queue.RUnlock()
} }
} }