Apache Airflow 密钥后端

2025 年 6 月 10 日 | 阅读 8 分钟

引言

Apache Airflow 是一个强大的平台，用于以编程方式创建、调度和监控工作流。但是，随着工作流变得越来越复杂并与各种系统集成，管理密码、API 密钥和其他凭据等敏感信息已成为一个关键问题。这时，Airflow 中的 Secrets Backend（密钥后端）概念就发挥了作用。

什么是 Airflow 中的 Secrets Backend？

Airflow 中的 Secrets Backend 是一种用于安全存储和检索工作流所需的敏感信息的机制。它不是在 DAG 或配置文件中硬编码凭据，而是允许集中管理密钥，从而提高安全性和合规性。

为什么使用 Secrets Backend？

安全性：将密钥存储在 Secrets Backend 中可以降低意外泄露的风险，例如通过版本控制系统。
合规性：许多组织都有要求对敏感信息进行加密和集中管理的政策。
集中化：密钥集中管理，简化了更新并减少了重复。
可伸缩性：随着工作流数量的增长，集中管理密钥可以使维护更加容易。

Secrets Backend 搜索路径

搜索路径决定了 Airflow 检查各种密钥来源的顺序。这种机制确保了灵活性，并允许在主来源找不到密钥时回退到其他选项。

默认搜索路径

Airflow 的默认搜索路径如下：

环境变量：密钥定义为环境变量。
密钥后端：已配置的外部密钥后端，如 AWS Secrets Manager 或 HashiCorp Vault。
回退到变量和连接：存储在 Airflow 元数据数据库中的密钥。

您可以通过在 airflow.cfg 中配置 providers.Secrets 部分来定制搜索路径。

定制搜索路径

要定制搜索路径，请按所需顺序指定后端：

[secrets]
backend = airflow.providers.amazon.aws.secrets.secrets_manager.SecretsManagerBackend
backend_kwargs = {
    "connections_prefix": "airflow/connections",
    "variables_prefix": "airflow/variables"
}

此示例优先使用 AWS Secrets Manager，同时仍允许使用环境变量和其他后端作为回退。

Secrets Backend 配置

正确配置 Secrets Backend 可确保无缝集成和安全检索密钥。以下是配置 Airflow 中 Secrets Backend 的一般步骤和示例：

配置的一般步骤

安装必要的包：根据后端，安装相应的提供程序包（例如，AWS Secrets Manager 的 apache-airflow-providers-amazon）。
更新 airflow.cfg：修改配置文件以包含所需的后端。
提供身份验证凭据：确保 Airflow 具有访问后端所必需的凭据，以保证安全。
测试配置：通过在测试环境中检索密钥来验证设置。

示例：配置 HashiCorp Vault

[secrets]
backend = airflow.providers.hashicorp.secrets.vault.VaultBackend
backend_kwargs = {
    "connections_path": "airflow/connections",
    "variables_path": "airflow/variables",
    "vault_url": "http://127.0.0.1:8200",
    "token": "my-secret-token"
}

示例：配置 AWS Secrets Manager

[secrets]
backend = airflow.providers.amazon.aws.secrets.secrets_manager.SecretsManagerBackend
backend_kwargs = {
    "connections_prefix": "airflow/connections",
    "variables_prefix": "airflow/variables"
}

其他配置示例

Google Cloud Secret Manager

[secrets]
backend = airflow.providers.google.cloud.secrets.secret_manager.CloudSecretManagerBackend
backend_kwargs = {
    "project_id": "my-gcp-project"
}

Kubernetes Secrets

[secrets]
backend = airflow.providers. Cf. Kubernetes. Secrets. Kubernetes.KubernetesSecretsBackend
backend_kwargs = {
    "namespace": "airflow",
    "in_cluster": True
}

Databricks Secret Scope

[secrets]
backend = airflow.providers. Data bricks. Secrets. Data bricks.DatabricksSecretBackend
backend_kwargs = {
    "scope": "my-data bricks-scope"
}

配置技巧

确保权限：验证 Airflow 服务帐户是否具有访问密钥后端所需的权限。
使用安全通道：与后端通信时，始终使用加密连接（例如 HTTPS 或 TLS）。
环境隔离：为开发、暂存和生产环境分离配置。

支持的核心后端

Airflow 开箱即用地支持多个核心密钥后端。它们包括：

环境变量：适用于小型工作流，设置简单快捷。
AWS Systems Manager Parameter Store：适用于以 AWS 为中心的工作流的可伸缩选项。
AWS Secrets Manager：提供高级密钥管理功能，例如轮换。
Google Secret Manager：与 GCP 托管的工作流无缝集成。
HashiCorp Vault：为企业级设置提供强大的安全功能。
Azure Key Vault：Azure 托管环境的安全解决方案。

功能比较

后端	轮换支持	访问控制	日志记录	云集成
环境变量	不能	基本功能	不能	不能
AWS Secrets Manager	是的	细粒度	是的	AWS
Google Secret Manager	是的	细粒度	是的	GCP
HashiCorp Vault	是的	细粒度	是的	不能
Azure Key Vault	是的	细粒度	是的	Azure

社区提供的 Secrets Backend

除了核心后端之外，Apache Airflow 社区还提供了额外的后端以实现扩展功能。这些后端作为 Airflow 提供程序包的一部分提供。

示例

Databricks Secret Scope
- 适用于与 Databricks 交互的工作流。
- 简化了 Databricks 密钥与 Airflow 的集成。
Kubernetes Secrets
- 适用于在 Kubernetes 集群上运行的工作流。
- 利用 Kubernetes 原生密钥管理功能。
AWS Parameter Store
- 提供 AWS Secrets Manager 的简单且经济高效的替代方案。

优点

利用现有基础设施。
简化与平台特定功能的集成。
减少自定义开发的需求。

自建 Secrets Backend

在内置或社区提供的后端不满足要求的场景中，您可以实现自定义的 Secrets Backend。

创建自定义后端的步骤

要实现自定义 Secrets Backend，您必须创建一个继承自 BaseSecretsBackend 并实现其方法的类。以下是关键组件：

1. 初始化 (__init__)

定义后端的配置参数（例如，API 端点、身份验证令牌）。

2. 检索连接密钥 (get_conn_value)

此方法为给定的连接 ID 获取连接密钥。

3. 检索变量密钥 (get_variable)

此方法从后端检索存储的变量。

4. 检索配置密钥 (get_config)

此方法从后端检索配置选项。

创建自定义 Secrets Backend 的分步指南

步骤 1：继承 BaseSecretsBackend

首先，从 Airflow 导入 BaseSecretsBackend 类并创建一个新的子类。

From airflow. Secrets import BaseSecretsBackend

class MyCustomSecretsBackend(BaseSecretsBackend):
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        # Initialize configuration, e.g., API URL, tokens
        self.api_url = kwargs.get("api_url")
        self.auth_token = kwargs.get("auth_token")

步骤 2：实现 get_conn_value

get_conn_value 方法获取给定 conn_id 的连接详细信息。以下是一个示例：

    def get_conn_value(self, conn_id: str) -> str:
        """
        Fetch connection details from the custom secrets backend.
        
        Args:
            conn_id (str): The connection ID in Airflow.
        
        Returns:
            Str: A connection URI or None if the connection is not found.
        """
        # Fetch data from the backend (simulated here with a dictionary)
        mock_secrets = {
            "my_database": {
                "login": "user123",
                "password": "pass123",
                "host": "db.example.com",
                "port": 5432
            }
        }

        secret = mock_secrets.get(conn_id)
        if not secret:
            return None

        # Convert the secret to an Airflow connection URI
        connection_uri = (
            f"PostgreSQL://{secret['login']}:{secret['password']}@"
            f"{secret['host']}:{secret['port']}"
        )

步骤 3：实现 get_variable

此方法从后端检索变量的值。

    def get_variable(self, key: str) -> str:
        """
        Fetch a variable value from the custom secrets backend.
        
        Args:
            Key (str): The variable name in Airflow.
        
        Returns:
            Str: The variable value or None if not found.
        """
        # Simulate a backend response
        mock_variables = {
            "my_variable": "my_value"
        }

        return mock_variables.get(key)

步骤 4：实现 get_config

此方法从后端检索配置选项。

从自定义密钥后端获取配置值。

        Args:
            Key (str): The configuration key in Airflow.
        
        Returns:
            Str: The configuration value or None if not found.
        """
        mock_configs = {
            "config_key": "config_value"
        }

        return mock_configs.get(key)

步骤 5：在 airflow.cfg 中注册自定义后端

实现自定义后端后，您需要在 Airflow 配置文件 (airflow.cfg) 中注册它。

backend = my_package.my_module.MyCustomSecretsBackend
backend_kwargs = {
    "api_url": "https://my-secrets-backend/api",
    "auth_token": "my-auth-token"
}

适应非 Airflow 兼容的连接密钥格式

许多密钥管理系统以不直接符合 Apache Airflow 预期结构的格式存储凭据。当用于连接、变量和配置时，Airflow 要求密钥遵循特定的架构。当使用存储密钥格式自定义或非标准格式的第三方密钥管理系统时，通常会出现此挑战。将这些密钥适配到与 Airflow 无缝配合，可确保工作流安全运行且不中断。

Airflow 的连接要求

在 Airflow 中，连接定义了工作流中的任务如何与外部系统（例如数据库、云服务、API）进行交互。连接通常包含以下详细信息：

登录凭据：用户名和密码。
主机信息：主机名或 IP 地址。
端口号：服务特定端口。
Extras：JSON 格式的其他配置，例如 API 令牌、服务特定设置或 SSL 配置。

Airflow 连接示例

典型的 Airflow 连接以以下结构表示：

{
  "login": "user123",
  "password": "pass123",
  "host": "example.com",
  "port": 5432,
  "schema": "database_name",
  "extras": "{\"key\": \"value\"}"
}

密钥后端通常以自定义格式存储密钥，这可能与 Airflow 的架构不同。例如：

扁平格式：键值对，无层级结构（例如，username=user123, password=pass123）。
嵌套格式：包含其他元数据的 JSON 或 YAML 对象。
编码格式：密钥以 Base64 字符串形式存储，需要解码。

适应不兼容密钥格式的挑战

非标准键名：Airflow 期望特定的字段名（例如，login, password），但密钥后端可能使用自定义键，如 username 或 pwd。
层级或嵌套数据：密钥可能以层级格式存储，例如 JSON 对象，必须对其进行解析才能提取单个字段。
数据编码：某些后端以编码或加密格式存储密钥，需要额外的解码或解密步骤。
单个键用于多个密钥：单个键可能包含多个捆绑在一起的凭据，例如包含用户名、密码和主机的 JSON 对象。
附加元数据：密钥后端可能在密钥旁边存储不相关的元数据，需要进行过滤或转换。
版本控制：某些后端对密钥实现版本控制，如果未自动返回最新版本，可能会使检索复杂化。

适应密钥格式的策略

将自定义键映射到 Airflow 键

如果密钥后端使用非标准键名，请在后端键和 Airflow 期望的键之间创建映射。

示例：键映射

backend_secret = {
    "username": "user123",
    "pwd": "pass123",
    "server": "example.com",
    "port_number": 5432
}

mapped_secret = {
    "login": backend_secret["username"],
    "password": backend_secret["pwd"],
    "host": backend_secret["server"],
    "port": backend_secret["port_number"]
}

解析嵌套或层级数据

如果后端以嵌套格式（例如 JSON）存储密钥，请编写一个解析器来提取所需的字段。

示例：解析 JSON 数据

import json
nested_secret = '{"credentials": {"user": "user123", "pass": "pass123"}, "host_info": {"host": "example.com", "port": 5432}}'

# Parse and map the secret:
secret_dict = json.loads(nested_secret)
parsed_secret = {
    "login": secret_dict["credentials"]["user"],
    "password": secret_dict["credentials"]["pass"],
    "host": secret_dict["host_info"]["host"],
    "port": secret_dict["host_info"]["port"]
}

解码编码的密钥

对于编码的密钥，在 Airflow 中使用它们之前进行解码。Base64 是一种常见的编码格式。

示例：解码 Base64 密钥

# Encoded secret:
encoded_secret = {
    "login": base64.b64encode(b"user123").decode("utf-8"),
    "password": base64.b64encode(b"pass123").decode("utf-8")
}

# Decode the secret:
decoded_secret = {
    "login": base64.b64decode(encoded_secret["login"]).decode("utf-8"),
    "password": base64.b64decode(encoded_secret["password"]).decode("utf-8")
}

捆绑密钥的自定义解析器

如果多个密钥存储在单个键中，请使用自定义解析逻辑提取必要的值。

示例：解析捆绑密钥

bundled_secret = '{"user": "user123", "password": "pass123", "host": "example.com", "port": 5432}'
import json
parsed_secret = json.loads(bundled_secret)
connection = {
    "login": parsed_secret["user"],
    "password": parsed_secret["password"],
    "host": parsed_secret["host"],
    "port": parsed_secret["port"]
}

在自定义 Secrets Backend 中实现适配器

Airflow 允许用户通过继承 BaseSecretsBackend 来创建自定义 Secrets Backend。此功能可用于为不兼容的密钥格式实现适配器。

自定义 Secrets Backend 示例

from airflow. secrets import BaseSecretsBackend
import json

class CustomSecretsBackend(BaseSecretsBackend):
    def get_conn_value(self, conn_id: str) -> str:
        # Fetch the secret from a backend (mocked here as a JSON string)
        backend_secret = '{"username": "user123", "password": "pass123", "server": "example.com", "port_number": 5432}'
        
        # Parse and map to Airflow connection format
        secret_dict = json.loads(backend_secret)
        connection = {
            "login": secret_dict["username"],
            "password": secret_dict["password"],
            "host": secret_dict["server"],
            "port": secret_dict["port_number"]
        }
        return connection

    def get_variable(self, key: str):
        # Logic for fetching variables
        pass

    def get_config(self, key: str):
        # Logic for fetching config
        pass

适应密钥格式

记录映射逻辑
- 维护关于后端密钥如何映射到 Airflow 预期格式的清晰文档。
进行广泛测试
- 在部署到生产环境之前，在暂存环境中验证适配逻辑。
使用特定于环境的密钥
- 为每个环境（例如，开发、暂存、生产）隔离密钥，以避免意外误用。
监控和审计
- 定期审计适配逻辑，以确保与后端密钥格式更新的兼容性。
优雅地处理故障
- 实现错误处理机制，以应对诸如缺少键、数据格式错误或解码错误等情况。

通过了解可用选项、定制配置以及适应独特要求，组织可以增强安全性、简化管理并确保合规性，无论是利用核心后端、社区提供的解决方案还是自定义实现；一个实现良好的 Secrets Backend 是强大工作流编排的基石。

下一主题Apache-airflow-security

Apache Airflow 密钥后端

引言

什么是 Airflow 中的 Secrets Backend？

为什么使用 Secrets Backend？

Secrets Backend 搜索路径

默认搜索路径

定制搜索路径

Secrets Backend 配置

其他配置示例

支持的核心后端

社区提供的 Secrets Backend

自建 Secrets Backend

创建自定义后端的步骤

创建自定义 Secrets Backend 的分步指南

适应非 Airflow 兼容的连接密钥格式

Airflow 的连接要求

适应不兼容密钥格式的挑战

适应密钥格式的策略

将自定义键映射到 Airflow 键

解析嵌套或层级数据

解码编码的密钥

捆绑密钥的自定义解析器

在自定义 Secrets Backend 中实现适配器

适应密钥格式

联系信息

关注我们

教程

面试题

在线编译器

Python

Java

.Net Framework

AI, ML and Data Science

Cloud Technology

B.Tech and MCA

Web Technology

PHP

Software Testing

Technical Interview

Java Interview

Python

Web Interview

Database Interview

B.Tech / MCA

Important Interview

Software Testing Interview

Company Interviews

Online Compilers

Multiple Choice Questions

Airflow 教程

Apache Airflow 密钥后端

引言

什么是 Airflow 中的 Secrets Backend？

为什么使用 Secrets Backend？

Secrets Backend 搜索路径

默认搜索路径

定制搜索路径

Secrets Backend 配置

其他配置示例

支持的核心后端

社区提供的 Secrets Backend

自建 Secrets Backend

创建自定义后端的步骤

创建自定义 Secrets Backend 的分步指南

适应非 Airflow 兼容的连接密钥格式

Airflow 的连接要求

适应不兼容密钥格式的挑战

适应密钥格式的策略

将自定义键映射到 Airflow 键

解析嵌套或层级数据

解码编码的密钥

捆绑密钥的自定义解析器

在自定义 Secrets Backend 中实现适配器

适应密钥格式

相关帖子

Apache Airflow Kerberos 集成

Apache Airflow 安全

使用公共接口与外部服务和应用程序集成

在反向代理后运行 Apache Airflow 负载均衡

Apache Airflow 血缘关系

Apache Airflow 故障排除

Apache Airflow DAG

Apache Airflow Providers Drill

使用 Apache Airflow 扩展机器学习工作流

Apache Airflow

订阅 Tpoint Tech

联系信息

关注我们

教程

面试题

在线编译器