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