From 6c79a025438dc4ec791feef7a11623858154d01e Mon Sep 17 00:00:00 2001 From: Shunping Huang Date: Wed, 14 Oct 2026 15:07:00 -0400 Subject: [PATCH 1/4] Add examples for AWS credentials from a secret manager Add examples under apache_beam/examples/secret_manager that set --aws_credentials_secret_manager to read AWS credentials from Google Cloud Secret Manager: * aws_credentials_s3_read.py: reads an S3 file outside a pipeline. * aws_credentials_s3_pipeline.py: reads an S3 file in a pipeline (DirectRunner or Dataflow). * aws_credentials_secret_read.py: reads a secret from AWS Secrets Manager with Secret.parse_secret_option and Secret.from_json. --- .../examples/secret_manager/__init__.py | 18 +++++ .../aws_credentials_s3_pipeline.py | 76 +++++++++++++++++++ .../secret_manager/aws_credentials_s3_read.py | 60 +++++++++++++++ .../aws_credentials_secret_read.py | 73 ++++++++++++++++++ 4 files changed, 227 insertions(+) create mode 100644 sdks/python/apache_beam/examples/secret_manager/__init__.py create mode 100644 sdks/python/apache_beam/examples/secret_manager/aws_credentials_s3_pipeline.py create mode 100644 sdks/python/apache_beam/examples/secret_manager/aws_credentials_s3_read.py create mode 100644 sdks/python/apache_beam/examples/secret_manager/aws_credentials_secret_read.py diff --git a/sdks/python/apache_beam/examples/secret_manager/__init__.py b/sdks/python/apache_beam/examples/secret_manager/__init__.py new file mode 100644 index 000000000000..92dc81c246f6 --- /dev/null +++ b/sdks/python/apache_beam/examples/secret_manager/__init__.py @@ -0,0 +1,18 @@ +# +# 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. +# + +"""Examples for SecretManager integration in Apache Beam.""" diff --git a/sdks/python/apache_beam/examples/secret_manager/aws_credentials_s3_pipeline.py b/sdks/python/apache_beam/examples/secret_manager/aws_credentials_s3_pipeline.py new file mode 100644 index 000000000000..704ea45254ad --- /dev/null +++ b/sdks/python/apache_beam/examples/secret_manager/aws_credentials_s3_pipeline.py @@ -0,0 +1,76 @@ +# +# 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. +# + +"""A pipeline that reads an S3 file and logs each line, using AWS credentials +stored in Google Cloud Secret Manager. + +Usage (DirectRunner): + python -m apache_beam.examples.secret_manager.aws_credentials_s3_pipeline \ + --input=s3://my-bucket/my-file.txt \ + --aws_region_name=us-east-1 \ + --aws_credentials_secret_manager=GoogleCloudSecretManager \ + --aws_access_key_id='{"name": "my-aws-key-id", "project": "my-project"}' \ + --aws_secret_access_key='{"name": "my-aws-secret-key", "project": "my-project"}' + +Usage (DataflowRunner): add the following flags to the command above. + --runner=DataflowRunner \ + --project=my-project \ + --region=us-central1 \ + --temp_location=gs://my-gcs-bucket/temp \ + --requirements_file=requirements.txt + +where requirements.txt contains: + boto3 + google-cloud-secret-manager + +Required permissions: + * The AWS IAM user that owns the access key needs ``s3:ListBucket`` on the + bucket (``arn:aws:s3:::my-bucket``, used to match the input path) and + ``s3:GetObject`` on the object (``arn:aws:s3:::my-bucket/my-file.txt``), + for example through the ``AmazonS3ReadOnlyAccess`` managed policy or a + narrower custom policy. If the object is encrypted with a customer managed + KMS key, it also needs ``kms:Decrypt`` on that key. + * The Google Cloud account that resolves the credentials needs the Secret + Manager Secret Accessor role (``roles/secretmanager.secretAccessor``) on + the two secrets that hold the access key ID and secret access key. The + credentials are resolved both when the pipeline is constructed and where + the file is read, so this is the account running the script and, on + Dataflow, also the worker service account. +""" + +import argparse +import logging + +import apache_beam as beam +from apache_beam.options.pipeline_options import PipelineOptions + + +def main(argv=None): + parser = argparse.ArgumentParser() + parser.add_argument('--input', required=True, help='S3 path to read.') + known_args, pipeline_args = parser.parse_known_args(argv) + + with beam.Pipeline(options=PipelineOptions(pipeline_args)) as p: + _ = ( + p + | 'Read' >> beam.io.ReadFromText(known_args.input) + | 'Log' >> beam.Map(logging.info)) + + +if __name__ == '__main__': + logging.getLogger().setLevel(logging.INFO) + main() diff --git a/sdks/python/apache_beam/examples/secret_manager/aws_credentials_s3_read.py b/sdks/python/apache_beam/examples/secret_manager/aws_credentials_s3_read.py new file mode 100644 index 000000000000..6ef1c1ba8049 --- /dev/null +++ b/sdks/python/apache_beam/examples/secret_manager/aws_credentials_s3_read.py @@ -0,0 +1,60 @@ +# +# 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. +# + +"""Reads an S3 file outside of a pipeline, using AWS credentials stored in +Google Cloud Secret Manager. + +Usage: + python -m apache_beam.examples.secret_manager.aws_credentials_s3_read \ + --input=s3://my-bucket/my-file.txt \ + --aws_region_name=us-east-1 \ + --aws_credentials_secret_manager=GoogleCloudSecretManager \ + --aws_access_key_id='{"name": "my-aws-key-id", "project": "my-project"}' \ + --aws_secret_access_key='{"name": "my-aws-secret-key", "project": "my-project"}' + +Required permissions: + * The AWS IAM user that owns the access key needs ``s3:GetObject`` on the + object (e.g. ``arn:aws:s3:::my-bucket/my-file.txt``), for example through + the ``AmazonS3ReadOnlyAccess`` managed policy or a narrower custom policy. + If the object is encrypted with a customer managed KMS key, it also needs + ``kms:Decrypt`` on that key. + * The Google Cloud account running this script (e.g. your Application + Default Credentials) needs the Secret Manager Secret Accessor role + (``roles/secretmanager.secretAccessor``) on the two secrets that hold the + access key ID and secret access key. +""" + +import argparse + +from apache_beam.io.filesystems import FileSystems +from apache_beam.options.pipeline_options import PipelineOptions + + +def main(argv=None): + parser = argparse.ArgumentParser() + parser.add_argument('--input', required=True, help='S3 path to read.') + known_args, pipeline_args = parser.parse_known_args(argv) + + # The S3 filesystem resolves the --aws_* credentials through the secret + # manager given by --aws_credentials_secret_manager. + FileSystems.set_options(PipelineOptions(pipeline_args)) + with FileSystems.open(known_args.input) as f: + print(f.read().decode('utf-8')) + + +if __name__ == '__main__': + main() diff --git a/sdks/python/apache_beam/examples/secret_manager/aws_credentials_secret_read.py b/sdks/python/apache_beam/examples/secret_manager/aws_credentials_secret_read.py new file mode 100644 index 000000000000..e702fa121d2b --- /dev/null +++ b/sdks/python/apache_beam/examples/secret_manager/aws_credentials_secret_read.py @@ -0,0 +1,73 @@ +# +# 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. +# + +"""Reads a secret from AWS Secrets Manager outside of a pipeline, using AWS +credentials stored in Google Cloud Secret Manager. + +Usage: + python -m apache_beam.examples.secret_manager.aws_credentials_secret_read \ + --secret_id=my-aws-secret \ + --aws_region_name=us-east-1 \ + --aws_credentials_secret_manager=GoogleCloudSecretManager \ + --aws_access_key_id='{"name": "my-aws-key-id", "project": "my-project"}' \ + --aws_secret_access_key='{"name": "my-aws-secret-key", "project": "my-project"}' + +Required permissions: + * The AWS IAM user that owns the access key needs + ``secretsmanager:GetSecretValue`` on the secret (e.g. + ``arn:aws:secretsmanager:us-east-1::secret:my-aws-secret-*``), + for example through the ``AWSSecretsManagerClientReadOnlyAccess`` managed + policy or a narrower custom policy. If the secret is encrypted with a + customer managed KMS key instead of the default ``aws/secretsmanager`` key, + it also needs ``kms:Decrypt`` on that key. + * The Google Cloud account running this script (e.g. your Application + Default Credentials) needs the Secret Manager Secret Accessor role + (``roles/secretmanager.secretAccessor``) on the two secrets that hold the + access key ID and secret access key. +""" + +import argparse + +from apache_beam.options.pipeline_options import PipelineOptions +from apache_beam.utils.secret import Secret + + +def main(argv=None): + parser = argparse.ArgumentParser() + parser.add_argument( + '--secret_id', + required=True, + help='Name or ARN of the secret in AWS Secrets Manager.') + known_args, pipeline_args = parser.parse_known_args(argv) + options = PipelineOptions(pipeline_args) + + # The AWS Secrets Manager client uses the --aws_* options, with credentials + # resolved through --aws_credentials_secret_manager. + + # Option 1: a 'type:;:' secret string. + secret = Secret.parse_secret_option( + f'type:awssecret;name:{known_args.secret_id}', options) + print(secret.get_str()) + + # Option 2: a JSON spec and a secret manager name. + secret = Secret.from_json( + f'{{"name": "{known_args.secret_id}"}}', 'AwsSecretsManager', options) + print(secret.get_str()) + + +if __name__ == '__main__': + main() From 15d9d5bd651016a23a06985e568db70eff0696f5 Mon Sep 17 00:00:00 2001 From: Shunping Huang Date: Thu, 15 Oct 2026 15:10:25 -0400 Subject: [PATCH 2/4] Add a use case of arn for reading AWS secrets --- .../aws_credentials_secret_read.py | 21 ++++++++++++++++--- 1 file changed, 18 insertions(+), 3 deletions(-) diff --git a/sdks/python/apache_beam/examples/secret_manager/aws_credentials_secret_read.py b/sdks/python/apache_beam/examples/secret_manager/aws_credentials_secret_read.py index e702fa121d2b..b69942e1ded8 100644 --- a/sdks/python/apache_beam/examples/secret_manager/aws_credentials_secret_read.py +++ b/sdks/python/apache_beam/examples/secret_manager/aws_credentials_secret_read.py @@ -21,6 +21,7 @@ Usage: python -m apache_beam.examples.secret_manager.aws_credentials_secret_read \ --secret_id=my-aws-secret \ + --secret_arn=arn:aws:secretsmanager:us-east-1::secret:my-aws-secret-AbCdEf \ --aws_region_name=us-east-1 \ --aws_credentials_secret_manager=GoogleCloudSecretManager \ --aws_access_key_id='{"name": "my-aws-key-id", "project": "my-project"}' \ @@ -51,7 +52,10 @@ def main(argv=None): parser.add_argument( '--secret_id', required=True, - help='Name or ARN of the secret in AWS Secrets Manager.') + help='Name of the secret in AWS Secrets Manager.') + parser.add_argument( + '--secret_arn', + help='Optional ARN of the secret in AWS Secrets Manager.') known_args, pipeline_args = parser.parse_known_args(argv) options = PipelineOptions(pipeline_args) @@ -61,12 +65,23 @@ def main(argv=None): # Option 1: a 'type:;:' secret string. secret = Secret.parse_secret_option( f'type:awssecret;name:{known_args.secret_id}', options) - print(secret.get_str()) + print(f'Retrieved secret of length {len(secret.get_str())}') # Option 2: a JSON spec and a secret manager name. secret = Secret.from_json( f'{{"name": "{known_args.secret_id}"}}', 'AwsSecretsManager', options) - print(secret.get_str()) + print(f'Retrieved secret of length {len(secret.get_str())}') + + # Options 3 and 4: the same as options 1 and 2, but identifying the secret by + # its ARN. + if known_args.secret_arn: + secret = Secret.parse_secret_option( + f'type:awssecret;arn:{known_args.secret_arn}', options) + print(f'Retrieved secret of length {len(secret.get_str())}') + + secret = Secret.from_json( + f'{{"arn": "{known_args.secret_arn}"}}', 'AwsSecretsManager', options) + print(f'Retrieved secret of length {len(secret.get_str())}') if __name__ == '__main__': From 22e408da009aedcc13e3bf630e3b8eba54c4bf31 Mon Sep 17 00:00:00 2001 From: Shunping Huang Date: Thu, 15 Oct 2026 18:01:30 -0400 Subject: [PATCH 3/4] Reformat --- .../examples/secret_manager/aws_credentials_secret_read.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/sdks/python/apache_beam/examples/secret_manager/aws_credentials_secret_read.py b/sdks/python/apache_beam/examples/secret_manager/aws_credentials_secret_read.py index b69942e1ded8..bf3dadedeb2d 100644 --- a/sdks/python/apache_beam/examples/secret_manager/aws_credentials_secret_read.py +++ b/sdks/python/apache_beam/examples/secret_manager/aws_credentials_secret_read.py @@ -54,8 +54,7 @@ def main(argv=None): required=True, help='Name of the secret in AWS Secrets Manager.') parser.add_argument( - '--secret_arn', - help='Optional ARN of the secret in AWS Secrets Manager.') + '--secret_arn', help='Optional ARN of the secret in AWS Secrets Manager.') known_args, pipeline_args = parser.parse_known_args(argv) options = PipelineOptions(pipeline_args) From 16a8777d7488d0c11ae5f37b68872469c25b8379 Mon Sep 17 00:00:00 2001 From: Shunping Huang Date: Fri, 16 Oct 2026 14:32:19 -0400 Subject: [PATCH 4/4] Added a requirements file for the example. --- .../aws_credentials_requirements.txt | 21 +++++++++++++++++++ .../aws_credentials_s3_pipeline.py | 7 +++---- 2 files changed, 24 insertions(+), 4 deletions(-) create mode 100644 sdks/python/apache_beam/examples/secret_manager/aws_credentials_requirements.txt diff --git a/sdks/python/apache_beam/examples/secret_manager/aws_credentials_requirements.txt b/sdks/python/apache_beam/examples/secret_manager/aws_credentials_requirements.txt new file mode 100644 index 000000000000..e0db2ef261db --- /dev/null +++ b/sdks/python/apache_beam/examples/secret_manager/aws_credentials_requirements.txt @@ -0,0 +1,21 @@ +# +# 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. +# + +# Worker dependencies for aws_credentials_s3_pipeline.py: boto3 for the S3 +# filesystem, and google-cloud-secret-manager to resolve the AWS credentials. +boto3>=1.9,<2 +google-cloud-secret-manager>=2.0,<3 diff --git a/sdks/python/apache_beam/examples/secret_manager/aws_credentials_s3_pipeline.py b/sdks/python/apache_beam/examples/secret_manager/aws_credentials_s3_pipeline.py index 704ea45254ad..535444653499 100644 --- a/sdks/python/apache_beam/examples/secret_manager/aws_credentials_s3_pipeline.py +++ b/sdks/python/apache_beam/examples/secret_manager/aws_credentials_s3_pipeline.py @@ -31,11 +31,10 @@ --project=my-project \ --region=us-central1 \ --temp_location=gs://my-gcs-bucket/temp \ - --requirements_file=requirements.txt + --requirements_file=apache_beam/examples/secret_manager/aws_credentials_requirements.txt -where requirements.txt contains: - boto3 - google-cloud-secret-manager +where aws_credentials_requirements.txt (in the same directory as this example) +lists the packages the workers need: boto3 and google-cloud-secret-manager. Required permissions: * The AWS IAM user that owns the access key needs ``s3:ListBucket`` on the