return conn
@classmethod
- def sendDOHQuery(cls, port, servername, baseurl, query, response=None, timeout=2.0, caFile=None, useQueue=True, rawQuery=False):
+ def sendDOHQuery(cls, port, servername, baseurl, query, response=None, timeout=2.0, caFile=None, useQueue=True, rawQuery=False, customHeaders=[]):
url = cls.getDOHGetURL(baseurl, query, rawQuery)
conn = cls.openDOHConnection(port, caFile=caFile, timeout=timeout)
#conn.setopt(pycurl.VERBOSE, True)
conn.setopt(pycurl.RESOLVE, ["%s:%d:127.0.0.1" % (servername, port)])
conn.setopt(pycurl.SSL_VERIFYPEER, 1)
conn.setopt(pycurl.SSL_VERIFYHOST, 2)
+ conn.setopt(pycurl.HTTPHEADER, customHeaders)
if caFile:
conn.setopt(pycurl.CAINFO, caFile)
addAction("drop.doh.tests.powerdns.com.", DropAction())
addAction("refused.doh.tests.powerdns.com.", RCodeAction(DNSRCode.REFUSED))
addAction("spoof.doh.tests.powerdns.com.", SpoofAction("1.2.3.4"))
+ addAction(HTTPHeaderRule("X-PowerDNS", "^[a]{5}$"), SpoofAction("2.3.4.5"))
+ addAction(HTTPPathRule("/PowerDNS"), SpoofAction("3.4.5.6"))
"""
_config_params = ['_testServerPort', '_dohServerPort', '_serverCert', '_serverKey']
name = 'drop.doh.tests.powerdns.com.'
query = dns.message.make_query(name, 'A', 'IN')
(_, receivedResponse) = self.sendDOHQuery(self._dohServerPort, self._serverName, self._dohBaseURL, caFile=self._caCert, query=query, response=None, useQueue=False)
- print(receivedResponse)
self.assertEquals(receivedResponse, None)
def testRefused(self):
self.checkQueryEDNSWithoutECS(expectedQuery, receivedQuery)
self.assertEquals(response, receivedResponse)
+ def testHeaderRule(self):
+ """
+ DOH: HeaderRule
+ """
+ name = 'header-rule.doh.tests.powerdns.com.'
+ query = dns.message.make_query(name, 'A', 'IN')
+ query.id = 0
+ query.flags &= ~dns.flags.RD
+ expectedResponse = dns.message.make_response(query)
+ rrset = dns.rrset.from_text(name,
+ 3600,
+ dns.rdataclass.IN,
+ dns.rdatatype.A,
+ '2.3.4.5')
+ expectedResponse.answer.append(rrset)
+
+ # this header should match
+ (_, receivedResponse) = self.sendDOHQuery(self._dohServerPort, self._serverName, self._dohBaseURL, caFile=self._caCert, query=query, response=None, useQueue=False, customHeaders=['x-powerdnS: aaaaa'])
+ self.assertEquals(receivedResponse, expectedResponse)
+
+ expectedQuery = dns.message.make_query(name, 'A', 'IN', use_edns=True, payload=4096)
+ expectedQuery.flags &= ~dns.flags.RD
+ expectedQuery.id = 0
+ 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)
+
+ # this content of the header should NOT match
+ (receivedQuery, receivedResponse) = self.sendDOHQuery(self._dohServerPort, self._serverName, self._dohBaseURL, query, response=response, caFile=self._caCert, customHeaders=['x-powerdnS: bbbbb'])
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = expectedQuery.id
+ self.assertEquals(expectedQuery, receivedQuery)
+ self.checkQueryEDNSWithoutECS(expectedQuery, receivedQuery)
+ self.assertEquals(response, receivedResponse)
+
+ def testHTTPPath(self):
+ """
+ DOH: HTTPPath
+ """
+ name = 'http-path.doh.tests.powerdns.com.'
+ query = dns.message.make_query(name, 'A', 'IN')
+ query.id = 0
+ query.flags &= ~dns.flags.RD
+ expectedResponse = dns.message.make_response(query)
+ rrset = dns.rrset.from_text(name,
+ 3600,
+ dns.rdataclass.IN,
+ dns.rdatatype.A,
+ '3.4.5.6')
+ expectedResponse.answer.append(rrset)
+
+ # this path should match
+ (_, receivedResponse) = self.sendDOHQuery(self._dohServerPort, self._serverName, self._dohBaseURL + 'PowerDNS', caFile=self._caCert, query=query, response=None, useQueue=False)
+ self.assertEquals(receivedResponse, expectedResponse)
+
+ expectedQuery = dns.message.make_query(name, 'A', 'IN', use_edns=True, payload=4096)
+ expectedQuery.id = 0
+ 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)
+
+ # this path should NOT match
+ (receivedQuery, receivedResponse) = self.sendDOHQuery(self._dohServerPort, self._serverName, self._dohBaseURL + "PowerDNS2", query, response=response, caFile=self._caCert)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = expectedQuery.id
+ self.assertEquals(expectedQuery, receivedQuery)
+ self.checkQueryEDNSWithoutECS(expectedQuery, receivedQuery)
+ self.assertEquals(response, receivedResponse)
+
class TestDOHAddingECS(DNSDistDOHTest):
_serverKey = 'server.key'