#!/usr/bin/env python
import base64
import dns
+import os
+import re
+import time
+import unittest
import clientsubnetoption
from dnsdisttests import DNSDistTest
#from hyper import HTTP20Connection
#from hyper.ssl_compat import SSLContext, PROTOCOL_TLSv1_2
+@unittest.skipIf('SKIP_DOH_TESTS' in os.environ, 'DNS over HTTPS tests are disabled')
class DNSDistDOHTest(DNSDistTest):
@classmethod
cls._response_headers = response_headers.getvalue()
return (receivedQuery, message)
+ def getHeaderValue(self, name):
+ for header in self._response_headers.decode().splitlines(False):
+ values = header.split(':')
+ key = values[0]
+ if key.lower() == name.lower():
+ return values[1].strip()
+ return None
+
+ def checkHasHeader(self, name, value):
+ got = self.getHeaderValue(name)
+ self.assertEquals(got, value)
+
+ def checkNoHeader(self, name):
+ self.checkHasHeader(name, None)
+
+ @classmethod
+ def setUpClass(cls):
+
+ # for some reason, @unittest.skipIf() is not applied to derived classes with some versions of Python
+ if 'SKIP_DOH_TESTS' in os.environ:
+ raise unittest.SkipTest('DNS over HTTPS tests are disabled')
+
+ cls.startResponders()
+ cls.startDNSDist()
+ cls.setUpSockets()
+
+ print("Launching tests..")
+
# @classmethod
# def openDOHConnection(cls, port, caFile, timeout=2.0):
# sslctx = SSLContext(PROTOCOL_TLSv1_2)
_config_template = """
newServer{address="127.0.0.1:%s"}
- addDOHLocal("127.0.0.1:%s", "%s", "%s", { "/" }, {customResponseHeaders={["access-control-allow-origin"]="*",["user-agent"]="derp",["UPPERCASE"]="VaLuE"}})
+ addDOHLocal("127.0.0.1:%s", "%s", "%s", { "/", "/coffee", "/PowerDNS", "/PowerDNS2", "/PowerDNS-999" }, {customResponseHeaders={["access-control-allow-origin"]="*",["user-agent"]="derp",["UPPERCASE"]="VaLuE"}})
dohFE = getDOHFrontend(0)
dohFE:setResponsesMap({newDOHResponseMapEntry('^/coffee$', 418, 'C0FFEE', {['FoO']='bar'})})
self.assertTrue((self._customResponseHeader2) in self._response_headers.decode())
self.assertFalse(('UPPERCASE: VaLuE' in self._response_headers.decode()))
self.assertTrue(('uppercase: VaLuE' in self._response_headers.decode()))
+ self.assertTrue(('cache-control: max-age=3600' in self._response_headers.decode()))
self.checkQueryEDNSWithoutECS(expectedQuery, receivedQuery)
self.assertEquals(response, receivedResponse)
+ self.checkHasHeader('cache-control', 'max-age=3600')
def testDOHSimplePOST(self):
"""
name = 'refused.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)
expectedResponse.set_rcode(dns.rcode.REFUSED)
self.checkQueryEDNSWithoutECS(expectedQuery, receivedQuery)
self.assertEquals(response, receivedResponse)
+ # this path is not in the URLs map and should lead to a 404
+ (_, receivedResponse) = self.sendDOHQuery(self._dohServerPort, self._serverName, self._dohBaseURL + "PowerDNS/something", query, caFile=self._caCert, useQueue=False, rawResponse=True)
+ self.assertTrue(receivedResponse)
+ self.assertEquals(receivedResponse, b'there is no endpoint configured for this path')
+ self.assertEquals(self._rcode, 404)
+
def testHTTPPathRegex(self):
"""
DOH: HTTPPathRegex
addDOHLocal("127.0.0.1:%s")
"""
_config_params = ['_testServerPort', '_dohServerPort']
+ _checkConfigExpectedOutput = b"""No certificate provided for DoH endpoint 127.0.0.1:8480, running in DNS over HTTP mode instead of DNS over HTTPS
+Configuration 'configs/dnsdist_TestDOHOverHTTP.conf' OK!
+"""
def testDOHSimple(self):
"""
self.checkQueryEDNSWithoutECS(expectedQuery, receivedQuery)
self.assertEquals(response, receivedResponse)
self.checkResponseNoEDNS(response, receivedResponse)
+
+class TestDOHWithCache(DNSDistDOHTest):
+
+ _serverKey = 'server.key'
+ _serverCert = 'server.chain'
+ _serverName = 'tls.tests.dnsdist.org'
+ _caCert = 'ca.pem'
+ _dohServerPort = 8443
+ _dohBaseURL = ("https://%s:%d/" % (_serverName, _dohServerPort))
+ _config_template = """
+ newServer{address="127.0.0.1:%s"}
+
+ addDOHLocal("127.0.0.1:%s", "%s", "%s")
+
+ pc = newPacketCache(100, {maxTTL=86400, minTTL=1})
+ getPool(""):setCache(pc)
+ """
+ _config_params = ['_testServerPort', '_dohServerPort', '_serverCert', '_serverKey']
+
+ def testDOHCacheLargeAnswer(self):
+ """
+ DOH with cache: Check that we can cache (and retrieve) large answers
+ """
+ numberOfQueries = 10
+ name = 'large.doh-with-cache.tests.powerdns.com.'
+ query = dns.message.make_query(name, 'A', 'IN', use_edns=False)
+ query.id = 0
+ expectedQuery = dns.message.make_query(name, 'A', 'IN', use_edns=True, payload=4096)
+ expectedQuery.id = 0
+ response = dns.message.make_response(query)
+ # we prepare a large answer
+ content = ""
+ for i in range(44):
+ if len(content) > 0:
+ content = content + ', '
+ content = content + (str(i)*50)
+ # pad up to 4096
+ content = content + 'A'*40
+
+ rrset = dns.rrset.from_text(name,
+ 3600,
+ dns.rdataclass.IN,
+ dns.rdatatype.TXT,
+ content)
+ response.answer.append(rrset)
+ self.assertEquals(len(response.to_wire()), 4096)
+
+ # first query to fill the cache
+ (receivedQuery, receivedResponse) = self.sendDOHQuery(self._dohServerPort, self._serverName, self._dohBaseURL, 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)
+ self.checkHasHeader('cache-control', 'max-age=3600')
+
+ for _ in range(numberOfQueries):
+ (_, receivedResponse) = self.sendDOHQuery(self._dohServerPort, self._serverName, self._dohBaseURL, query, caFile=self._caCert, useQueue=False)
+ self.assertEquals(receivedResponse, response)
+ self.checkHasHeader('cache-control', 'max-age=' + str(receivedResponse.answer[0].ttl))
+
+ time.sleep(1)
+
+ (_, receivedResponse) = self.sendDOHQuery(self._dohServerPort, self._serverName, self._dohBaseURL, query, caFile=self._caCert, useQueue=False)
+ self.assertEquals(receivedResponse, response)
+ self.checkHasHeader('cache-control', 'max-age=' + str(receivedResponse.answer[0].ttl))
+
+class TestDOHWithoutCacheControl(DNSDistDOHTest):
+
+ _serverKey = 'server.key'
+ _serverCert = 'server.chain'
+ _serverName = 'tls.tests.dnsdist.org'
+ _caCert = 'ca.pem'
+ _dohServerPort = 8443
+ _dohBaseURL = ("https://%s:%d/" % (_serverName, _dohServerPort))
+ _config_template = """
+ newServer{address="127.0.0.1:%s"}
+
+ addDOHLocal("127.0.0.1:%s", "%s", "%s", { "/" }, {sendCacheControlHeaders=false})
+ """
+ _config_params = ['_testServerPort', '_dohServerPort', '_serverCert', '_serverKey']
+
+ def testDOHSimple(self):
+ """
+ DOH without cache-control
+ """
+ name = 'simple.doh.tests.powerdns.com.'
+ query = dns.message.make_query(name, 'A', 'IN', use_edns=False)
+ query.id = 0
+ expectedQuery = dns.message.make_query(name, 'A', 'IN', use_edns=True, payload=4096)
+ 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)
+
+ (receivedQuery, receivedResponse) = self.sendDOHQuery(self._dohServerPort, self._serverName, self._dohBaseURL, query, response=response, caFile=self._caCert)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = expectedQuery.id
+ self.assertEquals(expectedQuery, receivedQuery)
+ self.checkNoHeader('cache-control')
+ self.checkQueryEDNSWithoutECS(expectedQuery, receivedQuery)
+ self.assertEquals(response, receivedResponse)
projects / thirdparty / pdns.git / blobdiff