# Add four groups
self.ldb.add({
"dn": "cn=g1,%s" % self.ou_groups,
- "objectclass": "group" })
+ "objectclass": "group"})
self.ldb.add({
"dn": "cn=g2,%s" % self.ou_groups,
- "objectclass": "group" })
+ "objectclass": "group"})
self.ldb.add({
"dn": "cn=g4,%s" % self.ou_groups,
- "objectclass": "group" })
+ "objectclass": "group"})
self.ldb.add({
"dn": "cn=g3,%s" % self.ou_groups,
- "objectclass": "group" })
+ "objectclass": "group"})
# Add four users
self.ldb.add({
self.ldb.add({
"dn": "cn=g1,%s" % self.ou_groups,
- "objectclass": "group" })
+ "objectclass": "group"})
self.ldb.add({
"dn": "cn=g2,%s" % self.ou_groups,
- "objectclass": "group" })
+ "objectclass": "group"})
self.ldb.add({
"dn": "cn=g3,%s" % self.ou_groups,
- "objectclass": "group" })
+ "objectclass": "group"})
self.ldb.add({
"dn": "cn=g4,%s" % self.ou_groups,
- "objectclass": "group" })
+ "objectclass": "group"})
self.ldb.add({
"dn": "cn=u1,%s" % self.ou_users,
set_attrs_seen.add(str(attrname).lower())
if syntax_oid in [dsdb.DSDB_SYNTAX_BINARY_DN, dsdb.DSDB_SYNTAX_OR_NAME,
- dsdb.DSDB_SYNTAX_STRING_DN, ldb.SYNTAX_DN ]:
+ 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)
else:
ctx.SPNs = ["HOST/%s" % ctx.myname,
"HOST/%s" % ctx.dnshostname,
- "GC/%s/%s" % (ctx.dnshostname, ctx.dnsforest) ]
+ "GC/%s/%s" % (ctx.dnshostname, ctx.dnsforest)]
res_rid_manager = ctx.samdb.search(scope=ldb.SCOPE_BASE,
attrs=["rIDManagerReference"],
"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_PARTIAL_SECRETS_ACCOUNT)
ctx.SPNs.extend(["RestrictedKrbHost/%s" % ctx.myname,
- "RestrictedKrbHost/%s" % ctx.dnshostname ])
+ "RestrictedKrbHost/%s" % ctx.dnshostname])
ctx.connection_dn = "CN=RODC Connection (FRS),%s" % ctx.ntds_dn
ctx.secure_channel_type = misc.SEC_CHAN_RODC
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',
- 'DNS_BOOT_METHOD_REGISTRY', 'DNS_BOOT_METHOD_DIRECTORY' ]
+ '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',
- 'DNS_ALLOW_MULTIBYTE_NAMES', 'DNS_ALLOW_ALL_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',
'DNS_ZONE_TYPE_SECONDARY', 'DNS_ZONE_TYPE_STUB',
- 'DNS_ZONE_TYPE_FORWARDER', 'DNS_ZONE_TYPE_SECONDARY_CACHE' ]
+ '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',
- 'DNS_ZONE_UPDATE_SECURE' ]
+ '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',
- 'DNS_ZONE_SECSECURE_LIST_ONLY', 'DNS_ZONE_SECSECURE_NO_XFER' ]
+ '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',
- 'DNS_ZONE_NOTIFY_LIST_ONLY' ]
+ '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',
- 'DNS_DP_FOREST_DEFAULT', 'DNS_DP_ENLISTED', 'DNS_DP_DELETED' ]
+ 'DNS_DP_FOREST_DEFAULT', 'DNS_DP_ENLISTED', 'DNS_DP_DELETED']
return bitmap_string(dnsserver, bitmap_defs, dp_flags)
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,
takes_options = [
Option("--dsa-option", help="DSA option to enable/disable", type="str",
- metavar="{+|-}IS_GC | {+|-}DISABLE_INBOUND_REPL | {+|-}DISABLE_OUTBOUND_REPL | {+|-}DISABLE_NTDSCONN_XLATE" ),
+ metavar="{+|-}IS_GC | {+|-}DISABLE_INBOUND_REPL | {+|-}DISABLE_OUTBOUND_REPL | {+|-}DISABLE_NTDSCONN_XLATE"),
]
option_map = {"IS_GC": 0x00000001,
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()
raise CommandError("Failed to find objectClass for user %s" % username)
session_info_flags = (AUTH_SESSION_INFO_DEFAULT_GROUPS |
- AUTH_SESSION_INFO_AUTHENTICATED )
+ AUTH_SESSION_INFO_AUTHENTICATED)
# When connecting to a remote server, don't look up the local privilege DB
if self.url is not None and self.url.startswith('ldap'):
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)
# Get new security descriptor
ds_sd_flags = (security.SECINFO_OWNER |
security.SECINFO_GROUP |
- security.SECINFO_DACL )
+ security.SECINFO_DACL)
msg = get_gpo_info(self.samdb, gpo=gpo, sd_flags=ds_sd_flags)[0]
ds_sd_ndr = msg['nTSecurityDescriptor'][0]
ds_sd = ndr_unpack(security.descriptor, ds_sd_ndr).as_sddl()
sio = (security.SECINFO_OWNER |
security.SECINFO_GROUP |
security.SECINFO_DACL |
- security.SECINFO_PROTECTED_DACL )
+ security.SECINFO_PROTECTED_DACL)
conn.set_acl(sharepath, fs_sd, sio)
# Copy GPO files over SMB
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,
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)
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,
# domain guid/sid
res6 = samdb.search(expression="(objectClass=*)", base=basedn,
scope=ldb.SCOPE_BASE, attrs=["objectGUID",
- "objectSid","msDS-Behavior-Version" ])
+ "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])
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")
mmr_serverids_config += read_and_sub_file(
setup_path("mmr_serverids.conf"), {
"SERVERID": str(serverid),
- "LDAPSERVER": url })
+ "LDAPSERVER": url})
rid = serverid * 10
rid = rid + 1
mmr_syncrepl_schema_config += read_and_sub_file(
"RID" : str(rid),
"MMRDN": self.names.domaindn,
"LDAPSERVER" : url,
- "MMR_PASSWORD": mmr_pass })
+ "MMR_PASSWORD": mmr_pass})
# OpenLDAP cn=config initialisation
olc_syncrepl_config = ""
olc_mmr_config = ""
serverid = serverid + 1
olc_serverids_config += read_and_sub_file(
setup_path("olc_serverid.conf"), {
- "SERVERID" : str(serverid), "LDAPSERVER" : url })
+ "SERVERID" : str(serverid), "LDAPSERVER" : url})
rid = rid + 1
olc_syncrepl_config += read_and_sub_file(
setup_path("cn=samba.ldif"),
{"LDAPADMINPASS": self.ldapadminpass,
"MMR_PASSWORD": mmr_pass,
- "MMR": mmr })
+ "MMR": mmr})
mapping = "schema-map-openldap-2.3"
backend_schema = "backend-schema.schema"
refint_config += read_and_sub_file(
setup_path("fedorads-refint-add.ldif"),
{"ARG_NUMBER" : str(argnum),
- "LINK_ATTR" : attr })
+ "LINK_ATTR" : attr})
memberof_config += read_and_sub_file(
setup_path("fedorads-linked-attributes.ldif"),
{"MEMBER_ATTR" : attr,
- "MEMBEROF_ATTR" : lnkattr[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" % \
# Remove references to dnsHostName in A, AAAA, NS, CNAME and SRV
values = [ndr_unpack(dnsp.DnssrvRpcRecord, v)
- for v in orig_values if not to_remove(v) ]
+ for v in orig_values if not to_remove(v)]
if len(values) != len(orig_values):
logger.info("updating %s keeping %d values, removing %s values" \
#
up = ndr_unpack(drsblobs.package_PrimaryUserPasswordBlob,
binascii.a2b_hex(up_package.data))
- self.checkUserPassword(up, [("{CRYPT}", "6",10000 )])
+ self.checkUserPassword(up, [("{CRYPT}", "6",10000)])
self.checkNtHash(USER_PASS, up.current_nt_hash.hash)
#
up = ndr_unpack(drsblobs.package_PrimaryUserPasswordBlob,
binascii.a2b_hex(up_package.data))
- self.checkUserPassword(up, [("{CRYPT}", "5",100 )])
+ self.checkUserPassword(up, [("{CRYPT}", "5",100)])
self.checkNtHash(USER_PASS, up.current_nt_hash.hash)
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",
- "../../../../testdata/samba3" ]
+ "../../../../testdata/samba3"]
for p in paths:
datadir = os.path.join(os.path.dirname(__file__), p)
if os.path.exists(datadir):
good_srv = ["SAMDOM.EXAMPLE.COM 65530 65530 65530"]
bad_srv = ["SAMDOM.EXAMPLE.COM 0 65536 0",
"SAMDOM.EXAMPLE.COM 0 0 65536",
- "SAMDOM.EXAMPLE.COM 65536 0 0" ]
+ "SAMDOM.EXAMPLE.COM 65536 0 0"]
for bad_dn in bad_dns:
bad_mx.append("%s 1" % bad_dn)
self.assertCmdSuccess(result, out, err, "Ensure syncpasswords --cache-ldb-initialize runs")
self.assertEqual(err,"","getpassword without url")
cache_attrs = {
- "objectClass": {"value": "userSyncPasswords" },
+ "objectClass": {"value": "userSyncPasswords"},
"samdbUrl": {},
"dirsyncFilter": {},
"dirsyncAttribute": {},
delta = ldb.Message()
delta.dn = ldb.Dn(samdb, str(res[0]["dn"]))
delta["oEMInformation"] = ldb.MessageElement(info, ldb.FLAG_MOD_REPLACE,
- "oEMInformation" )
+ "oEMInformation")
samdb.modify(delta)
def update_gpo(paths, samdb, names, lp, message):
"talloc",
"replace",
"tevent",
- "pidl" ]
+ "pidl"]
if os.environ.get("AUTOBUILD_SKIP_SAMBA_O3", "0") == "1":
defaulttasks.remove("samba-o3")
("install", "make install", "text/plain"),
("test", "make autotest", "text/plain"),
("check-clean-tree", "../script/clean-source-tree.sh", "text/plain"),
- ("clean", "make clean", "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"),
"text/plain"),
("install", "make install", "text/plain"),
("check-clean-tree", "script/clean-source-tree.sh", "text/plain"),
- ("clean", "make clean", "text/plain") ],
+ ("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"),
("test", "make test FAIL_IMMEDIATELY=1 TESTS='--include-env=nt4_dc --include-env=nt4_member'", "text/plain"),
("install", "make install", "text/plain"),
("check-clean-tree", "script/clean-source-tree.sh", "text/plain"),
- ("clean", "make clean", "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-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", '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"),
("test", "make quicktest FAIL_IMMEDIATELY=1 TESTS='--include-env=ad_dc'", "text/plain"),
("install", "make install", "text/plain"),
("check-clean-tree", "script/clean-source-tree.sh", "text/plain"),
- ("clean", "make clean", "text/plain") ],
+ ("clean", "make clean", "text/plain")],
"samba-ctdb" : [("random-sleep", "script/random-sleep.sh 60 600", "text/plain"),
# clean up:
("check-clean-tree", "script/clean-source-tree.sh", "text/plain"),
("clean", "make clean", "text/plain"),
- ("ctdb-clean", "cd ./ctdb && make clean", "text/plain") ],
+ ("ctdb-clean", "cd ./ctdb && make clean", "text/plain")],
"samba-libs" : [
("random-sleep", "script/random-sleep.sh 60 600", "text/plain"),
("install-no-lmdb", "make install", "text/plain"),
("check-clean-tree", "../../script/clean-source-tree.sh", "text/plain"),
("distcheck", "make distcheck", "text/plain"),
- ("clean", "make clean", "text/plain") ],
+ ("clean", "make clean", "text/plain")],
"tdb" : [
("random-sleep", "../../script/random-sleep.sh 60 600", "text/plain"),
("test", "make test", "text/plain"),
("check-clean-tree", "../../script/clean-source-tree.sh", "text/plain"),
("distcheck", "make distcheck", "text/plain"),
- ("clean", "make clean", "text/plain") ],
+ ("clean", "make clean", "text/plain")],
"talloc" : [
("random-sleep", "../../script/random-sleep.sh 60 600", "text/plain"),
("test", "make test", "text/plain"),
("check-clean-tree", "../../script/clean-source-tree.sh", "text/plain"),
("distcheck", "make distcheck", "text/plain"),
- ("clean", "make clean", "text/plain") ],
+ ("clean", "make clean", "text/plain")],
"replace" : [
("random-sleep", "../../script/random-sleep.sh 60 600", "text/plain"),
("test", "make test", "text/plain"),
("check-clean-tree", "../../script/clean-source-tree.sh", "text/plain"),
("distcheck", "make distcheck", "text/plain"),
- ("clean", "make clean", "text/plain") ],
+ ("clean", "make clean", "text/plain")],
"tevent" : [
("random-sleep", "../../script/random-sleep.sh 60 600", "text/plain"),
("test", "make test", "text/plain"),
("check-clean-tree", "../../script/clean-source-tree.sh", "text/plain"),
("distcheck", "make distcheck", "text/plain"),
- ("clean", "make clean", "text/plain") ],
+ ("clean", "make clean", "text/plain")],
"pidl" : [
- ("random-sleep", "../script/random-sleep.sh 60 600", "text/plain"),
- ("configure", "perl Makefile.PL PREFIX=${PREFIX_DIR}", "text/plain"),
- ("touch", "touch *.yp", "text/plain"),
- ("make", "make", "text/plain"),
- ("test", "make test", "text/plain"),
- ("install", "make install", "text/plain"),
- ("checkout-yapp-generated", "git checkout lib/Parse/Pidl/IDL.pm lib/Parse/Pidl/Expr.pm", "text/plain"),
- ("check-clean-tree", "../script/clean-source-tree.sh", "text/plain"),
- ("clean", "make clean", "text/plain") ],
+ ("random-sleep", "../script/random-sleep.sh 60 600", "text/plain"),
+ ("configure", "perl Makefile.PL PREFIX=${PREFIX_DIR}", "text/plain"),
+ ("touch", "touch *.yp", "text/plain"),
+ ("make", "make", "text/plain"),
+ ("test", "make test", "text/plain"),
+ ("install", "make install", "text/plain"),
+ ("checkout-yapp-generated", "git checkout lib/Parse/Pidl/IDL.pm lib/Parse/Pidl/Expr.pm", "text/plain"),
+ ("check-clean-tree", "../script/clean-source-tree.sh", "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):
rebase_remote,
rebase_remote, rebase_branch
),
- "test/plain" ) ]
+ "test/plain")]
self.retry = builder('retry', retry_task, cp=False)
self.need_retry = False
'enumlist' : enumlist,
'handler' : handler,
'deprecated' : deprecated,
- 'synonyms' : synonyms }
+ 'synonyms' : synonyms}
# map doc attributes to a section of the generated function
context_dict = {"G": "_GLOBAL", "S": "_LOCAL"}
#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)
plansmbtorture4testsuite(t, "simpleserver", "//%s/%s %s" % ('$SERVER_IP', s, ' -U$USERNAME%$PASSWORD'), description=s)
posix_tests = ["POSIX", "POSIX-APPEND", "POSIX-SYMLINK-ACL", "POSIX-SYMLINK-EA", "POSIX-OFD-LOCK",
- "POSIX-STREAM-DELETE", "WINDOWS-BAD-SYMLINK" ]
+ "POSIX-STREAM-DELETE", "WINDOWS-BAD-SYMLINK"]
for t in posix_tests:
plantestsuite("samba3.smbtorture_s3.plain(nt4_dc).%s" % t, "nt4_dc", [os.path.join(samba3srcdir, "script/tests/test_smbtorture_s3.sh"), t, '//$SERVER_IP/posix_share', '$USERNAME', '$PASSWORD', smbtorture3, "", "-l $LOCAL_PATH"])
plantestsuite("samba3.wbinfo_name_lookup", env,
[os.path.join(srcdir(),
"nsswitch/tests/test_wbinfo_name_lookup.sh"),
- '$DOMAIN', '$REALM', '$DC_USERNAME' ])
+ '$DOMAIN', '$REALM', '$DC_USERNAME'])
env = "ad_member:local"
plantestsuite("samba3.wbinfo_user_info", env,
[os.path.join(srcdir(),
"nsswitch/tests/test_wbinfo_user_info.sh"),
- '$DOMAIN', '$REALM', '$DOMAIN', 'alice', 'alice', 'jane', 'jane.doe' ])
+ '$DOMAIN', '$REALM', '$DOMAIN', 'alice', 'alice', 'jane', 'jane.doe'])
env = "fl2008r2dc:local"
plantestsuite("samba3.wbinfo_user_info", env,
[os.path.join(srcdir(),
"nsswitch/tests/test_wbinfo_user_info.sh"),
- '$TRUST_DOMAIN', '$TRUST_REALM', '$DOMAIN', 'alice', 'alice', 'jane', 'jane.doe' ])
+ '$TRUST_DOMAIN', '$TRUST_REALM', '$DOMAIN', 'alice', 'alice', 'jane', 'jane.doe'])
env = "ad_member"
t = "WBCLIENT-MULTI-PING"
unix = ["unix.info2", "unix.whoami"]
-nbt = ["nbt.dgram" ]
+nbt = ["nbt.dgram"]
libsmbclient = ["libsmbclient"]
plantestsuite("samba3.blackbox.net_tdb", "simpleserver:local",
[os.path.join(samba3srcdir, "script/tests/test_net_tdb.sh"),
smbclient3, '$SERVER', 'tmp', '$USERNAME', '$PASSWORD',
- configuration, '$LOCAL_PATH', '$LOCK_DIR' ])
+ 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"),
classinfo[name]["objectClassCategory"] = 0
for a in ["possSuperiors", "systemPossSuperiors",
"auxiliaryClass", "systemAuxiliaryClass",
- "subClassOf" ]:
+ "subClassOf"]:
classinfo[name][a] = []
if r.get(a):
for i in r[a]:
res = self.ldb_admin.search(self.base_dn,
expression="(distinguishedName=%s,%s)" %
("CN=test_add_group1,OU=test_add_ou2,OU=test_add_ou1",
- self.base_dn) )
+ self.base_dn))
self.assertEqual(len(res), 0)
def test_add_u4(self):
# Modify on attribute you have rights for
self.ldb_user.modify_ldif(ldif)
res = self.ldb_admin.search(self.base_dn, expression="(distinguishedName=%s)" \
- % self.get_user_dn(self.user_with_wp), attrs=["adminDescription"] )
+ % self.get_user_dn(self.user_with_wp), attrs=["adminDescription"])
self.assertEqual(res[0]["adminDescription"][0], "blah blah blah")
def test_modify_u5(self):
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),
"<SID=%s>" % security.SID_BUILTIN_ADMINISTRATORS,
"<SID=%s>" % security.SID_BUILTIN_SERVER_OPERATORS,
"<SID=%s>" % security.SID_BUILTIN_BACKUP_OPERATORS,
- "<SID=%s>" % security.SID_BUILTIN_ACCOUNT_OPERATORS ]
+ "<SID=%s>" % security.SID_BUILTIN_ACCOUNT_OPERATORS]
ctx.reveal_sid = "<SID=%s-%s>" % (ctx.domsid, security.DOMAIN_RID_RODC_ALLOW)
mysid = ctx.get_mysid()
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 ],
- "isDeleted": "FALSE" })
+ "member": [self.usr1, self.usr2],
+ "isDeleted": "FALSE"})
self.ldb.add({
"dn": self.sit1,
- "objectclass": "site" })
+ "objectclass": "site"})
self.ldb.add({
"dn": self.ss1,
- "objectclass": ["applicationSiteSettings", "nTDSSiteSettings"] })
+ "objectclass": ["applicationSiteSettings", "nTDSSiteSettings"]})
self.ldb.add({
"dn": self.srv1,
- "objectclass": "serversContainer" })
+ "objectclass": "serversContainer"})
self.ldb.add({
"dn": self.srv2,
- "objectClass": "server" })
+ "objectClass": "server"})
self.objLive1 = self.search_dn(self.usr1)
self.guid1=self.objLive1["objectGUID"][0]
delta.dn = Dn(self.ldb_admin, str(ouname))
delta["cn"] = MessageElement("test ou",
FLAG_MOD_ADD,
- "cn" )
+ "cn")
self.ldb_admin.modify(delta)
res = self.ldb_admin.search(self.base_dn,
expression="name=testou",
delta["cn"] = MessageElement("test ou",
FLAG_MOD_ADD,
- "cn" )
+ "cn")
self.ldb_admin.modify(delta)
res = self.ldb_admin.search(self.base_dn,
expression="(&(objectClass=organizationalUnit)(!(isDeleted=*)))",
delta.dn = Dn(self.ldb_admin, str(ouname))
delta["cn"] = MessageElement([],
FLAG_MOD_DELETE,
- "cn" )
+ "cn")
self.ldb_admin.modify(delta)
res = self.ldb_admin.search(self.base_dn,
expression="(&(objectClass=organizationalUnit)(!(isDeleted=*)))",
expression="(name=Administrators)",
controls=[control1])
- self.assertEqual(len(res[0].get("member")), size )
+ self.assertEqual(len(res[0].get("member")), size)
self.ldb_admin.newgroup("testgroup")
self.ldb_admin.add_remove_group_members("testgroup", [self.simple_user],
controls=["dirsync:1:0:1"])
self.assertEqual(len(res[0].get("member")), 1)
- self.assertTrue(res[0].get("member") != "" )
+ self.assertTrue(res[0].get("member") != "")
ctl = str(res.controls[0]).split(":")
ctl[1] = "1"
expression="(name=Administrators)",
controls=["dirsync:1:%d:1" % flag_incr_linked])
- self.assertTrue(res[0].get("member;range=1-1") != None )
+ self.assertTrue(res[0].get("member;range=1-1") != None)
self.assertTrue(len(res[0].get("member;range=1-1")) > 0)
size = len(res[0].get("member;range=1-1"))
expression="(name=Administrators)",
controls=[control1])
- self.assertEqual(res[0].get("member;range=1-1"), None )
+ self.assertEqual(res[0].get("member;range=1-1"), None)
self.assertEqual(len(res[0].get("member;range=0-0")), 1)
ctl = str(res.controls[0]).split(":")
expression="(name=Administrators)",
controls=[control2])
- self.assertEqual(res[0].get("member;range=1-1"), None )
+ self.assertEqual(res[0].get("member;range=1-1"), None)
self.assertEqual(len(res[0].get("member;range=0-0")), 1)
res = self.ldb_admin.search(self.base_dn,
expression="(name=Administrators)",
controls=[control1])
- self.assertEqual(res[0].get("member;range=1-1"), None )
+ self.assertEqual(res[0].get("member;range=1-1"), None)
self.assertEqual(len(res[0].get("member;range=0-0")), 2)
def test_dirsync_deleted_items(self):
try:
self.ldb.add({
"dn": "cn=ldaptestuser,cn=users," + self.base_dn,
- "objectClass": [] })
+ "objectClass": []})
self.fail()
except LdbError as e1:
(num, _) = e1.args
try:
self.ldb.add({
"dn": "cn=ldaptestuser,cn=users," + self.base_dn,
- "objectClass": "X" })
+ "objectClass": "X"})
self.fail()
except LdbError as e2:
(num, _) = e2.args
self.ldb.add({
"dn": "cn=ldaptestuser,cn=users," + self.base_dn,
"objectClass": "person",
- "objectCategory": self.base_dn })
+ "objectCategory": self.base_dn})
self.fail()
except LdbError as e3:
(num, _) = e3.args
self.ldb.add({
"dn": "cn=ldaptestuser,cn=users," + self.base_dn,
"objectClass": "person",
- "systemFlags": ["0", str(SYSTEM_FLAG_DOMAIN_DISALLOW_MOVE)] })
+ "systemFlags": ["0", str(SYSTEM_FLAG_DOMAIN_DISALLOW_MOVE)]})
self.fail()
except LdbError as e4:
(num, _) = e4.args
try:
self.ldb.add({
"dn": "cn=ldaptestuser,cn=users," + self.base_dn,
- "objectClass": "connectionPoint" })
+ "objectClass": "connectionPoint"})
self.fail()
except LdbError as e5:
(num, _) = e5.args
try:
self.ldb.add({
"dn": "cn=ldaptestuser,cn=users," + self.base_dn,
- "objectClass": ["person", "leaf"] })
+ "objectClass": ["person", "leaf"]})
self.fail()
except LdbError as e6:
(num, _) = e6.args
# subclasses) are allowed
self.ldb.add({
"dn": "cn=ldaptestuser,cn=users," + self.base_dn,
- "objectClass": ["top", "leaf", "connectionPoint", "serviceConnectionPoint"] })
+ "objectClass": ["top", "leaf", "connectionPoint", "serviceConnectionPoint"]})
delete_force(self.ldb, "cn=ldaptestuser,cn=users," + self.base_dn)
try:
self.ldb.add({
"dn": "cn=ldaptestuser,cn=users," + self.base_dn,
- "objectClass": ["person", "container"] })
+ "objectClass": ["person", "container"]})
self.fail()
except LdbError as e7:
(num, _) = e7.args
self.ldb.add({
"dn": "cn=ldaptestuser,cn=users," + self.base_dn,
"objectClass": "person",
- "systemFlags": str(~(SYSTEM_FLAG_CONFIG_ALLOW_RENAME | SYSTEM_FLAG_CONFIG_ALLOW_MOVE | SYSTEM_FLAG_CONFIG_ALLOW_LIMITED_MOVE)) })
+ "systemFlags": str(~(SYSTEM_FLAG_CONFIG_ALLOW_RENAME | SYSTEM_FLAG_CONFIG_ALLOW_MOVE | SYSTEM_FLAG_CONFIG_ALLOW_LIMITED_MOVE))})
res = ldb.search("cn=ldaptestuser,cn=users," + self.base_dn,
scope=SCOPE_BASE, attrs=["systemFlags"])
self.ldb.add({
"dn": "cn=ldaptestuser,cn=users," + self.base_dn,
- "objectClass": "person" })
+ "objectClass": "person"})
# We can remove derivation classes of the structural objectclass
# but they're going to be readded afterwards
self.ldb.add({
"dn": "cn=ldaptestuser,cn=users," + self.base_dn,
- "objectClass": "user" })
+ "objectClass": "user"})
# Add a new top-most structural class "container". This does not work
# since it stands in no direct relation to the current one.
ldb.add({
"dn": "cn=ldaptestuser,cn=users," + self.base_dn,
"objectClass": "person",
- "sn": "" })
+ "sn": ""})
self.fail()
except LdbError as e39:
(num, _) = e39.args
ldb.add({
"dn": "cn=ldaptestuser,cn=users," + self.base_dn,
- "objectClass": "person" })
+ "objectClass": "person"})
# Too short (min. 1)
m = Message()
self.ldb.add({
"dn": "cn=ldaptestgroup,cn=users," + self.base_dn,
"objectclass": "group",
- "instanceType": "1" })
+ "instanceType": "1"})
self.fail()
except LdbError as e47:
(num, _) = e47.args
self.ldb.add({
"dn": "cn=ldaptestgroup,cn=users," + self.base_dn,
"objectclass": "group",
- "instanceType": "32" })
+ "instanceType": "32"})
self.fail()
except LdbError as e48:
(num, _) = e48.args
self.ldb.add({
"dn": "cn=ldaptestuser2,cn=users," + self.base_dn,
"objectclass": "user",
- "instanceType": "3" })
+ "instanceType": "3"})
self.fail()
except LdbError as e52:
(num, _) = e52.args
self.ldb.add({
"dn": "cn=ldaptestuser2,cn=users," + self.base_dn,
- "objectclass": "user" })
+ "objectclass": "user"})
ldb.rename("cn=ldaptestuser2,cn=users," + self.base_dn, "cn=ldaptestuser2,cn=users," + self.base_dn)
ldb.rename("cn=ldaptestuser2,cn=users," + self.base_dn, "cn=ldaptestuser3,cn=users," + self.base_dn)
"""Tests the rename operation twice - this corresponds to a past bug"""
self.ldb.add({
"dn": "cn=ldaptestuser5,cn=users," + self.base_dn,
- "objectclass": "user" })
+ "objectclass": "user"})
ldb.rename("cn=ldaptestuser5,cn=users," + self.base_dn, "cn=ldaptestUSER5,cn=users," + self.base_dn)
delete_force(self.ldb, "cn=ldaptestuser5,cn=users," + self.base_dn)
self.ldb.add({
"dn": "cn=ldaptestuser5,cn=users," + self.base_dn,
- "objectclass": "user" })
+ "objectclass": "user"})
ldb.rename("cn=ldaptestuser5,cn=Users," + self.base_dn, "cn=ldaptestUSER5,cn=users," + self.base_dn)
res = ldb.search(expression="cn=ldaptestuser5")
self.assertEquals(len(res), 1, "Wrong number of hits for cn=ldaptestuser5")
self.ldb.add({
"dn": "cn=ldaptestcontainer," + self.base_dn,
- "objectClass": "container" })
+ "objectClass": "container"})
# The objectGUID cannot directly be changed
try:
self.ldb.add({
"dn": "cn=ldaptestcontainer," + self.base_dn,
- "objectClass": "container" })
+ "objectClass": "container"})
res = ldb.search("cn=ldaptestcontainer," + self.base_dn,
scope=SCOPE_BASE,
"uSNCreated" : "1",
"uSNChanged" : "1",
"whenCreated": timestring(long(time.time())),
- "whenChanged": timestring(long(time.time())) })
+ "whenChanged": timestring(long(time.time()))})
res = ldb.search("cn=ldaptestcontainer," + self.base_dn,
scope=SCOPE_BASE,
self.ldb.add({"dn": user_dn,
"objectClass": "user",
"sAMAccountName": user_name,
- "nTSecurityDescriptor": [] })
+ "nTSecurityDescriptor": []})
self.fail()
except LdbError as e107:
(num, _) = e107.args
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
delete_force(self.ldb, user_dn)
self.ldb.add({"dn": user_dn,
"objectClass": "user",
- "sAMAccountName": user_name })
+ "sAMAccountName": user_name})
m = Message()
m.dn = Dn(ldb, user_dn)
delete_force(self.ldb, user_dn)
self.ldb.add({"dn": user_dn,
"objectClass": "user",
- "sAMAccountName": user_name })
+ "sAMAccountName": user_name})
#
# We check the following values:
# 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":
self.ldb.add({
"dn": "cn=testuser2,cn=users," + self.base_dn,
"objectclass": "user",
- "userPassword": ["thatsAcomplPASS1", "thatsAcomplPASS2"] })
+ "userPassword": ["thatsAcomplPASS1", "thatsAcomplPASS2"]})
# This surprisingly should work
delete_force(self.ldb, "cn=testuser2,cn=users," + self.base_dn)
self.ldb.add({
"dn": "cn=testuser2,cn=users," + self.base_dn,
"objectclass": "user",
- "userPassword": ["thatsAcomplPASS1", "thatsAcomplPASS1"] })
+ "userPassword": ["thatsAcomplPASS1", "thatsAcomplPASS1"]})
def test_empty_passwords(self):
print("Performs some empty passwords testing")
self.ldb.add({
"dn": "cn=testuser2,cn=users," + self.base_dn,
"objectclass": "user",
- "unicodePwd": [] })
+ "unicodePwd": []})
self.fail()
except LdbError as e32:
(num, _) = e32.args
self.ldb.add({
"dn": "cn=testuser2,cn=users," + self.base_dn,
"objectclass": "user",
- "dBCSPwd": [] })
+ "dBCSPwd": []})
self.fail()
except LdbError as e33:
(num, _) = e33.args
self.ldb.add({
"dn": "cn=testuser2,cn=users," + self.base_dn,
"objectclass": "user",
- "userPassword": [] })
+ "userPassword": []})
self.fail()
except LdbError as e34:
(num, _) = e34.args
self.ldb.add({
"dn": "cn=testuser2,cn=users," + self.base_dn,
"objectclass": "user",
- "clearTextPassword": [] })
+ "clearTextPassword": []})
self.fail()
except LdbError as e35:
(num, _) = e35.args
ldb.add({
"dn": "cn=ldaptestuser,cn=users," + self.base_dn,
"objectclass": "user",
- "userAccountControl": str(UF_NORMAL_ACCOUNT | UF_PASSWD_NOTREQD) })
+ "userAccountControl": str(UF_NORMAL_ACCOUNT | UF_PASSWD_NOTREQD)})
res1 = ldb.search("cn=ldaptestuser,cn=users," + self.base_dn,
scope=SCOPE_BASE, attrs=["primaryGroupID"])
ldb.add({
"dn": "cn=ldaptestuser,cn=users," + self.base_dn,
"objectclass": "computer",
- "userAccountControl": str(UF_WORKSTATION_TRUST_ACCOUNT | UF_PASSWD_NOTREQD) })
+ "userAccountControl": str(UF_WORKSTATION_TRUST_ACCOUNT | UF_PASSWD_NOTREQD)})
res1 = ldb.search("cn=ldaptestuser,cn=users," + self.base_dn,
scope=SCOPE_BASE, attrs=["primaryGroupID"])
ldb.add({
"dn": "cn=ldaptestuser,cn=users," + self.base_dn,
"objectclass": "computer",
- "userAccountControl": str(UF_SERVER_TRUST_ACCOUNT | UF_PASSWD_NOTREQD) })
+ "userAccountControl": str(UF_SERVER_TRUST_ACCOUNT | UF_PASSWD_NOTREQD)})
res1 = ldb.search("cn=ldaptestuser,cn=users," + self.base_dn,
scope=SCOPE_BASE, attrs=["primaryGroupID"])
ldb.add({
"dn": "cn=ldaptestuser,cn=users," + self.base_dn,
"objectclass": "computer",
- "userAccountControl": str(UF_PARTIAL_SECRETS_ACCOUNT | UF_WORKSTATION_TRUST_ACCOUNT | UF_PASSWD_NOTREQD) })
+ "userAccountControl": str(UF_PARTIAL_SECRETS_ACCOUNT | UF_WORKSTATION_TRUST_ACCOUNT | UF_PASSWD_NOTREQD)})
res1 = ldb.search("cn=ldaptestuser,cn=users," + self.base_dn,
scope=SCOPE_BASE, attrs=["primaryGroupID"])
"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()
self.ldb.add({
"dn": "cn=ldaptestgroup,cn=users," + self.base_dn,
- "objectclass": "group" })
+ "objectclass": "group"})
m = Message()
m.dn = Dn(ldb, "cn=ldaptestgroup,cn=users," + self.base_dn)
self.ldb.add({
"dn": "cn=ldaptestgroup,cn=users," + self.base_dn,
"objectclass": "group",
- "fSMORoleOwner": [] })
+ "fSMORoleOwner": []})
self.fail()
except LdbError as e84:
(num, _) = e84.args
self.ldb.add({
"dn": "cn=ldaptestgroup,cn=users," + self.base_dn,
"objectclass": "group",
- "fSMORoleOwner": ds_service_name })
+ "fSMORoleOwner": ds_service_name})
delete_force(self.ldb, "cn=ldaptestgroup,cn=users," + self.base_dn)
self.ldb.add({
"dn": "cn=ldaptestgroup,cn=users," + self.base_dn,
- "objectclass": "group" })
+ "objectclass": "group"})
m = Message()
m.dn = Dn(ldb, "cn=ldaptestgroup,cn=users," + self.base_dn)
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
if orig_val is None:
continue
if not isinstance(orig_val, MessageElement):
- orig_val = MessageElement(str(orig_val), 0, attr )
+ orig_val = MessageElement(str(orig_val), 0, attr)
m = Message()
m.add(orig_val)
orig_ldif = self.samdb.write_ldif(m, 0)
'uSNChanged': '**',
'instanceType': '4',
'adminCount': '0',
- 'cn': groupname }
+ 'cn': groupname}
def test_plain_group(self):
print("Test restored Group attributes")
'uSNCreated': '**',
'uSNChanged': '**',
'instanceType': '4',
- rdn.lower(): name }
+ rdn.lower(): name}
def _create_test_ou(self, rdn, name=None, description=None):
ou_dn = "OU=%s,%s" % (rdn, self.base_dn)
if others is not None:
msg_dict = dict(msg_dict.items() + others.items())
- msg = ldb.Message.from_dict(self.samdb, msg_dict )
+ msg = ldb.Message.from_dict(self.samdb, msg_dict)
msg["sAMAccountName"] = samaccountname
print("Adding computer account %s" % computername)
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):
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),
"<SID=%s>" % security.SID_BUILTIN_ADMINISTRATORS,
"<SID=%s>" % security.SID_BUILTIN_SERVER_OPERATORS,
"<SID=%s>" % security.SID_BUILTIN_BACKUP_OPERATORS,
- "<SID=%s>" % security.SID_BUILTIN_ACCOUNT_OPERATORS ]
+ "<SID=%s>" % security.SID_BUILTIN_ACCOUNT_OPERATORS]
ctx.reveal_sid = "<SID=%s-%s>" % (ctx.domsid, security.DOMAIN_RID_RODC_ALLOW)
mysid = ctx.get_mysid()
casefold=True)
t.cmd_contains("bin/wbinfo -u",
["${WIN_DOMAIN}/administrator",
- "${WIN_DOMAIN}/krbtgt" ],
+ "${WIN_DOMAIN}/krbtgt"],
casefold=True)
t.cmd_contains("bin/wbinfo -g",
["${WIN_DOMAIN}/domain users",
t.setwinvars(vm)
t.info("Joining ${WIN_VM} as a member using net ads join")
t.port_wait("${WIN_IP}", 389)
- t.retry_cmd("host -t SRV _ldap._tcp.${WIN_REALM} ${WIN_IP}", ['has SRV record'] )
+ t.retry_cmd("host -t SRV _ldap._tcp.${WIN_REALM} ${WIN_IP}", ['has SRV record'])
t.cmd_contains("bin/net ads join -Uadministrator%${WIN_PASS}", ["Joined"])
t.cmd_contains("bin/net ads testjoin", ["Join is OK"])
t.cmd_contains("bin/net ads dns register ${HOSTNAME}.${WIN_REALM} -P", ["Successfully registered hostname with DNS"])
# 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"])
"CN=Configuration,${BASEDN}",
"Last success",
"CN=Configuration,${BASEDN}",
- "Last success" ],
+ "Last success"],
ordered=True,
regex=True)
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}",
"CN=Configuration,${BASEDN}",
"Last attempt.*was successful",
"CN=Configuration,${BASEDN}",
- "Last attempt.*was successful" ],
+ "Last attempt.*was successful"],
ordered=True,
regex=True)
t.setwinvars(vm)
t.info("Joining ${WIN_VM} as a second DC using samba-tool domain join DC")
t.port_wait("${WIN_IP}", 389)
- t.retry_cmd("host -t SRV _ldap._tcp.${WIN_REALM} ${WIN_IP}", ['has SRV record'] )
+ t.retry_cmd("host -t SRV _ldap._tcp.${WIN_REALM} ${WIN_IP}", ['has SRV record'])
- t.retry_cmd("bin/samba-tool drs showrepl ${WIN_HOSTNAME}.${WIN_REALM} -Uadministrator%${WIN_PASS}", ['INBOUND NEIGHBORS'] )
+ t.retry_cmd("bin/samba-tool drs showrepl ${WIN_HOSTNAME}.${WIN_REALM} -Uadministrator%${WIN_PASS}", ['INBOUND NEIGHBORS'])
t.run_cmd('bin/samba-tool domain join ${WIN_REALM} DC -Uadministrator%${WIN_PASS} -d${DEBUGLEVEL} --option=interfaces=${INTERFACE}')
t.run_cmd('bin/samba-tool drs kcc ${WIN_HOSTNAME}.${WIN_REALM} -Uadministrator@${WIN_REALM}%${WIN_PASS}')
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.setwinvars(vm)
t.info("Joining ${WIN_VM} as a RODC using samba-tool domain join DC")
t.port_wait("${WIN_IP}", 389)
- t.retry_cmd("host -t SRV _ldap._tcp.${WIN_REALM} ${WIN_IP}", ['has SRV record'] )
- t.retry_cmd("bin/samba-tool drs showrepl ${WIN_HOSTNAME}.${WIN_REALM} -Uadministrator%${WIN_PASS}", ['INBOUND NEIGHBORS'] )
+ t.retry_cmd("host -t SRV _ldap._tcp.${WIN_REALM} ${WIN_IP}", ['has SRV record'])
+ t.retry_cmd("bin/samba-tool drs showrepl ${WIN_HOSTNAME}.${WIN_REALM} -Uadministrator%${WIN_PASS}", ['INBOUND NEIGHBORS'])
t.run_cmd('bin/samba-tool domain join ${WIN_REALM} RODC -Uadministrator%${WIN_PASS} -d${DEBUGLEVEL} --option=interfaces=${INTERFACE}')
t.run_cmd('bin/samba-tool drs kcc ${WIN_HOSTNAME}.${WIN_REALM} -Uadministrator@${WIN_REALM}%${WIN_PASS}')
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
"CN=Configuration,${WIN_BASEDN}",
"Last attempt .* was successful",
"CN=Configuration,${WIN_BASEDN}",
- "Last attempt .* was successful" ],
+ "Last attempt .* was successful"],
ordered=True,
regex=True)
# Give DNS registration a kick
child.sendline("ipconfig /registerdns")
- self.retry_cmd("host -t SRV _ldap._tcp.${WIN_REALM} ${WIN_IP}", ['has SRV record'], retries=60, delay=5 )
+ self.retry_cmd("host -t SRV _ldap._tcp.${WIN_REALM} ${WIN_IP}", ['has SRV record'], retries=60, delay=5)
def start_winvm(self, vm):
git.samba.org / samba.git / commitdiff