This is an automated email from the ASF dual-hosted git repository.
RexXiong pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/celeborn.git
The following commit(s) were added to refs/heads/main by this push:
new 50c18eb6b6 [CELEBORN-2381] Authorize PbApplicationMetaRequest to
prevent secret disclosure
50c18eb6b6 is described below
commit 50c18eb6b66c1926d037441d954270b60f6f638f
Author: Nicholas Jiang <[email protected]>
AuthorDate: Tue Jul 14 21:58:10 2026 +0800
[CELEBORN-2381] Authorize PbApplicationMetaRequest to prevent secret
disclosure
### What changes were proposed in this pull request?
Call `checkAuth(context, appId)` before serving `PbApplicationMetaRequest`
in the Master, consistent with the other application-scoped handlers
(`RequestSlots`, `UnregisterShuffle`, `ApplicationLost`).
Workers fetch application meta over the internal channel, where the
connection has no per-application client id, so the check is a no-op for them;
it only rejects an external application that asks for another application's
secret.
### Why are the changes needed?
The Master served `PbApplicationMetaRequest` by returning the requested
application's SASL secret without an authorization check. With authentication
enabled, a caller on the external application port could read another
application's secret and impersonate it.
### Does this PR resolve a correctness bug?
- [ ] Yes
### Does this PR introduce _any_ user-facing change?
- [ ] Yes
### How was this patch tested?
- New `MasterApplicationMetaAuthSuite` verifying `PbApplicationMetaRequest`
is authorized against the registered application.
- New case in `MasterSuite` verifying a caller requesting another
application's secret is rejected.
Closes #3759 from SteNicholas/CELEBORN-2381.
Authored-by: Nicholas Jiang <[email protected]>
Signed-off-by: Shuang <[email protected]>
---
.../celeborn/service/deploy/master/Master.scala | 5 +-
.../master/MasterApplicationMetaAuthSuite.scala | 146 +++++++++++++++++++++
.../service/deploy/master/MasterSuite.scala | 50 ++++++-
3 files changed, 197 insertions(+), 4 deletions(-)
diff --git
a/master/src/main/scala/org/apache/celeborn/service/deploy/master/Master.scala
b/master/src/main/scala/org/apache/celeborn/service/deploy/master/Master.scala
index 6b25552060..8960934dd3 100644
---
a/master/src/main/scala/org/apache/celeborn/service/deploy/master/Master.scala
+++
b/master/src/main/scala/org/apache/celeborn/service/deploy/master/Master.scala
@@ -647,7 +647,10 @@ private[celeborn] class Master(
context))
case pb: PbApplicationMetaRequest =>
- // This request is from a worker
+ // Workers fetch application meta over the internal channel, where no
client id
+ // is set, so the check is a no-op for them; it only rejects an external
+ // application that asks for another application's secret.
+ checkAuth(context, pb.getAppId)
executeWithLeaderChecker(context,
handleRequestForApplicationMeta(context, pb))
case pb: PbRemoveWorkersUnavailableInfo =>
diff --git
a/master/src/test/scala/org/apache/celeborn/service/deploy/master/MasterApplicationMetaAuthSuite.scala
b/master/src/test/scala/org/apache/celeborn/service/deploy/master/MasterApplicationMetaAuthSuite.scala
new file mode 100644
index 0000000000..b672851cb0
--- /dev/null
+++
b/master/src/test/scala/org/apache/celeborn/service/deploy/master/MasterApplicationMetaAuthSuite.scala
@@ -0,0 +1,146 @@
+/*
+ * 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.celeborn.service.deploy.master
+
+import java.io.{PrintWriter, StringWriter}
+
+import scala.collection.mutable.ArrayBuffer
+
+import org.scalatest.BeforeAndAfterAll
+import org.scalatest.funsuite.AnyFunSuite
+
+import org.apache.celeborn.common.CelebornConf
+import org.apache.celeborn.common.metrics.source.Role
+import org.apache.celeborn.common.network.sasl.registration.RegistrationInfo
+import org.apache.celeborn.common.protocol.{PbApplicationMeta,
PbApplicationMetaRequest, RpcNameConstants, TransportModuleConstants}
+import org.apache.celeborn.common.rpc.{ClientSaslContextBuilder, RpcAddress,
RpcEndpointRef, RpcEnv, RpcSecurityContextBuilder}
+
+/**
+ * End-to-end authorization check for PbApplicationMetaRequest with auth
enabled, driving
+ * the real SASL registration path rather than a mocked client id (as in
[[MasterSuite]]).
+ *
+ * It covers the whole chain the security guarantee rests on: registration
sets the
+ * connection's client id, and checkAuth enforces it. A regression that
stopped setting
+ * the client id would silently turn checkAuth into a no-op and reopen the
cross-tenant
+ * secret leak while the mocked unit test still passed; this suite catches
that.
+ */
+class MasterApplicationMetaAuthSuite extends AnyFunSuite
+ with BeforeAndAfterAll
+ with MasterClusterFeature {
+
+ private val conf = new CelebornConf()
+ .set(CelebornConf.AUTH_ENABLED.key, "true")
+ .set(CelebornConf.INTERNAL_PORT_ENABLED.key, "true")
+
+ private var master: Master = _
+ private var externalAddress: RpcAddress = _
+ private var internalAddress: RpcAddress = _
+ private val clientEnvs = new ArrayBuffer[RpcEnv]()
+
+ override def beforeAll(): Unit = {
+ super.beforeAll()
+ master = setupMasterWithRandomPort(conf.getAll.toMap)
+ externalAddress = master.rpcEnv.address
+ internalAddress = master.internalRpcEnvInUse.address
+ }
+
+ override def afterAll(): Unit = {
+ clientEnvs.foreach(_.shutdown())
+ if (master != null) {
+ shutdownMaster()
+ }
+ super.afterAll()
+ }
+
+ private def metaRequest(appId: String): PbApplicationMetaRequest =
+ PbApplicationMetaRequest.newBuilder().setAppId(appId).build()
+
+ // A client env that authenticates and registers `appId` with the master
over the
+ // external port, exactly as an application (LifecycleManager) does.
+ private def registeredAppRef(appId: String, secret: String): RpcEndpointRef
= {
+ val securityContext = new RpcSecurityContextBuilder()
+ .withClientSaslContext(
+ new ClientSaslContextBuilder()
+ .withAddRegistrationBootstrap(true)
+ .withAppId(appId)
+ .withSaslUser(appId)
+ .withSaslPassword(secret)
+ .withRegistrationInfo(new RegistrationInfo())
+ .build())
+ .build()
+ val env = RpcEnv.create(
+ s"client-$appId",
+ TransportModuleConstants.RPC_SERVICE_MODULE,
+ "localhost",
+ 0,
+ conf,
+ Role.CLIENT,
+ Some(securityContext))
+ clientEnvs += env
+ env.setupEndpointRef(externalAddress, RpcNameConstants.MASTER_EP)
+ }
+
+ // A worker-like env reaching the master over the unauthenticated internal
port, where
+ // no per-application client id is set on the connection.
+ private def workerInternalRef(): RpcEndpointRef = {
+ val env = RpcEnv.create(
+ "worker-internal",
+ TransportModuleConstants.RPC_SERVICE_MODULE,
+ "localhost",
+ 0,
+ conf,
+ Role.WORKER,
+ None)
+ clientEnvs += env
+ env.setupEndpointRef(internalAddress, RpcNameConstants.MASTER_INTERNAL_EP)
+ }
+
+ private def stackTraceOf(t: Throwable): String = {
+ val sw = new StringWriter()
+ t.printStackTrace(new PrintWriter(sw))
+ sw.toString
+ }
+
+ test("PbApplicationMetaRequest is authorized against the registered
application") {
+ val victimApp = "victim-app"
+ val victimSecret = "victim-secret"
+ val attackerApp = "attacker-app"
+ val attackerSecret = "attacker-secret"
+
+ val victimRef = registeredAppRef(victimApp, victimSecret)
+ // Reading its own meta confirms the victim's secret is planted in the
master.
+
assert(victimRef.askSync[PbApplicationMeta](metaRequest(victimApp)).getSecret
== victimSecret)
+
+ // The attacker successfully registers its own app (so the connection is
genuine and
+ // authenticated), but the registration path set its client id to the
attacker's app,
+ // so checkAuth rejects the cross-application read of the victim's secret.
+ val attackerRef = registeredAppRef(attackerApp, attackerSecret)
+ assert(
+
attackerRef.askSync[PbApplicationMeta](metaRequest(attackerApp)).getSecret ==
attackerSecret)
+ val e = intercept[Exception] {
+ attackerRef.askSync[PbApplicationMeta](metaRequest(victimApp))
+ }
+ assert(stackTraceOf(e).contains(s"not authorized for application
$victimApp"))
+
+ // A worker over the internal port carries no client id, so it can still
fetch the
+ // victim's meta — the legitimate path must keep working.
+ assert(
+
workerInternalRef().askSync[PbApplicationMeta](metaRequest(victimApp)).getSecret
+ == victimSecret)
+ }
+}
diff --git
a/master/src/test/scala/org/apache/celeborn/service/deploy/master/MasterSuite.scala
b/master/src/test/scala/org/apache/celeborn/service/deploy/master/MasterSuite.scala
index 0a8a7b592f..d915320c95 100644
---
a/master/src/test/scala/org/apache/celeborn/service/deploy/master/MasterSuite.scala
+++
b/master/src/test/scala/org/apache/celeborn/service/deploy/master/MasterSuite.scala
@@ -21,16 +21,18 @@ import java.nio.file.Files
import java.util
import org.mockito.ArgumentCaptor
-import org.mockito.Mockito.{mock, verify}
+import org.mockito.Mockito.{mock, verify, when}
import org.scalatest.{BeforeAndAfterAll, BeforeAndAfterEach}
import org.scalatest.funsuite.AnyFunSuite
import org.apache.celeborn.common.CelebornConf
import org.apache.celeborn.common.identity.UserIdentifier
-import org.apache.celeborn.common.protocol.{PbCheckForWorkerTimeout,
PbRegisterWorker}
+import org.apache.celeborn.common.network.client.{RpcResponseCallback,
TransportClient}
+import org.apache.celeborn.common.protocol.{PbApplicationMetaRequest,
PbCheckForWorkerTimeout, PbRegisterWorker}
import
org.apache.celeborn.common.protocol.message.ControlMessages.{RequestSlots,
RequestSlotsResponse}
import org.apache.celeborn.common.protocol.message.StatusCode
-import org.apache.celeborn.common.rpc.RpcCallContext
+import org.apache.celeborn.common.rpc.{RpcAddress, RpcCallContext}
+import org.apache.celeborn.common.rpc.netty.{NettyRpcEnv,
RemoteNettyRpcCallContext}
import org.apache.celeborn.common.util.{CelebornExitKind, ThreadUtils}
class MasterSuite extends AnyFunSuite
@@ -197,4 +199,46 @@ class MasterSuite extends AnyFunSuite
master.rpcEnv.shutdown()
}
+
+ test("PbApplicationMetaRequest rejects a caller requesting another
application's secret") {
+ val conf = new CelebornConf()
+ val randomMasterPort = selectRandomPort()
+ val randomHttpPort = selectRandomPort()
+ conf.set(CelebornConf.HA_ENABLED.key, "false")
+ conf.set(CelebornConf.MASTER_HTTP_HOST.key, "127.0.0.1")
+ conf.set(CelebornConf.MASTER_HTTP_PORT.key, randomHttpPort.toString)
+
+ val args = Array("-h", "localhost", "-p", randomMasterPort.toString)
+ val masterArgs = new MasterArguments(args, conf)
+ val master = new Master(conf, masterArgs)
+
+ // Builds a remote call context whose connection is authenticated as
`clientId`;
+ // null models a worker on the internal channel, which sets no client id.
+ def contextForClient(clientId: String): RemoteNettyRpcCallContext = {
+ val client = mock(classOf[TransportClient])
+ when(client.getClientId).thenReturn(clientId)
+ new RemoteNettyRpcCallContext(
+ mock(classOf[NettyRpcEnv]),
+ mock(classOf[RpcResponseCallback]),
+ RpcAddress("localhost", 1234),
+ client)
+ }
+
+ val request =
PbApplicationMetaRequest.newBuilder().setAppId("victim-app").build()
+ val unhandled = (_: Any) => fail("PbApplicationMetaRequest was not
handled")
+
+ try {
+ // An application authenticated as "attacker-app" on the external port
must not
+ // be able to read "victim-app"'s secret.
+ val e = intercept[IllegalStateException] {
+
master.receiveAndReply(contextForClient("attacker-app")).applyOrElse(request,
unhandled)
+ }
+ assert(e.getMessage.contains("not authorized for application
victim-app"))
+
+ // A worker carries no client id, so the guard is a no-op and the
request is served.
+ master.receiveAndReply(contextForClient(null)).applyOrElse(request,
unhandled)
+ } finally {
+ master.rpcEnv.shutdown()
+ }
+ }
}
