国密证书过期监控实战：从零构建 Python 证书寿命追踪系统

┌─────────────────────────────────────────────────────────────┐
│                    证书监控调度器                              │
│                  (Celery Beat / APScheduler)                  │
└──────────────────┬──────────────────────────────────────────┘
                   │ 定时触发
       ┌───────────┼───────────┐
       ▼           ▼           ▼
┌──────────┐ ┌──────────┐ ┌──────────┐
│ 证书发现  │ │ 状态检查  │ │ 告警通知  │
│ 模块     │ │ 模块     │ │ 模块     │
└────┬─────┘ └────┬─────┘ └────┬─────┘
     │            │            │
     ▼            ▼            ▼
┌─────────────────────────────────────────┐
│         PostgreSQL / Redis 存储         │
│  (证书清单、告警历史、配置参数)            │
└─────────────────────────────────────────┘

#!/usr/bin/env python3
"""
国密证书扫描模块
支持 TCP 端口扫描、TLS 握手、证书解析
"""

import socket
import ssl
import threading
import concurrent.futures
import re
import os
from datetime import datetime, timezone
from cryptography import x509
from cryptography.x509.oid import NameOID
from gmssl import sm2, func
from typing import List, Dict, Optional, Tuple
import queue
import logging

logger = logging.getLogger(__name__)


class CertificateScanner:
    """证书扫描器 - 支持国密 SM2 证书"""
    
    def __init__(self, timeout: int = 5, threads: int = 50):
        self.timeout = timeout
        self.threads = threads
        self.results: List[Dict] = []
        self._lock = threading.Lock()
    
    def scan_host(
        self, 
        host: str, 
        ports: List[int] = None
    ) -> List[Dict]:
        """扫描单个主机的指定端口"""
        if ports is None:
            # 常见 TLS 端口
            ports = [443, 8443, 9500, 9501, 4433, 8444, 1443]
        
        findings = []
        for port in ports:
            cert_info = self._try_extract_cert(host, port)
            if cert_info:
                findings.append(cert_info)
        
        return findings
    
    def _try_extract_cert(
        self, 
        host: str, 
        port: int
    ) -> Optional[Dict]:
        """尝试建立 TLS 连接并提取证书"""
        try:
            context = ssl.create_default_context()
            context.check_hostname = False
            context.verify_mode = ssl.CERT_NONE
            
            with socket.create_connection(
                (host, port), 
                timeout=self.timeout
            ) as sock:
                with context.wrap_socket(sock, server_hostname=host) as ssock:
                    cert_der = ssock.getpeercert(binary_form=True)
                    if not cert_der:
                        return None
                    
                    cert = x509.load_der_x509_certificate(cert_der)
                    return self._parse_certificate(cert, host, port)
                    
        except (socket.timeout, socket.error, ssl.SSLError):
            return None
        except Exception as e:
            logger.debug(f"Connection failed {host}:{port}: {e}")
            return None
    
    def _parse_certificate(
        self, 
        cert: x509.Certificate,
        host: str,
        port: int
    ) -> Dict:
        """解析证书信息"""
        subject = cert.subject
        issuer = cert.issuer
        
        # 提取主题信息
        common_name = None
        for attr in subject:
            if attr.oid == NameOID.COMMON_NAME:
                common_name = attr.value
                break
        
        # 提取签发者信息
        issuer_cn = None
        for attr in issuer:
            if attr.oid == NameOID.COMMON_NAME:
                issuer_cn = attr.value
                break
        
        # 计算有效期
        not_before = cert.not_valid_before_utc
        not_after = cert.not_valid_after_utc
        now_utc = datetime.now(timezone.utc)
        days_remaining = (not_after - now_utc).days
        
        # 检查是否为国密证书（通过 OID 判断）
        is_gm_cert = self._is_gm_certificate(cert)
        
        return {
            'host': host,
            'port': port,
            'common_name': common_name,
            'issuer_cn': issuer_cn,
            'not_before': not_before.isoformat(),
            'not_after': not_after.isoformat(),
            'days_remaining': days_remaining,
            'is_expired': days_remaining < 0,
            'is_expiring_soon': 0 < days_remaining <= 30,
            'is_gm_certificate': is_gm_cert,
            'fingerprint_sha256': cert.fingerprint(x509.HashAlgorithm.SHA256()).hex(),
            'serial_number': format(cert.serial_number, 'x'),
            'signature_algorithm': cert.signature_algorithm_oid.dotted_string,
        }
    
    def _is_gm_certificate(self, cert: x509.Certificate) -> bool:
        """判断是否为国密证书（SM2）"""
        # 国密 SM2 签名算法 OID: 1.2.156.10197.1.501
        gm_signature_oids = [
            '1.2.156.10197.1.501',  # sm2WithSM3
            '1.2.156.10197.1.502',  # sm2Encryption
        ]
        
        sig_oid = cert.signature_algorithm_oid.dotted_string
        if sig_oid in gm_signature_oids:
            return True
        
        # 检查扩展项中的国密 OID
        try:
            for ext in cert.extensions:
                if ext.value.oid.dotted_string.startswith('1.2.156'):
                    return True
        except Exception:
            pass
        
        return False
    
    def scan_range(
        self,
        hosts: List[str],
        ports: List[int] = None
    ) -> List[Dict]:
        """并发扫描多个主机"""
        findings = []
        
        with concurrent.futures.ThreadPoolExecutor(
            max_workers=self.threads
        ) as executor:
            futures = {
                executor.submit(self.scan_host, host, ports): host
                for host in hosts
            }
            
            for future in concurrent.futures.as_completed(futures):
                host = futures[future]
                try:
                    results = future.result()
                    with self._lock:
                        findings.extend(results)
                except Exception as e:
                    logger.error(f"Scan failed for {host}: {e}")
        
        return findings

class PassiveCertificateCollector:
    """被动证书收集器 - 从已知源拉取证书"""
    
    def __init__(self, config: Dict):
        self.config = config
        self.sources = config.get('sources', [])
    
    def collect_from_files(self, file_patterns: List[str]) -> List[Dict]:
        """从文件系统中收集证书"""
        import glob
        findings = []
        
        for pattern in file_patterns:
            for filepath in glob.glob(pattern):
                try:
                    certs = self._load_pem_certs(filepath)
                    for cert_info in certs:
                        cert_info['source_file'] = filepath
                        findings.append(cert_info)
                except Exception as e:
                    logger.warning(f"Failed to load {filepath}: {e}")
        
        return findings
    
    def _load_pem_certs(self, filepath: str) -> List[Dict]:
        """加载 PEM 格式的证书文件"""
        with open(filepath, 'rb') as f:
            content = f.read()
        
        # 分割多个证书
        cert_blocks = re.findall(
            r'-----BEGIN CERTIFICATE-----.*?-----END CERTIFICATE-----',
            content.decode('utf-8'),
            re.DOTALL
        )
        
        results = []
        for block in cert_blocks:
            cert = x509.load_pem_x509_certificate(block.encode())
            results.append(self._parse_certificate(cert, filepath, None))
        
        return results
    
    def collect_from_hsm(
        self, 
        hsm_url: str, 
        username: str, 
        password: str
    ) -> List[Dict]:
        """从 HSM 设备拉取证书（需厂商 SDK）"""
        # 注意：实际实现需要调用 HSM 厂商提供的 SDK
        # 这里以通用方式示意
        logger.warning(
            f"HSM collection from {hsm_url} requires vendor SDK"
        )
        return []

from datetime import datetime, timedelta
from typing import List, Dict
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
import requests
import json

class CertificateMonitor:
    """证书监控引擎"""
    
    def __init__(self, config: Dict):
        self.config = config
        self.alert_thresholds = config.get('thresholds', {
            'warning_days': 90,
            'critical_days': 30,
            'emergency_days': 7,
        })
        self.notification_channels = config.get('channels', [])
    
    def check_certificates(
        self,
        certificates: List[Dict]
    ) -> List[Dict]:
        """批量检查证书状态"""
        alerts = []
        now_utc = datetime.now(timezone.utc)
        
        for cert in certificates:
            not_after = datetime.fromisoformat(cert['not_after'])
            days_remaining = (not_after - now_utc).days
            
            alert_level = self._determine_alert_level(days_remaining)
            
            if alert_level != 'normal':
                alert = {
                    **cert,
                    'days_remaining': days_remaining,
                    'alert_level': alert_level,
                    'checked_at': now_utc.isoformat(),
                }
                alerts.append(alert)
        
        return alerts
    
    def _determine_alert_level(
        self, 
        days_remaining: int
    ) -> str:
        """根据剩余天数确定告警级别"""
        if days_remaining <= 0:
            return 'expired'
        elif days_remaining <= self.alert_thresholds['emergency_days']:
            return 'emergency'
        elif days_remaining <= self.alert_thresholds['critical_days']:
            return 'critical'
        elif days_remaining <= self.alert_thresholds['warning_days']:
            return 'warning'
        return 'normal'
    
    def send_alerts(self, alerts: List[Dict]):
        """发送告警通知"""
        if not alerts:
            return
        
        for alert in alerts:
            level = alert['alert_level']
            channels = self._get_channels_for_level(level)
            
            for channel in channels:
                self._send_via_channel(alert, channel)
    
    def _get_channels_for_level(self, level: str) -> List[str]:
        """根据告警级别获取通知渠道"""
        channel_map = {
            'warning': ['email'],
            'critical': ['email', 'webhook'],
            'emergency': ['email', 'webhook', 'sms'],
            'expired': ['email', 'webhook', 'sms', 'phone'],
        }
        return channel_map.get(level, ['email'])
    
    def _send_via_channel(
        self, 
        alert: Dict, 
        channel: str
    ):
        """通过指定渠道发送告警"""
        if channel == 'email':
            self._send_email(alert)
        elif channel == 'webhook':
            self._send_webhook(alert)
        elif channel == 'sms':
            self._send_sms(alert)
    
    def _send_email(self, alert: Dict):
        """发送电子邮件告警"""
        smtp_config = self.config.get('smtp', {})
        if not smtp_config:
            return
        
        msg = MIMEMultipart()
        msg['From'] = smtp_config.get('from', 'monitor@example.com')
        msg['To'] = ', '.join(self.config.get('recipients', []))
        msg['Subject'] = f"[{alert['alert_level'].upper()}] 证书即将过期: {alert.get('common_name', 'Unknown')}"
        
        # 构建邮件正文
        body = self._build_email_body(alert)
        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'):
                server.starttls()
            if smtp_config.get('username'):
                server.login(
                    smtp_config['username'],
                    smtp_config['password']
                )
            server.send_message(msg)
    
    def _build_email_body(self, alert: Dict) -> str:
        """构建邮件正文"""
        return f"""
证书过期预警通知
================

证书信息:
- 主机: {alert.get('host', 'N/A')}
- 端口: {alert.get('port', 'N/A')}
- 域名: {alert.get('common_name', 'N/A')}
- 签发者: {alert.get('issuer_cn', 'N/A')}
- 过期时间: {alert.get('not_after', 'N/A')}
- 剩余天数: {alert.get('days_remaining', 'N/A')}

告警级别: {alert.get('alert_level', 'N/A').upper()}
检查时间: {alert.get('checked_at', 'N/A')}

请及时处理，避免影响业务。

-- 
国密证书监控系统
"""
    
    def _send_webhook(self, alert: Dict):
        """发送 Webhook 通知（钉钉/企业微信/飞书）"""
        webhook_url = self.config.get('webhook_url')
        if not webhook_url:
            return
        
        payload = {
            'msg_type': 'text',
            'text': {
                'content': f"[{alert['alert_level'].upper()}] 证书预警: {alert.get('common_name', 'Unknown')} 剩余 {alert.get('days_remaining', 0)} 天"
            }
        }
        
        requests.post(webhook_url, json=payload, timeout=10)

from jinja2 import Template
from reportlab.lib import colors
from reportlab.lib.pagesizes import A4
from reportlab.platypus import SimpleDocTemplate, Table, TableStyle
from reportlab.lib.styles import getSampleStyleSheet
import pdfkit

class ComplianceReportGenerator:
    """合规报告生成器"""
    
    def __init__(self, output_dir: str = './reports'):
        self.output_dir = output_dir
        os.makedirs(output_dir, exist_ok=True)
    
    def generate_html_report(
        self,
        certificates: List[Dict],
        alerts: List[Dict],
        report_date: str
    ) -> str:
        """生成 HTML 格式报告"""
        template = Template('''
<!DOCTYPE html>
<html>
<head>
    <meta charset="UTF-8">
    <title>国密证书合规报告 - {{ report_date }}</title>
    <style>
        body { font-family: Arial, sans-serif; margin: 40px; }
        h1 { color: #333; border-bottom: 2px solid #007bff; padding-bottom: 10px; }
        h2 { color: #555; margin-top: 30px; }
        table { border-collapse: collapse; width: 100%; margin: 20px 0; }
        th, td { border: 1px solid #ddd; padding: 8px; text-align: left; }
        th { background-color: #007bff; color: white; }
        tr:nth-child(even) { background-color: #f9f9f9; }
        .expired { color: #dc3545; font-weight: bold; }
        .critical { color: #fd7e14; font-weight: bold; }
        .warning { color: #ffc107; font-weight: bold; }
        .normal { color: #28a745; }
        .summary { background: #f8f9fa; padding: 20px; border-radius: 5px; margin: 20px 0; }
    </style>
</head>
<body>
    <h1>国密证书合规报告</h1>
    <p>报告日期: {{ report_date }}</p>
    
    <div class="summary">
        <h2>摘要</h2>
        <ul>
            <li>总证书数: {{ certificates|length }}</li>
            <li>正常证书: {{ normals }}</li>
            <li>即将过期: {{ warnings }}</li>
            <li>紧急过期: {{ emergencies }}</li>
            <li>已过期: {{ expireds }}</li>
        </ul>
    </div>
    
    <h2>证书清单</h2>
    <table>
        <tr>
            <th>主机</th>
            <th>域名</th>
            <th>过期时间</th>
            <th>剩余天数</th>
            <th>状态</th>
        </tr>
        {% for cert in certificates %}
        <tr class="{{ cert.status_class }}">
            <td>{{ cert.host }}:{{ cert.port }}</td>
            <td>{{ cert.common_name or 'N/A' }}</td>
            <td>{{ cert.not_after }}</td>
            <td>{{ cert.days_remaining }}</td>
            <td>{{ cert.alert_level }}</td>
        </tr>
        {% endfor %}
    </table>
    
    <h2>告警详情</h2>
    <table>
        <tr>
            <th>级别</th>
            <th>证书</th>
            <th>剩余天数</th>
            <th>建议操作</th>
        </tr>
        {% for alert in alerts %}
        <tr>
            <td>{{ alert.alert_level }}</td>
            <td>{{ alert.common_name or alert.host }}</td>
            <td>{{ alert.days_remaining }}</td>
            <td>{{ alert.recommended_action }}</td>
        </tr>
        {% endfor %}
    </table>
</body>
</html>
        ''')
        
        # 统计数据
        normals = sum(1 for c in certificates if c['alert_level'] == 'normal')
        warnings = sum(1 for c in certificates if c['alert_level'] == 'warning')
        emergencies = sum(1 for c in certificates if c['alert_level'] == 'emergency')
        expireds = sum(1 for c in certificates if c['alert_level'] == 'expired')
        
        # 为每个证书添加状态样式
        for cert in certificates:
            level = cert['alert_level']
            if level == 'expired':
                cert['status_class'] = 'expired'
            elif level == 'emergency':
                cert['status_class'] = 'critical'
            elif level == 'warning':
                cert['status_class'] = 'warning'
            else:
                cert['status_class'] = 'normal'
        
        # 为告警添加建议操作
        action_map = {
            'expired': '立即续期或替换证书',
            'emergency': '48小时内处理',
            'critical': '7天内处理',
            'warning': '安排下次维护窗口处理',
        }
        for alert in alerts:
            alert['recommended_action'] = action_map.get(alert['alert_level'], '继续监控')
        
        html_content = template.render(
            certificates=certificates,
            alerts=alerts,
            report_date=report_date,
            normals=normals,
            warnings=warnings,
            emergencies=emergencies,
            expireds=expireds,
        )
        
        return html_content
    
    def generate_pdf_report(
        self,
        html_content: str,
        filename: str
    ) -> str:
        """将 HTML 报告转换为 PDF"""
        output_path = os.path.join(self.output_dir, filename)
        
        # 使用 weasyprint 或 reportlab 生成 PDF
        try:
            import weasyprint
            weasyprint.HTML(string=html_content).write_pdf(output_path)
        except ImportError:
            # 备用方案：使用 reportlab
            self._generate_pdf_with_reportlab(html_content, output_path)
        
        return output_path
    
    def _generate_pdf_with_reportlab(
        self,
        html_content: str,
        output_path: str
    ):
        """使用 reportlab 生成基础 PDF（降级方案）"""
        doc = SimpleDocTemplate(output_path, pagesize=A4)
        styles = getSampleStyleSheet()
        
        # 这里简化处理，实际应解析 HTML 内容
        # 建议使用 weasyprint 或 pdftotext 库
        pass

# config.yaml
scan:
  targets:
    - host: "192.168.1.0/24"
      ports: [443, 8443, 9500]
    - host: "example.com"
      ports: [443]
  timeout: 5
  threads: 50

monitor:
  thresholds:
    warning_days: 90
    critical_days: 30
    emergency_days: 7
  
  channels:
    email:
      enabled: true
      recipients:
        - admin@example.com
        - security@example.com
    webhook:
      enabled: true
      url: "https://hooks.example.com/alerts"
    sms:
      enabled: false

report:
  output_dir: "./reports"
  schedule: "0 8 * * *"  # 每天 8:00 生成报告
  formats: ["html", "pdf"]

storage:
  type: sqlite
  path: "./data/cert_monitor.db"

#!/usr/bin/env python3
"""
国密证书生命周期监控系统
主入口
"""

import yaml
import logging
from datetime import datetime
from apscheduler.schedulers.blocking import BlockingScheduler
from scanner import CertificateScanner
from monitor import CertificateMonitor
from reporter import ComplianceReportGenerator

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)


def main():
    # 加载配置
    with open('config.yaml', 'r', encoding='utf-8') as f:
        config = yaml.safe_load(f)
    
    # 初始化组件
    scanner = CertificateScanner(
        timeout=config['scan']['timeout'],
        threads=config['scan']['threads']
    )
    monitor = CertificateMonitor(config['monitor'])
    reporter = ComplianceReportGenerator(
        config['report']['output_dir']
    )
    
    # 执行扫描
    logger.info("开始扫描证书...")
    certificates = []
    
    for target in config['scan']['targets']:
        findings = scanner.scan_host(
            target['host'],
            target.get('ports', [443])
        )
        certificates.extend(findings)
        logger.info(f"扫描 {target['host']}: 发现 {len(findings)} 个证书")
    
    logger.info(f"扫描完成，共发现 {len(certificates)} 个证书")
    
    # 检查状态
    alerts = monitor.check_certificates(certificates)
    logger.info(f"发现 {len(alerts)} 个需要关注的证书")
    
    # 发送告警
    if alerts:
        monitor.send_alerts(alerts)
        logger.info("告警已发送")
    
    # 生成报告
    report_date = datetime.now(timezone.utc).strftime('%Y-%m-%d')
    html_report = reporter.generate_html_report(
        certificates,
        alerts,
        report_date
    )
    report_path = os.path.join(
        config['report']['output_dir'],
        f'{report_date}_report.html'
    )
    with open(report_path, 'w', encoding='utf-8') as f:
        f.write(html_report)
    logger.info(f"报告已生成: {report_path}")
    
    return len(alerts)


if __name__ == '__main__':
    main()

from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.triggers.cron import CronTrigger

def setup_scheduler():
    scheduler = BlockingScheduler()
    
    # 每天凌晨 2:00 执行扫描
    scheduler.add_job(
        main,
        CronTrigger(hour=2, minute=0),
        id='daily_scan',
        name='每日证书扫描'
    )
    
    # 每周生成周报
    scheduler.add_job(
        generate_weekly_report,
        CronTrigger(day_of_week='mon', hour=9, minute=0),
        id='weekly_report',
        name='周报生成'
    )
    
    scheduler.start()


def generate_weekly_report():
    """生成周报（略）"""
    pass