source4/torture/drs/python/repl_schema.py

   1 #!/usr/bin/env python3
   2 # -*- coding: utf-8 -*-
   3 #
   4 # Tests various schema replication scenarios
   5 #
   6 # Copyright (C) Kamen Mazdrashki <kamenim@samba.org> 2010
   7 #
   8 # This program is free software; you can redistribute it and/or modify
   9 # it under the terms of the GNU General Public License as published by
  10 # the Free Software Foundation; either version 3 of the License, or
  11 # (at your option) any later version.
  12 #
  13 # This program is distributed in the hope that it will be useful,
  14 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  15 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  16 # GNU General Public License for more details.
  17 #
  18 # You should have received a copy of the GNU General Public License
  19 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
  20 #
  21
  22 #
  23 # Usage:
  24 #  export DC1=dc1_dns_name
  25 #  export DC2=dc2_dns_name
  26 #  export SUBUNITRUN=$samba4srcdir/scripting/bin/subunitrun
  27 #  PYTHONPATH="$PYTHONPATH:$samba4srcdir/torture/drs/python" $SUBUNITRUN repl_schema -U"$DOMAIN/$DC_USERNAME"%"$DC_PASSWORD"
  28 #
  29
  30 import time
  31 import random
  32 import ldb
  33 import drs_base
  34
  35 from ldb import (
  36     ERR_NO_SUCH_OBJECT,
  37     LdbError,
  38     SCOPE_BASE,
  39     Message,
  40     FLAG_MOD_ADD,
  41     FLAG_MOD_REPLACE
  42 )
  43 from samba.dcerpc import drsuapi, misc
  44 from samba.drs_utils import drs_DsBind
  45 from samba import dsdb
  46
  47
  48 class DrsReplSchemaTestCase(drs_base.DrsBaseTestCase):
  49
  50     # prefix for all objects created
  51     obj_prefix = None
  52     # current Class or Attribute object id
  53     obj_id = 0
  54
  55     def _ds_bind(self, server_name):
  56         binding_str = "ncacn_ip_tcp:%s[seal]" % server_name
  57
  58         drs = drsuapi.drsuapi(binding_str, self.get_loadparm(), self.get_credentials())
  59         (drs_handle, supported_extensions) = drs_DsBind(drs)
  60         return (drs, drs_handle)
  61
  62     def _exop_req8(self, dest_dsa, invocation_id, nc_dn_str, exop,
  63                    replica_flags=0, max_objects=0):
  64         req8 = drsuapi.DsGetNCChangesRequest8()
  65
  66         req8.destination_dsa_guid = misc.GUID(dest_dsa) if dest_dsa else misc.GUID()
  67         req8.source_dsa_invocation_id = misc.GUID(invocation_id)
  68         req8.naming_context = drsuapi.DsReplicaObjectIdentifier()
  69         req8.naming_context.dn = str(nc_dn_str)
  70         req8.highwatermark = drsuapi.DsReplicaHighWaterMark()
  71         req8.highwatermark.tmp_highest_usn = 0
  72         req8.highwatermark.reserved_usn = 0
  73         req8.highwatermark.highest_usn = 0
  74         req8.uptodateness_vector = None
  75         req8.replica_flags = replica_flags
  76         req8.max_object_count = max_objects
  77         req8.max_ndr_size = 402116
  78         req8.extended_op = exop
  79         req8.fsmo_info = 0
  80         req8.partial_attribute_set = None
  81         req8.partial_attribute_set_ex = None
  82         req8.mapping_ctr.num_mappings = 0
  83         req8.mapping_ctr.mappings = None
  84
  85         return req8
  86
  87     def setUp(self):
  88         super(DrsReplSchemaTestCase, self).setUp()
  89
  90         # disable automatic replication temporary
  91         self._disable_all_repl(self.dnsname_dc1)
  92         self._disable_all_repl(self.dnsname_dc2)
  93
  94         # make sure DCs are synchronized before the test
  95         self._net_drs_replicate(DC=self.dnsname_dc2, fromDC=self.dnsname_dc1, forced=True)
  96         self._net_drs_replicate(DC=self.dnsname_dc1, fromDC=self.dnsname_dc2, forced=True)
  97         # initialize objects prefix if not done yet
  98         if self.obj_prefix is None:
  99             t = time.strftime("%s", time.gmtime())
 100             DrsReplSchemaTestCase.obj_prefix = "DrsReplSchema-%s" % t
 101
 102     def tearDown(self):
 103         self._enable_all_repl(self.dnsname_dc1)
 104         self._enable_all_repl(self.dnsname_dc2)
 105         super(DrsReplSchemaTestCase, self).tearDown()
 106
 107     def _make_obj_names(self, base_name):
 108         '''Try to create a unique name for an object
 109            that is to be added to schema'''
 110         self.obj_id += 1
 111         obj_name = "%s-%d-%s" % (self.obj_prefix, self.obj_id, base_name)
 112         obj_ldn = obj_name.replace("-", "")
 113         obj_dn = ldb.Dn(self.ldb_dc1, "CN=X")
 114         obj_dn.add_base(ldb.Dn(self.ldb_dc1, self.schema_dn))
 115         obj_dn.set_component(0, "CN", obj_name)
 116         return (obj_dn, obj_name, obj_ldn)
 117
 118     def _schema_new_class(self, ldb_ctx, base_name, base_int, oc_cat=1, attrs=None):
 119         (class_dn, class_name, class_ldn) = self._make_obj_names(base_name)
 120         rec = {"dn": class_dn,
 121                "objectClass": ["top", "classSchema"],
 122                "cn": class_name,
 123                "lDAPDisplayName": class_ldn,
 124                "governsId": "1.3.6.1.4.1.7165.4.6.2.5."
 125                + str((100000 * base_int) + random.randint(1, 100000)) + ".1.5.13",
 126                "instanceType": "4",
 127                "objectClassCategory": "%d" % oc_cat,
 128                "subClassOf": "top",
 129                "systemOnly": "FALSE"}
 130         # allow overriding/adding attributes
 131         if attrs is not None:
 132             rec.update(attrs)
 133         # add it to the Schema
 134         try:
 135             ldb_ctx.add(rec)
 136         except LdbError as e:
 137             (enum, estr) = e.args
 138             self.fail("Adding record failed with %d/%s" % (enum, estr))
 139
 140         self._ldap_schemaUpdateNow(ldb_ctx)
 141         return (rec["lDAPDisplayName"], rec["dn"])
 142
 143     def _schema_new_attr(self, ldb_ctx, base_name, base_int, attrs=None):
 144         (attr_dn, attr_name, attr_ldn) = self._make_obj_names(base_name)
 145         rec = {"dn": attr_dn,
 146                "objectClass": ["top", "attributeSchema"],
 147                "cn": attr_name,
 148                "lDAPDisplayName": attr_ldn,
 149                "attributeId": "1.3.6.1.4.1.7165.4.6.1.5."
 150                + str((100000 * base_int) + random.randint(1, 100000)) + ".1.5.13",
 151                "attributeSyntax": "2.5.5.12",
 152                "omSyntax": "64",
 153                "instanceType": "4",
 154                "isSingleValued": "TRUE",
 155                "systemOnly": "FALSE"}
 156         # allow overriding/adding attributes
 157         if attrs is not None:
 158             rec.update(attrs)
 159         # add it to the Schema
 160         ldb_ctx.add(rec)
 161         self._ldap_schemaUpdateNow(ldb_ctx)
 162         return (rec["lDAPDisplayName"], rec["dn"])
 163
 164     def _check_object(self, obj_dn):
 165         '''Check if object obj_dn exists on both DCs'''
 166         res_dc1 = self.ldb_dc1.search(base=obj_dn,
 167                                       scope=SCOPE_BASE,
 168                                       attrs=["*"])
 169         self.assertEquals(len(res_dc1), 1,
 170                           "%s doesn't exists on %s" % (obj_dn, self.dnsname_dc1))
 171         try:
 172             res_dc2 = self.ldb_dc2.search(base=obj_dn,
 173                                           scope=SCOPE_BASE,
 174                                           attrs=["*"])
 175         except LdbError as e1:
 176             (enum, estr) = e1.args
 177             if enum == ERR_NO_SUCH_OBJECT:
 178                 self.fail("%s doesn't exists on %s" % (obj_dn, self.dnsname_dc2))
 179             raise
 180         self.assertEquals(len(res_dc2), 1,
 181                           "%s doesn't exists on %s" % (obj_dn, self.dnsname_dc2))
 182
 183     def test_class(self):
 184         """Simple test for classSchema replication"""
 185         # add new classSchema object
 186         (c_ldn, c_dn) = self._schema_new_class(self.ldb_dc1, "cls-S", 0)
 187         # force replication from DC1 to DC2
 188         self._net_drs_replicate(DC=self.dnsname_dc2, fromDC=self.dnsname_dc1, nc_dn=self.schema_dn, forced=True)
 189         # check object is replicated
 190         self._check_object(c_dn)
 191
 192     def test_classInheritance(self):
 193         """Test inheritance through subClassOf
 194            I think 5 levels of inheritance is pretty decent for now."""
 195         # add 5 levels deep hierarchy
 196         c_dn_list = []
 197         c_ldn_last = None
 198         for i in range(1, 6):
 199             base_name = "cls-I-%02d" % i
 200             (c_ldn, c_dn) = self._schema_new_class(self.ldb_dc1, base_name, i)
 201             c_dn_list.append(c_dn)
 202             if c_ldn_last:
 203                 # inherit from last class added
 204                 m = Message.from_dict(self.ldb_dc1,
 205                                       {"dn": c_dn,
 206                                        "subClassOf": c_ldn_last},
 207                                       FLAG_MOD_REPLACE)
 208                 self.ldb_dc1.modify(m)
 209             # store last class ldapDisplayName
 210             c_ldn_last = c_ldn
 211         # force replication from DC1 to DC2
 212         self._net_drs_replicate(DC=self.dnsname_dc2, fromDC=self.dnsname_dc1, nc_dn=self.schema_dn, forced=True)
 213         # check objects are replicated
 214         for c_dn in c_dn_list:
 215             self._check_object(c_dn)
 216
 217     def test_classWithCustomAttribute(self):
 218         """Create new Attribute and a Class,
 219            that has value for newly created attribute.
 220            This should check code path that searches for
 221            AttributeID_id in Schema cache"""
 222         # add new attributeSchema object
 223         (a_ldn, a_dn) = self._schema_new_attr(self.ldb_dc1, "attr-A", 7)
 224         # add a base classSchema class so we can use our new
 225         # attribute in class definition in a sibling class
 226         (c_ldn, c_dn) = self._schema_new_class(self.ldb_dc1, "cls-A", 8,
 227                                                1,
 228                                                {"systemMayContain": a_ldn,
 229                                                 "subClassOf": "classSchema"})
 230         # add new classSchema object with value for a_ldb attribute
 231         (c_ldn, c_dn) = self._schema_new_class(self.ldb_dc1, "cls-B", 9,
 232                                                1,
 233                                                {"objectClass": ["top", "classSchema", c_ldn],
 234                                                 a_ldn: "test_classWithCustomAttribute"})
 235         # force replication from DC1 to DC2
 236         self._net_drs_replicate(DC=self.dnsname_dc2, fromDC=self.dnsname_dc1, nc_dn=self.schema_dn, forced=True)
 237         # check objects are replicated
 238         self._check_object(c_dn)
 239         self._check_object(a_dn)
 240
 241     def test_classWithCustomLinkAttribute(self):
 242         """Create new Attribute and a Class,
 243            that has value for newly created attribute.
 244            This should check code path that searches for
 245            AttributeID_id in Schema cache"""
 246         # add new attributeSchema object
 247         (a_ldn, a_dn) = self._schema_new_attr(self.ldb_dc1, "attr-Link-X", 10,
 248                                               attrs={'linkID': "1.2.840.113556.1.2.50",
 249                                                      "attributeSyntax": "2.5.5.1",
 250                                                      "omSyntax": "127"})
 251         # add a base classSchema class so we can use our new
 252         # attribute in class definition in a sibling class
 253         (c_ldn, c_dn) = self._schema_new_class(self.ldb_dc1, "cls-Link-Y", 11,
 254                                                1,
 255                                                {"systemMayContain": a_ldn,
 256                                                 "subClassOf": "classSchema"})
 257         # add new classSchema object with value for a_ldb attribute
 258         (c_ldn, c_dn) = self._schema_new_class(self.ldb_dc1, "cls-Link-Z", 12,
 259                                                1,
 260                                                {"objectClass": ["top", "classSchema", c_ldn],
 261                                                 a_ldn: self.schema_dn})
 262         # force replication from DC1 to DC2
 263         self._net_drs_replicate(DC=self.dnsname_dc2, fromDC=self.dnsname_dc1, nc_dn=self.schema_dn, forced=True)
 264         # check objects are replicated
 265         self._check_object(c_dn)
 266         self._check_object(a_dn)
 267
 268         res = self.ldb_dc1.search(base="",
 269                                   scope=SCOPE_BASE,
 270                                   attrs=["domainFunctionality"])
 271
 272         if int(res[0]["domainFunctionality"][0]) > dsdb.DS_DOMAIN_FUNCTION_2000:
 273             res = self.ldb_dc1.search(base=a_dn,
 274                                       scope=SCOPE_BASE,
 275                                       attrs=["msDS-IntId"])
 276             self.assertEqual(1, len(res))
 277             self.assertTrue("msDS-IntId" in res[0])
 278             int_id = int(res[0]["msDS-IntId"][0])
 279             if int_id < 0:
 280                 int_id += (1 << 32)
 281
 282         dc_guid_1 = self.ldb_dc1.get_invocation_id()
 283
 284         drs, drs_handle = self._ds_bind(self.dnsname_dc1)
 285
 286         req8 = self._exop_req8(dest_dsa=None,
 287                                invocation_id=dc_guid_1,
 288                                nc_dn_str=c_dn,
 289                                exop=drsuapi.DRSUAPI_EXOP_REPL_OBJ,
 290                                replica_flags=drsuapi.DRSUAPI_DRS_SYNC_FORCED)
 291
 292         (level, ctr) = drs.DsGetNCChanges(drs_handle, 8, req8)
 293
 294         for link in ctr.linked_attributes:
 295             self.assertTrue(link.attid != int_id,
 296                             'Got %d for both' % link.attid)
 297
 298     def test_attribute(self):
 299         """Simple test for attributeSchema replication"""
 300         # add new attributeSchema object
 301         (a_ldn, a_dn) = self._schema_new_attr(self.ldb_dc1, "attr-S", 13)
 302         # force replication from DC1 to DC2
 303         self._net_drs_replicate(DC=self.dnsname_dc2, fromDC=self.dnsname_dc1, nc_dn=self.schema_dn, forced=True)
 304         # check object is replicated
 305         self._check_object(a_dn)
 306
 307     def test_attribute_on_ou(self):
 308         """Simple test having an OU with a custome attribute replicated correctly
 309
 310         This ensures that the server
 311         """
 312
 313        # add new attributeSchema object
 314         (a_ldn, a_dn) = self._schema_new_attr(self.ldb_dc1, "attr-OU-S", 14)
 315         (c_ldn, c_dn) = self._schema_new_class(self.ldb_dc1, "cls-OU-A", 15,
 316                                                3,
 317                                                {"mayContain": a_ldn})
 318         ou_dn = ldb.Dn(self.ldb_dc1, "ou=X")
 319         ou_dn.add_base(self.ldb_dc1.get_default_basedn())
 320         ou_dn.set_component(0, "OU", a_dn.get_component_value(0))
 321         rec = {"dn": ou_dn,
 322                "objectClass": ["top", "organizationalUnit", c_ldn],
 323                "ou": ou_dn.get_component_value(0),
 324                a_ldn: "test OU"}
 325         self.ldb_dc1.add(rec)
 326
 327         # force replication from DC1 to DC2
 328         self._net_drs_replicate(DC=self.dnsname_dc2, fromDC=self.dnsname_dc1, nc_dn=self.domain_dn, forced=True)
 329         # check objects are replicated
 330         self._check_object(c_dn)
 331         self._check_object(a_dn)
 332         self._check_object(ou_dn)
 333         self.ldb_dc1.delete(ou_dn)
 334
 335     def test_all(self):
 336         """Basic plan is to create bunch of classSchema
 337            and attributeSchema objects, replicate Schema NC
 338            and then check all objects are replicated correctly"""
 339
 340         # add new classSchema object
 341         (c_ldn, c_dn) = self._schema_new_class(self.ldb_dc1, "cls-A", 16)
 342         # add new attributeSchema object
 343         (a_ldn, a_dn) = self._schema_new_attr(self.ldb_dc1, "attr-A", 17)
 344
 345         # add attribute to the class we have
 346         m = Message.from_dict(self.ldb_dc1,
 347                               {"dn": c_dn,
 348                                "mayContain": a_ldn},
 349                               FLAG_MOD_ADD)
 350         self.ldb_dc1.modify(m)
 351
 352         # force replication from DC1 to DC2
 353         self._net_drs_replicate(DC=self.dnsname_dc2, fromDC=self.dnsname_dc1, nc_dn=self.schema_dn, forced=True)
 354
 355         # check objects are replicated
 356         self._check_object(c_dn)
 357         self._check_object(a_dn)
 358
 359     def test_classWithCustomBinaryDNLinkAttribute(self):
 360         # Add a new attribute to the schema, which has binary DN syntax (2.5.5.7)
 361         (bin_ldn, bin_dn) = self._schema_new_attr(self.ldb_dc1, "attr-Link-Bin", 18,
 362                                                   attrs={"linkID": "1.2.840.113556.1.2.50",
 363                                                          "attributeSyntax": "2.5.5.7",
 364                                                          "omSyntax": "127"})
 365
 366         (bin_ldn_b, bin_dn_b) = self._schema_new_attr(self.ldb_dc1, "attr-Link-Bin-Back", 19,
 367                                                       attrs={"linkID": bin_ldn,
 368                                                              "attributeSyntax": "2.5.5.1",
 369                                                              "omSyntax": "127"})
 370
 371         # Add a new class to the schema which can have the binary DN attribute
 372         (c_ldn, c_dn) = self._schema_new_class(self.ldb_dc1, "cls-Link-Bin", 20,
 373                                                3,
 374                                                {"mayContain": bin_ldn})
 375         (c_ldn_b, c_dn_b) = self._schema_new_class(self.ldb_dc1, "cls-Link-Bin-Back", 21,
 376                                                    3,
 377                                                    {"mayContain": bin_ldn_b})
 378
 379         link_end_dn = ldb.Dn(self.ldb_dc1, "ou=X")
 380         link_end_dn.add_base(self.ldb_dc1.get_default_basedn())
 381         link_end_dn.set_component(0, "OU", bin_dn_b.get_component_value(0))
 382
 383         ou_dn = ldb.Dn(self.ldb_dc1, "ou=X")
 384         ou_dn.add_base(self.ldb_dc1.get_default_basedn())
 385         ou_dn.set_component(0, "OU", bin_dn.get_component_value(0))
 386
 387         # Add an instance of the class to be pointed at
 388         rec = {"dn": link_end_dn,
 389                "objectClass": ["top", "organizationalUnit", c_ldn_b],
 390                "ou": link_end_dn.get_component_value(0)}
 391         self.ldb_dc1.add(rec)
 392
 393         # .. and one that does, and points to the first one
 394         rec = {"dn": ou_dn,
 395                "objectClass": ["top", "organizationalUnit", c_ldn],
 396                "ou": ou_dn.get_component_value(0)}
 397         self.ldb_dc1.add(rec)
 398
 399         m = Message.from_dict(self.ldb_dc1,
 400                               {"dn": ou_dn,
 401                                bin_ldn: "B:8:1234ABCD:%s" % str(link_end_dn)},
 402                               FLAG_MOD_ADD)
 403         self.ldb_dc1.modify(m)
 404
 405         self._net_drs_replicate(DC=self.dnsname_dc2, fromDC=self.dnsname_dc1,
 406                                 nc_dn=self.schema_dn, forced=True)
 407         self._net_drs_replicate(DC=self.dnsname_dc2, fromDC=self.dnsname_dc1,
 408                                 nc_dn=self.domain_dn, forced=True)
 409
 410         self._check_object(c_dn)
 411         self._check_object(bin_dn)
 412
 413         # Make sure we can delete the backlink
 414         self.ldb_dc1.delete(link_end_dn)
 415
 416         self._net_drs_replicate(DC=self.dnsname_dc2, fromDC=self.dnsname_dc1,
 417                                 nc_dn=self.schema_dn, forced=True)
 418         self._net_drs_replicate(DC=self.dnsname_dc2, fromDC=self.dnsname_dc1,
 419                                 nc_dn=self.domain_dn, forced=True)
 420
 421     def test_rename(self):
 422         """Basic plan is to create a classSchema
 423            and attributeSchema objects, replicate Schema NC
 424            and then check all objects are replicated correctly"""
 425
 426         # add new classSchema object
 427         (c_ldn, c_dn) = self._schema_new_class(self.ldb_dc1, "cls-B", 20)
 428
 429         self._net_drs_replicate(DC=self.dnsname_dc2, fromDC=self.dnsname_dc1,
 430                                 nc_dn=self.schema_dn, forced=True)
 431
 432         # check objects are replicated
 433         self._check_object(c_dn)
 434
 435         # rename the Class CN
 436         c_dn_new = ldb.Dn(self.ldb_dc1, str(c_dn))
 437         c_dn_new.set_component(0,
 438                                "CN",
 439                                c_dn.get_component_value(0) + "-NEW")
 440         try:
 441             self.ldb_dc1.rename(c_dn, c_dn_new)
 442         except LdbError as e2:
 443             (num, _) = e2.args
 444             self.fail("failed to change CN for %s: %s" % (c_dn, _))
 445
 446         # force replication from DC1 to DC2
 447         self._net_drs_replicate(DC=self.dnsname_dc2, fromDC=self.dnsname_dc1,
 448                                 nc_dn=self.schema_dn, forced=True)
 449
 450         # check objects are replicated
 451         self._check_object(c_dn_new)