claudevdm commented on code in PR #40547: URL: https://github.com/apache/beam/pull/40547#discussion_r4261885303
########## sdks/python/apache_beam/io/aws/clients/client_options.py: ########## @@ -0,0 +1,176 @@ +# +# 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, 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: Review Comment: Maybe also return early if secret manager is not set here? ########## sdks/python/apache_beam/io/aws/clients/client_options.py: ########## @@ -0,0 +1,176 @@ +# +# 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, 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 Review Comment: Is value expected to be a string here? What if it is passed as a dictionary? Maybe check if it is dictionary and convert to a str? ########## sdks/python/apache_beam/io/aws/clients/s3/boto3_client_test.py: ########## @@ -111,8 +115,123 @@ 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): Review Comment: Should this test be skipped if secretmanager is not installed similar to secret_test.py? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
