Python 自動化強化威脅回應與安全防護：台灣企業實戰指南

#!/usr/bin/env python3
# 這個腳本示範如何自動化偵測電子郵件中的網路釣魚連結
# 整合多個威脅情資來源進行綜合評估

import requests
import re
import sys
from urllib.parse import urlparse
from typing import Dict, List, Tuple

# 設定檔：包含各個威脅情資平台的 API 金鑰與端點
# 在實際企業環境中，這些敏感資訊應該透過環境變數或密碼管理服務提供
API_CONFIG = {
    "phishtank": {
        "url": "http://data.phishtank.com/data/{api_key}/online-valid.csv",
        "enabled": False  # PhishTank 需要申請 API 金鑰
    },
    "virustotal": {
        "url": "https://www.virustotal.com/vtapi/v2/url/report",
        "api_key": "YOUR_VT_API_KEY",  # 請替換為實際金鑰
        "enabled": True
    },
    "google_safebrowsing": {
        "url": "https://safebrowsing.googleapis.com/v4/threatMatches:find",
        "api_key": "YOUR_GSB_API_KEY",  # 請替換為實際金鑰
        "enabled": True
    }
}

def extract_domains_from_email(email_content: str) -> List[str]:
    """
    從電子郵件內容中提取所有網域名稱
    使用正規表達式比對 URL 模式，並解析出網域部分
    
    參數:
        email_content: 電子郵件的完整內容（包含 HTML 或純文字）
    
    返回:
        唯一網域名稱的清單
    """
    # 正規表達式模式：比對 http:// 或 https:// 開頭的 URL
    # 這個模式能夠捕捉大多數標準 URL 格式
    url_pattern = r'https?://(?:[a-zA-Z0-9-]+\.)+[a-zA-Z]{2,}(?:/[^"\s]*)?'
    
    # 尋找所有符合模式的字串
    urls = re.findall(url_pattern, email_content, re.IGNORECASE)
    
    # 使用集合確保網域唯一性，避免重複檢查
    domains = set()
    
    for url in urls:
        try:
            # 使用 urlparse 解析 URL，提取網域名稱
            parsed = urlparse(url)
            domain = parsed.netloc.lower()  # 轉換為小寫以標準化
            
            # 移除連接埠號（如果存在）
            if ':' in domain:
                domain = domain.split(':')[0]
            
            domains.add(domain)
        except Exception as e:
            # 記錄解析錯誤，但繼續處理其他 URL
            print(f"警告：無法解析 URL {url}: {e}", file=sys.stderr)
    
    return list(domains)

def check_virustotal(url: str) -> Tuple[bool, str]:
    """
    使用 VirusTotal API 檢查 URL 信譽
    VirusTotal 會掃描 URL 並回報多個防毒引擎的檢測結果
    
    參數:
        url: 要檢查的完整 URL
    
    返回:
        Tuple[是否惡意, 檢測結果描述]
    """
    if not API_CONFIG["virustotal"]["enabled"]:
        return False, "VirusTotal 檢查未啟用"
    
    api_key = API_CONFIG["virustotal"]["api_key"]
    if api_key == "YOUR_VT_API_KEY":
        return False, "VirusTotal API 金鑰未設定"
    
    try:
        # 準備 API 請求參數
        params = {
            "apikey": api_key,
            "resource": url,
            "scan": 0  # 不觸發即時掃描，僅查詢現有資料庫
        }
        
        # 發送 GET 請求至 VirusTotal API
        response = requests.get(
            API_CONFIG["virustotal"]["url"],
            params=params,
            timeout=10
        )
        
        # 檢查 HTTP 狀態碼
        if response.status_code != 200:
            return False, f"API 請求失敗，狀態碼：{response.status_code}"
        
        result = response.json()
        
        # 解析回應結果
        if result.get("response_code") == 1:
            # 計算檢測為惡意的引擎數量
            positives = result.get("positives", 0)
            total = result.get("total", 0)
            
            if positives > 0:
                return True, f"檢測為惡意（{positives}/{total} 引擎）"
            else:
                return False, f"未檢測為惡意（0/{total} 引擎）"
        elif result.get("response_code") == 0:
            return False, "URL 不在 VirusTotal 資料庫中"
        else:
            return False, f"無法判斷，回應碼：{result.get('response_code')}"
    
    except requests.exceptions.Timeout:
        return False, "請求超時"
    except requests.exceptions.ConnectionError:
        return False, "網路連線錯誤"
    except Exception as e:
        return False, f"檢查過程發生異常：{str(e)}"

def analyze_email_risk(email_content: str) -> Dict:
    """
    綜合分析電子郵件的網路釣魚風險
    整合多個威脅情資來源的結果
    
    參數:
        email_content: 電子郵件完整內容
    
    返回:
        包含風險評估結果的字典
    """
    # 提取郵件中的所有網域名稱
    domains = extract_domains_from_email(email_content)
    
    if not domains:
        return {
            "risk_level": "LOW",
            "message": "未發現可疑連結",
            "details": []
        }
    
    # 針對每個網域進行檢查
    malicious_domains = []
    safe_domains = []
    unclear_domains = []
    
    for domain in domains:
        # 建構完整 URL 進行檢查（VirusTotal 需要完整 URL）
        test_url = f"http://{domain}"
        
        is_malicious, description = check_virustotal(test_url)
        
        if is_malicious:
            malicious_domains.append({
                "domain": domain,
                "reason": description
            })
        elif "未檢測為惡意" in description:
            safe_domains.append({
                "domain": domain,
                "reason": description
            })
        else:
            unclear_domains.append({
                "domain": domain,
                "reason": description
            })
    
    # 根據檢查結果判定整體風險等級
    if malicious_domains:
        risk_level = "HIGH"
        message = f"發現 {len(malicious_domains)} 個惡意連結"
    elif unclear_domains:
        risk_level = "MEDIUM"
        message = f"發現 {len(unclear_domains)} 個無法確認的連結"
    else:
        risk_level = "LOW"
        message = "所有連結皆未檢測為惡意"
    
    return {
        "risk_level": risk_level,
        "message": message,
        "total_domains_checked": len(domains),
        "malicious_domains": malicious_domains,
        "safe_domains": safe_domains,
        "unclear_domains": unclear_domains
    }

def main():
    """
    主函式：示範如何使用網路釣魚偵測功能
    """
    # 模擬一封包含可疑連結的電子郵件
    # 實際應用中，這可以從郵件伺服器或郵件閘道器取得
    sample_email = """
    親愛的用戶您好，
    
    您的帳號即將到期，請立即登入以下連結更新您的密碼：
    http://www.microsoft-security-update.com/login
    或
    https://login.microsoftonline.com.phishing.example.com
    
    若未在24小時內完成更新，您的帳號將被暫停使用。
    
    敬祝 順心
    IT 支援團隊
    """
    
    print("開始分析電子郵件中的網路釣魚風險...")
    print("=" * 60)
    
    # 執行風險分析
    result = analyze_email_risk(sample_email)
    
    # 顯示分析結果
    print(f"風險等級：{result['risk_level']}")
    print(f"分析結果：{result['message']}")
    print(f"檢查的網域總數：{result['total_domains_checked']}")
    print()
    
    # 顯示詳細資訊
    if result['malicious_domains']:
        print("惡意連結清單：")
        for item in result['malicious_domains']:
            print(f"  - {item['domain']}: {item['reason']}")
        print()
    
    if result['unclear_domains']:
        print("無法確認的連結清單：")
        for item in result['unclear_domains']:
            print(f"  - {item['domain']}: {item['reason']}")
        print()
    
    if result['safe_domains']:
        print("安全連結清單：")
        for item in result['safe_domains']:
            print(f"  - {item['domain']}: {item['reason']}")
    
    # 根據風險等級建議後續行動
    print()
    print("建議行動：")
    if result['risk_level'] == "HIGH":
        print("  立即隔離此郵件，並通知所有收件人")
        print("  檢查是否有使用者已點擊連結")
        print("  考慮封鎖相關網域")
    elif result['risk_level'] == "MEDIUM":
        print("  將郵件移至隔離區，等待進一步調查")
        print("  手動檢查無法確認的連結")
    else:
        print("  郵件風險較低，可正常處理")
        print("  持續監控未來變化")

if __name__ == "__main__":
    # 執行主函式
    main()

#!/usr/bin/env python3
# 這個腳本示範如何自動化偵測與回應勒索軟體攻擊
# 整合端點監控、自動隔離與事件通知功能

import requests
import json
import time
from datetime import datetime
from typing import Dict, List, Any

class RansomwareResponseSystem:
    """
    勒索軟體自動回應系統類別
    負責監控端點活動、偵測可疑行為並執行自動化回應
    """
    
    def __init__(self, edr_api_url: str, edr_api_key: str, firewall_api_url: str, slack_webhook: str):
        """
        初始化回應系統，設定各項 API 連線資訊
        
        參數:
            edr_api_url: EDR 系統的 API 端點 URL
            edr_api_key: EDR API 的認證金鑰
            firewall_api_url: 防火牆 API 的端點 URL
            slack_webhook: Slack 通知的 Webhook URL
        """
        self.edr_api_url = edr_api_url
        self.edr_api_key = edr_api_key
        self.firewall_api_url = firewall_api_url
        self.slack_webhook = slack_webhook
        self.suspicious_processes = [
            "encrypt.exe", "lockbit.exe", "wannacry.exe",
            "ransomware.exe", "crypto.exe", "encryptor.exe"
        ]
        self.encryption_extensions = [
            ".locked", ".encrypted", ".wannacry", ".lockbit",
            ".crypto", ".ransom", ".enc"
        ]
    
    def get_endpoint_events(self, endpoint_id: str, time_window: int = 300) -> List[Dict[str, Any]]:
        """
        從 EDR 系統取得指定端點在特定時間範圍內的事件記錄
        
        參數:
            endpoint_id: 端點設備的唯一識別碼
            time_window: 回顧的時間範圍（秒），預設 300 秒（5 分鐘）
        
        返回:
            事件記錄的清單，每個事件為字典格式
        """
        headers = {
            "Authorization": f"Bearer {self.edr_api_key}",
            "Content-Type": "application/json"
        }
        
        # 計算時間範圍
        end_time = datetime.utcnow()
        start_time = datetime.utcfromtimestamp(end_time.timestamp() - time_window)
        
        # 準備 API 請求參數
        params = {
            "endpoint_id": endpoint_id,
            "start_time": start_time.isoformat() + "Z",
            "end_time": end_time.isoformat() + "Z",
            "event_types": "file_modification,process_creation,file_deletion"
        }
        
        try:
            response = requests.get(
                f"{self.edr_api_url}/api/v1/events",
                headers=headers,
                params=params,
                timeout=30
            )
            response.raise_for_status()
            return response.json().get("events", [])
        except requests.exceptions.RequestException as e:
            print(f"錯誤：無法從 EDR 取得事件資料 - {e}")
            return []
    
    def detect_ransomware_behavior(self, events: List[Dict[str, Any]]) -> Dict[str, Any]:
        """
        分析端點事件，偵測是否符合勒索軟體行為模式
        
        參數:
            events: 從 EDR 取得的事件清單
        
        返回:
            包含偵測結果的字典
        """
        detection_result = {
            "is_ransomware": False,
            "confidence": 0,
            "indicators": [],
            "affected_files": [],
            "malicious_processes": []
        }
        
        # 統計指標
        file_modifications = 0
        file_deletions = 0
        suspicious_processes = []
        encrypted_files = []
        
        for event in events:
            event_type = event.get("event_type")
            
            if event_type == "file_modification":
                file_modifications += 1
                file_path = event.get("file_path", "")
                
                # 檢查檔案副檔名是否為勒索軟體常見的加密副檔名
                for ext in self.encryption_extensions:
                    if file_path.endswith(ext):
                        encrypted_files.append(file_path)
                        detection_result["indicators"].append(f"發現加密檔案：{file_path}")
            
            elif event_type == "file_deletion":
                file_deletions += 1
            
            elif event_type == "process_creation":
                process_name = event.get("process_name", "").lower()
                if any(proc in process_name for proc in self.suspicious_processes):
                    suspicious_processes.append(process_name)
                    detection_result["indicators"].append(f"發現可疑程序：{process_name}")
        
        # 綜合評估風險
        # 勒索軟體通常會在短時間內大量修改檔案
        if file_modifications > 100:
            detection_result["confidence"] += 30
            detection_result["indicators"].append(f"短時間內大量檔案修改：{file_modifications} 個")
        
        # 大量刪除檔案也是勒索軟體的特徵
        if file_deletions > 50:
            detection_result["confidence"] += 20
            detection_result["indicators"].append(f"異常檔案刪除數量：{file_deletions} 個")
        
        # 發現加密副檔名的檔案
        if encrypted_files:
            detection_result["confidence"] += 40
            detection_result["affected_files"] = encrypted_files
        
        # 發現已知勒索軟體程序
        if suspicious_processes:
            detection_result["confidence"] += 30
            detection_result["malicious_processes"] = suspicious_processes
        
        # 當信心分數超過 50 分，判定為勒索軟體攻擊
        if detection_result["confidence"] >= 50:
            detection_result["is_ransomware"] = True
        
        return detection_result
    
    def isolate_endpoint(self, endpoint_id: str) -> bool:
        """
        隔離受感染的端點，防止威脅擴散
        
        參數:
            endpoint_id: 要隔離的端點設備識別碼
        
        返回:
            隔離是否成功
        """
        headers = {
            "Authorization": f"Bearer {self.edr_api_key}",
            "Content-Type": "application/json"
        }
        
        payload = {
            "action": "isolate",
            "endpoint_id": endpoint_id,
            "reason": "偵測到勒索軟體活動，自動隔離以防止擴散"
        }
        
        try:
            response = requests.post(
                f"{self.edr_api_url}/api/v1/endpoints/isolate",
                headers=headers,
                json=payload,
                timeout=30
            )
            
            if response.status_code == 200:
                print(f"端點 {endpoint_id} 已成功隔離")
                return True
            else:
                print(f"隔離端點 {endpoint_id} 失敗：{response.status_code}")
                return False
        
        except requests.exceptions.RequestException as e:
            print(f"錯誤：無法隔離端點 - {e}")
            return False
    
    def block_malicious_ip(self, ip_address: str) -> bool:
        """
        封鎖惡意 IP 位址，阻止與 C2 伺服器通訊
        
        參數:
            ip_address: 要封鎖的 IP 位址
        
        返回:
            封鎖是否成功
        """
        # 這是防火牆 API 的範例，實際實作需根據使用的防火牆品牌調整
        payload = {
            "action": "block",
            "ip_address": ip_address,
            "rule_name": f"Ransomware_Block_{int(time.time())}",
            "description": "自動封鎖勒索軟體 C2 伺服器"
        }
        
        try:
            response = requests.post(
                f"{self.firewall_api_url}/api/v1/firewall/rules",
                json=payload,
                timeout=30
            )
            
            if response.status_code in [200, 201]:
                print(f"已封鎖惡意 IP：{ip_address}")
                return True
            else:
                print(f"封鎖 IP 失敗：{response.status_code}")
                return False
        
        except requests.exceptions.RequestException as e:
            print(f"錯誤：無法封鎖 IP - {e}")
            return False
    
    def send_alert_notification(self, detection_result: Dict[str, Any], endpoint_id: str) -> None:
        """
        發送警報通知至企業通訊平台
        確保資安團隊能即時掌握事件
        
        參數:
            detection_result: 勒索軟體偵測結果
            endpoint_id: 受影響的端點識別碼
        """
        if not self.slack_webhook:
            print("警告：Slack Webhook 未設定，無法發送通知")
            return
        
        # 建構警報訊息
        message = {
            "attachments": [
                {
                    "color": "danger",
                    "title": "🚨 勒索軟體攻擊警報",
                    "fields": [
                        {
                            "title": "受影響端點",
                            "value": endpoint_id,
                            "short": True
                        },
                        {
                            "title": "偵測時間",
                            "value": datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S UTC"),
                            "short": True
                        },
                        {
                            "title": "風險信心度",
                            "value": f"{detection_result['confidence']}%",
                            "short": True
                        },
                        {
                            "title": "偵測指標數量",
                            "value": len(detection_result["indicators"]),
                            "short": True
                        },
                        {
                            "title": "受影響檔案數量",
                            "value": len(detection_result["affected_files"]),
                            "short": True
                        }
                    ],
                    "text": "已自動執行隔離與封鎖措施，請立即進行調查",
                    "footer": "自動化威脅回應系統"
                }
            ]
        }
        
        try:
            response = requests.post(
                self.slack_webhook,
                json=message,
                timeout=10
            )
            
            if response.status_code == 200:
                print("警報通知已發送至 Slack")
            else:
                print(f"發送通知失敗：{response.status_code}")
        
        except requests.exceptions.RequestException as e:
            print(f"錯誤：無法發送通知 - {e}")
    
    def automated_response(self, endpoint_id: str) -> bool:
        """
        完整的自動化回應流程
        從偵測到回應的完整處理
        
        參數:
            endpoint_id: 要監控的端點識別碼
        
        返回:
            回應是否成功執行
        """
        print(f"開始監控端點 {endpoint_id}...")
        
        # 步驟 1：取得端點事件
        events = self.get_endpoint_events(endpoint_id)
        if not events:
            print(f"無法取得端點 {endpoint_id} 的事件資料")
            return False
        
        # 步驟 2：偵測勒索軟體行為
        detection = self.detect_ransomware_behavior(events)
        
        if not detection["is_ransomware"]:
            print(f"端點 {endpoint_id} 未偵測到勒索軟體活動")
            return True
        
        # 步驟 3：執行自動化回應
        print(f"偵測到勒索軟體活動！信心度：{detection['confidence']}%")
        print("立即執行自動化回應措施...")
        
        # 隔離端點
        isolation_success = self.isolate_endpoint(endpoint_id)
        
        # 假設從事件中提取了惡意 IP，實際實作需解析相關欄位
        # 這裡示範封鎖一個假設的 C2 IP
        c2_ip = "192.168.1.100"  # 實際應從事件資料中提取
        block_success = self.block_malicious_ip(c2_ip)
        
        # 發送警報
        self.send_alert_notification(detection, endpoint_id)
        
        if isolation_success and block_success:
            print("自動化回應成功執行")
            return True
        else:
            print("自動化回應部分失敗，需要手動介入")
            return False

# 使用範例
if __name__ == "__main__":
    # 初始化回應系統
    # 實際使用時，請替換為真實的 API 端點與金鑰
    response_system = RansomwareResponseSystem(
        edr_api_url="https://edr.enterprise.example.com",
        edr_api_key="your_edr_api_key_here",
        firewall_api_url="https://firewall.enterprise.example.com",
        slack_webhook="https://hooks.slack.com/services/YOUR/WEBHOOK/URL"
    )
    
    # 監控特定端點
    endpoint_to_monitor = "WIN-PC-12345"
    response_system.automated_response(endpoint_to_monitor)