fluentdに個人情報をフィルターするプラグインを作ってみた

#目的、背景
log管理する基盤としては、fluentdはデファクトスタンダードになりつつあります。
複数のサーバーを集約して、分析や調査ができる。
ただし、個人情報を取り扱う案件などでは、収集したlogに個人情報が含まれないようにしないと分析作業に制約が出ます。
（制約とはセキュリティルームからしかできない、在宅からリモートで分析できないなど）
そこで、fluentdのpluginとして、個人情報を除去するプラグインを作りました。

プログラム本体

plugin/filter_keshipan.rb
-> /etc/td-agent/plugin/filter_keshipan.rb

• filterの機能で個人情報が検知された場合は、マスクする。
• /var/log/messageにエラー(マスクされたメッセージを付加して)として出力する。
• チェックする内容

• メールアドレス （正規表現）
• クレジットカード番号 (BINレンジ指定) イシュアシステム内の場合、存在しうるカード番号は、BINレンジが固定されるため誤マッチングを防ぐ
• 氏名（漢字、かな、カタカナ ローマ字）上位2000
• 住所(漢字、主要な地名)
  主要地名
  ■各都道府県において表示される基準地・重要地･主要地一覧表
  https://www.mlit.go.jp/road/sign/sign/annai/6-hyou-timei.htm

プログラム実装

filter_exclude_personal_info.rb

# Fluentd
#
#    Licensed under the Apache License, Version 2.0 (the "License");
#    you may not use this file except in compliance with the License.
#    You may obtain a copy of the License at
#
#        http://www.apache.org/licenses/LICENSE-2.0
#
#    Unless required by applicable law or agreed to in writing, software
#    distributed under the License is distributed on an "AS IS" BASIS,
#    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#    See the License for the specific language governing permissions and
#    limitations under the License.
#

require 'fluent/plugin/filter'
require 'fluent/config/error'
require 'fluent/plugin/string_util'
require 'syslog/logger'
require 'json'

module Fluent::Plugin
  class GrepFilter < Filter
    Fluent::Plugin.register_filter('keshipan', self)
    @@single_byte_dict_condtions=[]
    @@multi_byte_dict_condtions=[]
    @@email_check=false
    @@pan_check=false
    @@bin_regex_condtions=[]
    @@syslog = Syslog::Logger.new 'fluent/keshipan'
    @@syslog.info 'this line will be logged via syslog(3)'
    def initialize
      super
      @@dict_condtions=[]
      log.info "keshipan initialize"
    end

    helpers :record_accessor

    def configure(conf)
      super
      log.info "keshipan configure"
      count=0
      File.open(conf["single_byte_file"], mode = "rt"){|f|
           f.each_line{|line|
	       line.chomp!
	       if line.start_with?("#") == false and line.length>0
	           @@single_byte_dict_condtions.push(line.to_s.downcase.force_encoding('UTF-8'))
                   count+=1
	       end
           }
      }
      log.info @@single_byte_dict_condtions
      log.info "single_byte_file:"+conf["single_byte_file"] + " " +count.to_s
      count=0
      File.open(conf["multi_byte_file"], mode = "rt"){|f|
           f.each_line{|line|
	       line.chomp!
	       if line.start_with?("#") == false and line.length>0
	           @@multi_byte_dict_condtions.push(line.to_s.downcase.force_encoding('UTF-8'))
                   count+=1
	       end
           }
      }
      log.info @@multi_byte_dict_condtions
      log.info "multi_byte_file:"+conf["multi_byte_file"] + " "+count.to_s
      if conf["email_check"] == "true"
	  @@email_check = true
          log.info "email_check is true"
      end
      if conf["pan_check"] == "true"
	  @@pan_check = true
          log.info "pan_check is true"
          count=0
          File.open(conf["bin_file"], mode = "rt"){|f|
               f.each_line{|line|
                   line.chomp!
		   if line.start_with?("#") == false and line.length>0
                       @@bin_regex_condtions.push(line)
		       count+=1
	           end
               }
               log.info @@bin_regex_condtions
          }
          log.info "bin_file:"+conf["bin_file"] + " "+count.to_s
      end
    end
    # ある文字列がマルチバイト文字を含んでいるか
    def has_mb?(str)
        str.bytes do |b|
            return true if  (b & 0b10000000) != 0
        end
        return false
    end

    def filter(tag, time, record)
      personal_info_flg=false
      #ひらがな　カタカナ　漢字　氏名　地名
      if @@single_byte_dict_condtions.size>0 
          match_flg=false
          msg=record["message"].to_s.downcase
	  @@single_byte_dict_condtions.each { |word| 
	      if msg.gsub!(word,'***single byte match***') 
                   match_flg=true
              end
	  }
          if match_flg
	     personal_info_flg=true
	     record["message"]=msg
          end
      end
      #ひらがな　カタカナ　漢字　氏名　地名
      if @@multi_byte_dict_condtions.size>0 
          if has_mb?(record["message"]) 
              match_flg=false
              msg=record["message"].to_s.force_encoding('UTF-8')
              @@multi_byte_dict_condtions.each { |word| 
    	          if msg.gsub!(word,'***multi byte match***') 
                       match_flg=true
                  end
    	      }
              if match_flg
                  personal_info_flg=true
                  record["message"]=msg
              end
          end
      end
      if @@email_check
          msg=record["message"].to_s.downcase
	  if msg.gsub!(/([^@\s]+)@((?:[-a-z0-9]+\.)+[a-z]{2,})/,"***email match***")
	      personal_info_flg=true
	      record["message"]=msg
	  end   
      end   
      if @@pan_check
	  msg=record["message"].to_s
          match_flg=false
	  @@bin_regex_condtions.each { |jouken| 
	      if msg.gsub!(/#{jouken}/,"***pan match****")
                  match_flg=true
	      end   
	  }
          if match_flg
	     personal_info_flg=true
	     record["message"]=msg
          end
      end 
      if personal_info_flg
	  @@syslog.error "KESHIPAN personal info match!! msg:"+JSON.generate(record)
      end
      return record
    end

    Expression = Struct.new(:key, :pattern) do
      def match?(record)
        ::Fluent::StringUtil.match_regexp(pattern, key.call(record).to_s)
      end
    end
  end
end

• 設定ファイルに追記

/etc/td-agent/td-agent.conf

<source>
    @type forward
    port 24224
    bind 0.0.0.0
</source>
<filter *.*>
    @type keshipan
    single_byte_file /etc/td-agent/plugin/keshipan_single_byte.dat
    multi_byte_file /etc/td-agent/plugin/keshipan_multi_byte.dat
    email_check true
    pan_check true
    bin_file /etc/td-agent/plugin/keshipan_bin.dat
</filter>
<match **>
  @type file
  #format single_value
  append true
  # 受信したデータを出力ファイル
  path /var/log/fluentd/syslog.log
  # 日単位に出力
  time_slice_format %Y-%m-%d
  <buffer>
    # バッファタイプはファイル
    @type file
    # バッファファイルの出力先
    path /var/log/fluentd/out/
    flush_mode interval
    flush_interval 3s
  </buffer>
</match>

記述	意味
@type exclude_personal_info	プライグインを指定
single_byte_file	/etc/td-agent/plugin/keshipan_single_byte.dat 個人情報の英字 個人情報の辞書ファイル:フルパス指定：テキストファイルで一行ごとに個人情報を入れる
multi_byte_file	/etc/td-agent/plugin/keshipan_multi_byte.dat 個人情報のマルチバイト、氏名住所 個人情報の辞書ファイル:フルパス指定：テキストファイルで一行ごとに個人情報を入れる
email_check	emailアドレスをチェックする場合true しない場合はfalse
pan_check	クレジットカード番号をチェックする場合true しない場合はfalse
bin_file	/etc/td-agent/plugin/keshipan_bin_xxx.dat 案件別にBINレンジ別に記載

個人情報の辞書ファイル

サンプル

[vagrant@server112 plugin]$ head -10 /etc/td-agent/plugin/keshipan_single_byte.dat
佐藤
鈴木
高橋
田中
伊藤
渡辺
山本
中村
小林
加藤

サンプル

[vagrant@server112 plugin]$ head -10 /etc/td-agent/plugin/keshipan_bin_xxx.dat
34567512[0-9]{8}
123456[0-9]{10}

設定の反映

設定や辞書ファイルを変えた場合は再起動が必要

再起動

systemctl restart td-agent

実行結果

/var/log/fluentd/syslog.log.2021-01-10.log

2021-08-09T00:24:44+00:00       syslog.messages {"message":"Aug  9 00:24:43 server111 vagrant[1219]: HelloWorld***multi byte match***さん0","tailed_path":"/var/log/messages","host":"server111"}
2021-08-09T00:24:44+00:00       syslog.messages {"message":"Aug  9 00:24:43 server111 vagrant[1220]: HelloWorld***multi byte match***さん1","tailed_path":"/var/log/messages","host":"server111"}
2021-08-09T00:24:44+00:00       syslog.messages {"message":"Aug  9 00:24:43 server111 vagrant[1221]: HelloWorld***multi byte match***さん2","tailed_path":"/var/log/messages","host":"server111"}

個人情報が含まれるが含まれるので、検知の上　除外。


#実装して見て
- 頻出苗字を2000位をやってみたが、レスポンス低下は特にみられなかった。秒間100件程度は普通に処理している。
- 辞書ファイルを手厚くしていけば、個人情報を検知の可能性を上げられる。

#今後の課題
 - どのlogの何行目が、辞書ファイルの何行目と引っかかったのかを出力できるように検討