证书有效期缩短至 47 天：CA/B 论坛倒计时已启动，企业 PKI 自动化迫在眉睫

#!/usr/bin/env python3
"""
证书自动续期脚本 — 基于 ACME 协议
支持 certbot 和 step-ca 两种 ACME CA
"""

import subprocess
import json
import logging
import sys
from datetime import datetime, timedelta
from pathlib import Path

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s [%(levelname)s] %(message)s',
    handlers=[
        logging.FileHandler('/var/log/cert-renewal.log'),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger(__name__)

# ACME CA 配置
ACME_CONFIGS = {
    'production': {
        'ca_url': 'https://acme-v02.api.letsencrypt.org/directory',
        'email': 'admin@example.com',
        'cert_dir': '/etc/letsencrypt/live',
    },
    'internal': {
        'ca_url': 'https://internal-ca.example.com/acme/directory',
        'email': 'pki-admin@example.com',
        'cert_dir': '/etc/internal-certs',
    }
}

def check_cert_expiry(domain: str, cert_path: str, threshold_days: int = 30) -> dict:
    """检查证书剩余有效期"""
    try:
        result = subprocess.run(
            ['openssl', 'x509', '-in', cert_path, '-noout', '-enddate'],
            capture_output=True, text=True, check=True
        )
        # 解析: notAfter=Jun 15 12:00:00 2026 GMT
        end_date_str = result.stdout.strip().split('=')[1]
        end_date = datetime.strptime(end_date_str, '%b %d %H:%M:%S %Y %Z')
        remaining = (end_date - datetime.utcnow()).days
        
        return {
            'domain': domain,
            'expiry': end_date.isoformat(),
            'remaining_days': remaining,
            'needs_renewal': remaining <= threshold_days
        }
    except Exception as e:
        logger.error(f"检查证书失败 {domain}: {e}")
        return {'domain': domain, 'error': str(e), 'needs_renewal': True}

def renew_cert(domain: str, config: dict) -> bool:
    """通过 certbot 续期证书"""
    try:
        cmd = [
            'certbot', 'certonly',
            '--non-interactive',
            '--agree-tos',
            '--email', config['email'],
            '--server', config['ca_url'],
            '-d', domain,
            '--deploy-hook', f'/usr/local/bin/deploy-cert.sh {domain}'
        ]
        result = subprocess.run(cmd, capture_output=True, text=True, timeout=120)
        if result.returncode == 0:
            logger.info(f"证书续期成功: {domain}")
            return True
        else:
            logger.error(f"证书续期失败: {domain} - {result.stderr}")
            return False
    except subprocess.TimeoutExpired:
        logger.error(f"证书续期超时: {domain}")
        return False
    except Exception as e:
        logger.error(f"证书续期异常: {domain} - {e}")
        return False

def deploy_cert(domain: str) -> bool:
    """部署证书到 Nginx/负载均衡器"""
    try:
        # 验证新证书
        cert_path = f"/etc/letsencrypt/live/{domain}/fullchain.pem"
        key_path = f"/etc/letsencrypt/live/{domain}/privkey.pem"
        
        # 检查证书和密钥匹配
        cert_mod = subprocess.run(
            ['openssl', 'x509', '-noout', '-modulus', '-in', cert_path],
            capture_output=True, text=True, check=True
        )
        key_mod = subprocess.run(
            ['openssl', 'rsa', '-noout', '-modulus', '-in', key_path],
            capture_output=True, text=True, check=True
        )
        
        if cert_mod.stdout.strip() != key_mod.stdout.strip():
            logger.error(f"证书和密钥不匹配: {domain}")
            return False
        
        # 重载 Nginx 配置
        result = subprocess.run(
            ['nginx', '-s', 'reload'],
            capture_output=True, text=True, check=True
        )
        logger.info(f"证书部署成功: {domain}")
        return True
    except Exception as e:
        logger.error(f"证书部署失败: {domain} - {e}")
        return False

def main():
    """主流程：检查所有证书，续期即将过期的"""
    domains = [
        'www.example.com',
        'api.example.com',
        'admin.example.com',
        # 从配置文件或数据库加载完整列表
    ]
    
    renewal_count = 0
    failure_count = 0
    
    for domain in domains:
        cert_path = f"/etc/letsencrypt/live/{domain}/fullchain.pem"
        if not Path(cert_path).exists():
            logger.warning(f"证书文件不存在: {domain}")
            continue
        
        status = check_cert_expiry(domain, cert_path, threshold_days=30)
        logger.info(f"证书状态: {domain} | 剩余 {status.get('remaining_days', '?')} 天")
        
        if status.get('needs_renewal'):
            config = ACME_CONFIGS['production']
            if renew_cert(domain, config):
                renewal_count += 1
            else:
                failure_count += 1
    
    logger.info(f"续期完成: 成功 {renewal_count}, 失败 {failure_count}")
    
    # 发送告警通知
    if failure_count > 0:
        logger.error(f"有 {failure_count} 个证书续期失败，请立即检查！")
        sys.exit(1)

if __name__ == '__main__':
    main()