Make sure that a valid TCP over DNS message can be decoded

scapy/layers/dns.py

@@ -437,6 +437,23 @@ def compress(self):
         """Return the compressed DNS packet (using `dns_compress()`"""
         return dns_compress(self)
 
+    def pre_dissect(self, s):
+        """
+        Check that a valid DNS over TCP message can be decoded
+        """
+        if isinstance(self.underlayer, TCP):
+            # Compute the length of the DNS packet
+            dns_len = 0
+            if len(s) >= 2:
+                dns_len = struct.unpack("!H", s[:2])[0]
+
+            # Check if the length is valid
+            if dns_len < 14 or len(s) < dns_len:
+                message = "Malformed DNS message: invalid length!"
+                warning(message)
+                raise Scapy_Exception(message)
+        return s
+
 
 # https://www.iana.org/assignments/dns-parameters/dns-parameters.xhtml#dns-parameters-4
 dnstypes = {

test/regression.uts

@@ -7886,6 +7886,15 @@ p = DNS(raw(DNS(id=1,ra=1,an=DNSRR(rrname='secdev', type='TXT', rdata=["sweet",
 assert p[DNS].an.rdata == [b"sweet", b"celestia"]
 assert raw(p) == b'\x00\x01\x01\x80\x00\x00\x00\x01\x00\x00\x00\x00\x06secdev\x00\x00\x10\x00\x01\x00\x00\x00\x01\x00\x0f\x05sweet\x08celestia'
 
+= DNS - Malformed DNS over TCP message
+
+try:
+    p = IP(raw(IP()/TCP()/DNS(length=24, qdcount=1)))
+    assert False
+except Scapy_Exception as e:
+    assert str(e) == "Malformed DNS message: invalid length!"
+
+
 = Layer binding
 
 * Test DestMACField & DestIPField