在前面 24 天的開發中，我們已經陸續完成 Process Detection、Network Detection、File Monitoring、Persistence Detection、Correlation Engine，以及基本的 Response 功能。
其中 Day22～24，也已經透過受控測試，成功使用 SIGTERM 終止 Linux 程序，並利用 nftables 阻斷指定 IP 的網路流量。
不過，做到這裡我發現一個很大的問題：
雖然每個功能都能獨立執行，但是它們還沒有真正整合成一套完整的系統。
例如，我們可以使用 Process Monitor 偵測 Python 指令，也可以利用 Network Monitor 偵測可疑的網路連線，但兩者原本並沒有共用同一條完整的事件處理流程。
因此，Day25 的目標不是增加新的 Detection Rule，而是建立 Unified Event Pipeline（統一事件處理流程），讓不同監控來源的 Alert 能夠進入 Correlation Engine、Response Dispatcher，並留下完整的紀錄。
今天主要完成以下功能：
首先要了解 Pipeline 是什麼。
在程式設計中，Pipeline 可以理解成一條資料處理流水線。
例如，一筆可疑程序事件出現後，系統不應只是顯示警告，而是要讓它繼續進入後面的處理模組。
我們希望建立的流程如下：
kunyo
Linux System Activity
↓
Process / Network Monitor
↓
Detection Engine
↓
Unified Event Pipeline
↓
Alert Logger
↓
Correlation Engine
↓
Response Dispatcher
↓
Response Audit Log
這些模組各自有不同的工作。
| 模組 | 負責工作 |
|---|---|
| Monitor | 收集 Linux 程序與網路活動 |
| Detection Engine | 利用規則判斷是否可疑 |
| Unified Pipeline | 統一處理不同來源的 Alert |
| Alert Logger | 保存偵測事件 |
| Correlation Engine | 分析不同 Alert 是否有關聯 |
| Response Dispatcher | 決定如何回應 |
| Audit Logger | 記錄 Response 決策及結果 |
這樣的設計有一個好處：
未來我們要增加新的 Detection Rule，不需要為每個規則重新設計 Response 流程。
只要能產生符合格式的 Alert，就能交給統一的 Pipeline 處理。
今天新增的核心檔案是：
response/event_pipeline.py
主要使用兩個函式：
prepare_event(alert, source)
ingest_alert(alert, source)
不同的監控器可能會產生不同格式的資料。
例如 Process Alert 主要包含：
PID
PPID
Process
Command
Parent
Severity
Network Alert 則主要包含：
PID
Process
Local IP
Local Port
Remote IP
Remote Port
State
Severity
因此，我們需要一個共同的事件格式。
核心程式：
from uuid import uuid4
from datetime import datetime, timezone
def prepare_event(alert, source):
event = dict(alert)
event.setdefault("event_id", str(uuid4()))
event.setdefault(
"timestamp",
datetime.now(timezone.utc).isoformat()
)
event["source"] = source
return event
這裡使用了 Python 的 uuid4()。
UUID（Universally Unique Identifier）是一種用來識別資料的編號。
我們可以把它想像成每筆資安事件的身分證字號。
例如今天的真實 Network Alert：
Event ID:
efa4ec13-8df2-4126-a1dd-1e0f5e7714cc
同一筆 Alert 經過 Detection、Response Dispatcher 和 Audit Logger 時，可以使用相同的 Event ID。
如此一來，日後就能知道：
「這筆 Response 決策究竟是由哪個 Detection Alert 產生的？」
UUID4 透過隨機方式產生識別值，碰撞機率極低，但它本身並不是安全驗證或授權機制。
接下來建立：
def ingest_alert(alert, source):
這個函式負責將一筆 Alert 送入整個處理流程。
核心邏輯：
event = prepare_event(alert, source)
save_alert(event)
correlations = add_alert(event) or []
dispatch_alert(event)
for correlation in correlations:
corr_event = prepare_event(
correlation,
"CORRELATION"
)
save_alert(corr_event)
dispatch_alert(corr_event)
這段程式的工作順序是：
第一步，幫 Alert 建立 Event ID。
第二步，把原始 Detection Alert 保存到 alerts.jsonl。
第三步，將 Alert 送入 Correlation Engine。
第四步，讓原始 Alert 進入 Response Dispatcher。
第五步，如果 Correlation Engine 產生新的關聯警告，也會將它保存並交給 Response Dispatcher。
這讓原本分散在不同監控器的處理流程，開始使用同一個入口。
今天第一個比較重要的問題出現在 Correlation Engine。
我們原本已經完成 BG-CORR-002，用來判斷 Python Inline Execution 與外部網路連線是否存在關聯。
測試時，Terminal 確實顯示：
========== BehaviorGuard Correlated Alert ==========
Rule : BG-CORR-002
Name : Python Execution with External Connection
Severity : HIGH
Description :
Python inline code execution was associated
with an unexpected external network connection.
看起來 Correlation 已經成功。
但是測試程式最後卻出現：
Generated Correlations: []
RuntimeError:
BG-CORR-002 was not generated
為什麼明明已經偵測到 Correlation，程式卻說沒有產生？
原本的 Correlation Engine 會執行：
print_correlation_alert(
correlation_rule,
matched_alerts
)
但 print() 只負責把內容顯示到 Terminal。
Terminal 看得到 Alert，不代表其他 Python 模組也能取得這筆 Alert。
如果沒有 return，外面的 Unified Pipeline 就無法接收到 Correlation Engine 產生的結果。
因此，我們修改了 correlation/engine.py。
在規則成立時，除了原本的顯示功能，還要建立一筆結構化的 Correlation Alert：
emitted.append({
"rule": correlation_rule["id"],
"name": correlation_rule["name"],
"severity": correlation_rule["severity"],
"description": correlation_rule["description"],
"category": "Correlation",
"pid": matched_alerts[-1].get("pid"),
"related_event_ids": [
item["event_id"]
for item in matched_alerts
if item.get("event_id")
]
})
最後回傳：
return emitted
原本的：
add_alert(alert)
也改成回傳檢查結果：
return check_correlation_rules()
這樣 Correlation Engine 就不只是顯示告警，而是能把告警資料交給其他模組。
再次執行：
PYTHONPATH=. python3 test_day25_correlation_pipeline.py
最後成功顯示：
Generated Correlations: ['BG-CORR-002']
[PASS] Correlation rule triggered
[PASS] Related Event IDs verified
[PASS] Unified Pipeline integration successful
這次成功驗證：
related_event_ids 可以保留原始 Detection Event 的識別編號。需要注意，這次使用的是模擬的 Detection Event，但實際執行的是我們真正寫好的 Correlation Rules 和 Unified Pipeline。
完成 Correlation Engine 後，下一步是修改：
agent/process_monitor.py
以前 Process Monitor 在產生 Alert 時，會個別執行：
add_alert(normalized_alert)
save_alert(normalized_alert)
dispatch_alert(normalized_alert)
雖然可以運作，但這樣的設計會讓每個 Monitor 都必須自己處理 Correlation、Logging 和 Response。
未來增加新的 Monitor 時，就需要重複相同程式。
因此，我們改成：
ingest_alert(
normalized_alert,
"PROCESS"
)
只要呼叫 ingest_alert()，就能交由 Unified Pipeline 完成後續工作。
我們執行：
python3 -c 'import time; print("Day25 Process Test"); time.sleep(8)'
BehaviorGuard 成功產生：
Rule : BG-CMD-002
Category : Process/Command
Severity : MEDIUM
PID : 2828420
Decision : INVESTIGATE
Execution : DRY RUN
原本的 Response Engine 也正常執行：
Action : Recommend further investigation
Mode : DRY RUN
Executed : False
這次的 Executed: False 並不是程式失敗。
因為 python3 -c 本身不一定是惡意行為，正常開發人員也可能使用。
因此，系統目前只產生調查建議，不直接終止程序。
接下來是今天花比較多時間的部分。
我們原本的 Network Monitor 已經具備多條偵測規則，例如：
| Rule ID | 偵測內容 |
|---|---|
| BG-NET-001 | Suspicious Network Listener |
| BG-NET-002 | Suspicious Network Connection |
| BG-NET-003 | Unexpected External Connection |
| BG-NET-004 | Unusual Process Port Behavior |
| BG-NET-005 | Shell or Interpreter Network Activity |
但原本 Network Monitor 偵測到 Alert 後，只會：
print_alert(alert, event)
也就是把 Alert 顯示在 Terminal。
它還沒有將真正的 Network Alert 送進 Unified Pipeline。
所以今天新增：
ingest_alert(
normalized_alert,
"NETWORK"
)
並將 PID、PPID、Process、Command、Network Scope、Local IP、Remote IP 和 Port 等資訊整理成標準事件。
一開始我們使用 Python 建立本機 TCP Listener。
結果 Process Monitor 成功偵測到：
BG-CMD-002
Python Inline Code Execution
但 Network Monitor 沒有產生 BG-NET-001。
後來檢查 Detection Rule 才發現：
suspicious_tools = [
"nc",
"ncat",
"socat"
]
而且還需要符合：
state == "LISTEN"
因此，普通的 Python Listener 並不符合 BG-NET-001 的程序名稱條件。
這讓我了解到：
收集到網路活動，不代表一定需要發出資安警告。
Detection Engine 應該根據規則判斷，而不是將所有網路連線都視為威脅。
後來檢查 agent/network_monitor.py，又發現一個更重要的問題。
原本使用：
["ss", "-ntp"]
但 ss -ntp 預設不會列出 TCP LISTEN Socket。
我們的 Network Detection Rule 卻需要偵測 LISTEN。
因此，即使有符合條件的程序正在監聽，也可能完全不會被 Network Collector 收集到。
這是監控資料來源與 Detection Rule 不一致的問題。
我們將它改成：
["ss", "-antp"]
其中：
| 參數 | 意義 |
|---|---|
-a |
顯示所有 Socket 狀態 |
-n |
使用數字顯示 IP 和 Port |
-t |
只顯示 TCP Socket |
-p |
顯示相關程序資訊 |
修改後，Network Collector 才能同時取得 LISTEN 與 ESTAB 狀態的 TCP Socket。
我們建立真正的 Python TCP Listener，再呼叫 Network Collector。
結果：
[PASS] ss changed to -antp
[PASS] Python syntax check
[PASS] Real LISTEN socket collected
PID : 2840751
Process : python3
State : LISTEN
Local IP: 127.0.0.1
Port : 19999
這次證明 Network Collector 已能正確取得真正存在的 TCP Listener。
不過，因為程序名稱是 python3，所以仍不符合 BG-NET-001 的規則。
之前 Process Monitor 和 Network Monitor 需要分開執行。
例如：
python3 agent/process_monitor.py
python3 agent/network_monitor.py
這樣有一個問題：
兩支 Python 程式各自有自己的記憶體空間。
即使它們都使用相同的 Correlation Engine 程式碼，也不代表會共用同一份 Alert Buffer。
我們原本的 Correlation Engine 使用：
alert_buffer = []
保存最近 30 秒的 Detection Alert。
如果 Process Monitor 和 Network Monitor 分別在兩個 Python 行程執行，它們就無法直接共享這個 List。
因此，今天新增：
agent/behaviorguard.py
利用 Python 的 threading，讓兩個 Monitor 在同一個 Python 行程中執行。
核心概念：
import threading
threading.Thread(
target=continuous_monitor,
daemon=True
)
threading.Thread(
target=monitor_network,
daemon=True
)
這樣 Process Monitor 和 Network Monitor 就能使用同一個 Correlation Engine 記憶體狀態。
另外，我們也在 Unified Pipeline 使用 RLock，讓不同監控執行緒依序處理事件，降低同時存取 Correlation Buffer 造成資料競爭的風險。
現在只需要執行：
PYTHONPATH=. python3 -u agent/behaviorguard.py
實際結果：
========== BehaviorGuard ==========
[INFO] Unified monitoring started.
[INFO] Process + Network monitoring
[INFO] Response mode: DRY RUN
[INFO] Press Ctrl+C to stop.
[START] Process Monitor
[INFO] Continuous monitoring started.
[START] Network Monitor
[BehaviorGuard] Enriched Network Monitor started...
這代表兩種監控功能可以透過同一個入口啟動。
不過，目前還是 Python Threading 原型，尚未加入完整的服務管理、自動重啟和異常復原機制。
前面已經確認 Network Collector 能取得 LISTEN Socket。
接下來，我們要使用真正符合 Detection Rule 的工具進行測試。
這次選擇 socat。
socat 是 Linux 的網路工具，可以建立 TCP 連線、監聽 Socket，以及轉送資料。
它本身不是惡意程式，但也可能被攻擊者利用來建立網路通道。
因此，我們的 BG-NET-001 將它列為需要進一步注意的工具之一。
測試程式使用 socat 在本機建立 TCP 監聽：
127.0.0.1:19999
這是 Loopback 位址，不會直接對外部網路提供測試服務。
[PASS] Real TCP listener detected
PID : 2841956
Process : socat
State : LISTEN
Address : 127.0.0.1:19999
接著 Detection Engine 成功比對：
Matched Rules: ['BG-NET-001']
[PASS] BG-NET-001 rule matched
這次不再是模擬 Network Event，而是真的讓 Kali Linux 執行 socat，由 Network Monitor 收集 Socket 資訊。
Unified Monitor 顯示：
========== BehaviorGuard Network Alert ==========
Rule : BG-NET-001
Name : Suspicious Network Listener
Severity : HIGH
Process : socat
PID : 2841956
Parent : python3
Command :
socat TCP-LISTEN:19999,bind=127.0.0.1,reuseaddr -
Description:
socat opened a listening socket on 127.0.0.1:19999.
這代表 BG-NET-001 確實偵測到真正的 Linux 網路監聽行為。
但最重要的是後面的處理流程。
========== BehaviorGuard Pipeline ==========
Event ID : efa4ec13-8df2-4126-a1dd-1e0f5e7714cc
Source : NETWORK
Rule : BG-NET-001
[INFO] Alert saved to logs/alerts.jsonl
這證明真正的 Network Alert 已經成功進入 Unified Pipeline。
接著 Response Dispatcher 顯示：
========== BehaviorGuard Dispatcher ==========
Rule : BG-NET-001
Category : Network
Severity : HIGH
PID : 2841956
Decision : REVIEW_REQUIRED
Execution : DRY RUN
原本的 Response Engine 也正常提供建議：
Action : Recommend process termination
Mode : DRY RUN
Executed : False
最後產生：
[AUDIT] RESPONSE_DECISION -> REVIEW_REQUIRED
[INFO] Live response decision recorded.
我們另外比對了 Detection Log 與 Response Audit Log 的 Event ID。
測試結果：
[PASS] Real Network Alert logged
Event ID:
efa4ec13-8df2-4126-a1dd-1e0f5e7714cc
[PASS] Response Audit matched
這證明同一筆真正的 Network Alert，已經能夠從偵測一路追蹤到 Response 決策紀錄。
經過今天的整合與測試，BehaviorGuard 已經完成以下功能：
| 功能 | 結果 |
|---|---|
| Process Monitor 持續監控 | 成功 |
| Network Monitor 持續監控 | 成功 |
| Process / Network 統一啟動 | 成功 |
| Unified Event Pipeline | 成功 |
| 真實 Process Alert 進入 Dispatcher | 成功 |
| 真實 Network Alert 進入 Dispatcher | 成功 |
| Correlation Alert 回傳 Pipeline | 測試成功 |
| Detection Event ID | 成功 |
| Correlation related_event_ids | 測試成功 |
| Response Audit Log | 成功 |
| 真正自動終止未知程序 | 尚未開放 |
| 真正自動封鎖任意外部 IP | 尚未開放 |
其中 Correlation 是使用測試事件進行真實規則比對，尚未驗證一段真實攻擊鏈能在持續監控下自動產生關聯警告。
而 Day22～24 已經測試過的 SIGTERM 與 nftables 阻斷功能，今天沒有直接開放給正式監控器任意執行。
原因是目前還缺少更完整的自動回應策略。
例如今天的 socat 是我們自己建立的無害測試程序，但仍被判定為 HIGH。
如果系統只根據 Severity 自動終止或阻斷，很容易造成誤判處置。
因此，正式監控目前仍採用 Dry Run，先確保偵測、關聯、回應決策與紀錄可以穩定運作。
今天最大的收穫，是了解到資安系統並不是把很多 Detection Rule 寫出來，就能算完成一套 EDR。
前面幾天我們一直增加 Process、Network、File、Persistence 等偵測功能，也做出初步的 Correlation 與 Response。
但直到今天真正進行整合，才發現很多功能雖然單獨運作正常，彼此之間卻沒有完整的資料傳遞方式。
例如 Correlation Engine 原本可以成功顯示 BG-CORR-002，但因為沒有回傳 Alert，Response Dispatcher 根本收不到結果。
還有 Network Monitor 原本使用 ss -ntp，卻希望偵測 LISTEN Socket，導致資料收集方式與規則需求不一致。
這些問題讓我發現，資安系統除了 Detection Rule 的準確性，還需要注意資料格式、模組介面、事件處理順序及日誌追蹤。
今天加入 Event ID 後，我們也能將同一筆 Detection Alert 與 Response Audit Log 對應起來，而不是只看到一堆無法連結的警告訊息。
最後透過真正的 socat Listener 測試，確認 Network Monitor 能夠將 BG-NET-001 送入 Unified Pipeline，再由 Response Dispatcher 做出 REVIEW_REQUIRED 的決策。
這讓 BehaviorGuard 從原本各自獨立的功能，逐步變成可以共同運作的端點監控與回應決策原型。
接下來 Day26，我們將開始處理真正的 Auto Response Policy。
目標是讓正式 Detection Alert 經過安全策略驗證後，能夠對明確授權的受控測試目標，自動執行 SIGTERM 或 nftables IP Blocking。
同時加入更完整的 IP 保護、封鎖期限與誤判防護，讓 Response 不只是有能力執行，更要能安全地執行。
Day25 到這裡正式完成。
iThome鐵人賽
看影片追技術