diff --git a/docs/utilities/kafka.md b/docs/utilities/kafka.md index da179bc5c..2b9c4cb7a 100644 --- a/docs/utilities/kafka.md +++ b/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`, 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, String> { + + static { + ClassSecurityValidator.setGlobal(ClassSecurityValidator.composite( + ClassSecurityValidator.getGlobal(), + ClassSecurityValidator.builder().add(MyClass.class).build())); + } + + // ... +} +``` + + +!!! 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. diff --git a/examples/powertools-examples-kafka/pom.xml b/examples/powertools-examples-kafka/pom.xml index e9a409825..e0282134a 100644 --- a/examples/powertools-examples-kafka/pom.xml +++ b/examples/powertools-examples-kafka/pom.xml @@ -11,7 +11,7 @@ 17 17 1.9.20.1 - 1.12.1 + 1.12.2 4.35.0 diff --git a/examples/powertools-examples-kafka/tools/pom.xml b/examples/powertools-examples-kafka/tools/pom.xml index 5944df28f..7a65d7cc8 100644 --- a/examples/powertools-examples-kafka/tools/pom.xml +++ b/examples/powertools-examples-kafka/tools/pom.xml @@ -11,7 +11,7 @@ 17 17 - 1.12.1 + 1.12.2 4.31.0 4.0.2 diff --git a/powertools-kafka/pom.xml b/powertools-kafka/pom.xml index 09995da74..cef06b3e9 100644 --- a/powertools-kafka/pom.xml +++ b/powertools-kafka/pom.xml @@ -35,7 +35,7 @@ 4.3.1 - 1.12.1 + 1.12.2 4.35.0 1.4.1 diff --git a/powertools-kafka/src/main/java/software/amazon/lambda/powertools/kafka/serializers/KafkaAvroDeserializer.java b/powertools-kafka/src/main/java/software/amazon/lambda/powertools/kafka/serializers/KafkaAvroDeserializer.java index 70e4affac..dc192b419 100644 --- a/powertools-kafka/src/main/java/software/amazon/lambda/powertools/kafka/serializers/KafkaAvroDeserializer.java +++ b/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> trustedTypes = ConcurrentHashMap.newKeySet(); + private final Set trustedClassNames = ConcurrentHashMap.newKeySet(); + @Override protected T deserializeObject(byte[] data, Class 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 datumReader = new SpecificDatumReader<>(type); Decoder decoder = DecoderFactory.get().binaryDecoder(data, null); @@ -44,4 +53,38 @@ protected T deserializeObject(byte[] data, Class 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 + } + } + } } diff --git a/powertools-kafka/src/test/avro/TestOrder.avsc b/powertools-kafka/src/test/avro/TestOrder.avsc new file mode 100644 index 000000000..75d219bab --- /dev/null +++ b/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"]} + } + ] +} diff --git a/powertools-kafka/src/test/java/software/amazon/lambda/powertools/kafka/serializers/KafkaAvroDeserializerTest.java b/powertools-kafka/src/test/java/software/amazon/lambda/powertools/kafka/serializers/KafkaAvroDeserializerTest.java index f501d7d98..6c0bac413 100644 --- a/powertools-kafka/src/test/java/software/amazon/lambda/powertools/kafka/serializers/KafkaAvroDeserializerTest.java +++ b/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(); + } + }