def test_computer_account_bind(self):
        # create a computer acocount for the test
        delete_force(self.ldb, self.computer_dn)
        self.ldb.add_ldif("""
dn: """ + self.computer_dn + """
cn: CENTOS53
displayName: CENTOS53$
name: CENTOS53
sAMAccountName: CENTOS53$
countryCode: 0
objectClass: computer
objectClass: organizationalPerson
objectClass: person
objectClass: top
objectClass: user
codePage: 0
userAccountControl: 4096
dNSHostName: centos53.alabala.test
operatingSystemVersion: 5.2 (3790)
operatingSystem: Windows Server 2003
""")
        self.ldb.modify_ldif("""
dn: """ + self.computer_dn + """
changetype: modify
replace: unicodePwd
unicodePwd:: """ + base64.b64encode("\"[email protected]\"".encode('utf-16-le')) + """
""")

        # do a simple bind and search with the machine account
        creds_machine.set_bind_dn(self.computer_dn)
        creds_machine.set_password(self.password)
        print "BindTest with: " + creds_machine.get_bind_dn()
        ldb_machine = samba.tests.connect_samdb(host, credentials=creds_machine,
                                                lp=lp, ldap_only=True)
        res = ldb_machine.search(base="", expression="", scope=SCOPE_BASE, attrs=["*"])

    def test_dirsync_deleted_items(self):
        """Check that dirsync returnd deleted objects too"""
        # Let's create an OU
        ouname="OU=testou3,%s" % self.base_dn
        self.ouname = ouname
        self.ldb_admin.create_ou(ouname)
        res = self.ldb_admin.search(self.base_dn,
                                    expression="(&(objectClass=organizationalUnit)(!(isDeleted=*)))",
                                    controls=["dirsync:1:0:1"])
        guid = None
        for e in res:
            if str(e["name"]) == "testou3":
                guid = str(ndr_unpack(misc.GUID,e.get("objectGUID")[0]))

        ctl = str(res.controls[0]).split(":")
        ctl[1] = "1"
        ctl[2] = "0"
        ctl[3] = "10000"
        control1 = str(":".join(ctl))

        # So now delete the object and check that
        # we can see the object but deleted when admin
        delete_force(self.ldb_admin, ouname)

        res = self.ldb_admin.search(self.base_dn,
                                    expression="(objectClass=organizationalUnit)",
                                    controls=[control1])
        self.assertEqual(len(res), 1)
        guid2 = str(ndr_unpack(misc.GUID,res[0].get("objectGUID")[0]))
        self.assertEqual(guid2, guid)
        self.assertTrue(res[0].get("isDeleted"))
        self.assertTrue(res[0].get("name") != None)

    def create_user_account(self):
        self.user_pass = samba.generate_random_password(32, 32)
        self.user_name = USER_NAME
        self.user_dn = "cn=%s,%s" % (self.user_name, self.ldb.domain_dn())

        # remove the account if it exists, this will happen if a previous test
        # run failed
        delete_force(self.ldb, self.user_dn)

        utf16pw = unicode(
            '"' + self.user_pass.encode('utf-8') + '"', 'utf-8'
        ).encode('utf-16-le')
        self.ldb.add({
            "dn": self.user_dn,
            "objectclass": "user",
            "sAMAccountName": "%s" % self.user_name,
            "userAccountControl": str(UF_NORMAL_ACCOUNT),
            "unicodePwd": utf16pw})

        self.user_creds = Credentials()
        self.user_creds.guess(self.get_loadparm())
        self.user_creds.set_password(self.user_pass)
        self.user_creds.set_username(self.user_name)
        self.user_creds.set_workstation(self.machine_name)
        pass

    def create_machine_account(self):
        self.machine_pass = samba.generate_random_password(32, 32)
        self.machine_name = MACHINE_NAME
        self.machine_dn = "cn=%s,%s" % (self.machine_name, self.ldb.domain_dn())

        # remove the account if it exists, this will happen if a previous test
        # run failed
        delete_force(self.ldb, self.machine_dn)

        utf16pw = unicode(
            '"' + self.machine_pass.encode('utf-8') + '"', 'utf-8'
        ).encode('utf-16-le')
        self.ldb.add({
            "dn": self.machine_dn,
            "objectclass": "computer",
            "sAMAccountName": "%s$" % self.machine_name,
            "userAccountControl":
                str(UF_WORKSTATION_TRUST_ACCOUNT | UF_PASSWD_NOTREQD),
            "unicodePwd": utf16pw})

        self.machine_creds = Credentials()
        self.machine_creds.guess(self.get_loadparm())
        self.machine_creds.set_secure_channel_type(SEC_CHAN_WKSTA)
        self.machine_creds.set_kerberos_state(DONT_USE_KERBEROS)
        self.machine_creds.set_password(self.machine_pass)
        self.machine_creds.set_username(self.machine_name + "$")
        self.machine_creds.set_workstation(self.machine_name)

    def test_u1_member_of_g4(self):
        i = 0
        import time
        print "KEYS: total add modify delete"
        
        while True:
            group = i % self.n_groups + 1
            
            start = time.time()
            self.ldb.add({
                "dn": "cn=u%d,%s" % (group, self.ou_users), "objectclass": "user"})
            end_add = time.time()

            start_mod = time.time()
            
            m = Message()
            m.dn = Dn(self.ldb, "CN=g%d,%s" % (group, self.ou_groups))
            m["member"] = MessageElement("cn=u%d,%s" % (group, self.ou_users),
                                         FLAG_MOD_ADD, "member")
            self.ldb.modify(m)
            end_mod = time.time()

            delete_force(self.ldb, "cn=u%d,%s" % (group, self.ou_users))
            end = time.time()
            print end - start, end_add - start, end_mod - start_mod, end - end_mod
            i += 1

    def setUp(self):
        super(AuthLogPassChangeTests, self).setUp()

        self.remoteAddress = os.environ["CLIENT_IP"]
        self.server_ip = os.environ["SERVER_IP"]

        host = "ldap://%s" % os.environ["SERVER"]
        self.ldb = SamDB(url=host,
                         session_info=system_session(),
                         credentials=self.get_credentials(),
                         lp=self.get_loadparm())

        print("ldb %s" % type(self.ldb))
        # Gets back the basedn
        base_dn = self.ldb.domain_dn()
        print("base_dn %s" % base_dn)

        # permit password changes during this test
        PasswordCommon.allow_password_changes(self, self.ldb)

        self.base_dn = self.ldb.domain_dn()

        # (Re)adds the test user USER_NAME with password USER_PASS
        delete_force(self.ldb, "cn=" + USER_NAME + ",cn=users," + self.base_dn)
        self.ldb.add({
            "dn": "cn=" + USER_NAME + ",cn=users," + self.base_dn,
            "objectclass": "user",
            "sAMAccountName": USER_NAME,
            "userPassword": USER_PASS
        })

        # discard any auth log messages for the password setup
        self.discardMessages()

    def test_subClassOf(self):
        """ Testing usage of custom child schamaClass
        """

        class_name = "my-Class" + time.strftime("%s", time.gmtime())
        class_ldap_display_name = class_name.replace("-", "")

        ldif = """
dn: CN=%s,%s""" % (class_name, self.schema_dn) + """
objectClass: top
objectClass: classSchema
adminDescription: """ + class_name + """
adminDisplayName: """ + class_name + """
cn: """ + class_name + """
governsId: 1.2.840.""" + str(random.randint(1,100000)) + """.1.5.9939
instanceType: 4
objectClassCategory: 1
subClassOf: organizationalUnit
systemFlags: 16
systemOnly: FALSE
"""
        self.ldb.add_ldif(ldif)

        # Search for created objectclass
        res = []
        res = self.ldb.search("cn=%s,%s" % (class_name, self.schema_dn), scope=SCOPE_BASE,
                              attrs=["lDAPDisplayName", "defaultObjectCategory",
                                     "schemaIDGUID", "distinguishedName"])
        self.assertEquals(len(res), 1)
        self.assertEquals(res[0]["lDAPDisplayName"][0], class_ldap_display_name)
        self.assertEquals(res[0]["defaultObjectCategory"][0], res[0]["distinguishedName"][0])
        self.assertTrue("schemaIDGUID" in res[0])

        ldif = """
dn:
changetype: modify
add: schemaUpdateNow
schemaUpdateNow: 1
"""
        self.ldb.modify_ldif(ldif)

        object_name = "org" + time.strftime("%s", time.gmtime())

        ldif = """
dn: OU=%s,%s""" % (object_name, self.base_dn) + """
objectClass: """ + class_ldap_display_name + """
ou: """ + object_name + """
instanceType: 4
"""
        self.ldb.add_ldif(ldif)

        # Search for created object
        res = []
        res = self.ldb.search("ou=%s,%s" % (object_name, self.base_dn), scope=SCOPE_BASE, attrs=["dn"])
        self.assertEquals(len(res), 1)
        # Delete the object
        delete_force(self.ldb, "ou=%s,%s" % (object_name, self.base_dn))

    def tearDown(self):
        self.samdb.transaction_start()

        for site in self.sites:
            delete_force(self.samdb, site, controls=['tree_delete:1'])

        for site_link in self.site_links:
            delete_force(self.samdb, site_link)

        self.samdb.transaction_commit()

    def add_user(self, options=None, clear_text=False, ldb=None):
        # set any needed options
        if options is not None:
            for (option, value) in options:
                self.lp.set(option, value)

        if ldb is None:
            self.creds = Credentials()
            self.session = system_session()
            self.creds.guess(self.lp)
            self.session = system_session()
            self.ldb = SamDB(session_info=self.session,
                             credentials=self.creds,
                             lp=self.lp)
        else:
            self.ldb = ldb

        res = self.ldb.search(base=self.ldb.get_config_basedn(),
                              expression="ncName=%s" % self.ldb.get_default_basedn(),
                              attrs=["nETBIOSName"])
        self.netbios_domain = res[0]["nETBIOSName"][0]
        self.dns_domain = self.ldb.domain_dns_name()


        # Gets back the basedn
        base_dn = self.ldb.domain_dn()

        # Gets back the configuration basedn
        configuration_dn = self.ldb.get_config_basedn().get_linearized()

        # permit password changes during this test
        PasswordCommon.allow_password_changes(self, self.ldb)

        self.base_dn = self.ldb.domain_dn()

        account_control = 0
        if clear_text:
            # Restore the current domain setting on exit.
            pwdProperties = self.ldb.get_pwdProperties()
            self.addCleanup(self.ldb.set_pwdProperties, pwdProperties)
            # Update the domain setting
            self.set_store_cleartext(clear_text)
            account_control |= UF_ENCRYPTED_TEXT_PASSWORD_ALLOWED

        # (Re)adds the test user USER_NAME with password USER_PASS
        # and userPrincipalName UPN
        delete_force(self.ldb, "cn=" + USER_NAME + ",cn=users," + self.base_dn)
        self.ldb.add({
             "dn": "cn=" + USER_NAME + ",cn=users," + self.base_dn,
             "objectclass": "user",
             "sAMAccountName": USER_NAME,
             "userPassword": USER_PASS,
             "userPrincipalName": UPN,
             "userAccountControl": str(account_control)
        })

    def setUp(self):
        super(AuthLogPassChangeTests, self).setUp()

        self.remoteAddress = os.environ["CLIENT_IP"]
        self.server_ip = os.environ["SERVER_IP"]

        host = "ldap://%s" % os.environ["SERVER"]
        self.ldb = SamDB(url=host,
                         session_info=system_session(),
                         credentials=self.get_credentials(),
                         lp=self.get_loadparm())

        print "ldb %s" % type(self.ldb)
        # Gets back the basedn
        base_dn = self.ldb.domain_dn()
        print "base_dn %s" % base_dn

        # Gets back the configuration basedn
        configuration_dn = self.ldb.get_config_basedn().get_linearized()

        # Get the old "dSHeuristics" if it was set
        dsheuristics = self.ldb.get_dsheuristics()

        # Set the "dSHeuristics" to activate the correct "userPassword"
        # behaviour
        self.ldb.set_dsheuristics("000000001")

        # Reset the "dSHeuristics" as they were before
        self.addCleanup(self.ldb.set_dsheuristics, dsheuristics)

        # Get the old "minPwdAge"
        minPwdAge = self.ldb.get_minPwdAge()

        # Set it temporarily to "0"
        self.ldb.set_minPwdAge("0")
        self.base_dn = self.ldb.domain_dn()

        # Reset the "minPwdAge" as it was before
        self.addCleanup(self.ldb.set_minPwdAge, minPwdAge)

        # (Re)adds the test user USER_NAME with password USER_PASS
        delete_force(self.ldb, "cn=" + USER_NAME + ",cn=users," + self.base_dn)
        self.ldb.add({
             "dn": "cn=" + USER_NAME + ",cn=users," + self.base_dn,
             "objectclass": "user",
             "sAMAccountName": USER_NAME,
             "userPassword": USER_PASS
        })

        # discard any auth log messages for the password setup
        self.discardMessages()

 def tearDown(self):
     super(UserAccountControlTests, self).tearDown()
     for computername in self.computernames:
         delete_force(self.admin_samdb, "CN={0!s},OU=test_computer_ou1,{1!s}".format(computername, self.base_dn))
     delete_force(self.admin_samdb, "CN=testcomputer-t,OU=test_computer_ou1,{0!s}".format((self.base_dn)))
     delete_force(self.admin_samdb, "OU=test_computer_ou1,{0!s}".format((self.base_dn)))
     delete_force(self.admin_samdb, "CN={0!s},CN=Users,{1!s}".format(self.unpriv_user, self.base_dn))

 def tearDown(self):
     super(UserAccountControlTests, self).tearDown()
     for computername in self.computernames:
         delete_force(self.admin_samdb, "CN=%s,OU=test_computer_ou1,%s" % (computername, self.base_dn))
     delete_force(self.admin_samdb, "CN=testcomputer-t,OU=test_computer_ou1,%s" % (self.base_dn))
     delete_force(self.admin_samdb, "OU=test_computer_ou1,%s" % (self.base_dn))
     delete_force(self.admin_samdb, "CN=%s,CN=Users,%s" % (self.unpriv_user, self.base_dn))

    def setUp(self):
        self.message_type = MSG_DSDB_LOG
        self.event_type   = DSDB_EVENT_NAME
        super(AuditLogDsdbTests, self).setUp()

        self.remoteAddress = os.environ["CLIENT_IP"]
        self.server_ip = os.environ["SERVER_IP"]

        host = "ldap://%s" % os.environ["SERVER"]
        self.ldb = SamDB(url=host,
                         session_info=system_session(),
                         credentials=self.get_credentials(),
                         lp=self.get_loadparm())
        self.server = os.environ["SERVER"]

        # Gets back the basedn
        self.base_dn = self.ldb.domain_dn()

        # Get the old "dSHeuristics" if it was set
        dsheuristics = self.ldb.get_dsheuristics()

        # Set the "dSHeuristics" to activate the correct "userPassword"
        # behaviour
        self.ldb.set_dsheuristics("000000001")

        # Reset the "dSHeuristics" as they were before
        self.addCleanup(self.ldb.set_dsheuristics, dsheuristics)

        # Get the old "minPwdAge"
        minPwdAge = self.ldb.get_minPwdAge()

        # Set it temporarily to "0"
        self.ldb.set_minPwdAge("0")
        self.base_dn = self.ldb.domain_dn()

        # Reset the "minPwdAge" as it was before
        self.addCleanup(self.ldb.set_minPwdAge, minPwdAge)

        # (Re)adds the test user USER_NAME with password USER_PASS
        delete_force(self.ldb, "cn=" + USER_NAME + ",cn=users," + self.base_dn)
        self.ldb.add({
            "dn": "cn=" + USER_NAME + ",cn=users," + self.base_dn,
            "objectclass": "user",
            "sAMAccountName": USER_NAME,
            "userPassword": USER_PASS
        })

    def _readd_user(self, creds, lockOutObservationWindow=0):
        username = creds.get_username()
        userpass = creds.get_password()
        userdn = "cn=%s,cn=users,%s" % (username, self.base_dn)

        delete_force(self.ldb, userdn)
        self.ldb.add({
             "dn": userdn,
             "objectclass": "user",
             "sAMAccountName": username})

        self.addCleanup(delete_force, self.ldb, userdn)

        # Sets the initial user password with a "special" password change
        # I think that this internally is a password set operation and it can
        # only be performed by someone which has password set privileges on the
        # account (at least in s4 we do handle it like that).
        self.ldb.modify_ldif("""
dn: """ + userdn + """
changetype: modify
delete: userPassword
add: userPassword
userPassword: """ + userpass + """
""")
        # Enables the user account
        self.ldb.enable_account("(sAMAccountName=%s)" % username)

        use_kerberos = creds.get_kerberos_state()
        fail_creds = self.insta_creds(self.template_creds,
                                      username=username,
                                      userpass=userpass+"X",
                                      kerberos_state=use_kerberos)
        self._check_account_initial(userdn)

        # Fail once to get a badPasswordTime
        try:
            ldb = SamDB(url=self.host_url, credentials=fail_creds, lp=self.lp)
            self.fail()
        except LdbError as e:
            (num, msg) = e.args
            self.assertEquals(num, ERR_INVALID_CREDENTIALS)

        # Succeed to reset everything to 0
        ldb = SamDB(url=self.host_url, credentials=creds, lp=self.lp)

        return ldb

 def tearDown(self):
     super(DynamicTokenTest, self).tearDown()
     delete_force(self.admin_ldb, "CN=%s,%s,%s" %
                       (self.test_user, "cn=users", self.base_dn))
     delete_force(self.admin_ldb, "CN=%s,%s,%s" %
                       (self.test_group0, "cn=users", self.base_dn))
     delete_force(self.admin_ldb, "CN=%s,%s,%s" %
                       (self.test_group1, "cn=users", self.base_dn))
     delete_force(self.admin_ldb, "CN=%s,%s,%s" %
                       (self.test_group2, "cn=users", self.base_dn))

 def tearDown(self):
     super(SimpleDirsyncTests, self).tearDown()
     delete_force(self.ldb_admin, self.get_user_dn(self.dirsync_user))
     delete_force(self.ldb_admin, self.get_user_dn(self.simple_user))
     delete_force(self.ldb_admin, self.get_user_dn(self.admin_user))
     if self.ouname:
         delete_force(self.ldb_admin, self.ouname)
     self.sd_utils.modify_sd_on_dn(self.base_dn, self.desc_sddl)
     try:
         self.ldb_admin.deletegroup("testgroup")
     except Exception:
         pass

    def setUp(self):
        super(UserAccountControlTests, self).setUp()
        self.admin_creds = creds
        self.admin_samdb = SamDB(url=ldaphost,
                                 session_info=system_session(),
                                 credentials=self.admin_creds, lp=lp)
        self.domain_sid = security.dom_sid(self.admin_samdb.get_domain_sid())
        self.base_dn = self.admin_samdb.domain_dn()

        self.unpriv_user = "testuser1"
        self.unpriv_user_pw = "[email protected]"
        self.unpriv_creds = self.get_creds(self.unpriv_user, self.unpriv_user_pw)

        delete_force(self.admin_samdb, "CN=testcomputer-t,OU=test_computer_ou1,%s" % (self.base_dn))
        delete_force(self.admin_samdb, "OU=test_computer_ou1,%s" % (self.base_dn))
        delete_force(self.admin_samdb, "CN=%s,CN=Users,%s" % (self.unpriv_user, self.base_dn))

        self.admin_samdb.newuser(self.unpriv_user, self.unpriv_user_pw)
        res = self.admin_samdb.search("CN=%s,CN=Users,%s" % (self.unpriv_user, self.admin_samdb.domain_dn()),
                                      scope=SCOPE_BASE,
                                      attrs=["objectSid"])
        self.assertEqual(1, len(res))

        self.unpriv_user_sid = ndr_unpack(security.dom_sid, res[0]["objectSid"][0])
        self.unpriv_user_dn = res[0].dn

        self.samdb = SamDB(url=ldaphost, credentials=self.unpriv_creds, lp=lp)

        self.samr = samr.samr("ncacn_ip_tcp:%s[seal]" % host, lp, self.unpriv_creds)
        self.samr_handle = self.samr.Connect2(None, security.SEC_FLAG_MAXIMUM_ALLOWED)
        self.samr_domain = self.samr.OpenDomain(self.samr_handle, security.SEC_FLAG_MAXIMUM_ALLOWED, self.domain_sid)

        self.sd_utils = sd_utils.SDUtils(self.admin_samdb)

        self.admin_samdb.create_ou("OU=test_computer_ou1," + self.base_dn)
        self.unpriv_user_sid = self.sd_utils.get_object_sid(self.unpriv_user_dn)
        mod = "(OA;;CC;bf967a86-0de6-11d0-a285-00aa003049e2;;%s)" % str(self.unpriv_user_sid)

        old_sd = self.sd_utils.read_sd_on_dn("OU=test_computer_ou1," + self.base_dn)

        self.sd_utils.dacl_add_ace("OU=test_computer_ou1," + self.base_dn, mod)

        self.add_computer_ldap("testcomputer-t")

        self.sd_utils.modify_sd_on_dn("OU=test_computer_ou1," + self.base_dn, old_sd)

        self.computernames = ["testcomputer-0"]

        # Get the SD of the template account, then force it to match
        # what we expect for SeMachineAccountPrivilege accounts, so we
        # can confirm we created the accounts correctly
        self.sd_reference_cc = self.sd_utils.read_sd_on_dn("CN=testcomputer-t,OU=test_computer_ou1,%s" % (self.base_dn))

        self.sd_reference_modify = self.sd_utils.read_sd_on_dn("CN=testcomputer-t,OU=test_computer_ou1,%s" % (self.base_dn))
        for ace in self.sd_reference_modify.dacl.aces:
            if ace.type == security.SEC_ACE_TYPE_ACCESS_ALLOWED and ace.trustee == self.unpriv_user_sid:
                ace.access_mask = ace.access_mask | security.SEC_ADS_SELF_WRITE | security.SEC_ADS_WRITE_PROP

        # Now reconnect without domain admin rights
        self.samdb = SamDB(url=ldaphost, credentials=self.unpriv_creds, lp=lp)

    def test_uac_bits_add(self):
        computername=self.computernames[0]

        user_sid = self.sd_utils.get_object_sid(self.unpriv_user_dn)
        mod = "(OA;;CC;bf967a86-0de6-11d0-a285-00aa003049e2;;%s)" % str(user_sid)

        old_sd = self.sd_utils.read_sd_on_dn("OU=test_computer_ou1," + self.base_dn)

        self.sd_utils.dacl_add_ace("OU=test_computer_ou1," + self.base_dn, mod)

        invalid_bits = set([UF_TEMP_DUPLICATE_ACCOUNT, UF_PARTIAL_SECRETS_ACCOUNT])
        # These bits are privileged, but authenticated users have that CAR by default, so this is a pain to test
        priv_to_auth_users_bits = set([UF_PASSWD_NOTREQD, UF_ENCRYPTED_TEXT_PASSWORD_ALLOWED,
                                       UF_DONT_EXPIRE_PASSWD])

        # These bits really are privileged
        priv_bits = set([UF_INTERDOMAIN_TRUST_ACCOUNT, UF_SERVER_TRUST_ACCOUNT,
                         UF_TRUSTED_FOR_DELEGATION, UF_TRUSTED_TO_AUTHENTICATE_FOR_DELEGATION])

        for bit in bits:
            try:
                self.add_computer_ldap(computername, others={"userAccountControl": [str(bit)]})
                delete_force(self.admin_samdb, "CN=%s,OU=test_computer_ou1,%s" % (computername, self.base_dn))
                if bit in priv_bits:
                    self.fail("Unexpectdly able to set userAccountControl bit 0x%08X on %s" % (bit, computername))

            except LdbError as e4:
                (enum, estr) = e4.args
                if bit in invalid_bits:
                    self.assertEqual(enum, ldb.ERR_OTHER, "Invalid bit 0x%08X was able to be set on %s" % (bit, computername))
                    # No point going on, try the next bit
                    continue
                elif bit in priv_bits:
                    self.assertEqual(enum, ldb.ERR_INSUFFICIENT_ACCESS_RIGHTS)
                    continue
                else:
                    self.fail("Unable to set userAccountControl bit 0x%08X on %s: %s" % (bit, computername, estr))

    def create_machine_account(self):
        self.machine_pass = samba.generate_random_password(32, 32)
        self.machine_name = MACHINE_NAME
        self.machine_dn = "cn=%s,%s" % (self.machine_name, self.ldb.domain_dn())

        # remove the account if it exists, this will happen if a previous test
        # run failed
        delete_force(self.ldb, self.machine_dn)
        # get unicode str for both py2 and py3
        pass_unicode = self.machine_pass.encode('utf-8').decode('utf-8')
        utf16pw = u'"{}"'.format(pass_unicode).encode('utf-16-le')
        self.ldb.add({
            "dn": self.machine_dn,
            "objectclass": "computer",
            "sAMAccountName": "%s$" % self.machine_name,
            "userAccountControl":
                str(UF_WORKSTATION_TRUST_ACCOUNT | UF_PASSWD_NOTREQD),
            "unicodePwd": utf16pw})

        self.machine_creds = Credentials()
        self.machine_creds.guess(self.get_loadparm())
        self.machine_creds.set_password(self.machine_pass)
        self.machine_creds.set_username(self.machine_name + "$")
        self.machine_creds.set_workstation(self.machine_name)

    def test_samr_GetGroupsForUser_nomember(self):
        # Confirm that we get the correct results against SAMR also
        if not url.startswith("ldap://"):
            self.fail(msg="This test is only valid on ldap (so we an find the hostname and use SAMR)")
        host = url.split("://")[1]

        test_user = "tokengroups_user2"
        self.admin_ldb.newuser(test_user, self.test_user_pass)
        res = self.admin_ldb.search(base="cn=%s,cn=users,%s" % (test_user, self.base_dn),
                                    attrs=["objectSid"], scope=ldb.SCOPE_BASE)
        user_sid = ndr_unpack(samba.dcerpc.security.dom_sid, res[0]["objectSid"][0])

        (domain_sid, user_rid) = user_sid.split()
        samr_conn = samba.dcerpc.samr.samr("ncacn_ip_tcp:%s[seal]" % host, lp, creds)
        samr_handle = samr_conn.Connect2(None, security.SEC_FLAG_MAXIMUM_ALLOWED)
        samr_domain = samr_conn.OpenDomain(samr_handle, security.SEC_FLAG_MAXIMUM_ALLOWED,
                                           domain_sid)
        user_handle = samr_conn.OpenUser(samr_domain, security.SEC_FLAG_MAXIMUM_ALLOWED, user_rid)
        rids = samr_conn.GetGroupsForUser(user_handle)
        user_info = samr_conn.QueryUserInfo(user_handle, 1)
        delete_force(self.admin_ldb, "CN=%s,%s,%s" %
                          (test_user, "cn=users", self.base_dn))
        self.assertEqual(len(rids.rids), 1)
        self.assertEqual(rids.rids[0].rid, user_info.primary_gid)