Renames SetRemotePort to SetRemoteAddr

Source OnDataPathRekeyed from the WGWatcher's per-handshake callback
(onWGCheckSuccess), which fires only on a fresh handshake, and OnDataPathDown
from the handshake-timeout path. A fresh handshake clocks the next chained
KEM exchange pushed over the data-path UDP transport.

client/internal/engine.go

@@ -50,6 +50,7 @@ import (
 	icemaker "github.com/netbirdio/netbird/client/internal/peer/ice"
 	"github.com/netbirdio/netbird/client/internal/peerstore"
 	"github.com/netbirdio/netbird/client/internal/portforward"
+	"github.com/netbirdio/netbird/client/internal/pqkem"
 	"github.com/netbirdio/netbird/client/internal/profilemanager"
 	"github.com/netbirdio/netbird/client/internal/relay"
 	"github.com/netbirdio/netbird/client/internal/rosenpass"
@@ -197,6 +198,10 @@ type Engine struct {
 	// rpManager is a Rosenpass manager
 	rpManager *rosenpass.Manager
 
+	// pqkemManager runs the ML-KEM post-quantum PSK exchange (gated by NB_ENABLE_PQ_MLKEM).
+	// It owns the data-path transport and peer endpoint routing.
+	pqkemManager *pqkem.Manager
+
 	// syncMsgMux is used to guarantee sequential Management Service message processing
 	syncMsgMux *sync.Mutex
 
@@ -651,6 +656,19 @@ func (e *Engine) Start(netbirdConfig *mgmProto.NetbirdConfig, mgmtURL *url.URL)
 		e.rpManager.SetInterface(e.wgInterface)
 	}
 
+	// Start the ML-KEM PQ manager after the interface is up so its dedicated UDP
+	// transport can bind on the WG overlay IP.
+	if pqkem.Enabled() {
+		tr, pqErr := newPQTransport(e.config.WgAddr.IP)
+		if pqErr != nil {
+			log.Errorf("pqkem: transport bind failed, exchange disabled: %v", pqErr)
+		} else {
+			e.pqkemManager = pqkem.NewManager(pqkem.LocalID(publicKey.String()), pqCallbackHandler{wg: e.wgInterface}, pqkem.NewLogger())
+			e.pqkemManager.Start(tr)
+			log.Infof("pqkem: enabled (udp port %d on overlay %s)", e.pqkemManager.LocalPort(), e.config.WgAddr.IP)
+		}
+	}
+
 	// if inbound conns are blocked there is no need to create the ACL manager
 	if e.firewall != nil && !e.config.BlockInbound {
 		e.acl = acl.NewDefaultManager(e.firewall)
@@ -914,6 +932,10 @@ func (e *Engine) removePeer(peerKey string) error {
 
 	e.connMgr.RemovePeerConn(peerKey)
 
+	if e.pqkemManager != nil {
+		e.pqkemManager.RemovePeer(pqkem.RemoteID(peerKey))
+	}
+
 	err := e.statusRecorder.RemovePeer(peerKey)
 	if err != nil {
 		log.Warnf("received error when removing peer %s from status recorder: %v", peerKey, err)
@@ -1900,6 +1922,9 @@ func (e *Engine) createPeerConn(pubKey string, allowedIPs []netip.Prefix, agentV
 		},
 		ICEConfig: e.createICEConfig(),
 	}
+	if e.pqkemManager != nil {
+		config.PQ = pqHandshaker{mgr: e.pqkemManager}
+	}
 
 	serviceDependencies := peer.ServiceDependencies{
 		StatusRecorder:     e.statusRecorder,
@@ -2083,6 +2108,10 @@ func (e *Engine) close() {
 		_ = e.rpManager.Close()
 	}
 
+	if e.pqkemManager != nil {
+		e.pqkemManager.Stop()
+	}
+
 	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 	defer cancel()
 	if err := e.portForwardManager.GracefullyStop(ctx); err != nil {
@@ -2895,6 +2924,8 @@ func convertToOfferAnswer(msg *sProto.Message) (*peer.OfferAnswer, error) {
 		Version:         msg.GetBody().GetNetBirdVersion(),
 		RosenpassPubKey: rosenpassPubKey,
 		RosenpassAddr:   rosenpassAddr,
+		MlkemPayload:    msg.GetBody().GetMlkemPayload(),
+		MlkemPort:       int(msg.GetBody().GetMlkemPort()),
 		RelaySrvAddress: msg.GetBody().GetRelayServerAddress(),
 		RelaySrvIP:      relayIP,
 		SessionID:       sessionID,

client/internal/peer/conn.go

@@ -74,6 +74,32 @@ type RosenpassConfig struct {
 	PermissiveMode bool
 }
 
+// PQHandshaker attaches post-quantum ML-KEM material to signalling offers/answers and
+// feeds received material back. It is implemented by the engine over the pqkem
+// manager and is nil when the PQ exchange is disabled. remoteKey is the peer's
+// WireGuard public key.
+type PQHandshaker interface {
+	// OfferPayload returns the KEM offer to embed in an outgoing offer (nil if this
+	// peer is not the KEM initiator) and the local PQ data-path port to announce.
+	OfferPayload(remoteKey string) (payload []byte, port int)
+	// AnswerPayload processes a received KEM offer (nil if absent) and returns the KEM
+	// answer to embed in the outgoing answer (nil if none) and the local PQ port.
+	AnswerPayload(remoteKey string, recvOffer []byte) (payload []byte, port int)
+	// OnAnswer feeds a received KEM answer (nil if absent).
+	OnAnswer(remoteKey string, recvAnswer []byte)
+	// PSK returns the peer's latest derived post-quantum PSK to program at WG
+	// peer-config time (the pull path). ok is false until one has been derived.
+	PSK(remoteKey string) (wgtypes.Key, bool)
+	// SetRemoteAddr registers the peer's data-path endpoint learned from signalling:
+	// its WG overlay IP with the advertised pq UDP port.
+	SetRemoteAddr(remoteKey string, addr netip.AddrPort)
+	// OnDataPathRekeyed signals a fresh WireGuard handshake for the peer; it clocks the
+	// next chained PSK rotation pushed over the data path.
+	OnDataPathRekeyed(remoteKey string)
+	// OnDataPathDown signals the peer's tunnel went down.
+	OnDataPathDown(remoteKey string)
+}
+
 // ConnConfig is a peer Connection configuration
 type ConnConfig struct {
 	// Key is a public key of a remote peer
@@ -91,6 +117,9 @@ type ConnConfig struct {
 
 	RosenpassConfig RosenpassConfig
 
+	// PQ carries post-quantum ML-KEM material on offers/answers; nil when disabled.
+	PQ PQHandshaker
+
 	// ICEConfig ICE protocol configuration
 	ICEConfig icemaker.Config
 }
@@ -681,6 +710,10 @@ func (conn *Conn) onWGDisconnected(watcherCtx context.Context) {
 
 	conn.Log.Warnf("WireGuard handshake timeout detected, closing current connection")
 
+	if conn.config.PQ != nil {
+		conn.config.PQ.OnDataPathDown(conn.config.Key)
+	}
+
 	// Close the active connection based on current priority
 	switch conn.currentConnPriority {
 	case conntype.Relay:
@@ -946,6 +979,11 @@ func (conn *Conn) onWGCheckSuccess() {
 	conn.mu.Lock()
 	conn.wgTimeouts = 0
 	conn.mu.Unlock()
+
+	// A fresh WireGuard handshake is the clock for the post-quantum PSK rotation.
+	if conn.config.PQ != nil {
+		conn.config.PQ.OnDataPathRekeyed(conn.config.Key)
+	}
 }
 
 // recordConnectionMetrics records connection stage timestamps as metrics
@@ -987,6 +1025,15 @@ func (conn *Conn) AgentVersionString() string {
 }
 
 func (conn *Conn) presharedKey(remoteRosenpassKey []byte) *wgtypes.Key {
+	// Post-quantum: once the ML-KEM exchange has derived a PSK for this peer, program
+	// it here so the peer's next WireGuard handshake adopts it. Applied at peer-config
+	// time (bootstrap / reconnect); steady-state rotation is pushed separately.
+	if conn.config.PQ != nil {
+		if psk, ok := conn.config.PQ.PSK(conn.config.Key); ok {
+			return &psk
+		}
+	}
+
 	if conn.config.RosenpassConfig.PubKey == nil {
 		return conn.config.WgConfig.PreSharedKey
 	}

client/internal/peer/handshaker.go

@@ -39,6 +39,16 @@ type OfferAnswer struct {
 	// This value is the local Rosenpass server address when sending the message
 	RosenpassAddr string
 
+	// MlkemPayload carries the post-quantum X25519MLKEM768 handshake message
+	// (pqkem-framed offer on an OFFER, answer on an ANSWER) that seeds the
+	// WireGuard PSK. Opaque here — the pqkem library frames and parses it. Nil
+	// when the peer does not run the ML-KEM PQ exchange.
+	MlkemPayload []byte
+
+	// MlkemPort is the peer's ML-KEM PQ service UDP port (bound on its WG overlay
+	// IP) where data-path rekey messages are sent. Zero when not running the exchange.
+	MlkemPort int
+
 	// relay server address
 	RelaySrvAddress string
 	// RelaySrvIP is the IP the remote peer is connected to on its
@@ -120,6 +130,8 @@ func (h *Handshaker) Listen(ctx context.Context) {
 
 			h.updateRemoteICEState(&remoteOfferAnswer)
 
+			h.pqRegisterEndpoint(remoteOfferAnswer.MlkemPort)
+
 			if h.relayListener != nil {
 				h.relayListener.Notify(&remoteOfferAnswer)
 			}
@@ -128,7 +140,7 @@ func (h *Handshaker) Listen(ctx context.Context) {
 				h.iceListener(&remoteOfferAnswer)
 			}
 
-			if err := h.sendAnswer(); err != nil {
+			if err := h.sendAnswer(&remoteOfferAnswer); err != nil {
 				h.log.Errorf("failed to send remote offer confirmation: %s", err)
 				continue
 			}
@@ -142,6 +154,8 @@ func (h *Handshaker) Listen(ctx context.Context) {
 
 			h.updateRemoteICEState(&remoteOfferAnswer)
 
+			h.pqRegisterEndpoint(remoteOfferAnswer.MlkemPort)
+
 			if h.relayListener != nil {
 				h.relayListener.Notify(&remoteOfferAnswer)
 			}
@@ -149,6 +163,10 @@ func (h *Handshaker) Listen(ctx context.Context) {
 			if h.iceListener != nil && h.RemoteICESupported() {
 				h.iceListener(&remoteOfferAnswer)
 			}
+
+			if h.config.PQ != nil {
+				h.config.PQ.OnAnswer(h.config.Key, remoteOfferAnswer.MlkemPayload)
+			}
 		case <-ctx.Done():
 			h.log.Infof("stop listening for remote offers and answers")
 			return
@@ -156,6 +174,16 @@ func (h *Handshaker) Listen(ctx context.Context) {
 	}
 }
 
+// pqRegisterEndpoint feeds the post-quantum handshaker the peer's data-path endpoint
+// (its WG overlay IP plus the advertised pq UDP port) learned from a remote offer/answer.
+func (h *Handshaker) pqRegisterEndpoint(remotePort int) {
+	if h.config.PQ == nil || remotePort <= 0 || remotePort > 65535 || len(h.config.WgConfig.AllowedIps) == 0 {
+		return
+	}
+	addr := netip.AddrPortFrom(h.config.WgConfig.AllowedIps[0].Addr(), uint16(remotePort))
+	h.config.PQ.SetRemoteAddr(h.config.Key, addr)
+}
+
 func (h *Handshaker) SendOffer() error {
 	h.mu.Lock()
 	defer h.mu.Unlock()
@@ -195,13 +223,23 @@ func (h *Handshaker) sendOffer() error {
 	}
 
 	offer := h.buildOfferAnswer()
+	if h.config.PQ != nil {
+		offer.MlkemPayload, offer.MlkemPort = h.config.PQ.OfferPayload(h.config.Key)
+	}
 	h.log.Debugf("sending offer with serial: %s", offer.SessionIDString())
 
 	return h.signaler.SignalOffer(offer, h.config.Key)
 }
 
-func (h *Handshaker) sendAnswer() error {
+func (h *Handshaker) sendAnswer(remoteOffer *OfferAnswer) error {
 	answer := h.buildOfferAnswer()
+	if h.config.PQ != nil {
+		var recvOffer []byte
+		if remoteOffer != nil {
+			recvOffer = remoteOffer.MlkemPayload
+		}
+		answer.MlkemPayload, answer.MlkemPort = h.config.PQ.AnswerPayload(h.config.Key, recvOffer)
+	}
 	h.log.Debugf("sending answer with serial: %s", answer.SessionIDString())
 
 	return h.signaler.SignalAnswer(answer, h.config.Key)

client/internal/peer/signaler.go

@@ -63,6 +63,8 @@ func (s *Signaler) signalOfferAnswer(offerAnswer OfferAnswer, remoteKey string,
 		},
 		RosenpassPubKey: offerAnswer.RosenpassPubKey,
 		RosenpassAddr:   offerAnswer.RosenpassAddr,
+		MlkemPayload:    offerAnswer.MlkemPayload,
+		MlkemPort:       offerAnswer.MlkemPort,
 		RelaySrvAddress: offerAnswer.RelaySrvAddress,
 		RelaySrvIP:      offerAnswer.RelaySrvIP,
 		SessionID:       sessionIDBytes,

client/internal/pqkem/bench_test.go

@@ -0,0 +1,59 @@
+package pqkem
+
+import (
+	"crypto/ecdh"
+	"crypto/mlkem"
+	"crypto/rand"
+	"testing"
+)
+
+func BenchmarkX25519Keygen(b *testing.B) {
+	c := ecdh.X25519()
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		if _, err := c.GenerateKey(rand.Reader); err != nil {
+			b.Fatal(err)
+		}
+	}
+}
+
+func BenchmarkX25519ECDH(b *testing.B) {
+	c := ecdh.X25519()
+	a, _ := c.GenerateKey(rand.Reader)
+	p, _ := c.GenerateKey(rand.Reader)
+	pub := p.PublicKey()
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		if _, err := a.ECDH(pub); err != nil {
+			b.Fatal(err)
+		}
+	}
+}
+
+func BenchmarkMLKEMKeygen(b *testing.B) {
+	for i := 0; i < b.N; i++ {
+		if _, err := mlkem.GenerateKey768(); err != nil {
+			b.Fatal(err)
+		}
+	}
+}
+
+func BenchmarkMLKEMEncaps(b *testing.B) {
+	dk, _ := mlkem.GenerateKey768()
+	ek := dk.EncapsulationKey()
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		_, _ = ek.Encapsulate()
+	}
+}
+
+func BenchmarkMLKEMDecaps(b *testing.B) {
+	dk, _ := mlkem.GenerateKey768()
+	_, ct := dk.EncapsulationKey().Encapsulate()
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		if _, err := dk.Decapsulate(ct); err != nil {
+			b.Fatal(err)
+		}
+	}
+}

client/internal/pqkem/callbacks.go

@@ -0,0 +1,18 @@
+package pqkem
+
+// CallbackHandler is implemented by the host and invoked by the library. The
+// library only reports events; the host owns the reaction. Keeping this an
+// interface — rather than touching the transport or keying directly — is what lets
+// the KEM code be extracted as a standalone library.
+type CallbackHandler interface {
+	// OnNewPSKReady fires when a fresh post-quantum PSK has been derived for a peer
+	// and must be programmed into the consumer's secure channel. It is invoked at
+	// the commit point of each side: the initiator on receiving the answer, the
+	// responder on receiving the confirm.
+	OnNewPSKReady(remoteID RemoteID, psk PSK) error
+
+	// OnRekeyFailed fires when an exchange fails to converge within the allotted
+	// time. The host should tear the peer connection down so it re-establishes, and
+	// log a WARN. The library reports the event; it does not dictate the reaction.
+	OnRekeyFailed(remoteID RemoteID) error
+}

client/internal/pqkem/convergence.go

@@ -0,0 +1,224 @@
+package pqkem
+
+import (
+	"context"
+	"time"
+)
+
+// startExchange creates a fresh initiator exchange (acknowledging ackID, zero for a
+// bootstrap) and returns the framed offer for the caller to send — pushed over the
+// data path for a chained rekey, or handed to the host for signalling when viaSignal
+// is set. Any previous in-flight exchange for the peer is cancelled.
+func (m *Manager) startExchange(remoteID RemoteID, viaSignal bool, ackID ExchangeID) ([]byte, error) {
+	init, err := NewInitiator()
+	if err != nil {
+		return nil, err
+	}
+	id, err := newExchangeID()
+	if err != nil {
+		return nil, err
+	}
+	raw, err := (&OfferMsg{ExchangeID: id, AckID: ackID, KEMOffer: init.Offer()}).Encode()
+	if err != nil {
+		return nil, err
+	}
+
+	ctx, cancel := context.WithCancel(m.rootCtx)
+	m.mu.Lock()
+	if old := m.exchanges[remoteID]; old != nil && old.cancel != nil {
+		old.cancel()
+	}
+	m.exchanges[remoteID] = &exchangeCtl{
+		id:        id,
+		state:     stateAwaitingAnswer,
+		startedAt: time.Now(),
+		cancel:    cancel,
+		lastSent:  raw,
+		initiator: init,
+		viaSignal: viaSignal,
+	}
+	m.mu.Unlock()
+
+	m.wait.Add(1)
+	go m.initiatorLoop(ctx, remoteID, id)
+	return raw, nil
+}
+
+// processOffer (responder) first acknowledges the previous exchange the offer names
+// (that offer riding the data path under the freshly adopted key proves it worked),
+// then derives the PSK for the new offer, commits it optimistically, and returns the
+// framed answer. A duplicate offer returns the cached answer without re-deriving.
+func (m *Manager) processOffer(remoteID RemoteID, o *OfferMsg) ([]byte, error) {
+	if o.AckID != (ExchangeID{}) {
+		m.ackConverged(remoteID, o.AckID)
+	}
+
+	m.mu.Lock()
+	if ex := m.exchanges[remoteID]; ex != nil && ex.id == o.ExchangeID {
+		state, last := ex.state, ex.lastSent
+		m.mu.Unlock()
+		if state == stateReserved {
+			return nil, nil
+		}
+		return last, nil
+	}
+	// Reserve the slot so a concurrent duplicate offer bails.
+	m.exchanges[remoteID] = &exchangeCtl{id: o.ExchangeID, state: stateReserved, startedAt: time.Now()}
+	m.mu.Unlock()
+
+	answerBytes, psk, err := Respond(o.KEMOffer, m.binding(remoteID))
+	if err != nil {
+		return nil, err
+	}
+	raw, err := (&AnswerMsg{ExchangeID: o.ExchangeID, KEMAnswer: answerBytes}).Encode()
+	if err != nil {
+		return nil, err
+	}
+
+	m.mu.Lock()
+	ex := m.exchanges[remoteID]
+	if ex == nil || ex.id != o.ExchangeID {
+		m.mu.Unlock()
+		return nil, nil
+	}
+	ex.state = stateAwaitingAck
+	ex.lastSent = raw
+	ex.pendingPSK = psk
+	m.psks[remoteID] = psk
+	m.mu.Unlock()
+
+	// Commit optimistically so our data path can rekey to the new PSK.
+	if err := m.cbHandler.OnNewPSKReady(remoteID, psk); err != nil {
+		return nil, err
+	}
+	return raw, nil
+}
+
+// processAnswer (initiator) derives and commits the PSK and parks in
+// stateAwaitingRekey; the next offer (chained from OnDataPathRekeyed) will acknowledge
+// this exchange. Only valid in stateAwaitingAnswer; advancing the state under the
+// lock makes a concurrent/duplicate answer bail.
+func (m *Manager) processAnswer(remoteID RemoteID, a *AnswerMsg) error {
+	m.mu.Lock()
+	ex := m.exchanges[remoteID]
+	if ex == nil || ex.id != a.ExchangeID || ex.state != stateAwaitingAnswer {
+		m.mu.Unlock()
+		return nil
+	}
+	ex.state = stateAwaitingRekey
+	init := ex.initiator
+	ex.initiator = nil
+	m.mu.Unlock()
+
+	psk, err := init.Finish(a.KEMAnswer, m.binding(remoteID))
+	if err != nil {
+		return err
+	}
+
+	// The initiator has converged: the responder must have derived the key to answer.
+	m.mu.Lock()
+	m.established[remoteID] = true
+	m.failures[remoteID] = 0
+	m.psks[remoteID] = psk
+	m.mu.Unlock()
+
+	return m.cbHandler.OnNewPSKReady(remoteID, psk)
+}
+
+// ackConverged (responder) records convergence of the exchange named by ackID: a
+// later offer acknowledging it proves both sides operate on that exchange's key. Only
+// acts on a matching stateAwaitingAck exchange; anything else is ignored.
+func (m *Manager) ackConverged(remoteID RemoteID, ackID ExchangeID) {
+	m.mu.Lock()
+	ex := m.exchanges[remoteID]
+	if ex == nil || ex.id != ackID || ex.state != stateAwaitingAck {
+		m.mu.Unlock()
+		return
+	}
+	delete(m.exchanges, remoteID)
+	m.established[remoteID] = true
+	m.failures[remoteID] = 0
+	_ = time.Since(ex.startedAt) // convergence latency (metrics hook, later step)
+	m.mu.Unlock()
+}
+
+// initiatorLoop enforces the offer->answer convergence deadline and retransmits the
+// initiator's outstanding data-path offer while awaiting the answer (a
+// signalling-bootstrapped offer is retransmitted by the host, so it is not resent
+// here). Exhausting the deadline before the answer arrives is a failure. Once the
+// answer is in (state past awaitingAnswer) the loop exits: the next rotation is driven
+// by OnDataPathRekeyed, and the idle wait for it has no deadline.
+func (m *Manager) initiatorLoop(ctx context.Context, remoteID RemoteID, id ExchangeID) {
+	defer m.wait.Done()
+	t := time.NewTicker(m.retryInterval)
+	defer t.Stop()
+
+	attempts := 0
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-t.C:
+			m.mu.Lock()
+			ex := m.exchanges[remoteID]
+			if ex == nil || ex.id != id {
+				m.mu.Unlock()
+				return
+			}
+
+			switch ex.state {
+			case stateAwaitingAnswer:
+				if attempts >= m.maxRetries {
+					delete(m.exchanges, remoteID)
+					fail := m.registerFailureLocked(remoteID)
+					m.mu.Unlock()
+					m.raiseFailure(remoteID, fail)
+					return
+				}
+				viaSignal := ex.viaSignal
+				msg := ex.lastSent
+				attempts++
+				m.mu.Unlock()
+				if !viaSignal {
+					if err := m.pushDataPath(remoteID, msg); err != nil {
+						m.logger.Warn("pqkem: offer retransmit failed", "peer", remoteID, "err", err)
+					}
+				}
+
+			default:
+				// Past awaiting the answer (converged) or superseded: the loop's job
+				// is done. The next rotation is driven externally by OnDataPathRekeyed,
+				// so there is no deadline while idle-waiting for it (that wait can be
+				// as long as the transport's natural rekey interval).
+				m.mu.Unlock()
+				return
+			}
+		}
+	}
+}
+
+// registerFailureLocked applies policy B and reports whether OnRekeyFailed is due:
+// an initial exchange (peer never established) fails immediately; a rekey tolerates
+// up to maxRekeyFailures consecutive misses (we stay on the still-valid previous
+// PSK) before failing. Assumes m.mu is held.
+func (m *Manager) registerFailureLocked(remoteID RemoteID) bool {
+	if !m.established[remoteID] {
+		return true
+	}
+	m.failures[remoteID]++
+	if m.failures[remoteID] >= m.maxRekeyFailures {
+		m.failures[remoteID] = 0
+		return true
+	}
+	return false
+}
+
+func (m *Manager) raiseFailure(remoteID RemoteID, fail bool) {
+	if !fail {
+		m.logger.Warn("pqkem: rekey attempt timed out, will retry next cycle", "peer", remoteID)
+		return
+	}
+	if err := m.cbHandler.OnRekeyFailed(remoteID); err != nil {
+		m.logger.Error("pqkem: OnRekeyFailed handler error", "peer", remoteID, "err", err)
+	}
+}

client/internal/pqkem/convergence_test.go

@@ -0,0 +1,74 @@
+package pqkem
+
+import (
+	"net/netip"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+)
+
+// dropTransport is a pqkem.Transport that silently discards everything.
+type dropTransport struct{}
+
+func (dropTransport) Send(netip.AddrPort, []byte) error { return nil }
+func (dropTransport) LocalPort() int                    { return 0 }
+func (dropTransport) Run(func(netip.AddrPort, []byte))  {}
+func (dropTransport) Close() error                      { return nil }
+
+func failedCount(f *fakeWG) int {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	return len(f.failed)
+}
+
+func TestManager_InitialTimeoutFailsImmediately(t *testing.T) {
+	wg := newFakeWG()
+	d := NewManager("bbbb", wg, nil) // bbbb > aaaa -> initiator
+	d.Start(dropTransport{})
+	d.retryInterval = 5 * time.Millisecond
+	d.maxRetries = 3
+	defer d.Stop()
+
+	// Bootstrap offer is produced for signalling; no answer ever comes back -> the
+	// initial exchange fails fast.
+	offer, err := d.SignalOffer("aaaa")
+	require.NoError(t, err)
+	require.NotNil(t, offer)
+
+	require.Eventually(t, func() bool { return failedCount(wg) == 1 }, time.Second, 5*time.Millisecond)
+}
+
+func TestManager_RekeyToleratesKFailures(t *testing.T) {
+	dA, dB, _, wgB, lbB := pair(t)
+	defer dA.Stop()
+	defer dB.Stop()
+
+	// Tighten B's timings before any exchange loop spawns (the loop reads these
+	// fields, so writing them after a loop is running would race).
+	dB.retryInterval = 5 * time.Millisecond
+	dB.maxRetries = 2
+
+	// Establish: bootstrap + data-path-rekeyed so B becomes established and its data
+	// path is usable.
+	bootstrap(t, dA, dB)
+	dA.OnDataPathRekeyed("bbbb")
+	dB.OnDataPathRekeyed("aaaa")
+	require.NotEqual(t, PSK{}, wgB.psk("aaaa"))
+
+	// Drop B's outbound so rekeys can no longer converge.
+	lbB.drop.Store(true)
+
+	// K-1 data-path rekeys must NOT raise OnRekeyFailed.
+	for i := 0; i < DefaultMaxRekeyFailures-1; i++ {
+		_, err := dB.startExchange("aaaa", false, ExchangeID{})
+		require.NoError(t, err)
+		time.Sleep(50 * time.Millisecond)
+	}
+	require.Equal(t, 0, failedCount(wgB), "no failure before K attempts")
+
+	// The K-th failure raises it once.
+	_, err := dB.startExchange("aaaa", false, ExchangeID{})
+	require.NoError(t, err)
+	require.Eventually(t, func() bool { return failedCount(wgB) == 1 }, time.Second, 5*time.Millisecond)
+}

client/internal/pqkem/env.go

@@ -0,0 +1,59 @@
+package pqkem
+
+import (
+	"log/slog"
+	"os"
+	"strconv"
+	"strings"
+
+	log "github.com/sirupsen/logrus"
+)
+
+// EnvEnabled is the environment variable that turns the ML-KEM post-quantum
+// exchange on for this client. Accepts on/off aliases plus anything
+// strconv.ParseBool understands (true/false/1/0).
+const EnvEnabled = "NB_ENABLE_PQ_MLKEM"
+
+// Enabled reports whether the ML-KEM PQ exchange is enabled via the environment.
+// An empty or unrecognized value is treated as disabled.
+func Enabled() bool {
+	raw := strings.ToLower(strings.TrimSpace(os.Getenv(EnvEnabled)))
+	switch raw {
+	case "":
+		return false
+	case "on":
+		return true
+	case "off":
+		return false
+	}
+	enabled, err := strconv.ParseBool(raw)
+	if err != nil {
+		log.Warnf("failed to parse %s value %q: %v", EnvEnabled, raw, err)
+		return false
+	}
+	return enabled
+}
+
+// EnvLogLevel overrides the ML-KEM manager's slog level (debug/info/warn/error).
+// Defaults to info.
+const EnvLogLevel = "NB_PQ_MLKEM_LOG_LEVEL"
+
+// NewLogger builds the slog logger for the ML-KEM manager: a text handler to stdout
+// at the level from EnvLogLevel. Mirrors the Rosenpass manager's logger setup so PQ
+// components log consistently.
+func NewLogger() *slog.Logger {
+	return slog.New(slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{Level: logLevel()}))
+}
+
+func logLevel() slog.Level {
+	switch strings.ToLower(strings.TrimSpace(os.Getenv(EnvLogLevel))) {
+	case "debug":
+		return slog.LevelDebug
+	case "warn":
+		return slog.LevelWarn
+	case "error":
+		return slog.LevelError
+	default:
+		return slog.LevelInfo
+	}
+}

client/internal/pqkem/kem.go

@@ -0,0 +1,167 @@
+// Package pqkem is a spike (NET-1406) for a post-quantum pre-shared-key exchange
+// that could replace Rosenpass. It performs an X25519MLKEM768 hybrid key
+// encapsulation and derives a 32-byte pre-shared key (PSK).
+//
+// The exchange is a single round trip designed to ride the (already
+// authenticated) Signal offer/answer channel:
+//
+//	initiator --Offer(1216B)--> responder
+//	initiator <--Answer(1120B)-- responder
+//
+// Both sides then hold the same PSK, which is bound to the two peers' identities
+// (their peer identity keys) so the derived key cannot be transplanted
+// to a different peer pair even if the transport authentication were bypassed.
+//
+// Combiner note: this follows draft-ietf-tls-ecdhe-mlkem for X25519MLKEM768 — on
+// the wire ML-KEM ‖ X25519 (the draft deliberately reversed the share order for
+// this group), and ML-KEM_ss ‖ X25519_ss fed into the KDF. The spike uses SHA-256
+// (also binding the transcript and peer identities); a production version should
+// use HKDF — see TODO below.
+package pqkem
+
+import (
+	"crypto/ecdh"
+	"crypto/mlkem"
+	"crypto/rand"
+	"crypto/sha256"
+	"fmt"
+)
+
+const (
+	// OfferSize is the initiator message: ML-KEM-768 encapsulation key ‖ X25519 public key
+	// (share order per draft-ietf-tls-ecdhe-mlkem for X25519MLKEM768).
+	OfferSize = mlkem.EncapsulationKeySize768 + 32 // 1216
+	// AnswerSize is the responder message: ML-KEM-768 ciphertext ‖ X25519 public key.
+	AnswerSize = mlkem.CiphertextSize768 + 32 // 1120
+
+	pskLabel = "netbird-pq-psk-v1"
+)
+
+// PSK is the 32-byte derived pre-shared key handed to the consumer to key its channel.
+type PSK [32]byte
+
+// Binding identifies the peer pair the PSK is derived for. Callers set both
+// peer identity keys; the order does not matter (it is canonicalised).
+type Binding struct {
+	LocalID  []byte
+	RemoteID []byte
+}
+
+// Initiator holds the ephemeral secrets between Offer and Finish.
+type Initiator struct {
+	x25519  *ecdh.PrivateKey
+	mlkemDK *mlkem.DecapsulationKey768
+	offer   []byte
+}
+
+// NewInitiator generates the ephemeral X25519 + ML-KEM-768 keypairs.
+func NewInitiator() (*Initiator, error) {
+	x, err := ecdh.X25519().GenerateKey(rand.Reader)
+	if err != nil {
+		return nil, fmt.Errorf("x25519 keygen: %w", err)
+	}
+	dk, err := mlkem.GenerateKey768()
+	if err != nil {
+		return nil, fmt.Errorf("ml-kem keygen: %w", err)
+	}
+
+	offer := make([]byte, 0, OfferSize)
+	offer = append(offer, dk.EncapsulationKey().Bytes()...)
+	offer = append(offer, x.PublicKey().Bytes()...)
+
+	return &Initiator{x25519: x, mlkemDK: dk, offer: offer}, nil
+}
+
+// Offer returns the initiator message to send over Signal.
+func (i *Initiator) Offer() []byte {
+	return i.offer
+}
+
+// Finish consumes the responder's answer and derives the PSK.
+func (i *Initiator) Finish(answer []byte, b Binding) (PSK, error) {
+	if len(answer) != AnswerSize {
+		return PSK{}, fmt.Errorf("answer: got %d bytes, want %d", len(answer), AnswerSize)
+	}
+	ct := answer[:mlkem.CiphertextSize768]
+	peerX := answer[mlkem.CiphertextSize768:]
+
+	ssMLKEM, err := i.mlkemDK.Decapsulate(ct)
+	if err != nil {
+		return PSK{}, fmt.Errorf("ml-kem decapsulate: %w", err)
+	}
+	pub, err := ecdh.X25519().NewPublicKey(peerX)
+	if err != nil {
+		return PSK{}, fmt.Errorf("parse peer x25519: %w", err)
+	}
+	ssX, err := i.x25519.ECDH(pub)
+	if err != nil {
+		return PSK{}, fmt.Errorf("x25519 ecdh: %w", err)
+	}
+
+	return derivePSK(ssMLKEM, ssX, i.offer, answer, b), nil
+}
+
+// Respond consumes an initiator offer, produces the answer, and derives the PSK.
+func Respond(offer []byte, b Binding) (answer []byte, psk PSK, err error) {
+	if len(offer) != OfferSize {
+		return nil, PSK{}, fmt.Errorf("offer: got %d bytes, want %d", len(offer), OfferSize)
+	}
+	peerEK := offer[:mlkem.EncapsulationKeySize768]
+	peerX := offer[mlkem.EncapsulationKeySize768:]
+
+	ek, err := mlkem.NewEncapsulationKey768(peerEK)
+	if err != nil {
+		return nil, PSK{}, fmt.Errorf("parse peer ml-kem key: %w", err)
+	}
+	ssMLKEM, ct := ek.Encapsulate()
+
+	x, err := ecdh.X25519().GenerateKey(rand.Reader)
+	if err != nil {
+		return nil, PSK{}, fmt.Errorf("x25519 keygen: %w", err)
+	}
+	pub, err := ecdh.X25519().NewPublicKey(peerX)
+	if err != nil {
+		return nil, PSK{}, fmt.Errorf("parse peer x25519: %w", err)
+	}
+	ssX, err := x.ECDH(pub)
+	if err != nil {
+		return nil, PSK{}, fmt.Errorf("x25519 ecdh: %w", err)
+	}
+
+	answer = make([]byte, 0, AnswerSize)
+	answer = append(answer, ct...)
+	answer = append(answer, x.PublicKey().Bytes()...)
+
+	// derivePSK uses the same argument order on both sides; the responder's local
+	// binding is the mirror of the initiator's, canonicalised inside derivePSK.
+	return answer, derivePSK(ssMLKEM, ssX, offer, answer, b), nil
+}
+
+// derivePSK combines the two shared secrets and binds the result to the full
+// transcript (offer ‖ answer) and the canonicalised peer identities.
+//
+// TODO(NET-1406): replace the SHA-256 concat with the RFC HKDF combiner
+// (crypto/hkdf, Go 1.24+) and proper labels before this leaves spike status.
+func derivePSK(ssMLKEM, ssX, offer, answer []byte, b Binding) PSK {
+	lo, hi := canonicalPair(b.LocalID, b.RemoteID)
+
+	h := sha256.New()
+	h.Write([]byte(pskLabel))
+	h.Write(ssMLKEM)
+	h.Write(ssX)
+	h.Write(offer)
+	h.Write(answer)
+	h.Write(lo)
+	h.Write(hi)
+
+	var psk PSK
+	copy(psk[:], h.Sum(nil))
+	return psk
+}
+
+func canonicalPair(a, b []byte) (lo, hi []byte) {
+	if string(a) <= string(b) {
+		return a, b
+	}
+	return b, a
+}

client/internal/pqkem/kem_test.go

@@ -0,0 +1,89 @@
+package pqkem
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+)
+
+var (
+	wgA = []byte("peer-A-wireguard-pubkey-32bytes!")
+	wgB = []byte("peer-B-wireguard-pubkey-32bytes!")
+)
+
+func TestExchange_DerivesMatchingPSK(t *testing.T) {
+	init, err := NewInitiator()
+	require.NoError(t, err)
+
+	require.Len(t, init.Offer(), OfferSize)
+
+	answer, pskB, err := Respond(init.Offer(), Binding{LocalID: wgB, RemoteID: wgA})
+	require.NoError(t, err)
+	require.Len(t, answer, AnswerSize)
+
+	pskA, err := init.Finish(answer, Binding{LocalID: wgA, RemoteID: wgB})
+	require.NoError(t, err)
+
+	require.Equal(t, pskB, pskA, "both sides must derive the same PSK")
+	require.NotEqual(t, PSK{}, pskA, "PSK must not be zero")
+}
+
+func TestExchange_PSKBoundToPeerIdentities(t *testing.T) {
+	init, err := NewInitiator()
+	require.NoError(t, err)
+
+	// responder computes with the honest pair...
+	_, pskHonest, err := Respond(init.Offer(), Binding{LocalID: wgB, RemoteID: wgA})
+	require.NoError(t, err)
+
+	// ...a second responder run with a different peer identity yields a different PSK,
+	// even though the KEM material would otherwise combine identically.
+	wgC := []byte("peer-C-wireguard-pubkey-32bytes!")
+	_, pskWrong, err := Respond(init.Offer(), Binding{LocalID: wgC, RemoteID: wgA})
+	require.NoError(t, err)
+
+	require.NotEqual(t, pskHonest, pskWrong, "PSK must be bound to the peer pair")
+}
+
+func TestExchange_RejectsMalformedMessages(t *testing.T) {
+	init, err := NewInitiator()
+	require.NoError(t, err)
+
+	_, _, err = Respond(init.Offer()[:10], Binding{})
+	require.Error(t, err)
+
+	_, err = init.Finish([]byte("too short"), Binding{})
+	require.Error(t, err)
+}
+
+// TestExchange_ReportSizesAndTiming is a spike measurement, not a pass/fail gate.
+// Run with: go test -run TestExchange_ReportSizesAndTiming -v ./client/internal/pqkem/
+func TestExchange_ReportSizesAndTiming(t *testing.T) {
+	const iters = 200
+
+	var tInit, tResp, tFinish time.Duration
+	for i := 0; i < iters; i++ {
+		s0 := time.Now()
+		init, err := NewInitiator()
+		require.NoError(t, err)
+		tInit += time.Since(s0)
+
+		s1 := time.Now()
+		answer, _, err := Respond(init.Offer(), Binding{LocalID: wgB, RemoteID: wgA})
+		require.NoError(t, err)
+		tResp += time.Since(s1)
+
+		s2 := time.Now()
+		_, err = init.Finish(answer, Binding{LocalID: wgA, RemoteID: wgB})
+		require.NoError(t, err)
+		tFinish += time.Since(s2)
+	}
+
+	t.Logf("wire sizes:   offer=%d B   answer=%d B   (Rosenpass static pubkey ~524160 B)", OfferSize, AnswerSize)
+	t.Logf("total on-wire per handshake: %d B  (~%.0fx smaller than RP static key)", OfferSize+AnswerSize, 524160.0/float64(OfferSize+AnswerSize))
+	t.Logf("avg NewInitiator (keygen): %s", tInit/iters)
+	t.Logf("avg Respond (encaps+dh):   %s", tResp/iters)
+	t.Logf("avg Finish (decaps+dh):    %s", tFinish/iters)
+	t.Logf("avg full handshake CPU:    %s", (tInit+tResp+tFinish)/iters)
+}

client/internal/pqkem/manager.go

@@ -0,0 +1,372 @@
+package pqkem
+
+import (
+	"context"
+	"crypto/rand"
+	"fmt"
+	"log/slog"
+	"net/netip"
+	"sync"
+	"time"
+)
+
+const (
+	// DefaultRetryInterval is how often the initiator retransmits its outstanding
+	// data-path offer while awaiting the answer.
+	DefaultRetryInterval = 2 * time.Second
+	// DefaultMaxRetries bounds how many ticks an exchange may run before it is
+	// declared failed. The convergence deadline is thus MaxRetries * RetryInterval.
+	DefaultMaxRetries = 10
+	// DefaultMaxRekeyFailures is how many consecutive rekey (non-initial) failures
+	// are tolerated before OnRekeyFailed. The initial exchange fails immediately.
+	DefaultMaxRekeyFailures = 3
+)
+
+// LocalID and RemoteID are peer identity keys (e.g. WireGuard public keys). They are
+// distinct types so the local and a remote identity cannot be mixed up.
+type (
+	LocalID  string
+	RemoteID string
+)
+
+// Transport is the data-path socket the Manager drives (the analogue of
+// go-rosenpass's Conn). It is a dumb mover of bytes to/from endpoints: the Manager
+// owns the remoteID<->endpoint routing and hands the transport a resolved endpoint
+// to Send, and reverse-resolves the source of each inbound datagram. Its lifecycle
+// belongs to the Manager (Run at Start, Close at Stop).
+type Transport interface {
+	// Send delivers msg to the given data-path endpoint.
+	Send(endpoint netip.AddrPort, msg []byte) error
+	// LocalPort is the bound local UDP port, announced to peers so they know where
+	// to send data-path messages.
+	LocalPort() int
+	// Run starts delivering inbound datagrams as (source endpoint, msg) to onInbound
+	// and returns immediately; it runs until Close.
+	Run(onInbound func(src netip.AddrPort, msg []byte))
+	// Close stops delivery and releases the socket.
+	Close() error
+}
+
+// exchangeState is the single source of truth for an exchange's role and phase.
+type exchangeState uint8
+
+const (
+	stateReserved       exchangeState = iota // responder: deriving the answer
+	stateAwaitingAnswer                      // initiator: offer sent, awaiting the answer
+	stateAwaitingRekey                       // initiator: PSK derived+set, awaiting OnDataPathRekeyed to chain the next offer
+	stateAwaitingAck                         // responder: answer sent, awaiting the next offer that acks this exchange
+)
+
+// exchangeCtl holds all state for one in-flight exchange with a peer, under the
+// Manager's single lock. state drives every decision. lastSent is the current
+// data-path retransmit payload (the offer, for the initiator). initiator is the
+// ephemeral handle used at Finish; pendingPSK is the responder's derived key.
+// viaSignal records that the offer went to the host for the signalling channel, so
+// the loop does not retransmit it on the data path. Only the initiator runs a
+// retransmit loop, so only it sets cancel.
+type exchangeCtl struct {
+	id         ExchangeID
+	state      exchangeState
+	startedAt  time.Time
+	cancel     context.CancelFunc
+	lastSent   []byte
+	initiator  *Initiator
+	pendingPSK PSK
+	viaSignal  bool
+}
+
+// Manager is the stateful orchestrator — the analogue of go-rosenpass's Server. It
+// drives the X25519MLKEM768 exchange, owns the peer endpoint routing and the data-path
+// transport, and surfaces the derived PSK and convergence to the host via
+// CallbackHandler. It is event-driven: the bootstrap is triggered by the host
+// (SignalOffer) and each rotation is clocked by OnDataPathRekeyed. The cryptography is
+// the pure kem.go primitives; all state lives here under one lock.
+type Manager struct {
+	localID   LocalID
+	cbHandler CallbackHandler
+	logger    *slog.Logger
+
+	retryInterval    time.Duration
+	maxRetries       int
+	maxRekeyFailures int
+
+	rootCtx    context.Context
+	rootCancel context.CancelFunc
+
+	mu          sync.Mutex
+	transport   Transport
+	exchanges   map[RemoteID]*exchangeCtl   // in-flight exchange per peer
+	established map[RemoteID]bool           // peer has completed at least one exchange
+	failures    map[RemoteID]int            // consecutive rekey failures per peer
+	psks        map[RemoteID]PSK            // latest derived PSK per peer (pulled at WG peer-config time)
+	peerAddrs   map[RemoteID]netip.AddrPort // remoteID -> data-path endpoint (send routing)
+	peersByAddr map[netip.AddrPort]RemoteID // reverse: source endpoint -> remoteID (inbound)
+	wait        sync.WaitGroup
+}
+
+// NewManager builds a manager for the local peer identified by its peer identity key
+// (used for the deterministic initiator role and the identity binding). A nil logger
+// falls back to slog.Default(). Install the data-path transport with Start.
+func NewManager(localID LocalID, h CallbackHandler, logger *slog.Logger) *Manager {
+	if logger == nil {
+		logger = slog.Default()
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	return &Manager{
+		localID:          localID,
+		cbHandler:        h,
+		logger:           logger,
+		retryInterval:    DefaultRetryInterval,
+		maxRetries:       DefaultMaxRetries,
+		maxRekeyFailures: DefaultMaxRekeyFailures,
+		rootCtx:          ctx,
+		rootCancel:       cancel,
+		exchanges:        make(map[RemoteID]*exchangeCtl),
+		established:      make(map[RemoteID]bool),
+		failures:         make(map[RemoteID]int),
+		psks:             make(map[RemoteID]PSK),
+		peerAddrs:        make(map[RemoteID]netip.AddrPort),
+		peersByAddr:      make(map[netip.AddrPort]RemoteID),
+	}
+}
+
+// Start installs the data-path transport and begins its inbound delivery. The Manager
+// owns it from here; Stop closes it. Start/Stop are the transport lifecycle pair.
+func (m *Manager) Start(t Transport) {
+	m.mu.Lock()
+	m.transport = t
+	m.mu.Unlock()
+	if t != nil {
+		t.Run(m.onDataPathInbound)
+	}
+}
+
+// LocalPort is the data-path transport's bound UDP port (0 if no transport), to be
+// announced to peers.
+func (m *Manager) LocalPort() int {
+	m.mu.Lock()
+	t := m.transport
+	m.mu.Unlock()
+	if t == nil {
+		return 0
+	}
+	return t.LocalPort()
+}
+
+// IsInitiator reports whether the local peer drives the exchange for this remote
+// peer. Roles are deterministic (lexicographic identity-key compare) so exactly one
+// side initiates, mirroring how Rosenpass picks its handshake initiator.
+func (m *Manager) IsInitiator(remoteID RemoteID) bool {
+	return string(m.localID) > string(remoteID)
+}
+
+// PSK returns the latest PSK derived for the peer, for the host to program at WG
+// peer-config time (the pull path). ok is false until an exchange has derived one.
+func (m *Manager) PSK(remoteID RemoteID) (PSK, bool) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	psk, ok := m.psks[remoteID]
+	return psk, ok
+}
+
+// AddPeer registers where a peer's data-path messages are sent and received: its
+// overlay endpoint (IP:port). Re-adding updates the endpoint.
+func (m *Manager) AddPeer(remoteID RemoteID, endpoint netip.AddrPort) {
+	if !endpoint.IsValid() {
+		return
+	}
+	m.mu.Lock()
+	if old, ok := m.peerAddrs[remoteID]; ok {
+		delete(m.peersByAddr, old)
+	}
+	m.peerAddrs[remoteID] = endpoint
+	m.peersByAddr[endpoint] = remoteID
+	m.mu.Unlock()
+}
+
+// RemovePeer stops any in-flight exchange for a peer and drops its state and routing.
+func (m *Manager) RemovePeer(remoteID RemoteID) {
+	m.mu.Lock()
+	if ex, ok := m.exchanges[remoteID]; ok {
+		if ex.cancel != nil {
+			ex.cancel()
+		}
+		delete(m.exchanges, remoteID)
+	}
+	delete(m.established, remoteID)
+	delete(m.failures, remoteID)
+	delete(m.psks, remoteID)
+	if ep, ok := m.peerAddrs[remoteID]; ok {
+		delete(m.peersByAddr, ep)
+		delete(m.peerAddrs, remoteID)
+	}
+	m.mu.Unlock()
+}
+
+// Stop cancels all in-flight exchanges, closes the transport, and waits for the
+// exchange goroutines to exit.
+func (m *Manager) Stop() {
+	m.rootCancel()
+	m.wait.Wait()
+	m.mu.Lock()
+	t := m.transport
+	m.transport = nil
+	m.exchanges = make(map[RemoteID]*exchangeCtl)
+	m.psks = make(map[RemoteID]PSK)
+	m.mu.Unlock()
+	if t != nil {
+		if err := t.Close(); err != nil {
+			m.logger.Warn("pqkem: closing data-path transport", "err", err)
+		}
+	}
+}
+
+// ---- Signalling channel (host-driven; rides the host's negotiation) ----
+
+// SignalOffer returns the KEM offer for the host to embed in its outgoing offer to
+// remoteID (bootstrap). It returns (nil, nil) when the local peer is not the
+// initiator. It is idempotent for an in-flight bootstrap: a repeat call returns the
+// same offer rather than starting a new exchange.
+func (m *Manager) SignalOffer(remoteID RemoteID) ([]byte, error) {
+	if !m.IsInitiator(remoteID) {
+		return nil, nil
+	}
+	m.mu.Lock()
+	if ex := m.exchanges[remoteID]; ex != nil && ex.viaSignal && ex.state == stateAwaitingAnswer {
+		last := ex.lastSent
+		m.mu.Unlock()
+		return last, nil
+	}
+	m.mu.Unlock()
+	// bootstrap offer acknowledges nothing (zero AckID).
+	return m.startExchange(remoteID, true, ExchangeID{})
+}
+
+// SignalOnOffer processes a KEM offer the host extracted from an incoming offer and
+// returns the KEM answer for the host to embed in its outgoing answer.
+func (m *Manager) SignalOnOffer(remoteID RemoteID, offer []byte) ([]byte, error) {
+	typ, msg, err := Decode(offer)
+	if err != nil {
+		return nil, fmt.Errorf("decode signal offer from %s: %w", remoteID, err)
+	}
+	if typ != MsgOffer {
+		return nil, fmt.Errorf("expected offer from %s, got type %d", remoteID, typ)
+	}
+	return m.processOffer(remoteID, msg.(*OfferMsg))
+}
+
+// SignalOnAnswer processes a KEM answer the host extracted from an incoming answer.
+// There is no reply: the next offer (over the data path) acknowledges this exchange.
+func (m *Manager) SignalOnAnswer(remoteID RemoteID, answer []byte) error {
+	typ, msg, err := Decode(answer)
+	if err != nil {
+		return fmt.Errorf("decode signal answer from %s: %w", remoteID, err)
+	}
+	if typ != MsgAnswer {
+		return fmt.Errorf("expected answer from %s, got type %d", remoteID, typ)
+	}
+	return m.processAnswer(remoteID, msg.(*AnswerMsg))
+}
+
+// ---- Data path ----
+
+// onDataPathInbound is the transport's inbound handler: it reverse-resolves the
+// source endpoint to a peer and dispatches. Unknown sources are dropped.
+func (m *Manager) onDataPathInbound(src netip.AddrPort, msg []byte) {
+	m.mu.Lock()
+	remoteID, ok := m.peersByAddr[src]
+	m.mu.Unlock()
+	if !ok {
+		return
+	}
+	if err := m.OnDataPathMessage(remoteID, msg); err != nil {
+		m.logger.Debug("pqkem: inbound", "peer", remoteID, "err", err)
+	}
+}
+
+// OnDataPathMessage handles a KEM message received over the data path from remoteID
+// and pushes any reply back over the data path.
+func (m *Manager) OnDataPathMessage(remoteID RemoteID, raw []byte) error {
+	typ, msg, err := Decode(raw)
+	if err != nil {
+		return fmt.Errorf("decode data-path msg from %s: %w", remoteID, err)
+	}
+	switch typ {
+	case MsgOffer:
+		answer, err := m.processOffer(remoteID, msg.(*OfferMsg))
+		if err != nil {
+			return err
+		}
+		if answer == nil {
+			return nil
+		}
+		return m.pushDataPath(remoteID, answer)
+	case MsgAnswer:
+		return m.processAnswer(remoteID, msg.(*AnswerMsg))
+	default:
+		return fmt.Errorf("unhandled data-path message type %d from %s", typ, remoteID)
+	}
+}
+
+// OnDataPathRekeyed notifies that the peer's data path is up and freshly keyed with
+// the latest PSK (fired on first establishment AND every rekey). If we are the
+// initiator that just derived a PSK, it chains the next exchange: a fresh offer over
+// the data path that acknowledges the just-completed one (its arrival under the new
+// key proves to the responder that the key works).
+func (m *Manager) OnDataPathRekeyed(remoteID RemoteID) {
+	m.mu.Lock()
+	ex := m.exchanges[remoteID]
+	chain := ex != nil && ex.state == stateAwaitingRekey
+	var ackID ExchangeID
+	if chain {
+		ackID = ex.id
+	}
+	m.mu.Unlock()
+
+	if !chain {
+		return
+	}
+	offer, err := m.startExchange(remoteID, false, ackID)
+	if err != nil {
+		m.logger.Error("pqkem: chain offer failed to start", "peer", remoteID, "err", err)
+		return
+	}
+	if err := m.pushDataPath(remoteID, offer); err != nil {
+		m.logger.Warn("pqkem: send chain offer failed", "peer", remoteID, "err", err)
+	}
+}
+
+// OnDataPathDown notifies that the peer's data path went down. Rotations resume once
+// the host re-bootstraps over signalling on reconnect; in-flight data-path sends will
+// simply fail until then. Reserved as an explicit hook.
+func (m *Manager) OnDataPathDown(remoteID RemoteID) {}
+
+// ---- internals ----
+
+// pushDataPath resolves the peer's endpoint and sends over the data-path transport,
+// erroring if the peer is unknown or no transport is set.
+func (m *Manager) pushDataPath(remoteID RemoteID, msg []byte) error {
+	m.mu.Lock()
+	ep, ok := m.peerAddrs[remoteID]
+	t := m.transport
+	m.mu.Unlock()
+	if !ok {
+		return fmt.Errorf("no data-path endpoint for peer %s", remoteID)
+	}
+	if t == nil {
+		return fmt.Errorf("no data-path transport")
+	}
+	return t.Send(ep, msg)
+}
+
+func (m *Manager) binding(remoteID RemoteID) Binding {
+	return Binding{LocalID: []byte(m.localID), RemoteID: []byte(remoteID)}
+}
+
+func newExchangeID() (ExchangeID, error) {
+	var id ExchangeID
+	if _, err := rand.Read(id[:]); err != nil {
+		return ExchangeID{}, fmt.Errorf("generate exchange id: %w", err)
+	}
+	return id, nil
+}

client/internal/pqkem/manager_test.go

@@ -0,0 +1,169 @@
+package pqkem
+
+import (
+	"fmt"
+	"net/netip"
+	"sync"
+	"sync/atomic"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+// netSwitch is an in-memory UDP fabric: transports register their endpoint and get
+// datagrams delivered to their inbound handler.
+type netSwitch struct {
+	mu sync.Mutex
+	h  map[netip.AddrPort]func(netip.AddrPort, []byte)
+}
+
+func newSwitch() *netSwitch {
+	return &netSwitch{h: map[netip.AddrPort]func(netip.AddrPort, []byte){}}
+}
+
+func (s *netSwitch) register(ep netip.AddrPort, fn func(netip.AddrPort, []byte)) {
+	s.mu.Lock()
+	s.h[ep] = fn
+	s.mu.Unlock()
+}
+
+func (s *netSwitch) deliver(dst, src netip.AddrPort, msg []byte) error {
+	s.mu.Lock()
+	fn := s.h[dst]
+	s.mu.Unlock()
+	if fn == nil {
+		return fmt.Errorf("no route to %s", dst)
+	}
+	fn(src, msg)
+	return nil
+}
+
+// loopback is an endpoint-based pqkem.Transport over a netSwitch, with a switchable
+// drop flag.
+type loopback struct {
+	ep   netip.AddrPort
+	sw   *netSwitch
+	drop atomic.Bool
+}
+
+func (l *loopback) Send(dst netip.AddrPort, msg []byte) error {
+	if l.drop.Load() {
+		return nil
+	}
+	return l.sw.deliver(dst, l.ep, append([]byte(nil), msg...))
+}
+
+func (l *loopback) LocalPort() int                             { return int(l.ep.Port()) }
+func (l *loopback) Run(onInbound func(netip.AddrPort, []byte)) { l.sw.register(l.ep, onInbound) }
+func (l *loopback) Close() error                               { return nil }
+
+type fakeWG struct {
+	mu     sync.Mutex
+	psks   map[RemoteID]PSK
+	failed []RemoteID
+}
+
+func newFakeWG() *fakeWG { return &fakeWG{psks: map[RemoteID]PSK{}} }
+
+func (f *fakeWG) OnNewPSKReady(remoteID RemoteID, psk PSK) error {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.psks[remoteID] = psk
+	return nil
+}
+
+func (f *fakeWG) OnRekeyFailed(remoteID RemoteID) error {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.failed = append(f.failed, remoteID)
+	return nil
+}
+
+func (f *fakeWG) psk(peer RemoteID) PSK {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	return f.psks[peer]
+}
+
+var (
+	epA = netip.MustParseAddrPort("100.64.0.1:51833")
+	epB = netip.MustParseAddrPort("100.64.0.2:51833")
+)
+
+// pair builds two wired managers (B is the initiator, "bbbb" > "aaaa") sharing a
+// netSwitch, with each peer's data-path endpoint registered. lbB is B's loopback
+// (for toggling drop).
+func pair(t *testing.T) (dA, dB *Manager, wgA, wgB *fakeWG, lbB *loopback) {
+	t.Helper()
+	sw := newSwitch()
+	wgA = newFakeWG()
+	wgB = newFakeWG()
+	dA = NewManager("aaaa", wgA, nil)
+	dB = NewManager("bbbb", wgB, nil)
+	dA.Start(&loopback{ep: epA, sw: sw})
+	lbB = &loopback{ep: epB, sw: sw}
+	dB.Start(lbB)
+	dA.AddPeer("bbbb", epB)
+	dB.AddPeer("aaaa", epA)
+	return dA, dB, wgA, wgB, lbB
+}
+
+// bootstrap runs the signalling offer/answer (the test plays the host carrying bytes).
+func bootstrap(t *testing.T, dA, dB *Manager) {
+	t.Helper()
+	offer, err := dB.SignalOffer("aaaa")
+	require.NoError(t, err)
+	require.NotNil(t, offer)
+	answer, err := dA.SignalOnOffer("bbbb", offer)
+	require.NoError(t, err)
+	require.NotNil(t, answer)
+	require.NoError(t, dB.SignalOnAnswer("aaaa", answer))
+}
+
+func TestManager_BootstrapDerivesSamePSK(t *testing.T) {
+	dA, dB, wgA, wgB, _ := pair(t)
+	defer dA.Stop()
+	defer dB.Stop()
+
+	bootstrap(t, dA, dB)
+
+	pskA := wgA.psk("bbbb")
+	pskB := wgB.psk("aaaa")
+	require.NotEqual(t, PSK{}, pskA)
+	require.Equal(t, pskB, pskA, "both sides derive the same PSK from the bootstrap exchange")
+}
+
+func TestManager_ChainRotatesAndAcks(t *testing.T) {
+	dA, dB, wgA, wgB, _ := pair(t)
+	defer dA.Stop()
+	defer dB.Stop()
+
+	bootstrap(t, dA, dB)
+	psk1 := wgB.psk("aaaa")
+
+	// Data path up: B (initiator) chains the next offer over the data path, which
+	// rotates both to a fresh PSK and acknowledges A.
+	dA.OnDataPathRekeyed("bbbb")
+	dB.OnDataPathRekeyed("aaaa")
+
+	psk2A := wgA.psk("bbbb")
+	psk2B := wgB.psk("aaaa")
+	require.Equal(t, psk2B, psk2A, "both sides converge on the rotated PSK")
+	require.NotEqual(t, psk1, psk2B, "the chain rotated to a new PSK")
+}
+
+func TestManager_NonInitiatorReturnsNoOffer(t *testing.T) {
+	dA := NewManager("aaaa", newFakeWG(), nil)
+	defer dA.Stop()
+
+	offer, err := dA.SignalOffer("bbbb") // not the initiator vs "bbbb"
+	require.NoError(t, err)
+	require.Nil(t, offer)
+}
+
+func TestManager_StopIsIdempotent(t *testing.T) {
+	dA := NewManager("aaaa", newFakeWG(), nil)
+	dA.Start(&loopback{ep: epA, sw: newSwitch()})
+	dA.Stop()
+	dA.Stop() // must not panic or hang
+}

client/internal/pqkem/message.go

@@ -0,0 +1,121 @@
+package pqkem
+
+import (
+	"crypto/mlkem"
+	"fmt"
+)
+
+// Wire framing for the PQ-KEM exchange. Messages are self-contained, versioned,
+// transport-agnostic byte blobs: the same bytes ride the signalling channel
+// (initial bootstrap) or a data-tunnel packet (rekey). The library only ever sees
+// opaque []byte at the transport seam.
+//
+// Layout (all messages): [type:1][version:1][exchangeID:16][payload...]
+//
+// There is no confirm message: an exchange is acknowledged by the NEXT offer, which
+// carries the acked exchange's id (see OfferMsg.AckID) and — riding the data path
+// under the freshly adopted key — proves that key works.
+
+const (
+	// ProtocolVersion is bumped on any wire-incompatible change; a peer rejects
+	// messages it does not understand rather than misparsing them.
+	ProtocolVersion uint8 = 1
+
+	// ExchangeIDSize identifies one exchange so answers/acks correlate and stale
+	// messages are dropped.
+	ExchangeIDSize = 16
+
+	headerSize = 1 + 1 + ExchangeIDSize
+)
+
+// MsgType tags the two message kinds of the exchange.
+type MsgType uint8
+
+const (
+	MsgOffer MsgType = iota + 1
+	MsgAnswer
+)
+
+// ExchangeID is the per-exchange correlator. The zero value means "none" (an offer
+// that acknowledges nothing, i.e. the first exchange of a connection).
+type ExchangeID [ExchangeIDSize]byte
+
+// OfferMsg carries the initiator's public material (X25519 pub ‖ ML-KEM encap key)
+// and AckID, the id of the previous exchange this offer acknowledges (zero if none).
+type OfferMsg struct {
+	ExchangeID ExchangeID
+	AckID      ExchangeID
+	// KEMOffer is the raw Initiator.Offer() blob (OfferSize bytes).
+	KEMOffer []byte
+}
+
+// AnswerMsg carries the responder's reply (ML-KEM ciphertext ‖ X25519 pub) for the
+// round identified by ExchangeID.
+type AnswerMsg struct {
+	ExchangeID ExchangeID
+	// KEMAnswer is the raw Respond() answer blob (AnswerSize bytes).
+	KEMAnswer []byte
+}
+
+// Encode serialises the offer with its framed header (payload = AckID ‖ KEMOffer).
+func (m *OfferMsg) Encode() ([]byte, error) {
+	if len(m.KEMOffer) != OfferSize {
+		return nil, fmt.Errorf("offer payload: got %d, want %d", len(m.KEMOffer), OfferSize)
+	}
+	payload := make([]byte, 0, ExchangeIDSize+OfferSize)
+	payload = append(payload, m.AckID[:]...)
+	payload = append(payload, m.KEMOffer...)
+	return frame(MsgOffer, m.ExchangeID, payload), nil
+}
+
+// Encode serialises the answer with its framed header.
+func (m *AnswerMsg) Encode() ([]byte, error) {
+	if len(m.KEMAnswer) != AnswerSize {
+		return nil, fmt.Errorf("answer payload: got %d, want %d", len(m.KEMAnswer), AnswerSize)
+	}
+	return frame(MsgAnswer, m.ExchangeID, m.KEMAnswer), nil
+}
+
+// Decode parses a framed message into one of *OfferMsg / *AnswerMsg.
+func Decode(buf []byte) (MsgType, any, error) {
+	if len(buf) < headerSize {
+		return 0, nil, fmt.Errorf("message too short: %d bytes", len(buf))
+	}
+	typ := MsgType(buf[0])
+	if ver := buf[1]; ver != ProtocolVersion {
+		return typ, nil, fmt.Errorf("unsupported protocol version %d (want %d)", ver, ProtocolVersion)
+	}
+
+	var id ExchangeID
+	copy(id[:], buf[2:headerSize])
+	payload := buf[headerSize:]
+
+	switch typ {
+	case MsgOffer:
+		if len(payload) != ExchangeIDSize+OfferSize {
+			return typ, nil, fmt.Errorf("offer payload: got %d, want %d", len(payload), ExchangeIDSize+OfferSize)
+		}
+		var ack ExchangeID
+		copy(ack[:], payload[:ExchangeIDSize])
+		return typ, &OfferMsg{ExchangeID: id, AckID: ack, KEMOffer: payload[ExchangeIDSize:]}, nil
+	case MsgAnswer:
+		if len(payload) != AnswerSize {
+			return typ, nil, fmt.Errorf("answer payload: got %d, want %d", len(payload), AnswerSize)
+		}
+		return typ, &AnswerMsg{ExchangeID: id, KEMAnswer: payload}, nil
+	default:
+		return typ, nil, fmt.Errorf("unknown message type %d", typ)
+	}
+}
+
+func frame(typ MsgType, id ExchangeID, payload []byte) []byte {
+	buf := make([]byte, headerSize+len(payload))
+	buf[0] = byte(typ)
+	buf[1] = ProtocolVersion
+	copy(buf[2:], id[:])
+	copy(buf[headerSize:], payload)
+	return buf
+}
+
+// compile-time assurance the KEM blob sizes referenced here stay in sync with kem.go.
+var _ = [1]struct{}{}[OfferSize-(32+mlkem.EncapsulationKeySize768)]

client/internal/pqkem/message_test.go

@@ -0,0 +1,57 @@
+package pqkem
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestMessageRoundTrip(t *testing.T) {
+	init, err := NewInitiator()
+	require.NoError(t, err)
+	answer, _, err := Respond(init.Offer(), Binding{LocalID: wgB, RemoteID: wgA})
+	require.NoError(t, err)
+
+	id := ExchangeID{1, 2, 3, 4}
+	ack := ExchangeID{9, 9, 9}
+
+	offBytes, err := (&OfferMsg{ExchangeID: id, AckID: ack, KEMOffer: init.Offer()}).Encode()
+	require.NoError(t, err)
+	typ, decoded, err := Decode(offBytes)
+	require.NoError(t, err)
+	require.Equal(t, MsgOffer, typ)
+	require.Equal(t, id, decoded.(*OfferMsg).ExchangeID)
+	require.Equal(t, ack, decoded.(*OfferMsg).AckID)
+	require.Equal(t, init.Offer(), decoded.(*OfferMsg).KEMOffer)
+
+	ansBytes, err := (&AnswerMsg{ExchangeID: id, KEMAnswer: answer}).Encode()
+	require.NoError(t, err)
+	typ, decoded, err = Decode(ansBytes)
+	require.NoError(t, err)
+	require.Equal(t, MsgAnswer, typ)
+	require.Equal(t, answer, decoded.(*AnswerMsg).KEMAnswer)
+}
+
+func TestDecodeRejects(t *testing.T) {
+	// too short
+	_, _, err := Decode([]byte{1, 1})
+	require.Error(t, err)
+
+	// wrong version
+	bad := make([]byte, headerSize+ExchangeIDSize+OfferSize)
+	bad[0] = byte(MsgOffer)
+	bad[1] = ProtocolVersion + 1
+	_, _, err = Decode(bad)
+	require.Error(t, err)
+
+	// unknown type
+	bad2 := make([]byte, headerSize)
+	bad2[0] = 99
+	bad2[1] = ProtocolVersion
+	_, _, err = Decode(bad2)
+	require.Error(t, err)
+
+	// offer with wrong payload size
+	_, err = (&OfferMsg{KEMOffer: []byte{1, 2, 3}}).Encode()
+	require.Error(t, err)
+}

client/internal/pqkem_adapter.go

@@ -0,0 +1,102 @@
+package internal
+
+import (
+	"net/netip"
+
+	log "github.com/sirupsen/logrus"
+	"golang.zx2c4.com/wireguard/wgctrl/wgtypes"
+
+	"github.com/netbirdio/netbird/client/internal/pqkem"
+)
+
+// pqPresharedKeySetter is the subset of the WireGuard interface the ML-KEM callback
+// needs: programming a peer's preshared key. *iface.WGIface satisfies it.
+type pqPresharedKeySetter interface {
+	SetPresharedKey(peerKey string, psk wgtypes.Key, updateOnly bool) error
+}
+
+// pqCallbackHandler programs the derived PQ PSK onto the WireGuard peer. It is the
+// engine-side implementation of pqkem.CallbackHandler.
+type pqCallbackHandler struct {
+	wg pqPresharedKeySetter
+}
+
+// OnNewPSKReady programs the freshly derived PSK for the peer (updateOnly: a no-op
+// if the peer is not present, mirroring Rosenpass). remoteID is the peer's WG pubkey.
+func (h pqCallbackHandler) OnNewPSKReady(remoteID pqkem.RemoteID, psk pqkem.PSK) error {
+	// updateOnly: applies to an already-configured peer (rotation). At bootstrap the
+	// peer is not configured yet, so this is a no-op there and the PSK is instead
+	// pulled at peer-config time (pqHandshaker.PSK / conn.presharedKey).
+	log.Debugf("pqkem: programming PSK for peer %s", remoteID)
+	return h.wg.SetPresharedKey(string(remoteID), wgtypes.Key(psk), true)
+}
+
+// OnRekeyFailed reports a failed PQ (re)key convergence.
+// TODO(NET-1406): tear the peer connection down / trigger ICE reconnect.
+func (h pqCallbackHandler) OnRekeyFailed(remoteID pqkem.RemoteID) error {
+	log.Warnf("pqkem: post-quantum rekey failed for peer %s", remoteID)
+	return nil
+}
+
+// pqHandshaker adapts the pqkem manager to peer.PQHandshaker (string peer keys),
+// wiring the host's signalling offers/answers to the KEM exchange.
+type pqHandshaker struct {
+	mgr *pqkem.Manager
+}
+
+func (p pqHandshaker) OfferPayload(remoteKey string) ([]byte, int) {
+	payload, err := p.mgr.SignalOffer(pqkem.RemoteID(remoteKey))
+	if err != nil {
+		log.Warnf("pqkem: build offer for %s: %v", remoteKey, err)
+	}
+	return payload, p.mgr.LocalPort()
+}
+
+func (p pqHandshaker) AnswerPayload(remoteKey string, recvOffer []byte) ([]byte, int) {
+	if len(recvOffer) == 0 {
+		return nil, p.mgr.LocalPort()
+	}
+	payload, err := p.mgr.SignalOnOffer(pqkem.RemoteID(remoteKey), recvOffer)
+	if err != nil {
+		log.Warnf("pqkem: build answer for %s: %v", remoteKey, err)
+	}
+	return payload, p.mgr.LocalPort()
+}
+
+func (p pqHandshaker) OnAnswer(remoteKey string, recvAnswer []byte) {
+	if len(recvAnswer) == 0 {
+		return
+	}
+	if err := p.mgr.SignalOnAnswer(pqkem.RemoteID(remoteKey), recvAnswer); err != nil {
+		log.Warnf("pqkem: process answer from %s: %v", remoteKey, err)
+	}
+}
+
+// PSK exposes the peer's derived PSK for the conn to program at WG peer-config time.
+func (p pqHandshaker) PSK(remoteKey string) (wgtypes.Key, bool) {
+	psk, ok := p.mgr.PSK(pqkem.RemoteID(remoteKey))
+	if !ok {
+		return wgtypes.Key{}, false
+	}
+	return wgtypes.Key(psk), true
+}
+
+// SetRemoteAddr registers the peer's data-path endpoint (overlay IP + pq UDP port)
+// learned from signalling. Sends only ever fire once the tunnel is up (clocked by
+// OnDataPathRekeyed), so registering here is safe even before connection-up.
+func (p pqHandshaker) SetRemoteAddr(remoteKey string, addr netip.AddrPort) {
+	if !addr.IsValid() || addr.Port() == 0 {
+		return
+	}
+	p.mgr.AddPeer(pqkem.RemoteID(remoteKey), addr)
+}
+
+// OnDataPathRekeyed clocks the next chained PSK rotation on a fresh WG handshake.
+func (p pqHandshaker) OnDataPathRekeyed(remoteKey string) {
+	p.mgr.OnDataPathRekeyed(pqkem.RemoteID(remoteKey))
+}
+
+// OnDataPathDown signals the peer's tunnel went down.
+func (p pqHandshaker) OnDataPathDown(remoteKey string) {
+	p.mgr.OnDataPathDown(pqkem.RemoteID(remoteKey))
+}

client/internal/pqkem_transport.go

@@ -0,0 +1,72 @@
+package internal
+
+import (
+	"fmt"
+	"net"
+	"net/netip"
+
+	log "github.com/sirupsen/logrus"
+)
+
+// DefaultPort is the preferred UDP port for the ML-KEM data-path service, bound on
+// the WG overlay IP. Since each client owns a distinct overlay IP, this port is
+// almost always free, so it need not be announced (peers assume it). A peer only
+// announces Body.mlkemPort when a collision forced it onto a different port.
+const DefaultPort = 51833
+
+// pqTransport is the ML-KEM data-path transport: a dumb UDP socket bound on the WG
+// overlay IP. It implements pqkem.Transport — the manager owns the remoteID<->endpoint
+// routing and drives this socket's lifecycle (Run / Close).
+type pqTransport struct {
+	conn *net.UDPConn
+	port int
+}
+
+// newPQTransport binds a UDP socket on the WG overlay IP, preferring DefaultPort and
+// falling back to an OS-assigned ephemeral port if it is in use. Call it after the WG
+// interface is up so the overlay IP is assigned; when the bound port is not
+// DefaultPort it must be announced to peers via Body.mlkemPort.
+func newPQTransport(overlayIP netip.Addr) (*pqTransport, error) {
+	if !overlayIP.IsValid() {
+		return nil, fmt.Errorf("invalid overlay IP for pqkem transport")
+	}
+	ip := net.IP(overlayIP.AsSlice())
+	conn, err := net.ListenUDP("udp4", &net.UDPAddr{IP: ip, Port: DefaultPort})
+	if err != nil {
+		log.Debugf("pqkem: default port %d unavailable on %s (%v), using an ephemeral port", DefaultPort, overlayIP, err)
+		conn, err = net.ListenUDP("udp4", &net.UDPAddr{IP: ip, Port: 0})
+		if err != nil {
+			return nil, fmt.Errorf("bind pqkem udp on overlay %s: %w", overlayIP, err)
+		}
+	}
+	return &pqTransport{conn: conn, port: conn.LocalAddr().(*net.UDPAddr).Port}, nil
+}
+
+// Send implements pqkem.Transport.
+func (t *pqTransport) Send(endpoint netip.AddrPort, msg []byte) error {
+	_, err := t.conn.WriteToUDPAddrPort(msg, endpoint)
+	return err
+}
+
+// LocalPort implements pqkem.Transport.
+func (t *pqTransport) LocalPort() int { return t.port }
+
+// Run implements pqkem.Transport: the receive loop, delivering each datagram as
+// (source endpoint, msg). Exits when the socket is closed.
+func (t *pqTransport) Run(onInbound func(src netip.AddrPort, msg []byte)) {
+	go func() {
+		buf := make([]byte, 2048)
+		for {
+			n, src, err := t.conn.ReadFromUDPAddrPort(buf)
+			if err != nil {
+				return
+			}
+			msg := make([]byte, n)
+			copy(msg, buf[:n])
+			onInbound(src, msg)
+		}
+	}()
+}
+
+// Close implements pqkem.Transport.
+func (t *pqTransport) Close() error { return t.conn.Close() }

shared/signal/client/client.go

@@ -52,6 +52,11 @@ type CredentialPayload struct {
 	Credential      *Credential
 	RosenpassPubKey []byte
 	RosenpassAddr   string
+	// MlkemPayload is the opaque post-quantum KEM handshake message riding this
+	// OFFER/ANSWER (see Body.mlkemPayload). Nil when not running the PQ exchange.
+	MlkemPayload []byte
+	// MlkemPort is the sender's ML-KEM PQ service UDP port (0 when not running).
+	MlkemPort       int
 	RelaySrvAddress string
 	RelaySrvIP      netip.Addr
 	SessionID       []byte
@@ -89,6 +94,13 @@ func MarshalCredential(myKey wgtypes.Key, remoteKey string, p CredentialPayload)
 	if p.RelaySrvIP.IsValid() {
 		body.RelayServerIP = p.RelaySrvIP.Unmap().AsSlice()
 	}
+	if len(p.MlkemPayload) > 0 {
+		body.MlkemPayload = p.MlkemPayload
+	}
+	if p.MlkemPort > 0 {
+		port := uint32(p.MlkemPort)
+		body.MlkemPort = &port
+	}
 	return &proto.Message{
 		Key:       myKey.PublicKey().String(),
 		RemoteKey: remoteKey,

shared/signal/proto/signalexchange.pb.go

@@ -239,6 +239,16 @@ type Body struct {
 	// fallback dial target when DNS resolution of relayServerAddress fails.
 	// SNI/TLS verification still uses relayServerAddress.
 	RelayServerIP []byte `protobuf:"bytes,11,opt,name=relayServerIP,proto3,oneof" json:"relayServerIP,omitempty"`
+	// mlkemPayload carries a post-quantum X25519MLKEM768 handshake message that
+	// seeds the WireGuard PSK, riding this Body's OFFER/ANSWER: on an OFFER it is
+	// the KEM offer, on an ANSWER the KEM answer. It is opaque to signal — the
+	// pqkem library frames and parses it. Absent when the sender does not run the
+	// ML-KEM PQ exchange; unknown to older clients, which ignore it.
+	MlkemPayload []byte `protobuf:"bytes,12,opt,name=mlkemPayload,proto3,oneof" json:"mlkemPayload,omitempty"`
+	// mlkemPort is the UDP port of the sender's ML-KEM PQ service, bound on its
+	// WireGuard overlay IP. Peers send subsequent rekey messages there over the
+	// data path. Zero/absent when the ML-KEM PQ exchange is not running.
+	MlkemPort *uint32 `protobuf:"varint,13,opt,name=mlkemPort,proto3,oneof" json:"mlkemPort,omitempty"`
 }
 
 func (x *Body) Reset() {
@@ -343,6 +353,20 @@ func (x *Body) GetRelayServerIP() []byte {
 	return nil
 }
 
+func (x *Body) GetMlkemPayload() []byte {
+	if x != nil {
+		return x.MlkemPayload
+	}
+	return nil
+}
+
+func (x *Body) GetMlkemPort() uint32 {
+	if x != nil && x.MlkemPort != nil {
+		return *x.MlkemPort
+	}
+	return 0
+}
+
 // Mode indicates a connection mode
 type Mode struct {
 	state         protoimpl.MessageState
@@ -466,7 +490,7 @@ var file_signalexchange_proto_rawDesc = []byte{
 	0x52, 0x09, 0x72, 0x65, 0x6d, 0x6f, 0x74, 0x65, 0x4b, 0x65, 0x79, 0x12, 0x28, 0x0a, 0x04, 0x62,
 	0x6f, 0x64, 0x79, 0x18, 0x04, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x14, 0x2e, 0x73, 0x69, 0x67, 0x6e,
 	0x61, 0x6c, 0x65, 0x78, 0x63, 0x68, 0x61, 0x6e, 0x67, 0x65, 0x2e, 0x42, 0x6f, 0x64, 0x79, 0x52,
-	0x04, 0x62, 0x6f, 0x64, 0x79, 0x22, 0xd2, 0x04, 0x0a, 0x04, 0x42, 0x6f, 0x64, 0x79, 0x12, 0x2d,
+	0x04, 0x62, 0x6f, 0x64, 0x79, 0x22, 0xbd, 0x05, 0x0a, 0x04, 0x42, 0x6f, 0x64, 0x79, 0x12, 0x2d,
 	0x0a, 0x04, 0x74, 0x79, 0x70, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0e, 0x32, 0x19, 0x2e, 0x73,
 	0x69, 0x67, 0x6e, 0x61, 0x6c, 0x65, 0x78, 0x63, 0x68, 0x61, 0x6e, 0x67, 0x65, 0x2e, 0x42, 0x6f,
 	0x64, 0x79, 0x2e, 0x54, 0x79, 0x70, 0x65, 0x52, 0x04, 0x74, 0x79, 0x70, 0x65, 0x12, 0x18, 0x0a,
@@ -494,39 +518,46 @@ var file_signalexchange_proto_rawDesc = []byte{
 	0x52, 0x09, 0x73, 0x65, 0x73, 0x73, 0x69, 0x6f, 0x6e, 0x49, 0x64, 0x88, 0x01, 0x01, 0x12, 0x29,
 	0x0a, 0x0d, 0x72, 0x65, 0x6c, 0x61, 0x79, 0x53, 0x65, 0x72, 0x76, 0x65, 0x72, 0x49, 0x50, 0x18,
 	0x0b, 0x20, 0x01, 0x28, 0x0c, 0x48, 0x02, 0x52, 0x0d, 0x72, 0x65, 0x6c, 0x61, 0x79, 0x53, 0x65,
-	0x72, 0x76, 0x65, 0x72, 0x49, 0x50, 0x88, 0x01, 0x01, 0x22, 0x52, 0x0a, 0x04, 0x54, 0x79, 0x70,
-	0x65, 0x12, 0x09, 0x0a, 0x05, 0x4f, 0x46, 0x46, 0x45, 0x52, 0x10, 0x00, 0x12, 0x0a, 0x0a, 0x06,
-	0x41, 0x4e, 0x53, 0x57, 0x45, 0x52, 0x10, 0x01, 0x12, 0x0d, 0x0a, 0x09, 0x43, 0x41, 0x4e, 0x44,
-	0x49, 0x44, 0x41, 0x54, 0x45, 0x10, 0x02, 0x12, 0x08, 0x0a, 0x04, 0x4d, 0x4f, 0x44, 0x45, 0x10,
-	0x04, 0x12, 0x0b, 0x0a, 0x07, 0x47, 0x4f, 0x5f, 0x49, 0x44, 0x4c, 0x45, 0x10, 0x05, 0x12, 0x0d,
-	0x0a, 0x09, 0x48, 0x45, 0x41, 0x52, 0x54, 0x42, 0x45, 0x41, 0x54, 0x10, 0x06, 0x42, 0x15, 0x0a,
-	0x13, 0x5f, 0x72, 0x65, 0x6c, 0x61, 0x79, 0x53, 0x65, 0x72, 0x76, 0x65, 0x72, 0x41, 0x64, 0x64,
-	0x72, 0x65, 0x73, 0x73, 0x42, 0x0c, 0x0a, 0x0a, 0x5f, 0x73, 0x65, 0x73, 0x73, 0x69, 0x6f, 0x6e,
-	0x49, 0x64, 0x42, 0x10, 0x0a, 0x0e, 0x5f, 0x72, 0x65, 0x6c, 0x61, 0x79, 0x53, 0x65, 0x72, 0x76,
-	0x65, 0x72, 0x49, 0x50, 0x4a, 0x04, 0x08, 0x09, 0x10, 0x0a, 0x22, 0x2e, 0x0a, 0x04, 0x4d, 0x6f,
-	0x64, 0x65, 0x12, 0x1b, 0x0a, 0x06, 0x64, 0x69, 0x72, 0x65, 0x63, 0x74, 0x18, 0x01, 0x20, 0x01,
-	0x28, 0x08, 0x48, 0x00, 0x52, 0x06, 0x64, 0x69, 0x72, 0x65, 0x63, 0x74, 0x88, 0x01, 0x01, 0x42,
-	0x09, 0x0a, 0x07, 0x5f, 0x64, 0x69, 0x72, 0x65, 0x63, 0x74, 0x22, 0x6d, 0x0a, 0x0f, 0x52, 0x6f,
-	0x73, 0x65, 0x6e, 0x70, 0x61, 0x73, 0x73, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x12, 0x28, 0x0a,
-	0x0f, 0x72, 0x6f, 0x73, 0x65, 0x6e, 0x70, 0x61, 0x73, 0x73, 0x50, 0x75, 0x62, 0x4b, 0x65, 0x79,
-	0x18, 0x01, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x0f, 0x72, 0x6f, 0x73, 0x65, 0x6e, 0x70, 0x61, 0x73,
-	0x73, 0x50, 0x75, 0x62, 0x4b, 0x65, 0x79, 0x12, 0x30, 0x0a, 0x13, 0x72, 0x6f, 0x73, 0x65, 0x6e,
-	0x70, 0x61, 0x73, 0x73, 0x53, 0x65, 0x72, 0x76, 0x65, 0x72, 0x41, 0x64, 0x64, 0x72, 0x18, 0x02,
-	0x20, 0x01, 0x28, 0x09, 0x52, 0x13, 0x72, 0x6f, 0x73, 0x65, 0x6e, 0x70, 0x61, 0x73, 0x73, 0x53,
-	0x65, 0x72, 0x76, 0x65, 0x72, 0x41, 0x64, 0x64, 0x72, 0x32, 0xb9, 0x01, 0x0a, 0x0e, 0x53, 0x69,
-	0x67, 0x6e, 0x61, 0x6c, 0x45, 0x78, 0x63, 0x68, 0x61, 0x6e, 0x67, 0x65, 0x12, 0x4c, 0x0a, 0x04,
-	0x53, 0x65, 0x6e, 0x64, 0x12, 0x20, 0x2e, 0x73, 0x69, 0x67, 0x6e, 0x61, 0x6c, 0x65, 0x78, 0x63,
-	0x68, 0x61, 0x6e, 0x67, 0x65, 0x2e, 0x45, 0x6e, 0x63, 0x72, 0x79, 0x70, 0x74, 0x65, 0x64, 0x4d,
-	0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x1a, 0x20, 0x2e, 0x73, 0x69, 0x67, 0x6e, 0x61, 0x6c, 0x65,
+	0x72, 0x76, 0x65, 0x72, 0x49, 0x50, 0x88, 0x01, 0x01, 0x12, 0x27, 0x0a, 0x0c, 0x6d, 0x6c, 0x6b,
+	0x65, 0x6d, 0x50, 0x61, 0x79, 0x6c, 0x6f, 0x61, 0x64, 0x18, 0x0c, 0x20, 0x01, 0x28, 0x0c, 0x48,
+	0x03, 0x52, 0x0c, 0x6d, 0x6c, 0x6b, 0x65, 0x6d, 0x50, 0x61, 0x79, 0x6c, 0x6f, 0x61, 0x64, 0x88,
+	0x01, 0x01, 0x12, 0x21, 0x0a, 0x09, 0x6d, 0x6c, 0x6b, 0x65, 0x6d, 0x50, 0x6f, 0x72, 0x74, 0x18,
+	0x0d, 0x20, 0x01, 0x28, 0x0d, 0x48, 0x04, 0x52, 0x09, 0x6d, 0x6c, 0x6b, 0x65, 0x6d, 0x50, 0x6f,
+	0x72, 0x74, 0x88, 0x01, 0x01, 0x22, 0x52, 0x0a, 0x04, 0x54, 0x79, 0x70, 0x65, 0x12, 0x09, 0x0a,
+	0x05, 0x4f, 0x46, 0x46, 0x45, 0x52, 0x10, 0x00, 0x12, 0x0a, 0x0a, 0x06, 0x41, 0x4e, 0x53, 0x57,
+	0x45, 0x52, 0x10, 0x01, 0x12, 0x0d, 0x0a, 0x09, 0x43, 0x41, 0x4e, 0x44, 0x49, 0x44, 0x41, 0x54,
+	0x45, 0x10, 0x02, 0x12, 0x08, 0x0a, 0x04, 0x4d, 0x4f, 0x44, 0x45, 0x10, 0x04, 0x12, 0x0b, 0x0a,
+	0x07, 0x47, 0x4f, 0x5f, 0x49, 0x44, 0x4c, 0x45, 0x10, 0x05, 0x12, 0x0d, 0x0a, 0x09, 0x48, 0x45,
+	0x41, 0x52, 0x54, 0x42, 0x45, 0x41, 0x54, 0x10, 0x06, 0x42, 0x15, 0x0a, 0x13, 0x5f, 0x72, 0x65,
+	0x6c, 0x61, 0x79, 0x53, 0x65, 0x72, 0x76, 0x65, 0x72, 0x41, 0x64, 0x64, 0x72, 0x65, 0x73, 0x73,
+	0x42, 0x0c, 0x0a, 0x0a, 0x5f, 0x73, 0x65, 0x73, 0x73, 0x69, 0x6f, 0x6e, 0x49, 0x64, 0x42, 0x10,
+	0x0a, 0x0e, 0x5f, 0x72, 0x65, 0x6c, 0x61, 0x79, 0x53, 0x65, 0x72, 0x76, 0x65, 0x72, 0x49, 0x50,
+	0x42, 0x0f, 0x0a, 0x0d, 0x5f, 0x6d, 0x6c, 0x6b, 0x65, 0x6d, 0x50, 0x61, 0x79, 0x6c, 0x6f, 0x61,
+	0x64, 0x42, 0x0c, 0x0a, 0x0a, 0x5f, 0x6d, 0x6c, 0x6b, 0x65, 0x6d, 0x50, 0x6f, 0x72, 0x74, 0x4a,
+	0x04, 0x08, 0x09, 0x10, 0x0a, 0x22, 0x2e, 0x0a, 0x04, 0x4d, 0x6f, 0x64, 0x65, 0x12, 0x1b, 0x0a,
+	0x06, 0x64, 0x69, 0x72, 0x65, 0x63, 0x74, 0x18, 0x01, 0x20, 0x01, 0x28, 0x08, 0x48, 0x00, 0x52,
+	0x06, 0x64, 0x69, 0x72, 0x65, 0x63, 0x74, 0x88, 0x01, 0x01, 0x42, 0x09, 0x0a, 0x07, 0x5f, 0x64,
+	0x69, 0x72, 0x65, 0x63, 0x74, 0x22, 0x6d, 0x0a, 0x0f, 0x52, 0x6f, 0x73, 0x65, 0x6e, 0x70, 0x61,
+	0x73, 0x73, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x12, 0x28, 0x0a, 0x0f, 0x72, 0x6f, 0x73, 0x65,
+	0x6e, 0x70, 0x61, 0x73, 0x73, 0x50, 0x75, 0x62, 0x4b, 0x65, 0x79, 0x18, 0x01, 0x20, 0x01, 0x28,
+	0x0c, 0x52, 0x0f, 0x72, 0x6f, 0x73, 0x65, 0x6e, 0x70, 0x61, 0x73, 0x73, 0x50, 0x75, 0x62, 0x4b,
+	0x65, 0x79, 0x12, 0x30, 0x0a, 0x13, 0x72, 0x6f, 0x73, 0x65, 0x6e, 0x70, 0x61, 0x73, 0x73, 0x53,
+	0x65, 0x72, 0x76, 0x65, 0x72, 0x41, 0x64, 0x64, 0x72, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52,
+	0x13, 0x72, 0x6f, 0x73, 0x65, 0x6e, 0x70, 0x61, 0x73, 0x73, 0x53, 0x65, 0x72, 0x76, 0x65, 0x72,
+	0x41, 0x64, 0x64, 0x72, 0x32, 0xb9, 0x01, 0x0a, 0x0e, 0x53, 0x69, 0x67, 0x6e, 0x61, 0x6c, 0x45,
+	0x78, 0x63, 0x68, 0x61, 0x6e, 0x67, 0x65, 0x12, 0x4c, 0x0a, 0x04, 0x53, 0x65, 0x6e, 0x64, 0x12,
+	0x20, 0x2e, 0x73, 0x69, 0x67, 0x6e, 0x61, 0x6c, 0x65, 0x78, 0x63, 0x68, 0x61, 0x6e, 0x67, 0x65,
+	0x2e, 0x45, 0x6e, 0x63, 0x72, 0x79, 0x70, 0x74, 0x65, 0x64, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67,
+	0x65, 0x1a, 0x20, 0x2e, 0x73, 0x69, 0x67, 0x6e, 0x61, 0x6c, 0x65, 0x78, 0x63, 0x68, 0x61, 0x6e,
+	0x67, 0x65, 0x2e, 0x45, 0x6e, 0x63, 0x72, 0x79, 0x70, 0x74, 0x65, 0x64, 0x4d, 0x65, 0x73, 0x73,
+	0x61, 0x67, 0x65, 0x22, 0x00, 0x12, 0x59, 0x0a, 0x0d, 0x43, 0x6f, 0x6e, 0x6e, 0x65, 0x63, 0x74,
+	0x53, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x12, 0x20, 0x2e, 0x73, 0x69, 0x67, 0x6e, 0x61, 0x6c, 0x65,
 	0x78, 0x63, 0x68, 0x61, 0x6e, 0x67, 0x65, 0x2e, 0x45, 0x6e, 0x63, 0x72, 0x79, 0x70, 0x74, 0x65,
-	0x64, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x22, 0x00, 0x12, 0x59, 0x0a, 0x0d, 0x43, 0x6f,
-	0x6e, 0x6e, 0x65, 0x63, 0x74, 0x53, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x12, 0x20, 0x2e, 0x73, 0x69,
-	0x67, 0x6e, 0x61, 0x6c, 0x65, 0x78, 0x63, 0x68, 0x61, 0x6e, 0x67, 0x65, 0x2e, 0x45, 0x6e, 0x63,
-	0x72, 0x79, 0x70, 0x74, 0x65, 0x64, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x1a, 0x20, 0x2e,
-	0x73, 0x69, 0x67, 0x6e, 0x61, 0x6c, 0x65, 0x78, 0x63, 0x68, 0x61, 0x6e, 0x67, 0x65, 0x2e, 0x45,
-	0x6e, 0x63, 0x72, 0x79, 0x70, 0x74, 0x65, 0x64, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x22,
-	0x00, 0x28, 0x01, 0x30, 0x01, 0x42, 0x08, 0x5a, 0x06, 0x2f, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62,
-	0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
+	0x64, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x1a, 0x20, 0x2e, 0x73, 0x69, 0x67, 0x6e, 0x61,
+	0x6c, 0x65, 0x78, 0x63, 0x68, 0x61, 0x6e, 0x67, 0x65, 0x2e, 0x45, 0x6e, 0x63, 0x72, 0x79, 0x70,
+	0x74, 0x65, 0x64, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x22, 0x00, 0x28, 0x01, 0x30, 0x01,
+	0x42, 0x08, 0x5a, 0x06, 0x2f, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74,
+	0x6f, 0x33,
 }
 
 var (

shared/signal/proto/signalexchange.proto

@@ -75,6 +75,18 @@ message Body {
   // fallback dial target when DNS resolution of relayServerAddress fails.
   // SNI/TLS verification still uses relayServerAddress.
   optional bytes relayServerIP = 11;
+
+  // mlkemPayload carries a post-quantum X25519MLKEM768 handshake message that
+  // seeds the WireGuard PSK, riding this Body's OFFER/ANSWER: on an OFFER it is
+  // the KEM offer, on an ANSWER the KEM answer. It is opaque to signal — the
+  // pqkem library frames and parses it. Absent when the sender does not run the
+  // ML-KEM PQ exchange; unknown to older clients, which ignore it.
+  optional bytes mlkemPayload = 12;
+
+  // mlkemPort is the UDP port of the sender's ML-KEM PQ service, bound on its
+  // WireGuard overlay IP. Peers send subsequent rekey messages there over the
+  // data path. Zero/absent when the ML-KEM PQ exchange is not running.
+  optional uint32 mlkemPort = 13;
 }
 
 // Mode indicates a connection mode