-#!/usr/bin/env 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
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:
+ '''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):
+ '''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
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
+ 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)
- :flags: samba.dsdb.UF_* flags
- :on: on=True (default) => set, on=False => unset
- :strict: strict=False (default) ignore if no action is needed
+ :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=search_filter, attrs=["userAccountControl"])
if len(res) == 0:
- raise Exception('Unable to find user "%s"' % search_filter)
+ raise Exception("Unable to find account where '%s'" % search_filter)
assert(len(res) == 1)
account_dn = res[0].dn
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)
+ 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 = 'userAccountFlags[%d:0x%08X] not contain 0x%X' % (old_uac, old_uac, flags)
+ error = "Account flag(s) '%s' already unset" % flags_str
raise Exception(error)
new_uac = old_uac & ~flags
"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,
raise Exception('Unable to find group "%s"' % groupname)
assert(len(targetgroup) == 1)
self.delete(targetgroup[0].dn)
- except Exception:
+ 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 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
if modified is True:
self.modify_ldif(addtargettogroup)
- except Exception:
+ except:
self.transaction_cancel()
raise
else:
# 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:
+ 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:
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: unicodePwd
unicodePwd:: %s
-""" % (user_dn, base64.b64encode(("\"" + password + "\"").encode('utf-16-le')))
+""" % (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(search_filter)
- except Exception:
+ except:
self.transaction_cancel()
raise
else:
""" % (user_dn, userAccountControl, accountExpires)
self.modify_ldif(setexp)
- except Exception:
+ except:
self.transaction_cancel()
raise
else:
"""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)
- def set_schema(self, schema):
- self.set_schema_from_ldb(schema.ldb)
+ 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):
- dsdb._dsdb_set_schema_from_ldb(self, ldb_conn)
+ 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 dsdb_DsReplicaAttribute(self, ldb, ldap_display_name, ldif_elements):
'''convert a list of attribute values to a DRSUAPI DsReplicaAttribute'''
for the given attribute. None if the attribute is not replicated
"""
- res = self.search(expression="dn=%s" % dn,
+ res = self.search(expression="distinguishedName=%s" % dn,
scope=ldb.SCOPE_SUBTREE,
controls=["search_options:1:2"],
attrs=["replPropertyMetaData"])
def set_attribute_replmetadata_version(self, dn, att, value,
addifnotexist=False):
- res = self.search(expression="dn=%s" % dn,
+ res = self.search(expression="distinguishedName=%s" % dn,
scope=ldb.SCOPE_SUBTREE,
controls=["search_options:1:2"],
attrs=["replPropertyMetaData"])
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["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)
+ 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