xds: fail closed when ext_authz response processing throws

xds/src/main/java/io/grpc/xds/internal/extauthz/AuthzCallbackObserver.java

@@ -74,12 +74,17 @@ final class AuthzCallbackObserver<ReqT, RespT> implements StreamObserver<CheckRe
 
   @Override
   public void onNext(CheckResponse value) {
-    // Note: This implementation is currently exception-safe.
-    //
-    // TODO(sauravz): Revisit hardening if this invariant changes in the future.
-    // If an unhandled RuntimeException escapes onNext(), gRPC cancels the stream and
-    // invokes onError(). Under failure_mode_allow: true, this causes the call to fail
-    // open, which could inadvertently permit an unauthorized request.
+    try {
+      handleCheckResponse(value);
+    } catch (RuntimeException e) {
+      // A processing failure is not an authz communication failure, so failure_mode_allow
+      // must not apply here.
+      setCallAndDrain(new FailingClientCall<>(
+          Status.INTERNAL.withCause(e).withDescription("Failed to process authz response")));
+    }
+  }
+
+  private void handleCheckResponse(CheckResponse value) {
     AuthzResponse authzResponse = responseHandler.handleResponse(value);
     if (authzResponse.decision() == AuthzResponse.Decision.ALLOW) {
       ClientCall<ReqT, RespT> delegate = next.newCall(method, callOptions);

xds/src/main/java/io/grpc/xds/internal/extauthz/CheckResponseHandler.java

@@ -121,8 +121,10 @@ private ImmutableList<HeaderValueOption> convertHeaders(
       } else {
         internalHeader = HeaderValue.create(key, header.getValue());
       }
+      // TODO(sauravzg): Confirm failing the RPC is correct for gRPC-owned headers.
       if (HeaderValueValidationUtils.isDisallowed(internalHeader)) {
-        continue;
+        throw new HeaderMutationDisallowedException(
+            "Header mutation disallowed for gRPC-owned key: " + key);
       }
       HeaderValueOption.HeaderAppendAction action;
       switch (optionProto.getAppendAction()) {

xds/src/test/java/io/grpc/xds/internal/extauthz/AuthzCallbackObserverTest.java

@@ -585,8 +585,8 @@ public void allow_withHeadersToRemoveOnly_backendReceivesMutatedHeaders() {
     assertThat(capturedBackendMessage).isEqualTo(request);
   }
 
-  @Test(expected = IllegalArgumentException.class)
-  public void deny_withMissingStatus_throwsIllegalArgumentException() {
+  @Test
+  public void deny_withMissingStatus_failsCallWithInternal() {
     CheckResponseHandler mockHandler = mock(CheckResponseHandler.class);
     AuthzResponse fakeAuthzResponse = new AuthzResponse() {
       @Override
@@ -621,8 +621,13 @@ public HeaderMutations responseHeaderMutations() {
             CallOptions.DEFAULT,
             MoreExecutors.directExecutor(),
             mockHandler, failClosedConfig(), authzCtx);
+    CapturingListener<SimpleResponse> listener = new CapturingListener<>();
+    delayedCall.start(listener, new Metadata());
+    delayedCall.request(1);
 
     observer.onNext(CheckResponse.getDefaultInstance());
+
+    assertThat(listener.getCloseStatus().getCode()).isEqualTo(Status.Code.INTERNAL);
   }
 
   @Test
@@ -693,6 +698,55 @@ public void allow_whenDelayedCallNotStarted_setCallReturnsNull() {
   }
 
 
+  @Test
+  public void deny_withMalformedHeader_failOpen_doesNotReachBackend() {
+    capturedBackendHeaders = null;
+    capturedBackendMessage = null;
+    doAnswer(invocation -> {
+      StreamObserver<CheckResponse> obs = invocation.getArgument(1);
+      obs.onNext(CheckResponse.newBuilder()
+          .setStatus(com.google.rpc.Status.newBuilder()
+              .setCode(com.google.rpc.Code.PERMISSION_DENIED_VALUE))
+          .setDeniedResponse(DeniedHttpResponse.newBuilder()
+              .setStatus(HttpStatus.newBuilder().setCode(StatusCode.Forbidden))
+              .addHeaders(HeaderValueOption.newBuilder()
+                  .setHeader(HeaderValue.newBuilder().setKey("x-deny-reason")
+                      .setValue("policy\nviolation"))))
+          .build());
+      obs.onCompleted();
+      return null;
+    }).when(authzService).check(any(), any());
+
+    TestDelayedCall<SimpleRequest, SimpleResponse> delayedCall =
+        new TestDelayedCall<>(MoreExecutors.directExecutor(), scheduler, null);
+    Context.CancellableContext authzCtx = Context.current().withCancellation();
+    AuthzCallbackObserver<SimpleRequest, SimpleResponse> observer =
+        new AuthzCallbackObserver<>(
+            delayedCall, channel,
+            SimpleServiceGrpc.getUnaryRpcMethod(),
+            CallOptions.DEFAULT,
+            MoreExecutors.directExecutor(),
+            responseHandler,
+            failOpenConfig(/*headerAdd=*/false), authzCtx);
+
+    SimpleRequest request =
+        SimpleRequest.newBuilder().setRequestMessage("malformed-header-payload").build();
+    CapturingListener<SimpleResponse> listener = new CapturingListener<>();
+    delayedCall.start(listener, new Metadata());
+    delayedCall.sendMessage(request);
+    delayedCall.halfClose();
+    delayedCall.request(1);
+
+    authzCtx.run(() -> {
+      AuthorizationGrpc.newStub(channel)
+          .check(CheckRequest.getDefaultInstance(), observer);
+    });
+
+    assertThat(capturedBackendHeaders).isNull();
+    assertThat(capturedBackendMessage).isNull();
+    assertThat(listener.getCloseStatus().getCode()).isEqualTo(Status.Code.INTERNAL);
+  }
+
   private static final class TestDelayedCall<ReqT, RespT>
       extends DelayedClientCall<ReqT, RespT> {
     TestDelayedCall(

xds/src/test/java/io/grpc/xds/internal/extauthz/CheckResponseHandlerTest.java

@@ -213,12 +213,10 @@ public void handleResponse_okWithDisallowedMutation() throws HeaderMutationDisal
   }
 
   @Test
-  public void handleResponse_ok_binaryHeadersPreservedAndDisallowedHeadersDropped() {
+  public void handleResponse_ok_binaryHeadersPreserved() {
     HeaderValueOption binaryOption =
         HeaderValueOption.newBuilder().setHeader(HeaderValue.newBuilder().setKey("test-bin")
             .setRawValue(com.google.protobuf.ByteString.copyFromUtf8("test"))).build();
-    HeaderValueOption disallowedOption = HeaderValueOption.newBuilder()
-        .setHeader(HeaderValue.newBuilder().setKey("host").setValue("disallowed")).build();
 
     io.grpc.xds.internal.headermutations.HeaderValueOption expectedBinaryOption =
         io.grpc.xds.internal.headermutations.HeaderValueOption.create(
@@ -228,8 +226,7 @@ public void handleResponse_ok_binaryHeadersPreservedAndDisallowedHeadersDropped(
 
     CheckResponse checkResponse = CheckResponse.newBuilder()
         .setStatus(com.google.rpc.Status.newBuilder().setCode(Code.OK_VALUE).build())
-        .setOkResponse(OkHttpResponse.newBuilder().addHeaders(binaryOption)
-            .addHeaders(disallowedOption).build())
+        .setOkResponse(OkHttpResponse.newBuilder().addHeaders(binaryOption).build())
         .build();
     AuthzResponse authzResponse = responseHandler.handleResponse(checkResponse);
 
@@ -240,6 +237,23 @@ public void handleResponse_ok_binaryHeadersPreservedAndDisallowedHeadersDropped(
     assertThat(authzResponse.requestHeaderMutations()).isEqualTo(expectedRequestMutations);
   }
 
+  @Test
+  public void handleResponse_ok_grpcOwnedHeader_deniesCall() {
+    HeaderValueOption disallowedOption = HeaderValueOption.newBuilder()
+        .setHeader(HeaderValue.newBuilder().setKey("host").setValue("disallowed")).build();
+
+    CheckResponse checkResponse = CheckResponse.newBuilder()
+        .setStatus(com.google.rpc.Status.newBuilder().setCode(Code.OK_VALUE).build())
+        .setOkResponse(OkHttpResponse.newBuilder().addHeaders(disallowedOption).build())
+        .build();
+    AuthzResponse authzResponse = responseHandler.handleResponse(checkResponse);
+
+    assertThat(authzResponse.decision()).isEqualTo(Decision.DENY);
+    assertThat(authzResponse.status().get().getCode()).isEqualTo(Status.INTERNAL.getCode());
+    assertThat(authzResponse.status().get().getDescription())
+        .contains("Header mutation disallowed for gRPC-owned key: host");
+  }
+
   @Test
   public void handleResponse_ok_invalidAppendAction_deniesCall() {
     HeaderValueOption invalidActionOption = HeaderValueOption.newBuilder()