From 6208f5c09f9ea853920e6160709b6e55528716d4 Mon Sep 17 00:00:00 2001
From: Shunping Huang <shunping@google.com>
Date: Mon, 12 Oct 2026 11:31:06 -0400
Subject: [PATCH 1/7] [Python] Add --aws_credentials_secret_manager to resolve
 AWS credentials from a secret manager

In this case, --aws_access_key_id, --aws_secret_access_key and --aws_session_token
can be secret specs instead of plain-text values.
---
 .../io/aws/clients/s3/boto3_client.py         | 26 ++++++
 .../io/aws/clients/s3/boto3_client_test.py    | 92 +++++++++++++++++++
 .../apache_beam/options/pipeline_options.py   | 27 +++++-
 3 files changed, 142 insertions(+), 3 deletions(-)

diff --git a/sdks/python/apache_beam/io/aws/clients/s3/boto3_client.py b/sdks/python/apache_beam/io/aws/clients/s3/boto3_client.py
index 57c7c83f3caf..ea1dc5de3bd6 100644
--- a/sdks/python/apache_beam/io/aws/clients/s3/boto3_client.py
+++ b/sdks/python/apache_beam/io/aws/clients/s3/boto3_client.py
@@ -17,9 +17,12 @@
 
 # pytype: skip-file
 
+import functools
+
 from apache_beam.io.aws.clients.s3 import messages
 from apache_beam.options import pipeline_options
 from apache_beam.utils import retry
+from apache_beam.utils.secret import Secret
 
 try:
   # pylint: disable=wrong-import-order, wrong-import-position
@@ -36,6 +39,26 @@ def get_http_error_code(exc):
   return None
 
 
+# S3IO (and therefore Client) is created for almost every filesystem
+# operation, so cache parsed secrets per process to avoid calling the secret
+# manager each time. Note that a rotated secret is only picked up by new
+# worker processes.
+@functools.lru_cache(maxsize=32)
+def _fetch_secret(spec, secret_manager):
+  return Secret.from_json(spec, secret_manager).get_str()
+
+
+def _parse_credential(value, secret_manager):
+  """Parses a credential option with Secret.from_json.
+
+  With no secret_manager, Secret.from_json returns the value itself (as a
+  RawSecret), so plain credentials keep working.
+  """
+  if not value:
+    return value
+  return _fetch_secret(value, secret_manager)
+
+
 class Client(object):
   """
   Wrapper for boto3 library
@@ -55,6 +78,9 @@ def __init__(self, options):
       s3_options = pipeline_options.S3Options(flags=[], **options)
 
     settings = aws_options.get_client_settings(s3_options)
+    secret_manager = aws_options.aws_credentials_secret_manager
+    for name in ('access_key_id', 'secret_access_key', 'session_token'):
+      settings[name] = _parse_credential(settings[name], secret_manager)
 
     session = boto3.session.Session()
     self.client = session.client(
diff --git a/sdks/python/apache_beam/io/aws/clients/s3/boto3_client_test.py b/sdks/python/apache_beam/io/aws/clients/s3/boto3_client_test.py
index 490d7747294f..fc4cd7fb6da1 100644
--- a/sdks/python/apache_beam/io/aws/clients/s3/boto3_client_test.py
+++ b/sdks/python/apache_beam/io/aws/clients/s3/boto3_client_test.py
@@ -17,10 +17,12 @@
 
 """Tests for option handling in the boto3 S3 client."""
 
+import json
 import unittest
 from unittest import mock
 
 from apache_beam.options.pipeline_options import PipelineOptions
+from apache_beam.utils.secret import RawSecret
 
 try:
   import boto3  # pylint: disable=unused-import
@@ -33,6 +35,7 @@
 @unittest.skipIf(boto3_client is None, 'boto3 is not installed')
 class Boto3ClientOptionsTest(unittest.TestCase):
   def setUp(self):
+    boto3_client._fetch_secret.cache_clear()
     patcher = mock.patch('boto3.session.Session')
     self.mock_session = patcher.start().return_value
     self.addCleanup(patcher.stop)
@@ -111,8 +114,97 @@ def test_no_options_uses_boto3_defaults(self):
     kwargs = self._client_kwargs()
     self.assertIsNone(kwargs['aws_access_key_id'])
     self.assertIsNone(kwargs['aws_secret_access_key'])
+    self.assertIsNone(kwargs['aws_session_token'])
     self.assertTrue(kwargs['use_ssl'])
 
+  @mock.patch('apache_beam.utils.secret.Secret.from_json')
+  def test_credentials_parsed_by_secret_manager(self, mock_from_json):
+    key_id_spec = json.dumps({'name': 'key-id', 'project': 'p'})
+    secret_key_spec = json.dumps({'name': 'secret-key', 'project': 'p'})
+    token_spec = json.dumps({'name': 'token', 'project': 'p'})
+    resolved = {
+        key_id_spec: 'AKIDEXAMPLE',
+        secret_key_spec: 'SECRETEXAMPLE',
+        token_spec: 'TOKEN',
+    }
+    mock_from_json.side_effect = lambda spec, manager: RawSecret(resolved[spec])
+
+    boto3_client.Client(
+        PipelineOptions([
+            '--aws_credentials_secret_manager=GoogleCloudSecretManager',
+            f'--aws_access_key_id={key_id_spec}',
+            f'--aws_secret_access_key={secret_key_spec}',
+            f'--aws_session_token={token_spec}',
+        ]))
+
+    kwargs = self._client_kwargs()
+    self.assertEqual(kwargs['aws_access_key_id'], 'AKIDEXAMPLE')
+    self.assertEqual(kwargs['aws_secret_access_key'], 'SECRETEXAMPLE')
+    self.assertEqual(kwargs['aws_session_token'], 'TOKEN')
+    mock_from_json.assert_has_calls([
+        mock.call(key_id_spec, 'GoogleCloudSecretManager'),
+        mock.call(secret_key_spec, 'GoogleCloudSecretManager'),
+        mock.call(token_spec, 'GoogleCloudSecretManager'),
+    ])
+
+  @mock.patch('apache_beam.utils.secret.Secret.from_json')
+  def test_secret_manager_applies_to_s3_credential_overrides(
+      self, mock_from_json):
+    resolved = {
+        'projects/p/secrets/key-id/versions/1': 'AKIDEXAMPLE',
+        'projects/p/secrets/secret-key/versions/1': 'SECRETEXAMPLE',
+    }
+    mock_from_json.side_effect = lambda spec, manager: RawSecret(resolved[spec])
+
+    boto3_client.Client({
+        'aws_credentials_secret_manager': 'GoogleCloudSecretManager',
+        's3_access_key_id': 'projects/p/secrets/key-id/versions/1',
+        's3_secret_access_key': 'projects/p/secrets/secret-key/versions/1',
+    })
+
+    kwargs = self._client_kwargs()
+    self.assertEqual(kwargs['aws_access_key_id'], 'AKIDEXAMPLE')
+    self.assertEqual(kwargs['aws_secret_access_key'], 'SECRETEXAMPLE')
+
+  @mock.patch('google.cloud.secretmanager.SecretManagerServiceClient')
+  def test_gcp_secret_manager_end_to_end(self, mock_gcp_client):
+    payloads = {
+        'projects/p/secrets/key-id/versions/latest': b'AKIDEXAMPLE',
+        'projects/p/secrets/secret-key/versions/latest': b'SECRETEXAMPLE',
+    }
+
+    def access(request):
+      response = mock.MagicMock()
+      response.payload.data = payloads[request['name']]
+      return response
+
+    mock_gcp_client.return_value.access_secret_version.side_effect = access
+
+    boto3_client.Client(
+        PipelineOptions([
+            '--aws_credentials_secret_manager=GoogleCloudSecretManager',
+            '--aws_access_key_id={"name": "key-id", "project": "p"}',
+            '--aws_secret_access_key={"name": "secret-key", "project": "p"}',
+        ]))
+
+    kwargs = self._client_kwargs()
+    self.assertEqual(kwargs['aws_access_key_id'], 'AKIDEXAMPLE')
+    self.assertEqual(kwargs['aws_secret_access_key'], 'SECRETEXAMPLE')
+
+  @mock.patch('apache_beam.utils.secret.Secret.from_json')
+  def test_parsed_secrets_are_cached(self, mock_from_json):
+    mock_from_json.side_effect = lambda spec, manager: RawSecret('value')
+    options = PipelineOptions([
+        '--aws_credentials_secret_manager=GoogleCloudSecretManager',
+        '--aws_access_key_id=projects/p/secrets/key-id/versions/1',
+        '--aws_secret_access_key=projects/p/secrets/secret-key/versions/1',
+    ])
+
+    boto3_client.Client(options)
+    boto3_client.Client(options)
+
+    self.assertEqual(mock_from_json.call_count, 2)
+
 
 if __name__ == '__main__':
   unittest.main()
diff --git a/sdks/python/apache_beam/options/pipeline_options.py b/sdks/python/apache_beam/options/pipeline_options.py
index 24abcae4d3e3..84c0743ceaaa 100644
--- a/sdks/python/apache_beam/options/pipeline_options.py
+++ b/sdks/python/apache_beam/options/pipeline_options.py
@@ -2426,15 +2426,36 @@ def _add_argparse_args(cls, parser):
     parser.add_argument(
         '--aws_access_key_id',
         default=None,
-        help='The access key ID to use when creating AWS clients.')
+        help=(
+            'The access key ID to use when creating AWS clients. If '
+            '--aws_credentials_secret_manager is set, this is a secret spec '
+            'parsed by that secret manager.'))
     parser.add_argument(
         '--aws_secret_access_key',
         default=None,
-        help='The secret access key to use when creating AWS clients.')
+        help=(
+            'The secret access key to use when creating AWS clients. If '
+            '--aws_credentials_secret_manager is set, this is a secret spec '
+            'parsed by that secret manager.'))
     parser.add_argument(
         '--aws_session_token',
         default=None,
-        help='The session token to use when creating AWS clients.')
+        help=(
+            'The session token to use when creating AWS clients. If '
+            '--aws_credentials_secret_manager is set, this is a secret spec '
+            'parsed by that secret manager.'))
+    parser.add_argument(
+        '--aws_credentials_secret_manager',
+        default=None,
+        help=(
+            'Secret manager used to parse --aws_access_key_id, '
+            '--aws_secret_access_key and --aws_session_token (and their '
+            'service-specific overrides, e.g. --s3_access_key_id) as secret '
+            'specs (a JSON string or plain text), e.g. '
+            'GoogleCloudSecretManager. For example: '
+            '--aws_credentials_secret_manager=GoogleCloudSecretManager '
+            '--aws_secret_access_key=\'{"name": "my-key", "project": "p"}\'. '
+            'If not set, these options are used as raw credentials.'))
     parser.add_argument(
         '--aws_endpoint_url',
         default=None,

From a08950ce5f289197cfe233f97f95bbf3f68a1ab5 Mon Sep 17 00:00:00 2001
From: Shunping Huang <shunping@google.com>
Date: Mon, 12 Oct 2026 14:45:21 -0400
Subject: [PATCH 2/7] Refactor: Move AWS client option handling into shared
 client_options module

---
 .../io/aws/clients/client_options.py          | 111 ++++++++++++++++++
 .../io/aws/clients/s3/boto3_client.py         |  46 +-------
 .../io/aws/clients/s3/boto3_client_test.py    |   3 +-
 3 files changed, 118 insertions(+), 42 deletions(-)
 create mode 100644 sdks/python/apache_beam/io/aws/clients/client_options.py

diff --git a/sdks/python/apache_beam/io/aws/clients/client_options.py b/sdks/python/apache_beam/io/aws/clients/client_options.py
new file mode 100644
index 000000000000..12838ca15e38
--- /dev/null
+++ b/sdks/python/apache_beam/io/aws/clients/client_options.py
@@ -0,0 +1,111 @@
+#
+# 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.
+#
+
+"""Builds boto3 client arguments from :class:`AwsOptions`.
+
+Shared by the S3 filesystem client and :class:`AwsSecret` so that every boto3
+client created by Beam uses the same credentials and connection settings.
+"""
+
+# pytype: skip-file
+
+import functools
+from typing import Any
+from typing import Optional
+from typing import Union
+
+from apache_beam.options.pipeline_options import AwsOptions
+from apache_beam.options.pipeline_options import PipelineOptions
+from apache_beam.utils.secret import Secret
+
+# Secret manager names that resolve to AwsSecret (see _SECRET_CLASSES in
+# apache_beam.utils.secret). AWS credentials cannot be stored there, because
+# the credentials are needed to access it.
+_AWS_SECRET_MANAGERS = frozenset(
+    {'awssecretsmanager', 'awssecretmanager', 'awssecret'})
+
+
+def view_options(
+    options: Union[PipelineOptions, dict],
+    options_cls: type[PipelineOptions]) -> PipelineOptions:
+  """Returns an ``options_cls`` view of PipelineOptions or a dict of values."""
+  if isinstance(options, PipelineOptions):
+    return options.view_as(options_cls)
+  return options_cls(flags=[], **options)
+
+
+# S3IO (and therefore the S3 client) is created for almost every filesystem
+# operation, so cache parsed secrets per process to avoid calling the secret
+# manager each time. Note that a rotated secret is only picked up by new
+# worker processes.
+@functools.lru_cache(maxsize=32)
+def _fetch_secret(spec, secret_manager):
+  return Secret.from_json(spec, secret_manager).get_str()
+
+
+def _parse_credential(value, secret_manager):
+  """Parses a credential option with Secret.from_json.
+
+  With no secret_manager, Secret.from_json returns the value itself (as a
+  RawSecret), so plain credentials keep working.
+  """
+  if not value:
+    return value
+  if (secret_manager and
+      secret_manager.strip().lower() in _AWS_SECRET_MANAGERS):
+    raise ValueError(
+        '--aws_credentials_secret_manager cannot be AWS Secrets Manager, '
+        'because the AWS credentials are needed to access it. Store the '
+        'credentials in another secret manager, e.g. GoogleCloudSecretManager.')
+  return _fetch_secret(value, secret_manager)
+
+
+def get_client_kwargs(
+    options: Union[PipelineOptions, dict],
+    service_options_cls: Optional[type[PipelineOptions]] = None
+) -> dict[str, Any]:
+  """Returns common boto3 client keyword arguments.
+
+  Settings come from the --aws_* defaults in AwsOptions, with the overrides
+  from ``service_options_cls`` applied (e.g. --s3_* in S3Options; see
+  AwsOptions.get_client_settings). Credentials are then resolved through
+  --aws_credentials_secret_manager if it is set. Service-specific settings that
+  are not shared (e.g. --s3_api_version) are not included.
+
+  Args:
+    options: PipelineOptions, or a dict of option names to values.
+    service_options_cls: Optional options class whose overrides apply, e.g.
+      S3Options. If None, only the --aws_* defaults are used.
+  """
+  aws_options = view_options(options, AwsOptions)
+  service_options = (
+      view_options(options, service_options_cls)
+      if service_options_cls else None)
+  settings = aws_options.get_client_settings(service_options)
+  secret_manager = aws_options.aws_credentials_secret_manager
+  return {
+      'aws_access_key_id': _parse_credential(
+          settings['access_key_id'], secret_manager),
+      'aws_secret_access_key': _parse_credential(
+          settings['secret_access_key'], secret_manager),
+      'aws_session_token': _parse_credential(
+          settings['session_token'], secret_manager),
+      'region_name': settings['region_name'],
+      'endpoint_url': settings['endpoint_url'],
+      'use_ssl': not settings['disable_ssl'],
+      'verify': settings['verify'],
+  }
diff --git a/sdks/python/apache_beam/io/aws/clients/s3/boto3_client.py b/sdks/python/apache_beam/io/aws/clients/s3/boto3_client.py
index ea1dc5de3bd6..5a8dcb2a26cf 100644
--- a/sdks/python/apache_beam/io/aws/clients/s3/boto3_client.py
+++ b/sdks/python/apache_beam/io/aws/clients/s3/boto3_client.py
@@ -17,12 +17,10 @@
 
 # pytype: skip-file
 
-import functools
-
+from apache_beam.io.aws.clients import client_options
 from apache_beam.io.aws.clients.s3 import messages
 from apache_beam.options import pipeline_options
 from apache_beam.utils import retry
-from apache_beam.utils.secret import Secret
 
 try:
   # pylint: disable=wrong-import-order, wrong-import-position
@@ -39,26 +37,6 @@ def get_http_error_code(exc):
   return None
 
 
-# S3IO (and therefore Client) is created for almost every filesystem
-# operation, so cache parsed secrets per process to avoid calling the secret
-# manager each time. Note that a rotated secret is only picked up by new
-# worker processes.
-@functools.lru_cache(maxsize=32)
-def _fetch_secret(spec, secret_manager):
-  return Secret.from_json(spec, secret_manager).get_str()
-
-
-def _parse_credential(value, secret_manager):
-  """Parses a credential option with Secret.from_json.
-
-  With no secret_manager, Secret.from_json returns the value itself (as a
-  RawSecret), so plain credentials keep working.
-  """
-  if not value:
-    return value
-  return _fetch_secret(value, secret_manager)
-
-
 class Client(object):
   """
   Wrapper for boto3 library
@@ -70,29 +48,15 @@ def __init__(self, options):
         values. --s3_* options override the --aws_* defaults for S3.
     """
     assert boto3 is not None, 'Missing boto3 requirement'
-    if isinstance(options, pipeline_options.PipelineOptions):
-      aws_options = options.view_as(pipeline_options.AwsOptions)
-      s3_options = options.view_as(pipeline_options.S3Options)
-    else:
-      aws_options = pipeline_options.AwsOptions(flags=[], **options)
-      s3_options = pipeline_options.S3Options(flags=[], **options)
-
-    settings = aws_options.get_client_settings(s3_options)
-    secret_manager = aws_options.aws_credentials_secret_manager
-    for name in ('access_key_id', 'secret_access_key', 'session_token'):
-      settings[name] = _parse_credential(settings[name], secret_manager)
+    s3_options = client_options.view_options(
+        options, pipeline_options.S3Options)
 
     session = boto3.session.Session()
     self.client = session.client(
         service_name='s3',
-        region_name=settings['region_name'],
         api_version=s3_options.s3_api_version,
-        use_ssl=not settings['disable_ssl'],
-        verify=settings['verify'],
-        endpoint_url=settings['endpoint_url'],
-        aws_access_key_id=settings['access_key_id'],
-        aws_secret_access_key=settings['secret_access_key'],
-        aws_session_token=settings['session_token'])
+        **client_options.get_client_kwargs(
+            options, pipeline_options.S3Options))
 
     self._download_request = None
     self._download_stream = None
diff --git a/sdks/python/apache_beam/io/aws/clients/s3/boto3_client_test.py b/sdks/python/apache_beam/io/aws/clients/s3/boto3_client_test.py
index fc4cd7fb6da1..2e3eeec9d433 100644
--- a/sdks/python/apache_beam/io/aws/clients/s3/boto3_client_test.py
+++ b/sdks/python/apache_beam/io/aws/clients/s3/boto3_client_test.py
@@ -27,6 +27,7 @@
 try:
   import boto3  # pylint: disable=unused-import
 
+  from apache_beam.io.aws.clients import client_options
   from apache_beam.io.aws.clients.s3 import boto3_client
 except ImportError:
   boto3_client = None
@@ -35,7 +36,7 @@
 @unittest.skipIf(boto3_client is None, 'boto3 is not installed')
 class Boto3ClientOptionsTest(unittest.TestCase):
   def setUp(self):
-    boto3_client._fetch_secret.cache_clear()
+    client_options._fetch_secret.cache_clear()
     patcher = mock.patch('boto3.session.Session')
     self.mock_session = patcher.start().return_value
     self.addCleanup(patcher.stop)

From 78b4d4335c46fb19d6aa56b434aa29f71b7282d0 Mon Sep 17 00:00:00 2001
From: Shunping Huang <shunping@google.com>
Date: Mon, 12 Oct 2026 19:13:33 -0400
Subject: [PATCH 3/7] Update CHANGES.md for --aws_credentials_secret_manager

---
 CHANGES.md | 1 +
 1 file changed, 1 insertion(+)

diff --git a/CHANGES.md b/CHANGES.md
index fda4c9edc218..834e19eb61fe 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -74,6 +74,7 @@
 * (Go) Pipeline-level `--resource_hints` are now forwarded to expansion services, so they apply to cross-language transforms as they do in the Java and Python SDKs. The `max_active_bundles_per_worker` hint name is now also accepted by `--resource_hints` ([#23893](https://github.com/apache/beam/issues/23893)).
 * ReadFromBigQuery now supports Lakehouse runtime catalog (BigLake metastore) tables with `method=DIRECT_READ`, using 4-part `project.catalog.namespace.table` identifiers. Previously `project:catalog.namespace.table` was silently mis-parsed and tables that report no `numBytes` failed to split (Python) ([#39597](https://github.com/apache/beam/issues/39597)).
 * (Python) Added `AwsOptions` with `--aws_*` pipeline options (e.g. `--aws_access_key_id`, `--aws_region_name`, `--aws_endpoint_url`) that set defaults for all AWS service clients. The existing `--s3_*` options in `S3Options` override these defaults for the S3 client only ([#40528](https://github.com/apache/beam/issues/40528)).
+* (Python) Added `--aws_credentials_secret_manager` so AWS credentials (`--aws_access_key_id`, `--aws_secret_access_key`, `--aws_session_token` and their `--s3_*` overrides) can be secret specs resolved through a secret manager such as `GoogleCloudSecretManager`, instead of plain-text values ([#40528](https://github.com/apache/beam/issues/40528)).
 
 ## Breaking Changes
 

From 0314fd631db03577191cd3da4f36afcb9d3014c7 Mon Sep 17 00:00:00 2001
From: Shunping Huang <shunping@google.com>
Date: Mon, 12 Oct 2026 19:26:40 -0400
Subject: [PATCH 4/7] Reformat

---
 sdks/python/apache_beam/io/aws/clients/s3/boto3_client.py | 3 +--
 1 file changed, 1 insertion(+), 2 deletions(-)

diff --git a/sdks/python/apache_beam/io/aws/clients/s3/boto3_client.py b/sdks/python/apache_beam/io/aws/clients/s3/boto3_client.py
index 5a8dcb2a26cf..d4986eaad94a 100644
--- a/sdks/python/apache_beam/io/aws/clients/s3/boto3_client.py
+++ b/sdks/python/apache_beam/io/aws/clients/s3/boto3_client.py
@@ -55,8 +55,7 @@ def __init__(self, options):
     self.client = session.client(
         service_name='s3',
         api_version=s3_options.s3_api_version,
-        **client_options.get_client_kwargs(
-            options, pipeline_options.S3Options))
+        **client_options.get_client_kwargs(options, pipeline_options.S3Options))
 
     self._download_request = None
     self._download_stream = None

From e503b8a394da09803db3627fb0a109cadc671c79 Mon Sep 17 00:00:00 2001
From: Shunping Huang <shunping@google.com>
Date: Mon, 12 Oct 2026 21:49:39 -0400
Subject: [PATCH 5/7] Add connection-setting overrides to AWS client_options

Let callers override non-credential AWS client settings (e.g.
region_name, endpoint_url, verify, disable_ssl) on top of the --aws_*
options and the service overrides.
---
 .../io/aws/clients/client_options.py          | 97 ++++++++++++++++---
 1 file changed, 81 insertions(+), 16 deletions(-)

diff --git a/sdks/python/apache_beam/io/aws/clients/client_options.py b/sdks/python/apache_beam/io/aws/clients/client_options.py
index 12838ca15e38..89e33516fcf4 100644
--- a/sdks/python/apache_beam/io/aws/clients/client_options.py
+++ b/sdks/python/apache_beam/io/aws/clients/client_options.py
@@ -24,6 +24,8 @@
 # pytype: skip-file
 
 import functools
+import types
+from collections.abc import Mapping
 from typing import Any
 from typing import Optional
 from typing import Union
@@ -32,11 +34,70 @@
 from apache_beam.options.pipeline_options import PipelineOptions
 from apache_beam.utils.secret import Secret
 
-# Secret manager names that resolve to AwsSecret (see _SECRET_CLASSES in
-# apache_beam.utils.secret). AWS credentials cannot be stored there, because
-# the credentials are needed to access it.
-_AWS_SECRET_MANAGERS = frozenset(
-    {'awssecretsmanager', 'awssecretmanager', 'awssecret'})
+# Client settings (see AwsOptions.get_client_settings) that hold credentials.
+CREDENTIAL_SETTINGS = frozenset(
+    {'access_key_id', 'secret_access_key', 'session_token'})
+
+_TRUE_STRINGS = frozenset({'true', '1', 'yes', 'y', 'on'})
+_FALSE_STRINGS = frozenset({'false', '0', 'no', 'n', 'off', ''})
+
+
+@functools.lru_cache(maxsize=None)
+def get_connection_setting_defaults() -> Mapping[str, Any]:
+  """Returns the non-credential client settings and their default values.
+
+  The names (e.g. ``region_name``, ``endpoint_url``) come from
+  AwsOptions.get_client_settings, so new settings added to AwsOptions are
+  picked up automatically. The result is computed once and is read-only.
+  """
+  return types.MappingProxyType({
+      name: value
+      for name, value in AwsOptions(flags=[]).get_client_settings().items()
+      if name not in CREDENTIAL_SETTINGS
+  })
+
+
+def parse_connection_settings(settings: Mapping[str, Any]) -> dict[str, Any]:
+  """Validates connection settings, e.g. from a secret spec.
+
+  Values for boolean settings (e.g. ``disable_ssl``) may be strings such as
+  'true' or 'false'.
+
+  Args:
+    settings: A dict of client setting names (without the ``aws_`` prefix) to
+      values.
+
+  Returns:
+    A dict of the settings with values converted to the setting types.
+
+  Raises:
+    ValueError: If a setting is a credential or not a known setting, or a
+      boolean value is invalid.
+  """
+  credentials = sorted(CREDENTIAL_SETTINGS.intersection(settings))
+  if credentials:
+    raise ValueError(
+        f"Invalid secret parameter {', '.join(credentials)}: AWS credentials "
+        "cannot be set here. Use the "
+        f"{', '.join('--aws_' + name for name in credentials)} pipeline "
+        "option instead.")
+  defaults = get_connection_setting_defaults()
+  unknown = sorted(set(settings) - set(defaults))
+  if unknown:
+    raise ValueError(f"Invalid secret parameter {', '.join(unknown)}")
+
+  parsed = {}
+  for name, value in settings.items():
+    if isinstance(defaults[name], bool) and isinstance(value, str):
+      lowered = value.strip().lower()
+      if lowered in _TRUE_STRINGS:
+        value = True
+      elif lowered in _FALSE_STRINGS:
+        value = False
+      else:
+        raise ValueError(f'Invalid boolean value for {name}: {value}')
+    parsed[name] = value
+  return parsed
 
 
 def view_options(
@@ -65,37 +126,41 @@ def _parse_credential(value, secret_manager):
   """
   if not value:
     return value
-  if (secret_manager and
-      secret_manager.strip().lower() in _AWS_SECRET_MANAGERS):
-    raise ValueError(
-        '--aws_credentials_secret_manager cannot be AWS Secrets Manager, '
-        'because the AWS credentials are needed to access it. Store the '
-        'credentials in another secret manager, e.g. GoogleCloudSecretManager.')
   return _fetch_secret(value, secret_manager)
 
 
 def get_client_kwargs(
     options: Union[PipelineOptions, dict],
-    service_options_cls: Optional[type[PipelineOptions]] = None
-) -> dict[str, Any]:
+    service_options_cls: Optional[type[PipelineOptions]] = None,
+    overrides: Optional[dict[str, Any]] = None) -> dict[str, Any]:
   """Returns common boto3 client keyword arguments.
 
   Settings come from the --aws_* defaults in AwsOptions, with the overrides
   from ``service_options_cls`` applied (e.g. --s3_* in S3Options; see
-  AwsOptions.get_client_settings). Credentials are then resolved through
-  --aws_credentials_secret_manager if it is set. Service-specific settings that
-  are not shared (e.g. --s3_api_version) are not included.
+  AwsOptions.get_client_settings), and then ``overrides``. Credentials are then
+  resolved through --aws_credentials_secret_manager if it is set.
+  Service-specific settings that are not shared (e.g. --s3_api_version) are not
+  included.
 
   Args:
     options: PipelineOptions, or a dict of option names to values.
     service_options_cls: Optional options class whose overrides apply, e.g.
       S3Options. If None, only the --aws_* defaults are used.
+    overrides: Optional dict of connection settings (setting names without the
+      ``aws_`` prefix, e.g. ``region_name``) that take precedence over the
+      options. Values that are None are ignored. Credentials cannot be
+      overridden; see parse_connection_settings.
   """
   aws_options = view_options(options, AwsOptions)
   service_options = (
       view_options(options, service_options_cls)
       if service_options_cls else None)
   settings = aws_options.get_client_settings(service_options)
+  if overrides:
+    settings.update(
+        (name, value)
+        for name, value in parse_connection_settings(overrides).items()
+        if value is not None)
   secret_manager = aws_options.aws_credentials_secret_manager
   return {
       'aws_access_key_id': _parse_credential(

From b9d1d30ad4534ea3b87ad2126d7d3bf9484c7d28 Mon Sep 17 00:00:00 2001
From: Shunping Huang <shunping@google.com>
Date: Mon, 12 Oct 2026 22:29:14 -0400
Subject: [PATCH 6/7] Add unit test for AWS connection-setting overrides

---
 .../io/aws/clients/client_options.py          |  4 +--
 .../io/aws/clients/s3/boto3_client_test.py    | 26 +++++++++++++++++++
 2 files changed, 28 insertions(+), 2 deletions(-)

diff --git a/sdks/python/apache_beam/io/aws/clients/client_options.py b/sdks/python/apache_beam/io/aws/clients/client_options.py
index 89e33516fcf4..8038f0839a00 100644
--- a/sdks/python/apache_beam/io/aws/clients/client_options.py
+++ b/sdks/python/apache_beam/io/aws/clients/client_options.py
@@ -17,8 +17,8 @@
 
 """Builds boto3 client arguments from :class:`AwsOptions`.
 
-Shared by the S3 filesystem client and :class:`AwsSecret` so that every boto3
-client created by Beam uses the same credentials and connection settings.
+Shared across AWS service clients (e.g. the S3 filesystem client) so that every
+boto3 client created by Beam uses the same credentials and connection settings.
 """
 
 # pytype: skip-file
diff --git a/sdks/python/apache_beam/io/aws/clients/s3/boto3_client_test.py b/sdks/python/apache_beam/io/aws/clients/s3/boto3_client_test.py
index 2e3eeec9d433..ac7dd73d7269 100644
--- a/sdks/python/apache_beam/io/aws/clients/s3/boto3_client_test.py
+++ b/sdks/python/apache_beam/io/aws/clients/s3/boto3_client_test.py
@@ -206,6 +206,32 @@ def test_parsed_secrets_are_cached(self, mock_from_json):
 
     self.assertEqual(mock_from_json.call_count, 2)
 
+  def test_connection_setting_overrides(self):
+    kwargs = client_options.get_client_kwargs(
+        PipelineOptions([
+            '--aws_access_key_id=AKIDEXAMPLE',
+            '--aws_region_name=us-east-1',
+            '--s3_region_name=us-east-2',
+        ]),
+        overrides={
+            'region_name': 'us-west-2',
+            'endpoint_url': 'http://localhost:4566',
+            'disable_ssl': 'true',
+            'verify': None,
+        })
+    self.assertEqual(kwargs['aws_access_key_id'], 'AKIDEXAMPLE')
+    self.assertEqual(kwargs['region_name'], 'us-west-2')
+    self.assertEqual(kwargs['endpoint_url'], 'http://localhost:4566')
+    self.assertFalse(kwargs['use_ssl'])
+
+    for bad in (
+        {'access_key_id': 'x'},
+        {'unknown_setting': 'x'},
+        {'disable_ssl': 'maybe'},
+    ):
+      with self.assertRaises(ValueError):
+        client_options.parse_connection_settings(bad)
+
 
 if __name__ == '__main__':
   unittest.main()

From a877f854738e2187bb137e59848c137b2417c942 Mon Sep 17 00:00:00 2001
From: Shunping Huang <shunping@google.com>
Date: Tue, 13 Oct 2026 14:21:02 -0400
Subject: [PATCH 7/7] Address reviewer's feedback

---
 sdks/python/apache_beam/io/aws/clients/client_options.py    | 5 +++--
 .../apache_beam/io/aws/clients/s3/boto3_client_test.py      | 6 ++++++
 2 files changed, 9 insertions(+), 2 deletions(-)

diff --git a/sdks/python/apache_beam/io/aws/clients/client_options.py b/sdks/python/apache_beam/io/aws/clients/client_options.py
index 8038f0839a00..4057886a1528 100644
--- a/sdks/python/apache_beam/io/aws/clients/client_options.py
+++ b/sdks/python/apache_beam/io/aws/clients/client_options.py
@@ -114,11 +114,12 @@ def view_options(
 # manager each time. Note that a rotated secret is only picked up by new
 # worker processes.
 @functools.lru_cache(maxsize=32)
-def _fetch_secret(spec, secret_manager):
+def _fetch_secret(spec: str, secret_manager: Optional[str]) -> str:
   return Secret.from_json(spec, secret_manager).get_str()
 
 
-def _parse_credential(value, secret_manager):
+def _parse_credential(value: Optional[str],
+                      secret_manager: Optional[str]) -> Optional[str]:
   """Parses a credential option with Secret.from_json.
 
   With no secret_manager, Secret.from_json returns the value itself (as a
diff --git a/sdks/python/apache_beam/io/aws/clients/s3/boto3_client_test.py b/sdks/python/apache_beam/io/aws/clients/s3/boto3_client_test.py
index ac7dd73d7269..c0c853632de2 100644
--- a/sdks/python/apache_beam/io/aws/clients/s3/boto3_client_test.py
+++ b/sdks/python/apache_beam/io/aws/clients/s3/boto3_client_test.py
@@ -32,6 +32,11 @@
 except ImportError:
   boto3_client = None
 
+try:
+  from google.cloud import secretmanager  # pylint: disable=unused-import
+except ImportError:
+  secretmanager = None
+
 
 @unittest.skipIf(boto3_client is None, 'boto3 is not installed')
 class Boto3ClientOptionsTest(unittest.TestCase):
@@ -167,6 +172,7 @@ def test_secret_manager_applies_to_s3_credential_overrides(
     self.assertEqual(kwargs['aws_access_key_id'], 'AKIDEXAMPLE')
     self.assertEqual(kwargs['aws_secret_access_key'], 'SECRETEXAMPLE')
 
+  @unittest.skipIf(secretmanager is None, 'GCP dependencies are not installed')
   @mock.patch('google.cloud.secretmanager.SecretManagerServiceClient')
   def test_gcp_secret_manager_end_to_end(self, mock_gcp_client):
     payloads = {