Recently in class, we were discussing detection strategies for DNS cache poisoning attacks. One of the ideas was to look for duplicate DNS replies to the same request. This would be pretty difficult with signature detection tools and flow data wouldn't have enough details. Zeek would be perfect for this type of detection. Let's write a script!

For our script, we are going to focus on the DNS transaction ID and query. You could refine the matching criteria further by including the source and destination IP addresses and ports but the transaction ID and query seem to work fine in my testing. We will start by declaring a global variable to store the DNS query (string) and the DNS transaction ID (integer):

global query_and_id: set[string, int];

We want to be notified if Zeek sees any duplicate DNS query replies. The dns_query_reply event is "Generated for each entry in the Question section of a DNS reply". Sounds good, let's try it:

event dns_query_reply (c: connection, msg: dns_msg, query: string, qtype: count, qclass: count)

In our function, we check to see if Zeek has already seen a DNS reply against a previously stored query and transaction ID. If yes, alert us (we are using a simple print statement to the screen but you could easily substitute your favorite notice policy).

{
  if([c$dns$query, c$dns$trans_id] in query_and_id){
    print fmt ("Possible DNS cache poisoning attempt --> Source IP: %s, Destination IP: %s, Query: %s", c$id$orig_h, c$id$resp_h, c$dns$query);
    return;
  }

Then, if Zeek has not seen this query and transaction ID pairing before, we have Zeek store the query and transaction ID pairing in our global variable.

  if(!([c$dns$query, c$dns$trans_id] in query_and_id)){
    add query_and_id[c$dns$query, c$dns$trans_id];
  }
}

And that's all there is to it. :)

In a production environment, the query_and_id variable could very quickly get extremely large and become inefficient. We can mitigate this issue by using the &write_expire option on the query_and_id variable. Since the spoofed and real DNS replies would need to be with in seconds of each other, you could expire entries relatively quickly. Let's use 1 minute for now. Here is the entire script:

global query_and_id: set[string, int] &write_expire=1min;

event dns_query_reply (c: connection, msg: dns_msg, query: string, qtype: count, qclass: count)
{
  if([c$dns$query, c$dns$trans_id] in query_and_id){
    print fmt ("Possible DNS cache poisoning attempt --> Source IP: %s, Destination IP: %s, Query: %s", c$id$orig_h, c$id$resp_h, c$dns$query);
    return;
  }
  if(!([c$dns$query, c$dns$trans_id] in query_and_id)){
    add query_and_id[c$dns$query, c$dns$trans_id];
  }
}

Yes, you could do this correlation in your SIEM but as Dave points out in this blog post, why not perform the correlation as close to the data source as possible? Hopefully, this helps.

Andy Laman, a principal consultant with A4 InfoSec, has more than 25 years of information technology and security experience in multiple industries. Andy is a course contributor and teaches SEC503: Intrusion Detection In-Depth, for the SANS Institute. In addition to the CISSP, Andy holds multiple GIAC certifications including the prestigious GIAC Security Expert (GSE #142) certification as well as multiple other industry certifications.