Removed old signals
This commit is contained in:
parent
4d9f3a2f53
commit
7a83f2565a
5
event.go
5
event.go
@ -29,11 +29,10 @@ func (e *Event) Clear() {
|
||||
}
|
||||
|
||||
func (e *Event) Fire() {
|
||||
if e == nil || atomic.SwapInt32(&e.guard, 1) != 0 {
|
||||
if atomic.SwapInt32(&e.guard, 1) != 0 {
|
||||
return
|
||||
}
|
||||
now := time.Now()
|
||||
if e.next.After(now) {
|
||||
if now := time.Now(); now.After(e.next) {
|
||||
select {
|
||||
case e.C <- struct{}{}:
|
||||
default:
|
||||
|
1
main.go
1
main.go
@ -148,7 +148,6 @@ func main() {
|
||||
|
||||
logger.Debug.Println("Debug log enabled")
|
||||
|
||||
|
||||
if err != nil {
|
||||
logger.Error.Println("Failed to create TUN device:", err)
|
||||
os.Exit(ExitSetupFailed)
|
||||
|
@ -571,7 +571,7 @@ func (peer *Peer) NewKeyPair() *KeyPair {
|
||||
} else {
|
||||
kp.previous = kp.current
|
||||
kp.current = keyPair
|
||||
peer.signal.newKeyPair.Send()
|
||||
peer.event.newKeyPair.Fire()
|
||||
}
|
||||
|
||||
} else {
|
||||
|
43
peer.go
43
peer.go
@ -15,7 +15,7 @@ import (
|
||||
|
||||
const (
|
||||
PeerRoutineNumber = 4
|
||||
EventInterval = time.Millisecond
|
||||
EventInterval = 10 * time.Millisecond
|
||||
)
|
||||
|
||||
type Peer struct {
|
||||
@ -46,18 +46,14 @@ type Peer struct {
|
||||
dataReceived *Event
|
||||
anyAuthenticatedPacketReceived *Event
|
||||
anyAuthenticatedPacketTraversal *Event
|
||||
handshakeComplete *Event
|
||||
handshakeCompleted *Event
|
||||
handshakePushDeadline *Event
|
||||
handshakeBegin *Event
|
||||
ephemeralKeyCreated *Event
|
||||
newKeyPair *Event
|
||||
}
|
||||
|
||||
signal struct {
|
||||
newKeyPair Signal // size 1, new key pair was generated
|
||||
handshakeCompleted Signal // size 1, handshake completed
|
||||
handshakeBegin Signal // size 1, begin new handshake begin
|
||||
messageSend Signal // size 1, message was send to peer
|
||||
messageReceived Signal // size 1, authenticated message recv
|
||||
|
||||
flushNonceQueue chan struct{} // size 0, empty queued packets
|
||||
}
|
||||
|
||||
@ -115,6 +111,18 @@ func (device *Device) NewPeer(pk NoisePublicKey) (*Peer, error) {
|
||||
peer.device = device
|
||||
peer.isRunning.Set(false)
|
||||
|
||||
// events
|
||||
|
||||
peer.event.dataSent = newEvent(EventInterval)
|
||||
peer.event.dataReceived = newEvent(EventInterval)
|
||||
peer.event.anyAuthenticatedPacketReceived = newEvent(EventInterval)
|
||||
peer.event.anyAuthenticatedPacketTraversal = newEvent(EventInterval)
|
||||
peer.event.handshakeCompleted = newEvent(EventInterval)
|
||||
peer.event.handshakePushDeadline = newEvent(EventInterval)
|
||||
peer.event.handshakeBegin = newEvent(EventInterval)
|
||||
peer.event.ephemeralKeyCreated = newEvent(EventInterval)
|
||||
peer.event.newKeyPair = newEvent(EventInterval)
|
||||
|
||||
// map public key
|
||||
|
||||
_, ok := device.peers.keyMap[pk]
|
||||
@ -202,22 +210,8 @@ func (peer *Peer) Start() {
|
||||
peer.routines.starting.Wait()
|
||||
peer.routines.stopping.Wait()
|
||||
|
||||
// events
|
||||
|
||||
peer.event.dataSent = newEvent(EventInterval)
|
||||
peer.event.dataReceived = newEvent(EventInterval)
|
||||
peer.event.anyAuthenticatedPacketReceived = newEvent(EventInterval)
|
||||
peer.event.anyAuthenticatedPacketTraversal = newEvent(EventInterval)
|
||||
peer.event.handshakeComplete = newEvent(EventInterval)
|
||||
peer.event.handshakePushDeadline = newEvent(EventInterval)
|
||||
peer.event.ephemeralKeyCreated = newEvent(EventInterval)
|
||||
|
||||
// prepare queues and signals
|
||||
|
||||
peer.signal.newKeyPair = NewSignal()
|
||||
peer.signal.handshakeBegin = NewSignal()
|
||||
peer.signal.handshakeCompleted = NewSignal()
|
||||
|
||||
peer.signal.flushNonceQueue = make(chan struct{})
|
||||
|
||||
peer.queue.nonce = make(chan *QueueOutboundElement, QueueOutboundSize)
|
||||
@ -269,12 +263,7 @@ func (peer *Peer) Stop() {
|
||||
|
||||
// close signals
|
||||
|
||||
peer.signal.newKeyPair.Close()
|
||||
peer.signal.handshakeBegin.Close()
|
||||
peer.signal.handshakeCompleted.Close()
|
||||
|
||||
close(peer.signal.flushNonceQueue)
|
||||
|
||||
peer.signal.flushNonceQueue = nil
|
||||
|
||||
// clear key pairs
|
||||
|
20
receive.go
20
receive.go
@ -456,8 +456,8 @@ func (device *Device) RoutineHandshake() {
|
||||
|
||||
// update timers
|
||||
|
||||
peer.TimerAnyAuthenticatedPacketTraversal()
|
||||
peer.TimerAnyAuthenticatedPacketReceived()
|
||||
peer.event.anyAuthenticatedPacketTraversal.Fire()
|
||||
peer.event.anyAuthenticatedPacketReceived.Fire()
|
||||
|
||||
// update endpoint
|
||||
|
||||
@ -489,7 +489,7 @@ func (device *Device) RoutineHandshake() {
|
||||
|
||||
err = peer.SendBuffer(packet)
|
||||
if err == nil {
|
||||
peer.TimerAnyAuthenticatedPacketTraversal()
|
||||
peer.event.anyAuthenticatedPacketTraversal.Fire()
|
||||
} else {
|
||||
logError.Println(peer, ": Failed to send handshake response", err)
|
||||
}
|
||||
@ -529,9 +529,9 @@ func (device *Device) RoutineHandshake() {
|
||||
|
||||
// update timers
|
||||
|
||||
peer.TimerAnyAuthenticatedPacketTraversal()
|
||||
peer.TimerAnyAuthenticatedPacketReceived()
|
||||
peer.TimerHandshakeComplete()
|
||||
peer.event.anyAuthenticatedPacketTraversal.Fire()
|
||||
peer.event.anyAuthenticatedPacketReceived.Fire()
|
||||
peer.event.handshakeCompleted.Fire()
|
||||
|
||||
// derive key-pair
|
||||
|
||||
@ -584,8 +584,8 @@ func (peer *Peer) RoutineSequentialReceiver() {
|
||||
continue
|
||||
}
|
||||
|
||||
peer.TimerAnyAuthenticatedPacketTraversal()
|
||||
peer.TimerAnyAuthenticatedPacketReceived()
|
||||
peer.event.anyAuthenticatedPacketTraversal.Fire()
|
||||
peer.event.anyAuthenticatedPacketReceived.Fire()
|
||||
peer.KeepKeyFreshReceiving()
|
||||
|
||||
// check if using new key-pair
|
||||
@ -593,7 +593,7 @@ func (peer *Peer) RoutineSequentialReceiver() {
|
||||
kp := &peer.keyPairs
|
||||
kp.mutex.Lock()
|
||||
if kp.next == elem.keyPair {
|
||||
peer.TimerHandshakeComplete()
|
||||
peer.event.handshakeCompleted.Fire()
|
||||
if kp.previous != nil {
|
||||
device.DeleteKeyPair(kp.previous)
|
||||
}
|
||||
@ -615,7 +615,7 @@ func (peer *Peer) RoutineSequentialReceiver() {
|
||||
logDebug.Println(peer, ": Received keep-alive")
|
||||
continue
|
||||
}
|
||||
peer.TimerDataReceived()
|
||||
peer.event.dataReceived.Fire()
|
||||
|
||||
// verify source and strip padding
|
||||
|
||||
|
11
send.go
11
send.go
@ -222,6 +222,9 @@ func (peer *Peer) RoutineNonce() {
|
||||
// wait for key pair
|
||||
|
||||
for {
|
||||
|
||||
peer.event.newKeyPair.Clear()
|
||||
|
||||
keyPair = peer.keyPairs.Current()
|
||||
if keyPair != nil && keyPair.sendNonce < RejectAfterMessages {
|
||||
if time.Now().Sub(keyPair.created) < RejectAfterTime {
|
||||
@ -229,12 +232,12 @@ func (peer *Peer) RoutineNonce() {
|
||||
}
|
||||
}
|
||||
|
||||
peer.signal.handshakeBegin.Send()
|
||||
peer.event.handshakeBegin.Fire()
|
||||
|
||||
logDebug.Println(peer, ": Awaiting key-pair")
|
||||
|
||||
select {
|
||||
case <-peer.signal.newKeyPair.Wait():
|
||||
case <-peer.event.newKeyPair.C:
|
||||
logDebug.Println(peer, ": Obtained awaited key-pair")
|
||||
case <-peer.signal.flushNonceQueue:
|
||||
goto NextPacket
|
||||
@ -392,9 +395,9 @@ func (peer *Peer) RoutineSequentialSender() {
|
||||
|
||||
// update timers
|
||||
|
||||
peer.TimerAnyAuthenticatedPacketTraversal()
|
||||
peer.event.anyAuthenticatedPacketTraversal.Fire()
|
||||
if len(elem.packet) != MessageKeepaliveSize {
|
||||
peer.TimerDataSent()
|
||||
peer.event.dataSent.Fire()
|
||||
}
|
||||
peer.KeepKeyFreshSending()
|
||||
}
|
||||
|
106
timers.go
106
timers.go
@ -27,10 +27,10 @@ func (peer *Peer) KeepKeyFreshSending() {
|
||||
}
|
||||
nonce := atomic.LoadUint64(&kp.sendNonce)
|
||||
if nonce > RekeyAfterMessages {
|
||||
peer.signal.handshakeBegin.Send()
|
||||
peer.event.handshakeBegin.Fire()
|
||||
}
|
||||
if kp.isInitiator && time.Now().Sub(kp.created) > RekeyAfterTime {
|
||||
peer.signal.handshakeBegin.Send()
|
||||
peer.event.handshakeBegin.Fire()
|
||||
}
|
||||
}
|
||||
|
||||
@ -54,7 +54,7 @@ func (peer *Peer) KeepKeyFreshReceiving() {
|
||||
if send {
|
||||
// do a last minute attempt at initiating a new handshake
|
||||
peer.timer.sendLastMinuteHandshake.Set(true)
|
||||
peer.signal.handshakeBegin.Send()
|
||||
peer.event.handshakeBegin.Fire()
|
||||
}
|
||||
}
|
||||
|
||||
@ -74,55 +74,13 @@ func (peer *Peer) SendKeepAlive() bool {
|
||||
}
|
||||
}
|
||||
|
||||
/* Event:
|
||||
* Sent non-empty (authenticated) transport message
|
||||
*/
|
||||
func (peer *Peer) TimerDataSent() {
|
||||
peer.event.dataSent.Fire()
|
||||
}
|
||||
|
||||
/* Event:
|
||||
* Received non-empty (authenticated) transport message
|
||||
*
|
||||
* Action:
|
||||
* Set a timer to confirm the message using a keep-alive (if not already set)
|
||||
*/
|
||||
func (peer *Peer) TimerDataReceived() {
|
||||
peer.event.dataReceived.Fire()
|
||||
/*
|
||||
if !peer.timer.keepalivePassive.Start(KeepaliveTimeout) {
|
||||
peer.timer.needAnotherKeepalive.Set(true)
|
||||
}
|
||||
*/
|
||||
}
|
||||
|
||||
/* Event:
|
||||
* Any (authenticated) packet received
|
||||
*/
|
||||
func (peer *Peer) TimerAnyAuthenticatedPacketReceived() {
|
||||
peer.event.anyAuthenticatedPacketReceived.Fire()
|
||||
}
|
||||
|
||||
/* Event:
|
||||
* Any authenticated packet send / received.
|
||||
*
|
||||
* Action:
|
||||
* Push persistent keep-alive into the future
|
||||
*/
|
||||
func (peer *Peer) TimerAnyAuthenticatedPacketTraversal() {
|
||||
peer.event.anyAuthenticatedPacketTraversal.Fire()
|
||||
}
|
||||
|
||||
/* Called after successfully completing a handshake.
|
||||
* i.e. after:
|
||||
*
|
||||
* - Valid handshake response
|
||||
* - First transport message under the "next" key
|
||||
*/
|
||||
func (peer *Peer) TimerHandshakeComplete() {
|
||||
peer.signal.handshakeCompleted.Send()
|
||||
peer.device.log.Info.Println(peer, ": New handshake completed")
|
||||
}
|
||||
// peer.device.log.Info.Println(peer, ": New handshake completed")
|
||||
|
||||
/* Event:
|
||||
* An ephemeral key is generated
|
||||
@ -145,10 +103,6 @@ func (peer *Peer) TimerEphemeralKeyCreated() {
|
||||
*/
|
||||
func (peer *Peer) sendNewHandshake() error {
|
||||
|
||||
// temporarily disable the handshake complete signal
|
||||
|
||||
peer.signal.handshakeCompleted.Disable()
|
||||
|
||||
// create initiation message
|
||||
|
||||
msg, err := peer.device.CreateMessageInitiation(peer)
|
||||
@ -166,14 +120,9 @@ func (peer *Peer) sendNewHandshake() error {
|
||||
|
||||
// send to endpoint
|
||||
|
||||
peer.TimerAnyAuthenticatedPacketTraversal()
|
||||
peer.event.anyAuthenticatedPacketTraversal.Fire()
|
||||
|
||||
err = peer.SendBuffer(packet)
|
||||
if err == nil {
|
||||
peer.signal.handshakeCompleted.Enable()
|
||||
}
|
||||
|
||||
return err
|
||||
return peer.SendBuffer(packet)
|
||||
}
|
||||
|
||||
func newTimer() *time.Timer {
|
||||
@ -198,6 +147,8 @@ func (peer *Peer) RoutineTimerHandler() {
|
||||
|
||||
// reset all timers
|
||||
|
||||
enableHandshake := true
|
||||
|
||||
pendingHandshakeNew := false
|
||||
pendingKeepalivePassive := false
|
||||
|
||||
@ -309,12 +260,12 @@ func (peer *Peer) RoutineTimerHandler() {
|
||||
|
||||
// handshake timers
|
||||
|
||||
case <-timerHandshakeNew.C:
|
||||
logInfo.Println(peer, ": Retrying handshake (timer event)")
|
||||
peer.signal.handshakeBegin.Send()
|
||||
|
||||
case <-timerHandshakeTimeout.C:
|
||||
|
||||
// allow new handshake to be send
|
||||
|
||||
enableHandshake = true
|
||||
|
||||
// clear source (in case this is causing problems)
|
||||
|
||||
peer.mutex.Lock()
|
||||
@ -339,6 +290,11 @@ func (peer *Peer) RoutineTimerHandler() {
|
||||
logDebug.Println(peer, ": Send handshake initiation (subsequent)")
|
||||
}
|
||||
|
||||
// disable further handshakes
|
||||
|
||||
peer.event.handshakeBegin.Clear()
|
||||
enableHandshake = false
|
||||
|
||||
case <-timerHandshakeDeadline.C:
|
||||
|
||||
// clear all queued packets and stop keep-alive
|
||||
@ -348,13 +304,19 @@ func (peer *Peer) RoutineTimerHandler() {
|
||||
peer.flushNonceQueue()
|
||||
signalSend(peer.signal.flushNonceQueue)
|
||||
timerKeepalivePersistent.Stop()
|
||||
peer.signal.handshakeBegin.Enable()
|
||||
|
||||
/* signals */
|
||||
// disable further handshakes
|
||||
|
||||
case <-peer.signal.handshakeBegin.Wait():
|
||||
peer.event.handshakeBegin.Clear()
|
||||
enableHandshake = true
|
||||
|
||||
peer.signal.handshakeBegin.Disable()
|
||||
case <-peer.event.handshakeBegin.C:
|
||||
|
||||
if !enableHandshake {
|
||||
continue
|
||||
}
|
||||
|
||||
logDebug.Println(peer, ": Event, Handshake Begin")
|
||||
|
||||
err := peer.sendNewHandshake()
|
||||
|
||||
@ -372,7 +334,12 @@ func (peer *Peer) RoutineTimerHandler() {
|
||||
|
||||
timerHandshakeDeadline.Reset(RekeyAttemptTime)
|
||||
|
||||
case <-peer.signal.handshakeCompleted.Wait():
|
||||
// disable further handshakes
|
||||
|
||||
peer.event.handshakeBegin.Clear()
|
||||
enableHandshake = false
|
||||
|
||||
case <-peer.event.handshakeCompleted.C:
|
||||
|
||||
logInfo.Println(peer, ": Handshake completed")
|
||||
|
||||
@ -383,9 +350,12 @@ func (peer *Peer) RoutineTimerHandler() {
|
||||
|
||||
timerHandshakeTimeout.Stop()
|
||||
timerHandshakeDeadline.Stop()
|
||||
peer.signal.handshakeBegin.Enable()
|
||||
|
||||
peer.timer.sendLastMinuteHandshake.Set(false)
|
||||
|
||||
// allow further handshakes
|
||||
|
||||
peer.event.handshakeBegin.Clear()
|
||||
enableHandshake = true
|
||||
}
|
||||
}
|
||||
}
|
||||
|
Loading…
Reference in New Issue
Block a user