]> git.ipfire.org Git - thirdparty/pdns.git/blob - pdns/ixfrdist.cc
Merge pull request #6913 from Habbie/truncate-truncated
[thirdparty/pdns.git] / pdns / ixfrdist.cc
1 /*
2 * This file is part of PowerDNS or dnsdist.
3 * Copyright -- PowerDNS.COM B.V. and its contributors
4 *
5 * This program is free software; you can redistribute it and/or modify
6 * it under the terms of version 2 of the GNU General Public License as
7 * published by the Free Software Foundation.
8 *
9 * In addition, for the avoidance of any doubt, permission is granted to
10 * link this program with OpenSSL and to (re)distribute the binaries
11 * produced as the result of such linking.
12 *
13 * This program is distributed in the hope that it will be useful,
14 * but WITHOUT ANY WARRANTY; without even the implied warranty of
15 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
16 * GNU General Public License for more details.
17 *
18 * You should have received a copy of the GNU General Public License
19 * along with this program; if not, write to the Free Software
20 * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
21 */
22 #ifdef HAVE_CONFIG_H
23 #include "config.h"
24 #endif
25 #include <boost/program_options.hpp>
26 #include <arpa/inet.h>
27 #include <sys/types.h>
28 #include <grp.h>
29 #include <pwd.h>
30 #include <sys/stat.h>
31 #include <mutex>
32 #include <thread>
33 #include <dirent.h>
34 #include <queue>
35 #include <condition_variable>
36 #include "ixfr.hh"
37 #include "ixfrutils.hh"
38 #include "resolver.hh"
39 #include "dns_random.hh"
40 #include "sstuff.hh"
41 #include "mplexer.hh"
42 #include "misc.hh"
43 #include "iputils.hh"
44 #include "logger.hh"
45 #include <yaml-cpp/yaml.h>
46
47 /* BEGIN Needed because of deeper dependencies */
48 #include "arguments.hh"
49 #include "statbag.hh"
50 StatBag S;
51
52 ArgvMap &arg()
53 {
54 static ArgvMap theArg;
55 return theArg;
56 }
57 /* END Needed because of deeper dependencies */
58
59 // Allows reading/writing ComboAddresses and DNSNames in YAML-cpp
60 namespace YAML {
61 template<>
62 struct convert<ComboAddress> {
63 static Node encode(const ComboAddress& rhs) {
64 return Node(rhs.toStringWithPort());
65 }
66 static bool decode(const Node& node, ComboAddress& rhs) {
67 if (!node.IsScalar()) {
68 return false;
69 }
70 try {
71 rhs = ComboAddress(node.as<string>(), 53);
72 return true;
73 } catch(const runtime_error &e) {
74 return false;
75 } catch (const PDNSException &e) {
76 return false;
77 }
78 }
79 };
80
81 template<>
82 struct convert<DNSName> {
83 static Node encode(const DNSName& rhs) {
84 return Node(rhs.toStringRootDot());
85 }
86 static bool decode(const Node& node, DNSName& rhs) {
87 if (!node.IsScalar()) {
88 return false;
89 }
90 try {
91 rhs = DNSName(node.as<string>());
92 return true;
93 } catch(const runtime_error &e) {
94 return false;
95 } catch (const PDNSException &e) {
96 return false;
97 }
98 }
99 };
100 } // namespace YAML
101
102 struct ixfrdiff_t {
103 shared_ptr<SOARecordContent> oldSOA;
104 shared_ptr<SOARecordContent> newSOA;
105 vector<DNSRecord> removals;
106 vector<DNSRecord> additions;
107 };
108
109 struct ixfrinfo_t {
110 shared_ptr<SOARecordContent> soa; // The SOA of the latestAXFR
111 records_t latestAXFR; // The most recent AXFR
112 vector<std::shared_ptr<ixfrdiff_t>> ixfrDiffs;
113 };
114
115 // Why a struct? This way we can add more options to a domain in the future
116 struct ixfrdistdomain_t {
117 set<ComboAddress> masters; // A set so we can do multiple master addresses in the future
118 };
119
120 // This contains the configuration for each domain
121 static map<DNSName, ixfrdistdomain_t> g_domainConfigs;
122
123 // Map domains and their data
124 static std::map<DNSName, std::shared_ptr<ixfrinfo_t>> g_soas;
125 static std::mutex g_soas_mutex;
126
127 // Condition variable for TCP handling
128 static std::condition_variable g_tcpHandlerCV;
129 static std::queue<pair<int, ComboAddress>> g_tcpRequestFDs;
130 static std::mutex g_tcpRequestFDsMutex;
131
132 namespace po = boost::program_options;
133
134 static bool g_exiting = false;
135
136 static NetmaskGroup g_acl;
137 static bool g_compress = false;
138
139 static void handleSignal(int signum) {
140 g_log<<Logger::Notice<<"Got "<<strsignal(signum)<<" signal";
141 if (g_exiting) {
142 g_log<<Logger::Notice<<", this is the second time we were asked to stop, forcefully exiting"<<endl;
143 exit(EXIT_FAILURE);
144 }
145 g_log<<Logger::Notice<<", stopping, this may take a few second due to in-progress transfers and cleanup. Send this signal again to forcefully stop"<<endl;
146 g_exiting = true;
147 }
148
149 static void usage(po::options_description &desc) {
150 cerr << "Usage: ixfrdist [OPTION]..."<<endl;
151 cerr << desc << "\n";
152 }
153
154 // The compiler does not like using rfc1982LessThan in std::sort directly
155 static bool sortSOA(uint32_t i, uint32_t j) {
156 return rfc1982LessThan(i, j);
157 }
158
159 static void cleanUpDomain(const DNSName& domain, const uint16_t& keep, const string& workdir) {
160 string dir = workdir + "/" + domain.toString();
161 DIR *dp;
162 dp = opendir(dir.c_str());
163 if (dp == nullptr) {
164 return;
165 }
166 vector<uint32_t> zoneVersions;
167 struct dirent *d;
168 while ((d = readdir(dp)) != nullptr) {
169 if(!strcmp(d->d_name, ".") || !strcmp(d->d_name, "..")) {
170 continue;
171 }
172 zoneVersions.push_back(std::stoi(d->d_name));
173 }
174 closedir(dp);
175 g_log<<Logger::Info<<"Found "<<zoneVersions.size()<<" versions of "<<domain<<", asked to keep "<<keep<<", ";
176 if (zoneVersions.size() <= keep) {
177 g_log<<Logger::Info<<"not cleaning up"<<endl;
178 return;
179 }
180 g_log<<Logger::Info<<"cleaning up the oldest "<<zoneVersions.size() - keep<<endl;
181
182 // Sort the versions
183 std::sort(zoneVersions.begin(), zoneVersions.end(), sortSOA);
184
185 // And delete all the old ones
186 {
187 // Lock to ensure no one reads this.
188 std::lock_guard<std::mutex> guard(g_soas_mutex);
189 for (auto iter = zoneVersions.cbegin(); iter != zoneVersions.cend() - keep; ++iter) {
190 string fname = dir + "/" + std::to_string(*iter);
191 g_log<<Logger::Debug<<"Removing "<<fname<<endl;
192 unlink(fname.c_str());
193 }
194 }
195 }
196
197 static shared_ptr<SOARecordContent> getSOAFromRecords(const records_t& records) {
198 for (const auto& dnsrecord : records) {
199 if (dnsrecord.d_type == QType::SOA) {
200 auto soa = getRR<SOARecordContent>(dnsrecord);
201 if (soa == nullptr) {
202 throw PDNSException("Unable to determine SOARecordContent from old records");
203 }
204 return soa;
205 }
206 }
207 throw PDNSException("No SOA in supplied records");
208 }
209
210 static void makeIXFRDiff(const records_t& from, const records_t& to, std::shared_ptr<ixfrdiff_t>& diff, const shared_ptr<SOARecordContent>& fromSOA = nullptr, const shared_ptr<SOARecordContent>& toSOA = nullptr) {
211 set_difference(from.cbegin(), from.cend(), to.cbegin(), to.cend(), back_inserter(diff->removals), from.value_comp());
212 set_difference(to.cbegin(), to.cend(), from.cbegin(), from.cend(), back_inserter(diff->additions), from.value_comp());
213 diff->oldSOA = fromSOA;
214 if (fromSOA == nullptr) {
215 diff->oldSOA = getSOAFromRecords(from);
216 }
217 diff->newSOA = toSOA;
218 if (toSOA == nullptr) {
219 diff->newSOA = getSOAFromRecords(to);
220 }
221 }
222
223 /* you can _never_ alter the content of the resulting shared pointer */
224 static std::shared_ptr<ixfrinfo_t> getCurrentZoneInfo(const DNSName& domain)
225 {
226 std::lock_guard<std::mutex> guard(g_soas_mutex);
227 return g_soas[domain];
228 }
229
230 static void updateCurrentZoneInfo(const DNSName& domain, std::shared_ptr<ixfrinfo_t>& newInfo)
231 {
232 std::lock_guard<std::mutex> guard(g_soas_mutex);
233 g_soas[domain] = newInfo;
234 }
235
236 void updateThread(const string& workdir, const uint16_t& keep, const uint16_t& axfrTimeout) {
237 std::map<DNSName, time_t> lastCheck;
238
239 // Initialize the serials we have
240 for (const auto &domainConfig : g_domainConfigs) {
241 DNSName domain = domainConfig.first;
242 lastCheck[domain] = 0;
243 string dir = workdir + "/" + domain.toString();
244 try {
245 g_log<<Logger::Info<<"Trying to initially load domain "<<domain<<" from disk"<<endl;
246 auto serial = getSerialsFromDir(dir);
247 shared_ptr<SOARecordContent> soa;
248 {
249 string fname = workdir + "/" + domain.toString() + "/" + std::to_string(serial);
250 loadSOAFromDisk(domain, fname, soa);
251 records_t records;
252 if (soa != nullptr) {
253 loadZoneFromDisk(records, fname, domain);
254 }
255 auto zoneInfo = std::make_shared<ixfrinfo_t>();
256 zoneInfo->latestAXFR = std::move(records);
257 zoneInfo->soa = soa;
258 updateCurrentZoneInfo(domain, zoneInfo);
259 }
260 if (soa != nullptr) {
261 g_log<<Logger::Notice<<"Loaded zone "<<domain<<" with serial "<<soa->d_st.serial<<endl;
262 // Initial cleanup
263 cleanUpDomain(domain, keep, workdir);
264 }
265 } catch (runtime_error &e) {
266 // Most likely, the directory does not exist.
267 g_log<<Logger::Info<<e.what()<<", attempting to create"<<endl;
268 // Attempt to create it, if _that_ fails, there is no hope
269 if (mkdir(dir.c_str(), 0777) == -1 && errno != EEXIST) {
270 g_log<<Logger::Error<<"Could not create '"<<dir<<"': "<<strerror(errno)<<endl;
271 exit(EXIT_FAILURE);
272 }
273 }
274 }
275
276 g_log<<Logger::Notice<<"Update Thread started"<<endl;
277
278 while (true) {
279 if (g_exiting) {
280 g_log<<Logger::Notice<<"UpdateThread stopped"<<endl;
281 break;
282 }
283 time_t now = time(nullptr);
284 for (const auto &domainConfig : g_domainConfigs) {
285
286 if (g_exiting) {
287 break;
288 }
289
290 DNSName domain = domainConfig.first;
291 shared_ptr<SOARecordContent> current_soa;
292 const auto& zoneInfo = getCurrentZoneInfo(domain);
293 if (zoneInfo != nullptr) {
294 current_soa = zoneInfo->soa;
295 }
296
297 auto& zoneLastCheck = lastCheck[domain];
298 if ((current_soa != nullptr && now - zoneLastCheck < current_soa->d_st.refresh) || // Only check if we have waited `refresh` seconds
299 (current_soa == nullptr && now - zoneLastCheck < 30)) { // Or if we could not get an update at all still, every 30 seconds
300 continue;
301 }
302
303 // TODO Keep track of 'down' masters
304 set<ComboAddress>::const_iterator it(domainConfig.second.masters.begin());
305 std::advance(it, random() % domainConfig.second.masters.size());
306 ComboAddress master = *it;
307
308 string dir = workdir + "/" + domain.toString();
309 g_log<<Logger::Info<<"Attempting to retrieve SOA Serial update for '"<<domain<<"' from '"<<master.toStringWithPort()<<"'"<<endl;
310 shared_ptr<SOARecordContent> sr;
311 try {
312 zoneLastCheck = now;
313 auto newSerial = getSerialFromMaster(master, domain, sr); // TODO TSIG
314 if(current_soa != nullptr) {
315 g_log<<Logger::Info<<"Got SOA Serial for "<<domain<<" from "<<master.toStringWithPort()<<": "<< newSerial<<", had Serial: "<<current_soa->d_st.serial;
316 if (newSerial == current_soa->d_st.serial) {
317 g_log<<Logger::Info<<", not updating."<<endl;
318 continue;
319 }
320 g_log<<Logger::Info<<", will update."<<endl;
321 }
322 } catch (runtime_error &e) {
323 g_log<<Logger::Warning<<"Unable to get SOA serial update for '"<<domain<<"' from master "<<master.toStringWithPort()<<": "<<e.what()<<endl;
324 continue;
325 }
326 // Now get the full zone!
327 g_log<<Logger::Info<<"Attempting to receive full zonedata for '"<<domain<<"'"<<endl;
328 ComboAddress local = master.isIPv4() ? ComboAddress("0.0.0.0") : ComboAddress("::");
329 TSIGTriplet tt;
330
331 // The *new* SOA
332 shared_ptr<SOARecordContent> soa;
333 records_t records;
334 try {
335 AXFRRetriever axfr(master, domain, tt, &local);
336 unsigned int nrecords=0;
337 Resolver::res_t nop;
338 vector<DNSRecord> chunk;
339 time_t t_start = time(nullptr);
340 time_t axfr_now = time(nullptr);
341 while(axfr.getChunk(nop, &chunk, (axfr_now - t_start + axfrTimeout))) {
342 for(auto& dr : chunk) {
343 if(dr.d_type == QType::TSIG)
344 continue;
345 dr.d_name.makeUsRelative(domain);
346 records.insert(dr);
347 nrecords++;
348 if (dr.d_type == QType::SOA) {
349 soa = getRR<SOARecordContent>(dr);
350 }
351 }
352 axfr_now = time(nullptr);
353 if (axfr_now - t_start > axfrTimeout) {
354 throw PDNSException("Total AXFR time exceeded!");
355 }
356 }
357 if (soa == nullptr) {
358 g_log<<Logger::Warning<<"No SOA was found in the AXFR of "<<domain<<endl;
359 continue;
360 }
361 g_log<<Logger::Notice<<"Retrieved all zone data for "<<domain<<". Received "<<nrecords<<" records."<<endl;
362 } catch (PDNSException &e) {
363 g_log<<Logger::Warning<<"Could not retrieve AXFR for '"<<domain<<"': "<<e.reason<<endl;
364 continue;
365 } catch (runtime_error &e) {
366 g_log<<Logger::Warning<<"Could not retrieve AXFR for zone '"<<domain<<"': "<<e.what()<<endl;
367 continue;
368 }
369
370 try {
371
372 writeZoneToDisk(records, domain, dir);
373 g_log<<Logger::Notice<<"Wrote zonedata for "<<domain<<" with serial "<<soa->d_st.serial<<" to "<<dir<<endl;
374
375 const auto oldZoneInfo = getCurrentZoneInfo(domain);
376 auto zoneInfo = std::make_shared<ixfrinfo_t>();
377
378 if (oldZoneInfo && !oldZoneInfo->latestAXFR.empty()) {
379 auto diff = std::make_shared<ixfrdiff_t>();
380 zoneInfo->ixfrDiffs = oldZoneInfo->ixfrDiffs;
381 g_log<<Logger::Debug<<"Calculating diff for "<<domain<<endl;
382 makeIXFRDiff(oldZoneInfo->latestAXFR, records, diff, oldZoneInfo->soa, soa);
383 g_log<<Logger::Debug<<"Calculated diff for "<<domain<<", we had "<<diff->removals.size()<<" removals and "<<diff->additions.size()<<" additions"<<endl;
384 zoneInfo->ixfrDiffs.push_back(std::move(diff));
385 }
386
387 // Clean up the diffs
388 while (zoneInfo->ixfrDiffs.size() > keep) {
389 zoneInfo->ixfrDiffs.erase(zoneInfo->ixfrDiffs.begin());
390 }
391
392 g_log<<Logger::Debug<<"Zone "<<domain<<" previously contained "<<(oldZoneInfo ? oldZoneInfo->latestAXFR.size() : 0)<<" entries, "<<records.size()<<" now"<<endl;
393 zoneInfo->latestAXFR = std::move(records);
394 zoneInfo->soa = soa;
395 updateCurrentZoneInfo(domain, zoneInfo);
396 } catch (PDNSException &e) {
397 g_log<<Logger::Warning<<"Could not save zone '"<<domain<<"' to disk: "<<e.reason<<endl;
398 } catch (runtime_error &e) {
399 g_log<<Logger::Warning<<"Could not save zone '"<<domain<<"' to disk: "<<e.what()<<endl;
400 }
401
402 // Now clean up the directory
403 cleanUpDomain(domain, keep, workdir);
404 } /* for (const auto &domain : domains) */
405 sleep(1);
406 } /* while (true) */
407 } /* updateThread */
408
409 static bool checkQuery(const MOADNSParser& mdp, const ComboAddress& saddr, const bool udp = true, const string& logPrefix="") {
410 vector<string> info_msg;
411
412 g_log<<Logger::Debug<<logPrefix<<"Had "<<mdp.d_qname<<"|"<<QType(mdp.d_qtype).getName()<<" query from "<<saddr.toStringWithPort()<<endl;
413
414 if (udp && mdp.d_qtype != QType::SOA && mdp.d_qtype != QType::IXFR) {
415 info_msg.push_back("QType is unsupported (" + QType(mdp.d_qtype).getName() + " is not in {SOA,IXFR}");
416 }
417
418 if (!udp && mdp.d_qtype != QType::SOA && mdp.d_qtype != QType::IXFR && mdp.d_qtype != QType::AXFR) {
419 info_msg.push_back("QType is unsupported (" + QType(mdp.d_qtype).getName() + " is not in {SOA,IXFR,AXFR}");
420 }
421
422 {
423 if (g_domainConfigs.find(mdp.d_qname) == g_domainConfigs.end()) {
424 info_msg.push_back("Domain name '" + mdp.d_qname.toLogString() + "' is not configured for distribution");
425 }
426
427 const auto zoneInfo = getCurrentZoneInfo(mdp.d_qname);
428 if (zoneInfo == nullptr) {
429 info_msg.push_back("Domain has not been transferred yet");
430 }
431 }
432
433 if (!info_msg.empty()) {
434 g_log<<Logger::Warning<<logPrefix<<"Ignoring "<<mdp.d_qname<<"|"<<QType(mdp.d_qtype).getName()<<" query from "<<saddr.toStringWithPort();
435 g_log<<Logger::Warning<<": ";
436 bool first = true;
437 for (const auto& s : info_msg) {
438 if (!first) {
439 g_log<<Logger::Warning<<", ";
440 first = false;
441 }
442 g_log<<Logger::Warning<<s;
443 }
444 g_log<<Logger::Warning<<endl;
445 return false;
446 }
447
448 return true;
449 }
450
451 /*
452 * Returns a vector<uint8_t> that represents the full response to a SOA
453 * query. QNAME is read from mdp.
454 */
455 static bool makeSOAPacket(const MOADNSParser& mdp, vector<uint8_t>& packet) {
456
457 auto zoneInfo = getCurrentZoneInfo(mdp.d_qname);
458 if (zoneInfo == nullptr) {
459 return false;
460 }
461
462 DNSPacketWriter pw(packet, mdp.d_qname, mdp.d_qtype);
463 pw.getHeader()->id = mdp.d_header.id;
464 pw.getHeader()->rd = mdp.d_header.rd;
465 pw.getHeader()->qr = 1;
466
467 pw.startRecord(mdp.d_qname, QType::SOA);
468 zoneInfo->soa->toPacket(pw);
469 pw.commit();
470
471 return true;
472 }
473
474 static vector<uint8_t> getSOAPacket(const MOADNSParser& mdp, const shared_ptr<SOARecordContent>& soa) {
475 vector<uint8_t> packet;
476 DNSPacketWriter pw(packet, mdp.d_qname, mdp.d_qtype);
477 pw.getHeader()->id = mdp.d_header.id;
478 pw.getHeader()->rd = mdp.d_header.rd;
479 pw.getHeader()->qr = 1;
480
481 // Add the first SOA
482 pw.startRecord(mdp.d_qname, QType::SOA);
483 soa->toPacket(pw);
484 pw.commit();
485 return packet;
486 }
487
488 static bool sendPacketOverTCP(int fd, const std::vector<uint8_t>& packet)
489 {
490 char sendBuf[2];
491 sendBuf[0]=packet.size()/256;
492 sendBuf[1]=packet.size()%256;
493
494 ssize_t send = writen2(fd, sendBuf, 2);
495 send += writen2(fd, &packet[0], packet.size());
496 return true;
497 }
498
499 static bool addRecordToWriter(DNSPacketWriter& pw, const DNSName& zoneName, const DNSRecord& record, bool compress)
500 {
501 pw.startRecord(record.d_name + zoneName, record.d_type, record.d_ttl, QClass::IN, DNSResourceRecord::ANSWER, compress);
502 record.d_content->toPacket(pw);
503 if (pw.size() > 65535) {
504 pw.rollback();
505 return false;
506 }
507 return true;
508 }
509
510 template <typename T> static bool sendRecordsOverTCP(int fd, const MOADNSParser& mdp, const T& records)
511 {
512 vector<uint8_t> packet;
513
514 for (auto it = records.cbegin(); it != records.cend();) {
515 bool recordsAdded = false;
516 packet.clear();
517 DNSPacketWriter pw(packet, mdp.d_qname, mdp.d_qtype);
518 pw.getHeader()->id = mdp.d_header.id;
519 pw.getHeader()->rd = mdp.d_header.rd;
520 pw.getHeader()->qr = 1;
521
522 while (it != records.cend()) {
523 if (it->d_type == QType::SOA) {
524 it++;
525 continue;
526 }
527
528 if (addRecordToWriter(pw, mdp.d_qname, *it, g_compress)) {
529 recordsAdded = true;
530 it++;
531 }
532 else {
533 if (recordsAdded) {
534 pw.commit();
535 sendPacketOverTCP(fd, packet);
536 }
537 if (it == records.cbegin()) {
538 /* something is wrong */
539 return false;
540 }
541
542 break;
543 }
544 }
545
546 if (it == records.cend() && recordsAdded) {
547 pw.commit();
548 sendPacketOverTCP(fd, packet);
549 }
550 }
551
552 return true;
553 }
554
555
556 static bool handleAXFR(int fd, const MOADNSParser& mdp) {
557 /* we get a shared pointer of the zone info that we can't modify, ever.
558 A newer one may arise in the meantime, but this one will stay valid
559 until we release it.
560 */
561 auto zoneInfo = getCurrentZoneInfo(mdp.d_qname);
562 if (zoneInfo == nullptr) {
563 return false;
564 }
565
566 shared_ptr<SOARecordContent> soa = zoneInfo->soa;
567 const records_t& records = zoneInfo->latestAXFR;
568
569 // Initial SOA
570 const auto soaPacket = getSOAPacket(mdp, soa);
571 if (!sendPacketOverTCP(fd, soaPacket)) {
572 return false;
573 }
574
575 if (!sendRecordsOverTCP(fd, mdp, records)) {
576 return false;
577 }
578
579 // Final SOA
580 if (!sendPacketOverTCP(fd, soaPacket)) {
581 return false;
582 }
583
584 return true;
585 }
586
587 /* Produces an IXFR if one can be made according to the rules in RFC 1995 and
588 * creates a SOA or AXFR packet when required by the RFC.
589 */
590 static bool handleIXFR(int fd, const ComboAddress& destination, const MOADNSParser& mdp, const shared_ptr<SOARecordContent>& clientSOA) {
591 vector<std::shared_ptr<ixfrdiff_t>> toSend;
592
593 /* we get a shared pointer of the zone info that we can't modify, ever.
594 A newer one may arise in the meantime, but this one will stay valid
595 until we release it.
596 */
597 auto zoneInfo = getCurrentZoneInfo(mdp.d_qname);
598 if (zoneInfo == nullptr) {
599 return false;
600 }
601
602 uint32_t ourLatestSerial = zoneInfo->soa->d_st.serial;
603
604 if (rfc1982LessThan(ourLatestSerial, clientSOA->d_st.serial) || ourLatestSerial == clientSOA->d_st.serial) {
605 /* RFC 1995 Section 2
606 * If an IXFR query with the same or newer version number than that of
607 * the server is received, it is replied to with a single SOA record of
608 * the server's current version, just as in AXFR.
609 */
610 vector<uint8_t> packet;
611 bool ret = makeSOAPacket(mdp, packet);
612 if (ret) {
613 sendPacketOverTCP(fd, packet);
614 }
615 return ret;
616 }
617
618 // as we use push_back in the updater, we know the vector is sorted as oldest first
619 bool shouldAdd = false;
620 // Get all relevant IXFR differences
621 for (const auto& diff : zoneInfo->ixfrDiffs) {
622 if (shouldAdd) {
623 toSend.push_back(diff);
624 continue;
625 }
626 if (diff->oldSOA->d_st.serial == clientSOA->d_st.serial) {
627 toSend.push_back(diff);
628 // Add all consecutive diffs
629 shouldAdd = true;
630 }
631 }
632
633 if (toSend.empty()) {
634 g_log<<Logger::Warning<<"No IXFR available from serial "<<clientSOA->d_st.serial<<" for zone "<<mdp.d_qname<<", attempting to send AXFR"<<endl;
635 return handleAXFR(fd, mdp);
636 }
637
638 std::vector<std::vector<uint8_t>> packets;
639 for (const auto& diff : toSend) {
640 /* An IXFR packet's ANSWER section looks as follows:
641 * SOA new_serial
642 * SOA old_serial
643 * ... removed records ...
644 * SOA new_serial
645 * ... added records ...
646 * SOA new_serial
647 */
648
649 const auto newSOAPacket = getSOAPacket(mdp, diff->newSOA);
650 const auto oldSOAPacket = getSOAPacket(mdp, diff->oldSOA);
651
652 if (!sendPacketOverTCP(fd, newSOAPacket)) {
653 return false;
654 }
655
656 if (!sendPacketOverTCP(fd, oldSOAPacket)) {
657 return false;
658 }
659
660 if (!sendRecordsOverTCP(fd, mdp, diff->removals)) {
661 return false;
662 }
663
664 if (!sendPacketOverTCP(fd, newSOAPacket)) {
665 return false;
666 }
667
668 if (!sendRecordsOverTCP(fd, mdp, diff->additions)) {
669 return false;
670 }
671
672 if (!sendPacketOverTCP(fd, newSOAPacket)) {
673 return false;
674 }
675 }
676
677 return true;
678 }
679
680 static bool allowedByACL(const ComboAddress& addr) {
681 return g_acl.match(addr);
682 }
683
684 static void handleUDPRequest(int fd, boost::any&) {
685 // TODO make the buffer-size configurable
686 char buf[4096];
687 ComboAddress saddr;
688 socklen_t fromlen = sizeof(saddr);
689 int res = recvfrom(fd, buf, sizeof(buf), 0, (struct sockaddr*) &saddr, &fromlen);
690
691 if (res == 0) {
692 g_log<<Logger::Warning<<"Got an empty message from "<<saddr.toStringWithPort()<<endl;
693 return;
694 }
695
696 if(res < 0) {
697 auto savedErrno = errno;
698 g_log<<Logger::Warning<<"Could not read message from "<<saddr.toStringWithPort()<<": "<<strerror(savedErrno)<<endl;
699 return;
700 }
701
702 if (saddr == ComboAddress("0.0.0.0", 0)) {
703 g_log<<Logger::Warning<<"Could not determine source of message"<<endl;
704 return;
705 }
706
707 if (!allowedByACL(saddr)) {
708 g_log<<Logger::Warning<<"UDP query from "<<saddr.toString()<<" is not allowed, dropping"<<endl;
709 return;
710 }
711
712 MOADNSParser mdp(true, string(buf, res));
713 if (!checkQuery(mdp, saddr)) {
714 return;
715 }
716
717 /* RFC 1995 Section 2
718 * Transport of a query may be by either UDP or TCP. If an IXFR query
719 * is via UDP, the IXFR server may attempt to reply using UDP if the
720 * entire response can be contained in a single DNS packet. If the UDP
721 * reply does not fit, the query is responded to with a single SOA
722 * record of the server's current version to inform the client that a
723 * TCP query should be initiated.
724 *
725 * Let's not complicate this with IXFR over UDP (and looking if we need to truncate etc).
726 * Just send the current SOA and let the client try over TCP
727 */
728 vector<uint8_t> packet;
729 makeSOAPacket(mdp, packet);
730 if(sendto(fd, &packet[0], packet.size(), 0, (struct sockaddr*) &saddr, fromlen) < 0) {
731 auto savedErrno = errno;
732 g_log<<Logger::Warning<<"Could not send reply for "<<mdp.d_qname<<"|"<<QType(mdp.d_qtype).getName()<<" to "<<saddr.toStringWithPort()<<": "<<strerror(savedErrno)<<endl;
733 }
734 return;
735 }
736
737 static void handleTCPRequest(int fd, boost::any&) {
738 ComboAddress saddr;
739 int cfd = 0;
740
741 try {
742 cfd = SAccept(fd, saddr);
743 setBlocking(cfd);
744 } catch(runtime_error &e) {
745 g_log<<Logger::Error<<e.what()<<endl;
746 return;
747 }
748
749 if (saddr == ComboAddress("0.0.0.0", 0)) {
750 g_log<<Logger::Warning<<"Could not determine source of message"<<endl;
751 close(cfd);
752 return;
753 }
754
755 if (!allowedByACL(saddr)) {
756 g_log<<Logger::Warning<<"TCP query from "<<saddr.toString()<<" is not allowed, dropping"<<endl;
757 close(cfd);
758 return;
759 }
760
761 {
762 std::lock_guard<std::mutex> lg(g_tcpRequestFDsMutex);
763 g_tcpRequestFDs.push({cfd, saddr});
764 }
765 g_tcpHandlerCV.notify_one();
766 }
767
768 /* Thread to handle TCP traffic
769 */
770 static void tcpWorker(int tid) {
771 string prefix = "TCP Worker " + std::to_string(tid) + ": ";
772
773 while(true) {
774 g_log<<Logger::Debug<<prefix<<"ready for a new request!"<<endl;
775 std::unique_lock<std::mutex> lk(g_tcpRequestFDsMutex);
776 g_tcpHandlerCV.wait(lk, []{return g_tcpRequestFDs.size() || g_exiting ;});
777 if (g_exiting) {
778 g_log<<Logger::Debug<<prefix<<"Stopping thread"<<endl;
779 break;
780 }
781 g_log<<Logger::Debug<<prefix<<"Going to handle a query"<<endl;
782 auto request = g_tcpRequestFDs.front();
783 g_tcpRequestFDs.pop();
784 lk.unlock();
785
786 int cfd = request.first;
787 ComboAddress saddr = request.second;
788
789 char buf[4096];
790 ssize_t res;
791 try {
792 uint16_t toRead;
793 readn2(cfd, &toRead, sizeof(toRead));
794 toRead = std::min(ntohs(toRead), static_cast<uint16_t>(sizeof(buf)));
795 res = readn2WithTimeout(cfd, &buf, toRead, 2);
796 g_log<<Logger::Debug<<prefix<<"Had message of "<<std::to_string(toRead)<<" bytes from "<<saddr.toStringWithPort()<<endl;
797 } catch (runtime_error &e) {
798 g_log<<Logger::Warning<<prefix<<"Could not read message from "<<saddr.toStringWithPort()<<": "<<e.what()<<endl;
799 close(cfd);
800 continue;
801 }
802
803 try {
804 MOADNSParser mdp(true, string(buf, res));
805
806 if (!checkQuery(mdp, saddr, false, prefix)) {
807 close(cfd);
808 continue;
809 }
810
811 if (mdp.d_qtype == QType::SOA) {
812 vector<uint8_t> packet;
813 bool ret = makeSOAPacket(mdp, packet);
814 if (!ret) {
815 close(cfd);
816 continue;
817 }
818 sendPacketOverTCP(cfd, packet);
819 }
820 else if (mdp.d_qtype == QType::AXFR) {
821 if (!handleAXFR(cfd, mdp)) {
822 close(cfd);
823 continue;
824 }
825 }
826 else if (mdp.d_qtype == QType::IXFR) {
827 /* RFC 1995 section 3:
828 * The IXFR query packet format is the same as that of a normal DNS
829 * query, but with the query type being IXFR and the authority section
830 * containing the SOA record of client's version of the zone.
831 */
832 shared_ptr<SOARecordContent> clientSOA;
833 for (auto &answer : mdp.d_answers) {
834 // from dnsparser.hh:
835 // typedef vector<pair<DNSRecord, uint16_t > > answers_t;
836 if (answer.first.d_type == QType::SOA && answer.first.d_place == DNSResourceRecord::AUTHORITY) {
837 clientSOA = getRR<SOARecordContent>(answer.first);
838 if (clientSOA != nullptr) {
839 break;
840 }
841 }
842 } /* for (auto const &answer : mdp.d_answers) */
843
844 if (clientSOA == nullptr) {
845 g_log<<Logger::Warning<<prefix<<"IXFR request packet did not contain a SOA record in the AUTHORITY section"<<endl;
846 close(cfd);
847 continue;
848 }
849
850 if (!handleIXFR(cfd, saddr, mdp, clientSOA)) {
851 close(cfd);
852 continue;
853 }
854 } /* if (mdp.d_qtype == QType::IXFR) */
855
856 shutdown(cfd, 2);
857 } catch (MOADNSException &e) {
858 g_log<<Logger::Warning<<prefix<<"Could not parse DNS packet from "<<saddr.toStringWithPort()<<": "<<e.what()<<endl;
859 } catch (runtime_error &e) {
860 g_log<<Logger::Warning<<prefix<<"Could not write reply to "<<saddr.toStringWithPort()<<": "<<e.what()<<endl;
861 }
862 // bye!
863 close(cfd);
864
865 if (g_exiting) {
866 break;
867 }
868 }
869 }
870
871 /* Parses the configuration file in configpath into config, adding defaults for
872 * missing parameters (if applicable), returning true if the config file was
873 * good, false otherwise. Will log all issues with the config
874 */
875 static bool parseAndCheckConfig(const string& configpath, YAML::Node& config) {
876 g_log<<Logger::Info<<"Loading configuration file from "<<configpath<<endl;
877 try {
878 config = YAML::LoadFile(configpath);
879 } catch (const runtime_error &e) {
880 g_log<<Logger::Error<<"Unable to load configuration file '"<<configpath<<"': "<<e.what()<<endl;
881 return false;
882 }
883
884 bool retval = true;
885
886 if (config["keep"]) {
887 try {
888 config["keep"].as<uint16_t>();
889 } catch (const runtime_error &e) {
890 g_log<<Logger::Error<<"Unable to read 'keep' value: "<<e.what()<<endl;
891 retval = false;
892 }
893 } else {
894 config["keep"] = 20;
895 }
896
897 if (config["axfr-timeout"]) {
898 try {
899 config["axfr-timeout"].as<uint16_t>();
900 } catch (const runtime_error &e) {
901 g_log<<Logger::Error<<"Unable to read 'axfr-timeout' value: "<<e.what()<<endl;
902 }
903 } else {
904 config["axfr-timeout"] = 20;
905 }
906
907 if (config["tcp-in-threads"]) {
908 try {
909 config["tcp-in-threads"].as<uint16_t>();
910 } catch (const runtime_error &e) {
911 g_log<<Logger::Error<<"Unable to read 'tcp-in-threads' value: "<<e.what()<<endl;
912 }
913 } else {
914 config["tcp-in-threads"] = 10;
915 }
916
917 if (config["listen"]) {
918 try {
919 config["listen"].as<vector<ComboAddress>>();
920 } catch (const runtime_error &e) {
921 g_log<<Logger::Error<<"Unable to read 'listen' value: "<<e.what()<<endl;
922 retval = false;
923 }
924 } else {
925 config["listen"].push_back("127.0.0.1:53");
926 config["listen"].push_back("[::1]:53");
927 }
928
929 if (config["acl"]) {
930 try {
931 config["acl"].as<vector<string>>();
932 } catch (const runtime_error &e) {
933 g_log<<Logger::Error<<"Unable to read 'acl' value: "<<e.what()<<endl;
934 retval = false;
935 }
936 } else {
937 config["acl"].push_back("127.0.0.0/8");
938 config["acl"].push_back("::1/128");
939 }
940
941 if (config["work-dir"]) {
942 try {
943 config["work-dir"].as<string>();
944 } catch(const runtime_error &e) {
945 g_log<<Logger::Error<<"Unable to read 'work-dir' value: "<<e.what()<<endl;
946 retval = false;
947 }
948 } else {
949 char tmp[512];
950 config["work-dir"] = getcwd(tmp, sizeof(tmp)) ? string(tmp) : "";;
951 }
952
953 if (config["uid"]) {
954 try {
955 config["uid"].as<string>();
956 } catch(const runtime_error &e) {
957 g_log<<Logger::Error<<"Unable to read 'uid' value: "<<e.what()<<endl;
958 retval = false;
959 }
960 }
961
962 if (config["gid"]) {
963 try {
964 config["gid"].as<string>();
965 } catch(const runtime_error &e) {
966 g_log<<Logger::Error<<"Unable to read 'gid' value: "<<e.what()<<endl;
967 retval = false;
968 }
969 }
970
971 if (config["domains"]) {
972 if (config["domains"].size() == 0) {
973 g_log<<Logger::Error<<"No domains configured"<<endl;
974 retval = false;
975 }
976 for (auto const &domain : config["domains"]) {
977 try {
978 if (!domain["domain"]) {
979 g_log<<Logger::Error<<"An entry in 'domains' is missing a 'domain' key!"<<endl;
980 retval = false;
981 continue;
982 }
983 domain["domain"].as<DNSName>();
984 } catch (const runtime_error &e) {
985 g_log<<Logger::Error<<"Unable to read domain '"<<domain["domain"].as<string>()<<"': "<<e.what()<<endl;
986 }
987 try {
988 if (!domain["master"]) {
989 g_log<<Logger::Error<<"Domain '"<<domain["domain"].as<string>()<<"' has no master configured!"<<endl;
990 retval = false;
991 continue;
992 }
993 domain["master"].as<ComboAddress>();
994 } catch (const runtime_error &e) {
995 g_log<<Logger::Error<<"Unable to read domain '"<<domain["domain"].as<string>()<<"' master address: "<<e.what()<<endl;
996 retval = false;
997 }
998 }
999 } else {
1000 g_log<<Logger::Error<<"No domains configured"<<endl;
1001 retval = false;
1002 }
1003
1004 if (config["compress"]) {
1005 try {
1006 config["compress"].as<bool>();
1007 }
1008 catch (const runtime_error &e) {
1009 g_log<<Logger::Error<<"Unable to read 'compress' value: "<<e.what()<<endl;
1010 retval = false;
1011 }
1012 }
1013 else {
1014 config["compress"] = false;
1015 }
1016
1017 return retval;
1018 }
1019
1020 int main(int argc, char** argv) {
1021 g_log.setLoglevel(Logger::Notice);
1022 g_log.toConsole(Logger::Notice);
1023 g_log.setPrefixed(true);
1024 g_log.disableSyslog(true);
1025 g_log.setTimestamps(false);
1026 po::variables_map g_vm;
1027 try {
1028 po::options_description desc("IXFR distribution tool");
1029 desc.add_options()
1030 ("help", "produce help message")
1031 ("version", "Display the version of ixfrdist")
1032 ("verbose", "Be verbose")
1033 ("debug", "Be even more verbose")
1034 ("config", po::value<string>()->default_value(SYSCONFDIR + string("/ixfrdist.yml")), "Configuration file to use")
1035 ;
1036
1037 po::store(po::command_line_parser(argc, argv).options(desc).run(), g_vm);
1038 po::notify(g_vm);
1039
1040 if (g_vm.count("help") > 0) {
1041 usage(desc);
1042 return EXIT_SUCCESS;
1043 }
1044
1045 if (g_vm.count("version") > 0) {
1046 cout<<"ixfrdist "<<VERSION<<endl;
1047 return EXIT_SUCCESS;
1048 }
1049 } catch (po::error &e) {
1050 g_log<<Logger::Error<<e.what()<<". See `ixfrdist --help` for valid options"<<endl;
1051 return(EXIT_FAILURE);
1052 }
1053
1054 bool had_error = false;
1055
1056 if (g_vm.count("verbose")) {
1057 g_log.setLoglevel(Logger::Info);
1058 g_log.toConsole(Logger::Info);
1059 }
1060
1061 if (g_vm.count("debug") > 0) {
1062 g_log.setLoglevel(Logger::Debug);
1063 g_log.toConsole(Logger::Debug);
1064 }
1065
1066 g_log<<Logger::Notice<<"IXFR distributor version "<<VERSION<<" starting up!"<<endl;
1067
1068 YAML::Node config;
1069 if (!parseAndCheckConfig(g_vm["config"].as<string>(), config)) {
1070 // parseAndCheckConfig already logged whatever was wrong
1071 return EXIT_FAILURE;
1072 }
1073
1074 /* From hereon out, we known that all the values in config are valid. */
1075
1076 for (auto const &domain : config["domains"]) {
1077 set<ComboAddress> s;
1078 s.insert(domain["master"].as<ComboAddress>());
1079 g_domainConfigs[domain["domain"].as<DNSName>()].masters = s;
1080 }
1081
1082 for (const auto &addr : config["acl"].as<vector<string>>()) {
1083 try {
1084 g_acl.addMask(addr);
1085 } catch (const NetmaskException &e) {
1086 g_log<<Logger::Error<<e.reason<<endl;
1087 had_error = true;
1088 }
1089 }
1090 g_log<<Logger::Notice<<"ACL set to "<<g_acl.toString()<<"."<<endl;
1091
1092 if (config["compress"]) {
1093 g_compress = config["compress"].as<bool>();
1094 if (g_compress) {
1095 g_log<<Logger::Notice<<"Record compression is enabled."<<endl;
1096 }
1097 }
1098
1099 FDMultiplexer* fdm = FDMultiplexer::getMultiplexerSilent();
1100 if (fdm == nullptr) {
1101 g_log<<Logger::Error<<"Could not enable a multiplexer for the listen sockets!"<<endl;
1102 return EXIT_FAILURE;
1103 }
1104
1105 set<int> allSockets;
1106 for (const auto& addr : config["listen"].as<vector<ComboAddress>>()) {
1107 for (const auto& stype : {SOCK_DGRAM, SOCK_STREAM}) {
1108 try {
1109 int s = SSocket(addr.sin4.sin_family, stype, 0);
1110 setNonBlocking(s);
1111 setReuseAddr(s);
1112 SBind(s, addr);
1113 if (stype == SOCK_STREAM) {
1114 SListen(s, 30); // TODO make this configurable
1115 }
1116 fdm->addReadFD(s, stype == SOCK_DGRAM ? handleUDPRequest : handleTCPRequest);
1117 allSockets.insert(s);
1118 } catch(runtime_error &e) {
1119 g_log<<Logger::Error<<e.what()<<endl;
1120 had_error = true;
1121 continue;
1122 }
1123 }
1124 }
1125
1126 int newgid = 0;
1127
1128 if (config["gid"]) {
1129 string gid = config["gid"].as<string>();
1130 if (!(newgid = atoi(gid.c_str()))) {
1131 struct group *gr = getgrnam(gid.c_str());
1132 if (gr == nullptr) {
1133 g_log<<Logger::Error<<"Can not determine group-id for gid "<<gid<<endl;
1134 had_error = true;
1135 } else {
1136 newgid = gr->gr_gid;
1137 }
1138 }
1139 g_log<<Logger::Notice<<"Dropping effective group-id to "<<newgid<<endl;
1140 if (setgid(newgid) < 0) {
1141 g_log<<Logger::Error<<"Could not set group id to "<<newgid<<": "<<stringerror()<<endl;
1142 had_error = true;
1143 }
1144 }
1145
1146 int newuid = 0;
1147
1148 if (config["uid"]) {
1149 string uid = config["uid"].as<string>();
1150 if (!(newuid = atoi(uid.c_str()))) {
1151 struct passwd *pw = getpwnam(uid.c_str());
1152 if (pw == nullptr) {
1153 g_log<<Logger::Error<<"Can not determine user-id for uid "<<uid<<endl;
1154 had_error = true;
1155 } else {
1156 newuid = pw->pw_uid;
1157 }
1158 }
1159
1160 struct passwd *pw = getpwuid(newuid);
1161 if (pw == nullptr) {
1162 if (setgroups(0, nullptr) < 0) {
1163 g_log<<Logger::Error<<"Unable to drop supplementary gids: "<<stringerror()<<endl;
1164 had_error = true;
1165 }
1166 } else {
1167 if (initgroups(pw->pw_name, newgid) < 0) {
1168 g_log<<Logger::Error<<"Unable to set supplementary groups: "<<stringerror()<<endl;
1169 had_error = true;
1170 }
1171 }
1172
1173 g_log<<Logger::Notice<<"Dropping effective user-id to "<<newuid<<endl;
1174 if (setuid(newuid) < 0) {
1175 g_log<<Logger::Error<<"Could not set user id to "<<newuid<<": "<<stringerror()<<endl;
1176 had_error = true;
1177 }
1178 }
1179
1180 if (had_error) {
1181 // We have already sent the errors to stderr, just die
1182 return EXIT_FAILURE;
1183 }
1184
1185 // It all starts here
1186 signal(SIGTERM, handleSignal);
1187 signal(SIGINT, handleSignal);
1188 signal(SIGPIPE, SIG_IGN);
1189
1190 // Init the things we need
1191 reportAllTypes();
1192
1193 dns_random_init();
1194
1195 std::thread ut(updateThread,
1196 config["work-dir"].as<string>(),
1197 config["keep"].as<uint16_t>(),
1198 config["axfr-timeout"].as<uint16_t>());
1199
1200 vector<std::thread> tcpHandlers;
1201 tcpHandlers.reserve(config["tcp-in-threads"].as<uint16_t>());
1202 for (size_t i = 0; i < tcpHandlers.capacity(); ++i) {
1203 tcpHandlers.push_back(std::thread(tcpWorker, i));
1204 }
1205
1206 struct timeval now;
1207 for(;;) {
1208 gettimeofday(&now, 0);
1209 fdm->run(&now);
1210 if (g_exiting) {
1211 g_log<<Logger::Debug<<"Closing listening sockets"<<endl;
1212 for (const int& fd : allSockets) {
1213 try {
1214 closesocket(fd);
1215 } catch(PDNSException &e) {
1216 g_log<<Logger::Error<<e.reason<<endl;
1217 }
1218 }
1219 break;
1220 }
1221 }
1222 g_log<<Logger::Debug<<"Waiting for al threads to stop"<<endl;
1223 g_tcpHandlerCV.notify_all();
1224 ut.join();
1225 for (auto &t : tcpHandlers) {
1226 t.join();
1227 }
1228 g_log<<Logger::Notice<<"IXFR distributor stopped"<<endl;
1229 return EXIT_SUCCESS;
1230 }