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

import optparse
import sys
import os
import time

sys.path.insert(0, "bin/python")
import samba
from samba.tests.subunitrun import SubunitOptions, TestProgram

import samba.getopt as options

from samba.credentials import Credentials, DONT_USE_KERBEROS
from samba.auth import system_session
from samba.common import get_string

from ldb import SCOPE_BASE, LdbError
from ldb import ERR_NO_SUCH_OBJECT, ERR_ATTRIBUTE_OR_VALUE_EXISTS
from ldb import ERR_ENTRY_ALREADY_EXISTS, ERR_UNWILLING_TO_PERFORM
from ldb import ERR_OTHER, ERR_NO_SUCH_ATTRIBUTE
from ldb import ERR_OBJECT_CLASS_VIOLATION
from ldb import ERR_CONSTRAINT_VIOLATION
from ldb import ERR_UNDEFINED_ATTRIBUTE_TYPE
from ldb import ERR_INSUFFICIENT_ACCESS_RIGHTS
from ldb import ERR_INVALID_CREDENTIALS
from ldb import ERR_STRONG_AUTH_REQUIRED
from ldb import Message, MessageElement, Dn
from ldb import FLAG_MOD_ADD, FLAG_MOD_REPLACE, FLAG_MOD_DELETE
from samba.samdb import SamDB
from samba.dsdb import (UF_NORMAL_ACCOUNT, UF_ACCOUNTDISABLE,
                        UF_WORKSTATION_TRUST_ACCOUNT, UF_SERVER_TRUST_ACCOUNT,
                        UF_PARTIAL_SECRETS_ACCOUNT, UF_TEMP_DUPLICATE_ACCOUNT,
                        UF_INTERDOMAIN_TRUST_ACCOUNT, UF_SMARTCARD_REQUIRED,
                        UF_PASSWD_NOTREQD, UF_LOCKOUT, UF_PASSWORD_EXPIRED, ATYPE_NORMAL_ACCOUNT,
                        GTYPE_SECURITY_BUILTIN_LOCAL_GROUP, GTYPE_SECURITY_DOMAIN_LOCAL_GROUP,
                        GTYPE_SECURITY_GLOBAL_GROUP, GTYPE_SECURITY_UNIVERSAL_GROUP,
                        GTYPE_DISTRIBUTION_DOMAIN_LOCAL_GROUP, GTYPE_DISTRIBUTION_GLOBAL_GROUP,
                        GTYPE_DISTRIBUTION_UNIVERSAL_GROUP,
                        ATYPE_SECURITY_GLOBAL_GROUP, ATYPE_SECURITY_UNIVERSAL_GROUP,
                        ATYPE_SECURITY_LOCAL_GROUP, ATYPE_DISTRIBUTION_GLOBAL_GROUP,
                        ATYPE_DISTRIBUTION_UNIVERSAL_GROUP, ATYPE_DISTRIBUTION_LOCAL_GROUP,
                        ATYPE_WORKSTATION_TRUST)
from samba.dcerpc.security import (DOMAIN_RID_USERS, DOMAIN_RID_ADMINS,
                                   DOMAIN_RID_DOMAIN_MEMBERS, DOMAIN_RID_DCS, DOMAIN_RID_READONLY_DCS)

from samba.ndr import ndr_unpack
from samba.dcerpc import drsblobs
from samba.dcerpc import drsuapi
from samba.dcerpc import security
from samba.tests import delete_force
from samba import gensec
from samba import werror

parser = optparse.OptionParser("sam.py [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)
subunitopts = SubunitOptions(parser)
parser.add_option_group(subunitopts)
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)
creds.set_gensec_features(creds.get_gensec_features() | gensec.FEATURE_SEAL)


class SamTests(samba.tests.TestCase):

    def setUp(self):
        super(SamTests, self).setUp()
        self.ldb = ldb
        self.base_dn = ldb.domain_dn()

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

        delete_force(self.ldb, "cn=ldaptestuser,cn=users," + self.base_dn)
        delete_force(self.ldb, "cn=ldaptestuser2,cn=users," + self.base_dn)
        delete_force(self.ldb, r"cn=ldaptest\,specialuser,cn=users," + self.base_dn)
        delete_force(self.ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)
        delete_force(self.ldb, "cn=ldaptestcomputer2,cn=computers," + self.base_dn)
        delete_force(self.ldb, "cn=ldaptestgroup,cn=users," + self.base_dn)
        delete_force(self.ldb, "cn=ldaptestgroup2,cn=users," + self.base_dn)

    def test_users_groups(self):
        """This tests the SAM users and groups behaviour"""
        print("Testing users and groups behaviour\n")

        ldb.add({
            "dn": "cn=ldaptestgroup,cn=users," + self.base_dn,
            "objectclass": "group"})

        ldb.add({
            "dn": "cn=ldaptestgroup2,cn=users," + self.base_dn,
            "objectclass": "group"})

        res1 = ldb.search("cn=ldaptestgroup,cn=users," + self.base_dn,
                          scope=SCOPE_BASE, attrs=["objectSID"])
        self.assertTrue(len(res1) == 1)
        obj_sid = get_string(ldb.schema_format_value("objectSID",
                                                     res1[0]["objectSID"][0]))
        group_rid_1 = security.dom_sid(obj_sid).split()[1]

        res1 = ldb.search("cn=ldaptestgroup2,cn=users," + self.base_dn,
                          scope=SCOPE_BASE, attrs=["objectSID"])
        self.assertTrue(len(res1) == 1)
        obj_sid = get_string(ldb.schema_format_value("objectSID",
                                                     res1[0]["objectSID"][0]))
        group_rid_2 = security.dom_sid(obj_sid).split()[1]

        # Try to create a user with an invalid account name
        try:
            ldb.add({
                "dn": "cn=ldaptestuser,cn=users," + self.base_dn,
                "objectclass": "user",
                "sAMAccountName": "administrator"})
            self.fail()
        except LdbError as e9:
            (num, _) = e9.args
            self.assertEqual(num, ERR_ENTRY_ALREADY_EXISTS)
        delete_force(self.ldb, "cn=ldaptestuser,cn=users," + self.base_dn)

        # Try to create a user with an invalid account name
        try:
            ldb.add({
                "dn": "cn=ldaptestuser,cn=users," + self.base_dn,
                "objectclass": "user",
                "sAMAccountName": []})
            self.fail()
        except LdbError as e10:
            (num, _) = e10.args
            self.assertEqual(num, ERR_CONSTRAINT_VIOLATION)
        delete_force(self.ldb, "cn=ldaptestuser,cn=users," + self.base_dn)

        # Try to create a user with an invalid primary group
        try:
            ldb.add({
                "dn": "cn=ldaptestuser,cn=users," + self.base_dn,
                "objectclass": "user",
                "primaryGroupID": "0"})
            self.fail()
        except LdbError as e11:
            (num, _) = e11.args
            self.assertEqual(num, ERR_UNWILLING_TO_PERFORM)
        delete_force(self.ldb, "cn=ldaptestuser,cn=users," + self.base_dn)

        # Try to Create a user with a valid primary group
        try:
            ldb.add({
                "dn": "cn=ldaptestuser,cn=users," + self.base_dn,
                "objectclass": "user",
                "primaryGroupID": str(group_rid_1)})
            self.fail()
        except LdbError as e12:
            (num, _) = e12.args
            self.assertEqual(num, ERR_UNWILLING_TO_PERFORM)
        delete_force(self.ldb, "cn=ldaptestuser,cn=users," + self.base_dn)

        # Test to see how we should behave when the user account doesn't
        # exist
        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestuser,cn=users," + self.base_dn)
        m["primaryGroupID"] = MessageElement("0", FLAG_MOD_REPLACE,
                                             "primaryGroupID")
        try:
            ldb.modify(m)
            self.fail()
        except LdbError as e13:
            (num, _) = e13.args
            self.assertEqual(num, ERR_NO_SUCH_OBJECT)

        # Test to see how we should behave when the account isn't a user
        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestgroup,cn=users," + self.base_dn)
        m["primaryGroupID"] = MessageElement("0", FLAG_MOD_REPLACE,
                                             "primaryGroupID")
        try:
            ldb.modify(m)
            self.fail()
        except LdbError as e14:
            (num, _) = e14.args
            self.assertEqual(num, ERR_OBJECT_CLASS_VIOLATION)

        # Test default primary groups on add operations

        ldb.add({
            "dn": "cn=ldaptestuser,cn=users," + self.base_dn,
            "objectclass": "user"})

        res1 = ldb.search("cn=ldaptestuser,cn=users," + self.base_dn,
                          scope=SCOPE_BASE, attrs=["primaryGroupID"])
        self.assertTrue(len(res1) == 1)
        self.assertEqual(int(res1[0]["primaryGroupID"][0]), DOMAIN_RID_USERS)

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

        ldb.add({
            "dn": "cn=ldaptestuser,cn=users," + self.base_dn,
            "objectclass": "user",
            "userAccountControl": str(UF_NORMAL_ACCOUNT | UF_PASSWD_NOTREQD)})

        res1 = ldb.search("cn=ldaptestuser,cn=users," + self.base_dn,
                          scope=SCOPE_BASE, attrs=["primaryGroupID"])
        self.assertTrue(len(res1) == 1)
        self.assertEqual(int(res1[0]["primaryGroupID"][0]), DOMAIN_RID_USERS)

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

        # unfortunately the INTERDOMAIN_TRUST_ACCOUNT case cannot be tested
        # since such accounts aren't directly creatable (ACCESS_DENIED)

        ldb.add({
            "dn": "cn=ldaptestuser,cn=users," + self.base_dn,
            "objectclass": "computer",
            "userAccountControl": str(UF_WORKSTATION_TRUST_ACCOUNT |
                                      UF_PASSWD_NOTREQD)})

        res1 = ldb.search("cn=ldaptestuser,cn=users," + self.base_dn,
                          scope=SCOPE_BASE, attrs=["primaryGroupID"])
        self.assertTrue(len(res1) == 1)
        self.assertEqual(int(res1[0]["primaryGroupID"][0]),
                          DOMAIN_RID_DOMAIN_MEMBERS)

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

        ldb.add({
            "dn": "cn=ldaptestuser,cn=users," + self.base_dn,
            "objectclass": "computer",
            "userAccountControl": str(UF_SERVER_TRUST_ACCOUNT |
                                      UF_PASSWD_NOTREQD)})

        res1 = ldb.search("cn=ldaptestuser,cn=users," + self.base_dn,
                          scope=SCOPE_BASE, attrs=["primaryGroupID"])
        self.assertTrue(len(res1) == 1)
        self.assertEqual(int(res1[0]["primaryGroupID"][0]), DOMAIN_RID_DCS)

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

        # Read-only DC accounts are only creatable by
        # UF_WORKSTATION_TRUST_ACCOUNT and work only on DCs >= 2008 (therefore
        # we have a fallback in the assertion)
        ldb.add({
            "dn": "cn=ldaptestuser,cn=users," + self.base_dn,
            "objectclass": "computer",
            "userAccountControl": str(UF_PARTIAL_SECRETS_ACCOUNT |
                                      UF_WORKSTATION_TRUST_ACCOUNT |
                                      UF_PASSWD_NOTREQD)})

        res1 = ldb.search("cn=ldaptestuser,cn=users," + self.base_dn,
                          scope=SCOPE_BASE, attrs=["primaryGroupID"])
        self.assertTrue(len(res1) == 1)
        self.assertTrue(int(res1[0]["primaryGroupID"][0]) == DOMAIN_RID_READONLY_DCS or
                        int(res1[0]["primaryGroupID"][0]) == DOMAIN_RID_DOMAIN_MEMBERS)

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

        # Test default primary groups on modify operations

        ldb.add({
            "dn": "cn=ldaptestuser,cn=users," + self.base_dn,
            "objectclass": "user"})

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestuser,cn=users," + self.base_dn)
        m["userAccountControl"] = MessageElement(str(UF_NORMAL_ACCOUNT |
                                                     UF_PASSWD_NOTREQD),
                                                 FLAG_MOD_REPLACE,
                                                 "userAccountControl")
        ldb.modify(m)

        res1 = ldb.search("cn=ldaptestuser,cn=users," + self.base_dn,
                          scope=SCOPE_BASE, attrs=["primaryGroupID"])
        self.assertTrue(len(res1) == 1)
        self.assertEqual(int(res1[0]["primaryGroupID"][0]), DOMAIN_RID_USERS)

        # unfortunately the INTERDOMAIN_TRUST_ACCOUNT case cannot be tested
        # since such accounts aren't directly creatable (ACCESS_DENIED)

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

        ldb.add({
            "dn": "cn=ldaptestuser,cn=users," + self.base_dn,
            "objectclass": "computer",
            "userAccountControl": str(UF_NORMAL_ACCOUNT |
                                      UF_PASSWD_NOTREQD)})

        res1 = ldb.search("cn=ldaptestuser,cn=users," + self.base_dn,
                          scope=SCOPE_BASE, attrs=["primaryGroupID"])
        self.assertTrue(len(res1) == 1)
        self.assertEqual(int(res1[0]["primaryGroupID"][0]), DOMAIN_RID_USERS)

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestuser,cn=users," + self.base_dn)
        m["userAccountControl"] = MessageElement(str(UF_WORKSTATION_TRUST_ACCOUNT |
                                                     UF_PASSWD_NOTREQD),
                                                 FLAG_MOD_REPLACE,
                                                 "userAccountControl")
        ldb.modify(m)

        res1 = ldb.search("cn=ldaptestuser,cn=users," + self.base_dn,
                          scope=SCOPE_BASE, attrs=["primaryGroupID"])
        self.assertTrue(len(res1) == 1)
        self.assertEqual(int(res1[0]["primaryGroupID"][0]), DOMAIN_RID_DOMAIN_MEMBERS)

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestuser,cn=users," + self.base_dn)
        m["userAccountControl"] = MessageElement(str(UF_SERVER_TRUST_ACCOUNT |
                                                     UF_PASSWD_NOTREQD),
                                                 FLAG_MOD_REPLACE,
                                                 "userAccountControl")
        ldb.modify(m)

        res1 = ldb.search("cn=ldaptestuser,cn=users," + self.base_dn,
                          scope=SCOPE_BASE, attrs=["primaryGroupID"])
        self.assertTrue(len(res1) == 1)
        self.assertEqual(int(res1[0]["primaryGroupID"][0]), DOMAIN_RID_DCS)

        # Read-only DC accounts are only creatable by
        # UF_WORKSTATION_TRUST_ACCOUNT and work only on DCs >= 2008 (therefore
        # we have a fallback in the assertion)
        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestuser,cn=users," + self.base_dn)
        m["userAccountControl"] = MessageElement(str(UF_PARTIAL_SECRETS_ACCOUNT |
                                                     UF_WORKSTATION_TRUST_ACCOUNT |
                                                     UF_PASSWD_NOTREQD),
                                                 FLAG_MOD_REPLACE,
                                                 "userAccountControl")
        ldb.modify(m)

        res1 = ldb.search("cn=ldaptestuser,cn=users," + self.base_dn,
                          scope=SCOPE_BASE, attrs=["primaryGroupID"])
        self.assertTrue(len(res1) == 1)
        self.assertTrue(int(res1[0]["primaryGroupID"][0]) == DOMAIN_RID_READONLY_DCS or
                        int(res1[0]["primaryGroupID"][0]) == DOMAIN_RID_DOMAIN_MEMBERS)

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

        # Recreate account for further tests

        ldb.add({
            "dn": "cn=ldaptestuser,cn=users," + self.base_dn,
            "objectclass": "user"})

        # Try to set an invalid account name
        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestuser,cn=users," + self.base_dn)
        m["sAMAccountName"] = MessageElement("administrator", FLAG_MOD_REPLACE,
                                             "sAMAccountName")
        try:
            ldb.modify(m)
            self.fail()
        except LdbError as e15:
            (num, _) = e15.args
            self.assertEqual(num, ERR_ENTRY_ALREADY_EXISTS)

        # But to reset the actual "sAMAccountName" should still be possible
        res1 = ldb.search("cn=ldaptestuser,cn=users," + self.base_dn,
                          scope=SCOPE_BASE, attrs=["sAMAccountName"])
        self.assertTrue(len(res1) == 1)
        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestuser,cn=users," + self.base_dn)
        m["sAMAccountName"] = MessageElement(res1[0]["sAMAccountName"][0], FLAG_MOD_REPLACE,
                                             "sAMAccountName")
        ldb.modify(m)

        # And another (free) name should be possible as well
        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestuser,cn=users," + self.base_dn)
        m["sAMAccountName"] = MessageElement("xxx_ldaptestuser_xxx", FLAG_MOD_REPLACE,
                                             "sAMAccountName")
        ldb.modify(m)

        # We should be able to reset our actual primary group
        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestuser,cn=users," + self.base_dn)
        m["primaryGroupID"] = MessageElement(str(DOMAIN_RID_USERS), FLAG_MOD_REPLACE,
                                             "primaryGroupID")
        ldb.modify(m)

        # Try to add invalid primary group
        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestuser,cn=users," + self.base_dn)
        m["primaryGroupID"] = MessageElement("0", FLAG_MOD_REPLACE,
                                             "primaryGroupID")
        try:
            ldb.modify(m)
            self.fail()
        except LdbError as e16:
            (num, _) = e16.args
            self.assertEqual(num, ERR_UNWILLING_TO_PERFORM)

        # Try to make group 1 primary - should be denied since it is not yet
        # secondary
        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestuser,cn=users," + self.base_dn)
        m["primaryGroupID"] = MessageElement(str(group_rid_1),
                                             FLAG_MOD_REPLACE, "primaryGroupID")
        try:
            ldb.modify(m)
            self.fail()
        except LdbError as e17:
            (num, _) = e17.args
            self.assertEqual(num, ERR_UNWILLING_TO_PERFORM)

        # Make group 1 secondary
        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestgroup,cn=users," + self.base_dn)
        m["member"] = MessageElement("cn=ldaptestuser,cn=users," + self.base_dn,
                                     FLAG_MOD_REPLACE, "member")
        ldb.modify(m)

        # Make group 1 primary
        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestuser,cn=users," + self.base_dn)
        m["primaryGroupID"] = MessageElement(str(group_rid_1),
                                             FLAG_MOD_REPLACE, "primaryGroupID")
        ldb.modify(m)

        # Try to delete group 1 - should be denied
        try:
            ldb.delete("cn=ldaptestgroup,cn=users," + self.base_dn)
            self.fail()
        except LdbError as e18:
            (num, _) = e18.args
            self.assertEqual(num, ERR_ENTRY_ALREADY_EXISTS)

        # Try to add group 1 also as secondary - should be denied
        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestgroup,cn=users," + self.base_dn)
        m["member"] = MessageElement("cn=ldaptestuser,cn=users," + self.base_dn,
                                     FLAG_MOD_ADD, "member")
        try:
            ldb.modify(m)
            self.fail()
        except LdbError as e19:
            (num, _) = e19.args
            self.assertEqual(num, ERR_ENTRY_ALREADY_EXISTS)

        # Try to add invalid member to group 1 - should be denied
        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestgroup,cn=users," + self.base_dn)
        m["member"] = MessageElement(
            "cn=ldaptestuser3,cn=users," + self.base_dn,
            FLAG_MOD_ADD, "member")
        try:
            ldb.modify(m)
            self.fail()
        except LdbError as e20:
            (num, _) = e20.args
            self.assertEqual(num, ERR_NO_SUCH_OBJECT)

        # Make group 2 secondary
        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestgroup2,cn=users," + self.base_dn)
        m["member"] = MessageElement("cn=ldaptestuser,cn=users," + self.base_dn,
                                     FLAG_MOD_ADD, "member")
        ldb.modify(m)

        # Swap the groups
        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestuser,cn=users," + self.base_dn)
        m["primaryGroupID"] = MessageElement(str(group_rid_2),
                                             FLAG_MOD_REPLACE, "primaryGroupID")
        ldb.modify(m)

        # Swap the groups (does not really make sense but does the same)
        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestuser,cn=users," + self.base_dn)
        m["primaryGroupID"] = MessageElement(str(group_rid_1),
                                             FLAG_MOD_REPLACE, "primaryGroupID")
        m["primaryGroupID"] = MessageElement(str(group_rid_2),
                                             FLAG_MOD_REPLACE, "primaryGroupID")
        ldb.modify(m)

        # Old primary group should contain a "member" attribute for the user,
        # the new shouldn't contain anymore one
        res1 = ldb.search("cn=ldaptestgroup, cn=users," + self.base_dn,
                          scope=SCOPE_BASE, attrs=["member"])
        self.assertTrue(len(res1) == 1)
        self.assertTrue(len(res1[0]["member"]) == 1)
        self.assertEqual(str(res1[0]["member"][0]).lower(),
                          ("cn=ldaptestuser,cn=users," + self.base_dn).lower())

        res1 = ldb.search("cn=ldaptestgroup2, cn=users," + self.base_dn,
                          scope=SCOPE_BASE, attrs=["member"])
        self.assertTrue(len(res1) == 1)
        self.assertFalse("member" in res1[0])

        # Primary group member
        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestgroup2,cn=users," + self.base_dn)
        m["member"] = MessageElement("cn=ldaptestuser,cn=users," + self.base_dn,
                                     FLAG_MOD_DELETE, "member")
        try:
            ldb.modify(m)
            self.fail()
        except LdbError as e21:
            (num, _) = e21.args
            self.assertEqual(num, ERR_UNWILLING_TO_PERFORM)

        # Delete invalid group member
        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestgroup2,cn=users," + self.base_dn)
        m["member"] = MessageElement("cn=ldaptestuser1,cn=users," + self.base_dn,
                                     FLAG_MOD_DELETE, "member")
        try:
            ldb.modify(m)
            self.fail()
        except LdbError as e22:
            (num, _) = e22.args
            self.assertEqual(num, ERR_UNWILLING_TO_PERFORM)

        # Also this should be denied
        try:
            ldb.add({
                "dn": "cn=ldaptestuser2,cn=users," + self.base_dn,
                "objectclass": "user",
                "primaryGroupID": "0"})
            self.fail()
        except LdbError as e23:
            (num, _) = e23.args
            self.assertEqual(num, ERR_UNWILLING_TO_PERFORM)

        # Recreate user accounts

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

        ldb.add({
            "dn": "cn=ldaptestuser,cn=users," + self.base_dn,
            "objectclass": "user"})

        ldb.add({
            "dn": "cn=ldaptestuser2,cn=users," + self.base_dn,
            "objectclass": "user"})

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestgroup2,cn=users," + self.base_dn)
        m["member"] = MessageElement("cn=ldaptestuser,cn=users," + self.base_dn,
                                     FLAG_MOD_ADD, "member")
        ldb.modify(m)

        # Already added
        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestgroup2,cn=users," + self.base_dn)
        m["member"] = MessageElement("cn=ldaptestuser,cn=users," + self.base_dn,
                                     FLAG_MOD_ADD, "member")
        try:
            ldb.modify(m)
            self.fail()
        except LdbError as e24:
            (num, _) = e24.args
            self.assertEqual(num, ERR_ENTRY_ALREADY_EXISTS)

        # Already added, but as <SID=...>
        res1 = ldb.search("cn=ldaptestuser,cn=users," + self.base_dn,
                          scope=SCOPE_BASE, attrs=["objectSid"])
        self.assertTrue(len(res1) == 1)
        sid_bin = res1[0]["objectSid"][0]
        sid_str = ("<SID=" + get_string(ldb.schema_format_value("objectSid", sid_bin)) + ">").upper()

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestgroup2,cn=users," + self.base_dn)
        m["member"] = MessageElement(sid_str, FLAG_MOD_ADD, "member")
        try:
            ldb.modify(m)
            self.fail()
        except LdbError as e25:
            (num, _) = e25.args
            self.assertEqual(num, ERR_ENTRY_ALREADY_EXISTS)

        # Invalid member
        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestgroup2,cn=users," + self.base_dn)
        m["member"] = MessageElement("cn=ldaptestuser1,cn=users," + self.base_dn,
                                     FLAG_MOD_REPLACE, "member")
        try:
            ldb.modify(m)
            self.fail()
        except LdbError as e26:
            (num, _) = e26.args
            self.assertEqual(num, ERR_NO_SUCH_OBJECT)

        # Invalid member
        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestgroup2,cn=users," + self.base_dn)
        m["member"] = MessageElement(["cn=ldaptestuser,cn=users," + self.base_dn,
                                      "cn=ldaptestuser1,cn=users," + self.base_dn],
                                     FLAG_MOD_REPLACE, "member")
        try:
            ldb.modify(m)
            self.fail()
        except LdbError as e27:
            (num, _) = e27.args
            self.assertEqual(num, ERR_NO_SUCH_OBJECT)

        # Invalid member
        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestgroup2,cn=users," + self.base_dn)
        m["member"] = MessageElement("cn=ldaptestuser,cn=users," + self.base_dn,
                                     FLAG_MOD_REPLACE, "member")
        m["member"] = MessageElement("cn=ldaptestuser1,cn=users," + self.base_dn,
                                     FLAG_MOD_ADD, "member")
        try:
            ldb.modify(m)
            self.fail()
        except LdbError as e28:
            (num, _) = e28.args
            self.assertEqual(num, ERR_NO_SUCH_OBJECT)

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestgroup2,cn=users," + self.base_dn)
        m["member"] = MessageElement(["cn=ldaptestuser,cn=users," + self.base_dn,
                                      "cn=ldaptestuser2,cn=users," + self.base_dn],
                                     FLAG_MOD_REPLACE, "member")
        ldb.modify(m)

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

        # Make also a small test for accounts with special DNs ("," in this case)
        ldb.add({
            "dn": r"cn=ldaptest\,specialuser,cn=users," + self.base_dn,
            "objectclass": "user"})
        delete_force(self.ldb, r"cn=ldaptest\,specialuser,cn=users," + self.base_dn)

    def test_sam_attributes(self):
        """Test the behaviour of special attributes of SAM objects"""
        print("Testing the behaviour of special attributes of SAM objects\n")

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

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestgroup,cn=users," + self.base_dn)
        m["groupType"] = MessageElement(str(GTYPE_SECURITY_GLOBAL_GROUP), FLAG_MOD_ADD,
                                        "groupType")
        try:
            ldb.modify(m)
            self.fail()
        except LdbError as e29:
            (num, _) = e29.args
            self.assertEqual(num, ERR_ATTRIBUTE_OR_VALUE_EXISTS)

        # Delete protection tests

        for attr in ["nTSecurityDescriptor", "objectSid", "sAMAccountType",
                     "sAMAccountName", "groupType"]:

            m = Message()
            m.dn = Dn(ldb, "cn=ldaptestgroup,cn=users," + self.base_dn)
            m[attr] = MessageElement([], FLAG_MOD_REPLACE, attr)
            try:
                ldb.modify(m)
                self.fail()
            except LdbError as e:
                (num, _) = e.args
                self.assertEqual(num, ERR_UNWILLING_TO_PERFORM)

            m = Message()
            m.dn = Dn(ldb, "cn=ldaptestgroup,cn=users," + self.base_dn)
            m[attr] = MessageElement([], FLAG_MOD_DELETE, attr)
            try:
                ldb.modify(m)
                self.fail()
            except LdbError as e1:
                (num, _) = e1.args
                self.assertEqual(num, ERR_UNWILLING_TO_PERFORM)

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestuser,cn=users," + self.base_dn)
        m["primaryGroupID"] = MessageElement("513", FLAG_MOD_ADD,
                                             "primaryGroupID")
        try:
            ldb.modify(m)
            self.fail()
        except LdbError as e30:
            (num, _) = e30.args
            self.assertEqual(num, ERR_ATTRIBUTE_OR_VALUE_EXISTS)

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestuser,cn=users," + self.base_dn)
        m["userAccountControl"] = MessageElement(str(UF_NORMAL_ACCOUNT |
                                                     UF_PASSWD_NOTREQD),
                                                 FLAG_MOD_ADD,
                                                 "userAccountControl")
        try:
            ldb.modify(m)
            self.fail()
        except LdbError as e31:
            (num, _) = e31.args
            self.assertEqual(num, ERR_ATTRIBUTE_OR_VALUE_EXISTS)

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestuser,cn=users," + self.base_dn)
        m["objectSid"] = MessageElement("xxxxxxxxxxxxxxxx", FLAG_MOD_ADD,
                                        "objectSid")
        try:
            ldb.modify(m)
            self.fail()
        except LdbError as e32:
            (num, _) = e32.args
            self.assertEqual(num, ERR_UNWILLING_TO_PERFORM)

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestuser,cn=users," + self.base_dn)
        m["sAMAccountType"] = MessageElement("0", FLAG_MOD_ADD,
                                             "sAMAccountType")
        try:
            ldb.modify(m)
            self.fail()
        except LdbError as e33:
            (num, _) = e33.args
            self.assertEqual(num, ERR_UNWILLING_TO_PERFORM)

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestuser,cn=users," + self.base_dn)
        m["sAMAccountName"] = MessageElement("test", FLAG_MOD_ADD,
                                             "sAMAccountName")
        try:
            ldb.modify(m)
            self.fail()
        except LdbError as e34:
            (num, _) = e34.args
            self.assertEqual(num, ERR_ATTRIBUTE_OR_VALUE_EXISTS)

        # Delete protection tests

        for attr in ["nTSecurityDescriptor", "objectSid", "sAMAccountType",
                     "sAMAccountName", "primaryGroupID", "userAccountControl",
                     "accountExpires", "badPasswordTime", "badPwdCount",
                     "codePage", "countryCode", "lastLogoff", "lastLogon",
                     "logonCount", "pwdLastSet"]:

            m = Message()
            m.dn = Dn(ldb, "cn=ldaptestuser,cn=users," + self.base_dn)
            m[attr] = MessageElement([], FLAG_MOD_REPLACE, attr)
            try:
                ldb.modify(m)
                self.fail()
            except LdbError as e2:
                (num, _) = e2.args
                self.assertEqual(num, ERR_UNWILLING_TO_PERFORM)

            m = Message()
            m.dn = Dn(ldb, "cn=ldaptestuser,cn=users," + self.base_dn)
            m[attr] = MessageElement([], FLAG_MOD_DELETE, attr)
            try:
                ldb.modify(m)
                self.fail()
            except LdbError as e3:
                (num, _) = e3.args
                self.assertEqual(num, ERR_UNWILLING_TO_PERFORM)

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

    def test_primary_group_token_constructed(self):
        """Test the primary group token behaviour (hidden-generated-readonly attribute on groups) and some other constructed attributes"""
        print("Testing primary group token behaviour and other constructed attributes\n")

        try:
            ldb.add({
                "dn": "cn=ldaptestgroup,cn=users," + self.base_dn,
                "objectclass": "group",
                "primaryGroupToken": "100"})
            self.fail()
        except LdbError as e35:
            (num, _) = e35.args
            self.assertEqual(num, ERR_UNDEFINED_ATTRIBUTE_TYPE)
        delete_force(self.ldb, "cn=ldaptestgroup,cn=users," + self.base_dn)

        ldb.add({
            "dn": "cn=ldaptestuser,cn=users," + self.base_dn,
            "objectclass": "user"})

        ldb.add({
            "dn": "cn=ldaptestgroup,cn=users," + self.base_dn,
            "objectclass": "group"})

        # Testing for one invalid, and one valid operational attribute, but also the things they are built from
        res1 = ldb.search(self.base_dn,
                          scope=SCOPE_BASE, attrs=["primaryGroupToken", "canonicalName", "objectClass", "objectSid"])
        self.assertTrue(len(res1) == 1)
        self.assertFalse("primaryGroupToken" in res1[0])
        self.assertTrue("canonicalName" in res1[0])
        self.assertTrue("objectClass" in res1[0])
        self.assertTrue("objectSid" in res1[0])

        res1 = ldb.search(self.base_dn,
                          scope=SCOPE_BASE, attrs=["primaryGroupToken", "canonicalName"])
        self.assertTrue(len(res1) == 1)
        self.assertFalse("primaryGroupToken" in res1[0])
        self.assertFalse("objectSid" in res1[0])
        self.assertFalse("objectClass" in res1[0])
        self.assertTrue("canonicalName" in res1[0])

        res1 = ldb.search("cn=users," + self.base_dn,
                          scope=SCOPE_BASE, attrs=["primaryGroupToken"])
        self.assertTrue(len(res1) == 1)
        self.assertFalse("primaryGroupToken" in res1[0])

        res1 = ldb.search("cn=ldaptestuser, cn=users," + self.base_dn,
                          scope=SCOPE_BASE, attrs=["primaryGroupToken"])
        self.assertTrue(len(res1) == 1)
        self.assertFalse("primaryGroupToken" in res1[0])

        res1 = ldb.search("cn=ldaptestgroup,cn=users," + self.base_dn,
                          scope=SCOPE_BASE)
        self.assertTrue(len(res1) == 1)
        self.assertFalse("primaryGroupToken" in res1[0])

        res1 = ldb.search("cn=ldaptestgroup,cn=users," + self.base_dn,
                          scope=SCOPE_BASE, attrs=["primaryGroupToken", "objectSID"])
        self.assertTrue(len(res1) == 1)
        primary_group_token = int(res1[0]["primaryGroupToken"][0])

        obj_sid = get_string(ldb.schema_format_value("objectSID", res1[0]["objectSID"][0]))
        rid = security.dom_sid(obj_sid).split()[1]
        self.assertEqual(primary_group_token, rid)

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestgroup,cn=users," + self.base_dn)
        m["primaryGroupToken"] = "100"
        try:
            ldb.modify(m)
            self.fail()
        except LdbError as e36:
            (num, _) = e36.args
            self.assertEqual(num, ERR_CONSTRAINT_VIOLATION)

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

    def test_tokenGroups(self):
        """Test the tokenGroups behaviour (hidden-generated-readonly attribute on SAM objects)"""
        print("Testing tokenGroups behaviour\n")

        # The domain object shouldn't contain any "tokenGroups" entry
        res = ldb.search(self.base_dn, scope=SCOPE_BASE, attrs=["tokenGroups"])
        self.assertTrue(len(res) == 1)
        self.assertFalse("tokenGroups" in res[0])

        # The domain administrator should contain "tokenGroups" entries
        # (the exact number depends on the domain/forest function level and the
        # DC software versions)
        res = ldb.search("cn=Administrator,cn=Users," + self.base_dn,
                         scope=SCOPE_BASE, attrs=["tokenGroups"])
        self.assertTrue(len(res) == 1)
        self.assertTrue("tokenGroups" in res[0])

        ldb.add({
            "dn": "cn=ldaptestuser,cn=users," + self.base_dn,
            "objectclass": "user"})

        # This testuser should contain at least two "tokenGroups" entries
        # (exactly two on an unmodified "Domain Users" and "Users" group)
        res = ldb.search("cn=ldaptestuser,cn=users," + self.base_dn,
                         scope=SCOPE_BASE, attrs=["tokenGroups"])
        self.assertTrue(len(res) == 1)
        self.assertTrue(len(res[0]["tokenGroups"]) >= 2)

        # one entry which we need to find should point to domains "Domain Users"
        # group and another entry should point to the builtin "Users"group
        domain_users_group_found = False
        users_group_found = False
        for sid in res[0]["tokenGroups"]:
            obj_sid = get_string(ldb.schema_format_value("objectSID", sid))
            rid = security.dom_sid(obj_sid).split()[1]
            if rid == 513:
                domain_users_group_found = True
            if rid == 545:
                users_group_found = True

        self.assertTrue(domain_users_group_found)
        self.assertTrue(users_group_found)

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

    def test_groupType(self):
        """Test the groupType behaviour"""
        print("Testing groupType behaviour\n")

        # You can never create or change to a
        # "GTYPE_SECURITY_BUILTIN_LOCAL_GROUP"

        # Add operation

        # Invalid attribute
        try:
            ldb.add({
                "dn": "cn=ldaptestgroup,cn=users," + self.base_dn,
                "objectclass": "group",
                "groupType": "0"})
            self.fail()
        except LdbError as e37:
            (num, _) = e37.args
            self.assertEqual(num, ERR_UNWILLING_TO_PERFORM)
        delete_force(self.ldb, "cn=ldaptestgroup,cn=users," + self.base_dn)

        try:
            ldb.add({
                "dn": "cn=ldaptestgroup,cn=users," + self.base_dn,
                "objectclass": "group",
                "groupType": str(GTYPE_SECURITY_BUILTIN_LOCAL_GROUP)})
            self.fail()
        except LdbError as e38:
            (num, _) = e38.args
            self.assertEqual(num, ERR_UNWILLING_TO_PERFORM)
        delete_force(self.ldb, "cn=ldaptestgroup,cn=users," + self.base_dn)

        ldb.add({
            "dn": "cn=ldaptestgroup,cn=users," + self.base_dn,
            "objectclass": "group",
            "groupType": str(GTYPE_SECURITY_GLOBAL_GROUP)})

        res1 = ldb.search("cn=ldaptestgroup,cn=users," + self.base_dn,
                          scope=SCOPE_BASE, attrs=["sAMAccountType"])
        self.assertTrue(len(res1) == 1)
        self.assertEqual(int(res1[0]["sAMAccountType"][0]),
                          ATYPE_SECURITY_GLOBAL_GROUP)
        delete_force(self.ldb, "cn=ldaptestgroup,cn=users," + self.base_dn)

        ldb.add({
            "dn": "cn=ldaptestgroup,cn=users," + self.base_dn,
            "objectclass": "group",
            "groupType": str(GTYPE_SECURITY_UNIVERSAL_GROUP)})

        res1 = ldb.search("cn=ldaptestgroup,cn=users," + self.base_dn,
                          scope=SCOPE_BASE, attrs=["sAMAccountType"])
        self.assertTrue(len(res1) == 1)
        self.assertEqual(int(res1[0]["sAMAccountType"][0]),
                          ATYPE_SECURITY_UNIVERSAL_GROUP)
        delete_force(self.ldb, "cn=ldaptestgroup,cn=users," + self.base_dn)

        ldb.add({
            "dn": "cn=ldaptestgroup,cn=users," + self.base_dn,
            "objectclass": "group",
            "groupType": str(GTYPE_SECURITY_DOMAIN_LOCAL_GROUP)})

        res1 = ldb.search("cn=ldaptestgroup,cn=users," + self.base_dn,
                          scope=SCOPE_BASE, attrs=["sAMAccountType"])
        self.assertTrue(len(res1) == 1)
        self.assertEqual(int(res1[0]["sAMAccountType"][0]),
                          ATYPE_SECURITY_LOCAL_GROUP)
        delete_force(self.ldb, "cn=ldaptestgroup,cn=users," + self.base_dn)

        ldb.add({
            "dn": "cn=ldaptestgroup,cn=users," + self.base_dn,
            "objectclass": "group",
            "groupType": str(GTYPE_DISTRIBUTION_GLOBAL_GROUP)})

        res1 = ldb.search("cn=ldaptestgroup,cn=users," + self.base_dn,
                          scope=SCOPE_BASE, attrs=["sAMAccountType"])
        self.assertTrue(len(res1) == 1)
        self.assertEqual(int(res1[0]["sAMAccountType"][0]),
                          ATYPE_DISTRIBUTION_GLOBAL_GROUP)
        delete_force(self.ldb, "cn=ldaptestgroup,cn=users," + self.base_dn)

        ldb.add({
            "dn": "cn=ldaptestgroup,cn=users," + self.base_dn,
            "objectclass": "group",
            "groupType": str(GTYPE_DISTRIBUTION_UNIVERSAL_GROUP)})

        res1 = ldb.search("cn=ldaptestgroup,cn=users," + self.base_dn,
                          scope=SCOPE_BASE, attrs=["sAMAccountType"])
        self.assertTrue(len(res1) == 1)
        self.assertEqual(int(res1[0]["sAMAccountType"][0]),
                          ATYPE_DISTRIBUTION_UNIVERSAL_GROUP)
        delete_force(self.ldb, "cn=ldaptestgroup,cn=users," + self.base_dn)

        ldb.add({
            "dn": "cn=ldaptestgroup,cn=users," + self.base_dn,
            "objectclass": "group",
            "groupType": str(GTYPE_DISTRIBUTION_DOMAIN_LOCAL_GROUP)})

        res1 = ldb.search("cn=ldaptestgroup,cn=users," + self.base_dn,
                          scope=SCOPE_BASE, attrs=["sAMAccountType"])
        self.assertTrue(len(res1) == 1)
        self.assertEqual(int(res1[0]["sAMAccountType"][0]),
                          ATYPE_DISTRIBUTION_LOCAL_GROUP)
        delete_force(self.ldb, "cn=ldaptestgroup,cn=users," + self.base_dn)

        # Modify operation

        ldb.add({
            "dn": "cn=ldaptestgroup,cn=users," + self.base_dn,
            "objectclass": "group"})

        # We can change in this direction: global <-> universal <-> local
        # On each step also the group type itself (security/distribution) is
        # variable.

        # After creation we should have a "security global group"
        res1 = ldb.search("cn=ldaptestgroup,cn=users," + self.base_dn,
                          scope=SCOPE_BASE, attrs=["sAMAccountType"])
        self.assertTrue(len(res1) == 1)
        self.assertEqual(int(res1[0]["sAMAccountType"][0]),
                          ATYPE_SECURITY_GLOBAL_GROUP)

        # Invalid attribute
        try:
            m = Message()
            m.dn = Dn(ldb, "cn=ldaptestgroup,cn=users," + self.base_dn)
            m["groupType"] = MessageElement("0",
                                            FLAG_MOD_REPLACE, "groupType")
            ldb.modify(m)
            self.fail()
        except LdbError as e39:
            (num, _) = e39.args
            self.assertEqual(num, ERR_UNWILLING_TO_PERFORM)

        # Security groups

        # Default is "global group"

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestgroup,cn=users," + self.base_dn)
        m["groupType"] = MessageElement(
            str(GTYPE_SECURITY_GLOBAL_GROUP),
            FLAG_MOD_REPLACE, "groupType")
        ldb.modify(m)

        res1 = ldb.search("cn=ldaptestgroup,cn=users," + self.base_dn,
                          scope=SCOPE_BASE, attrs=["sAMAccountType"])
        self.assertTrue(len(res1) == 1)
        self.assertEqual(int(res1[0]["sAMAccountType"][0]),
                          ATYPE_SECURITY_GLOBAL_GROUP)

        # Change to "local" (shouldn't work)

        try:
            m = Message()
            m.dn = Dn(ldb, "cn=ldaptestgroup,cn=users," + self.base_dn)
            m["groupType"] = MessageElement(
                str(GTYPE_SECURITY_DOMAIN_LOCAL_GROUP),
                FLAG_MOD_REPLACE, "groupType")
            ldb.modify(m)
            self.fail()
        except LdbError as e40:
            (num, _) = e40.args
            self.assertEqual(num, ERR_UNWILLING_TO_PERFORM)

        # Change to "universal"

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestgroup,cn=users," + self.base_dn)
        m["groupType"] = MessageElement(
            str(GTYPE_SECURITY_UNIVERSAL_GROUP),
            FLAG_MOD_REPLACE, "groupType")
        ldb.modify(m)

        res1 = ldb.search("cn=ldaptestgroup,cn=users," + self.base_dn,
                          scope=SCOPE_BASE, attrs=["sAMAccountType"])
        self.assertTrue(len(res1) == 1)
        self.assertEqual(int(res1[0]["sAMAccountType"][0]),
                          ATYPE_SECURITY_UNIVERSAL_GROUP)

        # Change back to "global"

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestgroup,cn=users," + self.base_dn)
        m["groupType"] = MessageElement(
            str(GTYPE_SECURITY_GLOBAL_GROUP),
            FLAG_MOD_REPLACE, "groupType")
        ldb.modify(m)

        res1 = ldb.search("cn=ldaptestgroup,cn=users," + self.base_dn,
                          scope=SCOPE_BASE, attrs=["sAMAccountType"])
        self.assertTrue(len(res1) == 1)
        self.assertEqual(int(res1[0]["sAMAccountType"][0]),
                          ATYPE_SECURITY_GLOBAL_GROUP)

        # Change back to "universal"

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestgroup,cn=users," + self.base_dn)
        m["groupType"] = MessageElement(
            str(GTYPE_SECURITY_UNIVERSAL_GROUP),
            FLAG_MOD_REPLACE, "groupType")
        ldb.modify(m)

        res1 = ldb.search("cn=ldaptestgroup,cn=users," + self.base_dn,
                          scope=SCOPE_BASE, attrs=["sAMAccountType"])
        self.assertTrue(len(res1) == 1)
        self.assertEqual(int(res1[0]["sAMAccountType"][0]),
                          ATYPE_SECURITY_UNIVERSAL_GROUP)

        # Change to "local"

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestgroup,cn=users," + self.base_dn)
        m["groupType"] = MessageElement(
            str(GTYPE_SECURITY_DOMAIN_LOCAL_GROUP),
            FLAG_MOD_REPLACE, "groupType")
        ldb.modify(m)

        res1 = ldb.search("cn=ldaptestgroup,cn=users," + self.base_dn,
                          scope=SCOPE_BASE, attrs=["sAMAccountType"])
        self.assertTrue(len(res1) == 1)
        self.assertEqual(int(res1[0]["sAMAccountType"][0]),
                          ATYPE_SECURITY_LOCAL_GROUP)

        # Change to "global" (shouldn't work)

        try:
            m = Message()
            m.dn = Dn(ldb, "cn=ldaptestgroup,cn=users," + self.base_dn)
            m["groupType"] = MessageElement(
                str(GTYPE_SECURITY_GLOBAL_GROUP),
                FLAG_MOD_REPLACE, "groupType")
            ldb.modify(m)
            self.fail()
        except LdbError as e41:
            (num, _) = e41.args
            self.assertEqual(num, ERR_UNWILLING_TO_PERFORM)

        # Change to "builtin local" (shouldn't work)

        try:
            m = Message()
            m.dn = Dn(ldb, "cn=ldaptestgroup,cn=users," + self.base_dn)
            m["groupType"] = MessageElement(
                str(GTYPE_SECURITY_BUILTIN_LOCAL_GROUP),
                FLAG_MOD_REPLACE, "groupType")
            ldb.modify(m)
            self.fail()
        except LdbError as e42:
            (num, _) = e42.args
            self.assertEqual(num, ERR_UNWILLING_TO_PERFORM)

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestgroup,cn=users," + self.base_dn)

        # Change back to "universal"

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestgroup,cn=users," + self.base_dn)
        m["groupType"] = MessageElement(
            str(GTYPE_SECURITY_UNIVERSAL_GROUP),
            FLAG_MOD_REPLACE, "groupType")
        ldb.modify(m)

        res1 = ldb.search("cn=ldaptestgroup,cn=users," + self.base_dn,
                          scope=SCOPE_BASE, attrs=["sAMAccountType"])
        self.assertTrue(len(res1) == 1)
        self.assertEqual(int(res1[0]["sAMAccountType"][0]),
                          ATYPE_SECURITY_UNIVERSAL_GROUP)

        # Change to "builtin local" (shouldn't work)

        try:
            m = Message()
            m.dn = Dn(ldb, "cn=ldaptestgroup,cn=users," + self.base_dn)
            m["groupType"] = MessageElement(
                str(GTYPE_SECURITY_BUILTIN_LOCAL_GROUP),
                FLAG_MOD_REPLACE, "groupType")
            ldb.modify(m)
            self.fail()
        except LdbError as e43:
            (num, _) = e43.args
            self.assertEqual(num, ERR_UNWILLING_TO_PERFORM)

        # Change back to "global"

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestgroup,cn=users," + self.base_dn)
        m["groupType"] = MessageElement(
            str(GTYPE_SECURITY_GLOBAL_GROUP),
            FLAG_MOD_REPLACE, "groupType")
        ldb.modify(m)

        res1 = ldb.search("cn=ldaptestgroup,cn=users," + self.base_dn,
                          scope=SCOPE_BASE, attrs=["sAMAccountType"])
        self.assertTrue(len(res1) == 1)
        self.assertEqual(int(res1[0]["sAMAccountType"][0]),
                          ATYPE_SECURITY_GLOBAL_GROUP)

        # Change to "builtin local" (shouldn't work)

        try:
            m = Message()
            m.dn = Dn(ldb, "cn=ldaptestgroup,cn=users," + self.base_dn)
            m["groupType"] = MessageElement(
                str(GTYPE_SECURITY_BUILTIN_LOCAL_GROUP),
                FLAG_MOD_REPLACE, "groupType")
            ldb.modify(m)
            self.fail()
        except LdbError as e44:
            (num, _) = e44.args
            self.assertEqual(num, ERR_UNWILLING_TO_PERFORM)

        # Distribution groups

        # Default is "global group"

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestgroup,cn=users," + self.base_dn)
        m["groupType"] = MessageElement(
            str(GTYPE_DISTRIBUTION_GLOBAL_GROUP),
            FLAG_MOD_REPLACE, "groupType")
        ldb.modify(m)

        res1 = ldb.search("cn=ldaptestgroup,cn=users," + self.base_dn,
                          scope=SCOPE_BASE, attrs=["sAMAccountType"])
        self.assertTrue(len(res1) == 1)
        self.assertEqual(int(res1[0]["sAMAccountType"][0]),
                          ATYPE_DISTRIBUTION_GLOBAL_GROUP)

        # Change to local (shouldn't work)

        try:
            m = Message()
            m.dn = Dn(ldb, "cn=ldaptestgroup,cn=users," + self.base_dn)
            m["groupType"] = MessageElement(
                str(GTYPE_DISTRIBUTION_DOMAIN_LOCAL_GROUP),
                FLAG_MOD_REPLACE, "groupType")
            ldb.modify(m)
            self.fail()
        except LdbError as e45:
            (num, _) = e45.args
            self.assertEqual(num, ERR_UNWILLING_TO_PERFORM)

        # Change to "universal"

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestgroup,cn=users," + self.base_dn)
        m["groupType"] = MessageElement(
            str(GTYPE_DISTRIBUTION_UNIVERSAL_GROUP),
            FLAG_MOD_REPLACE, "groupType")
        ldb.modify(m)

        res1 = ldb.search("cn=ldaptestgroup,cn=users," + self.base_dn,
                          scope=SCOPE_BASE, attrs=["sAMAccountType"])
        self.assertTrue(len(res1) == 1)
        self.assertEqual(int(res1[0]["sAMAccountType"][0]),
                          ATYPE_DISTRIBUTION_UNIVERSAL_GROUP)

        # Change back to "global"

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestgroup,cn=users," + self.base_dn)
        m["groupType"] = MessageElement(
            str(GTYPE_DISTRIBUTION_GLOBAL_GROUP),
            FLAG_MOD_REPLACE, "groupType")
        ldb.modify(m)

        res1 = ldb.search("cn=ldaptestgroup,cn=users," + self.base_dn,
                          scope=SCOPE_BASE, attrs=["sAMAccountType"])
        self.assertTrue(len(res1) == 1)
        self.assertEqual(int(res1[0]["sAMAccountType"][0]),
                          ATYPE_DISTRIBUTION_GLOBAL_GROUP)

        # Change back to "universal"

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestgroup,cn=users," + self.base_dn)
        m["groupType"] = MessageElement(
            str(GTYPE_DISTRIBUTION_UNIVERSAL_GROUP),
            FLAG_MOD_REPLACE, "groupType")
        ldb.modify(m)

        res1 = ldb.search("cn=ldaptestgroup,cn=users," + self.base_dn,
                          scope=SCOPE_BASE, attrs=["sAMAccountType"])
        self.assertTrue(len(res1) == 1)
        self.assertEqual(int(res1[0]["sAMAccountType"][0]),
                          ATYPE_DISTRIBUTION_UNIVERSAL_GROUP)

        # Change to "local"

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestgroup,cn=users," + self.base_dn)
        m["groupType"] = MessageElement(
            str(GTYPE_DISTRIBUTION_DOMAIN_LOCAL_GROUP),
            FLAG_MOD_REPLACE, "groupType")
        ldb.modify(m)

        res1 = ldb.search("cn=ldaptestgroup,cn=users," + self.base_dn,
                          scope=SCOPE_BASE, attrs=["sAMAccountType"])
        self.assertTrue(len(res1) == 1)
        self.assertEqual(int(res1[0]["sAMAccountType"][0]),
                          ATYPE_DISTRIBUTION_LOCAL_GROUP)

        # Change to "global" (shouldn't work)

        try:
            m = Message()
            m.dn = Dn(ldb, "cn=ldaptestgroup,cn=users," + self.base_dn)
            m["groupType"] = MessageElement(
                str(GTYPE_DISTRIBUTION_GLOBAL_GROUP),
                FLAG_MOD_REPLACE, "groupType")
            ldb.modify(m)
            self.fail()
        except LdbError as e46:
            (num, _) = e46.args
            self.assertEqual(num, ERR_UNWILLING_TO_PERFORM)

        # Change back to "universal"

        # Try to add invalid member to group 1 - should be denied
        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestgroup,cn=users," + self.base_dn)
        m["member"] = MessageElement(
            "cn=ldaptestuser3,cn=users," + self.base_dn,
            FLAG_MOD_ADD, "member")
        try:
            ldb.modify(m)
            self.fail()
        except LdbError as e47:
            (num, _) = e47.args
            self.assertEqual(num, ERR_NO_SUCH_OBJECT)

        # Make group 2 secondary
        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestgroup,cn=users," + self.base_dn)
        m["groupType"] = MessageElement(
            str(GTYPE_DISTRIBUTION_UNIVERSAL_GROUP),
            FLAG_MOD_REPLACE, "groupType")
        ldb.modify(m)

        res1 = ldb.search("cn=ldaptestgroup,cn=users," + self.base_dn,
                          scope=SCOPE_BASE, attrs=["sAMAccountType"])
        self.assertTrue(len(res1) == 1)
        self.assertEqual(int(res1[0]["sAMAccountType"][0]),
                          ATYPE_DISTRIBUTION_UNIVERSAL_GROUP)

        # Change back to "global"

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestgroup,cn=users," + self.base_dn)
        m["groupType"] = MessageElement(
            str(GTYPE_DISTRIBUTION_GLOBAL_GROUP),
            FLAG_MOD_REPLACE, "groupType")
        ldb.modify(m)

        res1 = ldb.search("cn=ldaptestgroup,cn=users," + self.base_dn,
                          scope=SCOPE_BASE, attrs=["sAMAccountType"])
        self.assertTrue(len(res1) == 1)
        self.assertEqual(int(res1[0]["sAMAccountType"][0]),
                          ATYPE_DISTRIBUTION_GLOBAL_GROUP)

        # Both group types: this performs only random checks - all possibilities
        # would require too much code.

        # Default is "global group"

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestgroup,cn=users," + self.base_dn)
        m["groupType"] = MessageElement(
            str(GTYPE_SECURITY_GLOBAL_GROUP),
            FLAG_MOD_REPLACE, "groupType")
        ldb.modify(m)

        res1 = ldb.search("cn=ldaptestgroup,cn=users," + self.base_dn,
                          scope=SCOPE_BASE, attrs=["sAMAccountType"])
        self.assertTrue(len(res1) == 1)
        self.assertEqual(int(res1[0]["sAMAccountType"][0]),
                          ATYPE_SECURITY_GLOBAL_GROUP)

        # Change to "local" (shouldn't work)

        try:
            m = Message()
            m.dn = Dn(ldb, "cn=ldaptestgroup,cn=users," + self.base_dn)
            m["groupType"] = MessageElement(
                str(GTYPE_DISTRIBUTION_DOMAIN_LOCAL_GROUP),
                FLAG_MOD_REPLACE, "groupType")
            ldb.modify(m)
            self.fail()
        except LdbError as e48:
            (num, _) = e48.args
            self.assertEqual(num, ERR_UNWILLING_TO_PERFORM)

        # Change to "universal"

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestgroup,cn=users," + self.base_dn)
        m["groupType"] = MessageElement(
            str(GTYPE_DISTRIBUTION_UNIVERSAL_GROUP),
            FLAG_MOD_REPLACE, "groupType")
        ldb.modify(m)

        res1 = ldb.search("cn=ldaptestgroup,cn=users," + self.base_dn,
                          scope=SCOPE_BASE, attrs=["sAMAccountType"])
        self.assertTrue(len(res1) == 1)
        self.assertEqual(int(res1[0]["sAMAccountType"][0]),
                          ATYPE_DISTRIBUTION_UNIVERSAL_GROUP)

        # Change back to "global"

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestgroup,cn=users," + self.base_dn)
        m["groupType"] = MessageElement(
            str(GTYPE_SECURITY_GLOBAL_GROUP),
            FLAG_MOD_REPLACE, "groupType")
        ldb.modify(m)

        res1 = ldb.search("cn=ldaptestgroup,cn=users," + self.base_dn,
                          scope=SCOPE_BASE, attrs=["sAMAccountType"])
        self.assertTrue(len(res1) == 1)
        self.assertEqual(int(res1[0]["sAMAccountType"][0]),
                          ATYPE_SECURITY_GLOBAL_GROUP)

        # Change back to "universal"

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestgroup,cn=users," + self.base_dn)
        m["groupType"] = MessageElement(
            str(GTYPE_SECURITY_UNIVERSAL_GROUP),
            FLAG_MOD_REPLACE, "groupType")
        ldb.modify(m)

        res1 = ldb.search("cn=ldaptestgroup,cn=users," + self.base_dn,
                          scope=SCOPE_BASE, attrs=["sAMAccountType"])
        self.assertTrue(len(res1) == 1)
        self.assertEqual(int(res1[0]["sAMAccountType"][0]),
                          ATYPE_SECURITY_UNIVERSAL_GROUP)

        # Change to "local"

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestgroup,cn=users," + self.base_dn)
        m["groupType"] = MessageElement(
            str(GTYPE_DISTRIBUTION_DOMAIN_LOCAL_GROUP),
            FLAG_MOD_REPLACE, "groupType")
        ldb.modify(m)

        res1 = ldb.search("cn=ldaptestgroup,cn=users," + self.base_dn,
                          scope=SCOPE_BASE, attrs=["sAMAccountType"])
        self.assertTrue(len(res1) == 1)
        self.assertEqual(int(res1[0]["sAMAccountType"][0]),
                          ATYPE_DISTRIBUTION_LOCAL_GROUP)

        # Change to "global" (shouldn't work)

        try:
            m = Message()
            m.dn = Dn(ldb, "cn=ldaptestgroup,cn=users," + self.base_dn)
            m["groupType"] = MessageElement(
                str(GTYPE_DISTRIBUTION_GLOBAL_GROUP),
                FLAG_MOD_REPLACE, "groupType")
            ldb.modify(m)
            self.fail()
        except LdbError as e49:
            (num, _) = e49.args
            self.assertEqual(num, ERR_UNWILLING_TO_PERFORM)

        # Change back to "universal"

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestgroup,cn=users," + self.base_dn)
        m["groupType"] = MessageElement(
            str(GTYPE_SECURITY_UNIVERSAL_GROUP),
            FLAG_MOD_REPLACE, "groupType")
        ldb.modify(m)

        res1 = ldb.search("cn=ldaptestgroup,cn=users," + self.base_dn,
                          scope=SCOPE_BASE, attrs=["sAMAccountType"])
        self.assertTrue(len(res1) == 1)
        self.assertEqual(int(res1[0]["sAMAccountType"][0]),
                          ATYPE_SECURITY_UNIVERSAL_GROUP)

        # Change back to "global"

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestgroup,cn=users," + self.base_dn)
        m["groupType"] = MessageElement(
            str(GTYPE_SECURITY_GLOBAL_GROUP),
            FLAG_MOD_REPLACE, "groupType")
        ldb.modify(m)

        res1 = ldb.search("cn=ldaptestgroup,cn=users," + self.base_dn,
                          scope=SCOPE_BASE, attrs=["sAMAccountType"])
        self.assertTrue(len(res1) == 1)
        self.assertEqual(int(res1[0]["sAMAccountType"][0]),
                          ATYPE_SECURITY_GLOBAL_GROUP)

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

    def test_pwdLastSet(self):
        """Test the pwdLastSet behaviour"""
        print("Testing pwdLastSet behaviour\n")

        ldb.add({
            "dn": "cn=ldaptestuser,cn=users," + self.base_dn,
            "objectclass": "user",
            "pwdLastSet": "0"})

        res1 = ldb.search("cn=ldaptestuser,cn=users," + self.base_dn,
                          scope=SCOPE_BASE,
                          attrs=["sAMAccountType", "userAccountControl", "pwdLastSet"])
        self.assertTrue(len(res1) == 1)
        self.assertEqual(int(res1[0]["sAMAccountType"][0]),
                         ATYPE_NORMAL_ACCOUNT)
        self.assertEqual(int(res1[0]["userAccountControl"][0]),
                         UF_NORMAL_ACCOUNT | UF_ACCOUNTDISABLE | UF_PASSWD_NOTREQD)
        self.assertEqual(int(res1[0]["pwdLastSet"][0]), 0)
        delete_force(self.ldb, "cn=ldaptestuser,cn=users," + self.base_dn)

        ldb.add({
            "dn": "cn=ldaptestuser,cn=users," + self.base_dn,
            "objectclass": "user",
            "pwdLastSet": "-1"})

        res1 = ldb.search("cn=ldaptestuser,cn=users," + self.base_dn,
                          scope=SCOPE_BASE,
                          attrs=["sAMAccountType", "userAccountControl", "pwdLastSet"])
        self.assertTrue(len(res1) == 1)
        self.assertEqual(int(res1[0]["sAMAccountType"][0]),
                         ATYPE_NORMAL_ACCOUNT)
        self.assertEqual(int(res1[0]["userAccountControl"][0]),
                         UF_NORMAL_ACCOUNT | UF_ACCOUNTDISABLE | UF_PASSWD_NOTREQD)
        self.assertNotEqual(int(res1[0]["pwdLastSet"][0]), 0)
        lastset = int(res1[0]["pwdLastSet"][0])
        delete_force(self.ldb, "cn=ldaptestuser,cn=users," + self.base_dn)

        try:
            ldb.add({
                "dn": "cn=ldaptestuser,cn=users," + self.base_dn,
                "objectclass": "user",
                "pwdLastSet": str(1)})
            self.fail()
        except LdbError as e50:
            (num, msg) = e50.args
            self.assertEqual(num, ERR_OTHER)
            self.assertTrue('00000057' in msg)

        try:
            ldb.add({
                "dn": "cn=ldaptestuser,cn=users," + self.base_dn,
                "objectclass": "user",
                "pwdLastSet": str(lastset)})
            self.fail()
        except LdbError as e51:
            (num, msg) = e51.args
            self.assertEqual(num, ERR_OTHER)
            self.assertTrue('00000057' in msg)

        ldb.add({
            "dn": "cn=ldaptestuser,cn=users," + self.base_dn,
            "objectclass": "user"})

        res1 = ldb.search("cn=ldaptestuser,cn=users," + self.base_dn,
                          scope=SCOPE_BASE,
                          attrs=["sAMAccountType", "userAccountControl", "pwdLastSet"])
        self.assertTrue(len(res1) == 1)
        self.assertEqual(int(res1[0]["sAMAccountType"][0]),
                         ATYPE_NORMAL_ACCOUNT)
        self.assertEqual(int(res1[0]["userAccountControl"][0]),
                         UF_NORMAL_ACCOUNT | UF_ACCOUNTDISABLE | UF_PASSWD_NOTREQD)
        self.assertEqual(int(res1[0]["pwdLastSet"][0]), 0)

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestuser,cn=users," + self.base_dn)
        m["pls1"] = MessageElement(str(0),
                                   FLAG_MOD_REPLACE,
                                   "pwdLastSet")
        ldb.modify(m)

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestuser,cn=users," + self.base_dn)
        m["pls1"] = MessageElement(str(0),
                                   FLAG_MOD_DELETE,
                                   "pwdLastSet")
        m["pls2"] = MessageElement(str(0),
                                   FLAG_MOD_ADD,
                                   "pwdLastSet")
        ldb.modify(m)

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestuser,cn=users," + self.base_dn)
        m["pls1"] = MessageElement(str(-1),
                                   FLAG_MOD_REPLACE,
                                   "pwdLastSet")
        ldb.modify(m)
        res1 = ldb.search("cn=ldaptestuser,cn=users," + self.base_dn,
                          scope=SCOPE_BASE,
                          attrs=["sAMAccountType", "userAccountControl", "pwdLastSet"])
        self.assertTrue(len(res1) == 1)
        self.assertEqual(int(res1[0]["sAMAccountType"][0]),
                         ATYPE_NORMAL_ACCOUNT)
        self.assertEqual(int(res1[0]["userAccountControl"][0]),
                         UF_NORMAL_ACCOUNT | UF_ACCOUNTDISABLE | UF_PASSWD_NOTREQD)
        self.assertGreater(int(res1[0]["pwdLastSet"][0]), lastset)
        lastset = int(res1[0]["pwdLastSet"][0])

        try:
            m = Message()
            m.dn = Dn(ldb, "cn=ldaptestuser,cn=users," + self.base_dn)
            m["pls1"] = MessageElement(str(0),
                                       FLAG_MOD_DELETE,
                                       "pwdLastSet")
            m["pls2"] = MessageElement(str(0),
                                       FLAG_MOD_ADD,
                                       "pwdLastSet")
            ldb.modify(m)
            self.fail()
        except LdbError as e52:
            (num, msg) = e52.args
            self.assertEqual(num, ERR_NO_SUCH_ATTRIBUTE)
            self.assertTrue('00002085' in msg)

        try:
            m = Message()
            m.dn = Dn(ldb, "cn=ldaptestuser,cn=users," + self.base_dn)
            m["pls1"] = MessageElement(str(-1),
                                       FLAG_MOD_DELETE,
                                       "pwdLastSet")
            m["pls2"] = MessageElement(str(0),
                                       FLAG_MOD_ADD,
                                       "pwdLastSet")
            ldb.modify(m)
            self.fail()
        except LdbError as e53:
            (num, msg) = e53.args
            self.assertEqual(num, ERR_NO_SUCH_ATTRIBUTE)
            self.assertTrue('00002085' in msg)

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestuser,cn=users," + self.base_dn)
        m["pls1"] = MessageElement(str(lastset),
                                   FLAG_MOD_DELETE,
                                   "pwdLastSet")
        m["pls2"] = MessageElement(str(-1),
                                   FLAG_MOD_ADD,
                                   "pwdLastSet")
        time.sleep(0.2)
        ldb.modify(m)
        res1 = ldb.search("cn=ldaptestuser,cn=users," + self.base_dn,
                          scope=SCOPE_BASE,
                          attrs=["sAMAccountType", "userAccountControl", "pwdLastSet"])
        self.assertTrue(len(res1) == 1)
        self.assertEqual(int(res1[0]["sAMAccountType"][0]),
                         ATYPE_NORMAL_ACCOUNT)
        self.assertEqual(int(res1[0]["userAccountControl"][0]),
                         UF_NORMAL_ACCOUNT | UF_ACCOUNTDISABLE | UF_PASSWD_NOTREQD)
        self.assertEqual(int(res1[0]["pwdLastSet"][0]), lastset)

        try:
            m = Message()
            m.dn = Dn(ldb, "cn=ldaptestuser,cn=users," + self.base_dn)
            m["pls1"] = MessageElement(str(lastset),
                                       FLAG_MOD_DELETE,
                                       "pwdLastSet")
            m["pls2"] = MessageElement(str(lastset),
                                       FLAG_MOD_ADD,
                                       "pwdLastSet")
            ldb.modify(m)
            self.fail()
        except LdbError as e54:
            (num, msg) = e54.args
            self.assertEqual(num, ERR_OTHER)
            self.assertTrue('00000057' in msg)

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestuser,cn=users," + self.base_dn)
        m["pls1"] = MessageElement(str(lastset),
                                   FLAG_MOD_DELETE,
                                   "pwdLastSet")
        m["pls2"] = MessageElement(str(0),
                                   FLAG_MOD_ADD,
                                   "pwdLastSet")
        ldb.modify(m)
        res1 = ldb.search("cn=ldaptestuser,cn=users," + self.base_dn,
                          scope=SCOPE_BASE,
                          attrs=["sAMAccountType", "userAccountControl", "pwdLastSet"])
        self.assertTrue(len(res1) == 1)
        self.assertEqual(int(res1[0]["sAMAccountType"][0]),
                         ATYPE_NORMAL_ACCOUNT)
        self.assertEqual(int(res1[0]["userAccountControl"][0]),
                         UF_NORMAL_ACCOUNT | UF_ACCOUNTDISABLE | UF_PASSWD_NOTREQD)
        uac = int(res1[0]["userAccountControl"][0])
        self.assertEqual(int(res1[0]["pwdLastSet"][0]), 0)

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestuser,cn=users," + self.base_dn)
        m["uac1"] = MessageElement(str(uac |UF_PASSWORD_EXPIRED),
                                   FLAG_MOD_REPLACE,
                                   "userAccountControl")
        ldb.modify(m)
        res1 = ldb.search("cn=ldaptestuser,cn=users," + self.base_dn,
                          scope=SCOPE_BASE,
                          attrs=["sAMAccountType", "userAccountControl", "pwdLastSet"])
        self.assertTrue(len(res1) == 1)
        self.assertEqual(int(res1[0]["sAMAccountType"][0]),
                         ATYPE_NORMAL_ACCOUNT)
        self.assertEqual(int(res1[0]["userAccountControl"][0]),
                         UF_NORMAL_ACCOUNT | UF_ACCOUNTDISABLE | UF_PASSWD_NOTREQD)
        self.assertEqual(int(res1[0]["pwdLastSet"][0]), 0)

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

    def test_ldap_bind_must_change_pwd(self):
        """Test the error messages for failing LDAP binds"""
        print("Test the error messages for failing LDAP binds\n")

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

        def format_error_msg(hresult_v, dsid_v, werror_v):
            #
            # There are 4 lower case hex digits following 'v' at the end,
            # but different Windows Versions return different values:
            #
            # Windows 2008R2 uses 'v1db1'
            # Windows 2012R2 uses 'v2580'
            #
            return "%08X: LdapErr: DSID-%08X, comment: AcceptSecurityContext error, data %x, v" % (
                    hresult_v, dsid_v, werror_v)

        HRES_SEC_E_LOGON_DENIED = 0x8009030C
        HRES_SEC_E_INVALID_TOKEN = 0x80090308

        sasl_bind_dsid = 0x0C0904DC
        simple_bind_dsid = 0x0C0903A9

        error_msg_sasl_wrong_pw = format_error_msg(
                                HRES_SEC_E_LOGON_DENIED,
                                sasl_bind_dsid,
                                werror.WERR_LOGON_FAILURE)
        error_msg_sasl_must_change = format_error_msg(
                                HRES_SEC_E_LOGON_DENIED,
                                sasl_bind_dsid,
                                werror.WERR_PASSWORD_MUST_CHANGE)
        error_msg_simple_wrong_pw = format_error_msg(
                                HRES_SEC_E_INVALID_TOKEN,
                                simple_bind_dsid,
                                werror.WERR_LOGON_FAILURE)
        error_msg_simple_must_change = format_error_msg(
                                HRES_SEC_E_INVALID_TOKEN,
                                simple_bind_dsid,
                                werror.WERR_PASSWORD_MUST_CHANGE)

        username = "ldaptestuser"
        password = "thatsAcomplPASS2"
        utf16pw = ('"' + password + '"').encode('utf-16-le')

        ldb.add({
            "dn": "cn=ldaptestuser,cn=users," + self.base_dn,
            "objectclass": "user",
            "sAMAccountName": username,
            "userAccountControl": str(UF_NORMAL_ACCOUNT),
            "unicodePwd": utf16pw,
        })

        res1 = ldb.search("cn=ldaptestuser,cn=users," + self.base_dn,
                          scope=SCOPE_BASE,
                          attrs=["sAMAccountName", "sAMAccountType", "userAccountControl", "pwdLastSet"])
        self.assertTrue(len(res1) == 1)
        self.assertEqual(str(res1[0]["sAMAccountName"][0]), username)
        self.assertEqual(int(res1[0]["sAMAccountType"][0]), ATYPE_NORMAL_ACCOUNT)
        self.assertEqual(int(res1[0]["userAccountControl"][0]), UF_NORMAL_ACCOUNT)
        self.assertNotEqual(int(res1[0]["pwdLastSet"][0]), 0)

        # Open a second LDB connection with the user credentials. Use the
        # command line credentials for information like the domain, the realm
        # and the workstation.
        sasl_creds = Credentials()
        sasl_creds.set_username(username)
        sasl_creds.set_password(password)
        sasl_creds.set_domain(creds.get_domain())
        sasl_creds.set_workstation(creds.get_workstation())
        sasl_creds.set_gensec_features(creds.get_gensec_features() | gensec.FEATURE_SEAL)
        sasl_creds.set_kerberos_state(DONT_USE_KERBEROS)

        sasl_wrong_creds = Credentials()
        sasl_wrong_creds.set_username(username)
        sasl_wrong_creds.set_password("wrong")
        sasl_wrong_creds.set_domain(creds.get_domain())
        sasl_wrong_creds.set_workstation(creds.get_workstation())
        sasl_wrong_creds.set_gensec_features(creds.get_gensec_features() | gensec.FEATURE_SEAL)
        sasl_wrong_creds.set_kerberos_state(DONT_USE_KERBEROS)

        simple_creds = Credentials()
        simple_creds.set_bind_dn("cn=ldaptestuser,cn=users," + self.base_dn)
        simple_creds.set_username(username)
        simple_creds.set_password(password)
        simple_creds.set_domain(creds.get_domain())
        simple_creds.set_workstation(creds.get_workstation())
        simple_creds.set_gensec_features(creds.get_gensec_features() | gensec.FEATURE_SEAL)
        simple_creds.set_kerberos_state(DONT_USE_KERBEROS)

        simple_wrong_creds = Credentials()
        simple_wrong_creds.set_bind_dn("cn=ldaptestuser,cn=users," + self.base_dn)
        simple_wrong_creds.set_username(username)
        simple_wrong_creds.set_password("wrong")
        simple_wrong_creds.set_domain(creds.get_domain())
        simple_wrong_creds.set_workstation(creds.get_workstation())
        simple_wrong_creds.set_gensec_features(creds.get_gensec_features() | gensec.FEATURE_SEAL)
        simple_wrong_creds.set_kerberos_state(DONT_USE_KERBEROS)

        sasl_ldb = SamDB(url=host, credentials=sasl_creds, lp=lp)
        self.assertIsNotNone(sasl_ldb)
        del sasl_ldb

        requires_strong_auth = False
        try:
            simple_ldb = SamDB(url=host, credentials=simple_creds, lp=lp)
            self.assertIsNotNone(simple_ldb)
            del simple_ldb
        except LdbError as e55:
            (num, msg) = e55.args
            if num != ERR_STRONG_AUTH_REQUIRED:
                raise
            requires_strong_auth = True

        def assertLDAPErrorMsg(msg, expected_msg):
            self.assertTrue(expected_msg in msg,
                            "msg[%s] does not contain expected[%s]" % (
                                msg, expected_msg))

        try:
            SamDB(url=host, credentials=sasl_wrong_creds, lp=lp)
            self.fail()
        except LdbError as e56:
            (num, msg) = e56.args
            self.assertEqual(num, ERR_INVALID_CREDENTIALS)
            self.assertTrue(error_msg_sasl_wrong_pw in msg)

        if not requires_strong_auth:
            try:
                SamDB(url=host, credentials=simple_wrong_creds, lp=lp)
                self.fail()
            except LdbError as e4:
                (num, msg) = e4.args
                self.assertEqual(num, ERR_INVALID_CREDENTIALS)
                assertLDAPErrorMsg(msg, error_msg_simple_wrong_pw)

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestuser,cn=users," + self.base_dn)
        m["pls1"] = MessageElement(str(0),
                                   FLAG_MOD_REPLACE,
                                   "pwdLastSet")
        ldb.modify(m)

        res1 = ldb.search("cn=ldaptestuser,cn=users," + self.base_dn,
                          scope=SCOPE_BASE, attrs=["pwdLastSet"])
        self.assertEqual(int(res1[0]["pwdLastSet"][0]), 0)

        try:
            SamDB(url=host, credentials=sasl_wrong_creds, lp=lp)
            self.fail()
        except LdbError as e57:
            (num, msg) = e57.args
            self.assertEqual(num, ERR_INVALID_CREDENTIALS)
            assertLDAPErrorMsg(msg, error_msg_sasl_wrong_pw)

        try:
            SamDB(url=host, credentials=sasl_creds, lp=lp)
            self.fail()
        except LdbError as e58:
            (num, msg) = e58.args
            self.assertEqual(num, ERR_INVALID_CREDENTIALS)
            assertLDAPErrorMsg(msg, error_msg_sasl_must_change)

        if not requires_strong_auth:
            try:
                SamDB(url=host, credentials=simple_wrong_creds, lp=lp)
                self.fail()
            except LdbError as e5:
                (num, msg) = e5.args
                self.assertEqual(num, ERR_INVALID_CREDENTIALS)
                assertLDAPErrorMsg(msg, error_msg_simple_wrong_pw)

            try:
                SamDB(url=host, credentials=simple_creds, lp=lp)
                self.fail()
            except LdbError as e6:
                (num, msg) = e6.args
                self.assertEqual(num, ERR_INVALID_CREDENTIALS)
                assertLDAPErrorMsg(msg, error_msg_simple_must_change)

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

    def test_userAccountControl_user_add_0_uac(self):
        """Test the userAccountControl behaviour"""
        print("Testing userAccountControl behaviour\n")

        # With a user object

        # Add operation

        # As user you can only set a normal account.
        # The UF_PASSWD_NOTREQD flag is needed since we haven't requested a
        # password yet.
        # With SYSTEM rights you can set a interdomain trust account.

        ldb.add({
            "dn": "cn=ldaptestuser,cn=users," + self.base_dn,
            "objectclass": "user",
            "userAccountControl": "0"})

        res1 = ldb.search("cn=ldaptestuser,cn=users," + self.base_dn,
                          scope=SCOPE_BASE,
                          attrs=["sAMAccountType", "userAccountControl"])
        self.assertTrue(len(res1) == 1)
        self.assertEqual(int(res1[0]["sAMAccountType"][0]),
                          ATYPE_NORMAL_ACCOUNT)
        self.assertTrue(int(res1[0]["userAccountControl"][0]) & UF_ACCOUNTDISABLE == 0)
        self.assertTrue(int(res1[0]["userAccountControl"][0]) & UF_PASSWD_NOTREQD == 0)
        delete_force(self.ldb, "cn=ldaptestuser,cn=users," + self.base_dn)

    def test_userAccountControl_user_add_normal(self):
        """Test the userAccountControl behaviour"""
        ldb.add({
            "dn": "cn=ldaptestuser,cn=users," + self.base_dn,
            "objectclass": "user",
            "userAccountControl": str(UF_NORMAL_ACCOUNT)})
        delete_force(self.ldb, "cn=ldaptestuser,cn=users," + self.base_dn)

    def test_userAccountControl_user_add_normal_pwnotreq(self):
        ldb.add({
            "dn": "cn=ldaptestuser,cn=users," + self.base_dn,
            "objectclass": "user",
            "userAccountControl": str(UF_NORMAL_ACCOUNT | UF_PASSWD_NOTREQD)})

        res1 = ldb.search("cn=ldaptestuser,cn=users," + self.base_dn,
                          scope=SCOPE_BASE,
                          attrs=["sAMAccountType", "userAccountControl"])
        self.assertTrue(len(res1) == 1)
        self.assertEqual(int(res1[0]["sAMAccountType"][0]),
                          ATYPE_NORMAL_ACCOUNT)
        self.assertTrue(int(res1[0]["userAccountControl"][0]) & UF_ACCOUNTDISABLE == 0)
        delete_force(self.ldb, "cn=ldaptestuser,cn=users," + self.base_dn)

    def test_userAccountControl_user_add_normal_pwnotreq_lockout_expired(self):
        ldb.add({
            "dn": "cn=ldaptestuser,cn=users," + self.base_dn,
            "objectclass": "user",
            "userAccountControl": str(UF_NORMAL_ACCOUNT |
                                      UF_PASSWD_NOTREQD |
                                      UF_LOCKOUT |
                                      UF_PASSWORD_EXPIRED)})

        res1 = ldb.search("cn=ldaptestuser,cn=users," + self.base_dn,
                          scope=SCOPE_BASE,
                          attrs=["sAMAccountType", "userAccountControl", "lockoutTime", "pwdLastSet"])
        self.assertTrue(len(res1) == 1)
        self.assertEqual(int(res1[0]["sAMAccountType"][0]),
                          ATYPE_NORMAL_ACCOUNT)
        self.assertTrue(int(res1[0]["userAccountControl"][0]) & (UF_LOCKOUT | UF_PASSWORD_EXPIRED) == 0)
        self.assertFalse("lockoutTime" in res1[0])
        self.assertTrue(int(res1[0]["pwdLastSet"][0]) == 0)
        delete_force(self.ldb, "cn=ldaptestuser,cn=users," + self.base_dn)

    def test_userAccountControl_user_add_temp_dup(self):
        try:
            ldb.add({
                "dn": "cn=ldaptestuser,cn=users," + self.base_dn,
                "objectclass": "user",
                "userAccountControl": str(UF_TEMP_DUPLICATE_ACCOUNT)})
            self.fail()
        except LdbError as e59:
            (num, _) = e59.args
            self.assertEqual(num, ERR_OTHER)
        delete_force(self.ldb, "cn=ldaptestuser,cn=users," + self.base_dn)

    def test_userAccountControl_user_add_server(self):
        try:
            ldb.add({
                "dn": "cn=ldaptestuser,cn=users," + self.base_dn,
                "objectclass": "user",
                "userAccountControl": str(UF_SERVER_TRUST_ACCOUNT)})
            self.fail()
        except LdbError as e60:
            (num, _) = e60.args
            self.assertEqual(num, ERR_OBJECT_CLASS_VIOLATION)
        delete_force(self.ldb, "cn=ldaptestuser,cn=users," + self.base_dn)

    def test_userAccountControl_user_add_workstation(self):
        try:
            ldb.add({
                "dn": "cn=ldaptestuser,cn=users," + self.base_dn,
                "objectclass": "user",
                "userAccountControl": str(UF_WORKSTATION_TRUST_ACCOUNT)})
        except LdbError as e61:
            (num, _) = e61.args
            self.assertEqual(num, ERR_OBJECT_CLASS_VIOLATION)
        delete_force(self.ldb, "cn=ldaptestuser,cn=users," + self.base_dn)

    def test_userAccountControl_user_add_rodc(self):
        try:
            ldb.add({
                "dn": "cn=ldaptestuser,cn=users," + self.base_dn,
                "objectclass": "user",
                "userAccountControl": str(UF_WORKSTATION_TRUST_ACCOUNT | UF_PARTIAL_SECRETS_ACCOUNT)})
        except LdbError as e62:
            (num, _) = e62.args
            self.assertEqual(num, ERR_OBJECT_CLASS_VIOLATION)
        delete_force(self.ldb, "cn=ldaptestuser,cn=users," + self.base_dn)

    def test_userAccountControl_user_add_trust(self):
        try:
            ldb.add({
                "dn": "cn=ldaptestuser,cn=users," + self.base_dn,
                "objectclass": "user",
                "userAccountControl": str(UF_INTERDOMAIN_TRUST_ACCOUNT)})
            self.fail()
        except LdbError as e63:
            (num, _) = e63.args
            self.assertEqual(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
        delete_force(self.ldb, "cn=ldaptestuser,cn=users," + self.base_dn)

        # Modify operation

    def test_userAccountControl_user_modify(self):
        ldb.add({
            "dn": "cn=ldaptestuser,cn=users," + self.base_dn,
            "objectclass": "user"})

        # After creation we should have a normal account
        res1 = ldb.search("cn=ldaptestuser,cn=users," + self.base_dn,
                          scope=SCOPE_BASE,
                          attrs=["sAMAccountType", "userAccountControl"])
        self.assertTrue(len(res1) == 1)
        self.assertEqual(int(res1[0]["sAMAccountType"][0]),
                          ATYPE_NORMAL_ACCOUNT)
        self.assertTrue(int(res1[0]["userAccountControl"][0]) & UF_ACCOUNTDISABLE != 0)

        # As user you can only switch from a normal account to a workstation
        # trust account and back.
        # The UF_PASSWD_NOTREQD flag is needed since we haven't requested a
        # password yet.
        # With SYSTEM rights you can switch to a interdomain trust account.

        # Invalid attribute
        try:
            m = Message()
            m.dn = Dn(ldb, "cn=ldaptestuser,cn=users," + self.base_dn)
            m["userAccountControl"] = MessageElement("0",
                                                     FLAG_MOD_REPLACE, "userAccountControl")
            ldb.modify(m)
        except LdbError as e64:
            (num, _) = e64.args
            self.assertEqual(num, ERR_UNWILLING_TO_PERFORM)

        try:
            m = Message()
            m.dn = Dn(ldb, "cn=ldaptestuser,cn=users," + self.base_dn)
            m["userAccountControl"] = MessageElement(
                str(UF_NORMAL_ACCOUNT),
                FLAG_MOD_REPLACE, "userAccountControl")
            ldb.modify(m)
        except LdbError as e65:
            (num, _) = e65.args
            self.assertEqual(num, ERR_UNWILLING_TO_PERFORM)

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestuser,cn=users," + self.base_dn)
        m["userAccountControl"] = MessageElement(
            str(UF_NORMAL_ACCOUNT | UF_PASSWD_NOTREQD),
            FLAG_MOD_REPLACE, "userAccountControl")
        ldb.modify(m)

        res1 = ldb.search("cn=ldaptestuser,cn=users," + self.base_dn,
                          scope=SCOPE_BASE,
                          attrs=["sAMAccountType", "userAccountControl"])
        self.assertTrue(len(res1) == 1)
        self.assertEqual(int(res1[0]["sAMAccountType"][0]),
                          ATYPE_NORMAL_ACCOUNT)
        self.assertTrue(int(res1[0]["userAccountControl"][0]) & UF_ACCOUNTDISABLE == 0)

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestuser,cn=users," + self.base_dn)
        m["userAccountControl"] = MessageElement(
            str(UF_ACCOUNTDISABLE),
            FLAG_MOD_REPLACE, "userAccountControl")
        ldb.modify(m)

        res1 = ldb.search("cn=ldaptestuser,cn=users," + self.base_dn,
                          scope=SCOPE_BASE,
                          attrs=["sAMAccountType", "userAccountControl"])
        self.assertTrue(len(res1) == 1)
        self.assertEqual(int(res1[0]["sAMAccountType"][0]),
                          ATYPE_NORMAL_ACCOUNT)
        self.assertTrue(int(res1[0]["userAccountControl"][0]) & UF_NORMAL_ACCOUNT != 0)
        self.assertTrue(int(res1[0]["userAccountControl"][0]) & UF_ACCOUNTDISABLE != 0)

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestuser,cn=users," + self.base_dn)
        m["lockoutTime"] = MessageElement(str(samba.unix2nttime(0)), FLAG_MOD_REPLACE, "lockoutTime")
        m["pwdLastSet"] = MessageElement(str(samba.unix2nttime(0)), FLAG_MOD_REPLACE, "pwdLastSet")
        ldb.modify(m)

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestuser,cn=users," + self.base_dn)
        m["userAccountControl"] = MessageElement(
            str(UF_LOCKOUT | UF_PASSWORD_EXPIRED),
            FLAG_MOD_REPLACE, "userAccountControl")
        ldb.modify(m)

        res1 = ldb.search("cn=ldaptestuser,cn=users," + self.base_dn,
                          scope=SCOPE_BASE,
                          attrs=["sAMAccountType", "userAccountControl", "lockoutTime", "pwdLastSet"])
        self.assertTrue(len(res1) == 1)
        self.assertEqual(int(res1[0]["sAMAccountType"][0]),
                          ATYPE_NORMAL_ACCOUNT)
        self.assertTrue(int(res1[0]["userAccountControl"][0]) & UF_NORMAL_ACCOUNT != 0)
        self.assertTrue(int(res1[0]["userAccountControl"][0]) & (UF_LOCKOUT | UF_PASSWORD_EXPIRED) == 0)
        self.assertTrue(int(res1[0]["lockoutTime"][0]) == 0)
        self.assertTrue(int(res1[0]["pwdLastSet"][0]) == 0)

        try:
            m = Message()
            m.dn = Dn(ldb, "cn=ldaptestuser,cn=users," + self.base_dn)
            m["userAccountControl"] = MessageElement(
                str(UF_TEMP_DUPLICATE_ACCOUNT),
                FLAG_MOD_REPLACE, "userAccountControl")
            ldb.modify(m)
            self.fail()
        except LdbError as e66:
            (num, _) = e66.args
            self.assertEqual(num, ERR_OTHER)

        try:
            m = Message()
            m.dn = Dn(ldb, "cn=ldaptestuser,cn=users," + self.base_dn)
            m["userAccountControl"] = MessageElement(
                str(UF_SERVER_TRUST_ACCOUNT),
                FLAG_MOD_REPLACE, "userAccountControl")
            ldb.modify(m)
            self.fail()
        except LdbError as e67:
            (num, _) = e67.args
            self.assertEqual(num, ERR_OBJECT_CLASS_VIOLATION)

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestuser,cn=users," + self.base_dn)
        m["userAccountControl"] = MessageElement(
            str(UF_WORKSTATION_TRUST_ACCOUNT),
            FLAG_MOD_REPLACE, "userAccountControl")
        ldb.modify(m)

        try:
            m = Message()
            m.dn = Dn(ldb, "cn=ldaptestuser,cn=users," + self.base_dn)
            m["userAccountControl"] = MessageElement(
                str(UF_WORKSTATION_TRUST_ACCOUNT | UF_PARTIAL_SECRETS_ACCOUNT),
                FLAG_MOD_REPLACE, "userAccountControl")
            ldb.modify(m)
            self.fail()
        except LdbError as e68:
            (num, _) = e68.args
            self.assertEqual(num, ERR_OBJECT_CLASS_VIOLATION)

        res1 = ldb.search("cn=ldaptestuser,cn=users," + self.base_dn,
                          scope=SCOPE_BASE, attrs=["sAMAccountType"])
        self.assertTrue(len(res1) == 1)
        self.assertEqual(int(res1[0]["sAMAccountType"][0]),
                          ATYPE_WORKSTATION_TRUST)

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestuser,cn=users," + self.base_dn)
        m["userAccountControl"] = MessageElement(
            str(UF_NORMAL_ACCOUNT | UF_PASSWD_NOTREQD),
            FLAG_MOD_REPLACE, "userAccountControl")
        ldb.modify(m)

        res1 = ldb.search("cn=ldaptestuser,cn=users," + self.base_dn,
                          scope=SCOPE_BASE, attrs=["sAMAccountType"])
        self.assertTrue(len(res1) == 1)
        self.assertEqual(int(res1[0]["sAMAccountType"][0]),
                          ATYPE_NORMAL_ACCOUNT)

        try:
            m = Message()
            m.dn = Dn(ldb, "cn=ldaptestuser,cn=users," + self.base_dn)
            m["userAccountControl"] = MessageElement(
                str(UF_INTERDOMAIN_TRUST_ACCOUNT),
                FLAG_MOD_REPLACE, "userAccountControl")
            ldb.modify(m)
            self.fail()
        except LdbError as e69:
            (num, _) = e69.args
            self.assertEqual(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)

    def test_userAccountControl_computer_add_0_uac(self):
        # With a computer object

        # Add operation

        # As computer you can set a normal account and a server trust account.
        # The UF_PASSWD_NOTREQD flag is needed since we haven't requested a
        # password yet.
        # With SYSTEM rights you can set a interdomain trust account.

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

        res1 = ldb.search("cn=ldaptestcomputer,cn=computers," + self.base_dn,
                          scope=SCOPE_BASE,
                          attrs=["sAMAccountType", "userAccountControl"])
        self.assertTrue(len(res1) == 1)
        self.assertEqual(int(res1[0]["sAMAccountType"][0]),
                          ATYPE_WORKSTATION_TRUST)
        self.assertTrue(int(res1[0]["userAccountControl"][0]) & UF_ACCOUNTDISABLE == 0)
        self.assertTrue(int(res1[0]["userAccountControl"][0]) & UF_PASSWD_NOTREQD == 0)
        delete_force(self.ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)

    def test_userAccountControl_computer_add_normal(self):
        ldb.add({
            "dn": "cn=ldaptestcomputer,cn=computers," + self.base_dn,
            "objectclass": "computer",
            "userAccountControl": str(UF_NORMAL_ACCOUNT)})
        delete_force(self.ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)

    def test_userAccountControl_computer_add_normal_pwnotreqd(self):
        ldb.add({
            "dn": "cn=ldaptestcomputer,cn=computers," + self.base_dn,
            "objectclass": "computer",
            "userAccountControl": str(UF_NORMAL_ACCOUNT | UF_PASSWD_NOTREQD)})

        res1 = ldb.search("cn=ldaptestcomputer,cn=computers," + self.base_dn,
                          scope=SCOPE_BASE,
                          attrs=["sAMAccountType", "userAccountControl"])
        self.assertTrue(len(res1) == 1)
        self.assertEqual(int(res1[0]["sAMAccountType"][0]),
                          ATYPE_NORMAL_ACCOUNT)
        self.assertTrue(int(res1[0]["userAccountControl"][0]) & UF_ACCOUNTDISABLE == 0)
        delete_force(self.ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)

    def test_userAccountControl_computer_add_normal_pwnotreqd_lockout_expired(self):
        ldb.add({
            "dn": "cn=ldaptestcomputer,cn=computers," + self.base_dn,
            "objectclass": "computer",
            "userAccountControl": str(UF_NORMAL_ACCOUNT |
                                      UF_PASSWD_NOTREQD |
                                      UF_LOCKOUT |
                                      UF_PASSWORD_EXPIRED)})

        res1 = ldb.search("cn=ldaptestcomputer,cn=computers," + self.base_dn,
                          scope=SCOPE_BASE,
                          attrs=["sAMAccountType", "userAccountControl", "lockoutTime", "pwdLastSet"])
        self.assertTrue(len(res1) == 1)
        self.assertEqual(int(res1[0]["sAMAccountType"][0]),
                          ATYPE_NORMAL_ACCOUNT)
        self.assertTrue(int(res1[0]["userAccountControl"][0]) & (UF_LOCKOUT | UF_PASSWORD_EXPIRED) == 0)
        self.assertFalse("lockoutTime" in res1[0])
        self.assertTrue(int(res1[0]["pwdLastSet"][0]) == 0)
        delete_force(self.ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)

    def test_userAccountControl_computer_add_temp_dup(self):
        try:
            ldb.add({
                "dn": "cn=ldaptestcomputer,cn=computers," + self.base_dn,
                "objectclass": "computer",
                "userAccountControl": str(UF_TEMP_DUPLICATE_ACCOUNT)})
            self.fail()
        except LdbError as e70:
            (num, _) = e70.args
            self.assertEqual(num, ERR_OTHER)
        delete_force(self.ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)

    def test_userAccountControl_computer_add_server(self):
        ldb.add({
            "dn": "cn=ldaptestcomputer,cn=computers," + self.base_dn,
            "objectclass": "computer",
            "userAccountControl": str(UF_SERVER_TRUST_ACCOUNT)})

        res1 = ldb.search("cn=ldaptestcomputer,cn=computers," + self.base_dn,
                          scope=SCOPE_BASE, attrs=["sAMAccountType"])
        self.assertTrue(len(res1) == 1)
        self.assertEqual(int(res1[0]["sAMAccountType"][0]),
                          ATYPE_WORKSTATION_TRUST)
        delete_force(self.ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)

    def test_userAccountControl_computer_add_workstation(self):
        try:
            ldb.add({
                "dn": "cn=ldaptestcomputer,cn=computers," + self.base_dn,
                "objectclass": "computer",
                "userAccountControl": str(UF_WORKSTATION_TRUST_ACCOUNT)})
        except LdbError as e71:
            (num, _) = e71.args
            self.assertEqual(num, ERR_OBJECT_CLASS_VIOLATION)
        delete_force(self.ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)

    def test_userAccountControl_computer_add_trust(self):
        try:
            ldb.add({
                "dn": "cn=ldaptestcomputer,cn=computers," + self.base_dn,
                "objectclass": "computer",
                "userAccountControl": str(UF_INTERDOMAIN_TRUST_ACCOUNT)})
            self.fail()
        except LdbError as e72:
            (num, _) = e72.args
            self.assertEqual(num, ERR_OBJECT_CLASS_VIOLATION)
        delete_force(self.ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)

    def test_userAccountControl_computer_modify(self):
        # Modify operation

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

        # After creation we should have a normal account
        res1 = ldb.search("cn=ldaptestcomputer,cn=computers," + self.base_dn,
                          scope=SCOPE_BASE,
                          attrs=["sAMAccountType", "userAccountControl"])
        self.assertTrue(len(res1) == 1)
        self.assertEqual(int(res1[0]["sAMAccountType"][0]),
                          ATYPE_WORKSTATION_TRUST)
        self.assertTrue(int(res1[0]["userAccountControl"][0]) & UF_ACCOUNTDISABLE != 0)

        # As computer you can switch from a normal account to a workstation
        # or server trust account and back (also swapping between trust
        # accounts is allowed).
        # The UF_PASSWD_NOTREQD flag is needed since we haven't requested a
        # password yet.
        # With SYSTEM rights you can switch to a interdomain trust account.

        # Invalid attribute
        try:
            m = Message()
            m.dn = Dn(ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)
            m["userAccountControl"] = MessageElement("0",
                                                     FLAG_MOD_REPLACE, "userAccountControl")
            ldb.modify(m)
        except LdbError as e73:
            (num, _) = e73.args
            self.assertEqual(num, ERR_UNWILLING_TO_PERFORM)

        try:
            m = Message()
            m.dn = Dn(ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)
            m["userAccountControl"] = MessageElement(
                str(UF_NORMAL_ACCOUNT),
                FLAG_MOD_REPLACE, "userAccountControl")
            ldb.modify(m)
        except LdbError as e74:
            (num, _) = e74.args
            self.assertEqual(num, ERR_UNWILLING_TO_PERFORM)

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)
        m["userAccountControl"] = MessageElement(
            str(UF_NORMAL_ACCOUNT | UF_PASSWD_NOTREQD),
            FLAG_MOD_REPLACE, "userAccountControl")
        ldb.modify(m)

        res1 = ldb.search("cn=ldaptestcomputer,cn=computers," + self.base_dn,
                          scope=SCOPE_BASE,
                          attrs=["sAMAccountType", "userAccountControl"])
        self.assertTrue(len(res1) == 1)
        self.assertEqual(int(res1[0]["sAMAccountType"][0]),
                          ATYPE_NORMAL_ACCOUNT)
        self.assertTrue(int(res1[0]["userAccountControl"][0]) & UF_ACCOUNTDISABLE == 0)

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)
        m["userAccountControl"] = MessageElement(
            str(UF_ACCOUNTDISABLE),
            FLAG_MOD_REPLACE, "userAccountControl")
        ldb.modify(m)

        res1 = ldb.search("cn=ldaptestcomputer,cn=computers," + self.base_dn,
                          scope=SCOPE_BASE,
                          attrs=["sAMAccountType", "userAccountControl"])
        self.assertTrue(len(res1) == 1)
        self.assertEqual(int(res1[0]["sAMAccountType"][0]),
                          ATYPE_NORMAL_ACCOUNT)
        self.assertTrue(int(res1[0]["userAccountControl"][0]) & UF_NORMAL_ACCOUNT != 0)
        self.assertTrue(int(res1[0]["userAccountControl"][0]) & UF_ACCOUNTDISABLE != 0)

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)
        m["lockoutTime"] = MessageElement(str(samba.unix2nttime(0)), FLAG_MOD_REPLACE, "lockoutTime")
        m["pwdLastSet"] = MessageElement(str(samba.unix2nttime(0)), FLAG_MOD_REPLACE, "pwdLastSet")
        ldb.modify(m)

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)
        m["userAccountControl"] = MessageElement(
            str(UF_LOCKOUT | UF_PASSWORD_EXPIRED),
            FLAG_MOD_REPLACE, "userAccountControl")
        ldb.modify(m)

        res1 = ldb.search("cn=ldaptestcomputer,cn=computers," + self.base_dn,
                          scope=SCOPE_BASE,
                          attrs=["sAMAccountType", "userAccountControl", "lockoutTime", "pwdLastSet"])
        self.assertTrue(len(res1) == 1)
        self.assertEqual(int(res1[0]["sAMAccountType"][0]),
                          ATYPE_NORMAL_ACCOUNT)
        self.assertTrue(int(res1[0]["userAccountControl"][0]) & UF_NORMAL_ACCOUNT != 0)
        self.assertTrue(int(res1[0]["userAccountControl"][0]) & (UF_LOCKOUT | UF_PASSWORD_EXPIRED) == 0)
        self.assertTrue(int(res1[0]["lockoutTime"][0]) == 0)
        self.assertTrue(int(res1[0]["pwdLastSet"][0]) == 0)

        try:
            m = Message()
            m.dn = Dn(ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)
            m["userAccountControl"] = MessageElement(
                str(UF_TEMP_DUPLICATE_ACCOUNT),
                FLAG_MOD_REPLACE, "userAccountControl")
            ldb.modify(m)
            self.fail()
        except LdbError as e75:
            (num, _) = e75.args
            self.assertEqual(num, ERR_OTHER)

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)
        m["userAccountControl"] = MessageElement(
            str(UF_SERVER_TRUST_ACCOUNT),
            FLAG_MOD_REPLACE, "userAccountControl")
        ldb.modify(m)

        res1 = ldb.search("cn=ldaptestcomputer,cn=computers," + self.base_dn,
                          scope=SCOPE_BASE, attrs=["sAMAccountType"])
        self.assertTrue(len(res1) == 1)
        self.assertEqual(int(res1[0]["sAMAccountType"][0]),
                          ATYPE_WORKSTATION_TRUST)

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)
        m["userAccountControl"] = MessageElement(
            str(UF_NORMAL_ACCOUNT | UF_PASSWD_NOTREQD),
            FLAG_MOD_REPLACE, "userAccountControl")
        ldb.modify(m)

        res1 = ldb.search("cn=ldaptestcomputer,cn=computers," + self.base_dn,
                          scope=SCOPE_BASE, attrs=["sAMAccountType"])
        self.assertTrue(len(res1) == 1)
        self.assertEqual(int(res1[0]["sAMAccountType"][0]),
                          ATYPE_NORMAL_ACCOUNT)

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)
        m["userAccountControl"] = MessageElement(
            str(UF_WORKSTATION_TRUST_ACCOUNT),
            FLAG_MOD_REPLACE, "userAccountControl")
        ldb.modify(m)

        res1 = ldb.search("cn=ldaptestcomputer,cn=computers," + self.base_dn,
                          scope=SCOPE_BASE, attrs=["sAMAccountType"])
        self.assertTrue(len(res1) == 1)
        self.assertEqual(int(res1[0]["sAMAccountType"][0]),
                          ATYPE_WORKSTATION_TRUST)

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)
        m["userAccountControl"] = MessageElement(
            str(UF_NORMAL_ACCOUNT | UF_PASSWD_NOTREQD),
            FLAG_MOD_REPLACE, "userAccountControl")
        ldb.modify(m)

        res1 = ldb.search("cn=ldaptestcomputer,cn=computers," + self.base_dn,
                          scope=SCOPE_BASE, attrs=["sAMAccountType"])
        self.assertTrue(len(res1) == 1)
        self.assertEqual(int(res1[0]["sAMAccountType"][0]),
                          ATYPE_NORMAL_ACCOUNT)

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)
        m["userAccountControl"] = MessageElement(
            str(UF_SERVER_TRUST_ACCOUNT),
            FLAG_MOD_REPLACE, "userAccountControl")
        ldb.modify(m)

        res1 = ldb.search("cn=ldaptestcomputer,cn=computers," + self.base_dn,
                          scope=SCOPE_BASE, attrs=["sAMAccountType"])
        self.assertTrue(len(res1) == 1)
        self.assertEqual(int(res1[0]["sAMAccountType"][0]),
                          ATYPE_WORKSTATION_TRUST)

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)
        m["userAccountControl"] = MessageElement(
            str(UF_WORKSTATION_TRUST_ACCOUNT),
            FLAG_MOD_REPLACE, "userAccountControl")
        ldb.modify(m)

        res1 = ldb.search("cn=ldaptestcomputer,cn=computers," + self.base_dn,
                          scope=SCOPE_BASE, attrs=["sAMAccountType"])
        self.assertTrue(len(res1) == 1)
        self.assertEqual(int(res1[0]["sAMAccountType"][0]),
                          ATYPE_WORKSTATION_TRUST)

        try:
            m = Message()
            m.dn = Dn(ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)
            m["userAccountControl"] = MessageElement(
                str(UF_INTERDOMAIN_TRUST_ACCOUNT),
                FLAG_MOD_REPLACE, "userAccountControl")
            ldb.modify(m)
            self.fail()
        except LdbError as e76:
            (num, _) = e76.args
            self.assertEqual(num, ERR_OBJECT_CLASS_VIOLATION)

        # "primaryGroupID" does not change if account type remains the same

        # For a user account

        ldb.add({
            "dn": "cn=ldaptestuser2,cn=users," + self.base_dn,
            "objectclass": "user",
            "userAccountControl": str(UF_NORMAL_ACCOUNT |
                                      UF_PASSWD_NOTREQD |
                                      UF_ACCOUNTDISABLE)})

        res1 = ldb.search("cn=ldaptestuser2,cn=users," + self.base_dn,
                          scope=SCOPE_BASE,
                          attrs=["userAccountControl"])
        self.assertTrue(len(res1) == 1)
        self.assertEqual(int(res1[0]["userAccountControl"][0]),
                          UF_NORMAL_ACCOUNT | UF_PASSWD_NOTREQD | UF_ACCOUNTDISABLE)

        m = Message()
        m.dn = Dn(ldb, "<SID=" + ldb.get_domain_sid() + "-" + str(DOMAIN_RID_ADMINS) + ">")
        m["member"] = MessageElement(
            "cn=ldaptestuser2,cn=users," + self.base_dn, FLAG_MOD_ADD, "member")
        ldb.modify(m)

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestuser2,cn=users," + self.base_dn)
        m["primaryGroupID"] = MessageElement(str(DOMAIN_RID_ADMINS),
                                             FLAG_MOD_REPLACE, "primaryGroupID")
        ldb.modify(m)

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestuser2,cn=users," + self.base_dn)
        m["userAccountControl"] = MessageElement(
            str(UF_NORMAL_ACCOUNT | UF_PASSWD_NOTREQD),
            FLAG_MOD_REPLACE, "userAccountControl")
        ldb.modify(m)

        res1 = ldb.search("cn=ldaptestuser2,cn=users," + self.base_dn,
                          scope=SCOPE_BASE,
                          attrs=["userAccountControl", "primaryGroupID"])
        self.assertTrue(len(res1) == 1)
        self.assertTrue(int(res1[0]["userAccountControl"][0]) & UF_ACCOUNTDISABLE == 0)
        self.assertEqual(int(res1[0]["primaryGroupID"][0]), DOMAIN_RID_ADMINS)

        # For a workstation account

        res1 = ldb.search("cn=ldaptestcomputer,cn=computers," + self.base_dn,
                          scope=SCOPE_BASE,
                          attrs=["primaryGroupID"])
        self.assertTrue(len(res1) == 1)
        self.assertEqual(int(res1[0]["primaryGroupID"][0]), DOMAIN_RID_DOMAIN_MEMBERS)

        m = Message()
        m.dn = Dn(ldb, "<SID=" + ldb.get_domain_sid() + "-" + str(DOMAIN_RID_USERS) + ">")
        m["member"] = MessageElement(
            "cn=ldaptestcomputer,cn=computers," + self.base_dn, FLAG_MOD_ADD, "member")
        ldb.modify(m)

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)
        m["primaryGroupID"] = MessageElement(str(DOMAIN_RID_USERS),
                                             FLAG_MOD_REPLACE, "primaryGroupID")
        ldb.modify(m)

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)
        m["userAccountControl"] = MessageElement(
            str(UF_WORKSTATION_TRUST_ACCOUNT),
            FLAG_MOD_REPLACE, "userAccountControl")
        ldb.modify(m)

        res1 = ldb.search("cn=ldaptestcomputer,cn=computers," + self.base_dn,
                          scope=SCOPE_BASE,
                          attrs=["primaryGroupID"])
        self.assertTrue(len(res1) == 1)
        self.assertEqual(int(res1[0]["primaryGroupID"][0]), DOMAIN_RID_USERS)

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

    def find_repl_meta_data(self, rpmd, attid):
        for i in range(0, rpmd.ctr.count):
            m = rpmd.ctr.array[i]
            if m.attid == attid:
                return m
        return None

    def test_smartcard_required1(self):
        """Test the UF_SMARTCARD_REQUIRED behaviour"""
        print("Testing UF_SMARTCARD_REQUIRED behaviour\n")

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

        ldb.add({
            "dn": "cn=ldaptestuser,cn=users," + self.base_dn,
            "objectclass": "user",
            "userAccountControl": str(UF_NORMAL_ACCOUNT),
            "unicodePwd": "\"thatsAcomplPASS2\"".encode('utf-16-le')
        })

        res = ldb.search("cn=ldaptestuser,cn=users," + self.base_dn,
                         scope=SCOPE_BASE,
                         attrs=["sAMAccountType", "userAccountControl",
                                "pwdLastSet", "msDS-KeyVersionNumber",
                                "replPropertyMetaData"])
        self.assertTrue(len(res) == 1)
        self.assertEqual(int(res[0]["sAMAccountType"][0]),
                         ATYPE_NORMAL_ACCOUNT)
        self.assertEqual(int(res[0]["userAccountControl"][0]),
                         UF_NORMAL_ACCOUNT)
        self.assertNotEqual(int(res[0]["pwdLastSet"][0]), 0)
        lastset = int(res[0]["pwdLastSet"][0])
        self.assertEqual(int(res[0]["msDS-KeyVersionNumber"][0]), 1)
        self.assertTrue(len(res[0]["replPropertyMetaData"]) == 1)
        rpmd = ndr_unpack(drsblobs.replPropertyMetaDataBlob,
                          res[0]["replPropertyMetaData"][0])
        lastsetmd = self.find_repl_meta_data(rpmd,
                                             drsuapi.DRSUAPI_ATTID_pwdLastSet)
        self.assertIsNotNone(lastsetmd)
        self.assertEqual(lastsetmd.version, 1)
        nthashmd = self.find_repl_meta_data(rpmd,
                                            drsuapi.DRSUAPI_ATTID_unicodePwd)
        self.assertIsNotNone(nthashmd)
        self.assertEqual(nthashmd.version, 1)
        nthistmd = self.find_repl_meta_data(rpmd,
                                            drsuapi.DRSUAPI_ATTID_ntPwdHistory)
        self.assertIsNotNone(nthistmd)
        self.assertEqual(nthistmd.version, 1)
        lmhashmd = self.find_repl_meta_data(rpmd,
                                            drsuapi.DRSUAPI_ATTID_dBCSPwd)
        self.assertIsNotNone(lmhashmd)
        self.assertEqual(lmhashmd.version, 1)
        lmhistmd = self.find_repl_meta_data(rpmd,
                                            drsuapi.DRSUAPI_ATTID_lmPwdHistory)
        self.assertIsNotNone(lmhistmd)
        self.assertEqual(lmhistmd.version, 1)
        spcbmd = self.find_repl_meta_data(rpmd,
                                          drsuapi.DRSUAPI_ATTID_supplementalCredentials)
        self.assertIsNotNone(spcbmd)
        self.assertEqual(spcbmd.version, 1)

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestuser,cn=users," + self.base_dn)
        m["userAccountControl"] = MessageElement(
            str(UF_NORMAL_ACCOUNT |UF_SMARTCARD_REQUIRED),
            FLAG_MOD_REPLACE, "userAccountControl")
        ldb.modify(m)

        res = ldb.search("cn=ldaptestuser,cn=users," + self.base_dn,
                         scope=SCOPE_BASE,
                         attrs=["sAMAccountType", "userAccountControl",
                                "pwdLastSet", "msDS-KeyVersionNumber",
                                "replPropertyMetaData"])
        self.assertTrue(len(res) == 1)
        self.assertEqual(int(res[0]["sAMAccountType"][0]),
                         ATYPE_NORMAL_ACCOUNT)
        self.assertEqual(int(res[0]["userAccountControl"][0]),
                         UF_NORMAL_ACCOUNT |UF_SMARTCARD_REQUIRED)
        self.assertEqual(int(res[0]["pwdLastSet"][0]), lastset)
        self.assertEqual(int(res[0]["msDS-KeyVersionNumber"][0]), 2)
        self.assertTrue(len(res[0]["replPropertyMetaData"]) == 1)
        rpmd = ndr_unpack(drsblobs.replPropertyMetaDataBlob,
                          res[0]["replPropertyMetaData"][0])
        lastsetmd = self.find_repl_meta_data(rpmd,
                                             drsuapi.DRSUAPI_ATTID_pwdLastSet)
        self.assertIsNotNone(lastsetmd)
        self.assertEqual(lastsetmd.version, 1)
        nthashmd = self.find_repl_meta_data(rpmd,
                                            drsuapi.DRSUAPI_ATTID_unicodePwd)
        self.assertIsNotNone(nthashmd)
        self.assertEqual(nthashmd.version, 2)
        nthistmd = self.find_repl_meta_data(rpmd,
                                            drsuapi.DRSUAPI_ATTID_ntPwdHistory)
        self.assertIsNotNone(nthistmd)
        self.assertEqual(nthistmd.version, 2)
        lmhashmd = self.find_repl_meta_data(rpmd,
                                            drsuapi.DRSUAPI_ATTID_dBCSPwd)
        self.assertIsNotNone(lmhashmd)
        self.assertEqual(lmhashmd.version, 2)
        lmhistmd = self.find_repl_meta_data(rpmd,
                                            drsuapi.DRSUAPI_ATTID_lmPwdHistory)
        self.assertIsNotNone(lmhistmd)
        self.assertEqual(lmhistmd.version, 2)
        spcbmd = self.find_repl_meta_data(rpmd,
                                          drsuapi.DRSUAPI_ATTID_supplementalCredentials)
        self.assertIsNotNone(spcbmd)
        self.assertEqual(spcbmd.version, 2)

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

    def test_smartcard_required2(self):
        """Test the UF_SMARTCARD_REQUIRED behaviour"""
        print("Testing UF_SMARTCARD_REQUIRED behaviour\n")

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

        ldb.add({
            "dn": "cn=ldaptestuser,cn=users," + self.base_dn,
            "objectclass": "user",
            "userAccountControl": str(UF_NORMAL_ACCOUNT |UF_ACCOUNTDISABLE),
        })

        res = ldb.search("cn=ldaptestuser,cn=users," + self.base_dn,
                         scope=SCOPE_BASE,
                         attrs=["sAMAccountType", "userAccountControl",
                                "pwdLastSet", "msDS-KeyVersionNumber",
                                "replPropertyMetaData"])
        self.assertTrue(len(res) == 1)
        self.assertEqual(int(res[0]["sAMAccountType"][0]),
                         ATYPE_NORMAL_ACCOUNT)
        self.assertEqual(int(res[0]["userAccountControl"][0]),
                         UF_NORMAL_ACCOUNT |UF_ACCOUNTDISABLE)
        self.assertEqual(int(res[0]["pwdLastSet"][0]), 0)
        self.assertTrue("msDS-KeyVersionNumber" in res[0])
        self.assertEqual(int(res[0]["msDS-KeyVersionNumber"][0]), 1)
        self.assertTrue(len(res[0]["replPropertyMetaData"]) == 1)
        rpmd = ndr_unpack(drsblobs.replPropertyMetaDataBlob,
                          res[0]["replPropertyMetaData"][0])
        lastsetmd = self.find_repl_meta_data(rpmd,
                                             drsuapi.DRSUAPI_ATTID_pwdLastSet)
        self.assertIsNotNone(lastsetmd)
        self.assertEqual(lastsetmd.version, 1)
        nthashmd = self.find_repl_meta_data(rpmd,
                                            drsuapi.DRSUAPI_ATTID_unicodePwd)
        self.assertIsNotNone(nthashmd)
        self.assertEqual(nthashmd.version, 1)
        nthistmd = self.find_repl_meta_data(rpmd,
                                            drsuapi.DRSUAPI_ATTID_ntPwdHistory)
        self.assertIsNotNone(nthistmd)
        self.assertEqual(nthistmd.version, 1)
        lmhashmd = self.find_repl_meta_data(rpmd,
                                            drsuapi.DRSUAPI_ATTID_dBCSPwd)
        self.assertIsNotNone(lmhashmd)
        self.assertEqual(lmhashmd.version, 1)
        lmhistmd = self.find_repl_meta_data(rpmd,
                                            drsuapi.DRSUAPI_ATTID_lmPwdHistory)
        self.assertIsNotNone(lmhistmd)
        self.assertEqual(lmhistmd.version, 1)
        spcbmd = self.find_repl_meta_data(rpmd,
                                          drsuapi.DRSUAPI_ATTID_supplementalCredentials)
        self.assertIsNone(spcbmd)

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestuser,cn=users," + self.base_dn)
        m["userAccountControl"] = MessageElement(
            str(UF_NORMAL_ACCOUNT |UF_ACCOUNTDISABLE |UF_SMARTCARD_REQUIRED),
            FLAG_MOD_REPLACE, "userAccountControl")
        ldb.modify(m)

        res = ldb.search("cn=ldaptestuser,cn=users," + self.base_dn,
                         scope=SCOPE_BASE,
                         attrs=["sAMAccountType", "userAccountControl",
                                "pwdLastSet", "msDS-KeyVersionNumber",
                                "replPropertyMetaData"])
        self.assertTrue(len(res) == 1)
        self.assertEqual(int(res[0]["sAMAccountType"][0]),
                         ATYPE_NORMAL_ACCOUNT)
        self.assertEqual(int(res[0]["userAccountControl"][0]),
                         UF_NORMAL_ACCOUNT |UF_ACCOUNTDISABLE |UF_SMARTCARD_REQUIRED)
        self.assertEqual(int(res[0]["pwdLastSet"][0]), 0)
        self.assertEqual(int(res[0]["msDS-KeyVersionNumber"][0]), 2)
        self.assertTrue(len(res[0]["replPropertyMetaData"]) == 1)
        rpmd = ndr_unpack(drsblobs.replPropertyMetaDataBlob,
                          res[0]["replPropertyMetaData"][0])
        lastsetmd = self.find_repl_meta_data(rpmd,
                                             drsuapi.DRSUAPI_ATTID_pwdLastSet)
        self.assertIsNotNone(lastsetmd)
        self.assertEqual(lastsetmd.version, 1)
        nthashmd = self.find_repl_meta_data(rpmd,
                                            drsuapi.DRSUAPI_ATTID_unicodePwd)
        self.assertIsNotNone(nthashmd)
        self.assertEqual(nthashmd.version, 2)
        nthistmd = self.find_repl_meta_data(rpmd,
                                            drsuapi.DRSUAPI_ATTID_ntPwdHistory)
        self.assertIsNotNone(nthistmd)
        self.assertEqual(nthistmd.version, 2)
        lmhashmd = self.find_repl_meta_data(rpmd,
                                            drsuapi.DRSUAPI_ATTID_dBCSPwd)
        self.assertIsNotNone(lmhashmd)
        self.assertEqual(lmhashmd.version, 2)
        lmhistmd = self.find_repl_meta_data(rpmd,
                                            drsuapi.DRSUAPI_ATTID_lmPwdHistory)
        self.assertIsNotNone(lmhistmd)
        self.assertEqual(lmhistmd.version, 2)
        spcbmd = self.find_repl_meta_data(rpmd,
                                          drsuapi.DRSUAPI_ATTID_supplementalCredentials)
        self.assertIsNotNone(spcbmd)
        self.assertEqual(spcbmd.version, 1)

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestuser,cn=users," + self.base_dn)
        m["userAccountControl"] = MessageElement(
            str(UF_NORMAL_ACCOUNT |UF_SMARTCARD_REQUIRED),
            FLAG_MOD_REPLACE, "userAccountControl")
        ldb.modify(m)

        res = ldb.search("cn=ldaptestuser,cn=users," + self.base_dn,
                         scope=SCOPE_BASE,
                         attrs=["sAMAccountType", "userAccountControl",
                                "pwdLastSet", "msDS-KeyVersionNumber",
                                "replPropertyMetaData"])
        self.assertTrue(len(res) == 1)
        self.assertEqual(int(res[0]["sAMAccountType"][0]),
                         ATYPE_NORMAL_ACCOUNT)
        self.assertEqual(int(res[0]["userAccountControl"][0]),
                         UF_NORMAL_ACCOUNT |UF_SMARTCARD_REQUIRED)
        self.assertEqual(int(res[0]["pwdLastSet"][0]), 0)
        self.assertEqual(int(res[0]["msDS-KeyVersionNumber"][0]), 2)
        self.assertTrue(len(res[0]["replPropertyMetaData"]) == 1)
        rpmd = ndr_unpack(drsblobs.replPropertyMetaDataBlob,
                          res[0]["replPropertyMetaData"][0])
        lastsetmd = self.find_repl_meta_data(rpmd,
                                             drsuapi.DRSUAPI_ATTID_pwdLastSet)
        self.assertIsNotNone(lastsetmd)
        self.assertEqual(lastsetmd.version, 1)
        nthashmd = self.find_repl_meta_data(rpmd,
                                            drsuapi.DRSUAPI_ATTID_unicodePwd)
        self.assertIsNotNone(nthashmd)
        self.assertEqual(nthashmd.version, 2)
        nthistmd = self.find_repl_meta_data(rpmd,
                                            drsuapi.DRSUAPI_ATTID_ntPwdHistory)
        self.assertIsNotNone(nthistmd)
        self.assertEqual(nthistmd.version, 2)
        lmhashmd = self.find_repl_meta_data(rpmd,
                                            drsuapi.DRSUAPI_ATTID_dBCSPwd)
        self.assertIsNotNone(lmhashmd)
        self.assertEqual(lmhashmd.version, 2)
        lmhistmd = self.find_repl_meta_data(rpmd,
                                            drsuapi.DRSUAPI_ATTID_lmPwdHistory)
        self.assertIsNotNone(lmhistmd)
        self.assertEqual(lmhistmd.version, 2)
        spcbmd = self.find_repl_meta_data(rpmd,
                                          drsuapi.DRSUAPI_ATTID_supplementalCredentials)
        self.assertIsNotNone(spcbmd)
        self.assertEqual(spcbmd.version, 1)

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

    def test_smartcard_required3(self):
        """Test the UF_SMARTCARD_REQUIRED behaviour"""
        print("Testing UF_SMARTCARD_REQUIRED behaviour\n")

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

        ldb.add({
            "dn": "cn=ldaptestuser,cn=users," + self.base_dn,
            "objectclass": "user",
            "userAccountControl": str(UF_NORMAL_ACCOUNT |UF_SMARTCARD_REQUIRED |UF_ACCOUNTDISABLE),
        })

        res = ldb.search("cn=ldaptestuser,cn=users," + self.base_dn,
                         scope=SCOPE_BASE,
                         attrs=["sAMAccountType", "userAccountControl",
                                "pwdLastSet", "msDS-KeyVersionNumber",
                                "replPropertyMetaData"])
        self.assertTrue(len(res) == 1)
        self.assertEqual(int(res[0]["sAMAccountType"][0]),
                         ATYPE_NORMAL_ACCOUNT)
        self.assertEqual(int(res[0]["userAccountControl"][0]),
                         UF_NORMAL_ACCOUNT |UF_SMARTCARD_REQUIRED |UF_ACCOUNTDISABLE)
        self.assertEqual(int(res[0]["pwdLastSet"][0]), 0)
        self.assertEqual(int(res[0]["msDS-KeyVersionNumber"][0]), 1)
        self.assertTrue(len(res[0]["replPropertyMetaData"]) == 1)
        rpmd = ndr_unpack(drsblobs.replPropertyMetaDataBlob,
                          res[0]["replPropertyMetaData"][0])
        lastsetmd = self.find_repl_meta_data(rpmd,
                                             drsuapi.DRSUAPI_ATTID_pwdLastSet)
        self.assertIsNotNone(lastsetmd)
        self.assertEqual(lastsetmd.version, 1)
        nthashmd = self.find_repl_meta_data(rpmd,
                                            drsuapi.DRSUAPI_ATTID_unicodePwd)
        self.assertIsNotNone(nthashmd)
        self.assertEqual(nthashmd.version, 1)
        nthistmd = self.find_repl_meta_data(rpmd,
                                            drsuapi.DRSUAPI_ATTID_ntPwdHistory)
        self.assertIsNotNone(nthistmd)
        self.assertEqual(nthistmd.version, 1)
        lmhashmd = self.find_repl_meta_data(rpmd,
                                            drsuapi.DRSUAPI_ATTID_dBCSPwd)
        self.assertIsNotNone(lmhashmd)
        self.assertEqual(lmhashmd.version, 1)
        lmhistmd = self.find_repl_meta_data(rpmd,
                                            drsuapi.DRSUAPI_ATTID_lmPwdHistory)
        self.assertIsNotNone(lmhistmd)
        self.assertEqual(lmhistmd.version, 1)
        spcbmd = self.find_repl_meta_data(rpmd,
                                          drsuapi.DRSUAPI_ATTID_supplementalCredentials)
        self.assertIsNotNone(spcbmd)
        self.assertEqual(spcbmd.version, 1)

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestuser,cn=users," + self.base_dn)
        m["userAccountControl"] = MessageElement(
            str(UF_NORMAL_ACCOUNT |UF_SMARTCARD_REQUIRED),
            FLAG_MOD_REPLACE, "userAccountControl")
        ldb.modify(m)

        res = ldb.search("cn=ldaptestuser,cn=users," + self.base_dn,
                         scope=SCOPE_BASE,
                         attrs=["sAMAccountType", "userAccountControl",
                                "pwdLastSet", "msDS-KeyVersionNumber",
                                "replPropertyMetaData"])
        self.assertTrue(len(res) == 1)
        self.assertEqual(int(res[0]["sAMAccountType"][0]),
                         ATYPE_NORMAL_ACCOUNT)
        self.assertEqual(int(res[0]["userAccountControl"][0]),
                         UF_NORMAL_ACCOUNT |UF_SMARTCARD_REQUIRED)
        self.assertEqual(int(res[0]["pwdLastSet"][0]), 0)
        self.assertEqual(int(res[0]["msDS-KeyVersionNumber"][0]), 1)
        self.assertTrue(len(res[0]["replPropertyMetaData"]) == 1)
        rpmd = ndr_unpack(drsblobs.replPropertyMetaDataBlob,
                          res[0]["replPropertyMetaData"][0])
        lastsetmd = self.find_repl_meta_data(rpmd,
                                             drsuapi.DRSUAPI_ATTID_pwdLastSet)
        self.assertIsNotNone(lastsetmd)
        self.assertEqual(lastsetmd.version, 1)
        nthashmd = self.find_repl_meta_data(rpmd,
                                            drsuapi.DRSUAPI_ATTID_unicodePwd)
        self.assertIsNotNone(nthashmd)
        self.assertEqual(nthashmd.version, 1)
        nthistmd = self.find_repl_meta_data(rpmd,
                                            drsuapi.DRSUAPI_ATTID_ntPwdHistory)
        self.assertIsNotNone(nthistmd)
        self.assertEqual(nthistmd.version, 1)
        lmhashmd = self.find_repl_meta_data(rpmd,
                                            drsuapi.DRSUAPI_ATTID_dBCSPwd)
        self.assertIsNotNone(lmhashmd)
        self.assertEqual(lmhashmd.version, 1)
        lmhistmd = self.find_repl_meta_data(rpmd,
                                            drsuapi.DRSUAPI_ATTID_lmPwdHistory)
        self.assertIsNotNone(lmhistmd)
        self.assertEqual(lmhistmd.version, 1)
        spcbmd = self.find_repl_meta_data(rpmd,
                                          drsuapi.DRSUAPI_ATTID_supplementalCredentials)
        self.assertIsNotNone(spcbmd)
        self.assertEqual(spcbmd.version, 1)

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

    def test_isCriticalSystemObject_user(self):
        """Test the isCriticalSystemObject behaviour"""
        print("Testing isCriticalSystemObject behaviour\n")

        # Add tests (of a user)

        ldb.add({
            "dn": "cn=ldaptestuser,cn=users," + self.base_dn,
            "objectclass": "user"})

        res1 = ldb.search("cn=ldaptestuser,cn=users," + self.base_dn,
                          scope=SCOPE_BASE,
                          attrs=["isCriticalSystemObject"])
        self.assertTrue(len(res1) == 1)
        self.assertTrue("isCriticalSystemObject" not in res1[0])

        # Modification tests
        m = Message()

        m.dn = Dn(ldb, "cn=ldaptestuser,cn=users," + self.base_dn)
        m["userAccountControl"] = MessageElement(str(UF_WORKSTATION_TRUST_ACCOUNT),
                                                 FLAG_MOD_REPLACE, "userAccountControl")
        ldb.modify(m)

        res1 = ldb.search("cn=ldaptestuser,cn=users," + self.base_dn,
                          scope=SCOPE_BASE,
                          attrs=["isCriticalSystemObject"])
        self.assertTrue(len(res1) == 1)
        self.assertTrue("isCriticalSystemObject" in res1[0])
        self.assertEqual(str(res1[0]["isCriticalSystemObject"][0]), "FALSE")

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

    def test_isCriticalSystemObject(self):
        """Test the isCriticalSystemObject behaviour"""
        print("Testing isCriticalSystemObject behaviour\n")

        # Add tests

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

        res1 = ldb.search("cn=ldaptestcomputer,cn=computers," + self.base_dn,
                          scope=SCOPE_BASE,
                          attrs=["isCriticalSystemObject"])
        self.assertTrue(len(res1) == 1)
        self.assertTrue("isCriticalSystemObject" in res1[0])
        self.assertEqual(str(res1[0]["isCriticalSystemObject"][0]), "FALSE")

        delete_force(self.ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)

        ldb.add({
            "dn": "cn=ldaptestcomputer,cn=computers," + self.base_dn,
            "objectclass": "computer",
            "userAccountControl": str(UF_WORKSTATION_TRUST_ACCOUNT)})

        res1 = ldb.search("cn=ldaptestcomputer,cn=computers," + self.base_dn,
                          scope=SCOPE_BASE,
                          attrs=["isCriticalSystemObject"])
        self.assertTrue(len(res1) == 1)
        self.assertEqual(str(res1[0]["isCriticalSystemObject"][0]), "FALSE")

        delete_force(self.ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)

        ldb.add({
            "dn": "cn=ldaptestcomputer,cn=computers," + self.base_dn,
            "objectclass": "computer",
            "userAccountControl": str(UF_WORKSTATION_TRUST_ACCOUNT | UF_PARTIAL_SECRETS_ACCOUNT)})

        res1 = ldb.search("cn=ldaptestcomputer,cn=computers," + self.base_dn,
                          scope=SCOPE_BASE,
                          attrs=["isCriticalSystemObject"])
        self.assertTrue(len(res1) == 1)
        self.assertEqual(str(res1[0]["isCriticalSystemObject"][0]), "TRUE")

        delete_force(self.ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)

        ldb.add({
            "dn": "cn=ldaptestcomputer,cn=computers," + self.base_dn,
            "objectclass": "computer",
            "userAccountControl": str(UF_SERVER_TRUST_ACCOUNT)})

        res1 = ldb.search("cn=ldaptestcomputer,cn=computers," + self.base_dn,
                          scope=SCOPE_BASE,
                          attrs=["isCriticalSystemObject"])
        self.assertTrue(len(res1) == 1)
        self.assertEqual(str(res1[0]["isCriticalSystemObject"][0]), "TRUE")

        # Modification tests

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)
        m["userAccountControl"] = MessageElement(str(UF_NORMAL_ACCOUNT | UF_PASSWD_NOTREQD),
                                                 FLAG_MOD_REPLACE, "userAccountControl")
        ldb.modify(m)

        res1 = ldb.search("cn=ldaptestcomputer,cn=computers," + self.base_dn,
                          scope=SCOPE_BASE,
                          attrs=["isCriticalSystemObject"])
        self.assertTrue(len(res1) == 1)
        self.assertEqual(str(res1[0]["isCriticalSystemObject"][0]), "TRUE")

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)
        m["userAccountControl"] = MessageElement(str(UF_WORKSTATION_TRUST_ACCOUNT),
                                                 FLAG_MOD_REPLACE, "userAccountControl")
        ldb.modify(m)

        res1 = ldb.search("cn=ldaptestcomputer,cn=computers," + self.base_dn,
                          scope=SCOPE_BASE,
                          attrs=["isCriticalSystemObject"])
        self.assertTrue(len(res1) == 1)
        self.assertEqual(str(res1[0]["isCriticalSystemObject"][0]), "FALSE")

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)
        m["userAccountControl"] = MessageElement(
            str(UF_WORKSTATION_TRUST_ACCOUNT | UF_PARTIAL_SECRETS_ACCOUNT),
            FLAG_MOD_REPLACE, "userAccountControl")
        ldb.modify(m)

        res1 = ldb.search("cn=ldaptestcomputer,cn=computers," + self.base_dn,
                          scope=SCOPE_BASE,
                          attrs=["isCriticalSystemObject"])
        self.assertTrue(len(res1) == 1)
        self.assertEqual(str(res1[0]["isCriticalSystemObject"][0]), "TRUE")

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)
        m["userAccountControl"] = MessageElement(str(UF_NORMAL_ACCOUNT | UF_PASSWD_NOTREQD),
                                                 FLAG_MOD_REPLACE, "userAccountControl")
        ldb.modify(m)

        res1 = ldb.search("cn=ldaptestcomputer,cn=computers," + self.base_dn,
                          scope=SCOPE_BASE,
                          attrs=["isCriticalSystemObject"])
        self.assertTrue(len(res1) == 1)
        self.assertEqual(str(res1[0]["isCriticalSystemObject"][0]), "TRUE")

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)
        m["userAccountControl"] = MessageElement(str(UF_SERVER_TRUST_ACCOUNT),
                                                 FLAG_MOD_REPLACE, "userAccountControl")
        ldb.modify(m)

        res1 = ldb.search("cn=ldaptestcomputer,cn=computers," + self.base_dn,
                          scope=SCOPE_BASE,
                          attrs=["isCriticalSystemObject"])
        self.assertTrue(len(res1) == 1)
        self.assertEqual(str(res1[0]["isCriticalSystemObject"][0]), "TRUE")

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)
        m["userAccountControl"] = MessageElement(str(UF_WORKSTATION_TRUST_ACCOUNT),
                                                 FLAG_MOD_REPLACE, "userAccountControl")
        ldb.modify(m)

        res1 = ldb.search("cn=ldaptestcomputer,cn=computers," + self.base_dn,
                          scope=SCOPE_BASE,
                          attrs=["isCriticalSystemObject"])
        self.assertTrue(len(res1) == 1)
        self.assertEqual(str(res1[0]["isCriticalSystemObject"][0]), "FALSE")

        delete_force(self.ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)

    def test_service_principal_name_updates(self):
        """Test the servicePrincipalNames update behaviour"""
        print("Testing servicePrincipalNames update behaviour\n")

        ldb.add({
            "dn": "cn=ldaptestcomputer,cn=computers," + self.base_dn,
            "objectclass": "computer",
            "dNSHostName": "testname.testdom"})

        res = ldb.search("cn=ldaptestcomputer,cn=computers," + self.base_dn,
                         scope=SCOPE_BASE, attrs=["servicePrincipalName"])
        self.assertTrue(len(res) == 1)
        self.assertFalse("servicePrincipalName" in res[0])

        delete_force(self.ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)

        ldb.add({
            "dn": "cn=ldaptestcomputer,cn=computers," + self.base_dn,
            "objectclass": "computer",
            "servicePrincipalName": "HOST/testname.testdom"})

        res = ldb.search("cn=ldaptestcomputer,cn=computers," + self.base_dn,
                         scope=SCOPE_BASE, attrs=["dNSHostName"])
        self.assertTrue(len(res) == 1)
        self.assertFalse("dNSHostName" in res[0])

        delete_force(self.ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)

        ldb.add({
            "dn": "cn=ldaptestcomputer,cn=computers," + self.base_dn,
            "objectclass": "computer",
            "dNSHostName": "testname2.testdom",
            "servicePrincipalName": "HOST/testname.testdom"})

        res = ldb.search("cn=ldaptestcomputer,cn=computers," + self.base_dn,
                         scope=SCOPE_BASE, attrs=["dNSHostName"])
        self.assertTrue(len(res) == 1)
        self.assertEqual(str(res[0]["dNSHostName"][0]), "testname2.testdom")

        res = ldb.search("cn=ldaptestcomputer,cn=computers," + self.base_dn,
                         scope=SCOPE_BASE, attrs=["servicePrincipalName"])
        self.assertTrue(len(res) == 1)
        self.assertEqual(str(res[0]["servicePrincipalName"][0]),
                          "HOST/testname.testdom")

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)
        m["dNSHostName"] = MessageElement("testname.testdoM",
                                          FLAG_MOD_REPLACE, "dNSHostName")
        ldb.modify(m)

        res = ldb.search("cn=ldaptestcomputer,cn=computers," + self.base_dn,
                         scope=SCOPE_BASE, attrs=["servicePrincipalName"])
        self.assertTrue(len(res) == 1)
        self.assertEqual(str(res[0]["servicePrincipalName"][0]),
                          "HOST/testname.testdom")

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)
        m["dNSHostName"] = MessageElement("testname2.testdom2",
                                          FLAG_MOD_REPLACE, "dNSHostName")
        ldb.modify(m)

        res = ldb.search("cn=ldaptestcomputer,cn=computers," + self.base_dn,
                         scope=SCOPE_BASE, attrs=["servicePrincipalName"])
        self.assertTrue(len(res) == 1)
        self.assertEqual(str(res[0]["servicePrincipalName"][0]),
                          "HOST/testname2.testdom2")

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)
        m["dNSHostName"] = MessageElement([],
                                          FLAG_MOD_DELETE, "dNSHostName")
        ldb.modify(m)

        res = ldb.search("cn=ldaptestcomputer,cn=computers," + self.base_dn,
                         scope=SCOPE_BASE, attrs=["servicePrincipalName"])
        self.assertTrue(len(res) == 1)
        self.assertEqual(str(res[0]["servicePrincipalName"][0]),
                          "HOST/testname2.testdom2")

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)
        m["dNSHostName"] = MessageElement("testname.testdom3",
                                          FLAG_MOD_REPLACE, "dNSHostName")
        ldb.modify(m)

        res = ldb.search("cn=ldaptestcomputer,cn=computers," + self.base_dn,
                         scope=SCOPE_BASE, attrs=["servicePrincipalName"])
        self.assertTrue(len(res) == 1)
        self.assertEqual(str(res[0]["servicePrincipalName"][0]),
                          "HOST/testname2.testdom2")

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)
        m["dNSHostName"] = MessageElement("testname2.testdom2",
                                          FLAG_MOD_REPLACE, "dNSHostName")
        ldb.modify(m)

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)
        m["dNSHostName"] = MessageElement("testname3.testdom3",
                                          FLAG_MOD_REPLACE, "dNSHostName")
        m["servicePrincipalName"] = MessageElement("HOST/testname2.testdom2",
                                                   FLAG_MOD_REPLACE,
                                                   "servicePrincipalName")
        ldb.modify(m)

        res = ldb.search("cn=ldaptestcomputer,cn=computers," + self.base_dn,
                         scope=SCOPE_BASE, attrs=["servicePrincipalName"])
        self.assertTrue(len(res) == 1)
        self.assertEqual(str(res[0]["servicePrincipalName"][0]),
                          "HOST/testname3.testdom3")

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)
        m["servicePrincipalName"] = MessageElement("HOST/testname2.testdom2",
                                                   FLAG_MOD_REPLACE,
                                                   "servicePrincipalName")
        m["dNSHostName"] = MessageElement("testname4.testdom4",
                                          FLAG_MOD_REPLACE, "dNSHostName")
        ldb.modify(m)

        res = ldb.search("cn=ldaptestcomputer,cn=computers," + self.base_dn,
                         scope=SCOPE_BASE, attrs=["servicePrincipalName"])
        self.assertTrue(len(res) == 1)
        self.assertEqual(str(res[0]["servicePrincipalName"][0]),
                          "HOST/testname2.testdom2")

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)
        m["servicePrincipalName"] = MessageElement([],
                                                   FLAG_MOD_DELETE,
                                                   "servicePrincipalName")
        ldb.modify(m)

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)
        m["dNSHostName"] = MessageElement("testname2.testdom2",
                                          FLAG_MOD_REPLACE, "dNSHostName")
        ldb.modify(m)

        res = ldb.search("cn=ldaptestcomputer,cn=computers," + self.base_dn,
                         scope=SCOPE_BASE, attrs=["servicePrincipalName"])
        self.assertTrue(len(res) == 1)
        self.assertFalse("servicePrincipalName" in res[0])

        delete_force(self.ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)

        ldb.add({
            "dn": "cn=ldaptestcomputer,cn=computers," + self.base_dn,
            "objectclass": "computer",
            "sAMAccountName": "testname$"})

        res = ldb.search("cn=ldaptestcomputer,cn=computers," + self.base_dn,
                         scope=SCOPE_BASE, attrs=["servicePrincipalName"])
        self.assertTrue(len(res) == 1)
        self.assertFalse("servicePrincipalName" in res[0])

        delete_force(self.ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)

        ldb.add({
            "dn": "cn=ldaptestcomputer,cn=computers," + self.base_dn,
            "objectclass": "computer",
            "servicePrincipalName": "HOST/testname"})

        res = ldb.search("cn=ldaptestcomputer,cn=computers," + self.base_dn,
                         scope=SCOPE_BASE, attrs=["sAMAccountName"])
        self.assertTrue(len(res) == 1)
        self.assertTrue("sAMAccountName" in res[0])

        delete_force(self.ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)

        ldb.add({
            "dn": "cn=ldaptestcomputer,cn=computers," + self.base_dn,
            "objectclass": "computer",
            "sAMAccountName": "testname$",
            "servicePrincipalName": "HOST/testname"})

        res = ldb.search("cn=ldaptestcomputer,cn=computers," + self.base_dn,
                         scope=SCOPE_BASE, attrs=["sAMAccountName"])
        self.assertTrue(len(res) == 1)
        self.assertEqual(str(res[0]["sAMAccountName"][0]), "testname$")

        res = ldb.search("cn=ldaptestcomputer,cn=computers," + self.base_dn,
                         scope=SCOPE_BASE, attrs=["servicePrincipalName"])
        self.assertTrue(len(res) == 1)
        self.assertEqual(str(res[0]["servicePrincipalName"][0]),
                          "HOST/testname")

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)
        m["sAMAccountName"] = MessageElement("testnamE$",
                                             FLAG_MOD_REPLACE, "sAMAccountName")
        ldb.modify(m)

        res = ldb.search("cn=ldaptestcomputer,cn=computers," + self.base_dn,
                         scope=SCOPE_BASE, attrs=["servicePrincipalName"])
        self.assertTrue(len(res) == 1)
        self.assertEqual(str(res[0]["servicePrincipalName"][0]),
                          "HOST/testname")

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)
        m["sAMAccountName"] = MessageElement("testname",
                                             FLAG_MOD_REPLACE, "sAMAccountName")
        ldb.modify(m)

        res = ldb.search("cn=ldaptestcomputer,cn=computers," + self.base_dn,
                         scope=SCOPE_BASE, attrs=["servicePrincipalName"])
        self.assertTrue(len(res) == 1)
        self.assertEqual(str(res[0]["servicePrincipalName"][0]),
                          "HOST/testname")

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)
        m["sAMAccountName"] = MessageElement("test$name$",
                                             FLAG_MOD_REPLACE, "sAMAccountName")
        ldb.modify(m)

        res = ldb.search("cn=ldaptestcomputer,cn=computers," + self.base_dn,
                         scope=SCOPE_BASE, attrs=["servicePrincipalName"])
        self.assertTrue(len(res) == 1)
        self.assertEqual(str(res[0]["servicePrincipalName"][0]),
                          "HOST/test$name")

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)
        m["sAMAccountName"] = MessageElement("testname2",
                                             FLAG_MOD_REPLACE, "sAMAccountName")
        ldb.modify(m)

        res = ldb.search("cn=ldaptestcomputer,cn=computers," + self.base_dn,
                         scope=SCOPE_BASE, attrs=["servicePrincipalName"])
        self.assertTrue(len(res) == 1)
        self.assertEqual(str(res[0]["servicePrincipalName"][0]),
                          "HOST/testname2")

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)
        m["sAMAccountName"] = MessageElement("testname3",
                                             FLAG_MOD_REPLACE, "sAMAccountName")
        m["servicePrincipalName"] = MessageElement("HOST/testname2",
                                                   FLAG_MOD_REPLACE,
                                                   "servicePrincipalName")
        ldb.modify(m)

        res = ldb.search("cn=ldaptestcomputer,cn=computers," + self.base_dn,
                         scope=SCOPE_BASE, attrs=["servicePrincipalName"])
        self.assertTrue(len(res) == 1)
        self.assertEqual(str(res[0]["servicePrincipalName"][0]),
                          "HOST/testname3")

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)
        m["servicePrincipalName"] = MessageElement("HOST/testname2",
                                                   FLAG_MOD_REPLACE,
                                                   "servicePrincipalName")
        m["sAMAccountName"] = MessageElement("testname4",
                                             FLAG_MOD_REPLACE, "sAMAccountName")
        ldb.modify(m)

        res = ldb.search("cn=ldaptestcomputer,cn=computers," + self.base_dn,
                         scope=SCOPE_BASE, attrs=["servicePrincipalName"])
        self.assertTrue(len(res) == 1)
        self.assertEqual(str(res[0]["servicePrincipalName"][0]),
                          "HOST/testname2")

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)
        m["servicePrincipalName"] = MessageElement([],
                                                   FLAG_MOD_DELETE,
                                                   "servicePrincipalName")
        ldb.modify(m)

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)
        m["sAMAccountName"] = MessageElement("testname2",
                                             FLAG_MOD_REPLACE, "sAMAccountName")
        ldb.modify(m)

        res = ldb.search("cn=ldaptestcomputer,cn=computers," + self.base_dn,
                         scope=SCOPE_BASE, attrs=["servicePrincipalName"])
        self.assertTrue(len(res) == 1)
        self.assertFalse("servicePrincipalName" in res[0])

        delete_force(self.ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)

        ldb.add({
            "dn": "cn=ldaptestcomputer,cn=computers," + self.base_dn,
            "objectclass": "computer",
            "dNSHostName": "testname.testdom",
            "sAMAccountName": "testname$",
            "servicePrincipalName": ["HOST/testname.testdom", "HOST/testname"]
        })

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)
        m["dNSHostName"] = MessageElement("testname2.testdom",
                                          FLAG_MOD_REPLACE, "dNSHostName")
        m["sAMAccountName"] = MessageElement("testname2$",
                                             FLAG_MOD_REPLACE, "sAMAccountName")
        ldb.modify(m)

        res = ldb.search("cn=ldaptestcomputer,cn=computers," + self.base_dn,
                         scope=SCOPE_BASE, attrs=["dNSHostName", "sAMAccountName", "servicePrincipalName"])
        self.assertTrue(len(res) == 1)
        self.assertEqual(str(res[0]["dNSHostName"][0]), "testname2.testdom")
        self.assertEqual(str(res[0]["sAMAccountName"][0]), "testname2$")
        self.assertTrue(str(res[0]["servicePrincipalName"][0]) == "HOST/testname2" or
                        str(res[0]["servicePrincipalName"][1]) == "HOST/testname2")
        self.assertTrue(str(res[0]["servicePrincipalName"][0]) == "HOST/testname2.testdom" or
                        str(res[0]["servicePrincipalName"][1]) == "HOST/testname2.testdom")

        delete_force(self.ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)

        ldb.add({
            "dn": "cn=ldaptestcomputer,cn=computers," + self.base_dn,
            "objectclass": "computer",
            "dNSHostName": "testname.testdom",
            "sAMAccountName": "testname$",
            "servicePrincipalName": ["HOST/testname.testdom", "HOST/testname"]
        })

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)
        m["sAMAccountName"] = MessageElement("testname2$",
                                             FLAG_MOD_REPLACE, "sAMAccountName")
        m["dNSHostName"] = MessageElement("testname2.testdom",
                                          FLAG_MOD_REPLACE, "dNSHostName")
        ldb.modify(m)

        res = ldb.search("cn=ldaptestcomputer,cn=computers," + self.base_dn,
                         scope=SCOPE_BASE, attrs=["dNSHostName", "sAMAccountName", "servicePrincipalName"])
        self.assertTrue(len(res) == 1)
        self.assertEqual(str(res[0]["dNSHostName"][0]), "testname2.testdom")
        self.assertEqual(str(res[0]["sAMAccountName"][0]), "testname2$")
        self.assertTrue(len(res[0]["servicePrincipalName"]) == 2)
        self.assertTrue("HOST/testname2" in [str(x) for x in res[0]["servicePrincipalName"]])
        self.assertTrue("HOST/testname2.testdom" in [str(x) for x in res[0]["servicePrincipalName"]])

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)
        m["servicePrincipalName"] = MessageElement("HOST/testname2.testdom",
                                                   FLAG_MOD_ADD, "servicePrincipalName")
        try:
            ldb.modify(m)
            self.fail()
        except LdbError as e77:
            (num, _) = e77.args
            self.assertEqual(num, ERR_ATTRIBUTE_OR_VALUE_EXISTS)

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)
        m["servicePrincipalName"] = MessageElement("HOST/testname3",
                                                   FLAG_MOD_ADD, "servicePrincipalName")
        ldb.modify(m)

        res = ldb.search("cn=ldaptestcomputer,cn=computers," + self.base_dn,
                         scope=SCOPE_BASE, attrs=["dNSHostName", "sAMAccountName", "servicePrincipalName"])
        self.assertTrue(len(res) == 1)
        self.assertEqual(str(res[0]["dNSHostName"][0]), "testname2.testdom")
        self.assertEqual(str(res[0]["sAMAccountName"][0]), "testname2$")
        self.assertTrue(len(res[0]["servicePrincipalName"]) == 3)
        self.assertTrue("HOST/testname2" in [str(x) for x in res[0]["servicePrincipalName"]])
        self.assertTrue("HOST/testname3" in [str(x) for x in res[0]["servicePrincipalName"]])
        self.assertTrue("HOST/testname2.testdom" in [str(x) for x in res[0]["servicePrincipalName"]])

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)
        m["dNSHostName"] = MessageElement("testname3.testdom",
                                          FLAG_MOD_REPLACE, "dNSHostName")
        m["servicePrincipalName"] = MessageElement("HOST/testname3.testdom",
                                                   FLAG_MOD_ADD, "servicePrincipalName")
        ldb.modify(m)

        res = ldb.search("cn=ldaptestcomputer,cn=computers," + self.base_dn,
                         scope=SCOPE_BASE, attrs=["dNSHostName", "sAMAccountName", "servicePrincipalName"])
        self.assertTrue(len(res) == 1)
        self.assertEqual(str(res[0]["dNSHostName"][0]), "testname3.testdom")
        self.assertEqual(str(res[0]["sAMAccountName"][0]), "testname2$")
        self.assertTrue(len(res[0]["servicePrincipalName"]) == 3)
        self.assertTrue("HOST/testname2" in [str(x) for x in res[0]["servicePrincipalName"]])
        self.assertTrue("HOST/testname3" in [str(x) for x in res[0]["servicePrincipalName"]])
        self.assertTrue("HOST/testname3.testdom" in [str(x) for x in res[0]["servicePrincipalName"]])

        delete_force(self.ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)

    def test_service_principal_name_uniqueness(self):
        """Test the servicePrincipalName uniqueness behaviour"""
        print("Testing servicePrincipalName uniqueness behaviour")

        ldb.add({
            "dn": "cn=ldaptestcomputer,cn=computers," + self.base_dn,
            "objectclass": "computer",
            "servicePrincipalName": "HOST/testname.testdom"})

        try:
            ldb.add({
                "dn": "cn=ldaptestcomputer2,cn=computers," + self.base_dn,
                "objectclass": "computer",
                "servicePrincipalName": "HOST/testname.testdom"})
        except LdbError as e:
            num, _ = e.args
            self.assertEqual(num, ERR_CONSTRAINT_VIOLATION)
        else:
            self.fail()

    def test_sam_description_attribute(self):
        """Test SAM description attribute"""
        print("Test SAM description attribute")

        self.ldb.add({
            "dn": "cn=ldaptestgroup,cn=users," + self.base_dn,
            "objectclass": "group",
            "description": "desc1"
        })

        res = ldb.search("cn=ldaptestgroup,cn=users," + self.base_dn,
                         scope=SCOPE_BASE, attrs=["description"])
        self.assertTrue(len(res) == 1)
        self.assertTrue("description" in res[0])
        self.assertTrue(len(res[0]["description"]) == 1)
        self.assertEqual(str(res[0]["description"][0]), "desc1")

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

        self.ldb.add({
            "dn": "cn=ldaptestgroup,cn=users," + self.base_dn,
            "objectclass": "group",
            "description": ["desc1", "desc2"]})

        res = ldb.search("cn=ldaptestgroup,cn=users," + self.base_dn,
                         scope=SCOPE_BASE, attrs=["description"])
        self.assertTrue(len(res) == 1)
        self.assertTrue("description" in res[0])
        self.assertTrue(len(res[0]["description"]) == 2)
        self.assertTrue(str(res[0]["description"][0]) == "desc1" or
                        str(res[0]["description"][1]) == "desc1")
        self.assertTrue(str(res[0]["description"][0]) == "desc2" or
                        str(res[0]["description"][1]) == "desc2")

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestgroup,cn=users," + self.base_dn)
        m["description"] = MessageElement(["desc1", "desc2"], FLAG_MOD_REPLACE,
                                          "description")
        try:
            ldb.modify(m)
            self.fail()
        except LdbError as e78:
            (num, _) = e78.args
            self.assertEqual(num, ERR_ATTRIBUTE_OR_VALUE_EXISTS)

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestgroup,cn=users," + self.base_dn)
        m["description"] = MessageElement(["desc1", "desc2"], FLAG_MOD_DELETE,
                                          "description")
        ldb.modify(m)

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

        self.ldb.add({
            "dn": "cn=ldaptestgroup,cn=users," + self.base_dn,
            "objectclass": "group"})

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestgroup,cn=users," + self.base_dn)
        m["description"] = MessageElement("desc1", FLAG_MOD_REPLACE,
                                          "description")
        ldb.modify(m)

        res = ldb.search("cn=ldaptestgroup,cn=users," + self.base_dn,
                         scope=SCOPE_BASE, attrs=["description"])
        self.assertTrue(len(res) == 1)
        self.assertTrue("description" in res[0])
        self.assertTrue(len(res[0]["description"]) == 1)
        self.assertEqual(str(res[0]["description"][0]), "desc1")

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

        self.ldb.add({
            "dn": "cn=ldaptestgroup,cn=users," + self.base_dn,
            "objectclass": "group",
            "description": ["desc1", "desc2"]})

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestgroup,cn=users," + self.base_dn)
        m["description"] = MessageElement("desc1", FLAG_MOD_REPLACE,
                                          "description")
        ldb.modify(m)

        res = ldb.search("cn=ldaptestgroup,cn=users," + self.base_dn,
                         scope=SCOPE_BASE, attrs=["description"])
        self.assertTrue(len(res) == 1)
        self.assertTrue("description" in res[0])
        self.assertTrue(len(res[0]["description"]) == 1)
        self.assertEqual(str(res[0]["description"][0]), "desc1")

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestgroup,cn=users," + self.base_dn)
        m["description"] = MessageElement("desc3", FLAG_MOD_ADD,
                                          "description")
        try:
            ldb.modify(m)
            self.fail()
        except LdbError as e79:
            (num, _) = e79.args
            self.assertEqual(num, ERR_ATTRIBUTE_OR_VALUE_EXISTS)

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestgroup,cn=users," + self.base_dn)
        m["description"] = MessageElement(["desc1", "desc2"], FLAG_MOD_DELETE,
                                          "description")
        try:
            ldb.modify(m)
            self.fail()
        except LdbError as e80:
            (num, _) = e80.args
            self.assertEqual(num, ERR_NO_SUCH_ATTRIBUTE)

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestgroup,cn=users," + self.base_dn)
        m["description"] = MessageElement("desc1", FLAG_MOD_DELETE,
                                          "description")
        ldb.modify(m)
        res = ldb.search("cn=ldaptestgroup,cn=users," + self.base_dn,
                         scope=SCOPE_BASE, attrs=["description"])
        self.assertTrue(len(res) == 1)
        self.assertFalse("description" in res[0])

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestgroup,cn=users," + self.base_dn)
        m["description"] = MessageElement(["desc1", "desc2"], FLAG_MOD_REPLACE,
                                          "description")
        try:
            ldb.modify(m)
            self.fail()
        except LdbError as e81:
            (num, _) = e81.args
            self.assertEqual(num, ERR_ATTRIBUTE_OR_VALUE_EXISTS)

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestgroup,cn=users," + self.base_dn)
        m["description"] = MessageElement(["desc3", "desc4"], FLAG_MOD_ADD,
                                          "description")
        try:
            ldb.modify(m)
            self.fail()
        except LdbError as e82:
            (num, _) = e82.args
            self.assertEqual(num, ERR_ATTRIBUTE_OR_VALUE_EXISTS)

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestgroup,cn=users," + self.base_dn)
        m["description"] = MessageElement("desc1", FLAG_MOD_ADD,
                                          "description")
        ldb.modify(m)

        res = ldb.search("cn=ldaptestgroup,cn=users," + self.base_dn,
                         scope=SCOPE_BASE, attrs=["description"])
        self.assertTrue(len(res) == 1)
        self.assertTrue("description" in res[0])
        self.assertTrue(len(res[0]["description"]) == 1)
        self.assertEqual(str(res[0]["description"][0]), "desc1")

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestgroup,cn=users," + self.base_dn)
        m.add(MessageElement("desc1", FLAG_MOD_DELETE, "description"))
        m.add(MessageElement("desc2", FLAG_MOD_ADD, "description"))
        ldb.modify(m)

        res = ldb.search("cn=ldaptestgroup,cn=users," + self.base_dn,
                         scope=SCOPE_BASE, attrs=["description"])
        self.assertTrue(len(res) == 1)
        self.assertTrue("description" in res[0])
        self.assertTrue(len(res[0]["description"]) == 1)
        self.assertEqual(str(res[0]["description"][0]), "desc2")

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

    def test_fSMORoleOwner_attribute(self):
        """Test fSMORoleOwner attribute"""
        print("Test fSMORoleOwner attribute")

        ds_service_name = self.ldb.get_dsServiceName()

        # The "fSMORoleOwner" attribute can only be set to "nTDSDSA" entries,
        # invalid DNs return ERR_UNWILLING_TO_PERFORM

        try:
            self.ldb.add({
                "dn": "cn=ldaptestgroup,cn=users," + self.base_dn,
                "objectclass": "group",
                "fSMORoleOwner": self.base_dn})
            self.fail()
        except LdbError as e83:
            (num, _) = e83.args
            self.assertEqual(num, ERR_UNWILLING_TO_PERFORM)

        try:
            self.ldb.add({
                "dn": "cn=ldaptestgroup,cn=users," + self.base_dn,
                "objectclass": "group",
                "fSMORoleOwner": []})
            self.fail()
        except LdbError as e84:
            (num, _) = e84.args
            self.assertEqual(num, ERR_CONSTRAINT_VIOLATION)

        # We are able to set it to a valid "nTDSDSA" entry if the server is
        # capable of handling the role

        self.ldb.add({
            "dn": "cn=ldaptestgroup,cn=users," + self.base_dn,
            "objectclass": "group",
            "fSMORoleOwner": ds_service_name})

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

        self.ldb.add({
            "dn": "cn=ldaptestgroup,cn=users," + self.base_dn,
            "objectclass": "group"})

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestgroup,cn=users," + self.base_dn)
        m.add(MessageElement(self.base_dn, FLAG_MOD_REPLACE, "fSMORoleOwner"))
        try:
            ldb.modify(m)
            self.fail()
        except LdbError as e85:
            (num, _) = e85.args
            self.assertEqual(num, ERR_UNWILLING_TO_PERFORM)

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestgroup,cn=users," + self.base_dn)
        m.add(MessageElement([], FLAG_MOD_REPLACE, "fSMORoleOwner"))
        try:
            ldb.modify(m)
            self.fail()
        except LdbError as e86:
            (num, _) = e86.args
            self.assertEqual(num, ERR_UNWILLING_TO_PERFORM)

        # We are able to set it to a valid "nTDSDSA" entry if the server is
        # capable of handling the role

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestgroup,cn=users," + self.base_dn)
        m.add(MessageElement(ds_service_name, FLAG_MOD_REPLACE, "fSMORoleOwner"))
        ldb.modify(m)

        # A clean-out works on plain entries, not master (schema, PDC...) DNs

        m = Message()
        m.dn = Dn(ldb, "cn=ldaptestgroup,cn=users," + self.base_dn)
        m.add(MessageElement([], FLAG_MOD_DELETE, "fSMORoleOwner"))
        ldb.modify(m)

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

    def test_protected_sid_objects(self):
        """Test deletion of objects with RID < 1000"""
        # a list of some well-known sids
        # objects in Builtin are already covered by objectclass
        protected_list = [
            ["CN=Domain Admins", "CN=Users,"],
            ["CN=Schema Admins", "CN=Users,"],
            ["CN=Enterprise Admins", "CN=Users,"],
            ["CN=Administrator", "CN=Users,"],
            ["CN=Domain Controllers", "CN=Users,"],
            ["CN=Protected Users", "CN=Users,"],
        ]

        for pr_object in protected_list:
            try:
                self.ldb.delete(pr_object[0] + "," + pr_object[1] + self.base_dn)
            except LdbError as e7:
                (num, _) = e7.args
                self.assertEqual(num, ERR_OTHER)
            else:
                self.fail("Deleted " + pr_object[0])

            try:
                self.ldb.rename(pr_object[0] + "," + pr_object[1] + self.base_dn,
                                pr_object[0] + "2," + pr_object[1] + self.base_dn)
            except LdbError as e8:
                (num, _) = e8.args
                self.fail("Could not rename " + pr_object[0])

            self.ldb.rename(pr_object[0] + "2," + pr_object[1] + self.base_dn,
                            pr_object[0] + "," + pr_object[1] + self.base_dn)

    def test_new_user_default_attributes(self):
        """Test default attributes for new user objects"""
        print("Test default attributes for new User objects\n")

        user_name = "ldaptestuser"
        user_dn = "CN=%s,CN=Users,%s" % (user_name, self.base_dn)
        ldb.add({
            "dn": user_dn,
            "objectclass": "user",
            "sAMAccountName": user_name})

        res = ldb.search(user_dn, scope=SCOPE_BASE)
        self.assertTrue(len(res) == 1)
        user_obj = res[0]

        expected_attrs = {"primaryGroupID": MessageElement(["513"]),
                          "logonCount": MessageElement(["0"]),
                          "cn": MessageElement([user_name]),
                          "countryCode": MessageElement(["0"]),
                          "objectClass": MessageElement(["top", "person", "organizationalPerson", "user"]),
                          "instanceType": MessageElement(["4"]),
                          "distinguishedName": MessageElement([user_dn]),
                          "sAMAccountType": MessageElement(["805306368"]),
                          "objectSid": "**SKIP**",
                          "whenCreated": "**SKIP**",
                          "uSNCreated": "**SKIP**",
                          "badPasswordTime": MessageElement(["0"]),
                          "dn": Dn(ldb, user_dn),
                          "pwdLastSet": MessageElement(["0"]),
                          "sAMAccountName": MessageElement([user_name]),
                          "objectCategory": MessageElement(["CN=Person,%s" % ldb.get_schema_basedn().get_linearized()]),
                          "objectGUID": "**SKIP**",
                          "whenChanged": "**SKIP**",
                          "badPwdCount": MessageElement(["0"]),
                          "accountExpires": MessageElement(["9223372036854775807"]),
                          "name": MessageElement([user_name]),
                          "codePage": MessageElement(["0"]),
                          "userAccountControl": MessageElement(["546"]),
                          "lastLogon": MessageElement(["0"]),
                          "uSNChanged": "**SKIP**",
                          "lastLogoff": MessageElement(["0"])}
        # assert we have expected attribute names
        actual_names = set(user_obj.keys())
        # Samba does not use 'dSCorePropagationData', so skip it
        actual_names -= set(['dSCorePropagationData'])
        self.assertEqual(set(expected_attrs.keys()), actual_names, "Actual object does not have expected attributes")
        # check attribute values
        for name in expected_attrs.keys():
            actual_val = user_obj.get(name)
            self.assertFalse(actual_val is None, "No value for attribute '%s'" % name)
            expected_val = expected_attrs[name]
            if expected_val == "**SKIP**":
                # "**ANY**" values means "any"
                continue
            self.assertEqual(expected_val, actual_val,
                             "Unexpected value[%r] for '%s' expected[%r]" %
                             (actual_val, name, expected_val))
        # clean up
        delete_force(self.ldb, "cn=ldaptestuser,cn=users," + self.base_dn)


if "://" not in host:
    if os.path.isfile(host):
        host = "tdb://%s" % host
    else:
        host = "ldap://%s" % host

ldb = SamDB(host, credentials=creds, session_info=system_session(lp), lp=lp)

TestProgram(module=__name__, opts=subunitopts)
