This is an automated email from the ASF dual-hosted git repository.
rzo1 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/storm.git
The following commit(s) were added to refs/heads/master by this push:
new c1c3338fd Generate tuple tree ids from a key stream instead of a
recoverable linear congruential generator (#8989)
c1c3338fd is described below
commit c1c3338fd31c6c88f3371404745cbe0ae637275b
Author: Richard Zowalla <[email protected]>
AuthorDate: Sat Aug 22 19:41:23 2026 +0200
Generate tuple tree ids from a key stream instead of a recoverable linear
congruential generator (#8989)
---
.../jvm/org/apache/storm/executor/Executor.java | 3 +-
.../org/apache/storm/utils/KeyStreamRandom.java | 107 +++++++++++++++++++++
.../apache/storm/executor/SpoutExecutorTest.java | 28 ++++++
.../jvm/org/apache/storm/tuple/MessageIdTest.java | 99 +++++++++++++++++++
4 files changed, 236 insertions(+), 1 deletion(-)
diff --git a/storm-client/src/jvm/org/apache/storm/executor/Executor.java
b/storm-client/src/jvm/org/apache/storm/executor/Executor.java
index d488a6263..716ec48c4 100644
--- a/storm-client/src/jvm/org/apache/storm/executor/Executor.java
+++ b/storm-client/src/jvm/org/apache/storm/executor/Executor.java
@@ -84,6 +84,7 @@ import org.apache.storm.tuple.TupleImpl;
import org.apache.storm.tuple.Values;
import org.apache.storm.utils.ConfigUtils;
import org.apache.storm.utils.JCQueue;
+import org.apache.storm.utils.KeyStreamRandom;
import org.apache.storm.utils.ObjectReader;
import org.apache.storm.utils.Time;
import org.apache.storm.utils.Utils;
@@ -180,7 +181,7 @@ public abstract class Executor implements Callable,
JCQueue.Consumer {
this.reportErrorDie = new ReportErrorAndDie(reportError, suicideFn);
this.sampler = ConfigUtils.mkStatsSampler(topoConf);
this.isDebug =
ObjectReader.getBoolean(topoConf.get(Config.TOPOLOGY_DEBUG), false);
- this.rand = new Random(Utils.secureRandomLong());
+ this.rand = new KeyStreamRandom();
this.credentials = credentials;
this.hasEventLoggers = StormCommon.hasEventLoggers(topoConf);
this.ackingEnabled = StormCommon.hasAckers(topoConf);
diff --git a/storm-client/src/jvm/org/apache/storm/utils/KeyStreamRandom.java
b/storm-client/src/jvm/org/apache/storm/utils/KeyStreamRandom.java
new file mode 100644
index 000000000..a8c7903f5
--- /dev/null
+++ b/storm-client/src/jvm/org/apache/storm/utils/KeyStreamRandom.java
@@ -0,0 +1,107 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership. The
ASF licenses this file to you under the Apache License, Version
+ * 2.0 (the "License"); you may not use this file except in compliance with
the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions
+ * and limitations under the License.
+ */
+
+package org.apache.storm.utils;
+
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.security.GeneralSecurityException;
+import java.security.SecureRandom;
+import java.util.Random;
+import javax.crypto.Cipher;
+import javax.crypto.spec.IvParameterSpec;
+import javax.crypto.spec.SecretKeySpec;
+
+/**
+ * A {@link Random} that returns slices of an AES counter mode key stream,
keyed from {@link SecureRandom} when the instance is
+ * created and buffered a block at a time. It is meant for values that other
parties must not be able to guess, such as the tuple
+ * tree ids handed out by {@link
org.apache.storm.tuple.MessageId#generateId(Random)}: unlike {@link Random},
+ * {@link java.util.SplittableRandom} and {@link
java.util.concurrent.ThreadLocalRandom}, whose internal state is recoverable
from
+ * a couple of returned values, the values returned here say nothing about the
values returned next.
+ *
+ * <p>The buffering keeps it cheap enough for the emit path; it is measurably
faster than {@link Random}, whose seed update is a
+ * contended compare and set per value. It is not a drop in replacement for a
seeded {@link Random} though, because it cannot be
+ * reseeded and so cannot produce a repeatable sequence.</p>
+ */
+public class KeyStreamRandom extends Random {
+ private static final long serialVersionUID = 1L;
+ private static final String TRANSFORMATION = "AES/CTR/NoPadding";
+ private static final int KEY_BYTES = 16;
+ private static final int DEFAULT_BUFFER_LONGS = 512;
+
+ private final SecureRandom source = new SecureRandom();
+ private final Cipher cipher;
+ private final byte[] input;
+ private final byte[] output;
+ private final long[] buffer;
+ private int position;
+
+ public KeyStreamRandom() {
+ this(DEFAULT_BUFFER_LONGS);
+ }
+
+ KeyStreamRandom(int bufferLongs) {
+ this.cipher = newCipher(source);
+ this.input = new byte[bufferLongs * Long.BYTES];
+ this.output = new byte[bufferLongs * Long.BYTES];
+ this.buffer = new long[bufferLongs];
+ // start empty, the first call fills the buffer
+ this.position = bufferLongs;
+ }
+
+ private static Cipher newCipher(SecureRandom source) {
+ byte[] key = new byte[KEY_BYTES];
+ byte[] iv = new byte[KEY_BYTES];
+ source.nextBytes(key);
+ source.nextBytes(iv);
+ try {
+ Cipher cipher = Cipher.getInstance(TRANSFORMATION);
+ cipher.init(Cipher.ENCRYPT_MODE, new SecretKeySpec(key, "AES"),
new IvParameterSpec(iv));
+ return cipher;
+ } catch (GeneralSecurityException e) {
+ // no AES here, fall back to taking the values from the
SecureRandom itself
+ return null;
+ }
+ }
+
+ @Override
+ public synchronized long nextLong() {
+ if (position == buffer.length) {
+ fill();
+ }
+ return buffer[position++];
+ }
+
+ @Override
+ protected int next(int bits) {
+ return (int) (nextLong() >>> (Long.SIZE - bits));
+ }
+
+ @Override
+ public void setSeed(long seed) {
+ // there is nothing to seed, and Random's constructor calls this
before this class' fields exist
+ }
+
+ private void fill() {
+ if (cipher == null) {
+ source.nextBytes(output);
+ } else {
+ try {
+ cipher.update(input, 0, input.length, output, 0);
+ } catch (GeneralSecurityException e) {
+ throw Utils.wrapInRuntime(e);
+ }
+ }
+
ByteBuffer.wrap(output).order(ByteOrder.nativeOrder()).asLongBuffer().get(buffer);
+ position = 0;
+ }
+}
diff --git
a/storm-client/test/jvm/org/apache/storm/executor/SpoutExecutorTest.java
b/storm-client/test/jvm/org/apache/storm/executor/SpoutExecutorTest.java
index a44597cde..b83009f89 100644
--- a/storm-client/test/jvm/org/apache/storm/executor/SpoutExecutorTest.java
+++ b/storm-client/test/jvm/org/apache/storm/executor/SpoutExecutorTest.java
@@ -22,6 +22,7 @@ import org.apache.storm.metrics2.StormMetricRegistry;
import org.apache.storm.task.WorkerTopologyContext;
import org.apache.storm.tuple.AddressedTuple;
import org.apache.storm.tuple.TupleImpl;
+import org.apache.storm.utils.KeyStreamRandom;
import org.apache.storm.utils.RotatingMap;
import org.apache.storm.utils.Utils;
import org.junit.jupiter.api.Test;
@@ -33,6 +34,7 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
import static org.mockito.ArgumentMatchers.anyInt;
import static org.mockito.ArgumentMatchers.anyString;
@@ -84,4 +86,30 @@ public class SpoutExecutorTest {
Mockito.verify(rotatingMap,Mockito.times(1)).rotate();
}
+
+ @Test
+ public void testTupleTreeIdsComeFromAnUnguessableGenerator() {
+
+ RateCounter rateCounter = Mockito.mock(RateCounter.class);
+
+ StormMetricRegistry stormMetricRegistry =
Mockito.mock(StormMetricRegistry.class);
+
Mockito.when(stormMetricRegistry.rateCounter(anyString(),anyString(),anyInt())).thenReturn(rateCounter);
+
+ ComponentCommon componentCommon = Mockito.mock(ComponentCommon.class);
+ Mockito.when(componentCommon.get_json_conf()).thenReturn(null);
+
+ WorkerTopologyContext workerTopologyContext =
Mockito.mock(WorkerTopologyContext.class);
+
Mockito.when(workerTopologyContext.getComponentId(anyInt())).thenReturn("1");
+
Mockito.when(workerTopologyContext.getComponentCommon(anyString())).thenReturn(componentCommon);
+
+ WorkerState workerState = Mockito.mock(WorkerState.class);
+
Mockito.when(workerState.getWorkerTopologyContext()).thenReturn(workerTopologyContext);
+
Mockito.when(workerState.getStateStorage()).thenReturn(Mockito.mock(IStateStorage.class));
+
Mockito.when(workerState.getTopologyConf()).thenReturn(Utils.readDefaultConfig());
+
Mockito.when(workerState.getMetricRegistry()).thenReturn(stormMetricRegistry);
+
+ SpoutExecutor spoutExecutor = new
SpoutExecutor(workerState,List.of(1L,5L),new HashMap<>());
+
+ assertInstanceOf(KeyStreamRandom.class, spoutExecutor.rand);
+ }
}
diff --git a/storm-client/test/jvm/org/apache/storm/tuple/MessageIdTest.java
b/storm-client/test/jvm/org/apache/storm/tuple/MessageIdTest.java
new file mode 100644
index 000000000..a8ef0559b
--- /dev/null
+++ b/storm-client/test/jvm/org/apache/storm/tuple/MessageIdTest.java
@@ -0,0 +1,99 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership. The
ASF licenses this file to you under the Apache License, Version
+ * 2.0 (the "License"); you may not use this file except in compliance with
the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions
+ * and limitations under the License.
+ */
+
+package org.apache.storm.tuple;
+
+import java.util.HashSet;
+import java.util.Random;
+import java.util.Set;
+import org.apache.storm.utils.KeyStreamRandom;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+
+public class MessageIdTest {
+
+ private static final long LCG_MULTIPLIER = 0x5DEECE66DL;
+ private static final long LCG_ADDEND = 0xBL;
+ private static final long LCG_MASK = (1L << 48) - 1;
+
+ @Test
+ public void generateIdReturnsDistinctValues() {
+ Random rand = new KeyStreamRandom();
+ Set<Long> ids = new HashSet<>();
+ for (int i = 0; i < 100000; i++) {
+ ids.add(MessageId.generateId(rand));
+ }
+ assertEquals(100000, ids.size());
+ }
+
+ @Test
+ public void generatedIdsDoNotRevealTheFollowingIds() {
+ Random rand = new KeyStreamRandom();
+ for (int i = 0; i < 20; i++) {
+ long first = MessageId.generateId(rand);
+ long second = MessageId.generateId(rand);
+ assertNotEquals(second, predictNextFromLcgState(first));
+ }
+ }
+
+ /**
+ * Guards the test above: the same prediction does work against a
java.util.Random, so a failure of
+ * generatedIdsDoNotRevealTheFollowingIds means the ids really are
predictable rather than the prediction being broken.
+ */
+ @Test
+ public void lcgPredictionWorksAgainstJavaUtilRandom() {
+ Random rand = new Random(4242L);
+ for (int i = 0; i < 20; i++) {
+ long first = MessageId.generateId(rand);
+ long second = MessageId.generateId(rand);
+ assertEquals(second, predictNextFromLcgState(first));
+ }
+ }
+
+ @Test
+ public void makeRootIdKeepsTheGeneratedIds() {
+ Random rand = new KeyStreamRandom();
+ long id = MessageId.generateId(rand);
+ long val = MessageId.generateId(rand);
+ MessageId messageId = MessageId.makeRootId(id, val);
+ assertEquals(val, messageId.getAnchorsToIds().get(id));
+ assertNotNull(messageId.toString());
+ }
+
+ /**
+ * Recovers the 48 bit state of a java.util.Random from a single nextLong
output and returns the nextLong it would produce
+ * afterwards, or 0 if the state could not be recovered.
+ */
+ private static long predictNextFromLcgState(long observed) {
+ int low = (int) observed;
+ int high = (int) ((observed - low) >>> 32);
+ long partialSeed = (high & 0xFFFFFFFFL) << 16;
+ for (int guess = 0; guess < (1 << 16); guess++) {
+ long seed = nextSeed(partialSeed | guess);
+ if ((int) (seed >>> 16) == low) {
+ seed = nextSeed(seed);
+ long nextHigh = seed >>> 16;
+ seed = nextSeed(seed);
+ long nextLow = seed >>> 16;
+ return (nextHigh << 32) + (int) nextLow;
+ }
+ }
+ return 0;
+ }
+
+ private static long nextSeed(long seed) {
+ return (seed * LCG_MULTIPLIER + LCG_ADDEND) & LCG_MASK;
+ }
+}
