Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/scapy/layers/ntlm.py: 35%

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.compat import bytes_base64

33from scapy.error import log_runtime

34from scapy.fields import (

35 ByteEnumField,

36 ByteField,

37 ConditionalField,

38 Field,

39 FieldLenField,

40 FlagsField,

41 LEIntEnumField,

42 LEIntField,

43 LEShortEnumField,

44 LEShortField,

45 LEThreeBytesField,

46 MultipleTypeField,

47 PacketField,

48 PacketListField,

49 StrField,

50 StrFieldUtf16,

51 StrFixedLenField,

52 StrLenFieldUtf16,

53 UTCTimeField,

54 XStrField,

55 XStrFixedLenField,

56 XStrLenField,

57 _StrField,

58)

59from scapy.packet import Packet

60from scapy.sessions import StringBuffer

62from scapy.layers.gssapi import (

63 _GSSAPI_OIDS,

64 _GSSAPI_SIGNATURE_OIDS,

65 GSS_C_FLAGS,

66 GSS_C_NO_CHANNEL_BINDINGS,

67 GSS_S_BAD_BINDINGS,

68 GSS_S_COMPLETE,

69 GSS_S_CONTINUE_NEEDED,

70 GSS_S_DEFECTIVE_CREDENTIAL,

71 GSS_S_DEFECTIVE_TOKEN,

72 GSS_S_FLAGS,

73 GssChannelBindings,

74 SSP,

75)

77# Typing imports

78from typing import (

79 Any,

80 Callable,

81 List,

82 Optional,

83 Tuple,

84 Union,

85)

87# Crypto imports

89from scapy.layers.tls.crypto.hash import Hash_MD4, Hash_MD5

90from scapy.layers.tls.crypto.h_mac import Hmac_MD5

92##########

93# Fields #

94##########

97# NTLM structures are all in all very complicated. Many fields don't have a fixed

98# position, but are rather referred to with an offset (from the beginning of the

99# structure) and a length. In addition to that, there are variants of the structure

100# with missing fields when running old versions of Windows (sometimes also seen when

101# talking to products that reimplement NTLM, most notably backup applications).

102

103# We add `_NTLMPayloadField` and `_NTLMPayloadPacket` to parse fields that use an

104# offset, and `_NTLM_post_build` to be able to rebuild those offsets.

105# In addition, the `NTLM_VARIANT*` allows to select what flavor of NTLM to use

106# (NT, XP, or Recent). But in real world use only Recent should be used.

107

108

109class _NTLMPayloadField(_StrField[List[Tuple[str, Any]]]):

110 """Special field used to dissect NTLM payloads.

111 This isn't trivial because the offsets are variable."""

112

113 __slots__ = [

114 "fields",

115 "fields_map",

116 "offset",

117 "length_from",

118 "force_order",

119 "offset_name",

120 ]

121 islist = True

122

123 def __init__(

124 self,

125 name, # type: str

126 offset, # type: Union[int, Callable[[Packet], int]]

127 fields, # type: List[Field[Any, Any]]

128 length_from=None, # type: Optional[Callable[[Packet], int]]

129 force_order=None, # type: Optional[List[str]]

130 offset_name="BufferOffset", # type: str

131 ):

132 # type: (...) -> None

133 self.offset = offset

134 self.fields = fields

135 self.fields_map = {field.name: field for field in fields}

136 self.length_from = length_from

137 self.force_order = force_order # whether the order of fields is fixed

138 self.offset_name = offset_name

139 super(_NTLMPayloadField, self).__init__(

140 name,

141 [

142 (field.name, field.default)

143 for field in fields

144 if field.default is not None

145 ],

146 )

147

148 def _on_payload(self, pkt, x, func):

149 # type: (Optional[Packet], bytes, str) -> List[Tuple[str, Any]]

150 if not pkt or not x:

151 return []

152 results = []

153 for field_name, value in x:

154 if field_name not in self.fields_map:

155 continue

156 if not isinstance(

157 self.fields_map[field_name], PacketListField

158 ) and not isinstance(value, Packet):

159 value = getattr(self.fields_map[field_name], func)(pkt, value)

160 results.append((field_name, value))

161 return results

162

163 def i2h(self, pkt, x):

164 # type: (Optional[Packet], bytes) -> List[Tuple[str, str]]

165 return self._on_payload(pkt, x, "i2h")

166

167 def h2i(self, pkt, x):

168 # type: (Optional[Packet], bytes) -> List[Tuple[str, str]]

169 return self._on_payload(pkt, x, "h2i")

170

171 def i2repr(self, pkt, x):

172 # type: (Optional[Packet], bytes) -> str

173 return repr(self._on_payload(pkt, x, "i2repr"))

174

175 def _o_pkt(self, pkt):

176 # type: (Optional[Packet]) -> int

177 if callable(self.offset):

178 return self.offset(pkt)

179 return self.offset

180

181 def addfield(self, pkt, s, val):

182 # type: (Optional[Packet], bytes, Optional[List[Tuple[str, str]]]) -> bytes

183 # Create string buffer

184 buf = StringBuffer()

185 buf.append(s, 1)

186 # Calc relative offset

187 r_off = self._o_pkt(pkt) - len(s)

188 if self.force_order:

189 val.sort(key=lambda x: self.force_order.index(x[0]))

190 for field_name, value in val:

191 if field_name not in self.fields_map:

192 continue

193 field = self.fields_map[field_name]

194 offset = pkt.getfieldval(field_name + self.offset_name)

195 if offset is None:

196 # No offset specified: calc

197 offset = len(buf)

198 else:

199 # Calc relative offset

200 offset -= r_off

201 pad = offset + 1 - len(buf)

202 # Add padding if necessary

203 if pad > 0:

204 buf.append(pad * b"\x00", len(buf))

205 buf.append(field.addfield(pkt, bytes(buf), value)[len(buf) :], offset + 1)

206 return bytes(buf)

207

208 def getfield(self, pkt, s):

209 # type: (Packet, bytes) -> Tuple[bytes, List[Tuple[str, str]]]

210 if self.length_from is None:

211 ret, remain = b"", s

212 else:

213 len_pkt = self.length_from(pkt)

214 ret, remain = s[len_pkt:], s[:len_pkt]

215 if not pkt or not remain:

216 return s, []

217 results = []

218 max_offset = 0

219 o_pkt = self._o_pkt(pkt)

220 offsets = [

221 pkt.getfieldval(x.name + self.offset_name) - o_pkt for x in self.fields

222 ]

223 for i, field in enumerate(self.fields):

224 offset = offsets[i]

225 try:

226 length = pkt.getfieldval(field.name + "Len")

227 except AttributeError:

228 length = len(remain) - offset

229 # length can't be greater than the difference with the next offset

230 try:

231 length = min(length, min(x - offset for x in offsets if x > offset))

232 except ValueError:

233 pass

234 if offset < 0:

235 continue

236 max_offset = max(offset + length, max_offset)

237 if remain[offset : offset + length]:

238 results.append(

239 (

240 offset,

241 field.name,

242 field.getfield(pkt, remain[offset : offset + length])[1],

243 )

244 )

245 ret += remain[max_offset:]

246 results.sort(key=lambda x: x[0])

247 return ret, [x[1:] for x in results]

248

249

250class _NTLMPayloadPacket(Packet):

251 _NTLM_PAYLOAD_FIELD_NAME = "Payload"

252

253 def __init__(

254 self,

255 _pkt=b"", # type: Union[bytes, bytearray]

256 post_transform=None, # type: Any

257 _internal=0, # type: int

258 _underlayer=None, # type: Optional[Packet]

259 _parent=None, # type: Optional[Packet]

260 **fields, # type: Any

261 ):

262 # pop unknown fields. We can't process them until the packet is initialized

263 unknown = {

264 k: fields.pop(k)

265 for k in list(fields)

266 if not any(k == f.name for f in self.fields_desc)

267 }

268 super(_NTLMPayloadPacket, self).__init__(

269 _pkt=_pkt,

270 post_transform=post_transform,

271 _internal=_internal,

272 _underlayer=_underlayer,

273 _parent=_parent,

274 **fields,

275 )

276 # check unknown fields for implicit ones

277 local_fields = next(

278 [y.name for y in x.fields]

279 for x in self.fields_desc

280 if x.name == self._NTLM_PAYLOAD_FIELD_NAME

281 )

282 implicit_fields = {k: v for k, v in unknown.items() if k in local_fields}

283 for k, value in implicit_fields.items():

284 self.setfieldval(k, value)

285

286 def getfieldval(self, attr):

287 # Ease compatibility with _NTLMPayloadField

288 try:

289 return super(_NTLMPayloadPacket, self).getfieldval(attr)

290 except AttributeError:

291 try:

292 return next(

293 x[1]

294 for x in super(_NTLMPayloadPacket, self).getfieldval(

295 self._NTLM_PAYLOAD_FIELD_NAME

296 )

297 if x[0] == attr

298 )

299 except StopIteration:

300 raise AttributeError(attr)

301

302 def getfield_and_val(self, attr):

303 # Ease compatibility with _NTLMPayloadField

304 try:

305 return super(_NTLMPayloadPacket, self).getfield_and_val(attr)

306 except ValueError:

307 PayFields = self.get_field(self._NTLM_PAYLOAD_FIELD_NAME).fields_map

308 try:

309 return (

310 PayFields[attr],

311 PayFields[attr].h2i( # cancel out the i2h.. it's dumb i know

312 self,

313 next(

314 x[1]

315 for x in super(_NTLMPayloadPacket, self).__getattr__(

316 self._NTLM_PAYLOAD_FIELD_NAME

317 )

318 if x[0] == attr

319 ),

320 ),

321 )

322 except (StopIteration, KeyError):

323 raise ValueError(attr)

324

325 def setfieldval(self, attr, val):

326 # Ease compatibility with _NTLMPayloadField

327 try:

328 return super(_NTLMPayloadPacket, self).setfieldval(attr, val)

329 except AttributeError:

330 Payload = super(_NTLMPayloadPacket, self).__getattr__(

331 self._NTLM_PAYLOAD_FIELD_NAME

332 )

333 if attr not in self.get_field(self._NTLM_PAYLOAD_FIELD_NAME).fields_map:

334 raise AttributeError(attr)

335 try:

336 Payload.pop(

337 next(

338 i

339 for i, x in enumerate(

340 super(_NTLMPayloadPacket, self).__getattr__(

341 self._NTLM_PAYLOAD_FIELD_NAME

342 )

343 )

344 if x[0] == attr

345 )

346 )

347 except StopIteration:

348 pass

349 Payload.append([attr, val])

350 super(_NTLMPayloadPacket, self).setfieldval(

351 self._NTLM_PAYLOAD_FIELD_NAME, Payload

352 )

353

354

355class _NTLM_ENUM(IntEnum):

356 LEN = 0x0001

357 MAXLEN = 0x0002

358 OFFSET = 0x0004

359 COUNT = 0x0008

360 PAD8 = 0x1000

361

362

363_NTLM_CONFIG = [

364 ("Len", _NTLM_ENUM.LEN),

365 ("MaxLen", _NTLM_ENUM.MAXLEN),

366 ("BufferOffset", _NTLM_ENUM.OFFSET),

367]

368

369

370def _NTLM_post_build(self, p, pay_offset, fields, config=_NTLM_CONFIG):

371 """Util function to build the offset and populate the lengths"""

372 for field_name, value in self.fields[self._NTLM_PAYLOAD_FIELD_NAME]:

373 fld = self.get_field(self._NTLM_PAYLOAD_FIELD_NAME).fields_map[field_name]

374 length = fld.i2len(self, value)

375 count = fld.i2count(self, value)

376 offset = fields[field_name]

377 i = 0

378 r = lambda y: {2: "H", 4: "I", 8: "Q"}[y]

379 for fname, ftype in config:

380 if isinstance(ftype, dict):

381 ftype = ftype[field_name]

382 if ftype & _NTLM_ENUM.LEN:

383 fval = length

384 elif ftype & _NTLM_ENUM.OFFSET:

385 fval = pay_offset

386 elif ftype & _NTLM_ENUM.MAXLEN:

387 fval = length

388 elif ftype & _NTLM_ENUM.COUNT:

389 fval = count

390 else:

391 raise ValueError

392 if ftype & _NTLM_ENUM.PAD8:

393 fval += (-fval) % 8

394 sz = self.get_field(field_name + fname).sz

395 if self.getfieldval(field_name + fname) is None:

396 p = (

397 p[: offset + i]

398 + struct.pack("<%s" % r(sz), fval)

399 + p[offset + i + sz :]

400 )

401 i += sz

402 pay_offset += length

403 return p

404

405

406##############

407# Structures #

408##############

409

410

411# -- Util: VARIANT class

412

413

414class NTLM_VARIANT(IntEnum):

415 """

416 The message variant to use for NTLM.

417 """

418

419 NT_OR_2000 = 0

420 XP_OR_2003 = 1

421 RECENT = 2

422

423

424class _NTLM_VARIANT_Packet(_NTLMPayloadPacket):

425 def __init__(self, *args, **kwargs):

426 self.VARIANT = kwargs.pop("VARIANT", NTLM_VARIANT.RECENT)

427 super(_NTLM_VARIANT_Packet, self).__init__(*args, **kwargs)

428

429 def clone_with(self, *args, **kwargs):

430 pkt = super(_NTLM_VARIANT_Packet, self).clone_with(*args, **kwargs)

431 pkt.VARIANT = self.VARIANT

432 return pkt

433

434 def copy(self):

435 pkt = super(_NTLM_VARIANT_Packet, self).copy()

436 pkt.VARIANT = self.VARIANT

437

438 return pkt

439

440 def show2(self, dump=False, indent=3, lvl="", label_lvl=""):

441 return self.__class__(bytes(self), VARIANT=self.VARIANT).show(

442 dump, indent, lvl, label_lvl

443 )

446# Sect 2.2

449class NTLM_Header(Packet):

450 name = "NTLM Header"

451 fields_desc = [

452 StrFixedLenField("Signature", b"NTLMSSP\0", length=8),

453 LEIntEnumField(

454 "MessageType",

455 3,

456 {

457 1: "NEGOTIATE_MESSAGE",

458 2: "CHALLENGE_MESSAGE",

459 3: "AUTHENTICATE_MESSAGE",

460 },

461 ),

462 ]

463

464 @classmethod

465 def dispatch_hook(cls, _pkt=None, *args, **kargs):

466 if cls is NTLM_Header and _pkt and len(_pkt) >= 10:

467 MessageType = struct.unpack("<H", _pkt[8:10])[0]

468 if MessageType == 1:

469 return NTLM_NEGOTIATE

470 elif MessageType == 2:

471 return NTLM_CHALLENGE

472 elif MessageType == 3:

473 return NTLM_AUTHENTICATE_V2

474 return cls

475

476

477# Sect 2.2.2.5

478_negotiateFlags = [

479 "NEGOTIATE_UNICODE", # A

480 "NEGOTIATE_OEM", # B

481 "REQUEST_TARGET", # C

482 "r10",

483 "NEGOTIATE_SIGN", # D

484 "NEGOTIATE_SEAL", # E

485 "NEGOTIATE_DATAGRAM", # F

486 "NEGOTIATE_LM_KEY", # G

487 "r9",

488 "NEGOTIATE_NTLM", # H

489 "r8",

490 "J",

491 "NEGOTIATE_OEM_DOMAIN_SUPPLIED", # K

492 "NEGOTIATE_OEM_WORKSTATION_SUPPLIED", # L

493 "r7",

494 "NEGOTIATE_ALWAYS_SIGN", # M

495 "TARGET_TYPE_DOMAIN", # N

496 "TARGET_TYPE_SERVER", # O

497 "r6",

498 "NEGOTIATE_EXTENDED_SESSIONSECURITY", # P

499 "NEGOTIATE_IDENTIFY", # Q

500 "r5",

501 "REQUEST_NON_NT_SESSION_KEY", # R

502 "NEGOTIATE_TARGET_INFO", # S

503 "r4",

504 "NEGOTIATE_VERSION", # T

505 "r3",

506 "r2",

507 "r1",

508 "NEGOTIATE_128", # U

509 "NEGOTIATE_KEY_EXCH", # V

510 "NEGOTIATE_56", # W

511]

512

513

514def _NTLMStrField(name, default):

515 return MultipleTypeField(

516 [

517 (

518 StrFieldUtf16(name, default),

519 lambda pkt: pkt.NegotiateFlags.NEGOTIATE_UNICODE,

520 )

521 ],

522 StrField(name, default),

523 )

526# Sect 2.2.2.10

529class _NTLM_Version(Packet):

530 fields_desc = [

531 ByteField("ProductMajorVersion", 0),

532 ByteField("ProductMinorVersion", 0),

533 LEShortField("ProductBuild", 0),

534 LEThreeBytesField("res_ver", 0),

535 ByteEnumField("NTLMRevisionCurrent", 0x0F, {0x0F: "v15"}),

536 ]

539# Sect 2.2.1.1

542class NTLM_NEGOTIATE(_NTLM_VARIANT_Packet, NTLM_Header):

543 name = "NTLM Negotiate"

544 __slots__ = ["VARIANT"]

545 MessageType = 1

546 OFFSET = lambda pkt: (

547 32

548 if (

549 pkt.VARIANT == NTLM_VARIANT.NT_OR_2000

550 or (pkt.DomainNameBufferOffset or 40) <= 32

551 )

552 else 40

553 )

554 fields_desc = (

555 [

556 NTLM_Header,

557 FlagsField("NegotiateFlags", 0, -32, _negotiateFlags),

558 # DomainNameFields

559 LEShortField("DomainNameLen", None),

560 LEShortField("DomainNameMaxLen", None),

561 LEIntField("DomainNameBufferOffset", None),

562 # WorkstationFields

563 LEShortField("WorkstationNameLen", None),

564 LEShortField("WorkstationNameMaxLen", None),

565 LEIntField("WorkstationNameBufferOffset", None),

566 ]

567 + [

568 # VERSION

569 ConditionalField(

570 # (not present on some old Windows versions. We use a heuristic)

571 x,

572 lambda pkt: pkt.VARIANT >= NTLM_VARIANT.XP_OR_2003

573 and (

574 (

575 (

576 40

577 if pkt.DomainNameBufferOffset is None

578 else pkt.DomainNameBufferOffset or len(pkt.original or b"")

579 )

580 > 32

581 )

582 or pkt.fields.get(x.name, b"")

583 ),

584 )

585 for x in _NTLM_Version.fields_desc

586 ]

587 + [

588 # Payload

589 _NTLMPayloadField(

590 "Payload",

591 OFFSET,

592 [

593 _NTLMStrField("DomainName", b""),

594 _NTLMStrField("WorkstationName", b""),

595 ],

596 ),

597 ]

598 )

599

600 def post_build(self, pkt, pay):

601 # type: (bytes, bytes) -> bytes

602 return (

603 _NTLM_post_build(

604 self,

605 pkt,

606 self.OFFSET(),

607 {

608 "DomainName": 16,

609 "WorkstationName": 24,

610 },

611 )

612 + pay

613 )

616# Challenge

619class Single_Host_Data(Packet):

620 fields_desc = [

621 LEIntField("Size", None),

622 LEIntField("Z4", 0),

623 # "CustomData" guessed using LSAP_TOKEN_INFO_INTEGRITY.

624 FlagsField(

625 "Flags",

626 0,

627 -32,

628 {

629 0x00000001: "UAC-Restricted",

630 },

631 ),

632 LEIntEnumField(

633 "TokenIL",

634 0x00002000,

635 {

636 0x00000000: "Untrusted",

637 0x00001000: "Low",

638 0x00002000: "Medium",

639 0x00003000: "High",

640 0x00004000: "System",

641 0x00005000: "Protected process",

642 },

643 ),

644 XStrFixedLenField("MachineID", b"", length=32),

645 # KB 5068222 - still waiting for [MS-KILE] update (oct. 2025)

646 ConditionalField(

647 XStrFixedLenField("PermanentMachineID", None, length=32),

648 lambda pkt: pkt.Size is None or pkt.Size > 48,

649 ),

650 ]

651

652 def post_build(self, pkt, pay):

653 if self.Size is None:

654 pkt = struct.pack("<I", len(pkt)) + pkt[4:]

655 return pkt + pay

656

657 def default_payload_class(self, payload):

658 return conf.padding_layer

659

660

661class AV_PAIR(Packet):

662 name = "NTLM AV Pair"

663 fields_desc = [

664 LEShortEnumField(

665 "AvId",

666 0,

667 {

668 0x0000: "MsvAvEOL",

669 0x0001: "MsvAvNbComputerName",

670 0x0002: "MsvAvNbDomainName",

671 0x0003: "MsvAvDnsComputerName",

672 0x0004: "MsvAvDnsDomainName",

673 0x0005: "MsvAvDnsTreeName",

674 0x0006: "MsvAvFlags",

675 0x0007: "MsvAvTimestamp",

676 0x0008: "MsvAvSingleHost",

677 0x0009: "MsvAvTargetName",

678 0x000A: "MsvAvChannelBindings",

679 },

680 ),

681 FieldLenField("AvLen", None, length_of="Value", fmt="<H"),

682 MultipleTypeField(

683 [

684 (

685 LEIntEnumField(

686 "Value",

687 1,

688 {

689 0x0001: "constrained",

690 0x0002: "MIC integrity",

691 0x0004: "SPN from untrusted source",

692 },

693 ),

694 lambda pkt: pkt.AvId == 0x0006,

695 ),

696 (

697 UTCTimeField(

698 "Value",

699 None,

700 epoch=[1601, 1, 1, 0, 0, 0],

701 custom_scaling=1e7,

702 fmt="<Q",

703 ),

704 lambda pkt: pkt.AvId == 0x0007,

705 ),

706 (

707 PacketField("Value", Single_Host_Data(), Single_Host_Data),

708 lambda pkt: pkt.AvId == 0x0008,

709 ),

710 (

711 XStrLenField("Value", b"", length_from=lambda pkt: pkt.AvLen),

712 lambda pkt: pkt.AvId == 0x000A,

713 ),

714 ],

715 StrLenFieldUtf16("Value", b"", length_from=lambda pkt: pkt.AvLen),

716 ),

717 ]

718

719 def default_payload_class(self, payload):

720 return conf.padding_layer

721

722

723class NTLM_CHALLENGE(_NTLM_VARIANT_Packet, NTLM_Header):

724 name = "NTLM Challenge"

725 __slots__ = ["VARIANT"]

726 MessageType = 2

727 OFFSET = lambda pkt: (

728 48

729 if (

730 pkt.VARIANT == NTLM_VARIANT.NT_OR_2000

731 or (pkt.TargetInfoBufferOffset or 56) <= 48

732 )

733 else 56

734 )

735 fields_desc = (

736 [

737 NTLM_Header,

738 # TargetNameFields

739 LEShortField("TargetNameLen", None),

740 LEShortField("TargetNameMaxLen", None),

741 LEIntField("TargetNameBufferOffset", None),

742 #

743 FlagsField("NegotiateFlags", 0, -32, _negotiateFlags),

744 XStrFixedLenField("ServerChallenge", None, length=8),

745 XStrFixedLenField("Reserved", None, length=8),

746 # TargetInfoFields

747 LEShortField("TargetInfoLen", None),

748 LEShortField("TargetInfoMaxLen", None),

749 LEIntField("TargetInfoBufferOffset", None),

750 ]

751 + [

752 # VERSION

753 ConditionalField(

754 # (not present on some old Windows versions. We use a heuristic)

755 x,

756 lambda pkt: pkt.VARIANT >= NTLM_VARIANT.XP_OR_2003

757 and (

758 ((pkt.TargetInfoBufferOffset or 56) > 40)

759 or pkt.fields.get(x.name, b"")

760 ),

761 )

762 for x in _NTLM_Version.fields_desc

763 ]

764 + [

765 # Payload

766 _NTLMPayloadField(

767 "Payload",

768 OFFSET,

769 [

770 _NTLMStrField("TargetName", b""),

771 PacketListField("TargetInfo", [AV_PAIR()], AV_PAIR),

772 ],

773 ),

774 ]

775 )

776

777 def getAv(self, AvId):

778 try:

779 return next(x for x in self.TargetInfo if x.AvId == AvId)

780 except (StopIteration, AttributeError):

781 raise IndexError

782

783 def post_build(self, pkt, pay):

784 # type: (bytes, bytes) -> bytes

785 return (

786 _NTLM_post_build(

787 self,

788 pkt,

789 self.OFFSET(),

790 {

791 "TargetName": 12,

792 "TargetInfo": 40,

793 },

794 )

795 + pay

796 )

799# Authenticate

802class LM_RESPONSE(Packet):

803 fields_desc = [

804 StrFixedLenField("Response", b"", length=24),

805 ]

806

807

808class LMv2_RESPONSE(Packet):

809 fields_desc = [

810 StrFixedLenField("Response", b"", length=16),

811 StrFixedLenField("ChallengeFromClient", b"", length=8),

812 ]

813

814

815class NTLM_RESPONSE(Packet):

816 fields_desc = [

817 StrFixedLenField("Response", b"", length=24),

818 ]

819

820

821class NTLMv2_CLIENT_CHALLENGE(Packet):

822 fields_desc = [

823 ByteField("RespType", 1),

824 ByteField("HiRespType", 1),

825 LEShortField("Reserved1", 0),

826 LEIntField("Reserved2", 0),

827 UTCTimeField(

828 "TimeStamp", None, fmt="<Q", epoch=[1601, 1, 1, 0, 0, 0], custom_scaling=1e7

829 ),

830 StrFixedLenField("ChallengeFromClient", b"12345678", length=8),

831 LEIntField("Reserved3", 0),

832 PacketListField("AvPairs", [AV_PAIR()], AV_PAIR),

833 ]

834

835 def getAv(self, AvId):

836 try:

837 return next(x for x in self.AvPairs if x.AvId == AvId)

838 except StopIteration:

839 raise IndexError

840

841

842class NTLMv2_RESPONSE(NTLMv2_CLIENT_CHALLENGE):

843 fields_desc = [

844 XStrFixedLenField("NTProofStr", b"", length=16),

845 NTLMv2_CLIENT_CHALLENGE,

846 ]

847

848 def computeNTProofStr(self, ResponseKeyNT, ServerChallenge):

849 """

850 Set temp to ConcatenationOf(Responserversion, HiResponserversion,

851 Z(6), Time, ClientChallenge, Z(4), ServerName, Z(4))

852 Set NTProofStr to HMAC_MD5(ResponseKeyNT,

853 ConcatenationOf(CHALLENGE_MESSAGE.ServerChallenge,temp))

854

855 Remember ServerName = AvPairs

856 """

857 Responserversion = b"\x01"

858 HiResponserversion = b"\x01"

859

860 ServerName = b"".join(bytes(x) for x in self.AvPairs)

861 temp = b"".join(

862 [

863 Responserversion,

864 HiResponserversion,

865 b"\x00" * 6,

866 struct.pack("<Q", self.TimeStamp),

867 self.ChallengeFromClient,

868 b"\x00" * 4,

869 ServerName,

870 # Final Z(4) is the EOL AvPair

871 ]

872 )

873 return HMAC_MD5(ResponseKeyNT, ServerChallenge + temp)

874

875

876class NTLM_AUTHENTICATE(_NTLM_VARIANT_Packet, NTLM_Header):

877 name = "NTLM Authenticate"

878 __slots__ = ["VARIANT"]

879 MessageType = 3

880 NTLM_VERSION = 1

881 OFFSET = lambda pkt: (

882 64

883 if (

884 pkt.VARIANT == NTLM_VARIANT.NT_OR_2000

885 or (pkt.DomainNameBufferOffset or 88) <= 64

886 )

887 else (

888 72

889 if pkt.VARIANT == NTLM_VARIANT.XP_OR_2003

890 or ((pkt.DomainNameBufferOffset or 88) <= 72)

891 else 88

892 )

893 )

894 fields_desc = (

895 [

896 NTLM_Header,

897 # LmChallengeResponseFields

898 LEShortField("LmChallengeResponseLen", None),

899 LEShortField("LmChallengeResponseMaxLen", None),

900 LEIntField("LmChallengeResponseBufferOffset", None),

901 # NtChallengeResponseFields

902 LEShortField("NtChallengeResponseLen", None),

903 LEShortField("NtChallengeResponseMaxLen", None),

904 LEIntField("NtChallengeResponseBufferOffset", None),

905 # DomainNameFields

906 LEShortField("DomainNameLen", None),

907 LEShortField("DomainNameMaxLen", None),

908 LEIntField("DomainNameBufferOffset", None),

909 # UserNameFields

910 LEShortField("UserNameLen", None),

911 LEShortField("UserNameMaxLen", None),

912 LEIntField("UserNameBufferOffset", None),

913 # WorkstationFields

914 LEShortField("WorkstationLen", None),

915 LEShortField("WorkstationMaxLen", None),

916 LEIntField("WorkstationBufferOffset", None),

917 # EncryptedRandomSessionKeyFields

918 LEShortField("EncryptedRandomSessionKeyLen", None),

919 LEShortField("EncryptedRandomSessionKeyMaxLen", None),

920 LEIntField("EncryptedRandomSessionKeyBufferOffset", None),

921 # NegotiateFlags

922 FlagsField("NegotiateFlags", 0, -32, _negotiateFlags),

923 # VERSION

924 ]

925 + [

926 ConditionalField(

927 # (not present on some old Windows versions. We use a heuristic)

928 x,

929 lambda pkt: pkt.VARIANT >= NTLM_VARIANT.XP_OR_2003

930 and (

931 ((pkt.DomainNameBufferOffset or 88) > 64)

932 or pkt.fields.get(x.name, b"")

933 ),

934 )

935 for x in _NTLM_Version.fields_desc

936 ]

937 + [

938 # MIC

939 ConditionalField(

940 # (not present on some old Windows versions. We use a heuristic)

941 XStrFixedLenField("MIC", b"", length=16),

942 lambda pkt: pkt.VARIANT >= NTLM_VARIANT.RECENT

943 and (

944 ((pkt.DomainNameBufferOffset or 88) > 72)

945 or pkt.fields.get("MIC", b"")

946 ),

947 ),

948 # Payload

949 _NTLMPayloadField(

950 "Payload",

951 OFFSET,

952 [

953 MultipleTypeField(

954 [

955 (

956 PacketField(

957 "LmChallengeResponse",

958 LMv2_RESPONSE(),

959 LMv2_RESPONSE,

960 ),

961 lambda pkt: pkt.NTLM_VERSION == 2,

962 )

963 ],

964 PacketField("LmChallengeResponse", LM_RESPONSE(), LM_RESPONSE),

965 ),

966 MultipleTypeField(

967 [

968 (

969 PacketField(

970 "NtChallengeResponse",

971 NTLMv2_RESPONSE(),

972 NTLMv2_RESPONSE,

973 ),

974 lambda pkt: pkt.NTLM_VERSION == 2,

975 )

976 ],

977 PacketField(

978 "NtChallengeResponse", NTLM_RESPONSE(), NTLM_RESPONSE

979 ),

980 ),

981 _NTLMStrField("DomainName", b""),

982 _NTLMStrField("UserName", b""),

983 _NTLMStrField("Workstation", b""),

984 XStrField("EncryptedRandomSessionKey", b""),

985 ],

986 ),

987 ]

988 )

989

990 def post_build(self, pkt, pay):

991 # type: (bytes, bytes) -> bytes

992 return (

993 _NTLM_post_build(

994 self,

995 pkt,

996 self.OFFSET(),

997 {

998 "LmChallengeResponse": 12,

999 "NtChallengeResponse": 20,

1000 "DomainName": 28,

1001 "UserName": 36,

1002 "Workstation": 44,

1003 "EncryptedRandomSessionKey": 52,

1004 },

1005 )

1006 + pay

1007 )

1008

1009 def compute_mic(self, ExportedSessionKey, negotiate, challenge):

1010 self.MIC = b"\x00" * 16

1011 self.MIC = HMAC_MD5(

1012 ExportedSessionKey, bytes(negotiate) + bytes(challenge) + bytes(self)

1013 )

1014

1015

1016class NTLM_AUTHENTICATE_V2(NTLM_AUTHENTICATE):

1017 NTLM_VERSION = 2

1018

1019

1020def HTTP_ntlm_negotiate(ntlm_negotiate):

1021 """Create an HTTP NTLM negotiate packet from an NTLM_NEGOTIATE message"""

1022 assert isinstance(ntlm_negotiate, NTLM_NEGOTIATE)

1023 from scapy.layers.http import HTTP, HTTPRequest

1024

1025 return HTTP() / HTTPRequest(

1026 Authorization=b"NTLM " + bytes_base64(bytes(ntlm_negotiate))

1027 )

1028

1029

1030# Experimental - Reversed stuff

1031

1032# This is the GSSAPI NegoEX Exchange metadata blob. This is not documented

1033# but described as an "opaque blob": this was reversed and everything is a

1034# placeholder.

1035

1036

1037class NEGOEX_EXCHANGE_NTLM_ITEM(ASN1_Packet):

1038 ASN1_codec = ASN1_Codecs.BER

1039 ASN1_root = ASN1F_SEQUENCE(

1040 ASN1F_SEQUENCE(

1041 ASN1F_SEQUENCE(

1042 ASN1F_OID("oid", ""),

1043 ASN1F_PRINTABLE_STRING("token", ""),

1044 explicit_tag=0x31,

1045 ),

1046 explicit_tag=0x80,

1047 )

1048 )

1049

1050

1051class NEGOEX_EXCHANGE_NTLM(ASN1_Packet):

1052 """

1053 GSSAPI NegoEX Exchange metadata blob

1054 This was reversed and may be meaningless

1055 """

1056

1057 ASN1_codec = ASN1_Codecs.BER

1058 ASN1_root = ASN1F_SEQUENCE(

1059 ASN1F_SEQUENCE(

1060 ASN1F_SEQUENCE_OF("items", [], NEGOEX_EXCHANGE_NTLM_ITEM), implicit_tag=0xA0

1061 ),

1062 )

1063

1064

1065# Crypto - [MS-NLMP]

1066

1067

1068def HMAC_MD5(key, data):

1069 return Hmac_MD5(key=key).digest(data)

1070

1071

1072def MD4le(x):

1073 """

1074 MD4 over a string encoded as utf-16le

1075 """

1076 return Hash_MD4().digest(x.encode("utf-16le"))

1077

1078

1079def RC4Init(key):

1080 """Alleged RC4"""

1081 from cryptography.hazmat.primitives.ciphers import Cipher, algorithms

1082

1083 try:

1084 # cryptography > 43.0

1085 from cryptography.hazmat.decrepit.ciphers import (

1086 algorithms as decrepit_algorithms,

1087 )

1088 except ImportError:

1089 decrepit_algorithms = algorithms

1090

1091 algorithm = decrepit_algorithms.ARC4(key)

1092 cipher = Cipher(algorithm, mode=None)

1093 encryptor = cipher.encryptor()

1094 return encryptor

1095

1096

1097def RC4(handle, data):

1098 """The RC4 Encryption Algorithm"""

1099 return handle.update(data)

1100

1101

1102def RC4K(key, data):

1103 """Indicates the encryption of data item D with the key K using the

1104 RC4 algorithm.

1105 """

1106 from cryptography.hazmat.primitives.ciphers import Cipher, algorithms

1107

1108 try:

1109 # cryptography > 43.0

1110 from cryptography.hazmat.decrepit.ciphers import (

1111 algorithms as decrepit_algorithms,

1112 )

1113 except ImportError:

1114 decrepit_algorithms = algorithms

1115

1116 algorithm = decrepit_algorithms.ARC4(key)

1117 cipher = Cipher(algorithm, mode=None)

1118 encryptor = cipher.encryptor()

1119 return encryptor.update(data) + encryptor.finalize()

1120

1121

1122# sect 2.2.2.9 - With Extended Session Security

1123

1124

1125class NTLMSSP_MESSAGE_SIGNATURE(Packet):

1126 # [MS-RPCE] sect 2.2.2.9.1/2.2.2.9.2

1127 fields_desc = [

1128 LEIntField("Version", 0x00000001),

1129 XStrFixedLenField("Checksum", b"", length=8),

1130 LEIntField("SeqNum", 0x00000000),

1131 ]

1132

1133 def default_payload_class(self, payload):

1134 return conf.padding_layer

1135

1136

1137_GSSAPI_OIDS["1.3.6.1.4.1.311.2.2.10"] = NTLM_Header

1138_GSSAPI_SIGNATURE_OIDS["1.3.6.1.4.1.311.2.2.10"] = NTLMSSP_MESSAGE_SIGNATURE

1141# sect 3.3.2

1144def NTOWFv2(Passwd, User, UserDom, HashNt=None):

1145 """

1146 Computes the ResponseKeyNT (per [MS-NLMP] sect 3.3.2)

1147

1148 :param Passwd: the plain password

1149 :param User: the username

1150 :param UserDom: the domain name

1151 :param HashNt: (out of spec) if you have the HashNt, use this and set

1152 Passwd to None

1153 """

1154 if HashNt is None:

1155 HashNt = MD4le(Passwd)

1156 return HMAC_MD5(HashNt, (User.upper() + UserDom).encode("utf-16le"))

1157

1158

1159def NTLMv2_ComputeSessionBaseKey(ResponseKeyNT, NTProofStr):

1160 return HMAC_MD5(ResponseKeyNT, NTProofStr)

1161

1162

1163# sect 3.4.4.2 - With Extended Session Security

1164

1165

1166def MAC(Handle, SigningKey, SeqNum, Message):

1167 chksum = HMAC_MD5(SigningKey, struct.pack("<i", SeqNum) + Message)[:8]

1168 if Handle:

1169 chksum = RC4(Handle, chksum)

1170 return NTLMSSP_MESSAGE_SIGNATURE(

1171 Version=0x00000001,

1172 Checksum=chksum,

1173 SeqNum=SeqNum,

1174 )

1177# sect 3.4.2

1180def SIGN(Handle, SigningKey, SeqNum, Message):

1181 # append? where is this used?!

1182 return Message + MAC(Handle, SigningKey, SeqNum, Message)

1185# sect 3.4.3

1188def SEAL(Handle, SigningKey, SeqNum, Message):

1189 """

1190 SEAL() according to [MS-NLMP]

1191 """

1192 # this is unused. Use GSS_WrapEx

1193 sealed_message = RC4(Handle, Message)

1194 signature = MAC(Handle, SigningKey, SeqNum, Message)

1195 return sealed_message, signature

1196

1197

1198def UNSEAL(Handle, SigningKey, SeqNum, Message):

1199 """

1200 UNSEAL() according to [MS-NLMP]

1201 """

1202 # this is unused. Use GSS_UnwrapEx

1203 unsealed_message = RC4(Handle, Message)

1204 signature = MAC(Handle, SigningKey, SeqNum, Message)

1205 return unsealed_message, signature

1208# sect 3.4.5.2

1211def SIGNKEY(NegFlg, ExportedSessionKey, Mode):

1212 if NegFlg.NEGOTIATE_EXTENDED_SESSIONSECURITY:

1213 if Mode == "Client":

1214 return Hash_MD5().digest(

1215 ExportedSessionKey

1216 + b"session key to client-to-server signing key magic constant\x00"

1217 )

1218 elif Mode == "Server":

1219 return Hash_MD5().digest(

1220 ExportedSessionKey

1221 + b"session key to server-to-client signing key magic constant\x00"

1222 )

1223 else:

1224 raise ValueError("Unknown Mode")

1225 else:

1226 return None

1229# sect 3.4.5.3

1232def SEALKEY(NegFlg, ExportedSessionKey, Mode):

1233 if NegFlg.NEGOTIATE_EXTENDED_SESSIONSECURITY:

1234 if NegFlg.NEGOTIATE_128:

1235 SealKey = ExportedSessionKey

1236 elif NegFlg.NEGOTIATE_56:

1237 SealKey = ExportedSessionKey[:7]

1238 else:

1239 SealKey = ExportedSessionKey[:5]

1240 if Mode == "Client":

1241 return Hash_MD5().digest(

1242 SealKey

1243 + b"session key to client-to-server sealing key magic constant\x00"

1244 )

1245 elif Mode == "Server":

1246 return Hash_MD5().digest(

1247 SealKey

1248 + b"session key to server-to-client sealing key magic constant\x00"

1249 )

1250 else:

1251 raise ValueError("Unknown Mode")

1252 elif NegFlg.NEGOTIATE_LM_KEY:

1253 if NegFlg.NEGOTIATE_56:

1254 return ExportedSessionKey[:6] + b"\xa0"

1255 else:

1256 return ExportedSessionKey[:4] + b"\xe5\x38\xb0"

1257 else:

1258 return ExportedSessionKey

1261# --- SSP

1264class NTLMSSP(SSP):

1265 """

1266 The NTLM SSP

1267

1268 Common arguments:

1269

1270 :param auth_level: One of DCE_C_AUTHN_LEVEL

1271 :param USE_MIC: whether to use a MIC or not (default: True)

1272 :param NTLM_VALUES: a dictionary used to override the following values

1273

1274 In case of a client::

1275

1276 - NegotiateFlags

1277 - ProductMajorVersion

1278 - ProductMinorVersion

1279 - ProductBuild

1280

1281 In case of a server::

1282

1283 - NetbiosDomainName

1284 - NetbiosComputerName

1285 - DnsComputerName

1286 - DnsDomainName (defaults to DOMAIN)

1287 - DnsTreeName (defaults to DOMAIN)

1288 - Flags

1289 - Timestamp

1290

1291 Client-only arguments:

1292

1293 :param UPN: the UPN to use for NTLM auth. If no domain is specified, will

1294 use the one provided by the server (domain in a domain, local

1295 if without domain)

1296 :param HASHNT: the password to use for NTLM auth

1297 :param PASSWORD: the password to use for NTLM auth

1298

1299 Server-only arguments:

1300

1301 :param DOMAIN_FQDN: the domain FQDN (default: domain.local)

1302 :param DOMAIN_NB_NAME: the domain Netbios name (default: strip DOMAIN_FQDN)

1303 :param COMPUTER_NB_NAME: the server Netbios name (default: SRV)

1304 :param COMPUTER_FQDN: the server FQDN

1305 (default: <computer_nb_name>.<domain_fqdn>)

1306 :param IDENTITIES: a dict {"username": <HashNT>}

1307 Setting this value enables signature computation and

1308 authenticates inbound users.

1309 """

1310

1311 auth_type = 0x0A

1312

1313 class STATE(SSP.STATE):

1314 INIT = 1

1315 CLI_SENT_NEGO = 2

1316 CLI_SENT_AUTH = 3

1317 SRV_SENT_CHAL = 4

1318

1319 class CONTEXT(SSP.CONTEXT):

1320 __slots__ = [

1321 "SessionKey",

1322 "ExportedSessionKey",

1323 "IsAcceptor",

1324 "SendSignKey",

1325 "SendSealKey",

1326 "RecvSignKey",

1327 "RecvSealKey",

1328 "SendSealHandle",

1329 "RecvSealHandle",

1330 "SendSeqNum",

1331 "RecvSeqNum",

1332 "neg_tok",

1333 "chall_tok",

1334 "ServerHostname",

1335 "ServerDomain",

1336 ]

1337

1338 def __init__(self, IsAcceptor, req_flags=None):

1339 self.state = NTLMSSP.STATE.INIT

1340 self.SessionKey = None

1341 self.ExportedSessionKey = None

1342 self.SendSignKey = None

1343 self.SendSealKey = None

1344 self.SendSealHandle = None

1345 self.RecvSignKey = None

1346 self.RecvSealKey = None

1347 self.RecvSealHandle = None

1348 self.SendSeqNum = 0

1349 self.RecvSeqNum = 0

1350 self.neg_tok = None

1351 self.chall_tok = None

1352 self.ServerHostname = None

1353 self.ServerDomain = None

1354 self.IsAcceptor = IsAcceptor

1355 super(NTLMSSP.CONTEXT, self).__init__(req_flags=req_flags)

1356

1357 def clifailure(self):

1358 self.__init__(self.IsAcceptor, req_flags=self.flags)

1359

1360 def __repr__(self):

1361 return "NTLMSSP"

1362

1363 # [MS-NLMP] note <36>: "the maximum lifetime is 36 hours" (lol, Kerberos has 5min)

1364 NTLM_MaxLifetime = 36 * 3600

1365

1366 def __init__(

1367 self,

1368 UPN=None,

1369 HASHNT=None,

1370 PASSWORD=None,

1371 USE_MIC=True,

1372 VARIANT: NTLM_VARIANT = NTLM_VARIANT.RECENT,

1373 NTLM_VALUES={},

1374 DOMAIN_FQDN=None,

1375 DOMAIN_NB_NAME=None,

1376 COMPUTER_NB_NAME=None,

1377 COMPUTER_FQDN=None,

1378 IDENTITIES=None,

1379 DO_NOT_CHECK_LOGIN=False,

1380 SERVER_CHALLENGE=None,

1381 **kwargs,

1382 ):

1383 self.UPN = UPN

1384 if HASHNT is None and PASSWORD is not None:

1385 HASHNT = MD4le(PASSWORD)

1386 self.HASHNT = HASHNT

1387 self.VARIANT = VARIANT

1388 if self.VARIANT != NTLM_VARIANT.RECENT:

1389 log_runtime.warning(

1390 "VARIANT != NTLM_VARIANT.RECENT. You shouldn't touch this !"

1391 )

1392 self.USE_MIC = False

1393 else:

1394 self.USE_MIC = USE_MIC

1395 self.NTLM_VALUES = NTLM_VALUES

1396 if UPN is not None:

1397 # Populate values used only in server mode.

1398 from scapy.layers.kerberos import _parse_upn

1399

1400 try:

1401 user, realm = _parse_upn(UPN)

1402 if DOMAIN_FQDN is None:

1403 DOMAIN_FQDN = realm

1404 if COMPUTER_NB_NAME is None:

1405 COMPUTER_NB_NAME = user

1406 except ValueError:

1407 pass

1408

1409 # Compute various netbios/fqdn names

1410 self.DOMAIN_FQDN = DOMAIN_FQDN or "domain.local"

1411 self.DOMAIN_NB_NAME = (

1412 DOMAIN_NB_NAME or self.DOMAIN_FQDN.split(".")[0].upper()[:15]

1413 )

1414 self.COMPUTER_NB_NAME = COMPUTER_NB_NAME or "WIN10"

1415 self.COMPUTER_FQDN = COMPUTER_FQDN or (

1416 self.COMPUTER_NB_NAME.lower() + "." + self.DOMAIN_FQDN

1417 )

1418

1419 self.IDENTITIES = IDENTITIES

1420 self.DO_NOT_CHECK_LOGIN = DO_NOT_CHECK_LOGIN

1421 self.SERVER_CHALLENGE = SERVER_CHALLENGE

1422 super(NTLMSSP, self).__init__(**kwargs)

1423

1424 def LegsAmount(self, Context: CONTEXT):

1425 return 3

1426

1427 def GSS_Inquire_names_for_mech(self):

1428 return ["1.3.6.1.4.1.311.2.2.10"]

1429

1430 def GSS_GetMICEx(self, Context, msgs, qop_req=0):

1431 """

1432 [MS-NLMP] sect 3.4.8

1433 """

1434 # Concatenate the ToSign

1435 ToSign = b"".join(x.data for x in msgs if x.sign)

1436 sig = MAC(

1437 Context.SendSealHandle,

1438 Context.SendSignKey,

1439 Context.SendSeqNum,

1440 ToSign,

1441 )

1442 Context.SendSeqNum += 1

1443 return sig

1444

1445 def GSS_VerifyMICEx(self, Context, msgs, signature):

1446 """

1447 [MS-NLMP] sect 3.4.9

1448 """

1449 Context.RecvSeqNum = signature.SeqNum

1450 # Concatenate the ToSign

1451 ToSign = b"".join(x.data for x in msgs if x.sign)

1452 sig = MAC(

1453 Context.RecvSealHandle,

1454 Context.RecvSignKey,

1455 Context.RecvSeqNum,

1456 ToSign,

1457 )

1458 if sig.Checksum != signature.Checksum:

1459 raise ValueError("ERROR: Checksums don't match")

1460

1461 def GSS_WrapEx(self, Context, msgs, qop_req=0):

1462 """

1463 [MS-NLMP] sect 3.4.6

1464 """

1465 msgs_cpy = copy.deepcopy(msgs) # Keep copy for signature

1466 # Encrypt

1467 for msg in msgs:

1468 if msg.conf_req_flag:

1469 msg.data = RC4(Context.SendSealHandle, msg.data)

1470 # Sign

1471 sig = self.GSS_GetMICEx(Context, msgs_cpy, qop_req=qop_req)

1472 return (

1473 msgs,

1474 sig,

1475 )

1476

1477 def GSS_UnwrapEx(self, Context, msgs, signature):

1478 """

1479 [MS-NLMP] sect 3.4.7

1480 """

1481 # Decrypt

1482 for msg in msgs:

1483 if msg.conf_req_flag:

1484 msg.data = RC4(Context.RecvSealHandle, msg.data)

1485 # Check signature

1486 self.GSS_VerifyMICEx(Context, msgs, signature)

1487 return msgs

1488

1489 def SupportsMechListMIC(self):

1490 if not self.USE_MIC:

1491 # RFC 4178

1492 # "If the mechanism selected by the negotiation does not support integrity

1493 # protection, then no mechlistMIC token is used."

1494 return False

1495 if self.DO_NOT_CHECK_LOGIN:

1496 # In this mode, we won't negotiate any credentials.

1497 return False

1498 return True

1499

1500 def GetMechListMIC(self, Context, input):

1501 # [MS-SPNG]

1502 # "When NTLM is negotiated, the SPNG server MUST set OriginalHandle to

1503 # ServerHandle before generating the mechListMIC, then set ServerHandle to

1504 # OriginalHandle after generating the mechListMIC."

1505 OriginalHandle = Context.SendSealHandle

1506 Context.SendSealHandle = RC4Init(Context.SendSealKey)

1507 try:

1508 return super(NTLMSSP, self).GetMechListMIC(Context, input)

1509 finally:

1510 Context.SendSealHandle = OriginalHandle

1511

1512 def VerifyMechListMIC(self, Context, otherMIC, input):

1513 # [MS-SPNG]

1514 # "the SPNEGO Extension server MUST set OriginalHandle to ClientHandle before

1515 # validating the mechListMIC and then set ClientHandle to OriginalHandle after

1516 # validating the mechListMIC."

1517 OriginalHandle = Context.RecvSealHandle

1518 Context.RecvSealHandle = RC4Init(Context.RecvSealKey)

1519 try:

1520 return super(NTLMSSP, self).VerifyMechListMIC(Context, otherMIC, input)

1521 finally:

1522 Context.RecvSealHandle = OriginalHandle

1523

1524 def GSS_Init_sec_context(

1525 self,

1526 Context: CONTEXT,

1527 input_token=None,

1528 target_name: Optional[str] = None,

1529 req_flags: Optional[GSS_C_FLAGS] = None,

1530 chan_bindings: GssChannelBindings = GSS_C_NO_CHANNEL_BINDINGS,

1531 ):

1532 if Context is None:

1533 Context = self.CONTEXT(False, req_flags=req_flags)

1534

1535 if Context.state == self.STATE.INIT:

1536 # Client: negotiate

1537 # Create a default token

1538 tok = NTLM_NEGOTIATE(

1539 VARIANT=self.VARIANT,

1540 NegotiateFlags="+".join(

1541 [

1542 "NEGOTIATE_UNICODE",

1543 "REQUEST_TARGET",

1544 "NEGOTIATE_NTLM",

1545 "NEGOTIATE_ALWAYS_SIGN",

1546 "TARGET_TYPE_DOMAIN",

1547 "NEGOTIATE_EXTENDED_SESSIONSECURITY",

1548 "NEGOTIATE_TARGET_INFO",

1549 "NEGOTIATE_128",

1550 "NEGOTIATE_56",

1551 ]

1552 + (

1553 ["NEGOTIATE_VERSION"]

1554 if self.VARIANT >= NTLM_VARIANT.XP_OR_2003

1555 else []

1556 )

1557 + (

1558 [

1559 "NEGOTIATE_KEY_EXCH",

1560 ]

1561 if Context.flags

1562 & (GSS_C_FLAGS.GSS_C_INTEG_FLAG | GSS_C_FLAGS.GSS_C_CONF_FLAG)

1563 else []

1564 )

1565 + (

1566 [

1567 "NEGOTIATE_SIGN",

1568 ]

1569 if Context.flags & GSS_C_FLAGS.GSS_C_INTEG_FLAG

1570 else []

1571 )

1572 + (

1573 [

1574 "NEGOTIATE_SEAL",

1575 ]

1576 if Context.flags & GSS_C_FLAGS.GSS_C_CONF_FLAG

1577 else []

1578 )

1579 ),

1580 ProductMajorVersion=10,

1581 ProductMinorVersion=0,

1582 ProductBuild=19041,

1583 )

1584 if self.NTLM_VALUES:

1585 # Update that token with the customs one

1586 for key in [

1587 "NegotiateFlags",

1588 "ProductMajorVersion",

1589 "ProductMinorVersion",

1590 "ProductBuild",

1591 ]:

1592 if key in self.NTLM_VALUES:

1593 setattr(tok, key, self.NTLM_VALUES[key])

1594 Context.neg_tok = tok

1595 Context.SessionKey = None # Reset signing (if previous auth failed)

1596 Context.state = self.STATE.CLI_SENT_NEGO

1597 return Context, tok, GSS_S_CONTINUE_NEEDED

1598 elif Context.state == self.STATE.CLI_SENT_NEGO:

1599 # Client: auth (token=challenge)

1600 chall_tok = input_token

1601 if self.UPN is None or self.HASHNT is None:

1602 raise ValueError(

1603 "Must provide a 'UPN' and a 'HASHNT' or 'PASSWORD' when "

1604 "running in standalone !"

1605 )

1606

1607 from scapy.layers.kerberos import _parse_upn

1608

1609 # Check token sanity

1610 if not chall_tok or NTLM_CHALLENGE not in chall_tok:

1611 log_runtime.debug("NTLMSSP: Unexpected token. Expected NTLM Challenge")

1612 return Context, None, GSS_S_DEFECTIVE_TOKEN

1613

1614 # Some information from the CHALLENGE are stored

1615 try:

1616 Context.ServerHostname = chall_tok.getAv(0x0001).Value

1617 except IndexError:

1618 pass

1619 try:

1620 Context.ServerDomain = chall_tok.getAv(0x0002).Value

1621 except IndexError:

1622 pass

1623 try:

1624 # the server SHOULD set the timestamp in the CHALLENGE_MESSAGE

1625 ServerTimestamp = chall_tok.getAv(0x0007).Value

1626 ServerTime = (ServerTimestamp / 1e7) - 11644473600

1627

1628 if abs(ServerTime - time.time()) >= NTLMSSP.NTLM_MaxLifetime:

1629 log_runtime.warning(

1630 "Server and Client times are off by more than 36h !"

1631 )

1632 # We could error here, but we don't.

1633 except IndexError:

1634 pass

1635

1636 # Initialize a default token

1637 tok = NTLM_AUTHENTICATE_V2(

1638 VARIANT=self.VARIANT,

1639 NegotiateFlags=chall_tok.NegotiateFlags,

1640 ProductMajorVersion=10,

1641 ProductMinorVersion=0,

1642 ProductBuild=19041,

1643 )

1644 tok.LmChallengeResponse = LMv2_RESPONSE()

1645

1646 # Populate the token

1647 # 1. Set username

1648 try:

1649 tok.UserName, realm = _parse_upn(self.UPN)

1650 except ValueError:

1651 tok.UserName, realm = self.UPN, Context.ServerDomain

1652

1653 # 2. Set domain name

1654 if realm is None:

1655 log_runtime.warning(

1656 "No realm specified in UPN, nor provided by server."

1657 )

1658 tok.DomainName = self.DOMAIN_FQDN

1659 else:

1660 tok.DomainName = realm

1661

1662 # 3. Set workstation name

1663 tok.Workstation = self.COMPUTER_NB_NAME

1664

1665 # 4. Create and calculate the ChallengeResponse

1666 # 4.1 Build the payload

1667 cr = tok.NtChallengeResponse = NTLMv2_RESPONSE(

1668 ChallengeFromClient=os.urandom(8),

1669 )

1670 cr.TimeStamp = int((time.time() + 11644473600) * 1e7)

1671 cr.AvPairs = (

1672 # Repeat AvPairs from the server

1673 chall_tok.TargetInfo[:-1]

1674 + (

1675 [

1676 AV_PAIR(AvId="MsvAvFlags", Value="MIC integrity"),

1677 ]

1678 if self.USE_MIC

1679 else []

1680 )

1681 + [

1682 AV_PAIR(

1683 AvId="MsvAvSingleHost",

1684 Value=Single_Host_Data(MachineID=os.urandom(32)),

1685 ),

1686 ]

1687 + (

1688 [

1689 AV_PAIR(

1690 # [MS-NLMP] sect 2.2.2.1 refers to RFC 4121 sect 4.1.1.2

1691 # "The Bnd field contains the MD5 hash of channel bindings"

1692 AvId="MsvAvChannelBindings",

1693 Value=chan_bindings.digestMD5(),

1694 ),

1695 ]

1696 if chan_bindings != GSS_C_NO_CHANNEL_BINDINGS

1697 else []

1698 )

1699 + [

1700 AV_PAIR(

1701 AvId="MsvAvTargetName",

1702 Value=target_name or ("host/" + Context.ServerHostname),

1703 ),

1704 AV_PAIR(AvId="MsvAvEOL"),

1705 ]

1706 )

1707 if self.NTLM_VALUES:

1708 # Update that token with the customs one

1709 for key in [

1710 "NegotiateFlags",

1711 "ProductMajorVersion",

1712 "ProductMinorVersion",

1713 "ProductBuild",

1714 ]:

1715 if key in self.NTLM_VALUES:

1716 setattr(tok, key, self.NTLM_VALUES[key])

1717

1718 # 4.2 Compute the ResponseKeyNT

1719 ResponseKeyNT = NTOWFv2(

1720 None,

1721 tok.UserName,

1722 tok.DomainName,

1723 HashNt=self.HASHNT,

1724 )

1725

1726 # 4.3 Compute the NTProofStr

1727 cr.NTProofStr = cr.computeNTProofStr(

1728 ResponseKeyNT,

1729 chall_tok.ServerChallenge,

1730 )

1731

1732 # 4.4 Compute the Session Key

1733 SessionBaseKey = NTLMv2_ComputeSessionBaseKey(ResponseKeyNT, cr.NTProofStr)

1734 KeyExchangeKey = SessionBaseKey # Only true for NTLMv2

1735 if chall_tok.NegotiateFlags.NEGOTIATE_KEY_EXCH:

1736 ExportedSessionKey = os.urandom(16)

1737 tok.EncryptedRandomSessionKey = RC4K(

1738 KeyExchangeKey,

1739 ExportedSessionKey,

1740 )

1741 else:

1742 ExportedSessionKey = KeyExchangeKey

1743

1744 # 4.5 Compute the MIC

1745 if self.USE_MIC:

1746 tok.compute_mic(ExportedSessionKey, Context.neg_tok, chall_tok)

1747

1748 # 5. Perform key computations

1749 Context.ExportedSessionKey = ExportedSessionKey

1750 # [MS-SMB] 3.2.5.3

1751 Context.SessionKey = Context.ExportedSessionKey

1752 # Compute NTLM keys

1753 Context.SendSignKey = SIGNKEY(

1754 tok.NegotiateFlags, ExportedSessionKey, "Client"

1755 )

1756 Context.SendSealKey = SEALKEY(

1757 tok.NegotiateFlags, ExportedSessionKey, "Client"

1758 )

1759 Context.SendSealHandle = RC4Init(Context.SendSealKey)

1760 Context.RecvSignKey = SIGNKEY(

1761 tok.NegotiateFlags, ExportedSessionKey, "Server"

1762 )

1763 Context.RecvSealKey = SEALKEY(

1764 tok.NegotiateFlags, ExportedSessionKey, "Server"

1765 )

1766 Context.RecvSealHandle = RC4Init(Context.RecvSealKey)

1767

1768 # Update the state

1769 Context.state = self.STATE.CLI_SENT_AUTH

1770

1771 return Context, tok, GSS_S_COMPLETE

1772 elif Context.state == self.STATE.CLI_SENT_AUTH:

1773 if input_token:

1774 # what is that?

1775 status = GSS_S_DEFECTIVE_TOKEN

1776 else:

1777 status = GSS_S_COMPLETE

1778 return Context, None, status

1779 else:

1780 raise ValueError("NTLMSSP: unexpected state %s" % repr(Context.state))

1781

1782 def GSS_Accept_sec_context(

1783 self,

1784 Context: CONTEXT,

1785 input_token=None,

1786 req_flags: Optional[GSS_S_FLAGS] = GSS_S_FLAGS.GSS_S_ALLOW_MISSING_BINDINGS,

1787 chan_bindings: GssChannelBindings = GSS_C_NO_CHANNEL_BINDINGS,

1788 ):

1789 if Context is None:

1790 Context = self.CONTEXT(IsAcceptor=True, req_flags=req_flags)

1791

1792 if Context.state == self.STATE.INIT:

1793 # Server: challenge (input_token=negotiate)

1794 nego_tok = input_token

1795 if not nego_tok or NTLM_NEGOTIATE not in nego_tok:

1796 log_runtime.debug("NTLMSSP: Unexpected token. Expected NTLM Negotiate")

1797 return Context, None, GSS_S_DEFECTIVE_TOKEN

1798

1799 # Build the challenge token

1800 currentTime = (time.time() + 11644473600) * 1e7

1801 tok = NTLM_CHALLENGE(

1802 VARIANT=self.VARIANT,

1803 ServerChallenge=self.SERVER_CHALLENGE or os.urandom(8),

1804 NegotiateFlags="+".join(

1805 [

1806 "NEGOTIATE_UNICODE",

1807 "REQUEST_TARGET",

1808 "NEGOTIATE_NTLM",

1809 "NEGOTIATE_ALWAYS_SIGN",

1810 "NEGOTIATE_EXTENDED_SESSIONSECURITY",

1811 "NEGOTIATE_TARGET_INFO",

1812 "TARGET_TYPE_DOMAIN",

1813 "NEGOTIATE_128",

1814 "NEGOTIATE_KEY_EXCH",

1815 "NEGOTIATE_56",

1816 ]

1817 + (

1818 ["NEGOTIATE_VERSION"]

1819 if self.VARIANT >= NTLM_VARIANT.XP_OR_2003

1820 else []

1821 )

1822 + (

1823 ["NEGOTIATE_SIGN"]

1824 if nego_tok.NegotiateFlags.NEGOTIATE_SIGN

1825 else []

1826 )

1827 + (

1828 ["NEGOTIATE_SEAL"]

1829 if nego_tok.NegotiateFlags.NEGOTIATE_SEAL

1830 else []

1831 )

1832 ),

1833 ProductMajorVersion=10,

1834 ProductMinorVersion=0,

1835 Payload=[

1836 ("TargetName", ""),

1837 (

1838 "TargetInfo",

1839 [

1840 # MsvAvNbComputerName

1841 AV_PAIR(AvId=1, Value=self.COMPUTER_NB_NAME),

1842 # MsvAvNbDomainName

1843 AV_PAIR(AvId=2, Value=self.DOMAIN_NB_NAME),

1844 # MsvAvDnsComputerName

1845 AV_PAIR(AvId=3, Value=self.COMPUTER_FQDN),

1846 # MsvAvDnsDomainName

1847 AV_PAIR(AvId=4, Value=self.DOMAIN_FQDN),

1848 # MsvAvDnsTreeName

1849 AV_PAIR(AvId=5, Value=self.DOMAIN_FQDN),

1850 # MsvAvTimestamp

1851 AV_PAIR(AvId=7, Value=currentTime),

1852 # MsvAvEOL

1853 AV_PAIR(AvId=0),

1854 ],

1855 ),

1856 ],

1857 )

1858 if self.NTLM_VALUES:

1859 # Update that token with the customs one

1860 for key in [

1861 "ServerChallenge",

1862 "NegotiateFlags",

1863 "ProductMajorVersion",

1864 "ProductMinorVersion",

1865 "TargetName",

1866 ]:

1867 if key in self.NTLM_VALUES:

1868 setattr(tok, key, self.NTLM_VALUES[key])

1869 avpairs = {x.AvId: x.Value for x in tok.TargetInfo}

1870 tok.TargetInfo = [

1871 AV_PAIR(AvId=i, Value=self.NTLM_VALUES.get(x, avpairs[i]))

1872 for (i, x) in [

1873 (2, "NetbiosDomainName"),

1874 (1, "NetbiosComputerName"),

1875 (4, "DnsDomainName"),

1876 (3, "DnsComputerName"),

1877 (5, "DnsTreeName"),

1878 (6, "Flags"),

1879 (7, "Timestamp"),

1880 (0, None),

1881 ]

1882 if ((x in self.NTLM_VALUES) or (i in avpairs))

1883 and self.NTLM_VALUES.get(x, True) is not None

1884 ]

1885

1886 # Store for next step

1887 Context.chall_tok = tok

1888

1889 # Update the state

1890 Context.state = self.STATE.SRV_SENT_CHAL

1891

1892 return Context, tok, GSS_S_CONTINUE_NEEDED

1893 elif Context.state == self.STATE.SRV_SENT_CHAL:

1894 # server: OK or challenge again (input_token=auth)

1895 auth_tok = input_token

1896

1897 if not auth_tok or NTLM_AUTHENTICATE_V2 not in auth_tok:

1898 log_runtime.debug(

1899 "NTLMSSP: Unexpected token. Expected NTLM Authenticate v2"

1900 )

1901 return Context, None, GSS_S_DEFECTIVE_TOKEN

1902

1903 if self.DO_NOT_CHECK_LOGIN:

1904 # Just trust me bro. Typically used in "guest" mode.

1905 return Context, None, GSS_S_COMPLETE

1906

1907 # Compute the session key

1908 SessionBaseKey = self._getSessionBaseKey(Context, auth_tok)

1909 if SessionBaseKey:

1910 # [MS-NLMP] sect 3.2.5.1.2

1911 KeyExchangeKey = SessionBaseKey # Only true for NTLMv2

1912 if auth_tok.NegotiateFlags.NEGOTIATE_KEY_EXCH:

1913 try:

1914 EncryptedRandomSessionKey = auth_tok.EncryptedRandomSessionKey

1915 except AttributeError:

1916 # No EncryptedRandomSessionKey. libcurl for instance

1917 # hmm. this looks bad

1918 EncryptedRandomSessionKey = b"\x00" * 16

1919 ExportedSessionKey = RC4K(KeyExchangeKey, EncryptedRandomSessionKey)

1920 else:

1921 ExportedSessionKey = KeyExchangeKey

1922 Context.ExportedSessionKey = ExportedSessionKey

1923 # [MS-SMB] 3.2.5.3

1924 Context.SessionKey = Context.ExportedSessionKey

1925

1926 # Check the timestamp

1927 try:

1928 ClientTimestamp = auth_tok.NtChallengeResponse.getAv(0x0007).Value

1929 ClientTime = (ClientTimestamp / 1e7) - 11644473600

1930

1931 if abs(ClientTime - time.time()) >= NTLMSSP.NTLM_MaxLifetime:

1932 log_runtime.warning(

1933 "Server and Client times are off by more than 36h !"

1934 )

1935 # We could error here, but we don't.

1936 except IndexError:

1937 pass

1938

1939 # Check the channel bindings

1940 if chan_bindings != GSS_C_NO_CHANNEL_BINDINGS:

1941 try:

1942 Bnd = auth_tok.NtChallengeResponse.getAv(0x000A).Value

1943 if Bnd != chan_bindings.digestMD5():

1944 # Bad channel bindings

1945 return Context, None, GSS_S_BAD_BINDINGS

1946 except IndexError:

1947 if GSS_S_FLAGS.GSS_S_ALLOW_MISSING_BINDINGS not in req_flags:

1948 # Uhoh, we required channel bindings

1949 return Context, None, GSS_S_BAD_BINDINGS

1950

1951 if Context.SessionKey:

1952 # Compute NTLM keys

1953 Context.SendSignKey = SIGNKEY(

1954 auth_tok.NegotiateFlags, ExportedSessionKey, "Server"

1955 )

1956 Context.SendSealKey = SEALKEY(

1957 auth_tok.NegotiateFlags, ExportedSessionKey, "Server"

1958 )

1959 Context.SendSealHandle = RC4Init(Context.SendSealKey)

1960 Context.RecvSignKey = SIGNKEY(

1961 auth_tok.NegotiateFlags, ExportedSessionKey, "Client"

1962 )

1963 Context.RecvSealKey = SEALKEY(

1964 auth_tok.NegotiateFlags, ExportedSessionKey, "Client"

1965 )

1966 Context.RecvSealHandle = RC4Init(Context.RecvSealKey)

1967

1968 # Check the NTProofStr

1969 if self._checkLogin(Context, auth_tok):

1970 # Set negotiated flags

1971 if auth_tok.NegotiateFlags.NEGOTIATE_SIGN:

1972 Context.flags |= GSS_C_FLAGS.GSS_C_INTEG_FLAG

1973 if auth_tok.NegotiateFlags.NEGOTIATE_SEAL:

1974 Context.flags |= GSS_C_FLAGS.GSS_C_CONF_FLAG

1975 return Context, None, GSS_S_COMPLETE

1976

1977 # Bad NTProofStr or unknown user

1978 Context.SessionKey = None

1979 Context.state = self.STATE.INIT

1980 return Context, None, GSS_S_DEFECTIVE_CREDENTIAL

1981 else:

1982 raise ValueError("NTLMSSP: unexpected state %s" % repr(Context.state))

1983

1984 def MaximumSignatureLength(self, Context: CONTEXT):

1985 """

1986 Returns the Maximum Signature length.

1987

1988 This will be used in auth_len in DceRpc5, and is necessary for

1989 PFC_SUPPORT_HEADER_SIGN to work properly.

1990 """

1991 return 16 # len(NTLMSSP_MESSAGE_SIGNATURE())

1992

1993 def GSS_Passive(self, Context: CONTEXT, token=None, req_flags=None):

1994 if Context is None:

1995 Context = self.CONTEXT(True)

1996 Context.passive = True

1997

1998 # We capture the Negotiate, Challenge, then call the server's auth handling

1999 # and discard the output.

2000

2001 if Context.state == self.STATE.INIT:

2002 if not token or NTLM_NEGOTIATE not in token:

2003 log_runtime.warning("NTLMSSP: Expected NTLM Negotiate")

2004 return None, GSS_S_DEFECTIVE_TOKEN

2005 Context.neg_tok = token

2006 Context.state = self.STATE.CLI_SENT_NEGO

2007 return Context, GSS_S_CONTINUE_NEEDED

2008 elif Context.state == self.STATE.CLI_SENT_NEGO:

2009 if not token or NTLM_CHALLENGE not in token:

2010 log_runtime.warning("NTLMSSP: Expected NTLM Challenge")

2011 return None, GSS_S_DEFECTIVE_TOKEN

2012 Context.chall_tok = token

2013 Context.state = self.STATE.SRV_SENT_CHAL

2014 return Context, GSS_S_CONTINUE_NEEDED

2015 elif Context.state == self.STATE.SRV_SENT_CHAL:

2016 if not token or NTLM_AUTHENTICATE_V2 not in token:

2017 log_runtime.warning("NTLMSSP: Expected NTLM Authenticate")

2018 return None, GSS_S_DEFECTIVE_TOKEN

2019 Context, _, status = self.GSS_Accept_sec_context(Context, token)

2020 if status != GSS_S_COMPLETE:

2021 log_runtime.info("NTLMSSP: auth failed.")

2022 Context.state = self.STATE.INIT

2023 return Context, status

2024 else:

2025 raise ValueError("NTLMSSP: unexpected state %s" % repr(Context.state))

2026

2027 def GSS_Passive_set_Direction(self, Context: CONTEXT, IsAcceptor=False):

2028 if Context.IsAcceptor is not IsAcceptor:

2029 return

2030 # Swap everything

2031 Context.SendSignKey, Context.RecvSignKey = (

2032 Context.RecvSignKey,

2033 Context.SendSignKey,

2034 )

2035 Context.SendSealKey, Context.RecvSealKey = (

2036 Context.RecvSealKey,

2037 Context.SendSealKey,

2038 )

2039 Context.SendSealHandle, Context.RecvSealHandle = (

2040 Context.RecvSealHandle,

2041 Context.SendSealHandle,

2042 )

2043 Context.SendSeqNum, Context.RecvSeqNum = Context.RecvSeqNum, Context.SendSeqNum

2044 Context.IsAcceptor = not Context.IsAcceptor

2045

2046 def _getSessionBaseKey(self, Context, auth_tok):

2047 """

2048 Function that returns the SessionBaseKey from the ntlm Authenticate.

2049 """

2050 try:

2051 username = auth_tok.UserName

2052 except AttributeError:

2053 username = None

2054 try:

2055 domain = auth_tok.DomainName

2056 except AttributeError:

2057 domain = ""

2058 if self.IDENTITIES and username in self.IDENTITIES:

2059 ResponseKeyNT = NTOWFv2(

2060 None,

2061 username,

2062 domain,

2063 HashNt=self.IDENTITIES[username],

2064 )

2065 return NTLMv2_ComputeSessionBaseKey(

2066 ResponseKeyNT,

2067 auth_tok.NtChallengeResponse.NTProofStr,

2068 )

2069 elif self.IDENTITIES:

2070 log_runtime.debug("NTLMSSP: Bad credentials for %s" % username)

2071 return None

2072

2073 def _checkLogin(self, Context, auth_tok):

2074 """

2075 Function that checks the validity of an authentication.

2076

2077 Overwrite and return True to bypass.

2078 """

2079 # Create the NTLM AUTH

2080 try:

2081 username = auth_tok.UserName

2082 except AttributeError:

2083 username = None

2084 try:

2085 domain = auth_tok.DomainName

2086 except AttributeError:

2087 domain = ""

2088 if username in self.IDENTITIES:

2089 ResponseKeyNT = NTOWFv2(

2090 None,

2091 username,

2092 domain,

2093 HashNt=self.IDENTITIES[username],

2094 )

2095 NTProofStr = auth_tok.NtChallengeResponse.computeNTProofStr(

2096 ResponseKeyNT,

2097 Context.chall_tok.ServerChallenge,

2098 )

2099 if NTProofStr == auth_tok.NtChallengeResponse.NTProofStr:

2100 return True

2101 return False

2102

2103

2104class NTLMSSP_DOMAIN(NTLMSSP):

2105 """

2106 A variant of the NTLMSSP to be used in server mode that gets the session

2107 keys from the domain using a Netlogon channel.

2108

2109 This has the same arguments as NTLMSSP, but supports the following in server

2110 mode:

2111

2112 :param UPN: the UPN of the machine account to login for Netlogon.

2113 :param HASHNT: the HASHNT of the machine account (use Netlogon secure channel).

2114 :param ssp: a KerberosSSP to use (use Kerberos secure channel).

2115 :param PASSWORD: the PASSWORD of the machine account to use for Netlogon.

2116 :param DC_IP: (optional) specify the IP of the DC.

2117

2118 Netlogon example::

2119

2120 >>> mySSP = NTLMSSP_DOMAIN(

2121 ... UPN="Server1@domain.local",

2122 ... HASHNT=bytes.fromhex("8846f7eaee8fb117ad06bdd830b7586c"),

2123 ... )

2124

2125 Kerberos example::

2126

2127 >>> mySSP = NTLMSSP_DOMAIN(

2128 ... UPN="Server1@domain.local",

2129 ... KEY=Key(EncryptionType.AES256_CTS_HMAC_SHA1_96,

2130 ... key=bytes.fromhex(

2131 ... "85abb9b61dc2fa49d4cc04317bbd108f8f79df28"

2132 ... "239155ed7b144c5d2ebcf016"

2133 ... )

2134 ... ),

2135 ... )

2136 """

2137

2138 def __init__(self, UPN=None, *args, timeout=3, ssp=None, **kwargs):

2139 from scapy.layers.kerberos import KerberosSSP

2140

2141 # Either PASSWORD or HASHNT or ssp

2142 if (

2143 "HASHNT" not in kwargs

2144 and "PASSWORD" not in kwargs

2145 and "KEY" not in kwargs

2146 and ssp is None

2147 ):

2148 raise ValueError(

2149 "Must specify either 'HASHNT', 'PASSWORD' or "

2150 "provide a ssp=KerberosSSP()"

2151 )

2152 elif ssp is not None and not isinstance(ssp, KerberosSSP):

2153 raise ValueError("'ssp' can only be None or a KerberosSSP !")

2154

2155 self.KEY = kwargs.pop("KEY", None)

2156 self.PASSWORD = kwargs.get("PASSWORD", None)

2157

2158 # UPN is mandatory

2159 if UPN is None and ssp is not None and ssp.UPN:

2160 UPN = ssp.UPN

2161 elif UPN is None:

2162 raise ValueError("Must specify a 'UPN' !")

2163 kwargs["UPN"] = UPN

2164

2165 # Call parent

2166 super(NTLMSSP_DOMAIN, self).__init__(

2167 *args,

2168 **kwargs,

2169 )

2170

2171 # Treat specific parameters

2172 self.DC_FQDN = kwargs.pop("DC_FQDN", None)

2173 if self.DC_FQDN is None:

2174 # Get DC_FQDN from dclocator

2175 from scapy.layers.ldap import dclocator

2176

2177 dc = dclocator(

2178 self.DOMAIN_FQDN,

2179 timeout=timeout,

2180 debug=kwargs.get("debug", 0),

2181 )

2182 self.DC_FQDN = dc.samlogon.DnsHostName.decode().rstrip(".")

2183

2184 # If logging in via Kerberos

2185 self.ssp = ssp

2186

2187 def _getSessionBaseKey(self, Context, ntlm):

2188 """

2189 Return the Session Key by asking the DC.

2190 """

2191 # No user / no domain: skip.

2192 if not ntlm.UserNameLen or not ntlm.DomainNameLen:

2193 return super(NTLMSSP_DOMAIN, self)._getSessionBaseKey(Context, ntlm)

2194

2195 # Import RPC stuff

2196 from scapy.layers.dcerpc import NDRUnion

2197 from scapy.layers.msrpce.msnrpc import (

2198 NETLOGON_SECURE_CHANNEL_METHOD,

2199 NetlogonClient,

2200 )

2201 from scapy.layers.msrpce.raw.ms_nrpc import (

2202 NETLOGON_LOGON_IDENTITY_INFO,

2203 NetrLogonSamLogonWithFlags_Request,

2204 PNETLOGON_AUTHENTICATOR,

2205 PNETLOGON_NETWORK_INFO,

2206 STRING,

2207 UNICODE_STRING,

2208 )

2209

2210 # Create NetlogonClient with PRIVACY

2211 client = NetlogonClient()

2212 client.connect(self.DC_FQDN)

2213

2214 # Establish the Netlogon secure channel (this will bind)

2215 try:

2216 if self.ssp is None and self.KEY is None:

2217 # Login via classic NetlogonSSP

2218 client.establish_secure_channel(

2219 mode=NETLOGON_SECURE_CHANNEL_METHOD.NetrServerAuthenticate3,

2220 UPN=f"{self.COMPUTER_NB_NAME}@{self.DOMAIN_NB_NAME}",

2221 DC_FQDN=self.DC_FQDN,

2222 HASHNT=self.HASHNT,

2223 )

2224 else:

2225 # Login via KerberosSSP (Windows 2025)

2226 client.establish_secure_channel(

2227 mode=NETLOGON_SECURE_CHANNEL_METHOD.NetrServerAuthenticateKerberos,

2228 UPN=self.UPN,

2229 DC_FQDN=self.DC_FQDN,

2230 PASSWORD=self.PASSWORD,

2231 KEY=self.KEY,

2232 ssp=self.ssp,

2233 )

2234 except ValueError:

2235 log_runtime.warning(

2236 "Couldn't establish the Netlogon secure channel. "

2237 "Check the credentials for '%s' !" % self.COMPUTER_NB_NAME

2238 )

2239 return super(NTLMSSP_DOMAIN, self)._getSessionBaseKey(Context, ntlm)

2240

2241 # Request validation of the NTLM request

2242 req = NetrLogonSamLogonWithFlags_Request(

2243 LogonServer="",

2244 ComputerName=self.COMPUTER_NB_NAME,

2245 Authenticator=client.create_authenticator(),

2246 ReturnAuthenticator=PNETLOGON_AUTHENTICATOR(),

2247 LogonLevel=6, # NetlogonNetworkTransitiveInformation

2248 LogonInformation=NDRUnion(

2249 tag=6,

2250 value=PNETLOGON_NETWORK_INFO(

2251 Identity=NETLOGON_LOGON_IDENTITY_INFO(

2252 LogonDomainName=UNICODE_STRING(

2253 Buffer=ntlm.DomainName,

2254 ),

2255 ParameterControl=0x00002AE0,

2256 UserName=UNICODE_STRING(

2257 Buffer=ntlm.UserName,

2258 ),

2259 Workstation=UNICODE_STRING(

2260 Buffer=ntlm.Workstation,

2261 ),

2262 ),

2263 LmChallenge=Context.chall_tok.ServerChallenge,

2264 NtChallengeResponse=STRING(

2265 Buffer=bytes(ntlm.NtChallengeResponse),

2266 ),

2267 LmChallengeResponse=STRING(

2268 Buffer=bytes(ntlm.LmChallengeResponse),

2269 ),

2270 ),

2271 ),

2272 ValidationLevel=6,

2273 ExtraFlags=0,

2274 ndr64=client.ndr64,

2275 )

2276

2277 # Get response

2278 resp = client.sr1_req(req)

2279 if resp and resp.status == 0:

2280 # Success

2281

2282 # Validate DC authenticator

2283 client.validate_authenticator(resp.ReturnAuthenticator.value)

2284

2285 # Get and return the SessionKey

2286 UserSessionKey = resp.ValidationInformation.value.value.UserSessionKey

2287 return bytes(UserSessionKey)

2288 else:

2289 # Failed

2290 return super(NTLMSSP_DOMAIN, self)._getSessionBaseKey(Context, ntlm)

2291

2292 def _checkLogin(self, Context, auth_tok):

2293 # Always OK if we got the session key

2294 return True