Logstash でアプリログの個人情報をマスクして Elasticsearch へ連携する

2022年3月7日

Elasticsearch にアプリケーションのログを連携できると Kibana のグラフで可視化できたり、Kibana Query Language (KQL) でログの調査がし易かったりで良い事がある！けれどもセキュリティ的な観点でお客様の個人情報をKibana上に表示できない・・・という事があり、 Logstash のfilter 機能で一部個人情報をマスクして Elasticsearch へ連携した際の手順を備忘録も兼ねて残しておきます。

連携の流れ

今回の案件では Java で作られたサーバサイドのアプリケーションログを Elasticsearch に連携しました。

恐らく一般的な構成だとログ収集を行ってくれる Filebeat を用いて、

input (log file) → Filebeat → Logstash → Elasticsearch

こんな感じで連携すると思います。

しかし上記構成の { Logstash → Elasticsearch } の部分を別部署が管理していて、ここにも個人情報を連携してはならないという事。そこで以下の様な構成でログを連携する事になりました。

input (log file) → Logstash → Filebeat → { Logstash → Elasticsearch }

つまり 1 つ目の Logstash で個人情報のマスクを行いそれを別部署が管理している { Logstash → Elasticsearch } に送るという形です。

Logstash の設定

まずは上記流れの最初の Logstash の設定です。Logstash Pipeline は以下のようになっていてそれぞれの項目をlogstash/conf.d 配下の conf に記載していきます。

やりたい事としては以下のようなアプリログがあったとして個人情報に当たる phoneNumber を上記の Filter でマスクをしたいといった感じです。

/test/app_info.log

[2022/02/02 12:15:56.092] [INFO ] Controller : co.test.api.controller.LoginApiController#loginUsersPost 
[2022/02/02 12:15:56.092] [INFO ] API : Object[][{F000001,class UserLoginAuthDto {
    phoneNumber: 09000009999
    userLoginDto: class UserLoginAuthDto {
        class UserLoginDto {
            type: 01
        }
        testnumber: 20290520021255J
    }
}}]

それを行うための設定は以下のようになります。

logstash/conf.d/logstash.conf

input {
    file {
        mode => "tail"
        path => ["/test/app_info.log"]
        sincedb_path => "/home/logstash/output/sincedb/app_info.log"
        start_position => "beginning"
        codec => plain {
            charset => "UTF-8"
        }
    }
}
filter {
    mutate {
        remove_field => ["path", "@version", "host"]
        gsub => [
            "message", "phoneNumber:.*", "phoneNumber:XXXXX"
        ]
    }  
}
output {
    file {
        path => "/home/logstash/output/test_%{+YYYYMMdd}.log"
    }
}

input 部では読み込み元となるログファイルのパスなどを記載しております。

肝心の filter 部分で行っている事ですが gsub の簡単な正規表現で json の値部分を XXXXX で置き換えるようにしています。また remove_field では不要な項目を消しています。

上記 conf の設定によってフィルタされたログは以下のように /home/logstash/output/test_%{+YYYYMMdd}.log ファイルに記載されます。(YYYYMMdd で日付毎にファイルが分かれるようにしています。)

/logstash/output/test_20220202.log

{"message":"[2022/02/02 12:19:56.092] [INFO ] API : Object[][{F000001,class LoginRequestDto {","@timestamp":"2022-02-02T12:19:56.931Z"}
{"message":"    userLoginDto: class UserLoginAuthDto {","@timestamp":"2022-02-02T12:19:56.931Z"}
{"message":"            type: 01","@timestamp":"2022-02-02T12:19:56.931Z"}
{"message":"        testnumber: 20290520021255J","@timestamp":"2022-02-02T12:19:56.931Z"}
{"message":"[2022/02/02 12:19:56.092] [INFO ] Controller : co.test.api.controller.LoginApiController#loginUsersPost","@timestamp":"2022-02-02T12:19:56.931Z"}
{"message":"    phoneNumber:XXXXX","@timestamp":"2022-02-02T12:19:56.930Z"}
{"message":"        class UserLoginDto {","@timestamp":"2022-02-02T12:19:56.931Z"}
{"message":"        }","@timestamp":"2022-02-02T12:19:56.931Z"}
{"message":"    }","@timestamp":"2022-02-02T12:19:56.931Z"}
{"message":"}}] ","@timestamp":"2022-02-02T12:19:56.931Z"}

上記のファイルには message と @timestamp という項目があり message 部に元のログがフィルタされるものはフィルタされてマッピングされます。phoneNumber も phoneNumber:XXXXX となっています。

注意

注意 1

@timestamp は本来不要なので remove_field に入れて削除しようとしたのですが、そうすると output 部の /output/test_%{+YYYYMMdd}.log ← こちらの処理で日付を取得できないという問題がありました。そこでこの段階では output に含めて、後述の連携先の方の logstash 側で message フィールドだけ別のフィールドにマップして確認できるようにします。

注意 2

上記フィルタ後のログを元のログと見比べるとフィルタ後のログの順番が変わってしまっているのが分かると思います。調べてみると logstash は一部設定で指定しないとログの順番は担保してくれないのが原因でした。(参照)

そこで /logstash/logstash.yml にてそれぞれ以下のように設定を変更します。

pipeline.workers: 1

pipeline.ordered: true

すると以下のように正しい順序にてファイルにログが出力されます。

{"message":"[2022/02/02 12:19:56.092] [INFO ] API : Object[][{F000001,class LoginRequestDto {","@timestamp":"2022-02-02T12:19:56.931Z"}
{"message":"[2022/02/02 12:19:56.092] [INFO ] Controller : co.test.api.controller.LoginApiController#loginUsersPost","@timestamp":"2022-02-02T12:19:56.931Z"}
{"message":"    phoneNumber:XXXXX","@timestamp":"2022-02-02T12:19:56.930Z"}
{"message":"    userLoginDto: class UserLoginAuthDto {","@timestamp":"2022-02-02T12:19:56.931Z"}
{"message":"        class UserLoginDto {","@timestamp":"2022-02-02T12:19:56.931Z"}
{"message":"            type: 01","@timestamp":"2022-02-02T12:19:56.931Z"}
{"message":"        }","@timestamp":"2022-02-02T12:19:56.931Z"}
{"message":"        testnumber: 20290520021255J","@timestamp":"2022-02-02T12:19:56.931Z"}
{"message":"    }","@timestamp":"2022-02-02T12:19:56.931Z"}
{"message":"}}] ","@timestamp":"2022-02-02T12:19:56.931Z"}

Flilebeat の設定

こちらは単純にフィルターされたログのパスを yml ファイルに設定するだけです。

/filebeat/config/test.yml

- type: log
  tags: ["test_tag"]
  enabled: true
  paths:
       - /home/logstash/output/test_*.log

ファイル名はtest_{日付}としているので*を使ってtest_という名前が含まれるファイルを連携対象としています。

また合わせて /filebeat/filebeat.yml の Logstash output に接続先の Logstash の host の情報を追加する必要もあります。

Logstash の設定 2

連携先の Logstash 設定用の conf もこちらで用意して連携先の部署で管理してもらうという流れになっているため、その logstash.conf は以下のように設定しました。( ※ filter 部分のみ )

filter {
    json {
        source => "message"
        target => "json"
    }
    mutate {
        add_field => {
            “test_ap_log" => "%{[json][message]}" 
        }
        remove_field => ["json", "message"]
    }
}

こちらから送る test_20220202.log は json 形式になっているため json{} で json として読み込み mutate{} にて test_ap_log というフィールドを新たに追加してそこに message フィールドをマップしているという流れになります。

これで Kibana 上では test_ap_log というフィールドで個人情報がマスクされたログを確認できるようになります。