HSM 密钥轮转自动化实战：从手动操作到零宕机迁移

import json
from datetime import datetime, timezone, timedelta
from typing import Optional
from uuid import uuid4

@dataclass
class KeyMetadata:
    key_id: str                    # 如 "DEK-AES256-v1-001"
    key_type: str                  # AES-256-GCM
    algorithm: str                 # AES-GCM
    status: str                    # active / readonly / retired / revoked
    created_at: datetime
    rotation_reason: Optional[str]  # 轮转原因：routine/compromise/audit
    associated_data: dict          # 关联的系统、表、字段
    
    def to_dict(self) -> dict:
        return {
            'key_id': self.key_id,
            'key_type': self.key_type,
            'algorithm': self.algorithm,
            'status': self.status,
            'created_at': self.created_at.isoformat(),
            'rotation_reason': self.rotation_reason,
            'associated_data': self.associated_data,
        }
    
    @classmethod
    def from_dict(cls, data: dict) -> 'KeyMetadata':
        data['created_at'] = datetime.fromisoformat(data['created_at'])
        return cls(**data)

-- 存储到数据库
CREATE TABLE hsm_keys (
    key_id VARCHAR(64) PRIMARY KEY,
    key_type VARCHAR(32) NOT NULL,
    algorithm VARCHAR(32) NOT NULL,
    status VARCHAR(16) NOT NULL DEFAULT 'active',
    created_at TIMESTAMP NOT NULL,
    rotation_reason VARCHAR(256),
    associated_data TEXT,  -- JSON
    tags JSON,             -- 灵活标签
    UNIQUE(key_id, status)
);

CREATE INDEX idx_hsm_keys_status ON hsm_keys(status, created_at DESC);
CREATE INDEX idx_hsm_keys_algorithm ON hsm_keys(algorithm, status);

import logging
from contextlib import contextmanager
from typing import Callable
import time

logger = logging.getLogger(__name__)

@contextmanager
def key_rotation_lock(rotation_id: str, timeout: int = 30):
    """
    轮转过程锁，防止并发操作导致状态混乱
    使用数据库层面的乐观锁，而非文件锁
    """
    from app.database import SessionLocal
    from app.models import KeyRotation
    
    db = SessionLocal()
    try:
        # 检查是否已有进行中的轮转
        existing = db.query(KeyRotation).filter(
            KeyRotation.status.in_(['in_progress', 'preparing']),
            KeyRotation.completed_at.is_(None)
        ).first()
        
        if existing:
            raise RotationError(f"存在进行中的轮转任务: {existing.rotation_id}")
        
        # 创建轮转任务
        rotation = KeyRotation(
            rotation_id=rotation_id,
            status='preparing',
            started_at=datetime.now(timezone.utc),
        )
        db.add(rotation)
        db.commit()
        db.refresh(rotation)
        
        logger.info(f"轮转任务 {rotation_id} 已启动")
        yield rotation
        
    finally:
        db.close()

class RotationError(Exception):
    """轮转错误基类"""
    pass

class HSMKeyRotator:
    """
    HSM 密钥轮转器
    支持：平滑轮转、灰度验证、自动回滚
    """
    
    def __init__(self, hsm_client, key_registry):
        self.hsm = hsm_client
        self.registry = key_registry
    
    def rotate_key(
        self,
        key_id: str,
        reason: str = "routine",
        dry_run: bool = False,
        verify_threshold: float = 0.99,
    ) -> dict:
        """
        执行完整的密钥轮转流程
        
        参数:
            key_id: 要轮转的旧密钥ID
            reason: 轮转原因
            dry_run: 是否仅模拟不实际执行
            verify_threshold: 验证通过率阈值，低于此值自动回滚
        """
        rotation_id = f"rot-{int(time.time())}-{uuid4().hex[:8]}"
        
        with key_rotation_lock(rotation_id) as rotation_ctx:
            # 阶段1：生成新密钥
            if not dry_run:
                new_key = self._generate_new_key(key_id, rotation_id)
            else:
                new_key = self._simulate_key_generation(key_id, rotation_id)
            
            # 阶段2：验证新密钥可用性
            if not self._verify_new_key(new_key):
                raise RotationError("新密钥验证失败")
            
            # 阶段3：灰度切换流量
            switch_result = self._gradual_traffic_switch(
                old_key_id=key_id,
                new_key=new_key,
                threshold=verify_threshold,
            )
            
            # 阶段4：验证切换结果
            if switch_result['success_rate'] < verify_threshold:
                logger.error(f"切换成功率 {switch_result['success_rate']:.2%} 低于阈值")
                self._rollback_switch(key_id, rotation_id)
                raise RotationError(f"切换失败，已回滚。成功率: {switch_result['success_rate']:.2%}")
            
            # 阶段5：标记旧密钥为只读
            if not dry_run:
                self._set_key_readonly(key_id)
            
            # 阶段6：更新元数据
            self._update_rotation_metadata(
                rotation_id=rotation_id,
                old_key_id=key_id,
                new_key=new_key,
                result=switch_result,
            )
            
            return {
                'rotation_id': rotation_id,
                'status': 'completed',
                'new_key_id': new_key.key_id,
                'success_rate': switch_result['success_rate'],
                'duration_seconds': switch_result.get('duration', 0),
            }
    
    def _generate_new_key(self, old_key_id: str, rotation_id: str) -> KeyMetadata:
        """生成新密钥并注册"""
        old_meta = self.registry.get_key(old_key_id)
        if not old_meta:
            raise RotationError(f"密钥 {old_key_id} 不存在")
        
        # 解析版本信息，递增序列号
        parts = old_key_id.split('-')
        version = int(parts[-2])
        seq = int(parts[-1])
        new_seq = seq + 1
        
        new_key_id = f"{parts[0]}-{parts[1]}-v{version}-{new_seq:03d}"
        
        # 调用 HSM 生成密钥
        logger.info(f"正在 HSM 中生成新密钥: {new_key_id}")
        key_material = self.hsm.generate_key(
            key_id=new_key_id,
            algorithm=old_meta.algorithm,
            key_size=old_meta.key_size,
        )
        
        # 注册新密钥
        new_meta = KeyMetadata(
            key_id=new_key_id,
            key_type=old_meta.key_type,
            algorithm=old_meta.algorithm,
            status='active',
            created_at=datetime.now(timezone.utc),
            rotation_reason=f"replaces-{old_key_id}",
            associated_data=old_meta.associated_data.copy(),
        )
        self.registry.register(new_meta)
        
        logger.info(f"新密钥 {new_key_id} 已生成并注册")
        return new_meta
    
    def _gradual_traffic_switch(
        self,
        old_key_id: str,
        new_key: KeyMetadata,
        threshold: float = 0.99,
    ) -> dict:
        """
        灰度切换流量
        
        策略：
        1. 0% → 5% → 20% → 50% → 100%
        2. 每个阶段停留 30 秒，监控错误率
        3. 任意阶段错误率 > 1% 则中止并回滚
        """
        stages = [0, 5, 20, 50, 100]
        total_start = time.time()
        
        results = {
            'stages': [],
            'success_rate': 0.0,
            'errors': [],
            'duration': 0,
        }
        
        for stage in stages:
            # 设置流量路由比例
            self._set_traffic_ratio(old_key_id, new_key.key_id, stage)
            
            # 等待稳定
            time.sleep(30)
            
            # 统计该阶段的验证结果
            stats = self._collect_validation_stats(
                old_key_id=old_key_id,
                new_key_id=new_key.key_id,
                window_seconds=30,
            )
            
            results['stages'].append({
                'traffic_percent': stage,
                'total_requests': stats['total'],
                'success': stats['success'],
                'error_rate': stats['error_rate'],
                'avg_latency_ms': stats['avg_latency_ms'],
            })
            
            # 检查错误率
            if stats['error_rate'] > 0.01:
                results['errors'].append(f"阶段 {stage}% 错误率过高: {stats['error_rate']:.2%}")
                results['duration'] = time.time() - total_start
                results['success_rate'] = stats['success'] / max(stats['total'], 1)
                return results
            
            logger.info(
                f"阶段 {stage}%: 成功率 {stats['success_rate']:.2%}, "
                f"平均延迟 {stats['avg_latency_ms']:.1f}ms"
            )
        
        results['success_rate'] = results['stages'][-1]['success'] / \
            max(results['stages'][-1]['total_requests'], 1)
        results['duration'] = time.time() - total_start
        
        return results
    
    def _set_traffic_ratio(self, old_key_id: str, new_key_id: str, new_percent: int):
        """设置新旧密钥的流量比例"""
        # 通过配置中心或开关服务设置
        # 示例：更新 Redis 中的路由配置
        import redis
        r = redis.Redis(host='localhost', port=6379)
        pipe = r.pipeline()
        pipe.set(f"key_route:{old_key_id}", str(100 - new_percent))
        pipe.set(f"key_route:{new_key_id}", str(new_percent))
        pipe.expire(f"key_route:{old_key_id}", 3600)
        pipe.expire(f"key_route:{new_key_id}", 3600)
        pipe.execute()
        
        logger.debug(f"流量切换: {old_key_id}=100-{new_percent}%, {new_key_id}={new_percent}%")
    
    def _collect_validation_stats(
        self,
        old_key_id: str,
        new_key_id: str,
        window_seconds: int = 30,
    ) -> dict:
        """收集验证统计数据"""
        from app.database import SessionLocal
        from app.models import EncryptionRequest
        from sqlalchemy import func
        
        db = SessionLocal()
        
        # 查询最近窗口期内的请求
        cutoff = datetime.now(timezone.utc) - timedelta(seconds=window_seconds)
        
        total = db.query(EncryptionRequest).filter(
            EncryptionRequest.created_at >= cutoff,
            EncryptionRequest.key_id.in_([old_key_id, new_key_id]),
        ).count()
        
        success = db.query(EncryptionRequest).filter(
            EncryptionRequest.created_at >= cutoff,
            EncryptionRequest.key_id == new_key_id,
            EncryptionRequest.status == 'success',
        ).count()
        
        avg_latency = db.query(func.avg(EncryptionRequest.latency_ms)).filter(
            EncryptionRequest.created_at >= cutoff,
            EncryptionRequest.key_id == new_key_id,
        ).scalar() or 0
        
        error_rate = 1.0 - (success / max(total, 1))
        
        db.close()
        
        return {
            'total': total,
            'success': success,
            'error_rate': error_rate,
            'avg_latency_ms': float(avg_latency),
            'success_rate': success / max(total, 1),
        }
    
    def _set_key_readonly(self, key_id: str):
        """标记旧密钥为只读（保留但不用于新加密）"""
        self.registry.update_key_status(key_id, 'readonly')
        logger.info(f"密钥 {key_id} 已标记为只读")
    
    def _rollback_switch(self, old_key_id: str, rotation_id: str):
        """回滚流量切换，恢复旧密钥全量流量"""
        self._set_traffic_ratio(old_key_id, None, 0)
        logger.warning(f"轮转 {rotation_id} 已回滚，恢复旧密钥 {old_key_id} 全量流量")

import asyncio
from concurrent.futures import ThreadPoolExecutor
from typing import AsyncIterator

class DataReencryptionWorker:
    """
    数据重加密工作器
    后台异步处理历史数据，从旧密钥迁移到新密钥
    """
    
    def __init__(
        self,
        hsm_client,
        key_registry,
        db_pool,
        batch_size: int = 1000,
        max_concurrency: int = 10,
    ):
        self.hsm = hsm_client
        self.registry = key_registry
        self.db = db_pool
        self.batch_size = batch_size
        self.semaphore = asyncio.Semaphore(max_concurrency)
    
    async def reencrypt_table(
        self,
        table_name: str,
        encrypt_col: str,
        old_key_id: str,
        new_key_id: str,
        progress_callback: Callable[[int, int], None] = None,
    ) -> dict:
        """
        重加密指定表中的字段
        
        参数:
            table_name: 表名
            encrypt_col: 加密字段名
            old_key_id: 旧密钥ID
            new_key_id: 新密钥ID
            progress_callback: 进度回调函数 (processed, total)
        """
        total = await self._count_pending(table_name, encrypt_col, old_key_id)
        processed = 0
        errors = []
        
        # 分页查询未重加密的记录
        offset = 0
        while True:
            async with self.semaphore:
                batch = await self._fetch_batch(
                    table_name, encrypt_col, old_key_id, offset, self.batch_size
                )
            
            if not batch:
                break
            
            # 批量重加密
            reenc_results = await asyncio.gather(
                *[self._reencrypt_record(record, old_key_id, new_key_id) 
                  for record in batch],
                return_exceptions=True,
            )
            
            # 处理结果
            for i, result in enumerate(reenc_results):
                if isinstance(result, Exception):
                    errors.append({
                        'record_id': batch[i]['id'],
                        'error': str(result),
                    })
                else:
                    processed += 1
            
            offset += self.batch_size
            
            if progress_callback:
                progress_callback(processed, total)
        
        return {
            'table': table_name,
            'total': total,
            'processed': processed,
            'errors': len(errors),
            'error_details': errors[:10],  # 只保留前10条错误详情
        }
    
    async def _reencrypt_record(
        self,
        record: dict,
        old_key_id: str,
        new_key_id: str,
    ) -> dict:
        """重加密单条记录"""
        encrypted_value = record['encrypted_value']
        
        # 1. 用旧密钥解密
        plaintext = await asyncio.to_thread(
            self.hsm.decrypt,
            key_id=old_key_id,
            ciphertext=encrypted_value,
        )
        
        # 2. 用新密钥加密
        new_encrypted = await asyncio.to_thread(
            self.hsm.encrypt,
            key_id=new_key_id,
            plaintext=plaintext,
        )
        
        # 3. 更新数据库
        async with self.db.acquire() as conn:
            await conn.execute(
                f"""
                UPDATE {record['table']}
                SET encrypted_value = $1, key_version = $2, updated_at = NOW()
                WHERE id = $3
                """,
                new_encrypted,
                new_key_id,
                record['id'],
            )
        
        return {'id': record['id'], 'new_key': new_key_id}

# 错误示范：直接切换，没有过渡期
def bad_rotation(old_key, new_key):
    # 瞬间切换
    config['current_key'] = new_key
    # 此时正在处理的请求可能使用旧密钥，导致不一致

# 正确做法：双密钥并存 + 灰度切换
async def good_rotation(old_key_id, new_key_id):
    # 1. 注册新密钥，标记为 active
    await register_key(new_key_id, status='active')
    
    # 2. 灰度切换（见上文 gradual_traffic_switch）
    await gradual_switch(old_key_id, new_key_id)
    
    # 3. 确认切换成功后，再标记旧密钥为 readonly
    await set_key_status(old_key_id, 'readonly')
    
    # 4. 等待足够长的观察期（建议至少 24 小时）
    # 5. 然后归档旧密钥
    await archive_key(old_key_id)

# 在重加密前进行完整性检查
def pre_reencryption_check(table_name: str, encrypt_col: str, key_id: str) -> dict:
    """重加密前的完整性检查"""
    db = get_db_connection()
    
    # 1. 检查旧密钥是否存在且有效
    key_valid = db.execute(
        "SELECT COUNT(*) FROM key_registry WHERE key_id = ? AND status = 'active'",
        (key_id,)
    ).fetchone()[0] > 0
    
    if not key_valid:
        raise CheckFailedError(f"密钥 {key_id} 不存在或已失效")
    
    # 2. 抽样验证解密能力
    sample_count = 100
    sample = db.execute(
        f"SELECT id, encrypted_value FROM {table_name} WHERE key_version = ? LIMIT ?",
        (key_id, sample_count)
    ).fetchall()
    
    decrypt_failures = 0
    for record in sample:
        try:
            db_decrypt(key_id, record['encrypted_value'])
        except Exception:
            decrypt_failures += 1
    
    failure_rate = decrypt_failures / len(sample)
    
    # 3. 检查阈值
    if failure_rate > 0.01:  # 允许 1% 的错误率
        raise CheckFailedError(
            f"抽样解密失败率 {failure_rate:.1%} 超过阈值 1%"
        )
    
    return {
        'key_valid': True,
        'sample_size': len(sample),
        'decrypt_failure_rate': failure_rate,
        'recommendation': '可以继续重加密' if failure_rate < 0.001 else '建议先修复密钥',
    }

# 审计日志结构：包含完整的密钥版本链
@dataclass
class AuditRecord:
    event_id: str
    timestamp: datetime
    action: str                  # encrypt / decrypt / rotate
    key_id: str                  # 当前使用的密钥ID
    key_chain: list              # 密钥版本链：[old_key, new_key, ...]
    data_hash: str               # 数据哈希（用于完整性验证）
    source_key_id: Optional[str] = None  # 解密时使用的源密钥
    target_key_id: Optional[str] = None  # 加密时使用的目标密钥
    
    def to_log_entry(self) -> str:
        return json.dumps({
            'event_id': self.event_id,
            'timestamp': self.timestamp.isoformat(),
            'action': self.action,
            'key_id': self.key_id,
            'key_chain': self.key_chain,
            'data_hash': self.data_hash,
            'source_key_id': self.source_key_id,
            'target_key_id': self.target_key_id,
        })

# 写入审计日志
def log_audit_event(audit_record: AuditRecord):
    db.execute(
        """
        INSERT INTO audit_log (event_id, timestamp, action, key_id, 
                               key_chain, data_hash, source_key_id, target_key_id)
        VALUES (?, ?, ?, ?, ?, ?, ?, ?)
        """,
        audit_record.event_id,
        audit_record.timestamp,
        audit_record.action,
        audit_record.key_id,
        json.dumps(audit_record.key_chain),
        audit_record.data_hash,
        audit_record.source_key_id,
        audit_record.target_key_id,
    )

# 明确区分密钥轮转和证书吊销
from enum import Enum

class KeyOperation(Enum):
    ROTATE = "rotate"           # 密钥轮转：生成新密钥，迁移数据
    REVOKE = "revoke"           # 密钥吊销：立即失效，不可恢复
    ARCHIVE = "archive"         # 密钥归档：标记为只读，保留解密能力
    RESTORE = "restore"         # 密钥恢复：从归档状态恢复为活跃

class CertOperation(Enum):
    RENEW = "renew"             # 证书续期
    REVOKE = "revoke"           # 证书吊销
    SUSPEND = "suspend"         # 证书暂停

# 流程隔离
def rotate_key(key_id: str, reason: str):
    """密钥轮转：不影响证书"""
    # 1. 生成新密钥
    new_key = generate_key()
    
    # 2. 迁移数据
    migrate_data(key_id, new_key.key_id)
    
    # 3. 标记旧密钥为只读
    set_key_readonly(key_id)
    
    # 注意：不调用任何证书吊销逻辑

def revoke_certificate(cert_serial: str):
    """证书吊销：不影响密钥"""
    # 1. 添加到 CRL
    add_to_crl(cert_serial)
    
    # 2. 发布 CRL
    publish_crl()
    
    # 注意：不调用任何密钥操作逻辑

import schedule
import json
from pathlib import Path
from typing import Dict, List

class KeyRotationScheduler:
    """
    密钥轮转调度器
    支持：定时轮转、事件触发轮转、手动触发
    """
    
    def __init__(self, config_path: str = "/etc/hsm/rotation.yaml"):
        self.config = self._load_config(config_path)
        self.rotator = HSMKeyRotator(
            hsm_client=self._init_hsm(),
            key_registry=self._init_registry(),
        )
        self.reencryptor = DataReencryptionWorker(
            hsm_client=self._init_hsm(),
            key_registry=self._init_registry(),
            db_pool=self._init_db_pool(),
        )
    
    def _load_config(self, path: str) -> Dict:
        """加载轮转配置"""
        config_file = Path(path)
        if not config_file.exists():
            raise ConfigError(f"配置文件不存在: {path}")
        
        with open(config_file) as f:
            config = yaml.safe_load(f)
        
        # 验证必要字段
        required_fields = ['key_types', 'rotation_interval_days', 'reencryption_delay_days']
        for field in required_fields:
            if field not in config:
                raise ConfigError(f"缺少必要配置项: {field}")
        
        return config
    
    def _init_hsm(self) -> HSMClient:
        """初始化 HSM 客户端"""
        return HSMClient(
            endpoint=self.config['hsm_endpoint'],
            auth_method=self.config['hsm_auth']['method'],
            timeout=self.config.get('hsm_timeout', 30),
        )
    
    def execute_rotation(self, key_type: str, force: bool = False):
        """
        执行指定类型的密钥轮转
        
        参数:
            key_type: 密钥类型（如 AES-256-GCM）
            force: 是否强制执行（跳过时间间隔检查）
        """
        # 检查上次轮转时间
        last_rotation = self._get_last_rotation_time(key_type)
        interval_days = self.config['rotation_interval_days'].get(key_type, 90)
        
        if not force and last_rotation:
            days_since = (datetime.now() - last_rotation).days
            if days_since < interval_days:
                logger.info(
                    f"密钥 {key_type} 距离上次轮转仅 {days_since} 天，"
                    f"跳过本次轮转（间隔 {interval_days} 天）"
                )
                return
        
        logger.info(f"开始执行 {key_type} 密钥轮转")
        
        # 获取当前活跃密钥
        active_keys = self.registry.list_active_keys(key_type)
        if not active_keys:
            logger.error(f"未找到 {key_type} 的活跃密钥")
            return
        
        current_key = active_keys[0]
        
        try:
            # 执行轮转
            result = self.rotator.rotate_key(
                key_id=current_key.key_id,
                reason="scheduled" if not force else "manual",
            )
            
            logger.info(f"密钥轮转完成: {result['rotation_id']}")
            
            # 记录轮转时间
            self._record_rotation_time(key_type, result['new_key_id'])
            
            # 触发重加密（如果配置了自动重加密）
            if self.config.get('auto_reencrypt', False):
                self._trigger_reencryption(key_type)
                
        except RotationError as e:
            logger.error(f"密钥轮转失败: {e}")
            self._handle_rotation_failure(key_type, str(e))
    
    def _trigger_reencryption(self, key_type: str):
        """触发数据重加密"""
        # 等待冷却期，避免刚轮转就重加密
        delay_days = self.config['reencryption_delay_days']
        logger.info(f"将在 {delay_days} 天后开始数据重加密")
        
        # 使用 delayed job 队列
        self.job_queue.schedule(
            func=self._execute_reencryption,
            args=[key_type],
            delay_days=delay_days,
        )
    
    def _execute_reencryption(self, key_type: str):
        """执行数据重加密"""
        active_keys = self.registry.list_active_keys(key_type)
        if len(active_keys) < 2:
            logger.error("密钥数量不足，无法执行重加密")
            return
        
        # 获取最近两把密钥
        old_key = active_keys[-1]  # 较老的活跃密钥
        new_key = active_keys[0]   # 最新的活跃密钥
        
        logger.info(
            f"开始重加密 {key_type} 数据："
            f"{old_key.key_id} → {new_key.key_id}"
        )
        
        # 对每个需要重加密的表执行重加密
        tables = self.config.get('reencryption_tables', [])
        results = []
        
        for table_info in tables:
            result = asyncio.run(
                self.reencryptor.reencrypt_table(
                    table_name=table_info['name'],
                    encrypt_col=table_info['encrypt_col'],
                    old_key_id=old_key.key_id,
                    new_key_id=new_key.key_id,
                )
            )
            results.append(result)
        
        # 汇总报告
        total_processed = sum(r['processed'] for r in results)
        total_errors = sum(r['errors'] for r in results)
        
        logger.info(
            f"重加密完成：成功 {total_processed} 条，失败 {total_errors} 条"
        )
        
        # 发送通知
        self._send_rotation_report(key_type, results)

class KeyRotationHealthChecker:
    """
    密钥轮转健康检查器
    监控轮转状态、密钥可用性、数据一致性
    """
    
    def __init__(self, registry: KeyRegistry, db_pool: DatabasePool):
        self.registry = registry
        self.db = db_pool
    
    def run_health_check(self) -> dict:
        """执行完整健康检查"""
        checks = {
            'key_availability': self._check_key_availability(),
            'rotation_status': self._check_rotation_status(),
            'data_consistency': self._check_data_consistency(),
            'audit_compliance': self._check_audit_compliance(),
        }
        
        # 判断整体状态
        all_passed = all(check['status'] == 'healthy' for check in checks.values())
        
        return {
            'timestamp': datetime.now(timezone.utc).isoformat(),
            'overall_status': 'healthy' if all_passed else 'unhealthy',
            'checks': checks,
        }
    
    def _check_key_availability(self) -> dict:
        """检查密钥可用性"""
        active_keys = self.registry.list_active_keys()
        warnings = []
        
        # 检查是否有密钥即将过期
        for key in active_keys:
            if key.expires_at:
                days_until_expiry = (key.expires_at - datetime.now()).days
                if days_until_expiry < 30:
                    warnings.append(
                        f"密钥 {key.key_id} 将在 {days_until_expiry} 天后过期"
                    )
        
        return {
            'status': 'warning' if warnings else 'healthy',
            'active_keys_count': len(active_keys),
            'warnings': warnings,
        }
    
    def _check_data_consistency(self) -> dict:
        """检查数据加密一致性"""
        from sqlalchemy import text
        
        # 检查是否存在使用已废弃密钥加密的数据
        inconsistencies = self.db.execute(text(
            """
            SELECT table_name, column_name, key_version, COUNT(*) as cnt
            FROM encryption_audit
            WHERE key_version NOT IN (
                SELECT key_id FROM key_registry WHERE status = 'active'
            )
            GROUP BY table_name, column_name, key_version
            HAVING cnt > 100
            """
        )).fetchall()
        
        return {
            'status': 'warning' if inconsistencies else 'healthy',
            'inconsistencies': [
                {
                    'table': row[0],
                    'column': row[1],
                    'key_version': row[2],
                    'count': row[3],
                }
                for row in inconsistencies
            ],
        }
    
    def send_alert(self, check_result: dict):
        """发送告警通知"""
        if check_result['overall_status'] == 'unhealthy':
            # 构建告警消息
            alert_msg = f"密钥轮转健康检查失败:\n"
            for check_name, check_data in check_result['checks'].items():
                if check_data['status'] != 'healthy':
                    alert_msg += f"  - {check_name}: {check_data.get('warnings', '未知错误')}\n"
            
            # 发送告警
            self.alert_service.send(
                channel='ops-channel',
                severity='critical',
                message=alert_msg,
            )

class KeyRotationRollback:
    """
    密钥轮转失败回滚处理器
    """
    
    def __init__(self, registry: KeyRegistry, hsm_client: HSMClient):
        self.registry = registry
        self.hsm = hsm_client
    
    def rollback_rotation(self, rotation_id: str) -> dict:
        """
        回滚指定的密钥轮转
        
        回滚步骤：
        1. 恢复旧密钥为 active
        2. 撤销新密钥
        3. 清理轮转期间的缓存
        4. 恢复流量配置
        """
        logger.warning(f"开始回滚轮转任务: {rotation_id}")
        
        try:
            # 1. 查询轮转记录
            rotation = self._get_rotation_record(rotation_id)
            if not rotation:
                raise RollbackError(f"未找到轮转记录: {rotation_id}")
            
            old_key_id = rotation['old_key_id']
            new_key_id = rotation['new_key_id']
            
            # 2. 恢复旧密钥为 active
            self.registry.update_key_status(old_key_id, 'active')
            logger.info(f"已恢复旧密钥 {old_key_id} 为 active")
            
            # 3. 撤销新密钥
            self.registry.update_key_status(new_key_id, 'revoked')
            logger.info(f"已撤销新密钥 {new_key_id}")
            
            # 4. 清理缓存
            self._clear_key_cache()
            logger.info("已清理密钥缓存")
            
            # 5. 恢复流量配置
            self._restore_traffic_config(old_key_id)
            logger.info("已恢复流量配置")
            
            # 6. 更新轮转状态
            self._update_rotation_status(rotation_id, 'rolled_back')
            
            return {
                'status': 'success',
                'message': f"轮转 {rotation_id} 已成功回滚",
                'restored_key': old_key_id,
            }
            
        except Exception as e:
            logger.error(f"回滚失败: {e}")
            self._update_rotation_status(rotation_id, 'rollback_failed')
            raise RollbackError(f"回滚过程异常: {e}")
    
    def _restore_traffic_config(self, old_key_id: str):
        """恢复流量配置"""
        # 将流量全部切回旧密钥
        self._set_traffic_ratio(old_key_id, None, 0)

# 应急联系人配置
EMERGENCY_CONTACTS = {
    'key_rotation_failure': {
        'level1': {
            'role': 'On-call Engineer',
            'channels': ['pagerduty', 'slack'],
            'response_time': '15 minutes',
        },
        'level2': {
            'role': 'Security Team Lead',
            'channels': ['phone', 'slack'],
            'response_time': '30 minutes',
        },
        'level3': {
            'role': 'CISO / Security Director',
            'channels': ['phone'],
            'response_time': '1 hour',
        },
    },
    'data_breach_suspected': {
        'level1': {
            'role': 'Incident Response Team',
            'channels': ['pagerduty', 'slack'],
            'response_time': '15 minutes',
        },
    },
}