This is an automated email from the ASF dual-hosted git repository.
peter-toth pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new 2062ed91842d [SPARK-57889][CORE] Authorize StreamRequest consistently
with ChunkFetchRequest in the transport layer
2062ed91842d is described below
commit 2062ed91842dc531f374a3144b08266cb4225213
Author: Peter Toth <[email protected]>
AuthorDate: Fri Jul 3 10:24:25 2026 +0200
[SPARK-57889][CORE] Authorize StreamRequest consistently with
ChunkFetchRequest in the transport layer
### What changes were proposed in this pull request?
`TransportRequestHandler.processStreamRequest` now calls
`streamManager.checkAuthorization` before `openStream`, mirroring
`ChunkFetchRequestHandler`. A `checkAuthorization(TransportClient, String)`
overload is added to `StreamManager` (no-op default) and overridden in
`OneForOneStreamManager` to parse the `"streamId_chunkIndex"` id and delegate
to the existing `checkAuthorization(TransportClient, long)`.
### Why are the changes needed?
The transport layer serves both `ChunkFetchRequest` and `StreamRequest`
from the same `StreamManager`. `ChunkFetchRequestHandler` calls
`streamManager.checkAuthorization` to enforce per-application ownership
(`clientId == appId`), but `processStreamRequest` called `openStream` directly
with no such check. Because `OneForOneStreamManager.openStream` addresses the
same registered streams by their string id, a `StreamRequest` could read a
stream that the `ChunkFetchRequest` path would ha [...]
### Does this PR introduce _any_ user-facing change?
No. This is internal transport-layer authorization; like the existing
`ChunkFetchRequest` check it enforces only for authenticated clients (non-null
client id), and behavior is unchanged when authentication is disabled.
### How was this patch tested?
New `OneForOneStreamManagerSuite.testStreamRequestAuthorization` (owner
allowed, other-application rejected with `SecurityException`, null client id
no-op). Existing `OneForOneStreamManagerSuite`, `TransportRequestHandlerSuite`,
and `ChunkFetchRequestHandlerSuite` pass.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 4.8)
Closes #56970 from peter-toth/SPARK-57889-streamrequest-authz.
Authored-by: Peter Toth <[email protected]>
Signed-off-by: Peter Toth <[email protected]>
(cherry picked from commit fd05a3bd52e78e079df6763fffbf74f901756cc4)
Signed-off-by: Peter Toth <[email protected]>
---
.../network/server/OneForOneStreamManager.java | 8 ++++++
.../apache/spark/network/server/StreamManager.java | 10 ++++++++
.../network/server/TransportRequestHandler.java | 1 +
.../server/OneForOneStreamManagerSuite.java | 29 ++++++++++++++++++++++
4 files changed, 48 insertions(+)
diff --git
a/common/network-common/src/main/java/org/apache/spark/network/server/OneForOneStreamManager.java
b/common/network-common/src/main/java/org/apache/spark/network/server/OneForOneStreamManager.java
index cb53d565e7e8..e13bc1f143cf 100644
---
a/common/network-common/src/main/java/org/apache/spark/network/server/OneForOneStreamManager.java
+++
b/common/network-common/src/main/java/org/apache/spark/network/server/OneForOneStreamManager.java
@@ -178,6 +178,14 @@ public class OneForOneStreamManager extends StreamManager {
}
}
+ @Override
+ public void checkAuthorization(TransportClient client, String streamChunkId)
{
+ // StreamRequest addresses the same registered streams as
ChunkFetchRequest, via a
+ // "streamId_chunkIndex" string. Authorize on the underlying stream id so
that this path
+ // enforces the same per-application ownership check as the chunk-fetch
path.
+ checkAuthorization(client, parseStreamChunkId(streamChunkId).getLeft());
+ }
+
@Override
public void chunkBeingSent(long streamId) {
StreamState streamState = streams.get(streamId);
diff --git
a/common/network-common/src/main/java/org/apache/spark/network/server/StreamManager.java
b/common/network-common/src/main/java/org/apache/spark/network/server/StreamManager.java
index e48d27be1126..63230e15a41d 100644
---
a/common/network-common/src/main/java/org/apache/spark/network/server/StreamManager.java
+++
b/common/network-common/src/main/java/org/apache/spark/network/server/StreamManager.java
@@ -73,6 +73,16 @@ public abstract class StreamManager {
*/
public void checkAuthorization(TransportClient client, long streamId) { }
+ /**
+ * Verify that the client is authorized to read from the given stream. This
variant covers the
+ * {@link #openStream(String)} path (i.e. {@code StreamRequest}). The
default implementation is a
+ * no-op; subclasses that key streams on a string id should override it,
typically by delegating
+ * to {@link #checkAuthorization(TransportClient, long)}.
+ *
+ * @throws SecurityException If client is not authorized.
+ */
+ public void checkAuthorization(TransportClient client, String streamId) { }
+
/**
* Return the number of chunks being transferred and not finished yet in
this StreamManager.
*/
diff --git
a/common/network-common/src/main/java/org/apache/spark/network/server/TransportRequestHandler.java
b/common/network-common/src/main/java/org/apache/spark/network/server/TransportRequestHandler.java
index 464d4d9eb378..803b26c0678b 100644
---
a/common/network-common/src/main/java/org/apache/spark/network/server/TransportRequestHandler.java
+++
b/common/network-common/src/main/java/org/apache/spark/network/server/TransportRequestHandler.java
@@ -140,6 +140,7 @@ public class TransportRequestHandler extends
MessageHandler<RequestMessage> {
}
ManagedBuffer buf;
try {
+ streamManager.checkAuthorization(reverseClient, req.streamId);
buf = streamManager.openStream(req.streamId);
} catch (Exception e) {
logger.error("Error opening stream {} for request from {}", e,
diff --git
a/common/network-common/src/test/java/org/apache/spark/network/server/OneForOneStreamManagerSuite.java
b/common/network-common/src/test/java/org/apache/spark/network/server/OneForOneStreamManagerSuite.java
index 136154cc2aa6..31401f104fa5 100644
---
a/common/network-common/src/test/java/org/apache/spark/network/server/OneForOneStreamManagerSuite.java
+++
b/common/network-common/src/test/java/org/apache/spark/network/server/OneForOneStreamManagerSuite.java
@@ -29,6 +29,7 @@ import org.mockito.Mockito;
import org.apache.spark.network.TestManagedBuffer;
import org.apache.spark.network.buffer.ManagedBuffer;
+import org.apache.spark.network.client.TransportClient;
public class OneForOneStreamManagerSuite {
@@ -162,4 +163,32 @@ public class OneForOneStreamManagerSuite {
Mockito.verify(mockManagedBuffer, Mockito.times(1)).release();
Assertions.assertEquals(0, manager.numStreamStates());
}
+
+ @Test
+ public void testStreamRequestAuthorization() {
+ OneForOneStreamManager manager = new OneForOneStreamManager();
+ Channel dummyChannel = Mockito.mock(Channel.class,
Mockito.RETURNS_SMART_NULLS);
+ long streamId = manager.registerStream(
+ "app1", new ArrayList<ManagedBuffer>().iterator(), dummyChannel);
+ // StreamRequest addresses a registered stream via a "streamId_chunkIndex"
string.
+ String streamChunkId = OneForOneStreamManager.genStreamChunkId(streamId,
0);
+
+ // A client authenticated as a different application is rejected, matching
the
+ // ChunkFetchRequest path.
+ TransportClient otherApp = Mockito.mock(TransportClient.class);
+ Mockito.when(otherApp.getClientId()).thenReturn("app2");
+ Assertions.assertThrows(SecurityException.class,
+ () -> manager.checkAuthorization(otherApp, streamChunkId));
+
+ // The owning application is allowed.
+ TransportClient owner = Mockito.mock(TransportClient.class);
+ Mockito.when(owner.getClientId()).thenReturn("app1");
+ manager.checkAuthorization(owner, streamChunkId);
+
+ // With authentication disabled the client id is null and the check is a
no-op, preserving
+ // the default behavior.
+ TransportClient noAuth = Mockito.mock(TransportClient.class);
+ Mockito.when(noAuth.getClientId()).thenReturn(null);
+ manager.checkAuthorization(noAuth, streamChunkId);
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
