ACME 协议实战：用 Python 构建自动化证书管理工具，应对 47 天有效期挑战

┌──────────┐                        ┌──────────┐
│  Client  │                        │  ACME CA │
│ (Python) │                        │(Let's    │
│          │                        │ Encrypt) │
└────┬─────┘                        └────┬─────┘
     │  1. 创建账户                        │
     │ ──────────────────────────────────> │
     │  2. 返回账户 URL + 公钥指纹          │
     │ <────────────────────────────────── │
     │  3. 创建订单（申请证书）              │
     │ ──────────────────────────────────> │
     │  4. 返回订单 + 授权挑战               │
     │ <────────────────────────────────── │
     │  5. 完成挑战（HTTP-01/DNS-01）        │
     │ ──────────────────────────────────> │
     │  6. CA 验证通过，返回证书             │
     │ <────────────────────────────────── │
     │  7. 下载证书链                        │
     │ ──────────────────────────────────> │
     │  8. 返回完整证书 + 中间证书            │
     │ <────────────────────────────────── │

#!/usr/bin/env python3
"""
acme_client.py — ACME 协议核心客户端
支持 RFC 8555 标准流程：注册 → 下单 → 挑战 → 签发
"""

import json
import time
import base64
import hashlib
import logging
from pathlib import Path
from typing import Optional
from datetime import datetime, timedelta

import requests
from cryptography import x509
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import rsa, ec, padding
from cryptography.hazmat.backends import default_backend
from cryptography.x509.oid import NameOID

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s [%(levelname)s] %(message)s'
)
logger = logging.getLogger(__name__)


def b64url(data: bytes) -> str:
    """Base64url 编码（无填充）"""
    return base64.urlsafe_b64encode(data).rstrip(b'=').decode('ascii')


def generate_account_key(key_path: str = "account.key") -> ec.EllipticCurvePrivateKey:
    """生成 ACME 账户密钥（EC P-256）"""
    if Path(key_path).exists():
        with open(key_path, 'rb') as f:
            return serialization.load_pem_private_key(f.read(), password=None)
    
    key = ec.generate_private_key(ec.SECP256R1(), default_backend())
    with open(key_path, 'wb') as f:
        f.write(key.private_bytes(
            serialization.Encoding.PEM,
            serialization.PrivateFormat.PKCS8,
            serialization.NoEncryption()
        ))
    logger.info(f"账户密钥已生成: {key_path}")
    return key


def generate_csr_key(key_path: str = "domain.key") -> ec.EllipticCurvePrivateKey:
    """生成域名证书密钥"""
    if Path(key_path).exists():
        with open(key_path, 'rb') as f:
            return serialization.load_pem_private_key(f.read(), password=None)
    
    key = ec.generate_private_key(ec.SECP256R1(), default_backend())
    with open(key_path, 'wb') as f:
        f.write(key.private_bytes(
            serialization.Encoding.PEM,
            serialization.PrivateFormat.PKCS8,
            serialization.NoEncryption()
        ))
    logger.info(f"域名密钥已生成: {key_path}")
    return key


def sign_jws(key, payload: dict, url: str, nonce: str) -> dict:
    """构造 JWS (JSON Web Signature) 请求体"""
    # JWS Header
    header = {
        "alg": "ES256",
        "url": url,
        "nonce": nonce,
        "jwk": {
            "kty": "EC",
            "crv": "P-256",
            "x": b64url(key.public_key().public_numbers().x.to_bytes(32, 'big')),
            "y": b64url(key.public_key().public_numbers().y.to_bytes(32, 'big'))
        }
    }
    
    # 编码 header 和 payload
    header_b64 = b64url(json.dumps(header, separators=(',', ':')).encode())
    payload_b64 = b64url(json.dumps(payload, separators=(',', ':')).encode())
    
    # 签名
    signing_input = f"{header_b64}.{payload_b64}".encode()
    signature = key.sign(signing_input, ec.ECDSA(hashes.SHA256()))
    
    # 将 DER 签名转换为 P1363 格式 (IEEE 1363)
    from cryptography.hazmat.primitives.asymmetric.utils import decode_dss_signature
    r, s = decode_dss_signature(signature)
    sig_bytes = r.to_bytes(32, 'big') + s.to_bytes(32, 'big')
    
    return {
        "protected": header_b64,
        "payload": payload_b64,
        "signature": b64url(sig_bytes)
    }


class ACMEClient:
    """
    ACME 协议客户端
    支持 Let's Encrypt (生产/测试) 和自定义 ACME CA
    """
    
    # Let's Encrypt 端点
    STAGING_DIRECTORY = "https://acme-staging-v02.api.letsencrypt.org/directory"
    PRODUCTION_DIRECTORY = "https://acme-v02.api.letsencrypt.org/directory"
    
    def __init__(self, ca_url: str = None, email: str = None, staging: bool = True):
        """
        初始化 ACME 客户端
        
        Args:
            ca_url: 自定义 ACME CA URL，None 则使用 Let's Encrypt
            email: 账户邮箱（用于紧急通知和续期提醒）
            staging: 是否使用测试环境（建议先用测试环境验证）
        """
        self.staging = staging
        self.email = email
        self.ca_url = ca_url or (self.STAGING_DIRECTORY if staging else self.PRODUCTION_DIRECTORY)
        
        # 获取 CA 目录
        self.directory = requests.get(self.ca_url, timeout=30).json()
        logger.info(f"ACME CA: {self.directory.get('meta', {}).get('termsOfService', 'N/A')}")
        
        # 加载或生成账户密钥
        self.account_key = generate_account_key(
            "account.key" if staging else "account_prod.key"
        )
        
        # 账户 URL（注册后填充）
        self.account_url: Optional[str] = None
        
        # nonce 缓存
        self._last_nonce: Optional[str] = None
    
    def _get_nonce(self) -> str:
        """获取防重放攻击的 nonce"""
        if self._last_nonce:
            return self._last_nonce
        resp = requests.head(self.directory['newNonce'], timeout=30)
        self._last_nonce = resp.headers['Replay-Nonce']
        return self._last_nonce
    
    def _request(self, url: str, payload: dict = None) -> requests.Response:
        """发送 JWS 签名的 ACME 请求"""
        nonce = self._get_nonce()
        body = sign_jws(self.account_key, payload or {}, url, nonce)
        
        resp = requests.post(url, json=body, headers={
            'Content-Type': 'application/jose+json'
        }, timeout=30)
        
        # 保存下一个 nonce
        if 'Replay-Nonce' in resp.headers:
            self._last_nonce = resp.headers['Replay-Nonce']
        
        return resp
    
    def register_account(self) -> dict:
        """注册 ACME 账户"""
        payload = {
            "termsOfServiceAgreed": True,
            "contact": [f"mailto:{self.email}"] if self.email else []
        }
        
        resp = self._request(self.directory['newAccount'], payload)
        
        if resp.status_code == 201:
            self.account_url = resp.headers['Location']
            logger.info(f"账户注册成功: {self.account_url}")
        elif resp.status_code == 200:
            self.account_url = resp.headers['Location']
            logger.info(f"账户已存在: {self.account_url}")
        else:
            raise Exception(f"账户注册失败: {resp.status_code} {resp.text}")
        
        return resp.json()
    
    def create_order(self, domains: list[str]) -> dict:
        """
        创建证书申请订单
        
        Args:
            domains: 域名列表（第一个为主域名，其余为 SAN）
        """
        identifiers = [{"type": "dns", "value": d} for d in domains]
        payload = {"identifiers": identifiers}
        
        resp = self._request(self.directory['newOrder'], payload)
        
        if resp.status_code != 201:
            raise Exception(f"创建订单失败: {resp.status_code} {resp.text}")
        
        order = resp.json()
        order['url'] = resp.headers['Location']
        logger.info(f"订单创建成功: {order['url']}")
        logger.info(f"状态: {order['status']}, 域名: {domains}")
        
        return order
    
    def get_authorizations(self, order: dict) -> list[dict]:
        """获取订单的所有授权挑战"""
        authorizations = []
        for auth_url in order.get('authorizations', []):
            resp = self._request(auth_url, "")
            auth = resp.json()
            auth['url'] = auth_url
            authorizations.append(auth)
            
            # 提取挑战信息
            domain = auth['identifier']['value']
            challenges = auth.get('challenges', [])
            for c in challenges:
                logger.info(f"  域名 {domain}: 挑战类型={c['type']}, 状态={c['status']}")
        
        return authorizations
    
    def complete_http01_challenge(self, challenge: dict, domain: str) -> dict:
        """
        完成 HTTP-01 挑战
        
        需要在域名对应的 Web 服务器上部署验证文件：
        http://<domain>/.well-known/acme-challenge/<token>
        内容为 key_authorization
        """
        token = challenge['token']
        
        # 计算 key_authorization
        # 格式: token.thumbprint
        # thumbprint 是账户公钥的 SHA-256 摘要（JWK Thumbprint, RFC 7638）
        jwk = {
            "kty": "EC",
            "crv": "P-256",
            "x": b64url(self.account_key.public_key().public_numbers().x.to_bytes(32, 'big')),
            "y": b64url(self.account_key.public_key().public_numbers().y.to_bytes(32, 'big'))
        }
        jwk_json = json.dumps(jwk, sort_keys=True, separators=(',', ':'))
        thumbprint = hashlib.sha256(jwk_json.encode()).digest()
        key_auth = f"{token}.{b64url(thumbprint)}"
        
        # 部署验证文件
        challenge_dir = Path(f"/var/www/acme-challenge/{domain}/.well-known/acme-challenge")
        challenge_dir.mkdir(parents=True, exist_ok=True)
        challenge_file = challenge_dir / token
        challenge_file.write_text(key_auth)
        logger.info(f"验证文件已部署: {challenge_file}")
        
        # 通知 CA 挑战已完成
        resp = self._request(challenge['url'], {})
        result = resp.json()
        logger.info(f"挑战结果: {result['status']}")
        
        return result
    
    def wait_for_order_ready(self, order: dict, timeout: int = 300) -> dict:
        """等待订单状态变为 ready"""
        url = order['url']
        start = time.time()
        
        while time.time() - start < timeout:
            resp = self._request(url, "")
            order = resp.json()
            
            if order['status'] == 'ready':
                logger.info("订单已就绪，可以提交 CSR")
                return order
            elif order['status'] == 'invalid':
                raise Exception(f"订单失败: {order.get('errors', '未知错误')}")
            
            logger.info(f"订单状态: {order['status']}，等待中...")
            time.sleep(5)
        
        raise TimeoutError(f"等待订单就绪超时 ({timeout}s)")
    
    def finalize_order(self, order: dict, domains: list[str]) -> dict:
        """
        提交 CSR 并完成证书签发
        """
        # 生成域名密钥和 CSR
        domain_key = generate_csr_key()
        
        # 构造 CSR
        builder = x509.CertificateSigningRequestBuilder()
        builder = builder.subject_name(x509.Name([
            x509.NameAttribute(NameOID.COMMON_NAME, domains[0])
        ]))
        
        # 添加 SAN
        san_names = [x509.DNSName(d) for d in domains]
        builder = builder.add_extension(
            x509.SubjectAlternativeName(san_names),
            critical=False
        )
        
        csr = builder.sign(domain_key, hashes.SHA256(), default_backend())
        csr_der = csr.public_bytes(serialization.Encoding.DER)
        
        # 提交 CSR
        payload = {"csr": b64url(csr_der)}
        resp = self._request(order['finalize'], payload)
        
        if resp.status_code != 200:
            raise Exception(f"提交 CSR 失败: {resp.status_code} {resp.text}")
        
        result = resp.json()
        logger.info(f"CSR 提交成功，状态: {result['status']}")
        
        return result
    
    def download_certificate(self, order: dict) -> str:
        """下载签发的证书"""
        # 等待证书就绪
        url = order['url']
        for _ in range(60):
            resp = self._request(url, "")
            order = resp.json()
            if order['status'] == 'valid' and 'certificate' in order:
                break
            time.sleep(2)
        
        cert_url = order.get('certificate')
        if not cert_url:
            raise Exception("证书 URL 不可用")
        
        resp = self._request(cert_url, "")
        cert_pem = resp.text
        
        # 保存证书
        cert_path = "certificate.pem"
        with open(cert_path, 'w') as f:
            f.write(cert_pem)
        logger.info(f"证书已保存: {cert_path}")
        
        return cert_path


# ============================================================
# 使用示例
if __name__ == '__main__':
    # 1. 初始化客户端（测试环境）
    client = ACMEClient(
        email="admin@example.com",
        staging=True  # 先用测试环境！
    )
    
    # 2. 注册账户
    client.register_account()
    
    # 3. 创建订单
    order = client.create_order(["example.com", "www.example.com"])
    
    # 4. 获取并完成挑战
    authorizations = client.get_authorizations(order)
    for auth in authorizations:
        domain = auth['identifier']['value']
        for challenge in auth.get('challenges', []):
            if challenge['type'] == 'http-01':
                client.complete_http01_challenge(challenge, domain)
    
    # 5. 等待订单就绪
    order = client.wait_for_order_ready(order)
    
    # 6. 提交 CSR
    order = client.finalize_order(order, ["example.com", "www.example.com"])
    
    # 7. 下载证书
    cert_path = client.download_certificate(order)
    print(f"\n证书已签发: {cert_path}")

#!/usr/bin/env python3
"""
cert_monitor.py — 证书健康检查与续期决策
支持多域名、多环境、告警通知
"""

import subprocess
import json
import logging
from datetime import datetime, timedelta
from pathlib import Path
from dataclasses import dataclass, asdict
from typing import Optional

from cryptography import x509
from cryptography.hazmat.backends import default_backend

logger = logging.getLogger(__name__)


@dataclass
class CertStatus:
    """证书状态"""
    domain: str
    cert_path: str
    subject: str
    issuer: str
    not_before: str
    not_after: str
    remaining_days: int
    needs_renewal: bool
    serial_number: str
    signature_algorithm: str
    key_size: int
    san_domains: list[str]
    error: Optional[str] = None


def check_certificate(cert_path: str, domain: str = None, 
                      threshold_days: int = 30) -> CertStatus:
    """
    检查证书的详细信息和剩余有效期
    
    Args:
        cert_path: 证书文件路径（PEM 格式）
        domain: 域名（用于日志标识）
        threshold_days: 续期阈值（天数）
    
    Returns:
        CertStatus 对象
    """
    try:
        with open(cert_path, 'rb') as f:
            cert_data = f.read()
        
        cert = x509.load_pem_x509_certificate(cert_data, default_backend())
        
        # 提取域名
        try:
            san = cert.extensions.get_extension_for_class(x509.SubjectAlternativeName)
            san_domains = san.value.get_values_for_type(x509.DNSName)
        except x509.ExtensionNotFound:
            san_domains = []
            # 从 CN 提取
            cn = cert.subject.get_attributes_for_oid(x509.oid.NameOID.COMMON_NAME)
            if cn:
                san_domains = [cn[0].value]
        
        # 计算剩余天数
        now = datetime.utcnow()
        not_after = cert.not_valid_after_utc if hasattr(cert, 'not_valid_after_utc') else cert.not_valid_after.replace(tzinfo=None)
        remaining = (not_after - now).days
        
        # 密钥大小
        key_size = cert.public_key().key_size if hasattr(cert.public_key(), 'key_size') else 0
        
        return CertStatus(
            domain=domain or san_domains[0] if san_domains else "unknown",
            cert_path=cert_path,
            subject=str(cert.subject),
            issuer=str(cert.issuer),
            not_before=str(cert.not_valid_before_utc if hasattr(cert, 'not_valid_before_utc') else cert.not_valid_before.replace(tzinfo=None)),
            not_after=str(not_after),
            remaining_days=remaining,
            needs_renewal=remaining <= threshold_days,
            serial_number=str(cert.serial_number),
            signature_algorithm=cert.signature_algorithm_oid._name,
            key_size=key_size,
            san_domains=san_domains
        )
    except Exception as e:
        logger.error(f"检查证书失败 {cert_path}: {e}")
        return CertStatus(
            domain=domain or "unknown",
            cert_path=cert_path,
            subject="", issuer="", not_before="", not_after="",
            remaining_days=-1, needs_renewal=True,
            serial_number="", signature_algorithm="", key_size=0,
            san_domains=[], error=str(e)
        )


def scan_certificates(config_path: str = "/etc/cert-monitor/config.json") -> list[CertStatus]:
    """
    扫描所有证书并返回状态列表
    
    配置文件格式：
    {
        "certificates": [
            {
                "domain": "example.com",
                "cert_path": "/etc/letsencrypt/live/example.com/fullchain.pem",
                "key_path": "/etc/letsencrypt/live/example.com/privkey.pem"
            }
        ],
        "threshold_days": 30
    }
    """
    with open(config_path) as f:
        config = json.load(f)
    
    results = []
    threshold = config.get('threshold_days', 30)
    
    for cert_config in config.get('certificates', []):
        status = check_certificate(
            cert_config['cert_path'],
            cert_config.get('domain'),
            threshold
        )
        results.append(status)
        
        level = logging.WARNING if status.needs_renewal else logging.INFO
        logger.log(level, 
            f"  {status.domain}: 剩余 {status.remaining_days} 天 "
            f"({'⚠️ 需要续期' if status.needs_renewal else '✅ 正常'})"
        )
    
    return results


def generate_renewal_plan(cert_statuses: list[CertStatus]) -> dict:
    """生成续期计划"""
    plan = {
        'timestamp': datetime.utcnow().isoformat(),
        'total': len(cert_statuses),
        'needs_renewal': [],
        'healthy': [],
        'errors': []
    }
    
    for status in cert_statuses:
        if status.error:
            plan['errors'].append(asdict(status))
        elif status.needs_renewal:
            plan['needs_renewal'].append(asdict(status))
        else:
            plan['healthy'].append(asdict(status))
    
    return plan


if __name__ == '__main__':
    logging.basicConfig(level=logging.INFO, format='%(asctime)s [%(levelname)s] %(message)s')
    
    print("=" * 60)
    print("证书健康检查报告")
    print(f"检查时间: {datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S UTC')}")
    print("=" * 60)
    
    statuses = scan_certificates()
    plan = generate_renewal_plan(statuses)
    
    print(f"\n总计: {plan['total']} 张证书")
    print(f"健康: {len(plan['healthy'])} 张")
    print(f"需要续期: {len(plan['needs_renewal'])} 张")
    print(f"检查异常: {len(plan['errors'])} 张")
    
    if plan['needs_renewal']:
        print("\n--- 需要续期的证书 ---")
        for item in plan['needs_renewal']:
            print(f"  ⚠️  {item['domain']}: 剩余 {item['remaining_days']} 天")
    
    if plan['errors']:
        print("\n--- 检查异常 ---")
        for item in plan['errors']:
            print(f"  ❌ {item['domain']}: {item['error']}")

def should_renew(status: CertStatus, cert_validity_days: int = 200) -> tuple[bool, str]:
    """
    智能续期决策
    
    在 47 天证书时代，续期策略需要更激进：
    - 47 天证书：剩余 14 天就开始续期（留出 33 天的安全窗口）
    - 200 天证书：剩余 60 天开始续期
    - 398 天证书：剩余 30 天开始续期（传统策略）
    
    Returns:
        (should_renew, reason)
    """
    if status.error:
        return True, f"证书检查异常: {status.error}"
    
    # 根据证书有效期动态调整续期阈值
    if cert_validity_days <= 47:
        threshold = 14  # 47天证书：剩余14天（约30%）时续期
    elif cert_validity_days <= 200:
        threshold = 60  # 200天证书：剩余60天（约30%）时续期
    else:
        threshold = 30  # 传统策略
    
    if status.remaining_days <= threshold:
        return True, f"剩余 {status.remaining_days} 天，低于阈值 {threshold} 天"
    
    # 检查密钥强度
    if status.key_size < 256:  # EC 密钥小于 256 位
        return True, f"密钥强度不足: {status.key_size} 位"
    
    return False, f"剩余 {status.remaining_days} 天，状态正常"

#!/usr/bin/env python3
"""
cert_deploy.py — 证书自动部署到 Nginx
支持单证书和国密双证书场景
"""

import subprocess
import shutil
import logging
import hashlib
from pathlib import Path
from datetime import datetime

logger = logging.getLogger(__name__)

NGINX_CONF_DIR = Path("/etc/nginx")
NGINX_SSL_DIR = Path("/etc/nginx/ssl")
NGINX_PID_FILE = Path("/run/nginx/nginx.pid")


def verify_cert_chain(cert_path: str, key_path: str) -> bool:
    """验证证书链和密钥匹配"""
    try:
        # 1. 验证证书格式
        result = subprocess.run(
            ['openssl', 'x509', '-in', cert_path, '-noout', '-text'],
            capture_output=True, text=True, check=True
        )
        
        # 2. 验证证书和密钥匹配
        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("证书和密钥不匹配！")
            return False
        
        # 3. 验证证书链完整性
        result = subprocess.run(
            ['openssl', 'verify', '-CAfile', cert_path, cert_path],
            capture_output=True, text=True
        )
        
        logger.info("证书链验证通过")
        return True
    except Exception as e:
        logger.error(f"证书验证失败: {e}")
        return False


def deploy_certificate(domain: str, cert_path: str, key_path: str,
                       is_gm_cert: bool = False) -> bool:
    """
    部署证书到 Nginx
    
    Args:
        domain: 域名
        cert_path: 证书路径
        key_path: 私钥路径
        is_gm_cert: 是否为国密证书（双证书场景）
    """
    # 1. 验证证书
    if not verify_cert_chain(cert_path, key_path):
        return False
    
    # 2. 备份旧证书
    ssl_dir = NGINX_SSL_DIR / domain
    ssl_dir.mkdir(parents=True, exist_ok=True)
    
    timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
    
    old_cert = ssl_dir / ("gm-cert.pem" if is_gm_cert else "cert.pem")
    old_key = ssl_dir / ("gm-key.pem" if is_gm_cert else "key.pem")
    
    if old_cert.exists():
        shutil.copy2(old_cert, ssl_dir / f"cert.pem.bak.{timestamp}")
        logger.info(f"旧证书已备份: {ssl_dir / f'cert.pem.bak.{timestamp}'}")
    
    # 3. 部署新证书
    dest_cert = ssl_dir / ("gm-cert.pem" if is_gm_cert else "cert.pem")
    dest_key = ssl_dir / ("gm-key.pem" if is_gm_cert else "key.pem")
    
    shutil.copy2(cert_path, dest_cert)
    shutil.copy2(key_path, dest_key)
    
    # 设置权限
    dest_cert.chmod(0o644)
    dest_key.chmod(0o600)
    
    logger.info(f"证书已部署: {dest_cert}, {dest_key}")
    
    # 4. 测试 Nginx 配置
    result = subprocess.run(
        ['nginx', '-t'],
        capture_output=True, text=True
    )
    
    if result.returncode != 0:
        logger.error(f"Nginx 配置测试失败: {result.stderr}")
        # 回滚
        backup_cert = ssl_dir / f"cert.pem.bak.{timestamp}"
        if backup_cert.exists():
            shutil.copy2(backup_cert, old_cert)
            logger.info("已回滚到旧证书")
        return False
    
    # 5. 重载 Nginx（不中断服务）
    result = subprocess.run(
        ['nginx', '-s', 'reload'],
        capture_output=True, text=True
    )
    
    if result.returncode == 0:
        logger.info("Nginx 重载成功")
    else:
        logger.error(f"Nginx 重载失败: {result.stderr}")
        return False
    
    return True


def generate_nginx_ssl_config(domain: str, 
                               enable_gm: bool = False) -> str:
    """生成 Nginx SSL 配置片段"""
    ssl_dir = f"/etc/nginx/ssl/{domain}"
    
    config = f"""
    # SSL 配置 — {domain}
    ssl_protocols TLSv1.2 TLSv1.3;
    ssl_ciphers ECDHE-ECDSA-AES128-GCM-SHA256:ECDHE-RSA-AES128-GCM-SHA256:ECDHE-ECDSA-AES256-GCM-SHA384:ECDHE-RSA-AES256-GCM-SHA384;
    ssl_prefer_server_ciphers off;
    ssl_session_cache shared:SSL:10m;
    ssl_session_timeout 1d;
    ssl_session_tickets off;
    
    # OCSP Stapling
    ssl_stapling on;
    ssl_stapling_verify on;
    resolver 8.8.8.8 8.8.4.4 valid=300s;
    resolver_timeout 5s;
    
    # HSTS（谨慎启用，确认 HTTPS 完全正常后再开启）
    # add_header Strict-Transport-Security "max-age=63072000" always;
"""
    
    if enable_gm:
        # 国密双证书场景
        # Nginx 支持在 ssl_certificate 中指定多个证书文件（空格分隔），
        # 服务器会根据客户端支持的密码套件自动选择
        config += f"""
    # 国密双证书配置（Nginx 自动根据客户端能力选择）
    ssl_certificate {ssl_dir}/gm-fullchain.pem {ssl_dir}/intl-fullchain.pem;
    ssl_certificate_key {ssl_dir}/gm-key.pem;
    # 注意：ssl_certificate_key 只支持一个密钥，双证书场景下
    # 国密客户端使用 gm-fullchain + gm-key，国际客户端使用 intl-fullchain + gm-key
    # 如需国际证书使用独立密钥，需配置两个 server 块（SNI 分流）
"""

    else:
        config += f"""
    ssl_certificate {ssl_dir}/cert.pem;
    ssl_certificate_key {ssl_dir}/key.pem;
"""
    return config


if __name__ == '__main__':
    logging.basicConfig(level=logging.INFO, format='%(asctime)s [%(levelname)s] %(message)s')
    
    # 部署示例
    deploy_certificate(
        domain="example.com",
        cert_path="/tmp/certificate.pem",
        key_path="/tmp/domain.key"
    )

#!/usr/bin/env python3
"""
dual_cert_manager.py — 国密双证书自动管理
同时管理 SM2 国密证书和 RSA/ECDSA 国际证书
"""

import json
import logging
from pathlib import Path
from dataclasses import dataclass
from typing import Optional

from cryptography import x509
from cryptography.hazmat.backends import default_backend

logger = logging.getLogger(__name__)


@dataclass
class DualCertSet:
    """双证书集合"""
    domain: str
    # 国密证书
    gm_cert_path: Optional[str] = None
    gm_key_path: Optional[str] = None
    gm_issuer: Optional[str] = None
    gm_remaining_days: int = -1
    # 国际证书
    intl_cert_path: Optional[str] = None
    intl_key_path: Optional[str] = None
    intl_issuer: Optional[str] = None
    intl_remaining_days: int = -1


class DualCertManager:
    """
    国密双证书管理器
    
    核心策略：
    1. 国密证书和国际证书独立续期，互不影响
    2. 优先保证国密证书有效（密评合规要求）
    3. 国际证书作为备选，确保非国密客户端兼容
    """
    
    def __init__(self, config_path: str = "/etc/cert-monitor/dual-cert.json"):
        with open(config_path) as f:
            self.config = json.load(f)
        
        self.domains = self.config.get('domains', [])
        self.gm_ca_api = self.config.get('gm_ca_api', '')  # 国密 CA API
        self.intl_acme_url = self.config.get('intl_acme_url', 
            'https://acme-v02.api.letsencrypt.org/directory')
    
    def check_all(self) -> list[DualCertSet]:
        """检查所有域名的双证书状态"""
        results = []
        
        for domain_config in self.domains:
            domain = domain_config['domain']
            cert_dir = Path(f"/etc/nginx/ssl/{domain}")
            
            dc = DualCertSet(domain=domain)
            
            # 检查国密证书
            gm_cert = cert_dir / "gm-cert.pem"
            if gm_cert.exists():
                dc.gm_cert_path = str(gm_cert)
                dc.gm_key_path = str(cert_dir / "gm-key.pem")
                try:
                    with open(gm_cert, 'rb') as f:
                        cert = x509.load_pem_x509_certificate(f.read(), default_backend())
                    dc.gm_issuer = str(cert.issuer)
                    from datetime import datetime
                    not_after = cert.not_valid_after_utc if hasattr(cert, 'not_valid_after_utc') else cert.not_valid_after.replace(tzinfo=None)
                    dc.gm_remaining_days = (not_after - datetime.utcnow()).days
                except Exception as e:
                    logger.error(f"检查国密证书失败 {domain}: {e}")
            
            # 检查国际证书
            intl_cert = cert_dir / "cert.pem"
            if intl_cert.exists():
                dc.intl_cert_path = str(intl_cert)
                dc.intl_key_path = str(cert_dir / "key.pem")
                try:
                    with open(intl_cert, 'rb') as f:
                        cert = x509.load_pem_x509_certificate(f.read(), default_backend())
                    dc.intl_issuer = str(cert.issuer)
                    not_after = cert.not_valid_after_utc if hasattr(cert, 'not_valid_after_utc') else cert.not_valid_after.replace(tzinfo=None)
                    dc.intl_remaining_days = (not_after - datetime.utcnow()).days
                except Exception as e:
                    logger.error(f"检查国际证书失败 {domain}: {e}")
            
            results.append(dc)
            
            # 日志
            gm_status = f"剩余 {dc.gm_remaining_days} 天" if dc.gm_remaining_days >= 0 else "未部署"
            intl_status = f"剩余 {dc.intl_remaining_days} 天" if dc.intl_remaining_days >= 0 else "未部署"
            logger.info(f"  {domain}: 国密={gm_status}, 国际={intl_status}")
        
        return results
    
    def renewal_plan(self, results: list[DualCertSet]) -> dict:
        """生成续期计划"""
        plan = {'gm_renewals': [], 'intl_renewals': [], 'alerts': []}
        
        for dc in results:
            # 国密证书：剩余 14 天或已过期
            if dc.gm_remaining_days < 14:
                plan['gm_renewals'].append({
                    'domain': dc.domain,
                    'remaining_days': dc.gm_remaining_days,
                    'priority': 'HIGH' if dc.gm_remaining_days < 7 else 'MEDIUM'
                })
            
            # 国际证书：剩余 14 天或已过期
            if dc.intl_remaining_days < 14:
                plan['intl_renewals'].append({
                    'domain': dc.domain,
                    'remaining_days': dc.intl_remaining_days,
                    'priority': 'HIGH' if dc.intl_remaining_days < 7 else 'MEDIUM'
                })
            
            # 告警：国密证书缺失
            if not dc.gm_cert_path:
                plan['alerts'].append(f"⚠️ {dc.domain}: 国密证书未部署（密评不合规风险）")
        
        return plan


# 配置文件示例
DUAL_CERT_CONFIG = {
    "domains": [
        {"domain": "www.example.com", "gm_ca": "cfca"},
        {"domain": "api.example.com", "gm_ca": "bjca"},
        {"domain": "admin.example.com", "gm_ca": "cfca"}
    ],
    "gm_ca_api": "https://gm-ca.example.com/api/v1",
    "intl_acme_url": "https://acme-v02.api.letsencrypt.org/directory",
    "renewal_threshold_days": 14,
    "notification_email": "pki-admin@example.com"
}

# 47 天证书的续期策略
# 签发日: Day 0
# 续期启动: Day 14（剩余 33 天）
# CA 验证: 1-3 天
# 部署窗口: 1-2 天
# 安全余量: 28 天
# 过期日: Day 47

RENEWAL_THRESHOLDS = {
    47: 14,    # 47天证书：剩余14天
    200: 60,   # 200天证书：剩余60天
    398: 30,   # 398天证书：剩余30天
}

┌─────────────────────────────────────────────┐
│              证书管理平台                      │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐  │
│  │ certbot  │  │   lego   │  │ 自研工具  │  │
│  │ (国际)   │  │ (国际)   │  │ (国密)   │  │
│  └────┬─────┘  └────┬─────┘  └────┬─────┘  │
│       │              │              │        │
│  ┌────▼──────────────▼──────────────▼────┐  │
│  │          统一监控 & 续期引擎            │  │
│  │    (本文的 cert_monitor.py)            │  │
│  └────────────────┬──────────────────────┘  │
│                   │                          │
│  ┌────────────────▼──────────────────────┐  │
│  │          自动部署 & 配置管理            │  │
│  │    (本文的 cert_deploy.py)             │  │
│  └───────────────────────────────────────┘  │
└─────────────────────────────────────────────┘