in_s3/out_s3: add aws_profile / aws_credential_process parameters for credencials

docs/credentials.md

@@ -14,6 +14,23 @@ AWS access key id.
 
 AWS secret key.
 
+### aws_profile
+
+Specify a named profile from your shared AWS configuration files (`~/.aws/credentials` or `~/.aws/config`).
+This is useful for environments using IAM Roles or short-lived credentials.
+
+### aws_credential_process
+
+Specify a `credential_process` command directly in the Fluentd configuration.
+This enables integration with external credential helpers like IAM Roles Anywhere.
+
+    <match *>
+      @type s3
+      # Example for IAM Roles Anywhere
+      aws_credential_process "/path/to/aws_signing_helper credential-process --certificate /path/to/cert.pem --private-key /path/to/key.pem --trust-anchor-arn <ARN> --profile-arn <ARN> --role-arn <ARN>"
+      # ... other settings
+    </match>
+
 ## \<assume_role_credentials\> section
 
 Typically, you use AssumeRole for cross-account access or federation.

lib/fluent/plugin/in_s3.rb

@@ -8,6 +8,7 @@
 require 'zlib'
 require 'time'
 require 'tempfile'
+require 'shellwords'
 
 module Fluent::Plugin
   class S3Input < Input
@@ -30,6 +31,10 @@ def initialize
     config_param :aws_key_id, :string, default: nil, secret: true
     desc "AWS secret key."
     config_param :aws_sec_key, :string, default: nil, secret: true
+    desc "AWS profile name."
+    config_param :aws_profile, :string, default: nil
+    desc "The command to run to generate credentials."
+    config_param :aws_credential_process, :string, default: nil
     config_section :assume_role_credentials, multi: false do
       desc "The Amazon Resource Name (ARN) of the role to assume"
       config_param :role_arn, :string
@@ -211,7 +216,7 @@ def run
             if @match_regexp
               raw_key = get_raw_key(body)
               key = CGI.unescape(raw_key)
-              next unless @match_regexp.match?(key) 
+              next unless @match_regexp.match?(key)
             end
             process(body)
           rescue => e
@@ -231,7 +236,7 @@ def is_valid_queue(body)
       if @sqs.event_bridge_mode
         log.debug("checking for eventbridge property")
         !!body["detail"]
-      else 
+      else
         log.debug("checking for Records property")
         !!body["Records"]
       end
@@ -242,7 +247,7 @@ def get_raw_key(body)
         body["detail"]["object"]["key"]
       else
         body["Records"].first["s3"]["object"]["key"]
-      end     
+      end
     end
 
     def setup_credentials
@@ -252,6 +257,10 @@ def setup_credentials
       when @aws_key_id && @aws_sec_key
         options[:access_key_id] = @aws_key_id
         options[:secret_access_key] = @aws_sec_key
+      when @aws_profile
+        options[:profile] = @aws_profile
+      when @aws_credential_process
+        options[:credentials] = Aws::ProcessCredentials.new(Shellwords.split(@aws_credential_process))
       when @assume_role_credentials
         c = @assume_role_credentials
         credentials_options[:role_arn] = c.role_arn

lib/fluent/plugin/out_s3.rb

@@ -6,6 +6,7 @@
 require 'time'
 require 'tempfile'
 require 'securerandom'
+require 'shellwords'
 
 module Fluent::Plugin
   class S3Output < Output
@@ -29,6 +30,10 @@ def initialize
     config_param :aws_key_id, :string, default: nil, secret: true
     desc "AWS secret key."
     config_param :aws_sec_key, :string, default: nil, secret: true
+    desc "AWS profile name."
+    config_param :aws_profile, :string, default: nil
+    desc "The command to run to generate credentials."
+    config_param :aws_credential_process, :string, default: nil
     config_section :assume_role_credentials, multi: false do
       desc "The Amazon Resource Name (ARN) of the role to assume"
       config_param :role_arn, :string, secret: true
@@ -542,6 +547,10 @@ def setup_credentials
       when @aws_key_id && @aws_sec_key
         options[:access_key_id] = @aws_key_id
         options[:secret_access_key] = @aws_sec_key
+      when @aws_profile
+        options[:profile] = @aws_profile
+      when @aws_credential_process
+        options[:credentials] = Aws::ProcessCredentials.new(Shellwords.split(@aws_credential_process))
       when @web_identity_credentials
         c = @web_identity_credentials
         region = c.sts_region || @s3_region

test/test_in_s3.rb

@@ -71,6 +71,32 @@ def test_empty
       end
     end
 
+    def test_aws_profile
+      d = create_driver(%[
+        aws_profile test_profile
+        s3_bucket test_bucket
+        buffer_type memory
+        <sqs>
+          queue_name test_queue
+          queue_owner_aws_account_id 123456789123
+        </sqs>
+      ])
+      assert_equal 'test_profile', d.instance.aws_profile
+    end
+
+    def test_aws_credential_process
+      d = create_driver(%[
+        aws_credential_process "/path/to/aws_signing_helper credential-process"
+        s3_bucket test_bucket
+        buffer_type memory
+        <sqs>
+          queue_name test_queue
+          queue_owner_aws_account_id 123456789123
+        </sqs>
+      ])
+      assert_equal '/path/to/aws_signing_helper credential-process', d.instance.aws_credential_process
+    end
+
     def test_without_sqs_section
       conf = %[
         aws_key_id test_key_id
@@ -708,7 +734,7 @@ def test_event_bridge_mode
         }
       }
     }
-    
+
     message = Struct::StubMessage.new(1, 1, Yajl.dump(body))
     @sqs_poller.get_messages(anything, anything) do |config, stats|
       config.before_request.call(stats) if config.before_request

test/test_out_s3.rb

@@ -69,6 +69,30 @@ def test_configure
       assert_equal true, d.instance.check_object
     end
 
+    def test_aws_profile
+      d = create_driver(%[
+        aws_profile test_profile
+        s3_bucket test_bucket
+        path log
+        utc
+        buffer_type memory
+        time_slice_format %Y%m%d-%H
+      ])
+      assert_equal 'test_profile', d.instance.aws_profile
+    end
+
+    def test_aws_credential_process
+      d = create_driver(%[
+        aws_credential_process "/path/to/aws_signing_helper credential-process"
+        s3_bucket test_bucket
+        path log
+        utc
+        buffer_type memory
+        time_slice_format %Y%m%d-%H
+      ])
+      assert_equal '/path/to/aws_signing_helper credential-process', d.instance.aws_credential_process
+    end
+
     def test_s3_endpoint_with_valid_endpoint
       d = create_driver(CONFIG + 's3_endpoint riak-cs.example.com')
       assert_equal 'riak-cs.example.com', d.instance.s3_endpoint