Add cloud-based auth (aws_iam, gcp, azure_ad) to Akeyless secrets backend

providers/akeyless/docs/secrets-backend.rst

@@ -94,6 +94,81 @@ Connections can be stored in three formats:
         "port": 5432
     }
 
+Authentication Methods
+----------------------
+
+The secrets backend supports the following authentication types:
+
+.. list-table::
+   :header-rows: 1
+   :widths: 15 85
+
+   * - ``access_type``
+     - Description
+   * - ``api_key``
+     - Authenticate with Access ID + Access Key. The default method.
+   * - ``uid``
+     - Use a pre-existing Universal Identity token.
+   * - ``aws_iam``
+     - Authenticate using the host's AWS IAM role. Ideal for **Amazon MWAA**,
+       EC2, ECS, and EKS workloads. No static credentials required.
+   * - ``gcp``
+     - Authenticate using GCP workload identity. Ideal for **Google Managed
+       Service for Apache Airflow** (formerly Cloud Composer) and GCE/GKE
+       workloads.
+   * - ``azure_ad``
+     - Authenticate using Azure AD identity. Ideal for Azure-hosted workloads.
+
+Cloud-based auth types (``aws_iam``, ``gcp``, ``azure_ad``) require the
+optional ``akeyless_cloud_id`` package:
+
+.. code-block:: bash
+
+    pip install apache-airflow-providers-akeyless[cloud_id]
+
+Using with Amazon MWAA
+""""""""""""""""""""""
+
+On `Amazon MWAA <https://aws.amazon.com/managed-workflows-for-apache-airflow/>`__
+you can leverage the environment's IAM execution role to authenticate with
+Akeyless -- no static API keys needed.
+
+1. Add to your ``requirements.txt`` (uploaded to S3):
+
+   .. code-block:: text
+
+       apache-airflow-providers-akeyless[cloud_id]
+
+2. In the MWAA console, add these **Airflow configuration options**:
+
+   .. list-table::
+      :widths: 30 70
+
+      * - ``secrets.backend``
+        - ``airflow.providers.akeyless.secrets.akeyless.AkeylessBackend``
+      * - ``secrets.backend_kwargs``
+        - ``{"api_url": "https://api.akeyless.io", "access_id": "p-xxxxxxxxx", "access_type": "aws_iam"}``
+
+3. Ensure the MWAA VPC has outbound HTTPS access to your Akeyless API
+   endpoint (``api.akeyless.io`` or your Akeyless Gateway).
+
+4. Create an Akeyless ``aws_iam`` Auth Method associated with the MWAA
+   execution role ARN.
+
+Using with Google Managed Service for Apache Airflow
+""""""""""""""""""""""""""""""""""""""""""""""""""""
+
+.. code-block:: ini
+
+    [secrets]
+    backend = airflow.providers.akeyless.secrets.akeyless.AkeylessBackend
+    backend_kwargs = {
+        "api_url": "https://api.akeyless.io",
+        "access_id": "p-xxxxxxxxx",
+        "access_type": "gcp",
+        "gcp_audience": "akeyless.io"
+    }
+
 Parameters
 ----------
 
@@ -127,4 +202,13 @@ Parameters
      - Akeyless Access Key (for ``api_key`` auth).
    * - ``access_type``
      - ``api_key``
-     - Authentication method.
+     - Authentication method (``api_key``, ``uid``, ``aws_iam``, ``gcp``, or ``azure_ad``).
+   * - ``gcp_audience``
+     -
+     - GCP audience string (only for ``gcp`` auth).
+   * - ``azure_object_id``
+     -
+     - Azure AD Object ID (only for ``azure_ad`` auth).
+   * - ``token_ttl``
+     - ``600``
+     - Seconds to cache the API token before re-authenticating.

providers/akeyless/src/airflow/providers/akeyless/secrets/akeyless.py

@@ -32,7 +32,8 @@
 if TYPE_CHECKING:
     from airflow.models.connection import Connection
 
-_SUPPORTED_BACKEND_AUTH_TYPES = ("api_key", "uid")
+_SUPPORTED_BACKEND_AUTH_TYPES = ("api_key", "uid", "aws_iam", "gcp", "azure_ad")
+_CLOUD_AUTH_TYPES = ("aws_iam", "gcp", "azure_ad")
 _DEFAULT_TOKEN_TTL = 600  # 10 minutes
 
 
@@ -62,9 +63,21 @@ class AkeylessBackend(BaseSecretsBackend, LoggingMixin):
     (when ``global_secrets_path`` is set) or ``{base_path}/{key}`` (default).
     Team-scoped lookup can be disabled with ``use_team_secrets_path = False``.
 
-    Only ``api_key`` and ``uid`` authentication types are supported in the
-    secrets backend.  For cloud-based authentication (``aws_iam``, ``gcp``,
-    ``azure_ad``) or other advanced methods, use ``AkeylessHook`` directly.
+    Supported authentication types:
+
+    * ``api_key`` -- authenticate with Access ID + Access Key.
+    * ``uid`` -- use a pre-existing Universal Identity token.
+    * ``aws_iam`` -- authenticate using the host's AWS IAM role (ideal for
+      Amazon MWAA and EC2/ECS/EKS workloads).
+    * ``gcp`` -- authenticate using GCP workload identity (ideal for Google
+      Managed Service for Apache Airflow and GCE/GKE workloads).
+    * ``azure_ad`` -- authenticate using Azure AD identity (ideal for Azure
+      workloads).
+
+    Cloud-based auth types (``aws_iam``, ``gcp``, ``azure_ad``) require the
+    optional ``akeyless_cloud_id`` package::
+
+        pip install apache-airflow-providers-akeyless[cloud_id]
 
     :param connections_path: Akeyless path prefix for Connections (None to disable).
     :param variables_path: Akeyless path prefix for Variables (None to disable).
@@ -77,7 +90,10 @@ class AkeylessBackend(BaseSecretsBackend, LoggingMixin):
     :param api_url: Akeyless API endpoint.
     :param access_id: Access ID.
     :param access_key: Access Key (for ``api_key`` auth).
-    :param access_type: Auth type (``api_key`` or ``uid``).
+    :param access_type: Auth type (``api_key``, ``uid``, ``aws_iam``, ``gcp``,
+        or ``azure_ad``).
+    :param gcp_audience: GCP audience for ``gcp`` auth (optional).
+    :param azure_object_id: Azure AD Object ID for ``azure_ad`` auth (optional).
     :param token_ttl: Seconds to cache the API token before refreshing (default 600).
     """
 
@@ -132,6 +148,13 @@ def _authenticate(self) -> str:
 
         if self._access_type == "uid":
             token = self._extra["uid_token"]
+        elif self._access_type in _CLOUD_AUTH_TYPES:
+            body = akeyless.Auth(
+                access_id=self._access_id,
+                access_type=self._access_type,
+                cloud_id=self._get_cloud_id(),
+            )
+            token = self._client.auth(body).token
         else:
             body = akeyless.Auth(access_id=self._access_id, access_key=self._access_key)
             token = self._client.auth(body).token
@@ -140,6 +163,24 @@ def _authenticate(self) -> str:
         self._token_expiry = now + self._token_ttl
         return token
 
+    def _get_cloud_id(self) -> str:
+        """Generate a cloud identity token for AWS IAM / GCP / Azure AD auth."""
+        try:
+            from akeyless_cloud_id import CloudId
+        except ImportError:
+            raise ImportError(
+                f"`akeyless_cloud_id` is required for {self._access_type} authentication. "
+                "Install it with: pip install apache-airflow-providers-akeyless[cloud_id]"
+            )
+        cid = CloudId()
+        if self._access_type == "aws_iam":
+            return cid.generate()
+        if self._access_type == "gcp":
+            return cid.generateGcp(self._extra.get("gcp_audience"))
+        if self._access_type == "azure_ad":
+            return cid.generateAzure(self._extra.get("azure_object_id"))
+        raise ValueError(f"No cloud-id generator for {self._access_type!r}")
+
     def _get_secret(self, base_path: str | None, key: str) -> str | None:
         if base_path is None:
             return None

providers/akeyless/tests/unit/akeyless/secrets/test_akeyless.py

@@ -185,7 +185,7 @@ def test_custom_separator(self, mock_sdk):
 
     def test_unsupported_access_type_raises(self):
         with pytest.raises(ValueError, match="Unsupported access_type"):
-            _backend(access_type="aws_iam")
+            _backend(access_type="ldap")
 
     @patch(f"{BACKEND_MODULE}.akeyless")
     def test_token_caching(self, mock_sdk):
@@ -277,3 +277,118 @@ def test_get_variable_no_team_when_multi_team_off(self, mock_sdk):
         val = backend.get_variable("my_var", team_name="analytics")
         assert val == "plain"
         mock_sdk.GetSecretValue.assert_called_with(names=["/airflow/variables/my_var"], token="t")
+
+    # ------------------------------------------------------------------
+    # Cloud-based authentication tests
+    # ------------------------------------------------------------------
+
+    @patch(f"{BACKEND_MODULE}.akeyless")
+    def test_aws_iam_auth(self, mock_sdk):
+        """aws_iam auth generates a cloud ID and authenticates."""
+        api = mock_sdk.V2Api.return_value
+        api.auth.return_value = MagicMock(token="aws-token")
+        api.get_secret_value.return_value = {"/airflow/variables/v": "aws-val"}
+
+        mock_cloud_id = MagicMock()
+        mock_cloud_id.generate.return_value = "fake-aws-cloud-id"
+
+        with patch(f"{BACKEND_MODULE}.AkeylessBackend._get_cloud_id", return_value="fake-aws-cloud-id"):
+            backend = _backend(access_type="aws_iam", access_key=None)
+            val = backend.get_variable("v")
+
+        assert val == "aws-val"
+        mock_sdk.Auth.assert_called_once_with(
+            access_id="p-test123", access_type="aws_iam", cloud_id="fake-aws-cloud-id"
+        )
+
+    @patch(f"{BACKEND_MODULE}.akeyless")
+    def test_gcp_auth(self, mock_sdk):
+        """gcp auth generates a GCP cloud ID and authenticates."""
+        api = mock_sdk.V2Api.return_value
+        api.auth.return_value = MagicMock(token="gcp-token")
+        api.get_secret_value.return_value = {"/airflow/variables/v": "gcp-val"}
+
+        with patch(f"{BACKEND_MODULE}.AkeylessBackend._get_cloud_id", return_value="fake-gcp-cloud-id"):
+            backend = _backend(access_type="gcp", access_key=None, gcp_audience="my-audience")
+            val = backend.get_variable("v")
+
+        assert val == "gcp-val"
+        mock_sdk.Auth.assert_called_once_with(
+            access_id="p-test123", access_type="gcp", cloud_id="fake-gcp-cloud-id"
+        )
+
+    @patch(f"{BACKEND_MODULE}.akeyless")
+    def test_azure_ad_auth(self, mock_sdk):
+        """azure_ad auth generates an Azure cloud ID and authenticates."""
+        api = mock_sdk.V2Api.return_value
+        api.auth.return_value = MagicMock(token="azure-token")
+        api.get_secret_value.return_value = {"/airflow/variables/v": "azure-val"}
+
+        with patch(f"{BACKEND_MODULE}.AkeylessBackend._get_cloud_id", return_value="fake-azure-id"):
+            backend = _backend(access_type="azure_ad", access_key=None, azure_object_id="obj-123")
+            val = backend.get_variable("v")
+
+        assert val == "azure-val"
+        mock_sdk.Auth.assert_called_once_with(
+            access_id="p-test123", access_type="azure_ad", cloud_id="fake-azure-id"
+        )
+
+    @patch(f"{BACKEND_MODULE}.akeyless")
+    def test_aws_iam_cloud_id_integration(self, mock_sdk):
+        """aws_iam calls CloudId.generate() to produce the cloud identity."""
+        api = mock_sdk.V2Api.return_value
+        api.auth.return_value = MagicMock(token="t")
+        api.get_secret_value.return_value = {"/airflow/variables/v": "val"}
+
+        mock_cid_instance = MagicMock()
+        mock_cid_instance.generate.return_value = "real-aws-cloud-id"
+        mock_cloud_id_cls = MagicMock(return_value=mock_cid_instance)
+
+        with patch.dict("sys.modules", {"akeyless_cloud_id": MagicMock(CloudId=mock_cloud_id_cls)}):
+            backend = _backend(access_type="aws_iam", access_key=None)
+            backend.get_variable("v")
+
+        mock_cid_instance.generate.assert_called_once()
+
+    @patch(f"{BACKEND_MODULE}.akeyless")
+    def test_gcp_cloud_id_passes_audience(self, mock_sdk):
+        """gcp auth passes gcp_audience to CloudId.generateGcp()."""
+        api = mock_sdk.V2Api.return_value
+        api.auth.return_value = MagicMock(token="t")
+        api.get_secret_value.return_value = {"/airflow/variables/v": "val"}
+
+        mock_cid_instance = MagicMock()
+        mock_cid_instance.generateGcp.return_value = "gcp-id"
+        mock_cloud_id_cls = MagicMock(return_value=mock_cid_instance)
+
+        with patch.dict("sys.modules", {"akeyless_cloud_id": MagicMock(CloudId=mock_cloud_id_cls)}):
+            backend = _backend(access_type="gcp", access_key=None, gcp_audience="my-aud")
+            backend.get_variable("v")
+
+        mock_cid_instance.generateGcp.assert_called_once_with("my-aud")
+
+    @patch(f"{BACKEND_MODULE}.akeyless")
+    def test_azure_cloud_id_passes_object_id(self, mock_sdk):
+        """azure_ad auth passes azure_object_id to CloudId.generateAzure()."""
+        api = mock_sdk.V2Api.return_value
+        api.auth.return_value = MagicMock(token="t")
+        api.get_secret_value.return_value = {"/airflow/variables/v": "val"}
+
+        mock_cid_instance = MagicMock()
+        mock_cid_instance.generateAzure.return_value = "azure-id"
+        mock_cloud_id_cls = MagicMock(return_value=mock_cid_instance)
+
+        with patch.dict("sys.modules", {"akeyless_cloud_id": MagicMock(CloudId=mock_cloud_id_cls)}):
+            backend = _backend(access_type="azure_ad", access_key=None, azure_object_id="obj-456")
+            backend.get_variable("v")
+
+        mock_cid_instance.generateAzure.assert_called_once_with("obj-456")
+
+    def test_cloud_auth_missing_package_raises(self):
+        """Cloud auth raises ImportError when akeyless_cloud_id is not installed."""
+        import sys
+
+        backend = _backend(access_type="aws_iam", access_key=None)
+        with patch.dict(sys.modules, {"akeyless_cloud_id": None}):
+            with pytest.raises(ImportError, match="akeyless_cloud_id"):
+                backend._get_cloud_id()