kafka: close connection when rejectable request appears (#36979)

Commit Message: kafka: close connection when rejectable request appears
Additional Description: provides
https://github.com/envoyproxy/envoy/issues/36978 by making the broker
filter check the requests' API key against allow/denylists in the config
- if we find a request to reject, we just close the connection. There is
also a bonus: this sets up a framework for "_close connection if we
don't like the request_" that we can expand upon later.
Risk Level: Low
Testing: automated (no filtering) + manual (filtering)
Docs Changes: Kafka broker .rst
Release Notes: n/a
Platform Specific Features: n/a

---------

Signed-off-by: Adam Kotwasinski <adam.kotwasinski@gmail.com>

Mirrored from https://github.com/envoyproxy/envoy @ c30d173544a6ac6646201f8766eb851c7c9e0bc6

contrib/envoy/extensions/filters/network/kafka_broker/v3/kafka_broker.proto

@@ -15,7 +15,7 @@ option (udpa.annotations.file_status).package_version_status = ACTIVE;
 // [#protodoc-title: Kafka Broker]
 // Kafka Broker :ref:`configuration overview <config_network_filters_kafka_broker>`.
 // [#extension: envoy.filters.network.kafka_broker]
-
+// [#next-free-field: 6]
 message KafkaBroker {
   option (udpa.annotations.versioning).previous_message_type =
       "envoy.config.filter.network.kafka_broker.v2alpha1.KafkaBroker";
@@ -39,6 +39,16 @@ message KafkaBroker {
     // Broker address rewrite rules that match by broker ID.
     IdBasedBrokerRewriteSpec id_based_broker_address_rewrite_spec = 3;
   }
+
+  // Optional list of allowed Kafka API keys. Only requests with provided API keys will be
+  // routed, otherwise the connection will be closed. No effect if empty.
+  repeated uint32 api_keys_allowed = 4
+      [(validate.rules).repeated = {items {uint32 {lte: 32767 gte: 0}}}];
+
+  // Optional list of denied Kafka API keys. Requests with API keys matching this list will have
+  // the connection closed. No effect if empty.
+  repeated uint32 api_keys_denied = 5
+      [(validate.rules).repeated = {items {uint32 {lte: 32767 gte: 0}}}];
 }
 
 // Collection of rules matching by broker ID.

envoy/config/filter/network/kafka_broker/v2alpha1/kafka_broker.proto

@@ -17,7 +17,7 @@ option (udpa.annotations.file_status).package_version_status = FROZEN;
 // [#protodoc-title: Kafka Broker]
 // Kafka Broker :ref:`configuration overview <config_network_filters_kafka_broker>`.
 // [#extension: envoy.filters.network.kafka_broker]
-
+// [#next-free-field: 6]
 message KafkaBroker {
   // The prefix to use when emitting :ref:`statistics <config_network_filters_kafka_broker_stats>`.
   string stat_prefix = 1 [(validate.rules).string = {min_bytes: 1}];
@@ -38,6 +38,16 @@ message KafkaBroker {
     // Broker address rewrite rules that match by broker ID.
     IdBasedBrokerRewriteSpec id_based_broker_address_rewrite_spec = 3;
   }
+
+  // Optional list of allowed Kafka API keys. Only requests with provided API keys will be
+  // routed, otherwise the connection will be closed. No effect if empty.
+  repeated uint32 api_keys_allowed = 4
+      [(validate.rules).repeated = {items {uint32 {lte: 32767 gte: 0}}}];
+
+  // Optional list of denied Kafka API keys. Requests with API keys matching this list will have
+  // the connection closed. No effect if empty.
+  repeated uint32 api_keys_denied = 5
+      [(validate.rules).repeated = {items {uint32 {lte: 32767 gte: 0}}}];
 }
 
 // Collection of rules matching by broker ID.