fix: propagate access denied errors

aws_lambda_powertools/utilities/feature_flags/appconfig.py

@@ -1,4 +1,5 @@
 import logging
+import traceback
 from typing import Any, Dict, Optional, cast
 
 from botocore.config import Config
@@ -7,7 +8,7 @@
 
 from ...shared import jmespath_utils
 from .base import StoreProvider
-from .exceptions import ConfigurationStoreError
+from .exceptions import ConfigurationStoreError, StoreClientError
 
 logger = logging.getLogger(__name__)
 
@@ -85,4 +86,7 @@ def get_configuration(self) -> Dict[str, Any]:
 
             return config
         except (GetParameterError, TransformParameterError) as exc:
+            err_msg = traceback.format_exc()
+            if "AccessDenied" in err_msg:
+                raise StoreClientError(err_msg) from exc
             raise ConfigurationStoreError("Unable to get AWS AppConfig configuration file") from exc

aws_lambda_powertools/utilities/feature_flags/exceptions.py

@@ -4,3 +4,10 @@ class ConfigurationStoreError(Exception):
 
 class SchemaValidationError(Exception):
     """When feature flag schema fails validation"""
+
+
+class StoreClientError(Exception):
+    """When a store raises an exception that should be propagated to the client to fix
+
+    For example, Access Denied errors when the client doesn't permissions to fetch config
+    """

aws_lambda_powertools/utilities/feature_flags/feature_flags.py

@@ -140,11 +140,7 @@ def get_configuration(self) -> Union[Dict[str, Dict], Dict]:
         """
         # parse result conf as JSON, keep in cache for max age defined in store
         logger.debug(f"Fetching schema from registered store, store={self._store}")
-        try:
-            config = self._store.get_configuration()
-        except Exception as err:
-            raise ConfigurationStoreError("Unable to fetch schema from registered store") from err
-
+        config = self._store.get_configuration()
         validator = schema.SchemaValidator(schema=config)
         validator.validate()
 

tests/functional/feature_flags/test_feature_flags.py

@@ -5,6 +5,7 @@
 
 from aws_lambda_powertools.utilities.feature_flags import ConfigurationStoreError, schema
 from aws_lambda_powertools.utilities.feature_flags.appconfig import AppConfigStore
+from aws_lambda_powertools.utilities.feature_flags.exceptions import StoreClientError
 from aws_lambda_powertools.utilities.feature_flags.feature_flags import FeatureFlags
 from aws_lambda_powertools.utilities.feature_flags.schema import RuleAction
 from aws_lambda_powertools.utilities.parameters import GetParameterError
@@ -487,3 +488,16 @@ def test_match_condition_with_dict_value(mocker, config):
     ctx = {"tenant": {"tenant_id": "6", "username": "lessa"}}
     toggle = feature_flags.evaluate(name="my_feature", context=ctx, default=False)
     assert toggle == expected_value
+
+
+def test_get_feature_toggle_propagates_access_denied_error(mocker, config):
+    # GIVEN a schema fetch that raises a StoreClientError
+    # due to client invalid permissions to fetch from the store
+    err = "An error occurred (AccessDeniedException) when calling the GetConfiguration operation"
+    schema_fetcher = init_fetcher_side_effect(mocker, config, GetParameterError(err))
+    feature_flags = FeatureFlags(schema_fetcher)
+
+    # WHEN calling evaluate
+    # THEN raise StoreClientError error
+    with pytest.raises(StoreClientError, match="AccessDeniedException") as err:
+        feature_flags.evaluate(name="Foo", default=False)