Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/scapy/layers/ntlm.py: 35%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# SPDX-License-Identifier: GPL-2.0-only
2# This file is part of Scapy
3# See https://scapy.net/ for more information
4# Copyright (C) Gabriel Potter
6"""
7NTLM
9This is documented in [MS-NLMP]
11.. note::
12 You will find more complete documentation for this layer over at
13 `GSSAPI <https://scapy.readthedocs.io/en/latest/layers/gssapi.html#ntlm>`_
14"""
16import copy
17import time
18import os
19import struct
21from enum import IntEnum
23from scapy.asn1.asn1 import ASN1_Codecs
24from scapy.asn1.mib import conf # loads conf.mib
25from scapy.asn1fields import (
26 ASN1F_OID,
27 ASN1F_PRINTABLE_STRING,
28 ASN1F_SEQUENCE,
29 ASN1F_SEQUENCE_OF,
30)
31from scapy.asn1packet import ASN1_Packet
32from scapy.config import crypto_validator
33from scapy.compat import bytes_base64
34from scapy.error import log_runtime
35from scapy.fields import (
36 ByteEnumField,
37 ByteField,
38 ConditionalField,
39 Field,
40 FieldLenField,
41 FlagsField,
42 LEIntEnumField,
43 LEIntField,
44 LEShortEnumField,
45 LEShortField,
46 LEThreeBytesField,
47 MultipleTypeField,
48 PacketField,
49 PacketListField,
50 StrField,
51 StrFieldUtf16,
52 StrFixedLenField,
53 StrLenFieldUtf16,
54 UTCTimeField,
55 XStrField,
56 XStrFixedLenField,
57 XStrLenField,
58 _StrField,
59)
60from scapy.packet import Packet
61from scapy.sessions import StringBuffer
63from scapy.layers.gssapi import (
64 _GSSAPI_OIDS,
65 _GSSAPI_SIGNATURE_OIDS,
66 GSS_C_FLAGS,
67 GSS_C_NO_CHANNEL_BINDINGS,
68 GSS_S_BAD_BINDINGS,
69 GSS_S_COMPLETE,
70 GSS_S_CONTINUE_NEEDED,
71 GSS_S_DEFECTIVE_CREDENTIAL,
72 GSS_S_DEFECTIVE_TOKEN,
73 GSS_S_FLAGS,
74 GssChannelBindings,
75 SSP,
76)
78# Typing imports
79from typing import (
80 Any,
81 Callable,
82 List,
83 Optional,
84 Tuple,
85 Union,
86)
88# Crypto imports
90from scapy.layers.tls.crypto.hash import Hash_MD4, Hash_MD5
91from scapy.layers.tls.crypto.h_mac import Hmac_MD5
93##########
94# Fields #
95##########
98# NTLM structures are all in all very complicated. Many fields don't have a fixed
99# position, but are rather referred to with an offset (from the beginning of the
100# structure) and a length. In addition to that, there are variants of the structure
101# with missing fields when running old versions of Windows (sometimes also seen when
102# talking to products that reimplement NTLM, most notably backup applications).
104# We add `_NTLMPayloadField` and `_NTLMPayloadPacket` to parse fields that use an
105# offset, and `_NTLM_post_build` to be able to rebuild those offsets.
106# In addition, the `NTLM_VARIANT*` allows to select what flavor of NTLM to use
107# (NT, XP, or Recent). But in real world use only Recent should be used.
110class _NTLMPayloadField(_StrField[List[Tuple[str, Any]]]):
111 """Special field used to dissect NTLM payloads.
112 This isn't trivial because the offsets are variable."""
114 __slots__ = [
115 "fields",
116 "fields_map",
117 "fields_pad",
118 "offset",
119 "length_from",
120 "force_order",
121 "offset_name",
122 ]
123 islist = True
125 def __init__(
126 self,
127 name, # type: str
128 offset, # type: Union[int, Callable[[Packet], int]]
129 fields, # type: List[Field[Any, Any]]
130 fields_pad=0, # type: int
131 length_from=None, # type: Optional[Callable[[Packet], int]]
132 force_order=None, # type: Optional[List[str]]
133 offset_name="BufferOffset", # type: str
134 ):
135 # type: (...) -> None
136 self.offset = offset
137 self.fields = fields
138 self.fields_map = {field.name: field for field in fields}
139 self.length_from = length_from
140 self.force_order = force_order # whether the order of fields is fixed
141 self.offset_name = offset_name
142 self.fields_pad = fields_pad
143 super(_NTLMPayloadField, self).__init__(
144 name,
145 [
146 (field.name, field.default)
147 for field in fields
148 if field.default is not None
149 ],
150 )
152 def _on_payload(self, pkt, x, func):
153 # type: (Optional[Packet], bytes, str) -> List[Tuple[str, Any]]
154 if not pkt or not x:
155 return []
156 results = []
157 for field_name, value in x:
158 if field_name not in self.fields_map:
159 continue
160 if not isinstance(
161 self.fields_map[field_name], PacketListField
162 ) and not isinstance(value, Packet):
163 value = getattr(self.fields_map[field_name], func)(pkt, value)
164 results.append((field_name, value))
165 return results
167 def i2h(self, pkt, x):
168 # type: (Optional[Packet], bytes) -> List[Tuple[str, str]]
169 return self._on_payload(pkt, x, "i2h")
171 def h2i(self, pkt, x):
172 # type: (Optional[Packet], bytes) -> List[Tuple[str, str]]
173 return self._on_payload(pkt, x, "h2i")
175 def i2repr(self, pkt, x):
176 # type: (Optional[Packet], bytes) -> str
177 return repr(self._on_payload(pkt, x, "i2repr"))
179 def _o_pkt(self, pkt):
180 # type: (Optional[Packet]) -> int
181 if callable(self.offset):
182 return self.offset(pkt)
183 return self.offset
185 def addfield(self, pkt, s, val):
186 # type: (Optional[Packet], bytes, Optional[List[Tuple[str, str]]]) -> bytes
187 # Create string buffer
188 buf = StringBuffer()
189 buf.append(s, 1)
190 # Calc relative offset
191 r_off = self._o_pkt(pkt) - len(s)
192 if self.force_order:
193 val.sort(key=lambda x: self.force_order.index(x[0]))
194 for field_name, value in val:
195 if field_name not in self.fields_map:
196 continue
197 field = self.fields_map[field_name]
198 offset = pkt.getfieldval(field_name + self.offset_name)
199 if offset is None:
200 # No offset specified: calc
201 offset = len(buf)
202 if self.fields_pad:
203 pad = (-offset) % self.fields_pad
204 offset += pad
205 buf.append(pad * b"\x00", len(buf))
206 else:
207 # Calc relative offset
208 offset -= r_off
209 pad = offset + 1 - len(buf)
210 # Add padding if necessary
211 if pad > 0:
212 buf.append(pad * b"\x00", len(buf))
213 buf.append(field.addfield(pkt, bytes(buf), value)[len(buf) :], offset + 1)
214 return bytes(buf)
216 def getfield(self, pkt, s):
217 # type: (Packet, bytes) -> Tuple[bytes, List[Tuple[str, str]]]
218 if self.length_from is None:
219 ret, remain = b"", s
220 else:
221 len_pkt = self.length_from(pkt)
222 ret, remain = s[len_pkt:], s[:len_pkt]
223 if not pkt or not remain:
224 return s, []
225 results = []
226 max_offset = 0
227 o_pkt = self._o_pkt(pkt)
228 offsets = [
229 pkt.getfieldval(x.name + self.offset_name) - o_pkt for x in self.fields
230 ]
231 for i, field in enumerate(self.fields):
232 offset = offsets[i]
233 try:
234 length = pkt.getfieldval(field.name + "Len")
235 except AttributeError:
236 length = len(remain) - offset
237 # length can't be greater than the difference with the next offset
238 try:
239 length = min(length, min(x - offset for x in offsets if x > offset))
240 except ValueError:
241 pass
242 if offset < 0:
243 continue
244 max_offset = max(offset + length, max_offset)
245 if remain[offset : offset + length]:
246 results.append(
247 (
248 offset,
249 field.name,
250 field.getfield(pkt, remain[offset : offset + length])[1],
251 )
252 )
253 ret += remain[max_offset:]
254 results.sort(key=lambda x: x[0])
255 return ret, [x[1:] for x in results]
258class _NTLMPayloadPacket(Packet):
259 _NTLM_PAYLOAD_FIELD_NAME = "Payload"
261 def __init__(
262 self,
263 _pkt=b"", # type: Union[bytes, bytearray]
264 post_transform=None, # type: Any
265 _internal=0, # type: int
266 _underlayer=None, # type: Optional[Packet]
267 _parent=None, # type: Optional[Packet]
268 **fields, # type: Any
269 ):
270 # pop unknown fields. We can't process them until the packet is initialized
271 unknown = {
272 k: fields.pop(k)
273 for k in list(fields)
274 if not any(k == f.name for f in self.fields_desc)
275 }
276 super(_NTLMPayloadPacket, self).__init__(
277 _pkt=_pkt,
278 post_transform=post_transform,
279 _internal=_internal,
280 _underlayer=_underlayer,
281 _parent=_parent,
282 **fields,
283 )
284 # check unknown fields for implicit ones
285 local_fields = next(
286 [y.name for y in x.fields]
287 for x in self.fields_desc
288 if x.name == self._NTLM_PAYLOAD_FIELD_NAME
289 )
290 implicit_fields = {k: v for k, v in unknown.items() if k in local_fields}
291 for k, value in implicit_fields.items():
292 self.setfieldval(k, value)
294 def getfieldval(self, attr):
295 # Ease compatibility with _NTLMPayloadField
296 try:
297 return super(_NTLMPayloadPacket, self).getfieldval(attr)
298 except AttributeError:
299 try:
300 return next(
301 x[1]
302 for x in super(_NTLMPayloadPacket, self).getfieldval(
303 self._NTLM_PAYLOAD_FIELD_NAME
304 )
305 if x[0] == attr
306 )
307 except StopIteration:
308 raise AttributeError(attr)
310 def getfield_and_val(self, attr):
311 # Ease compatibility with _NTLMPayloadField
312 try:
313 return super(_NTLMPayloadPacket, self).getfield_and_val(attr)
314 except ValueError:
315 PayFields = self.get_field(self._NTLM_PAYLOAD_FIELD_NAME).fields_map
316 try:
317 return (
318 PayFields[attr],
319 PayFields[attr].h2i( # cancel out the i2h.. it's dumb i know
320 self,
321 next(
322 x[1]
323 for x in super(_NTLMPayloadPacket, self).__getattr__(
324 self._NTLM_PAYLOAD_FIELD_NAME
325 )
326 if x[0] == attr
327 ),
328 ),
329 )
330 except (StopIteration, KeyError):
331 raise ValueError(attr)
333 def setfieldval(self, attr, val):
334 # Ease compatibility with _NTLMPayloadField
335 try:
336 return super(_NTLMPayloadPacket, self).setfieldval(attr, val)
337 except AttributeError:
338 Payload = super(_NTLMPayloadPacket, self).__getattr__(
339 self._NTLM_PAYLOAD_FIELD_NAME
340 )
341 if attr not in self.get_field(self._NTLM_PAYLOAD_FIELD_NAME).fields_map:
342 raise AttributeError(attr)
343 try:
344 Payload.pop(
345 next(
346 i
347 for i, x in enumerate(
348 super(_NTLMPayloadPacket, self).__getattr__(
349 self._NTLM_PAYLOAD_FIELD_NAME
350 )
351 )
352 if x[0] == attr
353 )
354 )
355 except StopIteration:
356 pass
357 Payload.append([attr, val])
358 super(_NTLMPayloadPacket, self).setfieldval(
359 self._NTLM_PAYLOAD_FIELD_NAME, Payload
360 )
363class _NTLM_ENUM(IntEnum):
364 LEN = 0x0001
365 MAXLEN = 0x0002
366 OFFSET = 0x0004
367 COUNT = 0x0008
368 PAD8 = 0x1000
371_NTLM_CONFIG = [
372 ("Len", _NTLM_ENUM.LEN),
373 ("MaxLen", _NTLM_ENUM.MAXLEN),
374 ("BufferOffset", _NTLM_ENUM.OFFSET),
375]
378def _NTLM_post_build(self, p, pay_offset, fields, config=_NTLM_CONFIG):
379 """Util function to build the offset and populate the lengths"""
380 gl_fld = self.get_field(self._NTLM_PAYLOAD_FIELD_NAME)
381 for field_name, value in self.fields[self._NTLM_PAYLOAD_FIELD_NAME]:
382 fld = gl_fld.fields_map[field_name]
383 length = fld.i2len(self, value)
384 count = fld.i2count(self, value)
385 offset = fields[field_name]
386 i = 0
387 r = lambda y: {2: "H", 4: "I", 8: "Q"}[y]
388 for fname, ftype in config:
389 if isinstance(ftype, dict):
390 ftype = ftype[field_name]
391 if ftype & _NTLM_ENUM.LEN:
392 fval = length
393 elif ftype & _NTLM_ENUM.OFFSET:
394 fval = pay_offset
395 elif ftype & _NTLM_ENUM.MAXLEN:
396 fval = length
397 elif ftype & _NTLM_ENUM.COUNT:
398 fval = count
399 else:
400 raise ValueError
401 if ftype & _NTLM_ENUM.PAD8:
402 pad = (-fval) % 8
403 fval += pad
404 pay_offset += pad
405 sz = self.get_field(field_name + fname).sz
406 if self.getfieldval(field_name + fname) is None:
407 p = (
408 p[: offset + i]
409 + struct.pack("<%s" % r(sz), fval)
410 + p[offset + i + sz :]
411 )
412 i += sz
413 pay_offset += length
414 return p
417##############
418# Structures #
419##############
422# -- Util: VARIANT class
425class NTLM_VARIANT(IntEnum):
426 """
427 The message variant to use for NTLM.
428 """
430 NT_OR_2000 = 0
431 XP_OR_2003 = 1
432 RECENT = 2
435class _NTLM_VARIANT_Packet(_NTLMPayloadPacket):
436 def __init__(self, *args, **kwargs):
437 self.VARIANT = kwargs.pop("VARIANT", NTLM_VARIANT.RECENT)
438 super(_NTLM_VARIANT_Packet, self).__init__(*args, **kwargs)
440 def clone_with(self, *args, **kwargs):
441 pkt = super(_NTLM_VARIANT_Packet, self).clone_with(*args, **kwargs)
442 pkt.VARIANT = self.VARIANT
443 return pkt
445 def copy(self):
446 pkt = super(_NTLM_VARIANT_Packet, self).copy()
447 pkt.VARIANT = self.VARIANT
449 return pkt
451 def show2(self, dump=False, indent=3, lvl="", label_lvl=""):
452 return self.__class__(bytes(self), VARIANT=self.VARIANT).show(
453 dump, indent, lvl, label_lvl
454 )
457# Sect 2.2
460class NTLM_Header(Packet):
461 name = "NTLM Header"
462 fields_desc = [
463 StrFixedLenField("Signature", b"NTLMSSP\0", length=8),
464 LEIntEnumField(
465 "MessageType",
466 3,
467 {
468 1: "NEGOTIATE_MESSAGE",
469 2: "CHALLENGE_MESSAGE",
470 3: "AUTHENTICATE_MESSAGE",
471 },
472 ),
473 ]
475 @classmethod
476 def dispatch_hook(cls, _pkt=None, *args, **kargs):
477 if cls is NTLM_Header and _pkt and len(_pkt) >= 10:
478 MessageType = struct.unpack("<H", _pkt[8:10])[0]
479 if MessageType == 1:
480 return NTLM_NEGOTIATE
481 elif MessageType == 2:
482 return NTLM_CHALLENGE
483 elif MessageType == 3:
484 return NTLM_AUTHENTICATE_V2
485 return cls
488# Sect 2.2.2.5
489_negotiateFlags = [
490 "NEGOTIATE_UNICODE", # A
491 "NEGOTIATE_OEM", # B
492 "REQUEST_TARGET", # C
493 "r10",
494 "NEGOTIATE_SIGN", # D
495 "NEGOTIATE_SEAL", # E
496 "NEGOTIATE_DATAGRAM", # F
497 "NEGOTIATE_LM_KEY", # G
498 "r9",
499 "NEGOTIATE_NTLM", # H
500 "r8",
501 "J",
502 "NEGOTIATE_OEM_DOMAIN_SUPPLIED", # K
503 "NEGOTIATE_OEM_WORKSTATION_SUPPLIED", # L
504 "NEGOTIATE_LOCAL_CALL",
505 "NEGOTIATE_ALWAYS_SIGN", # M
506 "TARGET_TYPE_DOMAIN", # N
507 "TARGET_TYPE_SERVER", # O
508 "r6",
509 "NEGOTIATE_EXTENDED_SESSIONSECURITY", # P
510 "NEGOTIATE_IDENTIFY", # Q
511 "r5",
512 "REQUEST_NON_NT_SESSION_KEY", # R
513 "NEGOTIATE_TARGET_INFO", # S
514 "r4",
515 "NEGOTIATE_VERSION", # T
516 "r3",
517 "r2",
518 "r1",
519 "NEGOTIATE_128", # U
520 "NEGOTIATE_KEY_EXCH", # V
521 "NEGOTIATE_56", # W
522]
525def _NTLMStrField(name, default):
526 return MultipleTypeField(
527 [
528 (
529 StrFieldUtf16(name, default),
530 lambda pkt: pkt.NegotiateFlags.NEGOTIATE_UNICODE,
531 )
532 ],
533 StrField(name, default),
534 )
537# Sect 2.2.2.10
540class _NTLM_Version(Packet):
541 fields_desc = [
542 ByteField("ProductMajorVersion", 0),
543 ByteField("ProductMinorVersion", 0),
544 LEShortField("ProductBuild", 0),
545 LEThreeBytesField("res_ver", 0),
546 ByteEnumField("NTLMRevisionCurrent", 0x0F, {0x0F: "v15"}),
547 ]
550# Sect 2.2.1.1
553class NTLM_NEGOTIATE(_NTLM_VARIANT_Packet, NTLM_Header):
554 name = "NTLM Negotiate"
555 __slots__ = ["VARIANT"]
556 MessageType = 1
557 OFFSET = lambda pkt: (
558 32
559 if (
560 pkt.VARIANT == NTLM_VARIANT.NT_OR_2000
561 or (pkt.DomainNameBufferOffset or 40) <= 32
562 )
563 else 40
564 )
565 fields_desc = (
566 [
567 NTLM_Header,
568 FlagsField("NegotiateFlags", 0, -32, _negotiateFlags),
569 # DomainNameFields
570 LEShortField("DomainNameLen", None),
571 LEShortField("DomainNameMaxLen", None),
572 LEIntField("DomainNameBufferOffset", None),
573 # WorkstationFields
574 LEShortField("WorkstationNameLen", None),
575 LEShortField("WorkstationNameMaxLen", None),
576 LEIntField("WorkstationNameBufferOffset", None),
577 ]
578 + [
579 # VERSION
580 ConditionalField(
581 # (not present on some old Windows versions. We use a heuristic)
582 x,
583 lambda pkt: pkt.VARIANT >= NTLM_VARIANT.XP_OR_2003
584 and (
585 (
586 (
587 40
588 if pkt.DomainNameBufferOffset is None
589 else pkt.DomainNameBufferOffset or len(pkt.original or b"")
590 )
591 > 32
592 )
593 or pkt.fields.get(x.name, b"")
594 ),
595 )
596 for x in _NTLM_Version.fields_desc
597 ]
598 + [
599 # Payload
600 _NTLMPayloadField(
601 "Payload",
602 OFFSET,
603 [
604 # "MUST be encoded using the OEM character set"
605 StrField("DomainName", b""),
606 StrField("WorkstationName", b""),
607 ],
608 ),
609 ]
610 )
612 def post_build(self, pkt, pay):
613 # type: (bytes, bytes) -> bytes
614 return (
615 _NTLM_post_build(
616 self,
617 pkt,
618 self.OFFSET(),
619 {
620 "DomainName": 16,
621 "WorkstationName": 24,
622 },
623 )
624 + pay
625 )
628# Challenge
631class Single_Host_Data(Packet):
632 fields_desc = [
633 LEIntField("Size", None),
634 LEIntField("Z4", 0),
635 # "CustomData" guessed using LSAP_TOKEN_INFO_INTEGRITY.
636 FlagsField(
637 "Flags",
638 0,
639 -32,
640 {
641 0x00000001: "UAC-Restricted",
642 },
643 ),
644 LEIntEnumField(
645 "TokenIL",
646 0x00002000,
647 {
648 0x00000000: "Untrusted",
649 0x00001000: "Low",
650 0x00002000: "Medium",
651 0x00003000: "High",
652 0x00004000: "System",
653 0x00005000: "Protected process",
654 },
655 ),
656 XStrFixedLenField("MachineID", b"", length=32),
657 # KB 5068222 - still waiting for [MS-KILE] update (oct. 2025)
658 ConditionalField(
659 XStrFixedLenField("PermanentMachineID", None, length=32),
660 lambda pkt: pkt.Size is None or pkt.Size > 48,
661 ),
662 ]
664 def post_build(self, pkt, pay):
665 if self.Size is None:
666 pkt = struct.pack("<I", len(pkt)) + pkt[4:]
667 return pkt + pay
669 def default_payload_class(self, payload):
670 return conf.padding_layer
673class AV_PAIR(Packet):
674 name = "NTLM AV Pair"
675 fields_desc = [
676 LEShortEnumField(
677 "AvId",
678 0,
679 {
680 0x0000: "MsvAvEOL",
681 0x0001: "MsvAvNbComputerName",
682 0x0002: "MsvAvNbDomainName",
683 0x0003: "MsvAvDnsComputerName",
684 0x0004: "MsvAvDnsDomainName",
685 0x0005: "MsvAvDnsTreeName",
686 0x0006: "MsvAvFlags",
687 0x0007: "MsvAvTimestamp",
688 0x0008: "MsvAvSingleHost",
689 0x0009: "MsvAvTargetName",
690 0x000A: "MsvAvChannelBindings",
691 },
692 ),
693 FieldLenField("AvLen", None, length_of="Value", fmt="<H"),
694 MultipleTypeField(
695 [
696 (
697 LEIntEnumField(
698 "Value",
699 1,
700 {
701 0x0001: "constrained",
702 0x0002: "MIC integrity",
703 0x0004: "SPN from untrusted source",
704 },
705 ),
706 lambda pkt: pkt.AvId == 0x0006,
707 ),
708 (
709 UTCTimeField(
710 "Value",
711 None,
712 epoch=[1601, 1, 1, 0, 0, 0],
713 custom_scaling=1e7,
714 fmt="<Q",
715 ),
716 lambda pkt: pkt.AvId == 0x0007,
717 ),
718 (
719 PacketField("Value", Single_Host_Data(), Single_Host_Data),
720 lambda pkt: pkt.AvId == 0x0008,
721 ),
722 (
723 XStrLenField("Value", b"", length_from=lambda pkt: pkt.AvLen),
724 lambda pkt: pkt.AvId == 0x000A,
725 ),
726 ],
727 StrLenFieldUtf16("Value", b"", length_from=lambda pkt: pkt.AvLen),
728 ),
729 ]
731 def default_payload_class(self, payload):
732 return conf.padding_layer
735class NTLM_CHALLENGE(_NTLM_VARIANT_Packet, NTLM_Header):
736 name = "NTLM Challenge"
737 __slots__ = ["VARIANT"]
738 MessageType = 2
739 OFFSET = lambda pkt: (
740 48
741 if (
742 pkt.VARIANT == NTLM_VARIANT.NT_OR_2000
743 or (pkt.TargetInfoBufferOffset or 56) <= 48
744 )
745 else 56
746 )
747 fields_desc = (
748 [
749 NTLM_Header,
750 # TargetNameFields
751 LEShortField("TargetNameLen", None),
752 LEShortField("TargetNameMaxLen", None),
753 LEIntField("TargetNameBufferOffset", None),
754 #
755 FlagsField("NegotiateFlags", 0, -32, _negotiateFlags),
756 XStrFixedLenField("ServerChallenge", None, length=8),
757 XStrFixedLenField("Reserved", None, length=8),
758 # TargetInfoFields
759 LEShortField("TargetInfoLen", None),
760 LEShortField("TargetInfoMaxLen", None),
761 LEIntField("TargetInfoBufferOffset", None),
762 ]
763 + [
764 # VERSION
765 ConditionalField(
766 # (not present on some old Windows versions. We use a heuristic)
767 x,
768 lambda pkt: pkt.VARIANT >= NTLM_VARIANT.XP_OR_2003
769 and (
770 ((pkt.TargetInfoBufferOffset or 56) > 40)
771 or pkt.fields.get(x.name, b"")
772 ),
773 )
774 for x in _NTLM_Version.fields_desc
775 ]
776 + [
777 # Payload
778 _NTLMPayloadField(
779 "Payload",
780 OFFSET,
781 [
782 _NTLMStrField("TargetName", b""),
783 PacketListField("TargetInfo", [AV_PAIR()], AV_PAIR),
784 ],
785 ),
786 ]
787 )
789 def getAv(self, AvId):
790 try:
791 return next(x for x in self.TargetInfo if x.AvId == AvId)
792 except (StopIteration, AttributeError):
793 raise IndexError
795 def post_build(self, pkt, pay):
796 # type: (bytes, bytes) -> bytes
797 return (
798 _NTLM_post_build(
799 self,
800 pkt,
801 self.OFFSET(),
802 {
803 "TargetName": 12,
804 "TargetInfo": 40,
805 },
806 )
807 + pay
808 )
811# Authenticate
814class LM_RESPONSE(Packet):
815 fields_desc = [
816 StrFixedLenField("Response", b"", length=24),
817 ]
820class LMv2_RESPONSE(Packet):
821 fields_desc = [
822 StrFixedLenField("Response", b"", length=16),
823 StrFixedLenField("ChallengeFromClient", b"", length=8),
824 ]
827class NTLM_RESPONSE(Packet):
828 fields_desc = [
829 StrFixedLenField("Response", b"", length=24),
830 ]
833class NTLMv2_CLIENT_CHALLENGE(Packet):
834 fields_desc = [
835 ByteField("RespType", 1),
836 ByteField("HiRespType", 1),
837 LEShortField("Reserved1", 0),
838 LEIntField("Reserved2", 0),
839 UTCTimeField(
840 "TimeStamp", None, fmt="<Q", epoch=[1601, 1, 1, 0, 0, 0], custom_scaling=1e7
841 ),
842 StrFixedLenField("ChallengeFromClient", b"12345678", length=8),
843 LEIntField("Reserved3", 0),
844 PacketListField("AvPairs", [AV_PAIR()], AV_PAIR),
845 ]
847 def getAv(self, AvId):
848 try:
849 return next(x for x in self.AvPairs if x.AvId == AvId)
850 except StopIteration:
851 raise IndexError
854class NTLMv2_RESPONSE(NTLMv2_CLIENT_CHALLENGE):
855 fields_desc = [
856 XStrFixedLenField("NTProofStr", b"", length=16),
857 NTLMv2_CLIENT_CHALLENGE,
858 ]
860 def computeNTProofStr(self, ResponseKeyNT, ServerChallenge):
861 """
862 Set temp to ConcatenationOf(Responserversion, HiResponserversion,
863 Z(6), Time, ClientChallenge, Z(4), ServerName, Z(4))
864 Set NTProofStr to HMAC_MD5(ResponseKeyNT,
865 ConcatenationOf(CHALLENGE_MESSAGE.ServerChallenge,temp))
867 Remember ServerName = AvPairs
868 """
869 Responserversion = b"\x01"
870 HiResponserversion = b"\x01"
872 ServerName = b"".join(bytes(x) for x in self.AvPairs)
873 temp = b"".join(
874 [
875 Responserversion,
876 HiResponserversion,
877 b"\x00" * 6,
878 struct.pack("<Q", self.TimeStamp),
879 self.ChallengeFromClient,
880 b"\x00" * 4,
881 ServerName,
882 # Final Z(4) is the EOL AvPair
883 ]
884 )
885 return HMAC_MD5(ResponseKeyNT, ServerChallenge + temp)
888class NTLM_AUTHENTICATE(_NTLM_VARIANT_Packet, NTLM_Header):
889 name = "NTLM Authenticate"
890 __slots__ = ["VARIANT"]
891 MessageType = 3
892 NTLM_VERSION = 1
893 OFFSET = lambda pkt: (
894 64
895 if (
896 pkt.VARIANT == NTLM_VARIANT.NT_OR_2000
897 or (pkt.DomainNameBufferOffset or 88) <= 64
898 )
899 else (
900 72
901 if pkt.VARIANT == NTLM_VARIANT.XP_OR_2003
902 or ((pkt.DomainNameBufferOffset or 88) <= 72)
903 else 88
904 )
905 )
906 fields_desc = (
907 [
908 NTLM_Header,
909 # LmChallengeResponseFields
910 LEShortField("LmChallengeResponseLen", None),
911 LEShortField("LmChallengeResponseMaxLen", None),
912 LEIntField("LmChallengeResponseBufferOffset", None),
913 # NtChallengeResponseFields
914 LEShortField("NtChallengeResponseLen", None),
915 LEShortField("NtChallengeResponseMaxLen", None),
916 LEIntField("NtChallengeResponseBufferOffset", None),
917 # DomainNameFields
918 LEShortField("DomainNameLen", None),
919 LEShortField("DomainNameMaxLen", None),
920 LEIntField("DomainNameBufferOffset", None),
921 # UserNameFields
922 LEShortField("UserNameLen", None),
923 LEShortField("UserNameMaxLen", None),
924 LEIntField("UserNameBufferOffset", None),
925 # WorkstationFields
926 LEShortField("WorkstationLen", None),
927 LEShortField("WorkstationMaxLen", None),
928 LEIntField("WorkstationBufferOffset", None),
929 # EncryptedRandomSessionKeyFields
930 LEShortField("EncryptedRandomSessionKeyLen", None),
931 LEShortField("EncryptedRandomSessionKeyMaxLen", None),
932 LEIntField("EncryptedRandomSessionKeyBufferOffset", None),
933 # NegotiateFlags
934 FlagsField("NegotiateFlags", 0, -32, _negotiateFlags),
935 # VERSION
936 ]
937 + [
938 ConditionalField(
939 # (not present on some old Windows versions. We use a heuristic)
940 x,
941 lambda pkt: pkt.VARIANT >= NTLM_VARIANT.XP_OR_2003
942 and (
943 ((pkt.DomainNameBufferOffset or 88) > 64)
944 or pkt.fields.get(x.name, b"")
945 ),
946 )
947 for x in _NTLM_Version.fields_desc
948 ]
949 + [
950 # MIC
951 ConditionalField(
952 # (not present on some old Windows versions. We use a heuristic)
953 XStrFixedLenField("MIC", b"", length=16),
954 lambda pkt: pkt.VARIANT >= NTLM_VARIANT.RECENT
955 and (
956 ((pkt.DomainNameBufferOffset or 88) > 72)
957 or pkt.fields.get("MIC", b"")
958 ),
959 ),
960 # Payload
961 _NTLMPayloadField(
962 "Payload",
963 OFFSET,
964 [
965 MultipleTypeField(
966 [
967 (
968 PacketField(
969 "LmChallengeResponse",
970 LMv2_RESPONSE(),
971 LMv2_RESPONSE,
972 ),
973 lambda pkt: pkt.NTLM_VERSION == 2,
974 )
975 ],
976 PacketField("LmChallengeResponse", LM_RESPONSE(), LM_RESPONSE),
977 ),
978 MultipleTypeField(
979 [
980 (
981 PacketField(
982 "NtChallengeResponse",
983 NTLMv2_RESPONSE(),
984 NTLMv2_RESPONSE,
985 ),
986 lambda pkt: pkt.NTLM_VERSION == 2,
987 )
988 ],
989 PacketField(
990 "NtChallengeResponse", NTLM_RESPONSE(), NTLM_RESPONSE
991 ),
992 ),
993 _NTLMStrField("DomainName", b""),
994 _NTLMStrField("UserName", b""),
995 _NTLMStrField("Workstation", b""),
996 XStrField("EncryptedRandomSessionKey", b""),
997 ],
998 ),
999 ]
1000 )
1002 def post_build(self, pkt, pay):
1003 # type: (bytes, bytes) -> bytes
1004 return (
1005 _NTLM_post_build(
1006 self,
1007 pkt,
1008 self.OFFSET(),
1009 {
1010 "LmChallengeResponse": 12,
1011 "NtChallengeResponse": 20,
1012 "DomainName": 28,
1013 "UserName": 36,
1014 "Workstation": 44,
1015 "EncryptedRandomSessionKey": 52,
1016 },
1017 )
1018 + pay
1019 )
1021 def compute_mic(self, ExportedSessionKey, negotiate, challenge):
1022 self.MIC = b"\x00" * 16
1023 self.MIC = HMAC_MD5(
1024 ExportedSessionKey, bytes(negotiate) + bytes(challenge) + bytes(self)
1025 )
1028class NTLM_AUTHENTICATE_V2(NTLM_AUTHENTICATE):
1029 NTLM_VERSION = 2
1032def HTTP_ntlm_negotiate(ntlm_negotiate):
1033 """Create an HTTP NTLM negotiate packet from an NTLM_NEGOTIATE message"""
1034 assert isinstance(ntlm_negotiate, NTLM_NEGOTIATE)
1035 from scapy.layers.http import HTTP, HTTPRequest
1037 return HTTP() / HTTPRequest(
1038 Authorization=b"NTLM " + bytes_base64(bytes(ntlm_negotiate))
1039 )
1042# Experimental - Reversed stuff
1044# This is the GSSAPI NegoEX Exchange metadata blob. This is not documented
1045# but described as an "opaque blob": this was reversed and everything is a
1046# placeholder.
1049class NEGOEX_EXCHANGE_NTLM_ITEM(ASN1_Packet):
1050 ASN1_codec = ASN1_Codecs.BER
1051 ASN1_root = ASN1F_SEQUENCE(
1052 ASN1F_SEQUENCE(
1053 ASN1F_SEQUENCE(
1054 ASN1F_OID("oid", ""),
1055 ASN1F_PRINTABLE_STRING("token", ""),
1056 explicit_tag=0x31,
1057 ),
1058 explicit_tag=0x80,
1059 )
1060 )
1063class NEGOEX_EXCHANGE_NTLM(ASN1_Packet):
1064 """
1065 GSSAPI NegoEX Exchange metadata blob
1066 This was reversed and may be meaningless
1067 """
1069 ASN1_codec = ASN1_Codecs.BER
1070 ASN1_root = ASN1F_SEQUENCE(
1071 ASN1F_SEQUENCE(
1072 ASN1F_SEQUENCE_OF("items", [], NEGOEX_EXCHANGE_NTLM_ITEM), implicit_tag=0xA0
1073 ),
1074 )
1077# Crypto - [MS-NLMP]
1080def HMAC_MD5(key, data):
1081 return Hmac_MD5(key=key).digest(data)
1084def MD4le(x):
1085 """
1086 MD4 over a string encoded as utf-16le
1087 """
1088 return Hash_MD4().digest(x.encode("utf-16le"))
1091def RC4Init(key):
1092 """Alleged RC4"""
1093 from cryptography.hazmat.primitives.ciphers import Cipher, algorithms
1095 try:
1096 # cryptography > 43.0
1097 from cryptography.hazmat.decrepit.ciphers import (
1098 algorithms as decrepit_algorithms,
1099 )
1100 except ImportError:
1101 decrepit_algorithms = algorithms
1103 algorithm = decrepit_algorithms.ARC4(key)
1104 cipher = Cipher(algorithm, mode=None)
1105 encryptor = cipher.encryptor()
1106 return encryptor
1109def RC4(handle, data):
1110 """The RC4 Encryption Algorithm"""
1111 return handle.update(data)
1114def RC4K(key, data):
1115 """Indicates the encryption of data item D with the key K using the
1116 RC4 algorithm.
1117 """
1118 from cryptography.hazmat.primitives.ciphers import Cipher, algorithms
1120 try:
1121 # cryptography > 43.0
1122 from cryptography.hazmat.decrepit.ciphers import (
1123 algorithms as decrepit_algorithms,
1124 )
1125 except ImportError:
1126 decrepit_algorithms = algorithms
1128 algorithm = decrepit_algorithms.ARC4(key)
1129 cipher = Cipher(algorithm, mode=None)
1130 encryptor = cipher.encryptor()
1131 return encryptor.update(data) + encryptor.finalize()
1134# sect 2.2.2.9 - With Extended Session Security
1137class NTLMSSP_MESSAGE_SIGNATURE(Packet):
1138 # [MS-RPCE] sect 2.2.2.9.1/2.2.2.9.2
1139 fields_desc = [
1140 LEIntField("Version", 0x00000001),
1141 XStrFixedLenField("Checksum", b"", length=8),
1142 LEIntField("SeqNum", 0x00000000),
1143 ]
1145 def default_payload_class(self, payload):
1146 return conf.padding_layer
1149_GSSAPI_OIDS["1.3.6.1.4.1.311.2.2.10"] = NTLM_Header
1150_GSSAPI_SIGNATURE_OIDS["1.3.6.1.4.1.311.2.2.10"] = NTLMSSP_MESSAGE_SIGNATURE
1153# sect 3.3.2
1156def NTOWFv2(Passwd, User, UserDom, HashNt=None):
1157 """
1158 Computes the ResponseKeyNT (per [MS-NLMP] sect 3.3.2)
1160 :param Passwd: the plain password
1161 :param User: the username
1162 :param UserDom: the domain name
1163 :param HashNt: (out of spec) if you have the HashNt, use this and set
1164 Passwd to None
1165 """
1166 if HashNt is None:
1167 HashNt = MD4le(Passwd)
1168 return HMAC_MD5(HashNt, (User.upper() + UserDom).encode("utf-16le"))
1171def NTLMv2_ComputeSessionBaseKey(ResponseKeyNT, NTProofStr):
1172 return HMAC_MD5(ResponseKeyNT, NTProofStr)
1175# sect 3.4.4.2 - With Extended Session Security
1178def MAC(Handle, SigningKey, SeqNum, Message):
1179 chksum = HMAC_MD5(SigningKey, struct.pack("<i", SeqNum) + Message)[:8]
1180 if Handle:
1181 chksum = RC4(Handle, chksum)
1182 return NTLMSSP_MESSAGE_SIGNATURE(
1183 Version=0x00000001,
1184 Checksum=chksum,
1185 SeqNum=SeqNum,
1186 )
1189# sect 3.4.2
1192def SIGN(Handle, SigningKey, SeqNum, Message):
1193 # append? where is this used?!
1194 return Message + MAC(Handle, SigningKey, SeqNum, Message)
1197# sect 3.4.3
1200def SEAL(Handle, SigningKey, SeqNum, Message):
1201 """
1202 SEAL() according to [MS-NLMP]
1203 """
1204 # this is unused. Use GSS_WrapEx
1205 sealed_message = RC4(Handle, Message)
1206 signature = MAC(Handle, SigningKey, SeqNum, Message)
1207 return sealed_message, signature
1210def UNSEAL(Handle, SigningKey, SeqNum, Message):
1211 """
1212 UNSEAL() according to [MS-NLMP]
1213 """
1214 # this is unused. Use GSS_UnwrapEx
1215 unsealed_message = RC4(Handle, Message)
1216 signature = MAC(Handle, SigningKey, SeqNum, Message)
1217 return unsealed_message, signature
1220# sect 3.4.5.2
1223def SIGNKEY(NegFlg, ExportedSessionKey, Mode):
1224 if NegFlg.NEGOTIATE_EXTENDED_SESSIONSECURITY:
1225 if Mode == "Client":
1226 return Hash_MD5().digest(
1227 ExportedSessionKey
1228 + b"session key to client-to-server signing key magic constant\x00"
1229 )
1230 elif Mode == "Server":
1231 return Hash_MD5().digest(
1232 ExportedSessionKey
1233 + b"session key to server-to-client signing key magic constant\x00"
1234 )
1235 else:
1236 raise ValueError("Unknown Mode")
1237 else:
1238 return None
1241# sect 3.4.5.3
1244def SEALKEY(NegFlg, ExportedSessionKey, Mode):
1245 if NegFlg.NEGOTIATE_EXTENDED_SESSIONSECURITY:
1246 if NegFlg.NEGOTIATE_128:
1247 SealKey = ExportedSessionKey
1248 elif NegFlg.NEGOTIATE_56:
1249 SealKey = ExportedSessionKey[:7]
1250 else:
1251 SealKey = ExportedSessionKey[:5]
1252 if Mode == "Client":
1253 return Hash_MD5().digest(
1254 SealKey
1255 + b"session key to client-to-server sealing key magic constant\x00"
1256 )
1257 elif Mode == "Server":
1258 return Hash_MD5().digest(
1259 SealKey
1260 + b"session key to server-to-client sealing key magic constant\x00"
1261 )
1262 else:
1263 raise ValueError("Unknown Mode")
1264 elif NegFlg.NEGOTIATE_LM_KEY:
1265 if NegFlg.NEGOTIATE_56:
1266 return ExportedSessionKey[:6] + b"\xa0"
1267 else:
1268 return ExportedSessionKey[:4] + b"\xe5\x38\xb0"
1269 else:
1270 return ExportedSessionKey
1273# --- SSP
1276class NTLMSSP(SSP):
1277 """
1278 The NTLM SSP
1280 Common arguments:
1282 :param USE_MIC: whether to use a MIC or not (default: True)
1283 :param NTLM_VALUES: a dictionary used to override the following values
1285 In case of a client::
1287 - NegotiateFlags
1288 - ProductMajorVersion
1289 - ProductMinorVersion
1290 - ProductBuild
1292 In case of a server::
1294 - NetbiosDomainName
1295 - NetbiosComputerName
1296 - DnsComputerName
1297 - DnsDomainName (defaults to DOMAIN)
1298 - DnsTreeName (defaults to DOMAIN)
1299 - Flags
1300 - Timestamp
1302 Client-only arguments:
1304 :param UPN: the UPN to use for NTLM auth. If no domain is specified, will
1305 use the one provided by the server (domain in a domain, local
1306 if without domain)
1307 :param HASHNT: the password to use for NTLM auth
1308 :param PASSWORD: the password to use for NTLM auth
1309 :param LOCAL: use local authentication (must be running locally on Windows)
1311 Server-only arguments:
1313 :param DOMAIN_FQDN: the domain FQDN (default: domain.local)
1314 :param DOMAIN_NB_NAME: the domain Netbios name (default: strip DOMAIN_FQDN)
1315 :param COMPUTER_NB_NAME: the server Netbios name (default: SRV)
1316 :param COMPUTER_FQDN: the server FQDN
1317 (default: <computer_nb_name>.<domain_fqdn>)
1318 :param IDENTITIES: a dict {"username": <HashNT>}
1319 Setting this value enables signature computation and
1320 authenticates inbound users.
1321 """
1323 auth_type = 0x0A
1325 class STATE(SSP.STATE):
1326 INIT = 1
1327 CLI_SENT_NEGO = 2
1328 CLI_SENT_AUTH = 3
1329 SRV_SENT_CHAL = 4
1331 class CONTEXT(SSP.CONTEXT):
1332 __slots__ = [
1333 "SessionKey",
1334 "ExportedSessionKey",
1335 "IsAcceptor",
1336 "SendSignKey",
1337 "SendSealKey",
1338 "RecvSignKey",
1339 "RecvSealKey",
1340 "SendSealHandle",
1341 "RecvSealHandle",
1342 "SendSeqNum",
1343 "RecvSeqNum",
1344 "neg_tok",
1345 "chall_tok",
1346 "ServerHostname",
1347 "ServerDomain",
1348 ]
1350 @crypto_validator
1351 def __init__(self, IsAcceptor, req_flags=None):
1352 self.state = NTLMSSP.STATE.INIT
1353 self.SessionKey = None
1354 self.ExportedSessionKey = None
1355 self.SendSignKey = None
1356 self.SendSealKey = None
1357 self.SendSealHandle = None
1358 self.RecvSignKey = None
1359 self.RecvSealKey = None
1360 self.RecvSealHandle = None
1361 self.SendSeqNum = 0
1362 self.RecvSeqNum = 0
1363 self.neg_tok = None
1364 self.chall_tok = None
1365 self.ServerHostname = None
1366 self.ServerDomain = None
1367 self.IsAcceptor = IsAcceptor
1368 super(NTLMSSP.CONTEXT, self).__init__(req_flags=req_flags)
1370 def clifailure(self):
1371 self.__init__(self.IsAcceptor, req_flags=self.flags)
1373 def __repr__(self):
1374 return "NTLMSSP"
1376 # [MS-NLMP] note <36>: "the maximum lifetime is 36 hours" (lol, Kerberos has 5min)
1377 NTLM_MaxLifetime = 36 * 3600
1379 def __init__(
1380 self,
1381 UPN=None,
1382 HASHNT=None,
1383 PASSWORD=None,
1384 USE_MIC=True,
1385 VARIANT: NTLM_VARIANT = NTLM_VARIANT.RECENT,
1386 NTLM_VALUES={},
1387 DOMAIN_FQDN=None,
1388 DOMAIN_NB_NAME=None,
1389 COMPUTER_NB_NAME=None,
1390 COMPUTER_FQDN=None,
1391 IDENTITIES=None,
1392 DO_NOT_CHECK_LOGIN=False,
1393 SERVER_CHALLENGE=None,
1394 **kwargs,
1395 ):
1396 self.UPN = UPN
1397 if HASHNT is None and PASSWORD is not None:
1398 HASHNT = MD4le(PASSWORD)
1399 self.HASHNT = HASHNT
1400 self.VARIANT = VARIANT
1401 if self.VARIANT != NTLM_VARIANT.RECENT:
1402 log_runtime.warning(
1403 "VARIANT != NTLM_VARIANT.RECENT. You shouldn't touch this !"
1404 )
1405 self.USE_MIC = False
1406 else:
1407 self.USE_MIC = USE_MIC
1409 if UPN is not None:
1410 # Populate values used only in server mode.
1411 from scapy.layers.kerberos import _parse_upn
1413 try:
1414 user, realm = _parse_upn(UPN)
1415 if DOMAIN_FQDN is None:
1416 DOMAIN_FQDN = realm
1417 if COMPUTER_NB_NAME is None:
1418 COMPUTER_NB_NAME = user
1419 except ValueError:
1420 pass
1422 # Compute various netbios/fqdn names
1423 self.DOMAIN_FQDN = DOMAIN_FQDN or "WORKGROUP"
1424 self.DOMAIN_NB_NAME = (
1425 DOMAIN_NB_NAME or self.DOMAIN_FQDN.split(".")[0].upper()[:15]
1426 )
1427 self.COMPUTER_NB_NAME = COMPUTER_NB_NAME or "WIN10"
1428 self.COMPUTER_FQDN = COMPUTER_FQDN or (
1429 self.COMPUTER_NB_NAME.lower() + "." + self.DOMAIN_FQDN
1430 )
1431 self.NTLM_VALUES = NTLM_VALUES
1433 if IDENTITIES:
1434 self.IDENTITIES = {
1435 # Windows usernames are case insensitive
1436 user.upper(): hashnt
1437 for user, hashnt in IDENTITIES.items()
1438 }
1439 else:
1440 self.IDENTITIES = IDENTITIES
1442 self.DO_NOT_CHECK_LOGIN = DO_NOT_CHECK_LOGIN
1443 self.SERVER_CHALLENGE = SERVER_CHALLENGE
1444 super(NTLMSSP, self).__init__(**kwargs)
1446 def LegsAmount(self, Context: CONTEXT):
1447 return 3
1449 def GSS_Inquire_names_for_mech(self):
1450 return ["1.3.6.1.4.1.311.2.2.10"]
1452 def GSS_GetMICEx(self, Context, msgs, qop_req=0):
1453 """
1454 [MS-NLMP] sect 3.4.8
1455 """
1456 # Concatenate the ToSign
1457 ToSign = b"".join(x.data for x in msgs if x.sign)
1458 sig = MAC(
1459 Context.SendSealHandle,
1460 Context.SendSignKey,
1461 Context.SendSeqNum,
1462 ToSign,
1463 )
1464 Context.SendSeqNum += 1
1465 return sig
1467 def GSS_VerifyMICEx(self, Context, msgs, signature):
1468 """
1469 [MS-NLMP] sect 3.4.9
1470 """
1471 Context.RecvSeqNum = signature.SeqNum
1472 # Concatenate the ToSign
1473 ToSign = b"".join(x.data for x in msgs if x.sign)
1474 sig = MAC(
1475 Context.RecvSealHandle,
1476 Context.RecvSignKey,
1477 Context.RecvSeqNum,
1478 ToSign,
1479 )
1480 if sig.Checksum != signature.Checksum:
1481 raise ValueError("ERROR: Checksums don't match")
1483 def GSS_WrapEx(self, Context, msgs, qop_req=0):
1484 """
1485 [MS-NLMP] sect 3.4.6
1486 """
1487 msgs_cpy = copy.deepcopy(msgs) # Keep copy for signature
1488 # Encrypt
1489 for msg in msgs:
1490 if msg.conf_req_flag:
1491 msg.data = RC4(Context.SendSealHandle, msg.data)
1492 # Sign
1493 sig = self.GSS_GetMICEx(Context, msgs_cpy, qop_req=qop_req)
1494 return (
1495 msgs,
1496 sig,
1497 )
1499 def GSS_UnwrapEx(self, Context, msgs, signature):
1500 """
1501 [MS-NLMP] sect 3.4.7
1502 """
1503 # Decrypt
1504 for msg in msgs:
1505 if msg.conf_req_flag:
1506 msg.data = RC4(Context.RecvSealHandle, msg.data)
1507 # Check signature
1508 self.GSS_VerifyMICEx(Context, msgs, signature)
1509 return msgs
1511 def SupportsMechListMIC(self):
1512 if not self.USE_MIC:
1513 # RFC 4178
1514 # "If the mechanism selected by the negotiation does not support integrity
1515 # protection, then no mechlistMIC token is used."
1516 return False
1517 if self.DO_NOT_CHECK_LOGIN:
1518 # In this mode, we won't negotiate any credentials.
1519 return False
1520 return True
1522 def GetMechListMIC(self, Context, input):
1523 # [MS-SPNG]
1524 # "When NTLM is negotiated, the SPNG server MUST set OriginalHandle to
1525 # ServerHandle before generating the mechListMIC, then set ServerHandle to
1526 # OriginalHandle after generating the mechListMIC."
1527 OriginalHandle = Context.SendSealHandle
1528 Context.SendSealHandle = RC4Init(Context.SendSealKey)
1529 try:
1530 return super(NTLMSSP, self).GetMechListMIC(Context, input)
1531 finally:
1532 Context.SendSealHandle = OriginalHandle
1534 def VerifyMechListMIC(self, Context, otherMIC, input):
1535 # [MS-SPNG]
1536 # "the SPNEGO Extension server MUST set OriginalHandle to ClientHandle before
1537 # validating the mechListMIC and then set ClientHandle to OriginalHandle after
1538 # validating the mechListMIC."
1539 OriginalHandle = Context.RecvSealHandle
1540 Context.RecvSealHandle = RC4Init(Context.RecvSealKey)
1541 try:
1542 return super(NTLMSSP, self).VerifyMechListMIC(Context, otherMIC, input)
1543 finally:
1544 Context.RecvSealHandle = OriginalHandle
1546 def GSS_Init_sec_context(
1547 self,
1548 Context: CONTEXT,
1549 input_token=None,
1550 target_name: Optional[str] = None,
1551 req_flags: Optional[GSS_C_FLAGS] = None,
1552 chan_bindings: GssChannelBindings = GSS_C_NO_CHANNEL_BINDINGS,
1553 ):
1554 if Context is None:
1555 Context = self.CONTEXT(False, req_flags=req_flags)
1557 if Context.state == self.STATE.INIT:
1558 # Client: negotiate
1560 # Create a default token
1561 tok = NTLM_NEGOTIATE(
1562 VARIANT=self.VARIANT,
1563 NegotiateFlags="+".join(
1564 [
1565 "NEGOTIATE_UNICODE",
1566 "REQUEST_TARGET",
1567 "NEGOTIATE_NTLM",
1568 "NEGOTIATE_ALWAYS_SIGN",
1569 "TARGET_TYPE_DOMAIN",
1570 "NEGOTIATE_EXTENDED_SESSIONSECURITY",
1571 "NEGOTIATE_TARGET_INFO",
1572 "NEGOTIATE_128",
1573 "NEGOTIATE_56",
1574 ]
1575 + (
1576 ["NEGOTIATE_VERSION"]
1577 if self.VARIANT >= NTLM_VARIANT.XP_OR_2003
1578 else []
1579 )
1580 + (
1581 [
1582 "NEGOTIATE_KEY_EXCH",
1583 ]
1584 if Context.flags
1585 & (GSS_C_FLAGS.GSS_C_INTEG_FLAG | GSS_C_FLAGS.GSS_C_CONF_FLAG)
1586 else []
1587 )
1588 + (
1589 [
1590 "NEGOTIATE_SIGN",
1591 ]
1592 if Context.flags & GSS_C_FLAGS.GSS_C_INTEG_FLAG
1593 else []
1594 )
1595 + (
1596 [
1597 "NEGOTIATE_SEAL",
1598 ]
1599 if Context.flags & GSS_C_FLAGS.GSS_C_CONF_FLAG
1600 else []
1601 )
1602 + (
1603 [
1604 "NEGOTIATE_IDENTIFY",
1605 ]
1606 if Context.flags & GSS_C_FLAGS.GSS_C_IDENTIFY_FLAG
1607 else []
1608 )
1609 ),
1610 ProductMajorVersion=10,
1611 ProductMinorVersion=0,
1612 ProductBuild=26100,
1613 )
1615 # Update that token with the customs one
1616 if self.NTLM_VALUES:
1617 for key in [
1618 "NegotiateFlags",
1619 "ProductMajorVersion",
1620 "ProductMinorVersion",
1621 "ProductBuild",
1622 "DomainName",
1623 "WorkstationName",
1624 ]:
1625 if key in self.NTLM_VALUES:
1626 setattr(tok, key, self.NTLM_VALUES[key])
1627 Context.neg_tok = tok
1628 Context.SessionKey = None # Reset signing (if previous auth failed)
1629 Context.state = self.STATE.CLI_SENT_NEGO
1630 return Context, tok, GSS_S_CONTINUE_NEEDED
1631 elif Context.state == self.STATE.CLI_SENT_NEGO:
1632 # Client: auth (token=challenge)
1633 chall_tok = input_token
1635 if self.UPN is None or self.HASHNT is None:
1636 raise ValueError(
1637 "Must provide a 'UPN' and a 'HASHNT' or 'PASSWORD' when "
1638 "running in standalone !"
1639 )
1641 from scapy.layers.kerberos import _parse_upn
1643 # Check token sanity
1644 if not chall_tok or NTLM_CHALLENGE not in chall_tok:
1645 log_runtime.debug("NTLMSSP: Unexpected token. Expected NTLM Challenge")
1646 return Context, None, GSS_S_DEFECTIVE_TOKEN
1648 # Some information from the CHALLENGE are stored
1649 try:
1650 Context.ServerHostname = chall_tok.getAv(0x0001).Value
1651 except IndexError:
1652 pass
1653 try:
1654 Context.ServerDomain = chall_tok.getAv(0x0002).Value
1655 except IndexError:
1656 pass
1657 try:
1658 # the server SHOULD set the timestamp in the CHALLENGE_MESSAGE
1659 ServerTimestamp = chall_tok.getAv(0x0007).Value
1660 ServerTime = (ServerTimestamp / 1e7) - 11644473600
1662 if abs(ServerTime - time.time()) >= NTLMSSP.NTLM_MaxLifetime:
1663 log_runtime.warning(
1664 "Server and Client times are off by more than 36h !"
1665 )
1666 # We could error here, but we don't.
1667 except IndexError:
1668 pass
1670 # Initialize a default token
1671 tok = NTLM_AUTHENTICATE_V2(
1672 VARIANT=self.VARIANT,
1673 NegotiateFlags=chall_tok.NegotiateFlags,
1674 ProductMajorVersion=10,
1675 ProductMinorVersion=0,
1676 ProductBuild=26100,
1677 )
1679 # Populate the token
1680 tok.LmChallengeResponse = LMv2_RESPONSE()
1682 # 1. Set username
1683 try:
1684 tok.UserName, realm = _parse_upn(self.UPN)
1685 except ValueError:
1686 tok.UserName, realm = self.UPN, Context.ServerDomain
1688 # 2. Set domain name
1689 if realm is None:
1690 log_runtime.warning(
1691 "No realm specified in UPN, nor provided by server."
1692 )
1693 tok.DomainName = self.DOMAIN_FQDN
1694 else:
1695 tok.DomainName = realm
1697 # 3. Set workstation name
1698 tok.Workstation = self.COMPUTER_NB_NAME
1700 # 4. Create and calculate the ChallengeResponse
1701 # 4.1 Build the payload
1702 cr = tok.NtChallengeResponse = NTLMv2_RESPONSE(
1703 ChallengeFromClient=os.urandom(8),
1704 )
1705 cr.TimeStamp = int((time.time() + 11644473600) * 1e7)
1706 cr.AvPairs = (
1707 # Repeat AvPairs from the server
1708 chall_tok.TargetInfo[:-1]
1709 + (
1710 [
1711 AV_PAIR(AvId="MsvAvFlags", Value="MIC integrity"),
1712 ]
1713 if self.USE_MIC
1714 else []
1715 )
1716 + [
1717 AV_PAIR(
1718 AvId="MsvAvSingleHost",
1719 Value=Single_Host_Data(
1720 MachineID=os.urandom(32),
1721 PermanentMachineID=os.urandom(32),
1722 ),
1723 ),
1724 ]
1725 + (
1726 [
1727 AV_PAIR(
1728 # [MS-NLMP] sect 2.2.2.1 refers to RFC 4121 sect 4.1.1.2
1729 # "The Bnd field contains the MD5 hash of channel bindings"
1730 AvId="MsvAvChannelBindings",
1731 Value=chan_bindings.digestMD5(),
1732 ),
1733 ]
1734 if chan_bindings != GSS_C_NO_CHANNEL_BINDINGS
1735 else []
1736 )
1737 + [
1738 AV_PAIR(
1739 AvId="MsvAvTargetName",
1740 Value=target_name or ("host/" + Context.ServerHostname),
1741 ),
1742 AV_PAIR(AvId="MsvAvEOL"),
1743 ]
1744 )
1745 if self.NTLM_VALUES:
1746 # Update that token with the customs one
1747 for key in [
1748 "NegotiateFlags",
1749 "ProductMajorVersion",
1750 "ProductMinorVersion",
1751 "ProductBuild",
1752 ]:
1753 if key in self.NTLM_VALUES:
1754 setattr(tok, key, self.NTLM_VALUES[key])
1756 # 4.2 Compute the ResponseKeyNT
1757 ResponseKeyNT = NTOWFv2(
1758 None,
1759 tok.UserName,
1760 tok.DomainName,
1761 HashNt=self.HASHNT,
1762 )
1764 # 4.3 Compute the NTProofStr
1765 cr.NTProofStr = cr.computeNTProofStr(
1766 ResponseKeyNT,
1767 chall_tok.ServerChallenge,
1768 )
1770 # 4.4 Compute the Session Key
1771 SessionBaseKey = NTLMv2_ComputeSessionBaseKey(ResponseKeyNT, cr.NTProofStr)
1772 KeyExchangeKey = SessionBaseKey # Only true for NTLMv2
1773 if chall_tok.NegotiateFlags.NEGOTIATE_KEY_EXCH:
1774 ExportedSessionKey = os.urandom(16)
1775 tok.EncryptedRandomSessionKey = RC4K(
1776 KeyExchangeKey,
1777 ExportedSessionKey,
1778 )
1779 else:
1780 ExportedSessionKey = KeyExchangeKey
1782 # 4.5 Compute the MIC
1783 if self.USE_MIC:
1784 tok.compute_mic(ExportedSessionKey, Context.neg_tok, chall_tok)
1786 # 5. Perform key computations
1787 Context.ExportedSessionKey = ExportedSessionKey
1788 # [MS-SMB] 3.2.5.3
1789 Context.SessionKey = Context.ExportedSessionKey
1790 # Compute NTLM keys
1791 Context.SendSignKey = SIGNKEY(
1792 tok.NegotiateFlags, ExportedSessionKey, "Client"
1793 )
1794 Context.SendSealKey = SEALKEY(
1795 tok.NegotiateFlags, ExportedSessionKey, "Client"
1796 )
1797 Context.SendSealHandle = RC4Init(Context.SendSealKey)
1798 Context.RecvSignKey = SIGNKEY(
1799 tok.NegotiateFlags, ExportedSessionKey, "Server"
1800 )
1801 Context.RecvSealKey = SEALKEY(
1802 tok.NegotiateFlags, ExportedSessionKey, "Server"
1803 )
1804 Context.RecvSealHandle = RC4Init(Context.RecvSealKey)
1806 # Update the state
1807 Context.state = self.STATE.CLI_SENT_AUTH
1809 return Context, tok, GSS_S_COMPLETE
1810 elif Context.state == self.STATE.CLI_SENT_AUTH:
1811 if input_token:
1812 # what is that?
1813 status = GSS_S_DEFECTIVE_TOKEN
1814 else:
1815 status = GSS_S_COMPLETE
1816 return Context, None, status
1817 else:
1818 raise ValueError("NTLMSSP: unexpected state %s" % repr(Context.state))
1820 def GSS_Accept_sec_context(
1821 self,
1822 Context: CONTEXT,
1823 input_token=None,
1824 req_flags: Optional[GSS_S_FLAGS] = GSS_S_FLAGS.GSS_S_ALLOW_MISSING_BINDINGS,
1825 chan_bindings: GssChannelBindings = GSS_C_NO_CHANNEL_BINDINGS,
1826 ):
1827 if Context is None:
1828 Context = self.CONTEXT(IsAcceptor=True, req_flags=req_flags)
1830 if Context.state == self.STATE.INIT:
1831 # Server: challenge (input_token=negotiate)
1832 nego_tok = input_token
1833 if not nego_tok or NTLM_NEGOTIATE not in nego_tok:
1834 log_runtime.debug("NTLMSSP: Unexpected token. Expected NTLM Negotiate")
1835 return Context, None, GSS_S_DEFECTIVE_TOKEN
1837 # Build the challenge token
1838 currentTime = (time.time() + 11644473600) * 1e7
1839 tok = NTLM_CHALLENGE(
1840 VARIANT=self.VARIANT,
1841 ServerChallenge=self.SERVER_CHALLENGE or os.urandom(8),
1842 NegotiateFlags="+".join(
1843 [
1844 "NEGOTIATE_UNICODE",
1845 "REQUEST_TARGET",
1846 "NEGOTIATE_NTLM",
1847 "NEGOTIATE_ALWAYS_SIGN",
1848 "NEGOTIATE_EXTENDED_SESSIONSECURITY",
1849 "NEGOTIATE_TARGET_INFO",
1850 "TARGET_TYPE_DOMAIN",
1851 "NEGOTIATE_128",
1852 "NEGOTIATE_KEY_EXCH",
1853 "NEGOTIATE_56",
1854 ]
1855 + (
1856 ["NEGOTIATE_VERSION"]
1857 if self.VARIANT >= NTLM_VARIANT.XP_OR_2003
1858 else []
1859 )
1860 + (
1861 ["NEGOTIATE_SIGN"]
1862 if nego_tok.NegotiateFlags.NEGOTIATE_SIGN
1863 else []
1864 )
1865 + (
1866 ["NEGOTIATE_SEAL"]
1867 if nego_tok.NegotiateFlags.NEGOTIATE_SEAL
1868 else []
1869 )
1870 ),
1871 ProductMajorVersion=10,
1872 ProductMinorVersion=0,
1873 Payload=[
1874 ("TargetName", ""),
1875 (
1876 "TargetInfo",
1877 [
1878 # MsvAvNbComputerName
1879 AV_PAIR(AvId=1, Value=self.COMPUTER_NB_NAME),
1880 # MsvAvNbDomainName
1881 AV_PAIR(AvId=2, Value=self.DOMAIN_NB_NAME),
1882 # MsvAvDnsComputerName
1883 AV_PAIR(AvId=3, Value=self.COMPUTER_FQDN),
1884 # MsvAvDnsDomainName
1885 AV_PAIR(AvId=4, Value=self.DOMAIN_FQDN),
1886 # MsvAvDnsTreeName
1887 AV_PAIR(AvId=5, Value=self.DOMAIN_FQDN),
1888 # MsvAvTimestamp
1889 AV_PAIR(AvId=7, Value=currentTime),
1890 # MsvAvEOL
1891 AV_PAIR(AvId=0),
1892 ],
1893 ),
1894 ],
1895 )
1896 if self.NTLM_VALUES:
1897 # Update that token with the customs one
1898 for key in [
1899 "ServerChallenge",
1900 "NegotiateFlags",
1901 "ProductMajorVersion",
1902 "ProductMinorVersion",
1903 "TargetName",
1904 ]:
1905 if key in self.NTLM_VALUES:
1906 setattr(tok, key, self.NTLM_VALUES[key])
1907 avpairs = {x.AvId: x.Value for x in tok.TargetInfo}
1908 tok.TargetInfo = [
1909 AV_PAIR(AvId=i, Value=self.NTLM_VALUES.get(x, avpairs[i]))
1910 for (i, x) in [
1911 (2, "NetbiosDomainName"),
1912 (1, "NetbiosComputerName"),
1913 (4, "DnsDomainName"),
1914 (3, "DnsComputerName"),
1915 (5, "DnsTreeName"),
1916 (6, "Flags"),
1917 (7, "Timestamp"),
1918 (0, None),
1919 ]
1920 if ((x in self.NTLM_VALUES) or (i in avpairs))
1921 and self.NTLM_VALUES.get(x, True) is not None
1922 ]
1924 # Store for next step
1925 Context.chall_tok = tok
1927 # Update the state
1928 Context.state = self.STATE.SRV_SENT_CHAL
1930 return Context, tok, GSS_S_CONTINUE_NEEDED
1931 elif Context.state == self.STATE.SRV_SENT_CHAL:
1932 # server: OK or challenge again (input_token=auth)
1933 auth_tok = input_token
1935 if not auth_tok or NTLM_AUTHENTICATE_V2 not in auth_tok:
1936 log_runtime.debug(
1937 "NTLMSSP: Unexpected token. Expected NTLM Authenticate v2"
1938 )
1939 return Context, None, GSS_S_DEFECTIVE_TOKEN
1941 if self.DO_NOT_CHECK_LOGIN:
1942 # Just trust me bro. Typically used in "guest" mode.
1943 return Context, None, GSS_S_COMPLETE
1945 # Compute the session key
1946 SessionBaseKey = self._getSessionBaseKey(Context, auth_tok)
1947 if SessionBaseKey:
1948 # [MS-NLMP] sect 3.2.5.1.2
1949 KeyExchangeKey = SessionBaseKey # Only true for NTLMv2
1950 if auth_tok.NegotiateFlags.NEGOTIATE_KEY_EXCH:
1951 try:
1952 EncryptedRandomSessionKey = auth_tok.EncryptedRandomSessionKey
1953 except AttributeError:
1954 # No EncryptedRandomSessionKey. libcurl for instance
1955 # hmm. this looks bad
1956 EncryptedRandomSessionKey = b"\x00" * 16
1957 ExportedSessionKey = RC4K(KeyExchangeKey, EncryptedRandomSessionKey)
1958 else:
1959 ExportedSessionKey = KeyExchangeKey
1960 Context.ExportedSessionKey = ExportedSessionKey
1961 # [MS-SMB] 3.2.5.3
1962 Context.SessionKey = Context.ExportedSessionKey
1964 # Check the timestamp
1965 try:
1966 ClientTimestamp = auth_tok.NtChallengeResponse.getAv(0x0007).Value
1967 ClientTime = (ClientTimestamp / 1e7) - 11644473600
1969 if abs(ClientTime - time.time()) >= NTLMSSP.NTLM_MaxLifetime:
1970 log_runtime.warning(
1971 "Server and Client times are off by more than 36h !"
1972 )
1973 # We could error here, but we don't.
1974 except IndexError:
1975 pass
1977 # Check the channel bindings
1978 if chan_bindings != GSS_C_NO_CHANNEL_BINDINGS:
1979 try:
1980 Bnd = auth_tok.NtChallengeResponse.getAv(0x000A).Value
1981 if Bnd != chan_bindings.digestMD5():
1982 # Bad channel bindings
1983 return Context, None, GSS_S_BAD_BINDINGS
1984 except IndexError:
1985 if GSS_S_FLAGS.GSS_S_ALLOW_MISSING_BINDINGS not in req_flags:
1986 # Uhoh, we required channel bindings
1987 return Context, None, GSS_S_BAD_BINDINGS
1989 if Context.SessionKey:
1990 # Compute NTLM keys
1991 Context.SendSignKey = SIGNKEY(
1992 auth_tok.NegotiateFlags, ExportedSessionKey, "Server"
1993 )
1994 Context.SendSealKey = SEALKEY(
1995 auth_tok.NegotiateFlags, ExportedSessionKey, "Server"
1996 )
1997 Context.SendSealHandle = RC4Init(Context.SendSealKey)
1998 Context.RecvSignKey = SIGNKEY(
1999 auth_tok.NegotiateFlags, ExportedSessionKey, "Client"
2000 )
2001 Context.RecvSealKey = SEALKEY(
2002 auth_tok.NegotiateFlags, ExportedSessionKey, "Client"
2003 )
2004 Context.RecvSealHandle = RC4Init(Context.RecvSealKey)
2006 # Check the NTProofStr
2007 if self._checkLogin(Context, auth_tok):
2008 # Set negotiated flags
2009 if auth_tok.NegotiateFlags.NEGOTIATE_SIGN:
2010 Context.flags |= GSS_C_FLAGS.GSS_C_INTEG_FLAG
2011 if auth_tok.NegotiateFlags.NEGOTIATE_SEAL:
2012 Context.flags |= GSS_C_FLAGS.GSS_C_CONF_FLAG
2013 return Context, None, GSS_S_COMPLETE
2015 # Bad NTProofStr or unknown user
2016 Context.SessionKey = None
2017 Context.state = self.STATE.INIT
2018 return Context, None, GSS_S_DEFECTIVE_CREDENTIAL
2019 else:
2020 raise ValueError("NTLMSSP: unexpected state %s" % repr(Context.state))
2022 def MaximumSignatureLength(self, Context: CONTEXT):
2023 """
2024 Returns the Maximum Signature length.
2026 This will be used in auth_len in DceRpc5, and is necessary for
2027 PFC_SUPPORT_HEADER_SIGN to work properly.
2028 """
2029 return 16 # len(NTLMSSP_MESSAGE_SIGNATURE())
2031 def GSS_Passive(self, Context: CONTEXT, token=None, req_flags=None):
2032 if Context is None:
2033 Context = self.CONTEXT(True)
2034 Context.passive = True
2036 # We capture the Negotiate, Challenge, then call the server's auth handling
2037 # and discard the output.
2039 if Context.state == self.STATE.INIT:
2040 if not token or NTLM_NEGOTIATE not in token:
2041 log_runtime.warning("NTLMSSP: Expected NTLM Negotiate")
2042 return None, GSS_S_DEFECTIVE_TOKEN
2043 Context.neg_tok = token
2044 Context.state = self.STATE.CLI_SENT_NEGO
2045 return Context, GSS_S_CONTINUE_NEEDED
2046 elif Context.state == self.STATE.CLI_SENT_NEGO:
2047 if not token or NTLM_CHALLENGE not in token:
2048 log_runtime.warning("NTLMSSP: Expected NTLM Challenge")
2049 return None, GSS_S_DEFECTIVE_TOKEN
2050 Context.chall_tok = token
2051 Context.state = self.STATE.SRV_SENT_CHAL
2052 return Context, GSS_S_CONTINUE_NEEDED
2053 elif Context.state == self.STATE.SRV_SENT_CHAL:
2054 if not token or NTLM_AUTHENTICATE_V2 not in token:
2055 log_runtime.warning("NTLMSSP: Expected NTLM Authenticate")
2056 return None, GSS_S_DEFECTIVE_TOKEN
2057 Context, _, status = self.GSS_Accept_sec_context(Context, token)
2058 if status != GSS_S_COMPLETE:
2059 log_runtime.info("NTLMSSP: auth failed.")
2060 Context.state = self.STATE.INIT
2061 return Context, status
2062 else:
2063 raise ValueError("NTLMSSP: unexpected state %s" % repr(Context.state))
2065 def GSS_Passive_set_Direction(self, Context: CONTEXT, IsAcceptor=False):
2066 if Context.IsAcceptor is not IsAcceptor:
2067 return
2068 # Swap everything
2069 Context.SendSignKey, Context.RecvSignKey = (
2070 Context.RecvSignKey,
2071 Context.SendSignKey,
2072 )
2073 Context.SendSealKey, Context.RecvSealKey = (
2074 Context.RecvSealKey,
2075 Context.SendSealKey,
2076 )
2077 Context.SendSealHandle, Context.RecvSealHandle = (
2078 Context.RecvSealHandle,
2079 Context.SendSealHandle,
2080 )
2081 Context.SendSeqNum, Context.RecvSeqNum = Context.RecvSeqNum, Context.SendSeqNum
2082 Context.IsAcceptor = not Context.IsAcceptor
2084 def _getSessionBaseKey(self, Context, auth_tok):
2085 """
2086 Function that returns the SessionBaseKey from the ntlm Authenticate.
2087 """
2088 try:
2089 # Windows usernames are case insensitive
2090 username = auth_tok.UserName.upper()
2091 except AttributeError:
2092 username = None
2093 try:
2094 domain = auth_tok.DomainName
2095 except AttributeError:
2096 domain = ""
2097 if self.IDENTITIES and username in self.IDENTITIES:
2098 ResponseKeyNT = NTOWFv2(
2099 None,
2100 username,
2101 domain,
2102 HashNt=self.IDENTITIES[username],
2103 )
2104 return NTLMv2_ComputeSessionBaseKey(
2105 ResponseKeyNT,
2106 auth_tok.NtChallengeResponse.NTProofStr,
2107 )
2108 elif self.IDENTITIES:
2109 log_runtime.debug("NTLMSSP: Bad credentials for %s" % username)
2110 return None
2112 def _checkLogin(self, Context, auth_tok):
2113 """
2114 Function that checks the validity of an authentication.
2116 Overwrite and return True to bypass.
2117 """
2118 try:
2119 # Windows usernames are case insensitive
2120 username = auth_tok.UserName.upper()
2121 except AttributeError:
2122 username = None
2123 try:
2124 domain = auth_tok.DomainName
2125 except AttributeError:
2126 domain = ""
2127 if username in self.IDENTITIES:
2128 ResponseKeyNT = NTOWFv2(
2129 None,
2130 username,
2131 domain,
2132 HashNt=self.IDENTITIES[username],
2133 )
2134 NTProofStr = auth_tok.NtChallengeResponse.computeNTProofStr(
2135 ResponseKeyNT,
2136 Context.chall_tok.ServerChallenge,
2137 )
2138 if NTProofStr == auth_tok.NtChallengeResponse.NTProofStr:
2139 return True
2140 return False
2143class NTLMSSP_DOMAIN(NTLMSSP):
2144 """
2145 A variant of the NTLMSSP to be used in server mode that gets the session
2146 keys from the domain using a Netlogon channel.
2148 This has the same arguments as NTLMSSP, but supports the following in server
2149 mode:
2151 :param UPN: the UPN of the machine account to login for Netlogon.
2152 :param HASHNT: the HASHNT of the machine account (use Netlogon secure channel).
2153 :param ssp: a KerberosSSP to use (use Kerberos secure channel).
2154 :param PASSWORD: the PASSWORD of the machine account to use for Netlogon.
2155 :param DC_FQDN: (optional) specify the FQDN of the DC.
2157 Netlogon example::
2159 >>> mySSP = NTLMSSP_DOMAIN(
2160 ... UPN="Server1@domain.local",
2161 ... HASHNT=bytes.fromhex("8846f7eaee8fb117ad06bdd830b7586c"),
2162 ... )
2164 Kerberos example::
2166 >>> mySSP = NTLMSSP_DOMAIN(
2167 ... UPN="Server1@domain.local",
2168 ... KEY=Key(EncryptionType.AES256_CTS_HMAC_SHA1_96,
2169 ... key=bytes.fromhex(
2170 ... "85abb9b61dc2fa49d4cc04317bbd108f8f79df28"
2171 ... "239155ed7b144c5d2ebcf016"
2172 ... )
2173 ... ),
2174 ... )
2175 """
2177 def __init__(self, UPN=None, *args, timeout=3, ssp=None, **kwargs):
2178 from scapy.layers.kerberos import KerberosSSP
2180 # Either PASSWORD or HASHNT or ssp
2181 if (
2182 "HASHNT" not in kwargs
2183 and "PASSWORD" not in kwargs
2184 and "KEY" not in kwargs
2185 and ssp is None
2186 ):
2187 raise ValueError(
2188 "Must specify either 'HASHNT', 'PASSWORD' or "
2189 "provide a ssp=KerberosSSP()"
2190 )
2191 elif ssp is not None and not isinstance(ssp, KerberosSSP):
2192 raise ValueError("'ssp' can only be None or a KerberosSSP !")
2194 self.KEY = kwargs.pop("KEY", None)
2195 self.PASSWORD = kwargs.get("PASSWORD", None)
2197 # UPN is mandatory
2198 if UPN is None and ssp is not None and ssp.UPN:
2199 UPN = ssp.UPN
2200 elif UPN is None:
2201 raise ValueError("Must specify a 'UPN' !")
2202 kwargs["UPN"] = UPN
2204 # Call parent
2205 super(NTLMSSP_DOMAIN, self).__init__(
2206 *args,
2207 **kwargs,
2208 )
2210 # Treat specific parameters
2211 self.DC_FQDN = kwargs.pop("DC_FQDN", None)
2212 if self.DC_FQDN is None:
2213 # Get DC_FQDN from dclocator
2214 from scapy.layers.ldap import dclocator
2216 dc = dclocator(
2217 self.DOMAIN_FQDN,
2218 timeout=timeout,
2219 debug=kwargs.get("debug", 0),
2220 )
2221 self.DC_FQDN = dc.samlogon.DnsHostName.decode().rstrip(".")
2223 # If logging in via Kerberos
2224 self.ssp = ssp
2226 def _getSessionBaseKey(self, Context, ntlm):
2227 """
2228 Return the Session Key by asking the DC.
2229 """
2230 # No user / no domain: skip.
2231 if not ntlm.UserNameLen or not ntlm.DomainNameLen:
2232 return super(NTLMSSP_DOMAIN, self)._getSessionBaseKey(Context, ntlm)
2234 # Import RPC stuff
2235 from scapy.layers.dcerpc import NDRUnion
2236 from scapy.layers.msrpce.msnrpc import (
2237 NETLOGON_SECURE_CHANNEL_METHOD,
2238 NetlogonClient,
2239 )
2240 from scapy.layers.msrpce.raw.ms_nrpc import (
2241 NETLOGON_LOGON_IDENTITY_INFO,
2242 NetrLogonSamLogonWithFlags_Request,
2243 PNETLOGON_AUTHENTICATOR,
2244 PNETLOGON_NETWORK_INFO,
2245 STRING,
2246 UNICODE_STRING,
2247 )
2249 # Create NetlogonClient with PRIVACY
2250 client = NetlogonClient()
2251 client.connect(self.DC_FQDN)
2253 # Establish the Netlogon secure channel (this will bind)
2254 try:
2255 if self.ssp is None and self.KEY is None:
2256 # Login via classic NetlogonSSP
2257 client.establish_secure_channel(
2258 mode=NETLOGON_SECURE_CHANNEL_METHOD.NetrServerAuthenticate3,
2259 UPN=f"{self.COMPUTER_NB_NAME}@{self.DOMAIN_NB_NAME}",
2260 DC_FQDN=self.DC_FQDN,
2261 HASHNT=self.HASHNT,
2262 )
2263 else:
2264 # Login via KerberosSSP (Windows 2025)
2265 client.establish_secure_channel(
2266 mode=NETLOGON_SECURE_CHANNEL_METHOD.NetrServerAuthenticateKerberos,
2267 UPN=self.UPN,
2268 DC_FQDN=self.DC_FQDN,
2269 PASSWORD=self.PASSWORD,
2270 KEY=self.KEY,
2271 ssp=self.ssp,
2272 )
2273 except ValueError:
2274 log_runtime.warning(
2275 "Couldn't establish the Netlogon secure channel. "
2276 "Check the credentials for '%s' !" % self.COMPUTER_NB_NAME
2277 )
2278 return super(NTLMSSP_DOMAIN, self)._getSessionBaseKey(Context, ntlm)
2280 # Request validation of the NTLM request
2281 req = NetrLogonSamLogonWithFlags_Request(
2282 LogonServer="",
2283 ComputerName=self.COMPUTER_NB_NAME,
2284 Authenticator=client.create_authenticator(),
2285 ReturnAuthenticator=PNETLOGON_AUTHENTICATOR(),
2286 LogonLevel=6, # NetlogonNetworkTransitiveInformation
2287 LogonInformation=NDRUnion(
2288 tag=6,
2289 value=PNETLOGON_NETWORK_INFO(
2290 Identity=NETLOGON_LOGON_IDENTITY_INFO(
2291 LogonDomainName=UNICODE_STRING(
2292 Buffer=ntlm.DomainName,
2293 ),
2294 ParameterControl=0x00002AE0,
2295 UserName=UNICODE_STRING(
2296 Buffer=ntlm.UserName,
2297 ),
2298 Workstation=UNICODE_STRING(
2299 Buffer=ntlm.Workstation,
2300 ),
2301 ),
2302 LmChallenge=Context.chall_tok.ServerChallenge,
2303 NtChallengeResponse=STRING(
2304 Buffer=bytes(ntlm.NtChallengeResponse),
2305 ),
2306 LmChallengeResponse=STRING(
2307 Buffer=bytes(ntlm.LmChallengeResponse),
2308 ),
2309 ),
2310 ),
2311 ValidationLevel=6,
2312 ExtraFlags=0,
2313 ndr64=client.ndr64,
2314 )
2316 # Get response
2317 resp = client.sr1_req(req)
2318 if resp and resp.status == 0:
2319 # Success
2321 # Validate DC authenticator
2322 client.validate_authenticator(resp.ReturnAuthenticator.value)
2324 # Get and return the SessionKey
2325 UserSessionKey = resp.ValidationInformation.value.value.UserSessionKey
2326 return bytes(UserSessionKey)
2327 else:
2328 # Failed
2329 return super(NTLMSSP_DOMAIN, self)._getSessionBaseKey(Context, ntlm)
2331 def _checkLogin(self, Context, auth_tok):
2332 # Always OK if we got the session key
2333 return True