Xây dựng năng lực phát hiện sớm các hành vi thu thập thông tin (Reconnaissance) như Port Scanning là tối quan trọng đối với bất kỳ hệ thống SOC nào. Tuy nhiên, việc chuyển từ "phát hiện" sang "phản ứng tự động" (Active Response) thường gặp nhiều rào cản kỹ thuật.
Trong Use Case này, tôi sẽ chia sẻ cách chúng tôi cấu hình OpenSearch để thực hiện: Phát hiện distinct port > 5 ports/1 phút từ một IP nguồn ➡ Gửi Alert Telegram đích danh IP ➡ Gọi API Firewall để auto-block IP tấn công.
Dưới đây là nhật ký chi tiết và những bài học xương máu (lessons learned) trong quá trình triển khai.
=> Giải pháp: Sử dụng khối JSON chuẩn:
JSON
{<br> <span>"buckets_path"</span>: {<br> <span>"port_count"</span>: <span>"distinct_ports"</span><br> },<br> <span>"parent_bucket_path"</span>: <span>"ips"</span>, <span>// Cực kỳ quan trọng: Phải khớp với tên aggregation trong Query!</span><br> <span>"script"</span>: {<br> <span>"source"</span>: <span>"params.port_count > 5"</span><br> }<br>}<br>
Thử thách 1: Lỗi "now" kinh điểnChúng tôi không thể nạp log với @timestamp: "now" vì OpenSearch yêu cầu giờ thực tế (ISO format) trong dữ liệu document, dẫn đến lỗi 400 - Bad Request (image_cf1de5.png).
Thử thách 2: Dữ liệu quá "khứ"Chúng tôi chạy lệnh nạp single log (image_14.png) và bulk log 6 ports (image_12.png), nhưng lại gán mốc giờ cố định (ví dụ 08:00 AM) trong paypload. Tuy nhiên, Monitor chỉ quét dữ liệu trong vòng 5 phút qua (now-5m). Sự lệch pha tận 7 tiếng khiến Monitor bỏ qua log test.
Giải pháp cuối cùng: SOC phải tính toán giờ UTC thực tế ngay lúc này (ví dụ 15:05:00) và dán vào lệnh _bulk trong Dev Tools để đẩy lên hệ thống.
(Ghi chú: Mặc dù trong ảnh image_15.png, phần bóc tách biến số dynamic IP tấn công và Số Port đích bị quét chưa hiển thị giá trị thực do lỗi templating Mustache, nhưng tin nhắn nổ về đã chứng minh Monitor đã quét trúng log test, Trigger condition thỏa mãn và Action đã gọi Webhook Telegram thành công. Phần Firewall cũng sẽ nhận được payload tương tự để auto-block).
Trong Use Case này, tôi sẽ chia sẻ cách chúng tôi cấu hình OpenSearch để thực hiện: Phát hiện distinct port > 5 ports/1 phút từ một IP nguồn ➡ Gửi Alert Telegram đích danh IP ➡ Gọi API Firewall để auto-block IP tấn công.
Dưới đây là nhật ký chi tiết và những bài học xương máu (lessons learned) trong quá trình triển khai.
Những Rào Cản Ban Đầu
Ban đầu, chúng tôi cố gắng cấu hình Monitor bằng giao diện kéo thả (Visual Editor) thông thường. Tuy nhiên, các vấn đề lập tức xuất hiện:- Ô Time field trống rỗng: Khi chọn index, hệ thống không nhận diện được trường thời gian (image_1.png). Bài học: Hãy đảm bảo Index đã được mồi dữ liệu thực tế và mapping chuẩn xác trước khi tạo Monitor.
- Thiếu trường để Aggregation: Trong phần Data filter và Group by, các trường quan trọng để đếm (như source.ip, destination.port) bỗng dưng biến mất trong dropdown (image_2.png, image_3.png).
- Lesson Learned: Điều này thường xảy ra khi log được nạp dạng Dynamic Mapping, OpenSearch hiểu trường IP là text thay vì keyword. UI sẽ tự động ẩn các trường text để tránh treo hệ thống khi gom nhóm.
Giải Pháp Kỹ Thuật 3 Bước (Extraction Query & Per-bucket)
Bước 1: Smart Query đếm Distinct Port (Monitor)
Chúng tôi viết một câu truy vấn JSON thực hiện 2 việc cùng lúc: Lọc log trong 5 phút qua có chữ "failed" và gom nhóm theo IP nguồn, sau đó dùng hàm cardinality để đếm số lượng port đích khác biệt.
Bước 2: Chuẩn hóa logic Trigger (Condition)
Đây là bước dễ lỗi nhất. Do Monitor dùng Extraction Query, ô Trigger condition bắt buộc phải tuân theo cấu trúc JSON hoàn chỉnh để định nghĩa đường dẫn aggregation, chứ không được viết code Painless đơn thuần (như kiểu ctx.bucket...). Việc điền text bình thường sẽ dẫn đến lỗi Invalid JSON kinh điển.
=> Giải pháp: Sử dụng khối JSON chuẩn:
JSON
{<br> <span>"buckets_path"</span>: {<br> <span>"port_count"</span>: <span>"distinct_ports"</span><br> },<br> <span>"parent_bucket_path"</span>: <span>"ips"</span>, <span>// Cực kỳ quan trọng: Phải khớp với tên aggregation trong Query!</span><br> <span>"script"</span>: {<br> <span>"source"</span>: <span>"params.port_count > 5"</span><br> }<br>}<br>
Bước 3: Connecting the pipes (Telegram Bot & Firewall Webhook)
Chúng tôi thiết lập 2 Actions đồng thời khi Trigger nổ:- Cảnh báo Telegram: Đầu tiên, cần lấy Chat ID (cá nhân 545376996 qua @RawDataBot - image_11.png), sau đó tạo Kênh thông báo hướng tới Telegram Bot API. Payload tin nhắn được thiết kế để lấy động IP tấn công và số port bị quét.
- Auto-block Firewall: Thêm một Action thứ hai hướng tới Custom Webhook của Firewall API với payload JSON chứa lệnh chặn.
Giả Lập Tấn Công & "Bẫy Thời Gian" (Simulation & Success)
Để kiểm thử, SOC cần "mồi" dữ liệu giả lập (đóng vai kẻ tấn công quét port) qua Dev Tools.Thử thách 1: Lỗi "now" kinh điểnChúng tôi không thể nạp log với @timestamp: "now" vì OpenSearch yêu cầu giờ thực tế (ISO format) trong dữ liệu document, dẫn đến lỗi 400 - Bad Request (image_cf1de5.png).
Thử thách 2: Dữ liệu quá "khứ"Chúng tôi chạy lệnh nạp single log (image_14.png) và bulk log 6 ports (image_12.png), nhưng lại gán mốc giờ cố định (ví dụ 08:00 AM) trong paypload. Tuy nhiên, Monitor chỉ quét dữ liệu trong vòng 5 phút qua (now-5m). Sự lệch pha tận 7 tiếng khiến Monitor bỏ qua log test.
Giải pháp cuối cùng: SOC phải tính toán giờ UTC thực tế ngay lúc này (ví dụ 15:05:00) và dán vào lệnh _bulk trong Dev Tools để đẩy lên hệ thống.
Kết Quả
Ngay sau khi SOC nạp dữ liệu giả lập đúng giờ thực tế, điện thoại chúng tôi lập tức rung lên. Tin nhắn Telegram nổ về đích danh với tiêu đề " PHÁT HIỆN HÀNH VI PORT SCAN "
(Ghi chú: Mặc dù trong ảnh image_15.png, phần bóc tách biến số dynamic IP tấn công và Số Port đích bị quét chưa hiển thị giá trị thực do lỗi templating Mustache, nhưng tin nhắn nổ về đã chứng minh Monitor đã quét trúng log test, Trigger condition thỏa mãn và Action đã gọi Webhook Telegram thành công. Phần Firewall cũng sẽ nhận được payload tương tự để auto-block).