🎯 Initial Access (初始存取)
    ↓
🔍 Discovery (偵察發現)
    ↓
🏃 Lateral Movement (橫向移動)
    ↓
🎛️ Collection (資料收集)
    ↓
💥 Impact (影響破壞)

攻擊手法：
- 發送含有惡意附件的釣魚郵件
- 目標：電力公司員工
- 惡意檔案：偽裝成合法文件的巨集病毒

防護策略：
✓ 郵件安全閘道：檢測惡意附件
✓ 使用者意識培訓：識別釣魚郵件
✓ 端點防護：阻止巨集執行
✓ 網路分段：限制企業網路對OT網路的存取

攻擊行為：
- 使用 nmap 掃描內部網路
- 發現 SCADA 系統和 HMI 工作站
- 收集 Modbus 設備清單和配置

實際發現的漏洞：
- 未加密的 Modbus TCP 通訊
- 預設密碼的 HMI 系統
- 缺乏網路分段的扁平架構

防護建議：
✓ 部署網路監控：檢測異常掃描行為
✓ 資產清冊管理：了解所有連網設備
✓ 零信任架構：最小權限原則
✓ 蜜罐系統：誘捕和偵測攻擊者

最終攻擊：
- 遠端控制 HMI 系統
- 手動關閉變電站斷路器
- 覆寫 PLC 韌體造成設備無法重啟

關鍵失誤：
- HMI 系統可直接控制關鍵設備
- 缺乏操作驗證機制
- 沒有安全的設備復原程序

改善措施：
✓ 雙因素認證：所有關鍵操作需要驗證
✓ 操作確認：重要動作需要雙人授權
✓ 安全備份：離線儲存 PLC 程式備份
✓ 應急程序：手動操作備援機制

攻擊路徑：
企業網路 → DMZ → 工程師工作站 → SIS 網路

關鍵弱點：
- 工程師工作站同時連接企業網路和SIS網路
- SIS 網路缺乏入侵檢測
- Tristation 通訊協議未加密

MITRE 對應技術：
- T0866: Exploitation of Remote Services
- T0859: Valid Accounts (使用有效帳號)
- T0847: Replication Through Removable Media

攻擊目標：
- Schneider Electric Tristation 軟體
- 修改 SIS 邏輯程式
- 繞過安全聯鎖功能

實際影響：
- 幸運的是，攻擊者的惡意程式存在 bug
- SIS 系統進入安全模式而非被控制
- 工廠緊急停機但避免了災難

防護策略：
✓ 物理隔離：SIS 系統完全離線
✓ 單向通訊：數據只能由 SIS 向外傳送
✓ 專用工程師工作站：僅用於 SIS 維護
✓ 嚴格變更管控：所有程式修改需要核准

攻擊入口：
- 過期的 VPN 帳號仍然有效
- 缺乏多因素認證
- 弱密碼保護

DarkSide 勒索軟體特徵：
- 避免攻擊工控系統（RaaS 集團的「職業道德」）
- 主要加密企業 IT 系統
- 要求比特幣贖金

實際 Modbus 相關影響：
- SCADA 系統本身未受直接攻擊
- 但公司出於謹慎關閉了 OT 系統
- 展現了 IT/OT 系統相互依賴性

關鍵發現：
1. IT 系統被攻陷可能導致 OT 系統停機
2. 即使 OT 系統安全，業務仍可能中斷
3. 應急程序和復原能力同樣重要

防護建議：
✓ IT/OT 網路完全分離
✓ 建立離線的應急操作程序
✓ 定期進行業務連續性演練
✓ 建立安全的遠端存取機制

威脅場景：
攻擊者透過釣魚郵件進入企業網路，然後嘗試存取 Modbus 系統

防護策略：
🛡️ 技術控制：
- 郵件安全閘道 (SEG)
- 端點偵測與回應 (EDR)
- 網路存取控制 (NAC)

🎓 管理控制：
- 資安意識培訓
- 事件回應程序
- 定期滲透測試

📋 程序控制：
- 最小權限原則
- 職責分離
- 定期權限檢視

Modbus 特定風險：
- HMI Web 介面暴露在網際網路
- Modbus TCP 埠直接對外開放
- 預設密碼未更改

真實案例 - Shodan 搜尋結果：
"截至2024年，全球仍有超過1,000個 Modbus 設備
直接暴露在網際網路上，主要分布在：
- 美國：35%
- 德國：15%
- 中國：12%
- 俄羅斯：8%"

防護實務：
✓ 絕不將 Modbus 設備直接連網
✓ 使用 VPN 或專線進行遠端存取
✓ 實施 IP 白名單限制
✓ 定期進行外部掃描檢測

攻擊特徵：
- 使用 nmap 掃描 Modbus 設備
- 利用 Modbus 功能碼 0x2B 進行設備識別
- 透過 502 埠嗅探發現設備

檢測技術：
🔍 網路監控：
- 異常掃描行為偵測
- 大量連線嘗試警報
- 非標準 Modbus 請求監控

📊 行為分析：
- 建立正常流量基線
- 偵測異常功能碼使用
- 監控設備識別請求

實際部署案例：
某汽車工廠部署 Claroty 工控安全平台後，
成功檢測到內部工程師工作站的異常掃描行為，
原來是新安裝的監控軟體配置錯誤。

風險場景：
攻擊者獲得有效的 SCADA 帳號，透過 Modbus 操作設備

防護策略：
🔐 身份管理：
- 強制多因素認證
- 定期密碼更換
- 帳號生命週期管理

🔍 行為監控：
- 異常登入時間偵測
- 地理位置異常警報
- 權限提升監控

真實案例 - Stuxnet 借鑑：
雖然 Stuxnet 主要針對 Siemens PLC，但其使用
有效憑證的技術同樣適用於 Modbus 環境：
- 竊取的數位憑證
- 有效的使用者帳號
- 合法的工程軟體

最終攻擊目標：
- 竄改 Modbus 設定點
- 關閉關鍵設備
- 破壞生產程序

多層防護策略：

🏭 物理層：
- 安全儀錶系統 (SIS) 獨立運作
- 硬體安全開關
- 緊急停機按鈕

⚙️ 控制層：
- PLC 程式碼保護
- 安全功能塊
- 硬體寫保護

📡 通訊層：
- Modbus 功能碼白名單
- 地址範圍限制
- 操作速率限制

實際案例 - 台灣某半導體廠：
實施了四重保護機制：
1. SCADA 操作需雙人確認
2. 關鍵參數變更需主管授權
3. PLC 程式有硬體寫保護
4. 異常操作自動觸發安全程序

🌐 Level 5: 企業網路 (Enterprise Zone)
   ├── ERP 系統、郵件伺服器
   └── 防火牆 ↓

🏢 Level 4: DMZ 區域 (DMZ Zone)
   ├── 資料史學伺服器、報表系統
   └── 工業防火牆 ↓

📊 Level 3: SCADA 網路 (SCADA Zone)
   ├── HMI 工作站、歷史伺服器
   └── 深度封包檢測 ↓

⚙️ Level 2: 控制網路 (Control Zone)
   ├── PLC、Modbus RTU Gateway
   └── 單向通訊 ↓

🔒 Level 1: 安全網路 (Safety Zone)
   └── SIS 系統 (完全隔離)

影響設備：
- 多款工業 HMI 系統
- Modbus TCP/IP 通訊模組
- 第三方 Modbus 函式庫

技術細節：
- CVE-2021-22681: 緩衝區溢位
- CVE-2021-22682: 格式字串漏洞
- 攻擊向量：惡意 Modbus 封包

CVSS 評分：9.8 (嚴重)

受害者：美國某水處理廠
攻擊手法：
1. 透過網路掃描發現脆弱的 HMI 系統
2. 發送畸形 Modbus 封包觸發漏洞
3. 獲得 HMI 系統控制權
4. 嘗試修改水處理參數

幸運結果：
- 安全儀錶系統及時介入
- 異常參數被自動修正
- 未造成實際危害

防護措施：
✓ 立即套用安全更新
✓ 實施 Modbus 封包過濾
✓ 加強網路監控
✓ 檢討應急程序

## 0x03 應用層安全強化

## 0x05 事件回應與恢復實戰

### 案例一：NotPetya 對 Maersk 的衝擊 (2017年)

**事件背景：**
2017年6月，NotPetya 勒索軟體攻擊了全球航運巨頭 Maersk，雖然主要影響 IT 系統，但也波及到港口的 Modbus 控制系統。

**影響分析：**

**應急回應程序：**

**學到的教訓：**
1. **IT/OT 系統相互依賴性**：即使攻擊主要針對 IT 系統，OT 系統也會受到影響
2. **備援通訊的重要性**：需要獨立的通訊管道來協調恢復工作
3. **手動程序的必要性**：關鍵業務需要有手動操作的備援方案

### 案例二：Colonial Pipeline 恢復經驗 (2021年)

**快速恢復策略：**

#### 分段恢復方法

#### 業務連續性措施

### 案例三：台灣半導體廠地震應變

**背景：**
台灣某晶圓廠在 921 地震後，建立了完整的災害恢復計畫，包括 Modbus 控制系統的快速恢復程序。

**恢復優先級設定：**

**實際演練結果：**

## 0x06 國際合規與最佳實務

### IEC 62443 實施案例

**歐盟 NIS2 指令下的合規實務**

隨著歐盟 NIS2 指令的實施，關鍵基礎設施營運商必須符合更嚴格的網路安全要求。

#### 德國製造業合規經驗

#### 美國 NERC CIP 電力標準

### 全球最佳實務綜合

#### NIST Cybersecurity Framework 應用

#### 應用層安全強化

**身份驗證與授權系統**

```python
import hashlib
import hmac
import json
import time
from cryptography.fernet import Fernet

class ModbusSecurityManager:
    def __init__(self, secret_key):
        self.secret_key = secret_key
        self.cipher = Fernet(Fernet.generate_key())
        self.user_database = {}
        self.audit_log = []

    def authenticate_request(self, api_key, signature, request_data):
        """驗證請求身份"""
        # 查找使用者
        user_data = self.user_database.get(api_key)
        if not user_data:
            return False, "無效的 API 金鑰"

        # 驗證簽名
        expected_signature = hmac.new(
            self.secret_key.encode(),
            request_data.encode(),
            hashlib.sha256
        ).hexdigest()

        if not hmac.compare_digest(request_signature, expected_signature):
            return False, "請求簽名無效"

        return True, user_data

    def authorize_operation(self, user_data, operation):
        """授權檢查"""
        user_permissions = user_data['permissions']

        permission_mapping = {
            'read_coils': 'read',
            'read_discrete_inputs': 'read',
            'read_holding_registers': 'read',
            'read_input_registers': 'read',
            'write_single_coil': 'write',
            'write_single_register': 'write',
            'write_multiple_coils': 'write_bulk',
            'write_multiple_registers': 'write_bulk'
        }

        required_permission = permission_mapping.get(operation)
        if not required_permission:
            return False, f"未知的操作: {operation}"

        if required_permission not in user_permissions:
            return False, f"權限不足: 需要 {required_permission}"

        return True, "授權成功"

    def secure_modbus_request(self, api_key, signature, encrypted_request):
        """安全的 Modbus 請求處理"""
        try:
            # 1. 解密請求
            decrypted_request = self.cipher.decrypt(encrypted_request.encode())
            request_data = json.loads(decrypted_request.decode())

            # 2. 驗證身份
            auth_success, user_data = self.authenticate_request(
                api_key, signature, encrypted_request
            )

            if not auth_success:
                return self.create_error_response("認證失敗", user_data)

            # 3. 授權檢查
            operation = request_data.get('operation')
            auth_success, auth_message = self.authorize_operation(user_data, operation)

            if not auth_success:
                return self.create_error_response("授權失敗", auth_message)

            # 4. 執行 Modbus 操作
            result = self.execute_modbus_operation(request_data)

            # 5. 記錄稽核日誌
            self.log_operation(user_data['username'], operation, request_data, result)

            # 6. 加密回應
            response = {
                'success': True,
                'data': result,
                'timestamp': time.time()
            }

            encrypted_response = self.cipher.encrypt(json.dumps(response).encode())
            return encrypted_response.decode()

        except Exception as e:
            return self.create_error_response("處理錯誤", str(e))

    def execute_modbus_operation(self, request_data):
        """執行 Modbus 操作"""
        operation = request_data['operation']
        params = request_data['parameters']

        # 連線到實際的 Modbus 設備
        client = ModbusTCPClient(params['host'], params.get('port', 502))
        client.connect()

        try:
            if operation == 'read_holding_registers':
                result = client.read_holding_registers(
                    params['address'],
                    params['count']
                )
                return {'registers': result}

            elif operation == 'write_single_register':
                client.write_single_register(
                    params['address'],
                    params['value']
                )
                return {'status': 'success'}

            # 其他操作...

        finally:
            client.close()

    def log_operation(self, username, operation, request_data, result):
        """記錄操作日誌"""
        log_entry = {
            'timestamp': time.time(),
            'username': username,
            'operation': operation,
            'parameters': request_data.get('parameters', {}),
            'success': result.get('success', False),
            'ip_address': request_data.get('client_ip'),
            'user_agent': request_data.get('user_agent')
        }

        self.audit_log.append(log_entry)

        # 可以擴展為寫入資料庫或日誌檔案
        print(f"稽核日誌: {log_entry}")

# 安全配置管理
class SecurityConfiguration:
    def __init__(self):
        self.config = {
            'access_control': {
                'require_authentication': True,
                'session_timeout': 3600,  # 1 小時
                'max_failed_attempts': 3,
                'lockout_duration': 900   # 15 分鐘
            },
            'encryption': {
                'algorithm': 'AES-256',
                'key_rotation_interval': 86400,  # 24 小時
                'require_tls': True
            },
            'audit': {
                'log_all_operations': True,
                'log_failed_attempts': True,
                'retention_days': 90
            },
            'rate_limiting': {
                'requests_per_minute': 60,
                'burst_limit': 10
            }
        }

    def validate_configuration(self):
        """驗證安全配置"""
        issues = []

        # 檢查認證設定
        if not self.config['access_control']['require_authentication']:
            issues.append("建議啟用認證機制")

        if self.config['access_control']['session_timeout'] > 7200:
            issues.append("會話超時時間過長，建議設為 2 小時以內")

        # 檢查加密設定
        if not self.config['encryption']['require_tls']:
            issues.append("建議啟用 TLS 加密")

        # 檢查稽核設定
        if not self.config['audit']['log_all_operations']:
            issues.append("建議記錄所有操作以便稽核")

        return issues

    def generate_security_recommendations(self):
        """產生安全建議"""
        recommendations = []

        recommendations.extend([
            "實施多因素認證 (MFA)",
            "定期更新密碼和 API 金鑰",
            "使用強式加密演算法",
            "實施網路存取控制清單 (ACL)",
            "定期進行安全稽核",
            "建立事件回應程序",
            "進行定期的滲透測試",
            "實施資料備份和恢復程序"
        ])

        return recommendations

        if self.metrics['request_rate']:
            timestamps = [m['timestamp'] for m in self.metrics['request_rate']]
            values = [m['value'] for m in self.metrics['request_rate']]

            axes[0, 1].plot(timestamps, values, color='orange')
            axes[0, 1].set_title('請求速率')
            axes[0, 1].set_ylabel('請求/分鐘')

        # 錯誤率
        if self.metrics['error_rate']:
            timestamps = [m['timestamp'] for m in self.metrics['error_rate']]
            values = [m['value'] for m in self.metrics['error_rate']]

            axes[1, 0].plot(timestamps, values, color='red')
            axes[1, 0].set_title('錯誤率')
            axes[1, 0].set_ylabel('錯誤率')

        # 警報統計
        alert_types = {}
        for alert in self.alerts:
            alert_type = alert['type']
            alert_types[alert_type] = alert_types.get(alert_type, 0) + 1

        if alert_types:
            axes[1, 1].bar(alert_types.keys(), alert_types.values())
            axes[1, 1].set_title('警報類型分布')
            axes[1, 1].set_ylabel('數量')

        plt.tight_layout()
        plt.savefig('modbus_security_dashboard.png', dpi=300, bbox_inches='tight')
        plt.show()

# 威脅情報整合

class ThreatIntelligence:
    def __init__(self):
        self.threat_feeds = {}
        self.ioc_database = {
            'malicious_ips': set(),
            'suspicious_patterns': [],
            'known_vulnerabilities': []
        }

    def update_threat_feeds(self):
        """更新威脅情報"""
        # 這裡可以整合外部威脅情報源
        # 例如：CERT, ICS-CERT, 商業威脅情報服務

        # 模擬威脅情報更新
        new_malicious_ips = [
            '192.168.100.50',  # 已知的攻擊 IP
            '10.0.0.99',       # 另一個惡意 IP
        ]

        self.ioc_database['malicious_ips'].update(new_malicious_ips)

        # 更新攻擊模式
        new_patterns = [
            {
                'name': 'modbus_reconnaissance',
                'pattern': 'sequential_function_code_scan',
                'severity': 'medium'
            },
            {
                'name': 'data_exfiltration',
                'pattern': 'bulk_register_read',
                'severity': 'high'
            }
        ]

        self.ioc_database['suspicious_patterns'].extend(new_patterns)

    def check_indicators(self, ip_address, traffic_pattern):
        """檢查威脅指標"""
        threats_found = []

        # 檢查惡意 IP
        if ip_address in self.ioc_database['malicious_ips']:
            threats_found.append({
                'type': 'malicious_ip',
                'severity': 'high',
                'description': f'已知惡意 IP: {ip_address}'
            })

        # 檢查攻擊模式
        for pattern in self.ioc_database['suspicious_patterns']:
            if self.match_pattern(traffic_pattern, pattern['pattern']):
                threats_found.append({
                    'type': 'suspicious_pattern',
                    'severity': pattern['severity'],
                    'description': f'檢測到可疑模式: {pattern["name"]}'
                })

        return threats_found

    def match_pattern(self, traffic_pattern, threat_pattern):
        """模式匹配邏輯"""
        # 簡化的模式匹配，實際應用會更複雜
        if threat_pattern == 'sequential_function_code_scan':
            # 檢查是否有連續的功能碼掃描
            function_codes = traffic_pattern.get('function_codes', [])
            return len(set(function_codes)) > 5

        elif threat_pattern == 'bulk_register_read':
            # 檢查是否有大量暫存器讀取
            read_operations = traffic_pattern.get('read_operations', [])
            total_registers = sum(op.get('quantity', 0) for op in read_operations)
            return total_registers > 1000

        return False

## 0x06 系列總結與實務建議

### 安全檢查清單

```markdown
# Modbus TCP 安全檢查清單

## 網路安全 ✓

- [ ] 實施網路分段隔離
- [ ] 部署工業防火牆
- [ ] 啟用入侵檢測系統
- [ ] 實施 VPN 或專線連接
- [ ] 定期進行網路掃描

## 設備安全 ✓

- [ ] 更改預設密碼
- [ ] 停用不必要的服務
- [ ] 更新韌體版本
- [ ] 實施存取控制
- [ ] 定期安全稽核

## 通訊安全 ✓

- [ ] 使用 TLS 加密
- [ ] 實施訊息認證
- [ ] 啟用日誌記錄
- [ ] 監控異常流量
- [ ] 實施速率限制

## 操作安全 ✓

- [ ] 建立安全政策
- [ ] 進行人員訓練
- [ ] 實施變更管理
- [ ] 定期備份配置
- [ ] 準備事件回應計畫

## 監控與回應 ✓

- [ ] 部署 SIEM 系統
- [ ] 建立監控儀表板
- [ ] 設定警報機制
- [ ] 建立回應程序
- [ ] 定期演練