--- /dev/null
+#!/usr/bin/env python
+from datetime import datetime, timedelta
+import dns
+import os
+import subprocess
+import threading
+import time
+import unittest
+from dnsdisttests import DNSDistTest
+
+class TestAdvancedFixupCase(DNSDistTest):
+
+ _dnsDistPort = 5340
+ _config_template = """
+ truncateTC(true)
+ fixupCase(true)
+ newServer{address="127.0.0.1:%s"}
+ """
+
+ _dnsdistcmd = (os.environ['DNSDISTBIN'] + " -C dnsdist_fixupcase.conf --acl 127.0.0.1/32 -l 127.0.0.1:" + str(_dnsDistPort)).split()
+
+ @classmethod
+ def startDNSDist(cls, shutUp=True):
+ print("Launching dnsdist..")
+ with open('dnsdist_fixupcase.conf', 'w') as conf:
+ conf.write(cls._config_template % str(cls._testServerPort))
+
+ print(' '.join(cls._dnsdistcmd))
+ if shutUp:
+ with open(os.devnull, 'w') as fdDevNull:
+ cls._dnsdist = subprocess.Popen(cls._dnsdistcmd, close_fds=True, stdout=fdDevNull, stderr=fdDevNull)
+ else:
+ cls._dnsdist = subprocess.Popen(cls._dnsdistcmd, close_fds=True)
+
+ time.sleep(1)
+
+ if cls._dnsdist.poll() is not None:
+ cls._dnsdist.terminate()
+ cls._dnsdist.wait()
+ sys.exit(cls._dnsdist.returncode)
+
+ def testAdvancedFixupCase(self):
+ """
+ Send a query with lower and upper chars,
+ make the backend return a lowercase version,
+ check that dnsdist fixes the response.
+ """
+ name = 'fiXuPCasE.advanced.tests.powerdns.com.'
+ query = dns.message.make_query(name, 'A', 'IN')
+ lowercasequery = dns.message.make_query(name.lower(), 'A', 'IN')
+ response = dns.message.make_response(lowercasequery)
+ expectedResponse = dns.message.make_response(query)
+ rrset = dns.rrset.from_text(name,
+ 3600,
+ dns.rdataclass.IN,
+ dns.rdatatype.A,
+ '127.0.0.1')
+ response.answer.append(rrset)
+ expectedResponse.answer.append(rrset)
+
+ (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = query.id
+ receivedResponse.id = expectedResponse.id
+ self.assertEquals(query, receivedQuery)
+ self.assertEquals(expectedResponse, receivedResponse)
+
+ (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = query.id
+ receivedResponse.id = expectedResponse.id
+ self.assertEquals(query, receivedQuery)
+ self.assertEquals(expectedResponse, receivedResponse)
+
+
+class TestAdvancedRemoveRD(DNSDistTest):
+
+ _dnsDistPort = 5340
+ _config_template = """
+ addNoRecurseRule("norecurse.advanced.tests.powerdns.com.")
+ newServer{address="127.0.0.1:%s"}
+ """
+
+ _dnsdistcmd = (os.environ['DNSDISTBIN'] + " -C dnsdist_noRD.conf --acl 127.0.0.1/32 -l 127.0.0.1:" + str(_dnsDistPort)).split()
+
+ @classmethod
+ def startDNSDist(cls, shutUp=True):
+ print("Launching dnsdist..")
+ with open('dnsdist_noRD.conf', 'w') as conf:
+ conf.write(cls._config_template % str(cls._testServerPort))
+
+ print(' '.join(cls._dnsdistcmd))
+ if shutUp:
+ with open(os.devnull, 'w') as fdDevNull:
+ cls._dnsdist = subprocess.Popen(cls._dnsdistcmd, close_fds=True, stdout=fdDevNull, stderr=fdDevNull)
+ else:
+ cls._dnsdist = subprocess.Popen(cls._dnsdistcmd, close_fds=True)
+
+ time.sleep(1)
+
+ if cls._dnsdist.poll() is not None:
+ cls._dnsdist.terminate()
+ cls._dnsdist.wait()
+ sys.exit(cls._dnsdist.returncode)
+
+ def testAdvancedNoRD(self):
+ """
+ Send a query with RD,
+ check that dnsdist clears the RD flag.
+ """
+ name = 'norecurse.advanced.tests.powerdns.com.'
+ query = dns.message.make_query(name, 'A', 'IN')
+ expectedQuery = dns.message.make_query(name, 'A', 'IN')
+ expectedQuery.flags &= ~dns.flags.RD
+
+ response = dns.message.make_response(query)
+ rrset = dns.rrset.from_text(name,
+ 3600,
+ dns.rdataclass.IN,
+ dns.rdatatype.A,
+ '127.0.0.1')
+ response.answer.append(rrset)
+
+ (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = expectedQuery.id
+ receivedResponse.id = response.id
+ self.assertEquals(expectedQuery, receivedQuery)
+ self.assertEquals(response, receivedResponse)
+
+ (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = expectedQuery.id
+ receivedResponse.id = response.id
+ self.assertEquals(expectedQuery, receivedQuery)
+ self.assertEquals(response, receivedResponse)
+
+ def testAdvancedKeepRD(self):
+ """
+ Send a query with RD for a canary domain,
+ check that dnsdist does not clear the RD flag.
+ """
+ name = 'keeprecurse.advanced.tests.powerdns.com.'
+ query = dns.message.make_query(name, 'A', 'IN')
+
+ response = dns.message.make_response(query)
+ rrset = dns.rrset.from_text(name,
+ 3600,
+ dns.rdataclass.IN,
+ dns.rdatatype.A,
+ '127.0.0.1')
+ response.answer.append(rrset)
+
+ (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = query.id
+ receivedResponse.id = response.id
+ self.assertEquals(query, receivedQuery)
+ self.assertEquals(response, receivedResponse)
+
+ (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = query.id
+ receivedResponse.id = response.id
+ self.assertEquals(query, receivedQuery)
+ self.assertEquals(response, receivedResponse)
+
+
+class TestAdvancedAddCD(DNSDistTest):
+
+ _dnsDistPort = 5340
+ _config_template = """
+ addDisableValidationRule("setcd.advanced.tests.powerdns.com.")
+ newServer{address="127.0.0.1:%s"}
+ """
+
+ _dnsdistcmd = (os.environ['DNSDISTBIN'] + " -C dnsdist_setCD.conf --acl 127.0.0.1/32 -l 127.0.0.1:" + str(_dnsDistPort)).split()
+
+ @classmethod
+ def startDNSDist(cls, shutUp=True):
+ print("Launching dnsdist..")
+ with open('dnsdist_setCD.conf', 'w') as conf:
+ conf.write(cls._config_template % str(cls._testServerPort))
+
+ print(' '.join(cls._dnsdistcmd))
+ if shutUp:
+ with open(os.devnull, 'w') as fdDevNull:
+ cls._dnsdist = subprocess.Popen(cls._dnsdistcmd, close_fds=True, stdout=fdDevNull, stderr=fdDevNull)
+ else:
+ cls._dnsdist = subprocess.Popen(cls._dnsdistcmd, close_fds=True)
+
+ time.sleep(1)
+
+ if cls._dnsdist.poll() is not None:
+ cls._dnsdist.terminate()
+ cls._dnsdist.wait()
+ sys.exit(cls._dnsdist.returncode)
+
+ def testAdvancedSetCD(self):
+ """
+ Send a query with CD cleared,
+ check that dnsdist set the CD flag.
+ """
+ name = 'setcd.advanced.tests.powerdns.com.'
+ query = dns.message.make_query(name, 'A', 'IN')
+ expectedQuery = dns.message.make_query(name.lower(), 'A', 'IN')
+ expectedQuery.flags |= dns.flags.CD
+
+ response = dns.message.make_response(query)
+ rrset = dns.rrset.from_text(name,
+ 3600,
+ dns.rdataclass.IN,
+ dns.rdatatype.A,
+ '127.0.0.1')
+ response.answer.append(rrset)
+
+ (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = expectedQuery.id
+ receivedResponse.id = response.id
+ self.assertEquals(expectedQuery, receivedQuery)
+ self.assertEquals(response, receivedResponse)
+
+ (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = expectedQuery.id
+ receivedResponse.id = response.id
+ self.assertEquals(expectedQuery, receivedQuery)
+ self.assertEquals(response, receivedResponse)
+
+ def testAdvancedKeepNoCD(self):
+ """
+ Send a query without CD for a canary domain,
+ check that dnsdist does not set the CD flag.
+ """
+ name = 'keepnocd.advanced.tests.powerdns.com.'
+ query = dns.message.make_query(name, 'A', 'IN')
+
+ response = dns.message.make_response(query)
+ rrset = dns.rrset.from_text(name,
+ 3600,
+ dns.rdataclass.IN,
+ dns.rdatatype.A,
+ '127.0.0.1')
+ response.answer.append(rrset)
+
+ (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = query.id
+ receivedResponse.id = response.id
+ self.assertEquals(query, receivedQuery)
+ self.assertEquals(response, receivedResponse)
+
+ (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = query.id
+ receivedResponse.id = response.id
+ self.assertEquals(query, receivedQuery)
+ self.assertEquals(response, receivedResponse)
+
+class TestAdvancedSpoof(DNSDistTest):
+
+ _dnsDistPort = 5340
+ _config_template = """
+ addDomainSpoof("spoof.tests.powerdns.com.", "192.0.2.1", "2001:DB8::1")
+ newServer{address="127.0.0.1:%s"}
+ """
+
+ _dnsdistcmd = (os.environ['DNSDISTBIN'] + " -C dnsdist_spoof.conf --acl 127.0.0.1/32 -l 127.0.0.1:" + str(_dnsDistPort)).split()
+
+ @classmethod
+ def startDNSDist(cls, shutUp=True):
+ print("Launching dnsdist..")
+ with open('dnsdist_spoof.conf', 'w') as conf:
+ conf.write(cls._config_template % str(cls._testServerPort))
+
+ print(' '.join(cls._dnsdistcmd))
+ if shutUp:
+ with open(os.devnull, 'w') as fdDevNull:
+ cls._dnsdist = subprocess.Popen(cls._dnsdistcmd, close_fds=True, stdout=fdDevNull, stderr=fdDevNull)
+ else:
+ cls._dnsdist = subprocess.Popen(cls._dnsdistcmd, close_fds=True)
+
+ time.sleep(1)
+
+ if cls._dnsdist.poll() is not None:
+ cls._dnsdist.terminate()
+ cls._dnsdist.wait()
+ sys.exit(cls._dnsdist.returncode)
+
+ def testSpoofA(self):
+ """
+ Send an A query to "spoof.tests.powerdns.com.",
+ check that dnsdist sends a spoofed result.
+ """
+ name = 'spoof.tests.powerdns.com.'
+ query = dns.message.make_query(name, 'A', 'IN')
+ # dnsdist set RA = RD for spoofed responses
+ query.flags &= ~dns.flags.RD
+ expectedResponse = dns.message.make_response(query)
+ rrset = dns.rrset.from_text(name,
+ 60,
+ dns.rdataclass.IN,
+ dns.rdatatype.A,
+ '192.0.2.1')
+ expectedResponse.answer.append(rrset)
+
+ (_, receivedResponse) = self.sendUDPQuery(query, response=None, useQueue=False, timeout=2.0)
+ self.assertTrue(receivedResponse)
+ receivedResponse.id = expectedResponse.id
+ self.assertEquals(expectedResponse, receivedResponse)
+
+ (_, receivedResponse) = self.sendTCPQuery(query, response=None, useQueue=False, timeout=2.0)
+ self.assertTrue(receivedResponse)
+ receivedResponse.id = expectedResponse.id
+ self.assertEquals(expectedResponse, receivedResponse)
+
+ def testSpoofAAAA(self):
+ """
+ Send an AAAA query to "spoof.tests.powerdns.com.",
+ check that dnsdist sends a spoofed result.
+ """
+ name = 'spoof.tests.powerdns.com.'
+ query = dns.message.make_query(name, 'AAAA', 'IN')
+ # dnsdist set RA = RD for spoofed responses
+ query.flags &= ~dns.flags.RD
+ expectedResponse = dns.message.make_response(query)
+ rrset = dns.rrset.from_text(name,
+ 60,
+ dns.rdataclass.IN,
+ dns.rdatatype.AAAA,
+ '2001:DB8::1')
+ expectedResponse.answer.append(rrset)
+
+ (_, receivedResponse) = self.sendUDPQuery(query, response=None, useQueue=False, timeout=2.0)
+ self.assertTrue(receivedResponse)
+ receivedResponse.id = expectedResponse.id
+ self.assertEquals(expectedResponse, receivedResponse)
+
+ (_, receivedResponse) = self.sendTCPQuery(query, response=None, useQueue=False, timeout=2.0)
+ self.assertTrue(receivedResponse)
+ receivedResponse.id = expectedResponse.id
+ self.assertEquals(expectedResponse, receivedResponse)
+
+
+class TestAdvancedPoolRouting(DNSDistTest):
+
+ _dnsDistPort = 5340
+ _config_template = """
+ newServer{address="127.0.0.1:%s", pool="real"}
+ addPoolRule("pool.tests.powerdns.com", "real")
+ """
+
+ _dnsdistcmd = (os.environ['DNSDISTBIN'] + " -C dnsdist_pool.conf --acl 127.0.0.1/32 -l 127.0.0.1:" + str(_dnsDistPort)).split()
+
+ @classmethod
+ def startDNSDist(cls, shutUp=True):
+ print("Launching dnsdist..")
+ with open('dnsdist_pool.conf', 'w') as conf:
+ conf.write(cls._config_template % str(cls._testServerPort))
+
+ print(' '.join(cls._dnsdistcmd))
+ if shutUp:
+ with open(os.devnull, 'w') as fdDevNull:
+ cls._dnsdist = subprocess.Popen(cls._dnsdistcmd, close_fds=True, stdout=fdDevNull, stderr=fdDevNull)
+ else:
+ cls._dnsdist = subprocess.Popen(cls._dnsdistcmd, close_fds=True)
+
+ time.sleep(1)
+
+ if cls._dnsdist.poll() is not None:
+ cls._dnsdist.terminate()
+ cls._dnsdist.wait()
+ sys.exit(cls._dnsdist.returncode)
+
+ def testPolicyPool(self):
+ """
+ Send an A query to "pool.tests.powerdns.com.",
+ check that dnsdist routes the query to the "real" pool.
+ """
+ name = 'pool.tests.powerdns.com.'
+ query = dns.message.make_query(name, 'A', 'IN')
+ response = dns.message.make_response(query)
+ rrset = dns.rrset.from_text(name,
+ 60,
+ dns.rdataclass.IN,
+ dns.rdatatype.A,
+ '192.0.2.1')
+ response.answer.append(rrset)
+
+ (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
+ receivedQuery.id = query.id
+ receivedResponse.id = response.id
+ self.assertEquals(query, receivedQuery)
+ self.assertEquals(response, receivedResponse)
+
+ (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
+ receivedQuery.id = query.id
+ receivedResponse.id = response.id
+ self.assertEquals(query, receivedQuery)
+ self.assertEquals(response, receivedResponse)
+
+ def testDefaultPool(self):
+ """
+ Send an A query to "notpool.tests.powerdns.com.",
+ check that dnsdist sends no response (no servers
+ in the default pool).
+ """
+ name = 'notpool.tests.powerdns.com.'
+ query = dns.message.make_query(name, 'A', 'IN')
+
+ (_, receivedResponse) = self.sendUDPQuery(query, response=None, useQueue=False, timeout=2.0)
+ self.assertEquals(receivedResponse, None)
+
+ (_, receivedResponse) = self.sendTCPQuery(query, response=None, useQueue=False, timeout=2.0)
+ self.assertEquals(receivedResponse, None)
+
+class TestAdvancedRoundRobinLB(DNSDistTest):
+
+ _dnsDistPort = 5340
+ _testServer2Port = 5351
+ _config_template = """
+ setServerPolicy(roundrobin)
+ s1 = newServer{address="127.0.0.1:%s"}
+ s1:setUp()
+ s2 = newServer{address="127.0.0.1:%s"}
+ s2:setUp()
+ """
+
+ _dnsdistcmd = (os.environ['DNSDISTBIN'] + " -C dnsdist_rr_lb.conf --acl 127.0.0.1/32 -l 127.0.0.1:" + str(_dnsDistPort)).split()
+
+ @classmethod
+ def startDNSDist(cls, shutUp=True):
+ print("Launching dnsdist..")
+ with open('dnsdist_rr_lb.conf', 'w') as conf:
+ conf.write(cls._config_template % (str(cls._testServerPort), str(cls._testServer2Port)))
+
+ print(' '.join(cls._dnsdistcmd))
+ if shutUp:
+ with open(os.devnull, 'w') as fdDevNull:
+ cls._dnsdist = subprocess.Popen(cls._dnsdistcmd, close_fds=True, stdout=fdDevNull, stderr=fdDevNull)
+ else:
+ cls._dnsdist = subprocess.Popen(cls._dnsdistcmd, close_fds=True)
+
+ time.sleep(1)
+
+ if cls._dnsdist.poll() is not None:
+ cls._dnsdist.terminate()
+ cls._dnsdist.wait()
+ sys.exit(cls._dnsdist.returncode)
+
+ @classmethod
+ def startResponders(cls):
+ print("Launching responders..")
+ cls._UDPResponder = threading.Thread(name='UDP Responder', target=cls.UDPResponder, args=[cls._testServerPort])
+ cls._UDPResponder.setDaemon(True)
+ cls._UDPResponder.start()
+ cls._UDPResponder2 = threading.Thread(name='UDP Responder 2', target=cls.UDPResponder, args=[cls._testServer2Port])
+ cls._UDPResponder2.setDaemon(True)
+ cls._UDPResponder2.start()
+
+ cls._TCPResponder = threading.Thread(name='TCP Responder', target=cls.TCPResponder, args=[cls._testServerPort])
+ cls._TCPResponder.setDaemon(True)
+ cls._TCPResponder.start()
+
+ cls._TCPResponder2 = threading.Thread(name='TCP Responder 2', target=cls.TCPResponder, args=[cls._testServer2Port])
+ cls._TCPResponder2.setDaemon(True)
+ cls._TCPResponder2.start()
+
+ def testRR(self):
+ """
+ Send 100 A queries to "rr.tests.powerdns.com.",
+ check that dnsdist routes half of it to each backend.
+ """
+ numberOfQueries = 10
+ name = 'rr.tests.powerdns.com.'
+ query = dns.message.make_query(name, 'A', 'IN')
+ response = dns.message.make_response(query)
+ rrset = dns.rrset.from_text(name,
+ 60,
+ dns.rdataclass.IN,
+ dns.rdatatype.A,
+ '192.0.2.1')
+ response.answer.append(rrset)
+
+ # clear counters
+ for key in TestAdvancedRoundRobinLB._responsesCounter:
+ TestAdvancedRoundRobinLB._responsesCounter[key] = 0
+
+ # the round robin counter is shared for UDP and TCP,
+ # so we need to do UDP then TCP to have a clean count
+ for idx in range(numberOfQueries):
+ (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
+ receivedQuery.id = query.id
+ receivedResponse.id = response.id
+ self.assertEquals(query, receivedQuery)
+ self.assertEquals(response, receivedResponse)
+
+ for idx in range(numberOfQueries):
+ (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
+ receivedQuery.id = query.id
+ receivedResponse.id = response.id
+ self.assertEquals(query, receivedQuery)
+ self.assertEquals(response, receivedResponse)
+
+ for key in TestAdvancedRoundRobinLB._responsesCounter:
+ value = TestAdvancedRoundRobinLB._responsesCounter[key]
+ self.assertEquals(value, numberOfQueries / 2)
+
+class TestAdvancedRoundRobinLBOneDown(DNSDistTest):
+
+ _dnsDistPort = 5340
+ _testServer2Port = 5351
+ _config_template = """
+ setServerPolicy(roundrobin)
+ s1 = newServer{address="127.0.0.1:%s"}
+ s1:setUp()
+ s2 = newServer{address="127.0.0.1:%s"}
+ s2:setDown()
+ """
+
+ _dnsdistcmd = (os.environ['DNSDISTBIN'] + " -C dnsdist_rr_lb_down.conf --acl 127.0.0.1/32 -l 127.0.0.1:" + str(_dnsDistPort)).split()
+
+ @classmethod
+ def startDNSDist(cls, shutUp=True):
+ print("Launching dnsdist..")
+ with open('dnsdist_rr_lb_down.conf', 'w') as conf:
+ conf.write(cls._config_template % (str(cls._testServerPort), str(cls._testServer2Port)))
+
+ print(' '.join(cls._dnsdistcmd))
+ if shutUp:
+ with open(os.devnull, 'w') as fdDevNull:
+ cls._dnsdist = subprocess.Popen(cls._dnsdistcmd, close_fds=True, stdout=fdDevNull, stderr=fdDevNull)
+ else:
+ cls._dnsdist = subprocess.Popen(cls._dnsdistcmd, close_fds=True)
+
+ time.sleep(1)
+
+ if cls._dnsdist.poll() is not None:
+ cls._dnsdist.terminate()
+ cls._dnsdist.wait()
+ sys.exit(cls._dnsdist.returncode)
+
+ def testRRWithOneDown(self):
+ """
+ Send 100 A queries to "rr.tests.powerdns.com.",
+ check that dnsdist routes all of it to the only backend up.
+ """
+ numberOfQueries = 10
+ name = 'rr.tests.powerdns.com.'
+ query = dns.message.make_query(name, 'A', 'IN')
+ response = dns.message.make_response(query)
+ rrset = dns.rrset.from_text(name,
+ 60,
+ dns.rdataclass.IN,
+ dns.rdatatype.A,
+ '192.0.2.1')
+ response.answer.append(rrset)
+
+ # the round robin counter is shared for UDP and TCP,
+ # so we need to do UDP then TCP to have a clean count
+ for idx in range(numberOfQueries):
+ (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
+ receivedQuery.id = query.id
+ receivedResponse.id = response.id
+ self.assertEquals(query, receivedQuery)
+ self.assertEquals(response, receivedResponse)
+
+ for idx in range(numberOfQueries):
+ (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
+ receivedQuery.id = query.id
+ receivedResponse.id = response.id
+ self.assertEquals(query, receivedQuery)
+ self.assertEquals(response, receivedResponse)
+
+ print(TestAdvancedRoundRobinLB._responsesCounter)
+ total = 0
+ for key in TestAdvancedRoundRobinLB._responsesCounter:
+ value = TestAdvancedRoundRobinLB._responsesCounter[key]
+ self.assertTrue(value == numberOfQueries or value == 0)
+ total += value
+
+ self.assertEquals(total, numberOfQueries * 2)
+
+class TestAdvancedACL(DNSDistTest):
+
+ _dnsDistPort = 5340
+ _config_template = """
+ newServer{address="127.0.0.1:%s"}
+ """
+
+ _dnsdistcmd = (os.environ['DNSDISTBIN'] + " -C dnsdist_acl.conf --acl 192.0.2.1/32 -l 127.0.0.1:" + str(_dnsDistPort)).split()
+
+ @classmethod
+ def startDNSDist(cls, shutUp=True):
+ print("Launching dnsdist..")
+ with open('dnsdist_acl.conf', 'w') as conf:
+ conf.write(cls._config_template % str(cls._testServerPort))
+
+ print(' '.join(cls._dnsdistcmd))
+ if shutUp:
+ with open(os.devnull, 'w') as fdDevNull:
+ cls._dnsdist = subprocess.Popen(cls._dnsdistcmd, close_fds=True, stdout=fdDevNull, stderr=fdDevNull)
+ else:
+ cls._dnsdist = subprocess.Popen(cls._dnsdistcmd, close_fds=True)
+
+ time.sleep(1)
+
+ if cls._dnsdist.poll() is not None:
+ cls._dnsdist.terminate()
+ cls._dnsdist.wait()
+ sys.exit(cls._dnsdist.returncode)
+
+ def testACLBlocked(self):
+ """
+ Send an A query to "tests.powerdns.com.",
+ we expect no response since 127.0.0.1 is not on the
+ ACL.
+ """
+ name = 'tests.powerdns.com.'
+ query = dns.message.make_query(name, 'A', 'IN')
+
+ (_, receivedResponse) = self.sendUDPQuery(query, response=None, useQueue=False, timeout=2.0)
+ self.assertEquals(receivedResponse, None)
+
+ (_, receivedResponse) = self.sendTCPQuery(query, response=None, useQueue=False, timeout=2.0)
+ self.assertEquals(receivedResponse, None)
+
+class TestAdvancedDelay(DNSDistTest):
+
+ _dnsDistPort = 5340
+ _config_template = """
+ addAction(AllRule(), DelayAction(1000))
+ newServer{address="127.0.0.1:%s"}
+ """
+
+ _dnsdistcmd = (os.environ['DNSDISTBIN'] + " -C dnsdist_delay.conf --acl 127.0.0.1/32 -l 127.0.0.1:" + str(_dnsDistPort)).split()
+
+ @classmethod
+ def startDNSDist(cls, shutUp=True):
+ print("Launching dnsdist..")
+ with open('dnsdist_delay.conf', 'w') as conf:
+ conf.write(cls._config_template % str(cls._testServerPort))
+
+ print(' '.join(cls._dnsdistcmd))
+ if shutUp:
+ with open(os.devnull, 'w') as fdDevNull:
+ cls._dnsdist = subprocess.Popen(cls._dnsdistcmd, close_fds=True, stdout=fdDevNull, stderr=fdDevNull)
+ else:
+ cls._dnsdist = subprocess.Popen(cls._dnsdistcmd, close_fds=True)
+
+ time.sleep(1)
+
+ if cls._dnsdist.poll() is not None:
+ cls._dnsdist.terminate()
+ cls._dnsdist.wait()
+ sys.exit(cls._dnsdist.returncode)
+
+ def testDelayed(self):
+ """
+ Send an A query to "tests.powerdns.com.",
+ check that the response delay is longer than 1000 ms
+ over UDP, less than that over TCP.
+ """
+ name = 'tests.powerdns.com.'
+ query = dns.message.make_query(name, 'A', 'IN')
+ response = dns.message.make_response(query)
+ rrset = dns.rrset.from_text(name,
+ 60,
+ dns.rdataclass.IN,
+ dns.rdatatype.A,
+ '192.0.2.1')
+ response.answer.append(rrset)
+
+ begin = datetime.now()
+ (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
+ end = datetime.now()
+ receivedQuery.id = query.id
+ receivedResponse.id = response.id
+ self.assertEquals(query, receivedQuery)
+ self.assertEquals(response, receivedResponse)
+ self.assertTrue((end - begin) > timedelta(0, 1));
+
+ begin = datetime.now()
+ (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
+ end = datetime.now()
+ receivedQuery.id = query.id
+ receivedResponse.id = response.id
+ self.assertEquals(query, receivedQuery)
+ self.assertEquals(response, receivedResponse)
+ self.assertTrue((end - begin) < timedelta(0, 1));
self.assertEquals(query, receivedQuery)
self.assertEquals(response, receivedResponse)
+ def testAnyIsTruncated(self):
+ """
+ dnsdist is configured to reply with TC to ANY queries,
+ send an ANY query and check the result.
+ """
+ name = 'any.tests.powerdns.com.'
+ query = dns.message.make_query(name, 'ANY', 'IN')
+ expectedResponse = dns.message.make_response(query)
+ expectedResponse.flags |= dns.flags.TC
+
+ (_, receivedResponse) = self.sendUDPQuery(query, response=None, useQueue=False, timeout=2.0)
+ receivedResponse.id = expectedResponse.id
+ self.assertEquals(receivedResponse, expectedResponse)
+
+ (_, receivedResponse) = self.sendTCPQuery(query, response=None, useQueue=False, timeout=2.0)
+ receivedResponse.id = expectedResponse.id
+ self.assertEquals(receivedResponse, expectedResponse)
+
+ def testTruncateTC(self):
+ """
+ dnsdist is configured to truncate TC (default),
+ we make the backend send responses
+ with TC set and additional content,
+ and check that the received response has been fixed.
+ """
+ name = 'atruncatetc.tests.powerdns.com.'
+ query = dns.message.make_query(name, 'A', 'IN')
+ response = dns.message.make_response(query)
+ rrset = dns.rrset.from_text(name,
+ 3600,
+ dns.rdataclass.IN,
+ dns.rdatatype.A,
+ '127.0.0.1')
+
+ response.answer.append(rrset)
+ response.flags |= dns.flags.TC
+
+ (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
+ receivedQuery.id = query.id
+ receivedResponse.id = response.id
+ self.assertEquals(query, receivedQuery)
+ self.assertEquals(response.flags, receivedResponse.flags)
+ self.assertEquals(response.question, receivedResponse.question)
+ self.assertFalse(response.answer == receivedResponse.answer)
+ self.assertEquals(len(receivedResponse.answer), 0)
+ self.assertEquals(len(receivedResponse.authority), 0)
+ self.assertEquals(len(receivedResponse.additional), 0)
+
+ def testRegexReturnsRefused(self):
+ """
+ dnsdist is configured to reply 'refused' for query
+ matching "evil[0-9]{4,}\\.regex\\.tests\\.powerdns\\.com$".
+ We send a query for evil4242.powerdns.com
+ and check that the response is "refused".
+ """
+ name = 'evil4242.regex.tests.powerdns.com.'
+ query = dns.message.make_query(name, 'A', 'IN')
+ expectedResponse = dns.message.make_response(query)
+ expectedResponse.set_rcode(dns.rcode.REFUSED)
+
+ (_, receivedResponse) = self.sendUDPQuery(query, response=None, useQueue=False, timeout=2.0)
+ receivedResponse.id = expectedResponse.id
+ self.assertEquals(receivedResponse, expectedResponse)
+
+ (_, receivedResponse) = self.sendTCPQuery(query, response=None, useQueue=False, timeout=2.0)
+ receivedResponse.id = expectedResponse.id
+ self.assertEquals(receivedResponse, expectedResponse)
+
+ def testDomainAndQTypeReturnsNotImplemented(self):
+ """
+ dnsdist is configured to reply 'not implemented' for query
+ matching "nameAndQtype.tests.powerdns.com." AND qtype TXT/
+ We send a TXT query for "nameAndQtype.powerdns.com."
+ and check that the response is 'not implemented'.
+ """
+ name = 'nameAndQtype.tests.powerdns.com.'
+ query = dns.message.make_query(name, 'TXT', 'IN')
+ expectedResponse = dns.message.make_response(query)
+ expectedResponse.set_rcode(dns.rcode.NOTIMP)
+
+ (_, receivedResponse) = self.sendUDPQuery(query, response=None, useQueue=False, timeout=2.0)
+ receivedResponse.id = expectedResponse.id
+ self.assertEquals(receivedResponse, expectedResponse)
+
+ (_, receivedResponse) = self.sendTCPQuery(query, response=None, useQueue=False, timeout=2.0)
+ receivedResponse.id = expectedResponse.id
+ self.assertEquals(receivedResponse, expectedResponse)
+
+ def testDomainWithoutQTypeIsNotAffected(self):
+ """
+ dnsdist is configured to reply 'not implemented' for query
+ matching "nameAndQtype.tests.powerdns.com." AND qtype TXT/
+ We send a A query for "nameAndQtype.tests.powerdns.com."
+ and check that the response is OK.
+ """
+ name = 'nameAndQtype.tests.powerdns.com.'
+ query = dns.message.make_query(name, 'A', 'IN')
+ response = dns.message.make_response(query)
+ rrset = dns.rrset.from_text(name,
+ 3600,
+ dns.rdataclass.IN,
+ dns.rdatatype.A,
+ '127.0.0.1')
+ response.answer.append(rrset)
+
+ (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = query.id
+ receivedResponse.id = response.id
+ self.assertEquals(query, receivedQuery)
+ self.assertEquals(response, receivedResponse)
+
+ (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = query.id
+ receivedResponse.id = response.id
+ self.assertEquals(query, receivedQuery)
+ self.assertEquals(response, receivedResponse)
+
+ def testOtherDomainANDQTypeIsNotAffected(self):
+ """
+ dnsdist is configured to reply 'not implemented' for query
+ matching "nameAndQtype.tests.powerdns.com." AND qtype TXT/
+ We send a TXT query for "OtherNameAndQtype.tests.powerdns.com."
+ and check that the response is OK.
+ """
+ name = 'OtherNameAndQtype.tests.powerdns.com.'
+ query = dns.message.make_query(name, 'TXT', 'IN')
+ response = dns.message.make_response(query)
+ rrset = dns.rrset.from_text(name,
+ 3600,
+ dns.rdataclass.IN,
+ dns.rdatatype.TXT,
+ 'nothing to see here')
+ response.answer.append(rrset)
+
+ (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = query.id
+ receivedResponse.id = response.id
+ self.assertEquals(query, receivedQuery)
+ self.assertEquals(response, receivedResponse)
+
+ (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = query.id
+ receivedResponse.id = response.id
+ self.assertEquals(query, receivedQuery)
+ self.assertEquals(response, receivedResponse)
+
if __name__ == '__main__':
unittest.main()
projects / thirdparty / pdns.git / commitdiff
