""" 证书到期邮件通知服务 - 由 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"""

⚠️ SSL 证书即将到期

域名	{domain}
到期时间	{expire_date}
剩余天数	{days_left} 天
证书类型	Let's Encrypt (阿里云 DNS 验证)

系统已配置 acme.sh 自动续期，正常情况下会在到期前自动更新证书。
如续期失败，请检查：
1. 阿里云 AccessKey 是否有效
2. 服务器能否访问 Let's Encrypt API
3. DNS TXT 记录是否正常清理

此邮件由 Nginx 域名管理平台自动发送，连续提醒至证书续期成功。

""" 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()