Elasticsearch ingest pipeline - not extract data passed from filebeat

samiusan · July 19, 2021, 9:34am

Hi,

I'm slowly teaching myself the Elastic stack. Current project is attempting to ingest and modelling alerts from snort3 against the elastic common schema. I've run into an issue where an ingest pipeline is not correctly extracting fields out of a json file.

Approach being taken is: filebeat (reading alerts_json.txt file) -> elasticsearch (index template and ingestion pipeline defined).

A snippet from my index template is below:

PUT /_index_template/snort3_template 
{
  "index_patterns": [
    "snort3-*"
  ],
  "template": {
    "mappings": {
      "properties": {
        "destination": {
          "properties": {
            "address": {
              "type": "keyword"
            },
            "ip": {
              "type": "ip"
            },
            "mac": {
              "type": "keyword"
            }
          }
        }
}

A snippet from my ingestion pipeline is below:

PUT /_ingest/pipeline/snort-json-pipeline
{
  "description": "Pipeline for ingesting JSON snort3 data",
  "processors": [
    {
      "convert": {
        "field": "pkt_num",
        "type": "integer",
        "ignore_missing": true
      }
    },
    {
      "convert": {
        "field": "pkt_len",
        "type": "integer",
        "ignore_missing": true
      }
    },
    {
      "convert": {
        "field": "src_port",
        "type": "integer",
        "ignore_missing": true
      }
    },
    {
      "rename": {
        "field": "src_port",
        "target_field": "source.port",
        "ignore_missing": true
      }
    },
    {
      "convert": {
        "field": "dst_port",
        "type": "integer",
        "ignore_missing": true
      }
    },
    {
      "convert": {
        "field": "priority",
        "type": "integer",
        "ignore_missing": true
      }
    },
    {
      "rename": {
        "field": "src_addr",
        "target_field": "source.address",
        "if": "ctx._source?.src_addr != '' && ctx._source?.src_addr !=null"
      }
    },

If I simulate the ingestion pipeline I get the expected result:

POST /_ingest/pipeline/snort-json-pipeline/_simulate
{
  "docs": [{"_index":"index","_id":"id","_source":{ "seconds" : 1626683982, "action" : "allow", "class" : "none", "b64_data" : "AAAAAAAAAAAAAAAAAAAAAAAA", "dir" : "UNK", "dst_ap" : ":0", "eth_dst" : "F0:6E:0B:0F:7A:E4", "eth_len" : 60, "eth_src" : "F0:9F:C2:C7:69:1C", "eth_type" : "0x806", "gid" : 112, "iface" : "ens161", "msg" : "(arp_spoof) unicast ARP request", "pkt_gen" : "raw", "pkt_len" : 18, "pkt_num" : 694811210, "priority" : 3, "proto" : "ARP", "rev" : 1, "rule" : "112:1:1", "service" : "unknown", "sid" : 1, "src_ap" : ":0", "vlan" : 0, "timestamp" : "07/19-18:39:42.586277" }}]
}

If I ingest the data via filebeat it does not extract any of the fields from the json message. It will add all of the fields I add via append in the ingest pipeline.

Am stuck at the minute - trying to work out why the field extraction is not working correctly.

cheers,

Michael

spinscale · July 19, 2021, 9:54am

Hey,

have you specified the pipeline in your filebeat configuration? As it is not specified as a default pipeline in the template, you need to specify somewhere that you would like to make use of that pipeline.

--Alex

legoguy1000 · July 19, 2021, 10:18am

You need to set output.elasticsearch.pipeline: snort-json-pipeline in Filebeat. Can you post your config?

samiusan · July 19, 2021, 10:22am

Hi Alex,

The pipeline is being applied correctly. Below is what I have at the bottom of my index template.

    "settings": {
      "default_pipeline": "snort-json-pipeline"
    }

The approach I'm trying is filebeat sends the logs to an index which is matched by the index template. Elasticsearch then creates a index using the template and dynamically assigns the additional fields not mentioned in the template. As part of setting up the template the default ingestion pipeline is applied to the index.

I am adding a few fields to the indexed document via append statements inside the ingestion pipeline. This is working correctly so I know the ingestion pipeline is correctly applied to the index.

Current theory is filebeat is sending the data with extra fields. Due to this the processors can't find the right fields in the data I'm ingesting.

What is the best way of seeing an event as sent by filebeat prior to processing via the ingestion pipline? For example - if the passed value stores its data in nested objects how can I access those from the ingestion pipeline statement? I have tried things like message.pkt_num and that hasn't worked.

    {
      "convert": {
        "field": "pkt_num",
        "type": "integer",
        "ignore_missing": true
      }
    },

legoguy1000 · July 19, 2021, 10:26am

You can output to a file or stdout from filebeat or run in debugging more and it will output the events sent to elasticsearch in the log. Also do I understand, the data is making it to the ingest pipeline but it's just not processing properly? If so what processors specifically aren't working?

spinscale · July 19, 2021, 11:10am

Can you share the index settings of that concrete index? The index template is only applied on index creation, so changes on the template are not reflected in the indices.

legoguy1000 · July 19, 2021, 11:32am

What index? I'm asking about the ingest pipeline. What specifically isn't working with the pipeline?

spinscale · July 19, 2021, 1:49pm

I'm asking about the proper configuration of the indexing pipeline in the index settings, as you do not specify the pipeline for each indexing/bulk operation in the beat. Can you share the index settings of the index you are indexing into?

legoguy1000 · July 19, 2021, 8:56pm

@spinscale Whoops, sorry. I wasn't paying attention, thought the question came from the OP.

samiusan · July 19, 2021, 8:57pm

Hi,

Thank you for the help.

Settings of the current index are below. This is correctly picking up the ingestion template and attempting to apply it.

    "settings" : {
      "index" : {
        "routing" : {
          "allocation" : {
            "include" : {
              "_tier_preference" : "data_content"
            }
          }
        },
        "number_of_shards" : "1",
        "provided_name" : "snort3-2021.07.20",
        "default_pipeline" : "snort-json-pipeline",
        "creation_date" : "1626703207180",
        "number_of_replicas" : "1",
        "uuid" : "Wsw3OcyAQROOX6VqZQCeMA",
        "version" : {
          "created" : "7130399"
        }
      }
    }

From experimenting further I feel I'm close to the solution.

When filebeat passes the message to elasticsearch it includes a number of other nested data sets as part of _source. Details of the original message are passed in the message nested object.

{
  },
  "ecs": {
    "version": "1.8.0"
  },
  "input": {
    "type": "log"
  },
  "log": {
    "file": {
      "path": "/var/log/snort/alert_json.txt"
    },
    "offset": 93914389
  },
  "message": {
  "seconds": 1626589993,
  "action": "allow",
  "class": "none",
  "b64_data": "AAAAAAAAAAAAAAAAAAAAAAAA",
  "dir": "UNK",
  "dst_ap": ":0",
  "eth_dst": "F0:9F:C2:C7:69:1C",
  "eth_len": 60,
  "eth_src": "BA:30:D5:05:8D:1C",
  "eth_type": "0x806",
  "gid": 112,
  "iface": "ens161",
  "msg": "(arp_spoof) unicast ARP request",
  "pkt_gen": "raw",
  "pkt_len": 18,
  "pkt_num": 615506767,
  "priority": 3,
  "proto": "ARP",
  "rev": 1,
  "rule": "112:1:1",
  "service": "unknown",
  "sid": 1,
  "src_ap": ":0",
  "vlan": 0,
  "timestamp": "07/18-16:33:13.829379"
}
}

The fields the ingestion pipeline target are not in the main _source section. Due to that the ingestion pipeline is not finding them. How can I access the data stored in a nested json object under _source? I have tried the message.field notation but haven't had any luck so far.

cheers,

Michael

samiusan · July 20, 2021, 10:47am

Hi,

I have continued to work at this.

I can extract a whole event and run it through a simulated pipeline okay.

The current pipeline I'm working with is below. This is using the message. notation and extracting data out of the ingested data okay.

PUT /_ingest/pipeline/snort-json-pipeline
{
  "description": "Pipeline for ingesting JSON snort3 data",
  "processors": [
    { 
      "set": {
        "field" : "pkt_num",
        "copy_from": "message.pkt_num",
        "tag": "set pkt_num"
      }
    },
    {
      "set": {
        "field": "pkt_len",
        "copy_from": "message.pkt_len"
      }
    },
    {
      "set" : {
        "field": "source.port",
        "copy_from": "message.src_port",
        "ignore_empty_value": true
      }
    },
    {
      "set" : {
        "field": "destination.port",
        "copy_from": "message.dst_port",
        "ignore_empty_value": true
      }
    },
    {
      "set": {
        "field": "source.address", 
        "copy_from": "message.src_addr",
        "ignore_empty_value": true
      }
    },
    {
      "set": {
        "field": "source.ip",
        "copy_from": "message.src_addr",
        "ignore_empty_value": true
      }
    },
    {
      "set": {
        "field": "source.mac",
        "copy_from": "message.eth_src"
      }
    },
    {
      "set": {
        "field": "destination.address", 
        "copy_from": "message.dst_addr",
        "ignore_empty_value": true
      }
    },
    {
      "set": {
        "field": "destination.ip",
        "copy_from": "message.dst_addr",
        "ignore_empty_value": true
      }
    },
    {
      "set": {
        "field": "destination.mac",
        "copy_from": "message.eth_dst"
      }
    },
    {
      "set": {
        "field": "network.protocol",
        "copy_from": "message.service"
      }
    },
    {
      "set": {
        "field": "network.transport",
        "copy_from": "message.proto"
      }
    },
    {
      "lowercase": {
        "field": "network.transport"
      }
    },
    {
      "date": {
        "field": "message.seconds",
        "formats": [
          "UNIX"
        ],
        "ignore_failure": true
      }
    },
    {
      "set": {
        "field": "event.severity",
        "copy_from": "message.priority",
        "ignore_empty_value": true
      }
    },
    {
      "append": {
        "field": "event.category",
        "value": [
          "intrusion_detection"
        ]
      }
    },
    {
      "append": {
        "field": "event.kind",
        "value": [
          "alert"
        ]
      }
    },
    {
      "append": {
        "field": "event.module",
        "value": [
          "snort3"
        ]
      }
    },
    {
      "community_id": {
        "seed": "15",
        "ignore_missing": true
      }
    },
    {
      "network_direction": {
        "internal_networks": ["private"]
      }
    }
  ],
  "version": 1
}

When I try to point data at the pipeline using filebeat I'm getting the below error. I'm understanding this to mean the ingestion pipeline is not finding the path in the message it is being sent.

If data is being passed from filebeat under the message: {} object how do I access that in a filebeat processor?

2021-07-20T20:38:35.823+1000    WARN    [elasticsearch] elasticsearch/client.go:408     Cannot index event publisher.Event{Content:beat.Event{Timestamp:time.Time{wall:0xc035c7ca3ac8a866, ext:549832714, loc:(*time.Location)(0x55dde5819dc0)}, Meta:{"raw_index":"snort3-2021.07.20"}, Fields:{"agent":{"ephemeral_id":"301f064e-3102-461f-9541-6c78f85c155c","hostname":"audite","id":"91fdd338-2c8f-40ff-a54d-c37ad962898d","name":"audite","type":"filebeat","version":"7.13.3"},"ecs":{"version":"1.8.0"},"host":{"architecture":"x86_64","containerized":false,"hostname":"audite","id":"3ce8587ede524b83abbc7d20f13212c2","ip":["192.168.10.5","2403:5800:7700:3204:20c:29ff:fe53:ecfe","fe80::20c:29ff:fe53:ecfe"],"mac":["00:0c:29:53:ec:fe","00:0c:29:53:ec:1c","00:0c:29:53:ec:08","00:0c:29:53:ec:12"],"name":"audite","os":{"codename":"focal","family":"debian","kernel":"5.4.0-77-generic","name":"Ubuntu","platform":"ubuntu","type":"linux","version":"20.04.2 LTS (Focal Fossa)"}},"input":{"type":"log"},"log":{"file":{"path":"/var/log/snort/alert_json.txt"},"offset":94499421},"message":"{ \"seconds\" : 1626777398, \"action\" : \"allow\", \"class\" : \"none\", \"b64_data\" : \"AAAAAAAAAAAAAAAAAAAAAAAA\", \"dir\" : \"UNK\", \"dst_ap\" : \":0\", \"eth_dst\" : \"4C:72:B9:41:D5:62\", \"eth_len\" : 60, \"eth_src\" : \"F0:9F:C2:C7:69:1C\", \"eth_type\" : \"0x806\", \"gid\" : 112, \"iface\" : \"ens161\", \"msg\" : \"(arp_spoof) unicast ARP request\", \"pkt_gen\" : \"raw\", \"pkt_len\" : 18, \"pkt_num\" : 790495972, \"priority\" : 3, \"proto\" : \"ARP\", \"rev\" : 1, \"rule\" : \"112:1:1\", \"service\" : \"unknown\", \"sid\" : 1, \"src_ap\" : \":0\", \"vlan\" : 0, \"timestamp\" : \"07/20-20:36:38.165235\" }"}, Private:file.State{Id:"native::1992103-64768", PrevId:"", Finished:false, Fileinfo:(*os.fileStat)(0xc0001c2680), Source:"/var/log/snort/alert_json.txt", Offset:94499963, Timestamp:time.Time{wall:0xc035c7ca3ac2ec46, ext:549456907, loc:(*time.Location)(0x55dde5819dc0)}, TTL:-1, Type:"log", Meta:map[string]string(nil), FileStateOS:file.StateOS{Inode:0x1e65a7, Device:0xfd00}, IdentifierName:"native"}, TimeSeries:false}, Flags:0x1, Cache:publisher.EventCache{m:common.MapStr(nil)}} (status=400): {"type":"illegal_argument_exception","reason":"cannot resolve [pkt_num] from object of type [java.lang.String] as part of path [message.pkt_num]"}

legoguy1000 · July 20, 2021, 11:26pm

It appears that message is being read as a JSON string and needs to be decoded to be referenced as an object, JSON processor | Elasticsearch Guide [7.13] | Elastic

spinscale · July 21, 2021, 1:55pm

what Alex above wrote is a valid point. On top of that, I think all your objects are within the message field, so maybe your processors need to specify the correct field like message.pkt_num?

system · August 18, 2021, 1:55pm

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

Topic		Replies	Views
Filebeat to elasticsearch using ingest pipeline Elasticsearch ingest-pipeline	6	421	June 2, 2022
Alternative to ingest-convert.sh for YML Logstash	5	430	December 7, 2021
Some questions about Filebeat that are not clear in the docs Beats filebeat	6	633	July 12, 2019
Elasticsearch json logs with filebeat module and ingest pipeline Beats filebeat	6	2294	February 21, 2020
Detecting JSON in Ingest pipeline Elasticsearch	2	2134	November 6, 2018

Elasticsearch ingest pipeline - not extract data passed from filebeat

Related topics