source4/torture/drs/python/repl_schema.py

   1 #!/usr/bin/env python
   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
  33 from ldb import (
  34     ERR_NO_SUCH_OBJECT,
  35     LdbError,
  36     SCOPE_BASE,
  37     Message,
  38     FLAG_MOD_ADD,
  39     FLAG_MOD_REPLACE,
  40     )
  41 import ldb
  42
  43 import drs_base
  44 from samba.dcerpc import drsuapi, misc
  45 from samba.drs_utils import drs_DsBind
  46 from samba import dsdb
  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 = unicode(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 not attrs is None:
 132             rec.update(attrs)
 133         # add it to the Schema
 134         try:
 135             ldb_ctx.add(rec)
 136         except LdbError, (enum, estr):
 137             self.fail("Adding record failed with %d/%s" % (enum, estr))
 138
 139         self._ldap_schemaUpdateNow(ldb_ctx)
 140         return (rec["lDAPDisplayName"], rec["dn"])
 141
 142     def _schema_new_attr(self, ldb_ctx, base_name, base_int, attrs=None):
 143         (attr_dn, attr_name, attr_ldn) = self._make_obj_names(base_name)
 144         rec = {"dn": attr_dn,
 145                "objectClass": ["top", "attributeSchema"],
 146                "cn": attr_name,
 147                "lDAPDisplayName": attr_ldn,
 148                "attributeId": "1.3.6.1.4.1.7165.4.6.1.5." \
 149                 + str((100000 * base_int) + random.randint(1,100000)) + ".1.5.13",
 150                "attributeSyntax": "2.5.5.12",
 151                "omSyntax": "64",
 152                "instanceType": "4",
 153                "isSingleValued": "TRUE",
 154                "systemOnly": "FALSE"}
 155         # allow overriding/adding attributes
 156         if not attrs is None:
 157             rec.update(attrs)
 158         # add it to the Schema
 159         ldb_ctx.add(rec)
 160         self._ldap_schemaUpdateNow(ldb_ctx)
 161         return (rec["lDAPDisplayName"], rec["dn"])
 162
 163     def _check_object(self, obj_dn):
 164         '''Check if object obj_dn exists on both DCs'''
 165         res_dc1 = self.ldb_dc1.search(base=obj_dn,
 166                                       scope=SCOPE_BASE,
 167                                       attrs=["*"])
 168         self.assertEquals(len(res_dc1), 1,
 169                           "%s doesn't exists on %s" % (obj_dn, self.dnsname_dc1))
 170         try:
 171             res_dc2 = self.ldb_dc2.search(base=obj_dn,
 172                                           scope=SCOPE_BASE,
 173                                           attrs=["*"])
 174         except LdbError, (enum, estr):
 175             if enum == ERR_NO_SUCH_OBJECT:
 176                 self.fail("%s doesn't exists on %s" % (obj_dn, self.dnsname_dc2))
 177             raise
 178         self.assertEquals(len(res_dc2), 1,
 179                           "%s doesn't exists on %s" % (obj_dn, self.dnsname_dc2))
 180
 181     def test_class(self):
 182         """Simple test for classSchema replication"""
 183         # add new classSchema object
 184         (c_ldn, c_dn) = self._schema_new_class(self.ldb_dc1, "cls-S", 0)
 185         # force replication from DC1 to DC2
 186         self._net_drs_replicate(DC=self.dnsname_dc2, fromDC=self.dnsname_dc1, nc_dn=self.schema_dn, forced=True)
 187         # check object is replicated
 188         self._check_object(c_dn)
 189
 190     def test_classInheritance(self):
 191         """Test inheritance through subClassOf
 192            I think 5 levels of inheritance is pretty decent for now."""
 193         # add 5 levels deep hierarchy
 194         c_dn_list = []
 195         c_ldn_last = None
 196         for i in range(1, 6):
 197             base_name = "cls-I-%02d" % i
 198             (c_ldn, c_dn) = self._schema_new_class(self.ldb_dc1, base_name, i)
 199             c_dn_list.append(c_dn)
 200             if c_ldn_last:
 201                 # inherit from last class added
 202                 m = Message.from_dict(self.ldb_dc1,
 203                                       {"dn": c_dn,
 204                                        "subClassOf": c_ldn_last},
 205                                       FLAG_MOD_REPLACE)
 206                 self.ldb_dc1.modify(m)
 207             # store last class ldapDisplayName
 208             c_ldn_last = c_ldn
 209         # force replication from DC1 to DC2
 210         self._net_drs_replicate(DC=self.dnsname_dc2, fromDC=self.dnsname_dc1, nc_dn=self.schema_dn, forced=True)
 211         # check objects are replicated
 212         for c_dn in c_dn_list:
 213             self._check_object(c_dn)
 214
 215     def test_classWithCustomAttribute(self):
 216         """Create new Attribute and a Class,
 217            that has value for newly created attribute.
 218            This should check code path that searches for
 219            AttributeID_id in Schema cache"""
 220         # add new attributeSchema object
 221         (a_ldn, a_dn) = self._schema_new_attr(self.ldb_dc1, "attr-A", 7)
 222         # add a base classSchema class so we can use our new
 223         # attribute in class definition in a sibling class
 224         (c_ldn, c_dn) = self._schema_new_class(self.ldb_dc1, "cls-A", 8,
 225                                                1,
 226                                                {"systemMayContain": a_ldn,
 227                                                 "subClassOf": "classSchema"})
 228         # add new classSchema object with value for a_ldb attribute
 229         (c_ldn, c_dn) = self._schema_new_class(self.ldb_dc1, "cls-B", 9,
 230                                                1,
 231                                                {"objectClass": ["top", "classSchema", c_ldn],
 232                                                 a_ldn: "test_classWithCustomAttribute"})
 233         # force replication from DC1 to DC2
 234         self._net_drs_replicate(DC=self.dnsname_dc2, fromDC=self.dnsname_dc1, nc_dn=self.schema_dn, forced=True)
 235         # check objects are replicated
 236         self._check_object(c_dn)
 237         self._check_object(a_dn)
 238
 239     def test_classWithCustomLinkAttribute(self):
 240         """Create new Attribute and a Class,
 241            that has value for newly created attribute.
 242            This should check code path that searches for
 243            AttributeID_id in Schema cache"""
 244         # add new attributeSchema object
 245         (a_ldn, a_dn) = self._schema_new_attr(self.ldb_dc1, "attr-Link-X", 10,
 246                                               attrs={'linkID':"1.2.840.113556.1.2.50",
 247                                                      "attributeSyntax": "2.5.5.1",
 248                                                      "omSyntax": "127"})
 249         # add a base classSchema class so we can use our new
 250         # attribute in class definition in a sibling class
 251         (c_ldn, c_dn) = self._schema_new_class(self.ldb_dc1, "cls-Link-Y", 11,
 252                                                1,
 253                                                {"systemMayContain": a_ldn,
 254                                                 "subClassOf": "classSchema"})
 255         # add new classSchema object with value for a_ldb attribute
 256         (c_ldn, c_dn) = self._schema_new_class(self.ldb_dc1, "cls-Link-Z", 12,
 257                                                1,
 258                                                {"objectClass": ["top", "classSchema", c_ldn],
 259                                                 a_ldn: self.schema_dn})
 260         # force replication from DC1 to DC2
 261         self._net_drs_replicate(DC=self.dnsname_dc2, fromDC=self.dnsname_dc1, nc_dn=self.schema_dn, forced=True)
 262         # check objects are replicated
 263         self._check_object(c_dn)
 264         self._check_object(a_dn)
 265
 266         res = self.ldb_dc1.search(base="",
 267                                   scope=SCOPE_BASE,
 268                                   attrs=["domainFunctionality"])
 269
 270         if int(res[0]["domainFunctionality"][0]) > dsdb.DS_DOMAIN_FUNCTION_2000:
 271             res = self.ldb_dc1.search(base=a_dn,
 272                                       scope=SCOPE_BASE,
 273                                       attrs=["msDS-IntId"])
 274             self.assertEqual(1, len(res))
 275             self.assertTrue("msDS-IntId" in res[0])
 276             int_id = int(res[0]["msDS-IntId"][0])
 277             if int_id < 0:
 278                 int_id += (1 << 32)
 279
 280         dc_guid_1 = self.ldb_dc1.get_invocation_id()
 281
 282         drs, drs_handle = self._ds_bind(self.dnsname_dc1)
 283
 284         req8 = self._exop_req8(dest_dsa=None,
 285                 invocation_id=dc_guid_1,
 286                 nc_dn_str=c_dn,
 287                 exop=drsuapi.DRSUAPI_EXOP_REPL_OBJ,
 288                 replica_flags=drsuapi.DRSUAPI_DRS_SYNC_FORCED)
 289
 290         (level, ctr) = drs.DsGetNCChanges(drs_handle, 8, req8)
 291
 292         for link in ctr.linked_attributes:
 293             self.assertTrue(link.attid != int_id,
 294                             'Got %d for both' % link.attid)
 295
 296     def test_attribute(self):
 297         """Simple test for attributeSchema replication"""
 298         # add new attributeSchema object
 299         (a_ldn, a_dn) = self._schema_new_attr(self.ldb_dc1, "attr-S", 13)
 300         # force replication from DC1 to DC2
 301         self._net_drs_replicate(DC=self.dnsname_dc2, fromDC=self.dnsname_dc1, nc_dn=self.schema_dn, forced=True)
 302         # check object is replicated
 303         self._check_object(a_dn)
 304
 305     def test_attribute_on_ou(self):
 306         """Simple test having an OU with a custome attribute replicated correctly
 307
 308         This ensures that the server
 309         """
 310         # disable automatic replication temporary
 311         self._disable_all_repl(self.dnsname_dc1)
 312         self._disable_all_repl(self.dnsname_dc2)
 313
 314        # add new attributeSchema object
 315         (a_ldn, a_dn) = self._schema_new_attr(self.ldb_dc1, "attr-OU-S", 14)
 316         (c_ldn, c_dn) = self._schema_new_class(self.ldb_dc1, "cls-OU-A", 15,
 317                                                3,
 318                                                {"mayContain": a_ldn})
 319         ou_dn = ldb.Dn(self.ldb_dc1, "ou=X")
 320         ou_dn.add_base(self.ldb_dc1.get_default_basedn())
 321         ou_dn.set_component(0, "OU", a_dn.get_component_value(0))
 322         rec = {"dn": ou_dn,
 323                "objectClass": ["top", "organizationalUnit", c_ldn],
 324                "ou": ou_dn.get_component_value(0),
 325                a_ldn: "test OU"}
 326         self.ldb_dc1.add(rec)
 327
 328         # force replication from DC1 to DC2
 329         self._net_drs_replicate(DC=self.dnsname_dc2, fromDC=self.dnsname_dc1, nc_dn=self.domain_dn, forced=True)
 330         # check objects are replicated
 331         self._check_object(c_dn)
 332         self._check_object(a_dn)
 333         self._check_object(ou_dn)
 334         self.ldb_dc1.delete(ou_dn)
 335
 336     def test_all(self):
 337         """Basic plan is to create bunch of classSchema
 338            and attributeSchema objects, replicate Schema NC
 339            and then check all objects are replicated correctly"""
 340
 341         # add new classSchema object
 342         (c_ldn, c_dn) = self._schema_new_class(self.ldb_dc1, "cls-A", 16)
 343         # add new attributeSchema object
 344         (a_ldn, a_dn) = self._schema_new_attr(self.ldb_dc1, "attr-A", 17)
 345
 346         # add attribute to the class we have
 347         m = Message.from_dict(self.ldb_dc1,
 348                               {"dn": c_dn,
 349                                "mayContain": a_ldn},
 350                               FLAG_MOD_ADD)
 351         self.ldb_dc1.modify(m)
 352
 353         # force replication from DC1 to DC2
 354         self._net_drs_replicate(DC=self.dnsname_dc2, fromDC=self.dnsname_dc1, nc_dn=self.schema_dn, forced=True)
 355
 356         # check objects are replicated
 357         self._check_object(c_dn)
 358         self._check_object(a_dn)