#!/usr/bin/python # Unix SMB/CIFS implementation. # Copyright (C) Jelmer Vernooij 2007-2008 # # Based on the original in EJS: # Copyright (C) Andrew Tridgell 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 . # """Convenience functions for using the SAM.""" import samba import glue import ldb from samba.idmap import IDmapDB import pwd import time import base64 __docformat__ = "restructuredText" class SamDB(samba.Ldb): """The SAM database.""" def __init__(self, url=None, lp=None, modules_dir=None, session_info=None, credentials=None, flags=0, options=None): """Opens the Sam Database. For parameter meanings see the super class (samba.Ldb) """ self.lp = lp if url is 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) glue.dsdb_set_global_schema(self) def connect(self, url=None, flags=0, options=None): super(SamDB, self).connect(url=self.lp.private_path(url), flags=flags, options=options) def enable_account(self, user_dn): """Enable an account. :param user_dn: Dn of the account to enable. """ res = self.search(user_dn, ldb.SCOPE_BASE, None, ["userAccountControl"]) assert len(res) == 1 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, user_dn): """Force a password change at next login :param user_dn: Dn of the account to force password change on """ mod = """ dn: %s changetype: modify replace: pwdLastSet pwdLastSet: 0 """ % (user_dn) self.modify_ldif(mod) 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 newuser(self, username, unixname, password, force_password_change_at_next_login=False): """add a new user record. :param username: Name of the new user. :param unixname: Name of the unix user to map to. :param password: Password for the new user """ # connect to the sam self.transaction_start() try: user_dn = "CN=%s,CN=Users,%s" % (username, self.domain_dn()) # # the new user record. note the reliance on the samdb module to # fill in a sid, guid etc # # now the real work self.add({"dn": user_dn, "sAMAccountName": username, "userPassword": password, "objectClass": "user"}) res = self.search(user_dn, scope=ldb.SCOPE_BASE, expression="objectclass=*", attrs=["objectSid"]) assert len(res) == 1 user_sid = self.schema_format_value("objectSid", res[0]["objectSid"][0]) try: idmap = IDmapDB(lp=self.lp) user = pwd.getpwnam(unixname) # setup ID mapping for this UID idmap.setup_name_mapping(user_sid, idmap.TYPE_UID, user[2]) except KeyError: pass if force_password_change_at_next_login: self.force_password_change_at_next_login(user_dn) # modify the userAccountControl to remove the disabled bit self.enable_account(user_dn) except: self.transaction_cancel() raise self.transaction_commit() def setpassword(self, filter, password, force_password_change_at_next_login=False): """Set a password on a user record :param filter: LDAP filter to find the user (eg samccountname=name) :param password: Password for the user """ # connect to the sam self.transaction_start() try: res = self.search(base=self.domain_dn(), scope=ldb.SCOPE_SUBTREE, expression=filter, attrs=[]) assert(len(res) == 1) user_dn = res[0].dn mod = ldb.Message() mod.dn = user_dn glue.samdb_set_password(samdb=self, user_dn=str(user_dn), dom_dn=self.domain_dn(), mod=mod, new_password=password, user_change=False) self.modify(mod) if force_password_change_at_next_login: self.force_password_change_at_next_login(user_dn) # modify the userAccountControl to remove the disabled bit self.enable_account(user_dn) except: self.transaction_cancel() raise self.transaction_commit() def setexpiry(self, user, expiry_seconds, noexpiry): """Set the account expiry for a user :param expiry_seconds: expiry time from now in seconds :param noexpiry: if set, then don't expire password """ self.transaction_start() try: res = self.search(base=self.domain_dn(), scope=ldb.SCOPE_SUBTREE, expression=("(samAccountName=%s)" % user), attrs=["userAccountControl", "accountExpires"]) assert len(res) == 1 userAccountControl = int(res[0]["userAccountControl"][0]) accountExpires = int(res[0]["accountExpires"][0]) if noexpiry: userAccountControl = userAccountControl | 0x10000 accountExpires = 0 else: userAccountControl = userAccountControl & ~0x10000 accountExpires = glue.unix2nttime(expiry_seconds + int(time.time())) mod = """ dn: %s changetype: modify replace: userAccountControl userAccountControl: %u replace: accountExpires accountExpires: %u """ % (res[0].dn, userAccountControl, accountExpires) # now change the database self.modify_ldif(mod) except: self.transaction_cancel() raise self.transaction_commit();