Skip to content

Commit

Permalink
Prevent Netflow and IPFIX templates from being modified concurrently
Browse files Browse the repository at this point in the history
  • Loading branch information
@jorritfolmer
jorritfolmer committed Nov 19, 2017
1 parent f1b977f commit 752cbfc
Show file tree
Hide file tree
Showing 4 changed files with 59 additions and 18 deletions.
1 change: 1 addition & 0 deletions CHANGELOG.md
View file
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
## 3.8.1

- Prevent Netflow and IPFIX templates from being modified concurrently
- Improved Palo Alto support and added rspec test

## 3.8.0
Expand Down
2 changes: 2 additions & 0 deletions docs/index.asciidoc
View file
Original file line number Diff line number Diff line change
Expand Up @@ -51,8 +51,10 @@ The following Netflow/IPFIX exporters are known to work with the most recent ver
|OpenBSD pflow | y | n | y | http://man.openbsd.org/OpenBSD-current/man4/pflow.4
|Softflowd | y | y | y | IPFIX supported in https://github.com/djmdjm/softflowd
|Streamcore Streamgroomer | | y | |
|Palo Alto PAN-OS | | y | |
|Ubiquiti Edgerouter X | | y | | With MPLS labels
|VMware VDS | | | y | Still some unknown fields
|YAF | | | y | With silk and applabel, but no DPI plugin support
|===========================================================================================

==== Usage
Expand Down
44 changes: 26 additions & 18 deletions lib/logstash/codecs/netflow.rb
View file
Original file line number Diff line number Diff line change
Expand Up @@ -51,7 +51,9 @@ class LogStash::Codecs::Netflow < LogStash::Codecs::Base

def initialize(params = {})
super(params)
@threadsafe = false
@threadsafe = true
@decode_mutex_netflow = Mutex.new
@decode_mutex_ipfix = Mutex.new
end

def register
Expand Down Expand Up @@ -212,17 +214,20 @@ def decode_netflow9(flowset, record, metadata = nil)
else
key = "#{flowset.source_id}|#{template.template_id}"
end
@netflow_templates[key, @cache_ttl] = BinData::Struct.new(:endian => :big, :fields => fields)
@logger.debug("Received template #{template.template_id} with fields #{fields.inspect}")
@logger.debug("Received template #{template.template_id} of size #{template_length} bytes. Representing in #{@netflow_templates[key].num_bytes} BinData bytes")
if template_length != @netflow_templates[key].num_bytes
@logger.warn("Received template #{template.template_id} of size #{template_length} bytes doesn't match BinData representation we built (#{@netflow_templates[key].num_bytes} bytes)")
end
# Purge any expired templates
@netflow_templates.cleanup!
if @cache_save_path
@netflow_templates_cache[key] = fields
save_templates_cache(@netflow_templates_cache, "#{@cache_save_path}/netflow_templates.cache")
# Prevent netflow_templates array from being concurrently modified
@decode_mutex_netflow.synchronize do
@netflow_templates[key, @cache_ttl] = BinData::Struct.new(:endian => :big, :fields => fields)
@logger.debug("Received template #{template.template_id} with fields #{fields.inspect}")
@logger.debug("Received template #{template.template_id} of size #{template_length} bytes. Representing in #{@netflow_templates[key].num_bytes} BinData bytes")
if template_length != @netflow_templates[key].num_bytes
@logger.warn("Received template #{template.template_id} of size #{template_length} bytes doesn't match BinData representation we built (#{@netflow_templates[key].num_bytes} bytes)")
end
# Purge any expired templates
@netflow_templates.cleanup!
if @cache_save_path
@netflow_templates_cache[key] = fields
save_templates_cache(@netflow_templates_cache, "#{@cache_save_path}/netflow_templates.cache")
end
end
end
end
Expand Down Expand Up @@ -316,12 +321,15 @@ def decode_ipfix(flowset, record)
end
# FIXME Source IP address required in key
key = "#{flowset.observation_domain_id}|#{template.template_id}"
@ipfix_templates[key, @cache_ttl] = BinData::Struct.new(:endian => :big, :fields => fields)
# Purge any expired templates
@ipfix_templates.cleanup!
if @cache_save_path
@ipfix_templates_cache[key] = fields
save_templates_cache(@ipfix_templates_cache, "#{@cache_save_path}/ipfix_templates.cache")
# Prevent ipfix_templates array from being concurrently modified
@decode_mutex_ipfix.synchronize do
@ipfix_templates[key, @cache_ttl] = BinData::Struct.new(:endian => :big, :fields => fields)
# Purge any expired templates
@ipfix_templates.cleanup!
if @cache_save_path
@ipfix_templates_cache[key] = fields
save_templates_cache(@ipfix_templates_cache, "#{@cache_save_path}/ipfix_templates.cache")
end
end
end
end
Expand Down
30 changes: 30 additions & 0 deletions spec/codecs/netflow_stress.py
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,30 @@
import socket
import sys
import time
import random

## Standalone Netflow v9 stressor
## Used to reproduce issue 91 https://github.com/logstash-plugins/logstash-codec-netflow/issues/91

host = 'host02'
port = 2055

tpl = '\x00\t\x00\x01e\x9c\xc0_XF\x8eU\x01u\xc7\x03\x00\x00\x08\x81\x00\x00\x00d\x01\x04\x00\x17\x00\x02\x00\x04\x00\x01\x00\x04\x00\x08\x00\x04\x00\x0c\x00\x04\x00\n\x00\x04\x00\x0e\x00\x04\x00\x15\x00\x04\x00\x16\x00\x04\x00\x07\x00\x02\x00\x0b\x00\x02\x00\x10\x00\x04\x00\x11\x00\x04\x00\x12\x00\x04\x00\t\x00\x01\x00\r\x00\x01\x00\x04\x00\x01\x00\x06\x00\x01\x00\x05\x00\x01\x00=\x00\x01\x00Y\x00\x01\x000\x00\x02\x00\xea\x00\x04\x00\xeb\x00\x04'

# 21 flows:
data = '\x00\t\x00\x15e\x9c\xbcqXF\x8eT\x01u\xc6\xa1\x00\x00\x08\x81\x01\x04\x05\\\x00\x00\x00\x01\x00\x00\x00(\n\x00\t\x92\n\x00\x1fQ\x00\x00\x00n\x00\x00\x00\x9ee\x9cG\x05e\x9cG\x05\xd3\x01\x01\xbb\x00\x00\x00\x00\x00\x00\xfb\xf0\n\x00\x0e!\x10\x14\x06\x10\x00\x01@\x00\x01`\x00\x00\x00`\x00\x00\x00\x00\x00\x00\x02\x00\x00\x00h\n\x00\x11*\n\x00#\x04\x00\x00\x00W\x00\x00\x00\x9ee\x9cI\x88e\x9cG\x07\x8e\x84\x01\xbb\x00\x00\x00\x00\x00\x00\xfb\xf0\n\x00\x0e!\x15\x10\x06\x10\x00\x01@\x00\x01`\x00\x00\x00`\x00\x00\x00\x00\x00\x00\x01\x00\x00\x004\n\x00\x16o\n\x00"\x8d\x00\x00\x00h\x00\x00\x00\x9ee\x9cG\ne\x9cG\nA\xae\x01\xbb\x00\x00\x00\x00\x00\x00\xfb\xf0\n\x00\x0e!\x18\x10\x06\x11\x00\x01@\x00\x01`\x00\x00\x00`\x00\x00\x00\x00\x00\x00\x01\x00\x00\x01\xb3\n\x00\x17;\n\x00$\xaa\x00\x00\x00V\x00\x00\x00\x9ee\x9cG\x0ce\x9cG\x0c\x005\xfd,\x00\x00\x00\x00\x00\x00\xfb\xf1\n\x00\x0e\x1f\x19\x13\x11\x00\x00\x01@\x00\x01`\x00\x00\x00`\x00\x00\x00\x00\x00\x00\x01\x00\x00\x03\xc9\n\x00"G\n\x00\x14\xf2\x00\x00\x00\x9e\x00\x00\x00je\x9cG\re\x9cG\r\x01\xbb\x07\xdd\x00\x00\xfb\xf0\x00\x00\xff\xa2\n\x00\x12\x05\x10\x15\x06\x18\x00\x00@\x00\x01`\x00\x00\x00`\x00\x00\x00\x00\x00\x00\x02\x00\x00\x00h\n\x00\n\x85\n\x00\x1ef\x00\x00\x00n\x00\x00\x00\x9ee\x9cG\re\x9cF\xba\x89\xc9\x00P\x00\x00\x00\x00\x00\x00\xfb\xf0\n\x00\x0e!\x10\x10\x06\x10\x00\x01@\x00\x01`\x00\x00\x00`\x00\x00\x00\x00\x00\x00\x01\x00\x00\x004\n\x00%\x1d\n\x00\x06\x18\x00\x00\x00f\x00\x00\x00\xa2e\x9cG\x10e\x9cG\x10\x00P\xdd\xc3\x00\x00;\x1d\x00\x00\xff\x97\n\x00\x00\xf2\x18\x10\x06\x10 \x00@\x00\x01`\x00\x00\x00`\x00\x00\x00\x00\x00\x00\x01\x00\x00\x02f\n\x00 \xb0\n\x00\x0bq\x00\x00\x00\x9e\x00\x00\x00.e\x9cG\x10e\x9cG\x10\x01\xbb\xdd\xfe\x00\x00\xfb\xf0\x00\x00\xff\x98\n\x00\x12i\x14\x10\x06\x18\x00\x00@\x00\x01`\x00\x00\x00`\x00\x00\x00\x00\x00\x00\x03\x00\x00\x10\xfe\n\x00\x0c\x15\n\x00\x0f&\x00\x00\x00W\x00\x00\x00\x9ee\x9cG\x11e\x9c1\xe7\x01\xbb\x9c\x8e\x00\x00\x80\xa6\x00\x00\xfb\xf2\n\x00\x0e\x1b\x18\x18\x06\x10\x00\x01@\x00\x01`\x00\x00\x00`\x00\x00\x00\x00\x00\x00\x02\x00\x00\x02\x15\n\x00\x04\xd4\n\x00\x03n\x00\x00\x00\xa2\x00\x00\x00fe\x9cT\x07e\x9cG\x12\xc6\x03\x01\xbb\x00\x00\xff\x97\x00\x00\x00F\n\x00\x10e\x10\x11\x06\x18\x00\x01@\x00\x01`\x00\x00\x00`\x00\x00\x00\x00\x00\x01E\x00\x005\\\n\x00!z\n\x00\x01\x88\x00\x00\x00\x9e\x00\x00\x00he\x9co\xd0e\x9c"\x1a\xe5\xbe\x00P\x00\x00\xfb\xf1\x00\x00\x00\x00\x00\x00\x00\x00\x15\x1b\x06\x10\x00\x00@\x00\x01`\x00\x00\x00`\x00\x00\x00\x00\x00\x00\x01\x00\x00\x00Y\n\x00\x14\xf2\n\x00"G\x00\x00\x00j\x00\x00\x00\x9ee\x9cG\x14e\x9cG\x14\x07\xdd\x01\xbb\x00\x00\xff\xa2\x00\x00\xfb\xf0\n\x00\x0e!\x15\x10\x06\x18`\x01@\x00\x01`\x00\x00\x00`\x00\x00\x00\x00\x00\x00\x01\x00\x00\x03A\n\x00\r\x19\n\x00\x0f&\x00\x00\x00W\x00\x00\x00\x9ee\x9cG\x16e\x9cG\x16\x01\xbb\xc9\xa5\x00\x00\x80\xa6\x00\x00\xfb\xf2\n\x00\x0e\x1b\x18\x18\x06\x18\x00\x01@\x00\x01`\x00\x00\x00`\x00\x00\x00\x00\x00\x00\x02\x00\x00\x06Y\n\x00\x19;\n\x00\x02\x12\x00\x00\x00\x9e\x00\x00\x00ne\x9cG\x18e\x9cF\xbf\x01\xbb\xf4\x00\x00\x00\xfb\xf0\x00\x00\xff\x9d\n\x00\x12~\x10\x10\x06\x18\x00\x00@\x00\x01`\x00\x00\x00`\x00\x00\x00\x00\x00\x00a\x00\x02+h\n\x00\x07I\n\x00\x1b\xa8\x00\x00\x00V\x00\x00\x00\x9ee\x9cu\xabe\x9c1\xfe\xeb\x98\x01\xd1\x00\x00\xff\x9c\x00\x00\xfb\xf0\n\x00\x0e!\x10\x10\x06\x18\x00\x01@\x00\x01`\x00\x00\x00`\x00\x00\x00\x00\x00\x00:\x00\x00\x0b\xc8\n\x00\x132\n\x00\x1b\xa9\x00\x00\x00j\x00\x00\x00\x9ee\x9cO\xcbe\x9cE:\x86\x94\x03\xe3\x00\x00\xff\xb7\x00\x00\xfb\xf0\n\x00\x0e!\x12\x10\x06\x10\x00\x01@\x00\x01`\x00\x00\x00`\x00\x00\x00\x00\x00\x00\x15\x00\x00{\x0c\n\x00\x1c\x96\n\x00\x18\r\x00\x00\x00\x9e\x00\x00\x00he\x9cHYe\x9cF\xf0\x01\xbb\xc2\xfd\x00\x00\xfb\xf0\x00\x00\x00\x00\x00\x00\x00\x00\x10\x19\x06\x10\x00\x00@\x00\x01`\x00\x00\x00`\x00\x00\x00\x00\x00\x00\x03\x00\x00\x0bg\n\x00\x1a\xbc\n\x00\x15\xc8\x00\x00\x00\x9e\x00\x00\x00We\x9cGfe\x9cE\xec\x03\xe1\xc4N\x00\x00\xfb\xf0\x00\x00\x00\x00\x00\x00\x00\x00\x10\x19\x06\x18\x00\x00@\x00\x01`\x00\x00\x00`\x00\x00\x00\x00\x00\x00\x05\x00\x00\x11\xa2\n\x00\x1d"\n\x00\x0f&\x00\x00\x00K\x00\x00\x00\x9ee\x9cm`e\x9cA\xfe\x01\xbb\x8c\x8f\x00\x00;A\x00\x00\xfb\xf2\n\x00\x0e\x1b\x18\x18\x06\x18\x00\x01@\x00\x01`\x00\x00\x00`\x00\x00\x00\x00\x00\x00\x01\x00\x00\x01F\n\x00\x08\xc8\n\x00\x05\xe0\x00\x00\x00f\x00\x00\x00\xa2e\x9cG\x1de\x9cG\x1dZX\xc9\xd7\x00\x00\x03\x15\x00\x00\xff\x97\n\x00\x00\xf2\x10\x10\x06\x18\x00\x00@\x00\x01`\x00\x00\x00`\x00\x00\x00\x00\x00\x00\x02\x00\x00\x00p\n\x00\x1d.\n\x00\x0f&\x00\x00\x00K\x00\x00\x00\x9ee\x9cG\x1de\x9c@\xea\x01\xbb\xcc\x8c\x00\x00;A\x00\x00\xfb\xf2\n\x00\x0e\x1b\x18\x18\x06\x12\x00\x01@\x00\x01`\x00\x00\x00`\x00\x00\x00\x00\x00\x00'


sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)

print("NETFLOW v9: sending 1 template 1 data packet in an infinite loop")

duration = 0.0
while True:
for i in range(0,400):
sock.sendto(tpl, (host, port))
sock.sendto(data, (host, port))
sys.stdout.write('.')
sys.stdout.flush()
time.sleep(random.random())
print

0 comments on commit 752cbfc

Please sign in to comment.