PEP8: fix E127: continuation line over-indented for visual indent
[samba.git] / python / samba / tests / dcerpc / dnsserver.py
1 # Unix SMB/CIFS implementation.
2 # Copyright (C) Amitay Isaacs <amitay@gmail.com> 2011
3 #
4 # This program is free software; you can redistribute it and/or modify
5 # it under the terms of the GNU General Public License as published by
6 # the Free Software Foundation; either version 3 of the License, or
7 # (at your option) any later version.
8 #
9 # This program is distributed in the hope that it will be useful,
10 # but WITHOUT ANY WARRANTY; without even the implied warranty of
11 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
12 # GNU General Public License for more details.
13 #
14 # You should have received a copy of the GNU General Public License
15 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
16 #
17
18 from __future__ import print_function
19 """Tests for samba.dcerpc.dnsserver"""
20
21 import os
22 import ldb
23
24 from samba.auth import system_session
25 from samba.samdb import SamDB
26 from samba.ndr import ndr_unpack, ndr_pack
27 from samba.dcerpc import dnsp, dnsserver, security
28 from samba.tests import RpcInterfaceTestCase, env_get_var_value
29 from samba.netcmd.dns import ARecord, AAAARecord, PTRRecord, CNameRecord, NSRecord, MXRecord, SRVRecord, TXTRecord
30 from samba import sd_utils, descriptor
31
32 class DnsserverTests(RpcInterfaceTestCase):
33
34     @classmethod
35     def setUpClass(cls):
36         good_dns = ["SAMDOM.EXAMPLE.COM",
37                     "1.EXAMPLE.COM",
38                     "%sEXAMPLE.COM" % ("1."*100),
39                     "EXAMPLE",
40                     "\n.COM",
41                     "!@#$%^&*()_",
42                     "HIGH\xFFBYTE",
43                     "@.EXAMPLE.COM",
44                     "."]
45         bad_dns = ["...",
46                    ".EXAMPLE.COM",
47                    ".EXAMPLE.",
48                    "",
49                    "SAMDOM..EXAMPLE.COM"]
50
51         good_mx = ["SAMDOM.EXAMPLE.COM 65535"]
52         bad_mx = []
53
54         good_srv = ["SAMDOM.EXAMPLE.COM 65535 65535 65535"]
55         bad_srv = []
56
57         for bad_dn in bad_dns:
58             bad_mx.append("%s 1" % bad_dn)
59             bad_srv.append("%s 0 0 0" % bad_dn)
60         for good_dn in good_dns:
61             good_mx.append("%s 1" % good_dn)
62             good_srv.append("%s 0 0 0" % good_dn)
63
64         cls.good_records = {
65             "A": ["192.168.0.1",
66                   "255.255.255.255"],
67             "AAAA": ["1234:5678:9ABC:DEF0:0000:0000:0000:0000",
68                      "0000:0000:0000:0000:0000:0000:0000:0000",
69                      "1234:5678:9ABC:DEF0:1234:5678:9ABC:DEF0",
70                      "1234:1234:1234::",
71                      "1234:1234:1234:1234:1234::",
72                      "1234:5678:9ABC:DEF0::",
73                      "0000:0000::0000",
74                      "1234::5678:9ABC:0000:0000:0000:0000",
75                      "::1",
76                      "::",
77                      "1:1:1:1:1:1:1:1"],
78             "PTR": good_dns,
79             "CNAME": good_dns,
80             "NS": good_dns,
81             "MX": good_mx,
82             "SRV": good_srv,
83             "TXT": ["text", "", "@#!", "\n"]
84         }
85
86         cls.bad_records = {
87             "A": ["192.168.0.500",
88                   "255.255.255.255/32"],
89             "AAAA": ["GGGG:1234:5678:9ABC:0000:0000:0000:0000",
90                      "0000:0000:0000:0000:0000:0000:0000:0000/1",
91                      "AAAA:AAAA:AAAA:AAAA:G000:0000:0000:1234",
92                      "1234:5678:9ABC:DEF0:1234:5678:9ABC:DEF0:1234",
93                      "1234:5678:9ABC:DEF0:1234:5678:9ABC",
94                      "1111::1111::1111"],
95             "PTR": bad_dns,
96             "CNAME": bad_dns,
97             "NS": bad_dns,
98             "MX": bad_mx,
99             "SRV": bad_srv
100         }
101
102         # Because we use uint16_t for these numbers, we can't
103         # actually create these records.
104         invalid_mx = ["SAMDOM.EXAMPLE.COM -1",
105                       "SAMDOM.EXAMPLE.COM 65536",
106                       "%s 1" % "A"*256]
107         invalid_srv = ["SAMDOM.EXAMPLE.COM 0 65536 0",
108                        "SAMDOM.EXAMPLE.COM 0 0 65536",
109                        "SAMDOM.EXAMPLE.COM 65536 0 0"]
110         cls.invalid_records = {
111             "MX": invalid_mx,
112             "SRV": invalid_srv
113         }
114
115     def setUp(self):
116         super(DnsserverTests, self).setUp()
117         self.server = os.environ["DC_SERVER"]
118         self.zone = env_get_var_value("REALM").lower()
119         self.conn = dnsserver.dnsserver("ncacn_ip_tcp:%s[sign]" % (self.server),
120                                         self.get_loadparm(),
121                                         self.get_credentials())
122
123         self.samdb = SamDB(url="ldap://%s" % os.environ["DC_SERVER_IP"],
124                            lp = self.get_loadparm(),
125                            session_info=system_session(),
126                            credentials=self.get_credentials())
127
128
129         self.custom_zone = "zone"
130         zone_create_info = dnsserver.DNS_RPC_ZONE_CREATE_INFO_LONGHORN()
131         zone_create_info.pszZoneName = self.custom_zone
132         zone_create_info.dwZoneType = dnsp.DNS_ZONE_TYPE_PRIMARY
133         zone_create_info.fAging = 0
134         zone_create_info.fDsIntegrated = 1
135         zone_create_info.fLoadExisting = 1
136         zone_create_info.dwDpFlags = dnsserver.DNS_DP_DOMAIN_DEFAULT
137
138         self.conn.DnssrvOperation2(dnsserver.DNS_CLIENT_VERSION_LONGHORN,
139                                    0,
140                                    self.server,
141                                    None,
142                                    0,
143                                    'ZoneCreate',
144                                    dnsserver.DNSSRV_TYPEID_ZONE_CREATE,
145                                    zone_create_info)
146
147     def tearDown(self):
148         self.conn.DnssrvOperation2(dnsserver.DNS_CLIENT_VERSION_LONGHORN,
149                                    0,
150                                    self.server,
151                                    self.custom_zone,
152                                    0,
153                                    'DeleteZoneFromDs',
154                                    dnsserver.DNSSRV_TYPEID_NULL,
155                                    None)
156         super(DnsserverTests, self).tearDown()
157
158     # This test fails against Samba (but passes against Windows),
159     # because Samba does not return the record when we enum records.
160     # Records can be given DNS_RANK_NONE when the zone they are in
161     # does not have DNS_ZONE_TYPE_PRIMARY. Since such records can be
162     # deleted, however, we do not consider this urgent to fix and
163     # so this test is a knownfail.
164     def test_rank_none(self):
165         """
166         See what happens when we set a record's rank to
167         DNS_RANK_NONE.
168         """
169
170         record_str = "192.168.50.50"
171         record_type_str = "A"
172         self.add_record(self.custom_zone, "testrecord", record_type_str, record_str)
173
174         dn, record = self.get_record_from_db(self.custom_zone, "testrecord")
175         record.rank = 0 # DNS_RANK_NONE
176         res = self.samdb.dns_replace_by_dn(dn, [record])
177         if res is not None:
178             self.fail("Unable to update dns record to have DNS_RANK_NONE.")
179
180         self.assert_num_records(self.custom_zone, "testrecord", record_type_str)
181         self.add_record(self.custom_zone, "testrecord", record_type_str, record_str, assertion=False)
182         self.delete_record(self.custom_zone, "testrecord", record_type_str, record_str)
183         self.assert_num_records(self.custom_zone, "testrecord", record_type_str, 0)
184
185     def test_dns_tombstoned(self):
186         """
187         See what happens when we set a record to be tombstoned.
188         """
189
190         record_str = "192.168.50.50"
191         record_type_str = "A"
192         self.add_record(self.custom_zone, "testrecord", record_type_str, record_str)
193
194         dn, record = self.get_record_from_db(self.custom_zone, "testrecord")
195         record.wType = dnsp.DNS_TYPE_TOMBSTONE
196         res = self.samdb.dns_replace_by_dn(dn, [record])
197         if res is not None:
198             self.fail("Unable to update dns record to be tombstoned.")
199
200         self.assert_num_records(self.custom_zone, "testrecord", record_type_str)
201         self.delete_record(self.custom_zone, "testrecord", record_type_str, record_str)
202         self.assert_num_records(self.custom_zone, "testrecord", record_type_str, 0)
203
204     def get_record_from_db(self, zone_name, record_name):
205         """
206         Returns (dn of record, record)
207         """
208
209         zones = self.samdb.search(base="DC=DomainDnsZones,%s" % self.samdb.get_default_basedn(), scope=ldb.SCOPE_SUBTREE,
210                                   expression="(objectClass=dnsZone)",
211                                   attrs=["cn"])
212
213         zone_dn = None
214         for zone in zones:
215             if "DC=%s," % zone_name in str(zone.dn):
216                 zone_dn = zone.dn
217                 break
218
219         if zone_dn is None:
220             raise AssertionError("Couldn't find zone '%s'." % zone_name)
221
222         records = self.samdb.search(base=zone_dn, scope=ldb.SCOPE_SUBTREE,
223                                     expression="(objectClass=dnsNode)",
224                                     attrs=["dnsRecord"])
225
226         for old_packed_record in records:
227             if record_name in str(old_packed_record.dn):
228                 rec = ndr_unpack(dnsp.DnssrvRpcRecord, old_packed_record["dnsRecord"][0])
229                 return (old_packed_record.dn, rec)
230
231     def test_duplicate_matching(self):
232         """
233         Make sure that records which should be distinct from each other or duplicate
234         to each other behave as expected.
235         """
236
237         distinct_dns = [("SAMDOM.EXAMPLE.COM",
238                          "SAMDOM.EXAMPLE.CO",
239                          "EXAMPLE.COM", "SAMDOM.EXAMPLE")]
240         duplicate_dns = [("SAMDOM.EXAMPLE.COM", "samdom.example.com", "SAMDOM.example.COM"),
241                          ("EXAMPLE.", "EXAMPLE")]
242
243         # Every tuple has entries which should be considered duplicate to one another.
244         duplicates = {
245             "AAAA": [("AAAA::", "aaaa::"),
246                      ("AAAA::", "AAAA:0000::"),
247                      ("AAAA::", "AAAA:0000:0000:0000:0000:0000:0000:0000"),
248                      ("AAAA::", "AAAA:0:0:0:0:0:0:0"),
249                      ("0123::", "123::"),
250                      ("::", "::0", "0000:0000:0000:0000:0000:0000:0000:0000")],
251         }
252
253         # Every tuple has entries which should be considered distinct from one another.
254         distinct = {
255             "A": [("192.168.1.0", "192.168.1.1", "192.168.2.0", "192.169.1.0", "193.168.1.0")],
256             "AAAA": [("AAAA::1234:5678:9ABC", "::AAAA:1234:5678:9ABC"),
257                      ("1000::", "::1000"),
258                      ("::1", "::11", "::1111"),
259                      ("1234::", "0234::")],
260             "SRV": [("SAMDOM.EXAMPLE.COM 1 1 1", "SAMDOM.EXAMPLE.COM 1 1 0", "SAMDOM.EXAMPLE.COM 1 0 1",
261                      "SAMDOM.EXAMPLE.COM 0 1 1", "SAMDOM.EXAMPLE.COM 2 1 0", "SAMDOM.EXAMPLE.COM 2 2 2")],
262             "MX": [("SAMDOM.EXAMPLE.COM 1", "SAMDOM.EXAMPLE.COM 0")],
263             "TXT": [("A RECORD", "B RECORD", "a record")]
264         }
265
266         for record_type_str in ("PTR", "CNAME", "NS"):
267             distinct[record_type_str] = distinct_dns
268             duplicates[record_type_str] = duplicate_dns
269
270         for record_type_str in duplicates:
271             for duplicate_tuple in duplicates[record_type_str]:
272                 # Attempt to add duplicates and make sure that all after the first fails
273                 self.add_record(self.custom_zone, "testrecord", record_type_str, duplicate_tuple[0])
274                 for record in duplicate_tuple:
275                     self.add_record(self.custom_zone, "testrecord", record_type_str, record, assertion=False)
276                     self.assert_num_records(self.custom_zone, "testrecord", record_type_str)
277                 self.delete_record(self.custom_zone, "testrecord", record_type_str, duplicate_tuple[0])
278
279                 # Repeatedly: add the first duplicate, and attempt to remove all of the others, making sure this succeeds
280                 for record in duplicate_tuple:
281                     self.add_record(self.custom_zone, "testrecord", record_type_str, duplicate_tuple[0])
282                     self.delete_record(self.custom_zone, "testrecord", record_type_str, record)
283
284         for record_type_str in distinct:
285             for distinct_tuple in distinct[record_type_str]:
286                 # Attempt to add distinct and make sure that they all succeed within a tuple
287                 i = 0
288                 for record in distinct_tuple:
289                     i = i + 1
290                     try:
291                         self.add_record(self.custom_zone, "testrecord", record_type_str, record)
292                         # All records should have been added.
293                         self.assert_num_records(self.custom_zone, "testrecord", record_type_str, expected_num=i)
294                     except AssertionError as e:
295                         raise AssertionError("Failed to add %s, which should be distinct from all others in the set. "
296                                              "Original error: %s\nDistinct set: %s." % (record, e, distinct_tuple))
297                 for record in distinct_tuple:
298                     self.delete_record(self.custom_zone, "testrecord", record_type_str, record)
299                     # CNAMEs should not have been added, since they conflict.
300                     if record_type_str == 'CNAME':
301                         continue
302
303                 # Add the first distinct and attempt to remove all of the others, making sure this fails
304                 # Windows fails this test. This is probably due to weird tombstoning behavior.
305                 self.add_record(self.custom_zone, "testrecord", record_type_str, distinct_tuple[0])
306                 for record in distinct_tuple:
307                     if record == distinct_tuple[0]:
308                         continue
309                     try:
310                         self.delete_record(self.custom_zone, "testrecord", record_type_str, record, assertion=False)
311                     except AssertionError as e:
312                         raise AssertionError("Managed to remove %s by attempting to remove %s. Original error: %s"
313                                              % (distinct_tuple[0], record, e))
314                 self.delete_record(self.custom_zone, "testrecord", record_type_str, distinct_tuple[0])
315
316     def test_accept_valid_commands(self):
317         """
318         Make sure that we can add, update and delete a variety
319         of valid records.
320         """
321         for record_type_str in self.good_records:
322             for record_str in self.good_records[record_type_str]:
323                 self.add_record(self.custom_zone, "testrecord", record_type_str, record_str)
324                 self.assert_num_records(self.custom_zone, "testrecord", record_type_str)
325                 self.delete_record(self.custom_zone, "testrecord", record_type_str, record_str)
326
327     def check_params(self, wDataLength, rank, flags, dwTtlSeconds, dwReserved, data,
328                      wType, dwTimeStamp=0, zone="zone", rec_name="testrecord"):
329         res = self.get_record_from_db(zone, rec_name)
330         self.assertIsNotNone(res, "Expected record %s but was not found over LDAP." % data)
331         (rec_dn, rec) = res
332         self.assertEqual(wDataLength, rec.wDataLength, "Unexpected data length for record %s. Got %s, expected %s." % (data, rec.wDataLength, wDataLength))
333         self.assertEqual(rank, rec.rank, "Unexpected rank for record %s. Got %s, expected %s." % (data, rec.rank, rank))
334         self.assertEqual(flags, rec.flags, "Unexpected flags for record %s. Got %s, expected %s." % (data, rec.flags, flags))
335         self.assertEqual(dwTtlSeconds, rec.dwTtlSeconds, "Unexpected time to live for record %s. Got %s, expected %s." % (data, rec.dwTtlSeconds, dwTtlSeconds))
336         self.assertEqual(dwReserved, rec.dwReserved, "Unexpected dwReserved for record %s. Got %s, expected %s." % (data, rec.dwReserved, dwReserved))
337         self.assertEqual(data.lower(), rec.data.lower(), "Unexpected data for record %s. Got %s, expected %s." % (data, rec.data.lower(), data.lower()))
338         self.assertEqual(wType, rec.wType, "Unexpected wType for record %s. Got %s, expected %s." % (data, rec.wType, wType))
339         self.assertEqual(dwTimeStamp, rec.dwTimeStamp, "Unexpected timestamp for record %s. Got %s, expected %s." % (data, rec.dwTimeStamp, dwTimeStamp))
340
341     def test_record_params(self):
342         """
343         Make sure that, when we add records to the database,
344         they're added with reasonable parameters.
345         """
346         self.add_record(self.custom_zone, "testrecord", "A", "192.168.50.50")
347         self.check_params(4, 240, 0, 900, 0, "192.168.50.50", 1)
348         self.delete_record(self.custom_zone, "testrecord", "A", "192.168.50.50")
349         self.add_record(self.custom_zone, "testrecord", "AAAA", "AAAA:AAAA::")
350         self.check_params(16, 240, 0, 900, 0, "AAAA:AAAA:0000:0000:0000:0000:0000:0000", 28)
351         self.delete_record(self.custom_zone, "testrecord", "AAAA", "AAAA:AAAA::")
352         self.add_record(self.custom_zone, "testrecord", "CNAME", "cnamedest")
353         self.check_params(13, 240, 0, 900, 0, "cnamedest", 5)
354         self.delete_record(self.custom_zone, "testrecord", "CNAME", "cnamedest")
355
356     def test_reject_invalid_commands(self):
357         """
358         Make sure that we can't add a variety of invalid records,
359         and that we can't update valid records to invalid ones.
360         """
361         num_failures = 0
362         for record_type_str in self.bad_records:
363             for record_str in self.bad_records[record_type_str]:
364                 # Attempt to add the bad record, which should fail. Then, attempt to query for and delete
365                 # it. Since it shouldn't exist, these should fail too.
366                 try:
367                     self.add_record(self.custom_zone, "testrecord", record_type_str, record_str, assertion=False)
368                     self.assert_num_records(self.custom_zone, "testrecord", record_type_str, expected_num=0)
369                     self.delete_record(self.custom_zone, "testrecord", record_type_str, record_str, assertion=False)
370                 except AssertionError as e:
371                     print(e)
372                     num_failures = num_failures + 1
373
374         # Also try to update valid records to invalid ones, making sure this fails
375         for record_type_str in self.bad_records:
376             for record_str in self.bad_records[record_type_str]:
377                 good_record_str = self.good_records[record_type_str][0]
378                 self.add_record(self.custom_zone, "testrecord", record_type_str, good_record_str)
379                 try:
380                     self.add_record(self.custom_zone, "testrecord", record_type_str, record_str, assertion=False)
381                 except AssertionError as e:
382                     print(e)
383                     num_failures = num_failures + 1
384                 self.delete_record(self.custom_zone, "testrecord", record_type_str, good_record_str)
385
386         self.assertTrue(num_failures == 0, "Failed to reject invalid commands. Total failures: %d." % num_failures)
387
388     def test_add_duplicate_different_type(self):
389         """
390         Attempt to add some values which have the same name as
391         existing ones, just a different type.
392         """
393         num_failures = 0
394         for record_type_str_1 in self.good_records:
395             record1 = self.good_records[record_type_str_1][0]
396             self.add_record(self.custom_zone, "testrecord", record_type_str_1, record1)
397             for record_type_str_2 in self.good_records:
398                 if record_type_str_1 == record_type_str_2:
399                     continue
400
401                 record2 = self.good_records[record_type_str_2][0]
402
403                 has_a = record_type_str_1 == 'A' or record_type_str_2 == 'A'
404                 has_aaaa = record_type_str_1 == 'AAAA' or record_type_str_2 == 'AAAA'
405                 has_cname = record_type_str_1 == 'CNAME' or record_type_str_2 == 'CNAME'
406                 has_ptr = record_type_str_1 == 'PTR' or record_type_str_2 == 'PTR'
407                 has_mx = record_type_str_1 == 'MX' or record_type_str_2 == 'MX'
408                 has_srv = record_type_str_1 == 'SRV' or record_type_str_2 == 'SRV'
409                 has_txt = record_type_str_1 == 'TXT' or record_type_str_2 == 'TXT'
410
411                 # If we attempt to add any record except A or AAAA when we already have an NS record,
412                 # the add should fail.
413                 add_error_ok = False
414                 if record_type_str_1 == 'NS' and not has_a and not has_aaaa:
415                     add_error_ok = True
416                 # If we attempt to add a CNAME when an A, PTR or MX record exists, the add should fail.
417                 if record_type_str_2 == 'CNAME' and (has_ptr or has_mx or has_a or has_aaaa):
418                     add_error_ok = True
419                 # If we have a CNAME, adding an A, AAAA, SRV or TXT record should fail.
420                 # If we have an A, AAAA, SRV or TXT record, adding a CNAME should fail.
421                 if has_cname and (has_a or has_aaaa or has_srv or has_txt):
422                     add_error_ok = True
423
424                 try:
425                     self.add_record(self.custom_zone, "testrecord", record_type_str_2, record2)
426                     if add_error_ok:
427                         num_failures = num_failures + 1
428                         print("Expected error when adding %s while a %s existed."
429                               % (record_type_str_2, record_type_str_1))
430                 except AssertionError as e:
431                     if not add_error_ok:
432                         num_failures = num_failures + 1
433                         print("Didn't expect error when adding %s while a %s existed."
434                               % (record_type_str_2, record_type_str_1))
435
436                 if not add_error_ok:
437                     # In the "normal" case, we expect the add to work and us to have one of each type of record afterwards.
438                     expected_num_type_1 = 1
439                     expected_num_type_2 = 1
440
441                     # If we have an MX record, a PTR record should replace it when added.
442                     # If we have a PTR record, an MX record should replace it when added.
443                     if has_ptr and has_mx:
444                         expected_num_type_1 = 0
445
446                     # If we have a CNAME, SRV or TXT record, a PTR or MX record should replace it when added.
447                     if (has_cname or has_srv or has_txt) and (record_type_str_2 == 'PTR' or record_type_str_2 == 'MX'):
448                         expected_num_type_1 = 0
449
450                     if (record_type_str_1 == 'NS' and (has_a or has_aaaa)):
451                         expected_num_type_2 = 0
452
453                     try:
454                         self.assert_num_records(self.custom_zone, "testrecord", record_type_str_1, expected_num=expected_num_type_1)
455                     except AssertionError as e:
456                         num_failures = num_failures + 1
457                         print("Expected %s %s records after adding a %s record and a %s record already existed."
458                               % (expected_num_type_1, record_type_str_1, record_type_str_2, record_type_str_1))
459                     try:
460                         self.assert_num_records(self.custom_zone, "testrecord", record_type_str_2, expected_num=expected_num_type_2)
461                     except AssertionError as e:
462                         num_failures = num_failures + 1
463                         print("Expected %s %s records after adding a %s record and a %s record already existed."
464                               % (expected_num_type_2, record_type_str_2, record_type_str_2, record_type_str_1))
465
466                 try:
467                     self.delete_record(self.custom_zone, "testrecord", record_type_str_2, record2)
468                 except AssertionError as e:
469                     pass
470
471             self.delete_record(self.custom_zone, "testrecord", record_type_str_1, record1)
472
473         self.assertTrue(num_failures == 0, "Failed collision and replacement behavior. Total failures: %d." % num_failures)
474
475     # Windows fails this test in the same way we do.
476     def _test_cname(self):
477         """
478         Test some special properties of CNAME records.
479         """
480
481         # RFC 1912: When there is a CNAME record, there must not be any other records with the same alias
482         cname_record = self.good_records["CNAME"][1]
483         self.add_record(self.custom_zone, "testrecord", "CNAME", cname_record)
484
485         for record_type_str in self.good_records:
486             other_record = self.good_records[record_type_str][0]
487             self.add_record(self.custom_zone, "testrecord", record_type_str, other_record, assertion=False)
488             self.assert_num_records(self.custom_zone, "testrecord", record_type_str, expected_num=0)
489
490         # RFC 2181: MX & NS records must not be allowed to point to a CNAME alias
491         mx_record = "testrecord 1"
492         ns_record = "testrecord"
493
494         self.add_record(self.custom_zone, "mxrec", "MX", mx_record, assertion=False)
495         self.add_record(self.custom_zone, "nsrec", "NS", ns_record, assertion=False)
496
497         self.delete_record(self.custom_zone, "testrecord", "CNAME", cname_record)
498
499     def test_add_duplicate_value(self):
500         """
501         Make sure that we can't add duplicate values of any type.
502         """
503         for record_type_str in self.good_records:
504             record = self.good_records[record_type_str][0]
505
506             self.add_record(self.custom_zone, "testrecord", record_type_str, record)
507             self.add_record(self.custom_zone, "testrecord", record_type_str, record, assertion=False)
508             self.assert_num_records(self.custom_zone, "testrecord", record_type_str)
509             self.delete_record(self.custom_zone, "testrecord", record_type_str, record)
510
511     def test_add_similar_value(self):
512         """
513         Attempt to add values with the same name and type in the same
514         zone. This should work, and should result in both values
515         existing (except with some types).
516         """
517         for record_type_str in self.good_records:
518             for i in range(1, len(self.good_records[record_type_str])):
519                 record1 = self.good_records[record_type_str][i-1]
520                 record2 = self.good_records[record_type_str][i]
521
522                 if record_type_str == 'CNAME':
523                     continue
524                 # We expect CNAME records to override one another, as
525                 # an alias can only map to one CNAME record.
526                 # Also, on Windows, when the empty string is added and
527                 # another record is added afterwards, the empty string
528                 # will be silently overridden by the new one, so it
529                 # fails this test for the empty string.
530                 expected_num = 1 if record_type_str == 'CNAME' else 2
531
532                 self.add_record(self.custom_zone, "testrecord", record_type_str, record1)
533                 self.add_record(self.custom_zone, "testrecord", record_type_str, record2)
534                 self.assert_num_records(self.custom_zone, "testrecord", record_type_str, expected_num=expected_num)
535                 self.delete_record(self.custom_zone, "testrecord", record_type_str, record1)
536                 self.delete_record(self.custom_zone, "testrecord", record_type_str, record2)
537
538     def assert_record(self, zone, name, record_type_str, expected_record_str,
539                       assertion=True, client_version=dnsserver.DNS_CLIENT_VERSION_LONGHORN):
540         """
541         Asserts whether or not the given record with the given type exists in the
542         given zone.
543         """
544         try:
545             _, result = self.query_records(zone, name, record_type_str)
546         except RuntimeError as e:
547             if assertion:
548                 raise AssertionError("Record '%s' of type '%s' was not present when it should have been."
549                                      % (expected_record_str, record_type_str))
550             else:
551                 return
552
553         found = False
554         for record in result.rec[0].records:
555             if record.data == expected_record_str:
556                 found = True
557                 break
558
559         if found and not assertion:
560             raise AssertionError("Record '%s' of type '%s' was present when it shouldn't have been." % (expected_record_str, record_type_str))
561         elif not found and assertion:
562             raise AssertionError("Record '%s' of type '%s' was not present when it should have been." % (expected_record_str, record_type_str))
563
564     def assert_num_records(self, zone, name, record_type_str, expected_num=1,
565                            client_version=dnsserver.DNS_CLIENT_VERSION_LONGHORN):
566         """
567         Asserts that there are a given amount of records with the given type in
568         the given zone.
569         """
570         try:
571             _, result = self.query_records(zone, name, record_type_str)
572             num_results = len(result.rec[0].records)
573             if not num_results == expected_num:
574                 raise AssertionError("There were %d records of type '%s' with the name '%s' when %d were expected."
575                                      % (num_results, record_type_str, name, expected_num))
576         except RuntimeError:
577             if not expected_num == 0:
578                 raise AssertionError("There were no records of type '%s' with the name '%s' when %d were expected."
579                                      % (record_type_str, name, expected_num))
580
581     def query_records(self, zone, name, record_type_str, client_version=dnsserver.DNS_CLIENT_VERSION_LONGHORN):
582         return self.conn.DnssrvEnumRecords2(client_version,
583                                             0,
584                                             self.server,
585                                             zone,
586                                             name,
587                                             None,
588                                             self.record_type_int(record_type_str),
589                                             dnsserver.DNS_RPC_VIEW_AUTHORITY_DATA | dnsserver.DNS_RPC_VIEW_NO_CHILDREN,
590                                             None,
591                                             None)
592
593     def record_obj_from_str(self, record_type_str, record_str):
594         if record_type_str == 'A':
595             return ARecord(record_str)
596         elif record_type_str == 'AAAA':
597             return AAAARecord(record_str)
598         elif record_type_str == 'PTR':
599             return PTRRecord(record_str)
600         elif record_type_str == 'CNAME':
601             return CNameRecord(record_str)
602         elif record_type_str == 'NS':
603             return NSRecord(record_str)
604         elif record_type_str == 'MX':
605             split = record_str.split(' ')
606             return MXRecord(split[0], int(split[1]))
607         elif record_type_str == 'SRV':
608             split = record_str.split(' ')
609             target = split[0]
610             port = int(split[1])
611             priority = int(split[2])
612             weight = int(split[3])
613             return SRVRecord(target, port, priority, weight)
614         elif record_type_str == 'TXT':
615             return TXTRecord(record_str)
616
617     def record_type_int(self, record_type_str):
618         if record_type_str == 'A':
619             return dnsp.DNS_TYPE_A
620         elif record_type_str == 'AAAA':
621             return dnsp.DNS_TYPE_AAAA
622         elif record_type_str == 'PTR':
623             return dnsp.DNS_TYPE_PTR
624         elif record_type_str == 'CNAME':
625             return dnsp.DNS_TYPE_CNAME
626         elif record_type_str == 'NS':
627             return dnsp.DNS_TYPE_NS
628         elif record_type_str == 'MX':
629             return dnsp.DNS_TYPE_MX
630         elif record_type_str == 'SRV':
631             return dnsp.DNS_TYPE_SRV
632         elif record_type_str == 'TXT':
633             return dnsp.DNS_TYPE_TXT
634
635     def add_record(self, zone, name, record_type_str, record_str,
636                    assertion=True, client_version=dnsserver.DNS_CLIENT_VERSION_LONGHORN):
637         """
638         Attempts to add a map from the given name to a record of the given type,
639         in the given zone.
640         Also asserts whether or not the add was successful.
641         This can also update existing records if they have the same name.
642         """
643         record = self.record_obj_from_str(record_type_str, record_str)
644         add_rec_buf = dnsserver.DNS_RPC_RECORD_BUF()
645         add_rec_buf.rec = record
646
647         try:
648             self.conn.DnssrvUpdateRecord2(client_version,
649                                           0,
650                                           self.server,
651                                           zone,
652                                           name,
653                                           add_rec_buf,
654                                           None)
655             if not assertion:
656                 raise AssertionError("Successfully added record '%s' of type '%s', which should have failed."
657                                      % (record_str, record_type_str))
658         except RuntimeError as e:
659             if assertion:
660                 raise AssertionError("Failed to add record '%s' of type '%s', which should have succeeded. Error was '%s'."
661                                      % (record_str, record_type_str, str(e)))
662
663     def delete_record(self, zone, name, record_type_str, record_str,
664                       assertion=True, client_version=dnsserver.DNS_CLIENT_VERSION_LONGHORN):
665         """
666         Attempts to delete a record with the given name, record and record type
667         from the given zone.
668         Also asserts whether or not the deletion was successful.
669         """
670         record = self.record_obj_from_str(record_type_str, record_str)
671         del_rec_buf = dnsserver.DNS_RPC_RECORD_BUF()
672         del_rec_buf.rec = record
673
674         try:
675             self.conn.DnssrvUpdateRecord2(client_version,
676                                           0,
677                                           self.server,
678                                           zone,
679                                           name,
680                                           None,
681                                           del_rec_buf)
682             if not assertion:
683                 raise AssertionError("Successfully deleted record '%s' of type '%s', which should have failed." % (record_str, record_type_str))
684         except RuntimeError as e:
685             if assertion:
686                 raise AssertionError("Failed to delete record '%s' of type '%s', which should have succeeded. Error was '%s'." % (record_str, record_type_str, str(e)))
687
688     def test_query2(self):
689         typeid, result = self.conn.DnssrvQuery2(dnsserver.DNS_CLIENT_VERSION_W2K,
690                                                 0,
691                                                 self.server,
692                                                 None,
693                                                 'ServerInfo')
694         self.assertEquals(dnsserver.DNSSRV_TYPEID_SERVER_INFO_W2K, typeid)
695
696         typeid, result = self.conn.DnssrvQuery2(dnsserver.DNS_CLIENT_VERSION_DOTNET,
697                                                 0,
698                                                 self.server,
699                                                 None,
700                                                 'ServerInfo')
701         self.assertEquals(dnsserver.DNSSRV_TYPEID_SERVER_INFO_DOTNET, typeid)
702
703         typeid, result = self.conn.DnssrvQuery2(dnsserver.DNS_CLIENT_VERSION_LONGHORN,
704                                                 0,
705                                                 self.server,
706                                                 None,
707                                                 'ServerInfo')
708         self.assertEquals(dnsserver.DNSSRV_TYPEID_SERVER_INFO, typeid)
709
710     def test_operation2(self):
711         client_version = dnsserver.DNS_CLIENT_VERSION_LONGHORN
712         rev_zone = '1.168.192.in-addr.arpa'
713
714         zone_create = dnsserver.DNS_RPC_ZONE_CREATE_INFO_LONGHORN()
715         zone_create.pszZoneName = rev_zone
716         zone_create.dwZoneType = dnsp.DNS_ZONE_TYPE_PRIMARY
717         zone_create.fAllowUpdate = dnsp.DNS_ZONE_UPDATE_SECURE
718         zone_create.fAging = 0
719         zone_create.dwDpFlags = dnsserver.DNS_DP_DOMAIN_DEFAULT
720
721         # Create zone
722         self.conn.DnssrvOperation2(client_version,
723                                    0,
724                                    self.server,
725                                    None,
726                                    0,
727                                    'ZoneCreate',
728                                    dnsserver.DNSSRV_TYPEID_ZONE_CREATE,
729                                    zone_create)
730
731         request_filter = (dnsserver.DNS_ZONE_REQUEST_REVERSE |
732                           dnsserver.DNS_ZONE_REQUEST_PRIMARY)
733         _, zones = self.conn.DnssrvComplexOperation2(client_version,
734                                                      0,
735                                                      self.server,
736                                                      None,
737                                                      'EnumZones',
738                                                      dnsserver.DNSSRV_TYPEID_DWORD,
739                                                      request_filter)
740         self.assertEquals(1, zones.dwZoneCount)
741
742         # Delete zone
743         self.conn.DnssrvOperation2(client_version,
744                                    0,
745                                    self.server,
746                                    rev_zone,
747                                    0,
748                                    'DeleteZoneFromDs',
749                                    dnsserver.DNSSRV_TYPEID_NULL,
750                                    None)
751
752         typeid, zones = self.conn.DnssrvComplexOperation2(client_version,
753                                                           0,
754                                                           self.server,
755                                                           None,
756                                                           'EnumZones',
757                                                           dnsserver.DNSSRV_TYPEID_DWORD,
758                                                           request_filter)
759         self.assertEquals(0, zones.dwZoneCount)
760
761
762     def test_complexoperation2(self):
763         client_version = dnsserver.DNS_CLIENT_VERSION_LONGHORN
764         request_filter = (dnsserver.DNS_ZONE_REQUEST_FORWARD |
765                           dnsserver.DNS_ZONE_REQUEST_PRIMARY)
766
767         typeid, zones = self.conn.DnssrvComplexOperation2(client_version,
768                                                           0,
769                                                           self.server,
770                                                           None,
771                                                           'EnumZones',
772                                                           dnsserver.DNSSRV_TYPEID_DWORD,
773                                                           request_filter)
774         self.assertEquals(dnsserver.DNSSRV_TYPEID_ZONE_LIST, typeid)
775         self.assertEquals(3, zones.dwZoneCount)
776
777         request_filter = (dnsserver.DNS_ZONE_REQUEST_REVERSE |
778                           dnsserver.DNS_ZONE_REQUEST_PRIMARY)
779         typeid, zones = self.conn.DnssrvComplexOperation2(client_version,
780                                                           0,
781                                                           self.server,
782                                                           None,
783                                                           'EnumZones',
784                                                           dnsserver.DNSSRV_TYPEID_DWORD,
785                                                           request_filter)
786         self.assertEquals(dnsserver.DNSSRV_TYPEID_ZONE_LIST, typeid)
787         self.assertEquals(0, zones.dwZoneCount)
788
789     def test_enumrecords2(self):
790         client_version = dnsserver.DNS_CLIENT_VERSION_LONGHORN
791         record_type = dnsp.DNS_TYPE_NS
792         select_flags = (dnsserver.DNS_RPC_VIEW_ROOT_HINT_DATA |
793                         dnsserver.DNS_RPC_VIEW_ADDITIONAL_DATA)
794         _, roothints = self.conn.DnssrvEnumRecords2(client_version,
795                                                     0,
796                                                     self.server,
797                                                     '..RootHints',
798                                                     '.',
799                                                     None,
800                                                     record_type,
801                                                     select_flags,
802                                                     None,
803                                                     None)
804         self.assertEquals(14, roothints.count)  # 1 NS + 13 A records (a-m)
805
806     def test_updaterecords2(self):
807         client_version = dnsserver.DNS_CLIENT_VERSION_LONGHORN
808         record_type = dnsp.DNS_TYPE_A
809         select_flags = dnsserver.DNS_RPC_VIEW_AUTHORITY_DATA
810
811         name = 'dummy'
812         rec = ARecord('1.2.3.4')
813         rec2 = ARecord('5.6.7.8')
814
815         # Add record
816         add_rec_buf = dnsserver.DNS_RPC_RECORD_BUF()
817         add_rec_buf.rec = rec
818         self.conn.DnssrvUpdateRecord2(client_version,
819                                       0,
820                                       self.server,
821                                       self.zone,
822                                       name,
823                                       add_rec_buf,
824                                       None)
825
826         _, result = self.conn.DnssrvEnumRecords2(client_version,
827                                                  0,
828                                                  self.server,
829                                                  self.zone,
830                                                  name,
831                                                  None,
832                                                  record_type,
833                                                  select_flags,
834                                                  None,
835                                                  None)
836         self.assertEquals(1, result.count)
837         self.assertEquals(1, result.rec[0].wRecordCount)
838         self.assertEquals(dnsp.DNS_TYPE_A, result.rec[0].records[0].wType)
839         self.assertEquals('1.2.3.4', result.rec[0].records[0].data)
840
841         # Update record
842         add_rec_buf = dnsserver.DNS_RPC_RECORD_BUF()
843         add_rec_buf.rec = rec2
844         del_rec_buf = dnsserver.DNS_RPC_RECORD_BUF()
845         del_rec_buf.rec = rec
846         self.conn.DnssrvUpdateRecord2(client_version,
847                                       0,
848                                       self.server,
849                                       self.zone,
850                                       name,
851                                       add_rec_buf,
852                                       del_rec_buf)
853
854         buflen, result = self.conn.DnssrvEnumRecords2(client_version,
855                                                       0,
856                                                       self.server,
857                                                       self.zone,
858                                                       name,
859                                                       None,
860                                                       record_type,
861                                                       select_flags,
862                                                       None,
863                                                       None)
864         self.assertEquals(1, result.count)
865         self.assertEquals(1, result.rec[0].wRecordCount)
866         self.assertEquals(dnsp.DNS_TYPE_A, result.rec[0].records[0].wType)
867         self.assertEquals('5.6.7.8', result.rec[0].records[0].data)
868
869         # Delete record
870         del_rec_buf = dnsserver.DNS_RPC_RECORD_BUF()
871         del_rec_buf.rec = rec2
872         self.conn.DnssrvUpdateRecord2(client_version,
873                                       0,
874                                       self.server,
875                                       self.zone,
876                                       name,
877                                       None,
878                                       del_rec_buf)
879
880         self.assertRaises(RuntimeError, self.conn.DnssrvEnumRecords2,
881                           client_version,
882                           0,
883                           self.server,
884                           self.zone,
885                           name,
886                           None,
887                           record_type,
888                           select_flags,
889                           None,
890                           None)
891
892     # The following tests do not pass against Samba because the owner and
893     # group are not consistent with Windows, as well as some ACEs.
894     #
895     # The following ACE are also required for 2012R2:
896     #
897     # (OA;CIIO;WP;ea1b7b93-5e48-46d5-bc6c-4df4fda78a35;bf967a86-0de6-11d0-a285-00aa003049e2;PS)
898     # (OA;OICI;RPWP;3f78c3e5-f79a-46bd-a0b8-9d18116ddc79;;PS)"
899     #
900     # [TPM + Allowed-To-Act-On-Behalf-Of-Other-Identity]
901     def test_security_descriptor_msdcs_zone(self):
902         """
903         Make sure that security descriptors of the msdcs zone is
904         as expected.
905         """
906
907         zones = self.samdb.search(base="DC=ForestDnsZones,%s" % self.samdb.get_default_basedn(),
908                                   scope=ldb.SCOPE_SUBTREE,
909                                   expression="(&(objectClass=dnsZone)(name=_msdcs*))",
910                                   attrs=["nTSecurityDescriptor", "objectClass"])
911         self.assertEqual(len(zones), 1)
912         self.assertTrue("nTSecurityDescriptor" in zones[0])
913         tmp = zones[0]["nTSecurityDescriptor"][0]
914         utils = sd_utils.SDUtils(self.samdb)
915         sd = ndr_unpack(security.descriptor, tmp)
916
917         domain_sid = security.dom_sid(self.samdb.get_domain_sid())
918
919         res = self.samdb.search(base=self.samdb.get_default_basedn(), scope=ldb.SCOPE_SUBTREE,
920                                 expression="(sAMAccountName=DnsAdmins)",
921                                 attrs=["objectSid"])
922
923         dns_admin = str(ndr_unpack(security.dom_sid, res[0]['objectSid'][0]))
924
925         packed_sd = descriptor.sddl2binary("O:SYG:BA" \
926                                            "D:AI(A;;RPWPCRCCDCLCLORCWOWDSDDTSW;;;DA)" \
927                                            "(A;;CC;;;AU)" \
928                                            "(A;;RPLCLORC;;;WD)" \
929                                            "(A;;RPWPCRCCDCLCLORCWOWDSDDTSW;;;SY)" \
930                                            "(A;CI;RPWPCRCCDCLCRCWOWDSDDTSW;;;ED)",
931                                            domain_sid, {"DnsAdmins": dns_admin})
932         expected_sd = descriptor.get_clean_sd(ndr_unpack(security.descriptor, packed_sd))
933
934         diff = descriptor.get_diff_sds(expected_sd, sd, domain_sid)
935         self.assertEqual(diff, '', "SD of msdcs zone different to expected.\n"
936                          "Difference was:\n%s\nExpected: %s\nGot: %s" %
937                          (diff, expected_sd.as_sddl(utils.domain_sid),
938                           sd.as_sddl(utils.domain_sid)))
939
940     def test_security_descriptor_forest_zone(self):
941         """
942         Make sure that security descriptors of forest dns zones are
943         as expected.
944         """
945         forest_zone = "test_forest_zone"
946         zone_create_info = dnsserver.DNS_RPC_ZONE_CREATE_INFO_LONGHORN()
947         zone_create_info.dwZoneType = dnsp.DNS_ZONE_TYPE_PRIMARY
948         zone_create_info.fAging = 0
949         zone_create_info.fDsIntegrated = 1
950         zone_create_info.fLoadExisting = 1
951
952         zone_create_info.pszZoneName = forest_zone
953         zone_create_info.dwDpFlags = dnsserver.DNS_DP_FOREST_DEFAULT
954
955         self.conn.DnssrvOperation2(dnsserver.DNS_CLIENT_VERSION_LONGHORN,
956                                    0,
957                                    self.server,
958                                    None,
959                                    0,
960                                    'ZoneCreate',
961                                    dnsserver.DNSSRV_TYPEID_ZONE_CREATE,
962                                    zone_create_info)
963
964         partition_dn = self.samdb.get_default_basedn()
965         partition_dn.add_child("DC=ForestDnsZones")
966         zones = self.samdb.search(base=partition_dn, scope=ldb.SCOPE_SUBTREE,
967                                   expression="(name=%s)" % forest_zone,
968                                   attrs=["nTSecurityDescriptor"])
969         self.assertEqual(len(zones), 1)
970         current_dn = zones[0].dn
971         self.assertTrue("nTSecurityDescriptor" in zones[0])
972         tmp = zones[0]["nTSecurityDescriptor"][0]
973         utils = sd_utils.SDUtils(self.samdb)
974         sd = ndr_unpack(security.descriptor, tmp)
975
976         domain_sid = security.dom_sid(self.samdb.get_domain_sid())
977
978         res = self.samdb.search(base=self.samdb.get_default_basedn(),
979                                 scope=ldb.SCOPE_SUBTREE,
980                                 expression="(sAMAccountName=DnsAdmins)",
981                                 attrs=["objectSid"])
982
983         dns_admin = str(ndr_unpack(security.dom_sid, res[0]['objectSid'][0]))
984
985         packed_sd = descriptor.sddl2binary("O:DAG:DA" \
986                                            "D:AI(A;;RPWPCRCCDCLCLORCWOWDSDDTSW;;;DA)" \
987                                            "(A;;CC;;;AU)" \
988                                            "(A;;RPLCLORC;;;WD)" \
989                                            "(A;;RPWPCRCCDCLCLORCWOWDSDDTSW;;;SY)" \
990                                            "(A;CI;RPWPCRCCDCLCRCWOWDSDDTSW;;;ED)",
991                                            domain_sid, {"DnsAdmins": dns_admin})
992         expected_sd = descriptor.get_clean_sd(ndr_unpack(security.descriptor, packed_sd))
993
994         packed_msdns = descriptor.get_dns_forest_microsoft_dns_descriptor(domain_sid,
995                                                                           {"DnsAdmins": dns_admin})
996         expected_msdns_sd = descriptor.get_clean_sd(ndr_unpack(security.descriptor, packed_msdns))
997
998         packed_part_sd = descriptor.get_dns_partition_descriptor(domain_sid)
999         expected_part_sd = descriptor.get_clean_sd(ndr_unpack(security.descriptor,
1000                                                               packed_part_sd))
1001         try:
1002             msdns_dn = ldb.Dn(self.samdb, "CN=MicrosoftDNS,%s" % str(partition_dn))
1003             security_desc_dict = [(current_dn.get_linearized(),  expected_sd),
1004                                   (msdns_dn.get_linearized(), expected_msdns_sd),
1005                                   (partition_dn.get_linearized(), expected_part_sd)]
1006
1007             for (key, sec_desc) in security_desc_dict:
1008                 zones = self.samdb.search(base=key, scope=ldb.SCOPE_BASE,
1009                                           attrs=["nTSecurityDescriptor"])
1010                 self.assertTrue("nTSecurityDescriptor" in zones[0])
1011                 tmp = zones[0]["nTSecurityDescriptor"][0]
1012                 utils = sd_utils.SDUtils(self.samdb)
1013
1014                 sd = ndr_unpack(security.descriptor, tmp)
1015                 diff = descriptor.get_diff_sds(sec_desc, sd, domain_sid)
1016
1017                 self.assertEqual(diff, '', "Security descriptor of forest DNS zone with DN '%s' different to expected. Difference was:\n%s\nExpected: %s\nGot: %s"
1018                                  % (key, diff, sec_desc.as_sddl(utils.domain_sid), sd.as_sddl(utils.domain_sid)))
1019
1020         finally:
1021             self.conn.DnssrvOperation2(dnsserver.DNS_CLIENT_VERSION_LONGHORN,
1022                                        0,
1023                                        self.server,
1024                                        forest_zone,
1025                                        0,
1026                                        'DeleteZoneFromDs',
1027                                        dnsserver.DNSSRV_TYPEID_NULL,
1028                                        None)
1029
1030     def test_security_descriptor_domain_zone(self):
1031         """
1032         Make sure that security descriptors of domain dns zones are
1033         as expected.
1034         """
1035
1036         partition_dn = self.samdb.get_default_basedn()
1037         partition_dn.add_child("DC=DomainDnsZones")
1038         zones = self.samdb.search(base=partition_dn, scope=ldb.SCOPE_SUBTREE,
1039                                   expression="(name=%s)" % self.custom_zone,
1040                                   attrs=["nTSecurityDescriptor"])
1041         self.assertEqual(len(zones), 1)
1042         current_dn = zones[0].dn
1043         self.assertTrue("nTSecurityDescriptor" in zones[0])
1044         tmp = zones[0]["nTSecurityDescriptor"][0]
1045         utils = sd_utils.SDUtils(self.samdb)
1046         sd = ndr_unpack(security.descriptor, tmp)
1047         sddl = sd.as_sddl(utils.domain_sid)
1048
1049         domain_sid = security.dom_sid(self.samdb.get_domain_sid())
1050
1051         res = self.samdb.search(base=self.samdb.get_default_basedn(), scope=ldb.SCOPE_SUBTREE,
1052                                 expression="(sAMAccountName=DnsAdmins)",
1053                                 attrs=["objectSid"])
1054
1055         dns_admin = str(ndr_unpack(security.dom_sid, res[0]['objectSid'][0]))
1056
1057         packed_sd = descriptor.sddl2binary("O:DAG:DA" \
1058                                            "D:AI(A;;RPWPCRCCDCLCLORCWOWDSDDTSW;;;DA)" \
1059                                            "(A;;CC;;;AU)" \
1060                                            "(A;;RPLCLORC;;;WD)" \
1061                                            "(A;;RPWPCRCCDCLCLORCWOWDSDDTSW;;;SY)" \
1062                                            "(A;CI;RPWPCRCCDCLCRCWOWDSDDTSW;;;ED)",
1063                                            domain_sid, {"DnsAdmins": dns_admin})
1064         expected_sd = descriptor.get_clean_sd(ndr_unpack(security.descriptor, packed_sd))
1065
1066         packed_msdns = descriptor.get_dns_domain_microsoft_dns_descriptor(domain_sid,
1067                                                                           {"DnsAdmins": dns_admin})
1068         expected_msdns_sd = descriptor.get_clean_sd(ndr_unpack(security.descriptor, packed_msdns))
1069
1070         packed_part_sd = descriptor.get_dns_partition_descriptor(domain_sid)
1071         expected_part_sd = descriptor.get_clean_sd(ndr_unpack(security.descriptor,
1072                                                               packed_part_sd))
1073
1074         msdns_dn = ldb.Dn(self.samdb, "CN=MicrosoftDNS,%s" % str(partition_dn))
1075         security_desc_dict = [(current_dn.get_linearized(),  expected_sd),
1076                               (msdns_dn.get_linearized(), expected_msdns_sd),
1077                               (partition_dn.get_linearized(), expected_part_sd)]
1078
1079         for (key, sec_desc) in security_desc_dict:
1080             zones = self.samdb.search(base=key, scope=ldb.SCOPE_BASE,
1081                                       attrs=["nTSecurityDescriptor"])
1082             self.assertTrue("nTSecurityDescriptor" in zones[0])
1083             tmp = zones[0]["nTSecurityDescriptor"][0]
1084             utils = sd_utils.SDUtils(self.samdb)
1085
1086             sd = ndr_unpack(security.descriptor, tmp)
1087             diff = descriptor.get_diff_sds(sec_desc, sd, domain_sid)
1088
1089             self.assertEqual(diff, '', "Security descriptor of domain DNS zone with DN '%s' different to expected. Difference was:\n%s\nExpected: %s\nGot: %s"
1090                              % (key, diff, sec_desc.as_sddl(utils.domain_sid), sd.as_sddl(utils.domain_sid)))