CVE-2020-10745: pytests: hand-rolled invalid dns/nbt packet tests

The client libraries don't allow us to make packets that are broken in
certain ways, so we need to construct them as byte strings.

These tests all fail at present, proving the server is rendered
unresponsive, which is the crux of CVE-2020-10745.

BUG: https://bugzilla.samba.org/show_bug.cgi?id=14378

Signed-off-by: Douglas Bagnall <douglas.bagnall@catalyst.net.nz>

3 files changed, 223 insertions, 0 deletions

diff --git a/python/samba/tests/dns_packet.py b/python/samba/tests/dns_packet.py
new file mode 100644
index 00000000000..c4f843eb613
--- /dev/null
+++ b/python/samba/tests/dns_packet.py
@@ -0,0 +1,211 @@
+# Tests of malformed DNS packets
+# Copyright (C) Catalyst.NET ltd
+#
+# written by Douglas Bagnall <douglas.bagnall@catalyst.net.nz>
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation; either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+"""Sanity tests for DNS and NBT server parsing.
+
+We don't use a proper client library so we can make improper packets.
+"""
+
+import os
+import struct
+import socket
+import select
+from samba.dcerpc import dns, nbt
+
+from samba.tests import TestCase
+
+
+def _msg_id():
+    while True:
+        for i in range(1, 0xffff):
+            yield i
+
+
+SERVER = os.environ['SERVER_IP']
+SERVER_NAME = f"{os.environ['SERVER']}.{os.environ['REALM']}"
+TIMEOUT = 0.5
+
+
+def encode_netbios_bytes(chars):
+    """Even RFC 1002 uses distancing quotes when calling this "compression"."""
+    out = []
+    chars = (chars + b'                   ')[:16]
+    for c in chars:
+        out.append((c >> 4) + 65)
+        out.append((c & 15) + 65)
+    return bytes(out)
+
+
+class TestDnsPacketBase(TestCase):
+    msg_id = _msg_id()
+
+    def tearDown(self):
+        # we need to ensure the DNS server is responsive before
+        # continuing.
+        for i in range(40):
+            ok = self._known_good_query()
+            if ok:
+                return
+        print(f"the server is STILL unresponsive after {40 * TIMEOUT} seconds")
+
+    def decode_reply(self, data):
+        header = data[:12]
+        id, flags, n_q, n_a, n_rec, n_exta = struct.unpack('!6H',
+                                                           header)
+        return {
+            'rcode': flags & 0xf
+        }
+
+    def construct_query(self, names):
+        """Create a query packet containing one query record.
+
+        *names* is either a single string name in the usual dotted
+        form, or a list of names. In the latter case, each name can
+        be a dotted string or a list of byte components, which allows
+        dots in components. Where I say list, I mean non-string
+        iterable.
+
+        Examples:
+
+        # these 3 are all the same
+        "example.com"
+        ["example.com"]
+        [[b"example", b"com"]]
+
+        # this is three names in the same request
+        ["example.com",
+         [b"example", b"com", b"..!"],
+         (b"first component", b" 2nd component")]
+        """
+        header = struct.pack('!6H',
+                             next(self.msg_id),
+                             0x0100,       # query, with recursion
+                             len(names),   # number of queries
+                             0x0000,       # no answers
+                             0x0000,       # no records
+                             0x0000,       # no extra records
+        )
+        tail = struct.pack('!BHH',
+                           0x00,         # root node
+                           self.qtype,
+                           0x0001,       # class IN-ternet
+        )
+        encoded_bits = []
+        for name in names:
+            if isinstance(name, str):
+                bits = name.encode('utf8').split(b'.')
+            else:
+                bits = name
+
+            for b in bits:
+                encoded_bits.append(b'%c%s' % (len(b), b))
+            encoded_bits.append(tail)
+
+        return header + b''.join(encoded_bits)
+
+    def _test_query(self, names=(), expected_rcode=None):
+
+        if isinstance(names, str):
+            names = [names]
+
+        packet = self.construct_query(names)
+        s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+        s.sendto(packet, self.server)
+        r, _, _ = select.select([s], [], [], TIMEOUT)
+        s.close()
+        # It is reasonable to not reply to these packets (Windows
+        # doesn't), but it is not reasonable to render the server
+        # unresponsive.
+        if r != [s]:
+            ok = self._known_good_query()
+            self.assertTrue(ok, f"the server is unresponsive")
+
+    def _known_good_query(self):
+        if self.server[1] == 53:
+            name = SERVER_NAME
+            expected_rcode = dns.DNS_RCODE_OK
+        else:
+            name = [encode_netbios_bytes(b'nxdomain'), b'nxdomain']
+            expected_rcode = nbt.NBT_RCODE_NAM
+
+        packet = self.construct_query([name])
+        s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+        s.sendto(packet, self.server)
+        r, _, _ = select.select([s], [], [], TIMEOUT)
+        if r != [s]:
+            s.close()
+            return False
+
+        data, addr = s.recvfrom(4096)
+        s.close()
+        rcode = self.decode_reply(data)['rcode']
+        return expected_rcode == rcode
+
+
+class TestDnsPackets(TestDnsPacketBase):
+    server = (SERVER, 53)
+    qtype = 1     # dns type A
+
+    def _test_many_repeated_components(self, label, n, expected_rcode=None):
+        name = [label] * n
+        self._test_query([name],
+                         expected_rcode=expected_rcode)
+
+    def test_127_very_dotty_components(self):
+        label = b'.' * 63
+        self._test_many_repeated_components(label, 127)
+
+    def test_127_half_dotty_components(self):
+        label = b'x.' * 31 + b'x'
+        self._test_many_repeated_components(label, 127)
+
+
+class TestNbtPackets(TestDnsPacketBase):
+    server = (SERVER, 137)
+    qtype = 0x20  # NBT_QTYPE_NETBIOS
+
+    def _test_nbt_encode_query(self, names, *args, **kwargs):
+        if isinstance(names, str):
+            names = [names]
+
+        nbt_names = []
+        for name in names:
+            if isinstance(name, str):
+                bits = name.encode('utf8').split(b'.')
+            else:
+                bits = name
+
+            encoded = [encode_netbios_bytes(bits[0])]
+            encoded.extend(bits[1:])
+            nbt_names.append(encoded)
+
+        self._test_query(nbt_names, *args, **kwargs)
+
+    def _test_many_repeated_components(self, label, n, expected_rcode=None):
+        name = [label] * n
+        name[0] = encode_netbios_bytes(label)
+        self._test_query([name],
+                         expected_rcode=expected_rcode)
+
+    def test_127_very_dotty_components(self):
+        label = b'.' * 63
+        self._test_many_repeated_components(label, 127)
+
+    def test_127_half_dotty_components(self):
+        label = b'x.' * 31 + b'x'
+        self._test_many_repeated_components(label, 127)
diff --git a/selftest/knownfail.d/dns_packet b/selftest/knownfail.d/dns_packet
new file mode 100644
index 00000000000..6e2e5a699de
--- /dev/null
+++ b/selftest/knownfail.d/dns_packet
@@ -0,0 +1,2 @@
+samba.tests.dns_packet.samba.tests.dns_packet.TestDnsPackets.test_127_very_dotty_components
+samba.tests.dns_packet.samba.tests.dns_packet.TestNbtPackets.test_127_very_dotty_components
diff --git a/source4/selftest/tests.py b/source4/selftest/tests.py
index cbb38f0ee62..36a2f126b33 100755
--- a/source4/selftest/tests.py
+++ b/source4/selftest/tests.py
@@ -431,6 +431,16 @@ plantestsuite_loadlist("samba.tests.dns_wildcard", "ad_dc", [python, os.path.joi
 
 plantestsuite_loadlist("samba.tests.dns_invalid", "ad_dc", [python, os.path.join(srcdir(), "python/samba/tests/dns_invalid.py"), '$SERVER_IP', '--machine-pass', '-U"$USERNAME%$PASSWORD"', '--workgroup=$DOMAIN', '$LOADLIST', '$LISTOPT'])
 
+plantestsuite_loadlist("samba.tests.dns_packet",
+                       "ad_dc",
+                       [python,
+                        '-msamba.subunit.run',
+                        '$LOADLIST',
+                        "$LISTOPT"
+                        "samba.tests.dns_packet"
+                       ])
+
+
 for t in smbtorture4_testsuites("dns_internal."):
     plansmbtorture4testsuite(t, "ad_dc_default:local", '//$SERVER/whavever')