Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/ntlm_auth/session_security.py: 22%
115 statements
« prev ^ index » next coverage.py v7.3.1, created at 2023-09-25 07:03 +0000
1# Copyright: (c) 2018, Jordan Borean (@jborean93) <jborean93@gmail.com>
2# MIT License (see LICENSE or https://opensource.org/licenses/MIT)
4import binascii
5import hashlib
6import hmac
7import struct
9import ntlm_auth.compute_keys as compkeys
11from ntlm_auth.constants import NegotiateFlags, SignSealConstants
12from ntlm_auth.rc4 import ARC4
15class _NtlmMessageSignature1(object):
16 EXPECTED_BODY_LENGTH = 16
18 def __init__(self, random_pad, checksum, seq_num):
19 """
20 [MS-NLMP] v28.0 2016-07-14
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.
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
39 def get_data(self):
40 signature = self.version
41 signature += self.random_pad
42 signature += self.checksum
43 signature += self.seq_num
45 assert self.EXPECTED_BODY_LENGTH == len(signature), \
46 "BODY_LENGTH: %d != signature: %d" \
47 % (self.EXPECTED_BODY_LENGTH, len(signature))
49 return signature
52class _NtlmMessageSignature2(object):
53 EXPECTED_BODY_LENGTH = 16
55 def __init__(self, checksum, seq_num):
56 """
57 [MS-NLMP] v28.0 2016-07-14
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
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
72 def get_data(self):
73 signature = self.version
74 signature += self.checksum
75 signature += self.seq_num
77 assert self.EXPECTED_BODY_LENGTH == len(signature),\
78 "BODY_LENGTH: %d != signature: %d"\
79 % (self.EXPECTED_BODY_LENGTH, len(signature))
81 return signature
84class SessionSecurity(object):
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.
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)
111 self.outgoing_handle = None
112 self.incoming_handle = None
113 self.reset_rc4_state(True)
114 self.reset_rc4_state(False)
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)
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)
134 def wrap(self, message):
135 """
136 [MS-NLMP] v28.0 2016-07-14
138 3.4.6 GSS_WrapEx()
139 Emulates the GSS_Wrap() implementation to sign and seal messages if the
140 correct flags are set.
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
152 elif self.negotiate_flags & NegotiateFlags.NTLMSSP_NEGOTIATE_SIGN:
153 signature = self.get_signature(message)
154 else:
155 signature = None
157 return message, signature
159 def unwrap(self, message, signature):
160 """
161 [MS-NLMP] v28.0 2016-07-14
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
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)
176 elif self.negotiate_flags & NegotiateFlags.NTLMSSP_NEGOTIATE_SIGN:
177 self.verify_signature(message, signature)
179 return message
181 def _seal_message(self, message):
182 """
183 [MS-NLMP] v28.0 2016-07-14
185 3.4.3 Message Confidentiality
186 Will generate an encrypted message using RC4 based on the
187 ClientSealingKey
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
195 def _unseal_message(self, message):
196 """
197 [MS-NLMP] v28.0 2016-07-14
199 3.4.3 Message Confidentiality
200 Will generate a dencrypted message using RC4 based on the
201 ServerSealingKey
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
209 def get_signature(self, message):
210 """
211 [MS-NLMP] v28.0 2016-07-14
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.
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
227 return signature.get_data()
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
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]
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]
253 if actual_checksum != expected_checksum:
254 raise Exception("The signature checksum does not match, message "
255 "has been altered")
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")
261 self.incoming_seq_num += 1
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]
275 signature = _NtlmMessageSignature2(checksum, seq_num)
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)
285 signature = _NtlmMessageSignature1(random_pad, checksum, seq_num)
287 return signature