3 Giờ Sáng, Một Cảnh Báo và 50 Kết Nối Không Giải Thích Được
Cảnh báo kích hoạt lúc 3:12 sáng: băng thông chiều ra của một server production đột ngột tăng từ mức bình thường 20 Mbps lên 290 Mbps. SSH vào, chạy netstat -an | grep ESTABLISHED, và đó rồi — hơn 50 kết nối tới các địa chỉ IP hoàn toàn không liên quan đến hệ thống.
Tối hôm đó tôi làm điều mà hầu hết kỹ sư đều làm: mở Wireshark, bắt vài phút lưu lượng và tay click qua hàng nghìn gói tin để tìm một mẫu. Mất 40 phút để xác định nguồn gốc vấn đề. Không phải vì bài toán khó — mà vì công cụ không phù hợp cho công việc này.
Giải pháp là một script Python 150 dòng dùng PyShark. Giờ nó chạy trên mọi node production tôi quản lý. Khi có cảnh báo kích hoạt, tôi nhận được báo cáo được định dạng sẵn trong terminal trong vòng hai phút thay vì bốn mươi.
PyShark Thực Sự Là Gì
PyShark là một wrapper Python cho TShark — engine dòng lệnh cung cấp sức mạnh phân tích gói tin của Wireshark. Khi bạn dùng giao diện GUI của Wireshark, TShark chính là thứ đang thực sự phân tích dữ liệu mạng bên dưới. PyShark tiếp cận engine đó qua một Python API sạch sẽ.
Điểm khác biệt quan trọng so với raw socket hay Scapy: PyShark trả về cho bạn các gói tin đã được phân tích sẵn. Bạn không cần tự giải nén byte để tìm cổng đích. Chỉ cần packet.tcp.dstport. HTTP method? packet.http.request_method. DNS query? packet.dns.qry_name. Toàn bộ hơn 3.000 bộ phân tích giao thức của Wireshark, truy cập được từ Python.
Theo kinh nghiệm thực tế của tôi, đây là một trong những kỹ năng thiết yếu cần nắm vững nếu bạn làm việc với hạ tầng mạng. Khi bạn có thể phân tích lưu lượng theo cách lập trình, bạn không còn phản ứng với sự cố nữa mà bắt đầu phát hiện các mẫu trước khi chúng trở thành sự cố.
PyShark hỗ trợ hai chế độ bắt gói tin:
- LiveCapture — bắt gói tin theo thời gian thực từ một giao diện mạng
- FileCapture — đọc từ file
.pcapcó sẵn, lý tưởng để phân tích sau sự cố khi bạn đã lưu dump trước đó
Xây Dựng Công Cụ Phân Tích Lưu Lượng Tự Động
Cài Đặt Môi Trường
PyShark cần TShark được cài đặt làm backend. Trên Ubuntu/Debian:
# Cài đặt TShark (độc lập, không cần giao diện Wireshark)
sudo apt-get install tshark
# Cho phép bắt gói tin không cần root — thêm user vào nhóm wireshark
sudo usermod -aG wireshark $USER
newgrp wireshark
# Cài đặt PyShark
pip install pysharkTrên CentOS/RHEL: sudo dnf install wireshark-cli. Tên nhóm có thể là wireshark hoặc pcap tùy theo bản phân phối.
Bắt Gói Tin Theo Thời Gian Thực
Bắt đầu với lệnh bắt gói tin đơn giản nhất có ích — chỉ in nguồn/đích cho toàn bộ lưu lượng TCP:
import pyshark
capture = pyshark.LiveCapture(interface='eth0', bpf_filter='tcp')
capture.sniff(timeout=60)
for packet in capture:
try:
src = packet.ip.src
dst = packet.ip.dst
dport = packet.tcp.dstport
print(f"{src} -> {dst}:{dport}")
except AttributeError:
pass # Bỏ qua gói tin không phải IP hoặc bị lỗiTham số bpf_filter='tcp' dùng cú pháp Berkeley Packet Filter — giống như tcpdump. Lọc ngay lúc bắt gói tin giúp giảm đáng kể mức sử dụng bộ nhớ trên các giao diện bận.
Tổng Hợp Kết Nối và Phát Hiện Bất Thường
Output từng gói tin một không phải thứ bạn cần trong lúc xử lý sự cố. Bạn cần tổng hợp: đích nào đang nhận nhiều lưu lượng nhất? Cổng nào? Có IP nào đang nhận số lượng kết nối bất thường cao không?
import pyshark
from collections import defaultdict, Counter
import time
def analyze_traffic(interface='eth0', duration=120, threshold=20):
"""
Bắt lưu lượng trong `duration` giây.
Đánh dấu bất kỳ IP đích nào có hơn `threshold` kết nối.
"""
connection_counts = Counter()
bytes_per_ip = defaultdict(int)
port_counts = Counter()
# Bộ lọc BPF: chỉ TCP chiều ra, loại trừ các subnet nội bộ
bpf = 'tcp and not (dst net 10.0.0.0/8 or dst net 192.168.0.0/16 or dst net 172.16.0.0/12)'
capture = pyshark.LiveCapture(
interface=interface,
bpf_filter=bpf
)
print(f"[*] Đang bắt gói tin trên {interface} trong {duration}s...")
capture.sniff(timeout=duration)
for packet in capture:
try:
dst_ip = packet.ip.dst
dst_port = packet.tcp.dstport
pkt_len = int(packet.length)
connection_counts[dst_ip] += 1
bytes_per_ip[dst_ip] += pkt_len
port_counts[dst_port] += 1
except AttributeError:
continue
anomalies = {
ip: count
for ip, count in connection_counts.items()
if count > threshold
}
return connection_counts, bytes_per_ip, port_counts, anomaliesTạo Báo Cáo Dễ Đọc
Sau khi cửa sổ bắt gói tin đóng lại, định dạng dữ liệu thành thứ bạn thực sự có thể đọc lúc 3 giờ sáng:
def generate_report(connection_counts, bytes_per_ip, port_counts, anomalies):
lines = []
lines.append("=" * 60)
lines.append("BÁO CÁO PHÂN TÍCH LƯU LƯỢNG MẠNG")
lines.append(f"Tạo lúc: {time.strftime('%Y-%m-%d %H:%M:%S')}")
lines.append("=" * 60)
lines.append("\n[TOP 10 IP ĐÍCH THEO SỐ KẾT NỐI]")
for ip, count in connection_counts.most_common(10):
mb = bytes_per_ip[ip] / (1024 * 1024)
lines.append(f" {ip:<20} {count:>6} kết nối {mb:>8.2f} MB")
lines.append("\n[TOP 10 CỔNG ĐÍCH]")
for port, count in port_counts.most_common(10):
lines.append(f" Cổng {port:<8} {count:>6} gói tin")
if anomalies:
lines.append("\n[!! BẤT THƯỜNG — SỐ KẾT NỐI CAO]")
for ip, count in sorted(anomalies.items(), key=lambda x: -x[1]):
lines.append(f" !! {ip} — {count} kết nối")
else:
lines.append("\n[OK] Không có bất thường vượt ngưỡng")
report = "\n".join(lines)
print(report)
fname = f"traffic_report_{time.strftime('%Y%m%d_%H%M%S')}.txt"
with open(fname, 'w') as f:
f.write(report)
print(f"\n[*] Báo cáo đã lưu vào {fname}")
return reportScript Hoàn Chỉnh Có Thể Chạy Ngay
Kết nối tất cả lại với một giao diện CLI:
if __name__ == '__main__':
import argparse
parser = argparse.ArgumentParser(description='Công cụ Phân Tích Lưu Lượng PyShark')
parser.add_argument('--interface', default='eth0')
parser.add_argument('--duration', type=int, default=120)
parser.add_argument('--threshold', type=int, default=20)
args = parser.parse_args()
conn_counts, bytes_ip, port_counts, anomalies = analyze_traffic(
interface=args.interface,
duration=args.duration,
threshold=args.threshold
)
generate_report(conn_counts, bytes_ip, port_counts, anomalies)# Bắt gói tin cơ bản 2 phút trên eth0
sudo python3 traffic_analyzer.py
# Bắt gói tin lâu hơn với ngưỡng bất thường cao hơn
sudo python3 traffic_analyzer.py --interface ens3 --duration 300 --threshold 50Phân Tích File pcap Đã Lưu
Nếu bạn đã bắt pcap trong lúc xảy ra sự cố bằng tcpdump, hãy phân tích sau sự kiện bằng cách thay LiveCapture bằng FileCapture:
def analyze_pcap(filepath):
# display_filter dùng cú pháp Wireshark — mạnh hơn BPF
capture = pyshark.FileCapture(
filepath,
display_filter='tcp and not ip.dst == 192.168.0.0/16'
)
for packet in capture:
try:
ts = packet.sniff_time
src = packet.ip.src
dst = packet.ip.dst
port = packet.tcp.dstport
print(f"{ts} | {src} -> {dst}:{port}")
except AttributeError:
continue
capture.close() # Luôn đóng FileCapture một cách tường minhLưu ý display_filter dùng cú pháp display filter của Wireshark thay vì BPF — bạn có thể viết những thứ như 'http.request.method == "POST"' hoặc 'dns.qry.name contains "evil"'. Đó mới là sức mạnh thực sự của TShark khi làm backend.
Thêm Các Kiểm Tra Cấp Production
Phiên bản cơ bản xử lý được hầu hết sự cố. Để giám sát production, hãy thêm các tính năng sau:
Đánh dấu lưu lượng đến các cổng đáng ngờ đã biết:
SUSPICIOUS_PORTS = {'4444', '6666', '6667', '31337', '1337', '9001', '8080'}
def check_suspicious_ports(port_counts):
hits = {p: c for p, c in port_counts.items() if str(p) in SUSPICIOUS_PORTS}
if hits:
print(f"[!! CẢNH BÁO] Phát hiện lưu lượng trên các cổng đáng ngờ: {hits}")
return hitsGiám sát cửa sổ liên tục — bọc lệnh bắt gói tin trong một vòng lặp, bắt các cửa sổ 5 phút, so sánh mỗi cửa sổ với cửa sổ trước. Một đỉnh tăng 3× từ cửa sổ này sang cửa sổ tiếp theo có ý nghĩa hơn nhiều so với một ngưỡng tuyệt đối.
Cảnh báo qua webhook — khi phát hiện bất thường, POST đến webhook Slack hoặc Telegram. Kết hợp với cron job là bạn đã có giám sát liên tục thụ động mà không cần cả một SIEM stack.
Ngữ cảnh địa lý — thư viện geoip2 ánh xạ IP đến quốc gia bằng cơ sở dữ liệu MaxMind cục bộ. Thấy 80% kết nối đột nhiên định tuyến đến một quốc gia mà người dùng của bạn không sống ở đó là tín hiệu rõ ràng hơn nhiều so với chỉ đếm số kết nối.
Thứ Công Cụ Này Thay Thế (và Thứ Không Thay Thế)
Công cụ này xử lý việc phân loại tự động, so sánh baseline và tạo báo cáo — những phần lặp đi lặp lại của việc điều tra mạng mà không cần người ngồi click qua GUI.
Wireshark vẫn chiếm ưu thế khi bạn cần theo dõi một luồng TCP cụ thể, kiểm tra byte payload hoặc tái tạo phiên HTTP trực quan. Hãy giữ nó được cài đặt. Nhưng cho 10 phút đầu tiên của bất kỳ sự cố mạng nào — khi bạn cần hiểu hình dạng của lưu lượng trước khi đi sâu hơn — một lệnh bắt gói tin PyShark theo script luôn vượt trội hơn lọc thủ công.
Script bắt đầu như một bản vá lúc 3 giờ sáng đó đã chạy trên server của tôi được sáu tháng. Nó đã phát hiện ba mẫu lưu lượng bất thường mà lẽ ra sẽ không ai chú ý cho đến khi chúng chạm đến giới hạn billing hoặc băng thông. Đó là phần thưởng: phân tích tự động không ngủ khi bạn ngủ.
