This is an automated email from the ASF dual-hosted git repository.
eladkal pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 6e6d0af05e5 Add cloud-based auth (aws_iam, gcp, azure_ad) to Akeyless
secrets backend (#69772)
6e6d0af05e5 is described below
commit 6e6d0af05e5575d19ca46f0fbfd1098b39793d60
Author: baraka-akeyless <[email protected]>
AuthorDate: Sun Jul 12 14:06:03 2026 +0300
Add cloud-based auth (aws_iam, gcp, azure_ad) to Akeyless secrets backend
(#69772)
* Add cloud-based auth (aws_iam, gcp, azure_ad) to AkeylessBackend
The secrets backend previously only supported api_key and uid auth.
This adds aws_iam, gcp, and azure_ad authentication using the
akeyless_cloud_id package, enabling seamless integration with managed
Airflow services like Amazon MWAA, GCP Cloud Composer, and Azure
workloads without requiring static API keys.
Co-authored-by: Cursor <[email protected]>
* Update Cloud Composer references to Managed Service for Apache Airflow
Cloud Composer was rebranded to Managed Service for Apache Airflow
(Managed Airflow) as of April 2026.
Co-authored-by: Cursor <[email protected]>
---------
Co-authored-by: Cursor <[email protected]>
---
providers/akeyless/docs/secrets-backend.rst | 86 ++++++++++++++-
.../airflow/providers/akeyless/secrets/akeyless.py | 51 ++++++++-
.../tests/unit/akeyless/secrets/test_akeyless.py | 117 ++++++++++++++++++++-
3 files changed, 247 insertions(+), 7 deletions(-)
diff --git a/providers/akeyless/docs/secrets-backend.rst
b/providers/akeyless/docs/secrets-backend.rst
index ea970705e4a..8fa04e1c20a 100644
--- a/providers/akeyless/docs/secrets-backend.rst
+++ b/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.
diff --git
a/providers/akeyless/src/airflow/providers/akeyless/secrets/akeyless.py
b/providers/akeyless/src/airflow/providers/akeyless/secrets/akeyless.py
index 2e2dc265a3c..eedc6794810 100644
--- a/providers/akeyless/src/airflow/providers/akeyless/secrets/akeyless.py
+++ b/providers/akeyless/src/airflow/providers/akeyless/secrets/akeyless.py
@@ -32,7 +32,8 @@ from airflow.utils.log.logging_mixin import LoggingMixin
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 @@ class AkeylessBackend(BaseSecretsBackend, LoggingMixin):
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 @@ class AkeylessBackend(BaseSecretsBackend, LoggingMixin):
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
diff --git a/providers/akeyless/tests/unit/akeyless/secrets/test_akeyless.py
b/providers/akeyless/tests/unit/akeyless/secrets/test_akeyless.py
index aa400db3561..eb97bfecb38 100644
--- a/providers/akeyless/tests/unit/akeyless/secrets/test_akeyless.py
+++ b/providers/akeyless/tests/unit/akeyless/secrets/test_akeyless.py
@@ -185,7 +185,7 @@ class TestAkeylessBackend:
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 @@ class TestAkeylessBackend:
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()
