dns: stop recording unknown names in the answering machine

scapy/layers/dns.py

@@ -13,7 +13,6 @@
 """
 
 import abc
-import collections
 import operator
 import itertools
 import socket
@@ -1674,7 +1673,7 @@ def normk(k):
                 k += b"."
             return k
 
-        self.match = collections.defaultdict(lambda: (joker, joker6))
+        self.match = {}
         if match:
             if isinstance(match, (list, set)):
                 self.match.update({normk(k): (None, None) for k in match})
@@ -1808,7 +1807,10 @@ def make_reply(self, req):
                 # A or AAAA
                 if rq.qtype == 28:
                     # AAAA
-                    rdata = self.match[rqname][1]
+                    try:
+                        rdata = self.match[rqname][1]
+                    except KeyError:
+                        rdata = self.joker6
                     if rdata is None and not self.relay:
                         # 'None' resolves to the default IPv6
                         iface = resolve_iface(self.optsniff.get("iface", conf.iface))
@@ -1824,7 +1826,10 @@ def make_reply(self, req):
                         resp[IPv6].src = rdata
                 elif rq.qtype == 1:
                     # A
-                    rdata = self.match[rqname][0]
+                    try:
+                        rdata = self.match[rqname][0]
+                    except KeyError:
+                        rdata = self.joker
                     if rdata is None and not self.relay:
                         # 'None' resolves to the default IPv4
                         iface = resolve_iface(self.optsniff.get("iface", conf.iface))

test/answering_machines.uts

@@ -137,6 +137,18 @@ assert DNS_am().make_reply(
     Ether()/IP()/UDP()/DNS(b'q\xa04\x00\x00\xa0\x01\x00\xf3\x00\x01\x04\x01y')
 ) is None
 
+def check_DNS_am_unknown_names_not_cached(qtypes):
+    am = DNS_am(joker=False)
+    for i, qtype in enumerate(qtypes):
+        am.make_reply(
+            Ether()/IP()/UDP()/DNS(
+                qd=DNSQR(qname="unknown-%d.example" % i, qtype=qtype)
+            )
+        )
+    assert not am.match
+
+check_DNS_am_unknown_names_not_cached(("A", "AAAA"))
+
 = LLMNR_am
 def check_LLMNR_am_am_reply(packet):
     # assert packet[Ether].src == get_if_hwaddr(conf.iface)