Tshark + Elasticsearch 打造流量回溯分析系統

tshark是網絡分析工具wireshark下的一個工具，主要用於命令行環境進行抓包、分析，尤爲對協議深層解析時，tcpdump難以勝任的場景中。本文將介紹與tshark相關的流量解決方案。

Photo by hao wang on Unsplash

tshark + elastic stack

利用tshark，不只能夠對現有的pcap文件進行分析，因爲能夠輸出其餘格式，也就能夠結合ES的強大搜索能力，達到對數據報文進行記錄、分析處理的能力，能夠實現回溯分析，結合kibana可視化工具，甚至達到實時可視化監控。

elastic stack全家桶

最先ELK全家桶，logstash性能一直被詬病，後來另起爐竈，針對採集使用golang構建出一套beats，用於不一樣的採集場景。其中針對網絡流量，開發出packetbeat。

packetbeat的優點是定製了elasticsearch的mapping、kibana一系列可視化圖表，能夠知足通常對tcp、dns、udp等常規報文的分析。基本達到開箱即用程度。

但packetbeat也有不足，對報文的分析採用會話分析，沒有一個個報文單獨分析，傾向於應用層，對網絡層面分析不足（尤爲是故障排查時），此外，支持的協議有限，僅常見協議與tshark的2000多種存在明顯差距，當遇到不支持時，須要等待支持或手動寫插件，難度極高。

離線導入elasticsearch

tshark支持將pcap報文分析後生成json文件導入elasticsearch，同時支持elasticsearch的批量導入接口_bulk的格式，命令以下：

tshark -r test_trace.pcap -T ek > test_trace.pcap.json

以後能夠將json文件經過curl導入。

curl -s -H 「Content-Type: application/x-ndjson」 -XPOST 「localhost:9200/foo/_bulk」 — data-binary 「@/Users/test-elastic/test_trace.pcap.json」

注：

1. 導入時可能存在導入失敗，因爲_bulk接口對post的文件大小有限制，儘可能不要超過15MB，最好在10MB之內。若是超過，建議使用tshark的輸出條件生成多個json文件，使用curl依次導入。
2. 若是導入失敗，能夠在curl加-v查看提示信息。
3. 默認索引名爲相似 packets-2019-04-23（報文記錄的日期），能夠導入後從新索引

可使用相似如下命令查看導入狀況：

curl ‘http://127.0.0.1:9200/packets-2019-04-23/_search/?size=10&pretty=true'

實時監控方案

主要思路

• 使用tshark實時抓取報文，並啓用過濾策略
• 使用tshark解析捕獲的報文，提取指定的字段並寫入csv文件中，或者使用json格式在下一步進行ETL
• 使用filebeat持續檢測csv文件，併發個logstash用於字段過濾之類（如無需過濾能夠跳過logstash直接發給elasticsearch）
• logstash對字段進行過濾，格式轉化等，以後發到elasticsearch
• 使用kibana進行數據可視化，對報文統計分析

簡單實現

1. tshark部分

tshark -i phy0.mon -t ad -t ad -lT fields -E separator=, -E quote=d   -e _ws.col.Time  -e wlan.fc.type -e wlan.fc.type_subtype -e radiotap.dbm_antsignal -e frame.len -e radiotap.datarate   > tshark.csv
複製代碼

2. filebeat

簡單filebeat.yml配置文件

filebeat.modules:
- module: system
  syslog:
    enabled: false
  auth:
    enabled: true
    var.paths: ["/home/tshark.csv"]
name: test
output.logstash:
  hosts: ["localhost:5044"]
複製代碼

3. logstash

logstash.yml文件，主要分爲：

• 監聽5044端口接收filebeat的數據
• 對數據按照csv格式解析，字段分割
• 對日期處理，轉換格式
• 添加時、分、秒，便於索引
• 對部分字段轉換爲數字格式
• 替換部分字段
• 輸出到elasticsearch

input {
  beats {
    port => 5044
  }
}

csv {
     source => "message"
     columns => [ "col.time","frame.type","frame.subtype","rssi","frame.size","data.rate" ]
   }

date {
     match => [ "col.time", "YYYY-MM-DD HH:mm:ss.SSSSSSSSS" ]
     target => "@timestamp"
   }

mutate {
  add_field => {"[hour]" => "%{+HH}"}
  add_field => {"[minute]" => "%{+mm}"}
  add_field => {"[second]" => "%{+ss}"}

  }

mutate {
     convert => [ "rssi", "integer" ]
     convert => [ "frame.size", "integer" ]
     convert => [ "data.rate", "integer" ]   
     convert => [ "second", "integer" ]  
     convert => [ "minute", "integer" ]  
     convert => [ "hour", "integer" ]  

   }

if[frame.type]=="0"{
   mutate {
     replace => [ "frame.type", "Management" ]
   }}
   if[frame.type]=="1"{
   mutate {
     replace => [ "frame.type", "Control" ]
   }}
   if[frame.type]=="2"{
   mutate {
     replace => [ "frame.type", "Data" ]
   }}

output {
  elasticsearch {
    hosts => ["http://localhost:9200"]
    index => "%{[@metadata][beat]}-%{+YYYY.MM.dd}" 
    document_type => "%{[@metadata][type]}" 
  }
}
複製代碼

4. elasticsearch

能夠預先導入索引定義mapping，這塊能夠查elasticsearch文檔

5. kibana

利用圖表、面板等進行數據可視化，實現監控功能，這塊可根據業務需求進行發揮～

小結

使用tshak的報文解析、數據導出功能，能夠根據需求靈活處理，藉助開源的大數據工具，能夠實現更貼合業務的工具，實現快速對網絡分析、實時監控、故障排查、高級檢索、回溯分析、統計報表等，甚至在部分場景下能夠替代商業的網絡回溯分析系統。

參考

• moloch 網絡流量回溯分析系統
• Tshark Logstash Filebeat Kibana Elasticsearch

未經許可，禁止轉載