[1288] added a sanity check: making sure the question section has exactly

one question.

src/bin/xfrout/tests/xfrout_test.py.in

@@ -104,15 +104,15 @@ class TestXfroutSession(unittest.TestCase):
     def message_has_tsig(self, msg):
         return msg.get_tsig_record() is not None
 
-    def create_request_data(self, with_tsig=False):
+    def create_request_data(self, with_question=True, with_tsig=False):
         msg = Message(Message.RENDER)
         query_id = 0x1035
         msg.set_qid(query_id)
         msg.set_opcode(Opcode.QUERY())
         msg.set_rcode(Rcode.NOERROR())
-        query_question = Question(Name("example.com"), RRClass.IN(),
-                                  RRType.AXFR())
-        msg.add_question(query_question)
+        if with_question:
+            msg.add_question(Question(Name("example.com"), RRClass.IN(),
+                                      RRType.AXFR()))
 
         renderer = MessageRenderer()
         if with_tsig:
@@ -131,7 +131,7 @@ class TestXfroutSession(unittest.TestCase):
                                        isc.acl.dns.REQUEST_LOADER.load(
                                            [{"action": "ACCEPT"}]),
                                        {})
-        self.mdata = self.create_request_data(False)
+        self.mdata = self.create_request_data()
         self.soa_rrset = RRset(Name('example.com'), RRClass.IN(), RRType.SOA(),
                                RRTTL(3600))
         self.soa_rrset.add_rdata(Rdata(RRType.SOA(), RRClass.IN(),
@@ -143,8 +143,13 @@ class TestXfroutSession(unittest.TestCase):
         [get_rcode, get_msg] = self.xfrsess._parse_query_message(self.mdata)
         self.assertEqual(get_rcode.to_text(), "NOERROR")
 
+        # Broken request: no question
+        request_data = self.create_request_data(with_question=False)
+        rcode, msg = self.xfrsess._parse_query_message(request_data)
+        self.assertEqual(Rcode.FORMERR(), rcode)
+
         # tsig signed query message
-        request_data = self.create_request_data(True)
+        request_data = self.create_request_data(with_tsig=True)
         # BADKEY
         [rcode, msg] = self.xfrsess._parse_query_message(request_data)
         self.assertEqual(rcode.to_text(), "NOTAUTH")
@@ -181,7 +186,7 @@ class TestXfroutSession(unittest.TestCase):
         self.assertEqual(rcode.to_text(), "REFUSED")
 
         # TSIG signed request
-        request_data = self.create_request_data(True)
+        request_data = self.create_request_data(with_tsig=True)
 
         # If the TSIG check fails, it should not check ACL
         # (If it checked ACL as well, it would just drop the request)

src/bin/xfrout/xfrout.py.in

@@ -169,23 +169,29 @@ class XfroutSession():
 
         # TSIG related checks
         rcode = self._check_request_tsig(msg, mdata)
-
-        if rcode == Rcode.NOERROR():
-            # ACL checks
-            zone_name = msg.get_question()[0].get_name()
-            zone_class = msg.get_question()[0].get_class()
-            acl = self._get_transfer_acl(zone_name, zone_class)
-            acl_result = acl.execute(
-                isc.acl.dns.RequestContext(self._remote,
-                                           msg.get_tsig_record()))
-            if acl_result == DROP:
-                logger.info(XFROUT_QUERY_DROPPED, zone_name, zone_class,
-                            self._remote[0], self._remote[1])
-                return None, None
-            elif acl_result == REJECT:
-                logger.info(XFROUT_QUERY_REJECTED, zone_name, zone_class,
-                            self._remote[0], self._remote[1])
-                return Rcode.REFUSED(), msg
+        if rcode != Rcode.NOERROR():
+            return rcode, msg
+
+        # Make sure the question is valid.  This should be ensured by
+        # the auth server, but since it's far from our xfrout itself,
+        # we check it by ourselves.
+        if msg.get_rr_count(Message.SECTION_QUESTION) != 1:
+            return Rcode.FORMERR(), msg
+
+        # ACL checks
+        zone_name = msg.get_question()[0].get_name()
+        zone_class = msg.get_question()[0].get_class()
+        acl = self._get_transfer_acl(zone_name, zone_class)
+        acl_result = acl.execute(
+            isc.acl.dns.RequestContext(self._remote, msg.get_tsig_record()))
+        if acl_result == DROP:
+            logger.info(XFROUT_QUERY_DROPPED, zone_name, zone_class,
+                        self._remote[0], self._remote[1])
+            return None, None
+        elif acl_result == REJECT:
+            logger.info(XFROUT_QUERY_REJECTED, zone_name, zone_class,
+                        self._remote[0], self._remote[1])
+            return Rcode.REFUSED(), msg
 
         return rcode, msg