+class TestRPCRoundtrip(DNSTest):
+ def get_credentials(self, lp):
+ creds = credentials.Credentials()
+ creds.guess(lp)
+ creds.set_machine_account(lp)
+ creds.set_krb_forwardable(credentials.NO_KRB_FORWARDABLE)
+ return creds
+
+ def setUp(self):
+ super(TestRPCRoundtrip, self).setUp()
+ self.lp = self.get_loadparm()
+ self.creds = self.get_credentials(self.lp)
+ self.server = os.getenv("SERVER_IP")
+ self.rpc_conn = dnsserver.dnsserver("ncacn_ip_tcp:%s[sign]" % (self.server),
+ self.lp, self.creds)
+
+ def tearDown(self):
+ super(TestRPCRoundtrip, self).tearDown()
+
+ def test_update_add_null_padded_txt_record(self):
+ "test adding records works"
+ prefix, txt = 'pad1textrec', ['"This is a test"', '', '']
+ p = self.make_txt_update(prefix, txt)
+ response = self.dns_transaction_udp(p)
+ self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
+ self.check_query_txt(prefix, txt)
+ self.assertIsNotNone(dns_record_match(self.rpc_conn, self.server,
+ self.get_dns_domain(),
+ "%s.%s" % (prefix, self.get_dns_domain()),
+ dnsp.DNS_TYPE_TXT, '"\\"This is a test\\"" "" ""'))
+
+ prefix, txt = 'pad2textrec', ['"This is a test"', '', '', 'more text']
+ p = self.make_txt_update(prefix, txt)
+ response = self.dns_transaction_udp(p)
+ self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
+ self.check_query_txt(prefix, txt)
+ self.assertIsNotNone(dns_record_match(self.rpc_conn, self.server,
+ self.get_dns_domain(),
+ "%s.%s" % (prefix, self.get_dns_domain()),
+ dnsp.DNS_TYPE_TXT, '"\\"This is a test\\"" "" "" "more text"'))
+
+ prefix, txt = 'pad3textrec', ['', '', '"This is a test"']
+ p = self.make_txt_update(prefix, txt)
+ response = self.dns_transaction_udp(p)
+ self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
+ self.check_query_txt(prefix, txt)
+ self.assertIsNotNone(dns_record_match(self.rpc_conn, self.server,
+ self.get_dns_domain(),
+ "%s.%s" % (prefix, self.get_dns_domain()),
+ dnsp.DNS_TYPE_TXT, '"" "" "\\"This is a test\\""'))
+
+ # Test is incomplete due to strlen against txt records
+ def test_update_add_null_char_txt_record(self):
+ "test adding records works"
+ prefix, txt = 'nulltextrec', ['NULL\x00BYTE']
+ p = self.make_txt_update(prefix, txt)
+ response = self.dns_transaction_udp(p)
+ self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
+ self.check_query_txt(prefix, ['NULL'])
+ self.assertIsNotNone(dns_record_match(self.rpc_conn, self.server,
+ self.get_dns_domain(),
+ "%s.%s" % (prefix, self.get_dns_domain()),
+ dnsp.DNS_TYPE_TXT, '"NULL"'))
+
+ prefix, txt = 'nulltextrec2', ['NULL\x00BYTE', 'NULL\x00BYTE']
+ p = self.make_txt_update(prefix, txt)
+ response = self.dns_transaction_udp(p)
+ self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
+ self.check_query_txt(prefix, ['NULL', 'NULL'])
+ self.assertIsNotNone(dns_record_match(self.rpc_conn, self.server,
+ self.get_dns_domain(),
+ "%s.%s" % (prefix, self.get_dns_domain()),
+ dnsp.DNS_TYPE_TXT, '"NULL" "NULL"'))
+
+ def test_update_add_hex_char_txt_record(self):
+ "test adding records works"
+ prefix, txt = 'hextextrec', ['HIGH\xFFBYTE']
+ p = self.make_txt_update(prefix, txt)
+ response = self.dns_transaction_udp(p)
+ self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
+ self.check_query_txt(prefix, txt)
+ self.assertIsNotNone(dns_record_match(self.rpc_conn, self.server,
+ self.get_dns_domain(),
+ "%s.%s" % (prefix, self.get_dns_domain()),
+ dnsp.DNS_TYPE_TXT, '"HIGH\xFFBYTE"'))
+
+ def test_update_add_slash_txt_record(self):
+ "test adding records works"
+ prefix, txt = 'slashtextrec', ['Th\\=is=is a test']
+ p = self.make_txt_update(prefix, txt)
+ response = self.dns_transaction_udp(p)
+ self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
+ self.check_query_txt(prefix, txt)
+ self.assertIsNotNone(dns_record_match(self.rpc_conn, self.server,
+ self.get_dns_domain(),
+ "%s.%s" % (prefix, self.get_dns_domain()),
+ dnsp.DNS_TYPE_TXT, '"Th\\\\=is=is a test"'))
+
+ def test_update_add_two_txt_records(self):
+ "test adding two txt records works"
+ prefix, txt = 'textrec2', ['"This is a test"',
+ '"and this is a test, too"']
+ p = self.make_txt_update(prefix, txt)
+ response = self.dns_transaction_udp(p)
+ self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
+ self.check_query_txt(prefix, txt)
+ self.assertIsNotNone(dns_record_match(self.rpc_conn, self.server,
+ self.get_dns_domain(),
+ "%s.%s" % (prefix, self.get_dns_domain()),
+ dnsp.DNS_TYPE_TXT, '"\\"This is a test\\""' +
+ ' "\\"and this is a test, too\\""'))
+
+ def test_update_add_empty_txt_records(self):
+ "test adding two txt records works"
+ prefix, txt = 'emptytextrec', []
+ p = self.make_txt_update(prefix, txt)
+ response = self.dns_transaction_udp(p)
+ self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
+ self.check_query_txt(prefix, txt)
+ self.assertIsNotNone(dns_record_match(self.rpc_conn, self.server,
+ self.get_dns_domain(),
+ "%s.%s" % (prefix, self.get_dns_domain()),
+ dnsp.DNS_TYPE_TXT, ''))
+