MaxHandshakeAttemptTime = time.Second * 90
)
+const (
+ RekeyAfterTimeReceiving = RekeyAfterTime - KeepaliveTimeout - RekeyTimeout
+)
+
const (
QueueOutboundSize = 1024
QueueInboundSize = 1024
signal struct {
stop chan struct{}
}
- congestionState int32 // used as an atomic bool
- peers map[NoisePublicKey]*Peer
- mac MACStateDevice
+ underLoad int32 // used as an atomic bool
+ peers map[NoisePublicKey]*Peer
+ mac MACStateDevice
}
-const (
- CongestionStateUnderLoad = iota
- CongestionStateOkay
-)
-
func (device *Device) SetPrivateKey(sk NoisePrivateKey) {
device.mutex.Lock()
defer device.mutex.Unlock()
go device.RoutineDecryption()
go device.RoutineHandshake()
}
+
go device.RoutineBusyMonitor()
go device.RoutineReadFromTUN(tun)
go device.RoutineReceiveIncomming()
go device.RoutineWriteToTUN(tun)
+
return device
}
+++ /dev/null
-package main
-
-import (
- "bytes"
- "encoding/binary"
- "sync/atomic"
- "time"
-)
-
-/* Sends a keep-alive if no packets queued for peer
- *
- * Used by initiator of handshake and with active keep-alive
- */
-func (peer *Peer) SendKeepAlive() bool {
- elem := peer.device.NewOutboundElement()
- elem.packet = nil
- if len(peer.queue.nonce) == 0 {
- select {
- case peer.queue.nonce <- elem:
- return true
- default:
- return false
- }
- }
- return true
-}
-
-/* Called when a new authenticated message has been send
- *
- * TODO: This might be done in a faster way
- */
-func (peer *Peer) KeepKeyFreshSending() {
- send := func() bool {
- peer.keyPairs.mutex.RLock()
- defer peer.keyPairs.mutex.RUnlock()
-
- kp := peer.keyPairs.current
- if kp == nil {
- return false
- }
-
- if !kp.isInitiator {
- return false
- }
-
- nonce := atomic.LoadUint64(&kp.sendNonce)
- if nonce > RekeyAfterMessages {
- return true
- }
- return time.Now().Sub(kp.created) > RekeyAfterTime
- }()
- if send {
- sendSignal(peer.signal.handshakeBegin)
- }
-}
-
-/* This is the state machine for handshake initiation
- *
- * Associated with this routine is the signal "handshakeBegin"
- * The routine will read from the "handshakeBegin" channel
- * at most every RekeyTimeout seconds
- */
-func (peer *Peer) RoutineHandshakeInitiator() {
- device := peer.device
- logger := device.log.Debug
- timeout := stoppedTimer()
-
- var elem *QueueOutboundElement
-
- logger.Println("Routine, handshake initator, started for peer", peer.id)
-
- func() {
- for {
- var attempts uint
- var deadline time.Time
-
- // wait for signal
-
- select {
- case <-peer.signal.handshakeBegin:
- case <-peer.signal.stop:
- return
- }
-
- HandshakeLoop:
- for {
- // clear completed signal
-
- select {
- case <-peer.signal.handshakeCompleted:
- case <-peer.signal.stop:
- return
- default:
- }
-
- // create initiation
-
- if elem != nil {
- elem.Drop()
- }
- elem = device.NewOutboundElement()
-
- msg, err := device.CreateMessageInitiation(peer)
- if err != nil {
- device.log.Error.Println("Failed to create initiation message:", err)
- break
- }
-
- // marshal & schedule for sending
-
- writer := bytes.NewBuffer(elem.data[:0])
- binary.Write(writer, binary.LittleEndian, msg)
- elem.packet = writer.Bytes()
- peer.mac.AddMacs(elem.packet)
- addToOutboundQueue(peer.queue.outbound, elem)
-
- if attempts == 0 {
- deadline = time.Now().Add(MaxHandshakeAttemptTime)
- }
-
- // set timeout
-
- attempts += 1
- stopTimer(timeout)
- timeout.Reset(RekeyTimeout)
- device.log.Debug.Println("Handshake initiation attempt", attempts, "queued for peer", peer.id)
-
- // wait for handshake or timeout
-
- select {
- case <-peer.signal.stop:
- return
-
- case <-peer.signal.handshakeCompleted:
- device.log.Debug.Println("Handshake complete")
- break HandshakeLoop
-
- case <-timeout.C:
- device.log.Debug.Println("Timeout")
- if deadline.Before(time.Now().Add(RekeyTimeout)) {
- peer.signal.flushNonceQueue <- struct{}{}
- if !peer.timer.sendKeepalive.Stop() {
- <-peer.timer.sendKeepalive.C
- }
- break HandshakeLoop
- }
- }
- }
- }
- }()
-
- logger.Println("Routine, handshake initator, stopped for peer", peer.id)
-}
next *KeyPair // not yet "confirmed by transport"
}
-/* Called during recieving to confirm the handshake
- * was completed correctly
- */
-func (kp *KeyPairs) Used(key *KeyPair) {
- if key == kp.next {
- kp.mutex.Lock()
- kp.previous = kp.current
- kp.current = key
- kp.next = nil
- kp.mutex.Unlock()
- }
-}
-
func (kp *KeyPairs) Current() *KeyPair {
kp.mutex.RLock()
defer kp.mutex.RUnlock()
"time"
)
+/* We use int32 as atomic bools
+ * (since booleans are not natively supported by sync/atomic)
+ */
+const (
+ AtomicFalse = iota
+ AtomicTrue
+)
+
func min(a uint, b uint) uint {
if a > b {
return b
return a
}
-func sendSignal(c chan struct{}) {
+func signalSend(c chan struct{}) {
select {
case c <- struct{}{}:
default:
}
}
-func stopTimer(timer *time.Timer) {
+func signalClear(c chan struct{}) {
+ select {
+ case <-c:
+ default:
+ }
+}
+
+func timerStop(timer *time.Timer) {
if !timer.Stop() {
select {
case <-timer.C:
}
}
-func stoppedTimer() *time.Timer {
+func NewStoppedTimer() *time.Timer {
timer := time.NewTimer(time.Hour)
- stopTimer(timer)
+ timerStop(timer)
return timer
}
}
kp.previous = kp.current
kp.current = keyPair
- sendSignal(peer.signal.newKeyPair)
+ signalSend(peer.signal.newKeyPair)
} else {
kp.next = keyPair
}
txBytes uint64
rxBytes uint64
time struct {
+ mutex sync.RWMutex
lastSend time.Time // last send message
lastHandshake time.Time // last completed handshake
+ nextKeepalive time.Time
}
signal struct {
newKeyPair chan struct{} // (size 1) : a new key pair was generated
handshakeBegin chan struct{} // (size 1) : request that a new handshake be started ("queue handshake")
handshakeCompleted chan struct{} // (size 1) : handshake completed
flushNonceQueue chan struct{} // (size 1) : empty queued packets
+ messageSend chan struct{} // (size 1) : a message was send to the peer
+ messageReceived chan struct{} // (size 1) : an authenticated message was received
stop chan struct{} // (size 0) : close to stop all goroutines for peer
}
timer struct {
- sendKeepalive *time.Timer
- handshakeTimeout *time.Timer
+ /* Both keep-alive timers acts as one (see timers.go)
+ * They are kept seperate to simplify the implementation.
+ */
+ keepalivePersistent *time.Timer // set for persistent keepalives
+ keepaliveAcknowledgement *time.Timer // set upon recieving messages
+ zeroAllKeys *time.Timer // zero all key material after RejectAfterTime*3
}
queue struct {
nonce chan *QueueOutboundElement // nonce / pre-handshake queue
outbound chan *QueueOutboundElement // sequential ordering of work
inbound chan *QueueInboundElement // sequential ordering of work
}
+ flags struct {
+ keepaliveWaiting int32
+ }
mac MACStatePeer
}
peer.mac.Init(pk)
peer.device = device
- peer.timer.sendKeepalive = stoppedTimer()
+
+ peer.timer.keepalivePersistent = NewStoppedTimer()
+ peer.timer.keepaliveAcknowledgement = NewStoppedTimer()
+ peer.timer.zeroAllKeys = NewStoppedTimer()
+
+ peer.flags.keepaliveWaiting = AtomicFalse
// assign id for debugging
peer.queue.outbound = make(chan *QueueOutboundElement, QueueOutboundSize)
peer.queue.inbound = make(chan *QueueInboundElement, QueueInboundSize)
- // prepare signaling
+ // prepare signaling & routines
peer.signal.stop = make(chan struct{})
peer.signal.newKeyPair = make(chan struct{}, 1)
peer.signal.handshakeCompleted = make(chan struct{}, 1)
peer.signal.flushNonceQueue = make(chan struct{}, 1)
- // outbound pipeline
-
go peer.RoutineNonce()
+ go peer.RoutineTimerHandler()
go peer.RoutineHandshakeInitiator()
go peer.RoutineSequentialSender()
go peer.RoutineSequentialReceiver()
"time"
)
-const (
- ElementStateOkay = iota
- ElementStateDropped
-)
-
type QueueHandshakeElement struct {
msgType uint32
packet []byte
}
type QueueInboundElement struct {
- state uint32
+ dropped int32
mutex sync.Mutex
packet []byte
counter uint64
}
func (elem *QueueInboundElement) Drop() {
- atomic.StoreUint32(&elem.state, ElementStateDropped)
+ atomic.StoreInt32(&elem.dropped, AtomicTrue)
}
func (elem *QueueInboundElement) IsDropped() bool {
- return atomic.LoadUint32(&elem.state) == ElementStateDropped
+ return atomic.LoadInt32(&elem.dropped) == AtomicTrue
}
func addToInboundQueue(
// update busy state
if busy {
- atomic.StoreInt32(&device.congestionState, CongestionStateUnderLoad)
+ atomic.StoreInt32(&device.underLoad, AtomicTrue)
} else {
- atomic.StoreInt32(&device.congestionState, CongestionStateOkay)
+ atomic.StoreInt32(&device.underLoad, AtomicFalse)
}
timer.Reset(interval)
work := new(QueueInboundElement)
work.packet = packet
work.keyPair = keyPair
- work.state = ElementStateOkay
+ work.dropped = AtomicFalse
work.mutex.Lock()
// add to decryption queues
// verify mac2
- busy := atomic.LoadInt32(&device.congestionState) == CongestionStateUnderLoad
+ busy := atomic.LoadInt32(&device.underLoad) == AtomicTrue
if busy && !device.mac.CheckMAC2(elem.packet, elem.source) {
sender := binary.LittleEndian.Uint32(elem.packet[4:8]) // "sender" always follows "type"
)
return
}
- sendSignal(peer.signal.handshakeCompleted)
- logDebug.Println("Recieved valid response message for peer", peer.id)
kp := peer.NewKeyPair()
if kp == nil {
logDebug.Println("Failed to derieve key-pair")
}
peer.SendKeepAlive()
+ peer.EventHandshakeComplete()
default:
device.log.Error.Println("Invalid message type in handshake queue")
// check for replay
- // update timers
+ // time (passive) keep-alive
+
+ peer.TimerStartKeepalive()
+
+ // refresh key material (rekey)
- // refresh key material
+ peer.KeepKeyFreshReceiving()
+
+ // check if confirming handshake
+
+ kp := &peer.keyPairs
+ kp.mutex.Lock()
+ if kp.next == elem.keyPair {
+ peer.EventHandshakeComplete()
+ kp.previous = kp.current
+ kp.current = kp.next
+ kp.next = nil
+ }
+ kp.mutex.Unlock()
// check for keep-alive
}
default:
- device.log.Debug.Println("Receieved packet with unknown IP version")
+ logDebug.Println("Receieved packet with unknown IP version")
return
}
* (to allow the construction of transport messages in-place)
*/
type QueueOutboundElement struct {
- state uint32
+ dropped int32
mutex sync.Mutex
data [MaxMessageSize]byte
packet []byte // slice of "data" (always!)
}
func (elem *QueueOutboundElement) Drop() {
- atomic.StoreUint32(&elem.state, ElementStateDropped)
+ atomic.StoreInt32(&elem.dropped, AtomicTrue)
}
func (elem *QueueOutboundElement) IsDropped() bool {
- return atomic.LoadUint32(&elem.state) == ElementStateDropped
+ return atomic.LoadInt32(&elem.dropped) == AtomicTrue
}
func addToOutboundQueue(
}
}
+func addToEncryptionQueue(
+ queue chan *QueueOutboundElement,
+ element *QueueOutboundElement,
+) {
+ for {
+ select {
+ case queue <- element:
+ return
+ default:
+ select {
+ case old := <-queue:
+ old.Drop()
+ old.mutex.Unlock()
+ default:
+ }
+ }
+ }
+}
+
/* Reads packets from the TUN and inserts
* into nonce queue for peer
*
break
}
}
- logDebug.Println("Key pair:", keyPair)
-
- sendSignal(peer.signal.handshakeBegin)
+ signalSend(peer.signal.handshakeBegin)
logDebug.Println("Waiting for key-pair, peer", peer.id)
select {
elem.keyPair = keyPair
elem.nonce = atomic.AddUint64(&keyPair.sendNonce, 1) - 1
+ elem.dropped = AtomicFalse
elem.peer = peer
elem.mutex.Lock()
- // add to parallel processing and sequential consuming queue
+ // add to parallel and sequential queue
- addToOutboundQueue(device.queue.encryption, elem)
+ addToEncryptionQueue(device.queue.encryption, elem)
addToOutboundQueue(peer.queue.outbound, elem)
elem = nil
}
func (device *Device) RoutineEncryption() {
var nonce [chacha20poly1305.NonceSize]byte
for work := range device.queue.encryption {
+
+ // check if dropped
+
if work.IsDropped() {
continue
}
* The routine terminates then the outbound queue is closed.
*/
func (peer *Peer) RoutineSequentialSender() {
- logDebug := peer.device.log.Debug
- logDebug.Println("Routine, sequential sender, started for peer", peer.id)
-
device := peer.device
+ logDebug := device.log.Debug
+ logDebug.Println("Routine, sequential sender, started for peer", peer.id)
+
for {
select {
case <-peer.signal.stop:
logDebug.Println("Routine, sequential sender, stopped for peer", peer.id)
return
case work := <-peer.queue.outbound:
+ work.mutex.Lock()
if work.IsDropped() {
continue
}
- work.mutex.Lock()
+
func() {
- if work.packet == nil {
- return
- }
+
+ // send to endpoint
peer.mutex.RLock()
defer peer.mutex.RUnlock()
}
atomic.AddUint64(&peer.txBytes, uint64(len(work.packet)))
- // shift keep-alive timer
+ // reset keep-alive (passive keep-alives / acknowledgements)
- if peer.persistentKeepaliveInterval != 0 {
- interval := time.Duration(peer.persistentKeepaliveInterval) * time.Second
- peer.timer.sendKeepalive.Reset(interval)
- }
+ peer.TimerResetKeepalive()
}()
}
}
--- /dev/null
+package main
+
+import (
+ "bytes"
+ "encoding/binary"
+ "golang.org/x/crypto/blake2s"
+ "sync/atomic"
+ "time"
+)
+
+/* Called when a new authenticated message has been send
+ *
+ */
+func (peer *Peer) KeepKeyFreshSending() {
+ send := func() bool {
+ peer.keyPairs.mutex.RLock()
+ defer peer.keyPairs.mutex.RUnlock()
+
+ kp := peer.keyPairs.current
+ if kp == nil {
+ return false
+ }
+
+ if !kp.isInitiator {
+ return false
+ }
+
+ nonce := atomic.LoadUint64(&kp.sendNonce)
+ return nonce > RekeyAfterMessages || time.Now().Sub(kp.created) > RekeyAfterTime
+ }()
+ if send {
+ signalSend(peer.signal.handshakeBegin)
+ }
+}
+
+/* Called when a new authenticated message has been recevied
+ *
+ */
+func (peer *Peer) KeepKeyFreshReceiving() {
+ send := func() bool {
+ peer.keyPairs.mutex.RLock()
+ defer peer.keyPairs.mutex.RUnlock()
+
+ kp := peer.keyPairs.current
+ if kp == nil {
+ return false
+ }
+
+ if !kp.isInitiator {
+ return false
+ }
+
+ nonce := atomic.LoadUint64(&kp.sendNonce)
+ return nonce > RekeyAfterMessages || time.Now().Sub(kp.created) > RekeyAfterTimeReceiving
+ }()
+ if send {
+ signalSend(peer.signal.handshakeBegin)
+ }
+}
+
+/* Called after succesfully completing a handshake.
+ * i.e. after:
+ * - Valid handshake response
+ * - First transport message under the "next" key
+ */
+func (peer *Peer) EventHandshakeComplete() {
+ peer.device.log.Debug.Println("Handshake completed")
+ peer.timer.zeroAllKeys.Reset(RejectAfterTime * 3)
+ signalSend(peer.signal.handshakeCompleted)
+}
+
+/* Queues a keep-alive if no packets are queued for peer
+ */
+func (peer *Peer) SendKeepAlive() bool {
+ elem := peer.device.NewOutboundElement()
+ elem.packet = nil
+ if len(peer.queue.nonce) == 0 {
+ select {
+ case peer.queue.nonce <- elem:
+ return true
+ default:
+ return false
+ }
+ }
+ return true
+}
+
+/* Starts the "keep-alive" timer
+ * (if not already running),
+ * in response to incomming messages
+ */
+func (peer *Peer) TimerStartKeepalive() {
+
+ // check if acknowledgement timer set yet
+
+ var waiting int32 = AtomicTrue
+ waiting = atomic.SwapInt32(&peer.flags.keepaliveWaiting, waiting)
+ if waiting == AtomicTrue {
+ return
+ }
+
+ // timer not yet set, start it
+
+ wait := KeepaliveTimeout
+ interval := atomic.LoadUint64(&peer.persistentKeepaliveInterval)
+ if interval > 0 {
+ duration := time.Duration(interval) * time.Second
+ if duration < wait {
+ wait = duration
+ }
+ }
+}
+
+/* Resets both keep-alive timers
+ */
+func (peer *Peer) TimerResetKeepalive() {
+
+ // reset persistent timer
+
+ interval := atomic.LoadUint64(&peer.persistentKeepaliveInterval)
+ if interval > 0 {
+ peer.timer.keepalivePersistent.Reset(
+ time.Duration(interval) * time.Second,
+ )
+ }
+
+ // stop acknowledgement timer
+
+ timerStop(peer.timer.keepaliveAcknowledgement)
+ atomic.StoreInt32(&peer.flags.keepaliveWaiting, AtomicFalse)
+}
+
+func (peer *Peer) BeginHandshakeInitiation() (*QueueOutboundElement, error) {
+
+ // create initiation
+
+ elem := peer.device.NewOutboundElement()
+ msg, err := peer.device.CreateMessageInitiation(peer)
+ if err != nil {
+ return nil, err
+ }
+
+ // marshal & schedule for sending
+
+ writer := bytes.NewBuffer(elem.data[:0])
+ binary.Write(writer, binary.LittleEndian, msg)
+ elem.packet = writer.Bytes()
+ peer.mac.AddMacs(elem.packet)
+ addToOutboundQueue(peer.queue.outbound, elem)
+ return elem, err
+}
+
+func (peer *Peer) RoutineTimerHandler() {
+ device := peer.device
+
+ logDebug := device.log.Debug
+ logDebug.Println("Routine, timer handler, started for peer", peer.id)
+
+ for {
+ select {
+
+ case <-peer.signal.stop:
+ return
+
+ // keep-alives
+
+ case <-peer.timer.keepalivePersistent.C:
+
+ logDebug.Println("Sending persistent keep-alive to peer", peer.id)
+
+ peer.SendKeepAlive()
+ peer.TimerResetKeepalive()
+
+ case <-peer.timer.keepaliveAcknowledgement.C:
+
+ logDebug.Println("Sending passive persistent keep-alive to peer", peer.id)
+
+ peer.SendKeepAlive()
+ peer.TimerResetKeepalive()
+
+ // clear key material
+
+ case <-peer.timer.zeroAllKeys.C:
+
+ logDebug.Println("Clearing all key material for peer", peer.id)
+
+ // zero out key pairs
+
+ func() {
+ kp := &peer.keyPairs
+ kp.mutex.Lock()
+ // best we can do is wait for GC :( ?
+ kp.current = nil
+ kp.previous = nil
+ kp.next = nil
+ kp.mutex.Unlock()
+ }()
+
+ // zero out handshake
+
+ func() {
+ hs := &peer.handshake
+ hs.mutex.Lock()
+ hs.localEphemeral = NoisePrivateKey{}
+ hs.remoteEphemeral = NoisePublicKey{}
+ hs.chainKey = [blake2s.Size]byte{}
+ hs.hash = [blake2s.Size]byte{}
+ hs.mutex.Unlock()
+ }()
+ }
+ }
+}
+
+/* This is the state machine for handshake initiation
+ *
+ * Associated with this routine is the signal "handshakeBegin"
+ * The routine will read from the "handshakeBegin" channel
+ * at most every RekeyTimeout seconds
+ */
+func (peer *Peer) RoutineHandshakeInitiator() {
+ device := peer.device
+
+ var elem *QueueOutboundElement
+
+ logError := device.log.Error
+ logDebug := device.log.Debug
+ logDebug.Println("Routine, handshake initator, started for peer", peer.id)
+
+ for run := true; run; {
+ var err error
+ var attempts uint
+ var deadline time.Time
+
+ // wait for signal
+
+ select {
+ case <-peer.signal.handshakeBegin:
+ case <-peer.signal.stop:
+ return
+ }
+
+ // wait for handshake
+
+ run = func() bool {
+ for {
+ // clear completed signal
+
+ select {
+ case <-peer.signal.handshakeCompleted:
+ case <-peer.signal.stop:
+ return false
+ default:
+ }
+
+ // create initiation
+
+ if elem != nil {
+ elem.Drop()
+ }
+ elem, err = peer.BeginHandshakeInitiation()
+ if err != nil {
+ logError.Println("Failed to create initiation message:", err)
+ break
+ }
+
+ // set timeout
+
+ attempts += 1
+ if attempts == 1 {
+ deadline = time.Now().Add(MaxHandshakeAttemptTime)
+ }
+ timeout := time.NewTimer(RekeyTimeout)
+ logDebug.Println("Handshake initiation attempt", attempts, "queued for peer", peer.id)
+
+ // wait for handshake or timeout
+
+ select {
+ case <-peer.signal.stop:
+ return true
+
+ case <-peer.signal.handshakeCompleted:
+ <-timeout.C
+ return true
+
+ case <-timeout.C:
+ logDebug.Println("Timeout")
+
+ // check if sufficient time for retry
+
+ if deadline.Before(time.Now().Add(RekeyTimeout)) {
+ signalSend(peer.signal.flushNonceQueue)
+ timerStop(peer.timer.keepalivePersistent)
+ timerStop(peer.timer.keepaliveAcknowledgement)
+ return true
+ }
+ }
+ }
+ return true
+ }()
+
+ signalClear(peer.signal.handshakeBegin)
+ }
+}
projects / thirdparty / wireguard-go.git / commitdiff
