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..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 @@ -437,13 +437,15 @@ object SparkConnectService extends Logging { sb.permitKeepAliveWithoutCalls(true) sb.addService(sparkConnectService) + // Add all registered interceptors to the server builder. + SparkConnectInterceptorRegistry.chainInterceptors(sb, configuredInterceptors) + + // 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)) } - // 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") + } +}