package main import ( "bytes" "encoding/binary" "golang.org/x/crypto/chacha20poly1305" "net" "sync" "sync/atomic" "time" ) type QueueHandshakeElement struct { msgType uint32 packet []byte source *net.UDPAddr } type QueueInboundElement struct { dropped int32 mutex sync.Mutex packet []byte counter uint64 keyPair *KeyPair } func (elem *QueueInboundElement) Drop() { atomic.StoreInt32(&elem.dropped, AtomicTrue) } func (elem *QueueInboundElement) IsDropped() bool { return atomic.LoadInt32(&elem.dropped) == AtomicTrue } func addToInboundQueue( queue chan *QueueInboundElement, element *QueueInboundElement, ) { for { select { case queue <- element: return default: select { case old := <-queue: old.Drop() default: } } } } func addToHandshakeQueue( queue chan QueueHandshakeElement, element QueueHandshakeElement, ) { for { select { case queue <- element: return default: select { case <-queue: default: } } } } /* Routine determining the busy state of the interface * * TODO: prehaps nicer to do this in response to events * TODO: more well reasoned definition of "busy" */ func (device *Device) RoutineBusyMonitor() { samples := 0 interval := time.Second for timer := time.NewTimer(interval); ; { select { case <-device.signal.stop: return case <-timer.C: } // compute busy heuristic if len(device.queue.handshake) > QueueHandshakeBusySize { samples += 1 } else if samples > 0 { samples -= 1 } samples %= 30 busy := samples > 5 // update busy state if busy { atomic.StoreInt32(&device.underLoad, AtomicTrue) } else { atomic.StoreInt32(&device.underLoad, AtomicFalse) } timer.Reset(interval) } } func (device *Device) RoutineReceiveIncomming() { logDebug := device.log.Debug logDebug.Println("Routine, receive incomming, started") var buffer []byte for { // check if stopped select { case <-device.signal.stop: return default: } // read next datagram if buffer == nil { buffer = make([]byte, MaxMessageSize) } device.net.mutex.RLock() conn := device.net.conn device.net.mutex.RUnlock() if conn == nil { time.Sleep(time.Second) continue } conn.SetReadDeadline(time.Now().Add(time.Second)) size, raddr, err := conn.ReadFromUDP(buffer) if err != nil || size < MinMessageSize { continue } // handle packet packet := buffer[:size] msgType := binary.LittleEndian.Uint32(packet[:4]) func() { switch msgType { case MessageInitiationType, MessageResponseType: // add to handshake queue addToHandshakeQueue( device.queue.handshake, QueueHandshakeElement{ msgType: msgType, packet: packet, source: raddr, }, ) buffer = nil 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 { logDebug.Println("Failed to decode cookie reply") return } device.ConsumeMessageCookieReply(&reply) case MessageTransportType: // 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.dropped = AtomicFalse work.mutex.Lock() // add to decryption queues addToInboundQueue(device.queue.decryption, work) addToInboundQueue(peer.queue.inbound, work) buffer = nil default: // unknown message type logDebug.Println("Got unknown message from:", raddr) } }() } } func (device *Device) RoutineDecryption() { var elem *QueueInboundElement var nonce [chacha20poly1305.NonceSize]byte logDebug := device.log.Debug logDebug.Println("Routine, decryption, started for device") for { select { case elem = <-device.queue.decryption: case <-device.signal.stop: return } // check if dropped if elem.IsDropped() { elem.mutex.Unlock() continue } // split message into fields counter := elem.packet[MessageTransportOffsetCounter:MessageTransportOffsetContent] content := elem.packet[MessageTransportOffsetContent:] // decrypt with key-pair var err error copy(nonce[4:], counter) elem.counter = binary.LittleEndian.Uint64(counter) elem.packet, err = elem.keyPair.receive.Open(elem.packet[:0], nonce[:], content, nil) if err != nil { elem.Drop() } 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 logDebug.Println("Routine, handshake routine, started for device") var elem QueueHandshakeElement for { select { case elem = <-device.queue.handshake: case <-device.signal.stop: return } func() { // verify mac1 if !device.mac.CheckMAC1(elem.packet) { logDebug.Println("Received packet with invalid mac1") return } // verify mac2 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" reply, err := device.CreateMessageCookieReply(elem.packet, sender, elem.source) if err != nil { logError.Println("Failed to create cookie reply:", err) return } writer := bytes.NewBuffer(elem.packet[:0]) binary.Write(writer, binary.LittleEndian, reply) elem.packet = writer.Bytes() _, err = device.net.conn.WriteToUDP(elem.packet, elem.source) if err != nil { logDebug.Println("Failed to send cookie reply:", err) } return } // ratelimit // handle messages 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 } // create response response, err := device.CreateMessageResponse(peer) if err != nil { logError.Println("Failed to create response message:", err) return } outElem := device.NewOutboundElement() writer := bytes.NewBuffer(outElem.data[:0]) binary.Write(writer, binary.LittleEndian, response) elem.packet = writer.Bytes() peer.mac.AddMacs(elem.packet) addToOutboundQueue(peer.queue.outbound, outElem) 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 } 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") } }() } } 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() // process packet func() { if elem.IsDropped() { return } // check for replay if !elem.keyPair.replayFilter.ValidateCounter(elem.counter) { return } // time (passive) keep-alive peer.TimerStartKeepalive() // refresh key material (rekey) 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 if len(elem.packet) == 0 { return } // verify source and strip padding switch elem.packet[0] >> 4 { case IPv4version: // strip padding if len(elem.packet) < IPv4headerSize { return } field := elem.packet[IPv4offsetTotalLength : IPv4offsetTotalLength+2] length := binary.BigEndian.Uint16(field) elem.packet = elem.packet[:length] // verify IPv4 source dst := elem.packet[IPv4offsetDst : IPv4offsetDst+net.IPv4len] if device.routingTable.LookupIPv4(dst) != peer { return } case IPv6version: // strip padding if len(elem.packet) < IPv6headerSize { return } field := elem.packet[IPv6offsetPayloadLength : IPv6offsetPayloadLength+2] length := binary.BigEndian.Uint16(field) length += IPv6headerSize elem.packet = elem.packet[:length] // verify IPv6 source dst := elem.packet[IPv6offsetDst : IPv6offsetDst+net.IPv6len] if device.routingTable.LookupIPv6(dst) != peer { return } default: logDebug.Println("Receieved packet with unknown IP version") return } atomic.AddUint64(&peer.rxBytes, uint64(len(elem.packet))) addToInboundQueue(device.queue.inbound, elem) }() } } func (device *Device) RoutineWriteToTUN(tun TUNDevice) { logError := device.log.Error logDebug := device.log.Debug logDebug.Println("Routine, sequential tun writer, started") for { select { case <-device.signal.stop: return case elem := <-device.queue.inbound: _, err := tun.Write(elem.packet) if err != nil { logError.Println("Failed to write packet to TUN device:", err) } } } }