PEP8: fix E703: statement ends with a semicolon
[samba.git] / python / samba / tests / py_credentials.py
1 # Integration tests for pycredentials
2 #
3 # Copyright (C) Catalyst IT Ltd. 2017
4 #
5 # This program is free software; you can redistribute it and/or modify
6 # it under the terms of the GNU General Public License as published by
7 # the Free Software Foundation; either version 3 of the License, or
8 # (at your option) any later version.
9 #
10 # This program is distributed in the hope that it will be useful,
11 # but WITHOUT ANY WARRANTY; without even the implied warranty of
12 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
13 # GNU General Public License for more details.
14 #
15 # You should have received a copy of the GNU General Public License
16 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
17 #
18 from samba.tests import TestCase, delete_force
19 import os
20
21 import samba
22 from samba.auth import system_session
23 from samba.credentials import (
24     Credentials,
25     CLI_CRED_NTLMv2_AUTH,
26     CLI_CRED_NTLM_AUTH,
27     DONT_USE_KERBEROS)
28 from samba.dcerpc import netlogon, ntlmssp, srvsvc
29 from samba.dcerpc.netlogon import (
30     netr_Authenticator,
31     netr_WorkstationInformation,
32     MSV1_0_ALLOW_MSVCHAPV2
33 )
34 from samba.dcerpc.misc import SEC_CHAN_WKSTA
35 from samba.dsdb import (
36     UF_WORKSTATION_TRUST_ACCOUNT,
37     UF_PASSWD_NOTREQD,
38     UF_NORMAL_ACCOUNT)
39 from samba.ndr import ndr_pack
40 from samba.samdb import SamDB
41 from samba import NTSTATUSError, ntstatus
42 import ctypes
43
44 """
45 Integration tests for pycredentials
46 """
47
48 MACHINE_NAME = "PCTM"
49 USER_NAME    = "PCTU"
50
51
52 class PyCredentialsTests(TestCase):
53
54     def setUp(self):
55         super(PyCredentialsTests, self).setUp()
56
57         self.server      = os.environ["SERVER"]
58         self.domain      = os.environ["DOMAIN"]
59         self.host        = os.environ["SERVER_IP"]
60         self.lp          = self.get_loadparm()
61
62         self.credentials = self.get_credentials()
63
64         self.session     = system_session()
65         self.ldb = SamDB(url="ldap://%s" % self.host,
66                          session_info=self.session,
67                          credentials=self.credentials,
68                          lp=self.lp)
69
70         self.create_machine_account()
71         self.create_user_account()
72
73     def tearDown(self):
74         super(PyCredentialsTests, self).tearDown()
75         delete_force(self.ldb, self.machine_dn)
76         delete_force(self.ldb, self.user_dn)
77
78     # Until a successful netlogon connection has been established there will
79     # not be a valid authenticator associated with the credentials
80     # and new_client_authenticator should throw a ValueError
81     def test_no_netlogon_connection(self):
82         self.assertRaises(ValueError,
83                           self.machine_creds.new_client_authenticator)
84
85     # Once a netlogon connection has been established,
86     # new_client_authenticator should return a value
87     #
88     def test_have_netlogon_connection(self):
89         c = self.get_netlogon_connection()
90         a = self.machine_creds.new_client_authenticator()
91         self.assertIsNotNone(a)
92
93     # Get an authenticator and use it on a sequence of operations requiring
94     # an authenticator
95     def test_client_authenticator(self):
96         c = self.get_netlogon_connection()
97         (authenticator, subsequent) = self.get_authenticator(c)
98         self.do_NetrLogonSamLogonWithFlags(c, authenticator, subsequent)
99         (authenticator, subsequent) = self.get_authenticator(c)
100         self.do_NetrLogonGetDomainInfo(c, authenticator, subsequent)
101         (authenticator, subsequent) = self.get_authenticator(c)
102         self.do_NetrLogonGetDomainInfo(c, authenticator, subsequent)
103         (authenticator, subsequent) = self.get_authenticator(c)
104         self.do_NetrLogonGetDomainInfo(c, authenticator, subsequent)
105
106     def test_SamLogonEx(self):
107         c = self.get_netlogon_connection()
108
109         logon = samlogon_logon_info(self.domain,
110                                     self.machine_name,
111                                     self.user_creds)
112
113         logon_level = netlogon.NetlogonNetworkTransitiveInformation
114         validation_level = netlogon.NetlogonValidationSamInfo4
115         netr_flags = 0
116
117         try:
118             c.netr_LogonSamLogonEx(self.server,
119                                    self.user_creds.get_workstation(),
120                                    logon_level,
121                                    logon,
122                                    validation_level,
123                                    netr_flags)
124         except NTSTATUSError as e:
125             enum = ctypes.c_uint32(e.args[0]).value
126             if enum == ntstatus.NT_STATUS_WRONG_PASSWORD:
127                 self.fail("got wrong password error")
128             else:
129                 raise
130
131     def test_SamLogonEx_no_domain(self):
132         c = self.get_netlogon_connection()
133
134         self.user_creds.set_domain('')
135
136         logon = samlogon_logon_info(self.domain,
137                                     self.machine_name,
138                                     self.user_creds)
139
140         logon_level = netlogon.NetlogonNetworkTransitiveInformation
141         validation_level = netlogon.NetlogonValidationSamInfo4
142         netr_flags = 0
143
144         try:
145             c.netr_LogonSamLogonEx(self.server,
146                                    self.user_creds.get_workstation(),
147                                    logon_level,
148                                    logon,
149                                    validation_level,
150                                    netr_flags)
151         except NTSTATUSError as e:
152             enum = ctypes.c_uint32(e.args[0]).value
153             if enum == ntstatus.NT_STATUS_WRONG_PASSWORD:
154                 self.fail("got wrong password error")
155             else:
156                 self.fail("got unexpected error" + str(e))
157
158     def test_SamLogonExNTLM(self):
159         c = self.get_netlogon_connection()
160
161         logon = samlogon_logon_info(self.domain,
162                                     self.machine_name,
163                                     self.user_creds,
164                                     flags=CLI_CRED_NTLM_AUTH)
165
166         logon_level = netlogon.NetlogonNetworkTransitiveInformation
167         validation_level = netlogon.NetlogonValidationSamInfo4
168         netr_flags = 0
169
170         try:
171             c.netr_LogonSamLogonEx(self.server,
172                                    self.user_creds.get_workstation(),
173                                    logon_level,
174                                    logon,
175                                    validation_level,
176                                    netr_flags)
177         except NTSTATUSError as e:
178             enum = ctypes.c_uint32(e.args[0]).value
179             if enum == ntstatus.NT_STATUS_WRONG_PASSWORD:
180                 self.fail("got wrong password error")
181             else:
182                 raise
183
184     def test_SamLogonExMSCHAPv2(self):
185         c = self.get_netlogon_connection()
186
187         logon = samlogon_logon_info(self.domain,
188                                     self.machine_name,
189                                     self.user_creds,
190                                     flags=CLI_CRED_NTLM_AUTH)
191
192         logon.identity_info.parameter_control = MSV1_0_ALLOW_MSVCHAPV2
193
194         logon_level = netlogon.NetlogonNetworkTransitiveInformation
195         validation_level = netlogon.NetlogonValidationSamInfo4
196         netr_flags = 0
197
198         try:
199             c.netr_LogonSamLogonEx(self.server,
200                                    self.user_creds.get_workstation(),
201                                    logon_level,
202                                    logon,
203                                    validation_level,
204                                    netr_flags)
205         except NTSTATUSError as e:
206             enum = ctypes.c_uint32(e.args[0]).value
207             if enum == ntstatus.NT_STATUS_WRONG_PASSWORD:
208                 self.fail("got wrong password error")
209             else:
210                 raise
211
212     # Test Credentials.encrypt_netr_crypt_password
213     # By performing a NetrServerPasswordSet2
214     # And the logging on using the new password.
215
216     def test_encrypt_netr_password(self):
217         # Change the password
218         self.do_Netr_ServerPasswordSet2()
219         # Now use the new password to perform an operation
220         srvsvc.srvsvc("ncacn_np:%s" % (self.server),
221                       self.lp,
222                       self.machine_creds)
223
224    # Change the current machine account password with a
225    # netr_ServerPasswordSet2 call.
226
227     def do_Netr_ServerPasswordSet2(self):
228         c = self.get_netlogon_connection()
229         (authenticator, subsequent) = self.get_authenticator(c)
230         PWD_LEN  = 32
231         DATA_LEN = 512
232         newpass = samba.generate_random_password(PWD_LEN, PWD_LEN)
233         encoded = newpass.encode('utf-16-le')
234         pwd_len = len(encoded)
235         filler  = [ord(x) for x in os.urandom(DATA_LEN - pwd_len)]
236         pwd = netlogon.netr_CryptPassword()
237         pwd.length = pwd_len
238         pwd.data = filler + [ord(x) for x in encoded]
239         self.machine_creds.encrypt_netr_crypt_password(pwd)
240         c.netr_ServerPasswordSet2(self.server,
241                                   self.machine_creds.get_workstation(),
242                                   SEC_CHAN_WKSTA,
243                                   self.machine_name,
244                                   authenticator,
245                                   pwd)
246
247         self.machine_pass = newpass
248         self.machine_creds.set_password(newpass)
249
250     # Establish sealed schannel netlogon connection over TCP/IP
251     #
252     def get_netlogon_connection(self):
253         return netlogon.netlogon("ncacn_ip_tcp:%s[schannel,seal]" % self.server,
254                                  self.lp,
255                                  self.machine_creds)
256
257     #
258     # Create the machine account
259     def create_machine_account(self):
260         self.machine_pass = samba.generate_random_password(32, 32)
261         self.machine_name = MACHINE_NAME
262         self.machine_dn = "cn=%s,%s" % (self.machine_name, self.ldb.domain_dn())
263
264         # remove the account if it exists, this will happen if a previous test
265         # run failed
266         delete_force(self.ldb, self.machine_dn)
267
268         utf16pw = unicode(
269             '"' + self.machine_pass.encode('utf-8') + '"', 'utf-8'
270         ).encode('utf-16-le')
271         self.ldb.add({
272             "dn": self.machine_dn,
273             "objectclass": "computer",
274             "sAMAccountName": "%s$" % self.machine_name,
275             "userAccountControl":
276                 str(UF_WORKSTATION_TRUST_ACCOUNT | UF_PASSWD_NOTREQD),
277             "unicodePwd": utf16pw})
278
279         self.machine_creds = Credentials()
280         self.machine_creds.guess(self.get_loadparm())
281         self.machine_creds.set_secure_channel_type(SEC_CHAN_WKSTA)
282         self.machine_creds.set_kerberos_state(DONT_USE_KERBEROS)
283         self.machine_creds.set_password(self.machine_pass)
284         self.machine_creds.set_username(self.machine_name + "$")
285         self.machine_creds.set_workstation(self.machine_name)
286
287     #
288     # Create a test user account
289     def create_user_account(self):
290         self.user_pass = samba.generate_random_password(32, 32)
291         self.user_name = USER_NAME
292         self.user_dn = "cn=%s,%s" % (self.user_name, self.ldb.domain_dn())
293
294         # remove the account if it exists, this will happen if a previous test
295         # run failed
296         delete_force(self.ldb, self.user_dn)
297
298         utf16pw = unicode(
299             '"' + self.user_pass.encode('utf-8') + '"', 'utf-8'
300         ).encode('utf-16-le')
301         self.ldb.add({
302             "dn": self.user_dn,
303             "objectclass": "user",
304             "sAMAccountName": "%s" % self.user_name,
305             "userAccountControl": str(UF_NORMAL_ACCOUNT),
306             "unicodePwd": utf16pw})
307
308         self.user_creds = Credentials()
309         self.user_creds.guess(self.get_loadparm())
310         self.user_creds.set_password(self.user_pass)
311         self.user_creds.set_username(self.user_name)
312         self.user_creds.set_workstation(self.machine_name)
313         pass
314
315     #
316     # Get the authenticator from the machine creds.
317     def get_authenticator(self, c):
318         auth = self.machine_creds.new_client_authenticator()
319         current = netr_Authenticator()
320         current.cred.data = [ord(x) for x in auth["credential"]]
321         current.timestamp = auth["timestamp"]
322
323         subsequent = netr_Authenticator()
324         return (current, subsequent)
325
326     def do_NetrLogonSamLogonWithFlags(self, c, current, subsequent):
327         logon = samlogon_logon_info(self.domain,
328                                     self.machine_name,
329                                     self.user_creds)
330
331         logon_level = netlogon.NetlogonNetworkTransitiveInformation
332         validation_level = netlogon.NetlogonValidationSamInfo4
333         netr_flags = 0
334         c.netr_LogonSamLogonWithFlags(self.server,
335                                       self.user_creds.get_workstation(),
336                                       current,
337                                       subsequent,
338                                       logon_level,
339                                       logon,
340                                       validation_level,
341                                       netr_flags)
342
343     def do_NetrLogonGetDomainInfo(self, c, current, subsequent):
344         query = netr_WorkstationInformation()
345
346         c.netr_LogonGetDomainInfo(self.server,
347                                   self.user_creds.get_workstation(),
348                                   current,
349                                   subsequent,
350                                   2,
351                                   query)
352
353 #
354 # Build the logon data required by NetrLogonSamLogonWithFlags
355
356
357 def samlogon_logon_info(domain_name, computer_name, creds,
358                         flags=CLI_CRED_NTLMv2_AUTH):
359
360     target_info_blob = samlogon_target(domain_name, computer_name)
361
362     challenge = b"abcdefgh"
363     # User account under test
364     response = creds.get_ntlm_response(flags=flags,
365                                        challenge=challenge,
366                                        target_info=target_info_blob)
367
368     logon = netlogon.netr_NetworkInfo()
369
370     logon.challenge     = [ord(x) for x in challenge]
371     logon.nt            = netlogon.netr_ChallengeResponse()
372     logon.nt.length     = len(response["nt_response"])
373     logon.nt.data       = [ord(x) for x in response["nt_response"]]
374     logon.identity_info = netlogon.netr_IdentityInfo()
375
376     (username, domain)  = creds.get_ntlm_username_domain()
377     logon.identity_info.domain_name.string  = domain
378     logon.identity_info.account_name.string = username
379     logon.identity_info.workstation.string  = creds.get_workstation()
380
381     return logon
382
383 #
384 # Build the samlogon target info.
385
386
387 def samlogon_target(domain_name, computer_name):
388     target_info = ntlmssp.AV_PAIR_LIST()
389     target_info.count = 3
390     computername = ntlmssp.AV_PAIR()
391     computername.AvId = ntlmssp.MsvAvNbComputerName
392     computername.Value = computer_name
393
394     domainname = ntlmssp.AV_PAIR()
395     domainname.AvId = ntlmssp.MsvAvNbDomainName
396     domainname.Value = domain_name
397
398     eol = ntlmssp.AV_PAIR()
399     eol.AvId = ntlmssp.MsvAvEOL
400     target_info.pair = [domainname, computername, eol]
401
402     return ndr_pack(target_info)