dongjoon-hyun commented on code in PR #58392: URL: https://github.com/apache/spark/pull/58392#discussion_r3990447849
########## 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 { Review Comment: `SparkConnectAuthSuite` already extends the same `SparkConnectServerTest` base with the same `deadbeef` token, so this suite boots a second `TestSparkSession` plus a real Netty server just to hold one test. Adding ```scala override protected def extraServerConfs: Seq[(String, String)] = Seq( Connect.CONNECT_GRPC_INTERCEPTOR_CLASSES.key -> classOf[CallCountingInterceptor].getName) ``` to `SparkConnectAuthSuite` and moving this test there should be equivalent: `withSparkEnvConfs` applies the key before `start()` and removes it afterwards, the interceptor is pass-through, and the existing "not readable through the Config RPC" test only asserts on the token and marker keys. The "prove it is wired" half could then reuse the authenticated `range(5).collect()` that suite already makes. ########## 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 => Review Comment: Optional: the auth-outermost invariant now lives only in the order of these two statements plus the comment, while `SparkConnectInterceptorRegistry` documents itself as the owner of the chain order (and its 1-arg `chainInterceptors(sb)` overload knows nothing about auth). A future `sb.intercept(...)` added below this block, which is the natural place for a new line, would silently run ahead of authentication again. One way to encode it in one place is to let the registry append the auth interceptor as the final element, e.g. `chainInterceptors(sb, configuredInterceptors, authInterceptor: Option[ServerInterceptor])`. Fine to defer given the new suite guards it end-to-end. ########## 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 Review Comment: Neither `anonymous` nor `authenticated` is ever closed. `close()` is what releases the server session, shuts down the `ManagedChannel`, and closes the per-session `RootAllocator`; `SparkConnectServerTest.afterAll` only stops the server, so both channels and their executor/netty threads outlive the suite (the thread audit only warns). `withSession` cannot be used here because it does not expose `.token(...)` on the client builder, so a `try { ... } finally session.close()` around each would do. `SparkConnectAuthSuite` has the same pattern, so this is precedent rather than a regression. ########## 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) Review Comment: Nit: the count is only ever compared with `=== 0` / `> 0`, and `range(5).collect()` issues one `ExecutePlan` plus a variable number of fire-and-forget `ReleaseExecute` calls, so an exact count would be fragile. A `@volatile var ran = false` states the contract the assertions actually check. If more tests are added to this suite later, the global state will also need per-test isolation because those releases arrive asynchronously. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
