def setUp(self):
     super(MapTestCase, self).setUp()
     ldb = Ldb(self.ldburl, lp=cmdline_loadparm, session_info=system_session())
     ldif = read_datafile("provision_samba3sam.ldif")
     ldb.add_ldif(self.samba4.subst(ldif))
     self.setup_modules(ldb, self.samba3, self.samba4)
     del ldb
     self.ldb = Ldb(self.ldburl, lp=cmdline_loadparm, session_info=system_session())

 def setUp(self):
     super(Samba3SamTestCase, self).setUp()
     ldb = Ldb(self.ldburl, lp=self.lp, session_info=system_session())
     self.samba3.setup_data("samba3.ldif")
     ldif = read_datafile("provision_samba3sam.ldif")
     ldb.add_ldif(self.samba4.subst(ldif))
     self.setup_modules(ldb, self.samba3, self.samba4)
     del ldb
     self.ldb = Ldb(self.ldburl, lp=self.lp, session_info=system_session())

 def setUp(self):
     super(EncryptedSecretsTests, self).setUp()
     self.lp = samba.tests.env_loadparm()
     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)

 def setUp(self):
     super(MapTestCase, self).setUp()
     ldb = Ldb(self.ldburl, lp=self.lp, session_info=system_session())
     ldb.set_opaque("skip_allocate_sids", "true");
     ldif = read_datafile("provision_samba3sam.ldif")
     ldb.add_ldif(self.samba4.subst(ldif))
     self.setup_modules(ldb, self.samba3, self.samba4)
     del ldb
     self.ldb = Ldb(self.ldburl, lp=self.lp, session_info=system_session())
     self.ldb.set_opaque("skip_allocate_sids", "true");

    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 run(self, *accounts, **kwargs):
        sambaopts = kwargs.get("sambaopts")
        credopts = kwargs.get("credopts")
        versionpts = kwargs.get("versionopts")
        server = kwargs.get("server")
        accounts_file = kwargs.get("file")

        if server is None:
            raise Exception("You must supply a server")

        if accounts_file is not None:
            accounts = []
            if accounts_file == "-":
                for line in sys.stdin:
                    accounts.append(line.strip())
            else:
                for line in open(accounts_file, 'r'):
                    accounts.append(line.strip())

        lp = sambaopts.get_loadparm()

        creds = credopts.get_credentials(lp, fallback_machine=True)

        # connect to the remote and local SAMs
        samdb = SamDB(url="ldap://%s" % server,
                      session_info=system_session(),
                      credentials=creds, lp=lp)

        local_samdb = SamDB(url=None, session_info=system_session(),
                            credentials=creds, lp=lp)

        destination_dsa_guid = misc.GUID(local_samdb.get_ntds_GUID())

        repl = drs_Replicate("ncacn_ip_tcp:%s[seal,print]" % server, lp, creds,
                             local_samdb, destination_dsa_guid)
        for account in accounts:
            # work out the source and destination GUIDs
            dc_ntds_dn = samdb.get_dsServiceName()
            res = samdb.search(base=dc_ntds_dn, scope=ldb.SCOPE_BASE, attrs=["invocationId"])
            source_dsa_invocation_id = misc.GUID(local_samdb.schema_format_value("objectGUID", res[0]["invocationId"][0]))

            dn = self.get_dn(samdb, account)
            self.outf.write("Replicating DN %s\n" % dn)

            local_samdb.transaction_start()
            try:
                repl.replicate(dn, source_dsa_invocation_id, destination_dsa_guid,
                               exop=drsuapi.DRSUAPI_EXOP_REPL_SECRET, rodc=True)
            except Exception, e:
                local_samdb.transaction_cancel()
                raise CommandError("Error replicating DN %s" % dn, e)
            local_samdb.transaction_commit()

def ldif_to_samdb(dburl, lp, ldif_file, forced_local_dsa=None):
    """Routine to import all objects and attributes that are relevent
    to the KCC algorithms from a previously exported LDIF file.

    The point of this function is to allow a programmer/debugger to
    import an LDIF file with non-security relevent information that
    was previously extracted from a DC database.  The LDIF file is used
    to create a temporary abbreviated database.  The KCC algorithm can
    then run against this abbreviated database for debug or test
    verification that the topology generated is computationally the
    same between different OSes and algorithms.

    :param dburl: path to the temporary abbreviated db to create
    :param ldif_file: path to the ldif file to import
    """
    if os.path.exists(dburl):
        raise LdifError("Specify a database (%s) that doesn't already exist." %
                        dburl)

    # Use ["modules:"] as we are attempting to build a sam
    # database as opposed to start it here.
    tmpdb = Ldb(url=dburl, session_info=system_session(),
                lp=lp, options=["modules:"])

    tmpdb.transaction_start()
    try:
        data = read_and_sub_file(ldif_file, None)
        tmpdb.add_ldif(data, None)
        if forced_local_dsa:
            tmpdb.modify_ldif("""dn: @ROOTDSE
changetype: modify
replace: dsServiceName
dsServiceName: CN=NTDS Settings,%s
            """ % forced_local_dsa)

        tmpdb.add_ldif("""dn: @MODULES
@LIST: rootdse,extended_dn_in,extended_dn_out_ldb,objectguid
-
""")

    except Exception as estr:
        tmpdb.transaction_cancel()
        raise LdifError("Failed to import %s: %s" % (ldif_file, estr))

    tmpdb.transaction_commit()

    # We have an abbreviated list of options here because we have built
    # an abbreviated database.  We use the rootdse and extended-dn
    # modules only during this re-open
    samdb = SamDB(url=dburl, session_info=system_session(), lp=lp)
    return samdb

    def run(self, username=None, filter=None, credopts=None, sambaopts=None,
            versionopts=None, H=None, newpassword=None,
            must_change_at_next_login=False, random_password=False):
        if filter is None and username is None:
            raise CommandError("Either the username or '--filter' must be specified!")

        if random_password:
            password = generate_random_password(128, 255)
        else:
            password = newpassword

        while 1:
            if password is not None and password is not '':
                break
            password = getpass("New Password: ")

        if filter is None:
            filter = "(&(objectClass=user)(sAMAccountName={0!s}))".format((ldb.binary_encode(username)))

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

        creds.set_gensec_features(creds.get_gensec_features() | gensec.FEATURE_SEAL)

        samdb = SamDB(url=H, session_info=system_session(),
                      credentials=creds, lp=lp)

        try:
            samdb.setpassword(filter, password,
                              force_change_at_next_login=must_change_at_next_login,
                              username=username)
        except Exception, msg:
            # FIXME: catch more specific exception
            raise CommandError("Failed to set password for user '{0!s}': {1!s}".format(username or filter, msg))

    def setUp(self):
        super(AuthLogTestsNetLogonBadCreds, self).setUp()
        self.lp      = samba.tests.env_loadparm()
        self.creds   = Credentials()

        self.session = system_session()
        self.ldb = SamDB(
            session_info=self.session,
            credentials=self.creds,
            lp=self.lp)

        self.domain        = os.environ["DOMAIN"]
        self.netbios_name  = "NetLogonBad"
        self.machinepass   = "abcdefghij"
        self.remoteAddress = AS_SYSTEM_MAGIC_PATH_TOKEN
        self.base_dn       = self.ldb.domain_dn()
        self.dn            = ("cn=%s,cn=users,%s" %
                              (self.netbios_name, self.base_dn))

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

def deprovision(setup_path, lp, creds, firstorg=None, firstou=None, reporter=None):
    """Remote all configuration entries added by the OpenChange
    installation.

    :param setup_path: Path to the setup directory.
    :param names: provision names object.
    :param lp: Loadparm context
    :param creds: Credentials Context
    :param reporter: A progress reporter instance (subclass of AbstractProgressReporter)
    """

    if reporter is None:
        reporter = TextProgressReporter()

    session_info = system_session()

    lp.set("dsdb:schema update allowed", "yes")

    names = guess_names_from_smbconf(lp, None, None)

    samdb = SamDB(url=get_ldb_url(lp, creds, names), session_info=session_info,
                  credentials=creds, lp=lp)

    try:
        deprovision_schema(setup_path, names, lp, creds, reporter, "AD/oc_provision_configuration.ldif", "Remove Exchange configuration objects")
    except LdbError, ldb_error:
        print ("[!] error while deprovisioning the Exchange configuration"
               " objects (%d): %s" % ldb_error.args)

    def run(self, accountname, principal, H=None, credopts=None, sambaopts=None,
            versionopts=None):

        lp = sambaopts.get_loadparm()
        creds = credopts.get_credentials(lp)
        paths = provision.provision_paths_from_lp(lp, lp.get("realm"))
        if H == None:
            path = paths.samdb
        else:
            path = H

        sam = SamDB(path, session_info=system_session(),
                    credentials=creds, lp=lp)
        # TODO once I understand how, use the domain info to naildown
        # to the correct domain
        (cleanedaccount, realm, domain) = _get_user_realm_domain(accountname)

        res = sam.search(expression="sAMAccountName=%s" %
                         ldb.binary_encode(cleanedaccount),
                         scope=ldb.SCOPE_SUBTREE,
                         attrs=["msDS-AllowedToDelegateTo"])
        if len(res) == 0:
            raise CommandError("Unable to find account name '%s'" % accountname)
        assert(len(res) == 1)

        msg = ldb.Message()
        msg.dn = res[0].dn
        msg["msDS-AllowedToDelegateTo"] = ldb.MessageElement([principal],
                                          ldb.FLAG_MOD_DELETE,
                                          "msDS-AllowedToDelegateTo")
        try:
            sam.modify(msg)
        except Exception as err:
            raise CommandError(err)

    def run(self, groupname, new_parent_dn, credopts=None, sambaopts=None,
            versionopts=None, H=None):
        lp = sambaopts.get_loadparm()
        creds = credopts.get_credentials(lp, fallback_machine=True)
        samdb = SamDB(url=H, session_info=system_session(),
                      credentials=creds, lp=lp)
        domain_dn = ldb.Dn(samdb, samdb.domain_dn())

        filter = ("(&(sAMAccountName=%s)(objectClass=group))" %
                  groupname)
        try:
            res = samdb.search(base=domain_dn,
                               expression=filter,
                               scope=ldb.SCOPE_SUBTREE)
            group_dn = res[0].dn
        except IndexError:
            raise CommandError('Unable to find group "%s"' % (groupname))

        try:
            full_new_parent_dn = samdb.normalize_dn_in_domain(new_parent_dn)
        except Exception as e:
            raise CommandError('Invalid new_parent_dn "%s": %s' %
                               (new_parent_dn, e.message))

        full_new_group_dn = ldb.Dn(samdb, str(group_dn))
        full_new_group_dn.remove_base_components(len(group_dn)-1)
        full_new_group_dn.add_base(full_new_parent_dn)

        try:
            samdb.rename(group_dn, full_new_group_dn)
        except Exception as e:
            raise CommandError('Failed to move group "%s"' % groupname, e)
        self.outf.write('Moved group "%s" into "%s"\n' %
                        (groupname, full_new_parent_dn))

    def run(self, acl, file, use_ntvfs=False, use_s3fs=False,
            quiet=False,xattr_backend=None,eadb_file=None,
            credopts=None, sambaopts=None, versionopts=None,
            service=None):
        logger = self.get_logger()
        lp = sambaopts.get_loadparm()
        try:
            samdb = SamDB(session_info=system_session(),
                          lp=lp)
        except Exception as e:
            raise CommandError("Unable to open samdb:", e)

        if not use_ntvfs and not use_s3fs:
            use_ntvfs = "smb" in lp.get("server services")
        elif use_s3fs:
            use_ntvfs = False

        try:
            domain_sid = security.dom_sid(samdb.domain_sid)
        except:
            raise CommandError("Unable to read domain SID from configuration files")

        s3conf = s3param.get_context()
        s3conf.load(lp.configfile)
        # ensure we are using the right samba_dsdb passdb backend, no matter what
        s3conf.set("passdb backend", "samba_dsdb:%s" % samdb.url)

        setntacl(lp, file, acl, str(domain_sid), xattr_backend, eadb_file, use_ntvfs=use_ntvfs, service=service)

        if use_ntvfs:
            logger.warning("Please note that POSIX permissions have NOT been changed, only the stored NT ACL")

 def setUp(self):
     super(SitesBaseTests, self).setUp()
     self.ldb = SamDB(ldaphost, credentials=creds,
                      session_info=system_session(lp), lp=lp)
     self.base_dn = self.ldb.domain_dn()
     self.domain_sid = security.dom_sid(self.ldb.get_domain_sid())
     self.configuration_dn = self.ldb.get_config_basedn().get_linearized()

    def test_join_time_ridalloc(self):
        """Perform a join against the RID manager and assert we have a RID Set"""

        fsmo_dn = ldb.Dn(self.ldb_dc1, "CN=RID Manager$,CN=System," + self.ldb_dc1.domain_dn())
        (fsmo_owner, fsmo_not_owner) = self._determine_fSMORoleOwner(fsmo_dn)

        targetdir = self._test_join(fsmo_owner['dns_name'], "RIDALLOCTEST5")
        try:
            # Connect to the database
            ldb_url = "tdb://%s" % os.path.join(targetdir, "private/sam.ldb")
            smbconf = os.path.join(targetdir, "etc/smb.conf")

            lp = self.get_loadparm()
            new_ldb = SamDB(ldb_url, credentials=self.get_credentials(),
                            session_info=system_session(lp), lp=lp)

            # 1. Get server name
            res = new_ldb.search(base=ldb.Dn(new_ldb, new_ldb.get_serverName()),
                                 scope=ldb.SCOPE_BASE, attrs=["serverReference"])
            # 2. Get server reference
            server_ref_dn = ldb.Dn(new_ldb, res[0]['serverReference'][0])

            # 3. Assert we get the RID Set
            res = new_ldb.search(base=server_ref_dn,
                                 scope=ldb.SCOPE_BASE, attrs=['rIDSetReferences'])

            self.assertTrue("rIDSetReferences" in res[0])
        finally:
            self._test_force_demote(fsmo_owner['dns_name'], "RIDALLOCTEST5")
            shutil.rmtree(targetdir, ignore_errors=True)

    def setUp(self):
        super(RodcCmdTestCase, self).setUp()
        self.lp = samba.param.LoadParm()
        self.lp.load(os.environ["SMB_CONF_PATH"])
        self.creds = Credentials()
        self.creds.set_username(os.environ["DC_USERNAME"])
        self.creds.set_password(os.environ["DC_PASSWORD"])
        self.creds.guess(self.lp)
        self.session = system_session()
        self.ldb = SamDB("ldap://" + os.environ["DC_SERVER"],
            session_info=self.session, credentials=self.creds,lp=self.lp)

        self.base_dn = self.ldb.domain_dn()

        self.ldb.newuser("sambatool1", "[email protected]")
        self.ldb.newuser("sambatool2", "2wsxCDE#")
        self.ldb.newuser("sambatool3", "3edcVFR$")
        self.ldb.newuser("sambatool4", "4rfvBGT%")
        self.ldb.newuser("sambatool5", "5tjbNHY*")
        self.ldb.newuser("sambatool6", "6yknMJU*")

        self.ldb.add_remove_group_members("Allowed RODC Password Replication Group",
                                          ["sambatool1", "sambatool2", "sambatool3",
                                           "sambatool4", "sambatool5"],
                                          add_members_operation=True)

    def run(self, username=None, filter=None, credopts=None, sambaopts=None,
            versionopts=None, H=None, newpassword=None,
            must_change_at_next_login=None):
        if filter is None and username is None:
            raise CommandError("Either the username or '--filter' must be specified!")

        password = newpassword
        if password is None:
            password = getpass("New Password: ")

        if filter is None:
            filter = "(&(objectClass=user)(sAMAccountName=%s))" % (username)

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

        creds.set_gensec_features(creds.get_gensec_features() | gensec.FEATURE_SEAL)

        samdb = SamDB(url=H, session_info=system_session(),
                      credentials=creds, lp=lp)

        try:
            samdb.setpassword(filter, password,
                              force_change_at_next_login=must_change_at_next_login,
                              username=username)
        except Exception, e:
            raise CommandError('Failed to set password for user "%s"' % username, e)

 def setUp(self):
     super(SchemaTests_msDS_isRODC, self).setUp()
     self.ldb =  SamDB(host, credentials=creds,
         session_info=system_session(lp), lp=lp, options=ldb_options)
     res = self.ldb.search(base="", expression="", scope=SCOPE_BASE, attrs=["defaultNamingContext"])
     self.assertEquals(len(res), 1)
     self.base_dn = res[0]["defaultNamingContext"][0]

 def __init__(self):
     self.samba_lp = LoadParm()
     self.samba_lp.set("debug level", "0")
     self.samba_lp.load_default()
     url = self.samba_lp.get("dcerpc_mapiproxy:samdb_url") or self.samba_lp.private_path("sam.ldb")
     self.samdb = SamDB(url=url, lp=self.samba_lp, session_info=system_session())
     self.conn = self._open_mysql_connection()