xds: fail closed when ext_authz response processing throws

CheckResponseHandler.handleResponse() only catches the checked
HeaderMutationDisallowedException. An unchecked exception raised while
processing a CheckResponse -- for example HeaderValue.create() rejecting a
header value that is not valid ASCII -- escaped AuthzCallbackObserver.onNext().
gRPC then cancelled the stream and invoked onError(), where failure_mode_allow
sent the request to the backend, turning an explicit PERMISSION_DENIED into an
ALLOW.

failure_mode_allow covers the authorization service being unreachable or
returning an error. It does not cover a failure to process a response that the
service successfully returned, so onNext() now handles its own failures and
fails the call with INTERNAL instead of letting them reach onError().

Also fail the RPC, rather than silently dropping the mutation, when the authz
server attempts to mutate a gRPC-owned header.

Author: sauravzg

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()