Pipeline rule to generate a Object field with obtained IPs (multi_value) for a QNAME, from a DNS LookUp Table

Hi,

After days trying to get this working, I think that I need help, :’(

Basically, I have this:

  • Microsoft DNS Server Logs, collected with SilkETW, shipped with Winlogbeat and parsed as json, with parse_json, select_jsonpath, and key_value functions in a pipeline rule (this is working ok):

rule “Regla Pipeline - Windows DNS Server - SilkService - Extrae campos JSON del campo event_data_param1”
when
has_field(“event_data_param1”)
then
let json = parse_json(to_string($message.event_data_param1));
set_fields(to_map(json));
let json_fields = select_jsonpath(json, {json_XmlEventData: “$.XmlEventData”});
set_fields(json_fields);
let json_nested_fields = select_jsonpath(json, {json_FormattedMessage: “$.XmlEventData.FormattedMessage”});
let kv_fields = to_string(json_nested_fields);
set_fields(key_value(value: kv_fields, trim_value_chars: “;”));
end

  • One of the parsed fields, named QNAME, with contains the DNS LookUp Qualified Name, from which I’m trying to get mutiple resolved IP address corresponding to it, using a DNS LookUp Table.

  • Using the single_value response of the DNS LookUp Table to generate QNAME_IP field is working ok, both using extractor or pipeline rule, but I want the multi_value response, specifically the IP addresses, and at my understanding, that can only be obtained/used with a pileline rule, not a extractor.

Here is a example:

My pipeline rule:
rule “Regla Pipeline - Windows DNS Server - SilkService - Lookup DNS QNAME”
when
has_field(“QNAME”) AND NOT contains(to_string($message.QNAME), “empresa.com”)
then
let lookup_IPs_json = lookup(“DNS_QNAME_IP”, to_string($message.QNAME));
// Text variable for testing
set_field(“QNAME_TEXT”,to_string(lookup_IPs_json));
//
let QNAME_IPs = select_jsonpath(parse_json(to_string(lookup_IPs_json)), { ip_address: “$.results[*].ip_address”});
set_field(“QNAME_IPs”, QNAME_IPs);
end

With that rule, QNAME_IPs is always null.

For the JSonPath of the rule, I already tried with no luck (helped me with Json Path Online Evalutator from http://jsonpath.com):

  • $.results[*].ip_address
  • $.multi_value.results[*].ip_address
  • $.results.ip_address
  • $.string_list_value

The strange thing here, at least for me, is that the text field that I’m generating for troubleshooting, have this text (for the example above):
{results=[ADnsAnswer{ipAddress=35.163.157.37, ipVersion=null, dnsTTL=47}, ADnsAnswer{ipAddress=35.164.39.71, ipVersion=null, dnsTTL=47}]}

What I Finally need as a result, is to populate and Object field, named QNAME_IPs, with this (for the example above):
[ “35.163.157.37”, “35.164.39.71” ]

Other strange thing (again, at least for me), is that sometimes, using different DNS Servers configured in the DNS LookUp Table for troubleshooing, the lookup for a QNAME brings my a single IP, duplicated, but when I retry the lookup, is OK:


I can think that this last problem is maybe a DNS Server problem, but how is it happening configuring different DNS Servers (from my work, public) in the DNS LookUp Table in different times, maybe it is a bug in the DNS Lookup Table Adapter of Graylog.

Can anybody try to explain my what I’m doing wrong with my pipeline rule?

Thanks in advance for your help.

Regards,
Alejandro

Hi everybody.

An update of this: ussing this rule, I obtained all the results (ip addresses, which is what I need, but additionally TTLs):

rule “Regla Pipeline - Windows DNS Server - SilkService - Lookup DNS QNAME - MultiIP”
when
has_field(“QNAME”) AND NOT contains(to_string($message.QNAME), “empresa.com”)
then
let multiIPs = lookup(“DNS_QNAME_IP”, to_string($message.QNAME));
set_field(“QNAME_IPs_Prueba”, multiIPs.results);
end

With this, I Obtain:

The problem is that I only need to obtain the ip addresses values. I tried to use in the rule:

  • set_field(“QNAME_IPs_Prueba”, multiIPs.results.ip_address);
  • set_field(“QNAME_IPs_Prueba”, multiIPs.results.ip_address);

The first option Graylog not allow me to use it, the second one give me no results.

I’m thing I’m near, so any help is appreciated.

Thanks in advance.

Regards,
Alejandro

Use debug() to dump values out to the graylog log files. Maybe you are grabbing the quotes in the field data so DNS can’t work with it? You can see the results in the Graylog log file using

tail -f /var/log/graylog-server/server.log

ps. use the formatting tools in your posts to make the code/logs you post more readable… for instance the </> button will make

rule “Regla Pipeline - Windows DNS Server - SilkService - Lookup DNS QNAME - MultiIP”
when
has_field(“QNAME”) AND NOT contains(to_string($message.QNAME), “empresa.com”)
then
let multiIPs = lookup(“DNS_QNAME_IP”, to_string($message.QNAME));
set_field(“QNAME_IPs_Prueba”, multiIPs.results);
end

Look more like

rule "Regla Pipeline - Windows DNS Server - SilkService - Lookup DNS QNAME - MultiIP"
when
    has_field("QNAME") AND NOT contains(to_string($message.QNAME), "[empresa.com](http://empresa.com/)")
then
    let multiIPs = lookup("DNS_QNAME_IP", to_string($message.QNAME));
    set_field("QNAME_IPs_Prueba", multiIPs.results);
end

Hi @tmacgbay, thanks for your reply and for your suggestion to use the formatting tools, :slight_smile:

I put the debug() function in my rule for multiIPs.results.ip_address, and the field is clearly empty:

[root@appliance ~]# tail -f /var/log/graylog-server/server.log | grep "Dropped message from"
2020-04-13T14:40:46.184-03:00 INFO  [Function] PIPELINE DEBUG: Dropped message from 
2020-04-13T14:40:46.185-03:00 INFO  [Function] PIPELINE DEBUG: Dropped message from 
2020-04-13T14:40:46.184-03:00 INFO  [Function] PIPELINE DEBUG: Dropped message from 
2020-04-13T14:40:46.185-03:00 INFO  [Function] PIPELINE DEBUG: Dropped message from 
2020-04-13T14:40:48.371-03:00 INFO  [Function] PIPELINE DEBUG: Dropped message from 
2020-04-13T14:40:48.376-03:00 INFO  [Function] PIPELINE DEBUG: Dropped message from 
2020-04-13T14:40:48.401-03:00 INFO  [Function] PIPELINE DEBUG: Dropped message from 

So, I think obtain fields with the same name in a json as an Object is not posible in Graylog, because with the correct way to do it (based on https://jsonpath.com/), using multiIPs.results[*].ip_address
Graylog not allow me to use it:

Thanks for your help.

Regards,
Alejandro

Did you put info in the debug()?

then
  debug($message.QNAME);
  let multiIPs = lookup(“DNS_QNAME_IP”, to_string($message.QNAME));
  debug(multiIPs );
end

will tell you what data is in QNAME at the time of processing.

Hi again @tmacgbay

Debugging multiIPs fields, I obtained the same output mentioned in the main post, when I say “text field that I’m generating for troubleshooting” (before you give me the hint about using the debug() function, I put the output to a common text field, “QNAME_IPs_Text”):

[root@appliance ~]# tail -f /var/log/graylog-server/server.log | grep "Debug message to review"
2020-04-13T15:23:33.953-03:00 INFO  [Function] PIPELINE DEBUG: Debug message to review: {results=[ADnsAnswer{ipAddress=40.90.23.247, ipVersion=null, dnsTTL=42}, ADnsAnswer{ipAddress=40.90.137.127, ipVersion=null, dnsTTL=42}, ADnsAnswer{ipAddress=40.90.137.124, ipVersion=null, dnsTTL=42}]}
2020-04-13T15:23:33.953-03:00 INFO  [Function] PIPELINE DEBUG: Debug message to review: {results=[ADnsAnswer{ipAddress=40.90.23.247, ipVersion=null, dnsTTL=51}, ADnsAnswer{ipAddress=40.90.137.127, ipVersion=null, dnsTTL=51}, ADnsAnswer{ipAddress=40.90.137.124, ipVersion=null, dnsTTL=51}]}
2020-04-13T15:23:33.954-03:00 INFO  [Function] PIPELINE DEBUG: Debug message to review: {results=[ADnsAnswer{ipAddress=40.90.23.247, ipVersion=null, dnsTTL=51}, ADnsAnswer{ipAddress=40.90.137.127, ipVersion=null, dnsTTL=51}, ADnsAnswer{ipAddress=40.90.137.124, ipVersion=null, dnsTTL=51}]}
2020-04-13T15:23:33.954-03:00 INFO  [Function] PIPELINE DEBUG: Debug message to review: {results=[ADnsAnswer{ipAddress=40.90.23.247, ipVersion=null, dnsTTL=51}, ADnsAnswer{ipAddress=40.90.137.127, ipVersion=null, dnsTTL=51}, ADnsAnswer{ipAddress=40.90.137.124, ipVersion=null, dnsTTL=51}]}
2020-04-13T15:23:33.955-03:00 INFO  [Function] PIPELINE DEBUG: Debug message to review: {results=[ADnsAnswer{ipAddress=40.90.23.247, ipVersion=null, dnsTTL=51}, ADnsAnswer{ipAddress=40.90.137.127, ipVersion=null, dnsTTL=51}, ADnsAnswer{ipAddress=40.90.137.124, ipVersion=null, dnsTTL=51}]}
2020-04-13T15:23:33.956-03:00 INFO  [Function] PIPELINE DEBUG: Debug message to review: {results=[ADnsAnswer{ipAddress=40.90.23.247, ipVersion=null, dnsTTL=42}, ADnsAnswer{ipAddress=40.90.137.127, ipVersion=null, dnsTTL=42}, ADnsAnswer{ipAddress=40.90.137.124, ipVersion=null, dnsTTL=42}]}
2020-04-13T15:23:33.981-03:00 INFO  [Function] PIPELINE DEBUG: Debug message to review: {results=[ADnsAnswer{ipAddress=40.90.137.124, ipVersion=null, dnsTTL=3}, ADnsAnswer{ipAddress=40.90.23.154, ipVersion=null, dnsTTL=3}, ADnsAnswer{ipAddress=40.90.137.120, ipVersion=null, dnsTTL=3}]}

With that string I’m “fighting” to obtain only the IP addresses in the other post, Pipeline rule, multiple values using regex function, only possible to return value ["0"], but with no luck too, :frowning:

Thanks for all your help!

Regards,
Alejandro

It might make more sense to fight this on the client side - Silkservice isn’t one that you usually see associated with Graylog - Beats and nxlog allow you to manage configs form Graylog … but whatever - you likely already considered that. If you can trim SilkETW to give you data in a better format where it breaks out fields Looks like silk ETW can do this with Yara… Then you have better data to put the pipleine. This would shift some processing to the DNS server which may or may not be good with high volume DNS logs.

Thanks for your help @tmacgbay.

I decided to use SilkETW because it is the only free option that I found to collect ETW Windows Events.

At my understanding, Beats can’t collect ETW Windows Events by their own (that is the reason I’m using SilkETW to send the events to Graylog using WinlogBeat), and NXLog has a module to collect ETW, but it is not free.

For me, collecting from the Windows Event Log (in this case with SilkETW), where I can decide when I want to rotate the logs and some other things that give me more control of what is happening, is better than collect from a DNS debug log file, which is the other way to collect Windows DNS Server logs.

In this case, I understand that the problem is at the “matching level” between the DNS LookUp Table and one of the fields of the event (the QNAME in this case), so I cannot thing in a way to modify something at SilkETW collection to solve this problem.

Anyway. Thanks for all your help, really appreciate that. And sorry for my english if it’s not clear, I do my best.

Regards,
Alejandro

poking around: https://www.elastic.co/guide/en/beats/winlogbeat/current/configuration-winlogbeat-options.html#configuration-winlogbeat-options-event_logs-name

Hi @tmacgbay,

Thanks for that post, the commands on it are very useful to obtain the exact Channel names in Windows and to know how to configure them on Winlogbeat, :slight_smile:

About Microsoft Windows DNS Server logs, the problem is that normally, the only available Channels on Windows for DNS Server are these:
LogName : DNS Server
LogName : Microsoft-Windows-DNSServer/Audit
And they are “about” zone transfers, add or delete records internally, etc., but not about resolutions.

The log for resolutions is this one, but is not available by default, and if you enable it, you can collect events from it:

In this post it is explained, and they talk about ETW:


And in this other post they talk about SilkETW, which is the option I’m using right now:

That are te posts I read to decide to go for that option. I understand that is not the common one, but for my understanding, is the best right now.

I’m going to try to figure out how to obtain all the IPs from the QNAME field, maybe with the help of other tool before enter to Graylog.

Thanks again for all your help.

Regards,
Alejandro

Spent too long on this, here is what I came up with:

let lookup_IPs_json = parse_json(to_string(lookup("DNS_QNAME_IP", to_string(message.QNAME))));
debug(concat("Initial json: ", to_string(lookup_IPs_json)));

let the_ips = select_jsonpath(json: lookup_IPs_json, paths: { QNAME_IP: "$['string list value']" });
debug(concat("Set to fields: ", to_string(the_ips)));

set_fields(the_ips);

Hi @tmacgbay,

First, thanks to squeeze your brain to help me, really appreciate that, :slight_smile:

Unfortunately, the fields didn’t get the expected info, :frowning:

I think the problem is that the result of the DNS LookUp Table is not a JSON string.

To check that, I modified the code that you send in your last answer, like this:

let lookup_IPs = lookup("DNS_QNAME_IP", to_string($message.QNAME));
let lookup_IPs_json = parse_json(to_string(lookup_IPs));
debug(concat("String json pre parse_json: ", to_string(lookup_IPs)));
debug(concat("String json: ", to_string(lookup_IPs_json)));
debug(concat("String json is json?: ", to_string(is_json(to_string(lookup_IPs)))));
let the_ips = select_jsonpath(json: lookup_IPs_json, paths: { QNAME_IP_address: "$['string list value']" });
debug(concat("String json to fields: ", to_string(the_ips))); 

And for an example, in the logs, you can see this DEBUG:

2020-04-15T16:05:54.906-03:00 INFO  [Function] PIPELINE DEBUG: String json pre parse_json: {results=[ADnsAnswer{ipAddress=151.101.0.133, ipVersion=null, dnsTTL=29}, ADnsAnswer{ipAddress=151.101.64.133, ipVersion=null, dnsTTL=29}, ADnsAnswer{ipAddress=151.101.128.133, ipVersion=null, dnsTTL=29}, ADnsAnswer{ipAddress=151.101.192.133, ipVersion=null, dnsTTL=29}]}
2020-04-15T16:05:54.906-03:00 INFO  [Function] PIPELINE DEBUG: String json:
2020-04-15T16:05:54.906-03:00 INFO  [Function] PIPELINE DEBUG: String json is json?: false
2020-04-15T16:05:54.906-03:00 INFO  [Function] PIPELINE DEBUG: String json to fields: {QNAME_IP_address=null}

I did additional troubleshooting, and at my understanding, the problem is with the function to_string used over the multi.value lookup value of a DNS LookUp Table.

Let me explain what I’m saying:

Using this code in a pipeline rule to debug:

set_field("QNAME_IPs_Prueba", multiIPs.results);
set_field("QNAME_IPs_Text", to_string(multiIPs.results));
set_field("QNAME_IPs_json", is_json(multiIPs.results));

The result is this one:

I used this page, https://jsonlint.com/, to evaluate the string of “QNAME_IPs_Prueba”, which is recognized as JSON, and I used the same page to evaluate the string of “QNAME_IPs_Text”, which is not recognized as JSON.

But for Graylog, the evaluation to know if the string is JSON is false in both cases, converting previously to a string or not, :frowning:

And additionally, the only way to use the function parse_json with the variable “multiIP.results” is using the function to_string, which in theory “destroy” the JSON format:

So, I don’t know how to solve this, And I don’t understand why to_string_function convert the content of “QNAME_IPs_Prueba” in “QNAME_IPs_Text”.

Can anybody have a clue to help me?

Thanks in advance for your help.

Regards,
Alejandro

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