Merge branch 'invalid_pktlen_checks_testcase' of /git/repositories/labs/knot

tests-extra/functional/tests/security/dnsheader/test.py

+#!/usr/bin/env python3
+
+'''DNS packet header parsing tests. '''
+
+import socket
+import dnstest
+
+t = dnstest.DnsTest()
+knot = t.server("knot")
+zone = t.zone("example.com.")
+t.link(zone, knot)
+
+# Update configuration
+knot.ratelimit = 5 # Check for crashes also in rate-limit code
+
+t.start()
+
+# Packet lengths shorter than DNS header
+data = '\x00'
+max_len = (12 + 5) # Header + minimal question size
+udp_socket = knot.create_sock(socket.SOCK_DGRAM)
+for i in range(1, max_len):
+	knot.send_raw(data * i, udp_socket)
+udp_socket.close()
+
+# Check if the server is still alive 
+resp = knot.dig("example.com", "SOA", timeout=5, tries=1)
+resp.check(rcode="NOERROR")
+
+t.end()

tests-extra/functional/tools/dnstest.py

@@ -640,6 +640,21 @@ class DnsServer(object):
         raise Exception("Can't query %s for %s %s %s." % \
                         (self.name, rname, rclass, rtype))
 
+
+    def create_sock(self, socket_type):
+        family = socket.AF_INET
+        if self.ip == 6:
+            family = socket.AF_INET6
+        return socket.socket(family, socket_type)
+
+    def send_raw(self, data, sock=None):
+        if sock is None:
+            sock = self.create_sock(socket.SOCK_DGRAM)
+        sent = sock.sendto(bytes(data, 'utf-8'), (self.addr, self.port))
+        if sent != len(data):
+            raise Exception("Can't send RAW data (%d bytes) to %s." % (len(data), self.name))
+        return 0
+
     def zones_wait(self, zones):
         for zone in zones:
             # Try to get SOA record with NOERROR.
@@ -824,7 +839,7 @@ class Knot(DnsServer):
         if value == True:
             conf.item(name, "on")
         elif value:
-            if value[:2] == "0x":
+            if isinstance(value, int) or value[:2] == "0x":
                 conf.item(name, value)
             else:
                 conf.item_str(name, value)
@@ -835,6 +850,7 @@ class Knot(DnsServer):
         self._on_str_hex(s, "identity", self.ident)
         self._on_str_hex(s, "version", self.version)
         self._on_str_hex(s, "nsid", self.nsid)
+        self._on_str_hex(s, "rate-limit", self.ratelimit)
         s.item_str("storage", self.dir)
         s.item_str("rundir", self.dir)
         s.end()