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 acfb69b22 elastic: if credentials are specific, require TLS (#1810)
acfb69b22 is described below
commit acfb69b226664570d7b6b93250237b765fd66f38
Author: PJ Fanning <[email protected]>
AuthorDate: Sun Aug 9 09:44:54 2026 +0100
elastic: if credentials are specific, require TLS (#1810)
---
.../elasticsearch/impl/ElasticsearchApi.scala | 14 ++++
.../elasticsearch/impl/ElasticsearchApiSpec.scala | 86 ++++++++++++++++++++++
2 files changed, 100 insertions(+)
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 5f49006d5..e29b33ee5 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
@@ -15,6 +15,7 @@ package org.apache.pekko.stream.connectors.elasticsearch.impl
import org.apache.pekko
import pekko.annotation.InternalApi
+import pekko.event.Logging
import pekko.http.scaladsl.HttpExt
import pekko.http.scaladsl.model._
import pekko.http.scaladsl.model.headers.BasicHttpCredentials
@@ -23,10 +24,23 @@ import
pekko.stream.connectors.elasticsearch.ElasticsearchConnectionSettings
import scala.concurrent.Future
@InternalApi private[impl] object ElasticsearchApi {
+
+ private val logSource = "ElasticsearchApi"
+
def executeRequest(
request: HttpRequest,
connectionSettings: ElasticsearchConnectionSettings)(implicit http:
HttpExt): Future[HttpResponse] = {
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)
+ }
http.singleRequest(
request.addCredentials(BasicHttpCredentials(connectionSettings.username.get,
connectionSettings.password.get)))
} else {
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
new file mode 100644
index 000000000..24cb5ee1e
--- /dev/null
+++
b/elasticsearch/src/test/scala/org/apache/pekko/stream/connectors/elasticsearch/impl/ElasticsearchApiSpec.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.pekko.stream.connectors.elasticsearch.impl
+
+import org.apache.pekko
+import pekko.actor.ActorSystem
+import pekko.http.scaladsl.{ Http, HttpExt }
+import pekko.http.scaladsl.model._
+import pekko.stream.Materializer
+import pekko.stream.connectors.elasticsearch.ElasticsearchConnectionSettings
+import pekko.testkit.TestKit
+import org.scalatest.BeforeAndAfterAll
+import org.scalatest.matchers.should.Matchers
+import org.scalatest.wordspec.AnyWordSpecLike
+
+class ElasticsearchApiSpec
+ extends TestKit(ActorSystem("elasticsearch-api-spec"))
+ with AnyWordSpecLike
+ with Matchers
+ with BeforeAndAfterAll {
+
+ implicit val mat: Materializer = Materializer(system)
+ implicit val http: HttpExt = Http()
+
+ override def afterAll(): Unit = {
+ TestKit.shutdownActorSystem(system)
+ }
+
+ "ElasticsearchApi.executeRequest" should {
+
+ "reject plain HTTP requests when credentials are configured" in {
+ val connectionSettings =
+ ElasticsearchConnectionSettings("http://localhost:9200";)
+ .withCredentials("user", "pass")
+
+ val request = HttpRequest(HttpMethods.GET)
+ .withUri(Uri("http://localhost:9200/_search";))
+
+ val ex = intercept[IllegalStateException] {
+ ElasticsearchApi.executeRequest(request, connectionSettings)
+ }
+ ex.getMessage should include("not 'https'")
+ ex.getMessage should include("insecure")
+ }
+
+ "allow HTTPS requests when credentials are configured" in {
+ val connectionSettings =
+ ElasticsearchConnectionSettings("https://localhost:9200";)
+ .withCredentials("user", "pass")
+
+ val request = HttpRequest(HttpMethods.GET)
+ .withUri(Uri("https://localhost:9200/_search";))
+
+ // Validation passes — no IllegalStateException thrown synchronously.
+ // The future will fail at the network level (no server), which is
expected.
+ noException should be thrownBy ElasticsearchApi.executeRequest(request,
connectionSettings)
+ }
+
+ "allow plain HTTP requests when no credentials are configured" in {
+ val connectionSettings =
+ ElasticsearchConnectionSettings("http://localhost:9200";)
+
+ val request = HttpRequest(HttpMethods.GET)
+ .withUri(Uri("http://localhost:9200/_search";))
+
+ // No credentials means no HTTPS enforcement.
+ // The future will fail at the network level (no server), which is
expected.
+ noException should be thrownBy ElasticsearchApi.executeRequest(request,
connectionSettings)
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
