from samba import param
from samba.provision import (find_setup_dir, get_domain_descriptor,
get_config_descriptor, secretsdb_self_join,
- ProvisioningError, getLastProvisionUSN,
- get_max_usn, updateProvisionUSN)
+ ProvisioningError, get_last_provision_usn,
+ get_max_usn, update_provision_usn)
from samba.schema import get_linked_attributes, Schema, get_schema_descriptor
from samba.dcerpc import security, drsblobs
from samba.ndr import ndr_unpack
# We do most of the special case handle if we do not have the
# highest usn as otherwise the replPropertyMetaData will guide us more
# correctly
- if usn == None:
+ if usn is None:
if (att == "member" and flag == FLAG_MOD_REPLACE):
hash = {}
newval = []
delta.remove(att)
return True
- if (att == "gPLink" or att == "gPCFileSysPath") and \
- flag == FLAG_MOD_REPLACE and\
- str(new[0].dn).lower() == str(old[0].dn).lower():
+ if (att in ("gPLink", "gPCFileSysPath") and
+ flag == FLAG_MOD_REPLACE and
+ str(new[0].dn).lower() == str(old[0].dn).lower()):
delta.remove(att)
return True
ref == old and ref == abs(new)
return True
- if (att == "adminDisplayName" or att == "adminDescription"):
+ if att in ("adminDisplayName", "adminDescription"):
return True
- if (str(old[0].dn) == "CN=Samba4-Local-Domain, %s" % (str(names.schemadn))\
+ if (str(old[0].dn) == "CN=Samba4-Local-Domain, %s" % (names.schemadn)
and att == "defaultObjectCategory" and flag == FLAG_MOD_REPLACE):
return True
# This is a bit of special animal as we might have added
# already SPN entries to the list that has to be modified
# So we go in detail to try to find out what has to be added ...
- if ( att == "servicePrincipalName" and flag == FLAG_MOD_REPLACE):
+ if (att == "servicePrincipalName" and flag == FLAG_MOD_REPLACE):
hash = {}
newval = []
changeDelta=0
for e in range(0, len(current)):
message(CHANGE, "old %d : %s" % (i, str(current[e])))
i+=1
- if reference != None:
+ if reference is not None:
i = 0
for e in range(0, len(reference)):
message(CHANGE, "new %d : %s" % (i, str(reference[e])))
i+=1
else:
- message(CHANGE, "old : %s" % str(ndr_unpack( security.dom_sid, current[0])))
- message(CHANGE, "new : %s" % str(ndr_unpack( security.dom_sid, reference[0])))
+ message(CHANGE, "old : %s" % ndr_unpack(security.dom_sid, current[0]))
+ message(CHANGE, "new : %s" % ndr_unpack(security.dom_sid, reference[0]))
def handle_special_add(samdb, dn, names):
"""Handle special operation (like remove) on some object needed during
- upgrade
+ upgrade
This is mostly due to wrong creation of the object in previous provision.
:param samdb: An Ldb object representing the SAM database
:param dn: DN of the object to inspect
- :param names: list of key provision parameters"""
+ :param names: list of key provision parameters
+ """
dntoremove = None
objDn = Dn(samdb, "CN=IIS_IUSRS, CN=Builtin, %s" % names.rootdn)
#This entry was misplaced lets remove it if it exists
dntoremove = "CN=Event Log Readers, CN=Users, %s" % names.rootdn
- objDn = Dn(samdb,"CN=System,CN=WellKnown Security Principals,"\
+ objDn = Dn(samdb,"CN=System,CN=WellKnown Security Principals,"
"CN=Configuration,%s" % names.rootdn)
if dn == objDn:
- oldDn = Dn(samdb,"CN=Well-Known-Security-Id-System,"\
- "CN=WellKnown Security Principals,"\
+ oldDn = Dn(samdb,"CN=Well-Known-Security-Id-System,"
+ "CN=WellKnown Security Principals,"
"CN=Configuration,%s" % names.rootdn)
res = samdb.search(expression="(dn=%s)" % oldDn,
scope=SCOPE_SUBTREE, attrs=["dn"],
controls=["search_options:1:2"])
if len(res) > 0:
- message(CHANGE, "Existing object %s must be replaced by %s,"\
+ message(CHANGE, "Existing object %s must be replaced by %s,"
"Renaming old object" % (str(oldDn), str(dn)))
samdb.rename(oldDn, objDn)
return 1
- if dntoremove != None:
+ if dntoremove is not None:
res = samdb.search(expression="(dn=%s)" % dntoremove,
base=str(names.rootdn),
scope=SCOPE_SUBTREE, attrs=["dn"],
controls=["search_options:1:2"])
if len(res) > 0:
- message(CHANGE, "Existing object %s must be replaced by %s,"\
+ message(CHANGE, "Existing object %s must be replaced by %s,"
"removing old object" % (dntoremove, str(dn)))
samdb.delete(res[0]["dn"])
return 0
+
def check_dn_nottobecreated(hash, index, listdn):
"""Check if one of the DN present in the list has a creation order
greater than the current.
:param listdn: List of DNs on which the current DN depends on
:return: None if the current object do not depend on other
object or if all object have been created before."""
- if listdn == None:
+ if listdn is None:
return None
for dn in listdn:
key = str(dn).lower()
for att in dn_syntax_att:
depend_on_yet_tobecreated = check_dn_nottobecreated(hash, index,
delta.get(str(att)))
- if depend_on_yet_tobecreated != None:
- message(CHANGE, "Object %s depends on %s in attribute %s," \
- "delaying the creation" % (str(dn), \
- depend_on_yet_tobecreated, str(att)))
+ if depend_on_yet_tobecreated is not None:
+ message(CHANGE, "Object %s depends on %s in attribute %s,"
+ "delaying the creation" % (dn,
+ depend_on_yet_tobecreated, att))
return False
delta.dn = dn
attrs=["dn", "wellKnownObjects"])
targetWKO = "%s:%s" % (wkoPrefix, str(reference[0]["dn"]))
- found = 0
+ found = False
if len(res[0]) > 0:
wko = res[0]["wellKnownObjects"]
# The wellKnownObject that we want to add.
for o in wko:
if str(o) == targetWKO:
- found = 1
+ found = True
listwko.append(str(o))
if not found:
blacklist = {}
hash = {}
newlinklist = []
- changed = 0
+ changed = False
newlinklist.extend(value)
for e in ref_value:
if not blacklist.has_key(e) and not hash.has_key(e):
newlinklist.append(str(e))
- changed = 1
+ changed = True
if changed:
delta[att] = MessageElement(newlinklist, FLAG_MOD_REPLACE, att)
else:
delta.remove("name")
- if len(delta.items()) > 1 and usns != None:
+ if len(delta.items()) > 1 and usns is not None:
# Fetch the replPropertyMetaData
res = samdb.search(expression="dn=%s" % (str(dn)), base=basedn,
scope=SCOPE_SUBTREE, controls=controls,
txt = ""
for att in delta:
- if usns != None:
+ if usns is not None:
# We have updated by provision usn information so let's exploit
# replMetadataProperties
if forwardlinked.has_key(att):
# was done in handle_special_case
continue
attrUSN = hash_attr_usn.get(att)
- if att == "forceLogoff" and attrUSN == None:
+ if att == "forceLogoff" and attrUSN is None:
continue
- if attrUSN == None:
+ if attrUSN is None:
delta.remove(att)
continue
if attrUSN == -1:
# This attribute was last modified by another DC forget
# about it
- message(CHANGE, "%sAttribute: %s has been" \
+ message(CHANGE, "%sAttribute: %s has been"
"created/modified/deleted by another DC,"
" do nothing" % (txt, att ))
txt = ""
delta.remove(att)
continue
- elif usn_in_range(int(attrUSN), usns) == 0:
- message(CHANGE, "%sAttribute: %s has been" \
- "created/modified/deleted not during a" \
- " provision or upgradeprovision: current" \
+ elif not usn_in_range(int(attrUSN), usns):
+ message(CHANGE, "%sAttribute: %s has been"
+ "created/modified/deleted not during a"
+ " provision or upgradeprovision: current"
" usn %d , do nothing" % (txt, att, attrUSN))
txt = ""
delta.remove(att)
continue
else:
if att == "defaultSecurityDescriptor":
- defSDmodified = 1
+ defSDmodified = True
if attrUSN:
- message(CHANGE, "%sAttribute: %s will be modified" \
- "/deleted it was last modified" \
- "during a provision, current usn:" \
+ message(CHANGE, "%sAttribute: %s will be modified"
+ "/deleted it was last modified"
+ "during a provision, current usn:"
"%d" % (txt, att, attrUSN))
txt = ""
else:
- message(CHANGE, "%sAttribute: %s will be added because" \
+ message(CHANGE, "%sAttribute: %s will be added because"
" it hasn't existed before " % (txt, att))
txt = ""
continue
else:
# Old school way of handling things for pre alpha12 upgrade
- defSDmodified = 1
+ defSDmodified = True
msgElt = delta.get(att)
if att == "nTSecurityDescriptor":
delta.dn = dn
if len(delta.items()) >1:
attributes=", ".join(delta.keys())
- message(CHANGE, "%s is different from the reference one, changed" \
+ message(CHANGE, "%s is different from the reference one, changed"
" attributes: %s\n" % (dn, attributes))
- changed = changed + 1
+ changed += 1
samdb.modify(delta)
return changed
if sddl != hash[key]:
txt = get_diff_sddls(hash[key], sddl)
if txt != "":
- message(CHANGESD, "On object %s ACL is different"\
+ message(CHANGESD, "On object %s ACL is different"
" \n%s" % (current[i]["dn"], txt))
controls=["search_options:1:2"])
for obj in res:
if not (str(obj["dn"]) == str(names.rootdn) or
- str(obj["dn"]) == str(names.configdn) or \
+ str(obj["dn"]) == str(names.configdn) or
str(obj["dn"]) == str(names.schemadn)):
hash[str(obj["dn"])] = obj["whenCreated"]
key_version_number=kvno,
secure_channel_type=secChanType)
else:
- raise ProvisioningError("Unable to find a Secure Channel" \
+ raise ProvisioningError("Unable to find a Secure Channel"
"of type SEC_CHAN_BDC")
if __name__ == '__main__':
global defSDmodified
- defSDmodified = 0
+ defSDmodified = False
# From here start the big steps of the program
# 1) First get files paths
paths = get_paths(param, smbconf=smbconf)
names = find_provision_key_parameters(ldbs.sam, ldbs.secrets, ldbs.idmap,
paths, smbconf, lp)
# 4)
- lastProvisionUSNs = getLastProvisionUSN(ldbs.sam)
- if lastProvisionUSNs != None:
+ lastProvisionUSNs = get_last_provision_usn(ldbs.sam)
+ if lastProvisionUSNs is not None:
message(CHANGE,
"Find a last provision USN, %d range(s)" % len(lastProvisionUSNs))
# ldbs = get_ldbs(paths, creds, adm_session, lp)
if not sanitychecks(ldbs.sam, names):
- message(SIMPLE, "Sanity checks for the upgrade fails, checks messages" \
+ message(SIMPLE, "Sanity checks for the upgrade fails, checks messages"
" and correct them before rerunning upgradeprovision")
sys.exit(1)
if opts.full:
if not update_samdb(new_ldbs.sam, ldbs.sam, names, lastProvisionUSNs,
schema):
- message(SIMPLE, "Rollbacking every changes. Check the reason" \
+ message(SIMPLE, "Rollbacking every changes. Check the reason"
" of the problem")
- message(SIMPLE, "In any case your system as it was before" \
+ message(SIMPLE, "In any case your system as it was before"
" the upgrade")
ldbs.groupedRollback()
new_ldbs.groupedRollback()
# 18) We rebuild SD only if defaultSecurityDescriptor is modified
# But in fact we should do it also if one object has its SD modified as
# child might need rebuild
- if defSDmodified == 1:
+ if defSDmodified:
message(SIMPLE, "Updating SD")
ldbs.sam.set_session_info(adm_session)
# Alpha10 was a bit broken still
# 21)
check_for_DNS(newpaths.private_dir, paths.private_dir)
# 22)
- if lastProvisionUSNs != None:
- updateProvisionUSN(ldbs.sam, minUSN, maxUSN)
- if opts.full and (names.policyid == None or names.policyid_dc == None):
+ if lastProvisionUSNs is not None:
+ update_provision_usn(ldbs.sam, minUSN, maxUSN)
+ if opts.full and (names.policyid is None or names.policyid_dc is None):
update_policyids(names, ldbs.sam)
if opts.full or opts.resetfileacl:
try:
update_gpo(paths, ldbs.sam, names, lp, message, 1)
except ProvisioningError, e:
- message(ERROR, "The policy for domain controller is missing," \
+ message(ERROR, "The policy for domain controller is missing,"
" you should restart upgradeprovision with --full")
else:
try:
update_gpo(paths, ldbs.sam, names, lp, message, 0)
except ProvisioningError, e:
- message(ERROR, "The policy for domain controller is missing," \
+ message(ERROR, "The policy for domain controller is missing,"
" you should restart upgradeprovision with --full")
ldbs.groupedCommit()
new_ldbs.groupedCommit()
# hard coded at this point, but will probably be changed when
# we enable different fsmo roles
+
def get_config_descriptor(domain_sid):
sddl = "O:EAG:EAD:(OA;;CR;1131f6aa-9c07-11d1-f79f-00c04fc2dcd2;;ED)" \
"(OA;;CR;1131f6ab-9c07-11d1-f79f-00c04fc2dcd2;;ED)" \
self.sitename = None
self.smbconf = None
-def updateProvisionUSN(samdb, low, high, replace = 0):
+
+def update_provision_usn(samdb, low, high, replace=False):
"""Update the field provisionUSN in sam.ldb
+
This field is used to track range of USN modified by provision and
upgradeprovision.
This value is used afterward by next provision to figure out if
:param low: The lowest USN modified by this upgrade
:param high: The highest USN modified by this upgrade
:param replace: A boolean indicating if the range should replace any
- existing one or appended (default)"""
+ existing one or appended (default)
+ """
tab = []
if not replace:
entry = samdb.search(expression="(&(dn=@PROVISION)(%s=*))" % \
LAST_PROVISION_USN_ATTRIBUTE, base="",
scope=ldb.SCOPE_SUBTREE,
- attrs=[LAST_PROVISION_USN_ATTRIBUTE,"dn"])
+ attrs=[LAST_PROVISION_USN_ATTRIBUTE, "dn"])
for e in entry[0][LAST_PROVISION_USN_ATTRIBUTE]:
tab.append(str(e))
- tab.append("%s-%s"%(str(low), str(high)))
+ tab.append("%s-%s" % (low, high))
delta = ldb.Message()
- delta.dn = ldb.Dn(samdb,"@PROVISION")
+ delta.dn = ldb.Dn(samdb, "@PROVISION")
delta[LAST_PROVISION_USN_ATTRIBUTE] = ldb.MessageElement(tab,
ldb.FLAG_MOD_REPLACE,
LAST_PROVISION_USN_ATTRIBUTE)
samdb.modify(delta)
-def setProvisionUSN(samdb, low, high):
+
+def set_provision_usn(samdb, low, high):
"""Set the field provisionUSN in sam.ldb
This field is used to track range of USN modified by provision and
upgradeprovision.
:param low: The lowest USN modified by this upgrade
:param high: The highest USN modified by this upgrade"""
tab = []
- tab.append("%s-%s"%(str(low), str(high)))
+ tab.append("%s-%s" % (low, high))
delta = ldb.Message()
- delta.dn = ldb.Dn(samdb,"@PROVISION")
+ delta.dn = ldb.Dn(samdb, "@PROVISION")
delta[LAST_PROVISION_USN_ATTRIBUTE] = ldb.MessageElement(tab,
ldb.FLAG_MOD_ADD,
LAST_PROVISION_USN_ATTRIBUTE)
samdb.add(delta)
+
def get_max_usn(samdb,basedn):
""" This function return the biggest USN present in the provision
"paged_results:1:1"])
return res[0]["uSNChanged"]
-def getLastProvisionUSN(sam):
+def get_last_provision_usn(sam):
"""Get the lastest USN modified by a provision or an upgradeprovision
:param sam: An LDB object pointing to the sam.ldb
privdir = os.path.join(targetdir, "private")
else:
privdir = default_lp.get("private dir")
- posixeadb_line = "posix:eadb = " + os.path.abspath(os.path.join(privdir,"eadb.tdb"))
+ posixeadb_line = "posix:eadb = " + os.path.abspath(os.path.join(privdir, "eadb.tdb"))
else:
posixeadb_line = ""
set_dir_acl(policy_path,dsacl2fsacl(POLICIES_ACL, str(domainsid)),
lp, str(domainsid))
res = samdb.search(base="CN=Policies,CN=System,%s"%(domaindn),
- attrs=["cn","nTSecurityDescriptor"],
+ attrs=["cn", "nTSecurityDescriptor"],
expression="", scope=ldb.SCOPE_ONELEVEL)
for policy in res:
acl = ndr_unpack(security.descriptor,
if not os.path.exists(paths.private_dir):
os.mkdir(paths.private_dir)
- if not os.path.exists(os.path.join(paths.private_dir,"tls")):
- os.mkdir(os.path.join(paths.private_dir,"tls"))
+ if not os.path.exists(os.path.join(paths.private_dir, "tls")):
+ os.mkdir(os.path.join(paths.private_dir, "tls"))
ldapi_url = "ldapi://%s" % urllib.quote(paths.s4_ldapi_path, safe="")
logger.info("A Kerberos configuration suitable for Samba 4 has been "
"generated at %s", paths.krb5conf)
- lastProvisionUSNs = getLastProvisionUSN(samdb)
+ lastProvisionUSNs = get_last_provision_usn(samdb)
maxUSN = get_max_usn(samdb, str(names.rootdn))
if lastProvisionUSNs != None:
- updateProvisionUSN(samdb, 0, maxUSN, 1)
+ update_provision_usn(samdb, 0, maxUSN, 1)
else:
- setProvisionUSN(samdb, 0, maxUSN)
+ set_provision_usn(samdb, 0, maxUSN)
if serverrole == "domain controller":
create_dns_update_list(lp, logger, paths, setup_path)
logger.info("This slapd-Commandline is also stored under: %s/ldap_backend_startup.sh",
provision_backend.ldapdir)
-
result = ProvisionResult()
result.domaindn = domaindn
result.paths = paths
setup_file(setup_path("named.conf.update"), paths.namedconf_update)
+
def create_named_txt(path, setup_path, realm, dnsdomain,
private_dir, keytab_name):
"""Write out a file containing zone statements suitable for inclusion in a
"PRIVATE_DIR": private_dir
})
+
def create_krb5_conf(path, setup_path, dnsdomain, hostname, realm):
"""Write out a file containing zone statements suitable for inclusion in a
named.conf file (including GSS-TSIG configuration).
secrets_ldb.transaction_commit()
return secrets_ldb
+
class ProvisionTestCase(samba.tests.TestCaseInTempDir):
"""Some simple tests for individual functions in the provisioning code.
"""
+
def test_setup_secretsdb(self):
path = os.path.join(self.tempdir, "secrets.ldb")
ldb = setup_secretsdb(path, setup_path, None, None, lp=env_loadparm())
construct_existor_expr)
from samba.tests.provision import create_dummy_secretsdb
-from samba.tests import env_loadparm, TestCaseInTempDir
+from samba.tests import TestCaseInTempDir
from samba import Ldb
from ldb import SCOPE_SUBTREE
import samba.tests
-lp = env_loadparm()
-
def dummymessage(a=None, b=None):
- if 0:
- print "none"
+ pass
class UpgradeProvisionTestCase(TestCaseInTempDir):
self.assertEquals(dn_sort("dc=toto,dc=tata",
"cn=foo,dc=toto,dc=tata"), -1)
self.assertEquals(dn_sort("cn=bar, dc=toto,dc=tata",
- "cn=foo, dc=toto,dc=tata"),-1)
+ "cn=foo, dc=toto,dc=tata"), -1)
+
def test_get_diff_sddl(self):
sddl = "O:SAG:DUD:AI(A;CIID;RPWPCRCCLCLORCWOWDSW;;;SA)\
(A;CIID;RP LCLORC;;;AU)(A;CIID;RPWPCRCCDCLCLORCWOWDSDDTSW;;;SY)S:AI(AU;CIIDSA;WP;;;WD)"
sddl5 = "O:SAG:DUD:AI(A;CIID;RPWPCRCCLCLORCWOWDSW;;;SA)\
(A;CIID;RP LCLORC;;;AU)(A;CIID;RPWPCRCCDCLCLORCWOWDSDDTSW;;;SY)"
- self.assertEquals(get_diff_sddls(sddl, sddl1) ,"")
+ self.assertEquals(get_diff_sddls(sddl, sddl1), "")
txt = get_diff_sddls(sddl, sddl2)
- self.assertEquals(txt ,"\tOwner mismatch: SA (in ref) BA(in current)\n")
+ self.assertEquals(txt, "\tOwner mismatch: SA (in ref) BA(in current)\n")
txt = get_diff_sddls(sddl, sddl3)
- self.assertEquals(txt ,"\tGroup mismatch: DU (in ref) BA(in current)\n")
+ self.assertEquals(txt, "\tGroup mismatch: DU (in ref) BA(in current)\n")
txt = get_diff_sddls(sddl, sddl4)
txtmsg = "\tPart dacl is different between reference and current here\
is the detail:\n\t\t(A;CIID;RPWPCRCCLCLORCWOWDSW;;;BA) ACE is not present in\
the reference\n\t\t(A;CIID;RPWPCRCCLCLORCWOWDSW;;;SA) ACE is not present in\
the current\n"
- self.assertEquals(txt , txtmsg)
+ self.assertEquals(txt, txtmsg)
txt = get_diff_sddls(sddl, sddl5)
- self.assertEquals(txt ,"\tCurrent ACL hasn't a sacl part\n")
+ self.assertEquals(txt, "\tCurrent ACL hasn't a sacl part\n")
def test_construct_existor_expr(self):
res = construct_existor_expr([])
res = construct_existor_expr(["foo", "bar"])
self.assertEquals(res, "(|(foo=*)(bar=*))")
+
class UpdateSecretsTests(samba.tests.TestCaseInTempDir):
+
def setUp(self):
super(UpdateSecretsTests, self).setUp()
self.referencedb = create_dummy_secretsdb(
import os
import re
import shutil
+
+from samba import param
from samba.credentials import Credentials
from samba.auth import system_session
from samba.provision import getpolicypath
find_provision_key_parameters, identic_rename,
updateOEMInfo, getOEMInfo, update_gpo,
delta_update_basesamdb,search_constructed_attrs_stored)
-
-from samba.tests.provision import create_dummy_secretsdb
-from samba import param
from samba.tests import env_loadparm, TestCaseInTempDir
+from samba.tests.provision import create_dummy_secretsdb
import ldb
def dummymessage(a=None, b=None):
- if 0:
- print "none"
+ pass
-lp = env_loadparm()
-smbConfPath = "%s/%s/%s" % (os.environ["SELFTEST_PREFIX"], "dc", "etc/smb.conf")
+smb_conf_path = "%s/%s/%s" % (os.environ["SELFTEST_PREFIX"], "dc", "etc/smb.conf")
class UpgradeProvisionBasicLdbHelpersTestCase(TestCaseInTempDir):
"""Some simple tests for individual functions in the provisioning code.
"""
def test_get_ldbs(self):
- paths = get_paths(param, None, smbConfPath)
+ paths = get_paths(param, None, smb_conf_path)
creds = Credentials()
+ lp = env_loadparm()
creds.guess(lp)
get_ldbs(paths, creds, system_session(), lp)
def test_find_key_param(self):
- paths = get_paths(param, None, smbConfPath)
+ paths = get_paths(param, None, smb_conf_path)
creds = Credentials()
+ lp = env_loadparm()
creds.guess(lp)
rootdn = "dc=samba,dc=example,dc=com"
ldbs = get_ldbs(paths, creds, system_session(), lp)
names = find_provision_key_parameters(ldbs.sam, ldbs.secrets, ldbs.idmap,
- paths, smbConfPath, lp)
+ paths, smb_conf_path, lp)
self.assertEquals(names.realm, "SAMBA.EXAMPLE.COM")
- self.assertTrue(str(names.rootdn).lower() == rootdn.lower())
+ self.assertEquals(str(names.rootdn).lower(), rootdn.lower())
self.assertTrue(names.policyid_dc != None)
self.assertTrue(names.ntdsguid != "")
class UpgradeProvisionWithLdbTestCase(TestCaseInTempDir):
+
def _getEmptyDbName(self):
return os.path.join(self.tempdir, "sam.ldb")
def setUp(self):
super(UpgradeProvisionWithLdbTestCase, self).setUp()
- paths = get_paths(param, None, smbConfPath)
+ paths = get_paths(param, None, smb_conf_path)
self.creds = Credentials()
- self.creds.guess(lp)
+ self.lp = env_loadparm()
+ self.creds.guess(self.lp)
self.paths = paths
- self.ldbs = get_ldbs(paths, self.creds, system_session(), lp)
- self.lp = lp
+ self.ldbs = get_ldbs(paths, self.creds, system_session(), self.lp)
self.names = find_provision_key_parameters(self.ldbs.sam, self.ldbs.secrets,
- self.ldbs.idmap, paths, smbConfPath, lp)
+ self.ldbs.idmap, paths, smb_conf_path, self.lp)
self.referencedb = create_dummy_secretsdb(
os.path.join(self.tempdir, "ref.ldb"))
-
def test_search_constructed_attrs_stored(self):
hashAtt = search_constructed_attrs_stored(self.ldbs.sam,
self.names.rootdn,
["msds-KeyVersionNumber"])
self.assertFalse(hashAtt.has_key("msds-KeyVersionNumber"))
+
def test_identic_rename(self):
rootdn = "DC=samba,DC=example,DC=com"
return ldbs
+
def usn_in_range(usn, range):
"""Check if the usn is in one of the range provided.
To do so, the value is checked to be between the lower bound and
:param usn: A integer value corresponding to the usn that we want to update
:param range: A list of integer representing ranges, lower bounds are in
the even indices, higher in odd indices
- :return: 1 if the usn is in one of the range, 0 otherwise"""
+ :return: True if the usn is in one of the range, False otherwise
+ """
idx = 0
- cont = 1
- ok = 0
- while (cont == 1):
+ cont = True
+ ok = False
+ while cont:
if idx == len(range):
- cont = 0
+ cont = False
continue
if usn < int(range[idx]):
if idx %2 == 1:
- ok = 1
- cont = 0
+ ok = True
+ cont = False
if usn == int(range[idx]):
- cont = 0
- ok = 1
+ cont = False
+ ok = True
idx = idx + 1
return ok
+
def get_paths(param, targetdir=None, smbconf=None):
"""Get paths to important provision objects (smb.conf, ldb files, ...)
else:
names.policyid_dc = None
+
def find_provision_key_parameters(samdb, secretsdb, idmapdb, paths, smbconf, lp):
"""Get key provision parameters (realm, domain, ...) from a given provision
:param paths: A list of path to provision object
:param smbconf: Path to the smb.conf file
:param lp: A LoadParm object
- :return: A list of key provision parameters"""
-
+ :return: A list of key provision parameters
+ """
names = ProvisionNames()
names.adminpass = None
return -1
return ret
+
def identic_rename(ldbobj, dn):
"""Perform a back and forth rename to trigger renaming on attribute that
- can't be directly modified.
+ can't be directly modified.
:param lbdobj: An Ldb Object
- :param dn: DN of the object to manipulate """
+ :param dn: DN of the object to manipulate
+ """
(before, sep, after)=str(dn).partition('=')
ldbobj.rename(dn, ldb.Dn(ldbobj, "%s=foo%s" % (before, after)))
ldbobj.rename(ldb.Dn(ldbobj, "%s=foo%s" % (before, after)), dn)
+
def chunck_acl(acl):
"""Return separate ACE of an ACL
Set ACL correctly also.
Check ACLs for sysvol/netlogon dirs also
"""
- resetacls = 0
+ resetacls = False
try:
ntacls.checkset_backend(lp, None, None)
eadbname = lp.get("posix:eadb")
attribute = samba.xattr_native.wrap_getxattr(paths.sysvol,
xattr.XATTR_NTACL_NAME)
except:
- resetacls = 1
+ resetacls = True
if force:
- resetacls = 1
+ resetacls = True
dir = getpolicypath(paths.sysvol, names.dnsdomain, names.policyid)
if not os.path.isdir(dir):
git.samba.org / abartlet / samba.git / .git / commitdiff