wit2: address adversarial-review safety findings (TDD)

Three high-severity issues from the Codex adversarial review on PR #2208,
each with a TDD regression test added first then fixed:

1. Header-race: signed announce arriving before header was silently
   downgraded. handleSignedWitnessAnnouncements called peer.AddKnownAnnounce
   unconditionally before the verification gate, leaving a peer marked
   announce-known even on bad-signature / header-unknown rejection paths.
   That suppressed our own re-relay back to that peer if a valid version
   of the same hash arrived from someone else, killing the natural
   recovery path. Fix: gate AddKnownAnnounce on acceptSignedAnnouncement
   success so the announce-known bit only reflects verified delivery.

   Test: TestHandleSignedWitnessAnnouncementsBadSigDoesNotMarkAnnounceKnown.

2. pendingWitnessBodies TTL didn't actually evict. get() observed expiry
   and returned false but left the entry in the map; gcLocked only ran
   from put(), so a node that stopped receiving witnesses retained up to
   capacity (10) ~50MB blobs indefinitely. Fix: when get() observes an
   expired entry, upgrade to write lock and delete it (re-checking under
   the write lock to avoid clobbering a concurrent put).

   Test: TestPendingWitnessBodyCacheGetEvictsExpired.

3. Honest body-server dropped on bad producer commitment.
   verifyAgainstSignedHash dropped the byte-server on every signed-hash
   mismatch, but the announcement only proves *some* BP signed *some*
   hash — not that the hash matches the canonical witness. A faulty or
   malicious scheduled producer that signed a bogus hash would weaponise
   this to disconnect every honest peer serving the real witness. Fix:
   reject the bytes (don't cache for serving) and back off the request
   without dropping the byte-server. TODO comment left for follow-up
   signer-quarantine work, which needs (signer, relayer) provenance the
   manager doesn't currently have.

   Test: TestProcessWitnessResponseDoesNotDropOnByteMismatch (replaces
   the previous TestProcessWitnessResponseDropsOnHashMismatch, whose
   policy this commit reverses).

Author: lucca30

eth/fetcher/witness_manager.go

@@ -723,9 +723,18 @@ func (m *witnessManager) verifyAgainstSignedHash(peer string, hash common.Hash,
 	actual := stateless.WitnessCommitHash(encoded)
 	if actual != expected {
 		witnessByteMismatchMeter.Mark(1)
-		log.Warn("[wm] Witness bytes do not match BP-signed hash; dropping peer",
+		// We cannot blame the byte-server on signed-hash disagreement alone:
+		// the announcement only proves *some* BP signed *some* hash. A faulty
+		// or malicious scheduled producer that signed a bogus hash would
+		// otherwise weaponise this path to disconnect every honest peer
+		// serving the canonical witness. Reject the bytes (don't cache for
+		// serving), back off the pending request so another peer/announcement
+		// gets tried, and let import-time execution validation pin blame.
+		// TODO(wit2): wire signer-quarantine once the manager has access to
+		// (signer, announcement-relayer) provenance from the handler.
+		log.Warn("[wm] Witness bytes do not match BP-signed hash; not caching, retrying with another peer",
 			"peer", peer, "block", hash, "expected", expected, "actual", actual)
-		m.handleWitnessFetchFailureExt(hash, peer, errors.New("witness hash mismatch"), false)
+		m.handleWitnessFetchFailureExt(hash, "", errors.New("witness hash mismatch"), false)
 		return nil, common.Hash{}, false
 	}
 	return encoded, expected, true

eth/fetcher/witness_manager_wit2_test.go

@@ -8,7 +8,6 @@ import (
 
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/core/stateless"
-	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/eth/protocols/eth"
 )
 
@@ -26,65 +25,60 @@ func blockAnnounceForTest(origin string, hash common.Hash, number uint64) *block
 	}
 }
 
-// TestProcessWitnessResponseDropsOnHashMismatch is the load-bearing safety
-// guarantee for WIT2 pre-import serving: a peer that returns bytes whose
-// keccak256 doesn't match the BP-signed witnessHash must be dropped, even
-// if every other check passes.
+// TestProcessWitnessResponseDoesNotDropOnByteMismatch encodes the post-
+// adversarial-review safety policy: when the served witness bytes do not
+// match the BP-signed witnessHash on file, the manager must back off and
+// retry, but it MUST NOT drop the byte-server. The accepted announcement
+// only proves *some* BP signed *some* hash — not that the hash matches the
+// canonical witness. A faulty or malicious scheduled producer that signs a
+// bogus hash would otherwise weaponise this code path to disconnect every
+// honest peer serving the real witness.
 //
-// Without this, a malicious server could pollute downstream relayers with
-// bytes the BP never committed to, and the relayers would face state-root
-// failures during execution that they cannot attribute to the right party.
-func TestProcessWitnessResponseDropsOnHashMismatch(t *testing.T) {
+// The mismatched bytes are still rejected (not cached for serving), and the
+// pending state stays alive with a fresh back-off so another peer (or another
+// announcement) gets a chance. Blame-pinning belongs at execution time, where
+// import-side validation can attribute fault to signer vs. server vs. caller.
+func TestProcessWitnessResponseDoesNotDropOnByteMismatch(t *testing.T) {
 	tw := newTestWitnessManager()
 	defer tw.Close()
 
 	block := createTestBlock(101)
 	hash := block.Hash()
 
-	// Prepare a "correct" witness that the BP signed over.
-	correct := createTestWitnessForBlock(block)
-	var buf bytes.Buffer
-	if err := correct.EncodeRLP(&buf); err != nil {
-		t.Fatalf("encode: %v", err)
-	}
-	signedWitnessHash := stateless.WitnessCommitHash(buf.Bytes())
-
-	// The peer will return a different witness — same block number, but
-	// the trie differs, producing different bytes and a different hash.
-	differentHeader := types.CopyHeader(block.Header())
-	differentHeader.GasUsed = 999_999_999
-	differentBlock := types.NewBlockWithHeader(differentHeader)
-	rogueWitness := createTestWitnessForBlock(differentBlock)
+	// The honest server returns the canonical witness for this block — its
+	// keccak commitment is `canonical`.
+	canonical := createTestWitnessForBlock(block)
 
-	// Inject the signed-witness lookup so processWitnessResponse uses it.
+	// Simulate a malicious / faulty BP that signed a bogus, unrelated hash.
+	// processWitnessResponse will see canonical bytes whose hash does not
+	// match what parentSignedWitnessHash reports.
+	rogueSignedHash := common.HexToHash("0xdeadbeef")
 	tw.manager.parentSignedWitnessHash = func(h common.Hash) (common.Hash, bool) {
 		if h == hash {
-			return signedWitnessHash, true
+			return rogueSignedHash, true
 		}
 		return common.Hash{}, false
 	}
 
-	// Seed pending state so the failure handler back-off path is exercised.
 	tw.manager.mu.Lock()
 	tw.manager.pending[hash] = &witnessRequestState{
-		op:       &blockOrHeaderInject{origin: "rogue", block: block},
-		announce: blockAnnounceForTest("rogue", hash, block.NumberU64()),
+		op:       &blockOrHeaderInject{origin: "honest", block: block},
+		announce: blockAnnounceForTest("honest", hash, block.NumberU64()),
 	}
 	tw.manager.mu.Unlock()
 
-	// Fabricate the response container expected by processWitnessResponse.
 	res := &eth.Response{
 		Time: time.Millisecond,
 		Done: make(chan error, 1),
-		Res:  []*stateless.Witness{rogueWitness},
+		Res:  []*stateless.Witness{canonical},
 	}
 
-	tw.manager.processWitnessResponse("rogue", hash, res, time.Now())
+	tw.manager.processWitnessResponse("honest-server", hash, res, time.Now())
 
 	tw.mu.Lock()
 	defer tw.mu.Unlock()
-	if len(tw.droppedPeers) != 1 || tw.droppedPeers[0] != "rogue" {
-		t.Fatalf("expected the lying peer to be dropped, got drops=%v", tw.droppedPeers)
+	if len(tw.droppedPeers) != 0 {
+		t.Fatalf("byte-server must not be dropped on signed-hash mismatch (BP may have signed bogus); drops=%v", tw.droppedPeers)
 	}
 }
 

eth/handler_wit.go

@@ -167,14 +167,23 @@ func (h *witHandler) handleSignedWitnessAnnouncements(peer *wit.Peer, anns []wit
 	}
 
 	for _, ann := range anns {
-		// Sender saw this announcement; suppress relay back to them. Do NOT
-		// mark them as a body-holder — they may be relaying without bytes.
-		peer.AddKnownAnnounce(ann.BlockHash)
-
 		if !h.acceptSignedAnnouncement(peer, ann) {
+			// Verification failed (bad signature, signer ≠ producer, or
+			// header not yet local). MUST NOT mark the sender as
+			// announce-known: doing so would (a) suppress our own later
+			// re-relay back to this peer if we receive a valid version of
+			// the same hash from someone else, and (b) leave us no path
+			// to recover from a header-arrival race once a re-gossip for
+			// the same hash arrives. Recovery on this branch relies on
+			// re-receipt, which the empty knownAnnounces set permits.
 			continue
 		}
 
+		// Sender produced a valid announcement; suppress relay back to them.
+		// Do NOT mark them as a body-holder — they may be relaying without
+		// bytes. Body fetches are gated on knownWitnesses, set elsewhere.
+		peer.AddKnownAnnounce(ann.BlockHash)
+
 		// Cache + dedup. Skip relay if we've already relayed this hash recently.
 		if !h.signedWitnesses.putIfNewer(ann) {
 			wit2DuplicateMeter.Mark(1)

eth/handler_wit2.go

@@ -194,14 +194,27 @@ func (c *pendingWitnessBodyCache) put(blockHash common.Hash, bytes []byte, witne
 
 func (c *pendingWitnessBodyCache) get(blockHash common.Hash) ([]byte, common.Hash, bool) {
 	c.mu.RLock()
-	defer c.mu.RUnlock()
 	e, ok := c.entries[blockHash]
 	if !ok {
+		c.mu.RUnlock()
 		return nil, common.Hash{}, false
 	}
 	if time.Since(e.receivedAt) > wit2AnnounceTTL {
+		// Expired: drop the large byte slice now rather than waiting for the
+		// next put() to gc. Without this, a node that stops receiving witness
+		// bodies retains up to capacity (10) ~50MB blobs indefinitely past the
+		// TTL, since gcLocked() only fires on put().
+		c.mu.RUnlock()
+		c.mu.Lock()
+		// Re-check under the write lock: a concurrent put() may have replaced
+		// the entry with a fresh one we should not delete.
+		if cur, ok2 := c.entries[blockHash]; ok2 && cur == e {
+			delete(c.entries, blockHash)
+		}
+		c.mu.Unlock()
 		return nil, common.Hash{}, false
 	}
+	c.mu.RUnlock()
 	return e.bytes, e.witnessHash, true
 }
 

eth/handler_wit2_test.go

@@ -510,6 +510,81 @@ func TestVerifyScheduledProducerDeferredWhenHeaderUnknown(t *testing.T) {
 	}
 }
 
+// TestHandleSignedWitnessAnnouncementsBadSigDoesNotMarkAnnounceKnown is the
+// regression for the verification-ordering bug: handleSignedWitnessAnnouncements
+// must not mark a peer as announce-known until the announcement has passed the
+// signature/producer-binding gate. The previous order called
+// peer.AddKnownAnnounce(hash) unconditionally before acceptSignedAnnouncement,
+// so a peer relaying a structurally invalid announcement still became
+// announce-known for that hash. Two bad consequences flowed from that:
+//   - this node refused to ever relay a *valid* later announcement back to that
+//     peer for the same hash, leaving them unable to recover;
+//   - this node short-circuited its own re-evaluation paths when a good
+//     announcement for the same hash arrived from another peer, because the
+//     original sender's announce-known bit served as a relay-suppression hint.
+//
+// Using a structurally invalid signature (length 3) is sufficient to drive the
+// reject path through verifySignedAnnouncement → strikeWit2Peer without needing
+// a bor engine or block header.
+func TestHandleSignedWitnessAnnouncementsBadSigDoesNotMarkAnnounceKnown(t *testing.T) {
+	h := newTestHandler()
+	defer h.close()
+
+	witH := (*witHandler)(h.handler)
+	peer, cleanup := newTestWit2PeerWithReader()
+	defer cleanup()
+
+	blockHash := common.HexToHash("0xfeedface")
+	ann := wit.SignedWitnessAnnouncement{
+		BlockHash:   blockHash,
+		BlockNumber: 1,
+		WitnessHash: common.HexToHash("0xc0ffee"),
+		Signature:   []byte{0x00, 0x01, 0x02}, // structurally invalid
+	}
+
+	if err := witH.handleSignedWitnessAnnouncements(peer, []wit.SignedWitnessAnnouncement{ann}); err != nil {
+		t.Fatalf("handleSignedWitnessAnnouncements: %v", err)
+	}
+
+	if peer.KnownAnnounceContainsHash(blockHash) {
+		t.Fatal("peer marked announce-known despite invalid signature; verification ordering is broken")
+	}
+	if _, ok := h.handler.signedWitnesses.get(blockHash); ok {
+		t.Fatal("signed announcement cached despite invalid signature")
+	}
+}
+
+// TestPendingWitnessBodyCacheGetEvictsExpired pins the leak fix for the TTL
+// path. Before the fix, get() returned false on expiry but left the entry in
+// the map; gcLocked only ran from put(), so a node that stopped receiving new
+// witnesses retained up to capacity (10) full witness blobs (~50 MiB each)
+// indefinitely, producing a long-lived OOM risk under bursty traffic.
+//
+// The contract this test enforces: any get() that observes an expired entry
+// MUST delete it in place so memory pressure does not persist past the TTL.
+func TestPendingWitnessBodyCacheGetEvictsExpired(t *testing.T) {
+	c := newPendingWitnessBodyCache(4)
+	hash := common.HexToHash("0xfade")
+	c.put(hash, []byte("expensive-body"), common.HexToHash("0xab"))
+
+	// Force the entry's receivedAt back beyond the TTL, mirroring the same
+	// approach used by TestSignedWitnessCacheTTLExpiry above.
+	c.mu.Lock()
+	c.entries[hash].receivedAt = time.Now().Add(-2 * wit2AnnounceTTL)
+	c.mu.Unlock()
+
+	if _, _, ok := c.get(hash); ok {
+		t.Fatal("expired entry must not be returned")
+	}
+
+	c.mu.RLock()
+	entriesAfter := len(c.entries)
+	c.mu.RUnlock()
+	if entriesAfter != 0 {
+		t.Fatalf("expired entry must be deleted on get; len(entries)=%d, want 0", entriesAfter)
+	}
+}
+
 // TestVerifyScheduledProducerRejectsBlockNumberMismatch covers the case where
 // the local header is present but disagrees with the announce on block
 // number. This is a confirmed bad announce and the caller must strike, so