Implementing Hashicorp Vault Dynamic Secrets

实施 HashiCorp Vault 动态密钥

适用场景

• 应用程序使用存储在配置文件或环境变量中的静态数据库凭据
• AWS IAM 访问密钥是长期有效的并在服务间共享
• 需要通过生成短期按需密钥来消除凭据蔓延
• 合规要求强制凭据轮换（PCI-DSS 要求 8、NIST 800-53 IA-5）
• 实施零信任密钥管理，凭据不会静态存储
• 从手动凭据管理迁移到自动化密钥生命周期

不适用于存储无法动态生成的静态密钥（改用 Vault 的 KV 密钥引擎）；动态密钥适用于可以在目标系统上以编程方式创建和撤销的凭据。

前置条件

• HashiCorp Vault 1.15+（社区版或企业版）
• Vault 服务器已初始化并解封，并配置了自动解封（AWS KMS、Azure Key Vault 或 Transit）
• 目标数据库系统具有管理员凭据，供 Vault 创建/撤销动态账户
• AWS IAM 账户具有创建/删除 IAM 用户和访问密钥的权限
• 从 Vault 到所有目标系统的网络连接
• 已为使用应用程序配置了 Vault 策略和认证方法

工作流程

步骤 1：部署和配置 Vault 服务器

使用生产级配置初始化 Vault：

# vault-config.hcl - 生产 Vault 服务器配置
storage "raft" {
  path    = "/opt/vault/data"
  node_id = "vault-1"

  retry_join {
    leader_api_addr = "https://vault-2.corp.local:8200"
  }
  retry_join {
    leader_api_addr = "https://vault-3.corp.local:8200"
  }
}

listener "tcp" {
  address       = "0.0.0.0:8200"
  tls_cert_file = "/opt/vault/tls/vault-cert.pem"
  tls_key_file  = "/opt/vault/tls/vault-key.pem"
}

seal "awskms" {
  region     = "us-east-1"
  kms_key_id = "alias/vault-unseal-key"
}

api_addr      = "https://vault-1.corp.local:8200"
cluster_addr  = "https://vault-1.corp.local:8201"

telemetry {
  prometheus_retention_time = "24h"
  disable_hostname          = true
}

ui = true

# 初始化 Vault 集群
vault operator init -key-shares=5 -key-threshold=3

# 启用审计日志记录
vault audit enable file file_path=/var/log/vault/audit.log

# 为应用程序启用 AppRole 认证
vault auth enable approle

# 为数据库密钥使用方创建策略
vault policy write db-consumer - <<EOF
# 允许读取动态数据库凭据
path "database/creds/app-readonly" {
  capabilities = ["read"]
}
path "database/creds/app-readwrite" {
  capabilities = ["read"]
}

# 允许续约和撤销自身的租约
path "sys/leases/renew" {
  capabilities = ["update"]
}
path "sys/leases/revoke" {
  capabilities = ["update"]
}

# 允许读取自身的令牌信息
path "auth/token/lookup-self" {
  capabilities = ["read"]
}
EOF

# 为应用程序创建 AppRole
vault write auth/approle/role/webapp \
    token_policies="db-consumer" \
    token_ttl=1h \
    token_max_ttl=4h \
    secret_id_ttl=720h \
    secret_id_num_uses=0

步骤 2：配置数据库密钥引擎

为 PostgreSQL 和 MySQL 设置动态凭据生成：

# 启用数据库密钥引擎
vault secrets enable database

# 配置 PostgreSQL 连接
vault write database/config/production-postgres \
    plugin_name=postgresql-database-plugin \
    allowed_roles="app-readonly,app-readwrite,app-admin" \
    connection_url="postgresql://{{username}}:{{password}}@db-primary.corp.local:5432/appdb?sslmode=verify-full" \
    username="vault_admin" \
    password="$VAULT_DB_PASSWORD" \
    password_authentication="scram-sha-256"

# 轮换根凭据，使 Vault 独家管理它们
vault write -force database/rotate-root/production-postgres

# 创建只读角色（TTL：1 小时，最大 24 小时）
vault write database/roles/app-readonly \
    db_name=production-postgres \
    creation_statements="CREATE ROLE \"{{name}}\" WITH LOGIN PASSWORD '{{password}}' VALID UNTIL '{{expiration}}'; \
        GRANT SELECT ON ALL TABLES IN SCHEMA public TO \"{{name}}\"; \
        ALTER DEFAULT PRIVILEGES IN SCHEMA public GRANT SELECT ON TABLES TO \"{{name}}\";" \
    revocation_statements="REVOKE ALL PRIVILEGES ON ALL TABLES IN SCHEMA public FROM \"{{name}}\"; \
        DROP ROLE IF EXISTS \"{{name}}\";" \
    default_ttl="1h" \
    max_ttl="24h"

# 创建读写角色（TTL：30 分钟，最大 8 小时）
vault write database/roles/app-readwrite \
    db_name=production-postgres \
    creation_statements="CREATE ROLE \"{{name}}\" WITH LOGIN PASSWORD '{{password}}' VALID UNTIL '{{expiration}}'; \
        GRANT SELECT, INSERT, UPDATE, DELETE ON ALL TABLES IN SCHEMA public TO \"{{name}}\"; \
        ALTER DEFAULT PRIVILEGES IN SCHEMA public GRANT SELECT, INSERT, UPDATE, DELETE ON TABLES TO \"{{name}}\";" \
    revocation_statements="REVOKE ALL PRIVILEGES ON ALL TABLES IN SCHEMA public FROM \"{{name}}\"; \
        DROP ROLE IF EXISTS \"{{name}}\";" \
    default_ttl="30m" \
    max_ttl="8h"

# 配置 MySQL 连接
vault write database/config/production-mysql \
    plugin_name=mysql-database-plugin \
    allowed_roles="mysql-readonly,mysql-readwrite" \
    connection_url="{{username}}:{{password}}@tcp(mysql-primary.corp.local:3306)/" \
    username="vault_admin" \
    password="$VAULT_MYSQL_PASSWORD"

vault write database/roles/mysql-readonly \
    db_name=production-mysql \
    creation_statements="CREATE USER '{{name}}'@'%' IDENTIFIED BY '{{password}}'; \
        GRANT SELECT ON appdb.* TO '{{name}}'@'%';" \
    revocation_statements="DROP USER IF EXISTS '{{name}}'@'%';" \
    default_ttl="1h" \
    max_ttl="24h"

# 测试动态凭据生成
echo "测试 PostgreSQL 动态凭据："
vault read database/creds/app-readonly
# 返回：username=v-approle-app-read-xxxxx, password=<随机>, lease_id=database/creds/app-readonly/xxxxx

步骤 3：配置 AWS 密钥引擎

生成短暂的 AWS IAM 凭据：

# 启用 AWS 密钥引擎
vault secrets enable aws

# 使用根凭据配置 AWS 密钥引擎
vault write aws/config/root \
    access_key="$AWS_ACCESS_KEY_ID" \
    secret_key="$AWS_SECRET_ACCESS_KEY" \
    region="us-east-1"

# 配置租约设置
vault write aws/config/lease \
    lease="30m" \
    lease_max="1h"

# 创建 S3 只读访问的 IAM 用户角色
vault write aws/roles/s3-readonly \
    credential_type=iam_user \
    policy_document=-<<EOF
{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "s3:GetObject",
        "s3:ListBucket",
        "s3:GetBucketLocation"
      ],
      "Resource": [
        "arn:aws:s3:::app-data-bucket",
        "arn:aws:s3:::app-data-bucket/*"
      ]
    }
  ]
}
EOF

# 创建 EC2 管理的承担角色（优于 IAM 用户）
vault write aws/roles/ec2-admin \
    credential_type=assumed_role \
    role_arns="arn:aws:iam::123456789012:role/VaultEC2AdminRole" \
    default_sts_ttl="30m" \
    max_sts_ttl="1h"

# 创建跨账户访问的联合令牌角色
vault write aws/roles/cross-account-readonly \
    credential_type=federation_token \
    policy_document=-<<EOF
{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "sts:AssumeRole"
      ],
      "Resource": "arn:aws:iam::987654321098:role/CrossAccountReadOnly"
    }
  ]
}
EOF

# 测试 AWS 动态凭据
echo "测试 AWS STS 凭据："
vault read aws/creds/ec2-admin
# 返回：access_key, secret_key, security_token，TTL 为 30 分钟

步骤 4：配置 PKI 密钥引擎以动态颁发证书

按需生成短期 TLS 证书：

# 为根 CA 启用 PKI 密钥引擎
vault secrets enable -path=pki pki
vault secrets tune -max-lease-ttl=87600h pki

# 生成根 CA 证书
vault write pki/root/generate/internal \
    common_name="Corp Internal Root CA" \
    ttl=87600h \
    key_type=ec \
    key_bits=384

# 为中间 CA 启用 PKI
vault secrets enable -path=pki_int pki
vault secrets tune -max-lease-ttl=43800h pki_int

# 生成中间 CA CSR
vault write pki_int/intermediate/generate/internal \
    common_name="Corp Intermediate CA" \
    key_type=ec \
    key_bits=256

# 用根 CA 签署中间 CA
vault write pki/root/sign-intermediate \
    csr=@intermediate.csr \
    format=pem_bundle \
    ttl=43800h

# 配置颁发 URL
vault write pki_int/config/urls \
    issuing_certificates="https://vault.corp.local:8200/v1/pki_int/ca" \
    crl_distribution_points="https://vault.corp.local:8200/v1/pki_int/crl"

# 为 Web 服务器证书创建角色（TTL：30 天）
vault write pki_int/roles/web-server \
    allowed_domains="corp.local,internal.corp.com" \
    allow_subdomains=true \
    max_ttl=720h \
    key_type=ec \
    key_bits=256 \
    require_cn=true \
    enforce_hostnames=true

# 为服务网格证书创建角色（TTL：24 小时）
vault write pki_int/roles/service-mesh \
    allowed_domains="service.consul" \
    allow_subdomains=true \
    max_ttl=24h \
    key_type=ec \
    key_bits=256 \
    allow_ip_sans=true \
    server_flag=true \
    client_flag=true

# 颁发证书
vault write pki_int/issue/web-server \
    common_name="api.corp.local" \
    alt_names="api.internal.corp.com" \
    ttl=720h

步骤 5：将应用程序与 Vault 集成

配置应用程序使用动态密钥：

"""
与 HashiCorp Vault 集成用于动态数据库凭据的应用程序示例。
使用带有自动租约续约的 hvac Python 客户端。
"""
import hvac
import threading
import time
import logging

class VaultDynamicCredentialManager:
    def __init__(self, vault_addr, role_id, secret_id):
        self.client = hvac.Client(url=vault_addr)
        self.role_id = role_id
        self.secret_id = secret_id
        self.logger = logging.getLogger("vault_credentials")
        self._current_creds = None
        self._lease_id = None
        self._renewal_thread = None
        self._stop_event = threading.Event()

    def authenticate(self):
        """使用 AppRole 向 Vault 进行认证。"""
        response = self.client.auth.approle.login(
            role_id=self.role_id,
            secret_id=self.secret_id
        )
        self.client.token = response["auth"]["client_token"]
        self.logger.info("已通过 AppRole 向 Vault 认证")

    def get_database_credentials(self, role="app-readonly"):
        """从 Vault 请求动态数据库凭据。"""
        self.authenticate()

        response = self.client.secrets.database.generate_credentials(
            name=role
        )

        self._current_creds = {
            "username": response["data"]["username"],
            "password": response["data"]["password"],
        }
        self._lease_id = response["lease_id"]
        lease_duration = response["lease_duration"]

        self.logger.info(
            f"已获取动态凭据：user={self._current_creds['username']}, "
            f"lease={self._lease_id}, ttl={lease_duration}s"
        )

        # 启动后台租约续约
        self._start_renewal(lease_duration)

        return self._current_creds

    def _start_renewal(self, lease_duration):
        """启动后台线程在到期前续约租约。"""
        if self._renewal_thread and self._renewal_thread.is_alive():
            self._stop_event.set()
            self._renewal_thread.join()

        self._stop_event.clear()
        renewal_interval = lease_duration * 0.7  # 在 TTL 的 70% 时续约

        def renew_loop():
            while not self._stop_event.wait(renewal_interval):
                try:
                    self.client.sys.renew_lease(
                        lease_id=self._lease_id,
                        increment=lease_duration
                    )
                    self.logger.info(f"已续约租约：{self._lease_id}")
                except hvac.exceptions.InvalidRequest:
                    self.logger.warning("租约已过期，获取新凭据")
                    self.get_database_credentials()
                    break
                except Exception as e:
                    self.logger.error(f"租约续约失败：{e}")

        self._renewal_thread = threading.Thread(target=renew_loop, daemon=True)
        self._renewal_thread.start()

    def revoke_credentials(self):
        """显式撤销当前动态凭据。"""
        if self._lease_id:
            self._stop_event.set()
            self.client.sys.revoke_lease(self._lease_id)
            self.logger.info(f"已撤销租约：{self._lease_id}")
            self._current_creds = None
            self._lease_id = None

    def get_aws_credentials(self, role="s3-readonly"):
        """从 Vault 请求动态 AWS 凭据。"""
        self.authenticate()

        response = self.client.secrets.aws.generate_credentials(
            name=role
        )

        return {
            "access_key": response["data"]["access_key"],
            "secret_key": response["data"]["secret_key"],
            "security_token": response["data"].get("security_token"),
            "lease_id": response["lease_id"],
            "ttl": response["lease_duration"]
        }

# 使用示例
vault_mgr = VaultDynamicCredentialManager(
    vault_addr="https://vault.corp.local:8200",
    role_id="<approle-role-id>",
    secret_id="<approle-secret-id>"
)

db_creds = vault_mgr.get_database_credentials("app-readonly")
# 使用 db_creds["username"] 和 db_creds["password"] 进行数据库连接

步骤 6：监控 Vault 操作和租约管理

跟踪动态密钥使用情况和租约生命周期：

# 监控活跃租约
vault list sys/leases/lookup/database/creds/app-readonly
vault list sys/leases/lookup/aws/creds/s3-readonly

# 检查租约详情
vault write sys/leases/lookup lease_id="database/creds/app-readonly/abcd1234"

# 撤销特定路径的所有租约（紧急凭据轮换）
vault lease revoke -prefix database/creds/app-readonly

# 用于监控的 Vault 指标（Prometheus 格式）
# 需要监控的关键指标：
# vault.expire.num_leases - 总活跃租约数
# vault.expire.revoke - 每秒租约撤销次数
# vault.secret.kv.count - 存储的密钥总数
# vault.runtime.alloc_bytes - 内存分配

# 配置 Vault 审计日志分析
cat > vault_audit_monitor.sh << 'SCRIPT'
#!/bin/bash
# 监控 Vault 审计日志中的可疑活动

AUDIT_LOG="/var/log/vault/audit.log"

# 统计每小时的凭据请求
echo "=== 动态凭据请求（最近 1 小时）==="
jq -r 'select(.type == "response" and .request.path | startswith("database/creds/")) |
    "\(.time) \(.request.path) \(.auth.display_name)"' \
    "$AUDIT_LOG" | tail -100

# 检测异常凭据请求模式
echo ""
echo "=== 高频凭据使用者 ==="
jq -r 'select(.type == "request" and .request.path | startswith("database/creds/")) |
    .auth.display_name' \
    "$AUDIT_LOG" | sort | uniq -c | sort -rn | head -10

# 检查认证失败尝试
echo ""
echo "=== 认证失败尝试 ==="
jq -r 'select(.type == "response" and .error != null and
    .request.path | startswith("auth/")) |
    "\(.time) \(.request.path) \(.error)"' \
    "$AUDIT_LOG" | tail -20
SCRIPT
chmod +x vault_audit_monitor.sh

核心概念

工具与系统

• HashiCorp Vault：提供动态密钥、加密即服务和基于身份的访问控制的密钥管理平台
• Vault Agent：处理 Vault 认证、令牌续约和应用程序密钥缓存的 Sidecar 进程
• Vault Secrets Operator：将 Vault 密钥同步到 Kubernetes Secrets 以供 Pod 使用的 Kubernetes Operator
• hvac：用于 HashiCorp Vault API 操作的 Python 客户端库

常见场景

场景：消除微服务中的静态数据库凭据

背景：50 个微服务共享 3 个静态 PostgreSQL 凭据，这些凭据存储在 Kubernetes 部署的环境变量中。凭据泄露需要同时轮换所有 50 个服务。

处理方式：

1. 在 Kubernetes 中使用 Raft 存储部署 3 节点 HA 集群的 Vault
2. 使用管理员凭据配置数据库密钥引擎的 PostgreSQL 连接
3. 使用最小权限 SQL 授权为每个服务创建 Vault 角色
4. 部署 Vault Secrets Operator 将动态凭据注入 Pod 环境变量
5. 更新应用程序连接逻辑以通过租约续约处理凭据轮换
6. 轮换 Vault 根凭据以删除人对管理员密码的知晓
7. 监控租约生命周期并为续约失败设置告警

注意事项：

• 不在应用程序连接池中处理凭据轮换（使用过期凭据的连接会失败）
• 将 TTL 设置得太短会对数据库造成过多的凭据生成负载
• 不配置适当的撤销语句会导致租约过期后遗留孤儿数据库用户
• 在没有 HA 的情况下运行 Vault 会为所有应用程序认证创建单点故障

输出格式

HASHICORP VAULT 动态密钥报告
=========================================
Vault 版本：      1.16.2 Enterprise
集群状态：        HA 活跃（3 个节点）
封印类型：        AWS KMS（自动解封）

密钥引擎
database/：       PostgreSQL, MySQL（2 个连接）
aws/：            IAM 用户、承担角色、联合令牌
pki_int/：        内部 CA（EC P-256）

动态凭据指标（最近 24 小时）
生成的总凭据：          4,287
  数据库（PostgreSQL）：2,891
  数据库（MySQL）：      543
  AWS STS：             612
  PKI 证书：            241

活跃租约
总活跃：                      387
  database/creds/app-readonly：198
  database/creds/app-readwrite：89
  aws/creds/s3-readonly：       67
  pki_int/issue/web-server：    33

租约生命周期
平均 TTL：              45 分钟
续约（24 小时）：       12,847
撤销（24 小时）：       3,901
过期（未续约）：        12

安全
认证失败尝试（24 小时）：3
根凭据已轮换：           是（所有数据库）
审计日志记录：           已启用（文件 + syslog）
策略违规（24 小时）：    7（权限拒绝）