iT邦幫忙

2026 iThome 鐵人賽

DAY 7
0
Security

從情資收集到資安鑑識:30 天建構自動化威脅情資與鑑識平台系列 第 7 篇

[Day 07]GitHub Actions - utils.py 視覺優化風格

  • 分享至 

  • xImage
  •  

延續上一篇的架構藍圖,在自動化威脅情資的收集過程中,面臨到一個實務上的挑戰:如何將龐雜且生硬的原始情資(Raw Data),轉化為能夠一目了然的戰術情報?
為了解決這個問題,我們在架構中設計了一支核心腳本 utils.py。這支腳本扮演著「情報調度指揮(Orchestration)」與「視覺化呈現」的關鍵角色,讓整個情資管線具備確定性(deterministic)與高可讀性。

utils.py 的四大架構重點與實作機制:

1. 狀態記憶與防重播機制(State Management)

在自動化排程中,最忌諱的就是每次執行時重複發送相同的告警,造成分析師的「告警疲勞」。

  • 讀取與比對:透過 load_state 函式,系統會讀取本地端的 JSON 狀態檔,並載入 seen_urls 陣列,精準盤點哪些情資網址已經被處理過。
  • 寫入與更新:當新一批情資處理完畢後,save_all_results 函式不僅會將更新後的狀態覆寫回檔案,還會自動生成一份 Markdown 格式的資安監控日誌。日誌內容會整齊列出情資的標籤、來源、主旨與原文連結,方便後續查閱。

2. 威脅標籤與視覺化風險分級(Risk Scoring & Tagging)

為了讓資安團隊能在第一時間辨識出最致命的威脅,腳本內建了 get_tag_style 函式,透過解析情資標籤(Tag)來自動配置對應的警示色彩:

  • 🚨 深鮮紅:代表核心警示或金融與設備相關的高危險情報。
  • ⚡ 亮紅:用於標示重大技術漏洞。
  • ⚠️ 黃橘:代表一般高危險情資。
  • 針對特定領域也設有專屬分類,例如 👑 橘色(精選日報)、🏦 紫色(金融)、🎯 藍色(關鍵設備)以及默認的 灰色(一般情報)。

3. 多元情資報表渲染(Dynamic Email Rendering)

將冷冰冰的數據轉為情報,需要良好的介面設計。send_visual_email 函式負責將收集到的情報組裝為 HTML 格式的多媒體電子郵件,並透過 Gmail 的 SMTP SSL 連線安全地發送給訂閱者。系統設計了三種切換自如的顯示風格(REPORT_STYLE):

  • 傳統表格(TABLE):以緊湊的行列結構呈現,適合一次性檢視大量情資,具備「情資類別」、「分數」、「來源」與「情資摘要」等明確欄位。
  • 現代卡片(CARDS):採用現代化的區塊設計,將每一筆情資封裝為帶有陰影的獨立卡片,凸顯來源與權重,適合行動裝置閱讀。
  • 簡約串流(TIMELINE):以左側帶有顏色邊框的條列式時間軸呈現,去除多餘框線,讓閱讀體驗更加純粹。

4. 稽核軌跡與原始資料留存(Audit & Logging)

在數位鑑識領域,凡走過必留下痕跡,情資系統本身也必須具備嚴謹的稽核(Audit)能力。

  • 原始探索紀錄:log_raw_discovery 函式會將所有抓取到的情資,以 CSV 格式追加寫入 raw_discovery.csv 檔案中。檔案內詳實記錄了「紀錄時間」、「來源」、「主旨」與「網址」四個關鍵維度,作為未來溯源的基礎數據。
  • 告警發送軌跡:log_alert_sent 函式則專注於記錄通訊行為,將成功發送的告警時間與主旨逐行寫入 alert_history.log,確保系統的每一次調度都有跡可循。

我們的 GitHub Actions 無伺服器情資中心不僅能自動化「抓取」資料,更透過 utils.py 的精巧設計,能專業地「消化」並「含有特色」情報。

以下為示範原碼

# utils.py
import json
import os
import smtplib
import csv
from datetime import datetime
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from config import *

def init_env():
    if not os.path.exists(LOG_DIR): os.makedirs(LOG_DIR)

def load_state():
    if os.path.exists(STATE_FILE):
        try:
            with open(STATE_FILE, 'r', encoding='utf-8') as f:
                data = json.load(f)
                return data if isinstance(data, dict) and "seen_urls" in data else {"seen_urls": []}
        except: pass
    return {"seen_urls": []}

def get_tag_style(tag):
    if "🚨" in tag: return "#D63031"  # 深鮮紅 (核心警示/類別與設備)
    if "⚡" in tag: return "#E74C3C"  # 亮紅 (重大技術漏洞)
    if "👑" in tag: return "#E67E22"  # 橘色 (精選日報)
    if "🏦" in tag: return "#8E44AD"  # 紫色 (特殊關鍵)
    if "🎯" in tag: return "#2980B9"  # 藍色 (關鍵設備)
    if "⚠️" in tag: return "#F39C12"  # 黃橘 (高危)
    return "#7F8C8D"                 # 灰色 (一般)

def send_visual_email(items):
    if not items or not MAIL_USERNAME or not MAIL_PASSWORD: return

    msg = MIMEMultipart('alternative')
    msg['Subject'] = f"資安預警 [{REPORT_STYLE}] - 新增 {len(items)} 筆重要情資"
    msg['From'] = f"情資監控系統 <{MAIL_USERNAME}>"
    msg['To'] = MAIL_TO

    # --- 共通 Header ---
    header_html = f"""
    <div style="border-left: 5px solid #d93025; padding-left: 12px; margin-bottom: 25px; display: flex; align-items: baseline;">
        <span style="color: #202124; font-size: 18px; font-weight: bold; margin-right: 15px;">資安情資監控</span>
        <span style="font-size: 14px; color: #5f6368;">收集時間:{datetime.now(TZ).strftime('%Y-%m-%d %H:%M:%S')}</span>
    </div>
    """

    content_body = ""

    # --- 風格 1: 傳統表格 (TABLE) ---
    if REPORT_STYLE == "TABLE":
        rows = ""
        for item in items:
            color = get_tag_style(item['tag'])
            rows += f"""
            <tr style='background-color:#f8f9fa; border-bottom:1px solid #dee2e6;'>
                <td style='padding:12px; text-align:center; color:{color}; font-weight:bold; font-size:16px;'>{item['tag']}</td>
                <td style='padding:12px; text-align:center; font-size:16px;'>{item['score']}</td>
                <td style='padding:12px; text-align:center; color:#5f6368; font-size:16px;'>{item['source']}</td>
                <td style='padding:12px; text-align:left;'>
                    <div style='color:#202124; margin-bottom:4px; font-size:16px;'>{item['title']}</div>
                    <a href='{item['url']}' style='color:#1a73e8; text-decoration:none; font-size:12px;'>{item['url']}</a>
                </td>
            </tr>"""
        content_body = f"""
        <table style='border-collapse:collapse; width:100%; border:1px solid #dee2e6;'>
            <thead><tr style='background-color:#f1f3f4;'>
                <th style='padding:12px; width:15%; font-size:16px;'>情資類別</th>
                <th style='padding:12px; width:8%; font-size:16px;'>分數</th>
                <th style='padding:12px; width:12%; font-size:16px;'>來源</th>
                <th style='padding:12px; text-align:left; font-size:16px;'>情資摘要</th>
            </tr></thead>
            <tbody>{rows}</tbody>
        </table>"""

    # --- 風格 2: 現代卡片 (CARDS) ---
    elif REPORT_STYLE == "CARDS":
        cards = ""
        for item in items:
            color = get_tag_style(item['tag'])
            cards += f"""
            <div style="background: #ffffff; border: 1px solid #e0e0e0; border-radius: 8px; padding: 16px; margin-bottom: 16px; box-shadow: 0 2px 4px rgba(0,0,0,0.05);">
                <div style="display: flex; justify-content: space-between; align-items: center; margin-bottom: 10px;">
                    <span style="background: {color}; color: white; padding: 4px 12px; border-radius: 20px; font-size: 14px; font-weight: bold;">{item['tag']}</span>
                    <span style="color: #7f8c8d; font-size: 14px;">來源: {item['source']} | 權重: {item['score']}</span>
                </div>
                <div style="font-size: 16px; color: #2c3e50; line-height: 1.4; margin-bottom: 10px;">{item['title']}</div>
                <div style="border-top: 1px solid #f1f1f1; padding-top: 8px;">
                    <a href="{item['url']}" style="color: #3498db; text-decoration: none; font-size: 13px;">➔ 查看完整情資內容</a>
                </div>
            </div>"""
        content_body = f"<div style='background: #f4f7f9; padding: 15px; border-radius: 10px;'>{cards}</div>"

    # --- 風格 3: 簡約串流 (TIMELINE) ---
    else:
        stream = ""
        for item in items:
            color = get_tag_style(item['tag'])
            stream += f"""
            <div style="border-left: 4px solid {color}; padding-left: 20px; margin-bottom: 25px; position: relative;">
                <div style="color: {color}; font-size: 14px; font-weight: bold; text-transform: uppercase; margin-bottom: 5px;">{item['tag']} · {item['source']}</div>
                <div style="font-size: 16px; color: #202124; margin-bottom: 5px;">{item['title']}</div>
                <a href="{item['url']}" style="color: #1a73e8; font-size: 13px; text-decoration: underline;">原文連結</a>
            </div>"""
        content_body = f"<div style='padding: 10px;'>{stream}</div>"

    full_html = f"""
    <html>
    <body style="font-family: 'Segoe UI', Roboto, sans-serif; padding: 20px; color: #333;">
        {header_html}
        {content_body}
        <p style='font-size: 12px; color: #999; text-align: center; margin-top: 30px;'>Copyright © 2026 Waason</p>
    </body>
    </html>
    """
    
    msg.attach(MIMEText(full_html, 'html', 'utf-8'))
    try:
        with smtplib.SMTP_SSL("smtp.gmail.com", 465) as server:
            server.login(MAIL_USERNAME, MAIL_PASSWORD)
            server.send_message(msg)
        print(f"    ✅ [{REPORT_STYLE}] 風格報表發送成功")
    except Exception as e: print(f"    ❌ 發送失敗: {e}")

def log_raw_discovery(items):
    file_path = "raw_discovery.csv"
    file_exists = os.path.isfile(file_path)
    with open(file_path, 'a', encoding='utf-8-sig', newline='') as f:
        writer = csv.writer(f)
        if not file_exists: writer.writerow(["紀錄時間", "來源", "主旨", "網址"])
        for item in items:
            writer.writerow([item.get('raw_time'), item.get('source'), item.get('title'), item.get('url')])

def log_alert_sent(items):
    current_time = datetime.now(TZ).strftime('%Y-%m-%d %H:%M:%S')
    with open("alert_history.log", "a", encoding="utf-8") as f:
        for item in items: f.write(f"[{current_time}] 發送成功: {item['title']}\n")

def save_all_results(state, processed_items):
    with open(STATE_FILE, 'w', encoding='utf-8') as f: json.dump(state, f, ensure_ascii=False, indent=2)
    if processed_items:
        with open(EXEC_MD_FILE, 'w', encoding='utf-8') as f:
            f.write(f"# 資安監控日誌 - {datetime.now(TZ).strftime('%Y-%m-%d %H:%M:%S')}\n\n")
            for item in processed_items: f.write(f"- **{item['tag']}** [{item['source']}] {item['title']} [連結]({item['url']})\n")
            

上一篇
[Day 06]GitHub Actions - keywords.py 威脅評分引擎
下一篇
[Day 08]GitHub Actions - main.py 管理核心
系列文
從情資收集到資安鑑識:30 天建構自動化威脅情資與鑑識平台 共 11 篇
圖片
  熱門推薦
圖片
{{ item.channelVendor }} | {{ item.webinarstarted }} |
{{ formatDate(item.duration) }}
直播中

尚未有邦友留言

立即登入留言