_dohServerPort = 8480
_serverName = 'tls.tests.dnsdist.org'
- _dohBaseURL = ("http://%s:%d/" % (_serverName, _dohServerPort))
+ _dohBaseURL = ("http://%s:%d/dns-query" % (_serverName, _dohServerPort))
_config_template = """
newServer{address="127.0.0.1:%s"}
addDOHLocal("127.0.0.1:%s")
_serverName = 'tls.tests.dnsdist.org'
_caCert = 'ca.pem'
_dohServerPort = 8443
- _dohBaseURL = ("https://%s:%d/" % (_serverName, _dohServerPort))
+ _dohBaseURL = ("https://%s:%d/dns-query" % (_serverName, _dohServerPort))
_config_template = """
newServer{address="127.0.0.1:%s"}
self.assertEquals(self._rcode, 200)
self.assertTrue('content-type: text/plain' in self._response_headers.decode())
+class TestDOHForwardedFor(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"}
+
+ setACL('192.0.2.1/32')
+ addDOHLocal("127.0.0.1:%s", "%s", "%s", { "/" }, {trustForwardedForHeader=true})
+ """
+ _config_params = ['_testServerPort', '_dohServerPort', '_serverCert', '_serverKey']
+
+ def testDOHAllowedForwarded(self):
+ """
+ DOH with X-Forwarded-For allowed
+ """
+ name = 'allowed.forwarded.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, customHeaders=['x-forwarded-for: 127.0.0.1:42, 127.0.0.1, 192.0.2.1:4200'])
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = expectedQuery.id
+ self.assertEquals(expectedQuery, receivedQuery)
+ self.checkQueryEDNSWithoutECS(expectedQuery, receivedQuery)
+ self.assertEquals(response, receivedResponse)
+
+ def testDOHDeniedForwarded(self):
+ """
+ DOH with X-Forwarded-For not allowed
+ """
+ name = 'not-allowed.forwarded.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, useQueue=False, rawResponse=True, customHeaders=['x-forwarded-for: 127.0.0.1:42, 127.0.0.1'])
+
+ self.assertEquals(self._rcode, 403)
+ self.assertEquals(receivedResponse, b'dns query not allowed because of ACL')
+
+class TestDOHForwardedForNoTrusted(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"}
+
+ setACL('192.0.2.1/32')
+ addDOHLocal("127.0.0.1:%s", "%s", "%s", { "/" })
+ """
+ _config_params = ['_testServerPort', '_dohServerPort', '_serverCert', '_serverKey']
+
+ def testDOHForwardedUntrusted(self):
+ """
+ DOH with X-Forwarded-For not trusted
+ """
+ name = 'not-trusted.forwarded.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, useQueue=False, rawResponse=True, customHeaders=['x-forwarded-for: 192.0.2.1:4200'])
+
+ self.assertEquals(self._rcode, 403)
+ self.assertEquals(receivedResponse, b'dns query not allowed because of ACL')