Python 项目 secrets 管理：用 Infisical 或 Vault 保护 RAG 服务的 API 密钥

Python 项目 secrets 管理：用 Infisical 或 Vault 保护 RAG 服务的 API 密钥

一、深度引言与场景痛点

大家好，我是赵咕咕。

来，看一个真实的代码片段，我猜至少 60% 的人这么干过：

# config.py
OPENAI_API_KEY = "sk-proj-xxxxxxxxxxxxxxxxxxxx"
QDRANT_URL = "https://qdrant.example.com"
QDRANT_API_KEY = "qdrant-api-key-here"
REDIS_PASSWORD = "my-redis-password"

或者是"好一点"的版本：

import os
from dotenv import load_dotenv
load_dotenv()
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")

这有问题吗？对于个人项目，没问题。对于团队协作的生产项目，问题大了。

一个典型的 RAG 服务需要哪些 secrets？OpenAI API Key、Embedding 服务 Key、向量数据库密码、Redis 密码、对象存储 Access Key、Slack Bot Token、GitHub Token……七八个密钥散落在 .env 文件里，团队成员之间靠 Slack 传来传去，有人不小心 git commit 了 .env，有人离职了密钥还在用，有人把生产环境的密钥写到了测试环境。

这不是"不够安全"，这是根本就没有安全管理。

这篇文章，我聊聊生产级 Python 项目的 secrets 管理方案。核心思路是用 Infisical 做日常开发团队的首选，用 HashiCorp Vault 做高安全要求场景的备选，以及如何在代码层面写出一个优雅的 secrets 访问层。

二、底层机制与原理深度剖析

2.1 一个合格的 Secrets 管理系统需要解决什么？

六个核心能力，缺一不可：

2.2 Infisical vs Vault：选型维度

简单粗暴的结论：

• 5-50 人团队，做 RAG/Agent 服务 → Infisical，够用而且快。
• 100+ 人组织，金融/医疗/安全合规要求 → Vault，别省这个麻烦。

2.3 密钥的生命周期

一个 API Key 从创建到退役，应该经历以下阶段：

三、生产级代码实现

下面实现一个支持 Infisical 和 Vault 双后端的 Secrets Provider，通过配置切换：

import asyncio
import logging
import os
from abc import ABC, abstractmethod
from enum import Enum
from typing import Any

logger = logging.getLogger(__name__)

class SecretBackend(Enum):
INFISICAL = "infisical"
VAULT = "vault"
ENV = "env" # 降级方案：读环境变量

class SecretNotFoundError(Exception):
"""密钥不存在异常。"""
pass

class BackendUnavailableError(Exception):
"""后端不可用异常。"""
pass

# ── 抽象接口 ───────────────────────────────────────────

class SecretBackendBase(ABC):
"""Secrets 后端抽象基类。"""

@abstractmethod
async def get(self, key: str) -> str:
"""获取单个密钥。"""
…

@abstractmethod
async def get_all(self, prefix: str = "") -> dict[str, str]:
"""获取前缀下的所有密钥。"""
…

@abstractmethod
async def health_check(self) -> bool:
"""检查后端可用性。"""
…

# ── Env 降级实现 ───────────────────────────────────────

class EnvBackend(SecretBackendBase):
"""从环境变量读取密钥，作为降级方案。"""

async def get(self, key: str) -> str:
value = os.getenv(key)
if value is None:
raise SecretNotFoundError(f"环境变量 {key} 未设置")
return value

async def get_all(self, prefix: str = "") -> dict[str, str]:
result = {}
for k, v in os.environ.items():
if k.startswith(prefix):
result[k] = v
return result

async def health_check(self) -> bool:
return True # 环境变量始终可用

# ── Infisical 实现 ─────────────────────────────────────

class InfisicalBackend(SecretBackendBase):
"""Infisical 后端实现。"""

def __init__(
self,
token: str | None = None,
project_id: str | None = None,
environment: str = "dev",
base_url: str = "https://app.infisical.com",
):
self._token = token or os.getenv("INFISICAL_TOKEN", "")
self._project_id = project_id or os.getenv("INFISICAL_PROJECT_ID", "")
self._environment = environment
self._base_url = base_url
self._cache: dict[str, str] = {}
self._cache_ttl = 300 # 5 分钟缓存

async def get(self, key: str) -> str:
if key in self._cache:
return self._cache[key]

try:
value = await self._fetch_secret(key)
self._cache[key] = value
return value
except Exception as e:
logger.error("Infisical 获取密钥 %s 失败: %s", key, e)
raise BackendUnavailableError(f"Infisical 不可用: {e}") from e

async def get_all(self, prefix: str = "") -> dict[str, str]:
try:
all_secrets = await self._fetch_all_secrets()
if prefix:
return {k: v for k, v in all_secrets.items() if k.startswith(prefix)}
return all_secrets
except Exception as e:
logger.error("Infisical 批量获取密钥失败: %s", e)
return {}

async def health_check(self) -> bool:
try:
# 简单探活：尝试读取一个已知存在的密钥或调用 status API
import aiohttp
async with aiohttp.ClientSession() as session:
url = f"{self._base_url}/api/v2/status"
async with session.get(url, timeout=aiohttp.ClientTimeout(total=5)) as resp:
return resp.status == 200
except Exception:
return False

async def _fetch_secret(self, key: str) -> str:
"""通过 Infisical REST API 获取单个密钥。"""
import aiohttp
headers = {
"Authorization": f"Bearer {self._token}",
"Content-Type": "application/json",
}
params = {
"workspaceId": self._project_id,
"environment": self._environment,
"secretName": key,
}
async with aiohttp.ClientSession() as session:
async with session.get(
f"{self._base_url}/api/v3/secrets/{key}",
headers=headers,
params=params,
timeout=aiohttp.ClientTimeout(total=10),
) as resp:
if resp.status == 404:
raise SecretNotFoundError(f"密钥 {key} 在 Infisical 中不存在")
resp.raise_for_status()
data = await resp.json()
secret_value = data.get("secret", {}).get("secretValue", "")
if not secret_value:
raise SecretNotFoundError(f"密钥 {key} 值为空")
return secret_value

async def _fetch_all_secrets(self) -> dict[str, str]:
import aiohttp
headers = {"Authorization": f"Bearer {self._token}"}
params = {
"workspaceId": self._project_id,
"environment": self._environment,
}
async with aiohttp.ClientSession() as session:
async with session.get(
f"{self._base_url}/api/v3/secrets",
headers=headers,
params=params,
timeout=aiohttp.ClientTimeout(total=15),
) as resp:
resp.raise_for_status()
data = await resp.json()
secrets = data.get("secrets", [])
return {
s["secretName"]: s.get("secretValue", "")
for s in secrets
}

# ── Vault 实现 ─────────────────────────────────────────

class VaultBackend(SecretBackendBase):
"""HashiCorp Vault 后端实现。"""

def __init__(
self,
url: str | None = None,
token: str | None = None,
mount_point: str = "secret",
base_path: str = "rag-service",
):
self._url = url or os.getenv("VAULT_ADDR", "http://localhost:8200")
self._token = token or os.getenv("VAULT_TOKEN", "")
self._mount_point = mount_point
self._base_path = base_path
self._cache: dict[str, str] = {}

async def get(self, key: str) -> str:
if key in self._cache:
return self._cache[key]

try:
value = await self._read_vault(key)
self._cache[key] = value
return value
except Exception as e:
logger.error("Vault 获取密钥 %s 失败: %s", key, e)
raise BackendUnavailableError(f"Vault 不可用: {e}") from e

async def get_all(self, prefix: str = "") -> dict[str, str]:
try:
all_data = await self._list_vault("")
if prefix:
return {k: v for k, v in all_data.items() if k.startswith(prefix)}
return all_data
except Exception as e:
logger.error("Vault 批量获取失败: %s", e)
return {}

async def health_check(self) -> bool:
try:
import aiohttp
async with aiohttp.ClientSession() as session:
async with session.get(
f"{self._url}/v1/sys/health",
headers={"X-Vault-Token": self._token},
timeout=aiohttp.ClientTimeout(total=5),
) as resp:
return resp.status == 200
except Exception:
return False

async def _read_vault(self, key: str) -> str:
"""从 Vault KV v2 读取密钥。"""
import aiohttp
# KV v2 的路径格式：/v1/{mount_point}/data/{path}
path = f"{self._url}/v1/{self._mount_point}/data/{self._base_path}"
headers = {"X-Vault-Token": self._token}
async with aiohttp.ClientSession() as session:
async with session.get(
path, headers=headers, timeout=aiohttp.ClientTimeout(total=10)
) as resp:
if resp.status == 404:
raise SecretNotFoundError(f"密钥路径 {path} 不存在")
resp.raise_for_status()
data = await resp.json()
secrets = data.get("data", {}).get("data", {})
if key not in secrets:
raise SecretNotFoundError(f"密钥 {key} 在 Vault 中不存在")
return secrets[key]

async def _list_vault(self, sub_path: str) -> dict[str, str]:
import aiohttp
path = f"{self._url}/v1/{self._mount_point}/data/{self._base_path}/{sub_path}".rstrip("/")
headers = {"X-Vault-Token": self._token}
async with aiohttp.ClientSession() as session:
async with session.get(
path, headers=headers, timeout=aiohttp.ClientTimeout(total=10)
) as resp:
resp.raise_for_status()
data = await resp.json()
return data.get("data", {}).get("data", {})

# ── 统一的 Secrets Provider ────────────────────────────

class SecretsProvider:
"""统一密钥访问入口，支持多后端 + 降级。"""

def __init__(
self,
primary: SecretBackendBase | None = None,
fallback: SecretBackendBase | None = None,
):
self._primary = primary or self._detect_backend()
self._fallback = fallback or EnvBackend()

async def get(self, key: str) -> str:
"""获取密钥，主后端失败时降级。"""
try:
if await self._primary.health_check():
return await self._primary.get(key)
except Exception as e:
logger.warning("主后端 %s 不可用，降级到 fallback: %s",
type(self._primary).__name__, e)

try:
return await self._fallback.get(key)
except SecretNotFoundError:
raise SecretNotFoundError(
f"密钥 {key} 在所有后端均未找到。"
f"主: {type(self._primary).__name__}, 降级: {type(self._fallback).__name__}"
)

async def load_config(self) -> dict[str, str]:
"""加载所有密钥，用于初始化应用配置。"""
secrets = {}

# 定义需要的密钥列表
required_keys = [
"OPENAI_API_KEY",
"QDRANT_URL",
"QDRANT_API_KEY",
"REDIS_URL",
"REDIS_PASSWORD",
]

missing = []
for key in required_keys:
try:
secrets[key] = await self.get(key)
except SecretNotFoundError:
missing.append(key)

if missing:
logger.warning("以下密钥未配置: %s，服务可能功能受限", ", ".join(missing))

return secrets

@staticmethod
def _detect_backend() -> SecretBackendBase:
"""根据环境变量自动检测后端。"""
backend = os.getenv("SECRETS_BACKEND", "env")
if backend == "infisical":
return InfisicalBackend(
environment=os.getenv("INFISICAL_ENV", "prod"),
)
elif backend == "vault":
return VaultBackend()
return EnvBackend()

# ── 使用示例 ────────────────────────────────────────────

async def init_rag_service() -> dict[str, Any]:
"""RAG 服务初始化示例。"""
provider = SecretsProvider()

try:
config = await provider.load_config()

# config 现在是脱敏后的密钥字典，可以直接用来初始化服务
return {
"openai_key": config.get("OPENAI_API_KEY", ""),
"qdrant_url": config.get("QDRANT_URL", ""),
"qdrant_key": config.get("QDRANT_API_KEY", ""),
"redis_url": config.get("REDIS_URL", ""),
}

except SecretNotFoundError as e:
logger.critical("关键密钥缺失，服务无法启动: %s", e)
raise

if __name__ == "__main__":
result = asyncio.run(init_rag_service())
# 注意：生产环境不要打印密钥值
print("已加载密钥数量:", len(result))

设计关键点：

• 策略模式：SecretBackendBase 抽象接口 + Infisical/Vault/Env 三种实现。新增后端只需要实现三个方法。
• 自动降级：主后端不可用时自动 fallback 到环境变量。生产用 Infisical，本地开发用 .env。不需要改代码。
• 环境变量驱动：通过 SECRETS_BACKEND 环境变量切换后端。Docker Compose 里改一行就能切换。
• 缓存：每个后端内部有内存缓存，避免每次都调外部 API。缓存 TTL 由具体后端实现控制。
• 绝不打印密钥：日志里只记录密钥数量，不记录值。初始化失败时只报 missing 的 key name，不 dump 整个配置字典。

四、边界分析与架构权衡

4.1 要不要用 Sidecar 模式？

Kubernetes 环境下，Vault 有官方 sidecar injector：一个 sidecar 容器在 Pod 启动时从 Vault 拉取密钥写入共享 Volume，应用容器从文件读取。优点是不需要应用感知 Vault SDK，缺点是多了一个 sidecar 容器。

对于 Python 项目，我建议直接用 SDK 而不是 sidecar。原因：

• Python 生态的 aiohttp + async 模式调用 Vault API 开销很小
• Sidecar 增加了部署复杂度（两个容器共享生命周期）
• SDK 模式可以做更精细的错误处理和降级

4.2 CI/CD 中的密钥处理

CI/CD Pipeline 也需要密钥（推送 Docker 镜像、部署到 K8s），但这些密钥不应该写死在 .github/workflows/*.yml 里。

推荐方案：

• GitHub Actions：用 Repository Secrets 或直接接 Infisical GitHub App
• Docker Build：用 BuildKit 的 –secret 参数挂载密钥到构建上下文，不写进镜像层
• K8s Deploy：用 External Secrets Operator 自动同步 Infisical/Vault 的密钥到 K8s Secret

4.3 热重载 vs 冷重启

密钥轮转后，正在运行的 RAG 服务怎么拿到新密钥？

• 冷重启（推荐起步方案）：Key 轮转 → 通知运维 → 手动重启服务 → 启动时重新拉取密钥。简单粗暴但有中断。
• 热重载（进阶）：定期（如每 5 分钟）从 Secrets Manager 刷新密钥缓存。对 OpenAI Key 这种"服务启动时建立连接"的密钥，还需要重建连接池。
• 协商方案：不频繁变更的密钥（如 OpenAI Key）用冷重启。需要动态轮转的密钥（如数据库密码）用热重载。

4.4 开发体验不能丢

上 Secrets Manager 之后最常见的抱怨是"本地开发好麻烦，每次都要连外网"。

务实的做法：

• 本地开发：SECRETS_BACKEND=env，用 .env 文件。.env 加入 .gitignore。
• 测试/CI 环境：用 Infisical 的 Dev 环境，密钥值可以是非敏感测试用的 stub。
• 生产环境：SECRETS_BACKEND=infisical 或 vault。永远不落盘。

这个三层模型既不牺牲安全性，也不降低开发效率。

五、总结

Secrets 管理这件事，说到底是把密钥从"人管"变成"系统管"。

.env 文件管密钥是"人管"——人记住在哪、人负责复制、人记得更新。人一定会出错。一旦团队超过 3 个人，这种模式就不行了。

我的建议路线图：

这一步做好了，你从此不用在 Slack 里发"OpenAI Key 是多少？"这种消息。也不用担心谁把密钥写进了仓库。它应该是一种安心感——你知道密钥在安全的地方，轮转在自动发生，审计日志详细可查。

下一篇预告：RAG 在跨境电商选品中的应用，多语言产品描述怎么向量化和做竞品分析。