Remove deprecated BlowfishTupleSerializer (#8994)

docs/SECURITY.md

@@ -677,7 +677,7 @@ The size limits need a caveat. `maxbytes` counts stream bytes per object, not pe
 
 An invalid pattern is rejected when the topology is submitted, and otherwise fails worker startup.
 
-For tuple encryption, use TLS-based transport encryption (`storm.messaging.netty.tls.enable`) instead of the deprecated `BlowfishTupleSerializer`, which uses a 64-bit block cipher vulnerable to birthday attacks.
+For tuple encryption, use TLS-based transport encryption (`storm.messaging.netty.tls.enable`). The legacy `BlowfishTupleSerializer` has been removed because it uses a 64-bit block cipher vulnerable to birthday attacks.
 
 ### Log Cleanup
 The Logviewer daemon now is also responsible for cleaning up old log files for dead topologies.

storm-client/test/jvm/org/apache/storm/serialization/SerializationFactoryTest.java

@@ -13,9 +13,11 @@
 package org.apache.storm.serialization;
 
 import com.esotericsoftware.kryo.Kryo;
+import com.esotericsoftware.kryo.Serializer;
+import com.esotericsoftware.kryo.io.Input;
+import com.esotericsoftware.kryo.io.Output;
 import java.util.Map;
 import org.apache.storm.Config;
-import org.apache.storm.security.serialization.BlowfishTupleSerializer;
 import org.apache.storm.testing.TestWordSpout;
 import org.apache.storm.topology.TopologyBuilder;
 import org.apache.storm.utils.ListDelegate;
@@ -48,14 +50,11 @@ public void test_throws_runtimeexception_when_no_such_class() {
 
     @Test
     public void test_registers_when_valid_class_name() {
-        Class arbitraryClass = BlowfishTupleSerializer.class;
-        String secretKey = "0123456789abcdef";
+        Class arbitraryClass = MockTupleSerializer.class;
         Map<String, Object> conf = Utils.readDefaultConfig();
         conf.put(Config.TOPOLOGY_TUPLE_SERIALIZER, arbitraryClass.getName());
-        conf.put(BlowfishTupleSerializer.SECRET_KEY, secretKey);
         Kryo kryo = SerializationFactory.getKryo(conf);
         assertEquals(arbitraryClass, kryo.getSerializer(ListDelegate.class).getClass());
-
     }
 
     @Test
@@ -89,4 +88,15 @@ public void decorate(Kryo k, Map<String, Object> conf) {
     public static class MockSerObject {
     }
 
+    public static class MockTupleSerializer extends Serializer<ListDelegate> {
+        @Override
+        public void write(Kryo kryo, Output output, ListDelegate object) {
+        }
+
+        @Override
+        public ListDelegate read(Kryo kryo, Input input, Class<? extends ListDelegate> type) {
+            return null;
+        }
+    }
+
 }

storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java

@@ -200,7 +200,6 @@
 import org.apache.storm.security.auth.ThriftConnectionType;
 import org.apache.storm.security.auth.ThriftServer;
 import org.apache.storm.security.auth.workertoken.WorkerTokenManager;
-import org.apache.storm.security.serialization.BlowfishTupleSerializer;
 import org.apache.storm.shade.com.google.common.annotations.VisibleForTesting;
 import org.apache.storm.shade.com.google.common.base.Strings;
 import org.apache.storm.shade.com.google.common.collect.ImmutableMap;
@@ -806,20 +805,22 @@
         return tc.readTopology(topoId, NIMBUS_SUBJECT);
     }
 
+    @VisibleForTesting
+    static final String LEGACY_BLOWFISH_SECRET_KEY = "topology.tuple.serializer.blowfish.key";
+
     /**
      * Mask the credential values in a config map that is about to be serialized to a client. Nimbus serves config over
      * several read-only operations, and a caller authorized for those is not necessarily authorized to hold the cluster's
-     * or the topology's secrets. This covers what {@link ConfigUtils#maskCredentials(Map)} covers, plus the Blowfish
-     * tuple-serializer key, whose constant lives outside the config classes so the annotation scan cannot see it and
-     * whose name matches no credential pattern.
+     * or the topology's secrets. This covers what {@link ConfigUtils#maskCredentials(Map)} covers, plus the legacy Blowfish
+     * tuple-serializer key, so that legacy stored configs do not leak the key.
      *
      * @param conf the config about to be served
      * @return a copy of the config with credential values replaced
      */
     private static Map<String, Object> maskCredentialsForApi(Map<String, Object> conf) {
         Map<String, Object> masked = new HashMap<>(ConfigUtils.maskCredentials(conf));
-        if (masked.get(BlowfishTupleSerializer.SECRET_KEY) instanceof String) {
-            masked.put(BlowfishTupleSerializer.SECRET_KEY, "*****");
+        if (masked.get(LEGACY_BLOWFISH_SECRET_KEY) instanceof String) {
+            masked.put(LEGACY_BLOWFISH_SECRET_KEY, "*****");
         }
         return masked;
     }
@@ -1138,7 +1139,7 @@
         cleanable.addAll(Utils.OR(state.heartbeatStorms(), EMPTY_STRING_LIST));
         cleanable.addAll(Utils.OR(state.errorTopologies(), EMPTY_STRING_LIST));
         cleanable.addAll(Utils.OR(store.storedTopoIds(), EMPTY_STRING_SET));
         cleanable.addAll(Utils.OR(state.backpressureTopologies(), EMPTY_STRING_LIST));
         cleanable.addAll(Utils.OR(state.idsOfTopologiesWithPrivateWorkerKeys(), EMPTY_STRING_SET));
         Set<String> delayedCleanable = getExpiredTopologyIds(cleanable, conf);
         delayedCleanable.removeAll(Utils.OR(state.activeStorms(), EMPTY_STRING_LIST));
@@ -1258,8 +1259,8 @@
         ret.put(Config.TOPOLOGY_WORKER_NIMBUS_THRIFT_CLIENT_USE_TLS, workerNimbusClientTlsEnabled);
         ret.put(Config.NIMBUS_THRIFT_CLIENT_USE_TLS, workerNimbusClientTlsEnabled);
 
         if (!mergedConf.containsKey(Config.TOPOLOGY_METRICS_REPORTERS) && mergedConf.containsKey(Config.STORM_METRICS_REPORTERS)) {
             ret.put(Config.TOPOLOGY_METRICS_REPORTERS, mergedConf.get(Config.STORM_METRICS_REPORTERS));
         }
 
         // add any system metrics reporters to the topology metrics reporters
@@ -3177,7 +3178,7 @@
                 state.teardownHeartbeats(topoId);
                 state.teardownTopologyErrors(topoId);
                 state.removeAllPrivateWorkerKeys(topoId);
                 state.removeBackpressure(topoId);
                 rmDependencyBlobsInTopology(topoId, stillReferenced);
                 forceDeleteTopoDistDir(topoId);
                 rmTopologyKeys(topoId);
@@ -3697,8 +3698,8 @@
                 waitForDesiredCodeReplication(totalConf, topoId);
                 state.setupHeatbeats(topoId, topoConf);
                 state.setupErrors(topoId, topoConf);
                 if (ObjectReader.getBoolean(totalConf.get(Config.TOPOLOGY_BACKPRESSURE_ENABLE), false)) {
                     state.setupBackpressure(topoId, topoConf);
                 }
                 notifyTopologyActionListener(topoName, "submitTopology");
                 TopologyStatus status = null;

storm-server/src/test/java/org/apache/storm/daemon/nimbus/NimbusGetTopologyConfTest.java

@@ -29,7 +29,6 @@
 import org.apache.storm.cluster.IStormClusterState;
 import net.minidev.json.JSONValue;
 
-import org.apache.storm.security.serialization.BlowfishTupleSerializer;
 import org.junit.jupiter.api.Test;
 import org.mockito.ArgumentMatchers;
 import org.mockito.Mockito;
@@ -56,7 +55,7 @@ public void getTopologyConfMasksCredentialsAndKeepsOtherValues() throws Exceptio
         Map<String, Object> storedConf = new HashMap<>();
         storedConf.put(Config.TOPOLOGY_NAME, TOPO_NAME);
         storedConf.put(Config.STORM_ZOOKEEPER_TOPOLOGY_AUTH_PAYLOAD, "topology-zk-secret");
-        storedConf.put(BlowfishTupleSerializer.SECRET_KEY, "0123456789abcdef");
+        storedConf.put(Nimbus.LEGACY_BLOWFISH_SECRET_KEY, "0123456789abcdef");
         storedConf.put(Config.TOPOLOGY_WORKERS, 3);
 
         try (LocalCluster cluster = new LocalCluster.Builder()
@@ -77,15 +76,15 @@ public void getTopologyConfMasksCredentialsAndKeepsOtherValues() throws Exceptio
 
             assertEquals(MASKED, served.get(Config.STORM_ZOOKEEPER_TOPOLOGY_AUTH_PAYLOAD),
                 "the ZooKeeper auth payload should be masked");
-            assertEquals(MASKED, served.get(BlowfishTupleSerializer.SECRET_KEY),
+            assertEquals(MASKED, served.get(Nimbus.LEGACY_BLOWFISH_SECRET_KEY),
                 "the tuple serializer key should be masked");
 
             assertEquals(TOPO_NAME, served.get(Config.TOPOLOGY_NAME));
             assertEquals(3, ((Number) served.get(Config.TOPOLOGY_WORKERS)).intValue());
 
             assertEquals("topology-zk-secret", storedConf.get(Config.STORM_ZOOKEEPER_TOPOLOGY_AUTH_PAYLOAD),
                 "the stored conf should keep its own values");
-            assertEquals("0123456789abcdef", storedConf.get(BlowfishTupleSerializer.SECRET_KEY),
+            assertEquals("0123456789abcdef", storedConf.get(Nimbus.LEGACY_BLOWFISH_SECRET_KEY),
                 "the stored conf should keep its own values");
         }
     }

storm-server/src/test/java/org/apache/storm/daemon/nimbus/NimbusGetTopologyPageInfoTest.java

@@ -33,7 +33,6 @@
 import org.apache.storm.generated.TopologyPageInfo;
 import org.apache.storm.generated.TopologyStatus;
 import org.apache.storm.scheduler.resource.TestUtilsForResourceAwareScheduler.TestSpout;
-import org.apache.storm.security.serialization.BlowfishTupleSerializer;
 import org.apache.storm.topology.TopologyBuilder;
 import org.junit.jupiter.api.Test;
 import org.mockito.ArgumentMatchers;
@@ -76,7 +75,7 @@ private static Map<String, Object> storedTopoConf() {
         topoConf.put(Config.TOPOLOGY_EVENTLOGGER_EXECUTORS, 0);
         topoConf.put(Config.TOPOLOGY_MESSAGE_TIMEOUT_SECS, 30);
         topoConf.put(Config.STORM_ZOOKEEPER_TOPOLOGY_AUTH_PAYLOAD, "topology-zk-secret");
-        topoConf.put(BlowfishTupleSerializer.SECRET_KEY, "0123456789abcdef");
+        topoConf.put(Nimbus.LEGACY_BLOWFISH_SECRET_KEY, "0123456789abcdef");
         return topoConf;
     }
 
@@ -137,7 +136,7 @@ public void getTopologyPageInfoMasksDaemonAndTopologyCredentials() throws Except
             // topology-side credentials, masked on getTopologyConf and equally reachable here
             assertEquals(MASKED, served.get(Config.STORM_ZOOKEEPER_TOPOLOGY_AUTH_PAYLOAD),
                 "the topology ZooKeeper auth payload should be masked");
-            assertEquals(MASKED, served.get(BlowfishTupleSerializer.SECRET_KEY),
+            assertEquals(MASKED, served.get(Nimbus.LEGACY_BLOWFISH_SECRET_KEY),
                 "the tuple serializer key should be masked");
 
             // values that carry no credential are served untouched