Json payload masking using logstash

[Suman_Reddy1 (Suman Reddy)]

Team,

I am a newbie here, we are using ELK stack for log-management. We configured logstash such that it will read log messages from our logs and push them to Elastisearch. We are good till this part, now the issue came when our tollgate system raised that we should not be logging PI data into Elastisearch/Kibana. Now we are trying to mask the data. We are also logging payload as Json into our logs. Now is there a way or a plugin that could mask the specific field in a json.
ex:
"transfer": {
"debitAccountId": "17203389562810665151",
"creditAccountNumber": "1008041411",
"bankCode": "ABCBANK",
"payeeName": "Test",
"amount": "0.05",
"transferType": "INSTANT",
"partyId": "17202989565945410151",
"referenceId": "SCSH0001"
}

now in the above json I only want to mask some fields, such that the json will look something like

"transfer": {
"debitAccountId": "XXXX389562810665151",
"creditAccountNumber": "1008041411",
"bankCode": "XXCBANK",
"payeeName": "Test",
"amount": "0.05",
"transferType": "INSTANT",
"partyId": "1720298956594541XXX",
"referenceId": "XXXX0001"
}

Please help me or suggest a way.

Thanks in advance.

Thanks,
Suman

[magnusbaeck (Magnus Bäck)]

Use a mutate filter and its gsub option.

[Suman_Reddy1 (Suman Reddy)]

Don't mind, but could you please elaborate with an example. I edited my initial post to match my actual requirement

[magnusbaeck (Magnus Bäck)]

mutate {
  gsub => ["[transfer][debitAccountId]", "^....", "XXXX"]
}

[Suman_Reddy1 (Suman Reddy)]

Thanks Magnus.

The json strings are not fixed all times, they are dynamic in nature as we are doing it in API Gateway's. So I came up with a solution which is dumb I think.
I used ruby functions to do it.

Below is the code snippet.

require 'json'
                                maskingkeys = YAML.load_file('./config/pipelines/masking_keys.yml')
                                def iterative(parsedPayload, maskKeys)
                                        parsedPayload.each {
                                                |key,value|
                                                if value.is_a?(Hash)
                                                        iterative(value,maskKeys)
                                                else
                                                        if value.is_a?(Array)
                                                                value.each{
                                                                        |x|
                                                                        if x.is_a?(Hash)
                                                                                 iterative(x,maskKeys)
                                                                        else
                                                                                if maskKeys.include?(key)
                                                                                        length = value.to_s.length/2
                                                                                        value = value.to_s.gsub!(/.(?=.{#{length}})/,'X')
                                                                                end
                                                                        end
                                                                }
                                                        else
                                                                if maskKeys.include?(key)
                                                                   length = value.to_s.length/2
                                                                   value = value.to_s.gsub!(/.(?=.{#{length}})/,'X')
                                                                end
                                                        end
                                                end
                                        }
                                end
                                payload = event.get('payload')
                                parsedPayload = JSON.parse(payload)
                                iterative(parsedPayload,maskingkeys['json']['keys'])
                                event.set('message',event.get('message').gsub(event.get('payload'),JSON.generate(parsedPayload)))

Using recursion way of finding fields and values.

As the JSON size is increasing like 4K+ lines this solution is consuming CPU. Could you please take a look at my ruby function and provide your feedback.

Thanks