and not linkID):
set_attrs_seen.add(str(attrname).lower())
- if syntax_oid in [ dsdb.DSDB_SYNTAX_BINARY_DN, dsdb.DSDB_SYNTAX_OR_NAME,
+ if syntax_oid in [dsdb.DSDB_SYNTAX_BINARY_DN, dsdb.DSDB_SYNTAX_OR_NAME,
dsdb.DSDB_SYNTAX_STRING_DN, ldb.SYNTAX_DN ]:
# it's some form of DN, do specialised checking on those
error_count += self.check_dn(obj, attrname, syntax_oid)
self.gp_db.delete(str(self), self.attribute)
def mapper(self):
- return { 'kdc:user_ticket_lifetime': (self.set_kdc_tdb, self.explicit),
+ return {'kdc:user_ticket_lifetime': (self.set_kdc_tdb, self.explicit),
'kdc:service_ticket_lifetime': (self.set_kdc_tdb,
self.mins_to_hours),
'kdc:renewal_lifetime': (self.set_kdc_tdb,
def mapper(self):
'''ldap value : samba setter'''
- return { "minPwdAge" : (self.ch_minPwdAge, self.days2rel_nttime),
+ return {"minPwdAge" : (self.ch_minPwdAge, self.days2rel_nttime),
"maxPwdAge" : (self.ch_maxPwdAge, self.days2rel_nttime),
# Could be none, but I like the method assignment in
# update_samba
else:
ctx.topology_dn = None
- ctx.SPNs = [ "HOST/%s" % ctx.myname,
+ ctx.SPNs = ["HOST/%s" % ctx.myname,
"HOST/%s" % ctx.dnshostname,
"GC/%s/%s" % (ctx.dnshostname, ctx.dnsforest) ]
"systemFlags" : str(samba.dsdb.SYSTEM_FLAG_DISALLOW_MOVE_ON_DELETE),
"dMDLocation" : ctx.schema_dn}
- nc_list = [ ctx.base_dn, ctx.config_dn, ctx.schema_dn ]
+ nc_list = [ctx.base_dn, ctx.config_dn, ctx.schema_dn ]
if ctx.behavior_version >= samba.dsdb.DS_DOMAIN_FUNCTION_2003:
rec["msDS-Behavior-Version"] = str(samba.dsdb.DS_DOMAIN_FUNCTION_2008_R2)
# full_nc_list is the list of naming context (NC) we hold
# read/write copies of. These are not subsets of each other.
- ctx.nc_list = [ ctx.config_dn, ctx.schema_dn ]
- ctx.full_nc_list = [ ctx.base_dn, ctx.config_dn, ctx.schema_dn ]
+ ctx.nc_list = [ctx.config_dn, ctx.schema_dn ]
+ ctx.full_nc_list = [ctx.base_dn, ctx.config_dn, ctx.schema_dn ]
if ctx.subdomain and ctx.dns_backend != "NONE":
ctx.full_nc_list += [ctx.domaindns_zone]
samba.dsdb.UF_TRUSTED_TO_AUTHENTICATE_FOR_DELEGATION |
samba.dsdb.UF_PARTIAL_SECRETS_ACCOUNT)
- ctx.SPNs.extend([ "RestrictedKrbHost/%s" % ctx.myname,
+ ctx.SPNs.extend(["RestrictedKrbHost/%s" % ctx.myname,
"RestrictedKrbHost/%s" % ctx.dnshostname ])
ctx.connection_dn = "CN=RODC Connection (FRS),%s" % ctx.ntds_dn
ctx.secure_channel_type = misc.SEC_CHAN_RODC
ctx.RODC = True
- ctx.replica_flags |= ( drsuapi.DRSUAPI_DRS_SPECIAL_SECRET_PROCESSING |
+ ctx.replica_flags |= (drsuapi.DRSUAPI_DRS_SPECIAL_SECRET_PROCESSING |
drsuapi.DRSUAPI_DRS_GET_ALL_GROUP_MEMBERSHIP)
ctx.domain_replica_flags = ctx.replica_flags
if domain_critical_only:
type=str, metavar="URL", dest="H"),
]
- takes_args = [ "computername", "new_ou_dn" ]
+ takes_args = ["computername", "new_ou_dn" ]
takes_optiongroups = {
"sambaopts": options.SambaOptions,
"credopts": options.CredentialsOptions,
samdb_schema = SamDB(session_info=system_session(), url=None,
credentials=creds, lp=lp)
- scope_map = { "SUB": ldb.SCOPE_SUBTREE, "BASE": ldb.SCOPE_BASE, "ONE":ldb.SCOPE_ONELEVEL }
+ scope_map = {"SUB": ldb.SCOPE_SUBTREE, "BASE": ldb.SCOPE_BASE, "ONE":ldb.SCOPE_ONELEVEL }
scope = scope.upper()
if not scope in scope_map:
raise CommandError("Unknown scope %s" % scope)
def boot_method_string(boot_method):
- enum_defs = [ 'DNS_BOOT_METHOD_UNINITIALIZED', 'DNS_BOOT_METHOD_FILE',
+ enum_defs = ['DNS_BOOT_METHOD_UNINITIALIZED', 'DNS_BOOT_METHOD_FILE',
'DNS_BOOT_METHOD_REGISTRY', 'DNS_BOOT_METHOD_DIRECTORY' ]
return enum_string(dnsserver, enum_defs, boot_method)
def name_check_flag_string(check_flag):
- enum_defs = [ 'DNS_ALLOW_RFC_NAMES_ONLY', 'DNS_ALLOW_NONRFC_NAMES',
+ enum_defs = ['DNS_ALLOW_RFC_NAMES_ONLY', 'DNS_ALLOW_NONRFC_NAMES',
'DNS_ALLOW_MULTIBYTE_NAMES', 'DNS_ALLOW_ALL_NAMES' ]
return enum_string(dnsserver, enum_defs, check_flag)
def zone_type_string(zone_type):
- enum_defs = [ 'DNS_ZONE_TYPE_CACHE', 'DNS_ZONE_TYPE_PRIMARY',
+ enum_defs = ['DNS_ZONE_TYPE_CACHE', 'DNS_ZONE_TYPE_PRIMARY',
'DNS_ZONE_TYPE_SECONDARY', 'DNS_ZONE_TYPE_STUB',
'DNS_ZONE_TYPE_FORWARDER', 'DNS_ZONE_TYPE_SECONDARY_CACHE' ]
return enum_string(dnsp, enum_defs, zone_type)
def zone_update_string(zone_update):
- enum_defs = [ 'DNS_ZONE_UPDATE_OFF', 'DNS_ZONE_UPDATE_UNSECURE',
+ enum_defs = ['DNS_ZONE_UPDATE_OFF', 'DNS_ZONE_UPDATE_UNSECURE',
'DNS_ZONE_UPDATE_SECURE' ]
return enum_string(dnsp, enum_defs, zone_update)
def zone_secondary_security_string(security):
- enum_defs = [ 'DNS_ZONE_SECSECURE_NO_SECURITY', 'DNS_ZONE_SECSECURE_NS_ONLY',
+ enum_defs = ['DNS_ZONE_SECSECURE_NO_SECURITY', 'DNS_ZONE_SECSECURE_NS_ONLY',
'DNS_ZONE_SECSECURE_LIST_ONLY', 'DNS_ZONE_SECSECURE_NO_XFER' ]
return enum_string(dnsserver, enum_defs, security)
def zone_notify_level_string(notify_level):
- enum_defs = [ 'DNS_ZONE_NOTIFY_OFF', 'DNS_ZONE_NOTIFY_ALL_SECONDARIES',
+ enum_defs = ['DNS_ZONE_NOTIFY_OFF', 'DNS_ZONE_NOTIFY_ALL_SECONDARIES',
'DNS_ZONE_NOTIFY_LIST_ONLY' ]
return enum_string(dnsserver, enum_defs, notify_level)
def dp_flags_string(dp_flags):
- bitmap_defs = [ 'DNS_DP_AUTOCREATED', 'DNS_DP_LEGACY', 'DNS_DP_DOMAIN_DEFAULT',
+ bitmap_defs = ['DNS_DP_AUTOCREATED', 'DNS_DP_LEGACY', 'DNS_DP_DOMAIN_DEFAULT',
'DNS_DP_FOREST_DEFAULT', 'DNS_DP_ENLISTED', 'DNS_DP_DELETED' ]
return bitmap_string(dnsserver, bitmap_defs, dp_flags)
def zone_flags_string(flags):
- bitmap_defs = [ 'DNS_RPC_ZONE_PAUSED', 'DNS_RPC_ZONE_SHUTDOWN',
+ bitmap_defs = ['DNS_RPC_ZONE_PAUSED', 'DNS_RPC_ZONE_SHUTDOWN',
'DNS_RPC_ZONE_REVERSE', 'DNS_RPC_ZONE_AUTOCREATED',
'DNS_RPC_ZONE_DSINTEGRATED', 'DNS_RPC_ZONE_AGING',
'DNS_RPC_ZONE_UPDATE_UNSECURE', 'DNS_RPC_ZONE_UPDATE_SECURE',
synopsis = '%prog <server> [options]'
- takes_args = [ 'server' ]
+ takes_args = ['server' ]
takes_optiongroups = {
"sambaopts": options.SambaOptions,
synopsis = '%prog <server> <zone> [options]'
- takes_args = [ 'server', 'zone' ]
+ takes_args = ['server', 'zone' ]
takes_optiongroups = {
"sambaopts": options.SambaOptions,
synopsis = '%prog <server> [options]'
- takes_args = [ 'server' ]
+ takes_args = ['server' ]
takes_optiongroups = {
"sambaopts": options.SambaOptions,
synopsis = '%prog <server> <zone> [options]'
- takes_args = [ 'server', 'zone' ]
+ takes_args = ['server', 'zone' ]
takes_optiongroups = {
"sambaopts": options.SambaOptions,
synopsis = '%prog <server> <zone> [options]'
- takes_args = [ 'server', 'zone' ]
+ takes_args = ['server', 'zone' ]
takes_optiongroups = {
"sambaopts": options.SambaOptions,
synopsis = '%prog <server> <zone> <name> <A|AAAA|CNAME|MX|NS|SOA|SRV|TXT|ALL> [options]'
- takes_args = [ 'server', 'zone', 'name', 'rtype' ]
+ takes_args = ['server', 'zone', 'name', 'rtype' ]
takes_optiongroups = {
"sambaopts": options.SambaOptions,
synopsis = '%prog <server> [<name>] [options]'
- takes_args = [ 'server', 'name?' ]
+ takes_args = ['server', 'name?' ]
takes_optiongroups = {
"sambaopts": options.SambaOptions,
synopsis = '%prog <server> <zone> <name> <A|AAAA|PTR|CNAME|NS|MX|SRV|TXT> <data>'
- takes_args = [ 'server', 'zone', 'name', 'rtype', 'data' ]
+ takes_args = ['server', 'zone', 'name', 'rtype', 'data' ]
takes_optiongroups = {
"sambaopts": options.SambaOptions,
synopsis = '%prog <server> <zone> <name> <A|AAAA|PTR|CNAME|NS|MX|SOA|SRV|TXT> <olddata> <newdata>'
- takes_args = [ 'server', 'zone', 'name', 'rtype', 'olddata', 'newdata' ]
+ takes_args = ['server', 'zone', 'name', 'rtype', 'olddata', 'newdata' ]
takes_optiongroups = {
"sambaopts": options.SambaOptions,
synopsis = '%prog <server> <zone> <name> <A|AAAA|PTR|CNAME|NS|MX|SRV|TXT> <data>'
- takes_args = [ 'server', 'zone', 'name', 'rtype', 'data' ]
+ takes_args = ['server', 'zone', 'name', 'rtype', 'data' ]
takes_optiongroups = {
"sambaopts": options.SambaOptions,
try:
if remote_trust_info:
self.outf.write("Creating remote TDO.\n")
- current_request = { "location": "remote", "name": "CreateTrustedDomainEx2"}
+ current_request = {"location": "remote", "name": "CreateTrustedDomainEx2"}
remote_tdo_handle = \
remote_lsa.CreateTrustedDomainEx2(remote_policy,
remote_trust_info,
self.outf.write("Remote TDO created.\n")
if enc_types:
self.outf.write("Setting supported encryption types on remote TDO.\n")
- current_request = { "location": "remote", "name": "SetInformationTrustedDomain"}
+ current_request = {"location": "remote", "name": "SetInformationTrustedDomain"}
remote_lsa.SetInformationTrustedDomain(remote_tdo_handle,
lsa.LSA_TRUSTED_DOMAIN_SUPPORTED_ENCRYPTION_TYPES,
enc_types)
self.outf.write("Creating local TDO.\n")
- current_request = { "location": "local", "name": "CreateTrustedDomainEx2"}
+ current_request = {"location": "local", "name": "CreateTrustedDomainEx2"}
local_tdo_handle = local_lsa.CreateTrustedDomainEx2(local_policy,
local_trust_info,
local_auth_info,
self.outf.write("Local TDO created\n")
if enc_types:
self.outf.write("Setting supported encryption types on local TDO.\n")
- current_request = { "location": "local", "name": "SetInformationTrustedDomain"}
+ current_request = {"location": "local", "name": "SetInformationTrustedDomain"}
local_lsa.SetInformationTrustedDomain(local_tdo_handle,
lsa.LSA_TRUSTED_DOMAIN_SUPPORTED_ENCRYPTION_TYPES,
enc_types)
res = samdb.search(base=trusteedn, expression="(objectClass=*)",
scope=SCOPE_BASE)
assert(len(res) == 1)
- return ndr_unpack( security.dom_sid,res[0]["objectSid"][0])
+ return ndr_unpack(security.dom_sid,res[0]["objectSid"][0])
def modify_descriptor(self, samdb, object_dn, desc, controls=None):
assert(isinstance(desc, security.descriptor))
def get_domain_sid(self, samdb):
res = samdb.search(base=samdb.domain_dn(),
expression="(objectClass=*)", scope=SCOPE_BASE)
- return ndr_unpack( security.dom_sid,res[0]["objectSid"][0])
+ return ndr_unpack(security.dom_sid,res[0]["objectSid"][0])
def add_ace(self, samdb, object_dn, new_ace):
"""Add new ace explicitly."""
d = g.split(';')
if len(d) != 2 or not d[0].startswith("[LDAP://"):
raise RuntimeError("Badly formed gPLink '%s'" % g)
- ret.append({ 'dn' : d[0][8:], 'options' : int(d[1])})
+ ret.append({'dn' : d[0][8:], 'options' : int(d[1])})
return ret
def copy_directory_remote_to_local(conn, remotedir, localdir):
if not os.path.isdir(localdir):
os.mkdir(localdir)
- r_dirs = [ remotedir ]
- l_dirs = [ localdir ]
+ r_dirs = [remotedir ]
+ l_dirs = [localdir ]
while r_dirs:
r_dir = r_dirs.pop()
l_dir = l_dirs.pop()
ignore_existing=False):
if not conn.chkpath(remotedir):
conn.mkdir(remotedir)
- l_dirs = [ localdir ]
- r_dirs = [ remotedir ]
+ l_dirs = [localdir ]
+ r_dirs = [remotedir ]
while l_dirs:
l_dir = l_dirs.pop()
r_dir = r_dirs.pop()
except Exception:
raise CommandError("Failed to find objectClass for user %s" % username)
- session_info_flags = ( AUTH_SESSION_INFO_DEFAULT_GROUPS |
+ session_info_flags = (AUTH_SESSION_INFO_DEFAULT_GROUPS |
AUTH_SESSION_INFO_AUTHENTICATED )
# When connecting to a remote server, don't look up the local privilege DB
if found:
raise CommandError("GPO '%s' already linked to this container" % gpo)
else:
- gplist.insert(0, { 'dn' : gpo_dn, 'options' : gplink_options })
+ gplist.insert(0, {'dn' : gpo_dn, 'options' : gplink_options })
else:
gplist = []
- gplist.append({ 'dn' : gpo_dn, 'options' : gplink_options })
+ gplist.append({'dn' : gpo_dn, 'options' : gplink_options })
gplink_str = encode_gplink(gplist)
"credopts": options.CredentialsOptions,
}
- takes_args = [ 'container_dn', 'inherit_state' ]
+ takes_args = ['container_dn', 'inherit_state' ]
takes_options = [
Option("-H", help="LDB URL for database or target server", type=str)
self.samdb.add(m)
# Get new security descriptor
- ds_sd_flags = ( security.SECINFO_OWNER |
+ ds_sd_flags = (security.SECINFO_OWNER |
security.SECINFO_GROUP |
security.SECINFO_DACL )
msg = get_gpo_info(self.samdb, gpo=gpo, sd_flags=ds_sd_flags)[0]
create_directory_hier(conn, sharepath)
# Set ACL
- sio = ( security.SECINFO_OWNER |
+ sio = (security.SECINFO_OWNER |
security.SECINFO_GROUP |
security.SECINFO_DACL |
security.SECINFO_PROTECTED_DACL )
type=str, metavar="URL", dest="H"),
]
- takes_args = [ "groupname", "new_parent_dn" ]
+ takes_args = ["groupname", "new_parent_dn" ]
takes_optiongroups = {
"sambaopts": options.SambaOptions,
"credopts": options.CredentialsOptions,
attrs = group_attrs.split(",")
filter = ("(&(sAMAccountType=%d)(sAMAccountName=%s))" %
- ( ATYPE_SECURITY_GLOBAL_GROUP,
+ (ATYPE_SECURITY_GLOBAL_GROUP,
ldb.binary_encode(groupname)))
domaindn = samdb.domain_dn()
self_ace_fixed = "%s" % self.fix_sid(self_ace)
other_ace_fixed = "%s" % other.fix_sid(other_ace)
if self_ace_fixed != other_ace_fixed:
- res += "%60s * %s\n" % ( self_ace_fixed, other_ace_fixed )
+ res += "%60s * %s\n" % (self_ace_fixed, other_ace_fixed )
flag = False
else:
- res += "%60s | %s\n" % ( self_ace_fixed, other_ace_fixed )
+ res += "%60s | %s\n" % (self_ace_fixed, other_ace_fixed )
i += 1
return (flag, res)
other_aces = []
self_dacl_list_fixed = []
other_dacl_list_fixed = []
- [self_dacl_list_fixed.append( self.fix_sid(ace) ) for ace in self.dacl_list]
- [other_dacl_list_fixed.append( other.fix_sid(ace) ) for ace in other.dacl_list]
+ [self_dacl_list_fixed.append(self.fix_sid(ace) ) for ace in self.dacl_list]
+ [other_dacl_list_fixed.append(other.fix_sid(ace) ) for ace in other.dacl_list]
for ace in self_dacl_list_fixed:
try:
other_dacl_list_fixed.index(ace)
self.domain_attributes = [x.upper() for x in self.domain_attributes]
#
# May contain DOMAIN_NETBIOS and SERVER_NAME
- self.servername_attributes = [ "distinguishedName", "name", "CN", "sAMAccountName", "dNSHostName",
+ self.servername_attributes = ["distinguishedName", "name", "CN", "sAMAccountName", "dNSHostName",
"servicePrincipalName", "rIDSetReferences", "serverReference", "serverReferenceBL",
"msDS-IsDomainFor", "interSiteTopologyGenerator",]
self.servername_attributes = [x.upper() for x in self.servername_attributes]
#
- self.netbios_attributes = [ "servicePrincipalName", "CN", "distinguishedName", "nETBIOSName", "name",]
+ self.netbios_attributes = ["servicePrincipalName", "CN", "distinguishedName", "nETBIOSName", "name",]
self.netbios_attributes = [x.upper() for x in self.netbios_attributes]
#
- self.other_attributes = [ "name", "DC",]
+ self.other_attributes = ["name", "DC",]
self.other_attributes = [x.upper() for x in self.other_attributes]
#
self.ignore_attributes = [x.upper() for x in self.ignore_attributes]
def __eq__(self, other):
res = True
if self.size != other.size:
- self.log( "\n* DN lists have different size: %s != %s" % (self.size, other.size) )
+ self.log("\n* DN lists have different size: %s != %s" % (self.size, other.size) )
if not self.skip_missing_dn:
res = False
#
for x in self.dn_list:
if not x.upper() in [q.upper() for q in other.dn_list]:
if title and not self.skip_missing_dn:
- self.log( title )
+ self.log(title )
title = None
res = False
- self.log( 4*" " + x )
+ self.log(4*" " + x )
self.dn_list[self.dn_list.index(x)] = ""
self.dn_list = [x for x in self.dn_list if x]
#
for x in other.dn_list:
if not x.upper() in [q.upper() for q in self.dn_list]:
if title and not self.skip_missing_dn:
- self.log( title )
+ self.log(title )
title = None
res = False
- self.log( 4*" " + x )
+ self.log(4*" " + x )
other.dn_list[other.dn_list.index(x)] = ""
other.dn_list = [x for x in other.dn_list if x]
#
other.update_size()
assert self.size == other.size
assert sorted([x.upper() for x in self.dn_list]) == sorted([x.upper() for x in other.dn_list])
- self.log( "\n* Objects to be compared: %s" % self.size )
+ self.log("\n* Objects to be compared: %s" % self.size )
index = 0
while index < self.size:
except LdbError as e:
(enum, estr) = e.args
if enum == ERR_NO_SUCH_OBJECT:
- self.log( "\n!!! Object not found: %s" % self.dn_list[index] )
+ self.log("\n!!! Object not found: %s" % self.dn_list[index] )
skip = True
raise
try:
except LdbError as e1:
(enum, estr) = e1.args
if enum == ERR_NO_SUCH_OBJECT:
- self.log( "\n!!! Object not found: %s" % other.dn_list[index] )
+ self.log("\n!!! Object not found: %s" % other.dn_list[index] )
skip = True
raise
if skip:
continue
if object1 == object2:
if self.con.verbose:
- self.log( "\nComparing:" )
- self.log( "'%s' [%s]" % (object1.dn, object1.con.host) )
- self.log( "'%s' [%s]" % (object2.dn, object2.con.host) )
- self.log( 4*" " + "OK" )
+ self.log("\nComparing:" )
+ self.log("'%s' [%s]" % (object1.dn, object1.con.host) )
+ self.log("'%s' [%s]" % (object2.dn, object2.con.host) )
+ self.log(4*" " + "OK" )
else:
- self.log( "\nComparing:" )
- self.log( "'%s' [%s]" % (object1.dn, object1.con.host) )
- self.log( "'%s' [%s]" % (object2.dn, object2.con.host) )
- self.log( object1.screen_output )
- self.log( 4*" " + "FAILED" )
+ self.log("\nComparing:" )
+ self.log("'%s' [%s]" % (object1.dn, object1.con.host) )
+ self.log("'%s' [%s]" % (object2.dn, object2.con.host) )
+ self.log(object1.screen_output )
+ self.log(4*" " + "FAILED" )
res = False
self.summary = object1.summary
other.summary = object2.summary
self.summary["df_value_attrs"] = list(set(self.summary["df_value_attrs"]))
#
if self.summary["unique_attrs"]:
- self.log( "\nAttributes found only in %s:" % self.con.host )
- self.log( "".join([str("\n" + 4*" " + x) for x in self.summary["unique_attrs"]]) )
+ self.log("\nAttributes found only in %s:" % self.con.host )
+ self.log("".join([str("\n" + 4*" " + x) for x in self.summary["unique_attrs"]]) )
#
if self.summary["df_value_attrs"]:
- self.log( "\nAttributes with different values:" )
- self.log( "".join([str("\n" + 4*" " + x) for x in self.summary["df_value_attrs"]]) )
+ self.log("\nAttributes with different values:" )
+ self.log("".join([str("\n" + 4*" " + x) for x in self.summary["df_value_attrs"]]) )
self.summary["df_value_attrs"] = []
action='store_true', help="List objects recursively."),
]
- takes_args = [ "ou_dn" ]
+ takes_args = ["ou_dn" ]
takes_optiongroups = {
"sambaopts": options.SambaOptions,
"credopts": options.CredentialsOptions,
flags = searchflags.split(',')
# We have to normalise all the values. To achieve this predictably
# we title case (Fattrindex), then swapcase (fATTINDEX)
- flags = [ x.capitalize().swapcase() for x in flags ]
+ flags = [x.capitalize().swapcase() for x in flags ]
for flag in flags:
if flag not in bitFields['searchflags'].keys():
raise CommandError("Unknown flag '%s', please see --help" % flag)
type=str, metavar="URL", dest="H"),
]
- takes_args = [ "username", "new_parent_dn" ]
+ takes_args = ["username", "new_parent_dn" ]
takes_optiongroups = {
"sambaopts": options.SambaOptions,
"credopts": options.CredentialsOptions,
scope=ldb.SCOPE_BASE, attrs=["objectGUID",
"objectSid","msDS-Behavior-Version" ])
names.domainguid = str(ndr_unpack(misc.GUID, res6[0]["objectGUID"][0]))
- names.domainsid = ndr_unpack( security.dom_sid, res6[0]["objectSid"][0])
- names.forestsid = ndr_unpack( security.dom_sid, res6[0]["objectSid"][0])
+ names.domainsid = ndr_unpack(security.dom_sid, res6[0]["objectSid"][0])
+ names.forestsid = ndr_unpack(security.dom_sid, res6[0]["objectSid"][0])
if res6[0].get("msDS-Behavior-Version") is None or \
int(res6[0]["msDS-Behavior-Version"][0]) < DS_DOMAIN_FUNCTION_2000:
names.domainlevel = DS_DOMAIN_FUNCTION_2000
secretsdb.modify(msg)
secretsdb.rename(res[0].dn, msg.dn)
else:
- spn = [ 'HOST/%s' % shortname ]
+ spn = ['HOST/%s' % shortname ]
if secure_channel_type == SEC_CHAN_BDC and dnsname is not None:
# we are a domain controller then we add servicePrincipalName
# entries for the keytab code to update.
- spn.extend([ 'HOST/%s' % dnsname ])
+ spn.extend(['HOST/%s' % dnsname ])
msg["servicePrincipalName"] = spn
secretsdb.add(msg)
logger.info("Setting up sam.ldb rootDSE marking as synchronized")
setup_modify_ldif(samdb, setup_path("provision_rootdse_modify.ldif"),
- { 'NTDSGUID' : names.ntdsguid })
+ {'NTDSGUID' : names.ntdsguid })
# fix any dangling GUIDs from the provision
logger.info("Fixing provision GUIDs")
ldap_backend_extra_port=None, ldap_dryrun_mode=False,
ol_mmr_urls=None, nosync=False, ldap_backend_forced_uri=None):
from samba.provision import setup_path
- super(OpenLDAPBackend, self).__init__( backend_type=backend_type,
+ super(OpenLDAPBackend, self).__init__(backend_type=backend_type,
paths=paths, lp=lp,
names=names, logger=logger,
domainsid=domainsid, schema=schema, hostname=hostname,
cn_samba = read_and_sub_file(
setup_path("cn=samba.ldif"),
- { "LDAPADMINPASS": self.ldapadminpass,
+ {"LDAPADMINPASS": self.ldapadminpass,
"MMR_PASSWORD": mmr_pass,
"MMR": mmr })
if lnkattr[attr] is not None:
refint_config += read_and_sub_file(
setup_path("fedorads-refint-add.ldif"),
- { "ARG_NUMBER" : str(argnum),
- "LINK_ATTR" : attr })
+ {"ARG_NUMBER" : str(argnum),
+ "LINK_ATTR" : attr })
memberof_config += read_and_sub_file(
setup_path("fedorads-linked-attributes.ldif"),
- { "MEMBER_ATTR" : attr,
- "MEMBEROF_ATTR" : lnkattr[attr] })
+ {"MEMBER_ATTR" : attr,
+ "MEMBEROF_ATTR" : lnkattr[attr] })
index_config += read_and_sub_file(
- setup_path("fedorads-index.ldif"), { "ATTR" : attr })
+ setup_path("fedorads-index.ldif"), {"ATTR" : attr })
argnum += 1
f = open(self.refint_ldif, 'w')
attr = "nsUniqueId"
index_config += read_and_sub_file(
- setup_path("fedorads-index.ldif"), { "ATTR" : attr })
+ setup_path("fedorads-index.ldif"), {"ATTR" : attr })
f = open(self.index_ldif, 'w')
try:
raise DemoteException("lookup of %s failed: %s" % (a_name, estr))
orig_num_recs = len(a_recs)
- a_recs = [ r for r in a_recs if not a_rec_to_remove(r) ]
+ a_recs = [r for r in a_recs if not a_rec_to_remove(r) ]
if len(a_recs) != orig_num_recs:
logger.info("updating %s keeping %d values, removing %s values" % \
continue
# Remove references to dnsHostName in A, AAAA, NS, CNAME and SRV
- values = [ ndr_unpack(dnsp.DnssrvRpcRecord, v)
+ values = [ndr_unpack(dnsp.DnssrvRpcRecord, v)
for v in orig_values if not to_remove(v) ]
if len(values) != len(orig_values):
# testenv may still exist at the point that tests start on another testenv.
rand = randint(1, 10000000)
dn = ldb.Dn(samdb, "OU=%s%d,%s" %(name, rand, samdb.get_default_basedn()))
- samdb.add({ "dn": dn, "objectclass": "organizationalUnit"})
+ samdb.add({"dn": dn, "objectclass": "organizationalUnit"})
return dn
import os
from samba.tests import BlackboxTestCase
-for p in [ "../../../../../source4/librpc/tests", "../../../../../librpc/tests"]:
+for p in ["../../../../../source4/librpc/tests", "../../../../../librpc/tests"]:
data_path_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), p))
print(data_path_dir)
if os.path.exists(data_path_dir):
from samba.dcerpc.security import dom_sid
import os
-for p in [ "../../../../../testdata/samba3", "../../../../testdata/samba3" ]:
+for p in ["../../../../../testdata/samba3", "../../../../testdata/samba3" ]:
DATADIR = os.path.join(os.path.dirname(__file__), p)
if os.path.exists(DATADIR):
break
import os
-for p in [ "../../../../../testdata/samba3", "../../../../testdata/samba3" ]:
+for p in ["../../../../../testdata/samba3", "../../../../testdata/samba3" ]:
DATADIR = os.path.join(os.path.dirname(__file__), p)
if os.path.exists(DATADIR):
break
import os
-for p in [ "../../../../../testdata/samba3", "../../../../testdata/samba3" ]:
+for p in ["../../../../../testdata/samba3", "../../../../testdata/samba3" ]:
DATADIR = os.path.join(os.path.dirname(__file__), p)
if os.path.exists(DATADIR):
break
import os
-for p in [ "../../../../../testdata/samba3", "../../../../testdata/samba3" ]:
+for p in ["../../../../../testdata/samba3", "../../../../testdata/samba3" ]:
DATADIR = os.path.join(os.path.dirname(__file__), p)
if os.path.exists(DATADIR):
break
import os
-for p in [ "../../../../../testdata/samba3", "../../../../testdata/samba3" ]:
+for p in ["../../../../../testdata/samba3", "../../../../testdata/samba3" ]:
DATADIR = os.path.join(os.path.dirname(__file__), p)
if os.path.exists(DATADIR):
break
def read_datafile(filename):
- paths = [ "../../../../../testdata/samba3",
+ paths = ["../../../../../testdata/samba3",
"../../../../testdata/samba3" ]
for p in paths:
datadir = os.path.join(os.path.dirname(__file__), p)
self.assertCmdSuccess(result, out, err, "Ensure syncpasswords --cache-ldb-initialize runs")
self.assertEqual(err,"","getpassword without url")
cache_attrs = {
- "objectClass": { "value": "userSyncPasswords" },
- "samdbUrl": { },
- "dirsyncFilter": { },
- "dirsyncAttribute": { },
- "dirsyncControl": { "value": "dirsync:1:0:0"},
- "passwordAttribute": { },
- "decryptSambaGPG": { },
- "currentTime": { },
+ "objectClass": {"value": "userSyncPasswords" },
+ "samdbUrl": {},
+ "dirsyncFilter": {},
+ "dirsyncAttribute": {},
+ "dirsyncControl": {"value": "dirsync:1:0:0"},
+ "passwordAttribute": {},
+ "decryptSambaGPG": {},
+ "currentTime": {},
}
for a in cache_attrs.keys():
v = cache_attrs[a].get("value", "")
"retry" : "."
}
-defaulttasks = [ "ctdb",
+defaulttasks = ["ctdb",
"samba",
"samba-nt4",
"samba-fileserver",
extra_python = "--extra-python=/usr/bin/python3"
tasks = {
- "ctdb" : [ ("random-sleep", "../script/random-sleep.sh 60 600", "text/plain"),
+ "ctdb" : [("random-sleep", "../script/random-sleep.sh 60 600", "text/plain"),
("configure", "./configure " + ctdb_configure_params, "text/plain"),
("make", "make all", "text/plain"),
("install", "make install", "text/plain"),
("clean", "make clean", "text/plain") ],
# We have 'test' before 'install' because, 'test' should work without 'install (runs ad_dc_ntvfs and all the other envs)'
- "samba" : [ ("configure", "./configure.developer --with-selftest-prefix=./bin/ab" + samba_configure_params, "text/plain"),
+ "samba" : [("configure", "./configure.developer --with-selftest-prefix=./bin/ab" + samba_configure_params, "text/plain"),
("make", "make -j", "text/plain"),
("test", "make test FAIL_IMMEDIATELY=1 "
"TESTS='--exclude-env=none "
("clean", "make clean", "text/plain") ],
# We split out this so the isolated nt4_dc tests do not wait for ad_dc or ad_dc_ntvfs tests (which are long)
- "samba-nt4" : [ ("random-sleep", "script/random-sleep.sh 60 600", "text/plain"),
+ "samba-nt4" : [("random-sleep", "script/random-sleep.sh 60 600", "text/plain"),
("configure", "./configure.developer --without-ads --with-selftest-prefix=./bin/ab" + samba_configure_params, "text/plain"),
("make", "make -j", "text/plain"),
("test", "make test FAIL_IMMEDIATELY=1 TESTS='--include-env=nt4_dc --include-env=nt4_member'", "text/plain"),
("clean", "make clean", "text/plain") ],
# We split out this so the isolated ad_dc tests do not wait for ad_dc_ntvfs tests (which are long)
- "samba-fileserver" : [ ("random-sleep", "script/random-sleep.sh 60 600", "text/plain"),
+ "samba-fileserver" : [("random-sleep", "script/random-sleep.sh 60 600", "text/plain"),
("configure", "./configure.developer --without-ad-dc --without-ldap --without-ads --without-json-audit --with-selftest-prefix=./bin/ab" + samba_configure_params, "text/plain"),
("make", "make -j", "text/plain"),
("test", "make test FAIL_IMMEDIATELY=1 TESTS='--include-env=fileserver'", "text/plain"),
("check-clean-tree", "script/clean-source-tree.sh", "text/plain")],
# We split out this so the isolated ad_dc tests do not wait for ad_dc_ntvfs tests (which are long)
- "samba-ad-dc" : [ ("random-sleep", "script/random-sleep.sh 60 600", "text/plain"),
+ "samba-ad-dc" : [("random-sleep", "script/random-sleep.sh 60 600", "text/plain"),
("configure", "./configure.developer --with-selftest-prefix=./bin/ab" + samba_configure_params, "text/plain"),
("make", "make -j", "text/plain"),
("test", "make test FAIL_IMMEDIATELY=1 TESTS='"
("check-clean-tree", "script/clean-source-tree.sh", "text/plain")],
# We split out this so the isolated ad_dc tests do not wait for ad_dc_ntvfs tests (which are long)
- "samba-ad-dc-2" : [ ("random-sleep", "script/random-sleep.sh 60 600", "text/plain"),
+ "samba-ad-dc-2" : [("random-sleep", "script/random-sleep.sh 60 600", "text/plain"),
("configure", "./configure.developer --with-selftest-prefix=./bin/ab" + samba_configure_params, "text/plain"),
("make", "make -j", "text/plain"),
("test", "make test FAIL_IMMEDIATELY=1 TESTS='--include-env=chgdcpass --include-env=vampire_2000_dc --include-env=fl2000dc'", "text/plain"),
("check-clean-tree", "script/clean-source-tree.sh", "text/plain")],
- "samba-test-only" : [ ("configure", "./configure.developer --with-selftest-prefix=./bin/ab --abi-check-disable" + samba_configure_params, "text/plain"),
+ "samba-test-only" : [("configure", "./configure.developer --with-selftest-prefix=./bin/ab --abi-check-disable" + samba_configure_params, "text/plain"),
("make", "make -j", "text/plain"),
("test", 'make test FAIL_IMMEDIATELY=1 TESTS="${TESTS}"',"text/plain") ],
# Test cross-compile infrastructure
- "samba-xc" : [ ("random-sleep", "script/random-sleep.sh 60 600", "text/plain"),
+ "samba-xc" : [("random-sleep", "script/random-sleep.sh 60 600", "text/plain"),
("configure-native", "./configure.developer --with-selftest-prefix=./bin/ab" + samba_configure_params, "text/plain"),
("configure-cross-execute", "./configure.developer -b ./bin-xe --cross-compile --cross-execute=script/identity_cc.sh" \
" --cross-answers=./bin-xe/cross-answers.txt --with-selftest-prefix=./bin-xe/ab" + samba_configure_params, "text/plain"),
("compare-results", "script/compare_cc_results.py ./bin/c4che/default.cache.py ./bin-xe/c4che/default.cache.py ./bin-xa/c4che/default.cache.py", "text/plain")],
# test build with -O3 -- catches extra warnings and bugs, tests the ad_dc environments
- "samba-o3" : [ ("random-sleep", "script/random-sleep.sh 60 600", "text/plain"),
+ "samba-o3" : [("random-sleep", "script/random-sleep.sh 60 600", "text/plain"),
("configure", "ADDITIONAL_CFLAGS='-O3' ./configure.developer --with-selftest-prefix=./bin/ab --abi-check-disable" + samba_configure_params, "text/plain"),
("make", "make -j", "text/plain"),
("test", "make quicktest FAIL_IMMEDIATELY=1 TESTS='--include-env=ad_dc'", "text/plain"),
("check-clean-tree", "script/clean-source-tree.sh", "text/plain"),
("clean", "make clean", "text/plain") ],
- "samba-ctdb" : [ ("random-sleep", "script/random-sleep.sh 60 600", "text/plain"),
+ "samba-ctdb" : [("random-sleep", "script/random-sleep.sh 60 600", "text/plain"),
# make sure we have tdb around:
("tdb-configure", "cd lib/tdb && PYTHONPATH=${PYTHON_PREFIX}/site-packages:$PYTHONPATH PKG_CONFIG_PATH=$PKG_CONFIG_PATH:${PREFIX_DIR}/lib/pkgconfig ./configure --bundled-libraries=NONE --abi-check --enable-debug -C ${PREFIX}", "text/plain"),
("clean", "make clean", "text/plain") ],
# these are useful for debugging autobuild
- 'pass' : [ ("pass", 'echo passing && /bin/true', "text/plain") ],
- 'fail' : [ ("fail", 'echo failing && /bin/false', "text/plain") ]
+ 'pass' : [("pass", 'echo passing && /bin/true', "text/plain") ],
+ 'fail' : [("fail", 'echo failing && /bin/false', "text/plain") ]
}
def do_print(msg):
self.tlist.append(b)
if options.retry:
rebase_remote = "rebaseon"
- retry_task = [ ("retry",
+ retry_task = [("retry",
'''set -e
git remote add -t %s %s %s
git fetch %s
#print domain
#print domsid
-sids=[ domsid + '-512', 'S-1-5-32-545', domsid + '-513', 'S-1-1-0', 'S-1-3-1', 'S-1-5-1' ]
+sids=[domsid + '-512', 'S-1-5-32-545', domsid + '-513', 'S-1-1-0', 'S-1-3-1', 'S-1-5-1' ]
flush_cache(sids=sids)
t = "--krb5auth=$DOMAIN/$DC_USERNAME%$DC_PASSWORD"
plantestsuite("samba3.wbinfo_simple.(%s:local).%s" % (env, t), "%s:local" % env, [os.path.join(srcdir(), "nsswitch/tests/test_wbinfo_simple.sh"), t])
plantestsuite("samba3.wbinfo_name_lookup", env,
- [ os.path.join(srcdir(),
+ [os.path.join(srcdir(),
"nsswitch/tests/test_wbinfo_name_lookup.sh"),
'$DOMAIN', '$REALM', '$DC_USERNAME' ])
env = "ad_member:local"
plantestsuite("samba3.wbinfo_user_info", env,
- [ os.path.join(srcdir(),
+ [os.path.join(srcdir(),
"nsswitch/tests/test_wbinfo_user_info.sh"),
'$DOMAIN', '$REALM', '$DOMAIN', 'alice', 'alice', 'jane', 'jane.doe' ])
env = "fl2008r2dc:local"
plantestsuite("samba3.wbinfo_user_info", env,
- [ os.path.join(srcdir(),
+ [os.path.join(srcdir(),
"nsswitch/tests/test_wbinfo_user_info.sh"),
'$TRUST_DOMAIN', '$TRUST_REALM', '$DOMAIN', 'alice', 'alice', 'jane', 'jane.doe' ])
configuration, os.path.join(bindir(), "sharesec"), "tmp"])
plantestsuite("samba3.blackbox.net_tdb", "simpleserver:local",
- [ os.path.join(samba3srcdir, "script/tests/test_net_tdb.sh"),
+ [os.path.join(samba3srcdir, "script/tests/test_net_tdb.sh"),
smbclient3, '$SERVER', 'tmp', '$USERNAME', '$PASSWORD',
configuration, '$LOCAL_PATH', '$LOCK_DIR' ])
plantestsuite("samba3.blackbox.smbd_error", "simpleserver:local",
- [ os.path.join(samba3srcdir, "script/tests/test_smbd_error.sh") ])
+ [os.path.join(samba3srcdir, "script/tests/test_smbd_error.sh") ])
plantestsuite("samba3.blackbox.net_cache_samlogon", "ad_member:local",
- [ os.path.join(samba3srcdir, "script/tests/test_net_cache_samlogon.sh"),
+ [os.path.join(samba3srcdir, "script/tests/test_net_cache_samlogon.sh"),
'$SERVER', 'tmp', '$DC_USERNAME', '$DC_PASSWORD'])
plantestsuite("samba3.blackbox.net_dom_join_fail_dc", "nt4_dc",
classinfo[name]["objectClassCategory"] = int(r["objectClassCategory"][0])
else:
classinfo[name]["objectClassCategory"] = 0
- for a in [ "possSuperiors", "systemPossSuperiors",
+ for a in ["possSuperiors", "systemPossSuperiors",
"auxiliaryClass", "systemAuxiliaryClass",
"subClassOf" ]:
classinfo[name][a] = []
self.ldb_user = self.get_ldb_connection(self.user_with_wp, self.user_pass)
self.ldb_user2 = self.get_ldb_connection(self.user_with_sm, self.user_pass)
self.ldb_user3 = self.get_ldb_connection(self.user_with_group_sm, self.user_pass)
- self.user_sid = self.sd_utils.get_object_sid( self.get_user_dn(self.user_with_wp))
+ self.user_sid = self.sd_utils.get_object_sid(self.get_user_dn(self.user_with_wp))
self.ldb_admin.newgroup("test_modify_group2", grouptype=samba.dsdb.GTYPE_DISTRIBUTION_DOMAIN_LOCAL_GROUP)
self.ldb_admin.newgroup("test_modify_group3", grouptype=samba.dsdb.GTYPE_DISTRIBUTION_DOMAIN_LOCAL_GROUP)
self.ldb_admin.newuser("test_modify_user2", self.user_pass)
mod = "(OA;;SW;bf9679c0-0de6-11d0-a285-00aa003049e2;;%s)" % str(user_sid)
self.sd_utils.dacl_add_ace("CN=test_modify_group2,CN=Users," + self.base_dn, mod)
self.ldb_user2.modify_ldif(ldif)
- res = self.ldb_admin.search( self.base_dn, expression="(distinguishedName=%s)" \
+ res = self.ldb_admin.search(self.base_dn, expression="(distinguishedName=%s)" \
% ("CN=test_modify_group2,CN=Users," + self.base_dn), attrs=["Member"])
self.assertEqual(res[0]["Member"][0], self.get_user_dn(self.user_with_sm))
#but not other users
add: Member
Member: """ + self.get_user_dn(self.user_with_wp)
self.ldb_user.modify_ldif(ldif)
- res = self.ldb_admin.search( self.base_dn, expression="(distinguishedName=%s)" \
+ res = self.ldb_admin.search(self.base_dn, expression="(distinguishedName=%s)" \
% ("CN=test_modify_group2,CN=Users," + self.base_dn), attrs=["Member"])
self.assertEqual(res[0]["Member"][0], self.get_user_dn(self.user_with_wp))
ldif = """
add: Member
Member: CN=test_modify_user2,CN=Users,""" + self.base_dn
self.ldb_user.modify_ldif(ldif)
- res = self.ldb_admin.search( self.base_dn, expression="(distinguishedName=%s)" \
+ res = self.ldb_admin.search(self.base_dn, expression="(distinguishedName=%s)" \
% ("CN=test_modify_group2,CN=Users," + self.base_dn), attrs=["Member"])
self.assertEqual(res[0]["Member"][0], "CN=test_modify_user2,CN=Users," + self.base_dn)
ok_list = [Dn(self.ldb_admin, "OU=ou2,OU=ou1," + self.base_dn),
Dn(self.ldb_admin, "OU=ou1," + self.base_dn)]
- res_list = [ x["dn"] for x in res if x["dn"] in ok_list ]
+ res_list = [x["dn"] for x in res if x["dn"] in ok_list ]
self.assertEquals(sorted(res_list), sorted(ok_list))
#these users should see all ous
res = self.ldb_user.search("OU=ou1," + self.base_dn, expression="(objectClass=*)",
scope=SCOPE_SUBTREE)
self.assertEquals(len(res), 6)
- res_list = [ x["dn"] for x in res if x["dn"] in self.full_list ]
+ res_list = [x["dn"] for x in res if x["dn"] in self.full_list ]
self.assertEquals(sorted(res_list), sorted(self.full_list))
res = self.ldb_user2.search("OU=ou1," + self.base_dn, expression="(objectClass=*)",
scope=SCOPE_SUBTREE)
self.assertEquals(len(res), 6)
- res_list = [ x["dn"] for x in res if x["dn"] in self.full_list ]
+ res_list = [x["dn"] for x in res if x["dn"] in self.full_list ]
self.assertEquals(sorted(res_list), sorted(self.full_list))
def test_search2(self):
res = self.ldb_user3.search("OU=ou1," + self.base_dn, expression="(objectClass=*)",
scope=SCOPE_SUBTREE)
#this user should see all ous
- res_list = [ x["dn"] for x in res if x["dn"] in self.full_list ]
+ res_list = [x["dn"] for x in res if x["dn"] in self.full_list ]
self.assertEquals(sorted(res_list), sorted(self.full_list))
#these users should see ou1, 2, 5 and 6 but not 3 and 4
Dn(self.ldb_admin, "OU=ou1," + self.base_dn),
Dn(self.ldb_admin, "OU=ou5,OU=ou3,OU=ou2,OU=ou1," + self.base_dn),
Dn(self.ldb_admin, "OU=ou6,OU=ou4,OU=ou2,OU=ou1," + self.base_dn)]
- res_list = [ x["dn"] for x in res if x["dn"] in ok_list ]
+ res_list = [x["dn"] for x in res if x["dn"] in ok_list ]
self.assertEquals(sorted(res_list), sorted(ok_list))
res = self.ldb_user2.search("OU=ou1," + self.base_dn, expression="(objectClass=*)",
scope=SCOPE_SUBTREE)
self.assertEquals(len(res), 4)
- res_list = [ x["dn"] for x in res if x["dn"] in ok_list ]
+ res_list = [x["dn"] for x in res if x["dn"] in ok_list ]
self.assertEquals(sorted(res_list), sorted(ok_list))
def test_search3(self):
res = self.ldb_user3.search("OU=ou1," + self.base_dn, expression="(objectClass=*)",
scope=SCOPE_SUBTREE)
- res_list = [ x["dn"] for x in res if x["dn"] in ok_list ]
+ res_list = [x["dn"] for x in res if x["dn"] in ok_list ]
self.assertEquals(sorted(res_list), sorted(ok_list))
ok_list = [Dn(self.ldb_admin, "OU=ou2,OU=ou1," + self.base_dn),
res = self.ldb_user.search("OU=ou1," + self.base_dn, expression="(objectClass=*)",
scope=SCOPE_SUBTREE)
self.assertEquals(len(res), 4)
- res_list = [ x["dn"] for x in res if x["dn"] in ok_list ]
+ res_list = [x["dn"] for x in res if x["dn"] in ok_list ]
self.assertEquals(sorted(res_list), sorted(ok_list))
res = self.ldb_user2.search("OU=ou1," + self.base_dn, expression="(objectClass=*)",
scope=SCOPE_SUBTREE)
self.assertEquals(len(res), 4)
- res_list = [ x["dn"] for x in res if x["dn"] in ok_list ]
+ res_list = [x["dn"] for x in res if x["dn"] in ok_list ]
self.assertEquals(sorted(res_list), sorted(ok_list))
def test_search4(self):
res = self.ldb_user3.search("OU=ou1," + self.base_dn, expression="(objectClass=*)",
scope=SCOPE_SUBTREE)
self.assertEquals(len(res), 2)
- res_list = [ x["dn"] for x in res if x["dn"] in ok_list ]
+ res_list = [x["dn"] for x in res if x["dn"] in ok_list ]
self.assertEquals(sorted(res_list), sorted(ok_list))
res = self.ldb_user.search("OU=ou1," + self.base_dn, expression="(objectClass=*)",
scope=SCOPE_SUBTREE)
self.assertEquals(len(res), 2)
- res_list = [ x["dn"] for x in res if x["dn"] in ok_list ]
+ res_list = [x["dn"] for x in res if x["dn"] in ok_list ]
self.assertEquals(sorted(res_list), sorted(ok_list))
def test_search5(self):
# same as for join_RODC, but do not set any SPNs
def create_rodc(self, ctx):
- ctx.nc_list = [ ctx.base_dn, ctx.config_dn, ctx.schema_dn ]
- ctx.full_nc_list = [ ctx.base_dn, ctx.config_dn, ctx.schema_dn ]
+ ctx.nc_list = [ctx.base_dn, ctx.config_dn, ctx.schema_dn ]
+ ctx.full_nc_list = [ctx.base_dn, ctx.config_dn, ctx.schema_dn ]
ctx.krbtgt_dn = "CN=krbtgt_%s,CN=Users,%s" % (ctx.myname, ctx.base_dn)
- ctx.never_reveal_sid = [ "<SID=%s-%s>" % (ctx.domsid, security.DOMAIN_RID_RODC_DENY),
+ ctx.never_reveal_sid = ["<SID=%s-%s>" % (ctx.domsid, security.DOMAIN_RID_RODC_DENY),
"<SID=%s>" % security.SID_BUILTIN_ADMINISTRATORS,
"<SID=%s>" % security.SID_BUILTIN_SERVER_OPERATORS,
"<SID=%s>" % security.SID_BUILTIN_BACKUP_OPERATORS,
ctx.join_add_objects()
def create_dc(self, ctx):
- ctx.nc_list = [ ctx.base_dn, ctx.config_dn, ctx.schema_dn ]
- ctx.full_nc_list = [ ctx.base_dn, ctx.config_dn, ctx.schema_dn ]
+ ctx.nc_list = [ctx.base_dn, ctx.config_dn, ctx.schema_dn ]
+ ctx.full_nc_list = [ctx.base_dn, ctx.config_dn, ctx.schema_dn ]
ctx.userAccountControl = samba.dsdb.UF_SERVER_TRUST_ACCOUNT | samba.dsdb.UF_TRUSTED_FOR_DELEGATION
ctx.secure_channel_type = misc.SEC_CHAN_BDC
ctx.replica_flags = (drsuapi.DRSUAPI_DRS_WRIT_REP |
"objectclass": "group",
"description": "test group",
"samaccountname": grp_name,
- "member": [ self.usr1, self.usr2 ],
+ "member": [self.usr1, self.usr2 ],
"isDeleted": "FALSE" })
self.ldb.add({
#
delete_force(self.ldb, user_dn)
try:
- self.ldb.add({ "dn": user_dn,
+ self.ldb.add({"dn": user_dn,
"objectClass": "user",
"sAMAccountName": user_name,
"nTSecurityDescriptor": [] })
nTSecurityDescriptor: """ + sddl)
res = self.ldb.search(base=user_dn, attrs=["nTSecurityDescriptor"])
desc = res[0]["nTSecurityDescriptor"][0]
- desc = ndr_unpack( security.descriptor, desc )
- desc_sddl = desc.as_sddl( self.domain_sid )
+ desc = ndr_unpack(security.descriptor, desc )
+ desc_sddl = desc.as_sddl(self.domain_sid )
self.assertEqual(desc_sddl, sddl)
finally:
delete_force(self.ldb, user_dn)
try:
sddl = "O:DUG:DUD:AI(A;;RPWP;;;AU)S:PAI"
desc = security.descriptor.from_sddl(sddl, security.dom_sid('S-1-5-21'))
- desc_base64 = base64.b64encode( ndr_pack(desc) ).decode('utf8')
+ desc_base64 = base64.b64encode(ndr_pack(desc) ).decode('utf8')
self.ldb.add_ldif("""
dn: """ + user_dn + """
objectclass: user
# Test an empty security descriptor (naturally this shouldn't work)
#
delete_force(self.ldb, user_dn)
- self.ldb.add({ "dn": user_dn,
+ self.ldb.add({"dn": user_dn,
"objectClass": "user",
"sAMAccountName": user_name })
user_dn = "CN=%s,CN=Users,%s" % (user_name, self.base_dn)
delete_force(self.ldb, user_dn)
- self.ldb.add({ "dn": user_dn,
+ self.ldb.add({"dn": user_dn,
"objectClass": "user",
"sAMAccountName": user_name })
# 370101000000Z => 20370101000000.0Z
# 20370102000000.*Z => 20370102000000.0Z
#
- ext = [ "Z", ".0Z", ".Z", ".000Z", ".RandomIgnoredCharacters...987654321Z" ]
+ ext = ["Z", ".0Z", ".Z", ".000Z", ".RandomIgnoredCharacters...987654321Z" ]
for i in range(0, len(ext)):
v_raw = "203701%02d000000" % (i + 1)
if ext[i] == "Z":
"objectclass": "computer",
"dNSHostName": "testname.testdom",
"sAMAccountName": "testname$",
- "servicePrincipalName": [ "HOST/testname.testdom", "HOST/testname" ]
+ "servicePrincipalName": ["HOST/testname.testdom", "HOST/testname" ]
})
m = Message()
"objectclass": "computer",
"dNSHostName": "testname.testdom",
"sAMAccountName": "testname$",
- "servicePrincipalName": [ "HOST/testname.testdom", "HOST/testname" ]
+ "servicePrincipalName": ["HOST/testname.testdom", "HOST/testname" ]
})
m = Message()
# User is not a member of any additional groups but default
res = self.ldb_admin.search(user_dn, attrs=["*"])
res = [x.upper() for x in res[0].keys()]
- self.assertFalse( "MEMBEROF" in res)
+ self.assertFalse("MEMBEROF" in res)
def check_modify_inheritance(self, _ldb, object_dn, owner_group=""):
# Modify
object_dn = "OU=test_domain_ou1," + self.base_dn
delete_force(self.ldb_admin, object_dn)
self.ldb_admin.create_ou(object_dn)
- user_sid = self.sd_utils.get_object_sid( self.get_users_domain_dn(user_name) )
+ user_sid = self.sd_utils.get_object_sid(self.get_users_domain_dn(user_name) )
mod = "(A;CI;WPWDCC;;;%s)" % str(user_sid)
self.sd_utils.dacl_add_ace(object_dn, mod)
# Create additional object into the first one
object_dn = "OU=test_domain_ou1," + self.base_dn
delete_force(self.ldb_admin, object_dn)
self.ldb_admin.create_ou(object_dn)
- user_sid = self.sd_utils.get_object_sid( self.get_users_domain_dn(user_name) )
+ user_sid = self.sd_utils.get_object_sid(self.get_users_domain_dn(user_name) )
mod = "(A;CI;WPWDCC;;;%s)" % str(user_sid)
self.sd_utils.dacl_add_ace(object_dn, mod)
# Create additional object into the first one
object_dn = "OU=test_domain_ou1," + self.base_dn
delete_force(self.ldb_admin, object_dn)
self.ldb_admin.create_ou(object_dn)
- user_sid = self.sd_utils.get_object_sid( self.get_users_domain_dn(user_name) )
+ user_sid = self.sd_utils.get_object_sid(self.get_users_domain_dn(user_name) )
mod = "(A;CI;WOWDCC;;;%s)" % str(user_sid)
self.sd_utils.dacl_add_ace(object_dn, mod)
# Create a custom security descriptor
object_dn = "OU=test_domain_ou1," + self.base_dn
delete_force(self.ldb_admin, object_dn)
self.ldb_admin.create_ou(object_dn)
- user_sid = self.sd_utils.get_object_sid( self.get_users_domain_dn(user_name) )
+ user_sid = self.sd_utils.get_object_sid(self.get_users_domain_dn(user_name) )
mod = "(A;CI;WOWDCC;;;%s)" % str(user_sid)
self.sd_utils.dacl_add_ace(object_dn, mod)
# Create a custom security descriptor
object_dn = "OU=test_domain_ou1," + self.base_dn
delete_force(self.ldb_admin, object_dn)
self.ldb_admin.create_ou(object_dn)
- user_sid = self.sd_utils.get_object_sid( self.get_users_domain_dn(user_name) )
+ user_sid = self.sd_utils.get_object_sid(self.get_users_domain_dn(user_name) )
mod = "(D;CI;WP;;;S-1-3-0)"
#mod = ""
self.sd_utils.dacl_add_ace(object_dn, mod)
# Open Ldb connection with the tested user
_ldb = self.get_ldb_connection(user_name, "samba123@")
# Change Schema partition descriptor
- user_sid = self.sd_utils.get_object_sid( self.get_users_domain_dn(user_name) )
+ user_sid = self.sd_utils.get_object_sid(self.get_users_domain_dn(user_name) )
mod = "(A;;WDCC;;;AU)"
self.sd_utils.dacl_add_ace(self.schema_dn, mod)
# Create example Schema class
# Open Ldb connection with the tested user
_ldb = self.get_ldb_connection(user_name, "samba123@")
#Change Schema partition descriptor
- user_sid = self.sd_utils.get_object_sid( self.get_users_domain_dn(user_name) )
+ user_sid = self.sd_utils.get_object_sid(self.get_users_domain_dn(user_name) )
mod = "(A;CI;WDCC;;;AU)"
self.sd_utils.dacl_add_ace(self.schema_dn, mod)
# Create example Schema class
_ldb = self.get_ldb_connection(user_name, "samba123@")
# Create a custom security descriptor
# NB! Problematic owner part won't accept DA only <User Sid> !!!
- user_sid = self.sd_utils.get_object_sid( self.get_users_domain_dn(user_name) )
+ user_sid = self.sd_utils.get_object_sid(self.get_users_domain_dn(user_name) )
desc_sddl = "O:%sG:DAD:(A;;RP;;;DU)" % str(user_sid)
# Create example Schema class
class_dn = self.create_schema_class(_ldb, desc_sddl)
_ldb = self.get_ldb_connection(user_name, "samba123@")
# Create a custom security descriptor
# NB! Problematic owner part won't accept DA only <User Sid> !!!
- user_sid = self.sd_utils.get_object_sid( self.get_users_domain_dn(user_name) )
+ user_sid = self.sd_utils.get_object_sid(self.get_users_domain_dn(user_name) )
desc_sddl = "O:%sG:DAD:(A;;RP;;;DU)" % str(user_sid)
# Create example Schema class
class_dn = self.create_schema_class(_ldb, desc_sddl)
object_dn = "CN=test-container1,CN=DisplaySpecifiers," + self.configuration_dn
delete_force(self.ldb_admin, object_dn)
self.create_configuration_container(self.ldb_admin, object_dn, )
- user_sid = self.sd_utils.get_object_sid( self.get_users_domain_dn(user_name) )
+ user_sid = self.sd_utils.get_object_sid(self.get_users_domain_dn(user_name) )
mod = "(A;;WDCC;;;AU)"
self.sd_utils.dacl_add_ace(object_dn, mod)
# Create child object with user's credentials
object_dn = "CN=test-container1,CN=DisplaySpecifiers," + self.configuration_dn
delete_force(self.ldb_admin, object_dn)
self.create_configuration_container(self.ldb_admin, object_dn, )
- user_sid = self.sd_utils.get_object_sid( self.get_users_domain_dn(user_name) )
+ user_sid = self.sd_utils.get_object_sid(self.get_users_domain_dn(user_name) )
mod = "(A;CI;WDCC;;;AU)"
self.sd_utils.dacl_add_ace(object_dn, mod)
# Create child object with user's credentials
object_dn = "CN=test-container1,CN=DisplaySpecifiers," + self.configuration_dn
delete_force(self.ldb_admin, object_dn)
self.create_configuration_container(self.ldb_admin, object_dn, )
- user_sid = self.sd_utils.get_object_sid( self.get_users_domain_dn(user_name) )
+ user_sid = self.sd_utils.get_object_sid(self.get_users_domain_dn(user_name) )
mod = "(A;;CC;;;AU)"
self.sd_utils.dacl_add_ace(object_dn, mod)
# Create child object with user's credentials
object_dn = "CN=test-container1,CN=DisplaySpecifiers," + self.configuration_dn
delete_force(self.ldb_admin, object_dn)
self.create_configuration_container(self.ldb_admin, object_dn, )
- user_sid = self.sd_utils.get_object_sid( self.get_users_domain_dn(user_name) )
+ user_sid = self.sd_utils.get_object_sid(self.get_users_domain_dn(user_name) )
mod = "(A;;CC;;;AU)"
self.sd_utils.dacl_add_ace(object_dn, mod)
# Create child object with user's credentials
self.user_sid_dn = "<SID=%s>" % str(ndr_unpack(samba.dcerpc.security.dom_sid, res[0]["tokenGroups"][0]))
- session_info_flags = ( AUTH_SESSION_INFO_DEFAULT_GROUPS |
+ session_info_flags = (AUTH_SESSION_INFO_DEFAULT_GROUPS |
AUTH_SESSION_INFO_AUTHENTICATED |
AUTH_SESSION_INFO_SIMPLE_PRIVILEGES)
if creds.get_kerberos_state() == DONT_USE_KERBEROS:
self.test_user_dn = res[0].dn
- session_info_flags = ( AUTH_SESSION_INFO_DEFAULT_GROUPS |
+ session_info_flags = (AUTH_SESSION_INFO_DEFAULT_GROUPS |
AUTH_SESSION_INFO_AUTHENTICATED |
AUTH_SESSION_INFO_SIMPLE_PRIVILEGES)
for x in [1, 2, 3]:
start = time.time()
self.create_bundle(num)
- res_add = Decimal( str(time.time() - start) )
+ res_add = Decimal(str(time.time() - start) )
avg_add += res_add
- print(" Attempt %s ADD: %.3fs" % ( x, float(res_add) ))
+ print(" Attempt %s ADD: %.3fs" % (x, float(res_add) ))
#
start = time.time()
self.remove_bundle(num)
- res_del = Decimal( str(time.time() - start) )
+ res_del = Decimal(str(time.time() - start) )
avg_del += res_del
- print(" Attempt %s DEL: %.3fs" % ( x, float(res_del) ))
- print("Average ADD: %.3fs" % float( Decimal(avg_add) / Decimal("3.0") ))
- print("Average DEL: %.3fs" % float( Decimal(avg_del) / Decimal("3.0") ))
+ print(" Attempt %s DEL: %.3fs" % (x, float(res_del) ))
+ print("Average ADD: %.3fs" % float(Decimal(avg_add) / Decimal("3.0") ))
+ print("Average DEL: %.3fs" % float(Decimal(avg_del) / Decimal("3.0") ))
print("")
def test_00000(self):
for x in [1, 2, 3]:
start = time.time()
res = _ldb.search(base=self.base_dn, expression="(objectClass=*)", scope=SCOPE_SUBTREE)
- res_search = Decimal( str(time.time() - start) )
+ res_search = Decimal(str(time.time() - start) )
avg_search += res_search
- print(" Attempt %s SEARCH: %.3fs" % ( x, float(res_search) ))
- print("Average Search: %.3fs" % float( Decimal(avg_search) / Decimal("3.0") ))
+ print(" Attempt %s SEARCH: %.3fs" % (x, float(res_search) ))
+ print("Average Search: %.3fs" % float(Decimal(avg_search) / Decimal("3.0") ))
self.remove_bundle(num)
def get_user_dn(self, name):
plansmbtorture4testsuite("net.api.become.dc", env, '$SERVER[%s] -U$USERNAME%%$PASSWORD -W$DOMAIN' % validate)
for bindoptions in ["sign", "seal"]:
- plansmbtorture4testsuite('rpc.backupkey', "ad_dc_ntvfs", ["ncacn_np:$SERVER[%s]" % ( bindoptions), '-U$USERNAME%$PASSWORD', '--workgroup=$DOMAIN'], "samba4.rpc.backupkey with %s" % (bindoptions))
+ plansmbtorture4testsuite('rpc.backupkey', "ad_dc_ntvfs", ["ncacn_np:$SERVER[%s]" % (bindoptions), '-U$USERNAME%$PASSWORD', '--workgroup=$DOMAIN'], "samba4.rpc.backupkey with %s" % (bindoptions))
for transport in transports:
for bindoptions in ["sign", "seal"]:
planpythontestsuite("ad_dc_ntvfs:local", "samba.tests.kcc.kcc_utils")
-for env in [ "simpleserver", "fileserver", "nt4_dc", "ad_dc", "ad_dc_ntvfs",
- "ad_member", "offlinebackupdc", "restoredc", "renamedc", "labdc" ]:
+for env in ["simpleserver", "fileserver", "nt4_dc", "ad_dc", "ad_dc_ntvfs",
+ "ad_member", "offlinebackupdc", "restoredc", "renamedc", "labdc"]:
planoldpythontestsuite(env, "netlogonsvc",
extra_path=[os.path.join(srcdir(), 'python/samba/tests')],
name="samba.tests.netlogonsvc.python(%s)" % env)
-for env in [ "ktest", "ad_member", "ad_dc_no_ntlm" ]:
+for env in ["ktest", "ad_member", "ad_dc_no_ntlm" ]:
planoldpythontestsuite(env, "ntlmdisabled",
extra_path=[os.path.join(srcdir(), 'python/samba/tests')],
name="samba.tests.ntlmdisabled.python(%s)" % env)
def _create_rodc(self, ctx):
- ctx.nc_list = [ ctx.base_dn, ctx.config_dn, ctx.schema_dn ]
- ctx.full_nc_list = [ ctx.base_dn, ctx.config_dn, ctx.schema_dn ]
+ ctx.nc_list = [ctx.base_dn, ctx.config_dn, ctx.schema_dn ]
+ ctx.full_nc_list = [ctx.base_dn, ctx.config_dn, ctx.schema_dn ]
ctx.krbtgt_dn = "CN=krbtgt_%s,CN=Users,%s" % (ctx.myname, ctx.base_dn)
- ctx.never_reveal_sid = [ "<SID=%s-%s>" % (ctx.domsid, security.DOMAIN_RID_RODC_DENY),
+ ctx.never_reveal_sid = ["<SID=%s-%s>" % (ctx.domsid, security.DOMAIN_RID_RODC_DENY),
"<SID=%s>" % security.SID_BUILTIN_ADMINISTRATORS,
"<SID=%s>" % security.SID_BUILTIN_SERVER_OPERATORS,
"<SID=%s>" % security.SID_BUILTIN_BACKUP_OPERATORS,
# the first replication will transfer the dnsHostname attribute
t.cmd_contains("bin/samba-tool drs replicate ${HOSTNAME}.${LCREALM} ${WIN_HOSTNAME} CN=Configuration,${BASEDN} -k yes", ["was successful"])
- for nc in [ '${BASEDN}', 'CN=Configuration,${BASEDN}', 'CN=Schema,CN=Configuration,${BASEDN}' ]:
+ for nc in ['${BASEDN}', 'CN=Configuration,${BASEDN}', 'CN=Schema,CN=Configuration,${BASEDN}' ]:
t.cmd_contains("bin/samba-tool drs replicate ${HOSTNAME}.${LCREALM} ${WIN_HOSTNAME}.${LCREALM} %s -k yes" % nc, ["was successful"])
t.cmd_contains("bin/samba-tool drs replicate ${WIN_HOSTNAME}.${LCREALM} ${HOSTNAME}.${LCREALM} %s -k yes" % nc, ["was successful"])
t.cmd_contains("bin/samba-tool drs showrepl ${HOSTNAME}.${LCREALM} -k yes",
- [ "INBOUND NEIGHBORS",
+ ["INBOUND NEIGHBORS",
"${BASEDN}",
"Last attempt .* was successful",
"CN=Configuration,${BASEDN}",
regex=True)
t.cmd_contains("bin/samba-tool drs showrepl ${WIN_HOSTNAME}.${LCREALM} -k yes",
- [ "INBOUND NEIGHBORS",
+ ["INBOUND NEIGHBORS",
"${BASEDN}",
"Last attempt .* was successful",
"CN=Configuration,${BASEDN}",
child.expect("CN=Configuration,${BASEDN}")
child.expect("was successful")
- for nc in [ '${BASEDN}', 'CN=Configuration,${BASEDN}', 'CN=Schema,CN=Configuration,${BASEDN}' ]:
+ for nc in ['${BASEDN}', 'CN=Configuration,${BASEDN}', 'CN=Schema,CN=Configuration,${BASEDN}' ]:
t.cmd_contains("bin/samba-tool drs replicate --add-ref ${WIN_HOSTNAME}.${LCREALM} ${HOSTNAME}.${LCREALM} %s" % nc, ["was successful"])
t.cmd_contains("bin/samba-tool drs showrepl ${HOSTNAME}.${LCREALM}",
- [ "INBOUND NEIGHBORS",
+ ["INBOUND NEIGHBORS",
"OUTBOUND NEIGHBORS",
"${BASEDN}",
"Last attempt.*was successful",
t.run_cmd('bin/samba-tool drs kcc ${HOSTNAME}.${WIN_REALM} -Uadministrator@${WIN_REALM}%${WIN_PASS}')
t.kinit("administrator@${WIN_REALM}", "${WIN_PASS}")
- for nc in [ '${WIN_BASEDN}', 'CN=Configuration,${WIN_BASEDN}', 'CN=Schema,CN=Configuration,${WIN_BASEDN}' ]:
+ for nc in ['${WIN_BASEDN}', 'CN=Configuration,${WIN_BASEDN}', 'CN=Schema,CN=Configuration,${WIN_BASEDN}' ]:
t.cmd_contains("bin/samba-tool drs replicate ${HOSTNAME}.${WIN_REALM} ${WIN_HOSTNAME}.${WIN_REALM} %s -k yes" % nc, ["was successful"])
t.cmd_contains("bin/samba-tool drs replicate ${WIN_HOSTNAME}.${WIN_REALM} ${HOSTNAME}.${WIN_REALM} %s -k yes" % nc, ["was successful"])
t.run_cmd('bin/samba-tool drs kcc ${WIN_HOSTNAME}.${WIN_REALM} -Uadministrator@${WIN_REALM}%${WIN_PASS}')
t.kinit("administrator@${WIN_REALM}", "${WIN_PASS}")
- for nc in [ '${WIN_BASEDN}', 'CN=Configuration,${WIN_BASEDN}', 'CN=Schema,CN=Configuration,${WIN_BASEDN}' ]:
+ for nc in ['${WIN_BASEDN}', 'CN=Configuration,${WIN_BASEDN}', 'CN=Schema,CN=Configuration,${WIN_BASEDN}' ]:
t.cmd_contains("bin/samba-tool drs replicate ${HOSTNAME}.${WIN_REALM} ${WIN_HOSTNAME}.${WIN_REALM} %s -k yes" % nc, ["was successful"])
retries = 10
child.expect("DSA invocationID")
t.cmd_contains("bin/samba-tool drs showrepl ${WIN_HOSTNAME}.${WIN_REALM} -k yes",
- [ "INBOUND NEIGHBORS",
+ ["INBOUND NEIGHBORS",
"OUTBOUND NEIGHBORS",
"${WIN_BASEDN}",
"Last attempt .* was successful",
git.samba.org / samba.git / commitdiff