[providers] Consolidate GCP authentication in Hashicorp Vault

This commit unifies the GCP authentication logic in the Hashicorp Vault secrets backend.
The 'gcp' authentication type now supports both service account key-based authentication
and Application Default Credentials (ADC) automatically.

Changes:
- Consolidate 'gcp' authentication to handle both keys and managed identities.
- Automatically determine service account email from credentials, with a fallback
  to the GCE/GKE metadata server.
- Fix a bug where the 'sub' claim in the signed JWT payload was incorrectly
  set to the credentials object instead of the service account email.
- Update VaultHook and VaultBackend docstrings and initialization logic.
- Add unit tests for ADC-based GCP authentication and update existing tests.

Author: fpopic

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

@@ -171,10 +171,6 @@ def __init__(
                 raise VaultError("The 'gcp' authentication type requires 'gcp_scopes'")
             if not role_id:
                 raise VaultError("The 'gcp' authentication type requires 'role_id'")
-            if not gcp_key_path and not gcp_keyfile_dict:
-                raise VaultError(
-                    "The 'gcp' authentication type requires 'gcp_key_path' or 'gcp_keyfile_dict'"
-                )
 
         self.kv_engine_version = kv_engine_version or 2
         self.url = url
@@ -352,25 +348,35 @@ def _auth_gcp(self, _client: hvac.Client) -> None:
         import json
         import time
 
-        import googleapiclient
-
-        if self.gcp_keyfile_dict:
-            creds = self.gcp_keyfile_dict
-        elif self.gcp_key_path:
-            with open(self.gcp_key_path) as f:
-                creds = json.load(f)
-
-        service_account = creds["client_email"]
+        # Determine service account email
+        service_account_email = getattr(credentials, "service_account_email", None)
+        if not service_account_email or not isinstance(service_account_email, str):
+            service_account_email = getattr(credentials, "client_email", None)
+        
+        if not service_account_email or not isinstance(service_account_email, str):
+            # Fallback for Compute Engine credentials if email is not yet populated
+            try:
+                from google.auth import compute_engine
+                if isinstance(credentials, compute_engine.Credentials):
+                    if not getattr(credentials, "service_account_email", None):
+                        from google.auth import transport
+                        credentials.refresh(transport.requests.Request())
+                    service_account_email = credentials.service_account_email
+            except Exception:
+                pass
+        
+        if not service_account_email:
+            raise VaultError("Could not determine service account email from credentials")
 
         # Generate a payload for subsequent "signJwt()" call
-        # Reference: https://googleapis.dev/python/google-auth/latest/reference/google.auth.jwt.html#google.auth.jwt.Credentials
         now = int(time.time())
         expires = now + 900  # 15 mins in seconds, can't be longer.
-        payload = {"iat": now, "exp": expires, "sub": credentials, "aud": f"vault/{self.role_id}"}
+        payload = {"iat": now, "exp": expires, "sub": service_account_email, "aud": f"vault/{self.role_id}"}
         body = {"payload": json.dumps(payload)}
-        name = f"projects/{project_id}/serviceAccounts/{service_account}"
+        name = f"projects/{project_id}/serviceAccounts/{service_account_email}"
 
         # Perform the GCP API call
+        import googleapiclient.discovery
         iam = googleapiclient.discovery.build("iam", "v1", credentials=credentials)
         request = iam.projects().serviceAccounts().signJwt(name=name, body=body)
         resp = request.execute()

providers/hashicorp/src/airflow/providers/hashicorp/hooks/vault.py

@@ -79,7 +79,7 @@ class VaultHook(BaseHook):
 
     :param vault_conn_id: The id of the connection to use
     :param auth_type: Authentication Type for the Vault. Default is ``token``. Available values are:
-        ('approle', 'github', 'gcp', 'jwt', 'kubernetes', 'ldap', 'token', 'userpass')
+        ('approle', 'aws_iam', 'azure', 'github', 'gcp', 'jwt', 'kubernetes', 'ldap', 'radius', 'token', 'userpass')
     :param auth_mount_point: It can be used to define mount_point for authentication chosen
           Default depends on the authentication method used.
     :param kv_engine_version: Select the version of the engine to run (``1`` or ``2``). Defaults to
@@ -161,6 +161,10 @@ def __init__(
             if not region:
                 region = self.connection.extra_dejson.get("region")
 
+        if auth_type == "gcp":
+            if not role_id:
+                role_id = self.connection.extra_dejson.get("role_id") or self.connection.login
+
         azure_resource, azure_tenant_id = (
             self._get_azure_parameters_from_connection(azure_resource, azure_tenant_id)
             if auth_type == "azure"

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

@@ -255,21 +255,18 @@ def test_azure_missing_tenant_id(self, mock_hvac):
                 secret_id="pass",
             )
 
-    @mock.patch("builtins.open", create=True)
     @mock.patch("airflow.providers.google.cloud.utils.credentials_provider._get_scopes")
     @mock.patch("airflow.providers.google.cloud.utils.credentials_provider.get_credentials_and_project_id")
     @mock.patch("airflow.providers.hashicorp._internal_client.vault_client.hvac.Client")
     @mock.patch("googleapiclient.discovery.build")
-    def test_gcp(self, mock_google_build, mock_hvac_client, mock_get_credentials, mock_get_scopes, mock_open):
-        # Mock the content of the file 'path.json'
-        mock_file = mock.MagicMock()
-        mock_file.read.return_value = '{"client_email": "service_account_email"}'
-        mock_open.return_value.__enter__.return_value = mock_file
-
+    def test_gcp(self, mock_google_build, mock_hvac_client, mock_get_credentials, mock_get_scopes):
         mock_client = mock.MagicMock()
         mock_hvac_client.return_value = mock_client
         mock_get_scopes.return_value = ["scope1", "scope2"]
-        mock_get_credentials.return_value = ("credentials", "project_id")
+        
+        mock_credentials = mock.MagicMock()
+        mock_credentials.client_email = "service_account_email"
+        mock_get_credentials.return_value = (mock_credentials, "project_id")
 
         # Mock the current time to use for iat and exp
         current_time = int(time.time())
@@ -321,29 +318,73 @@ def mocked_json_dumps(payload):
         # Assert iat and exp values are as expected
         assert payload["iat"] == iat
         assert payload["exp"] == exp
+        assert payload["sub"] == "service_account_email"
         assert abs(payload["exp"] - (payload["iat"] + 3600)) < 10  # Validate exp is 3600 seconds after iat
 
         client.auth.gcp.login.assert_called_with(role="role", jwt="mocked_jwt")
         client.is_authenticated.assert_called_with()
         assert vault_client.kv_engine_version == 2
 
-    @mock.patch("builtins.open", create=True)
+    @mock.patch("airflow.providers.google.cloud.utils.credentials_provider._get_scopes")
+    @mock.patch("airflow.providers.google.cloud.utils.credentials_provider.get_credentials_and_project_id")
+    @mock.patch("airflow.providers.hashicorp._internal_client.vault_client.hvac.Client")
+    @mock.patch("googleapiclient.discovery.build")
+    def test_gcp_adc(self, mock_google_build, mock_hvac_client, mock_get_credentials, mock_get_scopes):
+        mock_client = mock.MagicMock()
+        mock_hvac_client.return_value = mock_client
+        mock_get_scopes.return_value = ["scope1", "scope2"]
+
+        mock_credentials = mock.MagicMock()
+        mock_credentials.service_account_email = "service_account_email"
+        mock_get_credentials.return_value = (mock_credentials, "project_id")
+
+        mock_sign_jwt = (
+            mock_google_build.return_value.projects.return_value.serviceAccounts.return_value.signJwt
+        )
+        mock_sign_jwt.return_value.execute.return_value = {"signedJwt": "mocked_jwt"}
+
+        vault_client = _VaultClient(
+            auth_type="gcp",
+            gcp_scopes="scope1,scope2",
+            role_id="role",
+            url="http://localhost:8180",
+            session=None,
+        )
+
+        client = vault_client.client  # Trigger the Vault client creation
+
+        # Validate that the HVAC client and other mocks are called correctly
+        mock_hvac_client.assert_called_with(url="http://localhost:8180", session=None)
+        mock_get_scopes.assert_called_with("scope1,scope2")
+        mock_get_credentials.assert_called_with(
+            key_path=None, keyfile_dict=None, scopes=["scope1", "scope2"]
+        )
+
+        # Extract the arguments passed to the mocked signJwt API
+        args, kwargs = mock_sign_jwt.call_args
+        payload = json.loads(kwargs["body"]["payload"])
+
+        # Assert sub is correctly set to service account email
+        assert payload["sub"] == "service_account_email"
+
+        client.auth.gcp.login.assert_called_with(role="role", jwt="mocked_jwt")
+        client.is_authenticated.assert_called_with()
+        assert vault_client.kv_engine_version == 2
+
     @mock.patch("airflow.providers.google.cloud.utils.credentials_provider._get_scopes")
     @mock.patch("airflow.providers.google.cloud.utils.credentials_provider.get_credentials_and_project_id")
     @mock.patch("airflow.providers.hashicorp._internal_client.vault_client.hvac.Client")
     @mock.patch("googleapiclient.discovery.build")
     def test_gcp_different_auth_mount_point(
-        self, mock_google_build, mock_hvac_client, mock_get_credentials, mock_get_scopes, mock_open
+        self, mock_google_build, mock_hvac_client, mock_get_credentials, mock_get_scopes,
     ):
-        # Mock the content of the file 'path.json'
-        mock_file = mock.MagicMock()
-        mock_file.read.return_value = '{"client_email": "service_account_email"}'
-        mock_open.return_value.__enter__.return_value = mock_file
-
         mock_client = mock.MagicMock()
         mock_hvac_client.return_value = mock_client
         mock_get_scopes.return_value = ["scope1", "scope2"]
-        mock_get_credentials.return_value = ("credentials", "project_id")
+        
+        mock_credentials = mock.MagicMock()
+        mock_credentials.client_email = "service_account_email"
+        mock_get_credentials.return_value = (mock_credentials, "project_id")
 
         mock_sign_jwt = (
             mock_google_build.return_value.projects.return_value.serviceAccounts.return_value.signJwt
@@ -394,26 +435,27 @@ def mocked_json_dumps(payload):
         # Assert iat and exp values are as expected
         assert payload["iat"] == iat
         assert payload["exp"] == exp
+        assert payload["sub"] == "service_account_email"
         assert abs(payload["exp"] - (payload["iat"] + 3600)) < 10  # Validate exp is 3600 seconds after iat
 
         client.auth.gcp.login.assert_called_with(role="role", jwt="mocked_jwt", mount_point="other")
         client.is_authenticated.assert_called_with()
         assert vault_client.kv_engine_version == 2
 
-    @mock.patch(
-        "builtins.open", new_callable=mock_open, read_data='{"client_email": "service_account_email"}'
-    )
     @mock.patch("airflow.providers.google.cloud.utils.credentials_provider._get_scopes")
     @mock.patch("airflow.providers.google.cloud.utils.credentials_provider.get_credentials_and_project_id")
     @mock.patch("airflow.providers.hashicorp._internal_client.vault_client.hvac.Client")
     @mock.patch("googleapiclient.discovery.build")
     def test_gcp_dict(
-        self, mock_google_build, mock_hvac_client, mock_get_credentials, mock_get_scopes, mock_file
+        self, mock_google_build, mock_hvac_client, mock_get_credentials, mock_get_scopes
     ):
         mock_client = mock.MagicMock()
         mock_hvac_client.return_value = mock_client
         mock_get_scopes.return_value = ["scope1", "scope2"]
-        mock_get_credentials.return_value = ("credentials", "project_id")
+        
+        mock_credentials = mock.MagicMock()
+        mock_credentials.client_email = "service_account_email"
+        mock_get_credentials.return_value = (mock_credentials, "project_id")
 
         mock_sign_jwt = (
             mock_google_build.return_value.projects.return_value.serviceAccounts.return_value.signJwt
@@ -463,6 +505,7 @@ def mocked_json_dumps(payload):
         # Assert iat and exp values are as expected
         assert payload["iat"] == iat
         assert payload["exp"] == exp
+        assert payload["sub"] == "service_account_email"
         assert abs(payload["exp"] - (payload["iat"] + 3600)) < 10  # Validate exp is 3600 seconds after iat
 
         client.auth.gcp.login.assert_called_with(role="role", jwt="mocked_jwt")