-#!/usr/bin/python
-
# Unix SMB/CIFS implementation.
# Copyright (C) Jelmer Vernooij <jelmer@samba.org> 2007-2010
# Copyright (C) Matthias Dieter Wallnoefer 2009
#
# Based on the original in EJS:
# Copyright (C) Andrew Tridgell <tridge@samba.org> 2005
-#
+# Copyright (C) Giampaolo Lauria <lauria2@yahoo.com> 2011
+#
# 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/>.
#
"""Convenience functions for using the SAM."""
-import dsdb
import samba
import ldb
import time
import base64
+import os
+from samba import dsdb
+from samba.ndr import ndr_unpack, ndr_pack
+from samba.dcerpc import drsblobs, misc
+from samba.common import normalise_int32
__docformat__ = "restructuredText"
+
class SamDB(samba.Ldb):
"""The SAM database."""
+ hash_oid_name = {}
+
def __init__(self, url=None, lp=None, modules_dir=None, session_info=None,
- credentials=None, flags=0, options=None, global_schema=True, auto_connect=True,
- am_rodc=False):
+ credentials=None, flags=0, options=None, global_schema=True,
+ auto_connect=True, am_rodc=None):
self.lp = lp
if not auto_connect:
url = None
elif url is None and lp is not None:
- url = lp.get("sam database")
+ url = lp.samdb_url()
+
+ self.url = url
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)
+ dsdb._dsdb_set_global_schema(self)
- dsdb.dsdb_set_am_rodc(self, am_rodc)
+ if am_rodc is not None:
+ dsdb._dsdb_set_am_rodc(self, am_rodc)
def connect(self, url=None, flags=0, options=None):
- if self.lp is not None:
+ '''connect to the database'''
+ if self.lp is not None and not os.path.exists(url):
url = self.lp.private_path(url)
+ self.url = url
super(SamDB, self).connect(url=url, flags=flags,
options=options)
+ def am_rodc(self):
+ '''return True if we are an RODC'''
+ return dsdb._am_rodc(self)
+
+ def am_pdc(self):
+ '''return True if we are an PDC emulator'''
+ return dsdb._am_pdc(self)
+
def domain_dn(self):
- # find the DNs for the domain
- res = self.search(base="",
- scope=ldb.SCOPE_BASE,
- expression="(defaultNamingContext=*)",
- attrs=["defaultNamingContext"])
- assert(len(res) == 1 and res[0]["defaultNamingContext"] is not None)
- return res[0]["defaultNamingContext"][0]
-
- def enable_account(self, filter):
+ '''return the domain DN'''
+ return str(self.get_default_basedn())
+
+ def disable_account(self, search_filter):
+ """Disables an account
+
+ :param search_filter: LDAP filter to find the user (eg
+ samccountname=name)
+ """
+
+ flags = samba.dsdb.UF_ACCOUNTDISABLE
+ self.toggle_userAccountFlags(search_filter, flags, on=True)
+
+ def enable_account(self, search_filter):
"""Enables an account
-
- :param filter: LDAP filter to find the user (eg samccountname=name)
+
+ :param search_filter: LDAP filter to find the user (eg
+ samccountname=name)
+ """
+
+ flags = samba.dsdb.UF_ACCOUNTDISABLE | samba.dsdb.UF_PASSWD_NOTREQD
+ self.toggle_userAccountFlags(search_filter, flags, on=False)
+
+ def toggle_userAccountFlags(self, search_filter, flags, flags_str=None,
+ on=True, strict=False):
+ """Toggle_userAccountFlags
+
+ :param search_filter: LDAP filter to find the user (eg
+ samccountname=name)
+ :param flags: samba.dsdb.UF_* flags
+ :param on: on=True (default) => set, on=False => unset
+ :param strict: strict=False (default) ignore if no action is needed
+ strict=True raises an Exception if...
"""
res = self.search(base=self.domain_dn(), scope=ldb.SCOPE_SUBTREE,
- expression=filter, attrs=["userAccountControl"])
+ expression=search_filter, attrs=["userAccountControl"])
+ if len(res) == 0:
+ raise Exception("Unable to find account where '%s'" % search_filter)
assert(len(res) == 1)
- user_dn = res[0].dn
+ account_dn = res[0].dn
+
+ old_uac = int(res[0]["userAccountControl"][0])
+ if on:
+ if strict and (old_uac & flags):
+ error = "Account flag(s) '%s' already set" % flags_str
+ raise Exception(error)
+
+ new_uac = old_uac | flags
+ else:
+ if strict and not (old_uac & flags):
+ error = "Account flag(s) '%s' already unset" % flags_str
+ raise Exception(error)
+
+ new_uac = old_uac & ~flags
- userAccountControl = int(res[0]["userAccountControl"][0])
- if (userAccountControl & 0x2):
- userAccountControl = userAccountControl & ~0x2 # remove disabled bit
- if (userAccountControl & 0x20):
- userAccountControl = userAccountControl & ~0x20 # remove 'no password required' bit
+ if old_uac == new_uac:
+ return
mod = """
dn: %s
changetype: modify
-replace: userAccountControl
+delete: userAccountControl
userAccountControl: %u
-""" % (user_dn, userAccountControl)
+add: userAccountControl
+userAccountControl: %u
+""" % (account_dn, old_uac, new_uac)
self.modify_ldif(mod)
-
- def force_password_change_at_next_login(self, filter):
+
+ def force_password_change_at_next_login(self, search_filter):
"""Forces a password change at next login
-
- :param 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=filter, attrs=[])
+ expression=search_filter, attrs=[])
+ if len(res) == 0:
+ raise Exception('Unable to find user "%s"' % search_filter)
assert(len(res) == 1)
user_dn = res[0].dn
self.modify_ldif(mod)
def newgroup(self, groupname, groupou=None, grouptype=None,
- description=None, mailaddress=None, notes=None):
+ description=None, mailaddress=None, notes=None, sd=None):
"""Adds a new group with additional parameters
:param groupname: Name of the new group
:param description: Description of the new group
:param mailaddress: Email address of the new group
:param notes: Notes of the new group
+ :param sd: security descriptor of the object
"""
group_dn = "CN=%s,%s,%s" % (groupname, (groupou or "CN=Users"), self.domain_dn())
"objectClass": "group"}
if grouptype is not None:
- ldbmessage["groupType"] = "%d" % ((grouptype)-2**32)
+ ldbmessage["groupType"] = normalise_int32(grouptype)
if description is not None:
ldbmessage["description"] = description
if notes is not None:
ldbmessage["info"] = notes
- self.transaction_start()
- try:
- self.add(ldbmessage)
- except:
- self.transaction_cancel()
- raise
- else:
- self.transaction_commit()
+ if sd is not None:
+ ldbmessage["nTSecurityDescriptor"] = ndr_pack(sd)
+
+ self.add(ldbmessage)
def deletegroup(self, groupname):
"""Deletes a group
:param groupname: Name of the target group
"""
- groupfilter = "(&(sAMAccountName=%s)(objectCategory=%s,%s))" % (groupname, "CN=Group,CN=Schema,CN=Configuration", self.domain_dn())
+ groupfilter = "(&(sAMAccountName=%s)(objectCategory=%s,%s))" % (ldb.binary_encode(groupname), "CN=Group,CN=Schema,CN=Configuration", self.domain_dn())
self.transaction_start()
try:
targetgroup = self.search(base=self.domain_dn(), scope=ldb.SCOPE_SUBTREE,
if len(targetgroup) == 0:
raise Exception('Unable to find group "%s"' % groupname)
assert(len(targetgroup) == 1)
- self.delete(targetgroup[0].dn);
+ self.delete(targetgroup[0].dn)
except:
self.transaction_cancel()
raise
else:
self.transaction_commit()
- def add_remove_group_members(self, groupname, listofmembers,
+ def add_remove_group_members(self, groupname, members,
add_members_operation=True):
"""Adds or removes group members
:param groupname: Name of the target group
- :param listofmembers: Comma-separated list of group members
- :param add_members_operation: Defines if its an add or remove operation
+ :param members: list of group members
+ :param add_members_operation: Defines if its an add or remove
+ operation
"""
- groupfilter = "(&(sAMAccountName=%s)(objectCategory=%s,%s))" % (groupname, "CN=Group,CN=Schema,CN=Configuration", self.domain_dn())
- groupmembers = listofmembers.split(',')
+ groupfilter = "(&(sAMAccountName=%s)(objectCategory=%s,%s))" % (
+ ldb.binary_encode(groupname), "CN=Group,CN=Schema,CN=Configuration", self.domain_dn())
self.transaction_start()
try:
changetype: modify
""" % (str(targetgroup[0].dn))
- for member in groupmembers:
+ for member in members:
targetmember = self.search(base=self.domain_dn(), scope=ldb.SCOPE_SUBTREE,
- expression="(|(sAMAccountName=%s)(CN=%s))" % (member, member), attrs=[])
+ expression="(|(sAMAccountName=%s)(CN=%s))" % (
+ ldb.binary_encode(member), ldb.binary_encode(member)), attrs=[])
if len(targetmember) != 1:
- continue
+ continue
if add_members_operation is True and (targetgroup[0].get('member') is None or str(targetmember[0].dn) not in targetgroup[0]['member']):
- modified = True
- addtargettogroup += """add: member
+ modified = True
+ addtargettogroup += """add: member
member: %s
""" % (str(targetmember[0].dn))
elif add_members_operation is False and (targetgroup[0].get('member') is not None and str(targetmember[0].dn) in targetgroup[0]['member']):
- modified = True
- addtargettogroup += """delete: member
+ modified = True
+ addtargettogroup += """delete: member
member: %s
""" % (str(targetmember[0].dn))
if modified is True:
- self.modify_ldif(addtargettogroup)
+ self.modify_ldif(addtargettogroup)
except:
self.transaction_cancel()
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):
+ 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
:param internetaddress: Home page of the new user
:param telephonenumber: Phone number of the new user
:param physicaldeliveryoffice: Office location of the new user
+ :param sd: security descriptor of the object
+ :param setpassword: optionally disable password reset
"""
- displayname = "";
+ displayname = ""
if givenname is not None:
displayname += givenname
user_dn = "CN=%s,%s,%s" % (cn, (userou or "CN=Users"), self.domain_dn())
+ dnsdomain = ldb.Dn(self, self.domain_dn()).canonical_str().replace("/", "")
+ user_principal_name = "%s@%s" % (username, dnsdomain)
# The new user record. Note the reliance on the SAMLDB module which
# fills in the default informations
ldbmessage = {"dn": user_dn,
- "sAMAccountName": username,
- "objectClass": "user"}
+ "sAMAccountName": username,
+ "userPrincipalName": user_principal_name,
+ "objectClass": "user"}
if surname is not None:
ldbmessage["sn"] = surname
if physicaldeliveryoffice is not None:
ldbmessage["physicalDeliveryOfficeName"] = physicaldeliveryoffice
+ if sd is not None:
+ ldbmessage["nTSecurityDescriptor"] = ndr_pack(sd)
+
self.transaction_start()
try:
self.add(ldbmessage)
# Sets the password for it
- self.setpassword("(dn=" + user_dn + ")", password,
- force_password_change_at_next_login_req)
+ if setpassword:
+ self.setpassword("(samAccountName=%s)" % ldb.binary_encode(username), password,
+ force_password_change_at_next_login_req)
+ except:
+ self.transaction_cancel()
+ raise
+ else:
+ self.transaction_commit()
+
+ def deleteuser(self, username):
+ """Deletes a user
+
+ :param username: Name of the target user
+ """
+
+ filter = "(&(sAMAccountName=%s)(objectCategory=%s,%s))" % (ldb.binary_encode(username), "CN=Person,CN=Schema,CN=Configuration", self.domain_dn())
+ self.transaction_start()
+ try:
+ target = self.search(base=self.domain_dn(), scope=ldb.SCOPE_SUBTREE,
+ expression=filter, attrs=[])
+ if len(target) == 0:
+ raise Exception('Unable to find user "%s"' % username)
+ assert(len(target) == 1)
+ self.delete(target[0].dn)
except:
self.transaction_cancel()
raise
else:
self.transaction_commit()
- def setpassword(self, filter, password,
- force_change_at_next_login=False,
- username=None):
+ def setpassword(self, search_filter, password,
+ force_change_at_next_login=False, username=None):
"""Sets the password for a user
-
- Note: This call uses the "userPassword" attribute to set the password.
- This works correctly on SAMBA 4 and on Windows DCs with
- "2003 Native" or higer domain function level.
- :param 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
"""
self.transaction_start()
try:
res = self.search(base=self.domain_dn(), scope=ldb.SCOPE_SUBTREE,
- expression=filter, attrs=[])
+ expression=search_filter, attrs=[])
if len(res) == 0:
- raise Exception('Unable to find user "%s"' % (username or filter))
- assert(len(res) == 1)
+ raise Exception('Unable to find user "%s"' % (username or search_filter))
+ if len(res) > 1:
+ raise Exception('Matched %u multiple users with filter "%s"' % (len(res), search_filter))
user_dn = res[0].dn
-
+ pw = unicode('"' + password + '"', 'utf-8').encode('utf-16-le')
setpw = """
dn: %s
changetype: modify
-replace: userPassword
-userPassword:: %s
-""" % (user_dn, base64.b64encode(password))
+replace: unicodePwd
+unicodePwd:: %s
+""" % (user_dn, base64.b64encode(pw))
self.modify_ldif(setpw)
if force_change_at_next_login:
self.force_password_change_at_next_login(
- "(dn=" + str(user_dn) + ")")
+ "(distinguishedName=" + str(user_dn) + ")")
# modify the userAccountControl to remove the disabled bit
- self.enable_account(filter)
+ self.enable_account(search_filter)
except:
self.transaction_cancel()
raise
else:
self.transaction_commit()
- def setexpiry(self, filter, expiry_seconds, no_expiry_req=False):
+ def setexpiry(self, search_filter, expiry_seconds, no_expiry_req=False):
"""Sets the account expiry for a user
-
- :param 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
"""
self.transaction_start()
try:
res = self.search(base=self.domain_dn(), scope=ldb.SCOPE_SUBTREE,
- expression=filter,
+ expression=search_filter,
attrs=["userAccountControl", "accountExpires"])
+ if len(res) == 0:
+ raise Exception('Unable to find user "%s"' % search_filter)
assert(len(res) == 1)
user_dn = res[0].dn
:param sid: The new domain sid to use.
"""
- dsdb.samdb_set_domain_sid(self, sid)
+ 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)
- """
- dsdb.samdb_get_domain_sid(self)
+ domain_sid = property(get_domain_sid, set_domain_sid,
+ "SID for the domain")
def set_invocation_id(self, invocation_id):
"""Set the invocation id for this SamDB handle.
:param invocation_id: GUID of the invocation id.
"""
- dsdb.dsdb_set_ntds_invocation_id(self, invocation_id)
+ dsdb._dsdb_set_ntds_invocation_id(self, invocation_id)
+
+ def get_invocation_id(self):
+ """Get the invocation_id id"""
+ return dsdb._samdb_ntds_invocation_id(self)
+
+ invocation_id = property(get_invocation_id, set_invocation_id,
+ "Invocation ID GUID")
def get_oid_from_attid(self, attid):
- return dsdb.dsdb_get_oid_from_attid(self, attid)
+ return dsdb._dsdb_get_oid_from_attid(self, attid)
- def get_invocation_id(self):
- "Get the invocation_id id"
- return dsdb.samdb_ntds_invocation_id(self)
+ def get_attid_from_lDAPDisplayName(self, ldap_display_name,
+ is_schema_nc=False):
+ '''return the attribute ID for a LDAP attribute as an integer as found in DRSUAPI'''
+ return dsdb._dsdb_get_attid_from_lDAPDisplayName(self,
+ ldap_display_name, is_schema_nc)
+
+ def get_syntax_oid_from_lDAPDisplayName(self, ldap_display_name):
+ '''return the syntax OID for a LDAP attribute as a string'''
+ return dsdb._dsdb_get_syntax_oid_from_lDAPDisplayName(self, ldap_display_name)
+
+ def get_systemFlags_from_lDAPDisplayName(self, ldap_display_name):
+ '''return the systemFlags for a LDAP attribute as a integer'''
+ return dsdb._dsdb_get_systemFlags_from_lDAPDisplayName(self, ldap_display_name)
+
+ def get_linkId_from_lDAPDisplayName(self, ldap_display_name):
+ '''return the linkID for a LDAP attribute as a integer'''
+ return dsdb._dsdb_get_linkId_from_lDAPDisplayName(self, ldap_display_name)
+
+ def get_lDAPDisplayName_by_attid(self, attid):
+ '''return the lDAPDisplayName from an integer DRS attribute ID'''
+ return dsdb._dsdb_get_lDAPDisplayName_by_attid(self, attid)
+
+ def get_backlink_from_lDAPDisplayName(self, ldap_display_name):
+ '''return the attribute name of the corresponding backlink from the name
+ of a forward link attribute. If there is no backlink return None'''
+ return dsdb._dsdb_get_backlink_from_lDAPDisplayName(self, ldap_display_name)
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
:param ntds_settings_dn: The new DN to use
"""
- dsdb.samdb_set_ntds_settings_dn(self, ntds_settings_dn)
-
- invocation_id = property(get_invocation_id, set_invocation_id)
-
- domain_sid = property(get_domain_sid, set_domain_sid)
+ dsdb._samdb_set_ntds_settings_dn(self, ntds_settings_dn)
def get_ntds_GUID(self):
- "Get the NTDS objectGUID"
- return dsdb.samdb_ntds_objectGUID(self)
+ """Get the NTDS objectGUID"""
+ return dsdb._samdb_ntds_objectGUID(self)
def server_site_name(self):
- "Get the server site name"
- return dsdb.samdb_server_site_name(self)
+ """Get the server site name"""
+ return dsdb._samdb_server_site_name(self)
+
+ def host_dns_name(self):
+ """return the DNS name of this host"""
+ res = self.search(base='', scope=ldb.SCOPE_BASE, attrs=['dNSHostName'])
+ return res[0]['dNSHostName'][0]
+
+ def domain_dns_name(self):
+ """return the DNS name of the domain root"""
+ domain_dn = self.get_default_basedn()
+ return domain_dn.canonical_str().split('/')[0]
+
+ def forest_dns_name(self):
+ """return the DNS name of the forest root"""
+ forest_dn = self.get_root_basedn()
+ return forest_dn.canonical_str().split('/')[0]
def load_partition_usn(self, base_dn):
- return dsdb.dsdb_load_partition_usn(self, base_dn)
+ return dsdb._dsdb_load_partition_usn(self, base_dn)
+
+ def set_schema(self, schema, write_indices_and_attributes=True):
+ self.set_schema_from_ldb(schema.ldb, write_indices_and_attributes=write_indices_and_attributes)
+
+ def set_schema_from_ldb(self, ldb_conn, write_indices_and_attributes=True):
+ dsdb._dsdb_set_schema_from_ldb(self, ldb_conn, write_indices_and_attributes)
- def set_schema(self, schema):
- self.set_schema_from_ldb(schema.ldb)
+ def dsdb_DsReplicaAttribute(self, ldb, ldap_display_name, ldif_elements):
+ '''convert a list of attribute values to a DRSUAPI DsReplicaAttribute'''
+ return dsdb._dsdb_DsReplicaAttribute(ldb, ldap_display_name, ldif_elements)
- def set_schema_from_ldb(self, ldb):
- dsdb.dsdb_set_schema_from_ldb(self, ldb)
+ def dsdb_normalise_attributes(self, ldb, ldap_display_name, ldif_elements):
+ '''normalise a list of attribute values'''
+ return dsdb._dsdb_normalise_attributes(ldb, ldap_display_name, ldif_elements)
+
+ 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
+ """
+ if len(self.hash_oid_name.keys()) == 0:
+ self._populate_oid_attid()
+ if self.hash_oid_name.has_key(self.get_oid_from_attid(attid)):
+ return self.hash_oid_name[self.get_oid_from_attid(attid)]
+ else:
+ return None
+
+ def _populate_oid_attid(self):
+ """Populate the hash hash_oid_name.
+
+ 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",
+ controls=["search_options:1:2"],
+ attrs=["attributeID",
+ "lDAPDisplayName"])
+ if len(res) > 0:
+ for e in res:
+ 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
+
+ :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="distinguishedName=%s" % dn,
+ scope=ldb.SCOPE_SUBTREE,
+ controls=["search_options:1:2"],
+ attrs=["replPropertyMetaData"])
+ if len(res) == 0:
+ return None
+
+ repl = ndr_unpack(drsblobs.replPropertyMetaDataBlob,
+ str(res[0]["replPropertyMetaData"]))
+ ctr = repl.ctr
+ if len(self.hash_oid_name.keys()) == 0:
+ self._populate_oid_attid()
+ for o in ctr.array:
+ # Search for Description
+ att_oid = self.get_oid_from_attid(o.attid)
+ if self.hash_oid_name.has_key(att_oid) and\
+ att.lower() == self.hash_oid_name[att_oid].lower():
+ return o.version
+ return None
+
+ def set_attribute_replmetadata_version(self, dn, att, value,
+ addifnotexist=False):
+ res = self.search(expression="distinguishedName=%s" % dn,
+ scope=ldb.SCOPE_SUBTREE,
+ controls=["search_options:1:2"],
+ attrs=["replPropertyMetaData"])
+ if len(res) == 0:
+ return None
+
+ repl = ndr_unpack(drsblobs.replPropertyMetaDataBlob,
+ str(res[0]["replPropertyMetaData"]))
+ ctr = repl.ctr
+ now = samba.unix2nttime(int(time.time()))
+ found = False
+ if len(self.hash_oid_name.keys()) == 0:
+ self._populate_oid_attid()
+ for o in ctr.array:
+ # Search for Description
+ att_oid = self.get_oid_from_attid(o.attid)
+ if self.hash_oid_name.has_key(att_oid) and\
+ att.lower() == self.hash_oid_name[att_oid].lower():
+ found = True
+ seq = self.sequence_number(ldb.SEQ_NEXT)
+ o.version = value
+ o.originating_change_time = now
+ o.originating_invocation_id = misc.GUID(self.get_invocation_id())
+ o.originating_usn = seq
+ o.local_usn = seq
+
+ if not found and addifnotexist and len(ctr.array) >0:
+ o2 = drsblobs.replPropertyMetaData1()
+ o2.attid = 589914
+ att_oid = self.get_oid_from_attid(o2.attid)
+ seq = self.sequence_number(ldb.SEQ_NEXT)
+ o2.version = value
+ o2.originating_change_time = now
+ o2.originating_invocation_id = misc.GUID(self.get_invocation_id())
+ o2.originating_usn = seq
+ o2.local_usn = seq
+ found = True
+ tab = ctr.array
+ tab.append(o2)
+ ctr.count = ctr.count + 1
+ ctr.array = tab
+
+ if found :
+ replBlob = ndr_pack(repl)
+ msg = ldb.Message()
+ msg.dn = res[0].dn
+ msg["replPropertyMetaData"] = ldb.MessageElement(replBlob,
+ ldb.FLAG_MOD_REPLACE,
+ "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)
+ dsdb._dsdb_write_prefixes_from_schema_to_ldb(self)
+
+ def get_partitions_dn(self):
+ return dsdb._dsdb_get_partitions_dn(self)
+
+ def get_nc_root(self, dn):
+ return dsdb._dsdb_get_nc_root(self, dn)
+
+ def get_wellknown_dn(self, nc_root, wkguid):
+ return dsdb._dsdb_get_wellknown_dn(self, nc_root, wkguid)
+
+ def set_minPwdAge(self, value):
+ m = ldb.Message()
+ m.dn = ldb.Dn(self, self.domain_dn())
+ m["minPwdAge"] = ldb.MessageElement(value, ldb.FLAG_MOD_REPLACE, "minPwdAge")
+ self.modify(m)
+
+ def get_minPwdAge(self):
+ res = self.search(self.domain_dn(), scope=ldb.SCOPE_BASE, attrs=["minPwdAge"])
+ if len(res) == 0:
+ return None
+ elif not "minPwdAge" in res[0]:
+ return None
+ else:
+ return res[0]["minPwdAge"][0]
+
+ def set_minPwdLength(self, value):
+ m = ldb.Message()
+ m.dn = ldb.Dn(self, self.domain_dn())
+ m["minPwdLength"] = ldb.MessageElement(value, ldb.FLAG_MOD_REPLACE, "minPwdLength")
+ self.modify(m)
+
+ def get_minPwdLength(self):
+ res = self.search(self.domain_dn(), scope=ldb.SCOPE_BASE, attrs=["minPwdLength"])
+ if len(res) == 0:
+ return None
+ elif not "minPwdLength" in res[0]:
+ return None
+ else:
+ return res[0]["minPwdLength"][0]
+
+ def set_pwdProperties(self, value):
+ m = ldb.Message()
+ m.dn = ldb.Dn(self, self.domain_dn())
+ m["pwdProperties"] = ldb.MessageElement(value, ldb.FLAG_MOD_REPLACE, "pwdProperties")
+ self.modify(m)
+
+ def get_pwdProperties(self):
+ res = self.search(self.domain_dn(), scope=ldb.SCOPE_BASE, attrs=["pwdProperties"])
+ if len(res) == 0:
+ return None
+ elif not "pwdProperties" in res[0]:
+ return None
+ else:
+ return res[0]["pwdProperties"][0]
+
+ def set_dsheuristics(self, dsheuristics):
+ m = ldb.Message()
+ 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")
+ else:
+ m["dSHeuristics"] = ldb.MessageElement([], ldb.FLAG_MOD_DELETE,
+ "dSHeuristics")
+ self.modify(m)
+
+ def get_dsheuristics(self):
+ res = self.search("CN=Directory Service,CN=Windows NT,CN=Services,%s"
+ % self.get_config_basedn().get_linearized(),
+ scope=ldb.SCOPE_BASE, attrs=["dSHeuristics"])
+ if len(res) == 0:
+ dsheuristics = None
+ elif "dSHeuristics" in res[0]:
+ dsheuristics = res[0]["dSHeuristics"][0]
+ else:
+ dsheuristics = None
+
+ return dsheuristics
+
+ def create_ou(self, ou_dn, description=None, name=None, sd=None):
+ """Creates an organizationalUnit object
+ :param ou_dn: dn of the new object
+ :param description: description attribute
+ :param name: name atttribute
+ :param sd: security descriptor of the object, can be
+ an SDDL string or security.descriptor type
+ """
+ m = {"dn": ou_dn,
+ "objectClass": "organizationalUnit"}
+
+ if description:
+ m["description"] = description
+ if name:
+ m["name"] = name
+
+ if sd:
+ m["nTSecurityDescriptor"] = ndr_pack(sd)
+ self.add(m)
+
+ def sequence_number(self, seq_type):
+ """Returns the value of the sequence number according to the requested type
+ :param seq_type: type of sequence number
+ """
+ self.transaction_start()
+ try:
+ seq = super(SamDB, self).sequence_number(seq_type)
+ except:
+ self.transaction_cancel()
+ raise
+ else:
+ self.transaction_commit()
+ return seq
+
+ def get_dsServiceName(self):
+ '''get the NTDS DN from the rootDSE'''
+ res = self.search(base="", scope=ldb.SCOPE_BASE, attrs=["dsServiceName"])
+ return res[0]["dsServiceName"][0]
+
+ def get_serverName(self):
+ '''get the server DN from the rootDSE'''
+ res = self.search(base="", scope=ldb.SCOPE_BASE, attrs=["serverName"])
+ return res[0]["serverName"][0]
git.samba.org / samba.git / blobdiff