Posted inPython

Phân tích PCAP bằng Scapy: Từ ‘soi’ gói tin đến tự động hóa săn tìm Port Scanning

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

Tại sao nên dùng Scapy thay vì Wireshark?

Sau 6 tháng cày cuốc hệ thống giám sát mạng cho một ISP, mình nhận ra thực tế phũ phàng: Wireshark rất tuyệt để soi lỗi lẻ tẻ. Tuy nhiên, nếu phải xử lý 500 file PCAP mỗi ngày hoặc cần cảnh báo tức thì, Wireshark sẽ khiến bạn kiệt sức. Đó là lúc Python + Scapy tỏa sáng.

Scapy không chỉ là thư viện đọc file đơn thuần. Nó cho phép bạn can thiệp vào từng bit dữ liệu từ tầng L2 đến L4. Dự án của mình ban đầu chỉ mất 200 dòng code để lọc IP. Sau khi thêm các module nhận diện hành vi, con số này lên tới 2000 dòng. Dưới đây là lộ trình giúp bạn làm chủ công cụ này mà không cần tốn hàng tuần tự học.

Quick Start: Đọc file PCAP trong 5 phút

Đầu tiên, hãy cài đặt Scapy vào môi trường ảo (venv) để giữ hệ thống sạch sẽ.

pip install scapy

Hãy thử với một file capture.pcap bất kỳ. Đây là đoạn code ngắn nhất để bạn khám phá nội dung bên trong:

from scapy.all import rdpcap

# Đọc file pcap
packets = rdpcap('capture.pcap')

# Xem tóm tắt 10 gói tin đầu tiên
for packet in packets[:10]:
    print(packet.summary())

Lưu ý quan trọng: Hàm rdpcap sẽ nạp toàn bộ dữ liệu vào RAM. Nếu file PCAP nặng khoảng 1GB, một chiếc laptop 8GB RAM có thể bị đứng máy ngay lập tức. Mình sẽ hướng dẫn cách xử lý file lớn ở phần cuối bài.

Bóc tách gói tin: Đi sâu vào các tầng giao thức

Scapy phân cấp gói tin theo các lớp (layers) giống như mô hình TCP/IP. Bạn có thể truy cập dữ liệu cực nhanh bằng cú pháp packet[Layer].

Trích xuất IP và TCP/UDP

Phần lớn công việc phân tích sẽ xoay quanh IP nguồn, IP đích và Port. Đoạn code dưới đây giúp bạn lọc nhanh các thông tin này:

from scapy.all import rdpcap, IP, TCP

pkts = rdpcap('capture.pcap')
for pkt in pkts:
    if pkt.haslayer(IP):
        src_ip = pkt[IP].src
        dst_ip = pkt[IP].dst
        
        if pkt.haslayer(TCP):
            src_port = pkt[TCP].sport
            dst_port = pkt[TCP].dport
            print(f"[{src_ip}:{src_port}] -> [{dst_ip}:{dst_port}]")

Săn tìm dữ liệu DNS và HTTP

Để lấy thông tin DNS, bạn cần kiểm tra layer DNSQR. Với HTTP, hãy tận dụng module scapy.layers.http để tiết kiệm thời gian parse thủ công.

from scapy.all import *
from scapy.layers.http import HTTPRequest

def process_packet(pkt):
    if pkt.haslayer(DNSQR):
        query = pkt[DNSQR].qname.decode()
        print(f"[DNS] Truy vấn: {query}")

    if pkt.haslayer(HTTPRequest):
        url = pkt[HTTPRequest].Host.decode() + pkt[HTTPRequest].Path.decode()
        print(f"[HTTP] Request: {url}")

# Dùng sniff để xử lý từng gói, giúp tiết kiệm RAM tối đa
sniff(offline='capture.pcap', prn=process_packet, store=0)

Thực chiến: Viết script phát hiện Port Scanning

Port Scanning giống như việc kẻ trộm đi gõ cửa từng căn phòng để tìm sơ hở. Kẻ tấn công thường gửi hàng loạt gói tin TCP SYN tới nhiều port khác nhau.

Logic triển khai: Nếu một IP nguồn gửi gói SYN tới hơn 100 port của một đích đến trong 10 giây, chúng ta sẽ gắn cờ cảnh báo.

from scapy.all import rdpcap, IP, TCP
from collections import defaultdict

def detect_port_scan(pcap_file, threshold=100):
    packets = rdpcap(pcap_file)
    scan_attempts = defaultdict(set)

    for pkt in packets:
        # Kiểm tra gói TCP có cờ SYN (0x02)
        if pkt.haslayer(IP) and pkt.haslayer(TCP) and pkt[TCP].flags == 0x02:
            src = pkt[IP].src
            dst = pkt[IP].dst
            dport = pkt[TCP].dport
            scan_attempts[(src, dst)].add(dport)

    for (src, dst), ports in scan_attempts.items():
        if len(ports) > threshold:
            print(f"[!] Cảnh báo: {src} đang quét {len(ports)} cổng trên {dst}")

detect_port_scan('capture.pcap')

Kinh nghiệm thực tế khi xử lý dữ liệu lớn

Làm việc với file PCAP thực tế khắc nghiệt hơn nhiều so với ví dụ trên Lab. Dưới đây là 3 bí kíp mình rút ra:

  • Quên rdpcap đi: Với file lớn, hãy dùng sniff(offline='file.pcap', store=0). Tham số store=0 ngăn Scapy lưu lại các gói tin đã xử lý, giúp RAM luôn ổn định.
  • Tận dụng BPF Filter: Đừng đọc toàn bộ gói tin nếu bạn chỉ cần traffic web. Dùng filter="tcp port 80 or 443" ngay trong hàm sniff. Cách này giúp tăng tốc độ xử lý lên tới 60% vì Scapy lọc dữ liệu ngay từ tầng kernel.
  • Phòng thủ trước dữ liệu rác: Dữ liệu mạng thực tế thường bị lỗi decode. Luôn bọc các lệnh .decode() trong try-except để script không bị “văng” khi đang chạy dở giữa hàng triệu dòng log.

Làm chủ Scapy mở ra khả năng tự động hóa không giới hạn. Thay vì ngồi click từng dòng trên màn hình, bạn có thể để script tự chạy và gửi báo cáo thẳng về Telegram. Chúc các bạn áp dụng thành công!

Share:
Tags: