This is an automated email from the ASF dual-hosted git repository.
shunping pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new a9f808e4c45 [Python] Resolve AWS credentials through a secret manager
(`--aws_credentials_secret_manager`) (#40547)
a9f808e4c45 is described below
commit a9f808e4c45e66189ce8861e67839163762e2a3a
Author: Shunping Huang <[email protected]>
AuthorDate: Tue Oct 13 16:30:46 2026 -0400
[Python] Resolve AWS credentials through a secret manager
(`--aws_credentials_secret_manager`) (#40547)
* [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.
* Refactor: Move AWS client option handling into shared client_options
module
* Update CHANGES.md for --aws_credentials_secret_manager
* Reformat
* 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.
* Add unit test for AWS connection-setting overrides
* Address reviewer's feedback
---
CHANGES.md | 1 +
.../apache_beam/io/aws/clients/client_options.py | 177 +++++++++++++++++++++
.../apache_beam/io/aws/clients/s3/boto3_client.py | 19 +--
.../io/aws/clients/s3/boto3_client_test.py | 125 +++++++++++++++
.../python/apache_beam/options/pipeline_options.py | 27 +++-
5 files changed, 331 insertions(+), 18 deletions(-)
diff --git a/CHANGES.md b/CHANGES.md
index d1910513da5..7693e65d78a 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
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 00000000000..4057886a152
--- /dev/null
+++ b/sdks/python/apache_beam/io/aws/clients/client_options.py
@@ -0,0 +1,177 @@
+#
+# 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 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
+
+import functools
+import types
+from collections.abc import Mapping
+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
+
+# 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', ''})
+
+
[email protected]_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(
+ 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.
[email protected]_cache(maxsize=32)
+def _fetch_secret(spec: str, secret_manager: Optional[str]) -> str:
+ return Secret.from_json(spec, secret_manager).get_str()
+
+
+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
+ RawSecret), so plain credentials keep working.
+ """
+ if not value:
+ return value
+ return _fetch_secret(value, secret_manager)
+
+
+def get_client_kwargs(
+ options: Union[PipelineOptions, dict],
+ 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), 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(
+ 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 57c7c83f3ca..d4986eaad94 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,6 +17,7 @@
# pytype: skip-file
+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
@@ -47,26 +48,14 @@ class Client(object):
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)
+ 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 490d7747294..c0c853632de 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,22 +17,31 @@
"""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
+ from apache_beam.io.aws.clients import client_options
from apache_beam.io.aws.clients.s3 import boto3_client
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):
def setUp(self):
+ client_options._fetch_secret.cache_clear()
patcher = mock.patch('boto3.session.Session')
self.mock_session = patcher.start().return_value
self.addCleanup(patcher.stop)
@@ -111,8 +120,124 @@ class Boto3ClientOptionsTest(unittest.TestCase):
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')
+
+ @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 = {
+ '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)
+
+ 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()
diff --git a/sdks/python/apache_beam/options/pipeline_options.py
b/sdks/python/apache_beam/options/pipeline_options.py
index 24abcae4d3e..84c0743ceaa 100644
--- a/sdks/python/apache_beam/options/pipeline_options.py
+++ b/sdks/python/apache_beam/options/pipeline_options.py
@@ -2426,15 +2426,36 @@ class AwsOptions(PipelineOptions):
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,
