selftest: Pass the dcerpc binding object to self.waitForMessages in auth_log
[nivanova/samba-autobuild/.git] / python / samba / tests / auth_log_samlogon.py
1 # Unix SMB/CIFS implementation.
2 # Copyright (C) Andrew Bartlett <abartlet@samba.org> 2017
3 #
4 # This program is free software; you can redistribute it and/or modify
5 # it under the terms of the GNU General Public License as published by
6 # the Free Software Foundation; either version 3 of the License, or
7 # (at your option) any later version.
8 #
9 # This program is distributed in the hope that it will be useful,
10 # but WITHOUT ANY WARRANTY; without even the implied warranty of
11 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
12 # GNU General Public License for more details.
13 #
14 # You should have received a copy of the GNU General Public License
15 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
16 #
17
18 """
19     Tests auth logging tests that exercise SamLogon
20 """
21
22 from samba import auth
23 import samba.tests
24 from samba.messaging import Messaging
25 from samba.dcerpc.messaging import MSG_AUTH_LOG, AUTH_EVENT_NAME
26 import time
27 import json
28 import os
29 from samba import smb
30 from samba.samdb import SamDB
31 import samba.tests.auth_log_base
32 from samba.credentials import (
33     Credentials,
34     DONT_USE_KERBEROS,
35     CLI_CRED_NTLMv2_AUTH
36 )
37 from samba.dcerpc import ntlmssp, netlogon
38 from samba.ndr import ndr_pack
39 from samba.auth import system_session
40 from samba.tests import delete_force
41 from samba.dsdb import UF_WORKSTATION_TRUST_ACCOUNT, UF_PASSWD_NOTREQD
42 from samba.dcerpc.misc import SEC_CHAN_WKSTA
43
44 class AuthLogTestsSamLogon(samba.tests.auth_log_base.AuthLogTestBase):
45
46     def setUp(self):
47         super(AuthLogTestsSamLogon, self).setUp()
48         self.lp      = samba.tests.env_loadparm()
49         self.creds   = Credentials()
50
51         self.session = system_session()
52         self.ldb = SamDB(
53             session_info=self.session,
54             credentials=self.creds,
55             lp=self.lp)
56
57         self.domain        = os.environ["DOMAIN"]
58         self.netbios_name  = "SamLogonTest"
59         self.machinepass   = "abcdefghij"
60         self.remoteAddress = "/root/ncalrpc_as_system"
61         self.base_dn       = self.ldb.domain_dn()
62         self.samlogon_dn   = ("cn=%s,cn=users,%s" %
63                               (self.netbios_name, self.base_dn))
64
65
66     def tearDown(self):
67         super(AuthLogTestsSamLogon , self).tearDown()
68         delete_force(self.ldb, self.samlogon_dn)
69
70     def _test_samlogon(self, binding, creds, checkFunction):
71
72         def isLastExpectedMessage(msg):
73             return (
74                 msg["type"] == "Authentication" and
75                 msg["Authentication"]["serviceDescription"]  == "SamLogon" and
76                 msg["Authentication"]["authDescription"]     == "network" and
77                 msg["Authentication"]["passwordType"]        == "NTLMv2")
78
79         if binding:
80             binding = "[schannel,%s]" % binding
81         else:
82             binding = "[schannel]"
83
84         utf16pw = unicode(
85             '"' + self.machinepass.encode('utf-8') + '"', 'utf-8'
86         ).encode('utf-16-le')
87         self.ldb.add({
88             "dn": self.samlogon_dn,
89             "objectclass": "computer",
90             "sAMAccountName": "%s$" % self.netbios_name,
91             "userAccountControl":
92                 str(UF_WORKSTATION_TRUST_ACCOUNT | UF_PASSWD_NOTREQD),
93             "unicodePwd": utf16pw})
94
95         machine_creds = Credentials()
96         machine_creds.guess(self.get_loadparm())
97         machine_creds.set_secure_channel_type(SEC_CHAN_WKSTA)
98         machine_creds.set_password(self.machinepass)
99         machine_creds.set_username(self.netbios_name + "$")
100
101         netlogon_conn = netlogon.netlogon("ncalrpc:%s" % binding,
102                                           self.get_loadparm(),
103                                           machine_creds)
104         challenge = b"abcdefgh"
105
106         target_info = ntlmssp.AV_PAIR_LIST()
107         target_info.count = 3
108
109         domainname = ntlmssp.AV_PAIR()
110         domainname.AvId = ntlmssp.MsvAvNbDomainName
111         domainname.Value = self.domain
112
113         computername = ntlmssp.AV_PAIR()
114         computername.AvId = ntlmssp.MsvAvNbComputerName
115         computername.Value = self.netbios_name
116
117         eol = ntlmssp.AV_PAIR()
118         eol.AvId = ntlmssp.MsvAvEOL
119         target_info.pair = [domainname, computername, eol]
120
121
122         target_info_blob = ndr_pack(target_info)
123
124         response = creds.get_ntlm_response(flags=CLI_CRED_NTLMv2_AUTH,
125                                            challenge=challenge,
126                                            target_info=target_info_blob)
127
128         netr_flags = 0
129
130         logon_level = netlogon.NetlogonNetworkTransitiveInformation
131         logon = samba.dcerpc.netlogon.netr_NetworkInfo()
132
133         logon.challenge = [ord(x) for x in challenge]
134         logon.nt = netlogon.netr_ChallengeResponse()
135         logon.nt.length = len(response["nt_response"])
136         logon.nt.data = [ord(x) for x in response["nt_response"]]
137         logon.identity_info = samba.dcerpc.netlogon.netr_IdentityInfo()
138         (username, domain) = creds.get_ntlm_username_domain()
139
140         logon.identity_info.domain_name.string = domain
141         logon.identity_info.account_name.string = username
142         logon.identity_info.workstation.string = creds.get_workstation()
143
144         validation_level = samba.dcerpc.netlogon.NetlogonValidationSamInfo4
145
146
147         result = netlogon_conn.netr_LogonSamLogonEx(os.environ["SERVER"],
148                                                machine_creds.get_workstation(),
149                                                logon_level, logon,
150                                                validation_level, netr_flags)
151
152         (validation, authoritative, netr_flags_out) = result
153
154
155         messages = self.waitForMessages(isLastExpectedMessage, netlogon_conn)
156         checkFunction(messages)
157
158     def samlogon_check(self, messages):
159
160         expected_messages = 5
161         self.assertEquals(expected_messages,
162                           len(messages),
163                           "Did not receive the expected number of messages")
164
165         # Check the first message it should be an Authorization
166         msg = messages[0]
167         self.assertEquals("Authorization", msg["type"])
168         self.assertEquals("DCE/RPC",
169                           msg["Authorization"]["serviceDescription"])
170         self.assertEquals("ncalrpc", msg["Authorization"]["authType"])
171         self.assertEquals("NONE", msg["Authorization"]["transportProtection"])
172
173
174     def test_ncalrpc_samlogon(self):
175
176         creds = self.insta_creds(template=self.get_credentials(),
177                                  kerberos_state=DONT_USE_KERBEROS)
178         try:
179             self._test_samlogon("SEAL", creds, self.samlogon_check)
180         except Exception as e:
181             self.fail("Unexpected exception: " + str(e))