python/samba/tests/samba_tool/dnscmd.py

   1 # Unix SMB/CIFS implementation.
   2 # Copyright (C) Andrew Bartlett <abartlet@catalyst.net.nz>
   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 import os
  19 import ldb
  20
  21 from samba.auth import system_session
  22 from samba.samdb import SamDB
  23 from samba.ndr import ndr_unpack, ndr_pack
  24 from samba.dcerpc import dnsp
  25 from samba.tests.samba_tool.base import SambaToolCmdTest
  26
  27
  28 class DnsCmdTestCase(SambaToolCmdTest):
  29     def setUp(self):
  30         super(DnsCmdTestCase, self).setUp()
  31
  32         self.dburl = "ldap://%s" % os.environ["SERVER"]
  33         self.creds_string = "-U%s%%%s" % (os.environ["DC_USERNAME"],
  34                                           os.environ["DC_PASSWORD"])
  35
  36         self.samdb = self.getSamDB("-H", self.dburl, self.creds_string)
  37         self.config_dn = str(self.samdb.get_config_basedn())
  38
  39         self.testip = "192.168.0.193"
  40         self.testip2 = "192.168.0.194"
  41
  42         self.addZone()
  43
  44         # Note: SOA types don't work (and shouldn't), as we only have one zone per DNS record.
  45
  46         good_dns = ["SAMDOM.EXAMPLE.COM",
  47                     "1.EXAMPLE.COM",
  48                     "%sEXAMPLE.COM" % ("1." * 100),
  49                     "EXAMPLE",
  50                     "\n.COM",
  51                     "!@#$%^&*()_",
  52                     "HIGH\xFFBYTE",
  53                     "@.EXAMPLE.COM",
  54                     "."]
  55         bad_dns = ["...",
  56                    ".EXAMPLE.COM",
  57                    ".EXAMPLE.",
  58                    "",
  59                    "SAMDOM..EXAMPLE.COM"]
  60
  61         good_mx = ["SAMDOM.EXAMPLE.COM 65530"]
  62         bad_mx = ["SAMDOM.EXAMPLE.COM -1",
  63                   "SAMDOM.EXAMPLE.COM",
  64                   " ",
  65                   "SAMDOM.EXAMPLE.COM 1 1",
  66                   "SAMDOM.EXAMPLE.COM SAMDOM.EXAMPLE.COM"]
  67
  68         good_srv = ["SAMDOM.EXAMPLE.COM 65530 65530 65530"]
  69         bad_srv = ["SAMDOM.EXAMPLE.COM 0 65536 0",
  70                    "SAMDOM.EXAMPLE.COM 0 0 65536",
  71                    "SAMDOM.EXAMPLE.COM 65536 0 0"]
  72
  73         for bad_dn in bad_dns:
  74             bad_mx.append("%s 1" % bad_dn)
  75             bad_srv.append("%s 0 0 0" % bad_dn)
  76         for good_dn in good_dns:
  77             good_mx.append("%s 1" % good_dn)
  78             good_srv.append("%s 0 0 0" % good_dn)
  79
  80         self.good_records = {
  81                 "A":["192.168.0.1", "255.255.255.255"],
  82                 "AAAA":["1234:5678:9ABC:DEF0:0000:0000:0000:0000",
  83                         "0000:0000:0000:0000:0000:0000:0000:0000",
  84                         "1234:5678:9ABC:DEF0:1234:5678:9ABC:DEF0",
  85                         "1234:1234:1234::",
  86                         "1234:5678:9ABC:DEF0::",
  87                         "0000:0000::0000",
  88                         "1234::5678:9ABC:0000:0000:0000:0000",
  89                         "::1",
  90                         "::",
  91                         "1:1:1:1:1:1:1:1"],
  92                 "PTR": good_dns,
  93                 "CNAME": good_dns,
  94                 "NS": good_dns,
  95                 "MX": good_mx,
  96                 "SRV": good_srv,
  97                 "TXT": ["text", "", "@#!", "\n"]
  98         }
  99
 100         self.bad_records = {
 101                 "A":["192.168.0.500",
 102                      "255.255.255.255/32"],
 103                 "AAAA":["GGGG:1234:5678:9ABC:0000:0000:0000:0000",
 104                         "0000:0000:0000:0000:0000:0000:0000:0000/1",
 105                         "AAAA:AAAA:AAAA:AAAA:G000:0000:0000:1234",
 106                         "1234:5678:9ABC:DEF0:1234:5678:9ABC:DEF0:1234",
 107                         "1234:5678:9ABC:DEF0:1234:5678:9ABC",
 108                         "1111::1111::1111"],
 109                 "PTR": bad_dns,
 110                 "CNAME": bad_dns,
 111                 "NS": bad_dns,
 112                 "MX": bad_mx,
 113                 "SRV": bad_srv
 114         }
 115
 116     def tearDown(self):
 117         self.deleteZone()
 118         super(DnsCmdTestCase, self).tearDown()
 119
 120     def resetZone(self):
 121         self.deleteZone()
 122         self.addZone()
 123
 124     def addZone(self):
 125         self.zone = "zone"
 126         result, out, err = self.runsubcmd("dns",
 127                                           "zonecreate",
 128                                           os.environ["SERVER"],
 129                                           self.zone,
 130                                           self.creds_string)
 131         self.assertCmdSuccess(result, out, err)
 132
 133     def deleteZone(self):
 134         result, out, err = self.runsubcmd("dns",
 135                                           "zonedelete",
 136                                           os.environ["SERVER"],
 137                                           self.zone,
 138                                           self.creds_string)
 139         self.assertCmdSuccess(result, out, err)
 140
 141     def get_record_from_db(self, zone_name, record_name):
 142         zones = self.samdb.search(base="DC=DomainDnsZones,%s"
 143                                   % self.samdb.get_default_basedn(),
 144                                   scope=ldb.SCOPE_SUBTREE,
 145                                   expression="(objectClass=dnsZone)",
 146                                   attrs=["cn"])
 147
 148         for zone in zones:
 149             if zone_name in str(zone.dn):
 150                 zone_dn = zone.dn
 151                 break
 152
 153         records = self.samdb.search(base=zone_dn, scope=ldb.SCOPE_SUBTREE,
 154                                     expression="(objectClass=dnsNode)",
 155                                     attrs=["dnsRecord"])
 156
 157         for old_packed_record in records:
 158             if record_name in str(old_packed_record.dn):
 159                 return (old_packed_record.dn,
 160                         ndr_unpack(dnsp.DnssrvRpcRecord,
 161                                    old_packed_record["dnsRecord"][0]))
 162
 163     def test_rank_none(self):
 164         record_str = "192.168.50.50"
 165         record_type_str = "A"
 166
 167         result, out, err = self.runsubcmd("dns", "add", os.environ["SERVER"],
 168                                           self.zone, "testrecord", record_type_str,
 169                                           record_str, self.creds_string)
 170         self.assertCmdSuccess(result, out, err,
 171                               "Failed to add record '%s' with type %s."
 172                               % (record_str, record_type_str))
 173
 174         dn, record = self.get_record_from_db(self.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         errors = []
 181
 182         # The record should still exist
 183         result, out, err = self.runsubcmd("dns", "query", os.environ["SERVER"],
 184                                           self.zone, "testrecord", record_type_str,
 185                                           self.creds_string)
 186         try:
 187             self.assertCmdSuccess(result, out, err,
 188                                   "Failed to query for a record" \
 189                                   "which had DNS_RANK_NONE.")
 190             self.assertTrue("testrecord" in out and record_str in out,
 191                             "Query for a record which had DNS_RANK_NONE" \
 192                             "succeeded but produced no resulting records.")
 193         except AssertionError as e:
 194             # Windows produces no resulting records
 195             pass
 196
 197         # We should not be able to add a duplicate
 198         result, out, err = self.runsubcmd("dns", "add", os.environ["SERVER"],
 199                                           self.zone, "testrecord", record_type_str,
 200                                           record_str, self.creds_string)
 201         try:
 202             self.assertCmdFail(result, "Successfully added duplicate record" \
 203                                "of one which had DNS_RANK_NONE.")
 204         except AssertionError as e:
 205             errors.append(e)
 206
 207         # We should be able to delete it
 208         result, out, err = self.runsubcmd("dns", "delete", os.environ["SERVER"],
 209                                           self.zone, "testrecord", record_type_str,
 210                                           record_str, self.creds_string)
 211         try:
 212             self.assertCmdSuccess(result, out, err, "Failed to delete record" \
 213                                   "which had DNS_RANK_NONE.")
 214         except AssertionError as e:
 215             errors.append(e)
 216
 217         # Now the record should not exist
 218         result, out, err = self.runsubcmd("dns", "query", os.environ["SERVER"],
 219                                           self.zone, "testrecord",
 220                                           record_type_str, self.creds_string)
 221         try:
 222             self.assertCmdFail(result, "Successfully queried for deleted record" \
 223                                "which had DNS_RANK_NONE.")
 224         except AssertionError as e:
 225             errors.append(e)
 226
 227         if len(errors) > 0:
 228             err_str = "Failed appropriate behaviour with DNS_RANK_NONE:"
 229             for error in errors:
 230                 err_str = err_str + "\n" + str(error)
 231             raise AssertionError(err_str)
 232
 233     def test_accept_valid_commands(self):
 234         """
 235         For all good records, attempt to add, query and delete them.
 236         """
 237         num_failures = 0
 238         failure_msgs = []
 239         for dnstype in self.good_records:
 240             for record in self.good_records[dnstype]:
 241                 try:
 242                     result, out, err = self.runsubcmd("dns", "add",
 243                                                       os.environ["SERVER"],
 244                                                       self.zone, "testrecord",
 245                                                       dnstype, record,
 246                                                       self.creds_string)
 247                     self.assertCmdSuccess(result, out, err, "Failed to add" \
 248                                           "record %s with type %s."
 249                                           % (record, dnstype))
 250
 251                     result, out, err = self.runsubcmd("dns", "query",
 252                                                       os.environ["SERVER"],
 253                                                       self.zone, "testrecord",
 254                                                       dnstype,
 255                                                       self.creds_string)
 256                     self.assertCmdSuccess(result, out, err, "Failed to query" \
 257                                           "record %s with qualifier %s."
 258                                           % (record, dnstype))
 259
 260                     result, out, err = self.runsubcmd("dns", "delete",
 261                                                       os.environ["SERVER"],
 262                                                       self.zone, "testrecord",
 263                                                       dnstype, record,
 264                                                       self.creds_string)
 265                     self.assertCmdSuccess(result, out, err, "Failed to remove" \
 266                                           "record %s with type %s."
 267                                           % (record, dnstype))
 268                 except AssertionError as e:
 269                     num_failures = num_failures + 1
 270                     failure_msgs.append(e)
 271
 272         if num_failures > 0:
 273             for msg in failure_msgs:
 274                 print(msg)
 275             self.fail("Failed to accept valid commands. %d total failures." \
 276                       "Errors above." % num_failures)
 277
 278     def test_reject_invalid_commands(self):
 279         """
 280         For all bad records, attempt to add them and update to them,
 281         making sure that both operations fail.
 282         """
 283         num_failures = 0
 284         failure_msgs = []
 285
 286         # Add invalid records and make sure they fail to be added
 287         for dnstype in self.bad_records:
 288             for record in self.bad_records[dnstype]:
 289                 try:
 290                     result, out, err = self.runsubcmd("dns", "add",
 291                                                       os.environ["SERVER"],
 292                                                       self.zone, "testrecord",
 293                                                       dnstype, record,
 294                                                       self.creds_string)
 295                     self.assertCmdFail(result, "Successfully added invalid" \
 296                                        "record '%s' of type '%s'."
 297                                        % (record, dnstype))
 298                 except AssertionError as e:
 299                     num_failures = num_failures + 1
 300                     failure_msgs.append(e)
 301                     self.resetZone()
 302                 try:
 303                     result, out, err = self.runsubcmd("dns", "delete",
 304                                                       os.environ["SERVER"],
 305                                                       self.zone, "testrecord",
 306                                                       dnstype, record,
 307                                                       self.creds_string)
 308                     self.assertCmdFail(result, "Successfully deleted invalid" \
 309                                        "record '%s' of type '%s' which" \
 310                                        "shouldn't exist." % (record, dnstype))
 311                 except AssertionError as e:
 312                     num_failures = num_failures + 1
 313                     failure_msgs.append(e)
 314                     self.resetZone()
 315
 316         # Update valid records to invalid ones and make sure they
 317         # fail to be updated
 318         for dnstype in self.bad_records:
 319             for bad_record in self.bad_records[dnstype]:
 320                 good_record = self.good_records[dnstype][0]
 321
 322                 try:
 323                     result, out, err = self.runsubcmd("dns", "add",
 324                                                       os.environ["SERVER"],
 325                                                       self.zone, "testrecord",
 326                                                       dnstype, good_record,
 327                                                       self.creds_string)
 328                     self.assertCmdSuccess(result, out, err, "Failed to add " \
 329                                           "record '%s' with type %s."
 330                                           % (record, dnstype))
 331
 332                     result, out, err = self.runsubcmd("dns", "update",
 333                                                       os.environ["SERVER"],
 334                                                       self.zone, "testrecord",
 335                                                       dnstype, good_record,
 336                                                       bad_record,
 337                                                       self.creds_string)
 338                     self.assertCmdFail(result, "Successfully updated valid " \
 339                                        "record '%s' of type '%s' to invalid " \
 340                                        "record '%s' of the same type."
 341                                        % (good_record, dnstype, bad_record))
 342
 343                     result, out, err = self.runsubcmd("dns", "delete",
 344                                                       os.environ["SERVER"],
 345                                                       self.zone, "testrecord",
 346                                                       dnstype, good_record,
 347                                                       self.creds_string)
 348                     self.assertCmdSuccess(result, out, err, "Could not delete " \
 349                                           "valid record '%s' of type '%s'."
 350                                           % (good_record, dnstype))
 351                 except AssertionError as e:
 352                     num_failures = num_failures + 1
 353                     failure_msgs.append(e)
 354                     self.resetZone()
 355
 356         if num_failures > 0:
 357             for msg in failure_msgs:
 358                 print(msg)
 359             self.fail("Failed to reject invalid commands. %d total failures. " \
 360                       "Errors above." % num_failures)
 361
 362     def test_update_invalid_type(self):
 363         """
 364         Make sure that a record can't be updated to one of a different type.
 365         """
 366         for dnstype1 in self.good_records:
 367             record1 = self.good_records[dnstype1][0]
 368             result, out, err = self.runsubcmd("dns", "add",
 369                                               os.environ["SERVER"],
 370                                               self.zone, "testrecord",
 371                                               dnstype1, record1,
 372                                               self.creds_string)
 373             self.assertCmdSuccess(result, out, err, "Failed to add " \
 374                                   "record %s with type %s."
 375                                   % (record1, dnstype1))
 376
 377             for dnstype2 in self.good_records:
 378                 record2 = self.good_records[dnstype2][0]
 379
 380                 # Make sure that record2 isn't a valid entry of dnstype1.
 381                 # For example, any A-type will also be a valid TXT-type.
 382                 result, out, err = self.runsubcmd("dns", "add",
 383                                                   os.environ["SERVER"],
 384                                                   self.zone, "testrecord",
 385                                                   dnstype1, record2,
 386                                                   self.creds_string)
 387                 try:
 388                     self.assertCmdFail(result)
 389                 except AssertionError:
 390                     continue  # Don't check this one, because record2 _is_ a valid entry of dnstype1.
 391
 392                 # Check both ways: Give the current type and try to update,
 393                 # and give the new type and try to update.
 394                 result, out, err = self.runsubcmd("dns", "update",
 395                                                   os.environ["SERVER"],
 396                                                   self.zone, "testrecord",
 397                                                   dnstype1, record1,
 398                                                   record2, self.creds_string)
 399                 self.assertCmdFail(result, "Successfully updated record '%s' " \
 400                                    "to '%s', even though the latter is of " \
 401                                    "type '%s' where '%s' was expected."
 402                                    % (record1, record2, dnstype2, dnstype1))
 403
 404                 result, out, err = self.runsubcmd("dns", "update",
 405                                                   os.environ["SERVER"],
 406                                                   self.zone, "testrecord",
 407                                                   dnstype2, record1, record2,
 408                                                   self.creds_string)
 409                 self.assertCmdFail(result, "Successfully updated record " \
 410                                    "'%s' to '%s', even though the former " \
 411                                    "is of type '%s' where '%s' was expected."
 412                                    % (record1, record2, dnstype1, dnstype2))
 413
 414     def test_update_valid_type(self):
 415         for dnstype in self.good_records:
 416             for record in self.good_records[dnstype]:
 417                 result, out, err = self.runsubcmd("dns", "add",
 418                                                   os.environ["SERVER"],
 419                                                   self.zone, "testrecord",
 420                                                   dnstype, record,
 421                                                   self.creds_string)
 422                 self.assertCmdSuccess(result, out, err, "Failed to add " \
 423                                       "record %s with type %s."
 424                                       % (record, dnstype))
 425
 426                 # Update the record to be the same.
 427                 result, out, err = self.runsubcmd("dns", "update",
 428                                                   os.environ["SERVER"],
 429                                                   self.zone, "testrecord",
 430                                                   dnstype, record, record,
 431                                                   self.creds_string)
 432                 self.assertCmdFail(result, "Successfully updated record " \
 433                                    "'%s' to be exactly the same." % record)
 434
 435                 result, out, err = self.runsubcmd("dns", "delete",
 436                                                   os.environ["SERVER"],
 437                                                   self.zone, "testrecord",
 438                                                   dnstype, record,
 439                                                   self.creds_string)
 440                 self.assertCmdSuccess(result, out, err, "Could not delete " \
 441                                       "valid record '%s' of type '%s'."
 442                                       % (record, dnstype))
 443
 444         for record in self.good_records["SRV"]:
 445             result, out, err = self.runsubcmd("dns", "add",
 446                                               os.environ["SERVER"],
 447                                               self.zone, "testrecord",
 448                                               "SRV", record,
 449                                               self.creds_string)
 450             self.assertCmdSuccess(result, out, err, "Failed to add " \
 451                                   "record %s with type 'SRV'." % record)
 452
 453             split = record.split(' ')
 454             new_bit = str(int(split[3]) + 1)
 455             new_record = '%s %s %s %s' % (split[0], split[1], split[2], new_bit)
 456
 457             result, out, err = self.runsubcmd("dns", "update",
 458                                               os.environ["SERVER"],
 459                                               self.zone, "testrecord",
 460                                               "SRV", record,
 461                                               new_record, self.creds_string)
 462             self.assertCmdSuccess(result, out, err, "Failed to update record " \
 463                                   "'%s' of type '%s' to '%s'."
 464                                   % (record, "SRV", new_record))
 465
 466             result, out, err = self.runsubcmd("dns", "query",
 467                                               os.environ["SERVER"],
 468                                               self.zone, "testrecord",
 469                                               "SRV", self.creds_string)
 470             self.assertCmdSuccess(result, out, err, "Failed to query for " \
 471                                   "record '%s' of type '%s'."
 472                                   % (new_record, "SRV"))
 473
 474             result, out, err = self.runsubcmd("dns", "delete",
 475                                               os.environ["SERVER"],
 476                                               self.zone, "testrecord",
 477                                               "SRV", new_record,
 478                                               self.creds_string)
 479             self.assertCmdSuccess(result, out, err, "Could not delete " \
 480                                   "valid record '%s' of type '%s'."
 481                                   % (new_record, "SRV"))
 482
 483         # Since 'dns update' takes the current value as a parameter, make sure
 484         # we can't enter the wrong current value for a given record.
 485         for dnstype in self.good_records:
 486             if len(self.good_records[dnstype]) < 3:
 487                 continue  # Not enough records of this type to do this test
 488
 489             used_record = self.good_records[dnstype][0]
 490             unused_record = self.good_records[dnstype][1]
 491             new_record = self.good_records[dnstype][2]
 492
 493             result, out, err = self.runsubcmd("dns", "add",
 494                                               os.environ["SERVER"],
 495                                               self.zone, "testrecord",
 496                                               dnstype, used_record,
 497                                               self.creds_string)
 498             self.assertCmdSuccess(result, out, err, "Failed to add record %s " \
 499                                   "with type %s." % (used_record, dnstype))
 500
 501             result, out, err = self.runsubcmd("dns", "update",
 502                                               os.environ["SERVER"],
 503                                               self.zone, "testrecord",
 504                                               dnstype, unused_record,
 505                                               new_record,
 506                                               self.creds_string)
 507             self.assertCmdFail(result, "Successfully updated record '%s' " \
 508                                "from '%s' to '%s', even though the given " \
 509                                "source record is incorrect."
 510                                % (used_record, unused_record, new_record))
 511
 512     def test_invalid_types(self):
 513         result, out, err = self.runsubcmd("dns", "add",
 514                                           os.environ["SERVER"],
 515                                           self.zone, "testrecord",
 516                                           "SOA", "test",
 517                                           self.creds_string)
 518         self.assertCmdFail(result, "Successfully added record of type SOA, " \
 519                            "when this type should not be available.")
 520         self.assertTrue("type SOA is not supported" in err,
 521                         "Invalid error message '%s' when attempting to " \
 522                         "add record of type SOA." % err)
 523
 524     def test_add_overlapping_different_type(self):
 525         """
 526         Make sure that we can add an entry with the same name as an existing one but a different type.
 527         """
 528
 529         i = 0
 530         for dnstype1 in self.good_records:
 531             record1 = self.good_records[dnstype1][0]
 532             for dnstype2 in self.good_records:
 533                 # Only do some subset of dns types, otherwise it takes a long time.
 534                 i += 1
 535                 if i % 4 != 0:
 536                     continue
 537
 538                 if dnstype1 == dnstype2:
 539                     continue
 540
 541                 record2 = self.good_records[dnstype2][0]
 542
 543                 result, out, err = self.runsubcmd("dns", "add",
 544                                                   os.environ["SERVER"],
 545                                                   self.zone, "testrecord",
 546                                                   dnstype1, record1,
 547                                                   self.creds_string)
 548                 self.assertCmdSuccess(result, out, err, "Failed to add record " \
 549                                       "'%s' of type '%s'." % (record1, dnstype1))
 550
 551                 result, out, err = self.runsubcmd("dns", "add",
 552                                                   os.environ["SERVER"],
 553                                                   self.zone, "testrecord",
 554                                                   dnstype2, record2,
 555                                                   self.creds_string)
 556                 self.assertCmdSuccess(result, out, err, "Failed to add record " \
 557                                       "'%s' of type '%s' when a record '%s' " \
 558                                       "of type '%s' with the same name exists."
 559                                       % (record1, dnstype1, record2, dnstype2))
 560
 561                 result, out, err = self.runsubcmd("dns", "query",
 562                                                   os.environ["SERVER"],
 563                                                   self.zone, "testrecord",
 564                                                   dnstype1, self.creds_string)
 565                 self.assertCmdSuccess(result, out, err, "Failed to query for " \
 566                                       "record '%s' of type '%s' when a new " \
 567                                       "record '%s' of type '%s' with the same " \
 568                                       "name was added."
 569                                       % (record1, dnstype1, record2, dnstype2))
 570
 571                 result, out, err = self.runsubcmd("dns", "query",
 572                                                   os.environ["SERVER"],
 573                                                   self.zone, "testrecord",
 574                                                   dnstype2, self.creds_string)
 575                 self.assertCmdSuccess(result, out, err, "Failed to query " \
 576                                       "record '%s' of type '%s' which should " \
 577                                       "have been added with the same name as " \
 578                                       "record '%s' of type '%s'."
 579                                       % (record2, dnstype2, record1, dnstype1))
 580
 581                 result, out, err = self.runsubcmd("dns", "delete",
 582                                                   os.environ["SERVER"],
 583                                                   self.zone, "testrecord",
 584                                                   dnstype1, record1,
 585                                                   self.creds_string)
 586                 self.assertCmdSuccess(result, out, err, "Failed to delete " \
 587                                       "record '%s' of type '%s'."
 588                                       % (record1, dnstype1))
 589
 590                 result, out, err = self.runsubcmd("dns", "delete",
 591                                                   os.environ["SERVER"],
 592                                                   self.zone, "testrecord",
 593                                                   dnstype2, record2,
 594                                                   self.creds_string)
 595                 self.assertCmdSuccess(result, out, err, "Failed to delete " \
 596                                       "record '%s' of type '%s'."
 597                                       % (record2, dnstype2))
 598
 599     def test_query_deleted_record(self):
 600         self.runsubcmd("dns", "add", os.environ["SERVER"], self.zone,
 601                        "testrecord", "A", self.testip, self.creds_string)
 602         self.runsubcmd("dns", "delete", os.environ["SERVER"], self.zone,
 603                        "testrecord", "A", self.testip, self.creds_string)
 604
 605         result, out, err = self.runsubcmd("dns", "query",
 606                                           os.environ["SERVER"],
 607                                           self.zone, "testrecord",
 608                                           "A", self.creds_string)
 609         self.assertCmdFail(result)
 610
 611     def test_add_duplicate_record(self):
 612         for record_type in self.good_records:
 613             result, out, err = self.runsubcmd("dns", "add",
 614                                               os.environ["SERVER"],
 615                                               self.zone, "testrecord",
 616                                               record_type,
 617                                               self.good_records[record_type][0],
 618                                               self.creds_string)
 619             self.assertCmdSuccess(result, out, err)
 620             result, out, err = self.runsubcmd("dns", "add",
 621                                               os.environ["SERVER"],
 622                                               self.zone, "testrecord",
 623                                               record_type,
 624                                               self.good_records[record_type][0],
 625                                               self.creds_string)
 626             self.assertCmdFail(result)
 627             result, out, err = self.runsubcmd("dns", "query",
 628                                               os.environ["SERVER"],
 629                                               self.zone, "testrecord",
 630                                               record_type, self.creds_string)
 631             self.assertCmdSuccess(result, out, err)
 632             result, out, err = self.runsubcmd("dns", "delete",
 633                                               os.environ["SERVER"],
 634                                               self.zone, "testrecord",
 635                                               record_type,
 636                                               self.good_records[record_type][0],
 637                                               self.creds_string)
 638             self.assertCmdSuccess(result, out, err)
 639
 640     def test_remove_deleted_record(self):
 641         self.runsubcmd("dns", "add", os.environ["SERVER"], self.zone,
 642                        "testrecord", "A", self.testip, self.creds_string)
 643         self.runsubcmd("dns", "delete", os.environ["SERVER"], self.zone,
 644                        "testrecord", "A", self.testip, self.creds_string)
 645
 646         # Attempting to delete a record that has already been deleted or has never existed should fail
 647         result, out, err = self.runsubcmd("dns", "delete",
 648                                           os.environ["SERVER"],
 649                                           self.zone, "testrecord",
 650                                           "A", self.testip, self.creds_string)
 651         self.assertCmdFail(result)
 652         result, out, err = self.runsubcmd("dns", "query",
 653                                           os.environ["SERVER"],
 654                                           self.zone, "testrecord",
 655                                           "A", self.creds_string)
 656         self.assertCmdFail(result)
 657         result, out, err = self.runsubcmd("dns", "delete",
 658                                           os.environ["SERVER"],
 659                                           self.zone, "testrecord2",
 660                                           "A", self.testip, self.creds_string)
 661         self.assertCmdFail(result)
 662
 663     def test_cleanup_record(self):
 664         """
 665         Test dns cleanup command is working fine.
 666         """
 667
 668         # add a A record
 669         self.runsubcmd("dns", "add", os.environ["SERVER"], self.zone,
 670                        'testa', "A", self.testip, self.creds_string)
 671
 672         # the above A record points to this host
 673         dnshostname = '{}.{}'.format('testa', self.zone.lower())
 674
 675         # add a CNAME record points to above host
 676         self.runsubcmd("dns", "add", os.environ["SERVER"], self.zone,
 677                        'testcname', "CNAME", dnshostname, self.creds_string)
 678
 679         # add a NS record
 680         self.runsubcmd("dns", "add", os.environ["SERVER"], self.zone,
 681                        'testns', "NS", dnshostname, self.creds_string)
 682
 683         # add a PTR record points to above host
 684         self.runsubcmd("dns", "add", os.environ["SERVER"], self.zone,
 685                        'testptr', "PTR", dnshostname, self.creds_string)
 686
 687         # add a SRV record points to above host
 688         srv_record = "{} 65530 65530 65530".format(dnshostname)
 689         self.runsubcmd("dns", "add", os.environ["SERVER"], self.zone,
 690                        'testsrv', "SRV", srv_record, self.creds_string)
 691
 692         # cleanup record for this dns host
 693         self.runsubcmd("dns", "cleanup", os.environ["SERVER"],
 694                        dnshostname, self.creds_string)
 695
 696         # all records should be marked as dNSTombstoned
 697         for record_name in ['testa', 'testcname', 'testns', 'testptr', 'testsrv']:
 698
 699             records = self.samdb.search(
 700                 base="DC=DomainDnsZones,{}".format(self.samdb.get_default_basedn()),
 701                 scope=ldb.SCOPE_SUBTREE,
 702                 expression="(&(objectClass=dnsNode)(name={}))".format(record_name),
 703                 attrs=["dNSTombstoned"])
 704
 705             self.assertEqual(len(records), 1)
 706             for record in records:
 707                 self.assertEqual(str(record['dNSTombstoned']), 'TRUE')
 708
 709     def test_cleanup_record_no_A_record(self):
 710         """
 711         Test dns cleanup command works with no A record.
 712         """
 713
 714         # add a A record
 715         self.runsubcmd("dns", "add", os.environ["SERVER"], self.zone,
 716                        'notesta', "A", self.testip, self.creds_string)
 717
 718         # the above A record points to this host
 719         dnshostname = '{}.{}'.format('testa', self.zone.lower())
 720
 721         # add a CNAME record points to above host
 722         self.runsubcmd("dns", "add", os.environ["SERVER"], self.zone,
 723                        'notestcname', "CNAME", dnshostname, self.creds_string)
 724
 725         # add a NS record
 726         self.runsubcmd("dns", "add", os.environ["SERVER"], self.zone,
 727                        'notestns', "NS", dnshostname, self.creds_string)
 728
 729         # add a PTR record points to above host
 730         self.runsubcmd("dns", "add", os.environ["SERVER"], self.zone,
 731                        'notestptr', "PTR", dnshostname, self.creds_string)
 732
 733         # add a SRV record points to above host
 734         srv_record = "{} 65530 65530 65530".format(dnshostname)
 735         self.runsubcmd("dns", "add", os.environ["SERVER"], self.zone,
 736                        'notestsrv', "SRV", srv_record, self.creds_string)
 737
 738         # Remove the initial A record (leading to hanging references)
 739         self.runsubcmd("dns", "delete", os.environ["SERVER"], self.zone,
 740                        'notesta', "A", self.testip, self.creds_string)
 741
 742         # cleanup record for this dns host
 743         self.runsubcmd("dns", "cleanup", os.environ["SERVER"],
 744                        dnshostname, self.creds_string)
 745
 746         # all records should be marked as dNSTombstoned
 747         for record_name in ['notestcname', 'notestns', 'notestptr', 'notestsrv']:
 748
 749             records = self.samdb.search(
 750                 base="DC=DomainDnsZones,{}".format(self.samdb.get_default_basedn()),
 751                 scope=ldb.SCOPE_SUBTREE,
 752                 expression="(&(objectClass=dnsNode)(name={}))".format(record_name),
 753                 attrs=["dNSTombstoned"])
 754
 755             self.assertEqual(len(records), 1)
 756             for record in records:
 757                 self.assertEqual(str(record['dNSTombstoned']), 'TRUE')
 758
 759     def test_cleanup_multi_srv_record(self):
 760         """
 761         Test dns cleanup command for multi-valued SRV record.
 762
 763         Steps:
 764         - Add 2 A records host1 and host2
 765         - Add a SRV record srv1 and points to both host1 and host2
 766         - Run cleanup command for host1
 767         - Check records for srv1, data for host1 should be gone and host2 is kept.
 768         """
 769
 770         hosts = ['host1', 'host2']  # A record names
 771         srv_name = 'srv1'
 772
 773         # add A records
 774         for host in hosts:
 775             self.runsubcmd("dns", "add", os.environ["SERVER"], self.zone,
 776                            host, "A", self.testip, self.creds_string)
 777
 778             # the above A record points to this host
 779             dnshostname = '{}.{}'.format(host, self.zone.lower())
 780
 781             # add a SRV record points to above host
 782             srv_record = "{} 65530 65530 65530".format(dnshostname)
 783             self.runsubcmd("dns", "add", os.environ["SERVER"], self.zone,
 784                            srv_name, "SRV", srv_record, self.creds_string)
 785
 786         records = self.samdb.search(
 787             base="DC=DomainDnsZones,{}".format(self.samdb.get_default_basedn()),
 788             scope=ldb.SCOPE_SUBTREE,
 789             expression="(&(objectClass=dnsNode)(name={}))".format(srv_name),
 790             attrs=['dnsRecord'])
 791         # should have 2 records here
 792         self.assertEqual(len(records[0]['dnsRecord']), 2)
 793
 794         # cleanup record for dns host1
 795         dnshostname1 = 'host1.{}'.format(self.zone.lower())
 796         self.runsubcmd("dns", "cleanup", os.environ["SERVER"],
 797                        dnshostname1, self.creds_string)
 798
 799         records = self.samdb.search(
 800             base="DC=DomainDnsZones,{}".format(self.samdb.get_default_basedn()),
 801             scope=ldb.SCOPE_SUBTREE,
 802             expression="(&(objectClass=dnsNode)(name={}))".format(srv_name),
 803             attrs=['dnsRecord', 'dNSTombstoned'])
 804
 805         # dnsRecord for host1 should be deleted
 806         self.assertEqual(len(records[0]['dnsRecord']), 1)
 807
 808         # unpack data
 809         dns_record_bin = records[0]['dnsRecord'][0]
 810         dns_record_obj = ndr_unpack(dnsp.DnssrvRpcRecord, dns_record_bin)
 811
 812         # dnsRecord for host2 is still there and is the only one
 813         dnshostname2 = 'host2.{}'.format(self.zone.lower())
 814         self.assertEqual(dns_record_obj.data.nameTarget, dnshostname2)
 815
 816         # assert that the record isn't spuriously tombstoned
 817         self.assertTrue('dNSTombstoned' not in records[0] or
 818                         str(record['dNSTombstoned']) == 'FALSE')
 819
 820     def test_dns_wildcards(self):
 821         """
 822         Ensure that DNS wild card entries can be added deleted and queried
 823         """
 824         num_failures = 0
 825         failure_msgs = []
 826         records = [("*.", "MISS", "A", "1.1.1.1"),
 827                    ("*.SAMDOM", "MISS.SAMDOM", "A", "1.1.1.2")]
 828         for (name, miss, dnstype, record) in records:
 829             try:
 830                 result, out, err = self.runsubcmd("dns", "add",
 831                                                   os.environ["SERVER"],
 832                                                   self.zone, name,
 833                                                   dnstype, record,
 834                                                   self.creds_string)
 835                 self.assertCmdSuccess(
 836                     result,
 837                     out,
 838                     err,
 839                     ("Failed to add record %s (%s) with type %s."
 840                      % (name, record, dnstype)))
 841
 842                 result, out, err = self.runsubcmd("dns", "query",
 843                                                   os.environ["SERVER"],
 844                                                   self.zone, name,
 845                                                   dnstype,
 846                                                   self.creds_string)
 847                 self.assertCmdSuccess(
 848                     result,
 849                     out,
 850                     err,
 851                     ("Failed to query record %s with qualifier %s."
 852                      % (record, dnstype)))
 853
 854                 # dns tool does not perform dns wildcard search if the name
 855                 # does not match
 856                 result, out, err = self.runsubcmd("dns", "query",
 857                                                   os.environ["SERVER"],
 858                                                   self.zone, miss,
 859                                                   dnstype,
 860                                                   self.creds_string)
 861                 self.assertCmdFail(
 862                     result,
 863                     ("Failed to query record %s with qualifier %s."
 864                      % (record, dnstype)))
 865
 866                 result, out, err = self.runsubcmd("dns", "delete",
 867                                                   os.environ["SERVER"],
 868                                                   self.zone, name,
 869                                                   dnstype, record,
 870                                                   self.creds_string)
 871                 self.assertCmdSuccess(
 872                     result,
 873                     out,
 874                     err,
 875                     ("Failed to remove record %s with type %s."
 876                      % (record, dnstype)))
 877             except AssertionError as e:
 878                 num_failures = num_failures + 1
 879                 failure_msgs.append(e)
 880
 881         if num_failures > 0:
 882             for msg in failure_msgs:
 883                 print(msg)
 884             self.fail("Failed to accept valid commands. %d total failures."
 885                       "Errors above." % num_failures)