XDP 程式與其他 eBPF 程式不同之處在於其載入方式，通常無需手動載入，Linux 系統已整合 XDP 載入器於 ip 命令中。開發者可運用 ip link set dev <介面名稱> xdp obj <程式檔名> sec <節名稱> 的指令格式，將編譯完成的 XDP 程式載入至指定的網路介面，並指定程式中特定節區。例如，將 program.o 檔案中 mysection 節區的 XDP 程式載入至 eth0 介面，只需執行 ip link set dev eth0 xdp obj program.o sec mysection 指令即可。此指令會修改網路介面屬性，並啟用 XDP 程式，開始進行封包處理。更進一步，我們可以結合 BCC 工具，使用 Python 指令碼實作更複雜的 XDP 程式載入與控制。
透過 BCC，我們不僅可以動態載入 XDP 程式，還能讀取 BPF 地圖資料，實作更精細的封包監控與統計。例如，可以統計不同 IP 協定的封包數量，並根據需求設定過濾規則，例如阻擋特定 TCP 連線。此外，文章也詳細介紹瞭如何使用 Python 單元測試框架和 BPF_PROG_TEST_RUN 命令，對 XDP 程式進行效能測試，確保程式的正確性和效率。此測試方法能模擬真實封包流程，並驗證 XDP 程式的行為，無需搭建複雜的虛擬化環境，有效提升開發效率。文章中提供的程式碼範例涵蓋了 XDP 程式的核心功能，包含封包解析、計數、過濾等，可作為讀者學習和實踐的參考。
XDP 程式載入
與其他 eBPF 程式不同，XDP 程式通常不需要手動載入。Linux 提供了一個內建的載入器，實作為 ip 命令的一部分，可以用來載入編譯好的 XDP 程式。
使用 iproute2 載入 XDP 程式
ip 命令提供了一個簡單的語法來載入 XDP 程式：
ip link set dev <介面名稱> xdp obj <程式檔名> sec <節名稱>
這裡， <介面名稱> 是要載入 XDP 程式的網路介面名稱， <程式檔名> 是編譯好的 XDP 程式的 ELF 檔案名稱， <節名稱> 是 XDP 程式中的節名稱。
範例分析
假設我們要載入一個名為 program.o 的 XDP 程式到 eth0 介面上，且程式位於 mysection 節中，可以使用以下命令：
ip link set dev eth0 xdp obj program.o sec mysection
這個命令會載入 program.o 中的 XDP 程式到 eth0 介面，並啟用它。
內容解密：
在上述範例中，我們使用 ip link set 命令來載入 XDP 程式。這個命令會修改網路介面的屬性，並啟用指定的 XDP 程式。 xdp obj 引數指定要載入的 XDP 程式檔案， sec 引數指定程式中的節名稱。
圖表翻譯：
graph LR
A[IP 命令] --> B[載入 XDP 程式]
B --> C[修改網路介面屬性]
C --> D[啟用 XDP 程式]
style A fill:#f9f,stroke:#333,stroke-width:4px
style B fill:#f9f,stroke:#333,stroke-width:4px
style C fill:#f9f,stroke:#333,stroke-width:4px
style D fill:#f9f,stroke:#333,stroke-width:4px
這個圖表展示了使用 ip 命令載入 XDP 程式的過程。首先，執行 ip 命令，然後載入 XDP 程式，修改網路介面的屬性，最後啟用 XDP 程式。
使用 XDP 來阻擋網路存取
在這個例子中，我們將使用 XDP（eXpress Data Path）來阻擋網路存取。XDP 是一個 Linux 核心功能，允許使用者定義的程式碼在網路封包處理過程中執行。
安裝 XDP
首先，我們需要安裝 XDP 相關的工具和函式庫。可以使用以下命令：
sudo apt-get install linux-tools-common
建立 XDP 程式
接下來，我們需要建立一個 XDP 程式來阻擋網路存取。以下是範例程式碼：
#include <linux/bpf.h>
#include <linux/if_ether.h>
#include <linux/ip.h>
#include <linux/tcp.h>
SEC("prog")
int xdp_prog(struct xdp_md *ctx) {
void *data = (void *)(long)ctx->data;
void *data_end = (void *)(long)ctx->data_end;
struct ethhdr *eth = data;
struct iphdr *iph = data + sizeof(*eth);
struct tcphdr *tcph = data + sizeof(*eth) + sizeof(*iph);
if (tcph->dest == htons(8000)) {
return XDP_DROP;
}
return XDP_PASS;
}
這個程式碼定義了一個 XDP 程式，名為 xdp_prog。它會檢查每個封包的目的地埠是否為 8000，如果是，則丟棄該封包。
載入 XDP 程式
接下來，我們需要載入 XDP 程式到核心中。可以使用以下命令：
sudo ip link set enp0s3 xdp object xdp_prog.o sec prog
這個命令會載入 xdp_prog.o 這個物件檔案到核心中，並將其繫結到 enp0s3 這個網路介面上。
測試 XDP 程式
現在，我們可以測試 XDP 程式是否正常工作。可以使用以下命令：
sudo ss -tulpn
這個命令會顯示所有的網路連線。如果我們嘗試連線到 http://localhost:8000，則會發現連線被拒絕。
移除 XDP 程式
如果我們想要移除 XDP 程式，可以使用以下命令：
sudo ip link set enp0s3 xdp off
這個命令會移除 XDP 程式從核心中，並還原網路介面的正常功能。
圖表翻譯：
以下是 XDP 程式的流程圖：
flowchart TD
A[封包到達] --> B[檢查目的地埠]
B -->|是 8000|> C[丟棄封包]
B -->|不是 8000|> D[允許封包透過]
C --> E[結束]
D --> E
這個圖表顯示了 XDP 程式的流程。如果封包的目的地埠是 8000，則會丟棄該封包；否則，則允許封包透過。
XDP 程式設計：封鎖 TCP 連線
在本文中，我們將探討如何使用 XDP (Express Data Path) 程式設計來封鎖 TCP 連線。XDP 是一種高效的網路封包處理框架，允許開發人員在 Linux 核心中執行自定義的網路程式。
網路拓撲
首先，讓我們觀察一下網路拓撲。這臺機器有三個網路介面：lo、enp0s3 和 enp0s8。其中，enp0s3 是管理網路介面，enp0s8 是公開網路介面，我們的網頁伺服器將在這個介面上執行。
檢查開放連線埠
在載入 XDP 程式之前，我們可以使用 nmap 工具檢查伺服器上的開放連線埠。在這個例子中，我們可以看到連線埠 22 (SSH) 和 8000 (HTTP) 是開放的。
# nmap -sS 192.168.33.11
Nmap scan report for 192.168.33.11
Host is up (0.0034s latency).
Not shown: 998 closed ports
PORT STATE SERVICE
22/tcp open ssh
8000/tcp open http-alt
XDP 程式設計
現在，我們需要設計一個 XDP 程式來封鎖 TCP 連線。首先，我們需要包含必要的標頭檔，例如 linux/bpf.h、linux/if_ether.h 和 linux/ip.h。
#include <linux/bpf.h>
#include <linux/if_ether.h>
#include <linux/in.h>
#include <linux/ip.h>
接下來，我們需要定義一個 SEC 宏來宣告 ELF 屬性。
#define SEC(NAME) __attribute__((section(NAME), used))
然後，我們可以定義 XDP 程式的主入口點 myprogram，並宣告它所屬的 ELF 節 mysection。
SEC("mysection")
int myprogram(struct xdp_md *ctx) {
//...
}
在 myprogram 中，我們需要宣告一些變數，例如資料指標、乙太網路層和 IP 層結構體。
int ipsize = 0;
void *data = (void *)(long)ctx->data;
void *data_end = (void *)(long)ctx->data_end;
struct ethhdr *eth = data;
struct iphdr *ip;
接下來，我們需要從乙太網路層中提取 IP 層，並檢查是否超出資料指標的範圍。
ipsize = sizeof(*eth);
ip = data + ipsize;
ipsize += sizeof(struct iphdr);
if (data + ipsize > data_end) {
return XDP_DROP;
}
最後，我們可以實作封鎖 TCP 連線的邏輯。如果 IP 層的協定是 TCP，我們就丟棄該封包。
if (ip->protocol == IPPROTO_TCP) {
return XDP_DROP;
}
return XDP_PASS;
編譯和載入 XDP 程式
編譯 XDP 程式可以使用 Clang 編譯器，並生成一個 ELF 檔案 program.o。
$ clang -O2 -target bpf -c program.c -o program.o
然後，我們可以使用 ip 工具將 XDP 程式載入到公開網路介面 enp0s8 上。
# ip link set dev enp0s8 xdp obj program.o sec mysection
如果載入成功，我們可以檢查網路介面是否已經套用了 XDP 程式。
XDP 程式設計與 BCC 載入
XDP（Express Data Path）是一種高效能的網路封包處理技術，允許開發人員在 Linux 核心中執行自定義的程式碼。BCC（BPF Compiler Collection）是一個工具集合，提供了一種載入和執行 BPF（Berkeley Packet Filter）程式的方法，包括 XDP 程式。
XDP 程式設計
XDP 程式設計涉及到建立一個 kernel-space 程式，該程式會在網路封包到達時被執行。以下是一個簡單的 XDP 程式範例：
#define KBUILD_MODNAME "program"
#include <linux/bpf.h>
#include <linux/in.h>
#include <linux/ip.h>
BPF_TABLE("percpu_array", uint32_t, long, packetcnt, 256);
int myprogram(struct xdp_md *ctx) {
int ipsize = 0;
void *data = (void *)(long)ctx->data;
void *data_end = (void *)(long)ctx->data_end;
struct ethhdr *eth = data;
struct iphdr *ip;
long *cnt;
__u32 idx;
ipsize = sizeof(*eth);
ip = data + ipsize;
ipsize += sizeof(struct iphdr);
if (data + ipsize > data_end) {
return XDP_DROP;
}
idx = ip->protocol;
cnt = packetcnt.lookup(&idx);
if (cnt) {
*cnt += 1;
}
if (ip->protocol == IPPROTO_TCP) {
return XDP_DROP;
}
return XDP_PASS;
}
這個程式會計算每個 IP 協定的封包數量，並丟棄所有 TCP 封包。
BCC 載入
要載入 XDP 程式，需要使用 BCC 的 BPF 類別。以下是一個簡單的載入範例：
#!/usr/bin/python
from bcc import BPF
import time
device = "enp0s8"
# 載入 XDP 程式
b = BPF(src_file="program.c")
# 取得 BPF 地圖
packetcnt = b.get_table("packetcnt")
# 載入 XDP 程式到指定的網路介面
b.load_func(b"myprogram", BPF.XDP)
# 執行 XDP 程式
while True:
# 印出封包數量
for k, v in packetcnt.items():
print(f"Protocol {k.value}: {v.value}")
time.sleep(1)
這個範例會載入 XDP 程式到指定的網路介面，並印出每個 IP 協定的封包數量。
實作XDP程式的封包計數與監控
XDP程式設計
在實作XDP（Express Data Path）程式時，我們需要定義一個可以處理網路封包的函式，並將其附加到指定的網路介面上。以下是使用BPF（Berkeley Packet Filter）實作XDP程式的範例：
// program.c
#include <linux/bpf.h>
#include <linux/if_ether.h>
#include <linux/ip.h>
struct bpf_map_def SEC("maps") packetcnt = {
.type = BPF_MAP_TYPE_ARRAY,
.key_size = sizeof(u32),
.value_size = sizeof(u64),
.max_entries = 256,
};
SEC("xdp")
int xdp_myprogram(struct xdp_md *ctx) {
void *data = (void *)(long)ctx->data;
void *data_end = (void *)(long)ctx->data_end;
struct ethhdr *eth = data;
if ((void *)(eth + 1) > data_end)
return XDP_ABORTED;
u32 protocol = eth->h_proto;
u64 *count = bpf_map_lookup_elem(&packetcnt, &protocol);
if (count) {
__u64 val = *count;
*count = val + 1;
}
return XDP_DROP;
}
載入並附加XDP程式
接下來，我們需要載入XDP程式並將其附加到指定的網路介面上：
import ctypes
from bpf import BPF
# 載入XDP程式
b = BPF(src_file="program.c")
# 載入XDP函式
fn = b.load_func("xdp_myprogram", BPF.XDP)
# 附加XDP程式到網路介面
b.attach_xdp(device, fn, 0)
取得封包計數
現在，我們需要取得封包計數並將其印出。首先，我們需要取得packetcnt地圖：
packetcnt = b.get_table("packetcnt")
接下來，我們需要實作一個迴圈來取得封包計數並將其印出：
prev = [0] * 256
print("Printing packet counts per IP protocol-number, hit CTRL+C to stop")
while True:
try:
for k in packetcnt.keys():
val = packetcnt.sum(k).value
i = k.value
if val:
delta = val - prev[i]
prev[i] = val
print("{}: {} pkt/s".format(i, delta))
time.sleep(1)
except KeyboardInterrupt:
print("Removing filter from device")
break
這個迴圈會不斷取得封包計數並將其印出，直到使用者按下CTRL+C鍵。
移除XDP程式
最後，當使用者按下CTRL+C鍵時，我們需要移除XDP程式：
b.remove_xdp(device, fn)
這樣就完成了XDP程式的實作和封包計數的監控。
測試 XDP 程式
當我們在開發 XDP 程式時，遇到的最大挑戰是，要測試實際的封包流程，需要重現一個環境，使所有元件都能夠提供正確的封包。雖然現代的虛擬化技術使得建立一個工作環境變得相對容易，但複雜的設定可能會限制測試環境的可重現性和可程式化。另外，在虛擬化環境中分析高頻率 XDP 程式的效能方面，虛擬化的成本使得測試無效，因為它遠遠大於實際的封包處理成本。
幸運的是，核心開發人員已經有了一個解決方案。他們實作了一個命令，可以用來測試 XDP 程式，稱為 BPF_PROG_TEST_RUN。基本上，BPF_PROG_TEST_RUN 可以讓一個 XDP 程式執行，並接受一個輸入封包和一個輸出封包。當程式執行時，輸出封包變數會被填充，傳回的 XDP 程式碼也會被傳回。這意味著您可以在測試斷言中使用輸出封包和傳回程式碼。
使用 Python 單元測試框架進行 XDP 測試
使用 BPF_PROG_TEST_RUN 和 Python 單元測試框架 unittest 來撰寫 XDP 測試是一個好主意，原因如下：
- 您可以使用 Python 的 BCC 函式庫載入和執行 BPF 程式。
- Python 有一個非常好的封包工藝和內省函式庫：Scapy。
- Python 可以使用
ctypes來與 C 結構進行整合。
首先，我們需要匯入所有需要的函式庫；這是我們在一個名為 test_xdp.py 的檔案中要做的第一件事：
from bcc import BPF, libbcc
from scapy.all import Ether, IP, raw, TCP, UDP
import ctypes
import unittest
接下來，我們可以建立一個測試案例類別，名為 XDPExampleTestCase。這個測試類別將包含所有我們的測試案例和一個成員方法 (_xdp_test_run)，我們將使用它來進行斷言和呼叫 bpf_prog_test_run。
以下是 _xdp_test_run 方法的實作：
def _xdp_test_run(self, given_packet, expected_packet, expected_return):
size = len(given_packet)
given_packet = ctypes.create_string_buffer(raw(given_packet), size)
packet_output = ctypes.create_string_buffer(self.SKB_OUT_SIZE)
packet_output_size = ctypes.c_uint32()
test_retval = ctypes.c_uint32()
duration = ctypes.c_uint32()
repeat = 1
ret = libbcc.lib.bpf_prog_test_run(self.bpf_function.fd,
repeat,
ctypes.byref(given_packet),
size,
ctypes.byref(packet_output),
ctypes.byref(packet_output_size),
ctypes.byref(test_retval),
ctypes.byref(duration))
圖表翻譯：
graph LR
A[開始] --> B[載入 BPF 程式]
B --> C[建立測試案例]
C --> D[呼叫 bpf_prog_test_run]
D --> E[進行斷言]
E --> F[輸出結果]
這個圖表展示了測試 XDP 程式的流程，從載入 BPF 程式開始，到建立測試案例、呼叫 bpf_prog_test_run、進行斷言，最後輸出結果。
XDP 程式測試
XDP（eXpress Data Path）是一種高效能的網路封包處理框架，允許開發人員在 Linux 核心中執行自定義的封包處理邏輯。為了確保 XDP 程式的正確性和可靠性，進行徹底的測試是非常重要的。
測試框架
首先，我們需要建立一個測試框架，以便於測試 XDP 程式。這個框架應該包括以下幾個部分：
- 測試使用案例：定義一系列的測試使用案例，以覆寫不同的封包型別、協定和場景。
- 測試函式：實作一個測試函式，負責執行 XDP 程式並驗證其輸出。
- setUp 方法：定義一個 setUp 方法，用於初始化 XDP 程式和相關的資源。
測試函式
測試函式 _xdp_test_run 負責執行 XDP 程式並驗證其輸出。它接受三個引數：
given_packet：輸入的封包。expected_packet：預期的輸出封包。expected_return：預期的 XDP 程式傳回值。
這個函式使用 ctypes 函式庫將 Python 物件轉換為 C 型別，並呼叫 libbcc 函式庫中的 bpf_prog_test_run 函式執行 XDP 程式。然後，它驗證 XDP 程式的輸出與預期的結果是否匹配。
測試使用案例
我們可以定義多個測試使用案例，以覆寫不同的場景。例如：
- TCP 封包：測試 XDP 程式是否能夠正確地處理 TCP 封包。
- UDP 封包：測試 XDP 程式是否能夠正確地處理 UDP 封包。
- TCP 封包轉發：測試 XDP 程式是否能夠正確地轉發 TCP 封包至指定的埠。
每個測試使用案例都應該呼叫 _xdp_test_run 函式，並傳入相應的引數。
setUp 方法
setUp 方法負責初始化 XDP 程式和相關的資源。它應該載入 XDP 程式並將其儲存在 self.bpf_function 屬性中。
程式碼範例
import ctypes
from bcc import BPF
class TestXDPProgram(unittest.TestCase):
def setUp(self):
self.bpf_prog = BPF(src_file=b"program.c")
self.bpf_function = self.bpf_prog.load_func(b"myprogram", BPF.XDP)
def _xdp_test_run(self, given_packet, expected_packet, expected_return):
# 將 Python 物件轉換為 C 型別
given_packet_ptr = ctypes.pointer(given_packet)
expected_packet_ptr = ctypes.pointer(expected_packet)
# 呼叫 libbcc 函式庫中的 bpf_prog_test_run 函式
ret = self.bpf_function(given_packet_ptr, expected_packet_ptr)
# 驗證 XDP 程式的輸出
self.assertEqual(ret, expected_return)
if expected_packet:
self.assertEqual(given_packet, expected_packet)
def test_drop_tcp(self):
given_packet = Ether() / IP() / TCP()
self._xdp_test_run(given_packet, None, BPF.XDP_DROP)
def test_pass_udp(self):
given_packet = Ether() / IP() / UDP()
expected_packet = Ether() / IP() / UDP()
self._xdp_test_run(given_packet, expected_packet, BPF.XDP_PASS)
def test_forward_tcp(self):
given_packet = Ether() / IP() / TCP(dport=9090)
expected_packet = Ether(dst="08:00:27:dd:38:2a") / IP() / TCP(dport=9090)
self._xdp_test_run(given_packet, expected_packet, BPF.XDP_PASS)
這個程式碼範例定義了一個 TestXDPProgram 類別，包含了 setUp 方法和多個測試使用案例。每個測試使用案例都呼叫 _xdp_test_run 函式，並傳入相應的引數。
XDP 程式開發與測試
從底層實作到高階應用的全面檢視顯示，XDP 程式提供了一個在網路堆積疊早期階段高效處理封包的機制。透過多維度效能指標的實測分析，XDP 展現了其在降低延遲和提升吞吐量方面的顯著優勢，尤其在應對 DDoS 攻擊等場景下更具價值。然而，技術限制深析指出，XDP 程式開發的複雜度和除錯的難度仍然是需要正視的挑戰，需要更完善的工具鏈和更友善的開發框架來降低門檻。
考量 XDP 程式與現有系統的整合，漸進式匯入策略更為穩妥。優先應用於效能瓶頸突出的關鍵路徑，例如網路安全防護和流量控制，可以最大化其價值。技術團隊應著重於解決核心挑戰，例如程式碼驗證和資源管理，才能釋放 XDP 的完整潛力。從技術演進角度，eBPF 和 XDP 代表了未來網路技術的重要方向，值得持續投入和關注。隨著硬體和軟體生態的日趨成熟，我們預見 XDP 的應用將更加普及，並在更多網路應用場景中發揮關鍵作用。玄貓認為，掌握 XDP 程式設計將成為網路工程師的必備技能。
