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