1 # Unix SMB/CIFS implementation.
2 # Copyright (C) Andrew Bartlett <abartlet@samba.org> 2017
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.
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.
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/>.
19 Tests auth logging tests that exercise SamLogon
24 from samba.samdb import SamDB
25 import samba.tests.auth_log_base
26 from samba.credentials import (
31 from samba.dcerpc import ntlmssp, netlogon
32 from samba.dcerpc.dcerpc import AS_SYSTEM_MAGIC_PATH_TOKEN
33 from samba.ndr import ndr_pack
34 from samba.auth import system_session
35 from samba.tests import delete_force
36 from samba.dsdb import UF_WORKSTATION_TRUST_ACCOUNT, UF_PASSWD_NOTREQD
37 from samba.dcerpc.misc import SEC_CHAN_WKSTA
38 from samba.compat import text_type
39 from samba.dcerpc.windows_event_ids import EVT_ID_SUCCESSFUL_LOGON
42 class AuthLogTestsSamLogon(samba.tests.auth_log_base.AuthLogTestBase):
45 super(AuthLogTestsSamLogon, self).setUp()
46 self.lp = samba.tests.env_loadparm()
47 self.creds = Credentials()
49 self.session = system_session()
51 session_info=self.session,
52 credentials=self.creds,
55 self.domain = os.environ["DOMAIN"]
56 self.netbios_name = "SamLogonTest"
57 self.machinepass = "abcdefghij"
58 self.remoteAddress = AS_SYSTEM_MAGIC_PATH_TOKEN
59 self.base_dn = self.ldb.domain_dn()
60 self.samlogon_dn = ("cn=%s,cn=users,%s" %
61 (self.netbios_name, self.base_dn))
64 super(AuthLogTestsSamLogon, self).tearDown()
65 delete_force(self.ldb, self.samlogon_dn)
67 def _test_samlogon(self, binding, creds, checkFunction):
69 def isLastExpectedMessage(msg):
71 msg["type"] == "Authentication" and
72 msg["Authentication"]["serviceDescription"] == "SamLogon" and
73 msg["Authentication"]["authDescription"] == "network" and
74 msg["Authentication"]["passwordType"] == "NTLMv2" and
75 msg["Authentication"]["eventId"] == EVT_ID_SUCCESSFUL_LOGON)
78 binding = "[schannel,%s]" % binding
80 binding = "[schannel]"
82 utf16pw = text_type('"' + self.machinepass + '"').encode('utf-16-le')
84 "dn": self.samlogon_dn,
85 "objectclass": "computer",
86 "sAMAccountName": "%s$" % self.netbios_name,
88 str(UF_WORKSTATION_TRUST_ACCOUNT | UF_PASSWD_NOTREQD),
89 "unicodePwd": utf16pw})
91 machine_creds = Credentials()
92 machine_creds.guess(self.get_loadparm())
93 machine_creds.set_secure_channel_type(SEC_CHAN_WKSTA)
94 machine_creds.set_password(self.machinepass)
95 machine_creds.set_username(self.netbios_name + "$")
97 netlogon_conn = netlogon.netlogon("ncalrpc:%s" % binding,
100 challenge = b"abcdefgh"
102 target_info = ntlmssp.AV_PAIR_LIST()
103 target_info.count = 3
105 domainname = ntlmssp.AV_PAIR()
106 domainname.AvId = ntlmssp.MsvAvNbDomainName
107 domainname.Value = self.domain
109 computername = ntlmssp.AV_PAIR()
110 computername.AvId = ntlmssp.MsvAvNbComputerName
111 computername.Value = self.netbios_name
113 eol = ntlmssp.AV_PAIR()
114 eol.AvId = ntlmssp.MsvAvEOL
115 target_info.pair = [domainname, computername, eol]
117 target_info_blob = ndr_pack(target_info)
119 response = creds.get_ntlm_response(flags=CLI_CRED_NTLMv2_AUTH,
121 target_info=target_info_blob)
125 logon_level = netlogon.NetlogonNetworkTransitiveInformation
126 logon = samba.dcerpc.netlogon.netr_NetworkInfo()
128 logon.challenge = [x if isinstance(x,int) else ord(x) for x in challenge]
129 logon.nt = netlogon.netr_ChallengeResponse()
130 logon.nt.length = len(response["nt_response"])
131 logon.nt.data = [x if isinstance(x,int) else ord(x) for x in response["nt_response"]]
132 logon.identity_info = samba.dcerpc.netlogon.netr_IdentityInfo()
133 (username, domain) = creds.get_ntlm_username_domain()
135 logon.identity_info.domain_name.string = domain
136 logon.identity_info.account_name.string = username
137 logon.identity_info.workstation.string = creds.get_workstation()
139 validation_level = samba.dcerpc.netlogon.NetlogonValidationSamInfo4
141 result = netlogon_conn.netr_LogonSamLogonEx(
142 os.environ["SERVER"],
143 machine_creds.get_workstation(),
145 validation_level, netr_flags)
147 (validation, authoritative, netr_flags_out) = result
149 messages = self.waitForMessages(isLastExpectedMessage, netlogon_conn)
150 checkFunction(messages)
152 def samlogon_check(self, messages):
154 messages = self.remove_netlogon_messages(messages)
155 expected_messages = 5
156 self.assertEquals(expected_messages,
158 "Did not receive the expected number of messages")
160 # Check the first message it should be an Authorization
162 self.assertEquals("Authorization", msg["type"])
163 self.assertEquals("DCE/RPC",
164 msg["Authorization"]["serviceDescription"])
165 self.assertEquals("ncalrpc", msg["Authorization"]["authType"])
166 self.assertEquals("NONE", msg["Authorization"]["transportProtection"])
167 self.assertTrue(self.is_guid(msg["Authorization"]["sessionId"]))
169 def test_ncalrpc_samlogon(self):
171 creds = self.insta_creds(template=self.get_credentials(),
172 kerberos_state=DONT_USE_KERBEROS)
174 self._test_samlogon("SEAL", creds, self.samlogon_check)
175 except Exception as e:
176 self.fail("Unexpected exception: " + str(e))