From dbbae8d5b5096f8a83a87916d1cebc7a0e4b07f2 Mon Sep 17 00:00:00 2001 From: Holden Karau Date: Thu, 27 Aug 2026 23:19:58 +0000 Subject: [PATCH 1/2] [SPARK-59087][CONNECT] Authenticate before the other gRPC interceptors run ### What changes were proposed in this pull request? `startGRPCService` registers `PreSharedKeyAuthenticationInterceptor` after the entries from `SparkConnectInterceptorRegistry` rather than before them. A `ServerBuilder` invokes interceptors in the reverse of the order they were added, so the one added last is the outermost and runs first. That is the documented contract of `ServerBuilder.intercept` -- "Interceptors run in the reverse order in which they are added, just as with consecutive calls to `ServerInterceptors.intercept()`": https://github.com/grpc/grpc-java/blob/v1.76.0/api/src/main/java/io/grpc/ServerBuilder.java#L134-L144 Adds `SparkConnectAuthInterceptorOrderSuite`, which configures an interceptor through `spark.connect.grpc.interceptor.classes` and asserts it does not run for a call that fails authentication, then that it does run for a call that passes -- so the first assertion cannot hold merely because the interceptor was never wired up. ### Why are the changes needed? Authentication was the innermost interceptor, so it ran last. That is not apparent from reading `startGRPCService`, because it follows from the reverse-order contract above rather than from the order the `intercept` calls appear in. The practical consequence is that a new interceptor gets pre-authentication placement by default, which is the wrong default to leave in place. Nothing here is exploitable today, and this is not a security fix. `RequestDecompressionInterceptor.interceptCall` allocates an `AtomicReference` and attaches a gRPC `Context`; it never reads `headers`, and all of its real work -- plan decompression and the associated logging -- happens in `onMessage`, which gRPC does not invoke once the authentication interceptor has closed the call. Under the old order an unauthenticated caller therefore reached two object allocations and nothing else. The change also does not make authentication the first thing that runs. Authentication is itself an interceptor, so TLS, HTTP/2 framing, header block decoding, gRPC metadata parsing and method dispatch all precede it either way. What it buys is a sane default for the interceptor layer: an interceptor that does do work in `interceptCall` or `onMessage`, in-tree or named by an operator in `spark.connect.grpc.interceptor.classes`, is no longer handed calls that are about to be rejected as unauthenticated. ### Does this PR introduce _any_ user-facing change? Yes, for a server that configures both an authentication token and `spark.connect.grpc.interceptor.classes`: those interceptors no longer see calls that fail authentication. An interceptor that supplied the `Authorization` header itself, for example translating another scheme into a bearer token, relied on running ahead of authentication and no longer does. That ordering was never specified, but it is a behavior change for anyone depending on it. ### How was this patch tested? `SparkConnectAuthInterceptorOrderSuite` plus `SparkConnectAuthSuite`, `InterceptorRegistrySuite` and `SparkConnectServiceKeepAliveSuite`: 16 tests, all passing, scalastyle clean. Reverting just the registration order and rerunning the new suite fails with "1 did not equal 0 a configured interceptor ran for a call that failed authentication", so the assertion is load-bearing. The reverse-invocation order and the fact that a closed call stops message delivery were both confirmed against grpc 1.76.0 with a standalone in-process harness: with the old registration order the outer interceptor's `interceptCall` ran once for an unauthenticated call while its listener saw zero `onMessage` and zero `onHalfClose` callbacks and the service handler was never invoked; with the new order it did not run at all, and on an authenticated call it ran and saw the message. ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Claude Code (Claude Opus 5) Co-Authored-By: Claude Opus 5 (1M context) Co-Authored-By: Holden Karau --- .../connect/service/SparkConnectService.scala | 11 ++- ...parkConnectAuthInterceptorOrderSuite.scala | 86 +++++++++++++++++++ 2 files changed, 94 insertions(+), 3 deletions(-) create mode 100644 sql/connect/server/src/test/scala/org/apache/spark/sql/connect/service/SparkConnectAuthInterceptorOrderSuite.scala diff --git a/sql/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectService.scala b/sql/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectService.scala index 1fa9fbeda1e5..4a14fdcea576 100644 --- a/sql/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectService.scala +++ b/sql/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectService.scala @@ -437,13 +437,18 @@ object SparkConnectService extends Logging { sb.permitKeepAliveWithoutCalls(true) sb.addService(sparkConnectService) + // Add all registered interceptors to the server builder. + SparkConnectInterceptorRegistry.chainInterceptors(sb, configuredInterceptors) + + // Registered after the others on purpose. A ServerBuilder invokes interceptors in the + // reverse of the order they were added, so the one added last is the outermost and runs + // first. Authenticating first means an unauthenticated call is turned away before the + // decompression interceptor and before anything an operator put in + // spark.connect.grpc.interceptor.classes runs on its behalf. getAuthenticateToken.foreach { token => sb.intercept(new PreSharedKeyAuthenticationInterceptor(token)) } - // Add all registered interceptors to the server builder. - SparkConnectInterceptorRegistry.chainInterceptors(sb, configuredInterceptors) - // If debug mode is configured, load the ProtoReflection service so that tools like // grpcurl can introspect the API for debugging. protoReflectionService.foreach(service => sb.addService(service)) diff --git a/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/service/SparkConnectAuthInterceptorOrderSuite.scala b/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/service/SparkConnectAuthInterceptorOrderSuite.scala new file mode 100644 index 000000000000..867f375bc030 --- /dev/null +++ b/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/service/SparkConnectAuthInterceptorOrderSuite.scala @@ -0,0 +1,86 @@ +/* + * 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.spark.sql.connect.service + +import java.util.concurrent.atomic.AtomicInteger + +import io.grpc.{Metadata, ServerCall, ServerCallHandler, ServerInterceptor} + +import org.apache.spark.{SparkConf, SparkException} +import org.apache.spark.sql.connect.{SparkConnectServerTest, SparkSession} +import org.apache.spark.sql.connect.config.Connect + +/** + * Counts the calls it is handed, so a test can tell whether it ran at all. Needs a no-argument + * constructor to be loadable from `spark.connect.grpc.interceptor.classes`. + */ +class CallCountingInterceptor extends ServerInterceptor { + override def interceptCall[ReqT, RespT]( + call: ServerCall[ReqT, RespT], + headers: Metadata, + next: ServerCallHandler[ReqT, RespT]): ServerCall.Listener[ReqT] = { + CallCountingInterceptor.calls.incrementAndGet() + next.startCall(call, headers) + } +} + +object CallCountingInterceptor { + val calls = new AtomicInteger(0) +} + +/** + * Tests that authentication runs ahead of the other interceptors, rather than behind them. + * + * A `ServerBuilder` invokes interceptors in the reverse of the order they were added, so where + * `PreSharedKeyAuthenticationInterceptor` is registered decides how much of the pipeline an + * unauthenticated caller can drive before being turned away. + */ +class SparkConnectAuthInterceptorOrderSuite extends SparkConnectServerTest { + + private val token = "deadbeef" + + override protected def sparkConf: SparkConf = + super.sparkConf.set(Connect.CONNECT_AUTHENTICATE_TOKEN.key, token) + + override protected def extraServerConfs: Seq[(String, String)] = Seq( + Connect.CONNECT_GRPC_INTERCEPTOR_CLASSES.key -> classOf[CallCountingInterceptor].getName) + + test("an unauthenticated call is rejected before the configured interceptors run") { + CallCountingInterceptor.calls.set(0) + + val anonymous = SparkSession + .builder() + .remote(s"sc://localhost:${SparkConnectService.localPort}/") + .create() + val e = intercept[SparkException](anonymous.range(5).collect()) + assert(e.getMessage.contains("No authentication token provided")) + assert( + CallCountingInterceptor.calls.get() === 0, + "a configured interceptor ran for a call that failed authentication") + + // Without this the assertion above would also hold if the interceptor were simply never + // wired up, so prove it does run once the caller authenticates. + val authenticated = SparkSession + .builder() + .remote(s"sc://localhost:${SparkConnectService.localPort}/;token=$token") + .create() + assert(authenticated.range(5).collect().length === 5) + assert( + CallCountingInterceptor.calls.get() > 0, + "the configured interceptor never ran, so this suite proves nothing") + } +} From dcba3667e16edb48131a1d02808cf92b0b9ec6b5 Mon Sep 17 00:00:00 2001 From: Holden Karau Date: Fri, 28 Aug 2026 20:48:56 +0000 Subject: [PATCH 2/2] Simplify comment Co-Authored-By: Holden Karau --- .../spark/sql/connect/service/SparkConnectService.scala | 7 ++----- 1 file changed, 2 insertions(+), 5 deletions(-) diff --git a/sql/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectService.scala b/sql/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectService.scala index 4a14fdcea576..1b95a7f126e8 100644 --- a/sql/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectService.scala +++ b/sql/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectService.scala @@ -440,11 +440,8 @@ object SparkConnectService extends Logging { // Add all registered interceptors to the server builder. SparkConnectInterceptorRegistry.chainInterceptors(sb, configuredInterceptors) - // Registered after the others on purpose. A ServerBuilder invokes interceptors in the - // reverse of the order they were added, so the one added last is the outermost and runs - // first. Authenticating first means an unauthenticated call is turned away before the - // decompression interceptor and before anything an operator put in - // spark.connect.grpc.interceptor.classes runs on its behalf. + // A ServerBuilder invokes interceptors in the reverse of the order they were added so add + // auth at the end so it runs first. getAuthenticateToken.foreach { token => sb.intercept(new PreSharedKeyAuthenticationInterceptor(token)) }