延續上一篇的架構藍圖,在自動化威脅情資的收集過程中,面臨到一個實務上的挑戰:如何將龐雜且生硬的原始情資(Raw Data),轉化為能夠一目了然的戰術情報?
為了解決這個問題,我們在架構中設計了一支核心腳本 utils.py。這支腳本扮演著「情報調度指揮(Orchestration)」與「視覺化呈現」的關鍵角色,讓整個情資管線具備確定性(deterministic)與高可讀性。
utils.py 的四大架構重點與實作機制:
在自動化排程中,最忌諱的就是每次執行時重複發送相同的告警,造成分析師的「告警疲勞」。
load_state 函式,系統會讀取本地端的 JSON 狀態檔,並載入 seen_urls 陣列,精準盤點哪些情資網址已經被處理過。save_all_results 函式不僅會將更新後的狀態覆寫回檔案,還會自動生成一份 Markdown 格式的資安監控日誌。日誌內容會整齊列出情資的標籤、來源、主旨與原文連結,方便後續查閱。為了讓資安團隊能在第一時間辨識出最致命的威脅,腳本內建了 get_tag_style 函式,透過解析情資標籤(Tag)來自動配置對應的警示色彩:
將冷冰冰的數據轉為情報,需要良好的介面設計。send_visual_email 函式負責將收集到的情報組裝為 HTML 格式的多媒體電子郵件,並透過 Gmail 的 SMTP SSL 連線安全地發送給訂閱者。系統設計了三種切換自如的顯示風格(REPORT_STYLE):
在數位鑑識領域,凡走過必留下痕跡,情資系統本身也必須具備嚴謹的稽核(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")