fix(kafka): trust declared Avro types under Avro 1.12.2 class allowlist

docs/utilities/kafka.md

@@ -560,6 +560,35 @@ Handle errors gracefully when processing Kafka messages to ensure your applicati
     }
     ```
 
+### Trusted Avro classes
+
+Starting with version 1.12.2, Apache Avro only loads classes that its JVM-wide `ClassSecurityValidator` trusts. By default, Avro trusts a small set of JDK types and no application classes.
+
+Powertools trusts the Avro class you declare in your handler signature, for example `Order` in `ConsumerRecords<String, Order>`, and the records, enums and fixed types nested in its compiled schema. Powertools trusts no other classes. This is safe because Powertools never reads a schema from the Kafka record. It always decodes with the schema compiled into your declared class, so a producer cannot make Avro load any other class.
+
+To trust additional classes, add them in a static initializer of your handler class:
+
+```java
+import org.apache.avro.util.ClassSecurityValidator;
+
+public class OrderHandler implements RequestHandler<ConsumerRecords<String, Order>, String> {
+
+    static {
+        ClassSecurityValidator.setGlobal(ClassSecurityValidator.composite(
+                ClassSecurityValidator.getGlobal(),
+                ClassSecurityValidator.builder().add(MyClass.class).build()));
+    }
+
+    // ...
+}
+```
+
+<!-- prettier-ignore -->
+!!! warning "Always compose with `ClassSecurityValidator.getGlobal()`"
+    Calling `ClassSecurityValidator.setGlobal` without composing `ClassSecurityValidator.getGlobal()` replaces the Powertools allowlist, and Avro rejects your declared classes.
+
+You can also list trusted classes or packages with the `org.apache.avro.SERIALIZABLE_CLASSES` and `org.apache.avro.SERIALIZABLE_PACKAGES` system properties, for example `JAVA_TOOL_OPTIONS=-Dorg.apache.avro.SERIALIZABLE_PACKAGES=com.example.model`. Setting `SERIALIZABLE_PACKAGES` to `*` turns off the protection for the whole JVM.
+
 ### Integrating with Idempotency
 
 When processing Kafka messages in Lambda, failed batches can result in message reprocessing. The idempotency utility prevents duplicate processing by tracking which messages have already been handled, ensuring each message is processed exactly once.

examples/powertools-examples-kafka/pom.xml

@@ -11,7 +11,7 @@
         <maven.compiler.source>17</maven.compiler.source>
         <maven.compiler.target>17</maven.compiler.target>
         <aspectj.version>1.9.20.1</aspectj.version>
-        <avro.version>1.12.1</avro.version>
+        <avro.version>1.12.2</avro.version>
         <protobuf.version>4.35.0</protobuf.version>
     </properties>
 

examples/powertools-examples-kafka/tools/pom.xml

@@ -11,7 +11,7 @@
     <properties>
         <maven.compiler.source>17</maven.compiler.source>
         <maven.compiler.target>17</maven.compiler.target>
-        <avro.version>1.12.1</avro.version>
+        <avro.version>1.12.2</avro.version>
         <protobuf.version>4.31.0</protobuf.version>
         <kafka-clients.version>4.0.2</kafka-clients.version>
     </properties>

powertools-kafka/pom.xml

@@ -35,7 +35,7 @@
 
     <properties>
         <kafka-clients.version>4.3.1</kafka-clients.version>
-        <avro.version>1.12.1</avro.version>
+        <avro.version>1.12.2</avro.version>
         <protobuf.version>4.35.0</protobuf.version>
         <lambda-serialization.version>1.4.1</lambda-serialization.version>
     </properties>

powertools-kafka/src/main/java/software/amazon/lambda/powertools/kafka/serializers/KafkaAvroDeserializer.java

@@ -13,24 +13,33 @@
 package software.amazon.lambda.powertools.kafka.serializers;
 
 import java.io.IOException;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
 
+import org.apache.avro.Schema;
 import org.apache.avro.io.DatumReader;
 import org.apache.avro.io.Decoder;
 import org.apache.avro.io.DecoderFactory;
+import org.apache.avro.specific.SpecificData;
 import org.apache.avro.specific.SpecificDatumReader;
 import org.apache.avro.specific.SpecificRecordBase;
+import org.apache.avro.util.ClassSecurityValidator;
 
 /**
  * Deserializer for Kafka records using Avro format.
  */
 public class KafkaAvroDeserializer extends AbstractKafkaDeserializer {
 
+    private final Set<Class<?>> trustedTypes = ConcurrentHashMap.newKeySet();
+    private final Set<String> trustedClassNames = ConcurrentHashMap.newKeySet();
+
     @Override
     protected <T> T deserializeObject(byte[] data, Class<T> type, SchemaRegistryType schemaRegistryType)
             throws IOException {
         // If no Avro generated class is passed we cannot deserialize using Avro
         if (SpecificRecordBase.class.isAssignableFrom(type)) {
             try {
+                trust(type);
                 DatumReader<T> datumReader = new SpecificDatumReader<>(type);
                 Decoder decoder = DecoderFactory.get().binaryDecoder(data, null);
 
@@ -44,4 +53,38 @@ protected <T> T deserializeObject(byte[] data, Class<T> type, SchemaRegistryType
                     + "Consider using an alternative Deserializer.");
         }
     }
+
+    // Avro 1.12.2+ only loads classes trusted by ClassSecurityValidator. We decode with the schema compiled into the
+    // declared type and never with a schema from the record, so Avro only loads classes the developer authored.
+    private void trust(Class<?> type) {
+        if (trustedTypes.contains(type)) {
+            return;
+        }
+        synchronized (this) {
+            if (trustedTypes.isEmpty()) {
+                ClassSecurityValidator.setGlobal(ClassSecurityValidator.composite(ClassSecurityValidator.getGlobal(),
+                        clazz -> trustedClassNames.contains(clazz.getName())));
+            }
+            collectNamedTypes(SpecificData.get().getSchema(type));
+            trustedTypes.add(type);
+        }
+    }
+
+    private void collectNamedTypes(Schema schema) {
+        switch (schema.getType()) {
+            case RECORD -> {
+                // Skip records we have already seen to stop on recursive schemas
+                if (trustedClassNames.add(SpecificData.getClassName(schema))) {
+                    schema.getFields().forEach(field -> collectNamedTypes(field.schema()));
+                }
+            }
+            case ENUM, FIXED -> trustedClassNames.add(SpecificData.getClassName(schema));
+            case ARRAY -> collectNamedTypes(schema.getElementType());
+            case MAP -> collectNamedTypes(schema.getValueType());
+            case UNION -> schema.getTypes().forEach(this::collectNamedTypes);
+            default -> {
+                // Primitive types have no generated class to trust
+            }
+        }
+    }
 }

powertools-kafka/src/test/avro/TestOrder.avsc

@@ -0,0 +1,23 @@
+{
+  "namespace": "software.amazon.lambda.powertools.kafka.serializers.test.avro",
+  "type": "record",
+  "name": "TestOrder",
+  "fields": [
+    {"name": "id", "type": "string"},
+    {
+      "name": "customer",
+      "type": ["null", {
+        "type": "record",
+        "name": "TestCustomer",
+        "fields": [
+          {"name": "name", "type": "string"}
+        ]
+      }],
+      "default": null
+    },
+    {
+      "name": "status",
+      "type": {"type": "enum", "name": "TestOrderStatus", "symbols": ["PENDING", "SHIPPED"]}
+    }
+  ]
+}

powertools-kafka/src/test/java/software/amazon/lambda/powertools/kafka/serializers/KafkaAvroDeserializerTest.java

@@ -13,25 +13,40 @@
 package software.amazon.lambda.powertools.kafka.serializers;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static software.amazon.lambda.powertools.kafka.testutils.TestUtils.serializeAvro;
 
 import java.io.IOException;
 
+import org.apache.avro.util.ClassSecurityValidator;
+import org.apache.avro.util.ClassSecurityValidator.ClassSecurityPredicate;
+import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
+import software.amazon.lambda.powertools.kafka.serializers.test.avro.TestCustomer;
+import software.amazon.lambda.powertools.kafka.serializers.test.avro.TestOrder;
+import software.amazon.lambda.powertools.kafka.serializers.test.avro.TestOrderStatus;
 import software.amazon.lambda.powertools.kafka.serializers.test.avro.TestProduct;
+import software.amazon.lambda.powertools.kafka.testutils.TestProductPojo;
 
 class KafkaAvroDeserializerTest {
 
     private KafkaAvroDeserializer deserializer;
+    private ClassSecurityPredicate originalValidator;
 
     @BeforeEach
     void setUp() {
+        originalValidator = ClassSecurityValidator.getGlobal();
         deserializer = new KafkaAvroDeserializer();
     }
 
+    @AfterEach
+    void tearDown() {
+        ClassSecurityValidator.setGlobal(originalValidator);
+    }
+
     @Test
     void shouldThrowExceptionWhenTypeIsNotAvroSpecificRecord() {
         // Given
@@ -73,4 +88,49 @@ void shouldThrowExceptionWhenDeserializingInvalidAvroData() {
                         .hasMessageContaining("Failed to deserialize Avro data");
     }
 
+    @Test
+    void shouldDeserializeRecordWithNestedRecordAndEnum() throws IOException {
+        // Given
+        TestOrder order = new TestOrder("order-1", new TestCustomer("Jane"), TestOrderStatus.SHIPPED);
+        byte[] avroData = serializeAvro(order);
+
+        // When
+        TestOrder result = deserializer.deserializeObject(avroData, TestOrder.class,
+                AbstractKafkaDeserializer.SchemaRegistryType.NONE);
+
+        // Then
+        assertThat(result).isEqualTo(order);
+    }
+
+    @Test
+    void shouldNotTrustClassesOutsideDeclaredSchema() throws IOException {
+        // Given
+        byte[] avroData = serializeAvro(new TestProduct(123, "Test Product", 99.99));
+
+        // When
+        deserializer.deserializeObject(avroData, TestProduct.class, AbstractKafkaDeserializer.SchemaRegistryType.NONE);
+
+        // Then
+        assertThatThrownBy(() -> ClassSecurityValidator.validate(TestProductPojo.class))
+                .isInstanceOf(SecurityException.class);
+    }
+
+    @Test
+    void shouldKeepValidatorConfiguredBeforeFirstUse() throws IOException {
+        // Given
+        ClassSecurityValidator.setGlobal(ClassSecurityValidator.builder().add(TestProductPojo.class).build());
+        byte[] avroData = serializeAvro(new TestProduct(123, "Test Product", 99.99));
+
+        // When
+        TestProduct result = deserializer.deserializeObject(avroData, TestProduct.class,
+                AbstractKafkaDeserializer.SchemaRegistryType.NONE);
+
+        // Then
+        assertThat(result.getId()).isEqualTo(123);
+        assertThatCode(() -> {
+            ClassSecurityValidator.validate(TestProductPojo.class);
+            ClassSecurityValidator.validate(TestProduct.class);
+        }).doesNotThrowAnyException();
+    }
+
 }