[client] pqkem: enforce exchange roles and re-bootstrap stale signal offers

Two convergence bugs surfaced by the security review:

- Role guard (finding B): processOffer accepted an offer even when we are the
  KEM initiator for the peer, and processAnswer accepted an answer when we are
  the responder. The KEM is unidirectional (initiator offers, responder
  answers), so a role-violating message is anomalous — a desync, a duplicate,
  or an injected/spoofed data-path packet. Processing it derived and committed
  a fresh PSK, overwriting a live one and silently dropping any in-flight
  exchange (whose retry loop then exited without raising a failure or
  re-bootstrapping). Reject offers when we are the initiator and answers when
  we are not; this drops only anomalous traffic and leaves the normal flow
  untouched.

- Re-bootstrap on signal re-negotiation (finding A): SignalOffer was idempotent
  in stateAwaitingRekey too, replaying the frozen bootstrap offer. After the
  responder restarted and lost its state it derived a different PSK from fresh
  material, which our awaitingRekey side then rejected — a permanent desync with
  no recovery (in strict mode the peer stays blocked). Make the idempotency
  apply only while a bootstrap is still in flight (awaitingAnswer); once a PSK
  is derived, a fresh signal offer starts a new exchange so both sides converge.
  The controller-double-offer case the idempotency guarded is already covered by
  ShouldSendBootstrapOffer. Reusing the cached offer also reused the same
  ephemeral keys across exchanges, reducing forward secrecy.

Both paths have a failing-without-the-fix regression test.

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 {

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")
+}

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

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")
+}