#
# Based on the original in EJS:
# Copyright (C) Andrew Tridgell <tridge@samba.org> 2005
-#
+#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 3 of the License, or
# (at your option) any later version.
-#
+#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
-#
+#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
import base64
from samba import dsdb
from samba.ndr import ndr_unpack, ndr_pack
-from samba.dcerpc import drsblobs, misc, security
+from samba.dcerpc import drsblobs, misc
__docformat__ = "restructuredText"
+
class SamDB(samba.Ldb):
"""The SAM database."""
url = lp.get("sam database")
super(SamDB, self).__init__(url=url, lp=lp, modules_dir=modules_dir,
- session_info=session_info, credentials=credentials, flags=flags,
- options=options)
+ session_info=session_info, credentials=credentials, flags=flags,
+ options=options)
if global_schema:
dsdb._dsdb_set_global_schema(self)
def enable_account(self, search_filter):
"""Enables an account
-
- :param search_filter: LDAP filter to find the user (eg samccountname=name)
+
+ :param search_filter: LDAP filter to find the user (eg
+ samccountname=name)
"""
res = self.search(base=self.domain_dn(), scope=ldb.SCOPE_SUBTREE,
expression=search_filter, attrs=["userAccountControl"])
userAccountControl: %u
""" % (user_dn, userAccountControl)
self.modify_ldif(mod)
-
+
def force_password_change_at_next_login(self, search_filter):
"""Forces a password change at next login
-
- :param search_filter: LDAP filter to find the user (eg samccountname=name)
+
+ :param search_filter: LDAP filter to find the user (eg
+ samccountname=name)
"""
res = self.search(base=self.domain_dn(), scope=ldb.SCOPE_SUBTREE,
expression=search_filter, attrs=[])
self.transaction_commit()
def newuser(self, username, password,
- force_password_change_at_next_login_req=False,
- useusernameascn=False, userou=None, surname=None, givenname=None, initials=None,
- profilepath=None, scriptpath=None, homedrive=None, homedirectory=None,
- jobtitle=None, department=None, company=None, description=None,
- mailaddress=None, internetaddress=None, telephonenumber=None,
- physicaldeliveryoffice=None, sd=None, setpassword=True):
+ force_password_change_at_next_login_req=False,
+ useusernameascn=False, userou=None, surname=None, givenname=None,
+ initials=None, profilepath=None, scriptpath=None, homedrive=None,
+ homedirectory=None, jobtitle=None, department=None, company=None,
+ description=None, mailaddress=None, internetaddress=None,
+ telephonenumber=None, physicaldeliveryoffice=None, sd=None,
+ setpassword=True):
"""Adds a new user with additional parameters
:param username: Name of the new user
:param password: Password for the new user
:param force_password_change_at_next_login_req: Force password change
- :param useusernameascn: Use username as cn rather that firstname + initials + lastname
+ :param useusernameascn: Use username as cn rather that firstname +
+ initials + lastname
:param userou: Object container (without domainDN postfix) for new user
:param surname: Surname of the new user
:param givenname: First name of the new user
self.transaction_commit()
def setpassword(self, search_filter, password,
- force_change_at_next_login=False,
- username=None):
+ force_change_at_next_login=False, username=None):
"""Sets the password for a user
-
- :param search_filter: LDAP filter to find the user (eg samccountname=name)
+
+ :param search_filter: LDAP filter to find the user (eg
+ samccountname=name)
:param password: Password for the user
:param force_change_at_next_login: Force password change
"""
def setexpiry(self, search_filter, expiry_seconds, no_expiry_req=False):
"""Sets the account expiry for a user
-
- :param search_filter: LDAP filter to find the user (eg samccountname=name)
+
+ :param search_filter: LDAP filter to find the user (eg
+ samaccountname=name)
:param expiry_seconds: expiry time from now in seconds
:param no_expiry_req: if set, then don't expire password
"""
dsdb._samdb_set_domain_sid(self, sid)
def get_domain_sid(self):
- """Read the domain SID used by this LDB.
-
- """
+ """Read the domain SID used by this LDB. """
return dsdb._samdb_get_domain_sid(self)
def set_invocation_id(self, invocation_id):
def get_oid_from_attid(self, attid):
return dsdb._dsdb_get_oid_from_attid(self, attid)
- def get_attid_from_lDAPDisplayName(self, ldap_display_name, is_schema_nc=False):
- return dsdb._dsdb_get_attid_from_lDAPDisplayName(self, ldap_display_name, is_schema_nc)
+ def get_attid_from_lDAPDisplayName(self, ldap_display_name,
+ is_schema_nc=False):
+ return dsdb._dsdb_get_attid_from_lDAPDisplayName(self,
+ ldap_display_name, is_schema_nc)
def get_invocation_id(self):
- "Get the invocation_id id"
+ """Get the invocation_id id"""
return dsdb._samdb_ntds_invocation_id(self)
def set_ntds_settings_dn(self, ntds_settings_dn):
- """Set the NTDS Settings DN, as would be returned on the dsServiceName rootDSE attribute
+ """Set the NTDS Settings DN, as would be returned on the dsServiceName
+ rootDSE attribute.
This allows the DN to be set before the database fully exists
domain_sid = property(get_domain_sid, set_domain_sid)
def get_ntds_GUID(self):
- "Get the NTDS objectGUID"
+ """Get the NTDS objectGUID"""
return dsdb._samdb_ntds_objectGUID(self)
def server_site_name(self):
- "Get the server site name"
+ """Get the server site name"""
return dsdb._samdb_server_site_name(self)
def load_partition_usn(self, base_dn):
def get_attribute_from_attid(self, attid):
""" Get from an attid the associated attribute
- :param attid: The attribute id for searched attribute
- :return: The name of the attribute associated with this id
+ :param attid: The attribute id for searched attribute
+ :return: The name of the attribute associated with this id
"""
if len(self.hash_oid_name.keys()) == 0:
self._populate_oid_attid()
else:
return None
-
def _populate_oid_attid(self):
- """Populate the hash hash_oid_name
+ """Populate the hash hash_oid_name.
- This hash contains the oid of the attribute as a key and
- its display name as a value
+ This hash contains the oid of the attribute as a key and
+ its display name as a value
"""
self.hash_oid_name = {}
res = self.search(expression="objectClass=attributeSchema",
strDisplay = str(e.get("lDAPDisplayName"))
self.hash_oid_name[str(e.get("attributeID"))] = strDisplay
-
def get_attribute_replmetadata_version(self, dn, att):
- """ Get the version field trom the replPropertyMetaData for
- the given field
+ """Get the version field trom the replPropertyMetaData for
+ the given field
- :param dn: The on which we want to get the version
- :param att: The name of the attribute
- :return: The value of the version field in the replPropertyMetaData
- for the given attribute. None if the attribute is not replicated
+ :param dn: The on which we want to get the version
+ :param att: The name of the attribute
+ :return: The value of the version field in the replPropertyMetaData
+ for the given attribute. None if the attribute is not replicated
"""
res = self.search(expression="dn=%s" % dn,
return o.version
return None
-
- def set_attribute_replmetadata_version(self, dn, att, value, addifnotexist=False):
+ def set_attribute_replmetadata_version(self, dn, att, value,
+ addifnotexist=False):
res = self.search(expression="dn=%s" % dn,
scope=ldb.SCOPE_SUBTREE,
controls=["search_options:1:2"],
"replPropertyMetaData")
self.modify(msg, ["local_oid:1.3.6.1.4.1.7165.4.3.14:0"])
-
def write_prefixes_from_schema(self):
dsdb._dsdb_write_prefixes_from_schema_to_ldb(self)
m.dn = ldb.Dn(self, "CN=Directory Service,CN=Windows NT,CN=Services,%s"
% self.get_config_basedn().get_linearized())
if dsheuristics is not None:
- m["dSHeuristics"] = ldb.MessageElement(dsheuristics, ldb.FLAG_MOD_REPLACE,
- "dSHeuristics")
+ m["dSHeuristics"] = ldb.MessageElement(dsheuristics,
+ ldb.FLAG_MOD_REPLACE, "dSHeuristics")
else:
- m["dSHeuristics"] = ldb.MessageElement([], ldb.FLAG_MOD_DELETE, "dSHeuristics")
+ m["dSHeuristics"] = ldb.MessageElement([], ldb.FLAG_MOD_DELETE,
+ "dSHeuristics")
self.modify(m)
def get_dsheuristics(self):
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 3 of the License, or
# (at your option) any later version.
-#
+#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
-#
+#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
sec = security.descriptor.from_sddl(sddl, domain_sid)
return ndr_pack(sec)
-
+
class Schema(object):
def __init__(self, setup_path, domain_sid, invocationid=None, schemadn=None,
files=None, override_prefixmap=None, additional_prefixmap=None):
- """Load schema for the SamDB from the AD schema files and samba4_schema.ldif
-
+ """Load schema for the SamDB from the AD schema files and
+ samba4_schema.ldif
+
:param samdb: Load a schema into a SamDB.
:param setup_path: Setup path function.
:param schemadn: DN of the schema
-
- Returns the schema data loaded, to avoid double-parsing when then needing to add it to the db
+
+ Returns the schema data loaded, to avoid double-parsing when then
+ needing to add it to the db
"""
self.schemadn = schemadn
else:
self.ldb.transaction_commit()
- # Return a hash with the forward attribute as a key and the back as the value
+ # Return a hash with the forward attribute as a key and the back as the
+ # value
def linked_attributes(self):
return get_linked_attributes(self.schemadn, self.ldb)
scope=SCOPE_SUBTREE)
if target is not None:
attributes[str(res[i]["lDAPDisplayName"])]=str(target)
-
+
return attributes
domainsid=None,
override_prefixmap=None):
"""Load schema for the SamDB from the AD schema files and samba4_schema.ldif
-
+
:param setup_dir: Setup path
:param schemadn: DN of the schema
:param serverdn: DN of the server
-
+
Returns the schema data loaded as an object, with .ldb being a
new ldb with the schema loaded. This allows certain tests to
operate without a remote or local schema.
"""
-
+
def setup_path(file):
return os.path.join(setup_dir, file)
domainsid = security.random_sid()
else:
domainsid = security.dom_sid(domainsid)
- return Schema(setup_path, domainsid, schemadn=schemadn, override_prefixmap=override_prefixmap)
+ return Schema(setup_path, domainsid, schemadn=schemadn,
+ override_prefixmap=override_prefixmap)
scope=SCOPE_ONELEVEL, attrs=["cn","displayName"])
names.policyid = str(res[0]["cn"]).replace("{","").replace("}","")
# dc policy guid
- res2 = samdb.search(expression="(displayName=Default Domain Controllers" \
+ res2 = samdb.search(expression="(displayName=Default Domain Controllers"
" Policy)",
base="CN=Policies,CN=System," + str(names.rootdn),
scope=SCOPE_ONELEVEL, attrs=["cn","displayName"])
names.realm = string.upper(names.realm)
# netbiosname
# Get the netbiosname first (could be obtained from smb.conf in theory)
- res = secretsdb.search(expression="(flatname=%s)" % \
+ res = secretsdb.search(expression="(flatname=%s)" %
names.domain,base="CN=Primary Domains",
scope=SCOPE_SUBTREE, attrs=["sAMAccountName"])
names.netbiosname = str(res[0]["sAMAccountName"]).replace("$","")
scope=SCOPE_ONELEVEL, attrs=["cn","displayName"])
names.policyid = str(res7[0]["cn"]).replace("{","").replace("}","")
# dc policy guid
- res8 = samdb.search(expression="(displayName=Default Domain Controllers" \
+ res8 = samdb.search(expression="(displayName=Default Domain Controllers"
" Policy)",
base="CN=Policies,CN=System," + basedn,
scope=SCOPE_ONELEVEL, attrs=["cn","displayName"])
names.policyid_dc = str(res8[0]["cn"]).replace("{","").replace("}","")
else:
names.policyid_dc = None
- res9 = idmapdb.search(expression="(cn=%s)" % \
+ res9 = idmapdb.search(expression="(cn=%s)" %
(security.SID_BUILTIN_ADMINISTRATORS),
attrs=["xidNumber"])
if len(res9) == 1:
since the latest upgrade in the current provision
:param names: List of provision parameters
- :param setup_dis: Directory where the setup files are stored
+ :param setup_dir: Directory where the setup files are stored
:param creds: Credentials for the authentification
:param session: Session object
:param smbconf: Path to the smb.conf file
:param provdir: Directory where the provision will be stored
- :param logger: A `Logger`
+ :param logger: A Logger
"""
if os.path.isdir(provdir):
shutil.rmtree(provdir)
return hash
+
def get_diff_sddls(refsddl, cursddl):
"""Get the difference between 2 sddl
- This function split the textual representation of ACL into smaller
- chunck in order to not to report a simple permutation as a difference
- :param refsddl: First sddl to compare
- :param cursddl: Second sddl to compare
- :return: A string that explain difference between sddls"""
+ This function split the textual representation of ACL into smaller
+ chunck in order to not to report a simple permutation as a difference
+
+ :param refsddl: First sddl to compare
+ :param cursddl: Second sddl to compare
+ :return: A string that explain difference between sddls
+ """
txt = ""
hash_new = chunck_sddl(cursddl)
"""Update secrets.ldb
:param newsecrets_ldb: An LDB object that is connected to the secrets.ldb
- of the reference provision
+ of the reference provision
:param secrets_ldb: An LDB object that is connected to the secrets.ldb
- of the updated provision
+ of the updated provision
"""
messagefunc(SIMPLE, "update secrets.ldb")
delta = secrets_ldb.msg_diff(empty, reference[0])
for att in hashAttrNotCopied.keys():
delta.remove(att)
- messagefunc(CHANGE, "Entry %s is missing from secrets.ldb" % \
+ messagefunc(CHANGE, "Entry %s is missing from secrets.ldb" %
reference[0].dn)
for att in delta:
messagefunc(CHANGE, " Adding attribute %s" % att)
delta.remove(att)
for att in delta:
if att == "name":
- messagefunc(CHANGE, "Found attribute name on %s," \
+ messagefunc(CHANGE, "Found attribute name on %s,"
" must rename the DN" % (current[0].dn))
identic_rename(secrets_ldb, reference[0].dn)
else:
delta.remove(att)
if att != "dn":
messagefunc(CHANGE,
- "Adding/Changing attribute %s to %s" % \
+ "Adding/Changing attribute %s to %s" %
(att, current[0].dn))
delta.dn = current[0].dn
messagefunc(SIMPLE, "Remove old dns account")
secrets_ldb.delete(res2[0]["dn"])
+
def getOEMInfo(samdb, rootdn):
- """Return OEM Information on the top level
- Samba4 use to store version info in this field
+ """Return OEM Information on the top level Samba4 use to store version
+ info in this field
:param samdb: An LDB object connect to sam.ldb
:param rootdn: Root DN of the domain
- :return: The content of the field oEMInformation (if any)"""
+ :return: The content of the field oEMInformation (if any)
+ """
res = samdb.search(expression="(objectClass=*)", base=str(rootdn),
scope=SCOPE_BASE, attrs=["dn", "oEMInformation"])
if len(res) > 0:
else:
return ""
+
def updateOEMInfo(samdb, rootdn):
"""Update the OEMinfo field to add information about upgrade
- :param samdb: an LDB object connected to the sam DB
- :param rootdn: The string representation of the root DN of
- the provision (ie. DC=...,DC=...)
+
+ :param samdb: an LDB object connected to the sam DB
+ :param rootdn: The string representation of the root DN of
+ the provision (ie. DC=...,DC=...)
"""
res = samdb.search(expression="(objectClass=*)", base=rootdn,
scope=SCOPE_BASE, attrs=["dn", "oEMInformation"])
return deltaattr
+
def construct_existor_expr(attrs):
"""Construct a exists or LDAP search expression.
- ie (|(foo=*)(bar=*)
:param attrs: List of attribute on which we want to create the search
- expression.
+ expression.
:return: A string representing the expression, if attrs is empty an
- empty string is returned"""
+ empty string is returned
+ """
expr = ""
if len(attrs) > 0:
expr = "(|"
git.samba.org / vlendec / samba-autobuild / .git / commitdiff