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