证书生命周期自动化管理实战：用 Python 构建企业级证书监控与告警系统

#!/usr/bin/env python3
"""
cert_monitor.py - 证书生命周期监控引擎
"""

import ssl
import socket
import datetime
import json
import logging
from dataclasses import dataclass, field, asdict
from typing import List, Optional, Dict, Any
from pathlib import Path

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


@dataclass
class CertInfo:
    """证书信息数据类"""
    hostname: str
    port: int
    subject: str
    issuer: str
    serial_number: str
    not_before: str
    not_after: str
    days_remaining: int
    signature_algorithm: str
    key_size: int
    san_domains: List[str] = field(default_factory=list)
    fingerprint_sha256: str = ""
    chain_valid: bool = True
    error: Optional[str] = None


def get_certificate_info(hostname: str, port: int = 443, timeout: int = 10) -> CertInfo:
    """
    获取远程主机的 TLS 证书信息
    
    参数:
        hostname: 目标主机名
        port: 端口号（默认 443）
        timeout: 连接超时秒数
    
    返回:
        CertInfo 对象
    """
    context = ssl.create_default_context()
    
    try:
        with socket.create_connection((hostname, port), timeout=timeout) as sock:
            with context.wrap_socket(sock, server_hostname=hostname) as ssock:
                # 获取对端证书（二进制 DER 格式）
                der_cert = ssock.getpeercert(binary_form=True)
                # 获取证书链
                cert_chain = ssock.get_verified_chain()
                
                # 使用 cryptography 解析证书
                from cryptography import x509
                cert = x509.load_der_x509_certificate(der_cert)
                
                # 提取 Subject
                subject_parts = []
                for attr in cert.subject:
                    subject_parts.append(f"{attr.oid._name}={attr.value}")
                subject = ", ".join(subject_parts)
                
                # 提取 Issuer
                issuer_parts = []
                for attr in cert.issuer:
                    issuer_parts.append(f"{attr.oid._name}={attr.value}")
                issuer = ", ".join(issuer_parts)
                
                # 提取 SAN（Subject Alternative Name）
                san_domains = []
                try:
                    san_ext = cert.extensions.get_extension_for_class(
                        x509.SubjectAlternativeName
                    )
                    san_domains = san_ext.value.get_values_for_type(x509.DNSName)
                except x509.ExtensionNotFound:
                    pass
                
                # 计算剩余有效期
                now = datetime.datetime.now(datetime.timezone.utc)
                not_after = cert.not_valid_after_utc if hasattr(cert, 'not_valid_after_utc') else cert.not_valid_after.replace(tzinfo=datetime.timezone.utc)
                not_before = cert.not_valid_before_utc if hasattr(cert, 'not_valid_before_utc') else cert.not_valid_before.replace(tzinfo=datetime.timezone.utc)
                days_remaining = (not_after - now).days
                
                # 签名算法
                sig_algo = cert.signature_algorithm_oid._name
                
                # 密钥大小
                key_size = cert.public_key().key_size if hasattr(cert.public_key(), 'key_size') else 0
                
                # SHA-256 指纹
                fingerprint = cert.fingerprint(
                    __import__('cryptography.hazmat.primitives.hashes', fromlist=['SHA256']).SHA256()
                ).hex()
                
                return CertInfo(
                    hostname=hostname,
                    port=port,
                    subject=subject,
                    issuer=issuer,
                    serial_number=format(cert.serial_number, 'x'),
                    not_before=not_before.isoformat(),
                    not_after=not_after.isoformat(),
                    days_remaining=days_remaining,
                    signature_algorithm=sig_algo,
                    key_size=key_size,
                    san_domains=san_domains,
                    fingerprint_sha256=fingerprint,
                    chain_valid=True,
                )
                
    except ssl.SSLCertVerificationError as e:
        logger.warning(f"[{hostname}:{port}] 证书验证失败: {e}")
        return CertInfo(
            hostname=hostname, port=port,
            subject="", issuer="", serial_number="",
            not_before="", not_after="", days_remaining=-1,
            signature_algorithm="", key_size=0,
            chain_valid=False, error=f"SSL验证失败: {e}"
        )
    except socket.timeout:
        logger.error(f"[{hostname}:{port}] 连接超时")
        return CertInfo(
            hostname=hostname, port=port,
            subject="", issuer="", serial_number="",
            not_before="", not_after="", days_remaining=-1,
            signature_algorithm="", key_size=0,
            chain_valid=False, error="连接超时"
        )
    except Exception as e:
        logger.error(f"[{hostname}:{port}] 扫描异常: {e}")
        return CertInfo(
            hostname=hostname, port=port,
            subject="", issuer="", serial_number="",
            not_before="", not_after="", days_remaining=-1,
            signature_algorithm="", key_size=0,
            chain_valid=False, error=str(e)
        )


def scan_domains(domains: List[Dict[str, Any]]) -> List[CertInfo]:
    """
    批量扫描域名证书
    
    参数:
        domains: 域名列表，每项为 {"hostname": "example.com", "port": 443}
    
    返回:
        CertInfo 列表
    """
    results = []
    for entry in domains:
        hostname = entry["hostname"]
        port = entry.get("port", 443)
        logger.info(f"扫描 {hostname}:{port} ...")
        info = get_certificate_info(hostname, port)
        results.append(info)
        if info.error:
            logger.warning(f"  ⚠️  {info.error}")
        else:
            logger.info(f"  ✅ 剩余 {info.days_remaining} 天 | 密钥 {info.key_size} 位 | {info.issuer}")
    return results


def generate_report(scan_results: List[CertInfo]) -> Dict[str, Any]:
    """
    生成证书扫描报告
    """
    report = {
        "scan_time": datetime.datetime.now().isoformat(),
        "total": len(scan_results),
        "healthy": 0,
        "warning": 0,
        "critical": 0,
        "error": 0,
        "certificates": [],
    }
    
    for cert in scan_results:
        if cert.error:
            report["error"] += 1
            status = "error"
        elif cert.days_remaining <= 7:
            report["critical"] += 1
            status = "critical"
        elif cert.days_remaining <= 30:
            report["warning"] += 1
            status = "warning"
        else:
            report["healthy"] += 1
            status = "healthy"
        
        cert_dict = asdict(cert)
        cert_dict["status"] = status
        report["certificates"].append(cert_dict)
    
    # 按剩余天数排序（问题证书在前）
    report["certificates"].sort(key=lambda x: x["days_remaining"])
    
    return report


if __name__ == "__main__":
    # 示例：扫描几个域名
    domains = [
        {"hostname": "example.com", "port": 443},
        {"hostname": "expired.badssl.com", "port": 443},  # 已过期
        {"hostname": "wrong.host.badssl.com", "port": 443},  # 域名不匹配
    ]
    
    results = scan_domains(domains)
    report = generate_report(results)
    print(json.dumps(report, indent=2, ensure_ascii=False))

#!/usr/bin/env python3
"""
acme_renewer.py - ACME 证书自动续期
基于 RFC 8555，兼容 Let's Encrypt 和 other ACME CA
"""

import json
import time
import datetime
import logging
import hashlib
import base64
from pathlib import Path
from typing import Optional, Dict, Any, Tuple
from dataclasses import dataclass
from urllib.parse import urljoin

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

logger = logging.getLogger("acme_renewer")


@dataclass
class ACMEConfig:
    """ACME 配置"""
    directory_url: str
    account_email: str
    key_type: str = "ecdsa"  # "ecdsa" 或 "rsa"
    key_size: int = 384       # ECDSA: 256 或 384; RSA: 2048 或 4096
    challenge_type: str = "http-01"  # "http-01" 或 "dns-01"
    webroot_path: str = "/var/www/acme-challenge"
    ca_bundle: Optional[str] = None  # 自定义 CA 证书路径


class ACMEClient:
    """ACME 协议客户端"""
    
    def __init__(self, config: ACMEConfig):
        self.config = config
        self.session = requests.Session()
        self.session.headers.update({
            "Content-Type": "application/jose+json",
            "User-Agent": "CertMonitor-ACME/1.0"
        })
        
        # ACME 服务器元数据
        self.directory: Dict[str, Any] = {}
        self.account_url: str = ""
        self.account_key = None
        
        # 初始化
        self._load_or_create_account()
    
    def _get_directory(self) -> Dict[str, Any]:
        """获取 ACME 服务器目录"""
        if not self.directory:
            resp = self.session.get(self.config.directory_url)
            resp.raise_for_status()
            self.directory = resp.json()
            logger.info(f"ACME 目录: {list(self.directory.keys())}")
        return self.directory
    
    def _generate_key(self):
        """生成账户密钥对"""
        if self.config.key_type == "ecdsa":
            self.account_key = ec.generate_private_key(
                ec.SECP384R1() if self.config.key_size == 384 else ec.SECP256R1()
            )
        else:
            self.account_key = rsa.generate_private_key(
                public_exponent=65537,
                key_size=self.config.key_size,
            )
    
    def _load_or_create_account(self):
        """加载已有账户或创建新账户"""
        account_file = Path(f"acme_account_{self.config.key_type}.json")
        
        if account_file.exists():
            with open(account_file) as f:
                data = json.load(f)
            self.account_url = data["account_url"]
            # 从 JWK 恢复密钥
            self.account_key = self._load_key_from_jwk(data["jwk"])
            logger.info(f"已加载 ACME 账户: {self.account_url}")
        else:
            self._generate_key()
            self._register_account(account_file)
    
    def _load_key_from_jwk(self, jwk: dict):
        """从 JWK 格式加载私钥"""
        if jwk.get("kty") == "EC":
            from cryptography.hazmat.primitives.asymmetric.ec import (
                EllipticCurvePublicNumbers, EllipticCurvePrivateNumbers
            )
            curve = ec.SECP384R1() if jwk.get("crv") == "P-384" else ec.SECP256R1()
            x = int.from_bytes(base64.urlsafe_b64decode(jwk["x"] + "=="), "big")
            y = int.from_bytes(base64.urlsafe_b64decode(jwk["y"] + "=="), "big")
            d = int.from_bytes(base64.urlsafe_b64decode(jwk["d"] + "=="), "big")
            
            pub_numbers = EllipticCurvePublicNumbers(x, y, curve)
            priv_numbers = EllipticCurvePrivateNumbers(d, pub_numbers)
            return priv_numbers.private_key()
        else:
            from cryptography.hazmat.primitives.asymmetric.rsa import (
                RSAPublicNumbers, RSAPrivateNumbers
            )
            n = int.from_bytes(base64.urlsafe_b64decode(jwk["n"] + "=="), "big")
            e = int.from_bytes(base64.urlsafe_b64decode(jwk["e"] + "=="), "big")
            d = int.from_bytes(base64.urlsafe_b64decode(jwk["d"] + "=="), "big")
            p = int.from_bytes(base64.urlsafe_b64decode(jwk["p"] + "=="), "big")
            q = int.from_bytes(base64.urlsafe_b64decode(jwk["q"] + "=="), "big")
            
            pub = RSAPublicNumbers(e, n)
            priv = RSAPrivateNumbers(p, q, d, pub)
            return priv.private_key()
    
    def _register_account(self, account_file: Path):
        """注册 ACME 账户"""
        directory = self._get_directory()
        
        # 构建 JWK 头部
        jwk = self._get_jwk_header()
        
        # 注册请求
        payload = {
            "termsOfServiceAgreed": True,
            "contact": [f"mailto:{self.config.account_email}"]
        }
        
        protected = self._sign_payload(jwk, directory["newAccount"], payload)
        
        resp = self.session.post(directory["newAccount"], json=protected)
        
        if resp.status_code == 201:
            self.account_url = resp.headers["Location"]
            logger.info(f"ACME 账户注册成功: {self.account_url}")
            
            # 保存账户信息
            account_data = {
                "account_url": self.account_url,
                "jwk": jwk,
            }
            with open(account_file, "w") as f:
                json.dump(account_data, f, indent=2)
        elif resp.status_code == 200:
            self.account_url = resp.headers["Location"]
            logger.info(f"ACME 账户已存在: {self.account_url}")
        else:
            raise Exception(f"账户注册失败: {resp.status_code} {resp.text}")
    
    def _get_jwk_header(self) -> dict:
        """获取 JWK 公钥表示"""
        pub_key = self.account_key.public_key()
        
        if self.config.key_type == "ecdsa":
            from cryptography.hazmat.primitives.asymmetric.ec import EllipticCurvePublicKey
            ec_pub = pub_key.public_numbers()
            x_bytes = ec_pub.x.to_bytes(48, "big")  # P-384
            y_bytes = ec_pub.y.to_bytes(48, "big")
            
            return {
                "kty": "EC",
                "crv": "P-384" if self.config.key_size == 384 else "P-256",
                "x": base64.urlsafe_b64encode(x_bytes).rstrip(b"=").decode(),
                "y": base64.urlsafe_b64encode(y_bytes).rstrip(b"=").decode(),
            }
        else:
            rsa_pub = pub_key.public_numbers()
            n_bytes = rsa_pub.n.to_bytes((rsa_pub.n.bit_length() + 7) // 8, "big")
            e_bytes = rsa_pub.e.to_bytes(3, "big")
            
            return {
                "kty": "RSA",
                "n": base64.urlsafe_b64encode(n_bytes).rstrip(b"=").decode(),
                "e": base64.urlsafe_b64encode(e_bytes).rstrip(b"=").decode(),
            }
    
    def _sign_payload(self, jwk: dict, url: str, payload: dict) -> dict:
        """使用 JWS (JSON Web Signature) 签名请求"""
        # 计算 JWK 指纹 (thumbprint)
        jwk_json = json.dumps(jwk, sort_keys=True, separators=(",", ":"))
        thumbprint = hashlib.sha256(jwk_json.encode()).digest()
        
        protected = {
            "alg": "ES256" if self.config.key_type == "ecdsa" else "RS256",
            "jwk": jwk,
            "nonce": self._get_nonce(),
            "url": url,
        }
        
        # Base64url 编码
        protected_b64 = base64.urlsafe_b64encode(
            json.dumps(protected, separators=(",", ":")).encode()
        ).rstrip(b"=").decode()
        
        payload_b64 = base64.urlsafe_b64encode(
            json.dumps(payload, separators=(",", ":")).encode()
        ).rstrip(b"=").decode()
        
        # 签名
        signing_input = f"{protected_b64}.{payload_b64}".encode()
        
        if self.config.key_type == "ecdsa":
            from cryptography.hazmat.primitives.asymmetric.ec import ECDSA
            signature = self.account_key.sign(
                signing_input, ECDSA(hashes.SHA256())
            )
            # ECDSA 签名需要转换为 IEEE P1363 格式 (r || s)
            from cryptography.hazmat.primitives.asymmetric.utils import (
                decode_dss_signature
            )
            r, s = decode_dss_signature(signature)
            key_len = 48 if self.config.key_size == 384 else 32
            sig_bytes = r.to_bytes(key_len, "big") + s.to_bytes(key_len, "big")
        else:
            signature = self.account_key.sign(
                signing_input,
                padding.PKCS1v15(),
                hashes.SHA256()
            )
            sig_bytes = signature
        
        sig_b64 = base64.urlsafe_b64encode(sig_bytes).rstrip(b"=").decode()
        
        return {
            "protected": protected_b64,
            "payload": payload_b64,
            "signature": sig_b64,
        }
    
    def _get_nonce(self) -> str:
        """获取 ACME nonce"""
        directory = self._get_directory()
        resp = self.session.head(directory["newNonce"])
        return resp.headers["Replay-Nonce"]
    
    def request_certificate(self, domains: list[str]) -> Tuple[str, str]:
        """
        申请证书
        
        参数:
            domains: 域名列表（第一个为 CN）
        
        返回:
            (证书 PEM, 私钥 PEM) 元组
        """
        directory = self._get_directory()
        
        # 1. 生成证书密钥对
        cert_key = ec.generate_private_key(ec.SECP256R1())
        
        # 2. 构建 CSR
        csr_builder = x509.CertificateSigningRequestBuilder()
        csr_builder = csr_builder.subject_name(x509.Name([
            x509.NameAttribute(NameOID.COMMON_NAME, domains[0])
        ]))
        
        # 添加 SAN
        san_names = [x509.DNSName(d) for d in domains]
        csr_builder = csr_builder.add_extension(
            x509.SubjectAlternativeName(san_names),
            critical=False,
        )
        
        csr = csr_builder.sign(cert_key, hashes.SHA256())
        csr_der = csr.public_bytes(serialization.Encoding.DER)
        csr_b64 = base64.urlsafe_b64encode(csr_der).rstrip(b"=").decode()
        
        # 3. 创建订单
        jwk = self._get_jwk_header()
        order_payload = {
            "identifiers": [
                {"type": "dns", "value": d} for d in domains
            ]
        }
        
        protected = self._sign_payload(jwk, directory["newOrder"], order_payload)
        resp = self.session.post(directory["newOrder"], json=protected)
        resp.raise_for_status()
        
        order = resp.json()
        order_url = resp.headers["Location"]
        logger.info(f"订单创建成功: {order_url}")
        
        # 4. 完成挑战
        auth_urls = order["authorizations"]
        for auth_url in auth_urls:
            self._complete_challenge(auth_url)
        
        # 5. 轮询订单状态
        while True:
            resp = self.session.post(
                order_url,
                json=self._sign_payload(jwk, order_url, {})
            )
            order = resp.json()
            if order["status"] == "valid":
                break
            elif order["status"] == "invalid":
                raise Exception(f"订单失败: {order.get('error', '未知错误')}")
            logger.info(f"订单状态: {order['status']}，等待 5 秒...")
            time.sleep(5)
        
        # 6. 下载证书
        final_url = order["finalize"]
        finalize_payload = {"csr": csr_b64}
        resp = self.session.post(
            final_url,
            json=self._sign_payload(jwk, final_url, finalize_payload)
        )
        resp.raise_for_status()
        
        # 轮询最终化
        while True:
            resp = self.session.post(
                order_url,
                json=self._sign_payload(jwk, order_url, {})
            )
            order = resp.json()
            if order["status"] == "valid":
                break
            time.sleep(5)
        
        # 下载证书链
        cert_url = order["certificate"]
        resp = self.session.get(cert_url)
        cert_pem = resp.text
        
        # 导出私钥
        key_pem = cert_key.private_bytes(
            serialization.Encoding.PEM,
            serialization.PrivateFormat.PKCS8,
            serialization.NoEncryption()
        ).decode()
        
        logger.info(f"证书申请成功: {domains[0]}")
        return cert_pem, key_pem
    
    def _complete_challenge(self, auth_url: str):
        """完成域名验证挑战"""
        # 获取挑战详情
        resp = self.session.post(
            auth_url,
            json=self._sign_payload(self._get_jwk_header(), auth_url, {})
        )
        auth = resp.json()
        
        # 找到 HTTP-01 挑战
        challenge = None
        for c in auth.get("challenges", []):
            if c["type"] == self.config.challenge_type:
                challenge = c
                break
        
        if not challenge:
            raise Exception(f"未找到 {self.config.challenge_type} 挑战")
        
        token = challenge["token"]
        
        if self.config.challenge_type == "http-01":
            self._handle_http01_challenge(token)
        elif self.config.challenge_type == "dns-01":
            self._handle_dns01_challenge(token, auth["identifier"]["value"])
        
        # 通知 CA 验证完成
        resp = self.session.post(
            challenge["url"],
            json=self._sign_payload(self._get_jwk_header(), challenge["url"], {})
        )
        
        # 轮询挑战状态
        while True:
            time.sleep(3)
            resp = self.session.post(
                auth_url,
                json=self._sign_payload(self._get_jwk_header(), auth_url, {})
            )
            auth = resp.json()
            if auth["status"] == "valid":
                logger.info(f"挑战完成: {auth['identifier']['value']}")
                break
            elif auth["status"] == "invalid":
                raise Exception(f"挑战失败: {auth}")
            logger.info(f"挑战状态: {auth['status']}...")
    
    def _handle_http01_challenge(self, token: str):
        """处理 HTTP-01 挑战：在 webroot 创建验证文件"""
        webroot = Path(self.config.webroot_path)
        webroot.mkdir(parents=True, exist_ok=True)
        
        # 计算 key authorization
        jwk = self._get_jwk_header()
        jwk_json = json.dumps(jwk, sort_keys=True, separators=(",", ":"))
        thumbprint = hashlib.sha256(jwk_json.encode()).digest()
        key_auth = token + "." + base64.urlsafe_b64encode(thumbprint).rstrip(b"=").decode()
        
        # 写入验证文件
        challenge_dir = webroot / ".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"HTTP-01 验证文件已创建: {challenge_file}")
        
        # 清理函数（续期成功后调用）
        self._pending_cleanup = getattr(self, "_pending_cleanup", [])
        self._pending_cleanup.append(challenge_file)
    
    def _handle_dns01_challenge(self, token: str, domain: str):
        """处理 DNS-01 挑战：返回需要设置的 TXT 记录"""
        jwk = self._get_jwk_header()
        jwk_json = json.dumps(jwk, sort_keys=True, separators=(",", ":"))
        thumbprint = hashlib.sha256(jwk_json.encode()).digest()
        key_auth = token + "." + base64.urlsafe_b64encode(thumbprint).rstrip(b"=").decode()
        
        # 计算 TXT 记录值
        txt_value = base64.urlsafe_b64encode(
            hashlib.sha256(key_auth.encode()).digest()
        ).rstrip(b"=").decode()
        
        logger.info(f"请在 DNS 中设置 TXT 记录:")
        logger.info(f"  名称: _acme-challenge.{domain}")
        logger.info(f"  值: {txt_value}")
        
        # 等待 DNS 传播
        logger.info("等待 DNS 传播（30 秒）...")
        time.sleep(30)

#!/usr/bin/env python3
"""
notifier.py - 多渠道告警通知
支持 Webhook、邮件、企业微信
"""

import json
import logging
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from typing import List, Dict, Any, Optional
from dataclasses import dataclass
from datetime import datetime

import requests

logger = logging.getLogger("notifier")


@dataclass
class AlertRule:
    """告警规则"""
    name: str
    days_threshold: int  # 剩余天数阈值
    severity: str       # "info", "warning", "critical"
    notify_channels: List[str]  # ["webhook", "email", "wechat"]


class Notifier:
    """多渠道通知器"""
    
    def __init__(self, config: Dict[str, Any]):
        self.config = config
        self.rules = [
            AlertRule("critical", 7, "critical", ["webhook", "email"]),
            AlertRule("warning", 30, "warning", ["webhook"]),
            AlertRule("info", 60, "info", ["webhook"]),
        ]
    
    def evaluate_and_notify(self, scan_results: list):
        """评估证书状态并发送告警"""
        alerts = self._evaluate(scan_results)
        
        if not alerts:
            logger.info("所有证书状态正常，无需告警")
            return
        
        for alert in alerts:
            for channel in alert["notify_channels"]:
                try:
                    if channel == "webhook":
                        self._send_webhook(alert)
                    elif channel == "email":
                        self._send_email(alert)
                    elif channel == "wechat":
                        self._send_wechat(alert)
                    logger.info(f"告警已发送 [{channel}]: {alert['message']}")
                except Exception as e:
                    logger.error(f"发送失败 [{channel}]: {e}")
    
    def _evaluate(self, scan_results: list) -> list[dict]:
        """评估证书状态，生成告警"""
        alerts = []
        
        for cert in scan_results:
            if cert.get("error"):
                alerts.append({
                    "severity": "critical",
                    "hostname": cert["hostname"],
                    "message": f"❌ {cert['hostname']}:{cert['port']} 证书异常 - {cert['error']}",
                    "notify_channels": ["webhook", "email"],
                    "data": cert,
                })
                continue
            
            # 从最严格到最宽松匹配，确保触发最高级别告警
            for rule in self.rules:
                if cert["days_remaining"] <= rule.days_threshold:
                    severity_emoji = {"critical": "🔴", "warning": "🟡", "info": "🔵"}
                    alerts.append({
                        "severity": rule.severity,
                        "hostname": cert["hostname"],
                        "message": (
                            f"{severity_emoji.get(rule.severity, '⚪')} "
                            f"{cert['hostname']} 证书将在 {cert['days_remaining']} 天后过期 "
                            f"(阈值: {rule.days_threshold} 天)"
                        ),
                        "notify_channels": rule.notify_channels,
                        "data": cert,
                    })
                    break  # 只触发最高级别的告警
        
        return alerts
    
    def _send_webhook(self, alert: dict):
        """发送 Webhook 通知"""
        webhook_url = self.config.get("webhook_url")
        if not webhook_url:
            return
        
        payload = {
            "text": alert["message"],
            "severity": alert["severity"],
            "hostname": alert["hostname"],
            "timestamp": datetime.now().isoformat(),
        }
        
        resp = requests.post(webhook_url, json=payload, timeout=10)
        resp.raise_for_status()
    
    def _send_email(self, alert: dict):
        """发送邮件通知"""
        smtp_config = self.config.get("smtp", {})
        if not smtp_config:
            return
        
        msg = MIMEMultipart()
        msg["From"] = smtp_config["from"]
        msg["To"] = ", ".join(smtp_config["to"])
        msg["Subject"] = f"[证书告警] {alert['severity'].upper()} - {alert['hostname']}"
        
        body = f"""
证书生命周期告警

{alert['message']}

详细信息:
- 主机: {alert['data']['hostname']}:{alert['data']['port']}
- 剩余天数: {alert['data']['days_remaining']}
- 颁发者: {alert['data']['issuer']}
- 到期时间: {alert['data']['not_after']}
- 签名算法: {alert['data']['signature_algorithm']}

<hr>
由 CertMonitor 自动发送
        """
        
        msg.attach(MIMEText(body, "plain", "utf-8"))
        
        with smtplib.SMTP(smtp_config["host"], smtp_config.get("port", 587)) as server:
            if smtp_config.get("use_tls", True):
                server.starttls()
            if smtp_config.get("username"):
                server.login(smtp_config["username"], smtp_config["password"])
            server.send_message(msg)
    
    def _send_wechat(self, alert: dict):
        """发送企业微信通知"""
        wechat_webhook = self.config.get("wechat_webhook")
        if not wechat_webhook:
            return
        
        payload = {
            "msgtype": "text",
            "text": {
                "content": alert["message"]
            }
        }
        
        resp = requests.post(wechat_webhook, json=payload, timeout=10)
        resp.raise_for_status()

#!/usr/bin/env python3
"""
main.py - 证书监控系统主入口
"""

import sys
import yaml
import logging
import argparse
from pathlib import Path

from cert_monitor import scan_domains, generate_report
from acme_renewer import ACMEClient, ACMEConfig
from notifier import Notifier

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(name)s - %(message)s"
)
logger = logging.getLogger("main")


def load_config(config_path: str) -> dict:
    """加载 YAML 配置文件"""
    with open(config_path) as f:
        return yaml.safe_load(f)


def main():
    parser = argparse.ArgumentParser(description="证书生命周期监控系统")
    parser.add_argument("-c", "--config", default="config.yaml", help="配置文件路径")
    parser.add_argument("--scan-only", action="store_true", help="仅扫描，不续期")
    parser.add_argument("--report", action="store_true", help="生成 JSON 报告")
    args = parser.parse_args()
    
    config = load_config(args.config)
    
    # 1. 扫描证书
    logger.info("=" * 60)
    logger.info("开始证书扫描...")
    domains = config.get("domains", [])
    scan_results = scan_domains(domains)
    
    # 2. 评估并告警
    notifier = Notifier(config.get("notifications", {}))
    notifier.evaluate_and_notify(scan_results)
    
    # 3. 自动续期（如果启用且非仅扫描模式）
    if not args.scan_only and config.get("auto_renewal", {}).get("enabled"):
        acme_config = ACMEConfig(
            directory_url=config["auto_renewal"]["acme_directory"],
            account_email=config["auto_renewal"]["account_email"],
            webroot_path=config["auto_renewal"].get("webroot", "/var/www/acme-challenge"),
        )
        
        client = ACMEClient(acme_config)
        
        # 找出需要续期的证书（30 天内到期）
        for cert in scan_results:
            if 0 < cert.days_remaining <= 30 and not cert.error:
                domains_to_renew = [cert.hostname] + [
                    d for d in cert.san_domains if d != cert.hostname
                ]
                logger.info(f"续期: {cert.hostname} (剩余 {cert.days_remaining} 天)")
                try:
                    cert_pem, key_pem = client.request_certificate(domains_to_renew)
                    # 保存证书
                    output_dir = Path(config["auto_renewal"].get("output_dir", "./certs"))
                    output_dir.mkdir(parents=True, exist_ok=True)
                    
                    cert_file = output_dir / f"{cert.hostname}.crt"
                    key_file = output_dir / f"{cert.hostname}.key"
                    
                    cert_file.write_text(cert_pem)
                    key_file.write_text(key_pem)
                    
                    logger.info(f"证书已保存: {cert_file}")
                except Exception as e:
                    logger.error(f"续期失败 {cert.hostname}: {e}")
    
    # 4. 生成报告
    if args.report:
        report = generate_report(scan_results)
        report_path = Path("cert_report.json")
        report_path.write_text(
            json.dumps(report, indent=2, ensure_ascii=False)
        )
        logger.info(f"报告已生成: {report_path}")
    
    # 输出摘要
    healthy = sum(1 for c in scan_results if not c.error and c.days_remaining > 30)
    warning = sum(1 for c in scan_results if not c.error and 7 < c.days_remaining <= 30)
    critical = sum(1 for c in scan_results if not c.error and c.days_remaining <= 7)
    errors = sum(1 for c in scan_results if c.error)
    
    logger.info("=" * 60)
    logger.info(f"扫描完成: 总计 {len(scan_results)} 个域名")
    logger.info(f"  ✅ 健康: {healthy}")
    logger.info(f"  ⚠️  警告: {warning}")
    logger.info(f"  🔴 紧急: {critical}")
    logger.info(f"  ❌ 错误: {errors}")


if __name__ == "__main__":
    main()

# 待监控的域名列表
domains:
  - hostname: "www.example.com"
    port: 443
  - hostname: "api.example.com"
    port: 443
  - hostname: "mail.example.com"
    port: 443

# 自动续期配置
auto_renewal:
  enabled: true
  acme_directory: "https://acme-v02.api.letsencrypt.org/directory"
  account_email: "admin@example.com"
  webroot: "/var/www/acme-challenge"
  output_dir: "/etc/nginx/ssl"

# 通知配置
notifications:
  webhook_url: "https://hooks.slack.com/services/YOUR/WEBHOOK/URL"
  wechat_webhook: "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=YOUR_KEY"
  smtp:
    host: "smtp.example.com"
    port: 587
    use_tls: true
    username: "alert@example.com"
    password: "your-password"
    from: "alert@example.com"
    to:
      - "admin@example.com"
      - "ops@example.com"

# /etc/systemd/system/cert-monitor.service
[Unit]
Description=Certificate Lifecycle Monitor
After=network.target

[Service]
Type=oneshot
User=certmon
Group=certmon
WorkingDirectory=/opt/cert-monitor
ExecStart=/opt/cert-monitor/cert-monitor-env/bin/python main.py --report
StandardOutput=journal
StandardError=journal

[Install]
WantedBy=multi-user.target

server {
    listen 80;
    server_name example.com;
    
    # ACME 验证必须放在重定向之前
    location /.well-known/acme-challenge/ {
        root /var/www/acme-challenge;
    }
    
    # 其他请求重定向到 HTTPS
    location / {
        return 301 https://$host$request_uri;
    }
}