Refactor api_key handling into resolve_api_key and validate the format

Move the api_key logic into a single resolve_api_key method: an already
base64-encoded key or an Elastic Cloud API key is used as-is, a raw
id:api_key pair is base64-encoded, and any other value is rejected at
startup instead of failing later with a 401. Reword the api_key doc line.

Author: arenard

CHANGELOG.md

@@ -1,5 +1,5 @@
 ## 12.1.5
-  - Support Elastic Cloud API keys (`essu_` prefixed) in the `api_key` option by passing them verbatim instead of base64 re-encoding them [#1274](https://github.com/logstash-plugins/logstash-output-elasticsearch/pull/1274)
+  - Support Elastic Cloud API keys in the `api_key` option, which now accepts an `id:api_key` pair, its base64-encoded form, or an `essu_` Cloud API key, and rejects an unrecognized format at startup [#1274](https://github.com/logstash-plugins/logstash-output-elasticsearch/pull/1274)
 
 ## 12.1.4
   - [Doc] Add note for index option [#1269](https://github.com/logstash-plugins/logstash-output-elasticsearch/pull/1269)

lib/logstash/outputs/elasticsearch/http_client_builder.rb

@@ -190,10 +190,7 @@ def self.setup_api_key(logger, params)
 
       return {} unless (api_key&.value)
 
-      # Elastic Cloud API keys (`essu_` prefixed) and already base64-encoded
-      # keys are passed verbatim; a raw `id:api_key` pair is encoded here.
-      value = (cloud_api_key?(api_key.value) || is_base64?(api_key.value)) ? api_key.value : Base64.strict_encode64(api_key.value)
-
+      value = resolve_api_key(api_key.value)
       { "Authorization" => "ApiKey #{value}" }
     end
 
@@ -214,12 +211,26 @@ def query_param_separator(url)
         url.match?(/\?[^\s#]+/) ? '&' : '?'
       end
 
-      # Elastic Cloud API keys scoped for Elasticsearch (such as the unified
-      # Serverless keys) are opaque `essu_` prefixed tokens that Elasticsearch
-      # accepts verbatim in the `Authorization: ApiKey` header, with no base64
-      # encoding.
+      # Resolves the `api_key` value into the credential used in the
+      # `Authorization: ApiKey` header. An already base64-encoded key and an
+      # Elastic Cloud API key are used as-is; a raw `id:api_key` pair is
+      # base64-encoded. An unrecognized value is rejected so a malformed key
+      # surfaces at startup rather than as a later authentication failure.
+      def resolve_api_key(key_value)
+        if is_base64?(key_value) || cloud_api_key?(key_value)
+          key_value
+        elsif key_value.match?(/\A[^:]+:[^:]+\z/)
+          Base64.strict_encode64(key_value)
+        else
+          raise LogStash::ConfigurationError, "Invalid api_key format. Expected a base64-encoded key, an 'id:api_key' pair, or a Cloud API key (essu_ prefix)."
+        end
+      end
+
+      # Elastic Cloud API keys (such as the unified Serverless keys) are opaque
+      # tokens prefixed with `essu_` that Elasticsearch accepts verbatim in the
+      # `Authorization: ApiKey` header, with no base64 encoding.
       def cloud_api_key?(string)
-        string.start_with?("essu_")
+        string.match?(/\Aessu_.+/)
       end
 
       def is_base64?(string)

lib/logstash/plugin_mixins/elasticsearch/api_configs.rb

@@ -16,7 +16,7 @@ module APIConfigs
         :password => { :validate => :password },
 
         # Authenticate using Elasticsearch API key.
-        # Either the `id:api_key` pair (as returned by https://www.elastic.co/guide/en/elasticsearch/reference/current/security-api-create-api-key.html[Create API key]),
+        # Format is either the `id:api_key` pair (as returned by https://www.elastic.co/guide/en/elasticsearch/reference/current/security-api-create-api-key.html[Create API key]),
         # its base64-encoded form, or an https://www.elastic.co/docs/deploy-manage/api-keys/elastic-cloud-api-keys[Elastic Cloud API key] (prefixed with `essu_`) can be used.
         :api_key => { :validate => :password },
 

spec/unit/http_client_builder_spec.rb

@@ -64,6 +64,8 @@
     end
 
     context "when api-key is an Elastic Cloud API key (essu_ prefix)" do
+      # The suffix is intentionally not canonical base64: a Cloud key is opaque
+      # and must be forwarded verbatim regardless of its payload encoding.
       let(:api_key) { "essu_VFZGblZreFhTekJ4ZDB4M2NHUnZRMEU2YzNWd1pYSnpaV055WlhRPQ==AAAAAAAA" }
       let(:api_key_secured) do
         secured = double("api_key")
@@ -79,6 +81,21 @@
         expect(api_key_header["Authorization"]).to eql(expected)
       end
     end
+
+    context "when api-key has an unrecognized format" do
+      let(:api_key) { "not-a-valid-key" }
+      let(:api_key_secured) do
+        secured = double("api_key")
+        allow(secured).to receive(:value).and_return(api_key)
+        secured
+      end
+      let(:options) { { "api_key" => api_key_secured } }
+      let(:logger) { double("logger") }
+
+      it "raises a configuration error" do
+        expect { klass.setup_api_key(logger, options) }.to raise_error(LogStash::ConfigurationError, /Invalid api_key format/)
+      end
+    end
   end
 
   describe "customizing action paths" do