Support cert and proxies in Vault client session setup

providers/hashicorp/src/airflow/providers/hashicorp/_internal_client/vault_client.py

@@ -215,8 +215,10 @@ 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"]
+
+            session.verify = self.kwargs.get("verify", session.verify)
+            session.cert = self.kwargs.get("cert", session.cert)
+            session.proxies = self.kwargs.get("proxies", session.proxies)
             self.kwargs["session"] = session
 
         _client = hvac.Client(url=self.url, **self.kwargs)

providers/hashicorp/tests/unit/hashicorp/_internal_client/test_vault_client.py

@@ -925,6 +925,76 @@ def test_get_existing_key_v1_trust_private_ca(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_with_proxies_applied(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,
+            proxies={
+                "http": "http://10.10.1.10:3128",
+                "https": "http://10.10.1.10:1080",
+            },
+        )
+        secret = vault_client.get_secret(secret_path="/path/to/secret")
+        assert secret == {"value": "world"}
+        assert vault_client.kwargs["session"].proxies["http"] == "http://10.10.1.10:3128"
+        assert vault_client.kwargs["session"].proxies["https"] == "http://10.10.1.10:1080"
+        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_with_client_cert_applied(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,
+            cert=("/path/client.cert", "/path/client.key"),
+        )
+        secret = vault_client.get_secret(secret_path="/path/to/secret")
+        assert secret == {"value": "world"}
+        assert vault_client.kwargs["session"].cert == ("/path/client.cert", "/path/client.key")
+        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()