# Source: https://github.com/CN-Grace/QinglongScripts/blob/main/SSL.py
# Raw: https://raw.githubusercontent.com/CN-Grace/QinglongScripts/main/SSL.py
# Repo: CN-Grace/QinglongScripts
# Path: SSL.py
# UploadedAt: 2026-05-23T16:04:54Z
# SHA256: c1f33bef544c897a6d3ea4c28fde98c6ee9729002ec45451f6197a1348d24589
# Category: APP版/抓包
# Evidence: cookie/token/authorization/header

#!/usr/bin/env python3
"""
cron: 0 0 * * *
new Env("SSL证书检查")
SSL 证书检查脚本
- 批量检查多个域名的 SSL 证书状态
- 获取证书到期时间、剩余天数、颁发者等信息
- 分类显示正常、警告、过期和检查失败的证书
"""

import os
import ssl
import socket
from datetime import datetime, timezone
from typing import List, Dict

from utils import log_info, log_success, log_warning, log_error, beijing_now, beijing_time_str
from notifier import send as notify_send

# ==================== 用户配置 ====================
DOMAINS_TO_CHECK = [d.strip() for d in os.environ.get("SSL_DOMAINS", "").split(",") if d.strip()]
WARNING_THRESHOLD = 30
CONNECTION_TIMEOUT = 10


def get_certificate_info(domain: str) -> Dict:
    """获取单个域名的 SSL 证书信息"""
    result = {"domain": domain, "expiry_date": datetime.min, "days_left": -1, "issuer": "", "is_valid": False, "error": None}
    try:
        hostname = domain if ":" in domain else f"{domain}:443"
        host, port = hostname.split(":")
        port = int(port)
        context = ssl.create_default_context()
        with socket.create_connection((host, port), timeout=CONNECTION_TIMEOUT) as sock:
            with context.wrap_socket(sock, server_hostname=host) as ssock:
                cert = ssock.getpeercert()

        expiry_str = cert["notAfter"]
        expiry_date = datetime.strptime(expiry_str, "%b %d %H:%M:%S %Y %Z").replace(tzinfo=timezone.utc)
        now = datetime.now(timezone.utc)
        days_left = (expiry_date - now).days
        issuer_dict = dict(x[0] for x in cert["issuer"])
        issuer = issuer_dict.get("organizationName", "Unknown")
        result.update({"expiry_date": expiry_date, "days_left": days_left, "issuer": issuer, "is_valid": days_left > 0})
        log_info(f"{domain} 证书剩余 {days_left} 天")
    except Exception as e:
        result["error"] = str(e)
        log_error(f"{domain} 检查失败: {e}")
    return result


def check_all_domains(domains: List[str]) -> List[Dict]:
    """检查所有域名的 SSL 证书"""
    results = []
    log_info(f"开始检查 {len(domains)} 个域名的 SSL 证书...")
    for domain in domains:
        log_info(f"正在检查: {domain}")
        cert_info = get_certificate_info(domain)
        results.append(cert_info)
        if cert_info["is_valid"]:
            if cert_info["days_left"] <= WARNING_THRESHOLD:
                log_warning(f"  ⚠️ {domain} 将在 {cert_info['days_left']} 天后过期")
            else:
                log_success(f"  ✅ {domain} 证书正常，剩余 {cert_info['days_left']} 天")
        elif cert_info["error"]:
            log_error(f"  🔧 {domain} 检查失败: {cert_info['error']}")
        else:
            log_error(f"  ❌ {domain} 证书已过期")
    return results


def categorize_certificates(certificates: List[Dict]) -> Dict[str, List[Dict]]:
    """将证书结果分类"""
    warning_certs, expired_certs, valid_certs, error_certs = [], [], [], []
    for cert in certificates:
        if cert["error"] is not None:
            error_certs.append(cert)
        elif cert["is_valid"]:
            if 0 < cert["days_left"] <= WARNING_THRESHOLD:
                warning_certs.append(cert)
            elif cert["days_left"] > WARNING_THRESHOLD:
                valid_certs.append(cert)
        else:
            expired_certs.append(cert)
    return {"warning": warning_certs, "expired": expired_certs, "valid": valid_certs, "error": error_certs}


def format_certificate_report(certificates: List[Dict]) -> str:
    """格式化证书检查报告"""
    cats = categorize_certificates(certificates)
    lines = [f"🔔 SSL 证书检查报告", "", f"⏰ 检查时间: {beijing_time_str()}", f"📊 总计: {len(certificates)} 个域名", ""]

    if cats["expired"]:
        lines.append("❌ 已过期的证书:")
        for cert in cats["expired"]:
            expiry = cert["expiry_date"].strftime("%Y-%m-%d") if cert["expiry_date"] != datetime.min else "未知"
            lines.append(f"   {cert['domain']} — 已过期 {abs(cert['days_left'])} 天 (到期: {expiry})")
        lines.append("")

    if cats["warning"]:
        lines.append("⚠️ 即将过期的证书 (30天内):")
        for cert in cats["warning"]:
            expiry = cert["expiry_date"].strftime("%Y-%m-%d")
            lines.append(f"   {cert['domain']} — 剩余 {cert['days_left']} 天 (到期: {expiry} | 颁发: {cert['issuer']})")
        lines.append("")

    if cats["valid"]:
        lines.append("✅ 证书状态正常:")
        for cert in cats["valid"]:
            lines.append(f"   {cert['domain']} — 剩余 {cert['days_left']} 天")
        lines.append("")

    if cats["error"]:
        lines.append("🔧 检查失败的域名:")
        for cert in cats["error"]:
            lines.append(f"   {cert['domain']} — 错误: {cert['error']}")
        lines.append("")

    lines.append("📈 统计信息:")
    lines.append(f"   ✅ 正常: {len(cats['valid'])}")
    lines.append(f"   ⚠️ 警告: {len(cats['warning'])}")
    lines.append(f"   ❌ 过期: {len(cats['expired'])}")
    lines.append(f"   🔧 失败: {len(cats['error'])}")
    lines.append("")
    lines.append("─" * 18)
    lines.append(f"🕒 执行时间: {beijing_time_str()}")
    return "\n".join(lines)


def main():
    log_info("=" * 50)
    log_info("SSL 证书检查脚本开始执行")
    log_info("=" * 50)

    cert_results = check_all_domains(DOMAINS_TO_CHECK)
    report = format_certificate_report(cert_results)
    notify_send("SSL 证书检查报告", report)

    cats = categorize_certificates(cert_results)
    print(f"\n{'=' * 60}")
    print(f"SSL 证书检查完成! 检查域名总数: {len(cert_results)}")
    print(f"正常: {len(cats['valid'])} | 警告: {len(cats['warning'])} | 过期: {len(cats['expired'])} | 失败: {len(cats['error'])}")
    print("=" * 60)

    if cats["expired"] or cats["warning"]:
        print("\n⚠️ 需要注意的域名:")
        for cert in cats["expired"]:
            print(f"  ❌ {cert['domain']} — 已过期")
        for cert in cats["warning"]:
            print(f"  ⚠️ {cert['domain']} — 剩余 {cert['days_left']} 天到期")

    log_info("=" * 50)
    log_info(f"脚本执行结束")
    log_info("=" * 50)


if __name__ == "__main__":
    main()