path: root/source4/scripting/python/samba/samdb.py

#!/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
#   
# 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
from samba.ndr import ndr_unpack, ndr_pack
from samba.dcerpc import drsblobs, misc

__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=None):
        self.lp = lp
        if not auto_connect:
            url = None
        elif url is None and lp is not None:
            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)

        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):
        if self.lp is not None:
            url = self.lp.private_path(url)

        super(SamDB, self).connect(url=url, flags=flags,
                options=options)

    def am_rodc(self):
        return dsdb._am_rodc(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):
        """Enables an account
        
        :param filter: LDAP filter to find the user (eg samccountname=name)
        """
        res = self.search(base=self.domain_dn(), scope=ldb.SCOPE_SUBTREE,
                          expression=filter, attrs=["userAccountControl"])
        assert(len(res) == 1)
        user_dn = res[0].dn

        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

        mod = """
dn: %s
changetype: modify
replace: userAccountControl
userAccountControl: %u
""" % (user_dn, userAccountControl)
        self.modify_ldif(mod)
        
    def force_password_change_at_next_login(self, filter):
        """Forces a password change at next login
        
        :param filter: LDAP filter to find the user (eg samccountname=name)
        """
        res = self.search(base=self.domain_dn(), scope=ldb.SCOPE_SUBTREE,
                          expression=filter, attrs=[])
        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 newgroup(self, groupname, groupou=None, grouptype=None,
                 description=None, mailaddress=None, notes=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
        """

        group_dn = "CN=%s,%s,%s" % (groupname, (groupou or "CN=Users"), self.domain_dn())

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

        if grouptype is not None:
            ldbmessage["groupType"] = "%d" % 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

        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())
        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 add_remove_group_members(self, groupname, listofmembers,
                                  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
        """

        groupfilter = "(&(sAMAccountName=%s)(objectCategory=%s,%s))" % (groupname, "CN=Group,CN=Schema,CN=Configuration", self.domain_dn())
        groupmembers = listofmembers.split(',')

        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 groupmembers:
                targetmember = self.search(base=self.domain_dn(), scope=ldb.SCOPE_SUBTREE,
                                    expression="(|(sAMAccountName=%s)(CN=%s))" % (member, member), attrs=[])

                if len(targetmember) != 1:
                    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
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
member: %s
""" % (str(targetmember[0].dn))

            if modified is True:
                self.modify_ldif(addtargettogroup)

        except:
            self.transaction_cancel()
            raise
        else:
            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):
        """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
        """

        displayname = ""
        if givenname is not None:
            displayname += givenname

        if initials is not None:
            displayname += ' %s.' % initials

        if surname is not None:
            displayname += ' %s' % surname

        cn = username
        if useusernameascn is None and displayname is not "":
            cn = displayname

        user_dn = "CN=%s,%s,%s" % (cn, (userou or "CN=Users"), self.domain_dn())

        # 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"}

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

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

        if displayname is not "":
            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

        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)
        except:
            self.transaction_cancel()
            raise
        else:
            self.transaction_commit()

    def setpassword(self, filter, password,
                    force_change_at_next_login=False,
                    username=None):
        """Sets the password for a user
        
        :param 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=[])
            if len(res) == 0:
                raise Exception('Unable to find user "%s"' % (username or filter))
            assert(len(res) == 1)
            user_dn = res[0].dn

            setpw = """
dn: %s
changetype: modify
replace: unicodePwd
unicodePwd:: %s
""" % (user_dn, base64.b64encode(("\"" + password + "\"").encode('utf-16-le')))

            self.modify_ldif(setpw)

            if force_change_at_next_login:
                self.force_password_change_at_next_login(
                  "(dn=" + str(user_dn) + ")")

            #  modify the userAccountControl to remove the disabled bit
            self.enable_account(filter)
        except:
            self.transaction_cancel()
            raise
        else:
            self.transaction_commit()

    def setexpiry(self, 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 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,
                          attrs=["userAccountControl", "accountExpires"])
            assert(len(res) == 1)
            user_dn = res[0].dn

            userAccountControl = int(res[0]["userAccountControl"][0])
            accountExpires     = int(res[0]["accountExpires"][0])
            if no_expiry_req:
                userAccountControl = userAccountControl | 0x10000
                accountExpires = 0
            else:
                userAccountControl = userAccountControl & ~0x10000
                accountExpires = samba.unix2nttime(expiry_seconds + int(time.time()))

            setexp = """
dn: %s
changetype: modify
replace: userAccountControl
userAccountControl: %u
replace: accountExpires
accountExpires: %u
""" % (user_dn, userAccountControl, accountExpires)

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

    def set_domain_sid(self, sid):
        """Change the domain SID used by this LDB.

        :param sid: The new domain sid to use.
        """
        dsdb._samdb_set_domain_sid(self, sid)

    def get_domain_sid(self):
        """Read the domain SID used by this LDB.

        """
        return dsdb._samdb_get_domain_sid(self)

    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)

    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_invocation_id(self):
        "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

        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)

    def get_ntds_GUID(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)

    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_from_ldb(self, ldb):
        dsdb._dsdb_set_schema_from_ldb(self, ldb)

    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="dn=%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="dn=%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)

    def get_partitions_dn(self):
        return dsdb._dsdb_get_partitions_dn(self)