Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/ntlm_auth/messages.py: 28%

1# Copyright: (c) 2018, Jordan Borean (@jborean93) <jborean93@gmail.com> 

2# MIT License (see LICENSE or https://opensource.org/licenses/MIT) 

3 

4import hashlib 

5import hmac 

6import os 

7import struct 

8 

9from ntlm_auth.compute_response import ComputeResponse 

10from ntlm_auth.constants import AvId, AvFlags, MessageTypes, NegotiateFlags, \ 

11 NTLM_SIGNATURE 

12from ntlm_auth.rc4 import ARC4 

13 

14try: 

15 from collections import OrderedDict 

16except ImportError: # pragma: no cover 

17 from ordereddict import OrderedDict 

18 

19 

20class TargetInfo(object): 

21 

22 def __init__(self): 

23 self.fields = OrderedDict() 

24 

25 def __setitem__(self, key, value): 

26 self.fields[key] = value 

27 

28 def __getitem__(self, key): 

29 return self.fields.get(key, None) 

30 

31 def __delitem__(self, key): 

32 del self.fields[key] 

33 

34 def pack(self): 

35 if AvId.MSV_AV_EOL in self.fields: 

36 del self[AvId.MSV_AV_EOL] 

37 

38 data = b'' 

39 for attribute_type, attribute_value in self.fields.items(): 

40 data += struct.pack("<HH", attribute_type, len(attribute_value)) 

41 data += attribute_value 

42 

43 # end with a NTLMSSP_AV_EOL 

44 data += struct.pack('<HH', AvId.MSV_AV_EOL, 0) 

45 return data 

46 

47 def unpack(self, data): 

48 attribute_type = None 

49 while attribute_type != AvId.MSV_AV_EOL: 

50 attribute_type = struct.unpack("<H", data[:2])[0] 

51 attribute_length = struct.unpack("<H", data[2:4])[0] 

52 self[attribute_type] = data[4:attribute_length + 4] 

53 data = data[4 + attribute_length:] 

54 

55 

56class NegotiateMessage(object): 

57 EXPECTED_BODY_LENGTH = 40 

58 

59 def __init__(self, negotiate_flags, domain_name, workstation): 

60 """ 

61 [MS-NLMP] v28.0 2016-07-14 

62 

63 2.2.1.1 NEGOTIATE_MESSAGE 

64 The NEGOTIATE_MESSAGE defines an NTLM Negotiate message that is sent 

65 from the client to the server. This message allows the client to 

66 specify its supported NTLM options to the server. 

67 

68 :param negotiate_flags: A NEGOTIATE structure that contains a set of 

69 bit flags. These flags are the options the client supports 

70 :param domain_name: The domain name of the user to authenticate with, 

71 default is None 

72 :param workstation: The worksation of the client machine, default is 

73 None 

74 

75 Attributes: 

76 signature: An 8-byte character array that MUST contain the ASCII 

77 string 'NTLMSSP\0' 

78 message_type: A 32-bit unsigned integer that indicates the message 

79 type. This field must be set to 0x00000001 

80 negotiate_flags: A NEGOTIATE structure that contains a set of bit 

81 flags. These flags are the options the client supports 

82 version: Contains the windows version info of the client. It is 

83 used only debugging purposes and are only set when 

84 NTLMSSP_NEGOTIATE_VERSION flag is set 

85 domain_name: A byte-array that contains the name of the client 

86 authentication domain that MUST Be encoded in the negotiated 

87 character set 

88 workstation: A byte-array that contains the name of the client 

89 machine that MUST Be encoded in the negotiated character set 

90 """ 

91 

92 self.signature = NTLM_SIGNATURE 

93 self.message_type = struct.pack('<L', MessageTypes.NTLM_NEGOTIATE) 

94 

95 # Check if the domain_name value is set, if it is, make sure the 

96 # negotiate_flag is also set 

97 if domain_name is None: 

98 self.domain_name = '' 

99 else: 

100 self.domain_name = domain_name 

101 negotiate_flags |= \ 

102 NegotiateFlags.NTLMSSP_NEGOTIATE_OEM_DOMAIN_SUPPLIED 

103 

104 # Check if the workstation value is set, if it is, make sure the 

105 # negotiate_flag is also set 

106 if workstation is None: 

107 self.workstation = '' 

108 else: 

109 self.workstation = workstation 

110 negotiate_flags |= \ 

111 NegotiateFlags.NTLMSSP_NEGOTIATE_OEM_WORKSTATION_SUPPLIED 

112 

113 # Set the encoding flag to use UNICODE, remove OEM if set. 

114 negotiate_flags |= NegotiateFlags.NTLMSSP_NEGOTIATE_UNICODE 

115 negotiate_flags &= ~NegotiateFlags.NTLMSSP_NEGOTIATE_OEM 

116 

117 # The domain name and workstation are always OEM encoded. 

118 self.domain_name = self.domain_name.encode('ascii') 

119 self.workstation = self.workstation.encode('ascii') 

120 

121 self.version = get_version(negotiate_flags) 

122 

123 self.negotiate_flags = struct.pack('<I', negotiate_flags) 

124 

125 def get_data(self): 

126 payload_offset = self.EXPECTED_BODY_LENGTH 

127 

128 # DomainNameFields - 8 bytes 

129 domain_name_len = struct.pack('<H', len(self.domain_name)) 

130 domain_name_max_len = struct.pack('<H', len(self.domain_name)) 

131 domain_name_buffer_offset = struct.pack('<I', payload_offset) 

132 payload_offset += len(self.domain_name) 

133 

134 # WorkstationFields - 8 bytes 

135 workstation_len = struct.pack('<H', len(self.workstation)) 

136 workstation_max_len = struct.pack('<H', len(self.workstation)) 

137 workstation_buffer_offset = struct.pack('<I', payload_offset) 

138 payload_offset += len(self.workstation) 

139 

140 # Payload - variable length 

141 payload = self.domain_name 

142 payload += self.workstation 

143 

144 # Bring the header values together into 1 message 

145 msg1 = self.signature 

146 msg1 += self.message_type 

147 msg1 += self.negotiate_flags 

148 msg1 += domain_name_len 

149 msg1 += domain_name_max_len 

150 msg1 += domain_name_buffer_offset 

151 msg1 += workstation_len 

152 msg1 += workstation_max_len 

153 msg1 += workstation_buffer_offset 

154 msg1 += self.version 

155 

156 assert self.EXPECTED_BODY_LENGTH == len(msg1),\ 

157 "BODY_LENGTH: %d != msg1: %d"\ 

158 % (self.EXPECTED_BODY_LENGTH, len(msg1)) 

159 

160 # Adding the payload data to the message 

161 msg1 += payload 

162 return msg1 

163 

164 

165class ChallengeMessage(object): 

166 

167 def __init__(self, msg2): 

168 """ 

169 [MS-NLMP] v28.0 2016-07-14 

170 

171 2.2.1.2 CHALLENGE_MESSAGE 

172 The CHALLENGE_MESSAGE defines an NTLM challenge message that is sent 

173 from the server to the client. The CHALLENGE_MESSAGE is used by the 

174 server to challenge the client to prove its identity, For 

175 connection-oriented requests, the CHALLENGE_MESSAGE generated by the 

176 server is in response to the NEGOTIATE_MESSAGE from the client. 

177 

178 :param msg2: The CHALLENGE_MESSAGE received from the server after 

179 sending our NEGOTIATE_MESSAGE. This has been decoded from a base64 

180 string 

181 

182 Attributes 

183 signature: An 8-byte character array that MUST contain the ASCII 

184 string 'NTLMSSP\0' 

185 message_type: A 32-bit unsigned integer that indicates the message 

186 type. This field must be set to 0x00000002 

187 negotiate_flags: A NEGOTIATE strucutre that contains a set of bit 

188 flags. The server sets flags to indicate options it supports 

189 server_challenge: A 64-bit value that contains the NTLM challenge. 

190 The challenge is a 64-bit nonce. Used in the 

191 AuthenticateMessage message 

192 reserved: An 8-byte array whose elements MUST be zero when sent and 

193 MUST be ignored on receipt 

194 version: When NTLMSSP_NEGOTIATE_VERSION flag is set in 

195 negotiate_flags field which contains the windows version info. 

196 Used only for debugging purposes 

197 target_name: When NTLMSSP_REQUEST_TARGET is set is a byte array 

198 that contains the name of the server authentication realm. In a 

199 domain environment this is the domain name not server name 

200 target_info: When NTLMSSP_NEGOTIATE_TARGET_INFO is set is a byte 

201 array that contains a sequence of AV_PAIR structures 

202 """ 

203 

204 self.data = msg2 

205 # Setting the object values from the raw_challenge_message 

206 self.signature = msg2[0:8] 

207 self.message_type = struct.unpack("<I", msg2[8:12])[0] 

208 self.negotiate_flags = struct.unpack("<I", msg2[20:24])[0] 

209 self.server_challenge = msg2[24:32] 

210 self.reserved = msg2[32:40] 

211 

212 if self.negotiate_flags & \ 

213 NegotiateFlags.NTLMSSP_NEGOTIATE_VERSION and \ 

214 self.negotiate_flags & \ 

215 NegotiateFlags.NTLMSSP_NEGOTIATE_EXTENDED_SESSIONSECURITY and \ 

216 len(msg2) > 48: 

217 self.version = struct.unpack("<q", msg2[48:56])[0] 

218 else: 

219 self.version = None 

220 

221 if self.negotiate_flags & NegotiateFlags.NTLMSSP_REQUEST_TARGET: 

222 target_name_len = struct.unpack("<H", msg2[12:14])[0] 

223 target_name_max_len = struct.unpack("<H", msg2[14:16])[0] 

224 target_name_offset_mix = struct.unpack("<I", msg2[16:20])[0] 

225 target_offset_max = target_name_offset_mix + target_name_len 

226 self.target_name = msg2[target_name_offset_mix:target_offset_max] 

227 else: 

228 self.target_name = None 

229 

230 if self.negotiate_flags & NegotiateFlags.NTLMSSP_NEGOTIATE_TARGET_INFO: 

231 target_info_len = struct.unpack("<H", msg2[40:42])[0] 

232 target_info_max_len = struct.unpack("<H", msg2[42:44])[0] 

233 target_info_offset_min = struct.unpack("<I", msg2[44:48])[0] 

234 target_info_offset_max = target_info_offset_min + target_info_len 

235 

236 target_info_raw = \ 

237 msg2[target_info_offset_min:target_info_offset_max] 

238 self.target_info = TargetInfo() 

239 self.target_info.unpack(target_info_raw) 

240 else: 

241 self.target_info = None 

242 

243 # Verify initial integrity of the message, it matches what should be 

244 # there 

245 assert self.signature == NTLM_SIGNATURE 

246 assert self.message_type == MessageTypes.NTLM_CHALLENGE 

247 

248 def get_data(self): 

249 return self.data 

250 

251 

252class AuthenticateMessage(object): 

253 EXPECTED_BODY_LENGTH = 72 

254 EXPECTED_BODY_LENGTH_WITH_MIC = 88 

255 

256 def __init__(self, user_name, password, domain_name, workstation, 

257 challenge_message, ntlm_compatibility, 

258 server_certificate_hash=None, cbt_data=None): 

259 """ 

260 [MS-NLMP] v28.0 2016-07-14 

261 

262 2.2.1.3 AUTHENTICATE_MESSAGE 

263 The AUTHENTICATE_MESSAGE defines an NTLM authenticate message that is 

264 sent from the client to the server after the CHALLENGE_MESSAGE is 

265 processed by the client. 

266 

267 :param user_name: The user name of the user we are trying to 

268 authenticate with 

269 :param password: The password of the user we are trying to authenticate 

270 with 

271 :param domain_name: The domain name of the user account we are 

272 authenticated with, default is None 

273 :param workstation: The workstation we are using to authenticate with, 

274 default is None 

275 :param challenge_message: A ChallengeMessage object that was received 

276 from the server after the negotiate_message 

277 :param ntlm_compatibility: The Lan Manager Compatibility Level, used to 

278 determine what NTLM auth version to use, see Ntlm in ntlm.py for 

279 more details 

280 :param server_certificate_hash: Deprecated, used cbt_data instead 

281 :param cbt_data: The GssChannelBindingsStruct that contains the CBT 

282 data to bind in the auth response 

283 

284 Message Attributes (Attributes used to compute the message structure): 

285 signature: An 8-byte character array that MUST contain the ASCII 

286 string 'NTLMSSP\0' 

287 message_type: A 32-bit unsigned integer that indicates the message 

288 type. This field must be set to 0x00000003 

289 negotiate_flags: A NEGOTIATE strucutre that contains a set of bit 

290 flags. These flags are the choices the client has made from the 

291 CHALLENGE_MESSAGE options 

292 version: Contains the windows version info of the client. It is 

293 used only debugging purposes and are only set when 

294 NTLMSSP_NEGOTIATE_VERSION flag is set 

295 mic: The message integrity for the NEGOTIATE_MESSAGE, 

296 CHALLENGE_MESSAGE and AUTHENTICATE_MESSAGE 

297 lm_challenge_response: An LM_RESPONSE of LMv2_RESPONSE structure 

298 that contains the computed LM response to the challenge 

299 nt_challenge_response: An NTLM_RESPONSE or NTLMv2_RESPONSE 

300 structure that contains the computed NT response to the 

301 challenge 

302 domain_name: The domain or computer name hosting the user account, 

303 MUST be encoded in the negotiated character set 

304 user_name: The name of the user to be authenticated, MUST be 

305 encoded in the negotiated character set 

306 workstation: The name of the computer to which the user is logged 

307 on, MUST Be encoded in the negotiated character set 

308 encrypted_random_session_key: The client's encrypted random session 

309 key 

310 

311 Non-Message Attributes (Attributes not used in auth message): 

312 exported_session_key: A randomly generated session key based on 

313 other keys, used to derive the SIGNKEY and SEALKEY 

314 target_info: The AV_PAIR structure used in the nt response 

315 calculation 

316 """ 

317 self.signature = NTLM_SIGNATURE 

318 self.message_type = struct.pack('<L', MessageTypes.NTLM_AUTHENTICATE) 

319 self.negotiate_flags = challenge_message.negotiate_flags 

320 self.version = get_version(self.negotiate_flags) 

321 self.mic = None 

322 

323 if domain_name is None: 

324 self.domain_name = '' 

325 else: 

326 self.domain_name = domain_name 

327 

328 if workstation is None: 

329 self.workstation = '' 

330 else: 

331 self.workstation = workstation 

332 

333 if self.negotiate_flags & NegotiateFlags.NTLMSSP_NEGOTIATE_UNICODE: 

334 self.negotiate_flags &= ~NegotiateFlags.NTLMSSP_NEGOTIATE_OEM 

335 encoding_value = 'utf-16-le' 

336 else: 

337 encoding_value = 'ascii' 

338 

339 self.domain_name = self.domain_name.encode(encoding_value) 

340 self.user_name = user_name.encode(encoding_value) 

341 self.workstation = self.workstation.encode(encoding_value) 

342 

343 compute_response = ComputeResponse(user_name, password, domain_name, 

344 challenge_message, 

345 ntlm_compatibility) 

346 

347 self.lm_challenge_response = \ 

348 compute_response.get_lm_challenge_response() 

349 self.nt_challenge_response, key_exchange_key, target_info = \ 

350 compute_response.get_nt_challenge_response( 

351 self.lm_challenge_response, server_certificate_hash, cbt_data) 

352 self.target_info = target_info 

353 

354 if self.negotiate_flags & NegotiateFlags.NTLMSSP_NEGOTIATE_KEY_EXCH: 

355 self.exported_session_key = get_random_export_session_key() 

356 

357 rc4_handle = ARC4(key_exchange_key) 

358 self.encrypted_random_session_key = \ 

359 rc4_handle.update(self.exported_session_key) 

360 else: 

361 self.exported_session_key = key_exchange_key 

362 self.encrypted_random_session_key = b'' 

363 

364 self.negotiate_flags = struct.pack('<I', self.negotiate_flags) 

365 

366 def get_data(self): 

367 if self.mic is None: 

368 mic = b'' 

369 expected_body_length = self.EXPECTED_BODY_LENGTH 

370 else: 

371 mic = self.mic 

372 expected_body_length = self.EXPECTED_BODY_LENGTH_WITH_MIC 

373 

374 payload_offset = expected_body_length 

375 

376 # DomainNameFields - 8 bytes 

377 domain_name_len = struct.pack('<H', len(self.domain_name)) 

378 domain_name_max_len = struct.pack('<H', len(self.domain_name)) 

379 domain_name_buffer_offset = struct.pack('<I', payload_offset) 

380 payload_offset += len(self.domain_name) 

381 

382 # UserNameFields - 8 bytes 

383 user_name_len = struct.pack('<H', len(self.user_name)) 

384 user_name_max_len = struct.pack('<H', len(self.user_name)) 

385 user_name_buffer_offset = struct.pack('<I', payload_offset) 

386 payload_offset += len(self.user_name) 

387 

388 # WorkstatonFields - 8 bytes 

389 workstation_len = struct.pack('<H', len(self.workstation)) 

390 workstation_max_len = struct.pack('<H', len(self.workstation)) 

391 workstation_buffer_offset = struct.pack('<I', payload_offset) 

392 payload_offset += len(self.workstation) 

393 

394 # LmChallengeResponseFields - 8 bytes 

395 lm_challenge_response_len = \ 

396 struct.pack('<H', len(self.lm_challenge_response)) 

397 lm_challenge_response_max_len = \ 

398 struct.pack('<H', len(self.lm_challenge_response)) 

399 lm_challenge_response_buffer_offset = struct.pack('<I', payload_offset) 

400 payload_offset += len(self.lm_challenge_response) 

401 

402 # NtChallengeResponseFields - 8 bytes 

403 nt_challenge_response_len = \ 

404 struct.pack('<H', len(self.nt_challenge_response)) 

405 nt_challenge_response_max_len = \ 

406 struct.pack('<H', len(self.nt_challenge_response)) 

407 nt_challenge_response_buffer_offset = struct.pack('<I', payload_offset) 

408 payload_offset += len(self.nt_challenge_response) 

409 

410 # EncryptedRandomSessionKeyFields - 8 bytes 

411 encrypted_random_session_key_len = \ 

412 struct.pack('<H', len(self.encrypted_random_session_key)) 

413 encrypted_random_session_key_max_len = \ 

414 struct.pack('<H', len(self.encrypted_random_session_key)) 

415 encrypted_random_session_key_buffer_offset = \ 

416 struct.pack('<I', payload_offset) 

417 payload_offset += len(self.encrypted_random_session_key) 

418 

419 # Payload - variable length 

420 payload = self.domain_name 

421 payload += self.user_name 

422 payload += self.workstation 

423 payload += self.lm_challenge_response 

424 payload += self.nt_challenge_response 

425 payload += self.encrypted_random_session_key 

426 

427 msg3 = self.signature 

428 msg3 += self.message_type 

429 msg3 += lm_challenge_response_len 

430 msg3 += lm_challenge_response_max_len 

431 msg3 += lm_challenge_response_buffer_offset 

432 msg3 += nt_challenge_response_len 

433 msg3 += nt_challenge_response_max_len 

434 msg3 += nt_challenge_response_buffer_offset 

435 msg3 += domain_name_len 

436 msg3 += domain_name_max_len 

437 msg3 += domain_name_buffer_offset 

438 msg3 += user_name_len 

439 msg3 += user_name_max_len 

440 msg3 += user_name_buffer_offset 

441 msg3 += workstation_len 

442 msg3 += workstation_max_len 

443 msg3 += workstation_buffer_offset 

444 msg3 += encrypted_random_session_key_len 

445 msg3 += encrypted_random_session_key_max_len 

446 msg3 += encrypted_random_session_key_buffer_offset 

447 msg3 += self.negotiate_flags 

448 msg3 += self.version 

449 msg3 += mic 

450 

451 # Adding the payload data to the message 

452 msg3 += payload 

453 

454 return msg3 

455 

456 def add_mic(self, negotiate_message, challenge_message): 

457 if self.target_info is not None: 

458 av_flags = self.target_info[AvId.MSV_AV_FLAGS] 

459 

460 if av_flags is not None and av_flags == \ 

461 struct.pack("<L", AvFlags.MIC_PROVIDED): 

462 self.mic = struct.pack("<IIII", 0, 0, 0, 0) 

463 negotiate_data = negotiate_message.get_data() 

464 challenge_data = challenge_message.get_data() 

465 authenticate_data = self.get_data() 

466 

467 hmac_data = negotiate_data + challenge_data + authenticate_data 

468 mic = hmac.new(self.exported_session_key, hmac_data, 

469 digestmod=hashlib.md5).digest() 

470 self.mic = mic 

471 

472 

473def get_version(negotiate_flags): 

474 # Check the negotiate_flag version is set, if it is make sure the version 

475 # info is added to the data 

476 if negotiate_flags & NegotiateFlags.NTLMSSP_NEGOTIATE_VERSION: 

477 # TODO: Get the major and minor version of Windows instead of using 

478 # default values 

479 product_major_version = struct.pack('<B', 6) 

480 product_minor_version = struct.pack('<B', 1) 

481 product_build = struct.pack('<H', 7601) 

482 version_reserved = b'\x00' * 3 

483 ntlm_revision_current = struct.pack('<B', 15) 

484 version = product_major_version 

485 version += product_minor_version 

486 version += product_build 

487 version += version_reserved 

488 version += ntlm_revision_current 

489 else: 

490 version = b'\x00' * 8 

491 

492 return version 

493 

494 

495def get_random_export_session_key(): 

496 return os.urandom(16)