Gecko [ urlab.iiitd.edu.in ]

/ usr / lib / python3 / dist-packages / samba / tests / [ HOME SHELL ]

Name	Size	Permission
__pycache__	[ DIR ]	drwxr-xr-x
blackbox	[ DIR ]	drwxr-xr-x
dcerpc	[ DIR ]	drwxr-xr-x
dns_forwarder_helpers	[ DIR ]	drwxr-xr-x
emulate	[ DIR ]	drwxr-xr-x
kcc	[ DIR ]	drwxr-xr-x
krb5	[ DIR ]	drwxr-xr-x
samba_tool	[ DIR ]	drwxr-xr-x
__init__.py	22.9 KB	-rw-r--r--
audit_log_base.py	6.98 KB	-rw-r--r--
audit_log_dsdb.py	24.46 KB	-rw-r--r--
audit_log_pass_change.py	12.86 KB	-rw-r--r--
auth.py	4.5 KB	-rw-r--r--
auth_log.py	65.62 KB	-rw-r--r--
auth_log_base.py	5.42 KB	-rw-r--r--
auth_log_ncalrpc.py	4.19 KB	-rw-r--r--
auth_log_netlogon.py	5.22 KB	-rw-r--r--
auth_log_netlogon_bad_creds.py	7.7 KB	-rw-r--r--
auth_log_pass_change.py	12.92 KB	-rw-r--r--
auth_log_samlogon.py	6.68 KB	-rw-r--r--
auth_log_winbind.py	19.13 KB	-rw-r--r--
common.py	3.06 KB	-rw-r--r--
complex_expressions.py	18.37 KB	-rw-r--r--
core.py	2.77 KB	-rw-r--r--
cred_opt.py	1.73 KB	-rw-r--r--
credentials.py	22.13 KB	-rw-r--r--
dckeytab.py	2.11 KB	-rw-r--r--
dns.py	84.34 KB	-rw-r--r--
dns_aging.py	108.76 KB	-rw-r--r--
dns_base.py	14.3 KB	-rw-r--r--
dns_forwarder.py	20.95 KB	-rw-r--r--
dns_invalid.py	2.66 KB	-rw-r--r--
dns_packet.py	7.13 KB	-rw-r--r--
dns_tkey.py	7.11 KB	-rw-r--r--
dns_wildcard.py	12.89 KB	-rw-r--r--
docs.py	18.25 KB	-rw-r--r--
domain_backup.py	26.76 KB	-rw-r--r--
domain_backup_offline.py	10.01 KB	-rw-r--r--
dsdb.py	41.91 KB	-rw-r--r--
dsdb_api.py	1.93 KB	-rw-r--r--
dsdb_dns.py	2.56 KB	-rw-r--r--
dsdb_lock.py	11.63 KB	-rw-r--r--
dsdb_schema_attributes.py	8.9 KB	-rw-r--r--
encrypted_secrets.py	3.02 KB	-rw-r--r--
gensec.py	8.91 KB	-rw-r--r--
get_opt.py	2.47 KB	-rw-r--r--
getdcname.py	18.52 KB	-rw-r--r--
glue.py	3.75 KB	-rw-r--r--
gpo.py	80.28 KB	-rw-r--r--
gpo_member.py	1.69 KB	-rw-r--r--
graph.py	26.45 KB	-rw-r--r--
group_audit.py	16.52 KB	-rw-r--r--
hostconfig.py	2.14 KB	-rw-r--r--
imports.py	1.12 KB	-rw-r--r--
join.py	6.71 KB	-rw-r--r--
krb5_credentials.py	3.8 KB	-rw-r--r--
ldap_raw.py	31.92 KB	-rw-r--r--
ldap_referrals.py	3.13 KB	-rw-r--r--
ldap_spn.py	34.19 KB	-rw-r--r--
ldap_upn_sam_account.py	17.63 KB	-rw-r--r--
libsmb.py	4.26 KB	-rw-r--r--
loadparm.py	3.25 KB	-rw-r--r--
lsa_string.py	2.22 KB	-rw-r--r--
messaging.py	4.97 KB	-rw-r--r--
ndr.py	4.09 KB	-rw-r--r--
net_join.py	2.29 KB	-rw-r--r--
net_join_no_spnego.py	3.46 KB	-rw-r--r--
netbios.py	2.08 KB	-rw-r--r--
netcmd.py	5.32 KB	-rw-r--r--
netlogonsvc.py	2.43 KB	-rw-r--r--
ntacls.py	3.68 KB	-rw-r--r--
ntacls_backup.py	6.58 KB	-rw-r--r--
ntlm_auth.py	13.08 KB	-rw-r--r--
ntlm_auth_base.py	8.5 KB	-rw-r--r--
ntlm_auth_krb5.py	3.63 KB	-rw-r--r--
ntlmdisabled.py	3.01 KB	-rw-r--r--
pam_winbind.py	2.52 KB	-rw-r--r--
pam_winbind_chauthtok.py	1.56 KB	-rw-r--r--
pam_winbind_warn_pwd_expire.py	2.01 KB	-rw-r--r--
param.py	3.58 KB	-rw-r--r--
password_hash.py	12.43 KB	-rw-r--r--
password_hash_fl2003.py	7.34 KB	-rw-r--r--
password_hash_fl2008.py	7.89 KB	-rw-r--r--
password_hash_gpgme.py	11.88 KB	-rw-r--r--
password_hash_ldap.py	4.85 KB	-rw-r--r--
password_quality.py	2.15 KB	-rw-r--r--
password_test.py	1.94 KB	-rw-r--r--
policy.py	1.17 KB	-rw-r--r--
posixacl.py	38.89 KB	-rw-r--r--
prefork_restart.py	16 KB	-rw-r--r--
process_limits.py	2.95 KB	-rw-r--r--
provision.py	6.35 KB	-rw-r--r--
pso.py	11.23 KB	-rw-r--r--
py_credentials.py	14.82 KB	-rw-r--r--
registry.py	2.5 KB	-rw-r--r--
s3_net_join.py	2.63 KB	-rw-r--r--
s3idmapdb.py	1.95 KB	-rw-r--r--
s3param.py	1.75 KB	-rw-r--r--
s3passdb.py	5.66 KB	-rw-r--r--
s3registry.py	1.82 KB	-rw-r--r--
s3windb.py	1.48 KB	-rw-r--r--
samba3sam.py	48.45 KB	-rw-r--r--
samba_upgradedns_lmdb.py	2.56 KB	-rw-r--r--
samdb.py	2.51 KB	-rw-r--r--
samdb_api.py	5.41 KB	-rw-r--r--
sddl.py	8.86 KB	-rw-r--r--
security.py	6.25 KB	-rw-r--r--
segfault.py	7 KB	-rw-r--r--
smb-notify.py	18.85 KB	-rw-r--r--
smb.py	9.47 KB	-rw-r--r--
smbd_base.py	1.76 KB	-rw-r--r--
smbd_fuzztest.py	3.5 KB	-rw-r--r--
source.py	8.74 KB	-rw-r--r--
strings.py	3.89 KB	-rw-r--r--
subunitrun.py	2.33 KB	-rw-r--r--
tdb_util.py	1.76 KB	-rw-r--r--
upgrade.py	1.36 KB	-rw-r--r--
upgradeprovision.py	6.65 KB	-rw-r--r--
upgradeprovisionneeddc.py	7.49 KB	-rw-r--r--
usage.py	12.98 KB	-rw-r--r--
xattr.py	5.27 KB	-rw-r--r--

Code Editor : auth_log.py

# Unix SMB/CIFS implementation.
# Copyright (C) Andrew Bartlett <abartlet@samba.org> 2017
#
# 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/>.
#

"""Tests for the Auth and AuthZ logging.
"""
import samba.tests
from samba.dcerpc import srvsvc, dnsserver
import os
from samba.samba3 import libsmb_samba_internal as libsmb
from samba.samba3 import param as s3param
from samba.samdb import SamDB
import samba.tests.auth_log_base
from samba.credentials import DONT_USE_KERBEROS, MUST_USE_KERBEROS
from samba import NTSTATUSError
from subprocess import call
from ldb import LdbError
from samba.dcerpc.windows_event_ids import (
    EVT_ID_SUCCESSFUL_LOGON,
    EVT_ID_UNSUCCESSFUL_LOGON,
    EVT_LOGON_NETWORK,
    EVT_LOGON_INTERACTIVE,
    EVT_LOGON_NETWORK_CLEAR_TEXT
)
import re

class AuthLogTests(samba.tests.auth_log_base.AuthLogTestBase):

def setUp(self):
        super(AuthLogTests, self).setUp()
        self.remoteAddress = os.environ["CLIENT_IP"]

def tearDown(self):
        super(AuthLogTests, self).tearDown()

def smb_connection(self, creds, use_spnego="yes", ntlmv2_auth="yes",
                       force_smb1=False):
        # the SMB bindings rely on having a s3 loadparm
        lp = self.get_loadparm()
        s3_lp = s3param.get_context()
        s3_lp.load(lp.configfile)

# Allow the testcase to skip SPNEGO or use NTLMv1
        s3_lp.set("client use spnego", use_spnego)
        s3_lp.set("client ntlmv2 auth", ntlmv2_auth)

return libsmb.Conn(self.server, "sysvol", lp=s3_lp, creds=creds,
                           force_smb1=force_smb1)

def _test_rpc_ncacn_np(self, authTypes, creds, service,
                           binding, protection, checkFunction):
        def isLastExpectedMessage(msg):
            return (msg["type"] == "Authorization" and
                    (msg["Authorization"]["serviceDescription"] == "DCE/RPC" or
                     msg["Authorization"]["serviceDescription"] == service) and
                    msg["Authorization"]["authType"] == authTypes[0] and
                    msg["Authorization"]["transportProtection"] == protection)

if binding:
            binding = "[%s]" % binding

if service == "dnsserver":
            x = dnsserver.dnsserver("ncacn_np:%s%s" % (self.server, binding),
                                    self.get_loadparm(),
                                    creds)
        elif service == "srvsvc":
            x = srvsvc.srvsvc("ncacn_np:%s%s" % (self.server, binding),
                              self.get_loadparm(),
                              creds)

# The connection is passed to ensure the server
        # messaging context stays up until all the messages have been received.
        messages = self.waitForMessages(isLastExpectedMessage, x)
        checkFunction(messages, authTypes, service, binding, protection)

def _assert_ncacn_np_serviceDescription(self, binding, serviceDescription):
        # Turn "[foo,bar]" into a list ("foo", "bar") to test
        # lambda x: x removes anything that evaluates to False,
        # including empty strings, so we handle "" as well
        binding_list = \
            list(filter(lambda x: x, re.compile('[\[,\]]').split(binding)))

# Handle explicit smb2, smb1 or auto negotiation
        if "smb2" in binding_list:
            self.assertEqual(serviceDescription, "SMB2")
        elif "smb1" in binding_list:
            self.assertEqual(serviceDescription, "SMB")
        else:
            self.assertIn(serviceDescription, ["SMB", "SMB2"])

def rpc_ncacn_np_ntlm_check(self, messages, authTypes, service,
                                binding, protection):

expected_messages = len(authTypes)
        self.assertEqual(expected_messages,
                          len(messages),
                          "Did not receive the expected number of messages")

# Check the first message it should be an Authentication
        msg = messages[0]
        self.assertEqual("Authentication", msg["type"])
        self.assertEqual("NT_STATUS_OK", msg["Authentication"]["status"])
        self.assertEqual(
            EVT_ID_SUCCESSFUL_LOGON, msg["Authentication"]["eventId"])
        self.assertEqual(
            EVT_LOGON_NETWORK, msg["Authentication"]["logonType"])
        self._assert_ncacn_np_serviceDescription(
            binding, msg["Authentication"]["serviceDescription"])
        self.assertEqual(authTypes[1],
                          msg["Authentication"]["authDescription"])

# Check the second message it should be an Authorization
        msg = messages[1]
        self.assertEqual("Authorization", msg["type"])
        self._assert_ncacn_np_serviceDescription(
            binding, msg["Authorization"]["serviceDescription"])
        self.assertEqual(authTypes[2], msg["Authorization"]["authType"])
        self.assertEqual("SMB", msg["Authorization"]["transportProtection"])
        self.assertTrue(self.is_guid(msg["Authorization"]["sessionId"]))

# Check the third message it should be an Authentication
        # if we are expecting 4 messages
        if expected_messages == 4:
            def checkServiceDescription(desc):
                return (desc == "DCE/RPC" or desc == service)

msg = messages[2]
            self.assertEqual("Authentication", msg["type"])
            self.assertEqual("NT_STATUS_OK", msg["Authentication"]["status"])
            self.assertTrue(
                checkServiceDescription(
                    msg["Authentication"]["serviceDescription"]))

self.assertEqual(authTypes[3],
                              msg["Authentication"]["authDescription"])
            self.assertEqual(
                EVT_ID_SUCCESSFUL_LOGON, msg["Authentication"]["eventId"])
            self.assertEqual(
                EVT_LOGON_NETWORK, msg["Authentication"]["logonType"])

def rpc_ncacn_np_krb5_check(
            self,
            messages,
            authTypes,
            service,
            binding,
            protection):

expected_messages = len(authTypes)
        self.assertEqual(expected_messages,
                          len(messages),
                          "Did not receive the expected number of messages")

# Check the first message it should be an Authentication
        # This is almost certainly Authentication over UDP, and is probably
        # returning message too big,
        msg = messages[0]
        self.assertEqual("Authentication", msg["type"])
        self.assertEqual("NT_STATUS_OK", msg["Authentication"]["status"])
        self.assertEqual("Kerberos KDC",
                          msg["Authentication"]["serviceDescription"])
        self.assertEqual(authTypes[1],
                          msg["Authentication"]["authDescription"])
        self.assertEqual(
            EVT_ID_SUCCESSFUL_LOGON, msg["Authentication"]["eventId"])
        self.assertEqual(
            EVT_LOGON_NETWORK, msg["Authentication"]["logonType"])

# Check the second message it should be an Authentication
        # This this the TCP Authentication in response to the message too big
        # response to the UDP Authentication
        msg = messages[1]
        self.assertEqual("Authentication", msg["type"])
        self.assertEqual("NT_STATUS_OK", msg["Authentication"]["status"])
        self.assertEqual("Kerberos KDC",
                          msg["Authentication"]["serviceDescription"])
        self.assertEqual(authTypes[2],
                          msg["Authentication"]["authDescription"])
        self.assertEqual(
            EVT_ID_SUCCESSFUL_LOGON, msg["Authentication"]["eventId"])
        self.assertEqual(
            EVT_LOGON_NETWORK, msg["Authentication"]["logonType"])

# Check the third message it should be an Authorization
        msg = messages[2]
        self.assertEqual("Authorization", msg["type"])
        self._assert_ncacn_np_serviceDescription(
            binding, msg["Authorization"]["serviceDescription"])
        self.assertEqual(authTypes[3], msg["Authorization"]["authType"])
        self.assertEqual("SMB", msg["Authorization"]["transportProtection"])
        self.assertTrue(self.is_guid(msg["Authorization"]["sessionId"]))

def test_rpc_ncacn_np_ntlm_dns_sign(self):
        creds = self.insta_creds(template=self.get_credentials(),
                                 kerberos_state=DONT_USE_KERBEROS)
        self._test_rpc_ncacn_np(["NTLMSSP",
                                 "NTLMSSP",
                                 "NTLMSSP",
                                 "NTLMSSP"],
                                creds, "dnsserver", "sign", "SIGN",
                                self.rpc_ncacn_np_ntlm_check)

def test_rpc_ncacn_np_ntlm_srv_sign(self):
        creds = self.insta_creds(template=self.get_credentials(),
                                 kerberos_state=DONT_USE_KERBEROS)
        self._test_rpc_ncacn_np(["NTLMSSP",
                                 "NTLMSSP",
                                 "NTLMSSP",
                                 "NTLMSSP"],
                                creds, "srvsvc", "sign", "SIGN",
                                self.rpc_ncacn_np_ntlm_check)

def test_rpc_ncacn_np_ntlm_dns(self):
        creds = self.insta_creds(template=self.get_credentials(),
                                 kerberos_state=DONT_USE_KERBEROS)
        self._test_rpc_ncacn_np(["ncacn_np",
                                 "NTLMSSP",
                                 "NTLMSSP"],
                                creds, "dnsserver", "", "SMB",
                                self.rpc_ncacn_np_ntlm_check)

def test_rpc_ncacn_np_ntlm_srv(self):
        creds = self.insta_creds(template=self.get_credentials(),
                                 kerberos_state=DONT_USE_KERBEROS)
        self._test_rpc_ncacn_np(["ncacn_np",
                                 "NTLMSSP",
                                 "NTLMSSP"],
                                creds, "srvsvc", "", "SMB",
                                self.rpc_ncacn_np_ntlm_check)

def test_rpc_ncacn_np_krb_dns_sign(self):
        creds = self.insta_creds(template=self.get_credentials(),
                                 kerberos_state=MUST_USE_KERBEROS)
        self._test_rpc_ncacn_np(["krb5",
                                 "ENC-TS Pre-authentication",
                                 "ENC-TS Pre-authentication",
                                 "krb5"],
                                creds, "dnsserver", "sign", "SIGN",
                                self.rpc_ncacn_np_krb5_check)

def test_rpc_ncacn_np_krb_srv_sign(self):
        creds = self.insta_creds(template=self.get_credentials(),
                                 kerberos_state=MUST_USE_KERBEROS)
        self._test_rpc_ncacn_np(["krb5",
                                 "ENC-TS Pre-authentication",
                                 "ENC-TS Pre-authentication",
                                 "krb5"],
                                creds, "srvsvc", "sign", "SIGN",
                                self.rpc_ncacn_np_krb5_check)

def test_rpc_ncacn_np_krb_dns(self):
        creds = self.insta_creds(template=self.get_credentials(),
                                 kerberos_state=MUST_USE_KERBEROS)
        self._test_rpc_ncacn_np(["ncacn_np",
                                 "ENC-TS Pre-authentication",
                                 "ENC-TS Pre-authentication",
                                 "krb5"],
                                creds, "dnsserver", "", "SMB",
                                self.rpc_ncacn_np_krb5_check)

def test_rpc_ncacn_np_krb_dns_smb2(self):
        creds = self.insta_creds(template=self.get_credentials(),
                                 kerberos_state=MUST_USE_KERBEROS)
        self._test_rpc_ncacn_np(["ncacn_np",
                                 "ENC-TS Pre-authentication",
                                 "ENC-TS Pre-authentication",
                                 "krb5"],
                                creds, "dnsserver", "smb2", "SMB",
                                self.rpc_ncacn_np_krb5_check)

def test_rpc_ncacn_np_krb_srv(self):
        creds = self.insta_creds(template=self.get_credentials(),
                                 kerberos_state=MUST_USE_KERBEROS)
        self._test_rpc_ncacn_np(["ncacn_np",
                                 "ENC-TS Pre-authentication",
                                 "ENC-TS Pre-authentication",
                                 "krb5"],
                                creds, "srvsvc", "", "SMB",
                                self.rpc_ncacn_np_krb5_check)

def _test_rpc_ncacn_ip_tcp(self, authTypes, creds, service,
                               binding, protection, checkFunction):
        def isLastExpectedMessage(msg):
            return (msg["type"] == "Authorization" and
                    msg["Authorization"]["serviceDescription"] == "DCE/RPC" and
                    msg["Authorization"]["authType"] == authTypes[0] and
                    msg["Authorization"]["transportProtection"] == protection)

if binding:
            binding = "[%s]" % binding

if service == "dnsserver":
            conn = dnsserver.dnsserver(
                "ncacn_ip_tcp:%s%s" % (self.server, binding),
                self.get_loadparm(),
                creds)
        elif service == "srvsvc":
            conn = srvsvc.srvsvc("ncacn_ip_tcp:%s%s" % (self.server, binding),
                                 self.get_loadparm(),
                                 creds)

messages = self.waitForMessages(isLastExpectedMessage, conn)
        checkFunction(messages, authTypes, service, binding, protection)

def rpc_ncacn_ip_tcp_ntlm_check(self, messages, authTypes, service,
                                    binding, protection):

expected_messages = len(authTypes)
        self.assertEqual(expected_messages,
                          len(messages),
                          "Did not receive the expected number of messages")

# Check the first message it should be an Authorization
        msg = messages[0]
        self.assertEqual("Authorization", msg["type"])
        self.assertEqual("DCE/RPC",
                          msg["Authorization"]["serviceDescription"])
        self.assertEqual(authTypes[1], msg["Authorization"]["authType"])
        self.assertEqual("NONE", msg["Authorization"]["transportProtection"])
        self.assertTrue(self.is_guid(msg["Authorization"]["sessionId"]))

# Check the second message it should be an Authentication
        msg = messages[1]
        self.assertEqual("Authentication", msg["type"])
        self.assertEqual("NT_STATUS_OK", msg["Authentication"]["status"])
        self.assertEqual("DCE/RPC",
                          msg["Authentication"]["serviceDescription"])
        self.assertEqual(authTypes[2],
                          msg["Authentication"]["authDescription"])
        self.assertEqual(
            EVT_ID_SUCCESSFUL_LOGON, msg["Authentication"]["eventId"])
        self.assertEqual(
            EVT_LOGON_NETWORK, msg["Authentication"]["logonType"])

def rpc_ncacn_ip_tcp_krb5_check(self, messages, authTypes, service,
                                    binding, protection):

expected_messages = len(authTypes)
        self.assertEqual(expected_messages,
                          len(messages),
                          "Did not receive the expected number of messages")

# Check the second message it should be an Authentication
        msg = messages[1]
        self.assertEqual("Authentication", msg["type"])
        self.assertEqual("NT_STATUS_OK", msg["Authentication"]["status"])
        self.assertEqual("Kerberos KDC",
                          msg["Authentication"]["serviceDescription"])
        self.assertEqual(authTypes[2],
                          msg["Authentication"]["authDescription"])
        self.assertEqual(
            EVT_ID_SUCCESSFUL_LOGON, msg["Authentication"]["eventId"])
        self.assertEqual(
            EVT_LOGON_NETWORK, msg["Authentication"]["logonType"])

# Check the third message it should be an Authentication
        msg = messages[2]
        self.assertEqual("Authentication", msg["type"])
        self.assertEqual("NT_STATUS_OK", msg["Authentication"]["status"])
        self.assertEqual("Kerberos KDC",
                          msg["Authentication"]["serviceDescription"])
        self.assertEqual(authTypes[2],
                          msg["Authentication"]["authDescription"])
        self.assertEqual(
            EVT_ID_SUCCESSFUL_LOGON, msg["Authentication"]["eventId"])
        self.assertEqual(
            EVT_LOGON_NETWORK, msg["Authentication"]["logonType"])

def test_rpc_ncacn_ip_tcp_ntlm_dns_sign(self):
        creds = self.insta_creds(template=self.get_credentials(),
                                 kerberos_state=DONT_USE_KERBEROS)
        self._test_rpc_ncacn_ip_tcp(["NTLMSSP",
                                     "ncacn_ip_tcp",
                                     "NTLMSSP"],
                                    creds, "dnsserver", "sign", "SIGN",
                                    self.rpc_ncacn_ip_tcp_ntlm_check)

def test_rpc_ncacn_ip_tcp_krb5_dns_sign(self):
        creds = self.insta_creds(template=self.get_credentials(),
                                 kerberos_state=MUST_USE_KERBEROS)
        self._test_rpc_ncacn_ip_tcp(["krb5",
                                     "ncacn_ip_tcp",
                                     "ENC-TS Pre-authentication",
                                     "ENC-TS Pre-authentication"],
                                    creds, "dnsserver", "sign", "SIGN",
                                    self.rpc_ncacn_ip_tcp_krb5_check)

def test_rpc_ncacn_ip_tcp_ntlm_dns(self):
        creds = self.insta_creds(template=self.get_credentials(),
                                 kerberos_state=DONT_USE_KERBEROS)
        self._test_rpc_ncacn_ip_tcp(["NTLMSSP",
                                     "ncacn_ip_tcp",
                                     "NTLMSSP"],
                                    creds, "dnsserver", "", "SIGN",
                                    self.rpc_ncacn_ip_tcp_ntlm_check)

def test_rpc_ncacn_ip_tcp_krb5_dns(self):
        creds = self.insta_creds(template=self.get_credentials(),
                                 kerberos_state=MUST_USE_KERBEROS)
        self._test_rpc_ncacn_ip_tcp(["krb5",
                                     "ncacn_ip_tcp",
                                     "ENC-TS Pre-authentication",
                                     "ENC-TS Pre-authentication"],
                                    creds, "dnsserver", "", "SIGN",
                                    self.rpc_ncacn_ip_tcp_krb5_check)

def test_rpc_ncacn_ip_tcp_ntlm_dns_connect(self):
        creds = self.insta_creds(template=self.get_credentials(),
                                 kerberos_state=DONT_USE_KERBEROS)
        self._test_rpc_ncacn_ip_tcp(["NTLMSSP",
                                     "ncacn_ip_tcp",
                                     "NTLMSSP"],
                                    creds, "dnsserver", "connect", "NONE",
                                    self.rpc_ncacn_ip_tcp_ntlm_check)

def test_rpc_ncacn_ip_tcp_krb5_dns_connect(self):
        creds = self.insta_creds(template=self.get_credentials(),
                                 kerberos_state=MUST_USE_KERBEROS)
        self._test_rpc_ncacn_ip_tcp(["krb5",
                                     "ncacn_ip_tcp",
                                     "ENC-TS Pre-authentication",
                                     "ENC-TS Pre-authentication"],
                                    creds, "dnsserver", "connect", "NONE",
                                    self.rpc_ncacn_ip_tcp_krb5_check)

def test_rpc_ncacn_ip_tcp_ntlm_dns_seal(self):
        creds = self.insta_creds(template=self.get_credentials(),
                                 kerberos_state=DONT_USE_KERBEROS)
        self._test_rpc_ncacn_ip_tcp(["NTLMSSP",
                                     "ncacn_ip_tcp",
                                     "NTLMSSP"],
                                    creds, "dnsserver", "seal", "SEAL",
                                    self.rpc_ncacn_ip_tcp_ntlm_check)

def test_rpc_ncacn_ip_tcp_krb5_dns_seal(self):
        creds = self.insta_creds(template=self.get_credentials(),
                                 kerberos_state=MUST_USE_KERBEROS)
        self._test_rpc_ncacn_ip_tcp(["krb5",
                                     "ncacn_ip_tcp",
                                     "ENC-TS Pre-authentication",
                                     "ENC-TS Pre-authentication"],
                                    creds, "dnsserver", "seal", "SEAL",
                                    self.rpc_ncacn_ip_tcp_krb5_check)

def test_ldap(self):

def isLastExpectedMessage(msg):
            return (msg["type"] == "Authorization" and
                    msg["Authorization"]["serviceDescription"] == "LDAP" and
                    msg["Authorization"]["transportProtection"] == "SEAL" and
                    msg["Authorization"]["authType"] == "krb5")

self.samdb = SamDB(url="ldap://%s" % os.environ["SERVER"],
                           lp=self.get_loadparm(),
                           credentials=self.get_credentials())

messages = self.waitForMessages(isLastExpectedMessage)
        self.assertEqual(3,
                          len(messages),
                          "Did not receive the expected number of messages")

# Check the first message it should be an Authentication
        msg = messages[0]
        self.assertEqual("Authentication", msg["type"])
        self.assertEqual("NT_STATUS_OK", msg["Authentication"]["status"])
        self.assertEqual("Kerberos KDC",
                          msg["Authentication"]["serviceDescription"])
        self.assertEqual("ENC-TS Pre-authentication",
                          msg["Authentication"]["authDescription"])
        self.assertTrue(msg["Authentication"]["duration"] > 0)
        self.assertEqual(
            EVT_ID_SUCCESSFUL_LOGON, msg["Authentication"]["eventId"])
        self.assertEqual(
            EVT_LOGON_NETWORK, msg["Authentication"]["logonType"])

# Check the second message it should be an Authentication
        msg = messages[1]
        self.assertEqual("Authentication", msg["type"])
        self.assertEqual("NT_STATUS_OK", msg["Authentication"]["status"])
        self.assertEqual("Kerberos KDC",
                          msg["Authentication"]["serviceDescription"])
        self.assertEqual("ENC-TS Pre-authentication",
                          msg["Authentication"]["authDescription"])
        self.assertTrue(msg["Authentication"]["duration"] > 0)
        self.assertEqual(
            EVT_ID_SUCCESSFUL_LOGON, msg["Authentication"]["eventId"])
        self.assertEqual(
            EVT_LOGON_NETWORK, msg["Authentication"]["logonType"])

def test_ldap_ntlm(self):

self.samdb = SamDB(url="ldap://%s" % os.environ["SERVER_IP"],
                           lp=self.get_loadparm(),
                           credentials=self.get_credentials())

messages = self.waitForMessages(isLastExpectedMessage)
        self.assertEqual(2,
                          len(messages),
                          "Did not receive the expected number of messages")
        # Check the first message it should be an Authentication
        msg = messages[0]
        self.assertEqual("Authentication", msg["type"])
        self.assertEqual("NT_STATUS_OK", msg["Authentication"]["status"])
        self.assertEqual("LDAP",
                          msg["Authentication"]["serviceDescription"])
        self.assertEqual("NTLMSSP", msg["Authentication"]["authDescription"])
        self.assertTrue(msg["Authentication"]["duration"] > 0)
        self.assertEqual(
            EVT_ID_SUCCESSFUL_LOGON, msg["Authentication"]["eventId"])
        self.assertEqual(
            EVT_LOGON_NETWORK, msg["Authentication"]["logonType"])

def test_ldap_simple_bind(self):
        def isLastExpectedMessage(msg):
            return (msg["type"] == "Authorization" and
                    msg["Authorization"]["serviceDescription"] == "LDAP" and
                    msg["Authorization"]["transportProtection"] == "TLS" and
                    msg["Authorization"]["authType"] == "simple bind")

creds = self.insta_creds(template=self.get_credentials())
        creds.set_bind_dn("%s\\%s" % (creds.get_domain(),
                                      creds.get_username()))

self.samdb = SamDB(url="ldaps://%s" % os.environ["SERVER"],
                           lp=self.get_loadparm(),
                           credentials=creds)

messages = self.waitForMessages(isLastExpectedMessage)
        self.assertEqual(2,
                          len(messages),
                          "Did not receive the expected number of messages")

# Check the first message it should be an Authentication
        msg = messages[0]
        self.assertEqual("Authentication", msg["type"])
        self.assertEqual("NT_STATUS_OK", msg["Authentication"]["status"])
        self.assertEqual("LDAP",
                          msg["Authentication"]["serviceDescription"])
        self.assertEqual("simple bind/TLS",
                          msg["Authentication"]["authDescription"])
        self.assertEqual(
            EVT_ID_SUCCESSFUL_LOGON, msg["Authentication"]["eventId"])
        self.assertEqual(
            EVT_LOGON_NETWORK_CLEAR_TEXT, msg["Authentication"]["logonType"])

def test_ldap_simple_bind_bad_password(self):
        def isLastExpectedMessage(msg):
            return (msg["type"] == "Authentication" and
                    msg["Authentication"]["serviceDescription"] == "LDAP" and
                    (msg["Authentication"]["status"] ==
                        "NT_STATUS_WRONG_PASSWORD") and
                    (msg["Authentication"]["authDescription"] ==
                        "simple bind/TLS") and
                    (msg["Authentication"]["eventId"] ==
                        EVT_ID_UNSUCCESSFUL_LOGON) and
                    (msg["Authentication"]["logonType"] ==
                        EVT_LOGON_NETWORK_CLEAR_TEXT))

creds = self.insta_creds(template=self.get_credentials())
        creds.set_password("badPassword")
        creds.set_bind_dn("%s\\%s" % (creds.get_domain(),
                                      creds.get_username()))

thrown = False
        try:
            self.samdb = SamDB(url="ldaps://%s" % os.environ["SERVER"],
                               lp=self.get_loadparm(),
                               credentials=creds)
        except LdbError:
            thrown = True
        self.assertEqual(thrown, True)

messages = self.waitForMessages(isLastExpectedMessage)
        self.assertEqual(1,
                          len(messages),
                          "Did not receive the expected number of messages")

def test_ldap_simple_bind_bad_user(self):
        def isLastExpectedMessage(msg):
            return (msg["type"] == "Authentication" and
                    msg["Authentication"]["serviceDescription"] == "LDAP" and
                    (msg["Authentication"]["status"] ==
                        "NT_STATUS_NO_SUCH_USER") and
                    (msg["Authentication"]["authDescription"] ==
                        "simple bind/TLS") and
                    (msg["Authentication"]["eventId"] ==
                        EVT_ID_UNSUCCESSFUL_LOGON) and
                    (msg["Authentication"]["logonType"] ==
                        EVT_LOGON_NETWORK_CLEAR_TEXT))

creds = self.insta_creds(template=self.get_credentials())
        creds.set_bind_dn("%s\\%s" % (creds.get_domain(), "badUser"))

def test_ldap_simple_bind_unparseable_user(self):
        def isLastExpectedMessage(msg):
            return (msg["type"] == "Authentication" and
                    msg["Authentication"]["serviceDescription"] == "LDAP" and
                    (msg["Authentication"]["status"] ==
                        "NT_STATUS_NO_SUCH_USER") and
                    (msg["Authentication"]["authDescription"] ==
                        "simple bind/TLS") and
                    (msg["Authentication"]["eventId"] ==
                        EVT_ID_UNSUCCESSFUL_LOGON) and
                    (msg["Authentication"]["logonType"] ==
                        EVT_LOGON_NETWORK_CLEAR_TEXT))

creds = self.insta_creds(template=self.get_credentials())
        creds.set_bind_dn("%s\\%s" % (creds.get_domain(), "abdcef"))

#
    # Note: as this test does not expect any messages it will
    #       time out in the call to self.waitForMessages.
    #       This is expected, but it will slow this test.
    def test_ldap_anonymous_access_bind_only(self):
        # Should be no logging for anonymous bind
        # so receiving any message indicates a failure.
        def isLastExpectedMessage(msg):
            return True

creds = self.insta_creds(template=self.get_credentials())
        creds.set_anonymous()

self.samdb = SamDB(url="ldaps://%s" % os.environ["SERVER"],
                           lp=self.get_loadparm(),
                           credentials=creds)

messages = self.waitForMessages(isLastExpectedMessage)
        self.assertEqual(0,
                          len(messages),
                          "Did not receive the expected number of messages")

def test_ldap_anonymous_access(self):
        def isLastExpectedMessage(msg):
            return (msg["type"] == "Authorization" and
                    msg["Authorization"]["serviceDescription"] == "LDAP" and
                    msg["Authorization"]["transportProtection"] == "TLS" and
                    msg["Authorization"]["account"] == "ANONYMOUS LOGON" and
                    msg["Authorization"]["authType"] == "no bind")

creds = self.insta_creds(template=self.get_credentials())
        creds.set_anonymous()

self.samdb = SamDB(url="ldaps://%s" % os.environ["SERVER"],
                           lp=self.get_loadparm(),
                           credentials=creds)

try:
            self.samdb.search(base=self.samdb.domain_dn())
            self.fail("Expected an LdbError exception")
        except LdbError:
            pass

def test_smb(self):
        def isLastExpectedMessage(msg):
            return (msg["type"] == "Authorization" and
                    "SMB" in msg["Authorization"]["serviceDescription"] and
                    msg["Authorization"]["authType"] == "krb5" and
                    msg["Authorization"]["transportProtection"] == "SMB")

creds = self.insta_creds(template=self.get_credentials())
        self.smb_connection(creds)

messages = self.waitForMessages(isLastExpectedMessage)
        self.assertEqual(3,
                          len(messages),
                          "Did not receive the expected number of messages")
        # Check the first message it should be an Authentication
        msg = messages[0]
        self.assertEqual("Authentication", msg["type"])
        self.assertEqual("NT_STATUS_OK", msg["Authentication"]["status"])
        self.assertEqual("Kerberos KDC",
                          msg["Authentication"]["serviceDescription"])
        self.assertEqual("ENC-TS Pre-authentication",
                          msg["Authentication"]["authDescription"])
        self.assertEqual(EVT_ID_SUCCESSFUL_LOGON,
                          msg["Authentication"]["eventId"])
        self.assertEqual(EVT_LOGON_NETWORK,
                          msg["Authentication"]["logonType"])

# Check the second message it should be an Authentication
        msg = messages[1]
        self.assertEqual("Authentication", msg["type"])
        self.assertEqual("NT_STATUS_OK", msg["Authentication"]["status"])
        self.assertEqual("Kerberos KDC",
                          msg["Authentication"]["serviceDescription"])
        self.assertEqual("ENC-TS Pre-authentication",
                          msg["Authentication"]["authDescription"])
        self.assertEqual(EVT_ID_SUCCESSFUL_LOGON,
                          msg["Authentication"]["eventId"])
        self.assertEqual(EVT_LOGON_NETWORK,
                          msg["Authentication"]["logonType"])

def test_smb_bad_password(self):
        def isLastExpectedMessage(msg):
            return (msg["type"] == "Authentication" and
                    (msg["Authentication"]["serviceDescription"] ==
                        "Kerberos KDC") and
                    (msg["Authentication"]["status"] ==
                        "NT_STATUS_WRONG_PASSWORD") and
                    (msg["Authentication"]["authDescription"] ==
                        "ENC-TS Pre-authentication"))

creds = self.insta_creds(template=self.get_credentials())
        creds.set_kerberos_state(MUST_USE_KERBEROS)
        creds.set_password("badPassword")

thrown = False
        try:
            self.smb_connection(creds)
        except NTSTATUSError:
            thrown = True
        self.assertEqual(thrown, True)

def test_smb_bad_user(self):
        def isLastExpectedMessage(msg):
            return (msg["type"] == "Authentication" and
                    (msg["Authentication"]["serviceDescription"] ==
                        "Kerberos KDC") and
                    (msg["Authentication"]["status"] ==
                        "NT_STATUS_NO_SUCH_USER") and
                    (msg["Authentication"]["authDescription"] ==
                        "ENC-TS Pre-authentication") and
                    (msg["Authentication"]["eventId"] ==
                        EVT_ID_UNSUCCESSFUL_LOGON) and
                    (msg["Authentication"]["logonType"] ==
                        EVT_LOGON_NETWORK))

creds = self.insta_creds(template=self.get_credentials())
        creds.set_kerberos_state(MUST_USE_KERBEROS)
        creds.set_username("badUser")

thrown = False
        try:
            self.smb_connection(creds)
        except NTSTATUSError:
            thrown = True
        self.assertEqual(thrown, True)

def test_smb1_anonymous(self):
        def isLastExpectedMessage(msg):
            return (msg["type"] == "Authorization" and
                    msg["Authorization"]["serviceDescription"] == "SMB" and
                    msg["Authorization"]["authType"] == "NTLMSSP" and
                    msg["Authorization"]["account"] == "ANONYMOUS LOGON" and
                    msg["Authorization"]["transportProtection"] == "SMB")

server = os.environ["SERVER"]

path = "//%s/IPC$" % server
        auth = "-N"
        call(["bin/smbclient", path, auth, "-mNT1", "-c quit"])

# Check the first message it should be an Authentication
        msg = messages[0]
        self.assertEqual("Authentication", msg["type"])
        self.assertEqual("NT_STATUS_NO_SUCH_USER",
                          msg["Authentication"]["status"])
        self.assertEqual("SMB",
                          msg["Authentication"]["serviceDescription"])
        self.assertEqual("NTLMSSP",
                          msg["Authentication"]["authDescription"])
        self.assertEqual("No-Password",
                          msg["Authentication"]["passwordType"])
        self.assertEqual(EVT_ID_UNSUCCESSFUL_LOGON,
                          msg["Authentication"]["eventId"])
        self.assertEqual(EVT_LOGON_NETWORK,
                          msg["Authentication"]["logonType"])

# Check the second message it should be an Authentication
        msg = messages[1]
        self.assertEqual("Authentication", msg["type"])
        self.assertEqual("NT_STATUS_OK",
                          msg["Authentication"]["status"])
        self.assertEqual("SMB",
                          msg["Authentication"]["serviceDescription"])
        self.assertEqual("NTLMSSP",
                          msg["Authentication"]["authDescription"])
        self.assertEqual("No-Password",
                          msg["Authentication"]["passwordType"])
        self.assertEqual("ANONYMOUS LOGON",
                          msg["Authentication"]["becameAccount"])
        self.assertEqual(EVT_ID_SUCCESSFUL_LOGON,
                          msg["Authentication"]["eventId"])
        self.assertEqual(EVT_LOGON_NETWORK,
                          msg["Authentication"]["logonType"])

def test_smb2_anonymous(self):
        def isLastExpectedMessage(msg):
            return (msg["type"] == "Authorization" and
                    msg["Authorization"]["serviceDescription"] == "SMB2" and
                    msg["Authorization"]["authType"] == "NTLMSSP" and
                    msg["Authorization"]["account"] == "ANONYMOUS LOGON" and
                    msg["Authorization"]["transportProtection"] == "SMB")

server = os.environ["SERVER"]

path = "//%s/IPC$" % server
        auth = "-N"
        call(["bin/smbclient", path, auth, "-mSMB3", "-c quit"])

# Check the first message it should be an Authentication
        msg = messages[0]
        self.assertEqual("Authentication", msg["type"])
        self.assertEqual("NT_STATUS_NO_SUCH_USER",
                          msg["Authentication"]["status"])
        self.assertEqual("SMB2",
                          msg["Authentication"]["serviceDescription"])
        self.assertEqual("NTLMSSP",
                          msg["Authentication"]["authDescription"])
        self.assertEqual("No-Password",
                          msg["Authentication"]["passwordType"])
        self.assertEqual(EVT_ID_UNSUCCESSFUL_LOGON,
                          msg["Authentication"]["eventId"])
        self.assertEqual(EVT_LOGON_NETWORK,
                          msg["Authentication"]["logonType"])

# Check the second message it should be an Authentication
        msg = messages[1]
        self.assertEqual("Authentication", msg["type"])
        self.assertEqual("NT_STATUS_OK",
                          msg["Authentication"]["status"])
        self.assertEqual("SMB2",
                          msg["Authentication"]["serviceDescription"])
        self.assertEqual("NTLMSSP",
                          msg["Authentication"]["authDescription"])
        self.assertEqual("No-Password",
                          msg["Authentication"]["passwordType"])
        self.assertEqual("ANONYMOUS LOGON",
                          msg["Authentication"]["becameAccount"])
        self.assertEqual(EVT_ID_SUCCESSFUL_LOGON,
                          msg["Authentication"]["eventId"])
        self.assertEqual(EVT_LOGON_NETWORK,
                          msg["Authentication"]["logonType"])

def test_smb_no_krb_spnego(self):
        def isLastExpectedMessage(msg):
            return (msg["type"] == "Authorization" and
                    "SMB" in msg["Authorization"]["serviceDescription"] and
                    msg["Authorization"]["authType"] == "NTLMSSP" and
                    msg["Authorization"]["transportProtection"] == "SMB")

creds = self.insta_creds(template=self.get_credentials(),
                                 kerberos_state=DONT_USE_KERBEROS)
        self.smb_connection(creds)

messages = self.waitForMessages(isLastExpectedMessage)
        self.assertEqual(2,
                          len(messages),
                          "Did not receive the expected number of messages")
        # Check the first message it should be an Authentication
        msg = messages[0]
        self.assertEqual("Authentication", msg["type"])
        self.assertEqual("NT_STATUS_OK", msg["Authentication"]["status"])
        self.assertIn(msg["Authentication"]["serviceDescription"],
                      ["SMB", "SMB2"])
        self.assertEqual("NTLMSSP",
                          msg["Authentication"]["authDescription"])
        self.assertEqual("NTLMv2",
                          msg["Authentication"]["passwordType"])
        self.assertEqual(EVT_ID_SUCCESSFUL_LOGON,
                          msg["Authentication"]["eventId"])
        self.assertEqual(EVT_LOGON_NETWORK,
                          msg["Authentication"]["logonType"])

def test_smb_no_krb_spnego_bad_password(self):
        def isLastExpectedMessage(msg):
            return (msg["type"] == "Authentication" and
                    "SMB" in msg["Authentication"]["serviceDescription"] and
                    msg["Authentication"]["authDescription"] == "NTLMSSP" and
                    msg["Authentication"]["passwordType"] == "NTLMv2" and
                    (msg["Authentication"]["status"] ==
                        "NT_STATUS_WRONG_PASSWORD") and
                    (msg["Authentication"]["eventId"] ==
                        EVT_ID_UNSUCCESSFUL_LOGON) and
                    (msg["Authentication"]["logonType"] ==
                        EVT_LOGON_NETWORK))

creds = self.insta_creds(template=self.get_credentials(),
                                 kerberos_state=DONT_USE_KERBEROS)
        creds.set_password("badPassword")

thrown = False
        try:
            self.smb_connection(creds)
        except NTSTATUSError:
            thrown = True
        self.assertEqual(thrown, True)

def test_smb_no_krb_spnego_bad_user(self):
        def isLastExpectedMessage(msg):
            return (msg["type"] == "Authentication" and
                    "SMB" in msg["Authentication"]["serviceDescription"] and
                    msg["Authentication"]["authDescription"] == "NTLMSSP" and
                    msg["Authentication"]["passwordType"] == "NTLMv2" and
                    (msg["Authentication"]["status"] ==
                        "NT_STATUS_NO_SUCH_USER") and
                    (msg["Authentication"]["eventId"] ==
                        EVT_ID_UNSUCCESSFUL_LOGON) and
                    (msg["Authentication"]["logonType"] ==
                        EVT_LOGON_NETWORK))

creds = self.insta_creds(template=self.get_credentials(),
                                 kerberos_state=DONT_USE_KERBEROS)
        creds.set_username("badUser")

thrown = False
        try:
            self.smb_connection(creds)
        except NTSTATUSError:
            thrown = True
        self.assertEqual(thrown, True)

def test_smb_no_krb_no_spnego_no_ntlmv2(self):
        def isLastExpectedMessage(msg):
            return (msg["type"] == "Authorization" and
                    msg["Authorization"]["serviceDescription"] == "SMB" and
                    msg["Authorization"]["authType"] == "bare-NTLM" and
                    msg["Authorization"]["transportProtection"] == "SMB")

creds = self.insta_creds(template=self.get_credentials(),
                                 kerberos_state=DONT_USE_KERBEROS)
        self.smb_connection(creds,
                            force_smb1=True,
                            ntlmv2_auth="no",
                            use_spnego="no")

messages = self.waitForMessages(isLastExpectedMessage)
        self.assertEqual(2,
                          len(messages),
                          "Did not receive the expected number of messages")
        # Check the first message it should be an Authentication
        msg = messages[0]
        self.assertEqual("Authentication", msg["type"])
        self.assertEqual("NT_STATUS_OK", msg["Authentication"]["status"])
        self.assertEqual("SMB",
                          msg["Authentication"]["serviceDescription"])
        self.assertEqual("bare-NTLM",
                          msg["Authentication"]["authDescription"])
        self.assertEqual("NTLMv1",
                          msg["Authentication"]["passwordType"])
        self.assertEqual(EVT_ID_SUCCESSFUL_LOGON,
                          msg["Authentication"]["eventId"])
        self.assertEqual(EVT_LOGON_NETWORK,
                          msg["Authentication"]["logonType"])

def test_smb_no_krb_no_spnego_no_ntlmv2_bad_password(self):
        def isLastExpectedMessage(msg):
            return (msg["type"] == "Authentication" and
                    msg["Authentication"]["serviceDescription"] == "SMB" and
                    msg["Authentication"]["authDescription"] == "bare-NTLM" and
                    msg["Authentication"]["passwordType"] == "NTLMv1" and
                    (msg["Authentication"]["status"] ==
                        "NT_STATUS_WRONG_PASSWORD") and
                    (msg["Authentication"]["eventId"] ==
                        EVT_ID_UNSUCCESSFUL_LOGON) and
                    (msg["Authentication"]["logonType"] ==
                        EVT_LOGON_NETWORK))

creds = self.insta_creds(template=self.get_credentials(),
                                 kerberos_state=DONT_USE_KERBEROS)
        creds.set_password("badPassword")

thrown = False
        try:
            self.smb_connection(creds,
                                force_smb1=True,
                                ntlmv2_auth="no",
                                use_spnego="no")
        except NTSTATUSError:
            thrown = True
        self.assertEqual(thrown, True)

def test_smb_no_krb_no_spnego_no_ntlmv2_bad_user(self):
        def isLastExpectedMessage(msg):
            return (msg["type"] == "Authentication" and
                    msg["Authentication"]["serviceDescription"] == "SMB" and
                    msg["Authentication"]["authDescription"] == "bare-NTLM" and
                    msg["Authentication"]["passwordType"] == "NTLMv1" and
                    (msg["Authentication"]["status"] ==
                        "NT_STATUS_NO_SUCH_USER") and
                    (msg["Authentication"]["eventId"] ==
                        EVT_ID_UNSUCCESSFUL_LOGON) and
                    (msg["Authentication"]["logonType"] ==
                        EVT_LOGON_NETWORK))

creds = self.insta_creds(template=self.get_credentials(),
                                 kerberos_state=DONT_USE_KERBEROS)
        creds.set_username("badUser")

def test_samlogon_interactive(self):

workstation = "AuthLogTests"

def isLastExpectedMessage(msg):
            return (msg["type"] == "Authentication" and
                    (msg["Authentication"]["serviceDescription"] ==
                        "SamLogon") and
                    (msg["Authentication"]["authDescription"] ==
                        "interactive") and
                    msg["Authentication"]["status"] == "NT_STATUS_OK" and
                    (msg["Authentication"]["workstation"] ==
                        r"\\%s" % workstation) and
                    (msg["Authentication"]["eventId"] ==
                        EVT_ID_SUCCESSFUL_LOGON) and
                    (msg["Authentication"]["logonType"] ==
                        EVT_LOGON_INTERACTIVE))

server = os.environ["SERVER"]
        user = os.environ["USERNAME"]
        password = os.environ["PASSWORD"]
        samlogon = "samlogon %s %s %s %d" % (user, password, workstation, 1)