ldap.py

samba最新软件

#!/usr/bin/python
# -*- coding: utf-8 -*-
# This is a port of the original in testprogs/ejs/ldap.js

import getopt
import optparse
import sys

sys.path.append("bin/python")

import samba.getopt as options

from samba.auth import system_session
from ldb import (SCOPE_SUBTREE, SCOPE_ONELEVEL, SCOPE_BASE, LdbError,
                 LDB_ERR_NO_SUCH_OBJECT, LDB_ERR_ATTRIBUTE_OR_VALUE_EXISTS,
                 LDB_ERR_ENTRY_ALREADY_EXISTS, LDB_ERR_UNWILLING_TO_PERFORM,
                 LDB_ERR_NOT_ALLOWED_ON_NON_LEAF, LDB_ERR_OTHER)
from samba import Ldb
from subunit import SubunitTestRunner
from samba import param
import unittest

parser = optparse.OptionParser("ldap [options] <host>")
sambaopts = options.SambaOptions(parser)
parser.add_option_group(sambaopts)
parser.add_option_group(options.VersionOptions(parser))
# use command line creds if available
credopts = options.CredentialsOptions(parser)
parser.add_option_group(credopts)
opts, args = parser.parse_args()

if len(args) < 1:
    parser.print_usage()
    sys.exit(1)

host = args[0]

lp = sambaopts.get_loadparm()
creds = credopts.get_credentials(lp)

class BasicTests(unittest.TestCase):
    def delete_force(self, ldb, dn):
        try:
            ldb.delete(dn)
        except LdbError, (num, _): 
            self.assertEquals(num, LDB_ERR_NO_SUCH_OBJECT)

    def find_basedn(self, ldb):
        res = ldb.search(base="", expression="", scope=SCOPE_BASE, 
                         attrs=["defaultNamingContext"])
        self.assertEquals(len(res), 1)
        return res[0]["defaultNamingContext"][0]

    def find_configurationdn(self, ldb):
        res = ldb.search(base="", expression="", scope=SCOPE_BASE, attrs=["configurationNamingContext"])
        self.assertEquals(len(res), 1)
        return res[0]["configurationNamingContext"][0]

    def find_schemadn(self, ldb):
        res = ldb.search(base="", expression="", scope=SCOPE_BASE, attrs=["schemaNamingContext"])
        self.assertEquals(len(res), 1)
        return res[0]["schemaNamingContext"][0]

    def setUp(self):
        self.ldb = ldb
        self.gc_ldb = gc_ldb
        self.base_dn = self.find_basedn(ldb)
        self.configuration_dn = self.find_configurationdn(ldb)
        self.schema_dn = self.find_schemadn(ldb)

        print "baseDN: %s\n" % self.base_dn

        self.delete_force(self.ldb, "cn=ldaptestuser,cn=users," + self.base_dn)
        self.delete_force(self.ldb, "cn=ldaptestgroup,cn=users," + self.base_dn)

    def test_group_add_invalid_member(self):
        """Testing group add with invalid member"""
        try:
            self.ldb.add({
                "dn": "cn=ldaptestgroup,cn=uSers," + self.base_dn,
                "objectclass": "group",
                "member": "cn=ldaptestuser,cn=useRs," + self.base_dn})
            self.fail()
        except LdbError, (num, _): 
            self.assertEquals(num, LDB_ERR_NO_SUCH_OBJECT)

    def test_all(self):
        """Basic tests"""

        self.delete_force(self.ldb, "cn=ldaptestuser,cn=users," + self.base_dn)

        print "Testing user add"
        ldb.add({
        "dn": "cn=ldaptestuser,cn=uSers," + self.base_dn,
        "objectclass": ["user", "person"],
        "cN": "LDAPtestUSER",
        "givenname": "ldap",
        "sn": "testy"})

        ldb.add({
            "dn": "cn=ldaptestgroup,cn=uSers," + self.base_dn,
            "objectclass": "group",
            "member": "cn=ldaptestuser,cn=useRs," + self.base_dn})

        self.delete_force(ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)
        ldb.add({
            "dn": "cn=ldaptestcomputer,cn=computers," + self.base_dn,
            "objectclass": "computer",
            "cN": "LDAPtestCOMPUTER"})

        self.delete_force(self.ldb, "cn=ldaptest2computer,cn=computers," + self.base_dn)
        ldb.add({"dn": "cn=ldaptest2computer,cn=computers," + self.base_dn,
            "objectClass": "computer",
            "cn": "LDAPtest2COMPUTER",
            "userAccountControl": "4096",
            "displayname": "ldap testy"})

        print "Testing attribute or value exists behaviour"
        try:
            ldb.modify_ldif("""
dn: cn=ldaptest2computer,cn=computers,""" + self.base_dn + """
changetype: modify
replace: servicePrincipalName
servicePrincipalName: host/ldaptest2computer
servicePrincipalName: host/ldaptest2computer
servicePrincipalName: cifs/ldaptest2computer
""")
        except LdbError, (num, msg):
            self.assertEquals(num, LDB_ERR_ATTRIBUTE_OR_VALUE_EXISTS)

            ldb.modify_ldif("""
dn: cn=ldaptest2computer,cn=computers,""" + self.base_dn + """
changetype: modify
replace: servicePrincipalName
servicePrincipalName: host/ldaptest2computer
servicePrincipalName: cifs/ldaptest2computer
""")
            try:
                ldb.modify_ldif("""
dn: cn=ldaptest2computer,cn=computers,""" + self.base_dn + """
changetype: modify
add: servicePrincipalName
servicePrincipalName: host/ldaptest2computer
""")
            except LdbError, (num, msg):
                self.assertEquals(num, LDB_ERR_ATTRIBUTE_OR_VALUE_EXISTS)
            
            print "Testing ranged results"
            ldb.modify_ldif("""
dn: cn=ldaptest2computer,cn=computers,""" + self.base_dn + """
changetype: modify
replace: servicePrincipalName
""")
            
            ldb.modify_ldif("""
dn: cn=ldaptest2computer,cn=computers,""" + self.base_dn + """
changetype: modify
add: servicePrincipalName
servicePrincipalName: host/ldaptest2computer0
servicePrincipalName: host/ldaptest2computer1
servicePrincipalName: host/ldaptest2computer2
servicePrincipalName: host/ldaptest2computer3
servicePrincipalName: host/ldaptest2computer4
servicePrincipalName: host/ldaptest2computer5
servicePrincipalName: host/ldaptest2computer6
servicePrincipalName: host/ldaptest2computer7
servicePrincipalName: host/ldaptest2computer8
servicePrincipalName: host/ldaptest2computer9
servicePrincipalName: host/ldaptest2computer10
servicePrincipalName: host/ldaptest2computer11
servicePrincipalName: host/ldaptest2computer12
servicePrincipalName: host/ldaptest2computer13
servicePrincipalName: host/ldaptest2computer14
servicePrincipalName: host/ldaptest2computer15
servicePrincipalName: host/ldaptest2computer16
servicePrincipalName: host/ldaptest2computer17
servicePrincipalName: host/ldaptest2computer18
servicePrincipalName: host/ldaptest2computer19
servicePrincipalName: host/ldaptest2computer20
servicePrincipalName: host/ldaptest2computer21
servicePrincipalName: host/ldaptest2computer22
servicePrincipalName: host/ldaptest2computer23
servicePrincipalName: host/ldaptest2computer24
servicePrincipalName: host/ldaptest2computer25
servicePrincipalName: host/ldaptest2computer26
servicePrincipalName: host/ldaptest2computer27
servicePrincipalName: host/ldaptest2computer28
servicePrincipalName: host/ldaptest2computer29
""")

            res = ldb.search(self.base_dn, expression="(cn=ldaptest2computer))", scope=SCOPE_SUBTREE, 
                             attrs=["servicePrincipalName;range=0-*"])
            self.assertEquals(len(res), 1, "Could not find (cn=ldaptest2computer)")
            #print len(res[0]["servicePrincipalName;range=0-*"])
            self.assertEquals(len(res[0]["servicePrincipalName;range=0-*"]), 30)

            res = ldb.search(self.base_dn, expression="(cn=ldaptest2computer))", scope=SCOPE_SUBTREE, attrs=["servicePrincipalName;range=0-19"])
            self.assertEquals(len(res), 1, "Could not find (cn=ldaptest2computer)")
            # print res[0]["servicePrincipalName;range=0-19"].length
            self.assertEquals(len(res[0]["servicePrincipalName;range=0-19"]), 20)

            
            res = ldb.search(self.base_dn, expression="(cn=ldaptest2computer))", scope=SCOPE_SUBTREE, attrs=["servicePrincipalName;range=0-30"])
            self.assertEquals(len(res), 1, "Could not find (cn=ldaptest2computer)")
            self.assertEquals(len(res[0]["servicePrincipalName;range=0-*"]), 30)

            res = ldb.search(self.base_dn, expression="(cn=ldaptest2computer))", scope=SCOPE_SUBTREE, attrs=["servicePrincipalName;range=0-40"])
            self.assertEquals(len(res), 1, "Could not find (cn=ldaptest2computer)")
            self.assertEquals(len(res[0]["servicePrincipalName;range=0-*"]), 30)

            res = ldb.search(self.base_dn, expression="(cn=ldaptest2computer))", scope=SCOPE_SUBTREE, attrs=["servicePrincipalName;range=30-40"])
            self.assertEquals(len(res), 1, "Could not find (cn=ldaptest2computer)")
            self.assertEquals(len(res[0]["servicePrincipalName;range=30-*"]), 0)

            
            res = ldb.search(self.base_dn, expression="(cn=ldaptest2computer))", scope=SCOPE_SUBTREE, attrs=["servicePrincipalName;range=10-40"])
            self.assertEquals(len(res), 1, "Could not find (cn=ldaptest2computer)")
            self.assertEquals(len(res[0]["servicePrincipalName;range=10-*"]), 20)
            # pos_11 = res[0]["servicePrincipalName;range=10-*"][18]

            res = ldb.search(self.base_dn, expression="(cn=ldaptest2computer))", scope=SCOPE_SUBTREE, attrs=["servicePrincipalName;range=11-40"])
            self.assertEquals(len(res), 1, "Could not find (cn=ldaptest2computer)")
            self.assertEquals(len(res[0]["servicePrincipalName;range=11-*"]), 19)
            # print res[0]["servicePrincipalName;range=11-*"][18]
            # print pos_11
            # self.assertEquals((res[0]["servicePrincipalName;range=11-*"][18]), pos_11)

            res = ldb.search(self.base_dn, expression="(cn=ldaptest2computer))", scope=SCOPE_SUBTREE, attrs=["servicePrincipalName;range=11-15"])
            self.assertEquals(len(res), 1, "Could not find (cn=ldaptest2computer)")
            self.assertEquals(len(res[0]["servicePrincipalName;range=11-15"]), 5)
            # self.assertEquals(res[0]["servicePrincipalName;range=11-15"][4], pos_11)

            res = ldb.search(self.base_dn, expression="(cn=ldaptest2computer))", scope=SCOPE_SUBTREE, attrs=["servicePrincipalName"])
            self.assertEquals(len(res), 1, "Could not find (cn=ldaptest2computer)")
            # print res[0]["servicePrincipalName"][18]
            # print pos_11
            self.assertEquals(len(res[0]["servicePrincipalName"]), 30)
            # self.assertEquals(res[0]["servicePrincipalName"][18], pos_11)

        self.delete_force(self.ldb, "cn=ldaptestuser2,cn=users," + self.base_dn)
        ldb.add({
            "dn": "cn=ldaptestuser2,cn=useRs," + self.base_dn,
            "objectClass": ["person", "user"],
            "cn": "LDAPtestUSER2",
            "givenname": "testy",
            "sn": "ldap user2"})

        print "Testing Ambigious Name Resolution"
        # Testing ldb.search for (&(anr=ldap testy)(objectClass=user))
        res = ldb.search(expression="(&(anr=ldap testy)(objectClass=user))")
        self.assertEquals(len(res), 3, "Could not find (&(anr=ldap testy)(objectClass=user))")

        # Testing ldb.search for (&(anr=testy ldap)(objectClass=user))
        res = ldb.search(expression="(&(anr=testy ldap)(objectClass=user))")
        self.assertEquals(len(res), 2, "Found only %d for (&(anr=testy ldap)(objectClass=user))" % len(res))

        # Testing ldb.search for (&(anr=ldap)(objectClass=user))
        res = ldb.search(expression="(&(anr=ldap)(objectClass=user))")
        self.assertEquals(len(res), 4, "Found only %d for (&(anr=ldap)(objectClass=user))" % len(res))

        # Testing ldb.search for (&(anr==ldap)(objectClass=user))
        res = ldb.search(expression="(&(anr==ldap)(objectClass=user))")
        self.assertEquals(len(res), 1, "Could not find (&(anr==ldap)(objectClass=user)). Found only %d for (&(anr=ldap)(objectClass=user))" % len(res))

        self.assertEquals(str(res[0].dn), ("CN=ldaptestuser,CN=Users," + self.base_dn))
        self.assertEquals(res[0]["cn"][0], "ldaptestuser")
        self.assertEquals(res[0]["name"], "ldaptestuser")

        # Testing ldb.search for (&(anr=testy)(objectClass=user))
        res = ldb.search(expression="(&(anr=testy)(objectClass=user))")
        self.assertEquals(len(res), 2, "Found only %d for (&(anr=testy)(objectClass=user))" % len(res))

        # Testing ldb.search for (&(anr=ldap testy)(objectClass=user))
        res = ldb.search(expression="(&(anr=testy ldap)(objectClass=user))")
        self.assertEquals(len(res), 2, "Found only %d for (&(anr=ldap testy)(objectClass=user))" % len(res))

        # Testing ldb.search for (&(anr==ldap testy)(objectClass=user))
        res = ldb.search(expression="(&(anr==testy ldap)(objectClass=user))")
        self.assertEquals(len(res), 1, "Found only %d for (&(anr==ldap testy)(objectClass=user))" % len(res))

        self.assertEquals(str(res[0].dn), ("CN=ldaptestuser,CN=Users," + self.base_dn))
        self.assertEquals(res[0]["cn"][0], "ldaptestuser")
        self.assertEquals(res[0]["name"][0], "ldaptestuser")

        # Testing ldb.search for (&(anr==testy ldap)(objectClass=user))
        res = ldb.search(expression="(&(anr==testy ldap)(objectClass=user))")
        self.assertEquals(len(res), 1, "Could not find (&(anr==testy ldap)(objectClass=user))")

        self.assertEquals(str(res[0].dn), ("CN=ldaptestuser,CN=Users," + self.base_dn))
        self.assertEquals(res[0]["cn"][0], "ldaptestuser")
        self.assertEquals(res[0]["name"][0], "ldaptestuser")