264 lines
9.5 KiB
Python
264 lines
9.5 KiB
Python
"""
|
||
证书到期邮件通知服务
|
||
- 由 cron 每日 10:00 触发
|
||
- 检查数据库中的 Let's Encrypt 证书
|
||
- 到期前 N 天起，连续每天发送提醒邮件
|
||
- 邮件通过 Mokee Gateway 对外开放 API 发送
|
||
|
||
使用方式:
|
||
python cert_notifier.py # 直接运行（cron 调用）
|
||
python cert_notifier.py --dry-run # 预览模式，不实际发送
|
||
python cert_notifier.py --check # 仅显示证书状态
|
||
"""
|
||
import os
|
||
import sys
|
||
import json
|
||
import logging
|
||
import urllib.request
|
||
import urllib.error
|
||
from datetime import datetime
|
||
|
||
# 数据库配置
|
||
DB_HOST = os.environ.get('MYSQL_HOST', '10.1.1.122')
|
||
DB_PORT = int(os.environ.get('MYSQL_PORT', '3306'))
|
||
DB_USER = os.environ.get('MYSQL_USER', 'nginxserver')
|
||
DB_PASS = os.environ.get('MYSQL_PASS', 'mokee.')
|
||
DB_NAME = os.environ.get('MYSQL_DB', 'nginxserver')
|
||
|
||
# 网关配置（可通过环境变量覆盖）
|
||
GATEWAY_URL = os.environ.get('MOKEE_GATEWAY_URL', 'https://gw.server.mokee.com')
|
||
SYSTEM_CODE = os.environ.get('MOKEE_SYSTEM_CODE', 'nginxServer')
|
||
|
||
logging.basicConfig(
|
||
level=logging.INFO,
|
||
format='%(asctime)s [%(levelname)s] %(message)s'
|
||
)
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
def get_db_connection():
|
||
"""获取数据库连接（延迟导入，避免非数据库场景报错）"""
|
||
import pymysql
|
||
return pymysql.connect(
|
||
host=DB_HOST, port=DB_PORT, user=DB_USER,
|
||
password=DB_PASS, database=DB_NAME,
|
||
charset='utf8mb4', connect_timeout=5
|
||
)
|
||
|
||
|
||
def get_token():
|
||
"""从 Mokee Gateway 获取一次性 Token"""
|
||
url = f"{GATEWAY_URL}/zgapi/v1/open/token"
|
||
data = json.dumps({"systemCode": SYSTEM_CODE}).encode('utf-8')
|
||
|
||
req = urllib.request.Request(url, data=data, method='POST')
|
||
req.add_header('Content-Type', 'application/json')
|
||
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=10) as resp:
|
||
result = json.loads(resp.read().decode('utf-8'))
|
||
if result.get('code') == 200:
|
||
token = result['data']['token']
|
||
logger.info(f"获取 Token 成功: {SYSTEM_CODE}")
|
||
return token
|
||
else:
|
||
logger.error(f"获取 Token 失败: {result}")
|
||
return None
|
||
except Exception as e:
|
||
logger.error(f"获取 Token 异常: {e}")
|
||
return None
|
||
|
||
|
||
def send_email(token, to, subject, content):
|
||
"""
|
||
通过 Mokee Gateway 发送邮件
|
||
成功 = Token 被消费，需要重新获取
|
||
"""
|
||
url = f"{GATEWAY_URL}/zgapi/v1/open/email/send"
|
||
data = json.dumps({
|
||
"to": to,
|
||
"subject": subject,
|
||
"content": content,
|
||
"contentType": "html"
|
||
}).encode('utf-8')
|
||
|
||
req = urllib.request.Request(url, data=data, method='POST')
|
||
req.add_header('Content-Type', 'application/json')
|
||
req.add_header('Authorization', f'Bearer {token}')
|
||
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=15) as resp:
|
||
result = json.loads(resp.read().decode('utf-8'))
|
||
return result.get('code') == 200, str(result)
|
||
except Exception as e:
|
||
return False, str(e)
|
||
|
||
|
||
def build_email_html(domain, days_left, expire_date):
|
||
"""生成 HTML 邮件内容"""
|
||
if days_left <= 5:
|
||
level = 'danger'
|
||
color = '#F56C6C'
|
||
elif days_left <= 10:
|
||
level = 'warning'
|
||
color = '#E6A23C'
|
||
else:
|
||
level = 'info'
|
||
color = '#409EFF'
|
||
|
||
return f"""<!DOCTYPE html>
|
||
<html><body style="font-family: 'Microsoft YaHei', Arial, sans-serif;">
|
||
<div style="max-width:600px;margin:0 auto;padding:20px;border:1px solid #EBEEF5;border-radius:8px;">
|
||
<div style="background:{color};padding:16px;border-radius:8px 8px 0 0;text-align:center;">
|
||
<h2 style="color:#fff;margin:0;">⚠️ SSL 证书即将到期</h2>
|
||
</div>
|
||
<div style="padding:20px;background:#fff;">
|
||
<table style="width:100%;border-collapse:collapse;">
|
||
<tr><td style="padding:8px 0;color:#909399;width:80px;">域名</td>
|
||
<td style="padding:8px 0;font-weight:bold;font-size:16px;">{domain}</td></tr>
|
||
<tr><td style="padding:8px 0;color:#909399;">到期时间</td>
|
||
<td style="padding:8px 0;">{expire_date}</td></tr>
|
||
<tr><td style="padding:8px 0;color:#909399;">剩余天数</td>
|
||
<td style="padding:8px 0;color:{color};font-weight:bold;font-size:20px;">{days_left} 天</td></tr>
|
||
<tr><td style="padding:8px 0;color:#909399;">证书类型</td>
|
||
<td style="padding:8px 0;">Let's Encrypt (阿里云 DNS 验证)</td></tr>
|
||
</table>
|
||
<hr style="border:0;border-top:1px solid #EBEEF5;margin:16px 0;">
|
||
<p style="color:#909399;font-size:13px;">
|
||
系统已配置 acme.sh 自动续期，正常情况下会在到期前自动更新证书。<br>
|
||
如续期失败，请检查：<br>
|
||
1. 阿里云 AccessKey 是否有效<br>
|
||
2. 服务器能否访问 Let's Encrypt API<br>
|
||
3. DNS TXT 记录是否正常清理
|
||
</p>
|
||
<p style="color:#C0C4CC;font-size:12px;">
|
||
此邮件由 Nginx 域名管理平台自动发送，连续提醒至证书续期成功。
|
||
</p>
|
||
</div>
|
||
</div>
|
||
</body></html>"""
|
||
|
||
|
||
def check_and_notify(dry_run=False):
|
||
"""检查证书到期情况并发送通知"""
|
||
conn = get_db_connection()
|
||
cursor = conn.cursor()
|
||
|
||
# 查询需要通知的 Let's Encrypt 证书
|
||
cursor.execute('''
|
||
SELECT id, domain, notify_email, notify_enabled, notify_days_before, expire_date
|
||
FROM certificates
|
||
WHERE cert_type = 'letsencrypt'
|
||
AND status = 'active'
|
||
AND notify_enabled = 1
|
||
AND notify_email IS NOT NULL
|
||
AND notify_email != ''
|
||
''')
|
||
|
||
certs = cursor.fetchall()
|
||
if not certs:
|
||
logger.info("无需通知的 Let's Encrypt 证书")
|
||
conn.close()
|
||
return
|
||
|
||
today = datetime.now()
|
||
token = None # 延迟获取
|
||
notified_count = 0
|
||
|
||
for cert_id, domain, notify_email, enabled, days_before, expire_date_db in certs:
|
||
# 从数据库获取过期日期
|
||
if expire_date_db:
|
||
expire_date = expire_date_db if isinstance(expire_date_db, datetime) else datetime.strptime(str(expire_date_db), '%Y-%m-%d %H:%M:%S')
|
||
days_left = (expire_date - today).days
|
||
else:
|
||
logger.warning(f"{domain}: 无过期日期信息，跳过")
|
||
continue
|
||
|
||
logger.info(f"{domain}: 剩余 {days_left} 天到期 (通知阈值: {days_before}天前开始)")
|
||
|
||
# 判断是否需要通知: days_before 天前开始，连续通知到到期前1天
|
||
if days_left > days_before or days_left <= 0:
|
||
logger.info(f"{domain}: 不在通知窗口 (days_left={days_left}, threshold={days_before})")
|
||
continue
|
||
|
||
# 检查今天是否已发送过通知
|
||
cursor.execute('''
|
||
SELECT id FROM cert_notify_log
|
||
WHERE cert_id = %s AND DATE(sent_at) = CURDATE()
|
||
''', (cert_id,))
|
||
if cursor.fetchone():
|
||
logger.info(f"{domain}: 今日已通知，跳过")
|
||
continue
|
||
|
||
# 生成邮件内容
|
||
expire_str = expire_date.strftime('%Y-%m-%d %H:%M')
|
||
subject = f"⚠️ SSL证书到期提醒 - {domain} (剩余{days_left}天)"
|
||
content = build_email_html(domain, days_left, expire_str)
|
||
|
||
if dry_run:
|
||
logger.info(f"[DRY-RUN] 将发送邮件: to={notify_email}, subject={subject}")
|
||
continue
|
||
|
||
# 获取 Token（每次发送前确保有效）
|
||
if token is None:
|
||
token = get_token()
|
||
if token is None:
|
||
logger.error(f"无法获取 Token，跳过 {domain}")
|
||
continue
|
||
|
||
# 发送邮件
|
||
success, resp = send_email(token, notify_email, subject, content)
|
||
if success:
|
||
# Token 已消费，下次需要重新获取
|
||
token = None
|
||
logger.info(f"{domain}: 邮件发送成功 → {notify_email}")
|
||
else:
|
||
logger.warning(f"{domain}: 邮件发送失败 → {resp}")
|
||
# 500 错误不消费 Token，可重试
|
||
|
||
# 记录日志
|
||
cursor.execute('''
|
||
INSERT INTO cert_notify_log (cert_id, domain, days_left, email_to, success, response_msg)
|
||
VALUES (%s, %s, %s, %s, %s, %s)
|
||
''', (cert_id, domain, days_left, notify_email, 1 if success else 0, resp[:1000]))
|
||
conn.commit()
|
||
|
||
if success:
|
||
notified_count += 1
|
||
|
||
conn.close()
|
||
logger.info(f"通知完成: 发送 {notified_count} 封邮件")
|
||
|
||
|
||
def check_status():
|
||
"""仅显示证书状态（--check 模式）"""
|
||
conn = get_db_connection()
|
||
cursor = conn.cursor()
|
||
cursor.execute('''
|
||
SELECT domain, cert_type, status, expire_date,
|
||
DATEDIFF(COALESCE(expire_date, NOW()), NOW()) AS days_left,
|
||
notify_enabled, notify_email, notify_days_before
|
||
FROM certificates
|
||
WHERE cert_type = 'letsencrypt'
|
||
ORDER BY days_left
|
||
''')
|
||
|
||
print(f"\n{'域名':25s} {'类型':12s} {'状态':8s} {'到期日期':20s} {'剩余天数':>8s} {'通知':>4s} {'收件人'}")
|
||
print('-' * 120)
|
||
for row in cursor.fetchall():
|
||
domain, cert_type, status, expire_date, days_left, enabled, email, days_before = row
|
||
dl = str(days_left) if days_left is not None else 'N/A'
|
||
print(f"{domain:25s} {cert_type:12s} {status:8s} {str(expire_date):20s} {dl:>8s} {'是' if enabled else '否':>4s} {email or '-'}")
|
||
|
||
conn.close()
|
||
|
||
|
||
if __name__ == '__main__':
|
||
if '--dry-run' in sys.argv:
|
||
print("=== 预览模式 (--dry-run) ===\n")
|
||
check_and_notify(dry_run=True)
|
||
elif '--check' in sys.argv:
|
||
check_status()
|
||
else:
|
||
check_and_notify()
|