ffi.cdef[[
typedef struct dnsdist_ffi_dnsquestion_t dnsdist_ffi_dnsquestion_t;
- typedef struct dnsdist_ednsoption {
- uint16_t optionCode;
- uint16_t len;
- const void* data;
- } dnsdist_ednsoption_t;
-
- typedef struct dnsdist_http_header {
- const char* name;
- const char* value;
- } dnsdist_http_header_t;
-
void dnsdist_ffi_dnsquestion_get_localaddr(const dnsdist_ffi_dnsquestion_t* dq, const void** addr, size_t* addrSize) __attribute__ ((visibility ("default")));
void dnsdist_ffi_dnsquestion_get_remoteaddr(const dnsdist_ffi_dnsquestion_t* dq, const void** addr, size_t* addrSize) __attribute__ ((visibility ("default")));
void dnsdist_ffi_dnsquestion_get_qname_raw(const dnsdist_ffi_dnsquestion_t* dq, const char** qname, size_t* qnameSize) __attribute__ ((visibility ("default")));
size_t dnsdist_ffi_dnsquestion_get_size(const dnsdist_ffi_dnsquestion_t* dq) __attribute__ ((visibility ("default")));
uint8_t dnsdist_ffi_dnsquestion_get_opcode(const dnsdist_ffi_dnsquestion_t* dq) __attribute__ ((visibility ("default")));
bool dnsdist_ffi_dnsquestion_get_tcp(const dnsdist_ffi_dnsquestion_t* dq) __attribute__ ((visibility ("default")));
- bool dnsdist_ffi_dnsquestion_get_skip_cache(const dnsdist_ffi_dnsquestion_t* dq) __attribute__ ((visibility ("default")));
- bool dnsdist_ffi_dnsquestion_get_use_ecs(const dnsdist_ffi_dnsquestion_t* dq) __attribute__ ((visibility ("default")));
- bool dnsdist_ffi_dnsquestion_get_ecs_override(const dnsdist_ffi_dnsquestion_t* dq) __attribute__ ((visibility ("default")));
- uint16_t dnsdist_ffi_dnsquestion_get_ecs_prefix_length(const dnsdist_ffi_dnsquestion_t* dq) __attribute__ ((visibility ("default")));
- bool dnsdist_ffi_dnsquestion_is_temp_failure_ttl_set(const dnsdist_ffi_dnsquestion_t* dq) __attribute__ ((visibility ("default")));
- uint32_t dnsdist_ffi_dnsquestion_get_temp_failure_ttl(const dnsdist_ffi_dnsquestion_t* dq) __attribute__ ((visibility ("default")));
bool dnsdist_ffi_dnsquestion_get_do(const dnsdist_ffi_dnsquestion_t* dq) __attribute__ ((visibility ("default")));
- void dnsdist_ffi_dnsquestion_get_sni(const dnsdist_ffi_dnsquestion_t* dq, const char** sni, size_t* sniSize) __attribute__ ((visibility ("default")));
const char* dnsdist_ffi_dnsquestion_get_tag(const dnsdist_ffi_dnsquestion_t* dq, const char* label) __attribute__ ((visibility ("default")));
- const char* dnsdist_ffi_dnsquestion_get_http_path(const dnsdist_ffi_dnsquestion_t* dq) __attribute__ ((visibility ("default")));
- const char* dnsdist_ffi_dnsquestion_get_http_query_string(const dnsdist_ffi_dnsquestion_t* dq) __attribute__ ((visibility ("default")));
- const char* dnsdist_ffi_dnsquestion_get_http_host(const dnsdist_ffi_dnsquestion_t* dq) __attribute__ ((visibility ("default")));
- const char* dnsdist_ffi_dnsquestion_get_http_scheme(const dnsdist_ffi_dnsquestion_t* dq) __attribute__ ((visibility ("default")));
-
- // returns the length of the resulting 'out' array. 'out' is not set if the length is 0
- size_t dnsdist_ffi_dnsquestion_get_edns_options(dnsdist_ffi_dnsquestion_t* ref, const dnsdist_ednsoption_t** out) __attribute__ ((visibility ("default")));
- size_t dnsdist_ffi_dnsquestion_get_http_headers(dnsdist_ffi_dnsquestion_t* ref, const dnsdist_http_header_t** out) __attribute__ ((visibility ("default")));
void dnsdist_ffi_dnsquestion_set_result(dnsdist_ffi_dnsquestion_t* dq, const char* str, size_t strSize) __attribute__ ((visibility ("default")));
void dnsdist_ffi_dnsquestion_set_rcode(dnsdist_ffi_dnsquestion_t* dq, int rcode) __attribute__ ((visibility ("default")));
- void dnsdist_ffi_dnsquestion_set_len(dnsdist_ffi_dnsquestion_t* dq, int len) __attribute__ ((visibility ("default")));
- void dnsdist_ffi_dnsquestion_set_skip_cache(dnsdist_ffi_dnsquestion_t* dq, bool skipCache) __attribute__ ((visibility ("default")));
- void dnsdist_ffi_dnsquestion_set_use_ecs(dnsdist_ffi_dnsquestion_t* dq, bool useECS) __attribute__ ((visibility ("default")));
- void dnsdist_ffi_dnsquestion_set_ecs_override(dnsdist_ffi_dnsquestion_t* dq, bool ecsOverride) __attribute__ ((visibility ("default")));
- void dnsdist_ffi_dnsquestion_set_ecs_prefix_length(dnsdist_ffi_dnsquestion_t* dq, uint16_t ecsPrefixLength) __attribute__ ((visibility ("default")));
- void dnsdist_ffi_dnsquestion_set_temp_failure_ttl(dnsdist_ffi_dnsquestion_t* dq, uint32_t tempFailureTTL) __attribute__ ((visibility ("default")));
void dnsdist_ffi_dnsquestion_set_tag(dnsdist_ffi_dnsquestion_t* dq, const char* label, const char* value) __attribute__ ((visibility ("default")));
- void dnsdist_ffi_dnsquestion_set_http_response(dnsdist_ffi_dnsquestion_t* dq, uint16_t statusCode, const char* body, const char* contentType) __attribute__ ((visibility ("default")));
]]
local expectingUDP = true
return false
end
+ local tag = ffi.C.dnsdist_ffi_dnsquestion_get_tag(dq, 'a-tag')
+ if ffi.string(tag) ~= 'a-value' then
+ print('invalid tag value')
+ print(ffi.string(tag))
+ return false
+ end
return true
end
end
end
+ function luaffiactionsettag(dq)
+ ffi.C.dnsdist_ffi_dnsquestion_set_tag(dq, 'a-tag', 'a-value')
+ return DNSAction.None
+ end
+
+ addAction(AllRule(), LuaFFIAction(luaffiactionsettag))
addAction(LuaFFIRule(luaffirulefunction), LuaFFIAction(luaffiactionfunction))
-- newServer{address="127.0.0.1:%s"}
"""
import pycurl
from io import BytesIO
-#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):
print("Launching tests..")
-# @classmethod
-# def openDOHConnection(cls, port, caFile, timeout=2.0):
-# sslctx = SSLContext(PROTOCOL_TLSv1_2)
-# sslctx.load_verify_locations(caFile)
-# return HTTP20Connection('127.0.0.1', port=port, secure=True, timeout=timeout, ssl_context=sslctx, force_proto='h2')
-
-# @classmethod
-# def sendDOHQueryOverConnection(cls, conn, baseurl, query, response=None, timeout=2.0):
-# url = cls.getDOHGetURL(baseurl, query)
-
-# if response:
-# cls._toResponderQueue.put(response, True, timeout)
-
-# conn.request('GET', url)
-
-# @classmethod
-# def recvDOHResponseOverConnection(cls, conn, useQueue=False, timeout=2.0):
-# message = None
-# data = conn.get_response()
-# if data:
-# data = data.read()
-# if data:
-# message = dns.message.from_wire(data)
-
-# if useQueue and not cls._fromResponderQueue.empty():
-# receivedQuery = cls._fromResponderQueue.get(True, timeout)
-# return (receivedQuery, message)
-# else:
-# return message
-
class TestDOH(DNSDistDOHTest):
_serverKey = 'server.key'
self.checkNoHeader('cache-control')
self.checkQueryEDNSWithoutECS(expectedQuery, receivedQuery)
self.assertEquals(response, receivedResponse)
+
+class TestDOHFFI(DNSDistDOHTest):
+
+ _serverKey = 'server.key'
+ _serverCert = 'server.chain'
+ _serverName = 'tls.tests.dnsdist.org'
+ _caCert = 'ca.pem'
+ _dohServerPort = 8443
+ _customResponseHeader1 = 'access-control-allow-origin: *'
+ _customResponseHeader2 = 'user-agent: derp'
+ _dohBaseURL = ("https://%s:%d/" % (_serverName, _dohServerPort))
+ _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"}})
+
+ local ffi = require("ffi")
+
+ ffi.cdef[[
+ typedef struct dnsdist_ffi_dnsquestion_t dnsdist_ffi_dnsquestion_t;
+
+ typedef struct dnsdist_http_header {
+ const char* name;
+ const char* value;
+ } dnsdist_http_header_t;
+
+ void dnsdist_ffi_dnsquestion_get_sni(const dnsdist_ffi_dnsquestion_t* dq, const char** sni, size_t* sniSize) __attribute__ ((visibility ("default")));
+
+ const char* dnsdist_ffi_dnsquestion_get_http_path(dnsdist_ffi_dnsquestion_t* dq) __attribute__ ((visibility ("default")));
+ const char* dnsdist_ffi_dnsquestion_get_http_query_string(dnsdist_ffi_dnsquestion_t* dq) __attribute__ ((visibility ("default")));
+ const char* dnsdist_ffi_dnsquestion_get_http_host(dnsdist_ffi_dnsquestion_t* dq) __attribute__ ((visibility ("default")));
+ const char* dnsdist_ffi_dnsquestion_get_http_scheme(dnsdist_ffi_dnsquestion_t* dq) __attribute__ ((visibility ("default")));
+
+ size_t dnsdist_ffi_dnsquestion_get_http_headers(dnsdist_ffi_dnsquestion_t* ref, const dnsdist_http_header_t** out) __attribute__ ((visibility ("default")));
+
+ void dnsdist_ffi_dnsquestion_set_http_response(dnsdist_ffi_dnsquestion_t* dq, uint16_t statusCode, const char* body, const char* contentType) __attribute__ ((visibility ("default")));
+ ]]
+
+ function dohHandler(dq)
+ local scheme = ffi.string(ffi.C.dnsdist_ffi_dnsquestion_get_http_scheme(dq))
+ local host = ffi.string(ffi.C.dnsdist_ffi_dnsquestion_get_http_host(dq))
+ local path = ffi.string(ffi.C.dnsdist_ffi_dnsquestion_get_http_path(dq))
+ local query_string = ffi.string(ffi.C.dnsdist_ffi_dnsquestion_get_http_query_string(dq))
+ if scheme == 'https' and host == '%s:%d' and path == '/' and query_string == '' then
+ local foundct = false
+ local headers_ptr = ffi.new("const dnsdist_http_header_t *[1]")
+ local headers_ptr_param = ffi.cast("const dnsdist_http_header_t **", headers_ptr)
+
+ local headers_count = tonumber(ffi.C.dnsdist_ffi_dnsquestion_get_http_headers(dq, headers_ptr_param))
+ if headers_count > 0 then
+ for idx = 0, headers_count-1 do
+ if ffi.string(headers_ptr[0][idx].name) == 'content-type' and ffi.string(headers_ptr[0][idx].value) == 'application/dns-message' then
+ foundct = true
+ break
+ end
+ end
+ end
+ if foundct then
+ ffi.C.dnsdist_ffi_dnsquestion_set_http_response(dq, 200, 'It works!', 'text/plain')
+ return DNSAction.HeaderModify
+ end
+ end
+ return DNSAction.None
+ end
+ addAction("http-lua-ffi.doh.tests.powerdns.com.", LuaFFIAction(dohHandler))
+ """
+ _config_params = ['_testServerPort', '_dohServerPort', '_serverCert', '_serverKey', '_serverName', '_dohServerPort']
+
+ def testHTTPLuaFFIResponse(self):
+ """
+ DOH: Lua FFI HTTP Response
+ """
+ name = 'http-lua-ffi.doh.tests.powerdns.com.'
+ query = dns.message.make_query(name, 'A', 'IN', use_edns=False)
+ query.id = 0
+
+ (_, receivedResponse) = self.sendDOHPostQuery(self._dohServerPort, self._serverName, self._dohBaseURL, query, caFile=self._caCert, useQueue=False, rawResponse=True)
+ self.assertTrue(receivedResponse)
+ self.assertEquals(receivedResponse, b'It works!')
+ self.assertEquals(self._rcode, 200)
+ self.assertTrue('content-type: text/plain' in self._response_headers.decode())
+
receivedQuery.id = query.id
self.assertEquals(receivedQuery, query)
self.assertEquals(receivedResponse, response)
+
+class TestEDNSOptionsLuaFFI(DNSDistTest):
+
+ _config_template = """
+ local ffi = require("ffi")
+
+ ffi.cdef[[
+ typedef struct dnsdist_ffi_dnsquestion_t dnsdist_ffi_dnsquestion_t;
+
+ typedef struct dnsdist_ednsoption {
+ uint16_t optionCode;
+ uint16_t len;
+ const void* data;
+ } dnsdist_ednsoption_t;
+
+ void dnsdist_ffi_dnsquestion_get_qname_raw(const dnsdist_ffi_dnsquestion_t* dq, const char** qname, size_t* qnameSize) __attribute__ ((visibility ("default")));
+ // returns the length of the resulting 'out' array. 'out' is not set if the length is 0
+ size_t dnsdist_ffi_dnsquestion_get_edns_options(dnsdist_ffi_dnsquestion_t* ref, const dnsdist_ednsoption_t** out) __attribute__ ((visibility ("default")));
+ ]]
+
+ function testEDNSOptions(dq)
+ local options_ptr = ffi.new("const dnsdist_ednsoption_t *[1]")
+ local ret_ptr_param = ffi.cast("const dnsdist_ednsoption_t **", options_ptr)
+
+ local options_count = tonumber(ffi.C.dnsdist_ffi_dnsquestion_get_edns_options(dq, ret_ptr_param))
+
+ local qname_ptr = ffi.new("const char *[1]")
+ local qname_ptr_param = ffi.cast("const char **", qname_ptr)
+ local qname_size = ffi.new("size_t[1]")
+ local qname_size_param = ffi.cast("size_t*", qname_size)
+ ffi.C.dnsdist_ffi_dnsquestion_get_qname_raw(dq, qname_ptr_param, qname_size_param)
+ local qname = ffi.string(qname_ptr[0])
+
+ if string.match(qname, 'noedns') then
+ if options_count ~= 0 then
+ local str = "192.0.2.255"
+ local buf = ffi.new("char[?]", #str + 1)
+ ffi.copy(buf, str)
+ ffi.C.dnsdist_ffi_dnsquestion_set_result(dq, buf, #str)
+ return DNSAction.Spoof
+ end
+ end
+
+ local cookies_count = 0
+ local ecs_count = 0
+ local ecs_index = -1
+ local first_cookie_index = -1
+ local last_cookie_index = -1
+ if options_count > 0 then
+ for idx = 0, options_count-1 do
+ if options_ptr[0][idx].optionCode == EDNSOptionCode.COOKIE then
+ cookies_count = cookies_count + 1
+ if first_cookie_index == -1 then
+ first_cookie_index = idx
+ end
+ last_cookie_index = idx
+ elseif options_ptr[0][idx].optionCode == EDNSOptionCode.ECS then
+ ecs_count = ecs_count + 1
+ ecs_index = idx
+ end
+ end
+ end
+
+ if string.match(qname, 'multiplecookies') then
+ if cookies_count == 0 then
+ local str = "192.0.2.1"
+ local buf = ffi.new("char[?]", #str + 1)
+ ffi.copy(buf, str)
+ ffi.C.dnsdist_ffi_dnsquestion_set_result(dq, buf, #str)
+ return DNSAction.Spoof
+ end
+ if cookies_count ~= 2 then
+ local str = "192.0.2.2"
+ local buf = ffi.new("char[?]", #str + 1)
+ ffi.copy(buf, str)
+ ffi.C.dnsdist_ffi_dnsquestion_set_result(dq, buf, #str)
+ return DNSAction.Spoof
+ end
+ if options_ptr[0][first_cookie_index].len ~= 16 then
+ local str = "192.0.2.3"
+ local buf = ffi.new("char[?]", #str + 1)
+ ffi.copy(buf, str)
+ ffi.C.dnsdist_ffi_dnsquestion_set_result(dq, buf, #str)
+ return DNSAction.Spoof
+ end
+ if options_ptr[0][last_cookie_index].len ~= 16 then
+ local str = "192.0.2.4"
+ local buf = ffi.new("char[?]", #str + 1)
+ ffi.copy(buf, str)
+ ffi.C.dnsdist_ffi_dnsquestion_set_result(dq, buf, #str)
+ return DNSAction.Spoof
+ end
+ elseif string.match(qname, 'cookie') then
+
+ if cookies_count == 0 then
+ local str = "192.0.2.1"
+ local buf = ffi.new("char[?]", #str + 1)
+ ffi.copy(buf, str)
+ ffi.C.dnsdist_ffi_dnsquestion_set_result(dq, buf, #str)
+ return DNSAction.Spoof
+ end
+ if cookies_count ~= 1 or options_ptr[0][first_cookie_index].len ~= 16 then
+ local str = "192.0.2.2"
+ local buf = ffi.new("char[?]", #str + 1)
+ ffi.copy(buf, str)
+ ffi.C.dnsdist_ffi_dnsquestion_set_result(dq, buf, #str)
+ return DNSAction.Spoof
+ end
+ end
+
+ if string.match(qname, 'ecs4') then
+ if ecs_count == 0 then
+ local str = "192.0.2.51"
+ local buf = ffi.new("char[?]", #str + 1)
+ ffi.copy(buf, str)
+ ffi.C.dnsdist_ffi_dnsquestion_set_result(dq, buf, #str)
+ return DNSAction.Spoof
+ end
+
+ if ecs_count ~= 1 or options_ptr[0][ecs_index].len ~= 8 then
+ local str = "192.0.2.52"
+ local buf = ffi.new("char[?]", #str + 1)
+ ffi.copy(buf, str)
+ ffi.C.dnsdist_ffi_dnsquestion_set_result(dq, buf, #str)
+ return DNSAction.Spoof
+ end
+ end
+
+ if string.match(qname, 'ecs6') then
+ if ecs_count == 0 then
+ local str = "192.0.2.101"
+ local buf = ffi.new("char[?]", #str + 1)
+ ffi.copy(buf, str)
+ ffi.C.dnsdist_ffi_dnsquestion_set_result(dq, buf, #str)
+ return DNSAction.Spoof
+ end
+ if ecs_count ~= 1 or options_ptr[0][ecs_index].len ~= 20 then
+ local str = "192.0.2.102"
+ local buf = ffi.new("char[?]", #str + 1)
+ ffi.copy(buf, str)
+ ffi.C.dnsdist_ffi_dnsquestion_set_result(dq, buf, #str)
+ return DNSAction.Spoof
+ end
+ end
+
+ return DNSAction.None
+
+ end
+
+ addAction(AllRule(), LuaFFIAction(testEDNSOptions))
+
+ newServer{address="127.0.0.1:%s"}
+ """
+
+ def testWithoutEDNSFFI(self):
+ """
+ EDNS Options: No EDNS (FFI)
+ """
+ name = 'noedns.ednsoptions.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,
+ '192.0.2.255')
+ response.answer.append(rrset)
+
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = query.id
+ self.assertEquals(receivedQuery, query)
+ self.assertEquals(receivedResponse, response)
+
+ def testCookieFFI(self):
+ """
+ EDNS Options: Cookie (FFI)
+ """
+ name = 'cookie.ednsoptions.tests.powerdns.com.'
+ eco = cookiesoption.CookiesOption(b'deadbeef', b'deadbeef')
+ query = dns.message.make_query(name, 'A', 'IN', use_edns=True, payload=4096, options=[eco])
+ 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)
+
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = query.id
+ self.assertEquals(receivedQuery, query)
+ self.assertEquals(receivedResponse, response)
+
+ def testECS4FFI(self):
+ """
+ EDNS Options: ECS4 (FFI)
+ """
+ name = 'ecs4.ednsoptions.tests.powerdns.com.'
+ ecso = clientsubnetoption.ClientSubnetOption('1.2.3.4', 32)
+ query = dns.message.make_query(name, 'A', 'IN', use_edns=True, payload=4096, options=[ecso])
+ 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)
+
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = query.id
+ self.assertEquals(receivedQuery, query)
+ self.assertEquals(receivedResponse, response)
+
+ def testECS6FFI(self):
+ """
+ EDNS Options: ECS6 (FFI)
+ """
+ name = 'ecs6.ednsoptions.tests.powerdns.com.'
+ ecso = clientsubnetoption.ClientSubnetOption('2001:DB8::1', 128)
+ query = dns.message.make_query(name, 'A', 'IN', use_edns=True, payload=4096, options=[ecso])
+ 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)
+
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = query.id
+ self.assertEquals(receivedQuery, query)
+ self.assertEquals(receivedResponse, response)
+
+ def testECS6CookieFFI(self):
+ """
+ EDNS Options: Cookie + ECS6 (FFI)
+ """
+ name = 'cookie-ecs6.ednsoptions.tests.powerdns.com.'
+ eco = cookiesoption.CookiesOption(b'deadbeef', b'deadbeef')
+ ecso = clientsubnetoption.ClientSubnetOption('2001:DB8::1', 128)
+ query = dns.message.make_query(name, 'A', 'IN', use_edns=True, payload=4096, options=[ecso,eco])
+ 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)
+
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = query.id
+ self.assertEquals(receivedQuery, query)
+ self.assertEquals(receivedResponse, response)
+
+ def testMultiCookiesECS6FFI(self):
+ """
+ EDNS Options: Two Cookies + ECS6 (FFI)
+ """
+ name = 'multiplecookies-ecs6.ednsoptions.tests.powerdns.com.'
+ eco1 = cookiesoption.CookiesOption(b'deadbeef', b'deadbeef')
+ ecso = clientsubnetoption.ClientSubnetOption('2001:DB8::1', 128)
+ eco2 = cookiesoption.CookiesOption(b'deadc0de', b'deadc0de')
+ query = dns.message.make_query(name, 'A', 'IN', use_edns=True, payload=4096, options=[eco1, ecso, eco2])
+ 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)
+
+ for method in ("sendUDPQuery", "sendTCPQuery"):
+ sender = getattr(self, method)
+ (receivedQuery, receivedResponse) = sender(query, response)
+ self.assertTrue(receivedQuery)
+ self.assertTrue(receivedResponse)
+ receivedQuery.id = query.id
+ self.assertEquals(receivedQuery, query)
+ self.assertEquals(receivedResponse, response)
projects / thirdparty / pdns.git / commitdiff
