Drop documents with empty field

Elastic Stack Logstash
1

I'm having difficulty finding the right syntax to drop a document with an empty field. I'm using aggregate maps. Here is my code....

filter {

  aggregate {
    task_id => "%{asset_id}%{vulnerability_id}"
    code =>
    "
      map ['Asset ID'] = event.get('asset_id')
      map ['Vulnerability ID'] = event.get('vulnerability_id')
      map ['Asset MAC Address'] = event.get('mac_address')
      map ['Site Name'] = event.get('sites')
      map ['Asset Name'] = event.get('host_name')
      map ['Asset IP Address'] = event.get('ip_address')
      map ['Asset OS Vendor'] = event.get('os_vendor')
      map ['Asset OS Name'] = event.get('os_name')
      map ['Asset OS Version'] = event.get('os_version')
      map ['Asset OS Family'] = event.get('os_family')
      map ['Asset Scan Credential Status'] = event.get('credential_status')
      map ['Last Assessed for Vulnerabilities'] = event.get('last_assessed_for_vulnerabilities')
      map ['Vulnerability Title'] = event.get('title')
      map ['Vulnerability Severity'] = event.get('severity')
      map ['Vulnerability CVSSv3 Score'] = event.get('cvss_v3_score')
      map ['Vulnerability CVSSv3 Vector'] = event.get('cvss_v3_vector')
      map ['Vulnerability Description'] = event.get('description')
      map ['Vulnerability Advisories'] ||= []
        var_advisory = {'Source' => event.get('source'),'Reference' => event.get('reference')}
        if ! map['Vulnerability Advisories'].include?(var_advisory)
          map['Vulnerability Advisories'] << var_advisory
        end

      map ['Vulnerability Advisories CSV'] ||= []
        var_advisory_csv = {'Source' => event.get('source'),'Reference' => event.get('reference')}
        if ! map['Vulnerability Advisories CSV'].include?(var_advisory_csv)
          map['Vulnerability Advisories CSV'] << var_advisory_csv
        end
      map ['Vulnerability Fix'] = event.get('fix')
      map ['Vulnerability Proof'] = event.get('proof')
      map ['Service Name'] = event.get('service')
      map ['Service Port'] = event.get('port')
      map ['Service Protocol'] = event.get('protocol')
      map ['type'] = event.get('type')
      map ['Vulnerability Test Date'] = event.get('test_date')
      map ['Vulnerable Since'] = event.get('found_date')
      map ['Vulnerability Published Date'] = event.get('date_published')
      event.cancel()
    "
    push_previous_map_as_event => true
    timeout => 3
    }

  if ([Vulnerability ID] == "") {
    drop {}
  }


  mutate {
    convert => {"Vulnerability Advisories CSV" => "string"}
  }
}

I'm trying to drop the document if the "Vulnerability ID" is blank. I've also used the pre-aggregate map field "vulnerability_id" and no dice. Here is my JSON output for the sample event in question...

{
  "_index": "test",
  "_type": "_doc",
  "_id": "TEST-336-%{Vulnerability ID}",
  "_version": 1,
  "_score": 0,
  "_source": {
    "Vulnerability CVSSv3 Score": null,
    "Asset MAC Address": null,
    "Vulnerability Title": null,
    "Asset Name": null,
    "Vulnerability Severity": null,
    "Service Port": null,
    "Vulnerability Published Date": null,
    "Vulnerable Since": null,
    "Vulnerability Advisories": [
      {
        "Source": null,
        "Reference": null
      }
    ],
    "Vulnerability Advisories CSV": [
      "{\"Source\"=>nil, \"Reference\"=>nil}"
    ],
    "Vulnerability Description": null,
    "Vulnerability ID": null,
    "type": "TEST",
    "Asset ID": 336,
    "Asset OS Family": "Linux",
    "Vulnerability CVSSv3 Vector": null,
    "Service Protocol": null,
    "Asset IP Address": "1.2.3.4/32",
    "Service Name": null,
    "@version": "1",
    "Asset Scan Credential Status": "N/A",
    "Asset OS Name": "Linux",
    "Last Assessed for Vulnerabilities": "2020-01-09T06:19:33.142Z",
    "Vulnerability Proof": null,
    "@timestamp": "2020-01-15T15:10:53.820Z",
    "Vulnerability Fix": null,
    "Asset OS Vendor": "CentOS",
    "Asset OS Version": "2.6.9",
    "Site Name": "EDC Staging",
    "Vulnerability Test Date": null
  },

Thank you!

2

You could try doing it in the aggregate filter

if ! event.get('vulnerability_id')
    event.cancel
end
3

Why are you using the aggregate filter? Doesn't seem to me you're aggregating anything.

Why not just use the ruby filter? That way you'll also be consistent in case one of those fields is not present in one of the documents.

Anyway, looking at that JSON response, that Vulnerability ID looks like a null more than an empty string.

4

Hi Fabio,

You are right, this particular event does not benefit from aggregation. However, I have many more that do. This is a vulnerability database and I'm aggregating on CVE and MSKB's for example. Having them present in a single document per IP vulnerability is my goal (and it works). My data source is a postgresql db.

I tried null as well.

5

Thx Badger... I'll give it a try.

6

Badger,

That didn't work. But what I did do is change my SQL statement to filter out that null value and keep the row from being parsed into a document. Probably the most efficient way in the end anyway.

Thanks!

7

This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.