パケットデータ(pcap)をフィルタしたり集約してJSON出力するツールを作ってみる

#はじめに

[Wireshark](https://www.wireshark.org/)のコマンド版である[TShark](https://www.wireshark.org/docs/man-pages/tshark.html)にはパケットキャプチャデータ(PCAPファイル)をJSON形式で出力するオプション(**-T json**)があります。パケットデータをテキストとして扱えるので非常に便利です。

また類似のオプションとして「**-T ek**」があり[Elasticsearch](https://www.elastic.co/jp/elasticsearch)の[Bulk API](https://www.elastic.co/guide/en/elasticsearch/reference/current/docs-bulk.html)で利用可能な形式で出力します。この機能を使ってElasticsearchへエキスポートし、例えば他の通信ログと対応するパケットデータを横串検索したりなどと活用できます。

この際パケットデータを特定のフィルタルールで絞り込んだり、複数パケットを単一フローへ集約することができるとElasticsearchへ格納するログ量を減らせて便利です。さらに[Elastic Common Schema (ECS)](https://github.com/elastic/ecs)に対応させたり、[Community ID Flow Hashing](https://github.com/corelight/community-id-spec)のフィールド値を追加すると他のログとのピボットがしやすくなります。

これまでそのたびにpythonスクリプトでマッチルールを書いたり、Bulk APIを使うにしても出力されたJSONデータを分割していたりしていたのですが（一度に多すぎる数のJSONデータをBulkPutするとElasticsearchのキャパ超えでエラーになる場合があります・・・）、そういった機能を簡単なツール（tsharkコマンドのラッパーツール）として実装してみました。今回は（持ち運びしやすい？ので）[Go](https://golang.org/)を使って実装してみます。

実装したツール(version 0.1)は[こちら(github)](https://github.com/rhpenguin/tshark-filter)へ置いておきました。

#実装機能の概要
* 出力するパケットフィールド名を指定可能
* パケットフィールド値をフィルタルールで絞り込み可能 (ANDed/ORed, exact/case_ignore/exists/regex)
* [Elastic Common Schema (ECS)](https://github.com/elastic/ecs)のフィールドを自動生成
* [Community ID Flow Hashing](https://github.com/corelight/community-id-spec)のフィールドを自動追加
* 同一フローの全パケットを単一ログへ集約（Community IDを利用）
* フローへ集約された各パケットフィールドの差分を抽出可能

#出力形式
1. 標準出力 (Stdout)
2. [Elasticsearch](https://www.elastic.co/jp/elasticsearch)へエキスポート：Bulk APIを利用。また認証およびTLSもサポートしてみます。

###出力例

以下はHTTPS(TLS)の[Server Name Indication](https://ja.wikipedia.org/wiki/Server_Name_Indication)を絞り込み出力する例です。

```json-doc:
[
...
{
  "@timestamp": "2020-03-06T12:30:57.813Z",
  "agent": {
    "type": "tshark-filter"
  },
  "destination": {
    "domain": "192.168.0.1",
    "ip": "192.168.0.1",
    "port": 443
  },
  "event": {
    "created": "2020-03-22T16:20:32.000Z",
    "module": "pcap",
    "type": "protocol"
  },
  "file": {
    "name": "examples/http1.pcapng"
  },
  "network": {
    "bytes": 251,
    "community_id": "1:9Us4XWwcOkDl1xAE6xlxxxxxxxx=",
    "iana_number": "6",
    "packets": 1,
    "protocol": "tls",
    "related": {
      "ip": [
        "192.168.1.1",
        "192.168.0.1"
      ]
    },
    "transport": "tcp",
    "type": "ipv4"
  },
  "packet": {
    "frame_interface_name": "ens33",
    "frame_len": "251",
    "frame_number": "2422",
    "frame_protocols": "eth:ethertype:ip:tcp:tls",
    "frame_time_delta": "0.002324240",
    "frame_time_epoch": "1583757057.813936138",
    "frame_time_relative": "62.307191989",
    "ip_dst": "192.168.0.1",
    "ip_dst_host": "192.168.0.1",
    "ip_len": "237",
    "ip_proto": "6",
    "ip_src": "192.168.1.1",
    "ip_src_host": "192.168.1.1",
    "ip_version": "4",
    "tcp_ack": "1",
    "tcp_dstport": "443",
    "tcp_flags": "0x00000018",
    "tcp_len": "197",
    "tcp_nxtseq": "198",
    "tcp_seq": "1",
    "tcp_srcport": "49907",
    "tls_handshake_extensions_server_name": "login.example.com" ★★★
  },
  "source": {
    "domain": "192.168.1.1",
    "ip": "192.168.1.1",
    "port": 49907
  }
},
...
]
```
※ TShark 2.6.8では「ssl_handshake_extensions_server_name」ですが、TShark **3.2.2**では「tls_handshake_extensions_server_name」とフィールド名が変更になったようです。ご注意ください。

#コマンド形式

```text
tshark-filter --pcap <pcap_file_path> --config <config_yml_path> --action <action_name> --output <output_name>
```

|コマンド引数|説明|
|---|---|
|--pcap pcap_file_path|PCAPファイル名（パケットキャプチャデータ）|
|--config config_yml_path|フィルタや集約ルール、Elasticsearchのアドレスや認証情報などを記述する設定ファイル名|
|--action action_name|操作指定。filter（フィルタ）またはagg（集約）を指定可能|
|--output output_name|出力指定。stdout（標準出力）またはelasticsearch（Elasticsearchへエキスポート）を指定可能|

```text:例：パケットデータをフィルタして標準出力
> ./tshark-filter --config ./default.yml --pcap http.pcapng --action filter --output stdout --pretty true
```

```text:例：パケットデータをフロー毎(Community-ID)に集約して標準出力
> ./tshark-filter --config ./default.yml --pcap http.pcapng --action agg --output stdout --pretty true
```

```text:例：パケットデータをフィルタしてElasticseachへエキスポート
> ./tshark-filter --config ./default.yml --pcap http.pcapng --action filter --output elasticsearch
```

#インストール

###依存するコンポーネント
* [TShark](https://www.wireshark.org/docs/man-pages/tshark.html)
* [elastic/go-elasticsearch for 7.x](https://github.com/elastic/go-elasticsearch)
* [adriansr/flowhash](https://github.com/adriansr/flowhash)

###ビルド方法
以下はLinux (e.g. Ubuntu 18.04) でのビルド手順です。Goで実装しているのでWindowsやMacでもビルドできると思います（たぶん）。

```text
1. cd $GOPATH/src
2. go get -u github.com/adriansr/flowhash
3. cd $GOPATH/src/github.com
4. mkdir elastic (if needed)
5. cd $GOPATH/src/github.com/elastic
6. git clone -b 7.x https://github.com/elastic/go-elasticsearch.git
7. cd $GOPATH/src/
8. git clone https://github.com/rhpenguin/tshark-filter.git
9. cd $GOPATH/src/tshark-filter/tshark-filter
10. go build
```
ビルドに成功すると「tshark-filter」というコマンドが生成されます。

#設定ファイル
フィルタルールや集約ルール、ElasticseachのURLや認証情報などを指定する設定ファイルはYAML形式で記述します。ここではそれを**default.yml**としています。

```yaml:default.yml
#
# 抽出するパケットフィールド名を指定。
# 指定されたフィールド名は実際にはtsharkコマンドの「-T ek -e <フィールド名>...」
# として渡されるためその形式で指定する。
# それらはtsharkコマンドを-T jsonまたは-T ekを指定して出力することで確認できる。
# 
# ただし、いくつかのフィールド(Frame情報、IP・TCP・UDPヘッダ情報)はデフォルトで
# 出力される（上述の出力例を参照）。
#
pcap_extracted_fields:
  - http.request.method
  - http.request.uri

#
# tsharkコマンドで出力されたフィールドへ適用するフィルタルールを指定。
# 以下の例ではhttp.request.methodフィールドの値が「GET」へ一致するものだけへ絞り
# 込まれる。
#
# ただし「tshark -T ek」の仕様によりフィールド名内の「.」が「_」へ置換される模様
# のため注意。
# （例）http.request.method => http_request_method
#  
pcap_field_conditions:
  - http_request_method: "GET"
```

###フィルタルール(pcap_field_conditions)のバリエーション

#### 1. マッチさせるフィールド値を複数指定 (exactマッチ)

```yaml:default.yml
pcap_extracted_fields:
  - http.host
  - http.request.method
  - http.request.uri
pcap_field_conditions:
  #
  # 配列で指定するとORedでマッチされる
  #
  # (http_request_method == "GET" OR http_request_method == "POST")
  # のパケットへマッチ
  # 
  - http_request_method: ["GET", "POST"]
```
または

```yaml:default.yml
pcap_extracted_fields:
  - http.host
  - http.request.method
  - http.request.uri
pcap_field_conditions:
  - http_request_method: 
    - "GET"
    - "POST"
```
 
#### 2. マッチさせるフィールド値をANDedで複数指定 (exactマッチ)

```yaml:default.yml
pcap_extracted_fields:
  - http.host
  - http.request.method
  - http.request.uri
pcap_field_conditions:
  #
  # オブジェクト型で複数フィールド指定するとANDedでマッチ
  #
  # (http_request_method == "GET" AND http_host == "www.example.com")
  # のパケットへマッチ
  #
  - http_request_method: "GET"
    http_host: "www.example.com"
```

#### 3. マッチさせるフィールド値をORedで複数指定 (exactマッチ)

```yaml:default.yml
pcap_extracted_fields:
  - http.host
  - http.request.method
  - http.request.uri
pcap_field_conditions:
  #
  # 配列型で複数フィールド指定するとORedでマッチ
  #
  # (http_request_method == "GET" OR http_host == "www.example.com")
  # のパケットへマッチ
  #
  - http_request_method: "GET"
  - http_host: "www.example.com"
```

#### 4. ANDedマッチさせる複数フィールドルールをORedで複数指定 (exactマッチ)

```yaml:default.yml
pcap_extracted_fields:
  - http.host
  - http.request.method
  - http.request.uri
pcap_field_conditions:
  #
  # オブジェクト型ルールを配列型で指定
  #
  # (http_request_method == "GET" AND http_host == "www1.example.com")
  #  OR
  # (http_request_method == "POST" AND http_host == "www2.example.com")
  # のルールでパケットへマッチ
  #
  - http_request_method: "GET"
    http_host: "www1.example.com"
  - http_request_method: "POST"
    http_host: "www2.example.com"
```

#### 5. exactマッチフィルタルールの冗長記述

```yaml:default.yml
pcap_extracted_fields:
  - http.host
  - http.request.method
  - http.request.uri
pcap_field_conditions:
  #
  # オブジェクト型で複数フィールド指定するとANDedでマッチ
  #
  # match: <マッチタイプ>
  # value: <マッチさせる値またはパターン>
  #
  # (http_request_method == "GET" OR http_request_method == "POST") AND http_host == "www.example.com"
  # のパケットへexactマッチ
  #
  - http_request_method:
      match: exact
      value: ["GET", "POST"]
    http_host: 
      match: exact
      value: "www.example.com"
    
```

#### 6. case_ignoreマッチのフィルタルール

```yaml:default.yml
pcap_extracted_fields:
  - http.host
  - http.request.method
  - http.request.uri
pcap_field_conditions:
  #
  # match: case_ignore 
  #  => 大文字小文字を無視してマッチさせる 
  #
  # オブジェクト型で複数フィールド指定するとANDedでマッチ
  #
  # (http_request_method == "GET" OR http_request_method == "POST") AND http_host == "www.example.com"
  # のパケットデータへ絞り込み
  #
  - http_request_method:
      match: case_ignore
      value: ["get", "post"]
    http_host: 
      match: case_ignore
      value: "WWW.EXAMPLE.COM"
    
```

#### 7. フィールド存在のフィルタルール

```yaml:default.yml
pcap_extracted_fields:
  - http.host
  - http.request.method
  - http.request.uri
pcap_field_conditions:
  #
  # match: exists
  #  => パケットデータに指定されたフィールドが存在する場合にマッチ 
  #
  # http_request_methodフィールドが存在するパケットデータのみへ絞り込み
  #
  - http_request_method:
      match: exists
    
```

#### 8. 正規表現によるフィルタルール

```yaml:default.yml
pcap_extracted_fields:
  - http.host
  - http.request.method
  - http.request.uri
pcap_field_conditions:
  #
  # match: regex
  # value: パターン文字列
  #  => パケットデータに指定されたフィールド値を正規表現マッチ 
  #
  # http_hostフィールドが".*\\.microsoft\\.com"または".*\\.facebook\\.com"へ
  # マッチするパケットデータのみへ絞り込み
  #
  - http_host:
      match: regex
      value:
        - ".*\\.microsoft\\.com"
        - ".*\\.facebook\\.com"    
```

または

```yaml:default.yml
pcap_extracted_fields:
  - http.host
  - http.request.method
  - http.request.uri
pcap_field_conditions:
  - http_host:
      match: regex
      value: [".*\\.microsoft\\.com", ".*\\.facebook\\.com"]
```

###パケットフィールドのマッチタイプ（まとめ）

|マッチタイプ|説明|
|---|---|
|exact|完全一致|
|case_ignore|大文字と小文字を区別しない一致|
|exists|該当の名前のフィールドがパケットデータに存在する場合にマッチ|
|regex|正規表現マッチ|

###集約ルールの設定

フィルタ処理されたパケットデータに対してさらに集約ルールを使うことで[Community ID Flow Hashing](https://github.com/corelight/community-id-spec)により複数パケットを単一のイベントログへまとめる（サマリー化する）こともできます。

これにより個々のパケット毎ではなく必要なフィールドだけを含んだ単一ログ（フローログ）として集約（サマリー化）されるため、ログ量を減らしたりまたログをピボットしたりサマライズすることが簡単になります。

```yaml:default.yml
pcap_extracted_fields:
  - http.host
  - http.request.method
  - tls.handshake.extensions_server_name
pcap_field_conditions:
  - tcp_dstport: [80, 443, 3128]
  - tcp_srcport: [80, 443, 3128]

#
# 集約ルールの記述
#
agg:
  #
  # 集約タイプ名。現状はCommunity IDのみサポート。値が一致するパケットを同一フローとして集約。
  # 
  type: "community_id"
  #
  # 抽出されたパケットーフィールドのうち重複しない値を列挙する指定
  #
  extracted_result_fields:
    - http_request_method
    - http_host
    - tls_handshake_extensions_server_name
    # SYN, ACK, FIN, RSTなど
    - tcp_flags
    - frame_protocols

#
# Elasticseach設定（後述）
#
elasticsearch:
  address: http://log.example.com:9200
  document_id: "community_id"
  bulk: 128
```

```text:コマンド実行例
> ./tshark-filter --config ./default.yml --pcap http.pcapng --action agg --output elasticsearch
```

以下の例では同一Community IDを持つ12パケットが単一フローのログへ集約されています。

```json:Elasticsearchへの出力例
{
  "_index": "tshark-filter-2020-03-09",
  "_type": "_doc",
  "_id": "27140429a5bd05a708ea719e80a8aa33fa01ec73dfb4f908bca37cb0dc83bf46",
  "_version": 1,
  "_score": null,
  "_source": {
    "@timestamp": "2020-03-09T12:31:12.580Z",
    "agent": {
      "type": "tshark-filter"
    },
    "destination": {
      "domain": "192.168.0.1",
      "ip": "192.168.0.1",
      "port": 80
    },
    "event": {
      "created": "2020-03-23T15:37:42.000Z",
      "module": "pcap",
      "type": "protocol"
    },
    "file": {
      "name": "examples/http1.pcapng"
    },
    "network": {
      # 集約された同一フロー内の全パケットサイズ合計
      "bytes": 967,
      # Community ID Flow Hash
      "community_id": "1:3hslQBzXeSXhqePQ1V6xxxxxxxx=",
      "iana_number": "6",
      # 集約された同一フロー内の全パケット数
      "packets": 12,
      "related": {
        "ip": [
          "192.168.2.150",
          "192.168.0.1"
        ]
      },
      "transport": "tcp",
      "type": "ipv4"
    },
    "packet": {
      "frame_interface_name": "ens33",
      "frame_len": "74",
      "frame_number": "3108",
      # フィールド値が重複しないものを列挙
      "frame_protocols": [
        "eth:ethertype:ip:tcp",
        "eth:ethertype:ip:tcp:http"
      ],
      "frame_time_delta": "0.965645454",
      "frame_time_epoch": "1583757072.580788038",
      "frame_time_relative": "77.074043889",
      # フィールド値が重複しないものを列挙
      "http_host": "www.example.com",
      "http_request_method": "GET",
      "ip_dst": "192.168.0.1",
      "ip_dst_host": "192.168.0.1",
      "ip_len": "60",
      "ip_proto": "6",
      "ip_src": "192.168.2.150",
      "ip_src_host": "192.168.2.150",
      "ip_version": "4",
      "tcp_ack": "0",
      "tcp_dstport": "80",
      # フィールド値が重複しないものを列挙
      "tcp_flags": [
        "0x00000002",
        "0x00000012",
        "0x00000010",
        "0x00000018",
        "0x00000019",
        "0x00000011"
      ],
      "tcp_len": "0",
      "tcp_nxtseq": "1",
      "tcp_seq": "0",
      "tcp_srcport": "50564"
    },
    "source": {
      "domain": "192.168.2.150",
      "ip": "192.168.2.150",
      "port": 50564
    }
  },
  "fields": {
    "@timestamp": [
      "2020-03-09T12:31:12.580Z"
    ],
    "event.created": [
      "2020-03-23T15:37:42.000Z"
    ]
  },
  "sort": [
    1583757072580
  ]
}
```

###Elasticsearchへのエキスポート設定

```yaml:default.yml
pcap_extracted_fields:
  - http.host
  - http.request.method
  - http.request.uri
pcap_field_conditions:
  - http_host:
      match: regex
      value: [".*\\.microsoft\\.com", ".*\\.facebook\\.com"]
#
# Elasticseach設定
#
elasticsearch:
  #
  # ElashticseachのURL
  #
  address: http://log.example.com:9200
  #
  # Document IDの生成方法
  #  - "auto" : Elasticseachが生成
  #  - "community_id" : 「frame_time_epoch + Community ID」のハッシュ値を利用。
  #                      => 同一パケットやフローデータを投入しても重複を避けることができる。
  #
  document_id: "community_id"
  #
  # 一度のBulkリクエストへ含める最大ログ数（JSONドキュメント数）
  #
  bulk: 128
```

Elasticseachとの通信に認証およびTLSを利用する場合には以下の通り。

```yaml:default.yml(認証＋TLS)
pcap_extracted_fields:
  - http.host
  - http.request.method
  - http.request.uri
pcap_field_conditions:
  - http_host:
      match: regex
      value: [".*\\.microsoft\\.com", ".*\\.facebook\\.com"]
elasticsearch:
  #
  # ElashticseachのURL (HTTPS)
  #
  address: https://log.example.com:9200
  #
  # 認証ユーザー名とパスワード
  #
  user_name: elastic
  password: elastic
  #
  # CA証明書（PEM形式）
  #
  ssl_ca_certificate: "./certs/ca/ca.crt"
  #
  # trueを指定すると証明書の検証をスキップ。
  #
  ssl_verification_disabled: false
  document_id: "community_id"
  bulk: 128
```

###その他の設定

```yaml:default.yml

#
# デフォルトの出力指定。記述することでコマンド実行で「--output」を省略できる。
#
# stdout または elasticsearch
#
#default_output: "elasticsearch"

#
# デフォルトの操作指定。記述することでコマンド実行で「--action」を省略できる。
#
# filter または agg
#
#default_action: "filter"

#
# フィールド値の最大サイズ。このサイズを超えるデータは切り捨てられます。
# （デフォルト(0)は全データを出力）
#
#max_field_len: 1024

#
# TSHarkの設定
#
#tshark:
  #
  # EXEのパス
  #
  #exe: "/usr/bin/tshark"
  #
  # tsharkへ渡す追加の引数がある場合には列挙。
  # 
  #args:
    #- "-d"
    #- "tcp.port==3128,http"
```

#設定ファイルのサンプル

|ファイル名|説明|
|---|---|
|[http.yml](https://github.com/rhpenguin/tshark-filter/blob/master/tshark-filter/examples/http.yml)|HTTPパケットをフィルタするサンプル設定ファイル|
|[agg_by_community_id.yml](https://github.com/rhpenguin/tshark-filter/blob/master/tshark-filter/examples/agg_by_community_id.yml) |Community IDによって単一フローへ集約するサンプル設定ファイル|
|[elasticsearch.yml](https://github.com/rhpenguin/tshark-filter/blob/master/tshark-filter/examples/elasticsearch.yml)|Elasticseachへエキスポートするサンプル設定ファイル|
|[download_winexe.yml](https://github.com/rhpenguin/tshark-filter/blob/master/tshark-filter/examples/download_winexe.yml)|Windows EXEファイルをダウンロードするHTTPパケットを抽出するサンプル設定。tcp_reassembled_dataフィールド値に「\r\n\r\nMZ」というパターンマッチするフィルタルールが定義されています|
|[CVE_2020_0796.yml](https://github.com/rhpenguin/tshark-filter/blob/master/tshark-filter/examples/CVE_2020_0796.yml)|最近話題の[Microsoft SMBv3 Wormable Vulnerability(CVE-2020-0796)](https://www.jpcert.or.jp/at/2020/at200011.html)のシグネチャにマッチするフィルタルールが定義されています。TShark 3.2.2でテスト済み|


#まとめ
tsharkコマンドのJSON出力を加工する簡単なラッパーツールを実装してみました。

Linux以外での動作実績レポートやバグレポート、その他PRなどありましたらこちらのコメント欄ではなく[github](https://github.com/rhpenguin/tshark-filter)の方へ送っていただけると助かります。