本文目录导读：
我来为您提供一个定时检查并更新数字证书的脚本,以下是针对不同场景的解决方案：
Python 版本（推荐，功能最完整）
cert_monitor.py
#!/usr/bin/env python3
"""
数字证书定时检查与更新脚本
支持：SSL/TLS证书、本地文件证书、ACME自动续期
"""
import os
import sys
import ssl
import socket
import logging
import subprocess
import json
from datetime import datetime, timedelta
from pathlib import Path
from typing import Dict, List, Optional, Tuple
import yaml
import OpenSSL
import requests
# 配置日志
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('cert_monitor.log'),
logging.StreamHandler(sys.stdout)
]
)
logger = logging.getLogger(__name__)
class CertificateMonitor:
"""证书监控器"""
def __init__(self, config_path: str = 'config.yaml'):
self.config = self._load_config(config_path)
self.check_interval = self.config.get('check_interval', 24) # 小时
self.warning_days = self.config.get('warning_days', 30) # 天数
self.critical_days = self.config.get('critical_days', 7) # 天数
def _load_config(self, config_path: str) -> Dict:
"""加载配置文件"""
try:
with open(config_path, 'r', encoding='utf-8') as f:
return yaml.safe_load(f)
except FileNotFoundError:
logger.error(f"配置文件 {config_path} 未找到")
sys.exit(1)
except yaml.YAMLError as e:
logger.error(f"配置文件解析错误: {e}")
sys.exit(1)
def check_remote_cert(self, hostname: str, port: int = 443) -> Dict:
"""检查远程服务器证书"""
try:
context = ssl.create_default_context()
with socket.create_connection((hostname, port), timeout=10) as sock:
with context.wrap_socket(sock, server_hostname=hostname) as ssock:
cert = ssock.getpeercert()
# 解析证书信息
not_after = datetime.strptime(
cert['notAfter'], '%b %d %H:%M:%S %Y %Z'
)
not_before = datetime.strptime(
cert['notBefore'], '%b %d %H:%M:%S %Y %Z'
)
return {
'hostname': hostname,
'port': port,
'subject': dict(cert['subject'][0]),
'issuer': dict(cert['issuer'][0]),
'not_before': not_before,
'not_after': not_after,
'days_remaining': (not_after - datetime.now()).days,
'valid': not_before <= datetime.now() <= not_after,
'serial_number': cert.get('serialNumber', 'N/A')
}
except Exception as e:
logger.error(f"检查 {hostname}:{port} 证书失败: {e}")
return {'hostname': hostname, 'port': port, 'error': str(e)}
def check_local_cert(self, cert_path: str) -> Dict:
"""检查本地证书文件"""
try:
with open(cert_path, 'rb') as f:
cert_data = f.read()
cert = OpenSSL.crypto.load_certificate(
OpenSSL.crypto.FILETYPE_PEM, cert_data
)
not_after = datetime.strptime(
cert.get_notAfter().decode('ascii'), '%Y%m%d%H%M%SZ'
)
not_before = datetime.strptime(
cert.get_notBefore().decode('ascii'), '%Y%m%d%H%M%SZ'
)
# 获取证书信息
subject = dict(cert.get_subject().get_components())
issuer = dict(cert.get_issuer().get_components())
return {
'path': cert_path,
'subject': subject,
'issuer': issuer,
'not_before': not_before,
'not_after': not_after,
'days_remaining': (not_after - datetime.now()).days,
'valid': not_before <= datetime.now() <= not_after,
'serial_number': cert.get_serial_number()
}
except Exception as e:
logger.error(f"检查本地证书 {cert_path} 失败: {e}")
return {'path': cert_path, 'error': str(e)}
def check_letsencrypt_cert(self, domain: str) -> Dict:
"""检查Let's Encrypt证书"""
cert_path = f"/etc/letsencrypt/live/{domain}/fullchain.pem"
if os.path.exists(cert_path):
return self.check_local_cert(cert_path)
else:
return {'domain': domain, 'error': '证书文件未找到'}
def renew_letsencrypt_cert(self, domain: str) -> bool:
"""续期Let's Encrypt证书"""
try:
logger.info(f"正在续期 {domain} 的Let's Encrypt证书...")
# 使用certbot续期
result = subprocess.run(
['certbot', 'renew', '--cert-name', domain, '--non-interactive'],
capture_output=True,
text=True
)
if result.returncode == 0:
logger.info(f"证书 {domain} 续期成功")
return True
else:
logger.error(f"证书 {domain} 续期失败: {result.stderr}")
return False
except Exception as e:
logger.error(f"续期证书 {domain} 时出错: {e}")
return False
def send_alert(self, message: str, level: str = 'warning'):
"""发送告警通知"""
alert_methods = self.config.get('alert_methods', [])
for method in alert_methods:
if method['type'] == 'email':
self._send_email_alert(message, method, level)
elif method['type'] == 'webhook':
self._send_webhook_alert(message, method, level)
elif method['type'] == 'telegram':
self._send_telegram_alert(message, method, level)
def _send_email_alert(self, message: str, config: Dict, level: str):
"""发送邮件告警"""
try:
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
msg = MIMEMultipart()
msg['From'] = config['from']
msg['To'] = config['to']
msg['Subject'] = f"[{level.upper()}] 证书监控告警"
body = f"""
<h2>证书监控告警</h2>
<p>级别: {level}</p>
<p>时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}</p>
<pre>{message}</pre>
"""
msg.attach(MIMEText(body, 'html'))
with smtplib.SMTP(config['smtp_server'], config['smtp_port']) as server:
server.starttls()
server.login(config['username'], config['password'])
server.send_message(msg)
logger.info(f"邮件告警已发送至 {config['to']}")
except Exception as e:
logger.error(f"发送邮件告警失败: {e}")
def _send_webhook_alert(self, message: str, config: Dict, level: str):
"""发送Webhook告警（如钉钉、企业微信等）"""
try:
payload = {
'msgtype': 'text',
'text': {
'content': f"[{level}] 证书监控告警\n{message}"
}
}
response = requests.post(
config['url'],
json=payload,
headers={'Content-Type': 'application/json'}
)
if response.status_code == 200:
logger.info(f"Webhook告警发送成功")
else:
logger.error(f"Webhook告警发送失败: {response.status_code}")
except Exception as e:
logger.error(f"发送Webhook告警失败: {e}")
def _send_telegram_alert(self, message: str, config: Dict, level: str):
"""发送Telegram告警"""
try:
url = f"https://api.telegram.org/bot{config['bot_token']}/sendMessage"
payload = {
'chat_id': config['chat_id'],
'text': f"[{level}] 证书监控告警\n\n{message}",
'parse_mode': 'HTML'
}
response = requests.post(url, json=payload)
if response.status_code == 200:
logger.info(f"Telegram告警发送成功")
else:
logger.error(f"Telegram告警发送失败: {response.status_code}")
except Exception as e:
logger.error(f"发送Telegram告警失败: {e}")
def run_check(self):
"""执行检查"""
results = []
alerts = []
# 检查远程证书
for server in self.config.get('remote_servers', []):
result = self.check_remote_cert(
server['hostname'],
server.get('port', 443)
)
results.append(result)
if 'error' not in result:
days = result['days_remaining']
if days <= self.critical_days:
alerts.append({
'level': 'critical',
'message': f"证书 {result['hostname']} 将在 {days} 天后过期！"
})
elif days <= self.warning_days:
alerts.append({
'level': 'warning',
'message': f"证书 {result['hostname']} 将在 {days} 天后过期"
})
# 检查本地证书
for cert_path in self.config.get('local_certs', []):
result = self.check_local_cert(cert_path)
results.append(result)
if 'error' not in result:
days = result['days_remaining']
if days <= self.critical_days:
alerts.append({
'level': 'critical',
'message': f"证书 {result['path']} 将在 {days} 天后过期！"
})
# 检查Let's Encrypt证书并自动续期
for domain in self.config.get('letsencrypt_domains', []):
result = self.check_letsencrypt_cert(domain)
results.append(result)
if 'error' not in result and result['days_remaining'] <= 30:
logger.info(f"证书 {domain} 将在 {result['days_remaining']} 天后过期，尝试续期...")
if self.renew_letsencrypt_cert(domain):
logger.info(f"证书 {domain} 续期成功")
else:
alerts.append({
'level': 'error',
'message': f"证书 {domain} 续期失败"
})
# 发送告警
for alert in alerts:
self.send_alert(alert['message'], alert['level'])
# 记录检查结果
self._save_results(results)
return results
def _save_results(self, results: List[Dict]):
"""保存检查结果"""
output = {
'timestamp': datetime.now().isoformat(),
'results': results
}
with open('cert_check_results.json', 'w') as f:
json.dump(output, f, indent=2, default=str)
def start_daemon(self):
"""以守护进程方式运行"""
import time
logger.info(f"证书监控守护进程启动，检查间隔: {self.check_interval} 小时")
while True:
try:
logger.info("开始执行证书检查...")
self.run_check()
logger.info("证书检查完成")
# 等待下一个检查周期
time.sleep(self.check_interval * 3600)
except KeyboardInterrupt:
logger.info("收到中断信号，退出监控")
break
except Exception as e:
logger.error(f"检查过程出错: {e}")
time.sleep(3600) # 出错后等待1小时重试
def main():
"""主函数"""
import argparse
parser = argparse.ArgumentParser(description='证书定时检查与更新工具')
parser.add_argument('--config', '-c', default='config.yaml', help='配置文件路径')
parser.add_argument('--run-once', '-r', action='store_true', help='只运行一次检查')
parser.add_argument('--daemon', '-d', action='store_true', help='以守护进程方式运行')
args = parser.parse_args()
monitor = CertificateMonitor(args.config)
if args.run_once:
monitor.run_check()
elif args.daemon:
monitor.start_daemon()
else:
# 默认先执行一次检查
monitor.run_check()
print("证书检查完成，结果已保存到 cert_check_results.json")
if __name__ == '__main__':
main()
配置文件
config.yaml
# 证书监控配置
# 检查间隔（小时）
check_interval: 24
# 告警天数
warning_days: 30
critical_days: 7
# 远程服务器检查列表
remote_servers:
- hostname: "example.com"
port: 443
- hostname: "api.example.com"
port: 443
- hostname: "mail.example.com"
port: 465
# 本地证书文件列表
local_certs:
- "/etc/nginx/ssl/server.crt"
- "/etc/apache2/ssl/server.crt"
- "/path/to/custom/cert.pem"
# Let's Encrypt 域名列表
letsencrypt_domains:
- "example.com"
- "www.example.com"
- "api.example.com"
# 告警配置（支持多种方式）
alert_methods:
- type: "email"
from: "monitor@example.com"
to: "admin@example.com"
smtp_server: "smtp.example.com"
smtp_port: 587
username: "monitor@example.com"
password: "your-email-password"
- type: "webhook"
url: "https://hooks.example.com/webhook/cert-monitor"
- type: "telegram"
bot_token: "your-bot-token"
chat_id: "your-chat-id"
依赖安装
# 安装Python依赖 pip install pyOpenSSL requests pyyaml # 安装certbot（用于Let's Encrypt自动续期） sudo apt install certbot # Ubuntu/Debian sudo yum install certbot # CentOS/RHEL
使用系统定时任务（Cron）
crontab_setup.sh
#!/bin/bash # 设置每天凌晨2点检查证书 echo "0 2 * * * /usr/bin/python3 /path/to/cert_monitor.py --run-once --config /path/to/config.yaml >> /var/log/cert_monitor.log 2>&1" | crontab - # 或者设置系统d服务（推荐） cat > /etc/systemd/system/cert-monitor.service << EOF [Unit] Description=Certificate Monitor Service After=network.target [Service] Type=simple User=root ExecStart=/usr/bin/python3 /path/to/cert_monitor.py --daemon --config /path/to/config.yaml Restart=always RestartSec=60 [Install] WantedBy=multi-user.target EOF # 启用并启动服务 systemctl daemon-reload systemctl enable cert-monitor systemctl start cert-monitor
简化版本（Bash脚本）
cert_check.sh
#!/bin/bash
# 数字证书检查脚本（简化版）
CERT_PATH="/etc/ssl/certs"
LOG_FILE="/var/log/cert_check.log"
WARNING_DAYS=30
ALERT_EMAIL="admin@example.com"
check_cert() {
local cert_file=$1
if [ ! -f "$cert_file" ]; then
echo "错误: 证书文件 $cert_file 不存在" | tee -a $LOG_FILE
return 1
fi
# 获取证书过期时间
local expiry_date=$(openssl x509 -in "$cert_file" -noout -enddate 2>/dev/null | cut -d= -f2)
if [ -z "$expiry_date" ]; then
echo "错误: 无法读取证书 $cert_file" | tee -a $LOG_FILE
return 1
fi
# 计算剩余天数
local expiry_epoch=$(date -d "$expiry_date" +%s 2>/dev/null)
local current_epoch=$(date +%s)
local days_remaining=$(( (expiry_epoch - current_epoch) / 86400 ))
echo "证书: $cert_file" | tee -a $LOG_FILE
echo "过期日期: $expiry_date" | tee -a $LOG_FILE
echo "剩余天数: $days_remaining" | tee -a $LOG_FILE
# 检查是否接近过期
if [ $days_remaining -le 0 ]; then
echo "严重: 证书已过期！" | tee -a $LOG_FILE
send_alert "证书 $cert_file 已过期！"
elif [ $days_remaining -le $WARNING_DAYS ]; then
echo "警告: 证书将在 $days_remaining 天后过期" | tee -a $LOG_FILE
if [ $days_remaining -le 7 ]; then
send_alert "证书 $cert_file 将在 $days_remaining 天后过期！"
fi
fi
echo "---" | tee -a $LOG_FILE
}
send_alert() {
local message=$1
echo "$message" | mail -s "证书过期告警" $ALERT_EMAIL
}
renew_letsencrypt() {
local domain=$1
echo "正在续期 Let's Encrypt 证书: $domain" | tee -a $LOG_FILE
certbot renew --cert-name "$domain" --non-interactive 2>&1 | tee -a $LOG_FILE
}
# 主检查流程
main() {
echo "========================================" | tee -a $LOG_FILE
echo "证书检查开始: $(date)" | tee -a $LOG_FILE
echo "========================================" | tee -a $LOG_FILE
# 检查指定目录下的所有证书
for cert in "$CERT_PATH"/*.crt; do
if [ -f "$cert" ]; then
check_cert "$cert"
fi
done
# 检查远程服务器（可选）
# echo | openssl s_client -connect example.com:443 2>/dev/null | openssl x509 -noout -dates
echo "证书检查完成: $(date)" | tee -a $LOG_FILE
echo "" | tee -a $LOG_FILE
}
# 执行主函数
main
使用方法
基本使用：
# 安装依赖 pip install pyOpenSSL requests pyyaml # 创建配置文件 cp config.yaml.example config.yaml # 编辑config.yaml，填入你的配置 # 单次运行 python3 cert_monitor.py --run-once # 守护进程模式 python3 cert_monitor.py --daemon # 使用cron定时执行（每天检查一次） 0 0 * * * /usr/bin/python3 /path/to/cert_monitor.py --run-once
功能特点
- ✅ 支持远程HTTPS证书检查
- ✅ 支持本地证书文件检查
- ✅ 支持Let's Encrypt自动续期
- ✅ 多种告警方式（邮件、Webhook、Telegram）
- ✅ 自定义检查间隔和告警阈值
- ✅ 结果保存到JSON文件
- ✅ 守护进程模式
- ✅ 详细日志记录
这个脚本可以根据您的具体需求进行修改和扩展！
