From c51c551e6b6165c7023c045a929d52a5362a4d4f Mon Sep 17 00:00:00 2001
From: Otto Moerbeek <otto.moerbeek@open-xchange.com>
Date: Fri, 11 Oct 2019 11:38:50 +0000
Subject: [PATCH] Tests, docs and validation of OOO setting.

Test required some framework work to allow for auths having
more than 1 thread.
---
 .circleci/config.yml                          |   2 +-
 build-scripts/travis.sh                       |   1 +
 docs/manpages/sdig.1.rst                      |   4 +-
 pdns/pdns_recursor.cc                         |  26 ++--
 pdns/recursordist/docs/settings.rst           |   5 +-
 pdns/syncres.hh                               |   2 +-
 .../recursortests.py                          |  83 ++++++++++--
 .../test_OOOTCP.py                            |  56 ++++++++
 .../test_SimpleTCP.py                         | 125 ++++++++++++++++++
 9 files changed, 279 insertions(+), 25 deletions(-)
 create mode 100644 regression-tests.recursor-dnssec/test_OOOTCP.py
 create mode 100644 regression-tests.recursor-dnssec/test_SimpleTCP.py

diff --git a/.circleci/config.yml b/.circleci/config.yml
index 3d6d3ddf3..6a7697b2f 100644
--- a/.circleci/config.yml
+++ b/.circleci/config.yml
@@ -897,7 +897,7 @@ jobs:
       - image: debian:buster
     steps:
       - add-auth-repo
-      - run: apt-get --no-install-recommends install -qq -y pdns-server pdns-backend-bind pdns-tools daemontools authbind jq libfaketime lua-posix moreutils bc virtualenv protobuf-compiler
+      - run: apt-get --no-install-recommends install -qq -y pdns-server pdns-backend-bind pdns-tools daemontools authbind jq libfaketime lua-posix lua-socket moreutils bc virtualenv protobuf-compiler
       - install-recursor-deps
       - run:
           name: Set up authbind
diff --git a/build-scripts/travis.sh b/build-scripts/travis.sh
index aa6934c62..c2d8a3092 100755
--- a/build-scripts/travis.sh
+++ b/build-scripts/travis.sh
@@ -343,6 +343,7 @@ install_recursor() {
     libfaketime \
     libsnmp-dev \
     lua-posix \
+    lua-socket \
     moreutils \
     snmpd"
   run "cd .."
diff --git a/docs/manpages/sdig.1.rst b/docs/manpages/sdig.1.rst
index 8d144357b..cddacafa7 100644
--- a/docs/manpages/sdig.1.rst
+++ b/docs/manpages/sdig.1.rst
@@ -12,6 +12,8 @@ Description
 :program:`sdig` sends a DNS query to *IP-ADDRESS-OR-DOH-URL* on port *PORT* and displays
 the answer in a formatted way.
 If the address starts with an ``h``, it is assumed to be a DoH endpoint, and *PORT* is ignored.
+If qname and qtype are both `-` and tcp is used, multiple lines are read
+form stdin, where each line contains a qname and a type.
 
 Options
 -------
@@ -34,4 +36,4 @@ showflags
 tcp
     Use TCP instead of UDP to send the query.
 xpf *XPFCODE* *XPFVERSION* *XPFPROTO* *XPFSRC* *XPFSRC*
-	Send an *XPF* additional with these parameters.
\ No newline at end of file
+	Send an *XPF* additional with these parameters.
diff --git a/pdns/pdns_recursor.cc b/pdns/pdns_recursor.cc
index adcbee595..a6077c378 100644
--- a/pdns/pdns_recursor.cc
+++ b/pdns/pdns_recursor.cc
@@ -746,16 +746,16 @@ static void writePid(void)
   }
 }
 
+uint16_t TCPConnection::s_maxInFlight;
+
 TCPConnection::TCPConnection(int fd, const ComboAddress& addr) : data(2, 0), d_remote(addr), d_fd(fd)
 {
-  d_maxInFlight = ::arg().asNum("max-concurrent-requests-per-tcp-connection");
   ++s_currentConnections;
   (*t_tcpClientCounts)[d_remote]++;
 }
 
 TCPConnection::~TCPConnection()
 {
-  g_log<<Logger::Warning<<"closing socket for TCPConnection " <<d_fd <<endl;
   try {
     if(closesocket(d_fd) < 0)
       g_log<<Logger::Error<<"Error closing socket for TCPConnection"<<endl;
@@ -1751,8 +1751,9 @@ static void startDoResolve(void *p)
         hadError=false;
 
       // update tcp connection status, closing if needed and doing the fd multiplexer accounting
-
-      dc->d_tcpConnection->d_requestsInFlight--;
+      if  (dc->d_tcpConnection->d_requestsInFlight > 0) {
+        dc->d_tcpConnection->d_requestsInFlight--;
+      }
 
       // In the code below, we try to remove the fd from the set, but
       // we don't know if another mthread already did the remove, so we can get a
@@ -1779,11 +1780,11 @@ static void startDoResolve(void *p)
         else {
           Utility::gettimeofday(&g_now, 0); // needs to be updated
           struct timeval ttd = g_now;
-          if (dc->d_tcpConnection->d_requestsInFlight == dc->d_tcpConnection->d_maxInFlight - 1) {
+          // If we cross from max to max-1 in flight requests, the fd was not listened to, add it back
+          if (dc->d_tcpConnection->d_requestsInFlight == TCPConnection::s_maxInFlight - 1) {
             // A read error might have happened. If we add the fd back, it will most likely error again.
             // This is not a big issue, the next handleTCPClientReadable() will see another read error
             // and take action.
-            cerr << "Reenabling " << dc->d_socket << ' ' << dc->d_tcpConnection->d_requestsInFlight << endl;
             ttd.tv_sec += g_tcpTimeout;
             t_fdm->addReadFD(dc->d_socket, handleRunningTCPQuestion, dc->d_tcpConnection, &ttd);
           } else {
@@ -2165,8 +2166,7 @@ static void handleRunningTCPQuestion(int fd, FDMultiplexer::funcparam_t& var)
         ++g_stats.qcounter;
         ++g_stats.tcpqcounter;
         ++conn->d_requestsInFlight;
-        if (conn->d_requestsInFlight >= conn->d_maxInFlight) {
-          cerr << "Disabling " << fd << ' ' << conn->d_requestsInFlight << endl;
+        if (conn->d_requestsInFlight >= TCPConnection::s_maxInFlight) {
           t_fdm->removeReadFD(fd); // should no longer awake ourselves when there is data to read
         } else {
           Utility::gettimeofday(&g_now, 0); // needed?
@@ -4013,6 +4013,16 @@ static int serviceMain(int argc, char*argv[])
   g_numThreads = g_numDistributorThreads + g_numWorkerThreads;
   g_maxMThreads = ::arg().asNum("max-mthreads");
 
+
+  int64_t maxInFlight = ::arg().asNum("max-concurrent-requests-per-tcp-connection");
+  if (maxInFlight < 1 || maxInFlight > USHRT_MAX || maxInFlight >= g_maxMThreads) {
+    g_log<<Logger::Warning<<"Asked to run with illegal max-concurrent-requests-per-tcp-connection, setting to default (10)"<<endl;
+    TCPConnection::s_maxInFlight = 10;
+  } else {
+    TCPConnection::s_maxInFlight = maxInFlight;
+  }
+    
+
   g_gettagNeedsEDNSOptions = ::arg().mustDo("gettag-needs-edns-options");
 
   g_statisticsInterval = ::arg().asNum("statistics-interval");
diff --git a/pdns/recursordist/docs/settings.rst b/pdns/recursordist/docs/settings.rst
index a76958ffe..a756210ce 100644
--- a/pdns/recursordist/docs/settings.rst
+++ b/pdns/recursordist/docs/settings.rst
@@ -868,7 +868,10 @@ Maximum number of seconds to cache an item in the DNS cache, no matter what the
 -  Integer
 -  Default: 10
 
-Maximum number of requests handled concurrently per tcp connection.
+Maximum number of incoming requests handled concurrently per tcp
+connection. This number must be larger than 0 and smaller than 65536
+and also smaller than `max-mthreads`.
+
 
 .. _setting-max-mthreads:
 
diff --git a/pdns/syncres.hh b/pdns/syncres.hh
index ee9571512..4f867d551 100644
--- a/pdns/syncres.hh
+++ b/pdns/syncres.hh
@@ -1014,7 +1014,7 @@ public:
   uint16_t bytesread{0};
   uint16_t d_requestsInFlight{0}; // number of mthreads spawned for this connection
   // The max number of concurrent TCP requests we're willing to process
-  uint16_t d_maxInFlight;
+  static uint16_t s_maxInFlight;
   static unsigned int getCurrentConnections() { return s_currentConnections; }
 private:
   const int d_fd;
diff --git a/regression-tests.recursor-dnssec/recursortests.py b/regression-tests.recursor-dnssec/recursortests.py
index cfbd87bad..ec611965f 100644
--- a/regression-tests.recursor-dnssec/recursortests.py
+++ b/regression-tests.recursor-dnssec/recursortests.py
@@ -114,6 +114,9 @@ sort.example.                      3600 IN A     17.38.42.80
 sort.example.                      3600 IN A     192.168.0.1
 sort.example.                      3600 IN A     17.238.240.5
 sort.example.                      3600 IN MX    25 mx
+
+delay.example.                     3600 IN NS   ns1.delay.example.
+ns1.delay.example.                 3600 IN A    {prefix}.16
         """,
         'secure.example': """
 secure.example.          3600 IN SOA  {soa}
@@ -232,6 +235,13 @@ undelegated.insecure.example.        3600 IN NS   ns1.undelegated.insecure.examp
 
 node1.undelegated.insecure.example.  3600 IN A    192.0.2.22
         """,
+
+        'delay.example': """
+delay.example.                       3600 IN SOA  {soa}
+delay.example.                       3600 IN NS n1.delay.example.
+ns1.delay.example.                   3600 IN A    {prefix}.16
+*.delay.example.                     0    LUA TXT ";" "local socket=require('socket')" "socket.sleep(tonumber(qname:getRawLabels()[1])/10)" "return 'a'"
+        """
     }
 
     # The private keys for the zones (note that DS records should go into
@@ -296,14 +306,25 @@ PrivateKey: Ep9uo6+wwjb4MaOmqq7LHav2FLrjotVOeZg8JT1Qk04=
     # is a list of zones hosted on that IP. Note that delegations should
     # go into the _zones's zonecontent
     _auth_zones = {
-        '8': ['ROOT'],
-        '9': ['secure.example', 'islandofsecurity.example'],
-        '10': ['example'],
-        '11': ['example'],
-        '12': ['bogus.example', 'undelegated.secure.example', 'undelegated.insecure.example'],
-        '13': ['insecure.example', 'insecure.sub2.secure.example', 'dname-secure.example'],
-        '14': ['optout.example'],
-        '15': ['insecure.optout.example', 'secure.optout.example', 'cname-secure.example']
+        '8': {'threads': 1,
+              'zones': ['ROOT']},
+        '9': {'threads': 1,
+              'zones': ['secure.example', 'islandofsecurity.example']},
+        '10': {'threads': 1,
+               'zones': ['example']},
+        '11': {'threads': 1,
+               'zones': ['example']},
+        '12': {'threads': 1,
+               'zones': ['bogus.example', 'undelegated.secure.example', 'undelegated.insecure.example']},
+        '13': {'threads': 1,
+               'zones': ['insecure.example', 'insecure.sub2.secure.example', 'dname-secure.example']},
+        '14': {'threads': 1,
+               'zones': ['optout.example']},
+        '15': {'threads': 1,
+               'zones': ['insecure.optout.example', 'secure.optout.example', 'cname-secure.example']},
+        # This zone need more threads so that the lua delay code does not cause serialization
+        '16': {'threads': 2,
+               'zones': ['delay.example']}
     }
 
     _auth_cmd = ['authbind',
@@ -342,7 +363,7 @@ options {
         };""" % (zone, zonename))
 
     @classmethod
-    def generateAuthConfig(cls, confdir):
+    def generateAuthConfig(cls, confdir, threads):
         bind_dnssec_db = os.path.join(confdir, 'bind-dnssec.sqlite3')
 
         with open(os.path.join(confdir, 'pdns.conf'), 'w') as pdnsconf:
@@ -360,9 +381,11 @@ query-cache-ttl=0
 log-dns-queries=yes
 log-dns-details=yes
 loglevel=9
+enable-lua-records
 dname-processing=yes
-distributor-threads=1""".format(confdir=confdir,
-                                bind_dnssec_db=bind_dnssec_db))
+distributor-threads={threads}""".format(confdir=confdir,
+                                        bind_dnssec_db=bind_dnssec_db,
+                                        threads=threads))
 
         pdnsutilCmd = [os.environ['PDNSUTIL'],
                        '--config-dir=%s' % confdir,
@@ -405,12 +428,14 @@ distributor-threads=1""".format(confdir=confdir,
     @classmethod
     def generateAllAuthConfig(cls, confdir):
         if cls._auth_zones:
-            for auth_suffix, zones in cls._auth_zones.items():
+            for auth_suffix, zoneinfo in cls._auth_zones.items():
+                threads = zoneinfo['threads']
+                zones = zoneinfo['zones']
                 authconfdir = os.path.join(confdir, 'auth-%s' % auth_suffix)
 
                 os.mkdir(authconfdir)
 
-                cls.generateAuthConfig(authconfdir)
+                cls.generateAuthConfig(authconfdir, threads)
                 cls.generateAuthNamedConf(authconfdir, zones)
 
                 for zone in zones:
@@ -657,6 +682,38 @@ distributor-threads=1""".format(confdir=confdir,
             message = dns.message.from_wire(data)
         return message
 
+    @classmethod
+    def sendTCPQueries(cls, queries, timeout=2.0):
+        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        if timeout:
+            sock.settimeout(timeout)
+
+        sock.connect(("127.0.0.1", cls._recursorPort))
+        data = []
+        try:
+            for query in queries:
+                wire = query.to_wire()
+                sock.send(struct.pack("!H", len(wire)))
+                sock.send(wire)
+            for i in range(len(queries)):
+                try:
+                    datalen = sock.recv(2)
+                    if datalen:
+                        (datalen,) = struct.unpack("!H", datalen)
+                        data.append(sock.recv(datalen))
+                except socket.timeout as e:
+                    continue
+        except socket.error as e:
+            print("Network error: %s" % (str(e)))
+            data = None
+        finally:
+            sock.close()
+
+        messages = []
+        for d in data:
+            messages.append(dns.message.from_wire(d))
+        return messages
+
     def setUp(self):
         # This function is called before every tests
         return
diff --git a/regression-tests.recursor-dnssec/test_OOOTCP.py b/regression-tests.recursor-dnssec/test_OOOTCP.py
new file mode 100644
index 000000000..bf9471d17
--- /dev/null
+++ b/regression-tests.recursor-dnssec/test_OOOTCP.py
@@ -0,0 +1,56 @@
+import dns
+import os
+import time
+from recursortests import RecursorTest
+
+class testOOOTCP(RecursorTest):
+    _confdir = 'OOOTCP'
+
+    _config_template = """dnssec=off
+"""
+
+    @classmethod
+    def generateRecursorConfig(cls, confdir):
+        super(testOOOTCP, cls).generateRecursorConfig(confdir)
+
+    def testOOOVeryBasic(self):
+        expected = {}
+        queries = []
+        for zone in ['5.delay.example.', '0.delay.example.']:
+            expected[zone] = dns.rrset.from_text(zone, 0, dns.rdataclass.IN, 'TXT', 'a')
+            query = dns.message.make_query(zone, 'TXT', want_dnssec=False)
+            query.flags |= dns.flags.AD
+            queries.append(query)
+
+        ress = self.sendTCPQueries(queries)
+
+        self.assertEqual(len(ress), len(expected))
+
+        i = 0
+        for exp in [expected['0.delay.example.'], expected['5.delay.example.']]:
+            print('ress0')
+            print(ress[i].answer[0].to_text())
+            print('exp')
+            print(exp.to_text())
+            #self.assertMessageIsAuthenticated(ress[i])
+            self.assertRRsetInAnswer(ress[i], exp)
+            #self.assertMatchingRRSIGInAnswer(ress[i], exp)
+            i = i + 1
+
+    def testOOOTimeout(self):
+        expected = {}
+        queries = []
+        for zone in ['25.delay.example.', '1.delay.example.']:
+            query = dns.message.make_query(zone, 'TXT', want_dnssec=False)
+            query.flags |= dns.flags.AD
+            queries.append(query)
+
+        ress = self.sendTCPQueries(queries)
+        self.assertEqual(len(ress), 2)
+        exp = dns.rrset.from_text(zone, 0, dns.rdataclass.IN, 'TXT', 'a')
+        self.assertRRsetInAnswer(ress[0], exp)
+        self.assertRcodeEqual(ress[1], dns.rcode.SERVFAIL)
+
+        # Let the auth timeout happen to not disturb other tests
+        time.sleep(1)
+
diff --git a/regression-tests.recursor-dnssec/test_SimpleTCP.py b/regression-tests.recursor-dnssec/test_SimpleTCP.py
new file mode 100644
index 000000000..60a38f6f1
--- /dev/null
+++ b/regression-tests.recursor-dnssec/test_SimpleTCP.py
@@ -0,0 +1,125 @@
+import dns
+import os
+from recursortests import RecursorTest
+
+class testSimpleTCP(RecursorTest):
+    _confdir = 'SimpleTCP'
+
+    _config_template = """dnssec=validate
+auth-zones=authzone.example=configs/%s/authzone.zone""" % _confdir
+
+    @classmethod
+    def generateRecursorConfig(cls, confdir):
+        authzonepath = os.path.join(confdir, 'authzone.zone')
+        with open(authzonepath, 'w') as authzone:
+            authzone.write("""$ORIGIN authzone.example.
+@ 3600 IN SOA {soa}
+@ 3600 IN A 192.0.2.88
+""".format(soa=cls._SOA))
+        super(testSimpleTCP, cls).generateRecursorConfig(confdir)
+
+    def testSOAs(self):
+        for zone in ['.', 'example.', 'secure.example.']:
+            expected = dns.rrset.from_text(zone, 0, dns.rdataclass.IN, 'SOA', self._SOA)
+            query = dns.message.make_query(zone, 'SOA', want_dnssec=True)
+            query.flags |= dns.flags.AD
+
+            res = self.sendTCPQuery(query)
+
+            self.assertMessageIsAuthenticated(res)
+            self.assertRRsetInAnswer(res, expected)
+            self.assertMatchingRRSIGInAnswer(res, expected)
+
+    def testA(self):
+        expected = dns.rrset.from_text('ns.secure.example.', 0, dns.rdataclass.IN, 'A', '{prefix}.9'.format(prefix=self._PREFIX))
+        query = dns.message.make_query('ns.secure.example', 'A', want_dnssec=True)
+        query.flags |= dns.flags.AD
+
+        res = self.sendTCPQuery(query)
+
+        self.assertMessageIsAuthenticated(res)
+        self.assertRRsetInAnswer(res, expected)
+        self.assertMatchingRRSIGInAnswer(res, expected)
+
+    def testDelegation(self):
+        query = dns.message.make_query('example', 'NS', want_dnssec=True)
+        query.flags |= dns.flags.AD
+
+        expectedNS = dns.rrset.from_text('example.', 0, 'IN', 'NS', 'ns1.example.', 'ns2.example.')
+
+        res = self.sendTCPQuery(query)
+
+        self.assertMessageIsAuthenticated(res)
+        self.assertRRsetInAnswer(res, expectedNS)
+
+    def testBogus(self):
+        query = dns.message.make_query('ted.bogus.example', 'A', want_dnssec=True)
+
+        res = self.sendTCPQuery(query)
+
+        self.assertRcodeEqual(res, dns.rcode.SERVFAIL)
+
+    def testAuthZone(self):
+        query = dns.message.make_query('authzone.example', 'A', want_dnssec=True)
+
+        expectedA = dns.rrset.from_text('authzone.example.', 0, 'IN', 'A', '192.0.2.88')
+
+        res = self.sendTCPQuery(query)
+
+        self.assertRcodeEqual(res, dns.rcode.NOERROR)
+        self.assertRRsetInAnswer(res, expectedA)
+
+    def testLocalhost(self):
+        queryA = dns.message.make_query('localhost', 'A', want_dnssec=True)
+        expectedA = dns.rrset.from_text('localhost.', 0, 'IN', 'A', '127.0.0.1')
+
+        queryPTR = dns.message.make_query('1.0.0.127.in-addr.arpa', 'PTR', want_dnssec=True)
+        expectedPTR = dns.rrset.from_text('1.0.0.127.in-addr.arpa.', 0, 'IN', 'PTR', 'localhost.')
+
+        resA = self.sendTCPQuery(queryA)
+        resPTR = self.sendTCPQuery(queryPTR)
+
+        self.assertRcodeEqual(resA, dns.rcode.NOERROR)
+        self.assertRRsetInAnswer(resA, expectedA)
+
+        self.assertRcodeEqual(resPTR, dns.rcode.NOERROR)
+        self.assertRRsetInAnswer(resPTR, expectedPTR)
+
+    def testIslandOfSecurity(self):
+        query = dns.message.make_query('cname-to-islandofsecurity.secure.example.', 'A', want_dnssec=True)
+
+        expectedCNAME = dns.rrset.from_text('cname-to-islandofsecurity.secure.example.', 0, 'IN', 'CNAME', 'node1.islandofsecurity.example.')
+        expectedA = dns.rrset.from_text('node1.islandofsecurity.example.', 0, 'IN', 'A', '192.0.2.20')
+
+        res = self.sendTCPQuery(query)
+
+        self.assertRcodeEqual(res, dns.rcode.NOERROR)
+        self.assertRRsetInAnswer(res, expectedA)
+
+
+    def testVeryBasicPipeline(self):
+        # This test does not enforce order, it will accept replies in any order. So
+        # it does not actually test OOO behaviour.
+        expected = {}
+        queries = []
+        for zone in ['.', 'example.', 'secure.example.']:
+            expected[zone] = dns.rrset.from_text(zone, 0, dns.rdataclass.IN, 'SOA', self._SOA)
+            query = dns.message.make_query(zone, 'SOA', want_dnssec=True)
+            query.flags |= dns.flags.AD
+            queries.append(query)
+
+        expected['ns.secure.example.'] = dns.rrset.from_text('ns.secure.example.', 0, dns.rdataclass.IN, 'A', '{prefix}.9'.format(prefix=self._PREFIX))
+        query = dns.message.make_query('ns.secure.example', 'A', want_dnssec=True)
+        query.flags |= dns.flags.AD
+        queries.append(query)
+
+        ress = self.sendTCPQueries(queries)
+
+        self.assertEqual(len(ress), len(expected))
+
+        for res in ress:
+            exp = expected[res.question[0].name.to_text()]
+            self.assertMessageIsAuthenticated(res)
+            self.assertRRsetInAnswer(res, exp)
+            self.assertMatchingRRSIGInAnswer(res, exp)
+
-- 
2.40.0