fix: filtering client suppliers MUST check the client region after creation as well as before

src/aws_encryption_sdk/keyrings/aws_kms/client_suppliers.py

@@ -114,21 +114,29 @@ class AllowRegionsClientSupplier(ClientSupplier):
     """
 
     allowed_regions = attr.ib(
-        validator=(deep_iterable(member_validator=instance_of(six.string_types)), value_is_not_a_string)
+        validator=(deep_iterable(member_validator=instance_of((type(None), six.string_types))), value_is_not_a_string)
     )
     _client_supplier = attr.ib(default=attr.Factory(DefaultClientSupplier), validator=optional(is_callable()))
 
+    def _check(self, region_name):
+        # type: (Union[None, str]) -> None
+        if region_name not in self.allowed_regions:
+            raise UnknownRegionError("Unable to provide client for region '{}'".format(region_name))
+
     def __call__(self, region_name):
         # type: (Union[None, str]) -> BaseClient
         """Return a client for the requested region.
 
         :rtype: BaseClient
         :raises UnknownRegionError: if a region is requested that is not in ``allowed_regions``
         """
-        if region_name not in self.allowed_regions:
-            raise UnknownRegionError("Unable to provide client for region '{}'".format(region_name))
+        self._check(region_name=region_name)
+
+        client = self._client_supplier(region_name)
+
+        self._check(region_name=client.meta.region_name)
 
-        return self._client_supplier(region_name)
+        return client
 
 
 @attr.s
@@ -142,18 +150,26 @@ class DenyRegionsClientSupplier(ClientSupplier):
     """
 
     denied_regions = attr.ib(
-        validator=(deep_iterable(member_validator=instance_of(six.string_types)), value_is_not_a_string)
+        validator=(deep_iterable(member_validator=instance_of((type(None), six.string_types))), value_is_not_a_string)
     )
     _client_supplier = attr.ib(default=attr.Factory(DefaultClientSupplier), validator=optional(is_callable()))
 
+    def _check(self, region_name):
+        # type: (Union[None, str]) -> None
+        if region_name in self.denied_regions:
+            raise UnknownRegionError("Unable to provide client for region '{}'".format(region_name))
+
     def __call__(self, region_name):
         # type: (Union[None, str]) -> BaseClient
         """Return a client for the requested region.
 
         :rtype: BaseClient
         :raises UnknownRegionError: if a region is requested that is in ``denied_regions``
         """
-        if region_name in self.denied_regions:
-            raise UnknownRegionError("Unable to provide client for region '{}'".format(region_name))
+        self._check(region_name=region_name)
+
+        client = self._client_supplier(region_name)
+
+        self._check(region_name=client.meta.region_name)
 
-        return self._client_supplier(region_name)
+        return client

test/functional/keyrings/aws_kms/test_client_suppliers.py

@@ -2,6 +2,7 @@
 # SPDX-License-Identifier: Apache-2.0
 """Functional tests for ``aws_encryption_sdk.keyrings.aws_kms.client_suppliers``."""
 import pytest
+from botocore.client import BaseClient
 from botocore.config import Config
 from botocore.session import Session
 
@@ -13,9 +14,28 @@
     DenyRegionsClientSupplier,
 )
 
+try:  # Python 3.5.0 and 3.5.1 have incompatible typing modules
+    from typing import Callable, Union  # noqa pylint: disable=unused-import
+except ImportError:  # pragma: no cover
+    # We only actually need these imports when running the mypy checks
+    pass
+
 pytestmark = [pytest.mark.functional, pytest.mark.local]
 
 
+class AlwaysOneRegionClientSupplier(ClientSupplier):
+    """Client supplier that always returns a client for a specific region."""
+
+    def __init__(self, region_name):
+        # type: (str) -> None
+        self._region_name = region_name
+        self._inner_supplier = DefaultClientSupplier()
+
+    def __call__(self, region_name):
+        # type: (Union[None, str]) -> BaseClient
+        return self._inner_supplier(self._region_name)
+
+
 def test_default_supplier_not_implemented():
     test = ClientSupplier()
 
@@ -114,3 +134,48 @@ def test_allow_deny_nested_supplier():
         test_deny("us-west-2")
 
     excinfo.match("Unable to provide client for region 'us-west-2'")
+
+
+def test_deny_region_allowed_region_resolved_by_client_supplier():
+    region = "us-west-2"
+    supplier = DenyRegionsClientSupplier(
+        denied_regions=["us-east-1"], client_supplier=AlwaysOneRegionClientSupplier(region_name=region)
+    )
+
+    # Resolves to a region that is not blocked.
+    test = supplier(None)
+
+    assert test.meta.region_name == region
+
+
+def test_allow_region_allowed_region_resolved_by_client_supplier():
+    region = "us-west-2"
+    supplier = AllowRegionsClientSupplier(
+        allowed_regions=[None, region], client_supplier=AlwaysOneRegionClientSupplier(region_name=region)
+    )
+
+    # Resolves to a region that is allowed
+    test = supplier(None)
+
+    assert test.meta.region_name == region
+
+
+def test_deny_region_block_on_response():
+    region = "us-west-2"
+    test = DenyRegionsClientSupplier(
+        denied_regions=[region], client_supplier=AlwaysOneRegionClientSupplier(region_name=region)
+    )
+
+    # Catch the blocked region on response.
+    with pytest.raises(UnknownRegionError):
+        test(None)
+
+
+def test_allow_region_block_on_response():
+    test = AllowRegionsClientSupplier(
+        allowed_regions=[None, "us-west-2"], client_supplier=AlwaysOneRegionClientSupplier(region_name="us-east-2")
+    )
+
+    # Catch the blocked region on response.
+    with pytest.raises(UnknownRegionError):
+        test(None)