From e88fc055759b4aa47df58d3cbdedc82ec4743b28 Mon Sep 17 00:00:00 2001 From: riccardom Date: Thu, 23 Jul 2026 16:52:26 +0200 Subject: [PATCH] Leave signal offer/answer as a pull/push operation not as an actual transport --- client/internal/pqkem/convergence.go | 105 ++++++---- client/internal/pqkem/convergence_test.go | 40 ++-- client/internal/pqkem/manager.go | 236 +++++++++++++--------- client/internal/pqkem/manager_test.go | 49 ++--- 4 files changed, 256 insertions(+), 174 deletions(-) diff --git a/client/internal/pqkem/convergence.go b/client/internal/pqkem/convergence.go index 3291e322f..26c064e0a 100644 --- a/client/internal/pqkem/convergence.go +++ b/client/internal/pqkem/convergence.go @@ -5,21 +5,57 @@ import ( "time" ) -// handleOffer (responder) derives the PSK, commits it optimistically, and sends the -// answer over the current channel. A duplicate offer (same exchangeID) resends the -// cached answer without re-deriving. The responder is otherwise reactive: no -// retransmit loop. -func (m *Manager) handleOffer(remoteID string, o *OfferMsg) error { +// startExchange creates a fresh initiator exchange and returns the framed offer for +// the caller to send (pushed over the data path for a rekey, or handed to the host +// for the signalling channel when viaSignal is set). Any previous in-flight exchange +// for the peer is cancelled. +func (m *Manager) startExchange(remoteID string, viaSignal bool) ([]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, 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) derives the PSK, commits it optimistically, and returns +// the framed answer for the caller to send. A duplicate offer (same exchangeID) +// returns the cached answer without re-deriving; a still-reserved slot returns nil. +func (m *Manager) processOffer(remoteID string, o *OfferMsg) ([]byte, error) { 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 { - // another goroutine reserved this exchange and is deriving the answer; - // dropping avoids a second (randomized -> divergent) derivation. - return nil + return nil, nil } - return m.send(remoteID, last) + return last, nil } // Reserve the slot so a concurrent duplicate offer bails. m.exchanges[remoteID] = &exchangeCtl{id: o.ExchangeID, state: stateReserved, startedAt: time.Now()} @@ -27,37 +63,36 @@ func (m *Manager) handleOffer(remoteID string, o *OfferMsg) error { answerBytes, psk, err := Respond(o.KEMOffer, m.binding(remoteID)) if err != nil { - return err + return nil, err } raw, err := (&AnswerMsg{ExchangeID: o.ExchangeID, KEMAnswer: answerBytes}).Encode() if err != nil { - return err + return nil, err } m.mu.Lock() ex := m.exchanges[remoteID] if ex == nil || ex.id != o.ExchangeID { m.mu.Unlock() - return nil + return nil, nil } ex.state = stateAwaitingConfirm ex.lastSent = raw ex.pendingPSK = psk m.mu.Unlock() - // Commit optimistically so our data path can rekey to the new PSK; a lost answer - // simply means the data path won't come up (initial) or the previous PSK keeps - // working until it does (rekey grace) — both self-heal via retry. + // Commit optimistically so our data path can rekey to the new PSK. if err := m.cbHandler.OnNewPSKReady(remoteID, psk); err != nil { - return err + return nil, err } - return m.send(remoteID, raw) + return raw, nil } -// handleAnswer (initiator) derives and commits the PSK, then waits for -// OnDataPathRekeyed to send the confirm. Only valid in stateAwaitingAnswer; -// advancing the state under the lock makes a concurrent/duplicate answer bail. -func (m *Manager) handleAnswer(remoteID string, a *AnswerMsg) error { +// processAnswer (initiator) derives and commits the PSK, then parks in +// stateAwaitingRekey; the confirm is sent later, over the data path, from +// OnDataPathRekeyed. Only valid in stateAwaitingAnswer; advancing the state under the +// lock makes a concurrent/duplicate answer bail. +func (m *Manager) processAnswer(remoteID string, a *AnswerMsg) error { m.mu.Lock() ex := m.exchanges[remoteID] if ex == nil || ex.id != a.ExchangeID || ex.state != stateAwaitingAnswer { @@ -73,16 +108,14 @@ func (m *Manager) handleAnswer(remoteID string, a *AnswerMsg) error { if err != nil { return err } - // Commit now; the confirm is deferred until OnDataPathRekeyed so it rides the - // data path under the new key. return m.cbHandler.OnNewPSKReady(remoteID, psk) } -// handleConfirm (responder) records convergence. The PSK was already committed in -// handleOffer; a confirm arriving over the data path with a matching exchangeID +// processConfirm (responder) records convergence. The PSK was already committed in +// processOffer; a confirm arriving over the data path with a matching exchangeID // proves we operate on the new key from this exchange. Stale/duplicate confirms find // the exchange gone and are ignored. -func (m *Manager) handleConfirm(remoteID string, c *ConfirmMsg) error { +func (m *Manager) processConfirm(remoteID string, c *ConfirmMsg) error { m.mu.Lock() ex := m.exchanges[remoteID] if ex == nil || ex.id != c.ExchangeID || ex.state != stateAwaitingConfirm { @@ -97,11 +130,12 @@ func (m *Manager) handleConfirm(remoteID string, c *ConfirmMsg) error { return nil } -// initiatorLoop retransmits the initiator's outstanding message and enforces the -// convergence deadline, keyed off the exchange state: resend the offer while -// awaiting the answer; wait (counting toward the deadline) while awaiting the data -// path rekey; resend the confirm a few best-effort times once sent. Exhausting the -// deadline before reaching the confirm phase is a failure. +// initiatorLoop enforces the convergence deadline and retransmits the initiator's +// outstanding DATA-PATH message: it resends the offer while awaiting the answer only +// when the offer went over the data path (a signalling-bootstrapped offer is +// retransmitted by the host with its own negotiation); it waits, counting toward the +// deadline, while awaiting the data-path rekey; it resends the confirm a few times +// once sent. Exhausting the deadline before the confirm phase is a failure. func (m *Manager) initiatorLoop(ctx context.Context, remoteID string, id ExchangeID) { defer m.wait.Done() t := time.NewTicker(m.retryInterval) @@ -129,11 +163,14 @@ func (m *Manager) initiatorLoop(ctx context.Context, remoteID string, id Exchang m.raiseFailure(remoteID, fail) return } + viaSignal := ex.viaSignal msg := ex.lastSent attempts++ m.mu.Unlock() - if err := m.send(remoteID, msg); err != nil { - m.logger.Warn("pqkem offer retransmit failed", "peer", remoteID, "err", err) + if !viaSignal { + if err := m.pushDataPath(remoteID, msg); err != nil { + m.logger.Warn("pqkem offer retransmit failed", "peer", remoteID, "err", err) + } } case stateAwaitingRekey: @@ -158,7 +195,7 @@ func (m *Manager) initiatorLoop(ctx context.Context, remoteID string, id Exchang msg := ex.lastSent confirmsSent++ m.mu.Unlock() - if err := m.transport.SendDataPath(remoteID, msg); err != nil { + if err := m.pushDataPath(remoteID, msg); err != nil { m.logger.Warn("pqkem confirm retransmit failed", "peer", remoteID, "err", err) } diff --git a/client/internal/pqkem/convergence_test.go b/client/internal/pqkem/convergence_test.go index 3676f8664..1b9e075a5 100644 --- a/client/internal/pqkem/convergence_test.go +++ b/client/internal/pqkem/convergence_test.go @@ -11,24 +11,21 @@ import ( type dropTransport struct{} func (dropTransport) SendDataPath(string, []byte) error { return nil } -func (dropTransport) SendSignal(string, []byte) error { return nil } -// gate is a loopback transport with a switchable drop flag. +// gate is a data-path loopback with a switchable drop flag. When dropping it reports +// success but does not deliver (mimics a lossy/broken tunnel). type gate struct { localID string peer *Manager drop atomic.Bool } -func (g *gate) SendDataPath(remoteID string, msg []byte) error { return g.deliver(msg) } -func (g *gate) SendSignal(remoteID string, msg []byte) error { return g.deliver(msg) } - -func (g *gate) deliver(msg []byte) error { +func (g *gate) SendDataPath(remoteID string, msg []byte) error { if g.drop.Load() { return nil } cp := append([]byte(nil), msg...) - return g.peer.HandleInbound(g.localID, cp) + return g.peer.OnDataPathMessage(g.localID, cp) } func TestManager_InitialTimeoutFailsImmediately(t *testing.T) { @@ -39,9 +36,12 @@ func TestManager_InitialTimeoutFailsImmediately(t *testing.T) { d.AddPeer("aaaa") defer d.Stop() - require.NoError(t, d.initiateRekey("aaaa")) + // 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) - // no answer will ever come -> the initial exchange fails fast. require.Eventually(t, func() bool { wg.mu.Lock() defer wg.mu.Unlock() @@ -66,25 +66,31 @@ func TestManager_RekeyToleratesKFailures(t *testing.T) { defer dA.Stop() defer dB.Stop() - // First exchange succeeds -> peer becomes established (subsequent failures are - // rekeys). Drive the data-path-rekeyed event so the confirm converges A. - require.NoError(t, dB.initiateRekey("aaaa")) + // Establish via a signalling bootstrap + data-path-rekeyed, so B becomes + // established and its data path is up. + offer, err := dB.SignalOffer("aaaa") + require.NoError(t, err) + answer, err := dA.SignalOnOffer("bbbb", offer) + require.NoError(t, err) + require.NoError(t, dB.SignalOnAnswer("aaaa", answer)) dA.OnDataPathRekeyed("bbbb") dB.OnDataPathRekeyed("aaaa") require.NotEqual(t, PSK{}, wgB.psk("aaaa")) - // Now drop B's outbound: rekeys can no longer converge. + // Now drop B's data-path delivery: rekeys can no longer converge. gB.drop.Store(true) - // K-1 failures must NOT raise OnRekeyFailed. + // K-1 data-path rekeys must NOT raise OnRekeyFailed. for i := 0; i < DefaultMaxRekeyFailures-1; i++ { - require.NoError(t, dB.initiateRekey("aaaa")) + _, err := dB.startExchange("aaaa", false) + 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. - require.NoError(t, dB.initiateRekey("aaaa")) + // The K-th failure raises it once. + _, err = dB.startExchange("aaaa", false) + require.NoError(t, err) require.Eventually(t, func() bool { return failedCount(wgB) == 1 }, time.Second, 5*time.Millisecond) } diff --git a/client/internal/pqkem/manager.go b/client/internal/pqkem/manager.go index 0de006a59..675d88956 100644 --- a/client/internal/pqkem/manager.go +++ b/client/internal/pqkem/manager.go @@ -15,34 +15,26 @@ const ( // forcing one. DefaultRekeyInterval = 2 * time.Minute // DefaultRetryInterval is how often the initiator retransmits its outstanding - // message (offer, then confirm) while an exchange is in flight. + // data-path message while an exchange is in flight. DefaultRetryInterval = 2 * time.Second - // DefaultMaxRetries bounds how many ticks an exchange may run before the offer - // or the wait for the data-path rekey is declared failed. The convergence - // deadline is thus derived as MaxRetries * RetryInterval — no separate timer. + // 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 // confirmRetransmits is how many times the initiator best-effort resends the - // confirm over the data path, to cover its loss without a dedicated goroutine. + // confirm over the data path, to cover its loss. confirmRetransmits = 3 ) -// Transport carries exchange messages over the two available channels; the library -// picks which. It is the analogue of go-rosenpass's Conn seam. -// -// Model: a message rides the channel keyed with the CURRENTLY valid key. Before a -// data path exists (initial bootstrap) that is the out-of-band signalling channel; -// once the data path is up (after OnDataPathRekeyed) offers/answers ride it, and the -// confirm ALWAYS rides the data path (so its arrival proves the new key works). +// Transport pushes a message over the peer's data path (e.g. a WireGuard tunnel). +// It is the library's only outbound send: the control-plane (signalling) channel is +// host-driven — the library hands the host offer/answer payloads to piggyback on the +// host's own offer/answer, it never pushes there itself (that would drive the host's +// connection negotiation, which is not the library's to control). type Transport interface { - // SendDataPath sends over the peer's established data path (e.g. a WireGuard - // tunnel). The consumer routes it accordingly. SendDataPath(remoteID string, msg []byte) error - // SendSignal sends over the out-of-band signalling channel, handled outside the - // library by the consumer. - SendSignal(remoteID string, msg []byte) error } // exchangeState is the single source of truth for an exchange's role and phase. @@ -58,9 +50,11 @@ const ( // 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 -// retransmit payload (offer, then confirm). initiator is the ephemeral handle used -// at Finish; pendingPSK is the responder's derived key. Only the initiator runs a -// retransmit loop, so only it sets cancel. +// data-path retransmit payload. initiator is the ephemeral handle used at Finish; +// pendingPSK is the responder's derived key. viaSignal records that the offer was +// handed to the host for the signalling channel (bootstrap), so the loop does not +// retransmit it on the data path (the host retransmits it with its own negotiation). +// Only the initiator runs a retransmit loop, so only it sets cancel. type exchangeCtl struct { id ExchangeID state exchangeState @@ -69,13 +63,13 @@ type exchangeCtl struct { lastSent []byte initiator *Initiator pendingPSK PSK + viaSignal bool } // Manager is the stateful orchestrator — the analogue of go-rosenpass's Server. It -// runs the per-peer rekey timer, drives the X25519MLKEM768 exchange over the -// Transport, and surfaces the derived PSK and convergence to the host via -// CallbackHandler. The cryptography is the pure kem.go primitives; all state lives -// here under one lock. +// runs the per-peer rekey timer, drives the X25519MLKEM768 exchange, and surfaces the +// derived PSK and convergence to the host via CallbackHandler. The cryptography is +// the pure kem.go primitives; all state lives here under one lock. type Manager struct { localID string transport Transport @@ -95,12 +89,15 @@ type Manager struct { exchanges map[string]*exchangeCtl // in-flight exchange per peer established map[string]bool // peer has completed at least one exchange failures map[string]int // consecutive rekey failures per peer - dataPathUp map[string]bool // peer's data path is up (offers/answers may ride it) - wait sync.WaitGroup + // dataSend holds the peer's data-path sender when the data path is up; nil (absent) + // means it is down. Toggled by OnDataPathRekeyed / OnDataPathDown. Its presence is + // the "a data path exists" signal. + dataSend map[string]func(string, []byte) error + 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 zero +// 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 zero // interval falls back to DefaultRekeyInterval; a nil logger to slog.Default(). func NewManager(localID string, t Transport, h CallbackHandler, interval time.Duration, logger *slog.Logger) *Manager { if interval <= 0 { @@ -125,7 +122,7 @@ func NewManager(localID string, t Transport, h CallbackHandler, interval time.Du exchanges: make(map[string]*exchangeCtl), established: make(map[string]bool), failures: make(map[string]int), - dataPathUp: make(map[string]bool), + dataSend: make(map[string]func(string, []byte) error), } } @@ -164,7 +161,7 @@ func (m *Manager) RemovePeer(remoteID string) { } delete(m.established, remoteID) delete(m.failures, remoteID) - delete(m.dataPathUp, remoteID) + delete(m.dataSend, remoteID) m.mu.Unlock() } @@ -178,14 +175,89 @@ func (m *Manager) Stop() { m.mu.Unlock() } -// OnDataPathRekeyed notifies the library that the peer's data path is up and freshly -// keyed with the latest PSK (fired on first establishment AND every rekey — the same -// event). It marks the data path usable and, if we are the initiator waiting to -// confirm, sends the confirm over the data path (its arrival proves to the responder -// that we operate on the new key, correlated by exchangeID). +// ---- Signalling channel (host-driven; rides the host's offer/answer) ---- + +// 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 (e.g. the +// host retransmitting its offer) returns the same offer rather than starting a new +// exchange. +func (m *Manager) SignalOffer(remoteID string) ([]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() + return m.startExchange(remoteID, true) +} + +// 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 string, 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 confirm rides the data path after OnDataPathRekeyed. +func (m *Manager) SignalOnAnswer(remoteID string, 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 (library-driven push) ---- + +// OnDataPathMessage feeds a KEM message received over the data path (tunnel) and +// pushes any reply back over the data path. +func (m *Manager) OnDataPathMessage(remoteID string, 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)) + case MsgConfirm: + return m.processConfirm(remoteID, msg.(*ConfirmMsg)) + 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 — the same event). It +// marks the data path usable and, if we are the initiator waiting to confirm, sends +// the confirm over the data path (its arrival proves to the responder that we operate +// on the new key, correlated by exchangeID). func (m *Manager) OnDataPathRekeyed(remoteID string) { m.mu.Lock() - m.dataPathUp[remoteID] = true + m.dataSend[remoteID] = m.transport.SendDataPath ex := m.exchanges[remoteID] if ex == nil || ex.state != stateAwaitingRekey { m.mu.Unlock() @@ -204,70 +276,21 @@ func (m *Manager) OnDataPathRekeyed(remoteID string) { _ = time.Since(ex.startedAt) // convergence latency (metrics hook, later step) m.mu.Unlock() - if err := m.transport.SendDataPath(remoteID, confirm); err != nil { + if err := m.pushDataPath(remoteID, confirm); err != nil { m.logger.Warn("pqkem send confirm failed", "peer", remoteID, "err", err) } } -// HandleInbound decodes an incoming message and drives the exchange. -func (m *Manager) HandleInbound(remoteID string, raw []byte) error { - typ, msg, err := Decode(raw) - if err != nil { - return fmt.Errorf("decode from %s: %w", remoteID, err) - } - switch typ { - case MsgOffer: - return m.handleOffer(remoteID, msg.(*OfferMsg)) - case MsgAnswer: - return m.handleAnswer(remoteID, msg.(*AnswerMsg)) - case MsgConfirm: - return m.handleConfirm(remoteID, msg.(*ConfirmMsg)) - default: - return fmt.Errorf("unhandled message type %d from %s", typ, remoteID) - } -} - -// initiateRekey starts a fresh exchange when the local peer is the initiator. The -// offer rides the data path if it is up (rekey) or the signalling channel otherwise -// (initial bootstrap). -func (m *Manager) initiateRekey(remoteID string) error { - if !m.IsInitiator(remoteID) { - return nil - } - init, err := NewInitiator() - if err != nil { - return err - } - id, err := newExchangeID() - if err != nil { - return err - } - raw, err := (&OfferMsg{ExchangeID: id, KEMOffer: init.Offer()}).Encode() - if err != nil { - return err - } - - ctx, cancel := context.WithCancel(m.rootCtx) +// OnDataPathDown notifies that the peer's data path went down; further rekeys wait +// until it is up again (the host re-bootstraps over signalling on reconnect). +func (m *Manager) OnDataPathDown(remoteID string) { 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, - } + delete(m.dataSend, remoteID) m.mu.Unlock() - - m.wait.Add(1) - go m.initiatorLoop(ctx, remoteID, id) - - return m.send(remoteID, raw) } +// ---- internals ---- + func (m *Manager) rekeyLoop(ctx context.Context, remoteID string) { defer m.wait.Done() t := time.NewTicker(m.rekeyInterval) @@ -277,23 +300,36 @@ func (m *Manager) rekeyLoop(ctx context.Context, remoteID string) { case <-ctx.Done(): return case <-t.C: - if err := m.initiateRekey(remoteID); err != nil { + m.mu.Lock() + _, dpUp := m.dataSend[remoteID] + _, inFlight := m.exchanges[remoteID] + m.mu.Unlock() + // Rekeys ride the data path only; skip when it is down (the host will + // re-bootstrap over signalling on reconnect) or an exchange is in flight. + if !dpUp || inFlight || !m.IsInitiator(remoteID) { + continue + } + offer, err := m.startExchange(remoteID, false) + if err != nil { m.logger.Error("pqkem rekey failed to start", "peer", remoteID, "err", err) + continue + } + if err := m.pushDataPath(remoteID, offer); err != nil { + m.logger.Warn("pqkem send rekey offer failed", "peer", remoteID, "err", err) } } } } -// send routes offer/answer over the data path when it is up, else the signalling -// channel. The confirm never goes through here — it always uses SendDataPath. -func (m *Manager) send(remoteID string, msg []byte) error { +// pushDataPath sends over the peer's data path, erroring if it is down. +func (m *Manager) pushDataPath(remoteID string, msg []byte) error { m.mu.Lock() - viaDataPath := m.dataPathUp[remoteID] + send := m.dataSend[remoteID] m.mu.Unlock() - if viaDataPath { - return m.transport.SendDataPath(remoteID, msg) + if send == nil { + return fmt.Errorf("no data path for peer %s", remoteID) } - return m.transport.SendSignal(remoteID, msg) + return send(remoteID, msg) } func (m *Manager) binding(remoteID string) Binding { diff --git a/client/internal/pqkem/manager_test.go b/client/internal/pqkem/manager_test.go index aea902fc4..065164a1e 100644 --- a/client/internal/pqkem/manager_test.go +++ b/client/internal/pqkem/manager_test.go @@ -8,20 +8,17 @@ import ( "github.com/stretchr/testify/require" ) -// loopback delivers a sent message synchronously to the peer manager's HandleInbound, -// attributing it to localID (the sender). Both channels deliver the same way — the -// test does not care which physical channel is used. +// loopback is a data-path transport: a SendDataPath delivers synchronously to the +// peer manager's OnDataPathMessage, attributing it to localID (the sender). The +// signalling channel is driven by the test directly via the SignalX methods. type loopback struct { localID string peer *Manager } -func (l *loopback) SendDataPath(remoteID string, msg []byte) error { return l.deliver(msg) } -func (l *loopback) SendSignal(remoteID string, msg []byte) error { return l.deliver(msg) } - -func (l *loopback) deliver(msg []byte) error { +func (l *loopback) SendDataPath(remoteID string, msg []byte) error { cp := append([]byte(nil), msg...) - return l.peer.HandleInbound(l.localID, cp) + return l.peer.OnDataPathMessage(l.localID, cp) } type fakeWG struct { @@ -58,23 +55,30 @@ func TestManager_ExchangeConverges(t *testing.T) { wgA := newFakeWG() wgB := newFakeWG() - // long interval so the ticker never fires during the test; we drive manually. dA := NewManager("aaaa", lbA, wgA, time.Hour, nil) dB := NewManager("bbbb", lbB, wgB, time.Hour, nil) - lbA.peer = dB // A sends -> B receives - lbB.peer = dA // B sends -> A receives + lbA.peer = dB // A's data-path sends -> B receives + lbB.peer = dA // B's data-path sends -> A receives dA.AddPeer("bbbb") dB.AddPeer("aaaa") defer dA.Stop() defer dB.Stop() - // B is the initiator ("bbbb" > "aaaa"). Offer/answer flow synchronously over the - // loopback; both commit their PSK, and B parks in stateAwaitingRekey. - require.NoError(t, dB.initiateRekey("aaaa")) + // Bootstrap over the signalling channel (the test plays the host carrying bytes). + // B is the initiator ("bbbb" > "aaaa"). + offer, err := dB.SignalOffer("aaaa") + require.NoError(t, err) + require.NotNil(t, offer) - // The consumer reports the data path is (re)keyed on both sides; this makes B - // send the confirm, which converges A. + answer, err := dA.SignalOnOffer("bbbb", offer) + require.NoError(t, err) + require.NotNil(t, answer) + + require.NoError(t, dB.SignalOnAnswer("aaaa", answer)) + + // Data path comes up on both sides; this makes B send the confirm over the data + // path, converging A. dA.OnDataPathRekeyed("bbbb") dB.OnDataPathRekeyed("aaaa") @@ -85,16 +89,15 @@ func TestManager_ExchangeConverges(t *testing.T) { require.Equal(t, pskB, pskA, "both sides converge on the same PSK") } -func TestManager_NonInitiatorDoesNothing(t *testing.T) { - lbA := &loopback{localID: "aaaa"} - wgA := newFakeWG() - dA := NewManager("aaaa", lbA, wgA, time.Hour, nil) +func TestManager_NonInitiatorReturnsNoOffer(t *testing.T) { + dA := NewManager("aaaa", &loopback{localID: "aaaa"}, newFakeWG(), time.Hour, nil) dA.AddPeer("bbbb") defer dA.Stop() - // A is NOT the initiator vs "bbbb" -> initiateRekey is a no-op, no Send. - require.NoError(t, dA.initiateRekey("bbbb")) - require.Equal(t, PSK{}, wgA.psk("bbbb")) + // A is NOT the initiator vs "bbbb" -> no offer to send. + offer, err := dA.SignalOffer("bbbb") + require.NoError(t, err) + require.Nil(t, offer) } func TestManager_StopIsIdempotent(t *testing.T) {