KAFKA-13613: Remove hard dependency on HmacSHA256 algorithm for Connect (#11894)

Reviewers: Mickael Maison <mickael.maison@gmail.com>

, Tom Bentley <tbentley@redhat.com>

connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedConfig.java

@@ -31,11 +31,16 @@ import javax.crypto.KeyGenerator;
 import javax.crypto.Mac;
 import java.security.InvalidParameterException;
 import java.security.NoSuchAlgorithmException;
+import java.security.Provider;
+import java.security.Security;
+import java.util.ArrayList;
 import java.util.Collections;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Locale;
 import java.util.Map;
 import java.util.Optional;
+import java.util.Set;
 import java.util.concurrent.TimeUnit;
 
 import static org.apache.kafka.common.config.ConfigDef.Range.atLeast;
@@ -176,8 +181,10 @@ public class DistributedConfig extends WorkerConfig {
     public static final int SCHEDULED_REBALANCE_MAX_DELAY_MS_DEFAULT = Math.toIntExact(TimeUnit.SECONDS.toMillis(300));
 
     public static final String INTER_WORKER_KEY_GENERATION_ALGORITHM_CONFIG = "inter.worker.key.generation.algorithm";
-    public static final String INTER_WORKER_KEY_GENERATION_ALGORITHM_DOC = "The algorithm to use for generating internal request keys";
     public static final String INTER_WORKER_KEY_GENERATION_ALGORITHM_DEFAULT = "HmacSHA256";
+    public static final String INTER_WORKER_KEY_GENERATION_ALGORITHM_DOC = "The algorithm to use for generating internal request keys. "
+            + "The algorithm '" + INTER_WORKER_KEY_GENERATION_ALGORITHM_DEFAULT + "' will be used as a default on JVMs that support it; "
+            + "on other JVMs, no default is used and a value for this property must be manually specified in the worker config.";
 
     public static final String INTER_WORKER_KEY_SIZE_CONFIG = "inter.worker.key.size";
     public static final String INTER_WORKER_KEY_SIZE_DOC = "The size of the key to use for signing internal requests, in bits. "
@@ -190,12 +197,17 @@ public class DistributedConfig extends WorkerConfig {
     public static final int INTER_WORKER_KEY_TTL_MS_MS_DEFAULT = Math.toIntExact(TimeUnit.HOURS.toMillis(1));
 
     public static final String INTER_WORKER_SIGNATURE_ALGORITHM_CONFIG = "inter.worker.signature.algorithm";
-    public static final String INTER_WORKER_SIGNATURE_ALGORITHM_DOC = "The algorithm used to sign internal requests";
     public static final String INTER_WORKER_SIGNATURE_ALGORITHM_DEFAULT = "HmacSHA256";
+    public static final String INTER_WORKER_SIGNATURE_ALGORITHM_DOC = "The algorithm used to sign internal requests"
+            + "The algorithm '" + INTER_WORKER_SIGNATURE_ALGORITHM_CONFIG + "' will be used as a default on JVMs that support it; "
+            + "on other JVMs, no default is used and a value for this property must be manually specified in the worker config.";
 
     public static final String INTER_WORKER_VERIFICATION_ALGORITHMS_CONFIG = "inter.worker.verification.algorithms";
-    public static final String INTER_WORKER_VERIFICATION_ALGORITHMS_DOC = "A list of permitted algorithms for verifying internal requests";
     public static final List<String> INTER_WORKER_VERIFICATION_ALGORITHMS_DEFAULT = Collections.singletonList(INTER_WORKER_SIGNATURE_ALGORITHM_DEFAULT);
+    public static final String INTER_WORKER_VERIFICATION_ALGORITHMS_DOC = "A list of permitted algorithms for verifying internal requests, "
+        + "which must include the algorithm used for the " + INTER_WORKER_SIGNATURE_ALGORITHM_CONFIG + " property. "
+        + "The algorithm(s) '" + INTER_WORKER_VERIFICATION_ALGORITHMS_DEFAULT + "' will be used as a default on JVMs that provide them; "
+        + "on other JVMs, no default is used and a value for this property must be manually specified in the worker config.";
 
     private enum ExactlyOnceSourceSupport {
         DISABLED(false),
@@ -225,6 +237,58 @@ public class DistributedConfig extends WorkerConfig {
             //       + "See the exactly-once source support documentation at [add docs link here] for more information on this feature.";
     public static final String EXACTLY_ONCE_SOURCE_SUPPORT_DEFAULT = ExactlyOnceSourceSupport.DISABLED.toString();
 
+    private static Object defaultKeyGenerationAlgorithm() {
+        try {
+            validateKeyAlgorithm(INTER_WORKER_KEY_GENERATION_ALGORITHM_CONFIG, INTER_WORKER_KEY_GENERATION_ALGORITHM_DEFAULT);
+            return INTER_WORKER_KEY_GENERATION_ALGORITHM_DEFAULT;
+        } catch (Throwable t) {
+            log.info(
+                    "The default key generation algorithm '{}' does not appear to be available on this worker."
+                            + "A key algorithm will have to be manually specified via the '{}' worker property",
+                    INTER_WORKER_KEY_GENERATION_ALGORITHM_DEFAULT,
+                    INTER_WORKER_KEY_GENERATION_ALGORITHM_CONFIG
+            );
+            return ConfigDef.NO_DEFAULT_VALUE;
+        }
+    }
+
+    private static Object defaultSignatureAlgorithm() {
+        try {
+            validateSignatureAlgorithm(INTER_WORKER_SIGNATURE_ALGORITHM_CONFIG, INTER_WORKER_SIGNATURE_ALGORITHM_DEFAULT);
+            return INTER_WORKER_SIGNATURE_ALGORITHM_DEFAULT;
+        } catch (Throwable t) {
+            log.info(
+                    "The default signature algorithm '{}' does not appear to be available on this worker."
+                            + "A signature algorithm will have to be manually specified via the '{}' worker property",
+                    INTER_WORKER_SIGNATURE_ALGORITHM_DEFAULT,
+                    INTER_WORKER_SIGNATURE_ALGORITHM_CONFIG
+            );
+            return ConfigDef.NO_DEFAULT_VALUE;
+        }
+    }
+
+    private static Object defaultVerificationAlgorithms() {
+        List<String> result = new ArrayList<>();
+        for (String verificationAlgorithm : INTER_WORKER_VERIFICATION_ALGORITHMS_DEFAULT) {
+            try {
+                validateSignatureAlgorithm(INTER_WORKER_VERIFICATION_ALGORITHMS_CONFIG, verificationAlgorithm);
+                result.add(verificationAlgorithm);
+            } catch (Throwable t) {
+                log.trace("Verification algorithm '{}' not found", verificationAlgorithm);
+            }
+        }
+        if (result.isEmpty()) {
+            log.info(
+                    "The default verification algorithm '{}' does not appear to be available on this worker."
+                            + "One or more verification algorithms will have to be manually specified via the '{}' worker property",
+                    INTER_WORKER_VERIFICATION_ALGORITHMS_DEFAULT,
+                    INTER_WORKER_VERIFICATION_ALGORITHMS_CONFIG
+            );
+            return ConfigDef.NO_DEFAULT_VALUE;
+        }
+        return result;
+    }
+
     @SuppressWarnings("unchecked")
     private static final ConfigDef CONFIG = baseConfigDef()
             .define(GROUP_ID_CONFIG,
@@ -406,11 +470,10 @@ public class DistributedConfig extends WorkerConfig {
                     INTER_WORKER_KEY_TTL_MS_MS_DOC)
             .define(INTER_WORKER_KEY_GENERATION_ALGORITHM_CONFIG,
                     ConfigDef.Type.STRING,
-                    INTER_WORKER_KEY_GENERATION_ALGORITHM_DEFAULT,
+                    defaultKeyGenerationAlgorithm(),
                     ConfigDef.LambdaValidator.with(
                             (name, value) -> validateKeyAlgorithm(name, (String) value),
-                        () -> "Any KeyGenerator algorithm supported by the worker JVM"
+                            () -> "Any KeyGenerator algorithm supported by the worker JVM"),
-                    ),
                     ConfigDef.Importance.LOW,
                     INTER_WORKER_KEY_GENERATION_ALGORITHM_DOC)
             .define(INTER_WORKER_KEY_SIZE_CONFIG,
@@ -420,7 +483,7 @@ public class DistributedConfig extends WorkerConfig {
                     INTER_WORKER_KEY_SIZE_DOC)
             .define(INTER_WORKER_SIGNATURE_ALGORITHM_CONFIG,
                     ConfigDef.Type.STRING,
-                    INTER_WORKER_SIGNATURE_ALGORITHM_DEFAULT,
+                    defaultSignatureAlgorithm(),
                     ConfigDef.LambdaValidator.with(
                             (name, value) -> validateSignatureAlgorithm(name, (String) value),
                             () -> "Any MAC algorithm supported by the worker JVM"),
@@ -428,11 +491,10 @@ public class DistributedConfig extends WorkerConfig {
                     INTER_WORKER_SIGNATURE_ALGORITHM_DOC)
             .define(INTER_WORKER_VERIFICATION_ALGORITHMS_CONFIG,
                     ConfigDef.Type.LIST,
-                    INTER_WORKER_VERIFICATION_ALGORITHMS_DEFAULT,
+                    defaultVerificationAlgorithms(),
                     ConfigDef.LambdaValidator.with(
-                        (name, value) -> validateSignatureAlgorithms(name, (List<String>) value),
+                            (name, value) -> validateVerificationAlgorithms(name, (List<String>) value),
-                        () -> "A list of one or more MAC algorithms, each supported by the worker JVM"
+                            () -> "A list of one or more MAC algorithms, each supported by the worker JVM"),
-                    ),
                     ConfigDef.Importance.LOW,
                     INTER_WORKER_VERIFICATION_ALGORITHMS_DOC);
 
@@ -487,8 +549,7 @@ public class DistributedConfig extends WorkerConfig {
     public DistributedConfig(Map<String, String> props) {
         super(CONFIG, props);
         exactlyOnceSourceSupport = ExactlyOnceSourceSupport.fromProperty(getString(EXACTLY_ONCE_SOURCE_SUPPORT_CONFIG));
-        getInternalRequestKeyGenerator(); // Check here for a valid key size + key algorithm to fail fast if either are invalid
+        validateInterWorkerKeyConfigs();
-        validateKeyAlgorithmAndVerificationAlgorithms();
     }
 
     public static void main(String[] args) {
@@ -537,19 +598,24 @@ public class DistributedConfig extends WorkerConfig {
         return topicSettings(STATUS_STORAGE_PREFIX);
     }
 
-    private void validateKeyAlgorithmAndVerificationAlgorithms() {
+    private void validateInterWorkerKeyConfigs() {
-        String keyAlgorithm = getString(INTER_WORKER_KEY_GENERATION_ALGORITHM_CONFIG);
+        getInternalRequestKeyGenerator();
+        ensureVerificationAlgorithmsIncludeSignatureAlgorithm();
+    }
+
+    private void ensureVerificationAlgorithmsIncludeSignatureAlgorithm() {
+        String signatureAlgorithm = getString(INTER_WORKER_SIGNATURE_ALGORITHM_CONFIG);
         List<String> verificationAlgorithms = getList(INTER_WORKER_VERIFICATION_ALGORITHMS_CONFIG);
-        if (!verificationAlgorithms.contains(keyAlgorithm)) {
+        if (!verificationAlgorithms.contains(signatureAlgorithm)) {
             throw new ConfigException(
-                INTER_WORKER_KEY_GENERATION_ALGORITHM_CONFIG,
+                INTER_WORKER_SIGNATURE_ALGORITHM_CONFIG,
-                keyAlgorithm,
+                signatureAlgorithm,
-                String.format("Key generation algorithm must be present in %s list", INTER_WORKER_VERIFICATION_ALGORITHMS_CONFIG)
+                String.format("Signature algorithm must be present in %s list", INTER_WORKER_VERIFICATION_ALGORITHMS_CONFIG)
             );
         }
     }
 
-    private static void validateSignatureAlgorithms(String configName, List<String> algorithms) {
+    private static void validateVerificationAlgorithms(String configName, List<String> algorithms) {
         if (algorithms.isEmpty()) {
             throw new ConfigException(
                     configName,
@@ -557,14 +623,20 @@ public class DistributedConfig extends WorkerConfig {
                     "At least one signature verification algorithm must be provided"
             );
         }
-        algorithms.forEach(algorithm -> validateSignatureAlgorithm(configName, algorithm));
+        for (String algorithm : algorithms) {
+            try {
+                Mac.getInstance(algorithm);
+            } catch (NoSuchAlgorithmException e) {
+                throw unsupportedAlgorithmException(configName, algorithm, "Mac");
+            }
+        }
     }
 
     private static void validateSignatureAlgorithm(String configName, String algorithm) {
         try {
             Mac.getInstance(algorithm);
         } catch (NoSuchAlgorithmException e) {
-            throw new ConfigException(configName, algorithm, e.getMessage());
+            throw unsupportedAlgorithmException(configName, algorithm, "Mac");
         }
     }
 
@@ -572,7 +644,29 @@ public class DistributedConfig extends WorkerConfig {
         try {
             KeyGenerator.getInstance(algorithm);
         } catch (NoSuchAlgorithmException e) {
-            throw new ConfigException(configName, algorithm, e.getMessage());
+            throw unsupportedAlgorithmException(configName, algorithm, "KeyGenerator");
+        }
+    }
+
+    private static ConfigException unsupportedAlgorithmException(String name, Object value, String type) {
+        return new ConfigException(
+                name,
+                value,
+                "the algorithm is not supported by this JVM; the supported algorithms are: " + supportedAlgorithms(type)
+        );
+    }
+
+    // Visible for testing
+    static Set<String> supportedAlgorithms(String type) {
+        Set<String> result = new HashSet<>();
+        for (Provider provider : Security.getProviders()) {
+            for (Provider.Service service : provider.getServices()) {
+                if (type.equals(service.getType())) {
+                    result.add(service.getAlgorithm());
                 }
             }
         }
+        return result;
+    }
+
+}

connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/DistributedConfigTest.java

@@ -20,22 +20,33 @@ package org.apache.kafka.connect.runtime.distributed;
 import org.apache.kafka.clients.CommonClientConfigs;
 import org.apache.kafka.common.config.ConfigException;
 import org.junit.Test;
+import org.mockito.MockedStatic;
 
 import javax.crypto.KeyGenerator;
+import javax.crypto.Mac;
+import java.security.NoSuchAlgorithmException;
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 
 import static org.apache.kafka.connect.runtime.distributed.DistributedConfig.EXACTLY_ONCE_SOURCE_SUPPORT_CONFIG;
 import static org.apache.kafka.connect.runtime.distributed.DistributedConfig.GROUP_ID_CONFIG;
+import static org.apache.kafka.connect.runtime.distributed.DistributedConfig.INTER_WORKER_KEY_GENERATION_ALGORITHM_CONFIG;
+import static org.apache.kafka.connect.runtime.distributed.DistributedConfig.INTER_WORKER_SIGNATURE_ALGORITHM_CONFIG;
+import static org.apache.kafka.connect.runtime.distributed.DistributedConfig.INTER_WORKER_VERIFICATION_ALGORITHMS_CONFIG;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotEquals;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertThrows;
 import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mockStatic;
 
 public class DistributedConfigTest {
 
@@ -57,13 +68,96 @@ public class DistributedConfigTest {
         assertNotNull(config.getInternalRequestKeyGenerator());
     }
 
+    @Test
+    public void testDefaultAlgorithmsNotPresent() {
+        final String fakeKeyGenerationAlgorithm = "FakeKeyGenerationAlgorithm";
+        final String fakeMacAlgorithm = "FakeMacAlgorithm";
+
+        final KeyGenerator fakeKeyGenerator = mock(KeyGenerator.class);
+        final Mac fakeMac = mock(Mac.class);
+
+        Map<String, String> configs = configs();
+        configs.put(DistributedConfig.INTER_WORKER_KEY_GENERATION_ALGORITHM_CONFIG, fakeKeyGenerationAlgorithm);
+        configs.put(DistributedConfig.INTER_WORKER_SIGNATURE_ALGORITHM_CONFIG, fakeMacAlgorithm);
+        configs.put(DistributedConfig.INTER_WORKER_VERIFICATION_ALGORITHMS_CONFIG, fakeMacAlgorithm);
+
+        try (
+                MockedStatic<KeyGenerator> keyGenerator = mockStatic(KeyGenerator.class);
+                MockedStatic<Mac> mac = mockStatic(Mac.class)
+        ) {
+            // Make it seem like the default key generation algorithm isn't available on this worker
+            keyGenerator.when(() -> KeyGenerator.getInstance(DistributedConfig.INTER_WORKER_KEY_GENERATION_ALGORITHM_DEFAULT))
+                    .thenThrow(new NoSuchAlgorithmException());
+            // But the one specified in the worker config file is
+            keyGenerator.when(() -> KeyGenerator.getInstance(fakeKeyGenerationAlgorithm))
+                    .thenReturn(fakeKeyGenerator);
+
+            // And for the signature algorithm
+            mac.when(() -> Mac.getInstance(DistributedConfig.INTER_WORKER_SIGNATURE_ALGORITHM_DEFAULT))
+                    .thenThrow(new NoSuchAlgorithmException());
+            // Likewise for key verification algorithms
+            DistributedConfig.INTER_WORKER_VERIFICATION_ALGORITHMS_DEFAULT.forEach(verificationAlgorithm ->
+                keyGenerator.when(() -> Mac.getInstance(verificationAlgorithm))
+                        .thenThrow(new NoSuchAlgorithmException())
+            );
+            mac.when(() -> Mac.getInstance(fakeMacAlgorithm))
+                    .thenReturn(fakeMac);
+
+            // Should succeed; even though the defaults aren't present, the manually-specified algorithms are valid
+            new DistributedConfig(configs);
+
+            // Should fail; the default key generation algorithm isn't present, and no override is specified
+            String removed = configs.remove(INTER_WORKER_KEY_GENERATION_ALGORITHM_CONFIG);
+            assertThrows(ConfigException.class, () -> new DistributedConfig(configs));
+            configs.put(INTER_WORKER_KEY_GENERATION_ALGORITHM_CONFIG, removed);
+
+            // Should fail; the default key generation algorithm isn't present, and no override is specified
+            removed = configs.remove(INTER_WORKER_SIGNATURE_ALGORITHM_CONFIG);
+            assertThrows(ConfigException.class, () -> new DistributedConfig(configs));
+            configs.put(INTER_WORKER_SIGNATURE_ALGORITHM_CONFIG, removed);
+
+            // Should fail; the default key generation algorithm isn't present, and no override is specified
+            removed = configs.remove(INTER_WORKER_VERIFICATION_ALGORITHMS_CONFIG);
+            assertThrows(ConfigException.class, () -> new DistributedConfig(configs));
+            configs.put(INTER_WORKER_VERIFICATION_ALGORITHMS_CONFIG, removed);
+        }
+    }
+
+    @Test
+    public void testSupportedMacAlgorithms() {
+        // These algorithms are required to be supported on JVMs ranging from at least Java 8 through Java 17; see
+        // https://docs.oracle.com/javase/8/docs/api/javax/crypto/Mac.html
+        // and https://docs.oracle.com/en/java/javase/17/docs/api/java.base/javax/crypto/Mac.html
+        testSupportedAlgorithms(
+                "Mac",
+                "HmacSHA1", "HmacSHA256"
+        );
+    }
+
+    @Test
+    public void testSupportedKeyGeneratorAlgorithms() {
+        // These algorithms are required to be supported on JVMs ranging from at least Java 8 through Java 17; see
+        // https://docs.oracle.com/javase/8/docs/api/javax/crypto/KeyGenerator.html
+        // and https://docs.oracle.com/en/java/javase/17/docs/api/java.base/javax/crypto/KeyGenerator.html
+        testSupportedAlgorithms(
+                "KeyGenerator",
+                "AES", "DESede", "HmacSHA1", "HmacSHA256"
+        );
+    }
+
+    private void testSupportedAlgorithms(String type, String... expectedAlgorithms) {
+        Set<String> supportedAlgorithms = DistributedConfig.supportedAlgorithms(type);
+        Set<String> unuspportedAlgorithms = new HashSet<>(Arrays.asList(expectedAlgorithms));
+        unuspportedAlgorithms.removeAll(supportedAlgorithms);
+        assertEquals(type + " algorithms were found that should be supported by this JVM but are not", Collections.emptySet(), unuspportedAlgorithms);
+    }
+
     @Test
     public void shouldCreateKeyGeneratorWithSpecificSettings() {
         final String algorithm = "HmacSHA1";
         Map<String, String> configs = configs();
         configs.put(DistributedConfig.INTER_WORKER_KEY_GENERATION_ALGORITHM_CONFIG, algorithm);
         configs.put(DistributedConfig.INTER_WORKER_KEY_SIZE_CONFIG, "512");
-        configs.put(DistributedConfig.INTER_WORKER_VERIFICATION_ALGORITHMS_CONFIG, algorithm);
         DistributedConfig config = new DistributedConfig(configs);
         KeyGenerator keyGenerator = config.getInternalRequestKeyGenerator();
         assertNotNull(keyGenerator);
@@ -79,13 +173,22 @@ public class DistributedConfigTest {
     }
 
     @Test
-    public void shouldFailIfKeyAlgorithmNotInVerificationAlgorithmsList() {
+    public void shouldFailIfSignatureAlgorithmNotInVerificationAlgorithmsList() {
         Map<String, String> configs = configs();
-        configs.put(DistributedConfig.INTER_WORKER_KEY_GENERATION_ALGORITHM_CONFIG, "HmacSHA1");
+        configs.put(DistributedConfig.INTER_WORKER_SIGNATURE_ALGORITHM_CONFIG, "HmacSHA1");
         configs.put(DistributedConfig.INTER_WORKER_VERIFICATION_ALGORITHMS_CONFIG, "HmacSHA256");
         assertThrows(ConfigException.class, () -> new DistributedConfig(configs));
     }
 
+    @Test
+    public void shouldNotFailIfKeyAlgorithmNotInVerificationAlgorithmsList() {
+        Map<String, String> configs = configs();
+        configs.put(DistributedConfig.INTER_WORKER_KEY_GENERATION_ALGORITHM_CONFIG, "HmacSHA1");
+        configs.put(DistributedConfig.INTER_WORKER_SIGNATURE_ALGORITHM_CONFIG, "HmacSHA256");
+        configs.put(DistributedConfig.INTER_WORKER_VERIFICATION_ALGORITHMS_CONFIG, "HmacSHA256");
+        new DistributedConfig(configs);
+    }
+
     @Test
     public void shouldFailWithInvalidKeyAlgorithm() {
         Map<String, String> configs = configs();