# create user
self.ldb.newuser(username=self.username, password=self.password)
ldb_res = self.ldb.search(base=self.domain_dn,
- scope=SCOPE_SUBTREE,
- expression="(samAccountName=%s)" % self.username)
+ scope=SCOPE_SUBTREE,
+ expression="(samAccountName=%s)" % self.username)
self.assertEquals(len(ldb_res), 1)
user_dn = ldb_res[0]["dn"]
self.addCleanup(delete_force, self.ldb, user_dn)
# create user
self.ldb.newuser(username=self.username, password=self.password)
ldb_res = self.ldb.search(base=self.domain_dn,
- scope=SCOPE_SUBTREE,
- expression="(samAccountName=%s)" % self.username)
+ scope=SCOPE_SUBTREE,
+ expression="(samAccountName=%s)" % self.username)
self.assertEquals(len(ldb_res), 1)
user_dn = ldb_res[0]["dn"]
self.addCleanup(delete_force, self.ldb, user_dn)
self.assertEqual(str(res1[0].dn).lower(), ("CN=g3,%s" % self.ou_groups).lower())
res1 = self.ldb.search(self.ou_groups,
- scope=SCOPE_SUBTREE,
- expression="member:1.2.840.113556.1.4.1941:=<SID=%s>" % sid)
+ scope=SCOPE_SUBTREE,
+ expression="member:1.2.840.113556.1.4.1941:=<SID=%s>" % sid)
self.assertEqual(len(res1), 2)
dn_list = [str(res.dn).lower() for res in res1]
self.assertTrue(("CN=g3,%s" % self.ou_groups).lower() in dn_list)
self.assertTrue(("CN=g4,%s" % self.ou_groups).lower() in dn_list)
res1 = self.ldb.search(self.ou_groups,
- scope=SCOPE_ONELEVEL,
- expression="member:1.2.840.113556.1.4.1941:=<SID=%s>" % sid)
+ scope=SCOPE_ONELEVEL,
+ expression="member:1.2.840.113556.1.4.1941:=<SID=%s>" % sid)
self.assertEqual(len(res1), 2)
dn_list = [str(res.dn).lower() for res in res1]
self.assertTrue(("CN=g3,%s" % self.ou_groups).lower() in dn_list)
def test_extended_dn_u4(self):
res1 = self.ldb.search("cn=u4,%s" % self.ou_users,
- scope=SCOPE_BASE,
- expression="objectClass=*",
- attrs=['objectSid', 'objectGUID'])
+ scope=SCOPE_BASE,
+ expression="objectClass=*",
+ attrs=['objectSid', 'objectGUID'])
self.assertEqual(len(res1), 1)
self.assertEqual(str(res1[0].dn).lower(), ("cn=u4,%s" % self.ou_users).lower())
guid = self.ldb.schema_format_value("objectGUID", res1[0]['objectGUID'][0])
res1 = self.ldb.search(self.ou_groups,
- scope=SCOPE_SUBTREE,
- expression="member=<SID=%s>" % sid)
+ scope=SCOPE_SUBTREE,
+ expression="member=<SID=%s>" % sid)
self.assertEqual(len(res1), 1)
self.assertEqual(str(res1[0].dn).lower(), ("CN=g4,%s" % self.ou_groups).lower())
res1 = self.ldb.search(self.ou_groups,
- scope=SCOPE_SUBTREE,
- expression="member=<GUID=%s>" % guid)
+ scope=SCOPE_SUBTREE,
+ expression="member=<GUID=%s>" % guid)
self.assertEqual(len(res1), 1)
self.assertEqual(str(res1[0].dn).lower(), ("CN=g4,%s" % self.ou_groups).lower())
res1 = self.ldb.search(self.ou_groups,
- scope=SCOPE_ONELEVEL,
- expression="member=<GUID=%s>" % guid)
+ scope=SCOPE_ONELEVEL,
+ expression="member=<GUID=%s>" % guid)
self.assertEqual(len(res1), 1)
self.assertEqual(str(res1[0].dn).lower(), ("CN=g4,%s" % self.ou_groups).lower())
res1 = self.ldb.search(self.ou_groups,
- scope=SCOPE_SUBTREE,
- expression="member:1.2.840.113556.1.4.1941:=<SID=%s>" % sid)
+ scope=SCOPE_SUBTREE,
+ expression="member:1.2.840.113556.1.4.1941:=<SID=%s>" % sid)
self.assertEqual(len(res1), 1)
self.assertEqual(str(res1[0].dn).lower(), ("CN=g4,%s" % self.ou_groups).lower())
res1 = self.ldb.search(self.ou_groups,
- scope=SCOPE_ONELEVEL,
- expression="member:1.2.840.113556.1.4.1941:=<SID=%s>" % sid)
+ scope=SCOPE_ONELEVEL,
+ expression="member:1.2.840.113556.1.4.1941:=<SID=%s>" % sid)
self.assertEqual(len(res1), 1)
self.assertEqual(str(res1[0].dn).lower(), ("CN=g4,%s" % self.ou_groups).lower())
res1 = self.ldb.search(self.ou_groups,
- scope=SCOPE_SUBTREE,
- expression="member:1.2.840.113556.1.4.1941:=<GUID=%s>" % guid)
+ scope=SCOPE_SUBTREE,
+ expression="member:1.2.840.113556.1.4.1941:=<GUID=%s>" % guid)
self.assertEqual(len(res1), 1)
self.assertEqual(str(res1[0].dn).lower(), ("CN=g4,%s" % self.ou_groups).lower())
res1 = self.ldb.search(self.ou_groups,
- scope=SCOPE_ONELEVEL,
- expression="member:1.2.840.113556.1.4.1941:=<GUID=%s>" % guid)
+ scope=SCOPE_ONELEVEL,
+ expression="member:1.2.840.113556.1.4.1941:=<GUID=%s>" % guid)
self.assertEqual(len(res1), 1)
self.assertEqual(str(res1[0].dn).lower(), ("CN=g4,%s" % self.ou_groups).lower())
def test_invalid_basedn(self):
res1 = self.ldb.search(self.base_dn,
- scope=SCOPE_SUBTREE,
- expression="memberOf:1.2.840.113556.1.4.1941:=cn=c1,ou=computers,ou=matchrulestest,%sXX" % self.base_dn)
+ scope=SCOPE_SUBTREE,
+ expression="memberOf:1.2.840.113556.1.4.1941:=cn=c1,ou=computers,ou=matchrulestest,%sXX" % self.base_dn)
self.assertEqual(len(res1), 0)
res1 = self.ldb.search(self.base_dn,
- scope=SCOPE_SUBTREE,
- expression="memberOf:1.2.840.113556.1.4.1941:=cn=XX,ou=computers,ou=matchrulestest,%s" % self.base_dn)
+ scope=SCOPE_SUBTREE,
+ expression="memberOf:1.2.840.113556.1.4.1941:=cn=XX,ou=computers,ou=matchrulestest,%s" % self.base_dn)
self.assertEqual(len(res1), 0)
def test_subtree(self):
def test_unknown_oid(self):
res1 = self.ldb.search("cn=g4,%s" % self.ou_groups,
- scope=SCOPE_BASE,
- expression="member:2.4.681.226012.2.8.3882:=cn=u1,%s" % self.ou_users)
+ scope=SCOPE_BASE,
+ expression="member:2.4.681.226012.2.8.3882:=cn=u1,%s" % self.ou_users)
self.assertEqual(len(res1), 0)
res1 = self.ldb.search("cn=g4,%s" % self.ou_groups,
- scope=SCOPE_BASE,
- expression="member:8.16.8720.1008448.8.32.15528:=cn=u1,%s" % self.ou_users)
+ scope=SCOPE_BASE,
+ expression="member:8.16.8720.1008448.8.32.15528:=cn=u1,%s" % self.ou_users)
self.assertEqual(len(res1), 0)
res1 = self.ldb.search("cn=g4,%s" % self.ou_groups,
- scope=SCOPE_BASE,
- expression="member:1.2.3.4:=cn=u1,%s" % self.ou_users)
+ scope=SCOPE_BASE,
+ expression="member:1.2.3.4:=cn=u1,%s" % self.ou_users)
self.assertEqual(len(res1), 0)
def test_nul_text(self):
def test_misc_matches(self):
res1 = self.ldb.search(self.ou_groups,
- scope=SCOPE_BASE,
- expression="member=cn=g1,%s" % self.ou_groups)
+ scope=SCOPE_BASE,
+ expression="member=cn=g1,%s" % self.ou_groups)
self.assertEqual(len(res1), 0)
res1 = self.ldb.search("cn=g1,%s" % self.ou_groups,
- scope=SCOPE_BASE,
- expression="member=cn=g1,%s" % self.ou_groups)
+ scope=SCOPE_BASE,
+ expression="member=cn=g1,%s" % self.ou_groups)
self.assertEqual(len(res1), 0)
res1 = self.ldb.search(self.ou_groups,
- scope=SCOPE_SUBTREE,
- expression="member=cn=g1,%s" % self.ou_groups)
+ scope=SCOPE_SUBTREE,
+ expression="member=cn=g1,%s" % self.ou_groups)
self.assertEqual(len(res1), 1)
self.assertEqual(str(res1[0].dn), "CN=g2,%s" % self.ou_groups)
res1 = self.ldb.search(self.ou_groups,
- scope=SCOPE_ONELEVEL,
- expression="member=cn=g1,%s" % self.ou_groups)
+ scope=SCOPE_ONELEVEL,
+ expression="member=cn=g1,%s" % self.ou_groups)
self.assertEqual(len(res1), 1)
self.assertEqual(str(res1[0].dn), "CN=g2,%s" % self.ou_groups)
res1 = self.ldb.search(self.ou_groups,
- scope=SCOPE_BASE,
- expression="member:1.2.840.113556.1.4.1941:=cn=g1,%s" % self.ou_groups)
+ scope=SCOPE_BASE,
+ expression="member:1.2.840.113556.1.4.1941:=cn=g1,%s" % self.ou_groups)
self.assertEqual(len(res1), 0)
res1 = self.ldb.search("cn=g1,%s" % self.ou_groups,
- scope=SCOPE_BASE,
- expression="member:1.2.840.113556.1.4.1941:=cn=g1,%s" % self.ou_groups)
+ scope=SCOPE_BASE,
+ expression="member:1.2.840.113556.1.4.1941:=cn=g1,%s" % self.ou_groups)
self.assertEqual(len(res1), 0)
res1 = self.ldb.search(self.ou_groups,
- scope=SCOPE_SUBTREE,
- expression="member:1.2.840.113556.1.4.1941:=cn=g1,%s" % self.ou_groups)
+ scope=SCOPE_SUBTREE,
+ expression="member:1.2.840.113556.1.4.1941:=cn=g1,%s" % self.ou_groups)
self.assertEqual(len(res1), 3)
dn_list = [str(res.dn) for res in res1]
self.assertTrue("CN=g2,%s" % self.ou_groups in dn_list)
self.assertTrue("CN=g4,%s" % self.ou_groups in dn_list)
res1 = self.ldb.search(self.ou_groups,
- scope=SCOPE_ONELEVEL,
- expression="member:1.2.840.113556.1.4.1941:=cn=g1,%s" % self.ou_groups)
+ scope=SCOPE_ONELEVEL,
+ expression="member:1.2.840.113556.1.4.1941:=cn=g1,%s" % self.ou_groups)
self.assertEqual(len(res1), 3)
dn_list = [str(res.dn) for res in res1]
self.assertTrue("CN=g2,%s" % self.ou_groups in dn_list)
self.assertTrue("CN=g4,%s" % self.ou_groups in dn_list)
res1 = self.ldb.search(self.ou_groups,
- scope=SCOPE_SUBTREE,
- expression="member:1.2.840.113556.1.4.1941:=cn=g4,%s" % self.ou_groups)
+ scope=SCOPE_SUBTREE,
+ expression="member:1.2.840.113556.1.4.1941:=cn=g4,%s" % self.ou_groups)
self.assertEqual(len(res1), 0)
res1 = self.ldb.search(self.ou_groups,
- scope=SCOPE_ONELEVEL,
- expression="member:1.2.840.113556.1.4.1941:=cn=g4,%s" % self.ou_groups)
+ scope=SCOPE_ONELEVEL,
+ expression="member:1.2.840.113556.1.4.1941:=cn=g4,%s" % self.ou_groups)
self.assertEqual(len(res1), 0)
res1 = self.ldb.search(self.ou_groups,
- scope=SCOPE_BASE,
- expression="memberOf=cn=g4,%s" % self.ou_groups)
+ scope=SCOPE_BASE,
+ expression="memberOf=cn=g4,%s" % self.ou_groups)
self.assertEqual(len(res1), 0)
res1 = self.ldb.search("cn=g4,%s" % self.ou_groups,
- scope=SCOPE_BASE,
- expression="memberOf=cn=g4,%s" % self.ou_groups)
+ scope=SCOPE_BASE,
+ expression="memberOf=cn=g4,%s" % self.ou_groups)
self.assertEqual(len(res1), 0)
res1 = self.ldb.search(self.ou_groups,
- scope=SCOPE_SUBTREE,
- expression="memberOf=cn=g4,%s" % self.ou_groups)
+ scope=SCOPE_SUBTREE,
+ expression="memberOf=cn=g4,%s" % self.ou_groups)
self.assertEqual(len(res1), 1)
self.assertEqual(str(res1[0].dn), ("CN=g3,%s" % self.ou_groups))
res1 = self.ldb.search(self.ou_groups,
- scope=SCOPE_ONELEVEL,
- expression="memberOf=cn=g4,%s" % self.ou_groups)
+ scope=SCOPE_ONELEVEL,
+ expression="memberOf=cn=g4,%s" % self.ou_groups)
self.assertEqual(len(res1), 1)
self.assertEqual(str(res1[0].dn), ("CN=g3,%s" % self.ou_groups))
res1 = self.ldb.search(self.ou_groups,
- scope=SCOPE_BASE,
- expression="memberOf:1.2.840.113556.1.4.1941:=cn=g4,%s" % self.ou_groups)
+ scope=SCOPE_BASE,
+ expression="memberOf:1.2.840.113556.1.4.1941:=cn=g4,%s" % self.ou_groups)
self.assertEqual(len(res1), 0)
res1 = self.ldb.search("cn=g4,%s" % self.ou_groups,
- scope=SCOPE_BASE,
- expression="memberOf:1.2.840.113556.1.4.1941:=cn=g4,%s" % self.ou_groups)
+ scope=SCOPE_BASE,
+ expression="memberOf:1.2.840.113556.1.4.1941:=cn=g4,%s" % self.ou_groups)
self.assertEqual(len(res1), 0)
res1 = self.ldb.search(self.ou_groups,
- scope=SCOPE_SUBTREE,
- expression="memberOf:1.2.840.113556.1.4.1941:=cn=g4,%s" % self.ou_groups)
+ scope=SCOPE_SUBTREE,
+ expression="memberOf:1.2.840.113556.1.4.1941:=cn=g4,%s" % self.ou_groups)
self.assertEqual(len(res1), 3)
dn_list = [str(res.dn) for res in res1]
self.assertTrue("CN=g1,%s" % self.ou_groups in dn_list)
self.assertTrue("CN=g3,%s" % self.ou_groups in dn_list)
res1 = self.ldb.search(self.ou_groups,
- scope=SCOPE_ONELEVEL,
- expression="memberOf:1.2.840.113556.1.4.1941:=cn=g4,%s" % self.ou_groups)
+ scope=SCOPE_ONELEVEL,
+ expression="memberOf:1.2.840.113556.1.4.1941:=cn=g4,%s" % self.ou_groups)
self.assertEqual(len(res1), 3)
dn_list = [str(res.dn) for res in res1]
self.assertTrue("CN=g1,%s" % self.ou_groups in dn_list)
self.assertTrue("CN=g3,%s" % self.ou_groups in dn_list)
res1 = self.ldb.search(self.ou_groups,
- scope=SCOPE_SUBTREE,
- expression="memberOf:1.2.840.113556.1.4.1941:=cn=g1,%s" % self.ou_groups)
+ scope=SCOPE_SUBTREE,
+ expression="memberOf:1.2.840.113556.1.4.1941:=cn=g1,%s" % self.ou_groups)
self.assertEqual(len(res1), 0)
res1 = self.ldb.search(self.ou_groups,
- scope=SCOPE_SUBTREE,
- expression="memberOf:1.2.840.113556.1.4.1941:=cn=g1,%s" % self.ou_groups)
+ scope=SCOPE_SUBTREE,
+ expression="memberOf:1.2.840.113556.1.4.1941:=cn=g1,%s" % self.ou_groups)
self.assertEqual(len(res1), 0)
class MatchRuleConditionTests(samba.tests.TestCase):
def test_g1_members(self):
res1 = self.ldb.search(self.ou,
- scope=SCOPE_SUBTREE,
- expression="memberOf=cn=g1,%s" % self.ou_groups)
+ scope=SCOPE_SUBTREE,
+ expression="memberOf=cn=g1,%s" % self.ou_groups)
self.assertEquals(len(res1), 2)
dn_list = [str(res.dn) for res in res1]
self.assertTrue("CN=g2,%s" % self.ou_groups in dn_list)
self.assertTrue("CN=u2,%s" % self.ou_users in dn_list)
res1 = self.ldb.search(self.ou,
- scope=SCOPE_SUBTREE,
- expression="memberOf:1.2.840.113556.1.4.1941:=cn=g1,%s" % self.ou_groups)
+ scope=SCOPE_SUBTREE,
+ expression="memberOf:1.2.840.113556.1.4.1941:=cn=g1,%s" % self.ou_groups)
self.assertEquals(len(res1), 6)
dn_list = [str(res.dn) for res in res1]
self.assertTrue("CN=u2,%s" % self.ou_users in dn_list)
self.assertTrue("CN=c3,%s" % self.ou_computers in dn_list)
res1 = self.ldb.search(self.ou,
- scope=SCOPE_SUBTREE,
- expression="member=cn=g1,%s" % self.ou_groups)
+ scope=SCOPE_SUBTREE,
+ expression="member=cn=g1,%s" % self.ou_groups)
self.assertEquals(len(res1), 0)
res1 = self.ldb.search(self.ou,
- scope=SCOPE_SUBTREE,
- expression="member:1.2.840.113556.1.4.1941:=cn=g1,%s" % self.ou_groups)
+ scope=SCOPE_SUBTREE,
+ expression="member:1.2.840.113556.1.4.1941:=cn=g1,%s" % self.ou_groups)
self.assertEquals(len(res1), 0)
def test_g2_members(self):
res1 = self.ldb.search(self.ou,
- scope=SCOPE_SUBTREE,
- expression="memberOf=cn=g2,%s" % self.ou_groups)
+ scope=SCOPE_SUBTREE,
+ expression="memberOf=cn=g2,%s" % self.ou_groups)
self.assertEquals(len(res1), 2)
dn_list = [str(res.dn) for res in res1]
self.assertTrue("CN=g3,%s" % self.ou_groups in dn_list)
self.assertTrue("CN=u2,%s" % self.ou_users in dn_list)
res1 = self.ldb.search(self.ou,
- scope=SCOPE_SUBTREE,
- expression="memberOf:1.2.840.113556.1.4.1941:=cn=g2,%s" % self.ou_groups)
+ scope=SCOPE_SUBTREE,
+ expression="memberOf:1.2.840.113556.1.4.1941:=cn=g2,%s" % self.ou_groups)
self.assertEquals(len(res1), 5)
dn_list = [str(res.dn) for res in res1]
self.assertTrue("CN=u2,%s" % self.ou_users in dn_list)
self.assertTrue("CN=c3,%s" % self.ou_computers in dn_list)
res1 = self.ldb.search(self.ou,
- scope=SCOPE_SUBTREE,
- expression="member=cn=g2,%s" % self.ou_groups)
+ scope=SCOPE_SUBTREE,
+ expression="member=cn=g2,%s" % self.ou_groups)
self.assertEquals(len(res1), 1)
self.assertEquals(str(res1[0].dn), "CN=g1,%s" % self.ou_groups)
res1 = self.ldb.search(self.ou,
- scope=SCOPE_SUBTREE,
- expression="member:1.2.840.113556.1.4.1941:=cn=g2,%s" % self.ou_groups)
+ scope=SCOPE_SUBTREE,
+ expression="member:1.2.840.113556.1.4.1941:=cn=g2,%s" % self.ou_groups)
self.assertEquals(len(res1), 1)
self.assertEquals(str(res1[0].dn), "CN=g1,%s" % self.ou_groups)
def test_g3_members(self):
res1 = self.ldb.search(self.ou,
- scope=SCOPE_SUBTREE,
- expression="memberOf=cn=g3,%s" % self.ou_groups)
+ scope=SCOPE_SUBTREE,
+ expression="memberOf=cn=g3,%s" % self.ou_groups)
self.assertEquals(len(res1), 2)
dn_list = [str(res.dn) for res in res1]
self.assertTrue("CN=u3,%s" % self.ou_users in dn_list)
self.assertTrue("CN=c3,%s" % self.ou_computers in dn_list)
res1 = self.ldb.search(self.ou,
- scope=SCOPE_SUBTREE,
- expression="memberOf:1.2.840.113556.1.4.1941:=cn=g3,%s" % self.ou_groups)
+ scope=SCOPE_SUBTREE,
+ expression="memberOf:1.2.840.113556.1.4.1941:=cn=g3,%s" % self.ou_groups)
self.assertEquals(len(res1), 3)
dn_list = [str(res.dn) for res in res1]
self.assertTrue("CN=u3,%s" % self.ou_users in dn_list)
self.assertTrue("CN=c3,%s" % self.ou_computers in dn_list)
res1 = self.ldb.search(self.ou,
- scope=SCOPE_SUBTREE,
- expression="member=cn=g3,%s" % self.ou_groups)
+ scope=SCOPE_SUBTREE,
+ expression="member=cn=g3,%s" % self.ou_groups)
self.assertEquals(len(res1), 1)
self.assertEquals(str(res1[0].dn), "CN=g2,%s" % self.ou_groups)
res1 = self.ldb.search(self.ou,
- scope=SCOPE_SUBTREE,
- expression="member:1.2.840.113556.1.4.1941:=cn=g3,%s" % self.ou_groups)
+ scope=SCOPE_SUBTREE,
+ expression="member:1.2.840.113556.1.4.1941:=cn=g3,%s" % self.ou_groups)
self.assertEquals(len(res1), 2)
dn_list = [str(res.dn) for res in res1]
self.assertTrue("CN=g1,%s" % self.ou_groups in dn_list)
def test_g4_members(self):
res1 = self.ldb.search(self.ou,
- scope=SCOPE_SUBTREE,
- expression="memberOf=cn=g4,%s" % self.ou_groups)
+ scope=SCOPE_SUBTREE,
+ expression="memberOf=cn=g4,%s" % self.ou_groups)
self.assertEquals(len(res1), 3)
dn_list = [str(res.dn) for res in res1]
self.assertTrue("CN=u3,%s" % self.ou_users in dn_list)
self.assertTrue("CN=c4,%s" % self.ou_computers in dn_list)
res1 = self.ldb.search(self.ou,
- scope=SCOPE_SUBTREE,
- expression="memberOf:1.2.840.113556.1.4.1941:=cn=g4,%s" % self.ou_groups)
+ scope=SCOPE_SUBTREE,
+ expression="memberOf:1.2.840.113556.1.4.1941:=cn=g4,%s" % self.ou_groups)
self.assertEquals(len(res1), 4)
dn_list = [str(res.dn) for res in res1]
self.assertTrue("CN=u3,%s" % self.ou_users in dn_list)
self.assertTrue("CN=c4,%s" % self.ou_computers in dn_list)
res1 = self.ldb.search(self.ou,
- scope=SCOPE_SUBTREE,
- expression="member=cn=g4,%s" % self.ou_groups)
+ scope=SCOPE_SUBTREE,
+ expression="member=cn=g4,%s" % self.ou_groups)
self.assertEquals(len(res1), 0)
res1 = self.ldb.search(self.ou,
- scope=SCOPE_SUBTREE,
- expression="member:1.2.840.113556.1.4.1941:=cn=g4,%s" % self.ou_groups)
+ scope=SCOPE_SUBTREE,
+ expression="member:1.2.840.113556.1.4.1941:=cn=g4,%s" % self.ou_groups)
self.assertEquals(len(res1), 0)
def test_u1_members(self):
res1 = self.ldb.search(self.ou,
- scope=SCOPE_SUBTREE,
- expression="memberOf=cn=u1,%s" % self.ou_users)
+ scope=SCOPE_SUBTREE,
+ expression="memberOf=cn=u1,%s" % self.ou_users)
self.assertEqual(len(res1), 2)
dn_list = [str(res.dn) for res in res1]
self.assertTrue("CN=c1,%s" % self.ou_computers in dn_list)
self.assertTrue("CN=c2,%s" % self.ou_computers in dn_list)
res1 = self.ldb.search(self.ou,
- scope=SCOPE_SUBTREE,
- expression="memberOf:1.2.840.113556.1.4.1941:=cn=u1,%s" % self.ou_users)
+ scope=SCOPE_SUBTREE,
+ expression="memberOf:1.2.840.113556.1.4.1941:=cn=u1,%s" % self.ou_users)
self.assertEqual(len(res1), 2)
dn_list = [str(res.dn) for res in res1]
self.assertTrue("CN=c1,%s" % self.ou_computers in dn_list)
self.assertTrue("CN=c2,%s" % self.ou_computers in dn_list)
res1 = self.ldb.search(self.ou,
- scope=SCOPE_SUBTREE,
- expression="member=cn=u1,%s" % self.ou_users)
+ scope=SCOPE_SUBTREE,
+ expression="member=cn=u1,%s" % self.ou_users)
self.assertEqual(len(res1), 0)
res1 = self.ldb.search(self.ou,
- scope=SCOPE_SUBTREE,
- expression="member:1.2.840.113556.1.4.1941:=cn=u1,%s" % self.ou_users)
+ scope=SCOPE_SUBTREE,
+ expression="member:1.2.840.113556.1.4.1941:=cn=u1,%s" % self.ou_users)
self.assertEqual(len(res1), 0)
def test_u2_members(self):
res1 = self.ldb.search(self.ou,
- scope=SCOPE_SUBTREE,
- expression="memberOf=cn=u2,%s" % self.ou_users)
+ scope=SCOPE_SUBTREE,
+ expression="memberOf=cn=u2,%s" % self.ou_users)
self.assertEqual(len(res1), 2)
dn_list = [str(res.dn) for res in res1]
self.assertTrue("CN=c1,%s" % self.ou_computers in dn_list)
self.assertTrue("CN=c3,%s" % self.ou_computers in dn_list)
res1 = self.ldb.search(self.ou,
- scope=SCOPE_SUBTREE,
- expression="memberOf:1.2.840.113556.1.4.1941:=cn=u2,%s" % self.ou_users)
+ scope=SCOPE_SUBTREE,
+ expression="memberOf:1.2.840.113556.1.4.1941:=cn=u2,%s" % self.ou_users)
self.assertEqual(len(res1), 2)
dn_list = [str(res.dn) for res in res1]
self.assertTrue("CN=c1,%s" % self.ou_computers in dn_list)
self.assertTrue("CN=c3,%s" % self.ou_computers in dn_list)
res1 = self.ldb.search(self.ou,
- scope=SCOPE_SUBTREE,
- expression="member=cn=u2,%s" % self.ou_users)
+ scope=SCOPE_SUBTREE,
+ expression="member=cn=u2,%s" % self.ou_users)
self.assertEqual(len(res1), 2)
dn_list = [str(res.dn) for res in res1]
self.assertTrue("CN=g1,%s" % self.ou_groups in dn_list)
self.assertTrue("CN=g2,%s" % self.ou_groups in dn_list)
res1 = self.ldb.search(self.ou,
- scope=SCOPE_SUBTREE,
- expression="member:1.2.840.113556.1.4.1941:=cn=u2,%s" % self.ou_users)
+ scope=SCOPE_SUBTREE,
+ expression="member:1.2.840.113556.1.4.1941:=cn=u2,%s" % self.ou_users)
self.assertEqual(len(res1), 2)
dn_list = [str(res.dn) for res in res1]
self.assertTrue("CN=g1,%s" % self.ou_groups in dn_list)
def test_u3_members(self):
res1 = self.ldb.search(self.ou,
- scope=SCOPE_SUBTREE,
- expression="member=cn=u3,%s" % self.ou_users)
+ scope=SCOPE_SUBTREE,
+ expression="member=cn=u3,%s" % self.ou_users)
self.assertEqual(len(res1), 2)
dn_list = [str(res.dn) for res in res1]
self.assertTrue("CN=g3,%s" % self.ou_groups in dn_list)
self.assertTrue("CN=g4,%s" % self.ou_groups in dn_list)
res1 = self.ldb.search(self.ou,
- scope=SCOPE_SUBTREE,
- expression="member:1.2.840.113556.1.4.1941:=cn=u3,%s" % self.ou_users)
+ scope=SCOPE_SUBTREE,
+ expression="member:1.2.840.113556.1.4.1941:=cn=u3,%s" % self.ou_users)
self.assertEqual(len(res1), 4)
dn_list = [str(res.dn) for res in res1]
self.assertTrue("CN=g1,%s" % self.ou_groups in dn_list)
self.assertTrue("CN=g4,%s" % self.ou_groups in dn_list)
res1 = self.ldb.search(self.ou,
- scope=SCOPE_SUBTREE,
- expression="memberOf=cn=u3,%s" % self.ou_users)
+ scope=SCOPE_SUBTREE,
+ expression="memberOf=cn=u3,%s" % self.ou_users)
self.assertEqual(len(res1), 1)
self.assertEqual(str(res1[0].dn), "CN=c1,%s" % self.ou_computers)
res1 = self.ldb.search(self.ou,
- scope=SCOPE_SUBTREE,
- expression="memberOf:1.2.840.113556.1.4.1941:=cn=u3,%s" % self.ou_users)
+ scope=SCOPE_SUBTREE,
+ expression="memberOf:1.2.840.113556.1.4.1941:=cn=u3,%s" % self.ou_users)
self.assertEqual(len(res1), 1)
self.assertEqual(str(res1[0].dn), "CN=c1,%s" % self.ou_computers)
def test_u4_members(self):
res1 = self.ldb.search(self.ou,
- scope=SCOPE_SUBTREE,
- expression="member=cn=u4,%s" % self.ou_users)
+ scope=SCOPE_SUBTREE,
+ expression="member=cn=u4,%s" % self.ou_users)
self.assertEqual(len(res1), 1)
self.assertEqual(str(res1[0].dn), "CN=g4,%s" % self.ou_groups)
res1 = self.ldb.search(self.ou,
- scope=SCOPE_SUBTREE,
- expression="member:1.2.840.113556.1.4.1941:=cn=u4,%s" % self.ou_users)
+ scope=SCOPE_SUBTREE,
+ expression="member:1.2.840.113556.1.4.1941:=cn=u4,%s" % self.ou_users)
self.assertEqual(len(res1), 1)
self.assertEqual(str(res1[0].dn), "CN=g4,%s" % self.ou_groups)
res1 = self.ldb.search(self.ou,
- scope=SCOPE_SUBTREE,
- expression="memberOf=cn=u4,%s" % self.ou_users)
+ scope=SCOPE_SUBTREE,
+ expression="memberOf=cn=u4,%s" % self.ou_users)
self.assertEqual(len(res1), 2)
dn_list = [str(res.dn) for res in res1]
self.assertTrue("CN=c1,%s" % self.ou_computers in dn_list)
self.assertTrue("CN=c4,%s" % self.ou_computers in dn_list)
res1 = self.ldb.search(self.ou,
- scope=SCOPE_SUBTREE,
- expression="memberOf:1.2.840.113556.1.4.1941:=cn=u4,%s" % self.ou_users)
+ scope=SCOPE_SUBTREE,
+ expression="memberOf:1.2.840.113556.1.4.1941:=cn=u4,%s" % self.ou_users)
self.assertEqual(len(res1), 2)
dn_list = [str(res.dn) for res in res1]
self.assertTrue("CN=c1,%s" % self.ou_computers in dn_list)
def test_c1_members(self):
res1 = self.ldb.search(self.ou,
- scope=SCOPE_SUBTREE,
- expression="member=cn=c1,%s" % self.ou_computers)
+ scope=SCOPE_SUBTREE,
+ expression="member=cn=c1,%s" % self.ou_computers)
self.assertEqual(len(res1), 4)
dn_list = [str(res.dn) for res in res1]
self.assertTrue("CN=u1,%s" % self.ou_users in dn_list)
self.assertTrue("CN=u4,%s" % self.ou_users in dn_list)
res1 = self.ldb.search(self.ou,
- scope=SCOPE_SUBTREE,
- expression="member:1.2.840.113556.1.4.1941:=cn=c1,%s" % self.ou_computers)
+ scope=SCOPE_SUBTREE,
+ expression="member:1.2.840.113556.1.4.1941:=cn=c1,%s" % self.ou_computers)
self.assertEqual(len(res1), 8)
dn_list = [str(res.dn) for res in res1]
self.assertTrue("CN=u1,%s" % self.ou_users in dn_list)
self.assertTrue("CN=g4,%s" % self.ou_groups in dn_list)
res1 = self.ldb.search(self.ou,
- scope=SCOPE_SUBTREE,
- expression="memberOf=cn=c1,%s" % self.ou_computers)
+ scope=SCOPE_SUBTREE,
+ expression="memberOf=cn=c1,%s" % self.ou_computers)
self.assertEqual(len(res1), 0)
res1 = self.ldb.search(self.ou,
- scope=SCOPE_SUBTREE,
- expression="memberOf:1.2.840.113556.1.4.1941:=cn=c1,%s" % self.ou_computers)
+ scope=SCOPE_SUBTREE,
+ expression="memberOf:1.2.840.113556.1.4.1941:=cn=c1,%s" % self.ou_computers)
self.assertEqual(len(res1), 0)
def test_c2_members(self):
res1 = self.ldb.search(self.ou,
- scope=SCOPE_SUBTREE,
- expression="member=cn=c2,%s" % self.ou_computers)
+ scope=SCOPE_SUBTREE,
+ expression="member=cn=c2,%s" % self.ou_computers)
self.assertEqual(len(res1), 1)
self.assertEqual(str(res1[0].dn), "CN=u1,%s" % self.ou_users)
res1 = self.ldb.search(self.ou,
- scope=SCOPE_SUBTREE,
- expression="member:1.2.840.113556.1.4.1941:=cn=c2,%s" % self.ou_computers)
+ scope=SCOPE_SUBTREE,
+ expression="member:1.2.840.113556.1.4.1941:=cn=c2,%s" % self.ou_computers)
self.assertEqual(len(res1), 1)
self.assertEqual(str(res1[0].dn), "CN=u1,%s" % self.ou_users)
res1 = self.ldb.search(self.ou,
- scope=SCOPE_SUBTREE,
- expression="memberOf=cn=c2,%s" % self.ou_computers)
+ scope=SCOPE_SUBTREE,
+ expression="memberOf=cn=c2,%s" % self.ou_computers)
self.assertEqual(len(res1), 0)
res1 = self.ldb.search(self.ou,
- scope=SCOPE_SUBTREE,
- expression="memberOf:1.2.840.113556.1.4.1941:=cn=c2,%s" % self.ou_computers)
+ scope=SCOPE_SUBTREE,
+ expression="memberOf:1.2.840.113556.1.4.1941:=cn=c2,%s" % self.ou_computers)
self.assertEqual(len(res1), 0)
def test_c3_members(self):
res1 = self.ldb.search(self.ou,
- scope=SCOPE_SUBTREE,
- expression="member=cn=c3,%s" % self.ou_computers)
+ scope=SCOPE_SUBTREE,
+ expression="member=cn=c3,%s" % self.ou_computers)
self.assertEqual(len(res1), 2)
dn_list = [str(res.dn) for res in res1]
self.assertTrue("CN=g3,%s" % self.ou_groups in dn_list)
self.assertTrue("CN=u2,%s" % self.ou_users in dn_list)
res1 = self.ldb.search(self.ou,
- scope=SCOPE_SUBTREE,
- expression="member:1.2.840.113556.1.4.1941:=cn=c3,%s" % self.ou_computers)
+ scope=SCOPE_SUBTREE,
+ expression="member:1.2.840.113556.1.4.1941:=cn=c3,%s" % self.ou_computers)
self.assertEqual(len(res1), 4)
dn_list = [str(res.dn) for res in res1]
self.assertTrue("CN=u2,%s" % self.ou_users in dn_list)
self.assertTrue("CN=g3,%s" % self.ou_groups in dn_list)
res1 = self.ldb.search(self.ou,
- scope=SCOPE_SUBTREE,
- expression="memberOf=cn=c3,%s" % self.ou_computers)
+ scope=SCOPE_SUBTREE,
+ expression="memberOf=cn=c3,%s" % self.ou_computers)
self.assertEqual(len(res1), 0)
res1 = self.ldb.search(self.ou,
- scope=SCOPE_SUBTREE,
- expression="memberOf:1.2.840.113556.1.4.1941:=cn=c3,%s" % self.ou_computers)
+ scope=SCOPE_SUBTREE,
+ expression="memberOf:1.2.840.113556.1.4.1941:=cn=c3,%s" % self.ou_computers)
self.assertEqual(len(res1), 0)
def test_c4_members(self):
res1 = self.ldb.search(self.ou,
- scope=SCOPE_SUBTREE,
- expression="member=cn=c4,%s" % self.ou_computers)
+ scope=SCOPE_SUBTREE,
+ expression="member=cn=c4,%s" % self.ou_computers)
self.assertEqual(len(res1), 2)
dn_list = [str(res.dn) for res in res1]
self.assertTrue("CN=g4,%s" % self.ou_groups in dn_list)
self.assertTrue("CN=u4,%s" % self.ou_users in dn_list)
res1 = self.ldb.search(self.ou,
- scope=SCOPE_SUBTREE,
- expression="member:1.2.840.113556.1.4.1941:=cn=c4,%s" % self.ou_computers)
+ scope=SCOPE_SUBTREE,
+ expression="member:1.2.840.113556.1.4.1941:=cn=c4,%s" % self.ou_computers)
self.assertEqual(len(res1), 2)
dn_list = [str(res.dn) for res in res1]
self.assertTrue("CN=u4,%s" % self.ou_users in dn_list)
self.assertTrue("CN=g4,%s" % self.ou_groups in dn_list)
res1 = self.ldb.search(self.ou,
- scope=SCOPE_SUBTREE,
- expression="memberOf=cn=c4,%s" % self.ou_computers)
+ scope=SCOPE_SUBTREE,
+ expression="memberOf=cn=c4,%s" % self.ou_computers)
self.assertEqual(len(res1), 0)
res1 = self.ldb.search(self.ou,
- scope=SCOPE_SUBTREE,
- expression="memberOf:1.2.840.113556.1.4.1941:=cn=c4,%s" % self.ou_computers)
+ scope=SCOPE_SUBTREE,
+ expression="memberOf:1.2.840.113556.1.4.1941:=cn=c4,%s" % self.ou_computers)
self.assertEqual(len(res1), 0)
def test_or_member_queries(self):
res1 = self.ldb.search(self.ou,
- scope=SCOPE_SUBTREE,
- expression=("(|(member:1.2.840.113556.1.4.1941:=cn=c1,%s)"
- "(member:1.2.840.113556.1.4.1941:=cn=c2,%s))") % (
- self.ou_computers, self.ou_computers))
+ scope=SCOPE_SUBTREE,
+ expression=("(|(member:1.2.840.113556.1.4.1941:=cn=c1,%s)"
+ "(member:1.2.840.113556.1.4.1941:=cn=c2,%s))") % (
+ self.ou_computers, self.ou_computers))
self.assertEqual(len(res1), 8)
dn_list = [str(res.dn) for res in res1]
self.assertTrue("CN=u1,%s" % self.ou_users in dn_list)
self.assertTrue("CN=g4,%s" % self.ou_groups in dn_list)
res1 = self.ldb.search(self.ou,
- scope=SCOPE_SUBTREE,
- expression=("(|(member:1.2.840.113556.1.4.1941:=cn=c2,%s)"
- "(member:1.2.840.113556.1.4.1941:=cn=c3,%s))") % (
- self.ou_computers, self.ou_computers))
+ scope=SCOPE_SUBTREE,
+ expression=("(|(member:1.2.840.113556.1.4.1941:=cn=c2,%s)"
+ "(member:1.2.840.113556.1.4.1941:=cn=c3,%s))") % (
+ self.ou_computers, self.ou_computers))
self.assertEqual(len(res1), 5)
dn_list = [str(res.dn) for res in res1]
self.assertTrue("CN=u1,%s" % self.ou_users in dn_list)
self.assertTrue("CN=g3,%s" % self.ou_groups in dn_list)
res1 = self.ldb.search(self.ou,
- scope=SCOPE_SUBTREE,
- expression=("(|(member:1.2.840.113556.1.4.1941:=cn=c2,%s)"
- "(member:1.2.840.113556.1.4.1941:=cn=c4,%s))") % (
- self.ou_computers, self.ou_computers))
+ scope=SCOPE_SUBTREE,
+ expression=("(|(member:1.2.840.113556.1.4.1941:=cn=c2,%s)"
+ "(member:1.2.840.113556.1.4.1941:=cn=c4,%s))") % (
+ self.ou_computers, self.ou_computers))
self.assertEqual(len(res1), 3)
dn_list = [str(res.dn) for res in res1]
self.assertTrue("CN=u1,%s" % self.ou_users in dn_list)
self.assertTrue("CN=g4,%s" % self.ou_groups in dn_list)
res1 = self.ldb.search(self.ou,
- scope=SCOPE_SUBTREE,
- expression=("(|(member:1.2.840.113556.1.4.1941:=cn=c3,%s)"
- "(member:1.2.840.113556.1.4.1941:=cn=c4,%s))") % (
- self.ou_computers, self.ou_computers))
+ scope=SCOPE_SUBTREE,
+ expression=("(|(member:1.2.840.113556.1.4.1941:=cn=c3,%s)"
+ "(member:1.2.840.113556.1.4.1941:=cn=c4,%s))") % (
+ self.ou_computers, self.ou_computers))
self.assertEqual(len(res1), 6)
dn_list = [str(res.dn) for res in res1]
self.assertTrue("CN=u2,%s" % self.ou_users in dn_list)
self.assertTrue("CN=g4,%s" % self.ou_groups in dn_list)
res1 = self.ldb.search(self.ou,
- scope=SCOPE_SUBTREE,
- expression=("(|(member:1.2.840.113556.1.4.1941:=cn=u1,%s)"
- "(member:1.2.840.113556.1.4.1941:=cn=c4,%s))") % (
- self.ou_users, self.ou_computers))
+ scope=SCOPE_SUBTREE,
+ expression=("(|(member:1.2.840.113556.1.4.1941:=cn=u1,%s)"
+ "(member:1.2.840.113556.1.4.1941:=cn=c4,%s))") % (
+ self.ou_users, self.ou_computers))
self.assertEqual(len(res1), 2)
dn_list = [str(res.dn) for res in res1]
self.assertTrue("CN=u4,%s" % self.ou_users in dn_list)
def test_and_member_queries(self):
res1 = self.ldb.search(self.ou,
- scope=SCOPE_SUBTREE,
- expression=("(&(member:1.2.840.113556.1.4.1941:=cn=c1,%s)"
- "(member:1.2.840.113556.1.4.1941:=cn=c2,%s))") % (
- self.ou_computers, self.ou_computers))
+ scope=SCOPE_SUBTREE,
+ expression=("(&(member:1.2.840.113556.1.4.1941:=cn=c1,%s)"
+ "(member:1.2.840.113556.1.4.1941:=cn=c2,%s))") % (
+ self.ou_computers, self.ou_computers))
self.assertEqual(len(res1), 1)
self.assertEqual(str(res1[0].dn), "CN=u1,%s" % self.ou_users)
res1 = self.ldb.search(self.ou,
- scope=SCOPE_SUBTREE,
- expression=("(&(member:1.2.840.113556.1.4.1941:=cn=c2,%s)"
- "(member:1.2.840.113556.1.4.1941:=cn=c3,%s))") % (
- self.ou_computers, self.ou_computers))
+ scope=SCOPE_SUBTREE,
+ expression=("(&(member:1.2.840.113556.1.4.1941:=cn=c2,%s)"
+ "(member:1.2.840.113556.1.4.1941:=cn=c3,%s))") % (
+ self.ou_computers, self.ou_computers))
self.assertEqual(len(res1), 0)
res1 = self.ldb.search(self.ou,
- scope=SCOPE_SUBTREE,
- expression=("(&(member:1.2.840.113556.1.4.1941:=cn=c3,%s)"
- "(member:1.2.840.113556.1.4.1941:=cn=u3,%s))") % (
- self.ou_computers, self.ou_users))
+ scope=SCOPE_SUBTREE,
+ expression=("(&(member:1.2.840.113556.1.4.1941:=cn=c3,%s)"
+ "(member:1.2.840.113556.1.4.1941:=cn=u3,%s))") % (
+ self.ou_computers, self.ou_users))
self.assertEqual(len(res1), 3)
dn_list = [str(res.dn) for res in res1]
self.assertTrue("CN=g1,%s" % self.ou_groups in dn_list)
self.assertTrue("CN=g3,%s" % self.ou_groups in dn_list)
res1 = self.ldb.search(self.ou,
- scope=SCOPE_SUBTREE,
- expression=("(&(member:1.2.840.113556.1.4.1941:=cn=c1,%s)"
- "(member:1.2.840.113556.1.4.1941:=cn=u4,%s))") % (
- self.ou_computers, self.ou_computers))
+ scope=SCOPE_SUBTREE,
+ expression=("(&(member:1.2.840.113556.1.4.1941:=cn=c1,%s)"
+ "(member:1.2.840.113556.1.4.1941:=cn=u4,%s))") % (
+ self.ou_computers, self.ou_computers))
self.assertEqual(len(res1), 0)
def test_or_memberOf_queries(self):
res1 = self.ldb.search(self.ou,
- scope=SCOPE_SUBTREE,
- expression=("(|(memberOf:1.2.840.113556.1.4.1941:=cn=g1,%s)"
- "(memberOf:1.2.840.113556.1.4.1941:=cn=g2,%s))") % (
- self.ou_groups, self.ou_groups))
+ scope=SCOPE_SUBTREE,
+ expression=("(|(memberOf:1.2.840.113556.1.4.1941:=cn=g1,%s)"
+ "(memberOf:1.2.840.113556.1.4.1941:=cn=g2,%s))") % (
+ self.ou_groups, self.ou_groups))
self.assertEqual(len(res1), 6)
dn_list = [str(res.dn) for res in res1]
self.assertTrue("CN=u2,%s" % self.ou_users in dn_list)
self.assertTrue("CN=c3,%s" % self.ou_computers in dn_list)
res1 = self.ldb.search(self.ou,
- scope=SCOPE_SUBTREE,
- expression=("(|(memberOf:1.2.840.113556.1.4.1941:=cn=g1,%s)"
- "(memberOf:1.2.840.113556.1.4.1941:=cn=g3,%s))") % (
- self.ou_groups, self.ou_groups))
+ scope=SCOPE_SUBTREE,
+ expression=("(|(memberOf:1.2.840.113556.1.4.1941:=cn=g1,%s)"
+ "(memberOf:1.2.840.113556.1.4.1941:=cn=g3,%s))") % (
+ self.ou_groups, self.ou_groups))
self.assertEqual(len(res1), 6)
dn_list = [str(res.dn) for res in res1]
self.assertTrue("CN=u2,%s" % self.ou_users in dn_list)
self.assertTrue("CN=c3,%s" % self.ou_computers in dn_list)
res1 = self.ldb.search(self.ou,
- scope=SCOPE_SUBTREE,
- expression=("(|(memberOf:1.2.840.113556.1.4.1941:=cn=g1,%s)"
- "(memberOf:1.2.840.113556.1.4.1941:=cn=g4,%s))") % (
- self.ou_groups, self.ou_groups))
+ scope=SCOPE_SUBTREE,
+ expression=("(|(memberOf:1.2.840.113556.1.4.1941:=cn=g1,%s)"
+ "(memberOf:1.2.840.113556.1.4.1941:=cn=g4,%s))") % (
+ self.ou_groups, self.ou_groups))
self.assertEqual(len(res1), 8)
dn_list = [str(res.dn) for res in res1]
self.assertTrue("CN=u2,%s" % self.ou_users in dn_list)
self.assertTrue("CN=c4,%s" % self.ou_computers in dn_list)
res1 = self.ldb.search(self.ou,
- scope=SCOPE_SUBTREE,
- expression=("(|(memberOf:1.2.840.113556.1.4.1941:=cn=g2,%s)"
- "(memberOf:1.2.840.113556.1.4.1941:=cn=g3,%s))") %
- (self.ou_groups, self.ou_groups))
+ scope=SCOPE_SUBTREE,
+ expression=("(|(memberOf:1.2.840.113556.1.4.1941:=cn=g2,%s)"
+ "(memberOf:1.2.840.113556.1.4.1941:=cn=g3,%s))") %
+ (self.ou_groups, self.ou_groups))
self.assertEqual(len(res1), 5)
dn_list = [str(res.dn) for res in res1]
self.assertTrue("CN=u2,%s" % self.ou_users in dn_list)
self.assertTrue("CN=c3,%s" % self.ou_computers in dn_list)
res1 = self.ldb.search(self.ou,
- scope=SCOPE_SUBTREE,
- expression=("(|(memberOf:1.2.840.113556.1.4.1941:=cn=g2,%s)"
- "(memberOf:1.2.840.113556.1.4.1941:=cn=g4,%s))") % (
- self.ou_groups, self.ou_groups))
+ scope=SCOPE_SUBTREE,
+ expression=("(|(memberOf:1.2.840.113556.1.4.1941:=cn=g2,%s)"
+ "(memberOf:1.2.840.113556.1.4.1941:=cn=g4,%s))") % (
+ self.ou_groups, self.ou_groups))
self.assertEqual(len(res1), 7)
dn_list = [str(res.dn) for res in res1]
self.assertTrue("CN=u2,%s" % self.ou_users in dn_list)
self.assertTrue("CN=c4,%s" % self.ou_computers in dn_list)
res1 = self.ldb.search(self.ou,
- scope=SCOPE_SUBTREE,
- expression=("(|(memberOf:1.2.840.113556.1.4.1941:=cn=g3,%s)"
- "(memberOf:1.2.840.113556.1.4.1941:=cn=g4,%s))") % (
- self.ou_groups, self.ou_groups))
+ scope=SCOPE_SUBTREE,
+ expression=("(|(memberOf:1.2.840.113556.1.4.1941:=cn=g3,%s)"
+ "(memberOf:1.2.840.113556.1.4.1941:=cn=g4,%s))") % (
+ self.ou_groups, self.ou_groups))
self.assertEqual(len(res1), 5)
dn_list = [str(res.dn) for res in res1]
self.assertTrue("CN=u3,%s" % self.ou_users in dn_list)
def test_and_memberOf_queries(self):
res1 = self.ldb.search(self.ou,
- scope=SCOPE_SUBTREE,
- expression=("(&(memberOf:1.2.840.113556.1.4.1941:=cn=g1,%s)"
- "(memberOf:1.2.840.113556.1.4.1941:=cn=g2,%s))") % (
- self.ou_groups, self.ou_groups))
+ scope=SCOPE_SUBTREE,
+ expression=("(&(memberOf:1.2.840.113556.1.4.1941:=cn=g1,%s)"
+ "(memberOf:1.2.840.113556.1.4.1941:=cn=g2,%s))") % (
+ self.ou_groups, self.ou_groups))
self.assertEqual(len(res1), 5)
dn_list = [str(res.dn) for res in res1]
self.assertTrue("CN=u2,%s" % self.ou_users in dn_list)
self.assertTrue("CN=c3,%s" % self.ou_computers in dn_list)
res1 = self.ldb.search(self.ou,
- scope=SCOPE_SUBTREE,
- expression=("(&(memberOf:1.2.840.113556.1.4.1941:=cn=g1,%s)"
- "(memberOf:1.2.840.113556.1.4.1941:=cn=g3,%s))") % (
- self.ou_groups, self.ou_groups))
+ scope=SCOPE_SUBTREE,
+ expression=("(&(memberOf:1.2.840.113556.1.4.1941:=cn=g1,%s)"
+ "(memberOf:1.2.840.113556.1.4.1941:=cn=g3,%s))") % (
+ self.ou_groups, self.ou_groups))
self.assertEqual(len(res1), 3)
dn_list = [str(res.dn) for res in res1]
self.assertTrue("CN=u3,%s" % self.ou_users in dn_list)
self.assertTrue("CN=c3,%s" % self.ou_computers in dn_list)
res1 = self.ldb.search(self.ou,
- scope=SCOPE_SUBTREE,
- expression=("(&(memberOf:1.2.840.113556.1.4.1941:=cn=g1,%s)"
- "(memberOf:1.2.840.113556.1.4.1941:=cn=g4,%s))") % (
- self.ou_groups, self.ou_groups))
+ scope=SCOPE_SUBTREE,
+ expression=("(&(memberOf:1.2.840.113556.1.4.1941:=cn=g1,%s)"
+ "(memberOf:1.2.840.113556.1.4.1941:=cn=g4,%s))") % (
+ self.ou_groups, self.ou_groups))
self.assertEqual(len(res1), 2)
dn_list = [str(res.dn) for res in res1]
self.assertTrue("CN=u3,%s" % self.ou_users in dn_list)
self.assertTrue("CN=c1,%s" % self.ou_computers in dn_list)
res1 = self.ldb.search(self.ou,
- scope=SCOPE_SUBTREE,
- expression=("(&(memberOf:1.2.840.113556.1.4.1941:=cn=g2,%s)"
- "(memberOf:1.2.840.113556.1.4.1941:=cn=g3,%s))") % (
- self.ou_groups, self.ou_groups))
+ scope=SCOPE_SUBTREE,
+ expression=("(&(memberOf:1.2.840.113556.1.4.1941:=cn=g2,%s)"
+ "(memberOf:1.2.840.113556.1.4.1941:=cn=g3,%s))") % (
+ self.ou_groups, self.ou_groups))
self.assertEqual(len(res1), 3)
dn_list = [str(res.dn) for res in res1]
self.assertTrue("CN=u3,%s" % self.ou_users in dn_list)
self.assertTrue("CN=c3,%s" % self.ou_computers in dn_list)
res1 = self.ldb.search(self.ou,
- scope=SCOPE_SUBTREE,
- expression=("(&(memberOf:1.2.840.113556.1.4.1941:=cn=g2,%s)"
- "(memberOf:1.2.840.113556.1.4.1941:=cn=g4,%s))") % (
- self.ou_groups, self.ou_groups))
+ scope=SCOPE_SUBTREE,
+ expression=("(&(memberOf:1.2.840.113556.1.4.1941:=cn=g2,%s)"
+ "(memberOf:1.2.840.113556.1.4.1941:=cn=g4,%s))") % (
+ self.ou_groups, self.ou_groups))
self.assertEqual(len(res1), 2)
dn_list = [str(res.dn) for res in res1]
self.assertTrue("CN=u3,%s" % self.ou_users in dn_list)
self.assertTrue("CN=c1,%s" % self.ou_computers in dn_list)
res1 = self.ldb.search(self.ou,
- scope=SCOPE_SUBTREE,
- expression=("(&(memberOf:1.2.840.113556.1.4.1941:=cn=g3,%s)"
- "(memberOf:1.2.840.113556.1.4.1941:=cn=g4,%s))") % (
- self.ou_groups, self.ou_groups))
+ scope=SCOPE_SUBTREE,
+ expression=("(&(memberOf:1.2.840.113556.1.4.1941:=cn=g3,%s)"
+ "(memberOf:1.2.840.113556.1.4.1941:=cn=g4,%s))") % (
+ self.ou_groups, self.ou_groups))
self.assertEqual(len(res1), 2)
dn_list = [str(res.dn) for res in res1]
self.assertTrue("CN=u3,%s" % self.ou_users in dn_list)
self.assertTrue("CN=c1,%s" % self.ou_computers in dn_list)
res1 = self.ldb.search(self.ou,
- scope=SCOPE_SUBTREE,
- expression=("(&(memberOf:1.2.840.113556.1.4.1941:=cn=g1,%s)"
- "(memberOf:1.2.840.113556.1.4.1941:=cn=c1,%s))") % (
- self.ou_groups, self.ou_computers))
+ scope=SCOPE_SUBTREE,
+ expression=("(&(memberOf:1.2.840.113556.1.4.1941:=cn=g1,%s)"
+ "(memberOf:1.2.840.113556.1.4.1941:=cn=c1,%s))") % (
+ self.ou_groups, self.ou_computers))
self.assertEqual(len(res1), 0)
parser = optparse.OptionParser("match_rules.py [options] <host>")
def test_add_dict_bytes_dn(self):
l = ldb.Ldb(self.url(), flags=self.flags())
m = {"dn": b"dc=foo6", "bla": b"bla",
- "objectUUID": b"0123456789abcdef"}
+ "objectUUID": b"0123456789abcdef"}
self.assertEqual(len(l.search()), 0)
l.add(m)
try:
class ARecord(dnsserver.DNS_RPC_RECORD):
def __init__(self, ip_addr, serial=1, ttl=900, rank=dnsp.DNS_RANK_ZONE,
- node_flag=0):
+ node_flag=0):
super(ARecord, self).__init__()
self.wType = dnsp.DNS_TYPE_A
self.dwFlags = rank | node_flag
class AAAARecord(dnsserver.DNS_RPC_RECORD):
def __init__(self, ip6_addr, serial=1, ttl=900, rank=dnsp.DNS_RANK_ZONE,
- node_flag=0):
+ node_flag=0):
super(AAAARecord, self).__init__()
self.wType = dnsp.DNS_TYPE_AAAA
self.dwFlags = rank | node_flag
class CNameRecord(dnsserver.DNS_RPC_RECORD):
def __init__(self, cname, serial=1, ttl=900, rank=dnsp.DNS_RANK_ZONE,
- node_flag=0):
+ node_flag=0):
super(CNameRecord, self).__init__()
self.wType = dnsp.DNS_TYPE_CNAME
self.dwFlags = rank | node_flag
class NSRecord(dnsserver.DNS_RPC_RECORD):
def __init__(self, dns_server, serial=1, ttl=900, rank=dnsp.DNS_RANK_ZONE,
- node_flag=0):
+ node_flag=0):
super(NSRecord, self).__init__()
self.wType = dnsp.DNS_TYPE_NS
self.dwFlags = rank | node_flag
client += 1
print(("we have %d conversations at rate %f" %
- (len(conversations), rate)), file=sys.stderr)
+ (len(conversations), rate)), file=sys.stderr)
conversations.sort()
return conversations
finally:
for s in (15, 15, 9):
print(("killing %d children with -%d" %
- (len(children), s)), file=sys.stderr)
+ (len(children), s)), file=sys.stderr)
for pid in children:
try:
os.kill(pid, s)
ctx.dnspass = samba.generate_random_password(128, 255)
recs = ctx.samdb.parse_ldif(read_and_sub_file(setup_path("provision_dns_add_samba.ldif"),
- {"DNSDOMAIN": ctx.dnsdomain,
- "DOMAINDN": ctx.base_dn,
- "HOSTNAME" : ctx.myname,
- "DNSPASS_B64": b64encode(ctx.dnspass.encode('utf-16-le')).decode('utf8'),
- "DNSNAME" : ctx.dnshostname}))
+ {"DNSDOMAIN": ctx.dnsdomain,
+ "DOMAINDN": ctx.base_dn,
+ "HOSTNAME" : ctx.myname,
+ "DNSPASS_B64": b64encode(ctx.dnspass.encode('utf-16-le')).decode('utf8'),
+ "DNSNAME" : ctx.dnshostname}))
for changetype, msg in recs:
assert changetype == ldb.CHANGETYPE_NONE
dns_acct_dn = msg["dn"]
if nc in ctx.nc_list:
print("Replicating %s" % (str(nc)))
repl.replicate(nc, source_dsa_invocation_id,
- destination_dsa_guid, rodc=ctx.RODC,
- replica_flags=ctx.replica_flags)
+ destination_dsa_guid, rodc=ctx.RODC,
+ replica_flags=ctx.replica_flags)
if ctx.RODC:
repl.replicate(ctx.acct_dn, source_dsa_invocation_id,
client_version = dnsserver.DNS_CLIENT_VERSION_LONGHORN
record_type = dnsp.DNS_TYPE_A
select_flags = dnsserver.DNS_RPC_VIEW_AUTHORITY_DATA |\
- dnsserver.DNS_RPC_VIEW_NO_CHILDREN
+ dnsserver.DNS_RPC_VIEW_NO_CHILDREN
zone = ctx.dnsdomain
msdcs_zone = "_msdcs.%s" % ctx.dnsforest
if ndr_unpack_in_fn is None:
raise TypeError("%r is not a NDR function object" % object)
ndr_unpack_in_fn(data, bigendian=bigendian, ndr64=ndr64,
- allow_remaining=allow_remaining)
+ allow_remaining=allow_remaining)
return object
if ndr_unpack_out_fn is None:
raise TypeError("%r is not a NDR function object" % object)
ndr_unpack_out_fn(data, bigendian=bigendian, ndr64=ndr64,
- allow_remaining=allow_remaining)
+ allow_remaining=allow_remaining)
return object
Option("-H", "--URL", help="LDB URL for database or target server",
type=str, metavar="URL", dest="H"),
Option("--computerou",
- help=("DN of alternative location (with or without domainDN "
- "counterpart) to default CN=Computers in which new "
- "computer object will be created. E.g. 'OU=<OU name>'"),
- type=str),
+ help=("DN of alternative location (with or without domainDN "
+ "counterpart) to default CN=Computers in which new "
+ "computer object will be created. E.g. 'OU=<OU name>'"),
+ type=str),
Option("--description", help="Computers's description", type=str),
Option("--prepare-oldjoin",
help="Prepare enabled machine account for oldjoin mechanism",
def boot_method_string(boot_method):
enum_defs = [ 'DNS_BOOT_METHOD_UNINITIALIZED', 'DNS_BOOT_METHOD_FILE',
- 'DNS_BOOT_METHOD_REGISTRY', 'DNS_BOOT_METHOD_DIRECTORY' ]
+ 'DNS_BOOT_METHOD_REGISTRY', 'DNS_BOOT_METHOD_DIRECTORY' ]
return enum_string(dnsserver, enum_defs, boot_method)
def name_check_flag_string(check_flag):
enum_defs = [ 'DNS_ALLOW_RFC_NAMES_ONLY', 'DNS_ALLOW_NONRFC_NAMES',
- 'DNS_ALLOW_MULTIBYTE_NAMES', 'DNS_ALLOW_ALL_NAMES' ]
+ 'DNS_ALLOW_MULTIBYTE_NAMES', 'DNS_ALLOW_ALL_NAMES' ]
return enum_string(dnsserver, enum_defs, check_flag)
def zone_type_string(zone_type):
enum_defs = [ 'DNS_ZONE_TYPE_CACHE', 'DNS_ZONE_TYPE_PRIMARY',
- 'DNS_ZONE_TYPE_SECONDARY', 'DNS_ZONE_TYPE_STUB',
- 'DNS_ZONE_TYPE_FORWARDER', 'DNS_ZONE_TYPE_SECONDARY_CACHE' ]
+ 'DNS_ZONE_TYPE_SECONDARY', 'DNS_ZONE_TYPE_STUB',
+ 'DNS_ZONE_TYPE_FORWARDER', 'DNS_ZONE_TYPE_SECONDARY_CACHE' ]
return enum_string(dnsp, enum_defs, zone_type)
def zone_update_string(zone_update):
enum_defs = [ 'DNS_ZONE_UPDATE_OFF', 'DNS_ZONE_UPDATE_UNSECURE',
- 'DNS_ZONE_UPDATE_SECURE' ]
+ 'DNS_ZONE_UPDATE_SECURE' ]
return enum_string(dnsp, enum_defs, zone_update)
def zone_secondary_security_string(security):
enum_defs = [ 'DNS_ZONE_SECSECURE_NO_SECURITY', 'DNS_ZONE_SECSECURE_NS_ONLY',
- 'DNS_ZONE_SECSECURE_LIST_ONLY', 'DNS_ZONE_SECSECURE_NO_XFER' ]
+ 'DNS_ZONE_SECSECURE_LIST_ONLY', 'DNS_ZONE_SECSECURE_NO_XFER' ]
return enum_string(dnsserver, enum_defs, security)
def zone_notify_level_string(notify_level):
enum_defs = [ 'DNS_ZONE_NOTIFY_OFF', 'DNS_ZONE_NOTIFY_ALL_SECONDARIES',
- 'DNS_ZONE_NOTIFY_LIST_ONLY' ]
+ 'DNS_ZONE_NOTIFY_LIST_ONLY' ]
return enum_string(dnsserver, enum_defs, notify_level)
if typeid != dnsserver.DNSSRV_TYPEID_SERVER_INFO:
outf.write(' aipServerAddrs : %s\n' %
- ip4_array_string(serverinfo.aipServerAddrs))
+ ip4_array_string(serverinfo.aipServerAddrs))
outf.write(' aipListenAddrs : %s\n' %
- ip4_array_string(serverinfo.aipListenAddrs))
+ ip4_array_string(serverinfo.aipListenAddrs))
outf.write(' aipForwarders : %s\n' %
- ip4_array_string(serverinfo.aipForwarders))
+ ip4_array_string(serverinfo.aipForwarders))
else:
outf.write(' aipServerAddrs : %s\n' %
- dns_addr_array_string(serverinfo.aipServerAddrs))
+ dns_addr_array_string(serverinfo.aipServerAddrs))
outf.write(' aipListenAddrs : %s\n' %
- dns_addr_array_string(serverinfo.aipListenAddrs))
+ dns_addr_array_string(serverinfo.aipListenAddrs))
outf.write(' aipForwarders : %s\n' %
- dns_addr_array_string(serverinfo.aipForwarders))
+ dns_addr_array_string(serverinfo.aipForwarders))
outf.write(' dwLogLevel : %d\n' % serverinfo.dwLogLevel)
outf.write(' dwDebugLevel : %d\n' % serverinfo.dwDebugLevel)
outf.write(' pszDataFile : %s\n' % zoneinfo.pszDataFile)
if typeid != dnsserver.DNSSRV_TYPEID_ZONE_INFO:
outf.write(' aipMasters : %s\n' %
- ip4_array_string(zoneinfo.aipMasters))
+ ip4_array_string(zoneinfo.aipMasters))
else:
outf.write(' aipMasters : %s\n' %
- dns_addr_array_string(zoneinfo.aipMasters))
+ dns_addr_array_string(zoneinfo.aipMasters))
outf.write(' fSecureSecondaries : %s\n' % zone_secondary_security_string(zoneinfo.fSecureSecondaries))
outf.write(' fNotifyLevel : %s\n' % zone_notify_level_string(zoneinfo.fNotifyLevel))
if typeid != dnsserver.DNSSRV_TYPEID_ZONE_INFO:
outf.write(' aipSecondaries : %s\n' %
- ip4_array_string(zoneinfo.aipSecondaries))
+ ip4_array_string(zoneinfo.aipSecondaries))
outf.write(' aipNotify : %s\n' %
- ip4_array_string(zoneinfo.aipNotify))
+ ip4_array_string(zoneinfo.aipNotify))
else:
outf.write(' aipSecondaries : %s\n' %
- dns_addr_array_string(zoneinfo.aipSecondaries))
+ dns_addr_array_string(zoneinfo.aipSecondaries))
outf.write(' aipNotify : %s\n' %
- dns_addr_array_string(zoneinfo.aipNotify))
+ dns_addr_array_string(zoneinfo.aipNotify))
outf.write(' fUseWins : %s\n' % bool_string(zoneinfo.fUseWins))
outf.write(' fUseNbstat : %s\n' % bool_string(zoneinfo.fUseNbstat))
outf.write(' fAging : %s\n' % bool_string(zoneinfo.fAging))
outf.write(' dwAvailForScavengeTime : %d\n' % zoneinfo.dwAvailForScavengeTime)
if typeid != dnsserver.DNSSRV_TYPEID_ZONE_INFO:
outf.write(' aipScavengeServers : %s\n' %
- ip4_array_string(zoneinfo.aipScavengeServers))
+ ip4_array_string(zoneinfo.aipScavengeServers))
else:
outf.write(' aipScavengeServers : %s\n' %
- dns_addr_array_string(zoneinfo.aipScavengeServers))
+ dns_addr_array_string(zoneinfo.aipScavengeServers))
if typeid != dnsserver.DNSSRV_TYPEID_ZONE_INFO_W2K:
outf.write(' dwRpcStructureVersion : 0x%x\n' % zoneinfo.dwRpcStructureVersion)
outf.write(' fForwarderSlave : %d\n' % zoneinfo.fForwarderSlave)
if typeid != dnsserver.DNSSRV_TYPEID_ZONE_INFO:
outf.write(' aipLocalMasters : %s\n' %
- ip4_array_string(zoneinfo.aipLocalMasters))
+ ip4_array_string(zoneinfo.aipLocalMasters))
else:
outf.write(' aipLocalMasters : %s\n' %
- dns_addr_array_string(zoneinfo.aipLocalMasters))
+ dns_addr_array_string(zoneinfo.aipLocalMasters))
outf.write(' dwDpFlags : %s\n' % dp_flags_string(zoneinfo.dwDpFlags))
outf.write(' pszDpFqdn : %s\n' % zoneinfo.pszDpFqdn)
outf.write(' pwszZoneDn : %s\n' % zoneinfo.pwszZoneDn)
takes_options = [
Option('--client-version', help='Client Version',
- default='longhorn', metavar='w2k|dotnet|longhorn',
- choices=['w2k','dotnet','longhorn'], dest='cli_ver'),
+ default='longhorn', metavar='w2k|dotnet|longhorn',
+ choices=['w2k','dotnet','longhorn'], dest='cli_ver'),
]
def run(self, server, cli_ver, sambaopts=None, credopts=None,
takes_options = [
Option('--client-version', help='Client Version',
- default='longhorn', metavar='w2k|dotnet|longhorn',
- choices=['w2k','dotnet','longhorn'], dest='cli_ver'),
+ default='longhorn', metavar='w2k|dotnet|longhorn',
+ choices=['w2k','dotnet','longhorn'], dest='cli_ver'),
]
def run(self, server, zone, cli_ver, sambaopts=None, credopts=None,
takes_options = [
Option('--client-version', help='Client Version',
- default='longhorn', metavar='w2k|dotnet|longhorn',
- choices=['w2k','dotnet','longhorn'], dest='cli_ver'),
+ default='longhorn', metavar='w2k|dotnet|longhorn',
+ choices=['w2k','dotnet','longhorn'], dest='cli_ver'),
Option('--primary', help='List primary zones (default)',
- action='store_true', dest='primary'),
+ action='store_true', dest='primary'),
Option('--secondary', help='List secondary zones',
- action='store_true', dest='secondary'),
+ action='store_true', dest='secondary'),
Option('--cache', help='List cached zones',
- action='store_true', dest='cache'),
+ action='store_true', dest='cache'),
Option('--auto', help='List automatically created zones',
- action='store_true', dest='auto'),
+ action='store_true', dest='auto'),
Option('--forward', help='List forward zones',
- action='store_true', dest='forward'),
+ action='store_true', dest='forward'),
Option('--reverse', help='List reverse zones',
- action='store_true', dest='reverse'),
+ action='store_true', dest='reverse'),
Option('--ds', help='List directory integrated zones',
- action='store_true', dest='ds'),
+ action='store_true', dest='ds'),
Option('--non-ds', help='List non-directory zones',
- action='store_true', dest='nonds')
+ action='store_true', dest='nonds')
]
def run(self, server, cli_ver, primary=False, secondary=False, cache=False,
- auto=False, forward=False, reverse=False, ds=False, nonds=False,
- sambaopts=None, credopts=None, versionopts=None):
+ auto=False, forward=False, reverse=False, ds=False, nonds=False,
+ sambaopts=None, credopts=None, versionopts=None):
request_filter = 0
if primary:
client_version = dns_client_version(cli_ver)
typeid, res = dns_conn.DnssrvComplexOperation2(client_version,
- 0, server, None,
- 'EnumZones',
- dnsserver.DNSSRV_TYPEID_DWORD,
- request_filter)
+ 0, server, None,
+ 'EnumZones',
+ dnsserver.DNSSRV_TYPEID_DWORD,
+ request_filter)
if client_version == dnsserver.DNS_CLIENT_VERSION_W2K:
typeid = dnsserver.DNSSRV_TYPEID_ZONE_W2K
takes_options = [
Option('--client-version', help='Client Version',
- default='longhorn', metavar='w2k|dotnet|longhorn',
- choices=['w2k','dotnet','longhorn'], dest='cli_ver')
+ default='longhorn', metavar='w2k|dotnet|longhorn',
+ choices=['w2k','dotnet','longhorn'], dest='cli_ver')
]
def run(self, server, zone, cli_ver, sambaopts=None, credopts=None,
takes_options = [
Option('--authority', help='Search authoritative records (default)',
- action='store_true', dest='authority'),
+ action='store_true', dest='authority'),
Option('--cache', help='Search cached records',
- action='store_true', dest='cache'),
+ action='store_true', dest='cache'),
Option('--glue', help='Search glue records',
- action='store_true', dest='glue'),
+ action='store_true', dest='glue'),
Option('--root', help='Search root hints',
- action='store_true', dest='root'),
+ action='store_true', dest='root'),
Option('--additional', help='List additional records',
- action='store_true', dest='additional'),
+ action='store_true', dest='additional'),
Option('--no-children', help='Do not list children',
- action='store_true', dest='no_children'),
+ action='store_true', dest='no_children'),
Option('--only-children', help='List only children',
- action='store_true', dest='only_children')
+ action='store_true', dest='only_children')
]
def run(self, server, zone, name, rtype, authority=False, cache=False,
}
def run(self, server, zone, name, rtype, olddata, newdata,
- sambaopts=None, credopts=None, versionopts=None):
+ sambaopts=None, credopts=None, versionopts=None):
if rtype.upper() not in ('A','AAAA','PTR','CNAME','NS','MX','SOA','SRV','TXT'):
raise CommandError('Updating record of type %s is not supported' % rtype)
]
common_ntvfs_options = [
- Option("--use-ntvfs", help="Use NTVFS for the fileserver (default = no)",
- action="store_true")
+ Option("--use-ntvfs", help="Use NTVFS for the fileserver (default = no)",
+ action="store_true")
]
def get_testparm_var(testparm, smbconf, varname):
}
takes_options = [
- Option("--interactive", help="Ask for names", action="store_true"),
- Option("--domain", type="string", metavar="DOMAIN",
- help="NetBIOS domain name to use"),
- Option("--domain-guid", type="string", metavar="GUID",
- help="set domainguid (otherwise random)"),
- Option("--domain-sid", type="string", metavar="SID",
- help="set domainsid (otherwise random)"),
- Option("--ntds-guid", type="string", metavar="GUID",
- help="set NTDS object GUID (otherwise random)"),
- Option("--invocationid", type="string", metavar="GUID",
- help="set invocationid (otherwise random)"),
- Option("--host-name", type="string", metavar="HOSTNAME",
- help="set hostname"),
- Option("--host-ip", type="string", metavar="IPADDRESS",
- help="set IPv4 ipaddress"),
- Option("--host-ip6", type="string", metavar="IP6ADDRESS",
- help="set IPv6 ipaddress"),
- Option("--site", type="string", metavar="SITENAME",
- help="set site name"),
- Option("--adminpass", type="string", metavar="PASSWORD",
- help="choose admin password (otherwise random)"),
- Option("--krbtgtpass", type="string", metavar="PASSWORD",
- help="choose krbtgt password (otherwise random)"),
- Option("--dns-backend", type="choice", metavar="NAMESERVER-BACKEND",
- choices=["SAMBA_INTERNAL", "BIND9_FLATFILE", "BIND9_DLZ", "NONE"],
- help="The DNS server backend. SAMBA_INTERNAL is the builtin name server (default), "
- "BIND9_FLATFILE uses bind9 text database to store zone information, "
- "BIND9_DLZ uses samba4 AD to store zone information, "
- "NONE skips the DNS setup entirely (not recommended)",
- default="SAMBA_INTERNAL"),
- Option("--dnspass", type="string", metavar="PASSWORD",
- help="choose dns password (otherwise random)"),
- Option("--root", type="string", metavar="USERNAME",
- help="choose 'root' unix username"),
- Option("--nobody", type="string", metavar="USERNAME",
- help="choose 'nobody' user"),
- Option("--users", type="string", metavar="GROUPNAME",
- help="choose 'users' group"),
- Option("--blank", action="store_true",
- help="do not add users or groups, just the structure"),
- Option("--server-role", type="choice", metavar="ROLE",
- choices=["domain controller", "dc", "member server", "member", "standalone"],
- help="The server role (domain controller | dc | member server | member | standalone). Default is dc.",
- default="domain controller"),
- Option("--function-level", type="choice", metavar="FOR-FUN-LEVEL",
- choices=["2000", "2003", "2008", "2008_R2"],
- help="The domain and forest function level (2000 | 2003 | 2008 | 2008_R2 - always native). Default is (Windows) 2008_R2 Native.",
- default="2008_R2"),
- Option("--base-schema", type="choice", metavar="BASE-SCHEMA",
- choices=["2008_R2", "2008_R2_old", "2012", "2012_R2"],
- help="The base schema files to use. Default is (Windows) 2008_R2.",
- default="2008_R2"),
- Option("--next-rid", type="int", metavar="NEXTRID", default=1000,
- help="The initial nextRid value (only needed for upgrades). Default is 1000."),
- Option("--partitions-only",
- help="Configure Samba's partitions, but do not modify them (ie, join a BDC)", action="store_true"),
- Option("--use-rfc2307", action="store_true", help="Use AD to store posix attributes (default = no)"),
+ Option("--interactive", help="Ask for names", action="store_true"),
+ Option("--domain", type="string", metavar="DOMAIN",
+ help="NetBIOS domain name to use"),
+ Option("--domain-guid", type="string", metavar="GUID",
+ help="set domainguid (otherwise random)"),
+ Option("--domain-sid", type="string", metavar="SID",
+ help="set domainsid (otherwise random)"),
+ Option("--ntds-guid", type="string", metavar="GUID",
+ help="set NTDS object GUID (otherwise random)"),
+ Option("--invocationid", type="string", metavar="GUID",
+ help="set invocationid (otherwise random)"),
+ Option("--host-name", type="string", metavar="HOSTNAME",
+ help="set hostname"),
+ Option("--host-ip", type="string", metavar="IPADDRESS",
+ help="set IPv4 ipaddress"),
+ Option("--host-ip6", type="string", metavar="IP6ADDRESS",
+ help="set IPv6 ipaddress"),
+ Option("--site", type="string", metavar="SITENAME",
+ help="set site name"),
+ Option("--adminpass", type="string", metavar="PASSWORD",
+ help="choose admin password (otherwise random)"),
+ Option("--krbtgtpass", type="string", metavar="PASSWORD",
+ help="choose krbtgt password (otherwise random)"),
+ Option("--dns-backend", type="choice", metavar="NAMESERVER-BACKEND",
+ choices=["SAMBA_INTERNAL", "BIND9_FLATFILE", "BIND9_DLZ", "NONE"],
+ help="The DNS server backend. SAMBA_INTERNAL is the builtin name server (default), "
+ "BIND9_FLATFILE uses bind9 text database to store zone information, "
+ "BIND9_DLZ uses samba4 AD to store zone information, "
+ "NONE skips the DNS setup entirely (not recommended)",
+ default="SAMBA_INTERNAL"),
+ Option("--dnspass", type="string", metavar="PASSWORD",
+ help="choose dns password (otherwise random)"),
+ Option("--root", type="string", metavar="USERNAME",
+ help="choose 'root' unix username"),
+ Option("--nobody", type="string", metavar="USERNAME",
+ help="choose 'nobody' user"),
+ Option("--users", type="string", metavar="GROUPNAME",
+ help="choose 'users' group"),
+ Option("--blank", action="store_true",
+ help="do not add users or groups, just the structure"),
+ Option("--server-role", type="choice", metavar="ROLE",
+ choices=["domain controller", "dc", "member server", "member", "standalone"],
+ help="The server role (domain controller | dc | member server | member | standalone). Default is dc.",
+ default="domain controller"),
+ Option("--function-level", type="choice", metavar="FOR-FUN-LEVEL",
+ choices=["2000", "2003", "2008", "2008_R2"],
+ help="The domain and forest function level (2000 | 2003 | 2008 | 2008_R2 - always native). Default is (Windows) 2008_R2 Native.",
+ default="2008_R2"),
+ Option("--base-schema", type="choice", metavar="BASE-SCHEMA",
+ choices=["2008_R2", "2008_R2_old", "2012", "2012_R2"],
+ help="The base schema files to use. Default is (Windows) 2008_R2.",
+ default="2008_R2"),
+ Option("--next-rid", type="int", metavar="NEXTRID", default=1000,
+ help="The initial nextRid value (only needed for upgrades). Default is 1000."),
+ Option("--partitions-only",
+ help="Configure Samba's partitions, but do not modify them (ie, join a BDC)", action="store_true"),
+ Option("--use-rfc2307", action="store_true", help="Use AD to store posix attributes (default = no)"),
]
openldap_options = [
help="Test initialisation support for unsupported LDAP backend type (fedora-ds or openldap) DO NOT USE",
choices=["fedora-ds", "openldap"]),
Option("--ol-mmr-urls", type="string", metavar="LDAPSERVER",
-
help="List of LDAP-URLS [ ldap://<FQHN>:<PORT>/ (where <PORT> has to be different than 389!) ] separated with comma (\",\") for use with OpenLDAP-MMR (Multi-Master-Replication), e.g.: \"ldap://s4dc1:9000,ldap://s4dc2:9000\""),
+ help="List of LDAP-URLS [ ldap://<FQHN>:<PORT>/ (where <PORT> has to be different than 389!) ] separated with comma (\",\") for use with OpenLDAP-MMR (Multi-Master-Replication), e.g.: \"ldap://s4dc1:9000,ldap://s4dc2:9000\""),
Option("--ldap-dryrun-mode", help="Configure LDAP backend, but do not run any binaries and exit early. Used only for the test environment. DO NOT USE",
action="store_true"),
Option("--slapd-path", type="string", metavar="SLAPD-PATH",
dsa_options = int(str(msg[0]['options']))
res = samdb.search(expression="(fSMORoleOwner=%s)" % str(ntds_dn),
- controls=["search_options:1:2"])
+ controls=["search_options:1:2"])
if len(res) != 0:
raise CommandError("Current DC is still the owner of %d role(s), "
self.errf.write("Asking partner server %s to synchronize from us\n"
% server)
for part in (samdb.get_schema_basedn(),
- samdb.get_config_basedn(),
- samdb.get_root_basedn()):
+ samdb.get_config_basedn(),
+ samdb.get_root_basedn()):
nc = drsuapi.DsReplicaObjectIdentifier()
nc.dn = str(part)
self.errf.write("Changing userControl and container\n")
res = remote_samdb.search(base=str(remote_samdb.domain_dn()),
expression="(&(objectClass=user)(sAMAccountName=%s$))" %
- netbios_name.upper(),
+ netbios_name.upper(),
attrs=["userAccountControl"])
dc_dn = res[0].dn
uac = int(str(res[0]["userAccountControl"]))
msg.dn = dc_dn
msg["userAccountControl"] = ldb.MessageElement("%d" % uac,
- ldb.FLAG_MOD_REPLACE,
- "userAccountControl")
+ ldb.FLAG_MOD_REPLACE,
+ "userAccountControl")
try:
remote_samdb.modify(msg)
except Exception as e:
if (len(res) != 0):
res = remote_samdb.search(base=computer_dn, expression="%s-%d" % (rdn, i),
- scope=ldb.SCOPE_ONELEVEL)
+ scope=ldb.SCOPE_ONELEVEL)
while(len(res) != 0 and i < 100):
i = i + 1
res = remote_samdb.search(base=computer_dn, expression="%s-%d" % (rdn, i),
- scope=ldb.SCOPE_ONELEVEL)
+ scope=ldb.SCOPE_ONELEVEL)
if i == 100:
if not (dsa_options & DS_NTDSDSA_OPT_DISABLE_OUTBOUND_REPL) and not samdb.am_rodc():
remote_samdb.modify(msg)
raise CommandError("Unable to find a slot for renaming %s,"
- " all names from %s-1 to %s-%d seemed used" %
- (str(dc_dn), rdn, rdn, i - 9))
+ " all names from %s-1 to %s-%d seemed used" %
+ (str(dc_dn), rdn, rdn, i - 9))
newrdn = "%s-%d" % (rdn, i)
takes_options = [
Option("--dbdir", type="string", metavar="DIR",
- help="Path to samba classic DC database directory"),
+ help="Path to samba classic DC database directory"),
Option("--testparm", type="string", metavar="PATH",
- help="Path to samba classic DC testparm utility from the previous installation. This allows the default paths of the previous installation to be followed"),
+ help="Path to samba classic DC testparm utility from the previous installation. This allows the default paths of the previous installation to be followed"),
Option("--targetdir", type="string", metavar="DIR",
- help="Path prefix where the new Samba 4.0 AD domain should be initialised"),
+ help="Path prefix where the new Samba 4.0 AD domain should be initialised"),
Option("-q", "--quiet", help="Be quiet", action="store_true"),
Option("-v", "--verbose", help="Be verbose", action="store_true"),
Option("--dns-backend", type="choice", metavar="NAMESERVER-BACKEND",
choices=["SAMBA_INTERNAL", "BIND9_FLATFILE", "BIND9_DLZ", "NONE"],
help="The DNS server backend. SAMBA_INTERNAL is the builtin name server (default), "
- "BIND9_FLATFILE uses bind9 text database to store zone information, "
- "BIND9_DLZ uses samba4 AD to store zone information, "
- "NONE skips the DNS setup entirely (this DC will not be a DNS server)",
+ "BIND9_FLATFILE uses bind9 text database to store zone information, "
+ "BIND9_DLZ uses samba4 AD to store zone information, "
+ "NONE skips the DNS setup entirely (this DC will not be a DNS server)",
default="SAMBA_INTERNAL")
]
self.outf.write("Creating local TDO.\n")
current_request = { "location": "local", "name": "CreateTrustedDomainEx2"}
local_tdo_handle = local_lsa.CreateTrustedDomainEx2(local_policy,
- local_trust_info,
- local_auth_info,
- lsa.LSA_TRUSTED_DOMAIN_ALL_ACCESS)
+ local_trust_info,
+ local_auth_info,
+ lsa.LSA_TRUSTED_DOMAIN_ALL_ACCESS)
self.outf.write("Local TDO created\n")
if enc_types:
self.outf.write("Setting supported encryption types on local TDO.\n")
if replace_upn:
update_msg['uPNSuffixes'] = ldb.MessageElement(update_upn_vals,
- ldb.FLAG_MOD_REPLACE,
- 'uPNSuffixes')
+ ldb.FLAG_MOD_REPLACE,
+ 'uPNSuffixes')
if replace_spn:
update_msg['msDS-SPNSuffixes'] = ldb.MessageElement(update_spn_vals,
ldb.FLAG_MOD_REPLACE,
try:
stored_forest_info = local_netlogon.netr_DsRGetForestTrustInformation(local_netlogon_info.dc_unc,
- None, 0)
+ None, 0)
except RuntimeError as error:
raise self.LocalRuntimeError(self, error, "netr_DsRGetForestTrustInformation() failed")
takes_options = [
Option("-H", "--URL", help="LDB URL for database or target server", type=str,
- metavar="URL", dest="H"),
+ metavar="URL", dest="H"),
Option("--current-time",
- help="The current time to evaluate the tombstone lifetime from, expressed as YYYY-MM-DD",
- type=str),
+ help="The current time to evaluate the tombstone lifetime from, expressed as YYYY-MM-DD",
+ type=str),
Option("--tombstone-lifetime", help="Number of days a tombstone should be preserved for", type=int),
]
help="The schema file to upgrade to. Default is (Windows) 2012_R2.",
default="2012_R2"),
Option("--ldf-file", type=str, default=None,
- help="Just apply the schema updates in the adprep/.LDF file(s) specified"),
+ help="Just apply the schema updates in the adprep/.LDF file(s) specified"),
Option("--base-dir", type=str, default=None,
help="Location of ldf files Default is ${SETUPDIR}/adprep.")
]
"ro-repl-secret-sync"],
help=car_help),
Option("--action", type="choice", choices=["allow", "deny"],
- help="""Deny or allow access"""),
+ help="""Deny or allow access"""),
Option("--objectdn", help="DN of the object whose SD to modify",
type="string"),
Option("--trusteedn", help="DN of the entity that gets access",
takes_options = [
Option("-H", "--URL", help="LDB URL for database or target server",
- type=str, metavar="URL", dest="H"),
+ type=str, metavar="URL", dest="H"),
]
def run(self, H=None, credopts=None, sambaopts=None, versionopts=None):
takes_options = [
Option("-H", "--URL", help="LDB URL for database or target server",
- type=str, metavar="URL", dest="H"),
+ type=str, metavar="URL", dest="H"),
]
takes_args = ["value"]
try:
msg = samdb.search(base=base_dn, scope=search_scope,
- expression=search_expr,
- attrs=['nTSecurityDescriptor',
- 'versionNumber',
- 'flags',
- 'name',
- 'displayName',
- 'gPCFileSysPath'],
- controls=['sd_flags:1:%d' % sd_flags])
+ expression=search_expr,
+ attrs=['nTSecurityDescriptor',
+ 'versionNumber',
+ 'flags',
+ 'name',
+ 'displayName',
+ 'gPCFileSysPath'],
+ controls=['sd_flags:1:%d' % sd_flags])
except Exception as e:
if gpo is not None:
mesg = "Cannot get information for GPO %s" % gpo
# Check if valid Container DN and get existing GPlinks
try:
msg = samdb.search(base=container_dn, scope=ldb.SCOPE_BASE,
- expression="(objectClass=*)",
- attrs=['gPLink'])[0]
+ expression="(objectClass=*)",
+ attrs=['gPLink'])[0]
except Exception as e:
raise CommandError("Container '%s' does not exist" % container_dn, e)
try:
msg = self.samdb.search(expression='(&(|(samAccountName=%s)(samAccountName=%s$))(objectClass=User))' %
- (ldb.binary_encode(username),ldb.binary_encode(username)))
+ (ldb.binary_encode(username),ldb.binary_encode(username)))
user_dn = msg[0].dn
except Exception:
raise CommandError("Failed to find account %s" % username)
]
def run(self, container_dn, H=None, sambaopts=None, credopts=None,
- versionopts=None):
+ versionopts=None):
self.lp = sambaopts.get_loadparm()
self.creds = credopts.get_credentials(self.lp, fallback_machine=True)
]
def run(self, container_dn, gpo, H=None, disabled=False, enforced=False,
- sambaopts=None, credopts=None, versionopts=None):
+ sambaopts=None, credopts=None, versionopts=None):
self.lp = sambaopts.get_loadparm()
self.creds = credopts.get_credentials(self.lp, fallback_machine=True)
]
def run(self, container, gpo, H=None, sambaopts=None, credopts=None,
- versionopts=None):
+ versionopts=None):
self.lp = sambaopts.get_loadparm()
self.creds = credopts.get_credentials(self.lp, fallback_machine=True)
]
def run(self, gpo, H=None, sambaopts=None, credopts=None,
- versionopts=None):
+ versionopts=None):
self.lp = sambaopts.get_loadparm()
self.creds = credopts.get_credentials(self.lp, fallback_machine=True)
]
def run(self, container_dn, H=None, sambaopts=None, credopts=None,
- versionopts=None):
+ versionopts=None):
self.lp = sambaopts.get_loadparm()
self.creds = credopts.get_credentials(self.lp, fallback_machine=True)
]
def run(self, container_dn, inherit_state, H=None, sambaopts=None, credopts=None,
- versionopts=None):
+ versionopts=None):
if inherit_state.lower() == 'block':
inheritance = dsdb.GPO_BLOCK_INHERITANCE
]
def run(self, gpo, H=None, sambaopts=None, credopts=None,
- versionopts=None):
+ versionopts=None):
self.lp = sambaopts.get_loadparm()
self.creds = credopts.get_credentials(self.lp, fallback_machine=True)
attrs = group_attrs.split(",")
filter = ("(&(sAMAccountType=%d)(sAMAccountName=%s))" %
- ( ATYPE_SECURITY_GLOBAL_GROUP,
- ldb.binary_encode(groupname)))
+ ( ATYPE_SECURITY_GLOBAL_GROUP,
+ ldb.binary_encode(groupname)))
domaindn = samdb.domain_dn()
takes_options = [
Option("--searchflags", help="Search Flags for the attribute", type=str),
Option("-H", "--URL", help="LDB URL for database or target server",
- type=str, metavar="URL", dest="H"),
+ type=str, metavar="URL", dest="H"),
]
takes_args = ["attribute"]
takes_options = [
Option("-H", "--URL", help="LDB URL for database or target server",
- type=str, metavar="URL", dest="H"),
+ type=str, metavar="URL", dest="H"),
]
takes_args = ["attribute"]
takes_options = [
Option("-H", "--URL", help="LDB URL for database or target server",
- type=str, metavar="URL", dest="H"),
+ type=str, metavar="URL", dest="H"),
]
takes_args = ["attribute"]
schema_dn = samdb.schema_dn()
may_filt = '(&(objectClass=classSchema)' \
- '(|(mayContain={0})(systemMayContain={0})))'.format(attribute)
+ '(|(mayContain={0})(systemMayContain={0})))'.format(attribute)
must_filt = '(&(objectClass=classSchema)' \
- '(|(mustContain={0})(systemMustContain={0})))'.format(attribute)
+ '(|(mustContain={0})(systemMustContain={0})))'.format(attribute)
may_res = samdb.search(base=schema_dn, scope=ldb.SCOPE_SUBTREE,
expression=may_filt, attrs=['cn'])
takes_options = [
Option("-H", "--URL", help="LDB URL for database or target server",
- type=str, metavar="URL", dest="H"),
+ type=str, metavar="URL", dest="H"),
]
takes_args = ["objectclass"]
(subnetname, site_of_subnet, e))
print(("Subnet %s shifted to site %s" %
- (subnetname, site_of_subnet)), file=self.outf)
+ (subnetname, site_of_subnet)), file=self.outf)
class cmd_sites_subnet(SuperCommand):
scope=ldb.SCOPE_SUBTREE)
if len(res) != 0 and not force:
raise CommandError("Service principal %s already"
- " affected to another user" % name)
+ " affected to another user" % name)
(cleaneduser, realm, domain) = _get_user_realm_domain(user)
res = sam.search(
sam.modify(msg)
else:
raise CommandError("Service principal %s already"
- " affected to %s" % (name, user))
+ " affected to %s" % (name, user))
else:
raise CommandError("User %s not found" % user)
result = elem
if result is None:
raise CommandError("Unable to find user %s with"
- " spn %s" % (user, name))
+ " spn %s" % (user, name))
else:
if len(res) != 1:
listUser = ""
except ImportError as e:
gpgme_support = False
decrypt_samba_gpg_help = "Decrypt the SambaGPG password not supported, " + \
- "python-gpgme required"
+ "python-gpgme required"
disabled_virtual_attributes = {
}
takes_options = [
Option("-H", "--URL", help="LDB URL for database or target server", type=str,
- metavar="URL", dest="H"),
+ metavar="URL", dest="H"),
Option("--must-change-at-next-login",
- help="Force password to be changed on next login",
- action="store_true"),
+ help="Force password to be changed on next login",
+ action="store_true"),
Option("--random-password",
- help="Generate random password",
- action="store_true"),
+ help="Generate random password",
+ action="store_true"),
Option("--smartcard-required",
- help="Require a smartcard for interactive logons",
- action="store_true"),
+ help="Require a smartcard for interactive logons",
+ action="store_true"),
Option("--use-username-as-cn",
- help="Force use of username as user's CN",
- action="store_true"),
+ help="Force use of username as user's CN",
+ action="store_true"),
Option("--userou",
- help="DN of alternative location (without domainDN counterpart) to default CN=Users in which new user object will be created. E. g. 'OU=<OU name>'",
- type=str),
+ help="DN of alternative location (without domainDN counterpart) to default CN=Users in which new user object will be created. E. g. 'OU=<OU name>'",
+ type=str),
Option("--surname", help="User's surname", type=str),
Option("--given-name", help="User's given name", type=str),
Option("--initials", help="User's initials", type=str),
Option("--telephone-number", help="User's phone number", type=str),
Option("--physical-delivery-office", help="User's office location", type=str),
Option("--rfc2307-from-nss",
- help="Copy Unix user attributes from NSS (will be overridden by explicit UID/GID/GECOS/shell)",
- action="store_true"),
+ help="Copy Unix user attributes from NSS (will be overridden by explicit UID/GID/GECOS/shell)",
+ action="store_true"),
Option("--nis-domain", help="User's Unix/RFC2307 NIS domain", type=str),
Option("--unix-home", help="User's Unix/RFC2307 home directory",
- type=str),
+ type=str),
Option("--uid", help="User's Unix/RFC2307 username", type=str),
Option("--uid-number", help="User's Unix/RFC2307 numeric UID", type=int),
Option("--gid-number", help="User's Unix/RFC2307 primary GID number", type=int),
credentials=creds, lp=lp)
filter = ("(&(sAMAccountName=%s)(sAMAccountType=805306368))" %
- ldb.binary_encode(username))
+ ldb.binary_encode(username))
try:
res = samdb.search(base=samdb.domain_dn(),
}
def run(self, credopts=None, sambaopts=None, versionopts=None,
- newpassword=None):
+ newpassword=None):
lp = sambaopts.get_loadparm()
creds = credopts.get_credentials(lp)
help="Force password to be changed on next login",
action="store_true"),
Option("--random-password",
- help="Generate random password",
- action="store_true"),
+ help="Generate random password",
+ action="store_true"),
Option("--smartcard-required",
- help="Require a smartcard for interactive logons",
- action="store_true"),
+ help="Require a smartcard for interactive logons",
+ action="store_true"),
Option("--clear-smartcard-required",
- help="Don't require a smartcard for interactive logons",
- action="store_true"),
+ help="Don't require a smartcard for interactive logons",
+ action="store_true"),
]
takes_args = ["username?"]
for h in up.hashes:
if (scheme_match is None and
- h.scheme == SCHEME and
- h.value.startswith(scheme_prefix)):
+ h.scheme == SCHEME and
+ h.value.startswith(scheme_prefix)):
scheme_match = h.value
if h.scheme == SCHEME and h.value.startswith(prefix):
return (h.value, scheme_match)
if (ldm & RIGHT_DS_READ_PROPERTY) and (ldm & RIGHT_DS_LIST_CONTENTS):
filemask = filemask | (SYNCHRONIZE | FILE_LIST_DIRECTORY |
- FILE_READ_ATTRIBUTES | FILE_READ_EA |
- FILE_READ_DATA | FILE_EXECUTE)
+ FILE_READ_ATTRIBUTES | FILE_READ_EA |
+ FILE_READ_DATA | FILE_EXECUTE)
if ldm & RIGHT_DS_WRITE_PROPERTY:
filemask = filemask | (SYNCHRONIZE | FILE_WRITE_DATA |
- FILE_APPEND_DATA | FILE_WRITE_EA |
- FILE_WRITE_ATTRIBUTES | FILE_ADD_FILE |
- FILE_ADD_SUBDIRECTORY)
+ FILE_APPEND_DATA | FILE_WRITE_EA |
+ FILE_WRITE_ATTRIBUTES | FILE_ADD_FILE |
+ FILE_ADD_SUBDIRECTORY)
if ldm & RIGHT_DS_CREATE_CHILD:
filemask = filemask | (FILE_ADD_SUBDIRECTORY | FILE_ADD_FILE)
# netbiosname
# Get the netbiosname first (could be obtained from smb.conf in theory)
res = secretsdb.search(expression="(flatname=%s)" %
- names.domain,base="CN=Primary Domains",
- scope=ldb.SCOPE_SUBTREE, attrs=["sAMAccountName"])
+ names.domain,base="CN=Primary Domains",
+ scope=ldb.SCOPE_SUBTREE, attrs=["sAMAccountName"])
names.netbiosname = str(res[0]["sAMAccountName"]).replace("$","")
names.smbconf = smbconf
names.hostname = str(res4[0]["dNSHostName"]).replace("." + names.dnsdomain, "")
server_res = samdb.search(expression="serverReference=%s" % res4[0].dn,
- attrs=[], base=names.configdn)
+ attrs=[], base=names.configdn)
names.serverdn = str(server_res[0].dn)
# invocation id/objectguid
:return: The biggest USN in the provision"""
res = samdb.search(expression="objectClass=*",base=basedn,
- scope=ldb.SCOPE_SUBTREE,attrs=["uSNChanged"],
- controls=["search_options:1:2",
- "server_sort:1:1:uSNChanged",
- "paged_results:1:1"])
+ scope=ldb.SCOPE_SUBTREE,attrs=["uSNChanged"],
+ controls=["search_options:1:2",
+ "server_sort:1:1:uSNChanged",
+ "paged_results:1:1"])
return res[0]["uSNChanged"]
use_ntvfs=use_ntvfs, skip_invalid_chown=True, passdb=passdb, service=SYSVOL_SERVICE)
res = samdb.search(base="CN=Policies,CN=System,%s"%(domaindn),
- attrs=["cn", "nTSecurityDescriptor"],
- expression="", scope=ldb.SCOPE_ONELEVEL)
+ attrs=["cn", "nTSecurityDescriptor"],
+ expression="", scope=ldb.SCOPE_ONELEVEL)
for policy in res:
acl = ndr_unpack(security.descriptor,
if fsacl_sddl != POLICIES_ACL:
raise ProvisioningError('%s ACL on policy root %s %s does not match expected value %s from provision' % (acl_type(direct_db_access), root_policy_path, fsacl_sddl, fsacl))
res = samdb.search(base="CN=Policies,CN=System,%s"%(domaindn),
- attrs=["cn", "nTSecurityDescriptor"],
- expression="", scope=ldb.SCOPE_ONELEVEL)
+ attrs=["cn", "nTSecurityDescriptor"],
+ expression="", scope=ldb.SCOPE_ONELEVEL)
for policy in res:
acl = ndr_unpack(security.descriptor,
mmr = "#"
cn_samba = read_and_sub_file(
- setup_path("cn=samba.ldif"),
- { "LDAPADMINPASS": self.ldapadminpass,
- "MMR_PASSWORD": mmr_pass,
- "MMR": mmr })
+ setup_path("cn=samba.ldif"),
+ { "LDAPADMINPASS": self.ldapadminpass,
+ "MMR_PASSWORD": mmr_pass,
+ "MMR": mmr })
mapping = "schema-map-openldap-2.3"
backend_schema = "backend-schema.schema"
sys.exit(0)
slapd_cmd = [self.slapd_path, "-Ttest", "-n", "0", "-f",
- self.slapdconf, "-F", self.olcdir]
+ self.slapdconf, "-F", self.olcdir]
retcode = subprocess.call(slapd_cmd, close_fds=True, shell=False)
if retcode != 0:
raise Exception("Unable to convert Samba 3 schema.")
self.schema = Schema(
- self.domainsid,
- schemadn=self.names.schemadn,
- files=[setup_path("schema_samba4.ldif"), self.samba3_ldif],
- additional_prefixmap=["1000:1.3.6.1.4.1.7165.2.1",
- "1001:1.3.6.1.4.1.7165.2.2"])
+ self.domainsid,
+ schemadn=self.names.schemadn,
+ files=[setup_path("schema_samba4.ldif"), self.samba3_ldif],
+ additional_prefixmap=["1000:1.3.6.1.4.1.7165.2.1",
+ "1001:1.3.6.1.4.1.7165.2.2"])
def provision(self):
from samba.provision import ProvisioningError, setup_path
# Set up domain container - DC=<DNSDOMAIN>,CN=MicrosoftDNS,<PREFIX>,<DOMAINDN>
domain_container_dn = ldb.Dn(samdb, "DC=%s,CN=MicrosoftDNS,%s,%s" %
- (dnsdomain, prefix, domaindn))
+ (dnsdomain, prefix, domaindn))
# DC=@ record
add_at_record(samdb, domain_container_dn, "DC=@", hostname, dnsdomain,
# DC=_ldap._tcp.DomainDnsZones
add_srv_record(samdb, domain_container_dn, "DC=_ldap._tcp.DomainDnsZones",
- fqdn_hostname, 389)
+ fqdn_hostname, 389)
# DC=_ldap._tcp.ForestDnsZones
add_srv_record(samdb, domain_container_dn, "DC=_ldap._tcp.ForestDnsZones",
- fqdn_hostname, 389)
+ fqdn_hostname, 389)
# DC=DomainDnsZones
add_host_record(samdb, domain_container_dn, "DC=DomainDnsZones", hostip,
def add_dc_msdcs_records(samdb, forestdn, prefix, site, dnsforest, hostname,
- hostip, hostip6, domainguid, ntdsguid):
+ hostip, hostip6, domainguid, ntdsguid):
fqdn_hostname = "%s.%s" % (hostname, dnsforest)
# Set up forest container - DC=<DNSDOMAIN>,CN=MicrosoftDNS,<PREFIX>,<DOMAINDN>
forest_container_dn = ldb.Dn(samdb, "DC=_msdcs.%s,CN=MicrosoftDNS,%s,%s" %
- (dnsforest, prefix, forestdn))
+ (dnsforest, prefix, forestdn))
# DC=@ record
add_at_record(samdb, forest_container_dn, "DC=@", hostname, dnsforest,
% (dn, samdb.get_config_basedn()))
if dn.add_child("CN=X") == False:
raise DemoteException("Failed constructing DN %s by adding child CN=X"\
- % (dn))
+ % (dn))
dn.set_component(0, "CN", dc_name)
try:
logger.info("Removing Sysvol reference: %s" % dn)
self.transaction_commit()
def add_remove_group_members(self, groupname, members,
- add_members_operation=True):
+ add_members_operation=True):
"""Adds or removes group members
:param groupname: Name of the target group
"""
self.hash_oid_name = {}
res = self.search(expression="objectClass=attributeSchema",
- controls=["search_options:1:2"],
- attrs=["attributeID",
+ controls=["search_options:1:2"],
+ attrs=["attributeID",
"lDAPDisplayName"])
if len(res) > 0:
for e in res:
"""
res = self.search(expression="distinguishedName=%s" % dn,
- scope=ldb.SCOPE_SUBTREE,
- controls=["search_options:1:2"],
- attrs=["replPropertyMetaData"])
+ scope=ldb.SCOPE_SUBTREE,
+ controls=["search_options:1:2"],
+ attrs=["replPropertyMetaData"])
if len(res) == 0:
return None
def set_attribute_replmetadata_version(self, dn, att, value,
addifnotexist=False):
res = self.search(expression="distinguishedName=%s" % dn,
- scope=ldb.SCOPE_SUBTREE,
- controls=["search_options:1:2"],
- attrs=["replPropertyMetaData"])
+ scope=ldb.SCOPE_SUBTREE,
+ controls=["search_options:1:2"],
+ attrs=["replPropertyMetaData"])
if len(res) == 0:
return None
""")
schema_dn_add = self.schema_dn_add \
- + "objectGUID: 24e2ca70-b093-4ae8-84c0-2d7ac652a1b8\n"
+ + "objectGUID: 24e2ca70-b093-4ae8-84c0-2d7ac652a1b8\n"
# These bits of LDIF are supplied when the Schema object is created
self.ldb.add_ldif(schema_dn_add)
tmp_desc = sd
m["nTSecurityDescriptor"] = MessageElement(ndr_pack(tmp_desc),
- FLAG_MOD_REPLACE,
- "nTSecurityDescriptor")
+ FLAG_MOD_REPLACE,
+ "nTSecurityDescriptor")
self.ldb.modify(m, controls)
def read_sd_on_dn(self, object_dn, controls=None):
try:
self.conn.DnssrvUpdateRecord2(client_version,
- 0,
- self.server,
- zone,
- name,
- None,
- del_rec_buf)
+ 0,
+ self.server,
+ zone,
+ name,
+ None,
+ del_rec_buf)
if not assertion:
raise AssertionError("Successfully deleted record '%s' of type '%s', which should have failed." % (record_str, record_type_str))
except RuntimeError as e:
# Create zone
self.conn.DnssrvOperation2(client_version,
- 0,
- self.server,
- None,
- 0,
- 'ZoneCreate',
- dnsserver.DNSSRV_TYPEID_ZONE_CREATE,
- zone_create)
+ 0,
+ self.server,
+ None,
+ 0,
+ 'ZoneCreate',
+ dnsserver.DNSSRV_TYPEID_ZONE_CREATE,
+ zone_create)
request_filter = (dnsserver.DNS_ZONE_REQUEST_REVERSE |
- dnsserver.DNS_ZONE_REQUEST_PRIMARY)
+ dnsserver.DNS_ZONE_REQUEST_PRIMARY)
_, zones = self.conn.DnssrvComplexOperation2(client_version,
0,
self.server,
# Delete zone
self.conn.DnssrvOperation2(client_version,
- 0,
- self.server,
- rev_zone,
- 0,
- 'DeleteZoneFromDs',
- dnsserver.DNSSRV_TYPEID_NULL,
- None)
+ 0,
+ self.server,
+ rev_zone,
+ 0,
+ 'DeleteZoneFromDs',
+ dnsserver.DNSSRV_TYPEID_NULL,
+ None)
typeid, zones = self.conn.DnssrvComplexOperation2(client_version,
- 0,
- self.server,
- None,
- 'EnumZones',
- dnsserver.DNSSRV_TYPEID_DWORD,
- request_filter)
+ 0,
+ self.server,
+ None,
+ 'EnumZones',
+ dnsserver.DNSSRV_TYPEID_DWORD,
+ request_filter)
self.assertEquals(0, zones.dwZoneCount)
def test_complexoperation2(self):
client_version = dnsserver.DNS_CLIENT_VERSION_LONGHORN
request_filter = (dnsserver.DNS_ZONE_REQUEST_FORWARD |
- dnsserver.DNS_ZONE_REQUEST_PRIMARY)
+ dnsserver.DNS_ZONE_REQUEST_PRIMARY)
typeid, zones = self.conn.DnssrvComplexOperation2(client_version,
- 0,
- self.server,
- None,
- 'EnumZones',
- dnsserver.DNSSRV_TYPEID_DWORD,
- request_filter)
+ 0,
+ self.server,
+ None,
+ 'EnumZones',
+ dnsserver.DNSSRV_TYPEID_DWORD,
+ request_filter)
self.assertEquals(dnsserver.DNSSRV_TYPEID_ZONE_LIST, typeid)
self.assertEquals(3, zones.dwZoneCount)
request_filter = (dnsserver.DNS_ZONE_REQUEST_REVERSE |
- dnsserver.DNS_ZONE_REQUEST_PRIMARY)
+ dnsserver.DNS_ZONE_REQUEST_PRIMARY)
typeid, zones = self.conn.DnssrvComplexOperation2(client_version,
- 0,
- self.server,
- None,
- 'EnumZones',
- dnsserver.DNSSRV_TYPEID_DWORD,
- request_filter)
+ 0,
+ self.server,
+ None,
+ 'EnumZones',
+ dnsserver.DNSSRV_TYPEID_DWORD,
+ request_filter)
self.assertEquals(dnsserver.DNSSRV_TYPEID_ZONE_LIST, typeid)
self.assertEquals(0, zones.dwZoneCount)
add_rec_buf = dnsserver.DNS_RPC_RECORD_BUF()
add_rec_buf.rec = rec
self.conn.DnssrvUpdateRecord2(client_version,
- 0,
- self.server,
- self.zone,
- name,
- add_rec_buf,
- None)
+ 0,
+ self.server,
+ self.zone,
+ name,
+ add_rec_buf,
+ None)
_, result = self.conn.DnssrvEnumRecords2(client_version,
0,
del_rec_buf = dnsserver.DNS_RPC_RECORD_BUF()
del_rec_buf.rec = rec
self.conn.DnssrvUpdateRecord2(client_version,
- 0,
- self.server,
- self.zone,
- name,
- add_rec_buf,
- del_rec_buf)
+ 0,
+ self.server,
+ self.zone,
+ name,
+ add_rec_buf,
+ del_rec_buf)
buflen, result = self.conn.DnssrvEnumRecords2(client_version,
- 0,
- self.server,
- self.zone,
- name,
- None,
- record_type,
- select_flags,
- None,
- None)
+ 0,
+ self.server,
+ self.zone,
+ name,
+ None,
+ record_type,
+ select_flags,
+ None,
+ None)
self.assertEquals(1, result.count)
self.assertEquals(1, result.rec[0].wRecordCount)
self.assertEquals(dnsp.DNS_TYPE_A, result.rec[0].records[0].wType)
del_rec_buf = dnsserver.DNS_RPC_RECORD_BUF()
del_rec_buf.rec = rec2
self.conn.DnssrvUpdateRecord2(client_version,
- 0,
- self.server,
- self.zone,
- name,
- None,
- del_rec_buf)
+ 0,
+ self.server,
+ self.zone,
+ name,
+ None,
+ del_rec_buf)
self.assertRaises(RuntimeError, self.conn.DnssrvEnumRecords2,
- client_version,
- 0,
- self.server,
- self.zone,
- name,
- None,
- record_type,
- select_flags,
- None,
- None)
+ client_version,
+ 0,
+ self.server,
+ self.zone,
+ name,
+ None,
+ record_type,
+ select_flags,
+ None,
+ None)
# The following tests do not pass against Samba because the owner and
# group are not consistent with Windows, as well as some ACEs.
# netr_ServerReqChallenge with given flags
req = self.generate_request(call_id = 2,
pfc_flags=dcerpc.DCERPC_PFC_FLAG_FIRST |
- dcerpc.DCERPC_PFC_FLAG_PENDING_CANCEL,
+ dcerpc.DCERPC_PFC_FLAG_PENDING_CANCEL,
context_id=ctx.context_id,
opnum=4,
stub=real_stub)
conn2.do_single_request(call_id = 1, ctx=ctx1, io=inq_if_ids,
fault_pfc_flags = samba.dcerpc.dcerpc.DCERPC_PFC_FLAG_FIRST |
- samba.dcerpc.dcerpc.DCERPC_PFC_FLAG_LAST |
- samba.dcerpc.dcerpc.DCERPC_PFC_FLAG_DID_NOT_EXECUTE,
+ samba.dcerpc.dcerpc.DCERPC_PFC_FLAG_LAST |
+ samba.dcerpc.dcerpc.DCERPC_PFC_FLAG_DID_NOT_EXECUTE,
fault_status=dcerpc.DCERPC_NCA_S_UNKNOWN_IF,
fault_context_id=0)
if hexdump:
sys.stderr.write("stub_out: %d\n%s" % (len(stub_out), self.hexdump(stub_out)))
ndr_unpack_out(io, stub_out, bigendian=bigendian, ndr64=ndr64,
- allow_remaining=allow_remaining)
+ allow_remaining=allow_remaining)
if ndr_print:
sys.stderr.write("out: %s" % samba.ndr.ndr_print_out(io))
self.assertEqual(rep_twr.tower.num_floors, 5)
self.assertEqual(len(rep_twr.tower.floors), 5)
self.assertEqual(rep_twr.tower.floors[3].lhs.protocol,
- samba.dcerpc.epmapper.EPM_PROTOCOL_TCP)
+ samba.dcerpc.epmapper.EPM_PROTOCOL_TCP)
self.assertEqual(rep_twr.tower.floors[3].lhs.protocol,
- samba.dcerpc.epmapper.EPM_PROTOCOL_TCP)
+ samba.dcerpc.epmapper.EPM_PROTOCOL_TCP)
# reconnect to the given port
self._disconnect("epmap_reconnect")
rpc_vers=5,
rpc_vers_minor=0,
pfc_flags = samba.dcerpc.dcerpc.DCERPC_PFC_FLAG_FIRST |
- samba.dcerpc.dcerpc.DCERPC_PFC_FLAG_LAST,
+ samba.dcerpc.dcerpc.DCERPC_PFC_FLAG_LAST,
drep = [samba.dcerpc.dcerpc.DCERPC_DREP_LE, 0, 0, 0],
ndr_print=None, hexdump=None):
rpc_vers=5,
rpc_vers_minor=0,
pfc_flags = samba.dcerpc.dcerpc.DCERPC_PFC_FLAG_FIRST |
- samba.dcerpc.dcerpc.DCERPC_PFC_FLAG_LAST,
+ samba.dcerpc.dcerpc.DCERPC_PFC_FLAG_LAST,
drep = [samba.dcerpc.dcerpc.DCERPC_DREP_LE, 0, 0, 0],
auth_length=None):
def generate_bind(self, call_id,
pfc_flags = samba.dcerpc.dcerpc.DCERPC_PFC_FLAG_FIRST |
- samba.dcerpc.dcerpc.DCERPC_PFC_FLAG_LAST,
+ samba.dcerpc.dcerpc.DCERPC_PFC_FLAG_LAST,
max_xmit_frag=5840,
max_recv_frag=5840,
assoc_group_id=0,
def generate_alter(self, call_id,
pfc_flags = samba.dcerpc.dcerpc.DCERPC_PFC_FLAG_FIRST |
- samba.dcerpc.dcerpc.DCERPC_PFC_FLAG_LAST,
+ samba.dcerpc.dcerpc.DCERPC_PFC_FLAG_LAST,
max_xmit_frag=5840,
max_recv_frag=5840,
assoc_group_id=0,
def generate_auth3(self, call_id,
pfc_flags = samba.dcerpc.dcerpc.DCERPC_PFC_FLAG_FIRST |
- samba.dcerpc.dcerpc.DCERPC_PFC_FLAG_LAST,
+ samba.dcerpc.dcerpc.DCERPC_PFC_FLAG_LAST,
auth_info="",
ndr_print=None, hexdump=None):
def generate_request(self, call_id,
pfc_flags = samba.dcerpc.dcerpc.DCERPC_PFC_FLAG_FIRST |
- samba.dcerpc.dcerpc.DCERPC_PFC_FLAG_LAST,
+ samba.dcerpc.dcerpc.DCERPC_PFC_FLAG_LAST,
alloc_hint=None,
context_id=None,
opnum=None,
def generate_co_cancel(self, call_id,
pfc_flags = samba.dcerpc.dcerpc.DCERPC_PFC_FLAG_FIRST |
- samba.dcerpc.dcerpc.DCERPC_PFC_FLAG_LAST,
+ samba.dcerpc.dcerpc.DCERPC_PFC_FLAG_LAST,
auth_info="",
ndr_print=None, hexdump=None):
def generate_orphaned(self, call_id,
pfc_flags = samba.dcerpc.dcerpc.DCERPC_PFC_FLAG_FIRST |
- samba.dcerpc.dcerpc.DCERPC_PFC_FLAG_LAST,
+ samba.dcerpc.dcerpc.DCERPC_PFC_FLAG_LAST,
auth_info="",
ndr_print=None, hexdump=None):
def generate_shutdown(self, call_id,
pfc_flags = samba.dcerpc.dcerpc.DCERPC_PFC_FLAG_FIRST |
- samba.dcerpc.dcerpc.DCERPC_PFC_FLAG_LAST,
+ samba.dcerpc.dcerpc.DCERPC_PFC_FLAG_LAST,
ndr_print=None, hexdump=None):
s = samba.dcerpc.dcerpc.shutdown()
result += c
principal_parts = self.principal.split('@')
assert principal_parts[0] in result and \
- principal_parts[1] in result, \
- 'Principal not found in generated keytab'
+ principal_parts[1] in result, \
+ 'Principal not found in generated keytab'
"Helper function to check opcode"
p_opcode = packet.operation & 0x7800
self.assertEquals(p_opcode, opcode, "Expected OPCODE %s, got %s" %
- (opcode, p_opcode))
+ (opcode, p_opcode))
def make_name_packet(self, opcode, qid=None):
"Helper creating a dns.name_packet"
def test_values(self):
self.assertEquals({b'DisplayName': (1, b'E\x00v\x00e\x00n\x00t\x00 \x00L\x00o\x00g\x00\x00\x00'),
b'ErrorControl': (4, b'\x01\x00\x00\x00')},
- self.registry.values(b"HKLM/SYSTEM/CURRENTCONTROLSET/SERVICES/EVENTLOG"))
+ self.registry.values(b"HKLM/SYSTEM/CURRENTCONTROLSET/SERVICES/EVENTLOG"))
search_filter = "(objectClass=organizationalUnit)"
oulist = self.samdb.search(base=self.samdb.domain_dn(),
- scope=ldb.SCOPE_SUBTREE,
- expression=search_filter,
- attrs=["name"])
+ scope=ldb.SCOPE_SUBTREE,
+ expression=search_filter,
+ attrs=["name"])
self.assertTrue(len(oulist) > 0, "no ous found in samdb")
nidx = nidx + 1
(kidx, kp) = find_package(sc.sub.packages, "Primary:Kerberos",
- start_idx=nidx)
+ start_idx=nidx)
self.assertIsNotNone(pp, "Primary:Kerberos required")
self.assertEqual(kidx, nidx, "Primary:Kerberos at wrong position")
nidx = nidx + 1
nidx = nidx + 1
(cidx, cp) = find_package(sc.sub.packages, "Primary:CLEARTEXT",
- start_idx=nidx)
+ start_idx=nidx)
if cidx is not None:
self.assertEqual(cidx, nidx, "Primary:CLEARTEXT at wrong position")
nidx = nidx + 1
# now run the expiration based on a filter
fourdays = time.time() + (4 * 24 * 60 * 60)
(result, out, err) = self.runsubcmd("user", "setexpiry",
- "--filter", "(&(objectClass=user)(company=comp2))",
- "--days=4",
- "-H", "ldap://%s" % os.environ["DC_SERVER"],
- "-U%s%%%s" % (os.environ["DC_USERNAME"], os.environ["DC_PASSWORD"]))
+ "--filter", "(&(objectClass=user)(company=comp2))",
+ "--days=4",
+ "-H", "ldap://%s" % os.environ["DC_SERVER"],
+ "-U%s%%%s" % (os.environ["DC_USERNAME"], os.environ["DC_PASSWORD"]))
self.assertCmdSuccess(result, out, err, "Can we run setexpiry with a filter")
for user in self.users:
})
# check if --rfc2307-from-nss sets the same values as we got from pwd.getpwuid()
(result, out, err) = self.runsubcmd("user", "create", user["name"], user["password"],
- "--surname=%s" % user["surname"],
- "--given-name=%s" % user["given-name"],
- "--job-title=%s" % user["job-title"],
- "--department=%s" % user["department"],
- "--description=%s" % user["description"],
- "--company=%s" % user["company"],
- "--gecos=%s" % user["gecos"],
- "--rfc2307-from-nss",
- "-H", "ldap://%s" % os.environ["DC_SERVER"],
- "-U%s%%%s" % (os.environ["DC_USERNAME"], os.environ["DC_PASSWORD"]))
+ "--surname=%s" % user["surname"],
+ "--given-name=%s" % user["given-name"],
+ "--job-title=%s" % user["job-title"],
+ "--department=%s" % user["department"],
+ "--description=%s" % user["description"],
+ "--company=%s" % user["company"],
+ "--gecos=%s" % user["gecos"],
+ "--rfc2307-from-nss",
+ "-H", "ldap://%s" % os.environ["DC_SERVER"],
+ "-U%s%%%s" % (os.environ["DC_USERNAME"], os.environ["DC_PASSWORD"]))
self.assertCmdSuccess(result, out, err)
self.assertEquals(err,"","Shouldn't be any error messages")
# create a user with posix attributes from nss but override all of them with the
# random ones just obtained
(result, out, err) = self.runsubcmd("user", "create", user["name"], user["password"],
- "--surname=%s" % user["surname"],
- "--given-name=%s" % user["given-name"],
- "--job-title=%s" % user["job-title"],
- "--department=%s" % user["department"],
- "--description=%s" % user["description"],
- "--company=%s" % user["company"],
- "--rfc2307-from-nss",
- "--gecos=%s" % user["gecos"],
- "--login-shell=%s" % user["loginShell"],
- "--uid=%s" % user["uid"],
- "--uid-number=%s" % user["uidNumber"],
- "--gid-number=%s" % user["gidNumber"],
- "-H", "ldap://%s" % os.environ["DC_SERVER"],
- "-U%s%%%s" % (os.environ["DC_USERNAME"], os.environ["DC_PASSWORD"]))
+ "--surname=%s" % user["surname"],
+ "--given-name=%s" % user["given-name"],
+ "--job-title=%s" % user["job-title"],
+ "--department=%s" % user["department"],
+ "--description=%s" % user["description"],
+ "--company=%s" % user["company"],
+ "--rfc2307-from-nss",
+ "--gecos=%s" % user["gecos"],
+ "--login-shell=%s" % user["loginShell"],
+ "--uid=%s" % user["uid"],
+ "--uid-number=%s" % user["uidNumber"],
+ "--gid-number=%s" % user["gidNumber"],
+ "-H", "ldap://%s" % os.environ["DC_SERVER"],
+ "-U%s%%%s" % (os.environ["DC_USERNAME"], os.environ["DC_PASSWORD"]))
self.assertCmdSuccess(result, out, err)
self.assertEquals(err,"","Shouldn't be any error messages")
def _create_user(self, user):
return self.runsubcmd("user", "create", user["name"], user["password"],
- "--surname=%s" % user["surname"],
- "--given-name=%s" % user["given-name"],
- "--job-title=%s" % user["job-title"],
- "--department=%s" % user["department"],
- "--description=%s" % user["description"],
- "--company=%s" % user["company"],
- "-H", "ldap://%s" % os.environ["DC_SERVER"],
- "-U%s%%%s" % (os.environ["DC_USERNAME"], os.environ["DC_PASSWORD"]))
+ "--surname=%s" % user["surname"],
+ "--given-name=%s" % user["given-name"],
+ "--job-title=%s" % user["job-title"],
+ "--department=%s" % user["department"],
+ "--description=%s" % user["description"],
+ "--company=%s" % user["company"],
+ "-H", "ldap://%s" % os.environ["DC_SERVER"],
+ "-U%s%%%s" % (os.environ["DC_USERNAME"], os.environ["DC_PASSWORD"]))
def _create_posix_user(self, user):
""" create a new user with RFC2307 attributes """
return self.runsubcmd("user", "create", user["name"], user["password"],
- "--surname=%s" % user["surname"],
- "--given-name=%s" % user["given-name"],
- "--job-title=%s" % user["job-title"],
- "--department=%s" % user["department"],
- "--description=%s" % user["description"],
- "--company=%s" % user["company"],
- "--gecos=%s" % user["gecos"],
- "--login-shell=%s" % user["loginShell"],
- "--uid=%s" % user["uid"],
- "--uid-number=%s" % user["uidNumber"],
- "--gid-number=%s" % user["gidNumber"],
- "-H", "ldap://%s" % os.environ["DC_SERVER"],
- "-U%s%%%s" % (os.environ["DC_USERNAME"], os.environ["DC_PASSWORD"]))
+ "--surname=%s" % user["surname"],
+ "--given-name=%s" % user["given-name"],
+ "--job-title=%s" % user["job-title"],
+ "--department=%s" % user["department"],
+ "--description=%s" % user["description"],
+ "--company=%s" % user["company"],
+ "--gecos=%s" % user["gecos"],
+ "--login-shell=%s" % user["loginShell"],
+ "--uid=%s" % user["uid"],
+ "--uid-number=%s" % user["uidNumber"],
+ "--gid-number=%s" % user["gidNumber"],
+ "-H", "ldap://%s" % os.environ["DC_SERVER"],
+ "-U%s%%%s" % (os.environ["DC_USERNAME"], os.environ["DC_PASSWORD"]))
def _find_user(self, name):
search_filter = "(&(sAMAccountName=%s)(objectCategory=%s,%s))" % (ldb.binary_encode(name), "CN=Person,CN=Schema,CN=Configuration", self.samdb.domain_dn())
msg = ldb.Message()
msg.dn = res[0].dn
msg["unicodePwd"] = ldb.MessageElement(b"ABCDEF1234567890",
- ldb.FLAG_MOD_REPLACE,
- "unicodePwd")
+ ldb.FLAG_MOD_REPLACE,
+ "unicodePwd")
self.ldb.modify(
msg,
controls=["local_oid:%s:0" %
self.assertEquals(dn_sort("dc=tata,dc=toto", "dc=toto"), 1)
self.assertEquals(dn_sort("dc=zata", "dc=tata"), 1)
self.assertEquals(dn_sort("dc=toto,dc=tata",
- "cn=foo,dc=toto,dc=tata"), -1)
+ "cn=foo,dc=toto,dc=tata"), -1)
self.assertEquals(dn_sort("cn=bar, dc=toto,dc=tata",
- "cn=foo, dc=toto,dc=tata"), -1)
+ "cn=foo, dc=toto,dc=tata"), -1)
def test_get_diff_sds(self):
domsid = security.dom_sid('S-1-5-21')
rootdn = "dc=samba,dc=example,dc=com"
ldbs = get_ldbs(paths, creds, system_session(), lp)
names = find_provision_key_parameters(ldbs.sam, ldbs.secrets, ldbs.idmap,
- paths, smb_conf_path, lp)
+ paths, smb_conf_path, lp)
self.assertEquals(names.realm, "SAMBA.EXAMPLE.COM")
self.assertEquals(str(names.rootdn).lower(), rootdn.lower())
self.assertNotEquals(names.policyid_dc, None)
def test_delta_update_basesamdb(self):
dummysampath = self._getEmptyDbName()
delta_update_basesamdb(self.paths.samdb, dummysampath,
- self.creds, system_session(), self.lp,
- dummymessage)
+ self.creds, system_session(), self.lp,
+ dummymessage)
def test_update_gpo_simple(self):
dir = getpolicypath(self.paths.sysvol, self.names.dnsdomain,
if found:
logger.warn('Group already exists sid=%s, groupname=%s existing_groupname=%s, Ignoring.',
- str(groupmap.sid), groupmap.nt_name, msg[0]['sAMAccountName'][0])
+ str(groupmap.sid), groupmap.nt_name, msg[0]['sAMAccountName'][0])
else:
if groupmap.sid_name_use == lsa.SID_NAME_WKN_GRP:
# In a lot of Samba3 databases, aliases are marked as well known groups
"""
# policy guid
res = samdb.search(expression="(displayName=Default Domain Policy)",
- base="CN=Policies,CN=System," + str(names.rootdn),
- scope=SCOPE_ONELEVEL, attrs=["cn","displayName"])
+ base="CN=Policies,CN=System," + str(names.rootdn),
+ scope=SCOPE_ONELEVEL, attrs=["cn","displayName"])
names.policyid = str(res[0]["cn"]).replace("{","").replace("}","")
# dc policy guid
res2 = samdb.search(expression="(displayName=Default Domain Controllers"
" Policy)",
- base="CN=Policies,CN=System," + str(names.rootdn),
- scope=SCOPE_ONELEVEL, attrs=["cn","displayName"])
+ base="CN=Policies,CN=System," + str(names.rootdn),
+ scope=SCOPE_ONELEVEL, attrs=["cn","displayName"])
if len(res2) == 1:
names.policyid_dc = str(res2[0]["cn"]).replace("{","").replace("}","")
else:
secrets_ldb.modify(delta)
reference = newsecrets_ldb.search(expression="objectClass=top", base="",
- scope=SCOPE_SUBTREE, attrs=["dn"])
+ scope=SCOPE_SUBTREE, attrs=["dn"])
current = secrets_ldb.search(expression="objectClass=top", base="",
- scope=SCOPE_SUBTREE, attrs=["dn"])
+ scope=SCOPE_SUBTREE, attrs=["dn"])
hash_new = {}
hash = {}
listMissing = []
for entry in listMissing:
reference = newsecrets_ldb.search(expression="distinguishedName=%s" % entry,
- base="", scope=SCOPE_SUBTREE)
+ base="", scope=SCOPE_SUBTREE)
current = secrets_ldb.search(expression="distinguishedName=%s" % entry,
- base="", scope=SCOPE_SUBTREE)
+ base="", scope=SCOPE_SUBTREE)
delta = secrets_ldb.msg_diff(empty, reference[0])
for att in hashAttrNotCopied:
delta.remove(att)
for entry in listPresent:
reference = newsecrets_ldb.search(expression="distinguishedName=%s" % entry,
- base="", scope=SCOPE_SUBTREE)
+ base="", scope=SCOPE_SUBTREE)
current = secrets_ldb.search(expression="distinguishedName=%s" % entry, base="",
- scope=SCOPE_SUBTREE)
+ scope=SCOPE_SUBTREE)
delta = secrets_ldb.msg_diff(current[0], reference[0])
for att in hashAttrNotCopied:
delta.remove(att)
for entry in listPresent:
reference = newsecrets_ldb.search(expression="distinguishedName=%s" % entry, base="",
- scope=SCOPE_SUBTREE)
+ scope=SCOPE_SUBTREE)
current = secrets_ldb.search(expression="distinguishedName=%s" % entry, base="",
- scope=SCOPE_SUBTREE)
+ scope=SCOPE_SUBTREE)
delta = secrets_ldb.msg_diff(current[0], reference[0])
for att in hashAttrNotCopied:
delta.remove(att)
secrets_ldb.modify(delta)
res2 = secrets_ldb.search(expression="(samaccountname=dns)",
- scope=SCOPE_SUBTREE, attrs=["dn"])
+ scope=SCOPE_SUBTREE, attrs=["dn"])
if len(res2) == 1:
messagefunc(SIMPLE, "Remove old dns account")
:return: The content of the field oEMInformation (if any)
"""
res = samdb.search(expression="(objectClass=*)", base=str(rootdn),
- scope=SCOPE_BASE, attrs=["dn", "oEMInformation"])
+ scope=SCOPE_BASE, attrs=["dn", "oEMInformation"])
if len(res) > 0 and res[0].get("oEMInformation"):
info = res[0]["oEMInformation"]
return info
the provision (ie. DC=...,DC=...)
"""
res = samdb.search(expression="(objectClass=*)", base=rootdn,
- scope=SCOPE_BASE, attrs=["dn", "oEMInformation"])
+ scope=SCOPE_BASE, attrs=["dn", "oEMInformation"])
if len(res) > 0:
if res[0].get("oEMInformation"):
info = str(res[0]["oEMInformation"])
delta = ldb.Message()
delta.dn = ldb.Dn(samdb, str(res[0]["dn"]))
delta["oEMInformation"] = ldb.MessageElement(info, ldb.FLAG_MOD_REPLACE,
- "oEMInformation" )
+ "oEMInformation" )
samdb.modify(delta)
def update_gpo(paths, samdb, names, lp, message):
if int(str(val)) < version:
done = done + 1
samdb.set_attribute_replmetadata_version(str(e.dn),
- "unicodePwd",
- version, True)
+ "unicodePwd",
+ version, True)
def delta_update_basesamdb(refsampath, sampath, creds, session, lp, message):
"""Update the provision container db: sam.ldb
This function is aimed for alpha9 and newer;
message(SIMPLE,
"Update base samdb by searching difference with reference one")
refsam = Ldb(refsampath, session_info=session, credentials=creds,
- lp=lp, options=["modules:"])
+ lp=lp, options=["modules:"])
sam = Ldb(sampath, session_info=session, credentials=creds, lp=lp,
- options=["modules:"])
+ options=["modules:"])
empty = ldb.Message()
deltaattr = None
for refentry in reference:
entry = sam.search(expression="distinguishedName=%s" % refentry["dn"],
- scope=SCOPE_SUBTREE)
+ scope=SCOPE_SUBTREE)
if not len(entry):
delta = sam.msg_diff(empty, refentry)
message(CHANGE, "Adding %s to sam db" % str(refentry.dn))
expression = "samAccountName=%s$" % names.netbiosname
secrets_msg = secrets_ldb.search(expression=expression,
- attrs=["secureChannelType"])
+ attrs=["secureChannelType"])
if int(secrets_msg[0]["secureChannelType"][0]) == SEC_CHAN_BDC:
res = samdb.search(expression=expression, attrs=[])
assert(len(res) == 1)
msg = ldb.Message(secrets_msg[0].dn)
msg["secret"] = ldb.MessageElement(machinepass,
- ldb.FLAG_MOD_REPLACE,
- "secret")
+ ldb.FLAG_MOD_REPLACE,
+ "secret")
msg["msDS-KeyVersionNumber"] = ldb.MessageElement(kvno,
ldb.FLAG_MOD_REPLACE,
"msDS-KeyVersionNumber")
hash_id = {}
res = samdb.search(base=basedn, expression="objectClass=*",
- scope=ldb.SCOPE_SUBTREE,
- attrs=["replPropertyMetaData"],
- controls=["search_options:1:2"])
+ scope=ldb.SCOPE_SUBTREE,
+ attrs=["replPropertyMetaData"],
+ controls=["search_options:1:2"])
for e in res:
nb_obj = nb_obj + 1
obj = ndr_unpack(drsblobs.replPropertyMetaDataBlob,
- str(e["replPropertyMetaData"])).ctr
+ str(e["replPropertyMetaData"])).ctr
for o in obj.array:
# like a timestamp but with the resolution of 1 minute
binding_string = "ncacn_np:$SERVER[%s%s%s]" % (a, s, e)
plantestsuite("samba3.blackbox.rpcclient krb5 ncacn_np with [%s%s%s] " % (a, s, e), "ktest:local", [os.path.join(samba3srcdir, "script/tests/test_rpcclient.sh"),
- "$PREFIX/ktest/krb5_ccache-3", binding_string, "-k", configuration])
+ "$PREFIX/ktest/krb5_ccache-3", binding_string, "-k", configuration])
plantestsuite("samba3.blackbox.rpcclient_samlogon", "ad_member:local", [os.path.join(samba3srcdir, "script/tests/test_rpcclient_samlogon.sh"),
"$DC_USERNAME", "$DC_PASSWORD", "ncacn_np:$DC_SERVER", configuration])
def tearDown(self):
super(AclAddTests, self).tearDown()
delete_force(self.ldb_admin, "CN=%s,%s,%s" %
- (self.test_user1, self.ou2, self.base_dn))
+ (self.test_user1, self.ou2, self.base_dn))
delete_force(self.ldb_admin, "CN=%s,%s,%s" %
- (self.test_group1, self.ou2, self.base_dn))
+ (self.test_group1, self.ou2, self.base_dn))
delete_force(self.ldb_admin, "%s,%s" % (self.ou2, self.base_dn))
delete_force(self.ldb_admin, "%s,%s" % (self.ou1, self.base_dn))
delete_force(self.ldb_admin, self.get_user_dn(self.usr_admin_owner))
self.ldb_owner.create_ou("OU=test_add_ou2,OU=test_add_ou1," + self.base_dn)
self.ldb_owner.newuser(self.test_user1, self.user_pass, userou=self.ou2)
self.ldb_owner.newgroup("test_add_group1", groupou="OU=test_add_ou2,OU=test_add_ou1",
- grouptype=samba.dsdb.GTYPE_DISTRIBUTION_DOMAIN_LOCAL_GROUP)
+ grouptype=samba.dsdb.GTYPE_DISTRIBUTION_DOMAIN_LOCAL_GROUP)
# Make sure we have successfully created the two objects -- user and group
res = self.ldb_admin.search(self.base_dn, expression="(distinguishedName=%s,%s)" % ("CN=test_add_user1,OU=test_add_ou2,OU=test_add_ou1", self.base_dn))
self.assertTrue(len(res) > 0)
self.fail()
try:
res = anonymous.search(anonymous.get_config_basedn(), expression="(objectClass=*)",
- scope=SCOPE_SUBTREE)
+ scope=SCOPE_SUBTREE)
except LdbError as e17:
(num, _) = e17.args
self.assertEquals(num, ERR_OPERATIONS_ERROR)
#these users should see all ous
res = self.ldb_user.search("OU=ou1," + self.base_dn, expression="(objectClass=*)",
- scope=SCOPE_SUBTREE)
+ scope=SCOPE_SUBTREE)
self.assertEquals(len(res), 6)
res_list = [ x["dn"] for x in res if x["dn"] in self.full_list ]
self.assertEquals(sorted(res_list), sorted(self.full_list))
#these users should see ou1, 2, 5 and 6 but not 3 and 4
res = self.ldb_user.search("OU=ou1," + self.base_dn, expression="(objectClass=*)",
- scope=SCOPE_SUBTREE)
+ scope=SCOPE_SUBTREE)
ok_list = [Dn(self.ldb_admin, "OU=ou2,OU=ou1," + self.base_dn),
Dn(self.ldb_admin, "OU=ou1," + self.base_dn),
Dn(self.ldb_admin, "OU=ou5,OU=ou3,OU=ou2,OU=ou1," + self.base_dn),
#should not see ou3 and ou4, but should see ou5 and ou6
res = self.ldb_user.search("OU=ou1," + self.base_dn, expression="(objectClass=*)",
- scope=SCOPE_SUBTREE)
+ scope=SCOPE_SUBTREE)
self.assertEquals(len(res), 4)
res_list = [ x["dn"] for x in res if x["dn"] in ok_list ]
self.assertEquals(sorted(res_list), sorted(ok_list))
self.assertEquals(sorted(res_list), sorted(ok_list))
res = self.ldb_user.search("OU=ou1," + self.base_dn, expression="(objectClass=*)",
- scope=SCOPE_SUBTREE)
+ scope=SCOPE_SUBTREE)
self.assertEquals(len(res), 2)
res_list = [ x["dn"] for x in res if x["dn"] in ok_list ]
self.assertEquals(sorted(res_list), sorted(ok_list))
self.ldb_admin.create_ou("OU=ou2,OU=ou1," + self.base_dn, sd=tmp_desc)
# assert user can only see dn
res = self.ldb_user.search("OU=ou2,OU=ou1," + self.base_dn, expression="(objectClass=*)",
- scope=SCOPE_SUBTREE)
+ scope=SCOPE_SUBTREE)
ok_list = ['dn']
self.assertEquals(len(res), 1)
res_list = list(res[0].keys())
self.assertEquals(res_list, ok_list)
res = self.ldb_user.search("OU=ou2,OU=ou1," + self.base_dn, expression="(objectClass=*)",
- scope=SCOPE_BASE, attrs=["ou"])
+ scope=SCOPE_BASE, attrs=["ou"])
self.assertEquals(len(res), 1)
res_list = list(res[0].keys())
self.sd_utils.dacl_add_ace("OU=ou1," + self.base_dn, mod)
self.sd_utils.dacl_add_ace("OU=ou2,OU=ou1," + self.base_dn, mod)
res = self.ldb_user.search("OU=ou2,OU=ou1," + self.base_dn, expression="(objectClass=*)",
- scope=SCOPE_SUBTREE)
+ scope=SCOPE_SUBTREE)
ok_list = ['dn', 'ou']
self.assertEquals(len(res), 1)
res_list = list(res[0].keys())
self.sd_utils.dacl_add_ace("OU=ou1," + self.base_dn, mod)
self.sd_utils.dacl_add_ace("OU=ou2,OU=ou1," + self.base_dn, mod)
res = self.ldb_user.search("OU=ou2,OU=ou1," + self.base_dn, expression="(objectClass=*)",
- scope=SCOPE_SUBTREE)
+ scope=SCOPE_SUBTREE)
ok_list = ['dn', 'objectClass', 'ou', 'distinguishedName', 'name', 'objectGUID', 'objectCategory']
res_list = list(res[0].keys())
self.ldb_user.create_ou("OU=ou3,OU=ou2,OU=ou1," + self.base_dn, sd=tmp_desc)
res = self.ldb_user.search("OU=ou1," + self.base_dn, expression="(ou=ou3)",
- scope=SCOPE_SUBTREE)
+ scope=SCOPE_SUBTREE)
#nothing should be returned as ou is not accessible
self.assertEquals(len(res), 0)
mod = "(OA;;RP;bf9679f0-0de6-11d0-a285-00aa003049e2;;%s)" % (str(self.user_sid))
self.sd_utils.dacl_add_ace("OU=ou3,OU=ou2,OU=ou1," + self.base_dn, mod)
res = self.ldb_user.search("OU=ou1," + self.base_dn, expression="(ou=ou3)",
- scope=SCOPE_SUBTREE)
+ scope=SCOPE_SUBTREE)
self.assertEquals(len(res), 1)
ok_list = ['dn', 'ou']
res_list = list(res[0].keys())
self.ldb_admin.newuser(self.testuser1, self.user_pass, userou=self.ou1)
try:
self.ldb_user.rename("CN=%s,%s,%s" % (self.testuser1, self.ou1, self.base_dn), \
- "CN=%s,%s,%s" % (self.testuser5, self.ou1, self.base_dn))
+ "CN=%s,%s,%s" % (self.testuser5, self.ou1, self.base_dn))
except LdbError as e21:
(num, _) = e21.args
self.assertEquals(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
msg = Message()
msg.dn = Dn(self.ldb_admin, dn)
msg["servicePrincipalName"] = MessageElement(spn, flag,
- "servicePrincipalName")
+ "servicePrincipalName")
_ldb.modify(msg)
def create_computer(self, computername, domainname):
# This does not pass against Windows, although it should according to docs
self.replace_spn(self.ldb_user1, self.computerdn, "HOST/%s" % (self.computername))
self.replace_spn(self.ldb_user1, self.computerdn, "HOST/%s.%s" %
- (self.computername, self.dcctx.dnsdomain))
+ (self.computername, self.dcctx.dnsdomain))
try:
self.replace_spn(self.ldb_user1, self.computerdn, "HOST/%s/%s" % (self.computername, netbiosdomain))
scope=SCOPE_SUBTREE,
attrs=['cn'])
print('%d %s took %s' % (i, expression,
- time.time() - t), file=sys.stderr)
+ time.time() - t), file=sys.stderr)
def _test_indexed_search(self):
expressions = ['(objectclass=group)',
scope=SCOPE_SUBTREE,
attrs=['cn'])
print('%d runs %s took %s' % (i, expression,
- time.time() - t), file=sys.stderr)
+ time.time() - t), file=sys.stderr)
def _test_add_many_users(self, n=BATCH_SIZE):
s = self.state.next_user_id
= self.ldb.get_wellknown_dn(self.ldb.get_default_basedn(),
dsdb.DS_GUID_DELETED_OBJECTS_CONTAINER)
sites_obj = self.search_dn("cn=sites,%s" \
- % self.ldb.get_config_basedn())
+ % self.ldb.get_config_basedn())
self.sites_dn = sites_obj.dn
self.sites_guid \
= sites_obj["objectGUID"][0]
res2 = self.ldb_admin.search(self.base_dn, scope=SCOPE_BASE, attrs=["objectGUID"])
res = self.ldb_admin.search(self.base_dn,
- expression="name=Configuration",
- controls=["dirsync:1:0:1"])
+ expression="name=Configuration",
+ controls=["dirsync:1:0:1"])
self.assertEqual(res2[0].get("objectGUID"), res[0].get("parentGUID"))
def test_ok_not_rootdc(self):
"""Test if it's ok to do dirsync on another NC that is not the root DC"""
self.ldb_admin.search(self.ldb_admin.get_config_basedn(),
- expression="samaccountname=*",
- controls=["dirsync:1:0:1"])
+ expression="samaccountname=*",
+ controls=["dirsync:1:0:1"])
def test_dirsync_errors(self):
"""Test if dirsync returns the correct LDAP errors in case of pb"""
delta = Message()
delta.dn = Dn(self.ldb_admin, str(ouname))
delta["cn"] = MessageElement("test ou",
- FLAG_MOD_ADD,
- "cn" )
+ FLAG_MOD_ADD,
+ "cn" )
self.ldb_admin.modify(delta)
res = self.ldb_admin.search(self.base_dn,
expression="name=testou",
delta.dn = Dn(self.ldb_admin, str(ouname))
delta["cn"] = MessageElement("test ou",
- FLAG_MOD_ADD,
- "cn" )
+ FLAG_MOD_ADD,
+ "cn" )
self.ldb_admin.modify(delta)
res = self.ldb_admin.search(self.base_dn,
expression="(&(objectClass=organizationalUnit)(!(isDeleted=*)))",
delta = Message()
delta.dn = Dn(self.ldb_admin, str(ouname))
delta["cn"] = MessageElement([],
- FLAG_MOD_DELETE,
- "cn" )
+ FLAG_MOD_DELETE,
+ "cn" )
self.ldb_admin.modify(delta)
res = self.ldb_admin.search(self.base_dn,
expression="(&(objectClass=organizationalUnit)(!(isDeleted=*)))",
# Make sure that the "member" attribute for "ldaptestuser" has been
# removed
res = ldb.search("cn=ldaptestgroup,cn=users," + self.base_dn,
- scope=SCOPE_BASE, attrs=["member"])
+ scope=SCOPE_BASE, attrs=["member"])
self.assertTrue(len(res) == 1)
self.assertFalse("member" in res[0])
# ensure we cannot add it again
try:
ldb.add({"dn": "cn=ldaptestuser3,cn=userS," + self.base_dn,
- "objectClass": "user",
- "cn": "LDAPtestUSER3"})
+ "objectClass": "user",
+ "cn": "LDAPtestUSER3"})
self.fail()
except LdbError as e98:
(num, _) = e98.args
# ensure can now use that name
ldb.add({"dn": "cn=ldaptestuser3,cn=users," + self.base_dn,
- "objectClass": "user",
- "cn": "LDAPtestUSER3"})
+ "objectClass": "user",
+ "cn": "LDAPtestUSER3"})
# ensure we now cannot rename
try:
lastLogon=(lastlogon_relation, 0),
lastLogonTimestamp=('greater', 0),
userAccountControl=
- dsdb.UF_NORMAL_ACCOUNT,
+ dsdb.UF_NORMAL_ACCOUNT,
msDSUserAccountControlComputed=0)
badPasswordTime = int(res[0]["badPasswordTime"][0])
logonCount = int(res[0]["logonCount"][0])
lastLogon=lastLogon,
lastLogonTimestamp=lastLogonTimestamp,
userAccountControl=
- dsdb.UF_NORMAL_ACCOUNT,
+ dsdb.UF_NORMAL_ACCOUNT,
msDSUserAccountControlComputed=0)
badPasswordTime = int(res[0]["badPasswordTime"][0])
lastLogon=lastLogon,
lastLogonTimestamp=lastLogonTimestamp,
userAccountControl=
- dsdb.UF_NORMAL_ACCOUNT,
+ dsdb.UF_NORMAL_ACCOUNT,
msDSUserAccountControlComputed=0)
# Wrong old password
lastLogon=lastLogon,
lastLogonTimestamp=lastLogonTimestamp,
userAccountControl=
- dsdb.UF_NORMAL_ACCOUNT,
+ dsdb.UF_NORMAL_ACCOUNT,
msDSUserAccountControlComputed=0)
badPasswordTime = int(res[0]["badPasswordTime"][0])
lastLogonTimestamp=lastLogonTimestamp,
lockoutTime=("greater", badPasswordTime),
userAccountControl=
- dsdb.UF_NORMAL_ACCOUNT,
+ dsdb.UF_NORMAL_ACCOUNT,
msDSUserAccountControlComputed=dsdb.UF_LOCKOUT)
badPasswordTime = int(res[0]["badPasswordTime"][0])
lockoutTime = int(res[0]["lockoutTime"][0])
lastLogonTimestamp=lastLogonTimestamp,
lockoutTime=lockoutTime,
userAccountControl=
- dsdb.UF_NORMAL_ACCOUNT,
+ dsdb.UF_NORMAL_ACCOUNT,
msDSUserAccountControlComputed=dsdb.UF_LOCKOUT)
# Wrong old password
lastLogon=lastLogon,
lastLogonTimestamp=lastLogonTimestamp,
userAccountControl=
- dsdb.UF_NORMAL_ACCOUNT,
+ dsdb.UF_NORMAL_ACCOUNT,
msDSUserAccountControlComputed=dsdb.UF_LOCKOUT)
try:
lastLogonTimestamp=lastLogonTimestamp,
lockoutTime=lockoutTime,
userAccountControl=
- dsdb.UF_NORMAL_ACCOUNT,
+ dsdb.UF_NORMAL_ACCOUNT,
msDSUserAccountControlComputed=dsdb.UF_LOCKOUT)
# Now reset the password, which does NOT change the lockout!
lastLogonTimestamp=lastLogonTimestamp,
lockoutTime=lockoutTime,
userAccountControl=
- dsdb.UF_NORMAL_ACCOUNT,
+ dsdb.UF_NORMAL_ACCOUNT,
msDSUserAccountControlComputed=dsdb.UF_LOCKOUT)
try:
lastLogonTimestamp=lastLogonTimestamp,
lockoutTime=lockoutTime,
userAccountControl=
- dsdb.UF_NORMAL_ACCOUNT,
+ dsdb.UF_NORMAL_ACCOUNT,
msDSUserAccountControlComputed=dsdb.UF_LOCKOUT)
m = Message()
lastLogonTimestamp=lastLogonTimestamp,
lockoutTime=lockoutTime,
userAccountControl=
- dsdb.UF_NORMAL_ACCOUNT,
+ dsdb.UF_NORMAL_ACCOUNT,
msDSUserAccountControlComputed=dsdb.UF_LOCKOUT)
# This shows that setting the UF_LOCKOUT flag makes no difference
lastLogon=lastLogon,
lastLogonTimestamp=lastLogonTimestamp,
userAccountControl=
- dsdb.UF_NORMAL_ACCOUNT,
+ dsdb.UF_NORMAL_ACCOUNT,
msDSUserAccountControlComputed=dsdb.UF_LOCKOUT)
self._reset_by_method(res, method)
lastLogon=lastLogon,
lastLogonTimestamp=lastLogonTimestamp,
userAccountControl=
- dsdb.UF_NORMAL_ACCOUNT,
+ dsdb.UF_NORMAL_ACCOUNT,
msDSUserAccountControlComputed=0)
# The correct password after doing the unlock
lastLogon=lastLogon,
lastLogonTimestamp=lastLogonTimestamp,
userAccountControl=
- dsdb.UF_NORMAL_ACCOUNT,
+ dsdb.UF_NORMAL_ACCOUNT,
msDSUserAccountControlComputed=0)
# Wrong old password
lastLogon=lastLogon,
lastLogonTimestamp=lastLogonTimestamp,
userAccountControl=
- dsdb.UF_NORMAL_ACCOUNT,
+ dsdb.UF_NORMAL_ACCOUNT,
msDSUserAccountControlComputed=0)
badPasswordTime = int(res[0]["badPasswordTime"][0])
lastLogon=lastLogon,
lastLogonTimestamp=lastLogonTimestamp,
userAccountControl=
- dsdb.UF_NORMAL_ACCOUNT,
+ dsdb.UF_NORMAL_ACCOUNT,
msDSUserAccountControlComputed=0)
badPasswordTime = int(res[0]["badPasswordTime"][0])
lastLogonTimestamp=lastLogonTimestamp,
lockoutTime=0,
userAccountControl=
- dsdb.UF_NORMAL_ACCOUNT,
+ dsdb.UF_NORMAL_ACCOUNT,
msDSUserAccountControlComputed=0)
# The following test lockout behaviour when modifying a user's password
lastLogon=("greater", 0),
lastLogonTimestamp=("greater", 0),
userAccountControl=
- dsdb.UF_NORMAL_ACCOUNT,
+ dsdb.UF_NORMAL_ACCOUNT,
msDSUserAccountControlComputed=0)
badPasswordTime = int(res[0]["badPasswordTime"][0])
logonCount = int(res[0]["logonCount"][0])
lastLogon=lastLogon,
lastLogonTimestamp=lastLogonTimestamp,
userAccountControl=
- dsdb.UF_NORMAL_ACCOUNT,
+ dsdb.UF_NORMAL_ACCOUNT,
msDSUserAccountControlComputed=0)
badPasswordTime = int(res[0]["badPasswordTime"][0])
lastLogon=lastLogon,
lastLogonTimestamp=lastLogonTimestamp,
userAccountControl=
- dsdb.UF_NORMAL_ACCOUNT,
+ dsdb.UF_NORMAL_ACCOUNT,
msDSUserAccountControlComputed=0)
# Wrong old password
lastLogon=lastLogon,
lastLogonTimestamp=lastLogonTimestamp,
userAccountControl=
- dsdb.UF_NORMAL_ACCOUNT,
+ dsdb.UF_NORMAL_ACCOUNT,
msDSUserAccountControlComputed=0)
badPasswordTime = int(res[0]["badPasswordTime"][0])
lastLogon=lastLogon,
lastLogonTimestamp=lastLogonTimestamp,
userAccountControl=
- dsdb.UF_NORMAL_ACCOUNT,
+ dsdb.UF_NORMAL_ACCOUNT,
msDSUserAccountControlComputed=0)
print("two failed password change")
lastLogonTimestamp=lastLogonTimestamp,
lockoutTime=("greater", badPasswordTime),
userAccountControl=
- dsdb.UF_NORMAL_ACCOUNT,
+ dsdb.UF_NORMAL_ACCOUNT,
msDSUserAccountControlComputed=dsdb.UF_LOCKOUT)
badPasswordTime = int(res[0]["badPasswordTime"][0])
lockoutTime = int(res[0]["lockoutTime"][0])
lastLogonTimestamp=lastLogonTimestamp,
lockoutTime=lockoutTime,
userAccountControl=
- dsdb.UF_NORMAL_ACCOUNT,
+ dsdb.UF_NORMAL_ACCOUNT,
msDSUserAccountControlComputed=dsdb.UF_LOCKOUT)
# Wrong old password
lastLogonTimestamp=lastLogonTimestamp,
lockoutTime=lockoutTime,
userAccountControl=
- dsdb.UF_NORMAL_ACCOUNT,
+ dsdb.UF_NORMAL_ACCOUNT,
msDSUserAccountControlComputed=dsdb.UF_LOCKOUT)
try:
lastLogonTimestamp=lastLogonTimestamp,
lockoutTime=lockoutTime,
userAccountControl=
- dsdb.UF_NORMAL_ACCOUNT,
+ dsdb.UF_NORMAL_ACCOUNT,
msDSUserAccountControlComputed=dsdb.UF_LOCKOUT)
# Now reset the lockout, by removing ACB_AUTOLOCK (which removes the lock, despite being a generated attribute)
lastLogonTimestamp=lastLogonTimestamp,
lockoutTime=0,
userAccountControl=
- dsdb.UF_NORMAL_ACCOUNT,
+ dsdb.UF_NORMAL_ACCOUNT,
msDSUserAccountControlComputed=0)
# Correct old password
lastLogonTimestamp=lastLogonTimestamp,
lockoutTime=0,
userAccountControl=
- dsdb.UF_NORMAL_ACCOUNT,
+ dsdb.UF_NORMAL_ACCOUNT,
msDSUserAccountControlComputed=0)
# Wrong old password
lastLogonTimestamp=lastLogonTimestamp,
lockoutTime=0,
userAccountControl=
- dsdb.UF_NORMAL_ACCOUNT,
+ dsdb.UF_NORMAL_ACCOUNT,
msDSUserAccountControlComputed=0)
badPasswordTime = int(res[0]["badPasswordTime"][0])
lastLogonTimestamp=lastLogonTimestamp,
lockoutTime=0,
userAccountControl=
- dsdb.UF_NORMAL_ACCOUNT,
+ dsdb.UF_NORMAL_ACCOUNT,
msDSUserAccountControlComputed=0)
badPasswordTime = int(res[0]["badPasswordTime"][0])
lastLogonTimestamp=lastLogonTimestamp,
lockoutTime=0,
userAccountControl=
- dsdb.UF_NORMAL_ACCOUNT,
+ dsdb.UF_NORMAL_ACCOUNT,
msDSUserAccountControlComputed=0)
# Wrong old password
lastLogonTimestamp=lastLogonTimestamp,
lockoutTime=("greater", badPasswordTime),
userAccountControl=
- dsdb.UF_NORMAL_ACCOUNT,
+ dsdb.UF_NORMAL_ACCOUNT,
msDSUserAccountControlComputed=dsdb.UF_LOCKOUT)
badPasswordTime = int(res[0]["badPasswordTime"][0])
lockoutTime = int(res[0]["lockoutTime"][0])
lastLogonTimestamp=lastLogonTimestamp,
lockoutTime=lockoutTime,
userAccountControl=
- dsdb.UF_NORMAL_ACCOUNT,
+ dsdb.UF_NORMAL_ACCOUNT,
msDSUserAccountControlComputed=0)
# SAMR doesn't have any impact if dsdb.UF_LOCKOUT isn't present.
lastLogon=lastLogon,
lastLogonTimestamp=lastLogonTimestamp,
userAccountControl=
- dsdb.UF_NORMAL_ACCOUNT,
+ dsdb.UF_NORMAL_ACCOUNT,
msDSUserAccountControlComputed=0)
def test_unicodePwd_lockout_with_clear_change_krb5(self):
lastLogon=0,
lastLogonTimestamp=('absent', None),
userAccountControl=
- dsdb.UF_NORMAL_ACCOUNT |
- dsdb.UF_ACCOUNTDISABLE |
- dsdb.UF_PASSWD_NOTREQD,
+ dsdb.UF_NORMAL_ACCOUNT |
+ dsdb.UF_ACCOUNTDISABLE |
+ dsdb.UF_PASSWD_NOTREQD,
msDSUserAccountControlComputed=
- dsdb.UF_PASSWORD_EXPIRED)
+ dsdb.UF_PASSWORD_EXPIRED)
# SAMR doesn't have any impact if dsdb.UF_LOCKOUT isn't present.
# It doesn't create "lockoutTime" = 0.
lastLogon=0,
lastLogonTimestamp=('absent', None),
userAccountControl=
- dsdb.UF_NORMAL_ACCOUNT |
- dsdb.UF_ACCOUNTDISABLE |
- dsdb.UF_PASSWD_NOTREQD,
+ dsdb.UF_NORMAL_ACCOUNT |
+ dsdb.UF_ACCOUNTDISABLE |
+ dsdb.UF_PASSWD_NOTREQD,
msDSUserAccountControlComputed=
- dsdb.UF_PASSWORD_EXPIRED)
+ dsdb.UF_PASSWORD_EXPIRED)
# Tests a password change when we don't have any password yet with a
# wrong old password
lastLogon=0,
lastLogonTimestamp=('absent', None),
userAccountControl=
- dsdb.UF_NORMAL_ACCOUNT |
- dsdb.UF_ACCOUNTDISABLE |
- dsdb.UF_PASSWD_NOTREQD,
+ dsdb.UF_NORMAL_ACCOUNT |
+ dsdb.UF_ACCOUNTDISABLE |
+ dsdb.UF_PASSWD_NOTREQD,
msDSUserAccountControlComputed=
- dsdb.UF_PASSWORD_EXPIRED)
+ dsdb.UF_PASSWORD_EXPIRED)
badPwdCount = int(res[0]["badPwdCount"][0])
badPasswordTime = int(res[0]["badPasswordTime"][0])
lastLogon=0,
lastLogonTimestamp=('absent', None),
userAccountControl=
- dsdb.UF_NORMAL_ACCOUNT |
- dsdb.UF_ACCOUNTDISABLE |
- dsdb.UF_PASSWD_NOTREQD,
+ dsdb.UF_NORMAL_ACCOUNT |
+ dsdb.UF_ACCOUNTDISABLE |
+ dsdb.UF_PASSWD_NOTREQD,
msDSUserAccountControlComputed=0)
# Enables the user account
lastLogon=0,
lastLogonTimestamp=('absent', None),
userAccountControl=
- dsdb.UF_NORMAL_ACCOUNT,
+ dsdb.UF_NORMAL_ACCOUNT,
msDSUserAccountControlComputed=0)
if lockOutObservationWindow != 0:
time.sleep(lockOutObservationWindow + 1)
lastLogon=0,
lastLogonTimestamp=('absent', None),
userAccountControl=
- dsdb.UF_NORMAL_ACCOUNT,
+ dsdb.UF_NORMAL_ACCOUNT,
msDSUserAccountControlComputed=0)
ldb = SamDB(url=self.host_url, credentials=creds, lp=self.lp)
lastLogon=(lastlogon_relation, 0),
lastLogonTimestamp=('greater', badPasswordTime),
userAccountControl=
- dsdb.UF_NORMAL_ACCOUNT,
+ dsdb.UF_NORMAL_ACCOUNT,
msDSUserAccountControlComputed=0)
logonCount = int(res[0]["logonCount"][0])
lastLogon=lastLogon,
lastLogonTimestamp=lastLogonTimestamp,
userAccountControl=
- dsdb.UF_NORMAL_ACCOUNT,
+ dsdb.UF_NORMAL_ACCOUNT,
msDSUserAccountControlComputed=0)
return ldb
lastLogon=("greater", 0),
lastLogonTimestamp=("greater", 0),
userAccountControl=
- dsdb.UF_NORMAL_ACCOUNT,
+ dsdb.UF_NORMAL_ACCOUNT,
msDSUserAccountControlComputed=0)
badPasswordTime = int(res[0]["badPasswordTime"][0])
logonCount = int(res[0]["logonCount"][0])
lastLogon=lastLogon,
lastLogonTimestamp=lastLogonTimestamp,
userAccountControl=
- dsdb.UF_NORMAL_ACCOUNT,
+ dsdb.UF_NORMAL_ACCOUNT,
msDSUserAccountControlComputed=0,
msg='lastlogontimestamp with wrong password')
badPasswordTime = int(res[0]["badPasswordTime"][0])
lastLogon=('greater', lastLogon),
lastLogonTimestamp=lastLogonTimestamp,
userAccountControl=
- dsdb.UF_NORMAL_ACCOUNT,
+ dsdb.UF_NORMAL_ACCOUNT,
msDSUserAccountControlComputed=0,
msg='LLTimestamp is updated to lastlogon')
lastLogon=lastLogon,
lastLogonTimestamp=lastLogonTimestamp,
userAccountControl=
- dsdb.UF_NORMAL_ACCOUNT,
+ dsdb.UF_NORMAL_ACCOUNT,
msDSUserAccountControlComputed=0)
badPasswordTime = int(res[0]["badPasswordTime"][0])
lastLogon=lastLogon,
lastLogonTimestamp=lastLogonTimestamp,
userAccountControl=
- dsdb.UF_NORMAL_ACCOUNT,
+ dsdb.UF_NORMAL_ACCOUNT,
msDSUserAccountControlComputed=0)
badPasswordTime = int(res[0]["badPasswordTime"][0])
lastLogonTimestamp=lastLogonTimestamp,
lockoutTime=("greater", badPasswordTime),
userAccountControl=
- dsdb.UF_NORMAL_ACCOUNT,
+ dsdb.UF_NORMAL_ACCOUNT,
msDSUserAccountControlComputed=dsdb.UF_LOCKOUT)
badPasswordTime = int(res[0]["badPasswordTime"][0])
lockoutTime = int(res[0]["lockoutTime"][0])
lastLogonTimestamp=lastLogonTimestamp,
lockoutTime=lockoutTime,
userAccountControl=
- dsdb.UF_NORMAL_ACCOUNT,
+ dsdb.UF_NORMAL_ACCOUNT,
msDSUserAccountControlComputed=dsdb.UF_LOCKOUT)
# The wrong password
lastLogonTimestamp=lastLogonTimestamp,
lockoutTime=lockoutTime,
userAccountControl=
- dsdb.UF_NORMAL_ACCOUNT,
+ dsdb.UF_NORMAL_ACCOUNT,
msDSUserAccountControlComputed=dsdb.UF_LOCKOUT)
# The correct password, but we are locked out
lastLogonTimestamp=lastLogonTimestamp,
lockoutTime=lockoutTime,
userAccountControl=
- dsdb.UF_NORMAL_ACCOUNT,
+ dsdb.UF_NORMAL_ACCOUNT,
msDSUserAccountControlComputed=dsdb.UF_LOCKOUT)
# wait for the lockout to end
lastLogon=lastLogon,
lastLogonTimestamp=lastLogonTimestamp,
userAccountControl=
- dsdb.UF_NORMAL_ACCOUNT,
+ dsdb.UF_NORMAL_ACCOUNT,
msDSUserAccountControlComputed=0)
# The correct password after letting the timeout expire
lastLogonTimestamp=lastLogonTimestamp,
lockoutTime=0,
userAccountControl=
- dsdb.UF_NORMAL_ACCOUNT,
+ dsdb.UF_NORMAL_ACCOUNT,
msDSUserAccountControlComputed=0,
msg="lastLogon is way off")
lastLogon=lastLogon,
lastLogonTimestamp=lastLogonTimestamp,
userAccountControl=
- dsdb.UF_NORMAL_ACCOUNT,
+ dsdb.UF_NORMAL_ACCOUNT,
msDSUserAccountControlComputed=0)
badPasswordTime = int(res[0]["badPasswordTime"][0])
lastLogon=lastLogon,
lastLogonTimestamp=lastLogonTimestamp,
userAccountControl=
- dsdb.UF_NORMAL_ACCOUNT,
+ dsdb.UF_NORMAL_ACCOUNT,
msDSUserAccountControlComputed=0)
badPasswordTime = int(res[0]["badPasswordTime"][0])
lastLogon=lastLogon,
lastLogonTimestamp=lastLogonTimestamp,
userAccountControl=
- dsdb.UF_NORMAL_ACCOUNT,
+ dsdb.UF_NORMAL_ACCOUNT,
msDSUserAccountControlComputed=0)
# The wrong password
lastLogon=lastLogon,
lastLogonTimestamp=lastLogonTimestamp,
userAccountControl=
- dsdb.UF_NORMAL_ACCOUNT,
+ dsdb.UF_NORMAL_ACCOUNT,
msDSUserAccountControlComputed=0)
badPasswordTime = int(res[0]["badPasswordTime"][0])
lastLogon=("greater", lastLogon),
lastLogonTimestamp=lastLogonTimestamp,
userAccountControl=
- dsdb.UF_NORMAL_ACCOUNT,
+ dsdb.UF_NORMAL_ACCOUNT,
msDSUserAccountControlComputed=0)
def _test_multiple_logon(self, creds):
lastLogon=("greater", 0),
lastLogonTimestamp=("greater", 0),
userAccountControl=
- dsdb.UF_NORMAL_ACCOUNT,
+ dsdb.UF_NORMAL_ACCOUNT,
msDSUserAccountControlComputed=0)
badPasswordTime = int(res[0]["badPasswordTime"][0])
logonCount = int(res[0]["logonCount"][0])
lastLogon=(lastlogon_relation, lastLogon),
lastLogonTimestamp=lastLogonTimestamp,
userAccountControl=
- dsdb.UF_NORMAL_ACCOUNT,
+ dsdb.UF_NORMAL_ACCOUNT,
msDSUserAccountControlComputed=0)
creds2.set_realm(creds.get_realm())
creds2.set_workstation(creds.get_workstation())
creds2.set_gensec_features(creds2.get_gensec_features()
- | gensec.FEATURE_SEAL)
+ | gensec.FEATURE_SEAL)
self.ldb2 = SamDB(url=host, credentials=creds2, lp=lp)
def test_unicodePwd_hash_set(self):
lastLogon=lastLogon,
lastLogonTimestamp=lastLogonTimestamp,
userAccountControl=
- dsdb.UF_NORMAL_ACCOUNT,
+ dsdb.UF_NORMAL_ACCOUNT,
msDSUserAccountControlComputed=0,
msg='lastlogontimestamp with wrong password')
badPasswordTime = int(res[0]["badPasswordTime"][0])
lastLogon=('greater', lastLogon),
lastLogonTimestamp=lastLogonTimestamp,
userAccountControl=
- dsdb.UF_NORMAL_ACCOUNT,
+ dsdb.UF_NORMAL_ACCOUNT,
msDSUserAccountControlComputed=0,
msg='badPwdCount not reset on RWDC')
lastLogon=lastLogon,
lastLogonTimestamp=lastLogonTimestamp,
userAccountControl=
- dsdb.UF_NORMAL_ACCOUNT,
+ dsdb.UF_NORMAL_ACCOUNT,
msDSUserAccountControlComputed=0,
msg='lastlogontimestamp with wrong password')
badPasswordTime = int(res[0]["badPasswordTime"][0])
lastLogon=('greater', lastLogon),
lastLogonTimestamp=lastLogonTimestamp,
userAccountControl=
- dsdb.UF_NORMAL_ACCOUNT,
+ dsdb.UF_NORMAL_ACCOUNT,
msDSUserAccountControlComputed=0,
msg='LLTimestamp is updated to lastlogon')
lastLogon=lastLogon,
lastLogonTimestamp=lastLogonTimestamp,
userAccountControl=
- dsdb.UF_NORMAL_ACCOUNT,
+ dsdb.UF_NORMAL_ACCOUNT,
msDSUserAccountControlComputed=0)
badPasswordTime = int(res[0]["badPasswordTime"][0])
lastLogon=lastLogon,
lastLogonTimestamp=lastLogonTimestamp,
userAccountControl=
- dsdb.UF_NORMAL_ACCOUNT,
+ dsdb.UF_NORMAL_ACCOUNT,
msDSUserAccountControlComputed=0)
badPasswordTime = int(res[0]["badPasswordTime"][0])
lastLogonTimestamp=lastLogonTimestamp,
lockoutTime=("greater", badPasswordTime),
userAccountControl=
- dsdb.UF_NORMAL_ACCOUNT,
+ dsdb.UF_NORMAL_ACCOUNT,
msDSUserAccountControlComputed=dsdb.UF_LOCKOUT)
badPasswordTime = int(res[0]["badPasswordTime"][0])
lockoutTime = int(res[0]["lockoutTime"][0])
lastLogonTimestamp=lastLogonTimestamp,
lockoutTime=lockoutTime,
userAccountControl=
- dsdb.UF_NORMAL_ACCOUNT,
+ dsdb.UF_NORMAL_ACCOUNT,
msDSUserAccountControlComputed=dsdb.UF_LOCKOUT)
# The wrong password
lastLogonTimestamp=lastLogonTimestamp,
lockoutTime=lockoutTime,
userAccountControl=
- dsdb.UF_NORMAL_ACCOUNT,
+ dsdb.UF_NORMAL_ACCOUNT,
msDSUserAccountControlComputed=dsdb.UF_LOCKOUT)
# The correct password, but we are locked out
lastLogonTimestamp=lastLogonTimestamp,
lockoutTime=lockoutTime,
userAccountControl=
- dsdb.UF_NORMAL_ACCOUNT,
+ dsdb.UF_NORMAL_ACCOUNT,
msDSUserAccountControlComputed=dsdb.UF_LOCKOUT)
# wait for the lockout to end
lastLogon=lastLogon,
lastLogonTimestamp=lastLogonTimestamp,
userAccountControl=
- dsdb.UF_NORMAL_ACCOUNT,
+ dsdb.UF_NORMAL_ACCOUNT,
msDSUserAccountControlComputed=0)
# The correct password after letting the timeout expire
lastLogonTimestamp=lastLogonTimestamp,
lockoutTime=lockoutTime,
userAccountControl=
- dsdb.UF_NORMAL_ACCOUNT,
+ dsdb.UF_NORMAL_ACCOUNT,
msDSUserAccountControlComputed=0,
msg="lastLogon is way off")
lastLogon=lastLogon,
lastLogonTimestamp=lastLogonTimestamp,
userAccountControl=
- dsdb.UF_NORMAL_ACCOUNT,
+ dsdb.UF_NORMAL_ACCOUNT,
msDSUserAccountControlComputed=0)
badPasswordTime = int(res[0]["badPasswordTime"][0])
lastLogon=lastLogon,
lastLogonTimestamp=lastLogonTimestamp,
userAccountControl=
- dsdb.UF_NORMAL_ACCOUNT,
+ dsdb.UF_NORMAL_ACCOUNT,
msDSUserAccountControlComputed=0)
badPasswordTime = int(res[0]["badPasswordTime"][0])
lastLogon=lastLogon,
lastLogonTimestamp=lastLogonTimestamp,
userAccountControl=
- dsdb.UF_NORMAL_ACCOUNT,
+ dsdb.UF_NORMAL_ACCOUNT,
msDSUserAccountControlComputed=0)
# The wrong password
lastLogon=lastLogon,
lastLogonTimestamp=lastLogonTimestamp,
userAccountControl=
- dsdb.UF_NORMAL_ACCOUNT,
+ dsdb.UF_NORMAL_ACCOUNT,
msDSUserAccountControlComputed=0)
badPasswordTime = int(res[0]["badPasswordTime"][0])
lastLogon=("greater", lastLogon),
lastLogonTimestamp=lastLogonTimestamp,
userAccountControl=
- dsdb.UF_NORMAL_ACCOUNT,
+ dsdb.UF_NORMAL_ACCOUNT,
msDSUserAccountControlComputed=0)
class RodcRwdcTests(password_lockout_base.BasePasswordTestCase):
"servicePrincipalName": "HOST/testname.testdom"})
res = ldb.search("cn=ldaptestcomputer,cn=computers," + self.base_dn,
- scope=SCOPE_BASE, attrs=["dNSHostName"])
+ scope=SCOPE_BASE, attrs=["dNSHostName"])
self.assertTrue(len(res) == 1)
self.assertFalse("dNSHostName" in res[0])
ldb.modify(m)
res = ldb.search("cn=ldaptestcomputer,cn=computers," + self.base_dn,
- scope=SCOPE_BASE, attrs=["servicePrincipalName"])
+ scope=SCOPE_BASE, attrs=["servicePrincipalName"])
self.assertTrue(len(res) == 1)
self.assertEquals(res[0]["servicePrincipalName"][0],
"HOST/testname3.testdom3")
def tearDown(self):
super(DynamicTokenTest, self).tearDown()
delete_force(self.admin_ldb, "CN=%s,%s,%s" %
- (self.test_user, "cn=users", self.base_dn))
+ (self.test_user, "cn=users", self.base_dn))
delete_force(self.admin_ldb, "CN=%s,%s,%s" %
- (self.test_group0, "cn=users", self.base_dn))
+ (self.test_group0, "cn=users", self.base_dn))
delete_force(self.admin_ldb, "CN=%s,%s,%s" %
- (self.test_group1, "cn=users", self.base_dn))
+ (self.test_group1, "cn=users", self.base_dn))
delete_force(self.admin_ldb, "CN=%s,%s,%s" %
- (self.test_group2, "cn=users", self.base_dn))
+ (self.test_group2, "cn=users", self.base_dn))
delete_force(self.admin_ldb, "CN=%s,%s,%s" %
- (self.test_group3, "cn=users", self.base_dn))
+ (self.test_group3, "cn=users", self.base_dn))
delete_force(self.admin_ldb, "CN=%s,%s,%s" %
- (self.test_group4, "cn=users", self.base_dn))
+ (self.test_group4, "cn=users", self.base_dn))
delete_force(self.admin_ldb, "CN=%s,%s,%s" %
- (self.test_group5, "cn=users", self.base_dn))
+ (self.test_group5, "cn=users", self.base_dn))
delete_force(self.admin_ldb, "CN=%s,%s,%s" %
- (self.test_group6, "cn=users", self.base_dn))
+ (self.test_group6, "cn=users", self.base_dn))
def test_rootDSE_tokenGroups(self):
"""Testing rootDSE tokengroups against internal calculation"""
rids = samr_conn.GetGroupsForUser(user_handle)
user_info = samr_conn.QueryUserInfo(user_handle, 1)
delete_force(self.admin_ldb, "CN=%s,%s,%s" %
- (test_user, "cn=users", self.base_dn))
+ (test_user, "cn=users", self.base_dn))
self.assertEqual(len(rids.rids), 1)
self.assertEqual(rids.rids[0].rid, user_info.primary_gid)
store_dn_len, store_dn_ofs, \
store_hwm_len, store_hwm_ofs, \
store_utdv_len, store_utdv_ofs) = \
- struct.unpack("<LLLLLLL", store_hdr)
+ struct.unpack("<LLLLLLL", store_hdr)
store_dn = store_blob[store_dn_ofs:store_dn_ofs+store_dn_len]
store_hwm_blob = store_blob[store_hwm_ofs:store_hwm_ofs+store_hwm_len]
len(dn), dn_ofs,
len(store_hwm_blob), hwm_ofs, \
len(store_utdv_blob), utdv_ofs) + \
- dn + store_hwm_blob + store_utdv_blob
+ dn + store_hwm_blob + store_utdv_blob
tmp_file = "%s.tmp" % cookie_file
f = open(tmp_file, 'wb')
plantestsuite("samba4.dsdb.samdb.ldb_modules.unique_object_sids" , "none",
[os.path.join(bindir(), "test_unique_object_sids")])
plantestsuite("samba4.dsdb.samdb.ldb_modules.encrypted_secrets", "none",
- [os.path.join(bindir(), "test_encrypted_secrets")])
+ [os.path.join(bindir(), "test_encrypted_secrets")])
plantestsuite("lib.audit_logging.audit_logging", "none",
- [os.path.join(bindir(), "audit_logging_test")])
+ [os.path.join(bindir(), "audit_logging_test")])
plantestsuite("lib.audit_logging.audit_logging.errors", "none",
- [os.path.join(bindir(), "audit_logging_error_test")])
+ [os.path.join(bindir(), "audit_logging_error_test")])
plantestsuite("samba4.dsdb.samdb.ldb_modules.audit_util", "none",
- [os.path.join(bindir(), "test_audit_util")])
+ [os.path.join(bindir(), "test_audit_util")])
plantestsuite("samba4.dsdb.samdb.ldb_modules.audit_log", "none",
- [os.path.join(bindir(), "test_audit_log")])
+ [os.path.join(bindir(), "test_audit_log")])
plantestsuite("samba4.dsdb.samdb.ldb_modules.audit_log.errors", "none",
- [os.path.join(bindir(), "test_audit_log_errors")])
+ [os.path.join(bindir(), "test_audit_log_errors")])
plantestsuite("samba4.dsdb.samdb.ldb_modules.group_audit", "none",
- [os.path.join(bindir(), "test_group_audit")])
+ [os.path.join(bindir(), "test_group_audit")])
plantestsuite("samba4.dsdb.samdb.ldb_modules.group_audit.errors", "none",
- [os.path.join(bindir(), "test_group_audit_errors")])
+ [os.path.join(bindir(), "test_group_audit_errors")])
m = ldb.Message()
m.dn = group_dn
m["member"] = ldb.MessageElement(str(user_dn),
- ldb.FLAG_MOD_ADD, "member")
+ ldb.FLAG_MOD_ADD, "member")
self.ldb_dc2.modify(m)
ldb_res = self.ldb_dc2.search(base=self.domain_dn,
m = ldb.Message()
m.dn = group_dn
m["member"] = ldb.MessageElement(str(user_dn),
- ldb.FLAG_MOD_ADD, "member")
+ ldb.FLAG_MOD_ADD, "member")
self.ldb_dc2.modify(m)
ldb_res = self.ldb_dc2.search(base=self.domain_dn,
print("More data: %d" %(ctr6.more_data))
def _get_replication(self, replica_flags,
- drs_error=drsuapi.DRSUAPI_EXOP_ERR_NONE, drs=None, drs_handle=None,
- highwatermark=None, uptodateness_vector=None,
- more_flags=0, max_objects=133, exop=0,
- dest_dsa=drsuapi.DRSUAPI_DS_BIND_GUID_W2K3,
- source_dsa=None, invocation_id=None, nc_dn_str=None):
+ drs_error=drsuapi.DRSUAPI_EXOP_ERR_NONE, drs=None, drs_handle=None,
+ highwatermark=None, uptodateness_vector=None,
+ more_flags=0, max_objects=133, exop=0,
+ dest_dsa=drsuapi.DRSUAPI_DS_BIND_GUID_W2K3,
+ source_dsa=None, invocation_id=None, nc_dn_str=None):
"""
Builds a DsGetNCChanges request based on the information provided
and returns the response received from the DC.
nc_dn_str=dc3,
exop=drsuapi.DRSUAPI_EXOP_REPL_OBJ,
replica_flags=drsuapi.DRSUAPI_DRS_WRIT_REP |
- drsuapi.DRSUAPI_DRS_GET_ANC,
+ drsuapi.DRSUAPI_DRS_GET_ANC,
more_flags=drsuapi.DRSUAPI_DRS_GET_TGT)
(level, ctr) = self.drs.DsGetNCChanges(self.drs_handle, 10, req)
self._check_ctr6(ctr, [dc3], expected_links=[dc3_link])
nc_dn_str=ou2,
exop=drsuapi.DRSUAPI_EXOP_REPL_OBJ,
replica_flags=drsuapi.DRSUAPI_DRS_CRITICAL_ONLY |
- drsuapi.DRSUAPI_DRS_GET_ANC,
+ drsuapi.DRSUAPI_DRS_GET_ANC,
more_flags=drsuapi.DRSUAPI_DRS_GET_TGT)
(level, ctr) = self.drs.DsGetNCChanges(self.drs_handle, 10, req)
self._check_ctr6(ctr, [ou2])
"cn": class_name,
"lDAPDisplayName": class_ldn,
"governsId": "1.3.6.1.4.1.7165.4.6.2.5." \
- + str((100000 * base_int) + random.randint(1,100000)) + ".1.5.13",
+ + str((100000 * base_int) + random.randint(1,100000)) + ".1.5.13",
"instanceType": "4",
"objectClassCategory": "%d" % oc_cat,
"subClassOf": "top",
"cn": attr_name,
"lDAPDisplayName": attr_ldn,
"attributeId": "1.3.6.1.4.1.7165.4.6.1.5." \
- + str((100000 * base_int) + random.randint(1,100000)) + ".1.5.13",
+ + str((100000 * base_int) + random.randint(1,100000)) + ".1.5.13",
"attributeSyntax": "2.5.5.12",
"omSyntax": "64",
"instanceType": "4",
def set_krb5_conf(t):
t.run_cmd("mkdir -p ${PREFIX}/etc")
t.write_file("${PREFIX}/etc/krb5.conf",
- '''[libdefaults]
+ '''[libdefaults]
dns_lookup_realm = false
dns_lookup_kdc = true''')
git.samba.org / metze / samba / wip.git / commitdiff