PEP8: fix E502: the backslash is redundant between brackets
[samba.git] / 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 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)