fix: validating undefined connection strings

azurefunctions-extensions-bindings-blob/azurefunctions/extensions/bindings/blob/blobClient.py

@@ -11,7 +11,6 @@
 
 class BlobClient(SdkType):
     def __init__(self, *, data: Union[bytes, Datum]) -> None:
-
         # model_binding_data properties
         self._data = data
         self._version = None
@@ -25,7 +24,7 @@ def __init__(self, *, data: Union[bytes, Datum]) -> None:
             self._source = data.source
             self._content_type = data.content_type
             content_json = json.loads(data.content)
-            self._connection = os.getenv(content_json["Connection"])
+            self._connection = validate_connection_string(content_json["Connection"])
             self._containerName = content_json["ContainerName"]
             self._blobName = content_json["BlobName"]
 
@@ -39,3 +38,16 @@ def get_sdk_type(self):
             )
         else:
             return None
+
+
+def validate_connection_string(connection_string: str) -> str:
+    """
+    Validates the connection string. If the connection string is
+    not an App Setting, an error will be thrown.
+    """
+    if not os.getenv(connection_string):
+        raise ValueError(
+            f"Storage account connection string {connection_string} does not exist. "
+            f"Please make sure that it is a defined App Setting."
+        )
+    return os.getenv(connection_string)

azurefunctions-extensions-bindings-blob/azurefunctions/extensions/bindings/blob/blobClientConverter.py

@@ -16,7 +16,6 @@ class BlobClientConverter(
     binding="blob",
     trigger="blobTrigger",
 ):
-
     @classmethod
     def check_input_type_annotation(cls, pytype: type) -> bool:
         return issubclass(

azurefunctions-extensions-bindings-blob/azurefunctions/extensions/bindings/blob/containerClient.py

@@ -11,7 +11,6 @@
 
 class ContainerClient(SdkType):
     def __init__(self, *, data: Union[bytes, Datum]) -> None:
-
         # model_binding_data properties
         self._data = data
         self._version = ""
@@ -25,7 +24,7 @@ def __init__(self, *, data: Union[bytes, Datum]) -> None:
             self._source = data.source
             self._content_type = data.content_type
             content_json = json.loads(data.content)
-            self._connection = os.getenv(content_json["Connection"])
+            self._connection = validate_connection_string(content_json["Connection"])
             self._containerName = content_json["ContainerName"]
             self._blobName = content_json["BlobName"]
 
@@ -37,3 +36,16 @@ def get_sdk_type(self):
             )
         else:
             return None
+
+
+def validate_connection_string(connection_string: str) -> str:
+    """
+    Validates the connection string. If the connection string is
+    not an App Setting, an error will be thrown.
+    """
+    if not os.getenv(connection_string):
+        raise ValueError(
+            f"Storage account connection string {connection_string} does not exist. "
+            f"Please make sure that it is a defined App Setting."
+        )
+    return os.getenv(connection_string)

azurefunctions-extensions-bindings-blob/azurefunctions/extensions/bindings/blob/storageStreamDownloader.py

@@ -11,7 +11,6 @@
 
 class StorageStreamDownloader(SdkType):
     def __init__(self, *, data: Union[bytes, Datum]) -> None:
-
         # model_binding_data properties
         self._data = data or {}
         self._version = ""
@@ -25,7 +24,7 @@ def __init__(self, *, data: Union[bytes, Datum]) -> None:
             self._source = data.source
             self._content_type = data.content_type
             content_json = json.loads(data.content)
-            self._connection = os.getenv(content_json["Connection"])
+            self._connection = validate_connection_string(content_json["Connection"])
             self._containerName = content_json["ContainerName"]
             self._blobName = content_json["BlobName"]
 
@@ -42,3 +41,16 @@ def get_sdk_type(self):
             return blob_client.download_blob()
         else:
             return None
+
+
+def validate_connection_string(connection_string: str) -> str:
+    """
+    Validates the connection string. If the connection string is
+    not an App Setting, an error will be thrown.
+    """
+    if not os.getenv(connection_string):
+        raise ValueError(
+            f"Storage account connection string {connection_string} does not exist. "
+            f"Please make sure that it is a defined App Setting."
+        )
+    return os.getenv(connection_string)

azurefunctions-extensions-bindings-blob/tests/test_blobclient.py

@@ -126,6 +126,31 @@ def test_input_populated(self):
         self.assertIsNotNone(sdk_result)
         self.assertIsInstance(sdk_result, BlobClientSdk)
 
+    def test_invalid_input_populated(self):
+        content = {
+            "Connection": "NotARealConnectionString",
+            "ContainerName": "test-blob",
+            "BlobName": "text.txt",
+        }
+
+        sample_mbd = MockMBD(
+            version="1.0",
+            source="AzureStorageBlobs",
+            content_type="application/json",
+            content=json.dumps(content),
+        )
+
+        with self.assertRaises(ValueError) as e:
+            datum: Datum = Datum(value=sample_mbd, type="model_binding_data")
+            result: BlobClient = BlobClientConverter.decode(
+                data=datum, trigger_metadata=None, pytype=BlobClient
+            )
+        self.assertEqual(
+            e.exception.args[0],
+            "Storage account connection string NotARealConnectionString does not exist. "
+            "Please make sure that it is a defined App Setting.",
+        )
+
     def test_input_invalid_pytype(self):
         content = {
             "Connection": "AzureWebJobsStorage",

azurefunctions-extensions-bindings-blob/tests/test_containerclient.py

@@ -124,6 +124,31 @@ def test_input_populated(self):
         self.assertIsNotNone(sdk_result)
         self.assertIsInstance(sdk_result, ContainerClientSdk)
 
+    def test_invalid_input_populated(self):
+        content = {
+            "Connection": "NotARealConnectionString",
+            "ContainerName": "test-blob",
+            "BlobName": "text.txt",
+        }
+
+        sample_mbd = MockMBD(
+            version="1.0",
+            source="AzureStorageBlobs",
+            content_type="application/json",
+            content=json.dumps(content),
+        )
+
+        with self.assertRaises(ValueError) as e:
+            datum: Datum = Datum(value=sample_mbd, type="model_binding_data")
+            result: ContainerClient = BlobClientConverter.decode(
+                data=datum, trigger_metadata=None, pytype=ContainerClient
+            )
+        self.assertEqual(
+            e.exception.args[0],
+            "Storage account connection string NotARealConnectionString does not exist. "
+            "Please make sure that it is a defined App Setting.",
+        )
+
     def test_input_invalid_pytype(self):
         content = {
             "Connection": "AzureWebJobsStorage",

azurefunctions-extensions-bindings-blob/tests/test_ssd.py

@@ -129,6 +129,31 @@ def test_input_populated(self):
         self.assertIsNotNone(sdk_result)
         self.assertIsInstance(sdk_result, SSDSdk)
 
+    def test_invalid_input_populated(self):
+        content = {
+            "Connection": "NotARealConnectionString",
+            "ContainerName": "test-blob",
+            "BlobName": "text.txt",
+        }
+
+        sample_mbd = MockMBD(
+            version="1.0",
+            source="AzureStorageBlobs",
+            content_type="application/json",
+            content=json.dumps(content),
+        )
+
+        with self.assertRaises(ValueError) as e:
+            datum: Datum = Datum(value=sample_mbd, type="model_binding_data")
+            result: StorageStreamDownloader = BlobClientConverter.decode(
+                data=datum, trigger_metadata=None, pytype=StorageStreamDownloader
+            )
+        self.assertEqual(
+            e.exception.args[0],
+            "Storage account connection string NotARealConnectionString does not exist. "
+            "Please make sure that it is a defined App Setting.",
+        )
+
     def test_input_invalid_pytype(self):
         content = {
             "Connection": "AzureWebJobsStorage",