)
const (
- QueueOutboundSize = 1024
+ QueueOutboundSize = 1024
+ QueueInboundSize = 1024
+ QueueHandshakeSize = 1024
+ QueueHandshakeBusySize = QueueHandshakeSize / 8
+ MinMessageSize = MessageTransportSize
)
routingTable RoutingTable
indices IndexTable
queue struct {
- encryption chan *QueueOutboundElement // parallel work queue
+ encryption chan *QueueOutboundElement
+ decryption chan *QueueInboundElement
+ handshake chan QueueHandshakeElement
+ inbound chan []byte // inbound queue for TUN
+ }
+ signal struct {
+ stop chan struct{}
}
peers map[NoisePublicKey]*Peer
mac MacStateDevice
defer device.mutex.Unlock()
device.log = NewLogger(logLevel)
+ device.mtu = tun.MTU()
device.peers = make(map[NoisePublicKey]*Peer)
device.indices.Init()
device.routingTable.Reset()
// create queues
device.queue.encryption = make(chan *QueueOutboundElement, QueueOutboundSize)
+ device.queue.handshake = make(chan QueueHandshakeElement, QueueHandshakeSize)
+ device.queue.decryption = make(chan *QueueInboundElement, QueueInboundSize)
+
+ // prepare signals
+
+ device.signal.stop = make(chan struct{})
// start workers
for i := 0; i < runtime.NumCPU(); i += 1 {
go device.RoutineEncryption()
+ go device.RoutineDecryption()
+ go device.RoutineHandshake()
}
go device.RoutineReadFromTUN(tun)
+ go device.RoutineReceiveIncomming()
return device
}
func (device *Device) Close() {
device.RemoveAllPeers()
+ close(device.signal.stop)
close(device.queue.encryption)
}
import (
"bytes"
"encoding/binary"
- "net"
"sync/atomic"
"time"
)
return true
}
-func StoppedTimer() *time.Timer {
- timer := time.NewTimer(time.Hour)
- if !timer.Stop() {
- <-timer.C
- }
- return timer
-}
-
/* Called when a new authenticated message has been send
*
* TODO: This might be done in a faster way
device := peer.device
buffer := make([]byte, 1024)
logger := device.log.Debug
- timeout := time.NewTimer(time.Hour)
+ timeout := stoppedTimer()
var work *QueueOutboundElement
// set timeout
- if !timeout.Stop() {
- select {
- case <-timeout.C:
- default:
- }
- }
attempts += 1
+ stopTimer(timeout)
timeout.Reset(RekeyTimeout)
device.log.Debug.Println("Handshake initiation attempt", attempts, "queued for peer", peer.id)
logger.Println("Routine, handshake initator, stopped for peer", peer.id)
}
-
-/* Handles incomming packets related to handshake
- *
- *
- */
-func (device *Device) HandshakeWorker(queue chan struct {
- msg []byte
- msgType uint32
- addr *net.UDPAddr
-}) {
- for {
- elem := <-queue
-
- switch elem.msgType {
- case MessageInitiationType:
- if len(elem.msg) != MessageInitiationSize {
- continue
- }
-
- // check for cookie
-
- var msg MessageInitiation
-
- binary.Read(nil, binary.LittleEndian, &msg)
-
- case MessageResponseType:
- if len(elem.msg) != MessageResponseSize {
- continue
- }
-
- // check for cookie
-
- case MessageCookieReplyType:
- if len(elem.msg) != MessageCookieReplySize {
- continue
- }
-
- default:
- device.log.Error.Println("Invalid message type in handshake queue")
- }
- }
-}
package main
+import (
+ "time"
+)
+
func min(a uint, b uint) uint {
if a > b {
return b
default:
}
}
+
+func stopTimer(timer *time.Timer) {
+ if !timer.Stop() {
+ select {
+ case <-timer.C:
+ default:
+ }
+ }
+}
+
+func stoppedTimer() *time.Timer {
+ timer := time.NewTimer(time.Hour)
+ stopTimer(timer)
+ return timer
+}
"golang.org/x/crypto/chacha20poly1305"
"golang.org/x/crypto/poly1305"
"sync"
+ "time"
)
const (
MessageInitiationSize = 148
MessageResponseSize = 92
MessageCookieReplySize = 64
+ MessageTransportSize = 16 + poly1305.TagSize // size of empty transport
+)
+
+const (
+ MessageTransportOffsetReceiver = 4
+ MessageTransportOffsetCounter = 8
+ MessageTransportOffsetContent = 16
)
/* Type is an 8-bit field, followed by 3 nul bytes,
type MessageResponse struct {
Type uint32
Sender uint32
- Reciever uint32
+ Receiver uint32
Ephemeral NoisePublicKey
Empty [poly1305.TagSize]byte
Mac1 [blake2s.Size128]byte
type MessageTransport struct {
Type uint32
- Reciever uint32
+ Receiver uint32
Counter uint64
Content []byte
}
var msg MessageResponse
msg.Type = MessageResponseType
msg.Sender = handshake.localIndex
- msg.Reciever = handshake.remoteIndex
+ msg.Receiver = handshake.remoteIndex
// create ephemeral key
}
msg.Ephemeral = handshake.localEphemeral.publicKey()
handshake.mixHash(msg.Ephemeral[:])
+ handshake.mixKey(msg.Ephemeral[:])
func() {
ss := handshake.localEphemeral.sharedSecret(handshake.remoteEphemeral)
// lookup handshake by reciever
- lookup := device.indices.Lookup(msg.Reciever)
+ lookup := device.indices.Lookup(msg.Receiver)
handshake := lookup.handshake
if handshake == nil {
return nil
// finish 3-way DH
hash = mixHash(handshake.hash, msg.Ephemeral[:])
- chainKey = handshake.chainKey
+ chainKey = mixKey(handshake.chainKey, msg.Ephemeral[:])
func() {
ss := handshake.localEphemeral.sharedSecret(msg.Ephemeral)
aead, _ := chacha20poly1305.New(key[:])
_, err := aead.Open(nil, ZeroNonce[:], msg.Empty[:], hash[:])
if err != nil {
+ device.log.Debug.Println("failed to open")
return false
}
hash = mixHash(hash, msg.Empty[:])
keyPair.recv, _ = chacha20poly1305.New(recvKey[:])
keyPair.sendNonce = 0
keyPair.recvNonce = 0
+ keyPair.created = time.Now()
// remap index
queue struct {
nonce chan []byte // nonce / pre-handshake queue
outbound chan *QueueOutboundElement // sequential ordering of work
+ inbound chan *QueueInboundElement // sequential ordering of work
}
mac MacStatePeer
}
peer := new(Peer)
peer.mutex.Lock()
defer peer.mutex.Unlock()
- peer.device = device
+
peer.mac.Init(pk)
- peer.queue.outbound = make(chan *QueueOutboundElement, QueueOutboundSize)
- peer.queue.nonce = make(chan []byte, QueueOutboundSize)
- peer.timer.sendKeepalive = StoppedTimer()
+ peer.device = device
+ peer.timer.sendKeepalive = stoppedTimer()
// assign id for debugging
handshake.precomputedStaticStatic = device.privateKey.sharedSecret(handshake.remoteStatic)
handshake.mutex.Unlock()
+ // prepare queuing
+
+ peer.queue.nonce = make(chan []byte, QueueOutboundSize)
+ peer.queue.inbound = make(chan *QueueInboundElement, QueueInboundSize)
+ peer.queue.outbound = make(chan *QueueOutboundElement, QueueOutboundSize)
+
// prepare signaling
peer.signal.stop = make(chan struct{})
go peer.RoutineNonce()
go peer.RoutineHandshakeInitiator()
go peer.RoutineSequentialSender()
+ go peer.RoutineSequentialReceiver()
return peer
}
--- /dev/null
+package main
+
+import (
+ "bytes"
+ "encoding/binary"
+ "golang.org/x/crypto/chacha20poly1305"
+ "net"
+ "sync"
+ "sync/atomic"
+ "time"
+)
+
+const (
+ ElementStateOkay = iota
+ ElementStateDropped
+)
+
+type QueueHandshakeElement struct {
+ msgType uint32
+ packet []byte
+ source *net.UDPAddr
+}
+
+type QueueInboundElement struct {
+ state uint32
+ mutex sync.Mutex
+ packet []byte
+ counter uint64
+ keyPair *KeyPair
+}
+
+func (elem *QueueInboundElement) Drop() {
+ atomic.StoreUint32(&elem.state, ElementStateDropped)
+ elem.mutex.Unlock()
+}
+
+func (device *Device) RoutineReceiveIncomming() {
+ var packet []byte
+
+ debugLog := device.log.Debug
+ debugLog.Println("Routine, receive incomming, started")
+
+ errorLog := device.log.Error
+
+ for {
+
+ // check if stopped
+
+ select {
+ case <-device.signal.stop:
+ return
+ default:
+ }
+
+ // read next datagram
+
+ if packet == nil {
+ packet = make([]byte, 1<<16)
+ }
+
+ device.net.mutex.RLock()
+ conn := device.net.conn
+ device.net.mutex.RUnlock()
+
+ conn.SetReadDeadline(time.Now().Add(time.Second))
+
+ size, raddr, err := conn.ReadFromUDP(packet)
+ if err != nil {
+ continue
+ }
+ if size < MinMessageSize {
+ continue
+ }
+
+ // handle packet
+
+ packet = packet[:size]
+ msgType := binary.LittleEndian.Uint32(packet[:4])
+
+ func() {
+ switch msgType {
+
+ case MessageInitiationType, MessageResponseType:
+
+ // verify mac1
+
+ if !device.mac.CheckMAC1(packet) {
+ debugLog.Println("Received packet with invalid mac1")
+ return
+ }
+
+ // check if busy, TODO: refine definition of "busy"
+
+ busy := len(device.queue.handshake) > QueueHandshakeBusySize
+ if busy && !device.mac.CheckMAC2(packet, raddr) {
+ sender := binary.LittleEndian.Uint32(packet[4:8]) // "sender" follows "type"
+ reply, err := device.CreateMessageCookieReply(packet, sender, raddr)
+ if err != nil {
+ errorLog.Println("Failed to create cookie reply:", err)
+ return
+ }
+ writer := bytes.NewBuffer(packet[:0])
+ binary.Write(writer, binary.LittleEndian, reply)
+ packet = writer.Bytes()
+ _, err = device.net.conn.WriteToUDP(packet, raddr)
+ if err != nil {
+ debugLog.Println("Failed to send cookie reply:", err)
+ }
+ return
+ }
+
+ // add to handshake queue
+
+ device.queue.handshake <- QueueHandshakeElement{
+ msgType: msgType,
+ packet: packet,
+ source: raddr,
+ }
+
+ case MessageCookieReplyType:
+
+ // verify and update peer cookie state
+
+ if len(packet) != MessageCookieReplySize {
+ return
+ }
+
+ var reply MessageCookieReply
+ reader := bytes.NewReader(packet)
+ err := binary.Read(reader, binary.LittleEndian, &reply)
+ if err != nil {
+ debugLog.Println("Failed to decode cookie reply")
+ return
+ }
+ device.ConsumeMessageCookieReply(&reply)
+
+ case MessageTransportType:
+
+ debugLog.Println("DEBUG: Got transport")
+
+ // lookup key pair
+
+ if len(packet) < MessageTransportSize {
+ return
+ }
+
+ receiver := binary.LittleEndian.Uint32(
+ packet[MessageTransportOffsetReceiver:MessageTransportOffsetCounter],
+ )
+ value := device.indices.Lookup(receiver)
+ keyPair := value.keyPair
+ if keyPair == nil {
+ return
+ }
+
+ // check key-pair expiry
+
+ if keyPair.created.Add(RejectAfterTime).Before(time.Now()) {
+ return
+ }
+
+ // add to peer queue
+
+ peer := value.peer
+ work := new(QueueInboundElement)
+ work.packet = packet
+ work.keyPair = keyPair
+ work.state = ElementStateOkay
+ work.mutex.Lock()
+
+ // add to parallel decryption queue
+
+ func() {
+ for {
+ select {
+ case device.queue.decryption <- work:
+ return
+ default:
+ select {
+ case elem := <-device.queue.decryption:
+ elem.Drop()
+ default:
+ }
+ }
+ }
+ }()
+
+ // add to sequential inbound queue
+
+ func() {
+ for {
+ select {
+ case peer.queue.inbound <- work:
+ break
+ default:
+ select {
+ case elem := <-peer.queue.inbound:
+ elem.Drop()
+ default:
+ }
+ }
+ }
+ }()
+
+ default:
+ // unknown message type
+ }
+ }()
+ }
+}
+
+func (device *Device) RoutineDecryption() {
+ var elem *QueueInboundElement
+ var nonce [chacha20poly1305.NonceSize]byte
+
+ for {
+ select {
+ case elem = <-device.queue.decryption:
+ case <-device.signal.stop:
+ return
+ }
+
+ // check if dropped
+
+ state := atomic.LoadUint32(&elem.state)
+ if state != ElementStateOkay {
+ continue
+ }
+
+ // split message into fields
+
+ counter := binary.LittleEndian.Uint64(
+ elem.packet[MessageTransportOffsetCounter:MessageTransportOffsetContent],
+ )
+ content := elem.packet[MessageTransportOffsetContent:]
+
+ // decrypt with key-pair
+
+ var err error
+ binary.LittleEndian.PutUint64(nonce[4:], counter)
+ elem.packet, err = elem.keyPair.recv.Open(elem.packet[:0], nonce[:], content, nil)
+ if err != nil {
+ elem.Drop()
+ continue
+ }
+
+ // release to consumer
+
+ elem.counter = counter
+ elem.mutex.Unlock()
+ }
+}
+
+/* Handles incomming packets related to handshake
+ *
+ *
+ */
+func (device *Device) RoutineHandshake() {
+
+ logInfo := device.log.Info
+ logError := device.log.Error
+ logDebug := device.log.Debug
+
+ var elem QueueHandshakeElement
+
+ for {
+ select {
+ case elem = <-device.queue.handshake:
+ case <-device.signal.stop:
+ return
+ }
+
+ func() {
+
+ switch elem.msgType {
+ case MessageInitiationType:
+
+ // unmarshal
+
+ if len(elem.packet) != MessageInitiationSize {
+ return
+ }
+
+ var msg MessageInitiation
+ reader := bytes.NewReader(elem.packet)
+ err := binary.Read(reader, binary.LittleEndian, &msg)
+ if err != nil {
+ logError.Println("Failed to decode initiation message")
+ return
+ }
+
+ // consume initiation
+
+ peer := device.ConsumeMessageInitiation(&msg)
+ if peer == nil {
+ logInfo.Println(
+ "Recieved invalid initiation message from",
+ elem.source.IP.String(),
+ elem.source.Port,
+ )
+ return
+ }
+ logDebug.Println("Recieved valid initiation message for peer", peer.id)
+
+ case MessageResponseType:
+
+ // unmarshal
+
+ if len(elem.packet) != MessageResponseSize {
+ return
+ }
+
+ var msg MessageResponse
+ reader := bytes.NewReader(elem.packet)
+ err := binary.Read(reader, binary.LittleEndian, &msg)
+ if err != nil {
+ logError.Println("Failed to decode response message")
+ return
+ }
+
+ // consume response
+
+ peer := device.ConsumeMessageResponse(&msg)
+ if peer == nil {
+ logInfo.Println(
+ "Recieved invalid response message from",
+ elem.source.IP.String(),
+ elem.source.Port,
+ )
+ return
+ }
+ sendSignal(peer.signal.handshakeCompleted)
+ logDebug.Println("Recieved valid response message for peer", peer.id)
+ peer.NewKeyPair()
+ peer.SendKeepAlive()
+
+ default:
+ device.log.Error.Println("Invalid message type in handshake queue")
+ }
+
+ }()
+ }
+}
+
+func (peer *Peer) RoutineSequentialReceiver() {
+ var elem *QueueInboundElement
+
+ device := peer.device
+ logDebug := device.log.Debug
+
+ logDebug.Println("Routine, sequential receiver, started for peer", peer.id)
+
+ for {
+ // wait for decryption
+
+ select {
+ case <-peer.signal.stop:
+ return
+ case elem = <-peer.queue.inbound:
+ }
+ elem.mutex.Lock()
+
+ // check if dropped
+
+ logDebug.Println("MESSSAGE:", elem)
+
+ state := atomic.LoadUint32(&elem.state)
+ if state != ElementStateOkay {
+ continue
+ }
+
+ // check for replay
+
+ // check for keep-alive
+
+ if len(elem.packet) == 0 {
+ continue
+ }
+
+ // insert into inbound TUN queue
+
+ device.queue.inbound <- elem.packet
+ }
+
+}
+
+func (device *Device) RoutineWriteToTUN(tun TUNDevice) {
+ for {
+ var packet []byte
+
+ select {
+ case <-device.signal.stop:
+ case packet = <-device.queue.inbound:
+ }
+
+ device.log.Debug.Println("GOT:", packet)
+
+ size, err := tun.Write(packet)
+ device.log.Debug.Println("DEBUG:", size, err)
+ if err != nil {
+
+ }
+ }
+}
* workers release lock when they have completed work on the packet.
*/
type QueueOutboundElement struct {
+ state uint32
mutex sync.Mutex
packet []byte
nonce uint64
}
}
+func (elem *QueueOutboundElement) Drop() {
+ atomic.StoreUint32(&elem.state, ElementStateDropped)
+}
+
+func (elem *QueueOutboundElement) IsDropped() bool {
+ return atomic.LoadUint32(&elem.state) == ElementStateDropped
+}
+
/* Reads packets from the TUN and inserts
* into nonce queue for peer
*
}
}
+ logger.Println("PACKET:", packet)
+
// wait for key pair
for {
break
}
}
+ logger.Println("Key pair:", keyPair)
sendSignal(peer.signal.handshakeBegin)
logger.Println("Waiting for key-pair, peer", peer.id)
work := new(QueueOutboundElement) // TODO: profile, maybe use pool
work.keyPair = keyPair
work.packet = packet
- work.nonce = atomic.AddUint64(&keyPair.sendNonce, 1)
+ work.nonce = atomic.AddUint64(&keyPair.sendNonce, 1) - 1
work.peer = peer
work.mutex.Lock()
+ logger.Println("WORK:", work)
+
packet = nil
// drop packets until there is space
case peer.device.queue.encryption <- work:
return
default:
- drop := <-peer.device.queue.encryption
- drop.packet = nil
- drop.mutex.Unlock()
+ select {
+ case elem := <-peer.device.queue.encryption:
+ elem.Drop()
+ default:
+ }
}
}
}()
func (device *Device) RoutineEncryption() {
var nonce [chacha20poly1305.NonceSize]byte
for work := range device.queue.encryption {
+ if work.IsDropped() {
+ continue
+ }
// pad packet
padding := device.mtu - len(work.packet)
if padding < 0 {
- // drop
- work.packet = nil
- work.mutex.Unlock()
+ work.Drop()
+ continue
}
+
for n := 0; n < padding; n += 1 {
work.packet = append(work.packet, 0)
}
+ device.log.Debug.Println(work.packet)
// encrypt
logger.Println("Routine, sequential sender, stopped for peer", peer.id)
return
case work := <-peer.queue.outbound:
+ if work.IsDropped() {
+ continue
+ }
work.mutex.Lock()
func() {
if work.packet == nil {
return
}
- logger.Println("Sending packet for peer", peer.id, work.packet)
+ logger.Println(work.packet)
_, err := device.net.conn.WriteToUDP(work.packet, peer.endpoint)
- logger.Println("SEND:", peer.endpoint, err)
+ if err != nil {
+ return
+ }
atomic.AddUint64(&peer.tx_bytes, uint64(len(work.packet)))
// shift keep-alive timer
peer.timer.sendKeepalive.Reset(interval)
}
}()
- work.mutex.Unlock()
}
}
}
Read([]byte) (int, error)
Write([]byte) (int, error)
Name() string
- MTU() uint
+ MTU() int
}
type NativeTun struct {
fd *os.File
name string
- mtu uint
+ mtu int
}
func (tun *NativeTun) Name() string {
return tun.name
}
-func (tun *NativeTun) MTU() uint {
+func (tun *NativeTun) MTU() int {
return tun.mtu
}
projects / thirdparty / wireguard-go.git / commitdiff
