package main import ( "bytes" "encoding/binary" "golang.org/x/crypto/chacha20poly1305" "golang.org/x/net/ipv4" "golang.org/x/net/ipv6" "net" "sync" "sync/atomic" "time" ) type QueueHandshakeElement struct { msgType uint32 packet []byte buffer *[MaxMessageSize]byte source *net.UDPAddr } type QueueInboundElement struct { dropped int32 mutex sync.Mutex buffer *[MaxMessageSize]byte 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 (device *Device) addToInboundQueue( queue chan *QueueInboundElement, element *QueueInboundElement, ) { for { select { case queue <- element: return default: select { case old := <-queue: old.Drop() default: } } } } func (device *Device) addToDecryptionQueue( queue chan *QueueInboundElement, element *QueueInboundElement, ) { for { select { case queue <- element: return default: select { case old := <-queue: // drop & release to potential consumer old.Drop() old.mutex.Unlock() default: } } } } func (device *Device) addToHandshakeQueue( queue chan QueueHandshakeElement, element QueueHandshakeElement, ) { for { select { case queue <- element: return default: select { case elem := <-queue: device.PutMessageBuffer(elem.buffer) default: } } } } func (device *Device) RoutineReceiveIncomming() { logDebug := device.log.Debug logDebug.Println("Routine, receive incomming, started") for { // wait for new conn logDebug.Println("Waiting for udp socket") select { case <-device.signal.stop: return case <-device.signal.newUDPConn: // fetch connection device.net.mutex.RLock() conn := device.net.conn device.net.mutex.RUnlock() if conn == nil { continue } logDebug.Println("Listening for inbound packets") // receive datagrams until conn is closed buffer := device.GetMessageBuffer() for { // read next datagram size, raddr, err := conn.ReadFromUDP(buffer[:]) // Blocks sometimes if err != nil { break } if size < MinMessageSize { continue } // check size of packet packet := buffer[:size] msgType := binary.LittleEndian.Uint32(packet[:4]) var okay bool switch msgType { // check if transport case MessageTransportType: // check size if len(packet) < MessageTransportType { continue } // lookup key pair receiver := binary.LittleEndian.Uint32( packet[MessageTransportOffsetReceiver:MessageTransportOffsetCounter], ) value := device.indices.Lookup(receiver) keyPair := value.keyPair if keyPair == nil { continue } // check key-pair expiry if keyPair.created.Add(RejectAfterTime).Before(time.Now()) { continue } // create work element peer := value.peer elem := &QueueInboundElement{ packet: packet, buffer: buffer, keyPair: keyPair, dropped: AtomicFalse, } elem.mutex.Lock() // add to decryption queues device.addToDecryptionQueue(device.queue.decryption, elem) device.addToInboundQueue(peer.queue.inbound, elem) buffer = device.GetMessageBuffer() continue // otherwise it is a handshake related packet case MessageInitiationType: okay = len(packet) == MessageInitiationSize case MessageResponseType: okay = len(packet) == MessageResponseSize case MessageCookieReplyType: okay = len(packet) == MessageCookieReplySize } if okay { device.addToHandshakeQueue( device.queue.handshake, QueueHandshakeElement{ msgType: msgType, buffer: buffer, packet: packet, source: raddr, }, ) buffer = device.GetMessageBuffer() } } } } } 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() { 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.buffer[: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 temp [MessageHandshakeSize]byte var elem QueueHandshakeElement for { select { case elem = <-device.queue.handshake: case <-device.signal.stop: return } // handle cookie fields and ratelimiting switch elem.msgType { case MessageCookieReplyType: // unmarshal packet logDebug.Println("Process cookie reply from:", elem.source.String()) var reply MessageCookieReply reader := bytes.NewReader(elem.packet) err := binary.Read(reader, binary.LittleEndian, &reply) if err != nil { logDebug.Println("Failed to decode cookie reply") return } // lookup peer and consume response entry := device.indices.Lookup(reply.Receiver) if entry.peer == nil { return } entry.peer.mac.ConsumeReply(&reply) continue case MessageInitiationType, MessageResponseType: // check mac fields and ratelimit if !device.mac.CheckMAC1(elem.packet) { logDebug.Println("Received packet with invalid mac1") return } if device.IsUnderLoad() { if !device.mac.CheckMAC2(elem.packet, elem.source) { // construct cookie reply logDebug.Println("Sending cookie reply to:", elem.source.String()) sender := binary.LittleEndian.Uint32(elem.packet[4:8]) // "sender" always follows "type" reply, err := device.mac.CreateReply(elem.packet, sender, elem.source) if err != nil { logError.Println("Failed to create cookie reply:", err) return } // marshal and send reply writer := bytes.NewBuffer(temp[:0]) binary.Write(writer, binary.LittleEndian, reply) _, err = device.net.conn.WriteToUDP( writer.Bytes(), elem.source, ) if err != nil { logDebug.Println("Failed to send cookie reply:", err) } continue } if !device.ratelimiter.Allow(elem.source.IP) { continue } } default: logError.Println("Invalid packet ended up in the handshake queue") continue } // handle handshake initation/response content switch elem.msgType { case MessageInitiationType: // unmarshal 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") continue } // consume initiation peer := device.ConsumeMessageInitiation(&msg) if peer == nil { logInfo.Println( "Recieved invalid initiation message from", elem.source.IP.String(), elem.source.Port, ) continue } // update timers peer.TimerAnyAuthenticatedPacketTraversal() peer.TimerAnyAuthenticatedPacketReceived() // update endpoint // TODO: Discover destination address also, only update on change peer.mutex.Lock() peer.endpoint = elem.source peer.mutex.Unlock() // create response response, err := device.CreateMessageResponse(peer) if err != nil { logError.Println("Failed to create response message:", err) continue } peer.TimerEphemeralKeyCreated() peer.NewKeyPair() logDebug.Println("Creating response message for", peer.String()) writer := bytes.NewBuffer(temp[:0]) binary.Write(writer, binary.LittleEndian, response) packet := writer.Bytes() peer.mac.AddMacs(packet) // send response _, err = peer.SendBuffer(packet) if err == nil { peer.TimerAnyAuthenticatedPacketTraversal() } case MessageResponseType: logDebug.Println("Process response") // unmarshal 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") continue } // consume response peer := device.ConsumeMessageResponse(&msg) if peer == nil { logInfo.Println( "Recieved invalid response message from", elem.source.IP.String(), elem.source.Port, ) continue } peer.TimerEphemeralKeyCreated() // update timers peer.TimerAnyAuthenticatedPacketTraversal() peer.TimerAnyAuthenticatedPacketReceived() peer.TimerHandshakeComplete() // derive key-pair peer.NewKeyPair() peer.SendKeepAlive() } } } func (peer *Peer) RoutineSequentialReceiver() { var elem *QueueInboundElement device := peer.device logInfo := device.log.Info logError := device.log.Error 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 if elem.IsDropped() { continue } // check for replay if !elem.keyPair.replayFilter.ValidateCounter(elem.counter) { continue } peer.TimerAnyAuthenticatedPacketTraversal() peer.TimerAnyAuthenticatedPacketReceived() peer.KeepKeyFreshReceiving() // check if using new key-pair kp := &peer.keyPairs kp.mutex.Lock() if kp.next == elem.keyPair { peer.TimerHandshakeComplete() kp.previous = kp.current kp.current = kp.next kp.next = nil } kp.mutex.Unlock() // check for keep-alive if len(elem.packet) == 0 { logDebug.Println("Received keep-alive from", peer.String()) continue } peer.TimerDataReceived() // verify source and strip padding switch elem.packet[0] >> 4 { case ipv4.Version: // strip padding if len(elem.packet) < ipv4.HeaderLen { continue } field := elem.packet[IPv4offsetTotalLength : IPv4offsetTotalLength+2] length := binary.BigEndian.Uint16(field) if int(length) > len(elem.packet) || int(length) < ipv4.HeaderLen { continue } elem.packet = elem.packet[:length] // verify IPv4 source src := elem.packet[IPv4offsetSrc : IPv4offsetSrc+net.IPv4len] if device.routingTable.LookupIPv4(src) != peer { logInfo.Println("Packet with unallowed source IP from", peer.String()) continue } case ipv6.Version: // strip padding if len(elem.packet) < ipv6.HeaderLen { continue } field := elem.packet[IPv6offsetPayloadLength : IPv6offsetPayloadLength+2] length := binary.BigEndian.Uint16(field) length += ipv6.HeaderLen if int(length) > len(elem.packet) { continue } elem.packet = elem.packet[:length] // verify IPv6 source src := elem.packet[IPv6offsetSrc : IPv6offsetSrc+net.IPv6len] if device.routingTable.LookupIPv6(src) != peer { logInfo.Println("Packet with unallowed source IP from", peer.String()) continue } default: logInfo.Println("Packet with invalid IP version from", peer.String()) continue } // write to tun atomic.AddUint64(&peer.stats.rxBytes, uint64(len(elem.packet))) _, err := device.tun.device.Write(elem.packet) device.PutMessageBuffer(elem.buffer) if err != nil { logError.Println("Failed to write packet to TUN device:", err) } } }