Type:flow missing source fields

sancroth · April 4, 2018, 1:56pm

In research to find a solution for capturing and visualizing all kinds of traffic on a few remote dedicated servers with the lack of firewall syslogs and tried out packetbeats. It does not do the same, i know, but it gives way more info than none and it has a better cpu and memory footprint than a manual tcpdump service -> log -> logstash -> elastic.

My big problem though is that the packetbeats do not send any source fields to the receiving end.
I have not done any other configuration other than adding the logstash output for some further manipulation if i found needed(already split type/index).

Stack Version used: 6.2

Do i need to add more configurations?

Example output:
{"type":"flow","flow_id":"EAT/////AP//////FP0BBBGrhqX7lPtQL0Inqt8","@timestamp":"2018-04-04T13:42:40.000Z","host":"MyHost","start_time":"2018-04-04T13:42:33.112Z","dest":{"ip":"151.151.151.151","stats":{"net_bytes_total":368,"net_packets_total":5},"port":1050},"last_time":"2018-04-04T13:42:33.113Z","final":false,"transport":"tcp","@version":"1","tags":["pbeat"]}

kvch · April 4, 2018, 2:17pm

Could you please share your debug logs and configuration with us?

sancroth · April 4, 2018, 2:43pm

Configurations:
#============================== Network device ================================

# Select the network interface to sniff the data. On Linux, you can use the
# "any" keyword to sniff on all connected interfaces.
packetbeat.interfaces.device: any

#================================== Flows =====================================

# Set `enabled: false` or comment out all options to disable flows reporting.
packetbeat.flows:
  # Set network flow timeout. Flow is killed if no packet is received before being
  # timed out.
  timeout: 30s

  # Configure reporting period. If set to -1, only killed flows will be reported
  period: 10s

#========================== Transaction protocols =============================

packetbeat.protocols:
- type: icmp
  # Enable ICMPv4 and ICMPv6 monitoring. Default: false
  enabled: true

- type: amqp
  # Configure the ports where to listen for AMQP traffic. You can disable
  # the AMQP protocol by commenting out the list of ports.
  ports: [5672]

- type: cassandra
  #Cassandra port for traffic monitoring.
  ports: [9042]

- type: dns
  # Configure the ports where to listen for DNS traffic. You can disable
  # the DNS protocol by commenting out the list of ports.
  ports: [53]

  # include_authorities controls whether or not the dns.authorities field
  # (authority resource records) is added to messages.
  include_authorities: true

  # include_additionals controls whether or not the dns.additionals field
  # (additional resource records) is added to messages.
  include_additionals: true

- type: http
  # Configure the ports where to listen for HTTP traffic. You can disable
  # the HTTP protocol by commenting out the list of ports.
  ports: [80, 8080, 8000, 5000, 8002]

- type: memcache
  # Configure the ports where to listen for memcache traffic. You can disable
  # the Memcache protocol by commenting out the list of ports.
  ports: [11211]

- type: mariadb
  # Configure the ports where to listen for MySQL traffic. You can disable
  # the MySQL protocol by commenting out the list of ports.
  ports: [3306]

- type: pgsql
  # Configure the ports where to listen for Pgsql traffic. You can disable
  # the Pgsql protocol by commenting out the list of ports.
  ports: [5432]

- type: redis
  # Configure the ports where to listen for Redis traffic. You can disable
  # the Redis protocol by commenting out the list of ports.
  ports: [6379]

- type: thrift
  # Configure the ports where to listen for Thrift-RPC traffic. You can disable
  # the Thrift-RPC protocol by commenting out the list of ports.
  ports: [9090]

- type: mongodb
  # Configure the ports where to listen for MongoDB traffic. You can disable
  # the MongoDB protocol by commenting out the list of ports.
  ports: [27017]

- type: nfs
  # Configure the ports where to listen for NFS traffic. You can disable
  # the NFS protocol by commenting out the list of ports.
  ports: [2049]

- type: tls
  # Configure the ports where to listen for TLS traffic. You can disable
  # the TLS protocol by commenting out the list of ports.
  ports: [443]


#==================== Elasticsearch template setting ==========================

setup.template.settings:
  index.number_of_shards: 3
  #index.codec: best_compression
  #_source.enabled: false

#================================ General =====================================

# The name of the shipper that publishes the network data. It can be used to group
# all the transactions sent by a single shipper in the web interface.
#name:

# The tags of the shipper are included in their own field with each
# transaction published.
#tags: ["service-X", "web-tier"]

# Optional fields that you can specify to add additional information to the
# output.
#fields:
#  env: staging


#============================== Dashboards =====================================
# These settings control loading the sample dashboards to the Kibana index. Loading
# the dashboards is disabled by default and can be enabled either by setting the
# options here, or by using the `-setup` CLI flag or the `setup` command.
#setup.dashboards.enabled: false

# The URL from where to download the dashboards archive. By default this URL
# has a value which is computed based on the Beat name and version. For released
# versions, this URL points to the dashboard archive on the artifacts.elastic.co
# website.
#setup.dashboards.url:

#============================== Kibana =====================================

# Starting with Beats version 6.0.0, the dashboards are loaded via the Kibana API.
# This requires a Kibana endpoint configuration.
setup.kibana:

  # Kibana Host
  # Scheme and port can be left out and will be set to the default (http and 5601)
  # In case you specify and additional path, the scheme is required: http://localhost:5601/path
  # IPv6 addresses should always be defined as: https://[2001:db8::1]:5601
  #host: "localhost:5601"

#============================= Elastic Cloud ==================================

# These settings simplify using packetbeat with the Elastic Cloud (https://cloud.elastic.co/).

# The cloud.id setting overwrites the `output.elasticsearch.hosts` and
# `setup.kibana.host` options.
# You can find the `cloud.id` in the Elastic Cloud web UI.
#cloud.id:

# The cloud.auth setting overwrites the `output.elasticsearch.username` and
# `output.elasticsearch.password` settings. The format is `<user>:<pass>`.
#cloud.auth:

#================================ Outputs =====================================

# Configure what output to use when sending the data collected by the beat.

#-------------------------- Elasticsearch output ------------------------------
#output.elasticsearch:
  # Array of hosts to connect to.
  #hosts: ["localhost:9200"]

  # Optional protocol and basic auth credentials.
  #protocol: "https"
  #username: "elastic"
  #password: "changeme"

#----------------------------- Logstash output --------------------------------
output.logstash:
  # The Logstash hosts
  hosts: ["xx.xx.xx.xx:xxxx"]

#================================ Logging =====================================

# Sets log level. The default log level is info.
# Available log levels are: error, warning, info, debug
logging.level: debug

sancroth · April 4, 2018, 2:43pm

Some part of debug logs:

8-04-04T16:22:43.790+0200    DEBUG   [flows] flows/flows.go:79       lookup flow: {3088 255 255 255 255 0 255 255 255 255 255 8 12 0 0 1} => [xx xx xx xx 178 146 86 22 187 1 198 167 115 2 0 0 0 0 0 0]
2018-04-04T16:22:43.790+0200    DEBUG   [flows] flows/flows.go:72       unlock flows
2018-04-04T16:22:43.790+0200    DEBUG   [sniffer]       sniffer/sniffer.go:192  Packet number: 173817
2018-04-04T16:22:43.790+0200    DEBUG   [decoder]       decoder/decoder.go:143  decode packet data
2018-04-04T16:22:43.790+0200    DEBUG   [flows] flows/flows.go:67       lock flows
2018-04-04T16:22:43.790+0200    DEBUG   [decoder]       decoder/decoder.go:216  IPv4 packet
2018-04-04T16:22:43.790+0200    DEBUG   [flows] flows/flowid.go:146     flowid: add ipv4
2018-04-04T16:22:43.790+0200    DEBUG   [decoder]       decoder/decoder.go:257  TCP packet
2018-04-04T16:22:43.790+0200    DEBUG   [flows] flows/flowid.go:213     flowid: add tcp
2018-04-04T16:22:43.790+0200    DEBUG   [flows] flows/flowid.go:219     flowid: add tcp connection id
2018-04-04T16:22:43.790+0200    DEBUG   [tcp]   tcp/tcp.go:146  tcp flow id: 0xc4215867c0
2018-04-04T16:22:43.790+0200    DEBUG   [tcp]   tcp/tcp.go:159  pkt.start_seq=571557482 pkt.last_seq=571572684 stream.last_seq=571557482 (len=15202)
2018-04-04T16:22:43.790+0200    DEBUG   [tls]   tls/parse.go:245        ignoring application data length 16408
2018-04-04T16:22:43.791+0200    DEBUG   [tls]   tls/parse.go:245        ignoring application data length 60
2018-04-04T16:22:43.791+0200    DEBUG   [decoder]       decoder/decoder.go:186  flow id flags: 3088
2018-04-04T16:22:43.791+0200    DEBUG   [flows] flows/flows.go:77       get flow
2018-04-04T16:22:43.791+0200    DEBUG   [flows] flows/flows.go:79       lookup flow: {3088 255 255 255 255 0 255 255 255 255 255 8 12 0 0 1} => [xx xx xx xx 178 146 86 22 187 1 198 167 115 2 0 0 0 0 0 0]
2018-04-04T16:22:43.791+0200    DEBUG   [flows] flows/flows.go:72       unlock flows
2018-04-04T16:22:43.791+0200    DEBUG   [sniffer]       sniffer/sniffer.go:192  Packet number: 173818
2018-04-04T16:22:43.791+0200    DEBUG   [decoder]       decoder/decoder.go:143  decode packet data
2018-04-04T16:22:43.791+0200    DEBUG   [flows] flows/flows.go:67       lock flows
2018-04-04T16:22:43.791+0200    DEBUG   [decoder]       decoder/decoder.go:216  IPv4 packet
2018-04-04T16:22:43.791+0200    DEBUG   [flows] flows/flowid.go:146     flowid: add ipv4
2018-04-04T16:22:43.791+0200    DEBUG   [decoder]       decoder/decoder.go:257  TCP packet
2018-04-04T16:22:43.791+0200    DEBUG   [flows] flows/flowid.go:213     flowid: add tcp
2018-04-04T16:22:43.791+0200    DEBUG   [flows] flows/flowid.go:219     flowid: add tcp connection id
2018-04-04T16:22:43.791+0200    DEBUG   [tcp]   tcp/tcp.go:146  tcp flow id: 0xc4215867c0
2018-04-04T16:22:43.791+0200    DEBUG   [tcp]   tcp/tcp.go:159  pkt.start_seq=571572684 pkt.last_seq=571580976 stream.last_seq=571572684 (len=8292)
2018-04-04T16:22:43.791+0200    DEBUG   [tls]   tls/parse.go:245        ignoring application data length 10251
2018-04-04T16:22:43.791+0200    DEBUG   [decoder]       decoder/decoder.go:186  flow id flags: 3088
2018-04-04T16:22:43.791+0200    DEBUG   [flows] flows/flows.go:77       get flow
2018-04-04T16:22:43.791+0200    DEBUG   [flows] flows/flows.go:79       lookup flow: {3088 255 255 255 255 0 255 255 255 255 255 8 12 0 0 1} => [xx xx xx xx 178 146 86 22 187 1 198 167 115 2 0 0 0 0 0 0]
2018-04-04T16:22:43.791+0200    DEBUG   [flows] flows/flows.go:72       unlock flows
2018-04-04T16:22:43.791+0200    DEBUG   [sniffer]       sniffer/sniffer.go:192  Packet number: 173819
2018-04-04T16:22:43.791+0200    DEBUG   [decoder]       decoder/decoder.go:143  decode packet data
2018-04-04T16:22:43.791+0200    DEBUG   [flows] flows/flows.go:67       lock flows
2018-04-04T16:22:43.791+0200    DEBUG   [decoder]       decoder/decoder.go:216  IPv4 packet
2018-04-04T16:22:43.791+0200    DEBUG   [flows] flows/flowid.go:146     flowid: add ipv4
2018-04-04T16:22:43.791+0200    DEBUG   [decoder]       decoder/decoder.go:257  TCP packet
2018-04-04T16:22:43.791+0200    DEBUG   [flows] flows/flowid.go:213     flowid: add tcp
2018-04-04T16:22:43.791+0200    DEBUG   [flows] flows/flowid.go:219     flowid: add tcp connection id
2018-04-04T16:22:43.791+0200    DEBUG   [tcp]   tcp/tcp.go:146  tcp flow id: 0xc4215867c0
2018-04-04T16:22:43.791+0200    DEBUG   [decoder]       decoder/decoder.go:186  flow id flags: 3088
2018-04-04T16:22:43.791+0200    DEBUG   [flows] flows/flows.go:77       get flow
2018-04-04T16:22:43.791+0200    DEBUG   [flows] flows/flows.go:79       lookup flow: {3088 255 255 255 255 0 255 255 255 255 255 8 12 0 0 1} => [xx xx xx xx 178 146 86 22 187 1 198 167 115 2 0 0 0 0 0 0]
2018-04-04T16:22:43.791+0200    DEBUG   [flows] flows/flows.go:72       unlock flows
2018-04-04T16:22:43.791+0200    DEBUG   [sniffer]       sniffer/sniffer.go:192  Packet number: 173820
2018-04-04T16:22:43.791+0200    DEBUG   [decoder]       decoder/decoder.go:143  decode packet data
2018-04-04T16:22:43.791+0200    DEBUG   [flows] flows/flows.go:67       lock flows
2018-04-04T16:22:43.791+0200    DEBUG   [decoder]       decoder/decoder.go:216  IPv4 packet
2018-04-04T16:22:43.791+0200    DEBUG   [flows] flows/flowid.go:146     flowid: add ipv4
2018-04-04T16:22:43.791+0200    DEBUG   [decoder]       decoder/decoder.go:257  TCP packet
2018-04-04T16:22:43.791+0200    DEBUG   [flows] flows/flowid.go:213     flowid: add tcp
2018-04-04T16:22:43.791+0200    DEBUG   [flows] flows/flowid.go:219     flowid: add tcp connection id
2018-04-04T16:22:43.791+0200    DEBUG   [tcp]   tcp/tcp.go:146  tcp flow id: 0xc4215867c0
2018-04-04T16:22:43.791+0200    DEBUG   [decoder]       decoder/decoder.go:186  flow id flags: 3088
2018-04-04T16:22:43.791+0200    DEBUG   [flows] flows/flows.go:77       get flow
2018-04-04T16:22:43.791+0200    DEBUG   [flows] flows/flows.go:79       lookup flow: {3088 255 255 255

sancroth · April 4, 2018, 2:59pm

Also on /etc/acketbeat/fields.yml the field source exists with all it's subfields.
Position above dest field and subfields.

So i expect the source field(s) to be created aswell.

steffens · April 4, 2018, 4:48pm

The flows analyzer handles the flow address reporting layer by layer. There is no difference when how source and dest fields are added to an event. Once the event is constructed, it is published as is. Packetbeat makes no further changes to the event (unless processors are configured to do so). Checking the code, I see no reason why source should be missing.

When running with debug enabled (either with -d '*' or just -d 'publish'), the debug logs should contain the event, before the event is being published to logstash. Can you post some sample events from the debug output?

You do any processing in Logstash?

sancroth · April 4, 2018, 7:07pm

I cannot do this atm since i am taking care of my Aunt at the hospital afterwork.
I shall update this post tommorow with your required logs.
Please forgive me

Also no further processing is happening on logstash as of now since i am testing out the information passed by packetbeat before trying to mix things up.

sancroth · April 5, 2018, 7:39am

You were correct source does exist on the fields sent.

2018-04-05T09:35:31.124+0200    DEBUG   [publish]       pipeline/processor.go:275       Publish event: {
  "@timestamp": "2018-04-05T07:35:30.790Z",
  "@metadata": {
    "beat": "packetbeat",
    "type": "doc",
    "version": "6.2.3"
  },
  "source": {
    "port": 44497,
    "stats": {
      "net_packets_total": 1,
      "net_bytes_total": 87
    },
    "ip": "xxxxxx"
  },
  "last_time": "2018-04-05T07:35:29.972Z",
  "flow_id": "EAL/////AP////8I//8AAAFegqX71YViYtGtNQA",
  "final": true,
  "transport": "udp",
  "dest": {
    "stats": {
      "net_packets_total": 1,
      "net_bytes_total": 217
    },
    "ip": "213.133.98.98",
    "port": 53
  },
  "start_time": "2018-04-05T07:35:29.972Z",
  "type": "flow",
  "beat": {
    "name": "test",
    "hostname": "test",
    "version": "6.2.3"
  }
}
2018-04-05T09:35:31.124+0200    DEBUG   [publish]       pipeline/processor.go:275       Publish event: {
  "@timestamp": "2018-04-05T07:35:30.790Z",
  "@metadata": {
    "beat": "packetbeat",
    "type": "doc",
    "version": "6.2.3"
  },
  "flow_id": "EAL/////AP////8I//8AAAFegqX71YViYnF2NQA",
  "transport": "udp",
  "source": {
    "stats": {
      "net_packets_total": 1,
      "net_bytes_total": 94
    },
    "ip": "xxxxxxxxxx",
    "port": 30321
  },
  "dest": {
    "ip": "213.133.98.98",
    "port": 53,
    "stats": {
      "net_packets_total": 1,
      "net_bytes_total": 224
    }
  },
  "beat": {
    "name": "test",
    "hostname": "test",
    "version": "6.2.3"
  },
  "start_time": "2018-04-05T07:35:29.972Z",
  "last_time": "2018-04-05T07:35:29.972Z",
  "type": "flow",
  "final": true
}

Which makes this case even more confusing. Gonna log the exact input right when logstash gets it and see if it actually gets messed up inside it. This is quite strange since it should not get inside any filter.

sancroth · April 5, 2018, 7:49am

While capturing the published event right after logstash inputs and with no further processing happening at input levels i get the same json as the one i posted before. No source fields yet.

How can the event being published correctly but the receiving event misses fields?

Unless there is further processing on the 'publish' event before being sent that removes the fields for some reason.

sancroth · April 5, 2018, 8:16am

netflow

Also more fields are missing like beats.hostname, beats.version etc.
And beats.hostname has been renamed into host(at least this makes sense since i dont create the filed manually).

As i said i am doing no other processing on logstash so this is the exact input i get which is the same that i log to a file at logstash input level.

Logstash conf to log the published event:

input {
     beats{
         port=>10444
         tags =>["some"]
    }

    beats{
        port=>11455
        tags =>["some"]
    }

    beats{
        port=>15046
        tags =>["pbeat"]
    }
}
output{
        if "pbeat" in [tags]{
                        file{
                                 path => "/opt/bucket.log"
                        }
        }
}

The above logs to the file this:

{"@version":"1","final":true,"start_time":"2018-04-05T07:59:07.555Z","dest":{"stats":{"net_packets_total":14,"net_bytes_total":160200},"ip":"127.0.0.1","port":9000},"last_time":"2018-04-05T07:59:08.772Z","type":"flow","@timestamp":"2018-04-05T07:59:09.995Z","flow_id":"EAT/////AP//////CP8AAAF/AAABfwAAASgjYFM","host":"some","transport":"tcp","tags":["pbeat","beats_input_raw_event"]}
{"@version":"1","final":true,"start_time":"2018-04-05T07:59:09.396Z","dest":{"stats":{"net_packets_total":5,"net_bytes_total":371},"ip":"xxxxxxxx","port":11050},"last_time":"2018-04-05T07:59:09.396Z","type":"flow","@timestamp":"2018-04-05T07:59:09.995Z","flow_id":"EAT/////AP//////CP8AAAFegqX7lPtQL0Inqs0","host":"some","transport":"tcp","tags":["pbeat","beats_input_raw_event"]}

sancroth · April 5, 2018, 1:31pm

Feel free to ask for any more info that could help you help me identify the problem.

sancroth · April 5, 2018, 2:20pm

By replacing the output from logstash to a file i get the desired events for some random reason:

{"@timestamp":"2018-04-05T14:14:50.000Z","@metadata":{"beat":"packetbeat","type":"doc","version":"6.2.3"},"start_time":"2018-04-05T14:14:40.126Z","final":false,"transport":"udp","beat":{"name":"xxxx","hostname":"xxxx","version":"6.2.3"},"source":{"ipv6":"xxxxxxxxxxxx","port":20669,"stats":{"net_packets_total":1,"net_bytes_total":112}},"dest":{"ipv6":"xxxxxxxxxxxx","port":123,"stats":{"net_packets_total":1,"net_bytes_total":112}},"last_time":"2018-04-05T14:14:40.126Z","type":"flow","flow_id":"QAL///////8A//8g//8AAAEqAQT4AAChEgAAAAAAAgACKgEE+AE7JkQAAAAAAAAAAnsAvVA"}
{"@timestamp":"2018-04-05T14:14:50.000Z","@metadata":{"beat":"packetbeat","type":"doc","version":"6.2.3"},"final":false,"beat":{"name":"xxx","hostname":"xxxx","version":"6.2.3"},"source":{"stats":{"net_packets_total":1,"net_bytes_total":88},"ipv6":"xxxx::xxxxxxxxxxe"},"dest":{"stats":{"net_packets_total":1,"net_bytes_total":80},"ipv6":"xxx::1"},"start_time":"2018-04-05T14:14:45.377Z","last_time":"2018-04-05T14:14:45.377Z","type":"flow","flow_id":"QAD///////8A//////8AAAH+gAAAAAAAAAAAAAAAAAAB/oAAAAAAAAASe0T//q/u/g"}

Although i could use filebeats or any shipper and get on with it i would like to find a proper solution for this and not keep even more logs on the server. Especially with the ammount of things being written regarding flow type records.

sancroth · April 5, 2018, 2:29pm

After testing it with the file output i was convinced it is a logstash problem.

In the end i closed every single input except the beats and allowed only flow type events from logstash.
It actually went through a huge filter that cleans up useless information from other logs, which was not wrapped correctly with it's conditionals.

In the end it was logstash processing but i couldnt trace it due to the large ammount of info coming in.
With reducing the information i could trace the problem.

Thank you for your time and sorry for the many posts.

system · May 3, 2018, 2:29pm

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