Consensys · zilm13 · Jun 12, 2026
diff --git a/.../src/main/java/tech/pegasys/teku/networking/eth2/rpc/core/Eth2IncomingRequestHandler.java b/.../src/main/java/tech/pegasys/teku/networking/eth2/rpc/core/Eth2IncomingRequestHandler.java
@@ -13,6 +13,8 @@
 
 package tech.pegasys.teku.networking.eth2.rpc.core;
 
+import static tech.pegasys.teku.networking.p2p.reputation.ReputationAdjustment.LARGE_PENALTY;
+
 import com.google.common.annotations.VisibleForTesting;
 import io.netty.buffer.ByteBuf;
 import java.time.Duration;
@@ -21,6 +23,7 @@
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import tech.pegasys.teku.infrastructure.async.AsyncRunner;
+import tech.pegasys.teku.infrastructure.async.SafeFuture;
 import tech.pegasys.teku.infrastructure.ssz.SszData;
 import tech.pegasys.teku.networking.eth2.peers.Eth2Peer;
 import tech.pegasys.teku.networking.eth2.peers.PeerLookup;
@@ -47,6 +50,7 @@ public class Eth2IncomingRequestHandler<
   private final String protocolId;
   private final AsyncRunner asyncRunner;
   private final AtomicBoolean requestHandled = new AtomicBoolean(false);
+  private Optional<Eth2Peer> cachedPeer = Optional.empty();
 
   public Eth2IncomingRequestHandler(
       final String protocolId,
@@ -70,26 +74,31 @@ public void active(final NodeId nodeId, final RpcStream rpcStream) {
 
   @Override
   public void processData(final NodeId nodeId, final RpcStream rpcStream, final ByteBuf data) {
+    final Optional<Eth2Peer> peer = getPeer(nodeId);
+    final PenalizingResponseCallback<TResponse> responseCallback =
+        createResponseCallback(rpcStream, peer);
     try {
-      Optional<Eth2Peer> peer = peerLookup.getConnectedPeer(nodeId);
       requestDecoder
           .decodeRequest(data)
-          .ifPresent(request -> handleRequest(peer, request, createResponseCallback(rpcStream)));
+          .ifPresent(request -> handleRequest(peer, request, responseCallback));
     } catch (final RpcException e) {
       requestHandled.set(true);
-      createResponseCallback(rpcStream).completeWithErrorResponse(e);
+      responseCallback.completeWithErrorResponse(e);
     }
   }
 
   @Override
   public void readComplete(final NodeId nodeId, final RpcStream rpcStream) {
+    final Optional<Eth2Peer> peer = getPeer(nodeId);
+    final PenalizingResponseCallback<TResponse> responseCallback =
+        createResponseCallback(rpcStream, peer);
     try {
-      Optional<Eth2Peer> peer = peerLookup.getConnectedPeer(nodeId);
       requestDecoder
           .complete()
-          .ifPresent(request -> handleRequest(peer, request, createResponseCallback(rpcStream)));
+          .ifPresent(request -> handleRequest(peer, request, responseCallback));
     } catch (RpcException e) {
-      createResponseCallback(rpcStream).completeWithErrorResponse(e);
+      requestHandled.set(true);
+      responseCallback.completeWithErrorResponse(e);
       LOG.debug("RPC Request stream closed prematurely {}", protocolId, e);
     }
   }
@@ -102,7 +111,7 @@ public void closed(final NodeId nodeId, final RpcStream rpcStream) {
   private void handleRequest(
       final Optional<Eth2Peer> peer,
       final TRequest request,
-      final ResponseCallback<TResponse> callback) {
+      final PenalizingResponseCallback<TResponse> callback) {
     try {
       requestHandled.set(true);
       final Optional<RpcException> requestValidationError =
@@ -117,10 +126,24 @@ private void handleRequest(
       callback.completeWithUnexpectedError(e);
     } catch (final Throwable t) {
       LOG.error("Unhandled error while processing request {}", protocolId, t);
+      callback.penalizePeer(
+          "unexpected RPC request handler exception: " + t.getClass().getSimpleName());
       callback.completeWithUnexpectedError(t);
     }
   }
 
+  private Optional<Eth2Peer> getPeer(final NodeId nodeId) {
+    if (cachedPeer.isPresent()) {
+      return cachedPeer;
+    }
+
+    final Optional<Eth2Peer> peer = peerLookup.getConnectedPeer(nodeId);
+    if (peer.isPresent()) {
+      cachedPeer = peer;
+    }
+    return peer;
+  }
+
   private void ensureRequestReceivedWithinTimeLimit(final RpcStream stream) {
     asyncRunner
         .runAfterDelay(
@@ -147,7 +170,73 @@ public String toString() {
     return "Eth2IncomingRequestHandler{" + "protocol=" + protocolId + '}';
   }
 
-  private RpcResponseCallback<TResponse> createResponseCallback(final RpcStream rpcStream) {
-    return new RpcResponseCallback<>(rpcStream, responseEncoder);
+  private PenalizingResponseCallback<TResponse> createResponseCallback(
+      final RpcStream rpcStream, final Optional<Eth2Peer> peer) {
+    return new PenalizingResponseCallback<>(
+        new RpcResponseCallback<>(rpcStream, responseEncoder), peer, protocolId);
+  }
+
+  private static class PenalizingResponseCallback<T> implements ResponseCallback<T> {
+    private final ResponseCallback<T> delegate;
+    private final Optional<Eth2Peer> peer;
+    private final String protocolId;
+    private final AtomicBoolean penaltyApplied = new AtomicBoolean(false);
+
+    private PenalizingResponseCallback(
+        final ResponseCallback<T> delegate,
+        final Optional<Eth2Peer> peer,
+        final String protocolId) {
+      this.delegate = delegate;
+      this.peer = peer;
+      this.protocolId = protocolId;
+    }
+
+    @Override
+    public SafeFuture<Void> respond(final T data) {
+      return delegate.respond(data);
+    }
+
+    @Override
+    public void respondAndCompleteSuccessfully(final T data) {
+      delegate.respondAndCompleteSuccessfully(data);
+    }
+
+    @Override
+    public void completeSuccessfully() {
+      delegate.completeSuccessfully();
+    }
+
+    @Override
+    public void completeWithErrorResponse(final RpcException error) {
+      if (isMalformedRequestError(error)) {
+        penalizePeer("malformed RPC request: " + error.getErrorMessageString());
+      }
+      delegate.completeWithErrorResponse(error);
+    }
+
+    @Override
+    public void completeWithUnexpectedError(final Throwable error) {
+      delegate.completeWithUnexpectedError(error);
+    }
+
+    private boolean isMalformedRequestError(final RpcException error) {
+      return error instanceof RpcException.ChunkTooLongException
+          || error instanceof RpcException.DecompressFailedException
+          || error instanceof RpcException.DeserializationFailedException
+          || error instanceof RpcException.ExtraDataAppendedException
+          || error instanceof RpcException.LengthOutOfBoundsException
+          || error instanceof RpcException.MessageTruncatedException
+          || error instanceof RpcException.PayloadTruncatedException;
+    }
+
+    private void penalizePeer(final String reason) {
+      if (penaltyApplied.compareAndSet(false, true)) {
+        peer.ifPresent(
+            p -> {
+              LOG.debug("Penalising peer {} for {} on protocol {}", p.getId(), reason, protocolId);
+              p.adjustReputation(LARGE_PENALTY);
+            });
+      }
+    }
   }
 }
diff --git a/.../test/java/tech/pegasys/teku/networking/eth2/rpc/core/Eth2IncomingRequestHandlerTest.java b/.../test/java/tech/pegasys/teku/networking/eth2/rpc/core/Eth2IncomingRequestHandlerTest.java
@@ -18,8 +18,10 @@
 import static org.mockito.ArgumentMatchers.argThat;
 import static org.mockito.Mockito.lenient;
 import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
+import static tech.pegasys.teku.networking.p2p.reputation.ReputationAdjustment.LARGE_PENALTY;
 
 import io.netty.buffer.ByteBuf;
 import java.util.Optional;
@@ -32,6 +34,7 @@
 import tech.pegasys.teku.networking.eth2.rpc.Utils;
 import tech.pegasys.teku.networking.eth2.rpc.beaconchain.BeaconChainMethods;
 import tech.pegasys.teku.networking.eth2.rpc.core.encodings.RpcEncoding;
+import tech.pegasys.teku.networking.eth2.rpc.core.encodings.context.RpcContextCodec;
 import tech.pegasys.teku.networking.eth2.rpc.core.methods.Eth2RpcMethod;
 import tech.pegasys.teku.spec.TestSpecFactory;
 import tech.pegasys.teku.spec.datastructures.blocks.SignedBeaconBlock;
@@ -111,6 +114,64 @@ public void shouldNotCloseStreamIfRequestReceivedInTime() throws Exception {
     verify(rpcStream, never()).closeAbruptly();
   }
 
+  @Test
+  public void shouldNotPenalizePeerForSemanticallyInvalidRequest() {
+    final BeaconBlocksByRangeRequestMessage invalidRequest =
+        new BeaconBlocksByRangeRequestMessage(UInt64.ONE, UInt64.ONE, UInt64.ZERO);
+    final Bytes invalidRequestData =
+        beaconChainMethods.beaconBlocksByRange().encodeRequest(invalidRequest);
+
+    reqHandler.processData(nodeId, rpcStream, Utils.toByteBuf(invalidRequestData));
+
+    verify(peer, never()).adjustReputation(LARGE_PENALTY);
+  }
+
+  @Test
+  public void shouldPenalizePeerForMalformedRequest() {
+    final Bytes malformedRequestData = Bytes.fromHexString("0xdeadbeef");
+
+    reqHandler.processData(nodeId, rpcStream, Utils.toByteBuf(malformedRequestData));
+    reqHandler.readComplete(nodeId, rpcStream);
+
+    verify(peer).adjustReputation(LARGE_PENALTY);
+  }
+
+  @Test
+  public void shouldUseCachedPeerWhenReadCompleteFailsAfterPeerRemoved() {
+    when(peerLookup.getConnectedPeer(nodeId))
+        .thenReturn(Optional.of(peer))
+        .thenReturn(Optional.empty());
+    final ByteBuf partialData = Utils.toByteBuf(requestData.slice(0, requestData.size() / 2));
+
+    reqHandler.processData(nodeId, rpcStream, partialData);
+    reqHandler.readComplete(nodeId, rpcStream);
+
+    verify(peerLookup, times(1)).getConnectedPeer(nodeId);
+    verify(peer).adjustReputation(LARGE_PENALTY);
+  }
+
+  @Test
+  public void shouldPenalizePeerForUnexpectedRequestHandlerException() {
+    final LocalMessageHandler<EmptyMessage, EmptyMessage> throwingMessageHandler =
+        (protocolId, peer, message, callback) -> {
+          throw new RuntimeException("boom");
+        };
+    final RpcResponseEncoder<EmptyMessage, Bytes> responseEncoder =
+        new RpcResponseEncoder<>(getRpcEncoding(), RpcContextCodec.noop(EmptyMessage.SSZ_SCHEMA));
+    final Eth2IncomingRequestHandler<EmptyMessage, EmptyMessage> requestHandler =
+        new Eth2IncomingRequestHandler<>(
+            "test",
+            responseEncoder,
+            new RpcRequestDecoder<>(EmptyMessage.SSZ_SCHEMA, getRpcEncoding()),
+            asyncRunner,
+            peerLookup,
+            throwingMessageHandler);
+
+    requestHandler.readComplete(nodeId, rpcStream);
+
+    verify(peer).adjustReputation(LARGE_PENALTY);
+  }
+
   @Test
   public void shouldReleaseBuffersWhenStreamClosedAfterPartialData() {
     // Send partial data (incomplete request) - this will cause the decoder to retain the buffer

diff --git a/...rking/p2p/src/main/java/tech/pegasys/teku/networking/p2p/libp2p/LibP2PNetworkBuilder.java b/...rking/p2p/src/main/java/tech/pegasys/teku/networking/p2p/libp2p/LibP2PNetworkBuilder.java
@@ -105,6 +105,7 @@ public P2PNetwork<Peer> build() {
     rpcHandlers = createRpcHandlers();
     // Setup peers
     peerManager = createPeerManager();
+    gossipNetwork.peerLookup(peerManager::getPeer);
 
     final PrivKey privKey = privateKeyProvider.get();
     final NodeId nodeId = new LibP2PNodeId(PeerId.fromPubKey(privKey.publicKey()));

diff --git a/...rking/p2p/src/main/java/tech/pegasys/teku/networking/p2p/libp2p/gossip/GossipHandler.java b/...rking/p2p/src/main/java/tech/pegasys/teku/networking/p2p/libp2p/gossip/GossipHandler.java
@@ -13,12 +13,17 @@
 
 package tech.pegasys.teku.networking.p2p.libp2p.gossip;
 
+import static tech.pegasys.teku.networking.p2p.reputation.ReputationAdjustment.LARGE_PENALTY;
+import static tech.pegasys.teku.networking.p2p.reputation.ReputationAdjustment.SMALL_PENALTY;
+
+import io.libp2p.core.PeerId;
 import io.libp2p.core.pubsub.MessageApi;
 import io.libp2p.core.pubsub.PubsubPublisherApi;
 import io.libp2p.core.pubsub.Topic;
 import io.libp2p.core.pubsub.ValidationResult;
 import io.libp2p.pubsub.PubsubMessage;
 import io.netty.buffer.Unpooled;
+import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
 import java.util.function.Function;
 import org.apache.logging.log4j.LogManager;
@@ -29,6 +34,10 @@
 import tech.pegasys.teku.infrastructure.async.SafeFuture;
 import tech.pegasys.teku.infrastructure.metrics.TekuMetricCategory;
 import tech.pegasys.teku.networking.p2p.gossip.TopicHandler;
+import tech.pegasys.teku.networking.p2p.libp2p.LibP2PNodeId;
+import tech.pegasys.teku.networking.p2p.peer.NodeId;
+import tech.pegasys.teku.networking.p2p.peer.Peer;
+import tech.pegasys.teku.networking.p2p.reputation.ReputationAdjustment;
 
 public class GossipHandler implements Function<MessageApi, CompletableFuture<ValidationResult>> {
   private static final Logger LOG = LogManager.getLogger();
@@ -40,15 +49,26 @@ public class GossipHandler implements Function<MessageApi, CompletableFuture<Val
   private final PubsubPublisherApi publisher;
   private final TopicHandler handler;
   private final Counter messageCounter;
+  private final Function<NodeId, Optional<Peer>> peerLookup;
 
   public GossipHandler(
       final MetricsSystem metricsSystem,
       final Topic topic,
       final PubsubPublisherApi publisher,
       final TopicHandler handler) {
+    this(metricsSystem, topic, publisher, handler, __ -> Optional.empty());
+  }
+
+  public GossipHandler(
+      final MetricsSystem metricsSystem,
+      final Topic topic,
+      final PubsubPublisherApi publisher,
+      final TopicHandler handler,
+      final Function<NodeId, Optional<Peer>> peerLookup) {
     this.topic = topic;
     this.publisher = publisher;
     this.handler = handler;
+    this.peerLookup = peerLookup;
     this.messageCounter =
         metricsSystem
             .createLabelledCounter(
@@ -69,6 +89,10 @@ public SafeFuture<ValidationResult> apply(final MessageApi message) {
           "Rejecting gossip message of length {} which exceeds maximum size of {}",
           messageSize,
           maxMessageSize);
+      penalizePeer(
+          message,
+          LARGE_PENALTY,
+          "gossip message size " + messageSize + " exceeds maximum " + maxMessageSize);
       return VALIDATION_FAILED;
     }
     LOG.trace("Received message for topic {}", topic);
@@ -78,7 +102,20 @@ public SafeFuture<ValidationResult> apply(final MessageApi message) {
       throw new IllegalArgumentException(
           "Don't know this PubsubMessage implementation: " + pubsubMessage.getClass());
     }
-    return handler.handleMessage(gossipPubsubMessage.getPreparedMessage());
+    return handler
+        .handleMessage(gossipPubsubMessage.getPreparedMessage())
+        .thenPeek(
+            validationResult -> {
+              if (validationResult == ValidationResult.Invalid) {
+                penalizePeer(message, SMALL_PENALTY, "invalid gossip message");
+              }
+            })
+        .catchAndRethrow(
+            error ->
+                penalizePeer(
+                    message,
+                    LARGE_PENALTY,
+                    "gossip validation exception: " + error.getClass().getSimpleName()));
   }
 
   public SafeFuture<Void> gossip(final Bytes bytes) {
@@ -90,4 +127,29 @@ public SafeFuture<Void> gossip(final Bytes bytes) {
   public String getTopic() {
     return topic.getTopic();
   }
+
+  private void penalizePeer(
+      final MessageApi message, final ReputationAdjustment penalty, final String reason) {
+    lookupPeer(message)
+        .ifPresent(
+            peer -> {
+              LOG.debug(
+                  "Penalising peer {} with {} for {} on topic {}",
+                  peer.getId(),
+                  penalty,
+                  reason,
+                  topic);
+              peer.adjustReputation(penalty);
+            });
+  }
+
+  private Optional<Peer> lookupPeer(final MessageApi message) {
+    final byte[] sender = message.getFrom();
+    if (sender == null || sender.length == 0) {
+      return Optional.empty();
+    }
+
+    final NodeId nodeId = new LibP2PNodeId(new PeerId(sender));
+    return peerLookup.apply(nodeId);
+  }
 }