xds: Enforce unconditional validation for ext_proc header values and remove HeaderValue.isValid

Per gRFC A93, the `disallow_is_error` configuration applies strictly to mutation operation rules (attempting to modify disallowed header keys). Conversely, malformed or invalid header values received from `ext_proc` represent protocol/response errors and must unconditionally throw an `IllegalArgumentException`.

Key changes:
- Remove `isValid` field and `createInvalid` factory from `HeaderValue` struct.
- Add unconditional `validateHeaderKey` and `validateHeaderValue` checks throwing `IllegalArgumentException` in `HeaderValueValidationUtils`.
- Restrict `HeaderValueValidationUtils.isDisallowed` exclusively to mutation operation rule checks (`host`, `:authority`, `:path`, `grpc-*`, uppercase keys).
- Update `ExternalProcessorClientInterceptor` and `GrpcServiceConfigParser` to enforce unconditional validation.
- Update unit tests across `grpc-xds` to verify unconditional failure on malformed headers even when `disallow_is_error` is false.

And other minor review comments.

Author: kannanjgithub

xds/src/main/java/io/grpc/xds/ExternalProcessorClientInterceptor.java

@@ -623,36 +623,40 @@ private void applyHeaderMutations(Metadata metadata,
       for (io.envoyproxy.envoy.config.core.v3.HeaderValueOption protoOption
           : mutation.getSetHeadersList()) {
         io.envoyproxy.envoy.config.core.v3.HeaderValue protoHeader = protoOption.getHeader();
-        HeaderValue headerValue;
-        
+        String key = protoHeader.getKey();
+        HeaderValueValidationUtils.validateHeaderKey(key);
+
         ByteString rawBytes = protoHeader.getRawValue();
         if (rawBytes.isEmpty()) {
           rawBytes = ByteString.copyFromUtf8(protoHeader.getValue());
         }
 
         if (rawBytes.size() > HeaderValueValidationUtils.MAX_HEADER_LENGTH) {
-          headerValue = HeaderValue.createInvalid(protoHeader.getKey());
-        } else if (protoHeader.getKey().endsWith(Metadata.BINARY_HEADER_SUFFIX)) {
-          try {
-            byte[] decodedBytes = BaseEncoding.base64().decode(rawBytes.toStringUtf8());
-            headerValue = HeaderValue.create(
-                protoHeader.getKey(), ByteString.copyFrom(decodedBytes));
-          } catch (IllegalArgumentException e) {
-            // Mark as invalid so HeaderMutationFilter can either silently ignore it or
-            // throw an exception based on the disallow_is_error configuration.
-            headerValue = HeaderValue.createInvalid(protoHeader.getKey());
-          }
+          throw new IllegalArgumentException(
+              "Header value length exceeds maximum allowed length: " + rawBytes.size());
+        }
+
+        HeaderValue headerValue;
+        if (key.endsWith(Metadata.BINARY_HEADER_SUFFIX)) {
+          byte[] decodedBytes = BaseEncoding.base64().decode(rawBytes.toStringUtf8());
+          headerValue = HeaderValue.create(key, ByteString.copyFrom(decodedBytes));
         } else {
-          headerValue = HeaderValue.create(protoHeader.getKey(), rawBytes.toStringUtf8());
+          headerValue = HeaderValue.create(key, rawBytes.toStringUtf8());
         }
         headersToModify.add(HeaderValueOption.create(
             headerValue,
             HeaderValueOption.HeaderAppendAction.valueOf(protoOption.getAppendAction().name())));
       }
 
+      ImmutableList.Builder<String> headersToRemove = ImmutableList.builder();
+      for (String headerToRemove : mutation.getRemoveHeadersList()) {
+        HeaderValueValidationUtils.validateHeaderKey(headerToRemove);
+        headersToRemove.add(headerToRemove);
+      }
+
       HeaderMutations mutations = HeaderMutations.create(
           headersToModify.build(),
-          ImmutableList.copyOf(mutation.getRemoveHeadersList()));
+          headersToRemove.build());
 
       HeaderMutations filteredMutations = mutationFilter.filter(mutations);
       mutator.applyMutations(filteredMutations, metadata);

xds/src/main/java/io/grpc/xds/ExternalProcessorFilter.java

@@ -32,6 +32,7 @@
 import io.envoyproxy.envoy.extensions.filters.http.ext_proc.v3.HeaderForwardingRules;
 import io.envoyproxy.envoy.extensions.filters.http.ext_proc.v3.ProcessingMode;
 import io.grpc.ClientInterceptor;
+import io.grpc.internal.GrpcUtil;
 import io.grpc.xds.Filter.FilterConfigParseContext;
 import io.grpc.xds.Filter.FilterContext;
 import io.grpc.xds.internal.MatcherParser;
@@ -82,7 +83,7 @@ public String[] typeUrls() {
 
     @Override
     public boolean isClientFilter() {
-      return true;
+      return GrpcUtil.getFlag("GRPC_EXPERIMENTAL_XDS_EXT_PROC_ON_CLIENT", false);
     }
 
     @Override

xds/src/main/java/io/grpc/xds/FilterRegistry.java

@@ -17,7 +17,6 @@
 package io.grpc.xds;
 
 import com.google.common.annotations.VisibleForTesting;
-import io.grpc.internal.GrpcUtil;
 import java.util.HashMap;
 import java.util.Map;
 import javax.annotation.Nullable;
@@ -39,10 +38,8 @@ static synchronized FilterRegistry getDefaultRegistry() {
               new FaultFilter.Provider(),
               new RouterFilter.Provider(),
               new RbacFilter.Provider(),
-              new GcpAuthenticationFilter.Provider());
-      if (GrpcUtil.getFlag("GRPC_EXPERIMENTAL_XDS_EXT_PROC_ON_CLIENT", false)) {
-        instance.register(new ExternalProcessorFilter.Provider());
-      }
+              new GcpAuthenticationFilter.Provider(),
+              new ExternalProcessorFilter.Provider());
     }
     return instance;
   }

xds/src/main/java/io/grpc/xds/GrpcServiceConfigParser.java

@@ -98,10 +98,14 @@ public static GrpcServiceConfig parse(GrpcService grpcServiceProto,
         .getInitialMetadataList()) {
       String key = header.getKey();
       HeaderValue headerValue;
-      if (key.endsWith(Metadata.BINARY_HEADER_SUFFIX)) {
-        headerValue = HeaderValue.create(key, header.getRawValue());
-      } else {
-        headerValue = HeaderValue.create(key, header.getValue());
+      try {
+        if (key.endsWith(Metadata.BINARY_HEADER_SUFFIX)) {
+          headerValue = HeaderValue.create(key, header.getRawValue());
+        } else {
+          headerValue = HeaderValue.create(key, header.getValue());
+        }
+      } catch (IllegalArgumentException e) {
+        throw new GrpcServiceParseException("Invalid initial metadata header: " + key, e);
       }
       if (HeaderValueValidationUtils.isDisallowed(headerValue)) {
         throw new GrpcServiceParseException("Invalid initial metadata header: " + key);

xds/src/main/java/io/grpc/xds/XdsNameResolver.java

@@ -739,10 +739,6 @@ private void updateActiveFilters(@Nullable List<NamedFilterConfig> filterConfigs
       Set<String> filtersToShutdown = new HashSet<>(activeFilters.keySet());
       for (NamedFilterConfig namedFilter : filterConfigs) {
         String typeUrl = namedFilter.filterConfig.typeUrl();
-        if (typeUrl.equals(ExternalProcessorFilter.TYPE_URL)
-            && !GrpcUtil.getFlag("GRPC_EXPERIMENTAL_XDS_EXT_PROC_ON_CLIENT", false)) {
-          continue;
-        }
         String filterKey = namedFilter.filterStateKey();
 
         Filter.Provider provider = filterRegistry.get(typeUrl);
@@ -893,12 +889,6 @@ private ClientInterceptor createFilters(
 
       ImmutableList.Builder<ClientInterceptor> filterInterceptors = ImmutableList.builder();
       for (NamedFilterConfig namedFilter : filterConfigs) {
-        String typeUrl = namedFilter.filterConfig.typeUrl();
-        if (typeUrl.equals(ExternalProcessorFilter.TYPE_URL)) {
-          if (!GrpcUtil.getFlag("GRPC_EXPERIMENTAL_XDS_EXT_PROC_ON_CLIENT", false)) {
-            continue;
-          }
-        }
         String name = namedFilter.name;
         FilterConfig config = namedFilter.filterConfig;
         FilterConfig overrideConfig = selectedOverrideConfigs.get(name);

xds/src/main/java/io/grpc/xds/XdsNameResolverProvider.java

@@ -63,7 +63,7 @@ public XdsNameResolverProvider() {
   }
 
   private XdsNameResolverProvider(String scheme,
-      @Nullable Map<String, ?> bootstrapOverride) {
+                                 @Nullable Map<String, ?> bootstrapOverride) {
     this.scheme = checkNotNull(scheme, "scheme");
     this.bootstrapOverride = bootstrapOverride;
   }
@@ -73,7 +73,7 @@ private XdsNameResolverProvider(String scheme,
    * and bootstrap.
    */
   public static XdsNameResolverProvider createForTest(String scheme,
-      @Nullable Map<String, ?> bootstrapOverride) {
+                                                      @Nullable Map<String, ?> bootstrapOverride) {
     return new XdsNameResolverProvider(scheme, bootstrapOverride);
   }
 

xds/src/main/java/io/grpc/xds/internal/grpcservice/ChannelCredsConfig.java

@@ -28,10 +28,12 @@
 public abstract class HeaderValue {
 
   public static HeaderValue create(String key, String value) {
+    HeaderValueValidationUtils.validateHeaderValue(key, value);
     return new AutoValue_HeaderValue(key, Optional.of(value), Optional.empty(), true);
   }
 
   public static HeaderValue create(String key, ByteString rawValue) {
+    HeaderValueValidationUtils.validateHeaderValue(key, rawValue);
     return new AutoValue_HeaderValue(key, Optional.empty(), Optional.of(rawValue), true);
   }
 

xds/src/main/java/io/grpc/xds/internal/grpcservice/HeaderValue.java

@@ -28,7 +28,47 @@ public final class HeaderValueValidationUtils {
   private HeaderValueValidationUtils() {}
 
   /**
-   * Returns true if the header key is disallowed for mutations or validation.
+   * Validates that the header key is non-empty and within allowed length.
+   * Throws {@link IllegalArgumentException} if invalid.
+   */
+  public static void validateHeaderKey(String key) {
+    if (key == null || key.isEmpty() || key.length() > MAX_HEADER_LENGTH) {
+      throw new IllegalArgumentException("Invalid header key: " + key);
+    }
+  }
+
+  /**
+   * Validates that the header value is within allowed length and contains valid ASCII characters.
+   * Throws {@link IllegalArgumentException} if invalid.
+   */
+  public static void validateHeaderValue(String key, String value) {
+    validateHeaderKey(key);
+    if (value == null || value.length() > MAX_HEADER_LENGTH) {
+      throw new IllegalArgumentException("Header value length exceeds maximum allowed length");
+    }
+    if (!key.endsWith("-bin") && !isValidAsciiHeaderValue(value)) {
+      throw new IllegalArgumentException(
+          "Invalid ASCII characters in header value for key: " + key);
+    }
+  }
+
+  /**
+   * Validates that the raw header value is within allowed length and contains valid ASCII
+   * characters. Throws {@link IllegalArgumentException} if invalid.
+   */
+  public static void validateHeaderValue(String key, ByteString rawValue) {
+    validateHeaderKey(key);
+    if (rawValue == null || rawValue.size() > MAX_HEADER_LENGTH) {
+      throw new IllegalArgumentException("Header value length exceeds maximum allowed length");
+    }
+    if (!key.endsWith("-bin") && !isValidAsciiHeaderValue(rawValue.toStringUtf8())) {
+      throw new IllegalArgumentException(
+          "Invalid ASCII characters in header value for key: " + key);
+    }
+  }
+
+  /**
+   * Returns true if the header key is disallowed for mutations.
    *
    * @param key The header key (e.g., "content-type")
    */
@@ -49,30 +89,12 @@ public static boolean isDisallowed(String key) {
   }
 
   /**
-   * Returns true if the header value is disallowed.
+   * Returns true if the header is disallowed for mutations.
    *
-   * @param header The HeaderValue containing key and values
+   * @param header The HeaderValue
    */
   public static boolean isDisallowed(HeaderValue header) {
-    if (!header.isValid()) {
-      return true;
-    }
-    if (isDisallowed(header.key())) {
-      return true;
-    }
-    if (header.value().isPresent()) {
-      String val = header.value().get();
-      if (!header.key().endsWith("-bin") && !isValidAsciiHeaderValue(val)) {
-        return true;
-      }
-    }
-    if (header.rawValue().isPresent()) {
-      ByteString rawVal = header.rawValue().get();
-      if (!header.key().endsWith("-bin") && !isValidAsciiHeaderValue(rawVal.toStringUtf8())) {
-        return true;
-      }
-    }
-    return false;
+    return isDisallowed(header.key());
   }
 
   /**