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"
elif url is None and lp is not None:
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)
dsdb._dsdb_set_am_rodc(self, am_rodc)
def connect(self, url=None, flags=0, options=None):
- if self.lp is not None:
+ 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)
: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, on=True, strict=False):
+ """toggle_userAccountFlags
+
+ :param search_filter: LDAP filter to find the user (eg
+ samccountname=name)
+ :flags: samba.dsdb.UF_* flags
+ :on: on=True (default) => set, on=False => unset
+ :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=search_filter, attrs=["userAccountControl"])
if len(res) == 0:
raise Exception('Unable to find user "%s"' % search_filter)
assert(len(res) == 1)
- user_dn = res[0].dn
+ account_dn = res[0].dn
- userAccountControl = int(res[0]["userAccountControl"][0])
- if userAccountControl & 0x2:
- # remove disabled bit
- userAccountControl = userAccountControl & ~0x2
- if userAccountControl & 0x20:
- # remove 'no password required' bit
- userAccountControl = userAccountControl & ~0x20
+ old_uac = int(res[0]["userAccountControl"][0])
+ if on:
+ if strict and (old_uac & flags):
+ error = 'userAccountFlags[%d:0x%08X] already contain 0x%X' % (old_uac, old_uac, flags)
+ raise Exception(error)
+
+ new_uac = old_uac | flags
+ else:
+ if strict and not (old_uac & flags):
+ error = 'userAccountFlags[%d:0x%08X] not contain 0x%X' % (old_uac, old_uac, flags)
+ raise Exception(error)
+
+ new_uac = old_uac & ~flags
+
+ 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, search_filter):
"objectClass": "group"}
if grouptype is not None:
- ldbmessage["groupType"] = self.normalise_int32(grouptype)
+ ldbmessage["groupType"] = normalise_int32(grouptype)
if description is not None:
ldbmessage["description"] = description
: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,
operation
"""
- 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())
groupmembers = listofmembers.split(',')
self.transaction_start()
for member in groupmembers:
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
# Sets the password for it
if setpassword:
- self.setpassword("(samAccountName=%s)" % username, password,
+ self.setpassword("(samAccountName=%s)" % ldb.binary_encode(username), password,
force_password_change_at_next_login_req)
except Exception:
self.transaction_cancel()
'''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.
if sd:
m["nTSecurityDescriptor"] = ndr_pack(sd)
self.add(m)
-
- def normalise_int32(self, ivalue):
- '''normalise a ldap integer to signed 32 bit'''
- if int(ivalue) & 0x80000000:
- return str(int(ivalue) - 0x100000000)
- return str(ivalue)
git.samba.org / amitay / samba.git / blobdiff