Posted inPython

Làm chủ Pyshark: ‘Soi’ traffic mạng thời gian thực bằng Python

Python tutorial - IT technology blog
Python tutorial - IT technology blog

Chọn công cụ nào để “soi” traffic mạng?

Anh em DevOps hay System Admin chắc không lạ gì cảnh server bỗng dưng “lag” mà không rõ nguyên nhân, API trả về 500 liên tục hoặc nghi ngờ có traffic lạ tấn công. Cách truyền thống là dùng tcpdump ghi file .pcap, sau đó tải về máy mở Wireshark để soi. Quy trình này ổn nhưng chỉ là xử lý hậu kỳ (post-analysis).

Nếu muốn xây dựng hệ thống giám sát tự động, cảnh báo ngay lập tức khi phát hiện mã độc hoặc lọc nhanh request theo IP, chúng ta cần can thiệp bằng code. Trong thế giới Python, có 3 hướng tiếp cận phổ biến:

1. Scapy

Scapy giống như một con dao đa năng, cho phép can thiệp sâu vào từng bit dữ liệu. Tuy nhiên, rào cản lớn nhất là đường cong học tập khá dốc (steep learning curve). Hiệu năng của nó khi xử lý lượng traffic lớn cũng không thực sự ấn tượng. Scapy thường được ưu tiên để tạo gói tin giả lập (Packet Crafting) hơn là bắt gói tin số lượng lớn.

2. Tcpdump / TShark (Dòng lệnh)

Đây là những tượng đài về tốc độ, nhẹ và có sẵn trên hầu hết các bản phân phối Linux. Thế nhưng, việc viết bash script để parse chuỗi JSON từ payload HTTP thực sự là một cực hình. Bạn sẽ tốn cả ngày chỉ để xử lý các ký tự lạ và xuống dòng.

3. Pyshark

Pyshark không tự mình bắt gói tin mà đóng vai trò là một “wrapper” quanh TShark. Điểm ăn tiền của nó là thừa hưởng khả năng phân tích hơn 3.000 giao thức từ Wireshark. Nó chuyển đổi dữ liệu thô thành các Object Python cực kỳ dễ thao tác.

Tại sao Pyshark là lựa chọn hàng đầu cho dự án thực tế?

Sau nhiều dự án, mình nhận thấy Pyshark là điểm giao thoa tốt nhất giữa sức mạnh và sự tiện dụng. Thay vì phải nhớ vị trí từng byte trong header, bạn chỉ cần gọi packet.ip.src.

Thực tế, chỉ cần Wireshark đọc được protocol đó, Pyshark sẽ xử lý được. Bạn không cần định nghĩa lại cấu trúc gói tin phức tạp. Dù hiệu năng có chậm hơn thư viện thuần C một chút, nhưng với các tác vụ giám sát thông thường, Pyshark hoàn toàn đáp ứng tốt.

Triển khai Pyshark: Từ cài đặt đến thực thi

Trước tiên, máy bạn cần cài sẵn TShark. Trên Ubuntu, hãy chạy lệnh sau:

sudo apt update && sudo apt install tshark -y

Tiếp theo, cài đặt thư viện thông qua pip:

pip install pyshark

Bắt gói tin thời gian thực (Live Capture)

Kịch bản phổ biến nhất là theo dõi traffic qua card mạng. Đoạn code sau sẽ bắt gói tin trên interface eth0 và in ra thông tin cơ bản.

import pyshark

def live_capture_basic(interface_name):
    # Khởi tạo LiveCapture
    capture = pyshark.LiveCapture(interface=interface_name)
    print(f"[*] Đang quét trên {interface_name}... Bấm Ctrl+C để dừng.")

    try:
        for packet in capture.sniff_continuously(packet_count=10):
            if 'IP' in packet:
                src = packet.ip.src
                dst = packet.ip.dst
                proto = packet.transport_layer
                print(f"[+] {proto}: {src} -> {dst}")
    except KeyboardInterrupt:
        print("\n[*] Đã dừng.")

if __name__ == "__main__":
    live_capture_basic('eth0')

Lọc dữ liệu thông minh với BPF Filter

Đừng bao giờ bắt tất cả traffic nếu bạn chỉ cần soi một dịch vụ nhất định. Việc này gây lãng phí CPU và RAM nghiêm trọng. Hãy tận dụng BPF (Berkeley Packet Filter) để lọc ngay từ tầng kernel.

# Chỉ bắt traffic HTTP (port 80) từ IP cụ thể
capture = pyshark.LiveCapture(
    interface='eth0', 
    bpf_filter='tcp port 80 and host 192.168.1.5'
)

Phân tích sâu Layer và Payload

Thứ khiến Pyshark mạnh mẽ là khả năng bóc tách layer. Ví dụ, để lấy URL từ request HTTP, bạn có thể viết như sau:

import pyshark

def analyze_http(interface_name):
    capture = pyshark.LiveCapture(interface=interface_name, display_filter='http')

    for packet in capture.sniff_continuously():
        try:
            if hasattr(packet, 'http'):
                host = packet.http.host
                uri = packet.http.request_uri
                print(f"[HTTP] {packet.http.request_method} {host}{uri}")
                
                if hasattr(packet.http, 'file_data'):
                    print(f" Payload: {packet.http.file_data}")
        except AttributeError:
            continue

analyze_http('eth0')

Khi xử lý payload, dữ liệu thường ở dạng hex hoặc bị mã hóa. Nếu cần trích xuất Token hay API Key bằng Regex, bạn nên kiểm tra kỹ pattern để tránh sai sót. Mình thường dùng regex tester tại toolcraft.app/vi/tools/developer/regex-tester để thử nghiệm nhanh các pattern trước khi đưa vào code chính thức. Việc này giúp tiết kiệm rất nhiều thời gian debug khi script đang chạy live.

Sử dụng Async để tránh rơi gói tin

Nếu xử lý logic quá nặng trong vòng lặp, script sẽ bị nghẽn và làm rơi (drop) các gói tin tiếp theo. Giải pháp là sử dụng AsyncLiveCapture để xử lý bất đồng bộ.

import asyncio
import pyshark

async def capture_packets():
    capture = pyshark.LiveCapture(interface='eth0', display_filter='icmp')
    async for packet in capture.sniff_continuously():
        print(f"Phát hiện ICMP từ: {packet.ip.src}")

loop = asyncio.get_event_loop()
loop.run_until_complete(capture_packets())

Lưu ý quan trọng về hiệu năng

  1. Quyền thực thi: Bắt gói tin yêu cầu quyền root. Hãy dùng sudo dpkg-reconfigure wireshark-common để cho phép user thường chạy TShark mà không cần sudo liên tục.
  2. Quản lý bộ nhớ: Pyshark mặc định lưu giữ gói tin cũ. Nếu chạy script 24/7, hãy cẩn thận kẻo tràn RAM. Luôn dọn dẹp list chứa dữ liệu định kỳ.
  3. BPF vs Display Filter: BPF Filter lọc ở tầng kernel nên cực nhanh. Display Filter lọc sau khi TShark đã nhận gói nên tốn tài nguyên hơn. Hãy ưu tiên BPF bất cứ khi nào có thể.

Kết hợp Pyshark với pandas để thống kê hoặc gửi cảnh báo qua Telegram sẽ giúp bạn có một hệ thống giám sát cực kỳ lợi hại. Hy vọng công cụ này sẽ giúp ích cho anh em trong công việc vận hành hệ thống hàng ngày.

Share:
Tags: