為確保事件追蹤一致性,我們建立統一格式的 incident_log.csv:
| 欄位名稱 | 說明 |
|---|---|
| incident_id | 事件唯一識別碼(UUID) |
| timestamp | 發生時間(UTC) |
| indicator | 網域或 IP |
| event_type | 事件類型(phishing / malware / botnet / unknown) |
| model_score | 模型預測分數 |
| ti_confidence | 威脅情報信心值 |
| source | 事件來源(API / Feed / Manual) |
| action_taken | 防禦行動(blocked / monitored / ignored) |
| status | 當前狀態(resolved / pending / false_positive) |
| related_incidents | 關聯事件 ID |
| remarks | 備註或人工說明 |
範例:
incident_id,timestamp,indicator,event_type,model_score,ti_confidence,source,action_taken,status,related_incidents,remarks
a12b34c,2025-10-23T02:35:00Z,secure-login.net,phishing,0.93,0.9,model,blocked,resolved,,Triggered by Ensemble Model
三、事件追蹤模組(Incident Tracer)
3.1 模組功能概述
從黑名單、Threat Feed 與模型結果中生成事件紀錄。
自動比對歷史事件,判斷是否為重複攻擊。
建立事件間的「時間與來源關聯圖」。
3.2 實作程式範例
python
複製程式碼
# incident_tracer.py
import pandas as pd, uuid, datetime
def create_incident(domain, event_type, model_score, confidence, source, action):
log = pd.read_csv("data/incident_log.csv") if "incident_log.csv" in os.listdir("data") else pd.DataFrame()
incident_id = str(uuid.uuid4())
record = {
"incident_id": incident_id,
"timestamp": datetime.datetime.utcnow().isoformat(),
"indicator": domain,
"event_type": event_type,
"model_score": model_score,
"ti_confidence": confidence,
"source": source,
"action_taken": action,
"status": "pending",
"related_incidents": "",
"remarks": "Generated automatically"
}
log = pd.concat([log, pd.DataFrame([record])], ignore_index=True)
log.to_csv("data/incident_log.csv", index=False)
print(f"✅ Incident {incident_id} created for {domain}")
# 範例:偵測到新釣魚網址時自動生成事件
create_incident("secure-login.net", "phishing", 0.94, 0.88, "model", "blocked")
四、事件回溯查詢(Incident Query Tool)
4.1 CLI 查詢版本
python
複製程式碼
# incident_query.py
import pandas as pd
df = pd.read_csv("data/incident_log.csv")
def query_indicator(indicator):
subset = df[df["indicator"].str.contains(indicator)]
if subset.empty:
print("No related incidents found.")
else:
print(subset[["incident_id","timestamp","event_type","action_taken","status"]])
query_indicator("secure-login.net")
執行結果:
nginx
複製程式碼
incident_id timestamp event_type action_taken status
a12b34c 2025-10-23T02:35:00Z phishing blocked resolved
五、事件關聯分析(Correlation Analysis)
5.1 基於 Domain / IP 關聯
當同一 domain 或 IP 多次出現在不同時間或來源,系統自動關聯成群組事件:
python
複製程式碼
# correlation.py
import pandas as pd
df = pd.read_csv("data/incident_log.csv")
df["count"] = df.groupby("indicator")["indicator"].transform("count")
hotlist = df[df["count"] > 1]
print("⚠️ Repeated indicators:")
print(hotlist[["indicator","count"]].drop_duplicates())
5.2 以時間區間建立攻擊鏈(Attack Timeline)
將近 24 小時內的事件視為同一攻擊波段:
python
複製程式碼
import datetime
cutoff = datetime.datetime.utcnow() - datetime.timedelta(hours=24)
recent = df[pd.to_datetime(df["timestamp"]) > cutoff]
print("Recent attack chain:")
print(recent[["timestamp","indicator","action_taken"]])
六、事件回溯視覺化介面(Streamlit)
6.1 Streamlit UI 範例
python
複製程式碼
# app_incident.py
import streamlit as st
import pandas as pd
st.title("🕵️♀️ 防禦事件回溯與調查系統")
df = pd.read_csv("data/incident_log.csv")
search = st.text_input("輸入 Domain 或 IP 以查詢:")
if search:
subset = df[df["indicator"].str.contains(search, case=False)]
if not subset.empty:
st.dataframe(subset)
else:
st.warning("查無相關事件。")
st.subheader("近期事件(24 小時內)")
recent = df.tail(10)
st.table(recent[["timestamp","indicator","event_type","action_taken","status"]])
執行:
bash
複製程式碼
streamlit run app_incident.py
七、事件報告導出(Incident Report Generator)
7.1 自動生成 PDF 報告
使用 reportlab 將事件摘要轉為 PDF:
python
複製程式碼
# report_generator.py
from reportlab.lib.pagesizes import A4
from reportlab.pdfgen import canvas
import pandas as pd
def export_incident_report(indicator):
df = pd.read_csv("data/incident_log.csv")
subset = df[df["indicator"].str.contains(indicator)]
if subset.empty:
print("No records.")
return
c = canvas.Canvas(f"report_{indicator}.pdf", pagesize=A4)
c.setFont("Helvetica-Bold", 14)
c.drawString(50, 800, f"Incident Report: {indicator}")
y = 760
for _, row in subset.iterrows():
c.setFont("Helvetica", 10)
c.drawString(50, y, f"{row['timestamp']} | {row['event_type']} | {row['action_taken']} | {row['status']}")
y -= 20
c.save()
print(f"✅ Report exported: report_{indicator}.pdf")
八、防禦事件生命週期(Incident Lifecycle)
階段 描述 負責模組
偵測 (Detection) 模型或 Threat Feed 偵測惡意指標 Day 18 / 20
記錄 (Logging) 自動生成事件紀錄 incident_tracer.py
分析 (Analysis) 事件關聯與時間序列分析 correlation.py
回應 (Response) 防火牆封鎖、通知安全人員 blacklist_service
回顧 (Review) 事件報告與模型修正 report_generator.py
九、安全與稽核考量
不可竄改性:事件紀錄採用追加式(append-only)結構,防止被修改。
時戳驗證:所有時間以 UTC ISO 格式存儲,便於跨區分析。
審核追蹤(Audit Trail):所有人工變更(例如解除封鎖)均需附註 reviewer_id。
法證保存:高風險事件報告需保存至少 90 天。