diff --git a/client/internal/pqkem/convergence.go b/client/internal/pqkem/convergence.go index f818590b1..98048706b 100644 --- a/client/internal/pqkem/convergence.go +++ b/client/internal/pqkem/convergence.go @@ -76,6 +76,16 @@ func (m *Manager) startExchangeLocked(remoteID RemoteID, viaSignal bool, ackID E func (m *Manager) processOffer(remoteID RemoteID, o *OfferMsg) ([]byte, error) { m.trace("pqkem: offer received", "peer", remoteID, "exchange", idHex(o.ExchangeID), "acks", idHex(o.AckID)) + // Role guard: only the responder processes offers. The KEM is unidirectional — the + // initiator sends offers, the responder answers — so an offer reaching the initiator + // is anomalous (desync, duplicate, or an injected/spoofed packet). Processing it would + // derive and commit a fresh PSK, overwriting a live one and silently discarding any + // in-flight exchange (whose retry loop then exits without recovery). Drop it. + if m.IsInitiator(remoteID) { + m.trace("pqkem: dropping offer, we are the initiator for this peer (role violation)", "peer", remoteID, "exchange", idHex(o.ExchangeID)) + return nil, nil + } + if o.AckID != (ExchangeID{}) { m.ackConverged(remoteID, o.AckID) } @@ -132,6 +142,14 @@ func (m *Manager) processOffer(remoteID RemoteID, o *OfferMsg) ([]byte, error) { // 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 { + // Role guard: only the initiator processes answers. An answer reaching the responder + // is anomalous (the responder sends answers, it never receives them) — drop it rather + // than let a stray/injected answer disturb the responder's state. + if !m.IsInitiator(remoteID) { + m.trace("pqkem: dropping answer, we are the responder for this peer (role violation)", "peer", remoteID, "answer_for", idHex(a.ExchangeID)) + return nil + } + m.mu.Lock() ex := m.exchanges[remoteID] if ex == nil || ex.id != a.ExchangeID || ex.state != stateAwaitingAnswer { diff --git a/client/internal/pqkem/convergence_test.go b/client/internal/pqkem/convergence_test.go index c8f4f1599..c2bef61ac 100644 --- a/client/internal/pqkem/convergence_test.go +++ b/client/internal/pqkem/convergence_test.go @@ -73,3 +73,24 @@ func TestManager_RekeyToleratesKFailures(t *testing.T) { require.NoError(t, err) assert.Eventually(t, func() bool { return failedCount(wgB) == 1 }, time.Second, 5*time.Millisecond) } + +// TestManager_InitiatorRejectsOfferFromResponderRole verifies the role guard: an offer +// reaching the peer that is the initiator (here dB) must be dropped, not adopted — else a +// stray/injected offer would overwrite the live PSK and silently drop any in-flight +// exchange. dA (the role-responder) crafts an offer and injects it into dB (the initiator). +func TestManager_InitiatorRejectsOfferFromResponderRole(t *testing.T) { + dA, dB, _, wgB, _ := pair(t) // dB ("bbbb") is the initiator, dA ("aaaa") the responder + defer dA.Stop() + defer dB.Stop() + + bootstrap(t, dA, dB) + psk1 := wgB.psk("aaaa") + require.NotEqual(t, PSK{}, psk1, "bootstrap established a PSK") + + rogue, err := dA.startExchangeTest("bbbb", false, ExchangeID{}) + require.NoError(t, err) + require.NoError(t, dB.OnDataPathMessage("aaaa", rogue)) + + assert.Equal(t, psk1, wgB.psk("aaaa"), + "the initiator must not adopt a PSK from an offer sent by the role-responder") +} diff --git a/client/internal/pqkem/manager.go b/client/internal/pqkem/manager.go index 22123d2ce..a58f9f8ef 100644 --- a/client/internal/pqkem/manager.go +++ b/client/internal/pqkem/manager.go @@ -295,13 +295,16 @@ func (m *Manager) SignalOffer(remoteID RemoteID) ([]byte, error) { m.mu.Unlock() return nil, nil // peer does not run the KEM; do not offer (avoids a failure/reoffer loop) } - // Idempotent while a signalling bootstrap is in flight OR already derived a PSK but - // not yet chained a rotation (awaitingRekey): return the SAME offer instead of - // starting a new exchange. This matters when the controller both offers on its own - // guard AND re-offers in response to the responder's offer — without this, the - // second call would start a fresh exchange (a different PSK) and desync the peers. - if ex := m.exchanges[remoteID]; ex != nil && ex.viaSignal && - (ex.state == stateAwaitingAnswer || ex.state == stateAwaitingRekey) { + // Idempotent ONLY while a signalling bootstrap is still in flight (awaitingAnswer): + // a repeat call returns the SAME offer instead of starting a duplicate exchange. Once + // the PSK is derived (awaitingRekey) a fresh signal offer must NOT reuse the frozen + // exchange: a signal re-negotiation means the remote may have restarted and lost its + // state, so it would Respond with fresh material and derive a different PSK, which our + // still-awaitingRekey side would reject — a permanent desync. Starting a fresh exchange + // re-bootstraps and both sides converge (the responder always adopts the latest). The + // controller-double-offer case the idempotency once guarded is already covered by + // ShouldSendBootstrapOffer (false while any exchange exists). + if ex := m.exchanges[remoteID]; ex != nil && ex.viaSignal && ex.state == stateAwaitingAnswer { last := ex.lastSent m.mu.Unlock() return last, nil diff --git a/client/internal/pqkem/manager_test.go b/client/internal/pqkem/manager_test.go index b7a13166a..94ab70490 100644 --- a/client/internal/pqkem/manager_test.go +++ b/client/internal/pqkem/manager_test.go @@ -7,6 +7,7 @@ import ( "sync/atomic" "testing" + "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) @@ -199,3 +200,41 @@ func TestManager_StopIsIdempotent(t *testing.T) { dA.Stop() dA.Stop() // must not panic or hang } + +// TestManager_SignalOfferRebootstrapsAfterResponderRestart verifies finding A: once a PSK +// is derived (initiator in awaitingRekey), a fresh signalling offer must start a NEW +// exchange, not replay the frozen one. Otherwise, if the responder restarted and lost its +// state, it would derive a different PSK the initiator then rejects — a permanent desync. +func TestManager_SignalOfferRebootstrapsAfterResponderRestart(t *testing.T) { + sw := newSwitch() + wgA, wgB := newFakeWG(), newFakeWG() + dA := NewManager("aaaa", wgA, nil) + dB := NewManager("bbbb", wgB, nil) + dA.Start(&loopback{ep: epA, sw: sw}) + dB.Start(&loopback{ep: epB, sw: sw}) + dA.AddPeer("bbbb", epB) + dB.AddPeer("aaaa", epA) + defer dB.Stop() + + bootstrap(t, dA, dB) + require.Equal(t, wgB.psk("aaaa"), wgA.psk("bbbb"), "bootstrap converged") + + // Responder ("aaaa") restarts: fresh manager, no state. + dA.Stop() + wgA2 := newFakeWG() + dA2 := NewManager("aaaa", wgA2, nil) + dA2.Start(&loopback{ep: epA, sw: sw}) + dA2.AddPeer("bbbb", epB) + defer dA2.Stop() + + offer, err := dB.SignalOffer("aaaa") + require.NoError(t, err) + require.NotNil(t, offer) + answer, err := dA2.SignalOnOffer("bbbb", offer) + require.NoError(t, err) + require.NotNil(t, answer) + require.NoError(t, dB.SignalOnAnswer("aaaa", answer)) + + assert.Equal(t, wgB.psk("aaaa"), wgA2.psk("bbbb"), + "after a responder restart both sides must converge on the same PSK") +}