JezK

Edit File: samdb.py

# 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 samba
import ldb
import time
import base64
import os
import re
from samba import dsdb, dsdb_dns
from samba.ndr import ndr_unpack, ndr_pack
from samba.dcerpc import drsblobs, misc
from samba.common import normalise_int32
from samba.common import get_bytes, cmp
from samba.dcerpc import security
from samba import is_ad_dc_built
import binascii

__docformat__ = "restructuredText"

def get_default_backend_store():
    return "tdb"

class SamDBError(Exception):
    pass

class SamDBNotFoundError(SamDBError):
    pass

class SamDB(samba.Ldb):
    """The SAM database."""

hash_oid_name = {}
    hash_well_known = {}

class _CleanUpOnError:
        def __init__(self, samdb, dn):
            self.samdb = samdb
            self.dn = dn

def __enter__(self):
            pass

def __exit__(self, exc_type, exc_val, exc_tb):
            if exc_type is not None:
                # We failed to modify the account. If we connected to the
                # database over LDAP, we don't have transactions, and so when
                # we call transaction_cancel(), the account will still exist in
                # a half-created state. We'll delete the account to ensure that
                # doesn't happen.
                self.samdb.delete(self.dn)

# Don't suppress any exceptions
            return False

def __init__(self, url=None, lp=None, modules_dir=None, session_info=None,
                 credentials=None, flags=ldb.FLG_DONT_CREATE_DB,
                 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.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)

if global_schema:
            dsdb._dsdb_set_global_schema(self)

if am_rodc is not None:
            dsdb._dsdb_set_am_rodc(self, am_rodc)

def connect(self, url=None, flags=0, options=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 schema_dn(self):
        '''return the schema partition dn'''
        return str(self.get_schema_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 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=search_filter, attrs=["userAccountControl"])
        if len(res) == 0:
                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 = "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

if old_uac == new_uac:
            return

mod = """
dn: %s
changetype: modify
delete: userAccountControl
userAccountControl: %u
add: userAccountControl
userAccountControl: %u
""" % (account_dn, old_uac, new_uac)
        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)
        """
        res = self.search(base=self.domain_dn(), scope=ldb.SCOPE_SUBTREE,
                          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

mod = """
dn: %s
changetype: modify
replace: pwdLastSet
pwdLastSet: 0
""" % (user_dn)
        self.modify_ldif(mod)

def unlock_account(self, search_filter):
        """Unlock a user account by resetting lockoutTime to 0.
        This does also reset the badPwdCount to 0.

:param search_filter: LDAP filter to find the user (e.g.
            sAMAccountName=username)
        """
        res = self.search(base=self.domain_dn(),
                          scope=ldb.SCOPE_SUBTREE,
                          expression=search_filter,
                          attrs=[])
        if len(res) == 0:
            raise SamDBNotFoundError('Unable to find user "%s"' % search_filter)
        if len(res) != 1:
            raise SamDBError('User "%s" is not unique' % search_filter)
        user_dn = res[0].dn

mod = """
dn: %s
changetype: modify
replace: lockoutTime
lockoutTime: 0
""" % (user_dn)
        self.modify_ldif(mod)

def newgroup(self, groupname, groupou=None, grouptype=None,
                 description=None, mailaddress=None, notes=None, sd=None,
                 gidnumber=None, nisdomain=None):
        """Adds a new group with additional parameters

:param groupname: Name of the new group
        :param grouptype: Type 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 gidnumber: GID Number of the new group
        :param nisdomain: NIS Domain Name of the new group
        :param sd: security descriptor of the object
        """

if groupou:
            group_dn = "CN=%s,%s,%s" % (groupname, groupou, self.domain_dn())
        else:
            group_dn = "CN=%s,%s" % (groupname, self.get_wellknown_dn(
                                        self.get_default_basedn(),
                                        dsdb.DS_GUID_USERS_CONTAINER))

# The new user record. Note the reliance on the SAMLDB module which
        # fills in the default information
        ldbmessage = {"dn": group_dn,
                      "sAMAccountName": groupname,
                      "objectClass": "group"}

if grouptype is not None:
            ldbmessage["groupType"] = normalise_int32(grouptype)

if description is not None:
            ldbmessage["description"] = description

if mailaddress is not None:
            ldbmessage["mail"] = mailaddress

if notes is not None:
            ldbmessage["info"] = notes

if gidnumber is not None:
            ldbmessage["gidNumber"] = normalise_int32(gidnumber)

if nisdomain is not None:
            ldbmessage["msSFU30Name"] = groupname
            ldbmessage["msSFU30NisDomain"] = nisdomain

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))" % (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,
                                      expression=groupfilter, attrs=[])
            if len(targetgroup) == 0:
                raise Exception('Unable to find group "%s"' % groupname)
            assert(len(targetgroup) == 1)
            self.delete(targetgroup[0].dn)
        except:
            self.transaction_cancel()
            raise
        else:
            self.transaction_commit()

def group_member_filter(self, member, member_types):
        filter = ""

all_member_types = [ 'user',
                             'group',
                             'computer',
                             'serviceaccount',
                             'contact',
                           ]

if 'all' in member_types:
            member_types = all_member_types

for member_type in member_types:
            if member_type not in all_member_types:
                raise Exception('Invalid group member type "%s". '
                                'Valid types are %s and all.' %
                                (member_type, ", ".join(all_member_types)))

if 'user' in member_types:
            filter += ('(&(sAMAccountName=%s)(samAccountType=%d))' %
                       (ldb.binary_encode(member), dsdb.ATYPE_NORMAL_ACCOUNT))
        if 'group' in member_types:
            filter += ('(&(sAMAccountName=%s)'
                       '(objectClass=group)'
                       '(!(groupType:1.2.840.113556.1.4.803:=1)))' %
                       ldb.binary_encode(member))
        if 'computer' in member_types:
            samaccountname = member
            if member[-1] != '$':
                samaccountname = "%s$" % member
            filter += ('(&(samAccountType=%d)'
                       '(!(objectCategory=msDS-ManagedServiceAccount))'
                       '(sAMAccountName=%s))' %
                       (dsdb.ATYPE_WORKSTATION_TRUST,
                        ldb.binary_encode(samaccountname)))
        if 'serviceaccount' in member_types:
            samaccountname = member
            if member[-1] != '$':
                samaccountname = "%s$" % member
            filter += ('(&(samAccountType=%d)'
                       '(objectCategory=msDS-ManagedServiceAccount)'
                       '(sAMAccountName=%s))' %
                       (dsdb.ATYPE_WORKSTATION_TRUST,
                        ldb.binary_encode(samaccountname)))
        if 'contact' in member_types:
            filter += ('(&(objectCategory=Person)(!(objectSid=*))(name=%s))' %
                       ldb.binary_encode(member))

filter = "(|%s)" % filter

return filter

def add_remove_group_members(self, groupname, members,
                                 add_members_operation=True,
                                 member_types=[ 'user', 'group', 'computer' ],
                                 member_base_dn=None):
        """Adds or removes group members

:param groupname: Name of the target group
        :param members: list of group members
        :param add_members_operation: Defines if its an add or remove
            operation
        """

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,
                                      expression=groupfilter, attrs=['member'])
            if len(targetgroup) == 0:
                raise Exception('Unable to find group "%s"' % groupname)
            assert(len(targetgroup) == 1)

modified = False

addtargettogroup = """
dn: %s
changetype: modify
""" % (str(targetgroup[0].dn))

for member in members:
                targetmember_dn = None
                if member_base_dn is None:
                    member_base_dn = self.domain_dn()

try:
                    membersid = security.dom_sid(member)
                    targetmember_dn = "<SID=%s>" % str(membersid)
                except TypeError as e:
                    pass

if targetmember_dn is None:
                    try:
                        member_dn = ldb.Dn(self, member)
                        if member_dn.get_linearized() == member_dn.extended_str(1):
                            full_member_dn = self.normalize_dn_in_domain(member_dn)
                        else:
                            full_member_dn = member_dn
                        targetmember_dn = full_member_dn.extended_str(1)
                    except ValueError as e:
                        pass

if targetmember_dn is None:
                    filter = self.group_member_filter(member, member_types)
                    targetmember = self.search(base=member_base_dn,
                                               scope=ldb.SCOPE_SUBTREE,
                                               expression=filter,
                                               attrs=[])

if len(targetmember) > 1:
                        targetmemberlist_str = ""
                        for msg in targetmember:
                            targetmemberlist_str += "%s\n" % msg.get("dn")
                        raise Exception('Found multiple results for "%s":\n%s' %
                                        (member, targetmemberlist_str))
                    if len(targetmember) != 1:
                        raise Exception('Unable to find "%s". Operation cancelled.' % member)
                    targetmember_dn = targetmember[0].dn.extended_str(1)

if add_members_operation is True and (targetgroup[0].get('member') is None or get_bytes(targetmember_dn) not in [str(x) for x in targetgroup[0]['member']]):
                    modified = True
                    addtargettogroup += """add: member
member: %s
""" % (str(targetmember_dn))

elif add_members_operation is False and (targetgroup[0].get('member') is not None and get_bytes(targetmember_dn) in targetgroup[0]['member']):
                    modified = True
                    addtargettogroup += """delete: member
member: %s
""" % (str(targetmember_dn))

if modified is True:
                self.modify_ldif(addtargettogroup)

except:
            self.transaction_cancel()
            raise
        else:
            self.transaction_commit()

def prepare_attr_replace(self, msg, old, attr_name, value):
        """Changes the MessageElement with the given attr_name of the
        given Message. If the value is "" set an empty value and the flag
        FLAG_MOD_DELETE, otherwise set the new value and FLAG_MOD_REPLACE.
        If the value is None or the Message contains the attr_name with this
        value, nothing will changed."""
        # skip unchanged attribute
        if value is None:
            return
        if attr_name in old and str(value) == str(old[attr_name]):
            return

# remove attribute
        if len(value) == 0:
            if attr_name in old:
                el = ldb.MessageElement([], ldb.FLAG_MOD_DELETE, attr_name)
                msg.add(el)
            return

# change attribute
        el = ldb.MessageElement(value, ldb.FLAG_MOD_REPLACE, attr_name)
        msg.add(el)

def fullname_from_names(self, given_name=None, initials=None, surname=None,
                            old_attrs={}, fallback_default=""):
        """Prepares new combined fullname, using the name parts.
        Used for things like displayName or cn.
        Use the original name values, if no new one is specified."""

attrs = {"givenName": given_name,
                 "initials": initials,
                 "sn": surname}

# if the attribute is not specified, try to use the old one
        for attr_name, attr_value in attrs.items():
            if attr_value is None and attr_name in old_attrs:
                attrs[attr_name] = str(old_attrs[attr_name])

# add '.' to initials if initals are not None and not "" and if the initials
        # don't have already a '.' at the end
        if attrs["initials"] and not attrs["initials"].endswith('.'):
            attrs["initials"] += '.'

# remove empty values (None and '')
        attrs_values = list(filter(None, attrs.values()))

# fullname is the combination of not-empty values as string, separated by ' '
        fullname = ' '.join(attrs_values)

if fullname == '':
            return fallback_default

return fullname

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, uidnumber=None, gidnumber=None, gecos=None,
                loginshell=None, uid=None, nisdomain=None, unixhome=None,
                smartcard_required=False):
        """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 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 initials: Initials of the new user
        :param profilepath: Profile path of the new user
        :param scriptpath: Logon script path of the new user
        :param homedrive: Home drive of the new user
        :param homedirectory: Home directory of the new user
        :param jobtitle: Job title of the new user
        :param department: Department of the new user
        :param company: Company of the new user
        :param description: of the new user
        :param mailaddress: Email address 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
        :param uidnumber: RFC2307 Unix numeric UID of the new user
        :param gidnumber: RFC2307 Unix primary GID of the new user
        :param gecos: RFC2307 Unix GECOS field of the new user
        :param loginshell: RFC2307 Unix login shell of the new user
        :param uid: RFC2307 Unix username of the new user
        :param nisdomain: RFC2307 Unix NIS domain of the new user
        :param unixhome: RFC2307 Unix home directory of the new user
        :param smartcard_required: set the UF_SMARTCARD_REQUIRED bit of the new user
        """

displayname = self.fullname_from_names(given_name=givenname,
                                               initials=initials,
                                               surname=surname)
        cn = username
        if useusernameascn is None and displayname != "":
            cn = displayname

if userou:
            user_dn = "CN=%s,%s,%s" % (cn, userou, self.domain_dn())
        else:
            user_dn = "CN=%s,%s" % (cn, self.get_wellknown_dn(
                                        self.get_default_basedn(),
                                        dsdb.DS_GUID_USERS_CONTAINER))

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 information
        ldbmessage = {"dn": user_dn,
                      "sAMAccountName": username,
                      "userPrincipalName": user_principal_name,
                      "objectClass": "user"}

if smartcard_required:
            ldbmessage["userAccountControl"] = str(dsdb.UF_NORMAL_ACCOUNT |
                                                   dsdb.UF_SMARTCARD_REQUIRED)
            setpassword = False

if surname is not None:
            ldbmessage["sn"] = surname

if givenname is not None:
            ldbmessage["givenName"] = givenname

if displayname != "":
            ldbmessage["displayName"] = displayname
            ldbmessage["name"] = displayname

if initials is not None:
            ldbmessage["initials"] = '%s.' % initials

if profilepath is not None:
            ldbmessage["profilePath"] = profilepath

if scriptpath is not None:
            ldbmessage["scriptPath"] = scriptpath

if homedrive is not None:
            ldbmessage["homeDrive"] = homedrive

if homedirectory is not None:
            ldbmessage["homeDirectory"] = homedirectory

if jobtitle is not None:
            ldbmessage["title"] = jobtitle

if department is not None:
            ldbmessage["department"] = department

if company is not None:
            ldbmessage["company"] = company

if description is not None:
            ldbmessage["description"] = description

if mailaddress is not None:
            ldbmessage["mail"] = mailaddress

if internetaddress is not None:
            ldbmessage["wWWHomePage"] = internetaddress

if telephonenumber is not None:
            ldbmessage["telephoneNumber"] = telephonenumber

if physicaldeliveryoffice is not None:
            ldbmessage["physicalDeliveryOfficeName"] = physicaldeliveryoffice

if sd is not None:
            ldbmessage["nTSecurityDescriptor"] = ndr_pack(sd)

ldbmessage2 = None
        if any(map(lambda b: b is not None, (uid, uidnumber, gidnumber, gecos,
                                             loginshell, nisdomain, unixhome))):
            ldbmessage2 = ldb.Message()
            ldbmessage2.dn = ldb.Dn(self, user_dn)
            if uid is not None:
                ldbmessage2["uid"] = ldb.MessageElement(str(uid), ldb.FLAG_MOD_REPLACE, 'uid')
            if uidnumber is not None:
                ldbmessage2["uidNumber"] = ldb.MessageElement(str(uidnumber), ldb.FLAG_MOD_REPLACE, 'uidNumber')
            if gidnumber is not None:
                ldbmessage2["gidNumber"] = ldb.MessageElement(str(gidnumber), ldb.FLAG_MOD_REPLACE, 'gidNumber')
            if gecos is not None:
                ldbmessage2["gecos"] = ldb.MessageElement(str(gecos), ldb.FLAG_MOD_REPLACE, 'gecos')
            if loginshell is not None:
                ldbmessage2["loginShell"] = ldb.MessageElement(str(loginshell), ldb.FLAG_MOD_REPLACE, 'loginShell')
            if unixhome is not None:
                ldbmessage2["unixHomeDirectory"] = ldb.MessageElement(
                    str(unixhome), ldb.FLAG_MOD_REPLACE, 'unixHomeDirectory')
            if nisdomain is not None:
                ldbmessage2["msSFU30NisDomain"] = ldb.MessageElement(
                    str(nisdomain), ldb.FLAG_MOD_REPLACE, 'msSFU30NisDomain')
                ldbmessage2["msSFU30Name"] = ldb.MessageElement(
                    str(username), ldb.FLAG_MOD_REPLACE, 'msSFU30Name')
                ldbmessage2["unixUserPassword"] = ldb.MessageElement(
                    'ABCD!efgh12345$67890', ldb.FLAG_MOD_REPLACE,
                    'unixUserPassword')

self.transaction_start()
        try:
            self.add(ldbmessage)

with self._CleanUpOnError(self, user_dn):
                if ldbmessage2:
                    self.modify(ldbmessage2)

# Sets the password for it
                if setpassword:
                    self.setpassword(("(distinguishedName=%s)" %
                                      ldb.binary_encode(user_dn)),
                                     password,
                                     force_password_change_at_next_login_req)
        except:
            self.transaction_cancel()
            raise
        else:
            self.transaction_commit()

def newcontact(self,
                   fullcontactname=None,
                   ou=None,
                   surname=None,
                   givenname=None,
                   initials=None,
                   displayname=None,
                   jobtitle=None,
                   department=None,
                   company=None,
                   description=None,
                   mailaddress=None,
                   internetaddress=None,
                   telephonenumber=None,
                   mobilenumber=None,
                   physicaldeliveryoffice=None):
        """Adds a new contact with additional parameters

:param fullcontactname: Optional full name of the new contact
        :param ou: Object container for new contact
        :param surname: Surname of the new contact
        :param givenname: First name of the new contact
        :param initials: Initials of the new contact
        :param displayname: displayName of the new contact
        :param jobtitle: Job title of the new contact
        :param department: Department of the new contact
        :param company: Company of the new contact
        :param description: Description of the new contact
        :param mailaddress: Email address of the new contact
        :param internetaddress: Home page of the new contact
        :param telephonenumber: Phone number of the new contact
        :param mobilenumber: Primary mobile number of the new contact
        :param physicaldeliveryoffice: Office location of the new contact
        """

# Prepare the contact name like the RSAT, using the name parts.
        cn = self.fullname_from_names(given_name=givenname,
                                      initials=initials,
                                      surname=surname)

# Use the specified fullcontactname instead of the previously prepared
        # contact name, if it is specified.
        # This is similar to the "Full name" value of the RSAT.
        if fullcontactname is not None:
            cn = fullcontactname

if fullcontactname is None and cn == "":
            raise Exception('No name for contact specified')

contactcontainer_dn = self.domain_dn()
        if ou:
            contactcontainer_dn = self.normalize_dn_in_domain(ou)

contact_dn = "CN=%s,%s" % (cn, contactcontainer_dn)

ldbmessage = {"dn": contact_dn,
                      "objectClass": "contact",
                      }

if surname is not None:
            ldbmessage["sn"] = surname

if givenname is not None:
            ldbmessage["givenName"] = givenname

if displayname is not None:
            ldbmessage["displayName"] = displayname