2 # -*- coding: utf-8 -*-
3 # This is a port of the original in testprogs/ejs/ldap.js
11 sys.path.append("bin/python")
13 samba.ensure_external_module("subunit", "subunit/python")
14 samba.ensure_external_module("testtools", "testtools")
16 import samba.getopt as options
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
26 from samba.dsdb import DS_DOMAIN_FUNCTION_2003
28 from subunit.run import SubunitTestRunner
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()
46 lp = sambaopts.get_loadparm()
47 creds = credopts.get_credentials(lp)
50 class SchemaTests(unittest.TestCase):
52 def delete_force(self, ldb, dn):
55 except LdbError, (num, _):
56 self.assertEquals(num, ERR_NO_SUCH_OBJECT)
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]
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]
70 super(SchemaTests, self).setUp()
72 self.schema_dn = self.find_schemadn(ldb)
73 self.base_dn = self.find_basedn(ldb)
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])
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,
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])
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("-", "")
99 dn: CN=%s,%s""" % (attr_name, self.schema_dn) + """
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
112 self.ldb.add_ldif(ldif)
114 # Search for created attribute
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])
121 class_name = "test-Class" + time.strftime("%s", time.gmtime())
122 class_ldap_display_name = class_name.replace("-", "")
124 # First try to create a class with a wrong "defaultObjectCategory"
126 dn: CN=%s,%s""" % (class_name, self.schema_dn) + """
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
135 objectClassCategory: 1
136 subClassOf: organizationalPerson
139 systemMustContain: cn
140 systemMustContain: """ + attr_ldap_display_name + """
144 self.ldb.add_ldif(ldif)
146 except LdbError, (num, _):
147 self.assertEquals(num, ERR_CONSTRAINT_VIOLATION)
150 dn: CN=%s,%s""" % (class_name, self.schema_dn) + """
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
158 objectClassCategory: 1
159 subClassOf: organizationalPerson
162 systemMustContain: cn
163 systemMustContain: """ + attr_ldap_display_name + """
166 self.ldb.add_ldif(ldif)
168 # Search for created objectclass
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])
182 self.ldb.modify_ldif(ldif)
184 object_name = "obj" + time.strftime("%s", time.gmtime())
187 dn: CN=%s,CN=Users,%s"""% (object_name, self.base_dn) + """
188 objectClass: organizationalPerson
190 objectClass: """ + class_ldap_display_name + """
192 cn: """ + object_name + """
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
199 self.ldb.add_ldif(ldif)
201 # Search for created object
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)
206 self.delete_force(self.ldb, "cn=%s,cn=Users,%s" % (object_name, self.base_dn))
209 class SchemaTests_msDS_IntId(unittest.TestCase):
212 super(SchemaTests_msDS_IntId, self).setUp()
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])
220 def _ldap_schemaUpdateNow(self):
227 self.ldb.modify_ldif(ldif)
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)
235 def _is_schema_base_object(self, ldb_msg):
236 """Test systemFlags for SYSTEM_FLAG_SCHEMA_BASE_OBJECT (16)"""
238 if "systemFlags" in ldb_msg:
239 systemFlags = int(ldb_msg["systemFlags"][0])
240 return (systemFlags & 16) != 0
242 def _make_attr_ldif(self, attr_name, attr_dn):
244 dn: """ + attr_dn + """
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
259 def test_msDS_IntId_on_attr(self):
260 """Testing msDs-IntId creation for Attributes.
261 See MS-ADTS - 3.1.1.Attributes
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
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)
279 # try to add msDS-IntId during Attribute creation
280 ldif_fail = ldif + "msDS-IntId: -1993108831\n"
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)
287 # add the new attribute and update schema
288 self.ldb.add_ldif(ldif)
289 self._ldap_schemaUpdateNow()
291 # Search for created attribute
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])
300 self.assertTrue("msDS-IntId" in res[0])
302 self.assertTrue("msDS-IntId" not in res[0])
305 msg.dn = Dn(self.ldb, attr_dn)
306 msg["msDS-IntId"] = MessageElement("-1993108831", FLAG_MOD_REPLACE, "msDS-IntId")
309 self.fail("Modifying msDS-IntId should return error")
310 except LdbError, (num, _):
311 self.assertEquals(num, ERR_CONSTRAINT_VIOLATION)
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"
321 # try to add msDS-IntId during Attribute creation
322 ldif_fail = ldif + "msDS-IntId: -1993108831\n"
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)
329 # add the new attribute and update schema
330 self.ldb.add_ldif(ldif)
331 self._ldap_schemaUpdateNow()
333 # Search for created attribute
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])
342 self.assertTrue("msDS-IntId" in res[0])
344 self.assertTrue("msDS-IntId" not in res[0])
347 msg.dn = Dn(self.ldb, attr_dn)
348 msg["msDS-IntId"] = MessageElement("-1993108831", FLAG_MOD_REPLACE, "msDS-IntId")
351 self.fail("Modifying msDS-IntId should return error")
352 except LdbError, (num, _):
353 self.assertEquals(num, ERR_CONSTRAINT_VIOLATION)
356 def _make_class_ldif(self, class_dn, class_name):
358 dn: """ + class_dn + """
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
366 objectClassCategory: 1
367 subClassOf: organizationalPerson
369 systemMustContain: cn
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"""
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)
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()
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")
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)
398 self.ldb.add_ldif(ldif)
399 self._ldap_schemaUpdateNow()
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])
407 msg.dn = Dn(self.ldb, class_dn)
408 msg["msDS-IntId"] = MessageElement("-1993108831", FLAG_MOD_REPLACE, "msDS-IntId")
411 self.fail("Modifying msDS-IntId should return error")
412 except LdbError, (num, _):
413 self.assertEquals(num, ERR_CONSTRAINT_VIOLATION)
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"
423 # try to add msDS-IntId during Class creation
424 ldif_add = ldif + "msDS-IntId: -1993108831\n"
425 self.ldb.add_ldif(ldif_add)
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")
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"
436 self.ldb.add_ldif(ldif)
437 self._ldap_schemaUpdateNow()
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])
445 msg.dn = Dn(self.ldb, class_dn)
446 msg["msDS-IntId"] = MessageElement("-1993108831", FLAG_MOD_REPLACE, "msDS-IntId")
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])
457 def test_verify_msDS_IntId(self):
458 """Verify msDS-IntId exists only on attributes without FLAG_SCHEMA_BASE_OBJECT flag set"""
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)
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)
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:
476 print "%3d warning: msDS-IntId expected on: %-30s %s" % (count, ldb_msg["attributeID"], ldb_msg["cn"])
478 self.assertTrue("msDS-IntId" not in ldb_msg)
481 class SchemaTests_msDS_isRODC(unittest.TestCase):
484 super(SchemaTests_msDS_isRODC, self).setUp()
486 res = ldb.search(base="", expression="", scope=SCOPE_BASE, attrs=["*"])
487 self.assertEquals(len(res), 1)
488 self.base_dn = res[0]["defaultNamingContext"][0]
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"])
494 self.assertTrue("msDS-isRODC" in ldb_msg)
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"])
500 ntds_search_dn = "CN=NTDS Settings,%s" % ldb_msg['dn']
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'])
507 self.assertTrue("objectCategory" in res_check[0])
508 self.assertTrue("msDS-isRODC" in ldb_msg)
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"])
514 if "serverReferenceBL" not in ldb_msg:
515 print("Computer entry %s doesn't have a serverReferenceBL attribute" % ldb_msg['dn'])
517 self.assertTrue("msDS-isRODC" in ldb_msg)
519 if not "://" in host:
520 if os.path.isfile(host):
521 host = "tdb://%s" % host
523 host = "ldap://%s" % host
526 if host.startswith("ldap://"):
527 # user 'paged_search' module when connecting remotely
528 ldb_options = ["modules:paged_searches"]
530 ldb = Ldb(host, credentials=creds, session_info=system_session(), lp=lp, options=ldb_options)
532 runner = SubunitTestRunner()
534 if not runner.run(unittest.makeSuite(SchemaTests)).wasSuccessful():
536 if not runner.run(unittest.makeSuite(SchemaTests_msDS_IntId)).wasSuccessful():
538 if not runner.run(unittest.makeSuite(SchemaTests_msDS_isRODC)).wasSuccessful():