跳至主要內容

密钥与凭据管理: 与Vault等 secrets manager 集成

老马啸西风2025/8/30大约 11 分钟CICDCICD

在现代软件开发和部署过程中，密钥与凭据管理是安全防护体系中的关键环节。不当的密钥管理可能导致严重的安全漏洞，包括数据泄露、系统入侵和合规性违规。通过与HashiCorp Vault等专业的secrets manager集成，CI/CD平台能够实现安全、动态、可审计的密钥管理。本文将深入探讨密钥管理的重要性、Vault集成实践以及最佳安全策略。

密钥管理的重要性与挑战

密钥与凭据是系统安全的核心要素，涵盖了数据库密码、API密钥、SSH密钥、证书等多种敏感信息。有效的密钥管理不仅能防止未授权访问，还能确保系统的合规性和可审计性。

密钥管理面临的挑战

1. 静态密钥的风险

传统的密钥管理方式通常将密钥硬编码在代码中或存储在配置文件里，这种方式存在显著的安全风险：

  • 代码泄露风险：一旦代码库被泄露，所有硬编码的密钥都会暴露
  • 版本控制问题：密钥变更会被记录在版本控制系统中，增加了泄露风险
  • 共享密钥问题：多个环境使用相同密钥，一旦泄露影响范围广泛

2. 密钥轮换的复杂性

定期轮换密钥是安全最佳实践，但在复杂的系统中实施密钥轮换面临诸多挑战：

  • 协调难度：需要同时更新多个系统和服务中的密钥
  • 停机风险：密钥更新过程中可能导致服务中断
  • 回滚困难：密钥更新失败时难以快速回滚

3. 访问控制与审计

在分布式系统中，确保密钥的访问控制和审计追踪是一项复杂任务：

  • 权限管理：需要精细化控制不同用户和服务对密钥的访问权限
  • 审计追踪：需要完整记录密钥的访问历史和使用情况
  • 合规要求：需要满足各种安全标准和合规性要求

密钥管理最佳实践

1. 零信任原则

采用零信任安全模型，确保密钥管理遵循以下原则：

  • 最小权限：每个服务和用户只能访问其必需的密钥
  • 动态密钥：使用短期有效的动态密钥替代长期静态密钥
  • 多因素认证：对密钥访问实施多因素认证

2. 密钥生命周期管理

建立完整的密钥生命周期管理流程：

  • 生成：使用安全的随机数生成器创建强密钥
  • 存储：使用专用的密钥存储系统
  • 分发：安全地将密钥分发给授权的服务和用户
  • 轮换：定期或按需轮换密钥
  • 撤销：及时撤销不再需要或已泄露的密钥
  • 销毁：安全地销毁废弃的密钥

HashiCorp Vault集成实践

HashiCorp Vault是一个开源的密钥管理解决方案，提供了安全的密钥存储、动态密钥生成和细粒度的访问控制功能。通过与CI/CD平台集成，Vault能够为应用提供安全的密钥管理服务。

Vault核心概念

1. Secret Engines

Vault通过Secret Engines提供不同类型的密钥存储和生成功能：

  • KV Secrets Engine：存储静态密钥，如数据库密码、API密钥
  • Dynamic Secrets Engine：动态生成临时密钥，如数据库凭证、SSH密钥
  • PKI Secrets Engine：生成和管理SSL/TLS证书
  • Transit Secrets Engine：提供加密即服务功能

2. Authentication Methods

Vault支持多种认证方式：

  • Token Authentication：基于令牌的认证
  • AppRole Authentication：面向应用的角色认证
  • Kubernetes Authentication：与Kubernetes集成的认证
  • LDAP Authentication：与LDAP集成的认证
  • JWT/OIDC Authentication：基于JWT/OIDC的认证

3. Policies

Vault通过策略控制用户和服务对密钥的访问权限：

  • 路径基础策略：基于密钥路径控制访问权限
  • 能力控制：控制用户对密钥的操作权限（读、写、删除等）
  • 元数据控制：控制用户对密钥元数据的访问

Vault部署与配置

1. 高可用部署

为了确保Vault服务的高可用性，建议采用以下部署方案：

# Vault高可用配置示例
storage "consul" {
  address = "127.0.0.1:8500"
  path    = "vault"
}

listener "tcp" {
  address     = "0.0.0.0:8200"
  tls_disable = 1
}

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

api_addr = "http://127.0.0.1:8200"
cluster_addr = "https://127.0.0.1:8201"
ui = true

2. 初始化与解封

Vault需要初始化和解封才能正常工作：

#!/bin/bash
# vault-init.sh - Vault初始化脚本

set -e

log() {
    echo "[$(date '+%Y-%m-%d %H:%M:%S')] $1"
}

# 初始化Vault
log "Initializing Vault..."
init_output=$(vault operator init -key-shares=5 -key-threshold=3 -format=json)

# 保存恢复密钥和根令牌
echo "$init_output" | jq -r '.unseal_keys_b64[]' > unseal-keys.txt
echo "$init_output" | jq -r '.root_token' > root-token.txt

log "Vault initialized successfully"
log "Unseal keys saved to unseal-keys.txt"
log "Root token saved to root-token.txt"

# 解封Vault（需要3个密钥）
log "Unsealing Vault..."
vault operator unseal $(head -n 1 unseal-keys.txt)
vault operator unseal $(head -n 2 unseal-keys.txt | tail -n 1)
vault operator unseal $(head -n 3 unseal-keys.txt | tail -n 1)

log "Vault unsealed successfully"

# 使用根令牌登录
export VAULT_TOKEN=$(cat root-token.txt)
vault login $VAULT_TOKEN

log "Vault login successful"

3. 配置Secret Engines

根据应用需求配置不同的Secret Engines：

#!/bin/bash
# vault-configure-engines.sh - 配置Vault Secret Engines

set -e

log() {
    echo "[$(date '+%Y-%m-%d %H:%M:%S')] $1"
}

# 启用KV Secrets Engine
log "Enabling KV Secrets Engine..."
vault secrets enable -path=secret kv-v2

# 启用Database Secrets Engine
log "Enabling Database Secrets Engine..."
vault secrets enable database

# 配置PostgreSQL数据库连接
log "Configuring PostgreSQL connection..."
vault write database/config/my-postgresql \
    plugin_name=postgresql-database-plugin \
    allowed_roles="*" \
    connection_url="postgresql://vault:vault@localhost:5432/myapp?sslmode=disable"

# 创建数据库角色
log "Creating database role..."
vault write database/roles/myapp-role \
    db_name=my-postgresql \
    creation_statements="CREATE ROLE \"{{name}}\" WITH LOGIN PASSWORD '{{password}}' VALID UNTIL '{{expiration}}'; \
        GRANT SELECT ON ALL TABLES IN SCHEMA public TO \"{{name}}\";" \
    default_ttl="1h" \
    max_ttl="24h"

# 启用PKI Secrets Engine
log "Enabling PKI Secrets Engine..."
vault secrets enable pki

# 配置PKI
log "Configuring PKI..."
vault write pki/root/generate/internal \
    common_name="myapp.example.com" \
    ttl=87600h

vault write pki/config/urls \
    issuing_certificates="http://127.0.0.1:8200/v1/pki/ca" \
    crl_distribution_points="http://127.0.0.1:8200/v1/pki/crl"

vault write pki/roles/myapp-role \
    allowed_domains="myapp.example.com" \
    allow_subdomains=true \
    max_ttl="720h"

log "Vault Secret Engines configured successfully"

CI/CD平台集成

1. AppRole认证集成

在CI/CD平台中使用AppRole认证访问Vault：

# GitLab CI中的Vault集成示例
stages:
  - build
  - deploy

variables:
  VAULT_ADDR: "http://vault.example.com:8200"

deploy:
  stage: deploy
  image: 
    name: hashicorp/vault:latest
    entrypoint: [""]
  script:
    # 使用AppRole认证获取Vault令牌
    - vault write -field=token auth/approle/login role_id=$VAULT_ROLE_ID secret_id=$VAULT_SECRET_ID > vault-token.txt
    - export VAULT_TOKEN=$(cat vault-token.txt)
    
    # 从Vault获取数据库密码
    - DB_PASSWORD=$(vault kv get -field=password secret/myapp/database)
    
    # 从Vault获取API密钥
    - API_KEY=$(vault kv get -field=api_key secret/myapp/external_service)
    
    # 部署应用
    - kubectl create secret generic myapp-secrets \
        --from-literal=db_password=$DB_PASSWORD \
        --from-literal=api_key=$API_KEY
    
    - kubectl apply -f k8s/deployment.yaml
    
    # 清理令牌
    - unset VAULT_TOKEN
  only:
    - master

2. Kubernetes集成

通过Vault Agent Injector实现与Kubernetes的深度集成：

# 应用Deployment配置
apiVersion: apps/v1
kind: Deployment
metadata:
  name: myapp
spec:
  replicas: 3
  selector:
    matchLabels:
      app: myapp
  template:
    metadata:
      labels:
        app: myapp
      annotations:
        # Vault Agent Injector注解
        vault.hashicorp.com/agent-inject: "true"
        vault.hashicorp.com/agent-inject-status: "update"
        vault.hashicorp.com/role: "myapp-role"
        vault.hashicorp.com/agent-inject-secret-database-config.txt: "secret/myapp/database"
        vault.hashicorp.com/agent-inject-template-database-config.txt: |
          {{- with secret "secret/myapp/database" -}}
          database:
            host: {{ .Data.host }}
            port: {{ .Data.port }}
            username: {{ .Data.username }}
            password: {{ .Data.password }}
          {{- end }}
        vault.hashicorp.com/agent-inject-secret-external-api.txt: "secret/myapp/external_service"
        vault.hashicorp.com/agent-inject-template-external-api.txt: |
          {{- with secret "secret/myapp/external_service" -}}
          external_service:
            api_key: {{ .Data.api_key }}
            endpoint: {{ .Data.endpoint }}
          {{- end }}
    spec:
      serviceAccountName: myapp-vault
      containers:
      - name: myapp
        image: myapp:v1.0
        volumeMounts:
        - name: vault-secrets
          mountPath: /vault/secrets
          readOnly: true
      volumes:
      - name: vault-secrets
        emptyDir:
          medium: Memory

3. 动态密钥集成

使用Vault的动态密钥功能为每次部署生成临时数据库凭证：

#!/usr/bin/env python3
"""
动态密钥集成工具
从Vault获取临时数据库凭证并配置应用
"""

import hvac
import json
import os
import subprocess
import time

class VaultDynamicSecrets:
    def __init__(self, vault_addr: str, role_id: str, secret_id: str):
        self.client = hvac.Client(url=vault_addr)
        self.authenticate(role_id, secret_id)
    
    def authenticate(self, role_id: str, secret_id: str):
        """使用AppRole认证"""
        auth_response = self.client.auth.approle.login(
            role_id=role_id,
            secret_id=secret_id
        )
        self.client.token = auth_response['auth']['client_token']
        print("Vault authentication successful")
    
    def get_database_credentials(self, role_name: str) -> dict:
        """获取临时数据库凭证"""
        try:
            response = self.client.secrets.database.generate_credentials(
                name=role_name
            )
            return response['data']
        except Exception as e:
            print(f"Error getting database credentials: {e}")
            return None
    
    def create_k8s_secret(self, secret_name: str, credentials: dict):
        """创建Kubernetes Secret"""
        # 创建临时文件存储凭证
        with open('/tmp/db-credentials.json', 'w') as f:
            json.dump(credentials, f)
        
        # 创建Kubernetes Secret
        cmd = [
            'kubectl', 'create', 'secret', 'generic', secret_name,
            '--from-file=/tmp/db-credentials.json',
            '--namespace=myapp'
        ]
        
        try:
            subprocess.run(cmd, check=True)
            print(f"Kubernetes secret {secret_name} created successfully")
        except subprocess.CalledProcessError as e:
            print(f"Error creating Kubernetes secret: {e}")
    
    def rotate_database_credentials(self, role_name: str, secret_name: str):
        """轮换数据库凭证"""
        print(f"Rotating database credentials for role: {role_name}")
        
        # 获取新凭证
        credentials = self.get_database_credentials(role_name)
        if not credentials:
            raise Exception("Failed to get database credentials")
        
        # 创建新的Kubernetes Secret
        timestamp = int(time.time())
        new_secret_name = f"{secret_name}-{timestamp}"
        self.create_k8s_secret(new_secret_name, credentials)
        
        # 更新应用Deployment使用新Secret
        self.update_deployment_secret(new_secret_name)
        
        # 删除旧Secret（保留最近的几个版本）
        self.cleanup_old_secrets(secret_name)
        
        print("Database credentials rotated successfully")
    
    def update_deployment_secret(self, secret_name: str):
        """更新Deployment使用的Secret"""
        patch = {
            "spec": {
                "template": {
                    "spec": {
                        "containers": [
                            {
                                "name": "myapp",
                                "envFrom": [
                                    {
                                        "secretRef": {
                                            "name": secret_name
                                        }
                                    }
                                ]
                            }
                        ]
                    }
                }
            }
        }
        
        cmd = [
            'kubectl', 'patch', 'deployment', 'myapp',
            '--namespace=myapp',
            '-p', json.dumps(patch)
        ]
        
        try:
            subprocess.run(cmd, check=True)
            print("Deployment updated with new secret")
        except subprocess.CalledProcessError as e:
            print(f"Error updating deployment: {e}")

# 使用示例
if __name__ == "__main__":
    # 从环境变量获取配置
    vault_addr = os.getenv('VAULT_ADDR', 'http://localhost:8200')
    role_id = os.getenv('VAULT_ROLE_ID')
    secret_id = os.getenv('VAULT_SECRET_ID')
    
    if not role_id or not secret_id:
        print("VAULT_ROLE_ID and VAULT_SECRET_ID environment variables are required")
        exit(1)
    
    vault_client = VaultDynamicSecrets(vault_addr, role_id, secret_id)
    
    # 获取临时数据库凭证
    credentials = vault_client.get_database_credentials('myapp-role')
    if credentials:
        print("Database credentials obtained successfully")
        print(f"Username: {credentials['username']}")
        print(f"Password expires at: {credentials['expiration']}")
        
        # 创建Kubernetes Secret
        vault_client.create_k8s_secret('myapp-db-credentials', credentials)

密钥管理最佳实践

1. 安全策略实施

访问控制策略

# Vault策略示例：应用访问策略
path "secret/data/myapp/database" {
  capabilities = ["read"]
}

path "secret/data/myapp/external_service" {
  capabilities = ["read"]
}

path "database/creds/myapp-role" {
  capabilities = ["read"]
}

# 不允许列出密钥
path "secret/metadata/myapp/*" {
  capabilities = ["deny"]
}

审计日志配置

#!/bin/bash
# vault-audit.sh - 配置Vault审计日志

set -e

log() {
    echo "[$(date '+%Y-%m-%d %H:%M:%S')] $1"
}

# 启用文件审计设备
log "Enabling file audit device..."
vault audit enable file file_path=/var/log/vault-audit.log

# 启用syslog审计设备
log "Enabling syslog audit device..."
vault audit enable syslog

# 配置审计选项
log "Configuring audit options..."
vault audit enable file \
    file_path=/var/log/vault-audit.log \
    log_raw=true \
    hmac_accessor=false

log "Vault audit logging configured successfully"

2. 密钥轮换策略

自动化密钥轮换

#!/usr/bin/env python3
"""
密钥轮换自动化工具
定期轮换静态密钥并更新相关服务
"""

import hvac
import json
import os
import subprocess
import time
import random
import string
from datetime import datetime, timedelta

class KeyRotator:
    def __init__(self, vault_addr: str, role_id: str, secret_id: str):
        self.client = hvac.Client(url=vault_addr)
        self.authenticate(role_id, secret_id)
    
    def authenticate(self, role_id: str, secret_id: str):
        """使用AppRole认证"""
        auth_response = self.client.auth.approle.login(
            role_id=role_id,
            secret_id=secret_id
        )
        self.client.token = auth_response['auth']['client_token']
        print("Vault authentication successful")
    
    def generate_secure_password(self, length: int = 32) -> str:
        """生成安全密码"""
        characters = string.ascii_letters + string.digits + "!@#$%^&*()_+-="
        return ''.join(random.choice(characters) for _ in range(length))
    
    def rotate_database_password(self, secret_path: str, db_role: str):
        """轮换数据库密码"""
        print(f"Rotating database password for {secret_path}")
        
        # 生成新密码
        new_password = self.generate_secure_password()
        
        # 获取现有配置
        try:
            existing = self.client.secrets.kv.v2.read_secret_version(path=secret_path)
            db_config = existing['data']['data']
        except Exception as e:
            print(f"Error reading existing secret: {e}")
            return False
        
        # 更新密码
        db_config['password'] = new_password
        
        # 保存到Vault
        try:
            self.client.secrets.kv.v2.create_or_update_secret(
                path=secret_path,
                secret=db_config
            )
            print("Database password updated in Vault")
        except Exception as e:
            print(f"Error updating secret in Vault: {e}")
            return False
        
        # 更新数据库用户密码
        if not self.update_database_user_password(
            db_config['host'],
            db_config['port'],
            db_config['admin_user'],
            db_config['admin_password'],
            db_config['username'],
            new_password
        ):
            return False
        
        # 通知相关服务更新
        self.notify_services(secret_path)
        
        return True
    
    def update_database_user_password(self, host: str, port: str, admin_user: str, 
                                    admin_password: str, target_user: str, new_password: str) -> bool:
        """更新数据库用户密码"""
        # 这里应该根据具体的数据库类型实现密码更新逻辑
        # 示例：PostgreSQL
        try:
            import psycopg2
            
            conn = psycopg2.connect(
                host=host,
                port=port,
                database="postgres",
                user=admin_user,
                password=admin_password
            )
            
            cursor = conn.cursor()
            cursor.execute(f"ALTER USER {target_user} WITH PASSWORD '{new_password}'")
            conn.commit()
            
            cursor.close()
            conn.close()
            
            print("Database user password updated successfully")
            return True
        except Exception as e:
            print(f"Error updating database user password: {e}")
            return False
    
    def notify_services(self, secret_path: str):
        """通知相关服务密钥已更新"""
        # 这里可以实现通知机制，如：
        # 1. 发送Webhook通知
        # 2. 更新Kubernetes ConfigMap
        # 3. 发送消息到消息队列
        print(f"Services notified about secret update: {secret_path}")
    
    def schedule_rotation(self, secret_path: str, db_role: str, interval_hours: int):
        """调度定期轮换"""
        print(f"Scheduling rotation for {secret_path} every {interval_hours} hours")
        
        while True:
            try:
                self.rotate_database_password(secret_path, db_role)
                print(f"Next rotation scheduled in {interval_hours} hours")
                time.sleep(interval_hours * 3600)
            except KeyboardInterrupt:
                print("Rotation scheduler stopped")
                break
            except Exception as e:
                print(f"Error during rotation: {e}")
                time.sleep(300)  # 5分钟后重试

# 使用示例
if __name__ == "__main__":
    # 从环境变量获取配置
    vault_addr = os.getenv('VAULT_ADDR', 'http://localhost:8200')
    role_id = os.getenv('VAULT_ROLE_ID')
    secret_id = os.getenv('VAULT_SECRET_ID')
    
    if not role_id or not secret_id:
        print("VAULT_ROLE_ID and VAULT_SECRET_ID environment variables are required")
        exit(1)
    
    rotator = KeyRotator(vault_addr, role_id, secret_id)
    
    # 立即执行一次轮换
    rotator.rotate_database_password('secret/myapp/database', 'myapp-role')
    
    # 启动定期轮换（每24小时）
    # rotator.schedule_rotation('secret/myapp/database', 'myapp-role', 24)

3. 监控与告警

Vault健康检查

#!/usr/bin/env python3
"""
Vault健康检查和监控工具
监控Vault服务状态和密钥使用情况
"""

import hvac
import requests
import json
import time
from datetime import datetime

class VaultMonitor:
    def __init__(self, vault_addr: str):
        self.vault_addr = vault_addr
        self.client = hvac.Client(url=vault_addr)
    
    def check_vault_health(self) -> dict:
        """检查Vault健康状态"""
        try:
            response = requests.get(f"{self.vault_addr}/v1/sys/health")
            return {
                'status': 'healthy' if response.status_code == 200 else 'unhealthy',
                'status_code': response.status_code,
                'timestamp': datetime.now().isoformat()
            }
        except Exception as e:
            return {
                'status': 'unreachable',
                'error': str(e),
                'timestamp': datetime.now().isoformat()
            }
    
    def check_token_validity(self, token: str) -> dict:
        """检查令牌有效性"""
        try:
            self.client.token = token
            response = self.client.auth.token.lookup_self()
            return {
                'valid': True,
                'ttl': response['data']['ttl'],
                'policies': response['data']['policies'],
                'timestamp': datetime.now().isoformat()
            }
        except Exception as e:
            return {
                'valid': False,
                'error': str(e),
                'timestamp': datetime.now().isoformat()
            }
    
    def audit_secret_access(self) -> dict:
        """审计密钥访问情况"""
        # 这里应该分析Vault审计日志
        # 简化示例：返回固定数据
        return {
            'total_requests': 100,
            'failed_requests': 5,
            'unique_clients': 10,
            'timestamp': datetime.now().isoformat()
        }
    
    def generate_health_report(self) -> dict:
        """生成健康报告"""
        return {
            'vault_health': self.check_vault_health(),
            'access_audit': self.audit_secret_access(),
            'timestamp': datetime.now().isoformat()
        }

# 监控脚本
def run_vault_monitoring():
    """运行Vault监控"""
    vault_addr = os.getenv('VAULT_ADDR', 'http://localhost:8200')
    monitor = VaultMonitor(vault_addr)
    
    while True:
        try:
            report = monitor.generate_health_report()
            print(json.dumps(report, indent=2))
            
            # 检查是否需要告警
            if report['vault_health']['status'] != 'healthy':
                print("ALERT: Vault is not healthy!")
                # 这里可以发送告警通知
            
            time.sleep(300)  # 每5分钟检查一次
        except KeyboardInterrupt:
            print("Monitoring stopped")
            break
        except Exception as e:
            print(f"Error during monitoring: {e}")
            time.sleep(60)

if __name__ == "__main__":
    run_vault_monitoring()

通过集成HashiCorp Vault等专业的secrets manager，CI/CD平台能够实现安全、动态、可审计的密钥管理。这不仅解决了传统密钥管理方式的安全风险，还提供了灵活的访问控制和完整的审计追踪功能。关键是要根据实际需求选择合适的认证方式和Secret Engines，并建立完善的密钥轮换和监控机制，确保整个密钥管理系统的安全性和可靠性。

我是老马，期待与你的下次重逢
Copyright © 2025 老马啸西风