如何编写证书过期提前预警脚本

本文目录导读：

1. 基础版Bash脚本（Linux）
2. Python版（功能更完善）
3. Docker版本配置
4. Crontab定时任务
5. 使用方法

基础版Bash脚本（Linux）

#!/bin/bash
# SSL证书过期预警脚本
# 使用方法: ./check_ssl.sh domain1.com domain2.com
# 配置
WARNING_DAYS=30  # 提前多少天预警
EMAIL="admin@example.com"  # 通知邮箱
# 检查单个域名证书
check_cert() {
    local domain=$1
    echo "检查域名: $domain"
    # 获取证书过期信息
    local expire_date=$(echo | openssl s_client -servername $domain -connect $domain:443 2>/dev/null | \
        openssl x509 -noout -enddate 2>/dev/null | cut -d= -f2)
    if [ -z "$expire_date" ]; then
        echo "警告: 无法获取 $domain 的证书信息"
        return 1
    fi
    # 计算剩余天数
    local expire_seconds=$(date -d "$expire_date" +%s)
    local now_seconds=$(date +%s)
    local remaining_days=$(( ($expire_seconds - $now_seconds) / 86400 ))
    echo "域名: $domain"
    echo "过期日期: $expire_date"
    echo "剩余天数: $remaining_days"
    # 判断是否需要预警
    if [ $remaining_days -le 0 ]; then
        echo "❌ 证书已过期!"
        send_alert "$domain 证书已过期"
    elif [ $remaining_days -le $WARNING_DAYS ]; then
        echo "⚠️  即将过期，剩余 $remaining_days 天"
        send_alert "$domain 证书将在 $remaining_days 天后过期"
    else
        echo "✅ 证书状态正常"
    fi
}
# 发送告警
send_alert() {
    local message=$1
    echo "$message" | mail -s "SSL证书预警" $EMAIL
}
# 主程序
if [ $# -eq 0 ]; then
    echo "使用方法: $0 domain1.com [domain2.com ...]"
    exit 1
fi
for domain in "$@"; do
    check_cert "$domain"
    echo "------------------------"
done

Python版（功能更完善）

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import ssl
import socket
import smtplib
import logging
from datetime import datetime, timedelta
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
import json
class SSLCertChecker:
    def __init__(self, config_file='config.json'):
        self.load_config(config_file)
        self.setup_logging()
    def load_config(self, config_file):
        """加载配置文件"""
        with open(config_file, 'r') as f:
            config = json.load(f)
        self.domains = config.get('domains', [])
        self.warning_days = config.get('warning_days', 30)
        self.email_config = config.get('email', {})
        self.webhook_url = config.get('webhook_url', '')
    def setup_logging(self):
        """设置日志"""
        logging.basicConfig(
            level=logging.INFO,
            format='%(asctime)s - %(levelname)s - %(message)s',
            handlers=[
                logging.FileHandler('ssl_check.log'),
                logging.StreamHandler()
            ]
        )
        self.logger = logging.getLogger(__name__)
    def check_domain_cert(self, domain, port=443):
        """检查单个域名的证书"""
        try:
            context = ssl.create_default_context()
            with socket.create_connection((domain, port), timeout=10) as sock:
                with context.wrap_socket(sock, server_hostname=domain) as ssock:
                    cert = ssock.getpeercert()
                    # 获取证书信息
                    subject = dict(x[0] for x in cert['subject'])
                    issuer = dict(x[0] for x in cert['issuer'])
                    # 获取过期时间
                    expire_date = datetime.strptime(
                        cert['notAfter'], 
                        '%b %d %H:%M:%S %Y %Z'
                    )
                    # 计算剩余天数
                    remaining_days = (expire_date - datetime.now()).days
                    cert_info = {
                        'domain': domain,
                        'subject': subject.get('CN', 'Unknown'),
                        'issuer': issuer.get('CN', 'Unknown'),
                        'expire_date': expire_date.strftime('%Y-%m-%d %H:%M:%S'),
                        'remaining_days': remaining_days,
                        'status': 'valid' if remaining_days > 0 else 'expired'
                    }
                    return cert_info
        except Exception as e:
            self.logger.error(f"检查 {domain} 失败: {str(e)}")
            return None
    def send_email_alert(self, alert_info):
        """发送邮件告警"""
        if not self.email_config.get('enabled'):
            return
        try:
            msg = MIMEMultipart()
            msg['From'] = self.email_config['from']
            msg['To'] = self.email_config['to']
            msg['Subject'] = f"SSL证书预警 - {alert_info['domain']}"
            body = f"""
            SSL证书过期预警
            域名: {alert_info['domain']}
            主题: {alert_info['subject']}
            颁发者: {alert_info['issuer']}
            过期时间: {alert_info['expire_date']}
            剩余天数: {alert_info['remaining_days']}
            请尽快更新证书！
            """
            msg.attach(MIMEText(body, 'plain', 'utf-8'))
            # 连接SMTP服务器
            server = smtplib.SMTP(
                self.email_config['smtp_server'],
                self.email_config['smtp_port']
            )
            server.starttls()
            server.login(
                self.email_config['username'],
                self.email_config['password']
            )
            server.send_message(msg)
            server.quit()
            self.logger.info(f"邮件告警发送成功: {alert_info['domain']}")
        except Exception as e:
            self.logger.error(f"邮件发送失败: {str(e)}")
    def send_webhook_alert(self, alert_info):
        """发送Webhook告警（如企业微信、钉钉等）"""
        if not self.webhook_url:
            return
        import requests
        try:
            data = {
                "msgtype": "text",
                "text": {
                    "content": f"SSL证书预警\n域名: {alert_info['domain']}\n"
                              f"剩余天数: {alert_info['remaining_days']}\n"
                              f"过期时间: {alert_info['expire_date']}"
                }
            }
            requests.post(self.webhook_url, json=data, timeout=5)
            self.logger.info(f"Webhook告警发送成功: {alert_info['domain']}")
        except Exception as e:
            self.logger.error(f"Webhook发送失败: {str(e)}")
    def run_check(self):
        """执行检查"""
        results = []
        alerts = []
        for domain in self.domains:
            cert_info = self.check_domain_cert(domain)
            if cert_info:
                results.append(cert_info)
                # 判断是否需要告警
                if cert_info['remaining_days'] <= 0:
                    self.logger.error(f"证书已过期: {domain}")
                    alerts.append(cert_info)
                elif cert_info['remaining_days'] <= self.warning_days:
                    self.logger.warning(f"证书即将过期: {domain} ({cert_info['remaining_days']}天)")
                    alerts.append(cert_info)
                else:
                    self.logger.info(f"证书正常: {domain} ({cert_info['remaining_days']}天)")
        # 发送告警
        for alert in alerts:
            self.send_email_alert(alert)
            self.send_webhook_alert(alert)
        return results
# 配置文件示例 (config.json)
"""
{
    "domains": ["example.com", "google.com", "github.com"],
    "warning_days": 30,
    "email": {
        "enabled": true,
        "smtp_server": "smtp.gmail.com",
        "smtp_port": 587,
        "from": "your-email@gmail.com",
        "to": "admin@example.com",
        "username": "your-email@gmail.com",
        "password": "your-app-password"
    },
    "webhook_url": "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=xxx"
}
"""
if __name__ == "__main__":
    checker = SSLCertChecker()
    results = checker.run_check()
    # 输出结果
    print("\n检查结果:")
    print("="*50)
    for result in results:
        print(f"\n域名: {result['domain']}")
        print(f"主题: {result['subject']}")
        print(f"颁发者: {result['issuer']}")
        print(f"过期时间: {result['expire_date']}")
        print(f"剩余天数: {result['remaining_days']}")
        if result['remaining_days'] <= 0:
            print("状态: ❌ 已过期")
        elif result['remaining_days'] <= 30:
            print("状态: ⚠️  即将过期")
        else:
            print("状态: ✅ 正常")

Docker版本配置

# docker-compose.yml
version: '3.8'
services:
  ssl-checker:
    image: python:3.9-slim
    container_name: ssl-cert-checker
    volumes:
      - ./ssl_checker:/app
      - ./logs:/app/logs
    working_dir: /app
    command: |
      sh -c "
        pip install requests &&
        python -u ssl_cert_checker.py
      "
    restart: unless-stopped
    environment:
      - TZ=Asia/Shanghai

Crontab定时任务

# 每天凌晨2点执行
0 2 * * * /path/to/ssl_check.sh example.com >> /var/log/ssl_check.log 2>&1
# 或者使用Python脚本
0 2 * * * cd /path/to && python3 ssl_cert_checker.py >> /var/log/ssl_check.log 2>&1

使用方法

1. Bash脚本：

   chmod +x ssl_check.sh
   ./ssl_check.sh domain1.com domain2.com

2. Python脚本：

   # 创建配置文件
   cp config.example.json config.json
   # 编辑配置文件
   vim config.json
   # 运行脚本
   python3 ssl_cert_checker.py

3. Docker部署：

   docker-compose up -d

这些脚本提供了多种方式和告警渠道,您可以根据实际需求选择适合的方案，建议结合定时任务自动运行，确保证书到期前能及时收到通知。