Process credentials

README.md

@@ -241,6 +241,27 @@ Path to the shared file. Defaults to "#{Dir.home}/.aws/credentials".
 
 Defaults to 'default' or `[ENV]('AWS_PROFILE')`.
 
+### process_credentials
+
+This loads AWS access credentials from an external process.
+
+    <match *>
+      @type kinesis_streams
+
+      <process_credentials>
+        process CMD
+      </process_credentials>
+    </match>
+
+See also:
+
+*   [Aws::ProcessCredentials](https://docs.aws.amazon.com/sdk-for-ruby/v3/api/Aws/ProcessCredentials.html)
+*   [Sourcing Credentials From External Processes](https://docs.aws.amazon.com/cli/latest/topic/config-vars.html#sourcing-credentials-from-external-processes)
+
+**process (required)**
+
+Command to be executed as an external process.
+
 ## Configuration: Format
 
 ### format (section)

lib/fluent/plugin/kinesis_helper/client.rb

@@ -63,6 +63,10 @@ module ClientParams
             desc "Profile name. Default to 'default' or ENV['AWS_PROFILE']"
             config_param :profile_name, :string, default: nil
           end
+          config_section :process_credentials, multi: false do
+            desc "External process to execute."
+            config_param :process, :string
+          end
         end
 
         def self.included(mod)
@@ -146,6 +150,13 @@ def setup_credentials
             credentials_options[:path] = c.path if c.path
             credentials_options[:profile_name] = c.profile_name if c.profile_name
             options[:credentials] = Aws::SharedCredentials.new(credentials_options)
+          when @process_credentials
+            if Gem::Version.new(Aws::CORE_GEM_VERSION) < Gem::Version.new('3.24.0')
+              raise Fluent::ConfigError, "Config process_credentials requires aws-sdk-core >= 3.24.0. Found aws-sdk-core #{Aws::CORE_GEM_VERSION} instead."
+            end
+            c = @process_credentials
+            process = c.process
+            options[:credentials] = Aws::ProcessCredentials.new(process)
           else
             # Use default credentials
             # See http://docs.aws.amazon.com/sdkforruby/api/Aws/S3/Client.html

test/kinesis_helper/test_client.rb

@@ -16,11 +16,39 @@
 require 'fluent/plugin/kinesis_helper/client'
 
 class KinesisHelperClientTest < Test::Unit::TestCase
-  class Mock
+  class ProcessCredentials
+    def self.process
+      expiration = Time.now.utc + 3600
+      response = {
+          :SecretAccessKey => "secret",
+          :SessionToken => "session-token",
+          :Version => 1,
+          :Expiration => expiration.strftime('%Y-%m-%dT%H:%M:%SZ'),
+          :AccessKeyId => "akid"
+      }.to_json
+
+      "echo '#{response}'"
+    end
+  end
+
+  class MockClientInstanceProfile
+    include Fluent::Plugin::KinesisHelper::Client
+
+    def initialize
+      @region = 'us-east-1'
+    end
+
+    def request_type
+      :firehose
+    end
+  end
+
+  class MockClientProcessCredentials
     include Fluent::Plugin::KinesisHelper::Client
 
     def initialize
       @region = 'us-east-1'
+      @process_credentials = ProcessCredentials
     end
 
     def request_type
@@ -35,8 +63,6 @@ def self.startup
   def setup
     WebMock.enable!
     FakeFS.activate!
-    @object = Mock.new
-    Aws.shared_config.fresh
   end
 
   def teardown
@@ -46,21 +72,38 @@ def teardown
   end
 
   def test_instance_profile
+    Aws.shared_config.fresh
     setup_instance_profile
-    credentials = @object.client.config.credentials
+    credentials = MockClientInstanceProfile.new.client.config.credentials
     assert_equal 'akid',          credentials.credentials.access_key_id
     assert_equal 'secret',        credentials.credentials.secret_access_key
     assert_equal 'session-token', credentials.credentials.session_token
   end
 
   def test_instance_profile_refresh
+    Aws.shared_config.fresh
     setup_instance_profile(Time.now.utc + 299)
-    credentials = @object.client.config.credentials
+    credentials = MockClientInstanceProfile.new.client.config.credentials
     assert_equal 'akid-2',          credentials.credentials.access_key_id
     assert_equal 'secret-2',        credentials.credentials.secret_access_key
     assert_equal 'session-token-2', credentials.credentials.session_token
   end
 
+  def test_process_credentials
+    omit_if(Gem::Version.new(Aws::CORE_GEM_VERSION) < Gem::Version.new('3.24.0'))
+    credentials = MockClientProcessCredentials.new.client.config.credentials
+    assert_equal 'akid',          credentials.credentials.access_key_id
+    assert_equal 'secret',        credentials.credentials.secret_access_key
+    assert_equal 'session-token', credentials.credentials.session_token
+  end
+
+  def test_process_credentials_config_error
+    omit_if(Gem::Version.new(Aws::CORE_GEM_VERSION) >= Gem::Version.new('3.24.0'))
+    assert_raise(Fluent::ConfigError) do
+      MockClientProcessCredentials.new.client.config.credentials
+    end
+  end
+
   private
 
   def setup_instance_profile(expiration = Time.now.utc + 3600)