inherit key "verify" from env into session so that kv_client can read…

… it properly (#38614)

airflow/providers/hashicorp/_internal_client/vault_client.py

@@ -212,6 +212,8 @@ def _client(self) -> hvac.Client:
             session = Session()
             session.mount("http://", adapter)
             session.mount("https://", adapter)
+            if self.kwargs and "verify" in self.kwargs:
+                session.verify = self.kwargs["verify"]
             self.kwargs["session"] = session
 
         _client = hvac.Client(url=self.url, **self.kwargs)

tests/providers/hashicorp/_internal_client/test_vault_client.py

@@ -837,6 +837,70 @@ def test_get_existing_key_v1(self, mock_hvac):
             mount_point="secret", path="/path/to/secret"
         )
 
+    @mock.patch("airflow.providers.hashicorp._internal_client.vault_client.hvac")
+    def test_get_existing_key_v1_ssl_verify_false(self, mock_hvac):
+        mock_client = mock.MagicMock()
+        mock_hvac.Client.return_value = mock_client
+
+        mock_client.secrets.kv.v1.read_secret.return_value = {
+            "request_id": "182d0673-618c-9889-4cba-4e1f4cfe4b4b",
+            "lease_id": "",
+            "renewable": False,
+            "lease_duration": 2764800,
+            "data": {"value": "world"},
+            "wrap_info": None,
+            "warnings": None,
+            "auth": None,
+        }
+
+        vault_client = _VaultClient(
+            auth_type="radius",
+            radius_host="radhost",
+            radius_port=8110,
+            radius_secret="pass",
+            kv_engine_version=1,
+            url="http://localhost:8180",
+            verify=False,
+        )
+        secret = vault_client.get_secret(secret_path="/path/to/secret")
+        assert {"value": "world"} == secret
+        assert not vault_client.kwargs["session"].verify
+        mock_client.secrets.kv.v1.read_secret.assert_called_once_with(
+            mount_point="secret", path="/path/to/secret"
+        )
+
+    @mock.patch("airflow.providers.hashicorp._internal_client.vault_client.hvac")
+    def test_get_existing_key_v1_trust_private_ca(self, mock_hvac):
+        mock_client = mock.MagicMock()
+        mock_hvac.Client.return_value = mock_client
+
+        mock_client.secrets.kv.v1.read_secret.return_value = {
+            "request_id": "182d0673-618c-9889-4cba-4e1f4cfe4b4b",
+            "lease_id": "",
+            "renewable": False,
+            "lease_duration": 2764800,
+            "data": {"value": "world"},
+            "wrap_info": None,
+            "warnings": None,
+            "auth": None,
+        }
+
+        vault_client = _VaultClient(
+            auth_type="radius",
+            radius_host="radhost",
+            radius_port=8110,
+            radius_secret="pass",
+            kv_engine_version=1,
+            url="http://localhost:8180",
+            verify="/etc/ssl/certificates/ca-bundle.pem",
+        )
+        secret = vault_client.get_secret(secret_path="/path/to/secret")
+        assert {"value": "world"} == secret
+        assert "/etc/ssl/certificates/ca-bundle.pem" == vault_client.kwargs["session"].verify
+        mock_client.secrets.kv.v1.read_secret.assert_called_once_with(
+            mount_point="secret", path="/path/to/secret"
+        )
+
     @mock.patch("airflow.providers.hashicorp._internal_client.vault_client.hvac")
     def test_get_existing_key_v1_without_preconfigured_mount_point(self, mock_hvac):
         mock_client = mock.MagicMock()