ixfrdist: handle incoming NOTIFY packets

diff --git a/pdns/ixfrdist.cc b/pdns/ixfrdist.cc
index fde9933454817e27eb608463790064ae122ac65b..52b6599d1c63751332f92d81ccaccb8ef0e82547 100644 (file)
--- a/pdns/ixfrdist.cc
+++ b/pdns/ixfrdist.cc
@@ -20,6 +20,7 @@
  * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
  */
 #include "dns.hh"
+#include "dnsparser.hh"
 #ifdef HAVE_CONFIG_H
 #include "config.h"
 #endif
@@ -152,6 +153,10 @@ static map<DNSName, ixfrdistdomain_t> g_domainConfigs;
 // Map domains and their data
 static LockGuarded<std::map<DNSName, std::shared_ptr<ixfrinfo_t>>> g_soas;
 
+// Queue of received NOTIFYs, already verified against their master IPs
+// Lazily implemented as a set
+static LockGuarded<std::set<DNSName>> g_notifiesReceived;
+
 // Condition variable for TCP handling
 static std::condition_variable g_tcpHandlerCV;
 static std::queue<pair<int, ComboAddress>> g_tcpRequestFDs;
@@ -161,7 +166,8 @@ namespace po = boost::program_options;
 
 static bool g_exiting = false;
 
-static NetmaskGroup g_acl;
+static NetmaskGroup g_acl;            // networks that can QUERY us
+static NetmaskGroup g_notifySources;  // networks (well, IPs) that can NOTIFY us
 static bool g_compress = false;
 
 static ixfrdistStats g_stats;
@@ -348,7 +354,9 @@ static void updateThread(const string& workdir, const uint16_t& keep, const uint
           refresh = std::min(refresh, domainConfig.second.maxSOARefresh);
         }
       }
-      if (now - zoneLastCheck < refresh) {
+
+
+      if (now - zoneLastCheck < refresh && g_notifiesReceived.lock()->erase(domain) == 0) {
         continue;
       }
 
@@ -476,12 +484,54 @@ static void updateThread(const string& workdir, const uint16_t& keep, const uint
 } /* updateThread */
 
 enum class ResponseType {
-  // Unknown,
+  Unknown,
   ValidQuery,
   RefusedOpcode,
-  RefusedQuery
+  RefusedQuery,
+  EmptyNoError
 };
 
+static ResponseType maybeHandleNotify(const MOADNSParser& mdp, const ComboAddress& saddr, const string& logPrefix="") {
+  if (mdp.d_header.opcode != Opcode::Notify) {
+    return ResponseType::Unknown;
+  }
+
+  g_log<<Logger::Info<<logPrefix<<"NOTIFY for "<<mdp.d_qname<<"|"<<QType(mdp.d_qtype).toString()<<" "<< Opcode::to_s(mdp.d_header.opcode) <<" from "<<saddr.toStringWithPort()<<endl;
+
+  auto found = g_domainConfigs.find(mdp.d_qname);
+  if (found == g_domainConfigs.end()) {
+    g_log<<Logger::Info<<("Domain name '" + mdp.d_qname.toLogString() + "' is not configured for notification")<<endl;
+    return ResponseType::RefusedQuery;
+  }
+
+  auto masters = found->second.masters;
+
+  bool masterFound = false;
+
+  auto saddrPort0 = saddr;
+  saddrPort0.setPort(0);
+
+  for(const auto &master : masters) {
+    auto masterPort0 = master;
+    masterPort0.setPort(0);
+    cerr<<masterPort0.toStringWithPort()<<"=="<<saddrPort0.toStringWithPort()<<endl;
+    if (masterPort0 == saddrPort0) {
+      masterFound = true;
+      cerr<<"found"<<endl;
+      break;
+    }
+  }
+
+  cerr<<"masterFound="<<masterFound<<endl;
+
+  if (masterFound) {
+    g_notifiesReceived.lock()->insert(mdp.d_qname);
+    return ResponseType::EmptyNoError;
+  }
+
+  return ResponseType::RefusedQuery;
+}
+
 static ResponseType checkQuery(const MOADNSParser& mdp, const ComboAddress& saddr, const bool udp = true, const string& logPrefix="") {
   vector<string> info_msg;
 
@@ -490,7 +540,7 @@ static ResponseType checkQuery(const MOADNSParser& mdp, const ComboAddress& sadd
   g_log<<Logger::Debug<<logPrefix<<"Had "<<mdp.d_qname<<"|"<<QType(mdp.d_qtype).toString()<<" query from "<<saddr.toStringWithPort()<<endl;
 
   if (mdp.d_header.opcode != Opcode::Query) {
-    info_msg.push_back("Opcode is unsupported (" + Opcode::to_s(mdp.d_header.opcode) + "), expected QUERY");
+    info_msg.push_back("Opcode is unsupported (" + Opcode::to_s(mdp.d_header.opcode) + "), expected QUERY"); // note that we also emit this for a NOTIFY from a wrong source
     ret = ResponseType::RefusedOpcode;
   }
   else {
@@ -537,6 +587,29 @@ static ResponseType checkQuery(const MOADNSParser& mdp, const ComboAddress& sadd
   return ret;
 }
 
+/*
+ * Returns a vector<uint8_t> that represents the full empty NOERROR response.
+ * QNAME is read from mdp.
+ */
+static bool makeEmptyNoErrorPacket(const MOADNSParser& mdp, vector<uint8_t>& packet) {
+
+  auto zoneInfo = getCurrentZoneInfo(mdp.d_qname);
+  if (zoneInfo == nullptr) {
+    return false;
+  }
+
+  DNSPacketWriter pw(packet, mdp.d_qname, mdp.d_qtype);
+  pw.getHeader()->opcode = mdp.d_header.opcode;
+  pw.getHeader()->id = mdp.d_header.id;
+  pw.getHeader()->rd = mdp.d_header.rd;
+  pw.getHeader()->qr = 1;
+  pw.getHeader()->aa = 1;
+
+  pw.commit();
+
+  return true;
+}
+
 /*
  * Returns a vector<uint8_t> that represents the full positive response to a SOA
  * query. QNAME is read from mdp.
@@ -549,6 +622,7 @@ static bool makeSOAPacket(const MOADNSParser& mdp, vector<uint8_t>& packet) {
   }
 
   DNSPacketWriter pw(packet, mdp.d_qname, mdp.d_qtype);
+  pw.getHeader()->opcode = mdp.d_header.opcode;
   pw.getHeader()->id = mdp.d_header.id;
   pw.getHeader()->rd = mdp.d_header.rd;
   pw.getHeader()->qr = 1;
@@ -567,6 +641,7 @@ static bool makeSOAPacket(const MOADNSParser& mdp, vector<uint8_t>& packet) {
  */
 static bool makeRefusedPacket(const MOADNSParser& mdp, vector<uint8_t>& packet) {
   DNSPacketWriter pw(packet, mdp.d_qname, mdp.d_qtype);
+  pw.getHeader()->opcode = mdp.d_header.opcode;
   pw.getHeader()->id = mdp.d_header.id;
   pw.getHeader()->rd = mdp.d_header.rd;
   pw.getHeader()->qr = 1;
@@ -581,11 +656,11 @@ static bool makeRefusedPacket(const MOADNSParser& mdp, vector<uint8_t>& packet)
  */
 static bool makeNotimpPacket(const MOADNSParser& mdp, vector<uint8_t>& packet) {
   DNSPacketWriter pw(packet, mdp.d_qname, mdp.d_qtype);
+  pw.getHeader()->opcode = mdp.d_header.opcode;
   pw.getHeader()->id = mdp.d_header.id;
   pw.getHeader()->rd = mdp.d_header.rd;
   pw.getHeader()->qr = 1;
   pw.getHeader()->rcode = RCode::NotImp;
-  pw.getHeader()->opcode = mdp.d_header.opcode;
 
   return true;
 }
@@ -814,7 +889,11 @@ static bool handleIXFR(int fd, const MOADNSParser& mdp, const shared_ptr<const S
   return true;
 }
 
-static bool allowedByACL(const ComboAddress& addr) {
+static bool allowedByACL(const ComboAddress& addr, bool forNotify = false) {
+  if (forNotify) {
+    return g_notifySources.match(addr);
+  }
+
   return g_acl.match(addr);
 }
 
@@ -843,14 +922,23 @@ try
     return;
   }
 
-  if (!allowedByACL(saddr)) {
+  MOADNSParser mdp(true, string(buf, res));
+  vector<uint8_t> packet;
+
+  ResponseType respt = ResponseType::Unknown;
+
+  if (allowedByACL(saddr, true)) {
+    respt = maybeHandleNotify(mdp, saddr);
+  }
+  else if (!allowedByACL(saddr)) {
     g_log<<Logger::Warning<<"UDP query from "<<saddr.toString()<<" is not allowed, dropping"<<endl;
     return;
   }
 
-  MOADNSParser mdp(true, string(buf, res));
-  vector<uint8_t> packet;
-  auto respt = checkQuery(mdp, saddr);
+  if (respt == ResponseType::Unknown) {
+    // query was not handled yet (so not a valid NOTIFY)
+    respt = checkQuery(mdp, saddr);
+  }
   if (respt == ResponseType::ValidQuery) {
     /* RFC 1995 Section 2
      *    Transport of a query may be by either UDP or TCP.  If an IXFR query
@@ -865,6 +953,8 @@ try
      */
     g_stats.incrementSOAinQueries(mdp.d_qname); // FIXME: this also counts IXFR queries (but the response is the same as to a SOA query)
     makeSOAPacket(mdp, packet);
+  } else if (respt == ResponseType::EmptyNoError) {
+    makeEmptyNoErrorPacket(mdp, packet);
   } else if (respt == ResponseType::RefusedQuery) {
     g_stats.incrementUnknownDomainInQueries(mdp.d_qname);
     makeRefusedPacket(mdp, packet);
@@ -902,7 +992,9 @@ static void handleTCPRequest(int fd, boost::any&) {
     return;
   }
 
-  if (!allowedByACL(saddr)) {
+  // we allow the connection if this is a legit client or a legit NOTIFY source
+  // need to check per-operation later
+  if (!allowedByACL(saddr) && !allowedByACL(saddr, true)) {
     g_log<<Logger::Warning<<"TCP query from "<<saddr.toString()<<" is not allowed, dropping"<<endl;
     close(cfd);
     return;
@@ -954,15 +1046,37 @@ static void tcpWorker(int tid) {
     try {
       MOADNSParser mdp(true, string(buf, res));
 
-      auto respt = checkQuery(mdp, saddr, false, prefix);
+      ResponseType respt = ResponseType::Unknown;
 
-      if (respt != ResponseType::ValidQuery) { // on TCP, we currently do not bother with sending useful errors
+      // this code is duplicated from the UDP path
+      if (allowedByACL(saddr, true)) {
+        respt = maybeHandleNotify(mdp, saddr);
+      }
+      else if (!allowedByACL(saddr)) {
         close(cfd);
         continue;
       }
 
-      if (mdp.d_qtype == QType::SOA) {
-        vector<uint8_t> packet;
+      if (respt == ResponseType::Unknown) {
+        respt = checkQuery(mdp, saddr, false, prefix);
+      }
+
+      if (respt != ResponseType::ValidQuery && respt != ResponseType::EmptyNoError) { // on TCP, we currently do not bother with sending useful errors
+        close(cfd);
+        continue;
+      }
+
+      vector<uint8_t> packet;
+
+      if (respt == ResponseType::EmptyNoError) {
+        bool ret = makeEmptyNoErrorPacket(mdp, packet);
+        if (!ret) {
+          close(cfd);
+          continue;
+        }
+        sendPacketOverTCP(cfd, packet);
+      }
+      else if (mdp.d_qtype == QType::SOA) {
         bool ret = makeSOAPacket(mdp, packet);
         if (!ret) {
           close(cfd);
@@ -1164,6 +1278,10 @@ static bool parseAndCheckConfig(const string& configpath, YAML::Node& config) {
           continue;
         }
         domain["master"].as<ComboAddress>();
+
+        auto notifySource = domain["master"].as<ComboAddress>();
+
+        g_notifySources.addMask(notifySource);
       } catch (const runtime_error &e) {
         g_log<<Logger::Error<<"Unable to read domain '"<<domain["domain"].as<string>()<<"' master address: "<<e.what()<<endl;
         retval = false;
@@ -1341,6 +1459,8 @@ static std::optional<IXFRDistConfiguration> parseConfiguration(int argc, char**
       g_log<<Logger::Error<<"Error printing ACL: "<<exp.what()<<endl;
     }
 
+    g_log<<Logger::Notice<<"NOTIFY accepted from "<<g_notifySources.toString()<<"."<<endl;
+
     if (config["compress"].IsDefined()) {
       g_compress = config["compress"].as<bool>();
       if (g_compress) {

diff --git a/regression-tests.ixfrdist/test_IXFR.py b/regression-tests.ixfrdist/test_IXFR.py
index c0788f874a2a6aca55305c238ded7dbc3827beae..159d85d4fba18c6a3484fbc54b30e3d21bf7459b 100644 (file)
--- a/regression-tests.ixfrdist/test_IXFR.py
+++ b/regression-tests.ixfrdist/test_IXFR.py
@@ -26,7 +26,7 @@ newrecord.example.        8484    A       192.0.2.42
 """,
     3: """
 $ORIGIN example.
-@        86400   SOA    foo bar 3 2 3 4 5
+@        86400   SOA    foo bar 3 1500 3 4 5
 @        4242    NS     ns1.example.
 @        4242    NS     ns2.example.
 ns1.example.    4242    A       192.0.2.1
@@ -72,11 +72,17 @@ class IXFRDistBasicTest(IXFRDistTest):
     def tearDownClass(cls):
         cls.tearDownIXFRDist()
 
-    def waitUntilCorrectSerialIsLoaded(self, serial, timeout=10):
+    def waitUntilCorrectSerialIsLoaded(self, serial, timeout=10, notify=False):
         global xfrServer
 
         xfrServer.moveToSerial(serial)
 
+        if notify:
+            notif = dns.message.make_query('example.', 'SOA')
+            notif.set_opcode(dns.opcode.NOTIFY)
+            notify_response = self.sendUDPQuery(notif)
+            assert notify_response.rcode() == dns.rcode.NOERROR
+
         def get_current_serial():
             query = dns.message.make_query('example.', 'SOA')
             response_message = self.sendUDPQuery(query)
@@ -227,7 +233,7 @@ class IXFRDistBasicTest(IXFRDistTest):
         self.checkIXFR(2,3)
         self.checkIXFR(1,3)
 
-        self.waitUntilCorrectSerialIsLoaded(4)
+        self.waitUntilCorrectSerialIsLoaded(serial=4, timeout=10, notify=True)
         self.checkFullZone(4)
         self.checkIXFR(3,4)
         self.checkIXFR(2,4)