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

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

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

3 

4import binascii 

5import hashlib 

6import hmac 

7import struct 

8 

9import ntlm_auth.compute_keys as compkeys 

10 

11from ntlm_auth.constants import NegotiateFlags, SignSealConstants 

12from ntlm_auth.rc4 import ARC4 

13 

14 

15class _NtlmMessageSignature1(object): 

16 EXPECTED_BODY_LENGTH = 16 

17 

18 def __init__(self, random_pad, checksum, seq_num): 

19 """ 

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

21 

22 2.2.2.9.1 NTLMSSP_MESSAGE_SIGNATURE 

23 This version of the NTLMSSP_MESSAGE_SIGNATURE structure MUST be used 

24 when the NTLMSSP_NEGOTIATE_EXTENDED_SESSIONSECURITY flag is not 

25 negotiated. 

26 

27 :param random_pad: A 4-byte array that contains the random pad for the 

28 message 

29 :param checksum: A 4-byte array that contains the checksum for the 

30 message 

31 :param seq_num: A 32-bit unsigned integer that contains the NTLM 

32 sequence number for this application message 

33 """ 

34 self.version = b"\x01\x00\x00\x00" 

35 self.random_pad = random_pad 

36 self.checksum = checksum 

37 self.seq_num = seq_num 

38 

39 def get_data(self): 

40 signature = self.version 

41 signature += self.random_pad 

42 signature += self.checksum 

43 signature += self.seq_num 

44 

45 assert self.EXPECTED_BODY_LENGTH == len(signature), \ 

46 "BODY_LENGTH: %d != signature: %d" \ 

47 % (self.EXPECTED_BODY_LENGTH, len(signature)) 

48 

49 return signature 

50 

51 

52class _NtlmMessageSignature2(object): 

53 EXPECTED_BODY_LENGTH = 16 

54 

55 def __init__(self, checksum, seq_num): 

56 """ 

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

58 

59 2.2.2.9.2 NTLMSSP_MESSAGE_SIGNATURE for Extended Session Security 

60 This version of the NTLMSSP_MESSAGE_SIGNATURE structure MUST be used 

61 when the NTLMSSP_NEGOTIATE_EXTENDED_SESSIONSECURITY flag is negotiated 

62 

63 :param checksum: An 8-byte array that contains the checksum for the 

64 message 

65 :param seq_num: A 32-bit unsigned integer that contains the NTLM 

66 sequence number for this application message 

67 """ 

68 self.version = b"\x01\x00\x00\x00" 

69 self.checksum = checksum 

70 self.seq_num = seq_num 

71 

72 def get_data(self): 

73 signature = self.version 

74 signature += self.checksum 

75 signature += self.seq_num 

76 

77 assert self.EXPECTED_BODY_LENGTH == len(signature),\ 

78 "BODY_LENGTH: %d != signature: %d"\ 

79 % (self.EXPECTED_BODY_LENGTH, len(signature)) 

80 

81 return signature 

82 

83 

84class SessionSecurity(object): 

85 

86 def __init__(self, negotiate_flags, exported_session_key, source="client"): 

87 """ 

88 Initialises a security session context that can be used by libraries 

89 that call ntlm-auth to sign and seal messages send to the server as 

90 well as verify and unseal messages that have been received from the 

91 server. This is similar to the GSS_Wrap functions specified in the 

92 MS-NLMP document which does the same task. 

93 

94 :param negotiate_flags: The negotiate flag structure that has been 

95 negotiated with the server 

96 :param exported_session_key: A 128-bit session key used to derive 

97 signing and sealing keys 

98 :param source: The source of the message, only used in test scenarios 

99 when testing out a server sealing and unsealing 

100 """ 

101 self.negotiate_flags = negotiate_flags 

102 self.exported_session_key = exported_session_key 

103 self.outgoing_seq_num = 0 

104 self.incoming_seq_num = 0 

105 self._source = source 

106 self._client_sealing_key = compkeys.get_seal_key(self.negotiate_flags, exported_session_key, 

107 SignSealConstants.CLIENT_SEALING) 

108 self._server_sealing_key = compkeys.get_seal_key(self.negotiate_flags, exported_session_key, 

109 SignSealConstants.SERVER_SEALING) 

110 

111 self.outgoing_handle = None 

112 self.incoming_handle = None 

113 self.reset_rc4_state(True) 

114 self.reset_rc4_state(False) 

115 

116 if source == "client": 

117 self.outgoing_signing_key = compkeys.get_sign_key(exported_session_key, SignSealConstants.CLIENT_SIGNING) 

118 self.incoming_signing_key = compkeys.get_sign_key(exported_session_key, SignSealConstants.SERVER_SIGNING) 

119 elif source == "server": 

120 self.outgoing_signing_key = compkeys.get_sign_key(exported_session_key, SignSealConstants.SERVER_SIGNING) 

121 self.incoming_signing_key = compkeys.get_sign_key(exported_session_key, SignSealConstants.CLIENT_SIGNING) 

122 else: 

123 raise ValueError("Invalid source parameter %s, must be client " 

124 "or server" % source) 

125 

126 def reset_rc4_state(self, outgoing=True): 

127 csk = self._client_sealing_key 

128 ssk = self._server_sealing_key 

129 if outgoing: 

130 self.outgoing_handle = ARC4(csk if self._source == 'client' else ssk) 

131 else: 

132 self.incoming_handle = ARC4(ssk if self._source == 'client' else csk) 

133 

134 def wrap(self, message): 

135 """ 

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

137 

138 3.4.6 GSS_WrapEx() 

139 Emulates the GSS_Wrap() implementation to sign and seal messages if the 

140 correct flags are set. 

141 

142 :param message: The message data that will be wrapped 

143 :return message: The message that has been sealed if flags are set 

144 :return signature: The signature of the message, None if flags are not 

145 set 

146 """ 

147 if self.negotiate_flags & NegotiateFlags.NTLMSSP_NEGOTIATE_SEAL: 

148 encrypted_message = self._seal_message(message) 

149 signature = self.get_signature(message) 

150 message = encrypted_message 

151 

152 elif self.negotiate_flags & NegotiateFlags.NTLMSSP_NEGOTIATE_SIGN: 

153 signature = self.get_signature(message) 

154 else: 

155 signature = None 

156 

157 return message, signature 

158 

159 def unwrap(self, message, signature): 

160 """ 

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

162 

163 3.4.7 GSS_UnwrapEx() 

164 Emulates the GSS_Unwrap() implementation to unseal messages and verify 

165 the signature sent matches what has been computed locally. Will throw 

166 an Exception if the signature doesn't match 

167 

168 :param message: The message data received from the server 

169 :param signature: The signature of the message 

170 :return message: The message that has been unsealed if flags are set 

171 """ 

172 if self.negotiate_flags & NegotiateFlags.NTLMSSP_NEGOTIATE_SEAL: 

173 message = self._unseal_message(message) 

174 self.verify_signature(message, signature) 

175 

176 elif self.negotiate_flags & NegotiateFlags.NTLMSSP_NEGOTIATE_SIGN: 

177 self.verify_signature(message, signature) 

178 

179 return message 

180 

181 def _seal_message(self, message): 

182 """ 

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

184 

185 3.4.3 Message Confidentiality 

186 Will generate an encrypted message using RC4 based on the 

187 ClientSealingKey 

188 

189 :param message: The message to be sealed (encrypted) 

190 :return encrypted_message: The encrypted message 

191 """ 

192 encrypted_message = self.outgoing_handle.update(message) 

193 return encrypted_message 

194 

195 def _unseal_message(self, message): 

196 """ 

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

198 

199 3.4.3 Message Confidentiality 

200 Will generate a dencrypted message using RC4 based on the 

201 ServerSealingKey 

202 

203 :param message: The message to be unsealed (dencrypted) 

204 :return decrypted_message: The decrypted message 

205 """ 

206 decrypted_message = self.incoming_handle.update(message) 

207 return decrypted_message 

208 

209 def get_signature(self, message): 

210 """ 

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

212 

213 3.4.4 Message Signature Functions 

214 Will create the signature based on the message to send to the server. 

215 Depending on the negotiate_flags set this could either be an NTLMv1 

216 signature or NTLMv2 with Extended Session Security signature. 

217 

218 :param message: The message data that will be signed 

219 :return signature: Either _NtlmMessageSignature1 or 

220 _NtlmMessageSignature2 depending on the flags set 

221 """ 

222 signature = calc_signature(message, self.negotiate_flags, 

223 self.outgoing_signing_key, 

224 self.outgoing_seq_num, self.outgoing_handle) 

225 self.outgoing_seq_num += 1 

226 

227 return signature.get_data() 

228 

229 def verify_signature(self, message, signature): 

230 """ 

231 Will verify that the signature received from the server matches up with 

232 the expected signature computed locally. Will throw an exception if 

233 they do not match 

234 

235 :param message: The message data that is received from the server 

236 :param signature: The signature of the message received from the server 

237 """ 

238 if self.negotiate_flags & \ 

239 NegotiateFlags.NTLMSSP_NEGOTIATE_EXTENDED_SESSIONSECURITY: 

240 actual_checksum = signature[4:12] 

241 actual_seq_num = struct.unpack("<I", signature[12:16])[0] 

242 else: 

243 actual_checksum = signature[8:12] 

244 actual_seq_num = struct.unpack("<I", signature[12:16])[0] 

245 

246 expected_signature = calc_signature(message, self.negotiate_flags, 

247 self.incoming_signing_key, 

248 self.incoming_seq_num, 

249 self.incoming_handle) 

250 expected_checksum = expected_signature.checksum 

251 expected_seq_num = struct.unpack("<I", expected_signature.seq_num)[0] 

252 

253 if actual_checksum != expected_checksum: 

254 raise Exception("The signature checksum does not match, message " 

255 "has been altered") 

256 

257 if actual_seq_num != expected_seq_num: 

258 raise Exception("The signature sequence number does not match up, " 

259 "message not received in the correct sequence") 

260 

261 self.incoming_seq_num += 1 

262 

263 

264def calc_signature(message, negotiate_flags, signing_key, seq_num, handle): 

265 seq_num = struct.pack("<I", seq_num) 

266 if negotiate_flags & \ 

267 NegotiateFlags.NTLMSSP_NEGOTIATE_EXTENDED_SESSIONSECURITY: 

268 checksum_hmac = hmac.new(signing_key, seq_num + message, 

269 digestmod=hashlib.md5) 

270 if negotiate_flags & NegotiateFlags.NTLMSSP_NEGOTIATE_KEY_EXCH: 

271 checksum = handle.update(checksum_hmac.digest()[:8]) 

272 else: 

273 checksum = checksum_hmac.digest()[:8] 

274 

275 signature = _NtlmMessageSignature2(checksum, seq_num) 

276 

277 else: 

278 message_crc = binascii.crc32(message) % (1 << 32) 

279 checksum = struct.pack("<I", message_crc) 

280 random_pad = handle.update(struct.pack("<I", 0)) 

281 checksum = handle.update(checksum) 

282 seq_num = handle.update(seq_num) 

283 random_pad = struct.pack("<I", 0) 

284 

285 signature = _NtlmMessageSignature1(random_pad, checksum, seq_num) 

286 

287 return signature