path: root/source4/dsdb/tests/python/password_lockout.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# This tests the password lockout behavior for AD implementations
#
# Copyright Matthias Dieter Wallnoefer 2010
# Copyright Andrew Bartlett 2013
# Copyright Stefan Metzmacher 2014
#

import optparse
import sys
import base64
import time

sys.path.insert(0, "bin/python")
import samba

from samba.tests.subunitrun import TestProgram, SubunitOptions

import samba.getopt as options

from samba.auth import system_session
from samba.credentials import Credentials, DONT_USE_KERBEROS, MUST_USE_KERBEROS
from ldb import SCOPE_BASE, LdbError
from ldb import ERR_CONSTRAINT_VIOLATION
from ldb import ERR_INVALID_CREDENTIALS
from ldb import Message, MessageElement, Dn
from ldb import FLAG_MOD_REPLACE
from samba import gensec, dsdb
from samba.samdb import SamDB
import samba.tests
from samba.tests import delete_force
from samba.dcerpc import security, samr
from samba.ndr import ndr_unpack

parser = optparse.OptionParser("passwords.py [options] <host>")
sambaopts = options.SambaOptions(parser)
parser.add_option_group(sambaopts)
parser.add_option_group(options.VersionOptions(parser))
# use command line creds if available
credopts = options.CredentialsOptions(parser)
parser.add_option_group(credopts)
subunitopts = SubunitOptions(parser)
parser.add_option_group(subunitopts)
opts, args = parser.parse_args()

if len(args) < 1:
    parser.print_usage()
    sys.exit(1)

host = args[0]

lp = sambaopts.get_loadparm()
creds = credopts.get_credentials(lp)

# Force an encrypted connection
creds.set_gensec_features(creds.get_gensec_features() | gensec.FEATURE_SEAL)

#
# Tests start here
#

class PasswordTests(samba.tests.TestCase):

    def _open_samr_user(self, res):
        self.assertTrue("objectSid" in res[0])

        (domain_sid, rid) = ndr_unpack(security.dom_sid, res[0]["objectSid"][0]).split()
        self.assertEquals(self.domain_sid, domain_sid)

        return self.samr.OpenUser(self.samr_domain, security.SEC_FLAG_MAXIMUM_ALLOWED, rid)

    def _reset_samr(self, res):

        # Now reset the lockout, by removing ACB_AUTOLOCK (which removes the lock, despite being a generated attribute)
        samr_user = self._open_samr_user(res)
        acb_info = self.samr.QueryUserInfo(samr_user, 16)
        acb_info.acct_flags &= ~samr.ACB_AUTOLOCK
        self.samr.SetUserInfo(samr_user, 16, acb_info)
        self.samr.Close(samr_user)

    def _reset_ldap_lockoutTime(self, res):
        self.ldb.modify_ldif("""
dn: """ + str(res[0].dn) + """
changetype: modify
replace: lockoutTime
lockoutTime: 0
""")

    def _reset_ldap_userAccountControl(self, res):
        self.assertTrue("userAccountControl" in res[0])
        self.assertTrue("msDS-User-Account-Control-Computed" in res[0])

        uac = int(res[0]["userAccountControl"][0])
        uacc = int(res[0]["msDS-User-Account-Control-Computed"][0])

        uac |= uacc
        uac = uac & ~dsdb.UF_LOCKOUT

        self.ldb.modify_ldif("""
dn: """ + str(res[0].dn) + """
changetype: modify
replace: userAccountControl
userAccountControl: %d
""" % uac)

    def _reset_by_method(self, res, method):
        if method is "ldap_userAccountControl":
            self._reset_ldap_userAccountControl(res)
        elif method is "ldap_lockoutTime":
            self._reset_ldap_lockoutTime(res)
        elif method is "samr":
            self._reset_samr(res)
        else:
            self.assertTrue(False, msg="Invalid reset method[%s]" % method)

    def _check_attribute(self, res, name, value):
        if value is None:
            self.assertTrue(name not in res[0],
                            msg="attr[%s]=%r on dn[%s]" %
                            (name, res[0], res[0].dn))
            return

        if isinstance(value, tuple):
            (mode, value) = value
        else:
            mode = "equal"

        if mode == "ignore":
            return

        self.assertTrue(name in res[0],
                        msg="attr[%s] missing on dn[%s]" %
                        (name, res[0].dn))
        self.assertTrue(len(res[0][name]) == 1,
                        msg="attr[%s]=%r on dn[%s]" %
                        (name, res[0][name], res[0].dn))

        if mode == "present":
            return
        if mode == "equal":
            self.assertTrue(str(res[0][name][0]) == str(value),
                            msg="attr[%s]=[%s] != [%s] on dn[%s]" %
                            (name, str(res[0][name][0]), str(value), res[0].dn))
            return
        if mode == "greater":
            v = int(res[0][name][0])
            self.assertTrue(v > int(value),
                            msg="attr[%s]=[%s] <= [%s] on dn[%s]" %
                            (name, v, int(value), res[0].dn))
            return
        if mode == "less":
            v = int(res[0][name][0])
            self.assertTrue(v < int(value),
                            msg="attr[%s]=[%s] >= [%s] on dn[%s]" %
                            (name, v, int(value), res[0].dn))
            return
        self.assertEqual(mode, not mode, "Invalid Mode[%s]" % mode)

    def _check_account(self, dn,
                       badPwdCount=None,
                       badPasswordTime=None,
                       lockoutTime=None,
                       userAccountControl=None,
                       msDSUserAccountControlComputed=None,
                       effective_bad_password_count=None):

        attrs = [
           "objectSid",
           "badPwdCount",
           "badPasswordTime",
           "lockoutTime",
           "userAccountControl",
           "msDS-User-Account-Control-Computed"
        ]

        # in order to prevent some time resolution problems we sleep for
        # 10 micro second
        time.sleep(0.01)

        res = self.ldb.search(dn, scope=SCOPE_BASE, attrs=attrs)
        self.assertTrue(len(res) == 1)
        self._check_attribute(res, "badPwdCount", badPwdCount)
        self._check_attribute(res, "badPasswordTime", badPasswordTime)
        self._check_attribute(res, "lockoutTime", lockoutTime)
        self._check_attribute(res, "userAccountControl", userAccountControl)
        self._check_attribute(res, "msDS-User-Account-Control-Computed",
                              msDSUserAccountControlComputed)

        samr_user = self._open_samr_user(res)
        uinfo3 = self.samr.QueryUserInfo(samr_user, 3)
        uinfo5 = self.samr.QueryUserInfo(samr_user, 5)
        uinfo16 = self.samr.QueryUserInfo(samr_user, 16)
        uinfo21 = self.samr.QueryUserInfo(samr_user, 21)
        self.samr.Close(samr_user)

        expected_acb_info = 0
        if userAccountControl & dsdb.UF_NORMAL_ACCOUNT:
            expected_acb_info |= samr.ACB_NORMAL
        if userAccountControl & dsdb.UF_ACCOUNTDISABLE:
            expected_acb_info |= samr.ACB_DISABLED
        if userAccountControl & dsdb.UF_PASSWD_NOTREQD:
            expected_acb_info |= samr.ACB_PWNOTREQ
        if msDSUserAccountControlComputed & dsdb.UF_LOCKOUT:
            expected_acb_info |= samr.ACB_AUTOLOCK
        if msDSUserAccountControlComputed & dsdb.UF_PASSWORD_EXPIRED:
            expected_acb_info |= samr.ACB_PW_EXPIRED

        expected_bad_password_count = 0
        if badPwdCount is not None:
            expected_bad_password_count = badPwdCount
        if effective_bad_password_count is None:
            effective_bad_password_count = expected_bad_password_count

        self.assertEquals(uinfo3.acct_flags, expected_acb_info)
        self.assertEquals(uinfo3.bad_password_count, expected_bad_password_count)

        self.assertEquals(uinfo5.acct_flags, expected_acb_info)
        self.assertEquals(uinfo5.bad_password_count, effective_bad_password_count)

        self.assertEquals(uinfo16.acct_flags, expected_acb_info)

        self.assertEquals(uinfo21.acct_flags, expected_acb_info)
        self.assertEquals(uinfo21.bad_password_count, effective_bad_passwo