Add Flexible Checksum v2 Config Options (#3271)

Add ``request_checksum_calculation`` and ``response_checksum_validation`` config options.

botocore/args.py

@@ -61,6 +61,15 @@
 # values result in a warning-level log message.
 USERAGENT_APPID_MAXLEN = 50
 
+VALID_REQUEST_CHECKSUM_CALCULATION_CONFIG = (
+    "when_supported",
+    "when_required",
+)
+VALID_RESPONSE_CHECKSUM_VALIDATION_CONFIG = (
+    "when_supported",
+    "when_required",
+)
+
 
 class ClientArgsCreator:
     def __init__(
@@ -271,11 +280,18 @@ def compute_client_args(
                 sigv4a_signing_region_set=(
                     client_config.sigv4a_signing_region_set
                 ),
+                request_checksum_calculation=(
+                    client_config.request_checksum_calculation
+                ),
+                response_checksum_validation=(
+                    client_config.response_checksum_validation
+                ),
             )
         self._compute_retry_config(config_kwargs)
         self._compute_connect_timeout(config_kwargs)
         self._compute_user_agent_appid_config(config_kwargs)
         self._compute_request_compression_config(config_kwargs)
+        self._compute_checksum_config(config_kwargs)
         s3_config = self.compute_s3_config(client_config)
 
         is_s3_service = self._is_s3_service(service_name)
@@ -771,3 +787,39 @@ def _compute_user_agent_appid_config(self, config_kwargs):
                 f'maximum length of {USERAGENT_APPID_MAXLEN} characters.'
             )
         config_kwargs['user_agent_appid'] = user_agent_appid
+
+    def _compute_checksum_config(self, config_kwargs):
+        self._handle_checksum_config(
+            config_kwargs,
+            config_key="request_checksum_calculation",
+            default_value="when_supported",
+            valid_options=VALID_REQUEST_CHECKSUM_CALCULATION_CONFIG,
+        )
+        self._handle_checksum_config(
+            config_kwargs,
+            config_key="response_checksum_validation",
+            default_value="when_supported",
+            valid_options=VALID_RESPONSE_CHECKSUM_VALIDATION_CONFIG,
+        )
+
+    def _handle_checksum_config(
+        self,
+        config_kwargs,
+        config_key,
+        default_value,
+        valid_options,
+    ):
+        value = config_kwargs.get(config_key)
+        if value is None:
+            value = (
+                self._config_store.get_config_variable(config_key)
+                or default_value
+            )
+        value = value.lower()
+        if value not in valid_options:
+            raise botocore.exceptions.InvalidChecksumConfigError(
+                config_key=config_key,
+                config_value=value,
+                valid_options=valid_options,
+            )
+        config_kwargs[config_key] = value

botocore/config.py

@@ -235,6 +235,37 @@ class Config:
         specified service will be ignored.
 
         Defaults to None.
+
+    :type request_checksum_calculation: str
+    :param request_checksum_calculation: Determines when a checksum will be
+        calculated for request payloads. Valid values are:
+
+        * ``when_supported`` -- When set, a checksum will be calculated for
+            all request payloads of operations modeled with the ``httpChecksum``
+            trait where ``requestChecksumRequired`` is ``true`` and/or a
+            ``requestAlgorithmMember`` is modeled.
+
+        * ``when_required`` -- When set, a checksum will only be calculated
+            for request payloads of operations modeled with the ``httpChecksum``
+            trait where ``requestChecksumRequired`` is ``true`` or where a
+            ``requestAlgorithmMember`` is modeled and supplied.
+
+        Defaults to None.
+
+    :type response_checksum_validation: str
+    :param response_checksum_validation: Determines when checksum validation
+        will be performed on response payloads. Valid values are:
+
+        * ``when_supported`` -- When set, checksum validation is performed on
+            all response payloads of operations modeled with the ``httpChecksum``
+            trait where ``responseAlgorithms`` is modeled, except when no modeled
+            checksum algorithms are supported.
+
+        * ``when_required`` -- When set, checksum validation is not performed
+            on response payloads of operations unless the checksum algorithm is
+            supported and the ``requestValidationModeMember`` member is set to ``ENABLED``.
+
+        Defaults to None.
     """
 
     OPTION_DEFAULTS = OrderedDict(
@@ -264,6 +295,8 @@ class Config:
             ('disable_request_compression', None),
             ('client_context_params', None),
             ('sigv4a_signing_region_set', None),
+            ('request_checksum_calculation', None),
+            ('response_checksum_validation', None),
         ]
     )
 

botocore/configprovider.py

@@ -168,6 +168,18 @@
         None,
         None,
     ),
+    'request_checksum_calculation': (
+        'request_checksum_calculation',
+        'AWS_REQUEST_CHECKSUM_CALCULATION',
+        "when_supported",
+        None,
+    ),
+    'response_checksum_validation': (
+        'response_checksum_validation',
+        'AWS_RESPONSE_CHECKSUM_VALIDATION',
+        "when_supported",
+        None,
+    ),
 }
 
 # Evaluate AWS_STS_REGIONAL_ENDPOINTS settings
@@ -468,7 +480,7 @@ def __copy__(self):
 
     def get_config_variable(self, logical_name):
         """
-        Retrieve the value associeated with the specified logical_name
+        Retrieve the value associated with the specified logical_name
         from the corresponding provider. If no value is found None will
         be returned.
 

botocore/exceptions.py

@@ -814,3 +814,12 @@ class EndpointResolutionError(EndpointProviderError):
 
 class UnknownEndpointResolutionBuiltInName(EndpointProviderError):
     fmt = 'Unknown builtin variable name: {name}'
+
+
+class InvalidChecksumConfigError(BotoCoreError):
+    """Error when invalid value supplied for checksum config"""
+
+    fmt = (
+        'Unsupported configuration value for {config_key}. '
+        'Expected one of {valid_options} but got {config_value}.'
+    )

tests/unit/test_args.py

@@ -614,6 +614,70 @@ def test_bad_value_disable_request_compression(self):
         config = client_args['client_config']
         self.assertFalse(config.disable_request_compression)
 
+    def test_checksum_default_client_config(self):
+        input_config = Config()
+        client_args = self.call_get_client_args(client_config=input_config)
+        config = client_args["client_config"]
+        self.assertEqual(config.request_checksum_calculation, "when_supported")
+        self.assertEqual(config.response_checksum_validation, "when_supported")
+
+    def test_checksum_client_config(self):
+        input_config = Config(
+            request_checksum_calculation="when_required",
+            response_checksum_validation="when_required",
+        )
+        client_args = self.call_get_client_args(client_config=input_config)
+        config = client_args['client_config']
+        self.assertEqual(config.request_checksum_calculation, "when_required")
+        self.assertEqual(config.response_checksum_validation, "when_required")
+
+    def test_checksum_config_store(self):
+        self.config_store.set_config_variable(
+            "request_checksum_calculation", "when_required"
+        )
+        self.config_store.set_config_variable(
+            "response_checksum_validation", "when_required"
+        )
+        config = self.call_get_client_args()['client_config']
+        self.assertEqual(config.request_checksum_calculation, "when_required")
+        self.assertEqual(config.response_checksum_validation, "when_required")
+
+    def test_checksum_client_config_overrides_config_store(self):
+        self.config_store.set_config_variable(
+            "request_checksum_calculation", "when_supported"
+        )
+        self.config_store.set_config_variable(
+            "response_checksum_validation", "when_supported"
+        )
+        input_config = Config(
+            request_checksum_calculation="when_required",
+            response_checksum_validation="when_required",
+        )
+        client_args = self.call_get_client_args(client_config=input_config)
+        config = client_args['client_config']
+        self.assertEqual(config.request_checksum_calculation, "when_required")
+        self.assertEqual(config.response_checksum_validation, "when_required")
+
+    def test_request_checksum_calculation_invalid_client_config(self):
+        with self.assertRaises(exceptions.InvalidChecksumConfigError):
+            config = Config(request_checksum_calculation="invalid_config")
+            self.call_get_client_args(client_config=config)
+        self.config_store.set_config_variable(
+            'request_checksum_calculation', "invalid_config"
+        )
+        with self.assertRaises(exceptions.InvalidChecksumConfigError):
+            self.call_get_client_args()
+
+    def test_response_checksum_validation_invalid_client_config(self):
+        with self.assertRaises(exceptions.InvalidChecksumConfigError):
+            config = Config(response_checksum_validation="invalid_config")
+            self.call_get_client_args(client_config=config)
+        self.config_store.set_config_variable(
+            'response_checksum_validation', "invalid_config"
+        )
+        with self.assertRaises(exceptions.InvalidChecksumConfigError):
+            self.call_get_client_args()
+
 
 class TestEndpointResolverBuiltins(unittest.TestCase):
     def setUp(self):