自動化SSL憑證監控與到期通知系統

def get_ssl_cert_expiry_date(domain):
    """
    取得 SSL 證書的過期日期。
    
    參數:
        domain (str): 需要檢查SSL證書過期時間的域名。
    
    返回:
        datetime: SSL證書的過期日期，如果獲取失敗則返回None。
    """
    # 建立SSL上下文，建立一個安全的“環境”來管理SSL設定和操作
    ssl_context = ssl.create_default_context()
    # 包裝socket對象，將基礎的socket通訊轉變為加密通訊
    conn = ssl_context.wrap_socket(socket.socket(socket.AF_INET), server_hostname=domain)
    # 設定連接的超時時間為 3 秒，防止長時間等待
    conn.settimeout(3.0)
    try:
        # 通過加密的連線嘗試連接到服務器的443端口（HTTPS）
        conn.connect((domain, 443))
        # 取得服務器的SSL證書訊息
        ssl_info = conn.getpeercert()
        # 解析證書中的過期時間
        expire_date = datetime.strptime(ssl_info['notAfter'], '%b %d %H:%M:%S %Y %Z')
        return expire_date
    except Exception as e:
        # 處理連接或取得證書訊息過程中的異常
        print(f"無法獲取 {domain} 的 SSL 證書過期日期，錯誤：{e}")
        return None
    finally:
        # 確保無論成功與否，都關閉與服務器的連接
        conn.close()

Mar  4 06:35:50 2024 GMT
月   日 時 分 秒  年   時區
%b  %d %H:%M:%S %Y   %Z

def check_ssl_expiration(domain):
    """
    檢查給定域名的SSL證書過期時間，並在證書即將過期時印出。
    
    參數:
        domain (str): 需要檢查SSL證書的域名。
    """
    # 調用先前定義的函數get_ssl_cert_expiry_date來獲取SSL證書的過期日期。
    expire_date = get_ssl_cert_expiry_date(domain)
    # 如果成功取得到過期日期
    if expire_date:
        # 計算證書的剩餘有效天數
        remaining_days = (expire_date - datetime.utcnow()).days
        # 如果剩餘天數不超過30天
        if remaining_days <= 30:
            # 印出一條log訊息，說明證書將在指定天數內過期
            print(f"{domain} 的 SSL 證書將在 {remaining_days} 天內過期。")
        else:
            # 如果證書的剩餘有效期超過30天，則打印證書的過期日
            print(f"{domain} 的 SSL 證書過期日期是 {expire_date.strftime('%Y-%m-%d')}。")

>>> from datetime import datetime
>>> datetime.utcnow()
## 返回一個`datetime.datetime`對象，這個對象包含了當前UTC時區的年、月、日、小時、分、秒，以及微秒。
datetime.datetime(2024, 4, 1, 15, 34, 21, 211636)

def send_notification(message, domain, telegram_bot_token, telegram_group_id):
    """
    通過指定的Webhook URL發送通知。

    參數:
        message (dict): 要發送的消息內容，格式為字典。
        domain (str): 域名，用於發送消息中表示哪個域名的SSL證書。
        telegram_token (str): 要使用的telegram bot token。
        telegram_group_id (str): 指定發送消息的群組。
    """
    telegram_send_message_url = f"https://api.telegram.org/bot{telegram_bot_token}/sendMessage"
    # 向telegram api發送HTTP POST請求。
    response = requests.post(telegram_send_message_url, data={
        "chat_id": telegram_group_id,
        "text": message
    })
    
    # 檢查響應的狀態碼。如果狀態碼為200，表示通知發送成功。
    if response.status_code == 200:
        print(f"已為 {domain} 發送通知")
    else:
        # 如果狀態碼不是200，表示發送失敗。
        print(f"為 {domain} 發送失敗")

def check_ssl_expiration(domain, env, platform, telegram_bot_token, telegram_group_id):
    """
    檢查給定域名的SSL證書過期時間，並在證書即將過期時通過webhook發送通知。
    
    參數:
        domain (str): 需要檢查SSL證書的域名。
        env (str): 環境標籤（例如：開發、測試、正式），用於消息中以區分不同環境。
        platform (str): 平台標籤（例如：AWS、GCP、Azure），用於消息中以標明證書部署的平台。
        telegram_token (str): 要使用的telegram bot token。
        telegram_group_id (str): 指定發送消息的群組。
    """
    expire_date = get_ssl_cert_expiry_date(domain)
    if expire_date:
        remaining_days = (expire_date - datetime.utcnow()).days
        if remaining_days <= 30:
            # 建立發送訊息，包含了證書到期的相關信息
            message = "\n".join([
			    "來源: Gitlab-Runner",
			    "標題: 憑證到期",
			    f"域名: {domain}",
			    f"到期日: {expire_date.strftime('%Y-%m-%d')}",
			    f"平台: {platform}",
			    f"環境: {env}",
			])
			
            print(f"{domain} 的 SSL 證書將在 {remaining_days} 天內過期。")
            send_notification(message, domain, telegram_bot_token, telegram_group_id)
        else:
            print(f"{domain} 的 SSL 證書過期日期是 {expire_date.strftime('%Y-%m-%d')}。")

domain_envs:
  live:
    - google.com
    - en.wikipedia.org

def load_data_from_yaml(yaml_file_path, key):
    """
    從YAML檔案加載指定鍵的配置。

    參數:
        yaml_file_path (str): YAML檔案的路徑。
        key (str): 要從YAML檔案中讀取的鍵名。

    返回:
        dict or None: 返回從YAML檔案中讀取的配置字典。如果指定的鍵不存在，則返回空字典。
    """
    try:
        # 嘗試打開指定的YAML檔案。'r'表示以讀取模式打開，'encoding='utf-8''確保文件正確讀取UTF-8編碼的內容。
        with open(yaml_file_path, 'r', encoding='utf-8') as file:
            # 使用yaml.safe_load(file)安全地加載YAML檔案的內容。
            # 此函數將YAML檔案的結構轉換為Python數據類型（通常是字典）。
            data = yaml.safe_load(file)
            # 嘗試從加載的數據中獲取特定鍵（key）的值。
            # 如果鍵不存在，則默認返回一個空字典{}。
            return data.get(key, {})
    except FileNotFoundError as e:
        # 如果嘗試打開的YAML檔案不存在，則捕獲FileNotFoundError異常。
        logging.error(f"YAML檔案未找到: {e}")
        # 錯誤日誌記錄後，返回一個空字典{}，表示沒有加載到任何數據。
        return {}
    except Exception as e:
        # 如果在加載或處理YAML檔案時發生了其他任何異常，則捕獲通用異常。
        logging.error(f"讀取YAML檔案時發生錯誤: {e}")
        # 同樣記錄錯誤日誌並返回一個空字典{}。
        return {}

yaml_file_path = 'config.yaml'
domain_envs = load_data_from_yaml(yaml_file_path, 'domain_envs')

{'live': ['google.com', 'en.wikipedia.org']}

for env, domains in domain_envs.items():
    for domain in domains:
        check_ssl_expiration(domain, env, platform, telegram_bot_token, telegram_group_id)

def get_env_variable(name, default_value="未設定"):
    """
    從系統環境變數中取得一個值。
    
    參數:
        name (str): 環境變數的鍵名。
        default_value (str): 如果找不到鍵，則返回的預設值。預設為"未設定"。
    
    返回:
        str: 環境變數的值，或者在找不到鍵時返回預設值。
    """
    return os.environ.get(name, default_value)

platform = get_env_variable("PLATFORM")
telegram_bot_token = get_env_variable("TELEGRAM_BOT_TOKEN")
telegram_group_id = get_env_variable("TELEGRAM_GROUP_ID")