利用Logtash进行PII信息脱敏

今年是金融行业大地震,凡是涉及到个人信息以及爬虫的,都会有所影响,更有新政出台来约束公司对个人信息的使用,数据脱敏必然提上议程。

如今各种应用都接入ELK,但凡开发一不留神,这万恶的PII信息就进了ES,去PII信息早已迫在眉睫,咋做呢?

当前情况:

  • 大部分应用直接怼kafka,以json的格式输出日志到kafka,修改应用代价太大

  • logstash消费kafka,在logstash的pipeline中完成日志的清洗,筛选,屏蔽比较合理

  • 进入elasticsearch,还有ingest层,这层也可以做脱敏,但是ingest方面不是太了解

  • kibana查询elasticsearch的时候,也可以选择脱敏展示,目前没找到好的工具

以上就是大致思路,时间紧,任务重,也就logstash的pipeline中添加处理流程,能够快速达成目标。

前日志的大致流程是:

app -> kafka -> logstash -> elasticsearch

专注于logstash部分,目前pipeline大致配置:

input(json-codec) -> filter(mutate) -> output(es)

能够修改识别PII信息以及替换的,就是在filtermutate中,使用gsub进行正则匹配和过滤;但是gsub只能针对多个字段进行处理,而我们日志中,PII信息所在字段并不确定,所以字段这块无法确认,怎么办呢

要是能够先针对整条event日志先进行PII信息处理,然后再进行json解析就好了,这样就无需关心PII信息所在位置了,于是思路如下:

在input中,将codec设置为plain,以纯文本的方式读取日志,然后在filter阶段进行PII信息替换,大致流程如下:

filter(plain-codec) -> mutate(gsub(message字段)) -> json(message字段) -> mutate(remove(message字段)) -> output

但是由于我们并不知道原json文本中是否有message字段,在后续删除原始message字段的时候,可能会将原json中的message字段给误删了,导致日志内容缺失,咋整呢?

祭出我们的字段改名大法,中间步骤如下:

rename(message => _message) -> gsub(_message字段) -> json(message字段) -> remove(_message字段)

只要确保改名后的字段不会出现在日志中即可,如果觉得_message字段不安全,可以使用__________message字段,我就不信哪个开发闲着没事定这么个业务字段。

最终测试使用的pipeline配置如下:

input {
file {
path => "/data/logs/mylogs.in"
codec => "plain"
}
}
filter {
mutate {
rename => ["message", "_message"]
gsub => [
"_message", "(\D[2-9]\d{2})\d{12}(\d{4}\D)", '\1************\2',
"_message", "(\D[1-9][0-9]{2})[0-9]{3}[12][089][0-9]{2}[01][0-9][0123][0-9]([0-9]{3}[0-9Xx]\D)", '\1***********\2',
"_message", "(\D[1-9]\d{2})\d{9}(\d{4}\D)", '\1************\2',
"_message", "(\D1[3456789][0-9]{1})[0-9]{4}([0-9][0-9]{3}\D)", '\1****\2'
]
}
json {
source => "_message"
}
mutate {
remove_field => "_message"
}
}
output {
file {
path => "/data/logs/mylogs.out"
}
}

四条正则忽略,为了提高精准度,分别为19位银行卡号,18位身份证号,16位银行卡号,11位手机号。可能存在误判,具体PII信息以更精准的正则为准,此处仅作为参考。

最终效果:

{"@timestamp":"2019-12-05T06:51:56.787Z","f1":"hahhha","host":"ubuntu","message":"wocaowuqing 188****3492","idcard":"lalla 321***********2276","credit_card":"621************6548","f2":"bababa","@version":"1","credit_card_2":"621************6846","phone":"woqunidaye188****3492","path":"/data/logs/mylogs.in"}

勉强蒙混过关吧。

补充:

由于日志信息的不确定性,所以存在误判的情况,比如timestamp等,会导致原本的number类型变为string类型,致使json日志无法解析.最终采用妥协的方式,将mask从*变为0.

如果有什么建议或者意见,可以联系笔者邮箱: troy.sdu@gmail.com