當代網路環境中的安全威脅持續演進,攻擊手法日益精密複雜。資安工程師必須掌握多層次的防禦技術,才能有效保護組織的數位資產。本文將深入探討網路安全領域的核心技術,包括網路探測技術、搜尋引擎安全評估工具開發、伺服器訊息區塊協定掃描,以及即時登入監控系統的完整實作方案。這些技術構成了現代網路安全防禦的重要基石,能夠協助安全團隊及早發現潛在威脅,並建立有效的防禦機制。

透過實務導向的Python程式開發,我們將展示如何建構實用的安全工具。同時,文章也會探討進階的網路封包分析技術,包括DNS協定處理、DHCPv6協定應用,以及自訂網路協定的設計與實作。這些技術的綜合運用,將幫助資安專業人員建立更完善的安全監控體系,提升組織整體的資訊安全防護能力。

網路位址探測技術實作

在企業網路環境中,某些特殊場景下的裝置可能無法透過動態主機組態協定自動取得網路位址。這種情況常見於隔離網段、特殊安全區域,或是進行網路滲透測試時。此時,網路管理人員需要透過探測技術來識別可用的網路位址範圍。本節將介紹如何使用Python開發一套智慧型的網路位址探測工具。

探測機制設計理念

網路位址探測的核心概念是透過系統化的方式,測試特定網路範圍內的各個位址是否可達。這個過程涉及網路介面的動態組態、網際網路控制訊息協定的測試封包傳送,以及回應結果的智慧分析。為了提高探測效率,我們採用隨機化策略,避免按順序掃描可能觸發的安全警報。同時,工具會自動測試常見的閘道器位址,這些位址通常位於網路範圍的起始或結束位置。

程式實作細節

#!/usr/bin/python3

import os
import re
import sys
from random import choice
import ipaddress

# 定義網路介面裝置名稱,根據實際環境調整
device = "wlan0"

# 定義要探測的IP網路範圍,使用CIDR標記法
ip_network = "192.168.0.0/24"

def ping_ip(ip):
    """
    嘗試對指定的IP位址發送ICMP echo請求
    
    參數:
        ip: 目標IP位址字串
    
    回傳:
        布林值,True表示收到回應,False表示無回應
    """
    try:
        # 使用系統ping指令進行測試
        # -c1: 只發送一個封包
        # -W1: 等待回應的時間限制為1秒
        fh = os.popen(f"ping -c1 -W1 {ip}")
        resp = fh.read()
        
        # 檢查回應內容中是否包含成功接收的標記
        if re.search("bytes from", resp, re.MULTILINE):
            print(f"[+] 收到來自 {ip} 的回應")
            return True
        return False
    except Exception as e:
        print(f"[-] ping {ip} 時發生錯誤: {e}")
        return False

def configure_interface(ip):
    """
    動態組態網路介面的IP位址
    
    參數:
        ip: 要組態的IP位址
    
    回傳:
        布林值,True表示組態成功,False表示組態失敗
    """
    try:
        # 清除介面現有的IP位址組態
        # 新增新的IP位址到介面
        # 啟動網路介面
        cmd = f"ip addr flush dev {device} && ip addr add {ip}/24 dev {device} && ip link set {device} up"
        result = os.system(cmd)
        
        if result == 0:
            print(f"[*] 成功組態 {device}{ip}/24")
            return True
        return False
    except Exception as e:
        print(f"[-] 組態 {device}{ip} 時發生錯誤: {e}")
        return False

def main():
    """
    主程式邏輯:執行網路位址探測流程
    """
    try:
        # 將網路範圍字串轉換為可操作的網路物件
        network = ipaddress.ip_network(ip_network, strict=False)
        
        # 生成網路範圍內所有可用的主機位址
        ips = list(network.hosts())
        
        print(f"[*] 開始探測網路範圍: {ip_network}")
        print(f"[*] 可用位址數量: {len(ips)}")
        
        # 當還有未測試的位址時,持續執行探測
        while ips:
            # 從剩餘位址中隨機選擇一個進行測試
            ip = choice(ips)
            ips.remove(ip)
            
            print(f"\n[*] 測試位址: {ip}")
            
            # 組態網路介面並執行連通性測試
            if configure_interface(str(ip)):
                # 測試當前組態的位址
                ping_ip(str(ip))
                
                # 測試常見的預設閘道器位址
                # 通常閘道器位於網路位址+1的位置
                gateway_ip = str(network.network_address + 1)
                print(f"[*] 測試預設閘道器位址: {gateway_ip}")
                if ping_ip(gateway_ip):
                    print(f"[+] 發現可用的網路組態!")
                    print(f"    主機位址: {ip}")
                    print(f"    閘道器位址: {gateway_ip}")
                
                # 某些網路環境的閘道器可能位於廣播位址-1的位置
                alt_gateway = str(network.broadcast_address - 1)
                print(f"[*] 測試備用閘道器位址: {alt_gateway}")
                ping_ip(alt_gateway)
                
    except KeyboardInterrupt:
        print("\n[!] 使用者中斷探測程序")
        sys.exit(0)
    except Exception as e:
        print(f"[-] 探測過程中發生錯誤: {e}")
        sys.exit(1)

if __name__ == "__main__":
    # 檢查是否具有root權限
    if os.geteuid() != 0:
        print("[-] 此程式需要root權限執行")
        sys.exit(1)
    
    main()

探測流程架構

@startuml
!define PLANTUML_FORMAT svg
!theme _none_

skinparam dpi auto
skinparam shadowing false
skinparam linetype ortho
skinparam roundcorner 5
skinparam defaultFontName "Microsoft JhengHei UI"
skinparam defaultFontSize 16
skinparam minClassWidth 100

start
:初始化探測程序;
:解析目標網路範圍;
:生成可用位址清單;

while (是否還有未測試位址?) is (是)
  :隨機選擇一個位址;
  :動態組態網路介面;
  
  if (組態是否成功?) then (成功)
    :測試主機位址連通性;
    :測試預設閘道器位址;
    :測試備用閘道器位址;
    
    if (發現可用組態?) then (是)
      :記錄有效的網路組態;
      :輸出詳細資訊;
    endif
  else (失敗)
    :記錄組態失敗;
  endif
  
  :從清單中移除已測試位址;
endwhile (否)

:完成探測程序;
stop

@enduml

這個探測工具的設計考量了實際網路環境的複雜性。透過隨機化的測試策略,工具能夠避免觸發入侵偵測系統的警報。程式會自動測試多個可能的閘道器位址,因為不同的網路環境可能採用不同的閘道器組態慣例。同時,完整的錯誤處理機制確保了程式在各種異常情況下都能妥善處理,提供清晰的執行狀態回饋。

搜尋引擎安全評估工具開發

搜尋引擎不僅是資訊檢索的工具,同時也是資安研究人員進行安全評估的重要資源。透過特定的搜尋語法,安全研究人員可以發現網路上公開的敏感資訊、組態錯誤的伺服器,以及潛在的安全漏洞。這種技術在資安領域被稱為搜尋引擎安全評估,是滲透測試和安全稽核的重要環節。

工具設計概念

搜尋引擎安全評估工具的核心功能是自動化執行特定的搜尋查詢,並對結果進行智慧過濾。工具需要支援彈性的查詢語法,能夠處理各種複雜的搜尋條件。同時,為了避免被搜尋引擎識別為自動化程式而遭到封鎖,工具必須實作適當的請求間隔控制和使用者代理輪換機制。結果過濾功能則能幫助使用者快速定位真正有價值的資訊,排除無關的干擾項目。

完整程式實作

#!/usr/bin/python3

import sys
import time
import re
from googlesearch import search
from urllib.parse import urlparse

def filter_results(url, blocked_patterns):
    """
    過濾不需要的搜尋結果
    
    參數:
        url: 搜尋結果的網址
        blocked_patterns: 要排除的網域模式清單
    
    回傳:
        布林值,True表示應該保留,False表示應該過濾
    """
    try:
        # 解析網址取得網域名稱
        parsed_url = urlparse(url)
        domain = parsed_url.netloc.lower()
        
        # 檢查是否符合任何排除模式
        for pattern in blocked_patterns:
            if re.search(pattern, domain, re.IGNORECASE):
                return False
        return True
    except Exception as e:
        print(f"[-] 過濾網址時發生錯誤: {e}")
        return True

def google_hacking(query, num_results=100, blocked_domains=None):
    """
    執行搜尋引擎安全評估查詢
    
    參數:
        query: 搜尋查詢字串
        num_results: 要取得的結果數量
        blocked_domains: 要排除的網域清單
    """
    if blocked_domains is None:
        # 預設排除的網域模式
        blocked_domains = [
            r"youtube\.com",
            r"reddit\.com",
            r"facebook\.com",
            r"twitter\.com",
            r"instagram\.com"
        ]
    
    try:
        print(f"[*] 執行搜尋查詢: {query}")
        print(f"[*] 目標結果數量: {num_results}")
        print(f"[*] 排除網域: {', '.join(blocked_domains)}")
        print("\n" + "=" * 80 + "\n")
        
        # 執行搜尋並取得結果
        results = search(query, num_results=num_results, sleep_interval=2)
        
        valid_count = 0
        for result in results:
            # 套用過濾規則
            if filter_results(result, blocked_domains):
                print(f"[+] {result}")
                valid_count += 1
                
                # 加入延遲避免請求過於頻繁
                time.sleep(0.5)
        
        print("\n" + "=" * 80)
        print(f"[*] 搜尋完成,找到 {valid_count} 個有效結果")
        
    except Exception as e:
        print(f"[-] 執行搜尋時發生錯誤: {e}")
        sys.exit(1)

def main():
    """
    主程式邏輯:處理命令列參數並執行搜尋
    """
    if len(sys.argv) < 2:
        print("使用方式:")
        print(f"  {sys.argv[0]} <搜尋查詢> [結果數量]")
        print("\n範例:")
        print(f"  {sys.argv[0]} 'intitle:\"index of\" password'")
        print(f"  {sys.argv[0]} 'filetype:pdf confidential' 50")
        print("\n常用查詢語法:")
        print("  intitle:     搜尋標題包含特定文字的頁面")
        print("  inurl:       搜尋網址包含特定文字的頁面")
        print("  filetype:    搜尋特定檔案型別")
        print("  site:        限制搜尋特定網站")
        print("  intext:      搜尋內容包含特定文字")
        sys.exit(1)

    # 取得搜尋查詢
    query = sys.argv[1]
    
    # 取得結果數量(如果有指定)
    num_results = 100
    if len(sys.argv) > 2:
        try:
            num_results = int(sys.argv[2])
        except ValueError:
            print("[-] 結果數量必須是整數")
            sys.exit(1)
    
    # 執行搜尋
    google_hacking(query, num_results)

if __name__ == "__main__":
    main()

工具執行流程

@startuml
!define PLANTUML_FORMAT svg
!theme _none_

skinparam dpi auto
skinparam shadowing false
skinparam linetype ortho
skinparam roundcorner 5
skinparam defaultFontName "Microsoft JhengHei UI"
skinparam defaultFontSize 16
skinparam minClassWidth 100

start
:接收搜尋查詢參數;
:驗證參數有效性;

if (參數是否有效?) then (是)
  :初始化排除網域清單;
  :組態搜尋參數;
  :執行搜尋引擎查詢;
  
  partition "結果處理迴圈" {
    while (是否還有結果?) is (是)
      :取得下一個搜尋結果;
      :解析結果網址;
      
      if (是否符合過濾條件?) then (通過)
        :輸出有效結果;
        :增加有效結果計數;
      else (過濾)
        :跳過此結果;
      endif
      
      :執行請求間隔延遲;
    endwhile (否)
  }
  
  :統計並輸出總結資訊;
else (否)
  :顯示使用說明;
  :列出查詢語法範例;
endif

stop

@enduml

這個工具的設計充分考慮了實務應用需求。透過可組態的排除清單,使用者可以快速過濾掉社群媒體等低價值的結果。程式內建的延遲機制能夠避免觸發搜尋引擎的反爬蟲機制。完整的錯誤處理和使用說明則確保了工具的可用性,即使是初學者也能快速上手使用。

伺服器訊息區塊協定掃描技術

伺服器訊息區塊協定是Windows網路環境中的核心通訊協定,負責處理檔案共享、印表機共享、使用者認證等重要功能。在企業網路中,不當組態的SMB服務可能成為攻擊者的入侵管道。因此,定期掃描網路中的SMB服務,評估其安全性組態,是網路安全維護的重要工作。

掃描技術原理

SMB協定掃描的核心是透過網路通訊埠的連通性測試,識別開放SMB服務的主機。現代SMB服務主要使用TCP 445埠,這是直接透過TCP通訊的SMB協定。透過對這個通訊埠的掃描,我們可以快速識別網路中啟用SMB服務的裝置。進一步地,我們還可以透過服務版本偵測功能,識別SMB服務的具體版本和組態資訊,這些資訊對於安全評估至關重要。

掃描工具實作

#!/usr/bin/python3

import nmap
import sys
from datetime import datetime

class SMBScanner:
    """
    SMB協定掃描器類別
    封裝了SMB服務掃描的完整功能
    """
    
    def __init__(self, target_range):
        """
        初始化掃描器
        
        參數:
            target_range: 目標IP範圍(支援CIDR標記法)
        """
        self.target_range = target_range
        self.nm = nmap.PortScanner()
        self.results = []
    
    def scan_host(self, ip):
        """
        掃描單一主機的SMB服務
        
        參數:
            ip: 目標主機IP位址
        
        回傳:
            字典,包含掃描結果資訊
        """
        try:
            print(f"[*] 正在掃描主機: {ip}")
            
            # 執行SMB通訊埠掃描,包含服務版本偵測
            # -sV: 啟用服務版本偵測
            # --script smb-os-discovery: 執行作業系統偵測腳本
            self.nm.scan(
                ip, 
                '445', 
                arguments='-sV --script smb-os-discovery'
            )
            
            # 檢查主機是否回應
            if ip not in self.nm.all_hosts():
                return None
            
            host_info = {
                'ip': ip,
                'hostname': self.nm[ip].hostname(),
                'state': self.nm[ip].state(),
                'protocols': {}
            }
            
            # 處理各個協定的掃描結果
            for proto in self.nm[ip].all_protocols():
                print(f"  [+] 協定: {proto}")
                host_info['protocols'][proto] = {}
                
                ports = self.nm[ip][proto].keys()
                for port in ports:
                    port_info = self.nm[ip][proto][port]
                    
                    # 只記錄開放的通訊埠
                    if port_info['state'] == 'open':
                        print(f"    [+] 埠號: {port}")
                        print(f"        狀態: {port_info['state']}")
                        print(f"        服務: {port_info['name']}")
                        
                        # 如果有版本資訊,一併輸出
                        if 'version' in port_info:
                            print(f"        版本: {port_info['version']}")
                        
                        # 如果有額外資訊,一併輸出
                        if 'extrainfo' in port_info:
                            print(f"        額外資訊: {port_info['extrainfo']}")
                        
                        host_info['protocols'][proto][port] = port_info
                        
                        # 檢查腳本執行結果
                        if 'script' in port_info:
                            print(f"        腳本結果:")
                            for script_name, script_output in port_info['script'].items():
                                print(f"          {script_name}:")
                                print(f"            {script_output}")
            
            return host_info
            
        except Exception as e:
            print(f"  [-] 掃描 {ip} 時發生錯誤: {e}")
            return None
    
    def run_scan(self):
        """
        執行完整的網路範圍掃描
        """
        try:
            print(f"\n[*] 開始掃描目標範圍: {self.target_range}")
            print(f"[*] 掃描時間: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
            print("=" * 80)
            
            # 先執行快速主機探索
            print(f"\n[*] 執行主機探索...")
            self.nm.scan(hosts=self.target_range, arguments='-sn')
            
            # 取得所有活躍的主機
            live_hosts = self.nm.all_hosts()
            print(f"[*] 發現 {len(live_hosts)} 個活躍主機")
            
            # 對每個活躍主機執行SMB掃描
            for host in live_hosts:
                result = self.scan_host(host)
                if result:
                    self.results.append(result)
                print()
            
            # 輸出掃描摘要
            self.print_summary()
            
        except KeyboardInterrupt:
            print("\n[!] 使用者中斷掃描")
        except Exception as e:
            print(f"[-] 掃描過程發生錯誤: {e}")
    
    def print_summary(self):
        """
        輸出掃描結果摘要
        """
        print("\n" + "=" * 80)
        print("掃描結果摘要")
        print("=" * 80)
        
        if not self.results:
            print("[!] 未發現開放SMB服務的主機")
            return
        
        print(f"\n[+] 發現 {len(self.results)} 個主機開放SMB服務:\n")
        
        for result in self.results:
            print(f"主機: {result['ip']}")
            if result['hostname']:
                print(f"  主機名稱: {result['hostname']}")
            print(f"  狀態: {result['state']}")
            
            for proto, ports in result['protocols'].items():
                for port, info in ports.items():
                    print(f"  服務: {info['name']} ({port}/{proto})")
                    if 'version' in info:
                        print(f"    版本: {info['version']}")
            print()

def main():
    """
    主程式邏輯
    """
    if len(sys.argv) < 2:
        print("使用方式:")
        print(f"  {sys.argv[0]} <目標IP範圍>")
        print("\n範例:")
        print(f"  {sys.argv[0]} 192.168.1.0/24")
        print(f"  {sys.argv[0]} 10.0.0.1-10.0.0.50")
        print(f"  {sys.argv[0]} 192.168.1.100")
        sys.exit(1)
    
    # 檢查是否具有root權限
    import os
    if os.geteuid() != 0:
        print("[-] 此程式需要root權限執行")
        sys.exit(1)
    
    target_range = sys.argv[1]
    
    # 建立掃描器實例並執行掃描
    scanner = SMBScanner(target_range)
    scanner.run_scan()

if __name__ == "__main__":
    main()

掃描執行流程

@startuml
!define PLANTUML_FORMAT svg
!theme _none_

skinparam dpi auto
skinparam shadowing false
skinparam linetype ortho
skinparam roundcorner 5
skinparam defaultFontName "Microsoft JhengHei UI"
skinparam defaultFontSize 16
skinparam minClassWidth 100

start
:初始化掃描器;
:解析目標範圍;

:執行主機探索掃描;
:取得活躍主機清單;

partition "主機掃描迴圈" {
  while (是否還有待掃描主機?) is (是)
    :選取下一個主機;
    :掃描SMB服務通訊埠;
    
    if (通訊埠是否開放?) then (開放)
      :執行服務版本偵測;
      :執行作業系統探測腳本;
      :收集服務資訊;
      :記錄掃描結果;
      
      if (是否有腳本輸出?) then (有)
        :解析腳本執行結果;
        :記錄額外資訊;
      endif
    else (關閉)
      :標記服務未開啟;
    endif
  endwhile (否)
}

:彙整掃描結果;
:生成摘要報告;
:輸出詳細資訊;

stop

@enduml

這個SMB掃描工具採用物件導向的設計方式,提供了清晰的程式結構和良好的擴充性。工具首先執行快速的主機探索,識別網路中的活躍裝置,然後針對這些裝置進行詳細的SMB服務掃描。透過整合Nmap的腳本引擎,工具能夠取得更豐富的服務資訊,包括作業系統類型、SMB版本等重要安全資訊。完整的結果摘要功能則幫助管理員快速掌握網路中的SMB服務分佈情況。

即時登入監控系統建置

企業網路環境中,使用者登入行為的監控是安全防護的重要環節。透過即時監控系統登入事件,安全團隊能夠及時發現異常的登入嘗試,並在潛在的入侵行為造成實質危害前採取防禦措施。本節將介紹如何建構一個功能完整的登入監控系統,該系統能夠追蹤登入嘗試、識別可疑行為,並在必要時觸發防禦機制。

系統架構設計

登入監控系統的核心是即時的日誌分析引擎。系統持續監控認證日誌檔案,透過正規表示式模式比對識別各種登入事件。針對每個來源位址,系統維護一個登入失敗計數器,當失敗次數超過設定的閾值時,系統會觸發預警機制。為了提供更直覺的安全警示,系統整合了語音通知功能,能夠即時向管理員報告重要的安全事件。

監控系統程式實作

#!/usr/bin/python3

import pygtail
import re
import subprocess
import pyttsx3
from datetime import datetime
import json
import os

class LoginMonitor:
    """
    登入監控系統類別
    提供完整的登入事件監控與處理功能
    """
    
    def __init__(self, config_file="monitor_config.json"):
        """
        初始化監控系統
        
        參數:
            config_file: 組態檔案路徑
        """
        # 載入組態
        self.load_config(config_file)
        
        # 初始化語音引擎
        self.engine = pyttsx3.init()
        self.configure_voice()
        
        # 初始化統計資料
        self.failed_logins = {}
        self.successful_logins = {}
        self.blocked_ips = set()
        
        # 編譯正規表示式模式
        self.compile_patterns()
    
    def load_config(self, config_file):
        """
        載入監控系統組態
        
        參數:
            config_file: 組態檔案路徑
        """
        try:
            if os.path.exists(config_file):
                with open(config_file, 'r', encoding='utf-8') as f:
                    config = json.load(f)
            else:
                # 使用預設組態
                config = {
                    "logfile": "/var/log/auth.log",
                    "max_failed_attempts": 3,
                    "shutdown_enabled": False,
                    "alert_enabled": True,
                    "voice_enabled": True,
                    "block_ip_enabled": True
                }
            
            self.logfile = config.get("logfile", "/var/log/auth.log")
            self.max_failed = config.get("max_failed_attempts", 3)
            self.shutdown_enabled = config.get("shutdown_enabled", False)
            self.alert_enabled = config.get("alert_enabled", True)
            self.voice_enabled = config.get("voice_enabled", True)
            self.block_ip_enabled = config.get("block_ip_enabled", True)
            
        except Exception as e:
            print(f"[-] 載入組態檔案失敗: {e}")
            print("[*] 使用預設組態")
            self.logfile = "/var/log/auth.log"
            self.max_failed = 3
            self.shutdown_enabled = False
            self.alert_enabled = True
            self.voice_enabled = True
            self.block_ip_enabled = True
    
    def configure_voice(self):
        """
        組態語音引擎參數
        """
        try:
            # 設定語音速率
            self.engine.setProperty('rate', 150)
            # 設定音量
            self.engine.setProperty('volume', 0.9)
        except Exception as e:
            print(f"[-] 組態語音引擎失敗: {e}")
    
    def compile_patterns(self):
        """
        編譯正規表示式模式
        """
        # 密碼錯誤模式(SSH登入失敗)
        self.failed_password_pattern = re.compile(
            r"Failed password for (?:invalid user )?(\w+) from (\d+\.\d+\.\d+\.\d+)"
        )
        
        # 登入成功模式
        self.success_pattern = re.compile(
            r"Accepted (?:password|publickey) for (\w+) from (\d+\.\d+\.\d+\.\d+)"
        )
        
        # 無效使用者模式
        self.invalid_user_pattern = re.compile(
            r"Invalid user (\w+) from (\d+\.\d+\.\d+\.\d+)"
        )
        
        # 連線關閉模式
        self.disconnect_pattern = re.compile(
            r"Disconnected from (?:invalid user )?(?:\w+ )?(\d+\.\d+\.\d+\.\d+)"
        )
    
    def speak(self, message):
        """
        使用語音引擎播報訊息
        
        參數:
            message: 要播報的訊息
        """
        if self.voice_enabled:
            try:
                self.engine.say(message)
                self.engine.runAndWait()
            except Exception as e:
                print(f"[-] 語音播報失敗: {e}")
    
    def log_event(self, event_type, username, ip, additional_info=""):
        """
        記錄事件到日誌
        
        參數:
            event_type: 事件類型
            username: 使用者名稱
            ip: 來源IP位址
            additional_info: 額外資訊
        """
        timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
        log_message = f"[{timestamp}] {event_type} - 使用者: {username}, IP: {ip}"
        if additional_info:
            log_message += f", {additional_info}"
        
        print(log_message)
        
        # 可以擴充為寫入自訂日誌檔案
        # with open('login_monitor.log', 'a') as f:
        #     f.write(log_message + '\n')
    
    def block_ip(self, ip):
        """
        封鎖可疑IP位址
        
        參數:
            ip: 要封鎖的IP位址
        """
        if not self.block_ip_enabled:
            return
        
        if ip in self.blocked_ips:
            return
        
        try:
            # 使用iptables封鎖IP
            cmd = f"iptables -A INPUT -s {ip} -j DROP"
            subprocess.run(cmd.split(), check=True)
            
            self.blocked_ips.add(ip)
            self.log_event("IP_BLOCKED", "system", ip, "已新增到防火牆封鎖清單")
            
            if self.voice_enabled:
                self.speak(f"已封鎖可疑IP位址 {ip}")
            
        except Exception as e:
            print(f"[-] 封鎖IP {ip} 失敗: {e}")
    
    def handle_failed_login(self, username, ip):
        """
        處理登入失敗事件
        
        參數:
            username: 使用者名稱
            ip: 來源IP位址
        """
        # 更新失敗計數
        if ip not in self.failed_logins:
            self.failed_logins[ip] = {'count': 0, 'users': set()}
        
        self.failed_logins[ip]['count'] += 1
        self.failed_logins[ip]['users'].add(username)
        
        failed_count = self.failed_logins[ip]['count']
        
        self.log_event(
            "LOGIN_FAILED",
            username,
            ip,
            f"失敗次數: {failed_count}/{self.max_failed}"
        )
        
        # 檢查是否超過閾值
        if failed_count >= self.max_failed:
            self.log_event(
                "ALERT",
                username,
                ip,
                f"超過失敗次數上限!"
            )
            
            if self.voice_enabled:
                warning_msg = f"警告! IP位址 {ip} 已嘗試登入失敗 {failed_count} 次"
                self.speak(warning_msg)
            
            # 封鎖IP
            self.block_ip(ip)
            
            # 如果啟用系統關閉功能
            if self.shutdown_enabled:
                shutdown_msg = "偵測到多次登入失敗,系統將在60秒後關閉"
                self.speak(shutdown_msg)
                subprocess.run(["shutdown", "-h", "+1"])
    
    def handle_successful_login(self, username, ip):
        """
        處理登入成功事件
        
        參數:
            username: 使用者名稱
            ip: 來源IP位址
        """
        # 重置該IP的失敗計數
        if ip in self.failed_logins:
            self.failed_logins[ip]['count'] = 0
        
        # 記錄成功登入
        if ip not in self.successful_logins:
            self.successful_logins[ip] = []
        
        self.successful_logins[ip].append({
            'username': username,
            'timestamp': datetime.now()
        })
        
        self.log_event("LOGIN_SUCCESS", username, ip)
        
        if self.voice_enabled and self.alert_enabled:
            success_msg = f"使用者 {username}{ip} 登入成功"
            self.speak(success_msg)
    
    def handle_invalid_user(self, username, ip):
        """
        處理無效使用者嘗試
        
        參數:
            username: 使用者名稱
            ip: 來源IP位址
        """
        self.log_event(
            "INVALID_USER",
            username,
            ip,
            "嘗試使用不存在的使用者名稱"
        )
        
        # 無效使用者嘗試也計入失敗次數
        self.handle_failed_login(username, ip)
    
    def monitor_log(self):
        """
        主要監控迴圈
        """
        print(f"[*] 開始監控日誌檔案: {self.logfile}")
        print(f"[*] 失敗次數閾值: {self.max_failed}")
        print(f"[*] 系統關閉功能: {'啟用' if self.shutdown_enabled else '停用'}")
        print(f"[*] IP封鎖功能: {'啟用' if self.block_ip_enabled else '停用'}")
        print(f"[*] 語音通知功能: {'啟用' if self.voice_enabled else '停用'}")
        print("=" * 80)
        
        try:
            # 使用pygtail即時追蹤日誌檔案
            for line in pygtail.Pygtail(self.logfile):
                # 檢查登入失敗
                failed_match = self.failed_password_pattern.search(line)
                if failed_match:
                    username = failed_match.group(1)
                    ip = failed_match.group(2)
                    self.handle_failed_login(username, ip)
                    continue
                
                # 檢查登入成功
                success_match = self.success_pattern.search(line)
                if success_match:
                    username = success_match.group(1)
                    ip = success_match.group(2)
                    self.handle_successful_login(username, ip)
                    continue
                
                # 檢查無效使用者
                invalid_match = self.invalid_user_pattern.search(line)
                if invalid_match:
                    username = invalid_match.group(1)
                    ip = invalid_match.group(2)
                    self.handle_invalid_user(username, ip)
                    continue
                
        except KeyboardInterrupt:
            print("\n[!] 監控系統已停止")
            self.print_statistics()
        except Exception as e:
            print(f"[-] 監控過程發生錯誤: {e}")
    
    def print_statistics(self):
        """
        輸出統計資訊
        """
        print("\n" + "=" * 80)
        print("監控統計資訊")
        print("=" * 80)
        
        print(f"\n已封鎖IP數量: {len(self.blocked_ips)}")
        if self.blocked_ips:
            for ip in self.blocked_ips:
                print(f"  - {ip}")
        
        print(f"\n有登入失敗記錄的IP數量: {len(self.failed_logins)}")
        for ip, data in self.failed_logins.items():
            if data['count'] > 0:
                print(f"  - {ip}: {data['count']} 次失敗")
                print(f"    嘗試的使用者: {', '.join(data['users'])}")
        
        print(f"\n成功登入的IP數量: {len(self.successful_logins)}")
        for ip, logins in self.successful_logins.items():
            print(f"  - {ip}: {len(logins)} 次成功登入")

def main():
    """
    主程式進入點
    """
    # 檢查是否具有root權限
    if os.geteuid() != 0:
        print("[-] 此程式需要root權限執行")
        sys.exit(1)
    
    # 建立監控器實例
    monitor = LoginMonitor()
    
    # 啟動監控
    monitor.monitor_log()

if __name__ == "__main__":
    main()

監控系統流程架構

@startuml
!define PLANTUML_FORMAT svg
!theme _none_

skinparam dpi auto
skinparam shadowing false
skinparam linetype ortho
skinparam roundcorner 5
skinparam defaultFontName "Microsoft JhengHei UI"
skinparam defaultFontSize 16
skinparam minClassWidth 100

start
:初始化監控系統;
:載入組態設定;
:初始化語音引擎;
:編譯正規表示式模式;

partition "即時監控迴圈" {
  while (持續監控日誌?) is (是)
    :讀取新的日誌行;
    :嘗試比對各種模式;
    
    if (比對到登入失敗?) then (是)
      :解析使用者名稱與IP;
      :更新失敗計數器;
      :記錄事件到日誌;
      
      if (失敗次數超過閾值?) then (是)
        :觸發警報機制;
        :執行語音通知;
        :封鎖來源IP位址;
        
        if (啟用系統關閉?) then (是)
          :排程系統關閉;
        endif
      endif
      
    else if (比對到登入成功?) then (是)
      :解析使用者名稱與IP;
      :重置失敗計數器;
      :記錄成功事件;
      :執行語音通知;
      
    else if (比對到無效使用者?) then (是)
      :記錄無效使用者嘗試;
      :計入失敗次數;
    endif
    
  endwhile (否)
}

:輸出監控統計資訊;
:清理資源;
stop

@enduml

這個登入監控系統採用事件驅動的架構設計,能夠即時回應各種登入事件。系統透過維護詳細的統計資料,為管理員提供全面的安全態勢感知。語音通知功能確保了重要事件能夠立即引起注意,而IP封鎖機制則提供了自動化的防禦能力。完整的組態系統讓管理員可以根據實際需求調整監控策略,在安全性和可用性之間取得平衡。

進階網路協定處理技術

現代網路環境中,各種標準協定和自訂協定共同構成了複雜的通訊體系。對於資安專業人員而言,深入理解這些協定的運作機制,並具備自訂協定處理的能力,是進行深度安全分析和防護的關鍵。Scapy框架提供了強大的網路協定處理能力,讓開發者能夠靈活地建構、解析和分析各種網路封包。

網域名稱系統協定處理

網域名稱系統是網際網路基礎設施的核心元件,負責將人類可讀的網域名稱轉換為機器可用的IP位址。透過Scapy,我們可以深入分析DNS協定的運作細節,建構自訂的DNS查詢工具,甚至實作簡單的DNS伺服器功能。

#!/usr/bin/python3

from scapy.all import *
import sys

class DNSHandler:
    """
    DNS協定處理器類別
    提供DNS查詢與回應的建構與解析功能
    """
    
    def __init__(self):
        """
        初始化DNS處理器
        """
        # DNS記錄類型對應
        self.query_types = {
            'A': 1,      # IPv4位址
            'NS': 2,     # 名稱伺服器
            'CNAME': 5,  # 標準名稱
            'SOA': 6,    # 授權起始點
            'PTR': 12,   # 指標記錄
            'MX': 15,    # 郵件交換
            'TXT': 16,   # 文字記錄
            'AAAA': 28,  # IPv6位址
        }
    
    def create_dns_query(self, domain_name, query_type='A', recursive=True):
        """
        建構DNS查詢封包
        
        參數:
            domain_name: 要查詢的網域名稱
            query_type: 查詢類型(A, AAAA, MX等)
            recursive: 是否要求遞迴查詢
        
        回傳:
            完整的DNS查詢封包
        """
        try:
            # 建構DNS查詢問題部分
            # qname: 查詢的網域名稱
            # qtype: 查詢的記錄類型
            dns_question = DNSQR(
                qname=domain_name,
                qtype=query_type
            )
            
            # 建構完整的DNS請求封包
            # rd: 遞迴查詢旗標(Recursion Desired)
            # qd: 查詢問題部分
            dns_request = DNS(
                rd=int(recursive),
                qd=dns_question
            )
            
            # 組合完整的網路封包
            # 使用UDP協定,目標埠53(DNS標準埠)
            packet = IP(dst="8.8.8.8") / UDP(dport=53) / dns_request
            
            return packet
            
        except Exception as e:
            print(f"[-] 建構DNS查詢封包失敗: {e}")
            return None
    
    def create_dns_response(self, domain_name, ip_address, ttl=3600):
        """
        建構DNS回應封包
        
        參數:
            domain_name: 網域名稱
            ip_address: 對應的IP位址
            ttl: 記錄的存活時間(秒)
        
        回傳:
            完整的DNS回應封包
        """
        try:
            # 建構DNS資源記錄(Resource Record)
            # rrname: 記錄名稱
            # type: 記錄類型
            # rdata: 記錄資料(IP位址)
            # ttl: 存活時間
            dns_answer = DNSRR(
                rrname=domain_name,
                type='A',
                rdata=ip_address,
                ttl=ttl
            )
            
            # 建構完整的DNS回應封包
            # qr: 查詢/回應旗標(1表示回應)
            # ra: 支援遞迴旗標(Recursion Available)
            # an: 回答部分
            dns_response = DNS(
                qr=1,
                ra=1,
                an=dns_answer
            )
            
            return dns_response
            
        except Exception as e:
            print(f"[-] 建構DNS回應封包失敗: {e}")
            return None
    
    def send_dns_query(self, domain_name, query_type='A', timeout=2):
        """
        發送DNS查詢並接收回應
        
        參數:
            domain_name: 要查詢的網域名稱
            query_type: 查詢類型
            timeout: 逾時時間(秒)
        
        回傳:
            DNS回應封包
        """
        try:
            # 建構查詢封包
            query = self.create_dns_query(domain_name, query_type)
            
            if not query:
                return None
            
            print(f"[*] 查詢網域: {domain_name}")
            print(f"[*] 查詢類型: {query_type}")
            
            # 發送封包並等待回應
            # sr1: 發送封包並接收一個回應
            response = sr1(query, timeout=timeout, verbose=0)
            
            if response and response.haslayer(DNS):
                return response
            else:
                print(f"[-] 未收到DNS回應")
                return None
                
        except Exception as e:
            print(f"[-] 發送DNS查詢失敗: {e}")
            return None
    
    def parse_dns_response(self, response):
        """
        解析DNS回應封包
        
        參數:
            response: DNS回應封包
        
        回傳:
            解析後的資訊字典
        """
        try:
            if not response or not response.haslayer(DNS):
                return None
            
            dns_layer = response.getlayer(DNS)
            results = {
                'query_id': dns_layer.id,
                'answers': [],
                'authorities': [],
                'additional': []
            }
            
            # 解析回答部分
            if dns_layer.ancount > 0:
                for i in range(dns_layer.ancount):
                    answer = dns_layer.an[i]
                    results['answers'].append({
                        'name': answer.rrname.decode() if hasattr(answer.rrname, 'decode') else str(answer.rrname),
                        'type': answer.type,
                        'data': answer.rdata,
                        'ttl': answer.ttl
                    })
            
            # 解析授權部分
            if dns_layer.nscount > 0:
                for i in range(dns_layer.nscount):
                    auth = dns_layer.ns[i]
                    results['authorities'].append({
                        'name': auth.rrname.decode() if hasattr(auth.rrname, 'decode') else str(auth.rrname),
                        'type': auth.type,
                        'data': auth.rdata
                    })
            
            return results
            
        except Exception as e:
            print(f"[-] 解析DNS回應失敗: {e}")
            return None

def main():
    """
    主程式:示範DNS協定處理功能
    """
    if len(sys.argv) < 2:
        print("使用方式:")
        print(f"  {sys.argv[0]} <網域名稱> [查詢類型]")
        print("\n範例:")
        print(f"  {sys.argv[0]} example.com")
        print(f"  {sys.argv[0]} example.com AAAA")
        sys.exit(1)
    
    domain = sys.argv[1]
    query_type = sys.argv[2] if len(sys.argv) > 2 else 'A'
    
    # 建立DNS處理器
    handler = DNSHandler()
    
    # 發送查詢
    response = handler.send_dns_query(domain, query_type)
    
    if response:
        # 解析回應
        results = handler.parse_dns_response(response)
        
        if results:
            print("\n" + "=" * 80)
            print("DNS查詢結果")
            print("=" * 80)
            
            if results['answers']:
                print("\n回答記錄:")
                for answer in results['answers']:
                    print(f"  名稱: {answer['name']}")
                    print(f"  類型: {answer['type']}")
                    print(f"  資料: {answer['data']}")
                    print(f"  TTL: {answer['ttl']} 秒")
                    print()

if __name__ == "__main__":
    main()

動態主機組態協定版本6處理

IPv6網路環境中,DHCPv6協定負責自動組態主機的網路參數。相較於IPv4的DHCP,DHCPv6提供了更豐富的組態選項和更靈活的位址分配機制。

#!/usr/bin/python3

from scapy.all import *
import random

class DHCPv6Handler:
    """
    DHCPv6協定處理器類別
    提供DHCPv6訊息的建構與處理功能
    """
    
    def __init__(self):
        """
        初始化DHCPv6處理器
        """
        # DHCPv6訊息類型
        self.message_types = {
            'SOLICIT': 1,      # 客戶端請求
            'ADVERTISE': 2,    # 伺服器通告
            'REQUEST': 3,      # 客戶端請求確認
            'CONFIRM': 4,      # 確認位址
            'RENEW': 5,        # 更新租約
            'REBIND': 6,       # 重新綁定
            'REPLY': 7,        # 伺服器回覆
            'RELEASE': 8,      # 釋放位址
            'DECLINE': 9,      # 拒絕位址
        }
    
    def create_solicit(self, client_mac, transaction_id=None):
        """
        建構DHCPv6 SOLICIT訊息
        
        參數:
            client_mac: 客戶端MAC位址
            transaction_id: 交易識別碼(若未提供則自動生成)
        
        回傳:
            完整的DHCPv6 SOLICIT封包
        """
        try:
            # 生成交易識別碼(如果未提供)
            if transaction_id is None:
                transaction_id = random.randint(0, 0xFFFFFF)
            
            # 建構DHCPv6 SOLICIT訊息
            dhcp6_msg = DHCP6_Solicit(trid=transaction_id)
            
            # 新增客戶端識別選項
            # DUID: DHCP Unique Identifier
            client_id = DHCP6OptClientId(
                duid=DUID_LLT(
                    lladdr=client_mac
                )
            )
            
            # 新增身份關聯選項(Identity Association)
            ia_na = DHCP6OptIA_NA(
                iaid=random.randint(0, 0xFFFFFFFF),
                T1=0,
                T2=0
            )
            
            # 新增選項請求清單
            opt_request = DHCP6OptOptReq(
                reqopts=[23, 24]  # 請求DNS伺服器和網域名稱
            )
            
            # 組合完整的DHCPv6訊息
            dhcp6_msg = dhcp6_msg / client_id / ia_na / opt_request
            
            # 建構IPv6和UDP標頭
            # DHCPv6使用連結本地位址和多播位址
            packet = IPv6(
                src="fe80::1",
                dst="ff02::1:2"  # 所有DHCPv6伺服器多播位址
            ) / UDP(
                sport=546,  # DHCPv6客戶端埠
                dport=547   # DHCPv6伺服器埠
            ) / dhcp6_msg
            
            return packet
            
        except Exception as e:
            print(f"[-] 建構DHCPv6 SOLICIT訊息失敗: {e}")
            return None
    
    def create_advertise(self, transaction_id, client_duid, server_duid, ipv6_address):
        """
        建構DHCPv6 ADVERTISE訊息
        
        參數:
            transaction_id: 交易識別碼
            client_duid: 客戶端DUID
            server_duid: 伺服器DUID
            ipv6_address: 要分配的IPv6位址
        
        回傳:
            完整的DHCPv6 ADVERTISE封包
        """
        try:
            # 建構DHCPv6 ADVERTISE訊息
            dhcp6_msg = DHCP6_Advertise(trid=transaction_id)
            
            # 新增伺服器識別選項
            server_id = DHCP6OptServerId(duid=server_duid)
            
            # 新增客戶端識別選項
            client_id = DHCP6OptClientId(duid=client_duid)
            
            # 新增身份關聯選項,包含位址資訊
            ia_addr = DHCP6OptIAAddress(
                addr=ipv6_address,
                preflft=3600,  # 偏好生命週期
                validlft=7200  # 有效生命週期
            )
            
            ia_na = DHCP6OptIA_NA(
                iaid=1,
                T1=1800,  # 更新時間
                T2=2880,  # 重新綁定時間
                ianaopts=ia_addr
            )
            
            # 組合完整的DHCPv6訊息
            dhcp6_msg = dhcp6_msg / server_id / client_id / ia_na
            
            # 建構IPv6和UDP標頭
            packet = IPv6(
                src="fe80::2",
                dst="fe80::1"
            ) / UDP(
                sport=547,
                dport=546
            ) / dhcp6_msg
            
            return packet
            
        except Exception as e:
            print(f"[-] 建構DHCPv6 ADVERTISE訊息失敗: {e}")
            return None

def main():
    """
    主程式:示範DHCPv6協定處理功能
    """
    print("[*] DHCPv6協定處理示範")
    print("=" * 80)
    
    # 建立DHCPv6處理器
    handler = DHCPv6Handler()
    
    # 建構SOLICIT訊息
    client_mac = "00:11:22:33:44:55"
    solicit_packet = handler.create_solicit(client_mac)
    
    if solicit_packet:
        print("\n[+] 成功建構DHCPv6 SOLICIT訊息")
        print(f"    客戶端MAC: {client_mac}")
        solicit_packet.show()

if __name__ == "__main__":
    main()

網路協定堆疊架構

@startuml
!define PLANTUML_FORMAT svg
!theme _none_

skinparam dpi auto
skinparam shadowing false
skinparam linetype ortho
skinparam roundcorner 5
skinparam defaultFontName "Microsoft JhengHei UI"
skinparam defaultFontSize 16
skinparam minClassWidth 100

package "應用層協定" {
  [HTTP/HTTPS]
  [DNS]
  [DHCP/DHCPv6]
  [FTP/SFTP]
  [SMTP/POP3]
}

package "傳輸層協定" {
  [TCP]
  [UDP]
  [SCTP]
}

package "網路層協定" {
  [IPv4]
  [IPv6]
  [ICMP/ICMPv6]
  [IPsec]
}

package "鏈路層協定" {
  [Ethernet]
  [Wi-Fi]
  [PPP]
  [ARP/NDP]
}

package "實體層" {
  [乙太網路線]
  [光纖]
  [無線訊號]
}

[HTTP/HTTPS] --> [TCP]
[DNS] --> [UDP]
[DHCP/DHCPv6] --> [UDP]
[FTP/SFTP] --> [TCP]
[SMTP/POP3] --> [TCP]

[TCP] --> [IPv4]
[TCP] --> [IPv6]
[UDP] --> [IPv4]
[UDP] --> [IPv6]
[SCTP] --> [IPv4]
[SCTP] --> [IPv6]

[IPv4] --> [Ethernet]
[IPv4] --> [Wi-Fi]
[IPv6] --> [Ethernet]
[IPv6] --> [Wi-Fi]
[ICMP/ICMPv6] --> [IPv4]
[ICMP/ICMPv6] --> [IPv6]
[IPsec] --> [IPv4]
[IPsec] --> [IPv6]

[Ethernet] --> [乙太網路線]
[Wi-Fi] --> [無線訊號]
[PPP] --> [乙太網路線]
[ARP/NDP] --> [Ethernet]

[乙太網路線] --> [實體層]
[光纖] --> [實體層]
[無線訊號] --> [實體層]

@enduml

自訂網路協定設計與實作

在某些特殊應用場景中,標準協定可能無法完全滿足需求。此時,設計自訂的網路協定成為必要選擇。Scapy提供了完整的自訂協定支援,讓開發者能夠定義新的封包格式和處理邏輯。

#!/usr/bin/python3

from scapy.all import *
import hashlib

class CustomSecurityProtocol(Packet):
    """
    自訂安全協定
    實作簡單的加密和完整性驗證功能
    """
    
    # 協定名稱
    name = "CustomSecurity"
    
    # 定義協定欄位結構
    fields_desc = [
        # 安全類型欄位(1位元組)
        # 0: 無加密, 1: AES加密, 2: DES加密
        ByteEnumField(
            "security_type",
            0,
            {0: "None", 1: "AES", 2: "DES", 3: "RSA"}
        ),
        
        # 協定版本欄位(1位元組)
        ByteField("version", 1),
        
        # 時間戳記欄位(4位元組)
        IntField("timestamp", 0),
        
        # 負載長度欄位(2位元組)
        ShortField("payload_length", 0),
        
        # 負載資料欄位(變動長度)
        StrLenField(
            "payload",
            "",
            length_from=lambda pkt: pkt.payload_length
        ),
        
        # 校驗和欄位(1位元組)
        XByteField("checksum", 0)
    ]
    
    def post_build(self, pkt, pay):
        """
        封包建構後的處理
        自動計算並設定校驗和
        
        參數:
            pkt: 建構完成的封包資料
            pay: 負載資料
        
        回傳:
            完整的封包
        """
        # 如果校驗和為0,則自動計算
        if self.checksum == 0:
            # 計算封包資料的校驗和(不包含校驗和欄位本身)
            checksum = self.calculate_checksum(pkt[:-1])
            # 更新封包中的校驗和欄位
            pkt = pkt[:-1] + bytes([checksum])
        
        return pkt + pay
    
    def calculate_checksum(self, data):
        """
        計算資料的校驗和
        
        參數:
            data: 要計算校驗和的資料
        
        回傳:
            校驗和值(1位元組)
        """
        # 使用MD5雜湊的第一個位元組作為簡單的校驗和
        hash_obj = hashlib.md5(data)
        return hash_obj.digest()[0]
    
    def verify_checksum(self):
        """
        驗證封包的校驗和
        
        回傳:
            布林值,True表示校驗和正確
        """
        # 取得封包的原始資料
        pkt_data = bytes(self)
        
        # 計算不含校驗和欄位的資料校驗和
        calculated = self.calculate_checksum(pkt_data[:-1])
        
        # 比較計算出的校驗和與封包中的校驗和
        return calculated == self.checksum

# 將自訂協定綁定到TCP協定上
# 當目標埠為12345時,使用自訂協定解析
bind_layers(TCP, CustomSecurityProtocol, dport=12345)
bind_layers(TCP, CustomSecurityProtocol, sport=12345)

def create_custom_packet(dest_ip, security_type, payload_data):
    """
    建立使用自訂協定的封包
    
    參數:
        dest_ip: 目標IP位址
        security_type: 安全類型(0-3)
        payload_data: 負載資料
    
    回傳:
        完整的網路封包
    """
    import time
    
    try:
        # 建立自訂協定層
        custom_layer = CustomSecurityProtocol(
            security_type=security_type,
            version=1,
            timestamp=int(time.time()),
            payload_length=len(payload_data),
            payload=payload_data
        )
        
        # 組合完整的網路封包
        packet = (
            IP(dst=dest_ip) /
            TCP(dport=12345, sport=random.randint(1024, 65535)) /
            custom_layer
        )
        
        return packet
        
    except Exception as e:
        print(f"[-] 建立自訂協定封包失敗: {e}")
        return None

def main():
    """
    主程式:示範自訂協定功能
    """
    print("[*] 自訂安全協定示範")
    print("=" * 80)
    
    # 建立測試封包
    test_data = b"Hello, this is a test message!"
    packet = create_custom_packet(
        dest_ip="192.168.1.100",
        security_type=1,  # AES加密
        payload_data=test_data
    )
    
    if packet:
        print("\n[+] 成功建立自訂協定封包")
        print("\n封包詳細資訊:")
        packet.show()
        
        # 驗證校驗和
        custom_layer = packet.getlayer(CustomSecurityProtocol)
        if custom_layer:
            is_valid = custom_layer.verify_checksum()
            print(f"\n[*] 校驗和驗證: {'通過' if is_valid else '失敗'}")
        
        # 顯示封包的十六進位表示
        print("\n封包十六進位表示:")
        hexdump(packet)

if __name__ == "__main__":
    main()

自訂協定應用場景架構

@startuml
!define PLANTUML_FORMAT svg
!theme _none_

skinparam dpi auto
skinparam shadowing false
skinparam linetype ortho
skinparam roundcorner 5
skinparam defaultFontName "Microsoft JhengHei UI"
skinparam defaultFontSize 16
skinparam minClassWidth 100

package "工業控制系統" {
  [PLC控制器]
  [SCADA系統]
  [感測器網路]
  note right of [PLC控制器]
    專用控制指令協定
    即時資料傳輸
    低延遲需求
  end note
}

package "物聯網環境" {
  [智慧型裝置]
  [閘道器]
  [雲端平台]
  note right of [智慧型裝置]
    輕量級通訊協定
    電池供電最佳化
    間歇性連線支援
  end note
}

package "安全通訊系統" {
  [認證伺服器]
  [加密模組]
  [金鑰管理]
  note right of [認證伺服器]
    專有認證機制
    端對端加密
    完整性驗證
  end note
}

package "特殊網路環境" {
  [衛星通訊]
  [水下聲納]
  [軍事通訊]
  note right of [衛星通訊]
    高延遲補償
    錯誤更正機制
    頻寬最佳化
  end note
}

[PLC控制器] ..> [自訂協定引擎] : 使用
[SCADA系統] ..> [自訂協定引擎] : 使用
[感測器網路] ..> [自訂協定引擎] : 使用

[智慧型裝置] ..> [自訂協定引擎] : 使用
[閘道器] ..> [自訂協定引擎] : 使用
[雲端平台] ..> [自訂協定引擎] : 使用

[認證伺服器] ..> [自訂協定引擎] : 使用
[加密模組] ..> [自訂協定引擎] : 使用
[金鑰管理] ..> [自訂協定引擎] : 使用

[衛星通訊] ..> [自訂協定引擎] : 使用
[水下聲納] ..> [自訂協定引擎] : 使用
[軍事通訊] ..> [自訂協定引擎] : 使用

package "核心元件" {
  [自訂協定引擎]
  [封包建構器]
  [協定解析器]
  [安全驗證模組]
}

[自訂協定引擎] --> [封包建構器]
[自訂協定引擎] --> [協定解析器]
[自訂協定引擎] --> [安全驗證模組]

@enduml

這些進階的網路協定處理技術展示了Scapy框架的強大能力。透過深入理解標準協定的運作機制,並掌握自訂協定的設計方法,資安專業人員能夠建構更專業的網路分析工具,並針對特殊需求開發客製化的網路通訊解決方案。無論是進行安全研究、協定分析,或是開發特殊應用的通訊系統,這些技術都提供了堅實的基礎。

綜合安全防護體系建構

本文深入探討了網路安全領域的多個關鍵技術面向,從基礎的網路位址探測、搜尋引擎安全評估、伺服器訊息區塊協定掃描,到進階的登入監控系統和網路協定處理技術。這些技術的綜合運用,構成了現代企業網路安全防護的重要基石。

網路安全是一個持續演進的領域,攻擊技術和防禦策略都在不斷發展。單一的防禦技術無法應對複雜多變的威脅環境,必須建立多層次、全方位的防護體系。本文介紹的各項技術相互補充,共同構建了完整的安全監控和防禦能力。網路位址探測技術幫助我們理解網路拓撲,搜尋引擎安全評估協助發現暴露的資訊資產,SMB掃描識別潛在的服務風險,而登入監控系統則提供即時的威脅偵測和回應能力。

未來的網路安全將更加注重主動防禦和智慧分析。人工智慧和機器學習技術的整合,將使安全系統能夠更準確地識別異常行為模式。威脅情報的共享和利用,將提升整體防禦的效率。零信任架構的推廣,將從根本上改變網路安全的設計理念。資安專業人員必須持續學習新技術,不斷更新知識體系,才能在這個快速變化的領域保持競爭力。

建立有效的安全防護體系需要技術、流程和人員的密切配合。技術工具提供了基礎能力,但合理的安全政策和訓練有素的安全團隊同樣重要。只有將這三者有機結合,才能建構真正堅固的網路安全防線,保護組織的數位資產免受各種威脅的侵害。