You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by gu...@apache.org on 2026/09/01 01:22:09 UTC
(spark-connect-rust) branch master updated: [SPARK-59139][FOLLOWUP] Reject bearer token sent over a cleartext connection
This is an automated email from the ASF dual-hosted git repository.
HyukjinKwon pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark-connect-rust.git
The following commit(s) were added to refs/heads/master by this push:
new 7c0d30a [SPARK-59139][FOLLOWUP] Reject bearer token sent over a cleartext connection
7c0d30a is described below
commit 7c0d30aabf05e3218671a7826ce0a85e0a875585
Author: Hyukjin Kwon <hy...@databricks.com>
AuthorDate: Tue Sep 1 10:22:01 2026 +0900
[SPARK-59139][FOLLOWUP] Reject bearer token sent over a cleartext connection
### What changes were proposed in this pull request?
Follow-up to #92 (addresses [this review comment](https://github.com/apache/spark-connect-rust/pull/92#issuecomment-5486670118)). That PR started attaching the connection-string `token` as an `Authorization: Bearer <token>` header, but the transport scheme is chosen from `use_ssl()` alone and the header was attached unconditionally. As a result `sc://host/;token=SECRET` (no `use_ssl`) would send the bearer token over plain `http://`.
This change refuses to attach the token when the connection is not TLS and the host is not loopback, returning an error that names `use_ssl=true`, instead of silently downgrading to cleartext. A token over plaintext loopback (`localhost`/`127.0.0.1`/`[::1]`) is still allowed for local development, mirroring the reference client's `local_channel_credentials()` exception in `core.py` `toChannel`.
Two minor cleanups the reviewer noted are folded in:
- the `"authorization"` header key is now a named `const` rather than a string literal;
- `test_token_bearer_in_metadata` now builds a real `Request`, runs it through `_attach_metadata`, and asserts on the resulting gRPC metadata rather than on the `client.metadata` field.
### Why are the changes needed?
To stop a mistyped or SSL-less connection string from leaking the authentication token in the clear. The reference client can never reach that state because grpcio refuses call credentials on an insecure channel; this restores parity by failing loudly.
### Does this PR introduce _any_ user-facing change?
Yes. `SparkConnectClient::connect` now returns an error for a connection string that supplies a `token` (or the `SPARK_CONNECT_AUTHENTICATE_TOKEN` env var) without `use_ssl=true` when the host is not loopback. The remedy is to set `use_ssl=true`.
### How was this patch tested?
New unit tests cover the rejection on a remote host, the loopback exception for `localhost`/`127.0.0.1`/`[::1]`, and that the bearer header reaches a real request's metadata. `cargo test -p apache-spark-connect-core` passes.
This pull request and its description were written by Isaac.
Closes #93 from HyukjinKwon/SPARK-59139-followup-cleartext-token.
Authored-by: Hyukjin Kwon <hy...@databricks.com>
Signed-off-by: Hyukjin Kwon <hy...@databricks.com>
---
crates/pyspark-rs/src/transport.rs | 9 +++-
crates/spark-connect-core/src/channel.rs | 18 ++++++-
crates/spark-connect-core/src/client.rs | 88 ++++++++++++++++++++++++++++----
crates/spark-connect-core/src/error.rs | 20 ++++++++
4 files changed, 121 insertions(+), 14 deletions(-)
diff --git a/crates/pyspark-rs/src/transport.rs b/crates/pyspark-rs/src/transport.rs
index 2e48d94..4386225 100644
--- a/crates/pyspark-rs/src/transport.rs
+++ b/crates/pyspark-rs/src/transport.rs
@@ -25,6 +25,8 @@ use spark_connect_core::retries::RetryPolicy;
use spark_connect_core::runtime::block_on;
use spark_connect_proto as proto;
+use crate::errors::ResultExt;
+
fn err<E: std::fmt::Display>(e: E) -> PyErr {
PyRuntimeError::new_err(e.to_string())
}
@@ -153,12 +155,15 @@ pub struct RustConnectStub {
impl RustConnectStub {
#[new]
fn new(py: Python<'_>, url: &str) -> PyResult<Self> {
- let builder = ChannelBuilder::parse(url).map_err(err)?;
+ // Map connection-setup failures (bad URL, cleartext-token refusal, …) through
+ // the typed error layer so they surface as the right `pyspark.errors` exception
+ // (e.g. `PySparkValueError`) rather than a bare `RuntimeError`.
+ let builder = ChannelBuilder::parse(url).to_pyerr()?;
// Release the GIL during the connect handshake so a slow/unresponsive server
// can't wedge the whole interpreter (incl. signal handling / Ctrl-C).
let mut client = py
.detach(|| block_on(SparkConnectClient::connect(&builder)))
- .map_err(err)?;
+ .to_pyerr()?;
// This stub is a drop-in for the grpcio stub: a single RPC per call. The reference
// client wraps calls in its own GrpcRetryHandler (per the client-side retry
// policy), so retrying inside our client too would double-retry and ignore that
diff --git a/crates/spark-connect-core/src/channel.rs b/crates/spark-connect-core/src/channel.rs
index ccdccf6..79a3063 100644
--- a/crates/spark-connect-core/src/channel.rs
+++ b/crates/spark-connect-core/src/channel.rs
@@ -154,11 +154,27 @@ impl ChannelBuilder {
.or_else(|| std::env::var("SPARK_CONNECT_AUTHENTICATE_TOKEN").ok())
}
- /// A channel is secure when SSL is requested or a token is present.
+ /// Mirrors the reference client's `secure` property (`use_ssl` OR a token is
+ /// present). NOTE: this does NOT gate the transport. It reads like a security
+ /// check but grants none: the token half was the source of the cleartext-token
+ /// bug, since a token alone reports "secure" while the scheme stays plaintext.
+ /// The real gate on sending the token is `use_ssl() && !is_loopback()` in
+ /// `SparkConnectClient::connect`; this method is kept only for parity/inspection.
pub fn secure(&self) -> bool {
self.use_ssl() || self.token().is_some()
}
+ /// Whether the endpoint is a loopback address.
+ ///
+ /// The reference client (`core.py` `toChannel`) permits a token over an
+ /// unencrypted channel only for `host == "localhost"` (via
+ /// `grpc.local_channel_credentials()`); every other host is forced onto TLS.
+ /// We treat the IPv4/IPv6 loopback literals as equivalent to `localhost` so
+ /// local development keeps working without `use_ssl=true`.
+ pub fn is_loopback(&self) -> bool {
+ matches!(self.host.as_str(), "localhost" | "127.0.0.1" | "[::1]")
+ }
+
pub fn user_id(&self) -> Option<&str> {
self.get(PARAM_USER_ID)
}
diff --git a/crates/spark-connect-core/src/client.rs b/crates/spark-connect-core/src/client.rs
index f000ab2..31820ed 100644
--- a/crates/spark-connect-core/src/client.rs
+++ b/crates/spark-connect-core/src/client.rs
@@ -29,6 +29,10 @@ use crate::retries::{RetryPolicy, RetryPolicyState};
/// to a down server would hang forever instead of raising `UNAVAILABLE`.
const READY_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);
+/// gRPC/HTTP header carrying the connection-string `token` as `Bearer <token>`.
+/// Named here (rather than as a string literal) so the key is written once.
+const AUTHORIZATION_HEADER: &str = "authorization";
+
/// Wait for `grpc` to become ready, bounded by [`READY_TIMEOUT`], mapping any failure
/// (or the timeout) to a gRPC `UNAVAILABLE` error. This matches how grpcio reports an
/// unreachable server, so the reference client raises `SparkConnectGrpcException` with
@@ -122,7 +126,25 @@ impl SparkConnectClient {
// excluded from `metadata()`, so attach it explicitly here; this then propagates
// to every request via `_attach_metadata` and the reattach stream.
if let Some(tok) = builder.token() {
- metadata.push(("authorization".to_string(), format!("Bearer {}", tok)));
+ // Never send the bearer token in cleartext. The scheme above is chosen from
+ // `use_ssl()` alone, so without this guard `sc://host/;token=SECRET` (no
+ // `use_ssl`) would ship `Authorization: Bearer SECRET` over plain HTTP. The
+ // reference client cannot reach that state: `toChannel` sends a token only
+ // over `ssl_channel_credentials()` or, for `localhost`, the loopback
+ // `local_channel_credentials()`, and grpcio refuses call credentials on an
+ // insecure channel outright. Mirror that by requiring TLS for a token unless
+ // the endpoint is loopback; fail loudly rather than downgrade silently.
+ if !builder.use_ssl() && !builder.is_loopback() {
+ // A misconfigured connection string is a value error (surfaces as
+ // `PySparkValueError`), not a generic runtime failure.
+ return Err(SparkError::value_msg(format!(
+ "Refusing to send the authentication token to '{}' over an insecure \
+ connection. Set 'use_ssl=true' in the connection string to enable TLS \
+ (a token is only allowed without TLS when connecting to localhost).",
+ builder.host()
+ )));
+ }
+ metadata.push((AUTHORIZATION_HEADER.to_string(), format!("Bearer {}", tok)));
}
Ok(Self {
@@ -970,22 +992,66 @@ mod tests {
#[tokio::test]
async fn test_token_bearer_in_metadata() {
- // A connection-string token must actually be SENT as an `Authorization: Bearer`
- // header, not merely parsed. Assert it is attached to the client's request metadata.
- let builder = ChannelBuilder::parse("sc://localhost/;token=my_token_123").unwrap();
+ // A connection-string token must actually reach the wire as an
+ // `Authorization: Bearer` header, not merely be parsed. Build a real request,
+ // run it through `_attach_metadata`, and assert on the resulting gRPC metadata
+ // so this survives a future change to how the header is stored or inserted.
+ let builder =
+ ChannelBuilder::parse("sc://localhost/;use_ssl=true;token=my_token_123").unwrap();
assert!(builder.secure());
assert_eq!(builder.token(), Some("my_token_123".to_string()));
let client = SparkConnectClient::connect(&builder).await.unwrap();
- assert!(
- client
- .metadata
- .iter()
- .any(|(k, v)| k.eq_ignore_ascii_case("authorization") && v == "Bearer my_token_123"),
- "expected an `Authorization: Bearer` header in client metadata, got {:?}",
- client.metadata
+
+ let mut req = Request::new(());
+ client._attach_metadata(&mut req);
+ assert_eq!(
+ req.metadata()
+ .get("authorization")
+ .map(|v| v.to_str().unwrap()),
+ Some("Bearer my_token_123"),
+ "expected an `Authorization: Bearer` header on the request, got {:?}",
+ req.metadata()
);
}
+ #[tokio::test]
+ async fn test_token_without_ssl_to_remote_host_is_rejected() {
+ // A token over a non-loopback host without `use_ssl=true` would ship the bearer
+ // credential in cleartext; the client must refuse rather than downgrade silently.
+ let builder = ChannelBuilder::parse("sc://example.com/;token=SECRET").unwrap();
+ match SparkConnectClient::connect(&builder).await {
+ Ok(_) => panic!("expected a token over cleartext to a remote host to be rejected"),
+ Err(e) => assert!(
+ e.to_string().contains("use_ssl=true"),
+ "expected the error to point at use_ssl=true, got: {e}"
+ ),
+ }
+ }
+
+ #[tokio::test]
+ async fn test_token_without_ssl_to_localhost_is_allowed() {
+ // Local development mirrors upstream's `local_channel_credentials()` exception:
+ // a token over plaintext loopback is permitted and the header is attached.
+ for url in [
+ "sc://localhost/;token=SECRET",
+ "sc://127.0.0.1/;token=SECRET",
+ "sc://[::1]/;token=SECRET",
+ ] {
+ let builder = ChannelBuilder::parse(url).unwrap();
+ let client = SparkConnectClient::connect(&builder)
+ .await
+ .unwrap_or_else(|e| panic!("{url} should be allowed, got: {e}"));
+ assert!(
+ client
+ .metadata
+ .iter()
+ .any(|(k, v)| k == "authorization" && v == "Bearer SECRET"),
+ "{url}: expected the bearer header to be attached, got {:?}",
+ client.metadata
+ );
+ }
+ }
+
#[tokio::test]
async fn test_get_configs_request_structure() {
let builder = ChannelBuilder::parse("sc://localhost").unwrap();
diff --git a/crates/spark-connect-core/src/error.rs b/crates/spark-connect-core/src/error.rs
index 1965498..ae5fa3a 100644
--- a/crates/spark-connect-core/src/error.rs
+++ b/crates/spark-connect-core/src/error.rs
@@ -268,6 +268,26 @@ impl SparkError {
}
}
+ /// Create a ValueError-kind error with a plain message (no error class).
+ ///
+ /// For invalid client-side configuration (e.g. a malformed connection string)
+ /// that has no registered error class. Surfaces as `PySparkValueError` on the
+ /// Python side, matching how the reference client reports connection-string
+ /// problems, rather than the generic runtime kind.
+ pub fn value_msg(message: impl Into<String>) -> Self {
+ Self {
+ kind: SparkErrorKind::ValueError,
+ error_class: String::new(),
+ params: BTreeMap::new(),
+ message: message.into(),
+ sql_state: None,
+ contexts: Vec::new(),
+ server_stacktrace: None,
+ grpc_code: None,
+ grpc_details: None,
+ }
+ }
+
/// Create an error of a specific kind with an error class and parameters.
pub fn classed(kind: SparkErrorKind, error_class: &str, params: &[(&str, &str)]) -> Self {
Self {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org