feat(api): bound field occurrences of all gRPC API requests

common/src/main/java/org/tron/common/prometheus/MetricKeys.java

@@ -26,6 +26,7 @@ public static class Counter {
     public static final String BLOCK_FETCH_ARMED = "tron:block_fetch_armed";
     public static final String BLOCK_FETCH_SECONDARY = "tron:block_fetch_secondary";
     public static final String BLOCK_ALREADY_KNOWN = "tron:block_already_known";
+    public static final String GRPC_ADMISSION_REJECT = "tron:grpc_admission_reject";
 
     private Counter() {
       throw new IllegalStateException("Counter");

common/src/main/java/org/tron/common/prometheus/MetricsCounter.java

@@ -30,6 +30,8 @@ class MetricsCounter {
             + "was already known when the response was handled (best-effort: concurrent "
             + "arrivals may be missed; a duplicate is not attributed to secondary "
             + "fetches).");
+    init(MetricKeys.Counter.GRPC_ADMISSION_REJECT, "gRPC requests rejected by admission check.",
+        "method", "reason");
   }
 
   private MetricsCounter() {

framework/src/main/java/org/tron/common/application/RpcService.java

@@ -16,15 +16,22 @@
 package org.tron.common.application;
 
 import io.grpc.Server;
+import io.grpc.ServerMethodDefinition;
+import io.grpc.ServerServiceDefinition;
+import io.grpc.ServiceDescriptor;
 import io.grpc.netty.NettyServerBuilder;
 import io.grpc.protobuf.services.ProtoReflectionService;
+import java.util.ArrayList;
+import java.util.List;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.TimeUnit;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.tron.common.es.ExecutorServiceManager;
 import org.tron.common.parameter.CommonParameter;
+import org.tron.core.admission.GuardedRequestMarshaller;
+import org.tron.core.admission.ProtobufAdmissionGuard;
 import org.tron.core.config.args.Args;
 import org.tron.core.services.filter.LiteFnQueryGrpcInterceptor;
 import org.tron.core.services.ratelimiter.PrometheusInterceptor;
@@ -84,14 +91,14 @@ public void innerStop() throws Exception {
   @Override
   public CompletableFuture<Boolean> start() {
     NettyServerBuilder serverBuilder = initServerBuilder();
-    addService(serverBuilder);
+    addServices(serverBuilder);
     addInterceptor(serverBuilder);
     initServer(serverBuilder);
     this.rateLimiterInterceptor.init(this.apiServer);
     return super.start();
   }
 
-  protected NettyServerBuilder initServerBuilder() {
+  protected final NettyServerBuilder initServerBuilder() {
     CommonParameter parameter = Args.getInstance();
     NettyServerBuilder serverBuilder = NettyServerBuilder.forPort(this.port);
     if (parameter.getRpcThreadNum() > 0) {
@@ -116,9 +123,40 @@ protected NettyServerBuilder initServerBuilder() {
     return serverBuilder;
   }
 
-  protected abstract void addService(NettyServerBuilder serverBuilder);
+  /** The services this port serves; {@link #addServices} guards and registers each of them. */
+  protected abstract List<ServerServiceDefinition> services();
 
-  protected void addInterceptor(NettyServerBuilder serverBuilder) {
+  private void addServices(NettyServerBuilder serverBuilder) {
+    int maxMessageSize = Args.getInstance().getMaxMessageSize();
+    for (ServerServiceDefinition service : services()) {
+      serverBuilder.addService(guardRequestMethods(service, maxMessageSize,
+          ProtobufAdmissionGuard.MAX_OCCURRENCES));
+    }
+  }
+
+  /** Guards every method; fails if a request marshaller gives no protobuf prototype. */
+  static ServerServiceDefinition guardRequestMethods(ServerServiceDefinition definition,
+      int maxMessageSize, int maxOccurrences) {
+    ServiceDescriptor service = definition.getServiceDescriptor();
+    ServiceDescriptor.Builder serviceBuilder = ServiceDescriptor.newBuilder(service.getName())
+        .setSchemaDescriptor(service.getSchemaDescriptor());
+    List<ServerMethodDefinition<?, ?>> methods = new ArrayList<>();
+    for (ServerMethodDefinition<?, ?> method : definition.getMethods()) {
+      methods.add(GuardedRequestMarshaller.guard(method, maxMessageSize, maxOccurrences));
+    }
+    // grpc requires the same MethodDescriptor instances in both descriptors.
+    for (ServerMethodDefinition<?, ?> method : methods) {
+      serviceBuilder.addMethod(method.getMethodDescriptor());
+    }
+    ServerServiceDefinition.Builder builder =
+        ServerServiceDefinition.builder(serviceBuilder.build());
+    for (ServerMethodDefinition<?, ?> method : methods) {
+      builder.addMethod(method);
+    }
+    return builder.build();
+  }
+
+  protected final void addInterceptor(NettyServerBuilder serverBuilder) {
     // add a ratelimiter interceptor
     serverBuilder.intercept(this.rateLimiterInterceptor);
 
@@ -134,7 +172,7 @@ protected void addInterceptor(NettyServerBuilder serverBuilder) {
     }
   }
 
-  protected void initServer(NettyServerBuilder serverBuilder) {
+  protected final void initServer(NettyServerBuilder serverBuilder) {
     this.apiServer = serverBuilder.build();
   }
 

framework/src/main/java/org/tron/core/admission/AdmissionCheckedHandler.java

@@ -0,0 +1,84 @@
+package org.tron.core.admission;
+
+import io.grpc.ForwardingServerCallListener.SimpleForwardingServerCallListener;
+import io.grpc.Grpc;
+import io.grpc.Metadata;
+import io.grpc.ServerCall;
+import io.grpc.ServerCallHandler;
+import io.grpc.Status;
+import lombok.extern.slf4j.Slf4j;
+import org.tron.common.prometheus.MetricKeys;
+import org.tron.common.prometheus.Metrics;
+import org.tron.core.admission.ProtobufAdmissionException.Reason;
+
+/**
+ * Closes the call with {@code INVALID_ARGUMENT}, or {@code RESOURCE_EXHAUSTED} for a request over
+ * the size limit, when {@link GuardedRequestMarshaller} hands over a rejection sentinel; the
+ * service method is never invoked.
+ */
+@Slf4j(topic = "rpc")
+public final class AdmissionCheckedHandler<ReqT, RespT> implements ServerCallHandler<ReqT, RespT> {
+
+  private final ServerCallHandler<ReqT, RespT> delegate;
+  private final GuardedRequestMarshaller<ReqT> marshaller;
+
+  AdmissionCheckedHandler(ServerCallHandler<ReqT, RespT> delegate,
+      GuardedRequestMarshaller<ReqT> marshaller) {
+    this.delegate = delegate;
+    this.marshaller = marshaller;
+  }
+
+  public ServerCallHandler<ReqT, RespT> getDelegate() {
+    return delegate;
+  }
+
+  public GuardedRequestMarshaller<ReqT> getMarshaller() {
+    return marshaller;
+  }
+
+  @Override
+  public ServerCall.Listener<ReqT> startCall(ServerCall<ReqT, RespT> call, Metadata headers) {
+    ServerCall.Listener<ReqT> next = delegate.startCall(call, headers);
+    return new SimpleForwardingServerCallListener<ReqT>(next) {
+      // Listener callbacks of one call are serialized by grpc.
+      private boolean rejected;
+
+      @Override
+      public void onMessage(ReqT message) {
+        if (rejected) {
+          return;
+        }
+        Reason reason = marshaller.rejectionReason(message);
+        if (reason != null) {
+          rejected = true;
+          String method = call.getMethodDescriptor().getFullMethodName();
+          if (logger.isDebugEnabled()) {
+            logger.debug("gRPC {} from {} rejected by admission check: {}", method,
+                call.getAttributes().get(Grpc.TRANSPORT_ATTR_REMOTE_ADDR), reason);
+          }
+          Metrics.counterInc(MetricKeys.Counter.GRPC_ADMISSION_REJECT, 1, method, reason.name());
+          // The same status grpc gives an uncompressed message over the size limit.
+          Status status = reason == Reason.TOO_LARGE
+              ? Status.RESOURCE_EXHAUSTED : Status.INVALID_ARGUMENT;
+          call.close(status.withDescription(marshaller.rejectionMessage()), new Metadata());
+          return;
+        }
+        super.onMessage(message);
+      }
+
+      @Override
+      public void onHalfClose() {
+        if (!rejected) {
+          super.onHalfClose();
+        }
+      }
+
+      @Override
+      public void onReady() {
+        if (!rejected) {
+          super.onReady();
+        }
+      }
+    };
+  }
+}