This is an automated email from the ASF dual-hosted git repository.
pjfanning pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/pekko-connectors.git
The following commit(s) were added to refs/heads/main by this push:
new 00a8f6d19 Elasticsearch: fix insecure-credentials check message and
failure mode (#1871)
00a8f6d19 is described below
commit 00a8f6d193c6c0967a3fb27fae3dc920ea09c63f
Author: PJ Fanning <[email protected]>
AuthorDate: Tue Sep 22 10:45:52 2026 +0100
Elasticsearch: fix insecure-credentials check message and failure mode
(#1871)
Motivation:
The HTTPS enforcement added in #1810 has three problems.
1. The error message never interpolates the scheme. In
"... scheme is '%s' (not 'https'). " +
"... insecure. " +
"Configure a HTTPS base URL ...".format(scheme)
`.format` binds to the last string literal only, which contains no
`%s`, so the message is emitted with a literal '%s'.
2. The check throws synchronously. ElasticsearchSimpleFlowStage.onPush
calls executeRequest outside any try/catch, so the exception escapes
the stage instead of going through the connector's failure handling.
ElasticsearchSourceStage happens to wrap its call, so the two stages
behaved differently.
3. There is no way to opt out. The documented #connection-settings
snippet pairs a http:// base URL with withCredentials, so a
previously working local setup now fails outright with no override.
Additionally, the credentials branch did not pass connectionContext, so
a connection context supplied via withSSLContext was dropped whenever
credentials were configured.
Modification:
Parenthesise the concatenation so format applies to the whole message.
Return a failed Future rather than throwing, so both stages route the
error through their own failure handling. Add
withAllowInsecureCredentialsTransport (default false) as an opt-out that
downgrades the rejection to a warning. Pass connectionContext in the
credentials branch as well.
The scheme decision is extracted into checkCredentialsTransport so it can
be tested without opening a socket. The documented snippets now use a
https:// base URL with credentials, which is the configuration users
should copy.
Result:
The message names the actual scheme. Failures arrive as a failed Future
and are reported by the connector rather than escaping the stage. Users
who need plain HTTP with credentials for local testing can opt in. A
user-supplied connection context is honoured with credentials set.
Tests:
Extended ElasticsearchApiSpec: the rejection is asserted on the failed
Future, a new test asserts the scheme is interpolated and no '%s'
remains, and four tests cover checkCredentialsTransport (reject http,
accept https, accept HTTPS case-insensitively, warn when opted in).
14 tests pass. Ran elasticsearch/mimaReportBinaryIssues (clean) and the
scalafmt checks for the module.
References:
Follow-up to #1810.
---
.../ElasticsearchConnectionSettings.scala | 41 +++++++++++----
.../elasticsearch/impl/ElasticsearchApi.scala | 47 ++++++++++++-----
.../javadsl/ElasticsearchParameterizedTest.java | 2 +-
.../docs/javadsl/OpensearchParameterizedTest.java | 2 +-
.../scaladsl/ElasticsearchConnectorBehaviour.scala | 2 +-
.../elasticsearch/impl/ElasticsearchApiSpec.scala | 60 ++++++++++++++++++++--
6 files changed, 125 insertions(+), 29 deletions(-)
diff --git
a/elasticsearch/src/main/scala/org/apache/pekko/stream/connectors/elasticsearch/ElasticsearchConnectionSettings.scala
b/elasticsearch/src/main/scala/org/apache/pekko/stream/connectors/elasticsearch/ElasticsearchConnectionSettings.scala
index 3540ef358..940488347 100644
---
a/elasticsearch/src/main/scala/org/apache/pekko/stream/connectors/elasticsearch/ElasticsearchConnectionSettings.scala
+++
b/elasticsearch/src/main/scala/org/apache/pekko/stream/connectors/elasticsearch/ElasticsearchConnectionSettings.scala
@@ -27,7 +27,8 @@ final class ElasticsearchConnectionSettings private (
val username: Option[String],
val password: Option[String],
val headers: List[HttpHeader],
- val connectionContext: Option[HttpsConnectionContext]) {
+ val connectionContext: Option[HttpsConnectionContext],
+ val allowInsecureCredentialsTransport: Boolean) {
def withBaseUrl(value: String): ElasticsearchConnectionSettings =
copy(baseUrl = value)
@@ -63,30 +64,52 @@ final class ElasticsearchConnectionSettings private (
def hasConnectionContextDefined: Boolean = connectionContext.isDefined
+ /**
+ * Permit sending credentials over a plain HTTP connection. Defaults to
`false`,
+ * in which case requests that would send credentials over a non-HTTPS
connection
+ * fail instead. Enable only for local development and testing, where the
traffic
+ * cannot be observed.
+ *
+ * @since 2.0.0
+ */
+ def withAllowInsecureCredentialsTransport(value: Boolean):
ElasticsearchConnectionSettings =
+ copy(allowInsecureCredentialsTransport = value)
+
private def copy(
baseUrl: String = baseUrl,
username: Option[String] = username,
password: Option[String] = password,
headers: List[HttpHeader] = headers,
- connectionContext: Option[HttpsConnectionContext] = connectionContext):
ElasticsearchConnectionSettings =
+ connectionContext: Option[HttpsConnectionContext] = connectionContext,
+ allowInsecureCredentialsTransport: Boolean =
allowInsecureCredentialsTransport)
+ : ElasticsearchConnectionSettings =
new ElasticsearchConnectionSettings(baseUrl = baseUrl,
username = username,
password = password,
headers = headers,
- connectionContext = connectionContext)
-
- override def toString =
-
s"""ElasticsearchConnectionSettings(baseUrl=$baseUrl,username=$username,password=${password.fold("")(_
=>
-
"***")},headers=${headers.mkString(";")},connectionContext=$connectionContext)"""
+ connectionContext = connectionContext,
+ allowInsecureCredentialsTransport = allowInsecureCredentialsTransport)
+
+ override def toString = {
+ val maskedPassword = password.fold("")(_ => "***")
+ val renderedHeaders = headers.mkString(";")
+ "ElasticsearchConnectionSettings(" +
+ s"baseUrl=$baseUrl," +
+ s"username=$username," +
+ s"password=$maskedPassword," +
+ s"headers=$renderedHeaders," +
+ s"connectionContext=$connectionContext," +
+ s"allowInsecureCredentialsTransport=$allowInsecureCredentialsTransport)"
+ }
}
object ElasticsearchConnectionSettings {
/** Scala API */
def apply(baseUrl: String): ElasticsearchConnectionSettings =
- new ElasticsearchConnectionSettings(baseUrl, None, None, List.empty, None)
+ new ElasticsearchConnectionSettings(baseUrl, None, None, List.empty, None,
false)
/** Java API */
def create(baseUrl: String): ElasticsearchConnectionSettings =
- new ElasticsearchConnectionSettings(baseUrl, None, None, List.empty, None)
+ new ElasticsearchConnectionSettings(baseUrl, None, None, List.empty, None,
false)
}
diff --git
a/elasticsearch/src/main/scala/org/apache/pekko/stream/connectors/elasticsearch/impl/ElasticsearchApi.scala
b/elasticsearch/src/main/scala/org/apache/pekko/stream/connectors/elasticsearch/impl/ElasticsearchApi.scala
index e29b33ee5..d69ab4cb9 100644
---
a/elasticsearch/src/main/scala/org/apache/pekko/stream/connectors/elasticsearch/impl/ElasticsearchApi.scala
+++
b/elasticsearch/src/main/scala/org/apache/pekko/stream/connectors/elasticsearch/impl/ElasticsearchApi.scala
@@ -27,26 +27,45 @@ import scala.concurrent.Future
private val logSource = "ElasticsearchApi"
+ private def insecureCredentialsMessage(scheme: String): String =
+ ("Credentials are configured but the request URI scheme is '%s' (not
'https'). " +
+ "Sending BasicAuth credentials over plain HTTP is insecure. " +
+ "Configure a HTTPS base URL to use with credentials, or set " +
+ "withAllowInsecureCredentialsTransport(true) to allow
this.").format(scheme)
+
+ /**
+ * Decide whether credentials may be sent for this request.
+ *
+ * Returns `Left` with the error message when the request must be rejected,
+ * or `Right` with an optional warning message when it may proceed.
+ */
+ private[impl] def checkCredentialsTransport(
+ scheme: String,
+ connectionSettings: ElasticsearchConnectionSettings): Either[String,
Option[String]] =
+ if (scheme.toLowerCase == "https") Right(None)
+ else if (connectionSettings.allowInsecureCredentialsTransport)
Right(Some(insecureCredentialsMessage(scheme)))
+ else Left(insecureCredentialsMessage(scheme))
+
def executeRequest(
request: HttpRequest,
connectionSettings: ElasticsearchConnectionSettings)(implicit http:
HttpExt): Future[HttpResponse] = {
+ val connectionContext =
connectionSettings.connectionContext.getOrElse(http.defaultClientHttpsContext)
if (connectionSettings.hasCredentialsDefined) {
- val scheme = request.uri.scheme.toLowerCase
- if (scheme != "https") {
- val log = Logging(http.system, logSource)
- val msg =
- "Credentials are configured but the request URI scheme is '%s' (not
'https'). " +
- "Sending BasicAuth credentials over plain HTTP is insecure. " +
- "Configure a HTTPS base URL to use with credentials.".format(scheme)
- log.error(msg)
- throw new IllegalStateException(msg)
+ checkCredentialsTransport(request.uri.scheme, connectionSettings) match {
+ case Left(msg) =>
+ Logging(http.system, logSource).error(msg)
+ // a failed Future rather than a thrown exception, so that the
calling
+ // stages can route this through their own failure handling
+ Future.failed(new IllegalStateException(msg))
+ case Right(warning) =>
+ warning.foreach(Logging(http.system, logSource).warning(_))
+ http.singleRequest(
+ request.addCredentials(
+ BasicHttpCredentials(connectionSettings.username.get,
connectionSettings.password.get)),
+ connectionContext = connectionContext)
}
- http.singleRequest(
-
request.addCredentials(BasicHttpCredentials(connectionSettings.username.get,
connectionSettings.password.get)))
} else {
- http.singleRequest(request,
- connectionContext =
-
connectionSettings.connectionContext.getOrElse(http.defaultClientHttpsContext))
+ http.singleRequest(request, connectionContext = connectionContext)
}
}
}
diff --git
a/elasticsearch/src/test/java/docs/javadsl/ElasticsearchParameterizedTest.java
b/elasticsearch/src/test/java/docs/javadsl/ElasticsearchParameterizedTest.java
index 6f3b7d79d..5df0b82dd 100644
---
a/elasticsearch/src/test/java/docs/javadsl/ElasticsearchParameterizedTest.java
+++
b/elasticsearch/src/test/java/docs/javadsl/ElasticsearchParameterizedTest.java
@@ -49,7 +49,7 @@ public class ElasticsearchParameterizedTest extends
ElasticsearchTestBase {
private void documentation() {
// #connection-settings
ElasticsearchConnectionSettings connectionSettings =
- ElasticsearchConnectionSettings.create("http://localhost:9200";)
+ ElasticsearchConnectionSettings.create("https://localhost:9200";)
.withCredentials("user", "password");
// #connection-settings
diff --git
a/elasticsearch/src/test/java/docs/javadsl/OpensearchParameterizedTest.java
b/elasticsearch/src/test/java/docs/javadsl/OpensearchParameterizedTest.java
index fbf46bd8e..2a46cc02a 100644
--- a/elasticsearch/src/test/java/docs/javadsl/OpensearchParameterizedTest.java
+++ b/elasticsearch/src/test/java/docs/javadsl/OpensearchParameterizedTest.java
@@ -50,7 +50,7 @@ public class OpensearchParameterizedTest extends
ElasticsearchTestBase {
private void documentation() {
// #connection-settings
ElasticsearchConnectionSettings connectionSettings =
- OpensearchConnectionSettings.create("http://localhost:9200";)
+ OpensearchConnectionSettings.create("https://localhost:9200";)
.withCredentials("user", "password");
// #connection-settings
diff --git
a/elasticsearch/src/test/scala/docs/scaladsl/ElasticsearchConnectorBehaviour.scala
b/elasticsearch/src/test/scala/docs/scaladsl/ElasticsearchConnectorBehaviour.scala
index e16e22136..65d368d2f 100644
---
a/elasticsearch/src/test/scala/docs/scaladsl/ElasticsearchConnectorBehaviour.scala
+++
b/elasticsearch/src/test/scala/docs/scaladsl/ElasticsearchConnectorBehaviour.scala
@@ -667,7 +667,7 @@ trait ElasticsearchConnectorBehaviour {
lazy val _ = {
// #connection-settings
- val connectionSettings =
ElasticsearchConnectionSettings("http://localhost:9200";)
+ val connectionSettings =
ElasticsearchConnectionSettings("https://localhost:9200";)
.withCredentials("user", "password")
// #connection-settings
// #source-settings
diff --git
a/elasticsearch/src/test/scala/org/apache/pekko/stream/connectors/elasticsearch/impl/ElasticsearchApiSpec.scala
b/elasticsearch/src/test/scala/org/apache/pekko/stream/connectors/elasticsearch/impl/ElasticsearchApiSpec.scala
index 24cb5ee1e..abfef8e67 100644
---
a/elasticsearch/src/test/scala/org/apache/pekko/stream/connectors/elasticsearch/impl/ElasticsearchApiSpec.scala
+++
b/elasticsearch/src/test/scala/org/apache/pekko/stream/connectors/elasticsearch/impl/ElasticsearchApiSpec.scala
@@ -25,6 +25,7 @@ import pekko.stream.Materializer
import pekko.stream.connectors.elasticsearch.ElasticsearchConnectionSettings
import pekko.testkit.TestKit
import org.scalatest.BeforeAndAfterAll
+import org.scalatest.concurrent.ScalaFutures
import org.scalatest.matchers.should.Matchers
import org.scalatest.wordspec.AnyWordSpecLike
@@ -32,6 +33,7 @@ class ElasticsearchApiSpec
extends TestKit(ActorSystem("elasticsearch-api-spec"))
with AnyWordSpecLike
with Matchers
+ with ScalaFutures
with BeforeAndAfterAll {
implicit val mat: Materializer = Materializer(system)
@@ -51,13 +53,65 @@ class ElasticsearchApiSpec
val request = HttpRequest(HttpMethods.GET)
.withUri(Uri("http://localhost:9200/_search";))
- val ex = intercept[IllegalStateException] {
- ElasticsearchApi.executeRequest(request, connectionSettings)
- }
+ // the rejection arrives as a failed Future, not as a synchronous throw,
+ // so that the calling stages can route it through their failure handling
+ noException should be thrownBy ElasticsearchApi.executeRequest(request,
connectionSettings)
+
+ val ex = ElasticsearchApi
+ .executeRequest(request, connectionSettings)
+ .failed
+ .futureValue
+ ex shouldBe an[IllegalStateException]
ex.getMessage should include("not 'https'")
ex.getMessage should include("insecure")
}
+ "interpolate the actual scheme into the rejection message" in {
+ val connectionSettings =
+ ElasticsearchConnectionSettings("http://localhost:9200";)
+ .withCredentials("user", "pass")
+
+ val request = HttpRequest(HttpMethods.GET)
+ .withUri(Uri("http://localhost:9200/_search";))
+
+ val ex = ElasticsearchApi
+ .executeRequest(request, connectionSettings)
+ .failed
+ .futureValue
+ ex.getMessage should include("scheme is 'http'")
+ (ex.getMessage should not).include("%s")
+ }
+
+ }
+
+ "ElasticsearchApi.checkCredentialsTransport" should {
+
+ val withCredentials =
+
ElasticsearchConnectionSettings("http://localhost:9200";).withCredentials("user",
"pass")
+
+ "reject a plain HTTP scheme by default" in {
+ val result = ElasticsearchApi.checkCredentialsTransport("http",
withCredentials)
+ result.isLeft shouldBe true
+ result.left.toOption.get should include("scheme is 'http'")
+ }
+
+ "accept an HTTPS scheme without warning" in {
+ ElasticsearchApi.checkCredentialsTransport("https", withCredentials)
shouldBe Right(None)
+ }
+
+ "accept an HTTPS scheme regardless of case" in {
+ ElasticsearchApi.checkCredentialsTransport("HTTPS", withCredentials)
shouldBe Right(None)
+ }
+
+ "accept a plain HTTP scheme with a warning when insecure transport is
allowed" in {
+ val settings =
withCredentials.withAllowInsecureCredentialsTransport(true)
+ val result = ElasticsearchApi.checkCredentialsTransport("http", settings)
+ result.isRight shouldBe true
+ val warning = result.toOption.get
+ warning should be(defined)
+ warning.get should include("insecure")
+ }
+
"allow HTTPS requests when credentials are configured" in {
val connectionSettings =
ElasticsearchConnectionSettings("https://localhost:9200";)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
