python/samba/tests/audit_log_dsdb.py

   1 # Tests for SamDb password change audit logging.
   2 # Copyright (C) Andrew Bartlett <abartlet@samba.org> 2018
   3 #
   4 # This program is free software; you can redistribute it and/or modify
   5 # it under the terms of the GNU General Public License as published by
   6 # the Free Software Foundation; either version 3 of the License, or
   7 # (at your option) any later version.
   8 #
   9 # This program is distributed in the hope that it will be useful,
  10 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  11 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  12 # GNU General Public License for more details.
  13 #
  14 # You should have received a copy of the GNU General Public License
  15 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
  16 #
  17
  18 from __future__ import print_function
  19 """Tests for the SamDb logging of password changes.
  20 """
  21
  22 import samba.tests
  23 from samba.dcerpc.messaging import MSG_DSDB_LOG, DSDB_EVENT_NAME
  24 from ldb import ERR_NO_SUCH_OBJECT
  25 from samba.samdb import SamDB
  26 from samba.auth import system_session
  27 import os
  28 import time
  29 from samba.tests.audit_log_base import AuditLogTestBase
  30 from samba.tests import delete_force
  31 from samba.net import Net
  32 import samba
  33 from samba.dcerpc import security, lsa
  34
  35 USER_NAME = "auditlogtestuser"
  36 USER_PASS = samba.generate_random_password(32, 32)
  37
  38
  39 class AuditLogDsdbTests(AuditLogTestBase):
  40
  41     def setUp(self):
  42         self.message_type = MSG_DSDB_LOG
  43         self.event_type = DSDB_EVENT_NAME
  44         super(AuditLogDsdbTests, self).setUp()
  45
  46         self.server_ip = os.environ["SERVER_IP"]
  47
  48         host = "ldap://%s" % os.environ["SERVER"]
  49         self.ldb = SamDB(url=host,
  50                          session_info=system_session(),
  51                          credentials=self.get_credentials(),
  52                          lp=self.get_loadparm())
  53         self.server = os.environ["SERVER"]
  54
  55         # Gets back the basedn
  56         self.base_dn = self.ldb.domain_dn()
  57
  58         # Get the old "dSHeuristics" if it was set
  59         dsheuristics = self.ldb.get_dsheuristics()
  60
  61         # Set the "dSHeuristics" to activate the correct "userPassword"
  62         # behaviour
  63         self.ldb.set_dsheuristics("000000001")
  64
  65         # Reset the "dSHeuristics" as they were before
  66         self.addCleanup(self.ldb.set_dsheuristics, dsheuristics)
  67
  68         # Get the old "minPwdAge"
  69         minPwdAge = self.ldb.get_minPwdAge()
  70
  71         # Set it temporarily to "0"
  72         self.ldb.set_minPwdAge("0")
  73         self.base_dn = self.ldb.domain_dn()
  74
  75         # Reset the "minPwdAge" as it was before
  76         self.addCleanup(self.ldb.set_minPwdAge, minPwdAge)
  77
  78         # (Re)adds the test user USER_NAME with password USER_PASS
  79         delete_force(self.ldb, "cn=" + USER_NAME + ",cn=users," + self.base_dn)
  80         self.ldb.add({
  81             "dn": "cn=" + USER_NAME + ",cn=users," + self.base_dn,
  82             "objectclass": "user",
  83             "sAMAccountName": USER_NAME,
  84             "userPassword": USER_PASS
  85         })
  86
  87     #
  88     # Discard the messages from the setup code
  89     #
  90     def discardSetupMessages(self, dn):
  91         self.waitForMessages(2, dn=dn)
  92         self.discardMessages()
  93
  94     def tearDown(self):
  95         self.discardMessages()
  96         super(AuditLogDsdbTests, self).tearDown()
  97
  98     def haveExpectedTxn(self, expected):
  99         if self.context["txnMessage"] is not None:
 100             txn = self.context["txnMessage"]["dsdbTransaction"]
 101             if txn["transactionId"] == expected:
 102                 return True
 103         return False
 104
 105     def waitForTransaction(self, expected, connection=None):
 106         """Wait for a transaction message to arrive
 107         The connection is passed through to keep the connection alive
 108         until all the logging messages have been received.
 109         """
 110
 111         self.connection = connection
 112
 113         start_time = time.time()
 114         while not self.haveExpectedTxn(expected):
 115             self.msg_ctx.loop_once(0.1)
 116             if time.time() - start_time > 1:
 117                 self.connection = None
 118                 return ""
 119
 120         self.connection = None
 121         return self.context["txnMessage"]
 122
 123     def test_net_change_password(self):
 124
 125         dn = "CN=" + USER_NAME + ",CN=Users," + self.base_dn
 126         self.discardSetupMessages(dn)
 127
 128         creds = self.insta_creds(template=self.get_credentials())
 129
 130         lp = self.get_loadparm()
 131         net = Net(creds, lp, server=self.server)
 132         password = "newPassword!!42"
 133
 134         net.change_password(newpassword=password,
 135                             username=USER_NAME,
 136                             oldpassword=USER_PASS)
 137
 138         messages = self.waitForMessages(1, net, dn=dn)
 139         print("Received %d messages" % len(messages))
 140         self.assertEqual(1,
 141                           len(messages),
 142                           "Did not receive the expected number of messages")
 143
 144         audit = messages[0]["dsdbChange"]
 145         self.assertEqual("Modify", audit["operation"])
 146         self.assertFalse(audit["performedAsSystem"])
 147         self.assertTrue(dn.lower(), audit["dn"].lower())
 148         self.assertRegexpMatches(audit["remoteAddress"],
 149                                  self.remoteAddress)
 150         session_id = self.get_session()
 151         self.assertEqual(session_id, audit["sessionId"])
 152         # We skip the check for self.get_service_description() as this
 153         # is subject to a race between smbd and the s4 rpc_server code
 154         # as to which will set the description as it is DCE/RPC over SMB
 155
 156         self.assertTrue(self.is_guid(audit["transactionId"]))
 157
 158         attributes = audit["attributes"]
 159         self.assertEqual(1, len(attributes))
 160         actions = attributes["clearTextPassword"]["actions"]
 161         self.assertEqual(1, len(actions))
 162         self.assertTrue(actions[0]["redacted"])
 163         self.assertEqual("replace", actions[0]["action"])
 164
 165     def test_net_set_password(self):
 166
 167         dn = "CN=" + USER_NAME + ",CN=Users," + self.base_dn
 168         self.discardSetupMessages(dn)
 169
 170         creds = self.insta_creds(template=self.get_credentials())
 171
 172         lp = self.get_loadparm()
 173         net = Net(creds, lp, server=self.server)
 174         password = "newPassword!!42"
 175         domain = lp.get("workgroup")
 176
 177         net.set_password(newpassword=password,
 178                          account_name=USER_NAME,
 179                          domain_name=domain)
 180         messages = self.waitForMessages(1, net, dn=dn)
 181         print("Received %d messages" % len(messages))
 182         self.assertEqual(1,
 183                           len(messages),
 184                           "Did not receive the expected number of messages")
 185         audit = messages[0]["dsdbChange"]
 186         self.assertEqual("Modify", audit["operation"])
 187         self.assertFalse(audit["performedAsSystem"])
 188         self.assertEqual(dn, audit["dn"])
 189         self.assertRegexpMatches(audit["remoteAddress"],
 190                                  self.remoteAddress)
 191         session_id = self.get_session()
 192         self.assertEqual(session_id, audit["sessionId"])
 193         # We skip the check for self.get_service_description() as this
 194         # is subject to a race between smbd and the s4 rpc_server code
 195         # as to which will set the description as it is DCE/RPC over SMB
 196
 197         self.assertTrue(self.is_guid(audit["transactionId"]))
 198
 199         attributes = audit["attributes"]
 200         self.assertEqual(1, len(attributes))
 201         actions = attributes["clearTextPassword"]["actions"]
 202         self.assertEqual(1, len(actions))
 203         self.assertTrue(actions[0]["redacted"])
 204         self.assertEqual("replace", actions[0]["action"])
 205
 206     def test_ldap_change_password(self):
 207
 208         dn = "cn=" + USER_NAME + ",cn=users," + self.base_dn
 209         self.discardSetupMessages(dn)
 210
 211         new_password = samba.generate_random_password(32, 32)
 212         dn = "cn=" + USER_NAME + ",cn=users," + self.base_dn
 213         self.ldb.modify_ldif(
 214             "dn: " + dn + "\n" +
 215             "changetype: modify\n" +
 216             "delete: userPassword\n" +
 217             "userPassword: " + USER_PASS + "\n" +
 218             "add: userPassword\n" +
 219             "userPassword: " + new_password + "\n")
 220
 221         messages = self.waitForMessages(1)
 222         print("Received %d messages" % len(messages))
 223         self.assertEqual(1,
 224                           len(messages),
 225                           "Did not receive the expected number of messages")
 226
 227         audit = messages[0]["dsdbChange"]
 228         self.assertEqual("Modify", audit["operation"])
 229         self.assertFalse(audit["performedAsSystem"])
 230         self.assertEqual(dn, audit["dn"])
 231         self.assertRegexpMatches(audit["remoteAddress"],
 232                                  self.remoteAddress)
 233         self.assertTrue(self.is_guid(audit["sessionId"]))
 234         session_id = self.get_session()
 235         self.assertEqual(session_id, audit["sessionId"])
 236         service_description = self.get_service_description()
 237         self.assertEqual(service_description, "LDAP")
 238
 239         attributes = audit["attributes"]
 240         self.assertEqual(1, len(attributes))
 241         actions = attributes["userPassword"]["actions"]
 242         self.assertEqual(2, len(actions))
 243         self.assertTrue(actions[0]["redacted"])
 244         self.assertEqual("delete", actions[0]["action"])
 245         self.assertTrue(actions[1]["redacted"])
 246         self.assertEqual("add", actions[1]["action"])
 247
 248     def test_ldap_replace_password(self):
 249
 250         dn = "cn=" + USER_NAME + ",cn=users," + self.base_dn
 251         self.discardSetupMessages(dn)
 252
 253         new_password = samba.generate_random_password(32, 32)
 254         self.ldb.modify_ldif(
 255             "dn: " + dn + "\n" +
 256             "changetype: modify\n" +
 257             "replace: userPassword\n" +
 258             "userPassword: " + new_password + "\n")
 259
 260         messages = self.waitForMessages(1, dn=dn)
 261         print("Received %d messages" % len(messages))
 262         self.assertEqual(1,
 263                           len(messages),
 264                           "Did not receive the expected number of messages")
 265
 266         audit = messages[0]["dsdbChange"]
 267         self.assertEqual("Modify", audit["operation"])
 268         self.assertFalse(audit["performedAsSystem"])
 269         self.assertTrue(dn.lower(), audit["dn"].lower())
 270         self.assertRegexpMatches(audit["remoteAddress"],
 271                                  self.remoteAddress)
 272         self.assertTrue(self.is_guid(audit["sessionId"]))
 273         session_id = self.get_session()
 274         self.assertEqual(session_id, audit["sessionId"])
 275         service_description = self.get_service_description()
 276         self.assertEqual(service_description, "LDAP")
 277         self.assertTrue(self.is_guid(audit["transactionId"]))
 278
 279         attributes = audit["attributes"]
 280         self.assertEqual(1, len(attributes))
 281         actions = attributes["userPassword"]["actions"]
 282         self.assertEqual(1, len(actions))
 283         self.assertTrue(actions[0]["redacted"])
 284         self.assertEqual("replace", actions[0]["action"])
 285
 286     def test_ldap_add_user(self):
 287
 288         # The setup code adds a user, so we check for the dsdb events
 289         # generated by it.
 290         dn = "cn=" + USER_NAME + ",cn=users," + self.base_dn
 291         messages = self.waitForMessages(2, dn=dn)
 292         print("Received %d messages" % len(messages))
 293         self.assertEqual(2,
 294                           len(messages),
 295                           "Did not receive the expected number of messages")
 296
 297         audit = messages[1]["dsdbChange"]
 298         self.assertEqual("Add", audit["operation"])
 299         self.assertFalse(audit["performedAsSystem"])
 300         self.assertEqual(dn, audit["dn"])
 301         self.assertRegexpMatches(audit["remoteAddress"],
 302                                  self.remoteAddress)
 303         session_id = self.get_session()
 304         self.assertEqual(session_id, audit["sessionId"])
 305         service_description = self.get_service_description()
 306         self.assertEqual(service_description, "LDAP")
 307         self.assertTrue(self.is_guid(audit["sessionId"]))
 308         self.assertTrue(self.is_guid(audit["transactionId"]))
 309
 310         attributes = audit["attributes"]
 311         self.assertEqual(3, len(attributes))
 312
 313         actions = attributes["objectclass"]["actions"]
 314         self.assertEqual(1, len(actions))
 315         self.assertEqual("add", actions[0]["action"])
 316         self.assertEqual(1, len(actions[0]["values"]))
 317         self.assertEqual("user", actions[0]["values"][0]["value"])
 318
 319         actions = attributes["sAMAccountName"]["actions"]
 320         self.assertEqual(1, len(actions))
 321         self.assertEqual("add", actions[0]["action"])
 322         self.assertEqual(1, len(actions[0]["values"]))
 323         self.assertEqual(USER_NAME, actions[0]["values"][0]["value"])
 324
 325         actions = attributes["userPassword"]["actions"]
 326         self.assertEqual(1, len(actions))
 327         self.assertEqual("add", actions[0]["action"])
 328         self.assertTrue(actions[0]["redacted"])
 329
 330     def test_samdb_delete_user(self):
 331
 332         dn = "cn=" + USER_NAME + ",cn=users," + self.base_dn
 333         self.discardSetupMessages(dn)
 334
 335         self.ldb.deleteuser(USER_NAME)
 336
 337         messages = self.waitForMessages(1, dn=dn)
 338         print("Received %d messages" % len(messages))
 339         self.assertEqual(1,
 340                           len(messages),
 341                           "Did not receive the expected number of messages")
 342
 343         audit = messages[0]["dsdbChange"]
 344         self.assertEqual("Delete", audit["operation"])
 345         self.assertFalse(audit["performedAsSystem"])
 346         self.assertTrue(dn.lower(), audit["dn"].lower())
 347         self.assertRegexpMatches(audit["remoteAddress"],
 348                                  self.remoteAddress)
 349         self.assertTrue(self.is_guid(audit["sessionId"]))
 350         self.assertEqual(0, audit["statusCode"])
 351         self.assertEqual("Success", audit["status"])
 352         session_id = self.get_session()
 353         self.assertEqual(session_id, audit["sessionId"])
 354         service_description = self.get_service_description()
 355         self.assertEqual(service_description, "LDAP")
 356
 357         transactionId = audit["transactionId"]
 358         message = self.waitForTransaction(transactionId)
 359         audit = message["dsdbTransaction"]
 360         self.assertEqual("commit", audit["action"])
 361         self.assertTrue(self.is_guid(audit["transactionId"]))
 362         self.assertTrue(audit["duration"] > 0)
 363
 364     def test_samdb_delete_non_existent_dn(self):
 365
 366         DOES_NOT_EXIST = "doesNotExist"
 367         dn = "cn=" + USER_NAME + ",cn=users," + self.base_dn
 368         self.discardSetupMessages(dn)
 369
 370         dn = "cn=" + DOES_NOT_EXIST + ",cn=users," + self.base_dn
 371         try:
 372             self.ldb.delete(dn)
 373             self.fail("Exception not thrown")
 374         except Exception:
 375             pass
 376
 377         messages = self.waitForMessages(1)
 378         print("Received %d messages" % len(messages))
 379         self.assertEqual(1,
 380                           len(messages),
 381                           "Did not receive the expected number of messages")
 382
 383         audit = messages[0]["dsdbChange"]
 384         self.assertEqual("Delete", audit["operation"])
 385         self.assertFalse(audit["performedAsSystem"])
 386         self.assertTrue(dn.lower(), audit["dn"].lower())
 387         self.assertRegexpMatches(audit["remoteAddress"],
 388                                  self.remoteAddress)
 389         self.assertEqual(ERR_NO_SUCH_OBJECT, audit["statusCode"])
 390         self.assertEqual("No such object", audit["status"])
 391         self.assertTrue(self.is_guid(audit["sessionId"]))
 392         session_id = self.get_session()
 393         self.assertEqual(session_id, audit["sessionId"])
 394         service_description = self.get_service_description()
 395         self.assertEqual(service_description, "LDAP")
 396
 397         transactionId = audit["transactionId"]
 398         message = self.waitForTransaction(transactionId)
 399         audit = message["dsdbTransaction"]
 400         self.assertEqual("rollback", audit["action"])
 401         self.assertTrue(self.is_guid(audit["transactionId"]))
 402         self.assertTrue(audit["duration"] > 0)
 403
 404     def test_create_and_delete_secret_over_lsa(self):
 405
 406         dn = "cn=Test Secret,CN=System," + self.base_dn
 407         self.discardSetupMessages(dn)
 408
 409         creds = self.insta_creds(template=self.get_credentials())
 410         lsa_conn = lsa.lsarpc(
 411             "ncacn_np:%s" % self.server,
 412             self.get_loadparm(),
 413             creds)
 414         lsa_handle = lsa_conn.OpenPolicy2(
 415             system_name="\\",
 416             attr=lsa.ObjectAttribute(),
 417             access_mask=security.SEC_FLAG_MAXIMUM_ALLOWED)
 418         secret_name = lsa.String()
 419         secret_name.string = "G$Test"
 420         lsa_conn.CreateSecret(
 421             handle=lsa_handle,
 422             name=secret_name,
 423             access_mask=security.SEC_FLAG_MAXIMUM_ALLOWED)
 424
 425         messages = self.waitForMessages(1, dn=dn)
 426         print("Received %d messages" % len(messages))
 427         self.assertEqual(1,
 428                           len(messages),
 429                           "Did not receive the expected number of messages")
 430
 431         audit = messages[0]["dsdbChange"]
 432         self.assertEqual("Add", audit["operation"])
 433         self.assertTrue(audit["performedAsSystem"])
 434         self.assertTrue(dn.lower(), audit["dn"].lower())
 435         self.assertRegexpMatches(audit["remoteAddress"],
 436                                  self.remoteAddress)
 437         self.assertTrue(self.is_guid(audit["sessionId"]))
 438         session_id = self.get_session()
 439         self.assertEqual(session_id, audit["sessionId"])
 440
 441         # We skip the check for self.get_service_description() as this
 442         # is subject to a race between smbd and the s4 rpc_server code
 443         # as to which will set the description as it is DCE/RPC over SMB
 444
 445         attributes = audit["attributes"]
 446         self.assertEqual(2, len(attributes))
 447
 448         object_class = attributes["objectClass"]
 449         self.assertEqual(1, len(object_class["actions"]))
 450         action = object_class["actions"][0]
 451         self.assertEqual("add", action["action"])
 452         values = action["values"]
 453         self.assertEqual(1, len(values))
 454         self.assertEqual("secret", values[0]["value"])
 455
 456         cn = attributes["cn"]
 457         self.assertEqual(1, len(cn["actions"]))
 458         action = cn["actions"][0]
 459         self.assertEqual("add", action["action"])
 460         values = action["values"]
 461         self.assertEqual(1, len(values))
 462         self.assertEqual("Test Secret", values[0]["value"])
 463
 464         #
 465         # Now delete the secret.
 466         self.discardMessages()
 467         h = lsa_conn.OpenSecret(
 468             handle=lsa_handle,
 469             name=secret_name,
 470             access_mask=security.SEC_FLAG_MAXIMUM_ALLOWED)
 471
 472         lsa_conn.DeleteObject(h)
 473         messages = self.waitForMessages(1, dn=dn)
 474         print("Received %d messages" % len(messages))
 475         self.assertEqual(1,
 476                           len(messages),
 477                           "Did not receive the expected number of messages")
 478
 479         dn = "cn=Test Secret,CN=System," + self.base_dn
 480         audit = messages[0]["dsdbChange"]
 481         self.assertEqual("Delete", audit["operation"])
 482         self.assertTrue(audit["performedAsSystem"])
 483         self.assertTrue(dn.lower(), audit["dn"].lower())
 484         self.assertRegexpMatches(audit["remoteAddress"],
 485                                  self.remoteAddress)
 486         self.assertTrue(self.is_guid(audit["sessionId"]))
 487         session_id = self.get_session()
 488         self.assertEqual(session_id, audit["sessionId"])
 489
 490         # We skip the check for self.get_service_description() as this
 491         # is subject to a race between smbd and the s4 rpc_server code
 492         # as to which will set the description as it is DCE/RPC over SMB
 493
 494     def test_modify(self):
 495
 496         dn = "cn=" + USER_NAME + ",cn=users," + self.base_dn
 497         self.discardSetupMessages(dn)
 498
 499         #
 500         # Add an attribute value
 501         #
 502         self.ldb.modify_ldif(
 503             "dn: " + dn + "\n" +
 504             "changetype: modify\n" +
 505             "add: carLicense\n" +
 506             "carLicense: license-01\n")
 507
 508         messages = self.waitForMessages(1, dn=dn)
 509         print("Received %d messages" % len(messages))
 510         self.assertEqual(1,
 511                           len(messages),
 512                           "Did not receive the expected number of messages")
 513
 514         audit = messages[0]["dsdbChange"]
 515         self.assertEqual("Modify", audit["operation"])
 516         self.assertFalse(audit["performedAsSystem"])
 517         self.assertEqual(dn, audit["dn"])
 518         self.assertRegexpMatches(audit["remoteAddress"],
 519                                  self.remoteAddress)
 520         self.assertTrue(self.is_guid(audit["sessionId"]))
 521         session_id = self.get_session()
 522         self.assertEqual(session_id, audit["sessionId"])
 523         service_description = self.get_service_description()
 524         self.assertEqual(service_description, "LDAP")
 525
 526         attributes = audit["attributes"]
 527         self.assertEqual(1, len(attributes))
 528         actions = attributes["carLicense"]["actions"]
 529         self.assertEqual(1, len(actions))
 530         self.assertEqual("add", actions[0]["action"])
 531         values = actions[0]["values"]
 532         self.assertEqual(1, len(values))
 533         self.assertEqual("license-01", values[0]["value"])
 534
 535         #
 536         # Add an another value to the attribute
 537         #
 538         self.discardMessages()
 539         self.ldb.modify_ldif(
 540             "dn: " + dn + "\n" +
 541             "changetype: modify\n" +
 542             "add: carLicense\n" +
 543             "carLicense: license-02\n")
 544
 545         messages = self.waitForMessages(1, dn=dn)
 546         print("Received %d messages" % len(messages))
 547         self.assertEqual(1,
 548                           len(messages),
 549                           "Did not receive the expected number of messages")
 550         attributes = messages[0]["dsdbChange"]["attributes"]
 551         self.assertEqual(1, len(attributes))
 552         actions = attributes["carLicense"]["actions"]
 553         self.assertEqual(1, len(actions))
 554         self.assertEqual("add", actions[0]["action"])
 555         values = actions[0]["values"]
 556         self.assertEqual(1, len(values))
 557         self.assertEqual("license-02", values[0]["value"])
 558
 559         #
 560         # Add an another two values to the attribute
 561         #
 562         self.discardMessages()
 563         self.ldb.modify_ldif(
 564             "dn: " + dn + "\n" +
 565             "changetype: modify\n" +
 566             "add: carLicense\n" +
 567             "carLicense: license-03\n" +
 568             "carLicense: license-04\n")
 569
 570         messages = self.waitForMessages(1, dn=dn)
 571         print("Received %d messages" % len(messages))
 572         self.assertEqual(1,
 573                           len(messages),
 574                           "Did not receive the expected number of messages")
 575         attributes = messages[0]["dsdbChange"]["attributes"]
 576         self.assertEqual(1, len(attributes))
 577         actions = attributes["carLicense"]["actions"]
 578         self.assertEqual(1, len(actions))
 579         self.assertEqual("add", actions[0]["action"])
 580         values = actions[0]["values"]
 581         self.assertEqual(2, len(values))
 582         self.assertEqual("license-03", values[0]["value"])
 583         self.assertEqual("license-04", values[1]["value"])
 584
 585         #
 586         # delete two values to the attribute
 587         #
 588         self.discardMessages()
 589         self.ldb.modify_ldif(
 590             "dn: " + dn + "\n" +
 591             "changetype: delete\n" +
 592             "delete: carLicense\n" +
 593             "carLicense: license-03\n" +
 594             "carLicense: license-04\n")
 595
 596         messages = self.waitForMessages(1, dn=dn)
 597         print("Received %d messages" % len(messages))
 598         self.assertEqual(1,
 599                           len(messages),
 600                           "Did not receive the expected number of messages")
 601         attributes = messages[0]["dsdbChange"]["attributes"]
 602         self.assertEqual(1, len(attributes))
 603         actions = attributes["carLicense"]["actions"]
 604         self.assertEqual(1, len(actions))
 605         self.assertEqual("delete", actions[0]["action"])
 606         values = actions[0]["values"]
 607         self.assertEqual(2, len(values))
 608         self.assertEqual("license-03", values[0]["value"])
 609         self.assertEqual("license-04", values[1]["value"])
 610
 611         #
 612         # replace two values to the attribute
 613         #
 614         self.discardMessages()
 615         self.ldb.modify_ldif(
 616             "dn: " + dn + "\n" +
 617             "changetype: delete\n" +
 618             "replace: carLicense\n" +
 619             "carLicense: license-05\n" +
 620             "carLicense: license-06\n")
 621
 622         messages = self.waitForMessages(1, dn=dn)
 623         print("Received %d messages" % len(messages))
 624         self.assertEqual(1,
 625                           len(messages),
 626                           "Did not receive the expected number of messages")
 627         attributes = messages[0]["dsdbChange"]["attributes"]
 628         self.assertEqual(1, len(attributes))
 629         actions = attributes["carLicense"]["actions"]
 630         self.assertEqual(1, len(actions))
 631         self.assertEqual("replace", actions[0]["action"])
 632         values = actions[0]["values"]
 633         self.assertEqual(2, len(values))
 634         self.assertEqual("license-05", values[0]["value"])
 635         self.assertEqual("license-06", values[1]["value"])