ELK栈PCAP文件分析工具选型及PacketBeat目录监听配置问询

Great question—you’re right that the official docs don’t make this super clear, and PacketBeat alone can’t natively monitor a directory for new PCAP files (its -I flag only works with single files or live network interfaces). But we can adapt your ideal workflow with a couple of solid workarounds using other Elastic Stack tools.

Option 1: FileBeat + Script + PacketBeat (Closest to Your Original Vision) #

This setup uses FileBeat to handle directory monitoring, triggers a script to process new PCAPs with PacketBeat (for ECS-aligned JSON output), then sends the normalized data to Logstash.

Step 1: Build a PCAP Processing Script #

Create a script (e.g., process_pcap.sh) that converts new PCAPs to PacketBeat’s structured JSON and archives processed files to avoid reprocessing:

#!/bin/bash
PCAP_FILE=$1
OUTPUT_DIR="/var/pcap_json/"
ARCHIVE_DIR="/var/pcap_archive/"

# Ensure directories exist
mkdir -p $OUTPUT_DIR $ARCHIVE_DIR

# Use PacketBeat to parse PCAP and output JSON
packetbeat -e -I $PCAP_FILE \
  -c /etc/packetbeat/packetbeat.yml \
  -E output.file.path=$OUTPUT_DIR \
  -E output.file.enabled=true \
  -E output.elasticsearch.enabled=false \
  -E output.logstash.enabled=false

# Move processed PCAP to archive
mv $PCAP_FILE $ARCHIVE_DIR

Make it executable: chmod +x process_pcap.sh

Step 2: Monitor the PCAP Directory #

Use a background script with inotifywait to trigger processing when new PCAPs are added (run this as a systemd service for persistence):

#!/bin/bash
PCAP_DIR="/var/pcap_input/"

mkdir -p $PCAP_DIR

# Watch directory for new files
inotifywait -m -e create -e moved_to $PCAP_DIR | while read dir action file; do
  if [[ $file == *.pcap || $file == *.pcapng ]]; then
    echo "Processing new PCAP: $file"
    /path/to/process_pcap.sh "$dir$file"
  fi
done

Step 3: Configure FileBeat to Send JSON to Logstash #

Update filebeat.yml to monitor the JSON output directory and forward data to Logstash:

filebeat.inputs:
- type: log
  enabled: true
  paths:
    - /var/pcap_json/*.json
  json.keys_under_root: true
  json.add_error_key: true

output.logstash:
  hosts: ["your-logstash-host:5044"]

Step 4: Logstash Filtering & ElasticSearch Output #

Set up logstash.conf to clean and route data:

input {
  beats {
    port => 5044
  }
}

filter {
  # Customize filters to fit your needs
  if [source] == "packetbeat" {
    mutate {
      remove_field => ["@version", "agent", "ecs"]
      add_tag => ["pcap_data"]
    }
  }
}

output {
  elasticsearch {
    hosts => ["your-es-host:9200"]
    index => "pcap-packetbeat-%{+YYYY.MM.dd}"
  }
}

Option 2: Logstash Directly (Use TShark Instead of PacketBeat) #

If you prefer skipping PacketBeat, Logstash can watch the PCAP directory directly and use TShark to convert files to JSON. This gives raw packet data that you’ll need to map manually.

Logstash Configuration Example #

input {
  file {
    path => "/var/pcap_input/*.pcap"
    start_position => "beginning"
    sincedb_path => "/dev/null" # Set to a valid path if you want to track processed files
  }
}

filter {
  # Use Ruby to call TShark and parse PCAP into packet-level events
  ruby {
    code => '
      require "open3"
      require "json"
      pcap_path = event.get("path")
      stdout, stderr, status = Open3.capture3("tshark -r #{pcap_path} -T json")
      
      if status.success?
        json_packets = JSON.parse(stdout)
        json_packets.each do |packet|
          # Emit a separate event for each packet
          cloned_event = event.clone
          cloned_event.set("packet_details", packet)
          cloned_event.emit
        end
        # Cancel the original directory-watch event
        event.cancel
      else
        event.set("tshark_error", stderr)
      end
    '
  }

  # Flatten TShark fields for easier Kibana querying
  mutate {
    rename => { "[packet_details][_source][layers][ip][ip_src]" => "source.ip" }
    rename => { "[packet_details][_source][layers][ip][ip_dst]" => "destination.ip" }
    rename => { "[packet_details][_source][layers][tcp][tcp_srcport]" => "source.port" }
  }
}

output {
  elasticsearch {
    hosts => ["your-es-host:9200"]
    index => "pcap-tshark-%{+YYYY.MM.dd}"
  }
}

Kibana Visualization Steps #

Once data is in ElasticSearch:

1. Go to Stack Management > Index Patterns and create a pattern for your PCAP index (e.g., pcap-*).
2. Use the Discover tab to explore fields like source.ip, destination.ip, network.transport, and packet.size.
3. Build visualizations:
   • Bar charts for top communication protocols
   • Pie charts for source/destination IP distributions
   • Line charts for packet volume over time

Key Notes #

• PacketBeat vs TShark: PacketBeat normalizes data to Elastic’s ECS (Elastic Common Schema), making Kibana visualization smoother. TShark provides raw packet data that requires manual field mapping.
• Performance: Processing large PCAP files is resource-heavy—test with small files first, and add resource limits if running on shared infrastructure.

内容的提问来源于stack exchange，提问作者75inchpianist