[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

@@ -348,24 +348,35 @@ def _auth_gcp(self, _client: hvac.Client) -> None:
         import json
         import time
 
-        type = getattr(credentials, "type", None)
-        if type != "service_account":
-            raise VaultError("Credentials are not of type service_account")
-
-        service_account_email = getattr(credentials, "client_email", None)
+        # 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_email}"
 
         # Perform the GCP API call
-        import googleapiclient
+        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
@@ -152,13 +152,13 @@ def __init__(
         if kwargs:
             client_kwargs = merge_dicts(client_kwargs, kwargs)
 
-        if auth_type == "approle" and self.connection.login:
+        if auth_type in ("approle", "gcp") and self.connection.login:
             role_id = self.connection.login
 
-        if auth_type == "aws_iam":
+        if auth_type in ("aws_iam", "gcp"):
             if not role_id:
                 role_id = self.connection.extra_dejson.get("role_id")
-            if not region:
+            if not region and auth_type == "aws_iam":
                 region = self.connection.extra_dejson.get("region")
 
         azure_resource, azure_tenant_id = (

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

@@ -263,7 +263,10 @@ def test_gcp(self, mock_google_build, mock_hvac_client, mock_get_credentials, mo
         mock_client = mock.MagicMock()
         mock_hvac_client.return_value = mock_client
         mock_get_scopes.return_value = ["scope1", "scope2"]
-        mock_get_credentials.return_value = ('{"client_email": "service_account_email"}', "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())
@@ -315,12 +318,59 @@ 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("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")
@@ -331,7 +381,10 @@ def test_gcp_different_auth_mount_point(
         mock_client = mock.MagicMock()
         mock_hvac_client.return_value = mock_client
         mock_get_scopes.return_value = ["scope1", "scope2"]
-        mock_get_credentials.return_value = ('{"client_email": "service_account_email"}', "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
@@ -382,6 +435,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", mount_point="other")
@@ -398,7 +452,10 @@ def test_gcp_dict(
         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
@@ -448,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")