path: root/src/send.go

package main

import (
	"encoding/binary"
	"golang.org/x/crypto/chacha20poly1305"
	"net"
	"sync"
	"sync/atomic"
	"time"
)

/* Handles outbound flow
 *
 * 1. TUN queue
 * 2. Routing (sequential)
 * 3. Nonce assignment (sequential)
 * 4. Encryption (parallel)
 * 5. Transmission (sequential)
 *
 * The order of packets (per peer) is maintained.
 * The functions in this file occure (roughly) in the order packets are processed.
 */

/* A work unit
 *
 * The sequential consumers will attempt to take the lock,
 * workers release lock when they have completed work on the packet.
 *
 * If the element is inserted into the "encryption queue",
 * the content is preceeded by enough "junk" to contain the header
 * (to allow the constuction of transport messages in-place)
 */
type QueueOutboundElement struct {
	state   uint32
	mutex   sync.Mutex
	data    [MaxMessageSize]byte
	packet  []byte   // slice of packet (sending)
	nonce   uint64   // nonce for encryption
	keyPair *KeyPair // key-pair for encryption
	peer    *Peer    // related peer
}

func (peer *Peer) FlushNonceQueue() {
	elems := len(peer.queue.nonce)
	for i := 0; i < elems; i += 1 {
		select {
		case <-peer.queue.nonce:
		default:
			return
		}
	}
}

func (device *Device) NewOutboundElement() *QueueOutboundElement {
	elem := new(QueueOutboundElement) // TODO: profile, consider sync.Pool
	return elem
}

func (elem *QueueOutboundElement) Drop() {
	atomic.StoreUint32(&elem.state, ElementStateDropped)
}

func (elem *QueueOutboundElement) IsDropped() bool {
	return atomic.LoadUint32(&elem.state) == ElementStateDropped
}

func addToOutboundQueue(
	queue chan *QueueOutboundElement,
	element *QueueOutboundElement,
) {
	for {
		select {
		case queue <- element:
			return
		default:
			select {
			case old := <-queue:
				old.Drop()
			default:
			}
		}
	}
}

/* Reads packets from the TUN and inserts
 * into nonce queue for peer
 *
 * Obs. Single instance per TUN device
 */
func (device *Device) RoutineReadFromTUN(tun TUNDevice) {
	if tun == nil {
		// dummy
		return
	}

	elem := device.NewOutboundElement()

	device.log.Debug.Println("Routine, TUN Reader: started")
	for {
		// read packet

		if elem == nil {
			elem = device.NewOutboundElement()
		}

		elem.packet = elem.data[MessageTransportHeaderSize:]
		size, err := tun.Read(elem.packet)
		if err != nil {
			device.log.Error.Println("Failed to read packet from TUN device:", err)
			continue
		}
		elem.packet = elem.packet[:size]
		if len(elem.packet) < IPv4headerSize {
			device.log.Error.Println("Packet too short, length:", size)
			continue
		}

		// lookup peer

		var peer *Peer
		switch elem.packet[0] >> 4 {
		case IPv4version:
			dst := elem.packet[IPv4offsetDst : IPv4offsetDst+net.IPv4len]
			peer = device.routingTable.LookupIPv4(dst)

		case IPv6version:
			dst := elem.packet[IPv6offsetDst : IPv6offsetDst+net.IPv6len]
			peer = device.routingTable.LookupIPv6(dst)

		default:
			device.log.Debug.Println("Receieved packet with unknown IP version")
		}

		if peer == nil {
			continue
		}
		if peer.endpoint == nil {
			device.log.Debug.Println("No known endpoint for peer", peer.id)
			continue
		}

		// insert into nonce/pre-handshake queue

		addToOutboundQueue(peer.queue.nonce, elem)
		elem = nil

	}
}

/* Queues packets when there is no handshake.
 * Then assigns nonces to packets sequentially
 * and creates "work" structs for workers
 *
 * TODO: Avoid dynamic allocation of work queue elements
 *
 * Obs. A single instance per peer
 */
func (peer *Peer) RoutineNonce() {
	var keyPair *KeyPair
	var elem *QueueOutboundElement

	device := peer.device
	logger := device.log.Debug

	logger.Println("Routine, nonce worker, started for peer", peer.id)

	func() {

		for {
		NextPacket:

			// wait for packet

			if elem == nil {
				select {
				case elem = <-peer.queue.nonce:
				case <-peer.signal.stop:
					return
				}
			}

			// wait for key pair

			for {
				select {
				case <-peer.signal.newKeyPair:
				default:
				}

				keyPair = peer.keyPairs.Current()
				if keyPair != nil && keyPair.sendNonce < RejectAfterMessages {
					if time.Now().Sub(keyPair.created) < RejectAfterTime {
						break
					}
				}
				logger.Println("Key pair:", keyPair)

				sendSignal(peer.signal.handshakeBegin)
				logger.Println("Waiting for key-pair, peer", peer.id)

				select {
				case <-peer.signal.newKeyPair:
					logger.Println("Key-pair negotiated for peer", peer.id)
					goto NextPacket

				case <-peer.signal.flushNonceQueue:
					logger.Println("Clearing queue for peer", peer.id)
					peer.FlushNonceQueue()
					elem = nil
					goto NextPacket

				case <-peer.signal.stop:
					return
				}
			}

			// process current packet

			if elem != nil {

				// create work element

				elem.keyPair = keyPair
				elem.nonce = atomic.AddUint64(&keyPair.sendNonce, 1) - 1
				elem.peer = peer
				elem.mutex.Lock()

				// add to parallel processing and sequential consuming queue

				addToOutboundQueue(device.queue.encryption, elem)
				addToOutboundQueue(peer.queue.outbound, elem)
				elem = nil
			}
		}
	}()

	logger.Println("Routine, nonce worker, stopped for peer", peer.id)
}

/* Encrypts the elements in the queue
 * and marks them for sequential consumption (by releasing the mutex)
 *
 * Obs. One instance per core
 */
func (device *Device) RoutineEncryption() {
	var nonce [chacha20poly1305.NonceSize]byte
	for work := range device.queue.encryption {
		if work.IsDropped() {
			continue
		}

		// populate header fields

		func() {
			header := work.data[:MessageTransportHeaderSize]

			fieldType := header[0:4]
			fieldReceiver := header[4:8]
			fieldNonce := header[8:16]

			binary.LittleEndian.PutUint32(fieldType, MessageTransportType)
			binary.LittleEndian.PutUint32(fieldReceiver, work.keyPair.remoteIndex)
			binary.LittleEndian.PutUint64(fieldNonce, work.nonce)
		}()

		// encrypt content

		func() {
			binary.LittleEndian.PutUint64(nonce[4:], work.nonce)
			work.packet = work.keyPair.send.Seal(
				work.packet[:0],
				nonce[:],
				work.packet,
				nil,
			)
			work.mutex.Unlock()
		}()

		// reslice to include header

		work.packet = work.data[:MessageTransportHeaderSize+len(work.packet)]

		// refresh key if necessary

		work.peer.KeepKeyFreshSending()
	}
}

/* Sequentially reads packets from queue and sends to endpoint
 *
 * Obs. Single instance per peer.
 * The routine terminates then the outbound queue is closed.
 */
func (peer *Peer) RoutineSequentialSender() {
	logger := peer.device.log.Debug
	logger.Println("Routine, sequential sender, started for peer", peer.id)

	device := peer.device

	for {
		select {
		case <-peer.signal.stop:
			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
				}

				peer.mutex.RLock()
				defer peer.mutex.RUnlock()

				if peer.endpoint == nil {
					logger.Println("No endpoint for peer:", peer.id)
					return
				}

				device.net.mutex.RLock()
				defer device.net.mutex.RUnlock()

				if device.net.conn == nil {
					logger.Println("No source for device")
					return
				}

				_, err := device.net.conn.WriteToUDP(work.packet, peer.endpoint)
				if err != nil {
					return
				}
				atomic.AddUint64(&peer.tx_bytes, uint64(len(work.packet)))

				// shift keep-alive timer

				if peer.persistentKeepaliveInterval != 0 {
					interval := time.Duration(peer.persistentKeepaliveInterval) * time.Second
					peer.timer.sendKeepalive.Reset(interval)
				}
			}()
		}
	}
}