core: Do not leak server state when application callbacks throw exceptions

core/src/main/java/io/grpc/ServerInterceptors.java

@@ -17,6 +17,8 @@
 package io.grpc;
 
 import com.google.common.base.Preconditions;
+import io.grpc.ForwardingServerCallListener.SimpleForwardingServerCallListener;
+import io.grpc.internal.SynchronizedServerCall;
 import java.io.BufferedInputStream;
 import java.io.InputStream;
 import java.util.ArrayList;
@@ -288,4 +290,81 @@ public void onMessage(WReqT message) {
       }
     };
   }
+
+  /**
+   * A class that intercepts uncaught exceptions of type {@link StatusRuntimeException} and handles
+   * them by closing the {@link ServerCall}, and transmitting the exception's details to the client.
+   *
+   * <p>Without this interceptor, gRPC will strip all details and close the {@link ServerCall} with
+   * a generic {@link Status#UNKNOWN} code.
+   *
+   * <p>Security warning: the {@link Status} and {@link Metadata} may contain sensitive server-side
+   * state information, and generally should not be sent to clients. Only install this interceptor
+   * if all clients are trusted.
+   */
+  @ExperimentalApi("https://github.com/grpc/grpc-java/issues/2189")
+  public static final ServerInterceptor STATUSRUNTIMEXCEPTION_TRANSMITTING_INTERCEPTOR =
+      new ServerInterceptor() {
+        @Override
+        public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall(
+            ServerCall<ReqT, RespT> call, Metadata headers, ServerCallHandler<ReqT, RespT> next) {
+          final ServerCall<ReqT, RespT> syncServerCall =
+              new SynchronizedServerCall<ReqT, RespT>(call);
+          ServerCall.Listener<ReqT> listener = next.startCall(syncServerCall, headers);
+          return new SimpleForwardingServerCallListener<ReqT>(listener) {
+            @Override
+            public void onMessage(ReqT message) {
+              try {
+                super.onMessage(message);
+              } catch (StatusRuntimeException t) {
+                closeWithException(t);
+              }
+            }
+
+            @Override
+            public void onHalfClose() {
+              try {
+                super.onHalfClose();
+              } catch (StatusRuntimeException t) {
+                closeWithException(t);
+              }
+            }
+
+            @Override
+            public void onCancel() {
+              try {
+                super.onCancel();
+              } catch (StatusRuntimeException t) {
+                closeWithException(t);
+              }
+            }
+
+            @Override
+            public void onComplete() {
+              try {
+                super.onComplete();
+              } catch (StatusRuntimeException t) {
+                closeWithException(t);
+              }
+            }
+
+            @Override
+            public void onReady() {
+              try {
+                super.onReady();
+              } catch (StatusRuntimeException t) {
+                closeWithException(t);
+              }
+            }
+
+            private void closeWithException(StatusRuntimeException t) {
+              Metadata metadata = Status.trailersFromThrowable(t);
+              if (metadata == null) {
+                metadata = new Metadata();
+              }
+              syncServerCall.close(Status.fromThrowable(t), metadata);
+            }
+          };
+        }
+      };
 }

core/src/main/java/io/grpc/internal/ServerImpl.java

@@ -532,9 +532,9 @@ void setListener(ServerStreamListener listener) {
     /**
      * Like {@link ServerCall#close(Status, Metadata)}, but thread-safe for internal use.
      */
-    private void internalClose(Status status, Metadata trailers) {
+    private void internalClose() {
       // TODO(ejona86): this is not thread-safe :)
-      stream.close(status, trailers);
+      stream.close(Status.UNKNOWN, new Metadata());
     }
 
     @Override
@@ -545,10 +545,10 @@ public void runInContext() {
           try {
             getListener().messageRead(message);
           } catch (RuntimeException e) {
-            internalClose(Status.fromThrowable(e), new Metadata());
+            internalClose();
             throw e;
           } catch (Error e) {
-            internalClose(Status.fromThrowable(e), new Metadata());
+            internalClose();
             throw e;
           }
         }
@@ -563,10 +563,10 @@ public void runInContext() {
           try {
             getListener().halfClosed();
           } catch (RuntimeException e) {
-            internalClose(Status.fromThrowable(e), new Metadata());
+            internalClose();
             throw e;
           } catch (Error e) {
-            internalClose(Status.fromThrowable(e), new Metadata());
+            internalClose();
             throw e;
           }
         }
@@ -601,10 +601,10 @@ public void runInContext() {
           try {
             getListener().onReady();
           } catch (RuntimeException e) {
-            internalClose(Status.fromThrowable(e), new Metadata());
+            internalClose();
             throw e;
           } catch (Error e) {
-            internalClose(Status.fromThrowable(e), new Metadata());
+            internalClose();
             throw e;
           }
         }

core/src/main/java/io/grpc/internal/SynchronizedServerCall.java

@@ -0,0 +1,87 @@
+/*
+ * Copyright 2017, gRPC Authors All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.grpc.internal;
+
+import io.grpc.Attributes;
+import io.grpc.ForwardingServerCall;
+import io.grpc.Metadata;
+import io.grpc.ServerCall;
+import io.grpc.Status;
+import javax.annotation.Nullable;
+
+/**
+ * A {@link ServerCall} that wraps around a non thread safe delegate and provides thread safe
+ * access.
+ */
+public class SynchronizedServerCall<ReqT, RespT> extends
+    ForwardingServerCall.SimpleForwardingServerCall<ReqT, RespT> {
+
+  public SynchronizedServerCall(ServerCall<ReqT, RespT> delegate) {
+    super(delegate);
+  }
+
+  @Override
+  public synchronized void sendMessage(RespT message) {
+    super.sendMessage(message);
+  }
+
+  @Override
+  public synchronized void request(int numMessages) {
+    super.request(numMessages);
+  }
+
+  @Override
+  public synchronized void sendHeaders(Metadata headers) {
+    super.sendHeaders(headers);
+  }
+
+  @Override
+  public synchronized boolean isReady() {
+    return super.isReady();
+  }
+
+  @Override
+  public synchronized void close(Status status, Metadata trailers) {
+    super.close(status, trailers);
+  }
+
+  @Override
+  public synchronized boolean isCancelled() {
+    return super.isCancelled();
+  }
+
+  @Override
+  public synchronized void setMessageCompression(boolean enabled) {
+    super.setMessageCompression(enabled);
+  }
+
+  @Override
+  public synchronized void setCompression(String compressor) {
+    super.setCompression(compressor);
+  }
+
+  @Override
+  public synchronized Attributes getAttributes() {
+    return super.getAttributes();
+  }
+
+  @Nullable
+  @Override
+  public synchronized String getAuthority() {
+    return super.getAuthority();
+  }
+}

core/src/test/java/io/grpc/ServerInterceptorsTest.java

@@ -20,7 +20,10 @@
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertSame;
 import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+import static org.mockito.Matchers.any;
 import static org.mockito.Matchers.same;
+import static org.mockito.Mockito.doThrow;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
@@ -68,6 +71,8 @@ public class ServerInterceptorsTest {
 
   private final Metadata headers = new Metadata();
 
+  private boolean allowListenerInteraction;
+
   /** Set up for test. */
   @Before
   public void setUp() {
@@ -85,14 +90,17 @@ public void setUp() {
 
     serviceDefinition = ServerServiceDefinition.builder(new ServiceDescriptor("basic", flowMethod))
         .addMethod(flowMethod, handler).build();
+    allowListenerInteraction = false;
   }
 
   /** Final checks for all tests. */
   @After
   public void makeSureExpectedMocksUnused() {
     verifyZeroInteractions(requestMarshaller);
     verifyZeroInteractions(responseMarshaller);
-    verifyZeroInteractions(listener);
+    if (!allowListenerInteraction) {
+      verifyZeroInteractions(listener);
+    }
   }
 
   @Test(expected = NullPointerException.class)
@@ -407,6 +415,36 @@ public void onMessage(ReqT message) {
         order);
   }
 
+  @Test
+  public void statusRuntimeExceptionTransmittingInterceptor() {
+    allowListenerInteraction = true;
+    final Status expectedStatus = Status.UNAVAILABLE;
+    final Metadata expectedMetadata = new Metadata();
+    call = Mockito.spy(call);
+    final StatusRuntimeException exception =
+        new StatusRuntimeException(expectedStatus, expectedMetadata);
+    doThrow(exception).when(listener).onMessage(any(String.class));
+    doThrow(exception).when(listener).onCancel();
+    doThrow(exception).when(listener).onComplete();
+    doThrow(exception).when(listener).onHalfClose();
+    doThrow(exception).when(listener).onReady();
+
+    ServerServiceDefinition intercepted = ServerInterceptors.intercept(
+        serviceDefinition,
+        Arrays.asList(ServerInterceptors.STATUSRUNTIMEXCEPTION_TRANSMITTING_INTERCEPTOR));
+    try {
+      getSoleMethod(intercepted).getServerCallHandler().startCall(call, headers).onMessage("hello");
+      getSoleMethod(intercepted).getServerCallHandler().startCall(call, headers).onCancel();
+      getSoleMethod(intercepted).getServerCallHandler().startCall(call, headers).onComplete();
+      getSoleMethod(intercepted).getServerCallHandler().startCall(call, headers).onHalfClose();
+      getSoleMethod(intercepted).getServerCallHandler().startCall(call, headers).onReady();
+    } catch (Throwable t) {
+      fail("The interceptor should have handled the error by directly closing the ServerCall, "
+          + "and should not propagate it to the method's caller.");
+    }
+    verify(call, times(5)).close(same(expectedStatus), same(expectedMetadata));
+  }
+
   @SuppressWarnings("unchecked")
   private static ServerMethodDefinition<String, Integer> getSoleMethod(
       ServerServiceDefinition serviceDef) {

core/src/test/java/io/grpc/internal/ServerImplTest.java

@@ -145,6 +145,8 @@ public <ReqT, RespT> Context filterContext(Context context) {
   @Captor
   private ArgumentCaptor<Status> statusCaptor;
   @Captor
+  private ArgumentCaptor<Metadata> metadataCaptor;
+  @Captor
   private ArgumentCaptor<ServerStreamListener> streamListenerCaptor;
 
   @Mock
@@ -958,8 +960,7 @@ public void messageRead_errorCancelsCall() throws Exception {
       fail("Expected exception");
     } catch (Throwable t) {
       assertSame(expectedT, t);
-      verify(stream).close(statusCaptor.capture(), any(Metadata.class));
-      assertSame(expectedT, statusCaptor.getValue().getCause());
+      ensureServerStateNotLeaked();
     }
   }
 
@@ -983,8 +984,7 @@ public void messageRead_runtimeExceptionCancelsCall() throws Exception {
       fail("Expected exception");
     } catch (Throwable t) {
       assertSame(expectedT, t);
-      verify(stream).close(statusCaptor.capture(), any(Metadata.class));
-      assertSame(expectedT, statusCaptor.getValue().getCause());
+      ensureServerStateNotLeaked();
     }
   }
 
@@ -1007,8 +1007,7 @@ public void halfClosed_errorCancelsCall() {
       fail("Expected exception");
     } catch (Throwable t) {
       assertSame(expectedT, t);
-      verify(stream).close(statusCaptor.capture(), any(Metadata.class));
-      assertSame(expectedT, statusCaptor.getValue().getCause());
+      ensureServerStateNotLeaked();
     }
   }
 
@@ -1031,8 +1030,7 @@ public void halfClosed_runtimeExceptionCancelsCall() {
       fail("Expected exception");
     } catch (Throwable t) {
       assertSame(expectedT, t);
-      verify(stream).close(statusCaptor.capture(), any(Metadata.class));
-      assertSame(expectedT, statusCaptor.getValue().getCause());
+      ensureServerStateNotLeaked();
     }
   }
 
@@ -1055,8 +1053,7 @@ public void onReady_errorCancelsCall() {
       fail("Expected exception");
     } catch (Throwable t) {
       assertSame(expectedT, t);
-      verify(stream).close(statusCaptor.capture(), any(Metadata.class));
-      assertSame(expectedT, statusCaptor.getValue().getCause());
+      ensureServerStateNotLeaked();
     }
   }
 
@@ -1079,8 +1076,7 @@ public void onReady_runtimeExceptionCancelsCall() {
       fail("Expected exception");
     } catch (Throwable t) {
       assertSame(expectedT, t);
-      verify(stream).close(statusCaptor.capture(), any(Metadata.class));
-      assertSame(expectedT, statusCaptor.getValue().getCause());
+      ensureServerStateNotLeaked();
     }
   }
 
@@ -1114,6 +1110,13 @@ private void verifyExecutorsReturned() {
     verifyNoMoreInteractions(timerPool);
   }
 
+  private void ensureServerStateNotLeaked() {
+    verify(stream).close(statusCaptor.capture(), metadataCaptor.capture());
+    assertEquals(Status.UNKNOWN, statusCaptor.getValue());
+    assertNull(statusCaptor.getValue().getCause());
+    assertTrue(metadataCaptor.getValue().keys().isEmpty());
+  }
+
   private static class SimpleServer implements io.grpc.internal.InternalServer {
     ServerListener listener;