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