Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 18 additions & 0 deletions sdks/python/apache_beam/examples/secret_manager/__init__.py
Original file line number Diff line number Diff line change
@@ -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."""
Original file line number Diff line number Diff line change
@@ -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
Original file line number Diff line number Diff line change
@@ -0,0 +1,75 @@
#
# 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=apache_beam/examples/secret_manager/aws_credentials_requirements.txt

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
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()
Original file line number Diff line number Diff line change
@@ -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()
Original file line number Diff line number Diff line change
@@ -0,0 +1,87 @@
#
# 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 \
--secret_arn=arn:aws:secretsmanager:us-east-1:<account-id>: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"}' \
--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:<account-id>: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 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)

# The AWS Secrets Manager client uses the --aws_* options, with credentials
# resolved through --aws_credentials_secret_manager.

# Option 1: a 'type:<secret_type>;<param>:<value>' secret string.
secret = Secret.parse_secret_option(
f'type:awssecret;name:{known_args.secret_id}', options)
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(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__':
main()
Loading