GrayLog Stream Lookup (SLookup) Pipeline Processor function

Plugin SLookup 2.0.0 - Multiple Return Fields

Stream Lookup function for GrayLog2 Pipeline Processor

billmurrin

free!

Download from Github View on Github 2 15

Published

17 Jan 00:59

Last Push

17 Jan 01:06

Marketplace Rating

Discussion

4 Comments

Readme from Github

The GrayLog Stream Lookup (SLookup) Pipeline Processor function

SLookup facilitates the lookup of a local stream's field value on a remote stream field, and if it matches, returns the requested fields for enrichment in the source stream.

For example, say there are two streams, one contains some http logs with source IPs (E.g. src_ip) from internal hosts and the other stream contains information about the systems on the network such as IP address (E.g. ip_address), computer name (E.g. computer_name), MAC address (E.g. mac_address), OU, make/model, etc.

In the example above, you might want to return the computer_name and mac_address fields where the value of src_ip matches ip_address.

The thought behind this function is to implement a similar functionality to the VLOOKUP function in Excel.

With features like index sets being introduced in Graylog 2.x, it is possible to use data in one stream to enrich data in another with Pipeline Processor rules.

Version 2.0.0 tested to work with Graylog 2.3.2 and 2.4.0

Function Breakdown

Function	Description
slookup(stream, srcField, dstField, rtnField, timeRange, sortOrder) : List	Conduct a lookup in a remote stream and return a list of field(s) values based on a matching source field. Similar to VLOOKUP in Excel

Parameter	Type	Required	Description
stream	String	Y	The stream to look up the source field.
srcField	String	Y	The source field. The value to query for in the remote stream.
dstField	String	Y	The destination field that will be queried against.
rtnField	List	Y	The field(s) to return if the query is successful.
timeRange	String	Y	Relative Time Range (Seconds)
sortOrder	String	Y	Timestamp sort order either "asc" or "desc".

Use Case and Rule Examples

Below are a couple example rules that help demonstrate how to use slookup.

In the following examples, the remote stream named Systems with stream_id 5a5d8854315d00059dbea98f contains system information (IP, MAC, Operating System). The source of this data might be Directory Service Computer Objects, NBTScan results, Discovery Scan, etc.

The slookup function constructs a search query using the value of winlogbeat_computer_name on the computer_name field (computer_name:VALUE_OF_FIELD). If the search is successful, the requested field(s) are returned. The list of values can then be added to the current stream message in the pipeline.

The sortOrder parameter instructs the function to either return the oldest match (ascending), or the newest match (descending) if multiple records are found during the query.

The Return Fields is a List object starting with index 0. The order of the indexes is based on the order you specified them in the return field List. If no search result is found for the field, it will return "No match found".

Match on Computer Name, Return IP, Operating System and Mac Address, Use Newest (Descending Sort Order) result.

rule "Log Enrichment - Ascending"
when
    has_field("winlogbeat_computer_name")
then
    //StreamID, Source Field, Destination Field, Return Field(s), Relative Time, Ascending SortOrder
    let system_info = slookup("5a5d8854315d00059dbea98f", "winlogbeat_computer_name", "computer_name", ["ip_address","operating_system","mac_address"], "300", "desc");
    set_field("ip_address", system_info[0]);
    set_field("operating_system", to_string(system_info[1]));
    set_field("mac_address", system_info[2]);
end

Match on Computer Name, Return only IP, Use Oldest (Ascending Sort Order) result.

rule "IP Lookup - Descending"
when
    has_field("winlogbeat_computer_name")
then
    //StreamID, Source Field, Destination Field, Return Field, Relative Time, Descending SortOrder
    let system_info = slookup("5a5d8854315d00059dbea98f", "winlogbeat_computer_name", "computer_name", ["ip_address"], "14400", "asc");
    set_field("ip_address", to_ip(system_info));
end

Additional Info

This function's performance impact on very large remote streams and very large relative data timeframes, remains unknown.

If you experience an ingestion slow-down enriching a large volume of data, you can attempt increasing processbuffer_processors in the graylog server.conf file.

More information about writing a Graylog2 processor pipeline function. https://www.graylog.org/blog/71-writing-your-own-graylog-processing-pipeline-functions

Your Rating

Please sign in to rate this add-on.

Comments

y2kman about 1 month ago

Hello Billmurrin,

I love this plugin as it saves a step in tracing down logs. I have a question about an anomaly though. I am using your plugin to take internet logs from our firewall, and then do a lookup to compare the MAC address with the MAC address in the NPS logs. This allows me to add the NPS Username to the internet log. This works perfectly 95% of the time. But sometimes it appears to do a random match with superfluous data. So, I can dump logs for a certain device, and most of the logs show the correct associated username, but some show random usernames that are not associated. My pipeline rule looks like this:

rule "srcMACandRadius"
when
has_field("srcMac") && ((to_string($message.srcInterface) == "X3") || (to_string($message.srcInterface) == "X3-V144") || (to_string($message.srcInterface) == "X2") || (to_string($message.srcInterface) == "X2-V150") || (to_string($message.srcInterface) == "X2-V140") || (to_string($message.srcInterface) == "X2-V124"))
then
//StreamID, Source Field, Destination Field, Return Field(s), Relative Time, Ascending SortOrder
let Radius = slookup("5eb937667803e50440f36831", "srcMac", "CallingStationID_with_colons", ["SubjectUserName", "AP_Name"], "4000", "desc");
set_field("RadiusUserName", to_string(Radius[0]));
set_field("RadiusAP_Name", to_string(Radius[1]));
end

Have I set this up correctly? Any idea why it might be occasionally mismatching? I can send examples if needed.

KnowMoreIT almost 2 years ago

I think I figured out the problem. Since the messages come in exactly at the same time, it may not be in the search results yet when the pipeline runs. I got it working in the simulator by adjusting the timestamp because I was thinking it was going off the source timestamp for some reason and it is actually the relative time from the time you run the simulator. This plugin is going to be very useful though. Thank you!

KnowMoreIT almost 2 years ago

I wasn't able to get this to work. I was trying to use it for looking up data from a NPS server for failed logins (unauthorized Mac Addresses). The problem is NPS has one log for the computer trying which includes the switch port number and the failed login afterwards contains the failure.

rule "NPS Server - Get Port Information"
when
has_field("Acct-Session-Id") == true && has_field("NAS-Port-Id") == false
then
let getport = slookup("5cb4f45fd4f7df652d151e12", "Acct-Session-Id", "Acct-Session-Id", ["NAS-Port-Id"], "600", "desc");
set_field("NAS-Port-ID", getport);
end

So this hsould be getting the NAS-Port-Id value from the stream 5cb4f45fd4f7df652d151e12 using the the search query: Acct-Session-Id: but it always returns no match found (even in the simulator).

This is a fantastic idea though.

helpitorg about 3 years ago

Hello Billmurrin,
thank you very much for this great plugin. It did a great job for me.I was ordered to find a solution to analyse the logs of a pfsense-guest-wlan of a bigger environment regarding which voucher is linked to which ip-address. The pfsense-filterlogs only shows the source and destination ip and the mapping of IPs is only shown in a different log-file. Thanks to your "slookup" I managed that graylog adds a field "IP_MAPS_VOUCHER" to the filterlog-messages which shows the voucher a SourceIP is actually connected to. I hope that I will find the time as soon as possible to present my solution to the graylog and the git-hub community soon. I can't believe that your plugin is not yet one of graylogs default ones. I nearly was about to try to solve it witch an ELK-stack. Your plugin should be a must for future releases.
Thanks a lot!
helpitorg

Please sign in to comment.

Back to listing