s4-ldap_schema.py: Remove unused LDB connection to GC port
[abartlet/samba.git/.git] / source4 / dsdb / tests / python / ldap_schema.py
1 #!/usr/bin/env python
2 # -*- coding: utf-8 -*-
3 # This is a port of the original in testprogs/ejs/ldap.js
4
5 import optparse
6 import sys
7 import time
8 import random
9 import os
10
11 sys.path.append("bin/python")
12 import samba
13 samba.ensure_external_module("subunit", "subunit/python")
14 samba.ensure_external_module("testtools", "testtools")
15
16 import samba.getopt as options
17
18 from samba.auth import system_session
19 from ldb import SCOPE_ONELEVEL, SCOPE_BASE, LdbError
20 from ldb import ERR_NO_SUCH_OBJECT
21 from ldb import ERR_UNWILLING_TO_PERFORM
22 from ldb import ERR_CONSTRAINT_VIOLATION
23 from ldb import Message, MessageElement, Dn
24 from ldb import FLAG_MOD_REPLACE
25 from samba import Ldb
26 from samba.dsdb import DS_DOMAIN_FUNCTION_2003
27
28 from subunit.run import SubunitTestRunner
29 import unittest
30
31 parser = optparse.OptionParser("ldap [options] <host>")
32 sambaopts = options.SambaOptions(parser)
33 parser.add_option_group(sambaopts)
34 parser.add_option_group(options.VersionOptions(parser))
35 # use command line creds if available
36 credopts = options.CredentialsOptions(parser)
37 parser.add_option_group(credopts)
38 opts, args = parser.parse_args()
39
40 if len(args) < 1:
41     parser.print_usage()
42     sys.exit(1)
43
44 host = args[0]
45
46 lp = sambaopts.get_loadparm()
47 creds = credopts.get_credentials(lp)
48
49
50 class SchemaTests(unittest.TestCase):
51
52     def delete_force(self, ldb, dn):
53         try:
54             ldb.delete(dn)
55         except LdbError, (num, _):
56             self.assertEquals(num, ERR_NO_SUCH_OBJECT)
57
58     def find_schemadn(self, ldb):
59         res = ldb.search(base="", expression="", scope=SCOPE_BASE, attrs=["schemaNamingContext"])
60         self.assertEquals(len(res), 1)
61         return res[0]["schemaNamingContext"][0]
62
63     def find_basedn(self, ldb):
64         res = ldb.search(base="", expression="", scope=SCOPE_BASE,
65                          attrs=["defaultNamingContext"])
66         self.assertEquals(len(res), 1)
67         return res[0]["defaultNamingContext"][0]
68
69     def setUp(self):
70         super(SchemaTests, self).setUp()
71         self.ldb = ldb
72         self.schema_dn = self.find_schemadn(ldb)
73         self.base_dn = self.find_basedn(ldb)
74
75     def test_generated_schema(self):
76         """Testing we can read the generated schema via LDAP"""
77         res = self.ldb.search("cn=aggregate,"+self.schema_dn, scope=SCOPE_BASE,
78                 attrs=["objectClasses", "attributeTypes", "dITContentRules"])
79         self.assertEquals(len(res), 1)
80         self.assertTrue("dITContentRules" in res[0])
81         self.assertTrue("objectClasses" in res[0])
82         self.assertTrue("attributeTypes" in res[0])
83
84     def test_generated_schema_is_operational(self):
85         """Testing we don't get the generated schema via LDAP by default"""
86         res = self.ldb.search("cn=aggregate,"+self.schema_dn, scope=SCOPE_BASE,
87                 attrs=["*"])
88         self.assertEquals(len(res), 1)
89         self.assertFalse("dITContentRules" in res[0])
90         self.assertFalse("objectClasses" in res[0])
91         self.assertFalse("attributeTypes" in res[0])
92
93     def test_schemaUpdateNow(self):
94         """Testing schemaUpdateNow"""
95         attr_name = "test-Attr" + time.strftime("%s", time.gmtime())
96         attr_ldap_display_name = attr_name.replace("-", "")
97
98         ldif = """
99 dn: CN=%s,%s""" % (attr_name, self.schema_dn) + """
100 objectClass: top
101 objectClass: attributeSchema
102 adminDescription: """ + attr_name + """
103 adminDisplayName: """ + attr_name + """
104 cn: """ + attr_name + """
105 attributeId: 1.2.840.""" + str(random.randint(1,100000)) + """.1.5.9940
106 attributeSyntax: 2.5.5.12
107 omSyntax: 64
108 instanceType: 4
109 isSingleValued: TRUE
110 systemOnly: FALSE
111 """
112         self.ldb.add_ldif(ldif)
113
114         # Search for created attribute
115         res = []
116         res = self.ldb.search("cn=%s,%s" % (attr_name, self.schema_dn), scope=SCOPE_BASE, attrs=["*"])
117         self.assertEquals(len(res), 1)
118         self.assertEquals(res[0]["lDAPDisplayName"][0], attr_ldap_display_name)
119         self.assertTrue("schemaIDGUID" in res[0])
120
121         class_name = "test-Class" + time.strftime("%s", time.gmtime())
122         class_ldap_display_name = class_name.replace("-", "")
123
124         # First try to create a class with a wrong "defaultObjectCategory"
125         ldif = """
126 dn: CN=%s,%s""" % (class_name, self.schema_dn) + """
127 objectClass: top
128 objectClass: classSchema
129 defaultObjectCategory: CN=_
130 adminDescription: """ + class_name + """
131 adminDisplayName: """ + class_name + """
132 cn: """ + class_name + """
133 governsId: 1.2.840.""" + str(random.randint(1,100000)) + """.1.5.9939
134 instanceType: 4
135 objectClassCategory: 1
136 subClassOf: organizationalPerson
137 systemFlags: 16
138 rDNAttID: cn
139 systemMustContain: cn
140 systemMustContain: """ + attr_ldap_display_name + """
141 systemOnly: FALSE
142 """
143         try:
144                  self.ldb.add_ldif(ldif)
145                  self.fail()
146         except LdbError, (num, _):
147                  self.assertEquals(num, ERR_CONSTRAINT_VIOLATION)
148
149         ldif = """
150 dn: CN=%s,%s""" % (class_name, self.schema_dn) + """
151 objectClass: top
152 objectClass: classSchema
153 adminDescription: """ + class_name + """
154 adminDisplayName: """ + class_name + """
155 cn: """ + class_name + """
156 governsId: 1.2.840.""" + str(random.randint(1,100000)) + """.1.5.9939
157 instanceType: 4
158 objectClassCategory: 1
159 subClassOf: organizationalPerson
160 systemFlags: 16
161 rDNAttID: cn
162 systemMustContain: cn
163 systemMustContain: """ + attr_ldap_display_name + """
164 systemOnly: FALSE
165 """
166         self.ldb.add_ldif(ldif)
167
168         # Search for created objectclass
169         res = []
170         res = self.ldb.search("cn=%s,%s" % (class_name, self.schema_dn), scope=SCOPE_BASE, attrs=["*"])
171         self.assertEquals(len(res), 1)
172         self.assertEquals(res[0]["lDAPDisplayName"][0], class_ldap_display_name)
173         self.assertEquals(res[0]["defaultObjectCategory"][0], res[0]["distinguishedName"][0])
174         self.assertTrue("schemaIDGUID" in res[0])
175
176         ldif = """
177 dn:
178 changetype: modify
179 add: schemaUpdateNow
180 schemaUpdateNow: 1
181 """
182         self.ldb.modify_ldif(ldif)
183
184         object_name = "obj" + time.strftime("%s", time.gmtime())
185
186         ldif = """
187 dn: CN=%s,CN=Users,%s"""% (object_name, self.base_dn) + """
188 objectClass: organizationalPerson
189 objectClass: person
190 objectClass: """ + class_ldap_display_name + """
191 objectClass: top
192 cn: """ + object_name + """
193 instanceType: 4
194 objectCategory: CN=%s,%s"""% (class_name, self.schema_dn) + """
195 distinguishedName: CN=%s,CN=Users,%s"""% (object_name, self.base_dn) + """
196 name: """ + object_name + """
197 """ + attr_ldap_display_name + """: test
198 """
199         self.ldb.add_ldif(ldif)
200
201         # Search for created object
202         res = []
203         res = self.ldb.search("cn=%s,cn=Users,%s" % (object_name, self.base_dn), scope=SCOPE_BASE, attrs=["*"])
204         self.assertEquals(len(res), 1)
205         # Delete the object
206         self.delete_force(self.ldb, "cn=%s,cn=Users,%s" % (object_name, self.base_dn))
207
208
209 class SchemaTests_msDS_IntId(unittest.TestCase):
210
211     def setUp(self):
212         super(SchemaTests_msDS_IntId, self).setUp()
213         self.ldb = ldb
214         res = ldb.search(base="", expression="", scope=SCOPE_BASE, attrs=["*"])
215         self.assertEquals(len(res), 1)
216         self.schema_dn = res[0]["schemaNamingContext"][0]
217         self.base_dn = res[0]["defaultNamingContext"][0]
218         self.forest_level = int(res[0]["forestFunctionality"][0])
219
220     def _ldap_schemaUpdateNow(self):
221         ldif = """
222 dn:
223 changetype: modify
224 add: schemaUpdateNow
225 schemaUpdateNow: 1
226 """
227         self.ldb.modify_ldif(ldif)
228
229     def _make_obj_names(self, prefix):
230         class_name = prefix + time.strftime("%s", time.gmtime())
231         class_ldap_name = class_name.replace("-", "")
232         class_dn = "CN=%s,%s" % (class_name, self.schema_dn)
233         return (class_name, class_ldap_name, class_dn)
234
235     def _is_schema_base_object(self, ldb_msg):
236         """Test systemFlags for SYSTEM_FLAG_SCHEMA_BASE_OBJECT (16)"""
237         systemFlags = 0
238         if "systemFlags" in ldb_msg:
239             systemFlags = int(ldb_msg["systemFlags"][0])
240         return (systemFlags & 16) != 0
241
242     def _make_attr_ldif(self, attr_name, attr_dn):
243         ldif = """
244 dn: """ + attr_dn + """
245 objectClass: top
246 objectClass: attributeSchema
247 adminDescription: """ + attr_name + """
248 adminDisplayName: """ + attr_name + """
249 cn: """ + attr_name + """
250 attributeId: 1.2.840.""" + str(random.randint(1,100000)) + """.1.5.9940
251 attributeSyntax: 2.5.5.12
252 omSyntax: 64
253 instanceType: 4
254 isSingleValued: TRUE
255 systemOnly: FALSE
256 """
257         return ldif
258
259     def test_msDS_IntId_on_attr(self):
260         """Testing msDs-IntId creation for Attributes.
261         See MS-ADTS - 3.1.1.Attributes
262
263         This test should verify that:
264         - Creating attribute with 'msDS-IntId' fails with ERR_UNWILLING_TO_PERFORM
265         - Adding 'msDS-IntId' on existing attribute fails with ERR_CONSTRAINT_VIOLATION
266         - Creating attribute with 'msDS-IntId' set and FLAG_SCHEMA_BASE_OBJECT flag
267           set fails with ERR_UNWILLING_TO_PERFORM
268         - Attributes created with FLAG_SCHEMA_BASE_OBJECT not set have
269           'msDS-IntId' attribute added internally
270         """
271
272         # 1. Create attribute without systemFlags
273         # msDS-IntId should be created if forest functional
274         # level is >= DS_DOMAIN_FUNCTION_2003
275         # and missing otherwise
276         (attr_name, attr_ldap_name, attr_dn) = self._make_obj_names("msDS-IntId-Attr-1-")
277         ldif = self._make_attr_ldif(attr_name, attr_dn)
278
279         # try to add msDS-IntId during Attribute creation
280         ldif_fail = ldif + "msDS-IntId: -1993108831\n"
281         try:
282             self.ldb.add_ldif(ldif_fail)
283             self.fail("Adding attribute with preset msDS-IntId should fail")
284         except LdbError, (num, _):
285             self.assertEquals(num, ERR_UNWILLING_TO_PERFORM)
286
287         # add the new attribute and update schema
288         self.ldb.add_ldif(ldif)
289         self._ldap_schemaUpdateNow()
290
291         # Search for created attribute
292         res = []
293         res = self.ldb.search(attr_dn, scope=SCOPE_BASE, attrs=["*"])
294         self.assertEquals(len(res), 1)
295         self.assertEquals(res[0]["lDAPDisplayName"][0], attr_ldap_name)
296         if self.forest_level >= DS_DOMAIN_FUNCTION_2003:
297             if self._is_schema_base_object(res[0]):
298                 self.assertTrue("msDS-IntId" not in res[0])
299             else:
300                 self.assertTrue("msDS-IntId" in res[0])
301         else:
302             self.assertTrue("msDS-IntId" not in res[0])
303
304         msg = Message()
305         msg.dn = Dn(self.ldb, attr_dn)
306         msg["msDS-IntId"] = MessageElement("-1993108831", FLAG_MOD_REPLACE, "msDS-IntId")
307         try:
308             self.ldb.modify(msg)
309             self.fail("Modifying msDS-IntId should return error")
310         except LdbError, (num, _):
311             self.assertEquals(num, ERR_CONSTRAINT_VIOLATION)
312
313         # 2. Create attribute with systemFlags = FLAG_SCHEMA_BASE_OBJECT
314         # msDS-IntId should be created if forest functional
315         # level is >= DS_DOMAIN_FUNCTION_2003
316         # and missing otherwise
317         (attr_name, attr_ldap_name, attr_dn) = self._make_obj_names("msDS-IntId-Attr-2-")
318         ldif = self._make_attr_ldif(attr_name, attr_dn)
319         ldif += "systemFlags: 16\n"
320
321         # try to add msDS-IntId during Attribute creation
322         ldif_fail = ldif + "msDS-IntId: -1993108831\n"
323         try:
324             self.ldb.add_ldif(ldif_fail)
325             self.fail("Adding attribute with preset msDS-IntId should fail")
326         except LdbError, (num, _):
327             self.assertEquals(num, ERR_UNWILLING_TO_PERFORM)
328
329         # add the new attribute and update schema
330         self.ldb.add_ldif(ldif)
331         self._ldap_schemaUpdateNow()
332
333         # Search for created attribute
334         res = []
335         res = self.ldb.search(attr_dn, scope=SCOPE_BASE, attrs=["*"])
336         self.assertEquals(len(res), 1)
337         self.assertEquals(res[0]["lDAPDisplayName"][0], attr_ldap_name)
338         if self.forest_level >= DS_DOMAIN_FUNCTION_2003:
339             if self._is_schema_base_object(res[0]):
340                 self.assertTrue("msDS-IntId" not in res[0])
341             else:
342                 self.assertTrue("msDS-IntId" in res[0])
343         else:
344             self.assertTrue("msDS-IntId" not in res[0])
345
346         msg = Message()
347         msg.dn = Dn(self.ldb, attr_dn)
348         msg["msDS-IntId"] = MessageElement("-1993108831", FLAG_MOD_REPLACE, "msDS-IntId")
349         try:
350             self.ldb.modify(msg)
351             self.fail("Modifying msDS-IntId should return error")
352         except LdbError, (num, _):
353             self.assertEquals(num, ERR_CONSTRAINT_VIOLATION)
354
355
356     def _make_class_ldif(self, class_dn, class_name):
357         ldif = """
358 dn: """ + class_dn + """
359 objectClass: top
360 objectClass: classSchema
361 adminDescription: """ + class_name + """
362 adminDisplayName: """ + class_name + """
363 cn: """ + class_name + """
364 governsId: 1.2.840.""" + str(random.randint(1,100000)) + """.1.5.9939
365 instanceType: 4
366 objectClassCategory: 1
367 subClassOf: organizationalPerson
368 rDNAttID: cn
369 systemMustContain: cn
370 systemOnly: FALSE
371 """
372         return ldif
373
374     def test_msDS_IntId_on_class(self):
375         """Testing msDs-IntId creation for Class
376            Reference: MS-ADTS - 3.1.1.2.4.8 Class classSchema"""
377
378         # 1. Create Class without systemFlags
379         # msDS-IntId should be created if forest functional
380         # level is >= DS_DOMAIN_FUNCTION_2003
381         # and missing otherwise
382         (class_name, class_ldap_name, class_dn) = self._make_obj_names("msDS-IntId-Class-1-")
383         ldif = self._make_class_ldif(class_dn, class_name)
384
385         # try to add msDS-IntId during Class creation
386         ldif_add = ldif + "msDS-IntId: -1993108831\n"
387         self.ldb.add_ldif(ldif_add)
388         self._ldap_schemaUpdateNow()
389
390         res = self.ldb.search(class_dn, scope=SCOPE_BASE, attrs=["*"])
391         self.assertEquals(len(res), 1)
392         self.assertEquals(res[0]["msDS-IntId"][0], "-1993108831")
393
394         # add a new Class and update schema
395         (class_name, class_ldap_name, class_dn) = self._make_obj_names("msDS-IntId-Class-2-")
396         ldif = self._make_class_ldif(class_dn, class_name)
397
398         self.ldb.add_ldif(ldif)
399         self._ldap_schemaUpdateNow()
400
401         # Search for created Class
402         res = self.ldb.search(class_dn, scope=SCOPE_BASE, attrs=["*"])
403         self.assertEquals(len(res), 1)
404         self.assertFalse("msDS-IntId" in res[0])
405
406         msg = Message()
407         msg.dn = Dn(self.ldb, class_dn)
408         msg["msDS-IntId"] = MessageElement("-1993108831", FLAG_MOD_REPLACE, "msDS-IntId")
409         try:
410             self.ldb.modify(msg)
411             self.fail("Modifying msDS-IntId should return error")
412         except LdbError, (num, _):
413             self.assertEquals(num, ERR_CONSTRAINT_VIOLATION)
414
415         # 2. Create Class with systemFlags = FLAG_SCHEMA_BASE_OBJECT
416         # msDS-IntId should be created if forest functional
417         # level is >= DS_DOMAIN_FUNCTION_2003
418         # and missing otherwise
419         (class_name, class_ldap_name, class_dn) = self._make_obj_names("msDS-IntId-Class-3-")
420         ldif = self._make_class_ldif(class_dn, class_name)
421         ldif += "systemFlags: 16\n"
422
423         # try to add msDS-IntId during Class creation
424         ldif_add = ldif + "msDS-IntId: -1993108831\n"
425         self.ldb.add_ldif(ldif_add)
426
427         res = self.ldb.search(class_dn, scope=SCOPE_BASE, attrs=["*"])
428         self.assertEquals(len(res), 1)
429         self.assertEquals(res[0]["msDS-IntId"][0], "-1993108831")
430
431         # add the new Class and update schema
432         (class_name, class_ldap_name, class_dn) = self._make_obj_names("msDS-IntId-Class-4-")
433         ldif = self._make_class_ldif(class_dn, class_name)
434         ldif += "systemFlags: 16\n"
435
436         self.ldb.add_ldif(ldif)
437         self._ldap_schemaUpdateNow()
438
439         # Search for created Class
440         res = self.ldb.search(class_dn, scope=SCOPE_BASE, attrs=["*"])
441         self.assertEquals(len(res), 1)
442         self.assertFalse("msDS-IntId" in res[0])
443
444         msg = Message()
445         msg.dn = Dn(self.ldb, class_dn)
446         msg["msDS-IntId"] = MessageElement("-1993108831", FLAG_MOD_REPLACE, "msDS-IntId")
447         try:
448             self.ldb.modify(msg)
449             self.fail("Modifying msDS-IntId should return error")
450         except LdbError, (num, _):
451             self.assertEquals(num, ERR_CONSTRAINT_VIOLATION)
452         res = self.ldb.search(class_dn, scope=SCOPE_BASE, attrs=["*"])
453         self.assertEquals(len(res), 1)
454         self.assertFalse("msDS-IntId" in res[0])
455
456
457     def test_verify_msDS_IntId(self):
458         """Verify msDS-IntId exists only on attributes without FLAG_SCHEMA_BASE_OBJECT flag set"""
459         count = 0
460         res = self.ldb.search(self.schema_dn, scope=SCOPE_ONELEVEL,
461                               expression="objectClass=attributeSchema",
462                               attrs=["systemFlags", "msDS-IntId", "attributeID", "cn"])
463         self.assertTrue(len(res) > 1)
464         for ldb_msg in res:
465             if self.forest_level >= DS_DOMAIN_FUNCTION_2003:
466                 if self._is_schema_base_object(ldb_msg):
467                     self.assertTrue("msDS-IntId" not in ldb_msg)
468                 else:
469                     # don't assert here as there are plenty of
470                     # attributes under w2k8 that are not part of
471                     # Base Schema (SYSTEM_FLAG_SCHEMA_BASE_OBJECT flag not set)
472                     # has not msDS-IntId attribute set
473                     #self.assertTrue("msDS-IntId" in ldb_msg, "msDS-IntId expected on: %s" % ldb_msg.dn)
474                     if "msDS-IntId" not in ldb_msg:
475                         count = count + 1
476                         print "%3d warning: msDS-IntId expected on: %-30s %s" % (count, ldb_msg["attributeID"], ldb_msg["cn"])
477             else:
478                 self.assertTrue("msDS-IntId" not in ldb_msg)
479
480
481 class SchemaTests_msDS_isRODC(unittest.TestCase):
482
483     def setUp(self):
484         super(SchemaTests_msDS_isRODC, self).setUp()
485         self.ldb = ldb
486         res = ldb.search(base="", expression="", scope=SCOPE_BASE, attrs=["*"])
487         self.assertEquals(len(res), 1)
488         self.base_dn = res[0]["defaultNamingContext"][0]
489
490     def test_objectClass_ntdsdsa(self):
491         res = self.ldb.search(self.base_dn, expression="objectClass=nTDSDSA",
492                               attrs=["msDS-isRODC"], controls=["search_options:1:2"])
493         for ldb_msg in res:
494             self.assertTrue("msDS-isRODC" in ldb_msg)
495
496     def test_objectClass_server(self):
497         res = self.ldb.search(self.base_dn, expression="objectClass=server",
498                               attrs=["msDS-isRODC"], controls=["search_options:1:2"])
499         for ldb_msg in res:
500             ntds_search_dn = "CN=NTDS Settings,%s" % ldb_msg['dn']
501             try:
502                 res_check = self.ldb.search(ntds_search_dn, attrs=["objectCategory"])
503             except LdbError, (num, _):
504                 self.assertEquals(num, ERR_NO_SUCH_OBJECT)
505                 print("Server entry %s doesn't have a NTDS settings object" % res[0]['dn'])
506             else:
507                 self.assertTrue("objectCategory" in res_check[0])
508                 self.assertTrue("msDS-isRODC" in ldb_msg)
509
510     def test_objectClass_computer(self):
511         res = self.ldb.search(self.base_dn, expression="objectClass=computer",
512                               attrs=["serverReferenceBL","msDS-isRODC"], controls=["search_options:1:2"])
513         for ldb_msg in res:
514             if "serverReferenceBL" not in ldb_msg:
515                 print("Computer entry %s doesn't have a serverReferenceBL attribute" % ldb_msg['dn'])
516             else:
517                 self.assertTrue("msDS-isRODC" in ldb_msg)
518
519 if not "://" in host:
520     if os.path.isfile(host):
521         host = "tdb://%s" % host
522     else:
523         host = "ldap://%s" % host
524
525 ldb_options = []
526 if host.startswith("ldap://"):
527     # user 'paged_search' module when connecting remotely
528     ldb_options = ["modules:paged_searches"]
529
530 ldb = Ldb(host, credentials=creds, session_info=system_session(), lp=lp, options=ldb_options)
531
532 runner = SubunitTestRunner()
533 rc = 0
534 if not runner.run(unittest.makeSuite(SchemaTests)).wasSuccessful():
535     rc = 1
536 if not runner.run(unittest.makeSuite(SchemaTests_msDS_IntId)).wasSuccessful():
537     rc = 1
538 if not runner.run(unittest.makeSuite(SchemaTests_msDS_isRODC)).wasSuccessful():
539     rc = 1
540
541 sys.exit(rc)