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

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 GSS_C_FLAGS,

65 GSS_C_NO_CHANNEL_BINDINGS,

66 GSS_S_BAD_BINDINGS,

67 GSS_S_BAD_MIC,

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 _GSSAPI_OIDS,

76 _GSSAPI_SIGNATURE_OIDS,

77)

79# Typing imports

80from typing import (

81 Any,

82 Callable,

83 List,

84 Optional,

85 Tuple,

86 Union,

87)

89# Crypto imports

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

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

94##########

95# Fields #

96##########

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

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

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

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

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

104

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

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

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

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

109

110

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

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

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

114

115 __slots__ = [

116 "fields",

117 "fields_map",

118 "fields_pad",

119 "offset",

120 "length_from",

121 "force_order",

122 "offset_name",

123 ]

124 islist = True

125

126 def __init__(

127 self,

128 name, # type: str

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

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

131 fields_pad=0, # type: int

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

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

134 offset_name="BufferOffset", # type: str

135 ):

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

137 self.offset = offset

138 self.fields = fields

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

140 self.length_from = length_from

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

142 self.offset_name = offset_name

143 self.fields_pad = fields_pad

144 super(_NTLMPayloadField, self).__init__(

145 name,

146 [

147 (field.name, field.default)

148 for field in fields

149 if field.default is not None

150 ],

151 )

152

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

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

155 if not pkt or not x:

156 return []

157 results = []

158 for field_name, value in x:

159 if field_name not in self.fields_map:

160 continue

161 if not isinstance(

162 self.fields_map[field_name], PacketListField

163 ) and not isinstance(value, Packet):

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

165 results.append((field_name, value))

166 return results

167

168 def i2h(self, pkt, x):

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

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

171

172 def h2i(self, pkt, x):

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

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

175

176 def i2repr(self, pkt, x):

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

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

179

180 def _o_pkt(self, pkt):

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

182 if callable(self.offset):

183 return self.offset(pkt)

184 return self.offset

185

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

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

188 # Create string buffer

189 buf = StringBuffer()

190 buf.append(s, 1)

191 # Calc relative offset

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

193 if self.force_order:

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

195 for field_name, value in val:

196 if field_name not in self.fields_map:

197 continue

198 field = self.fields_map[field_name]

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

200 if offset is None:

201 # No offset specified: calc

202 offset = len(buf)

203 if self.fields_pad:

204 pad = (-offset) % self.fields_pad

205 offset += pad

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

207 else:

208 # Calc relative offset

209 offset -= r_off

210 pad = offset + 1 - len(buf)

211 # Add padding if necessary

212 if pad > 0:

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

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

215 return bytes(buf)

216

217 def getfield(self, pkt, s):

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

219 if self.length_from is None:

220 ret, remain = b"", s

221 else:

222 len_pkt = self.length_from(pkt)

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

224 if not pkt or not remain:

225 return s, []

226 results = []

227 max_offset = 0

228 o_pkt = self._o_pkt(pkt)

229 offsets = [

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

231 ]

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

233 offset = offsets[i]

234 try:

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

236 except AttributeError:

237 length = len(remain) - offset

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

239 try:

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

241 except ValueError:

242 pass

243 if offset < 0:

244 continue

245 max_offset = max(offset + length, max_offset)

246 if remain[offset : offset + length]:

247 results.append(

248 (

249 offset,

250 field.name,

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

252 )

253 )

254 ret += remain[max_offset:]

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

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

257

258

259class _NTLMPayloadPacket(Packet):

260 _NTLM_PAYLOAD_FIELD_NAME = "Payload"

261

262 def __init__(

263 self,

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

265 post_transform=None, # type: Any

266 _internal=0, # type: int

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

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

269 **fields, # type: Any

270 ):

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

272 unknown = {

273 k: fields.pop(k)

274 for k in list(fields)

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

276 }

277 super(_NTLMPayloadPacket, self).__init__(

278 _pkt=_pkt,

279 post_transform=post_transform,

280 _internal=_internal,

281 _underlayer=_underlayer,

282 _parent=_parent,

283 **fields,

284 )

285 # check unknown fields for implicit ones

286 local_fields = next(

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

288 for x in self.fields_desc

289 if x.name == self._NTLM_PAYLOAD_FIELD_NAME

290 )

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

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

293 self.setfieldval(k, value)

294

295 def getfieldval(self, attr):

296 # Ease compatibility with _NTLMPayloadField

297 try:

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

299 except AttributeError:

300 try:

301 return next(

302 x[1]

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

304 self._NTLM_PAYLOAD_FIELD_NAME

305 )

306 if x[0] == attr

307 )

308 except StopIteration:

309 raise AttributeError(attr)

310

311 def getfield_and_val(self, attr):

312 # Ease compatibility with _NTLMPayloadField

313 try:

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

315 except ValueError:

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

317 try:

318 return (

319 PayFields[attr],

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

321 self,

322 next(

323 x[1]

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

325 self._NTLM_PAYLOAD_FIELD_NAME

326 )

327 if x[0] == attr

328 ),

329 ),

330 )

331 except (StopIteration, KeyError):

332 raise ValueError(attr)

333

334 def setfieldval(self, attr, val):

335 # Ease compatibility with _NTLMPayloadField

336 try:

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

338 except AttributeError:

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

340 self._NTLM_PAYLOAD_FIELD_NAME

341 )

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

343 raise AttributeError(attr)

344 try:

345 Payload.pop(

346 next(

347 i

348 for i, x in enumerate(

349 super(_NTLMPayloadPacket, self).__getattr__(

350 self._NTLM_PAYLOAD_FIELD_NAME

351 )

352 )

353 if x[0] == attr

354 )

355 )

356 except StopIteration:

357 pass

358 Payload.append([attr, val])

359 super(_NTLMPayloadPacket, self).setfieldval(

360 self._NTLM_PAYLOAD_FIELD_NAME, Payload

361 )

362

363

364class _NTLM_ENUM(IntEnum):

365 LEN = 0x0001

366 MAXLEN = 0x0002

367 OFFSET = 0x0004

368 COUNT = 0x0008

369 PAD8 = 0x1000

370

371

372_NTLM_CONFIG = [

373 ("Len", _NTLM_ENUM.LEN),

374 ("MaxLen", _NTLM_ENUM.MAXLEN),

375 ("BufferOffset", _NTLM_ENUM.OFFSET),

376]

377

378

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

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

381 gl_fld = self.get_field(self._NTLM_PAYLOAD_FIELD_NAME)

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

383 fld = gl_fld.fields_map[field_name]

384 length = fld.i2len(self, value)

385 count = fld.i2count(self, value)

386 offset = fields[field_name]

387 i = 0

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

389 for fname, ftype in config:

390 if isinstance(ftype, dict):

391 ftype = ftype[field_name]

392 if ftype & _NTLM_ENUM.LEN:

393 fval = length

394 elif ftype & _NTLM_ENUM.OFFSET:

395 fval = pay_offset

396 elif ftype & _NTLM_ENUM.MAXLEN:

397 fval = length

398 elif ftype & _NTLM_ENUM.COUNT:

399 fval = count

400 else:

401 raise ValueError

402 if ftype & _NTLM_ENUM.PAD8:

403 pad = (-fval) % 8

404 fval += pad

405 pay_offset += pad

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

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

408 p = (

409 p[: offset + i]

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

411 + p[offset + i + sz :]

412 )

413 i += sz

414 pay_offset += length

415 return p

416

417

418##############

419# Structures #

420##############

421

422

423# -- Util: VARIANT class

424

425

426class NTLM_VARIANT(IntEnum):

427 """

428 The message variant to use for NTLM.

429 """

430

431 NT_OR_2000 = 0

432 XP_OR_2003 = 1

433 RECENT = 2

434

435

436class _NTLM_VARIANT_Packet(_NTLMPayloadPacket):

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

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

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

440

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

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

443 pkt.VARIANT = self.VARIANT

444 return pkt

445

446 def copy(self):

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

448 pkt.VARIANT = self.VARIANT

449

450 return pkt

451

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

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

454 dump, indent, lvl, label_lvl

455 )

458# Sect 2.2

461class NTLM_Header(Packet):

462 name = "NTLM Header"

463 fields_desc = [

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

465 LEIntEnumField(

466 "MessageType",

467 3,

468 {

469 1: "NEGOTIATE_MESSAGE",

470 2: "CHALLENGE_MESSAGE",

471 3: "AUTHENTICATE_MESSAGE",

472 },

473 ),

474 ]

475

476 @classmethod

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

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

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

480 if MessageType == 1:

481 return NTLM_NEGOTIATE

482 elif MessageType == 2:

483 return NTLM_CHALLENGE

484 elif MessageType == 3:

485 return NTLM_AUTHENTICATE_V2

486 return cls

487

488

489# Sect 2.2.2.5

490_negotiateFlags = [

491 "NEGOTIATE_UNICODE", # A

492 "NEGOTIATE_OEM", # B

493 "REQUEST_TARGET", # C

494 "r10",

495 "NEGOTIATE_SIGN", # D

496 "NEGOTIATE_SEAL", # E

497 "NEGOTIATE_DATAGRAM", # F

498 "NEGOTIATE_LM_KEY", # G

499 "r9",

500 "NEGOTIATE_NTLM", # H

501 "r8",

502 "J",

503 "NEGOTIATE_OEM_DOMAIN_SUPPLIED", # K

504 "NEGOTIATE_OEM_WORKSTATION_SUPPLIED", # L

505 "NEGOTIATE_LOCAL_CALL",

506 "NEGOTIATE_ALWAYS_SIGN", # M

507 "TARGET_TYPE_DOMAIN", # N

508 "TARGET_TYPE_SERVER", # O

509 "r6",

510 "NEGOTIATE_EXTENDED_SESSIONSECURITY", # P

511 "NEGOTIATE_IDENTIFY", # Q

512 "r5",

513 "REQUEST_NON_NT_SESSION_KEY", # R

514 "NEGOTIATE_TARGET_INFO", # S

515 "r4",

516 "NEGOTIATE_VERSION", # T

517 "r3",

518 "r2",

519 "r1",

520 "NEGOTIATE_128", # U

521 "NEGOTIATE_KEY_EXCH", # V

522 "NEGOTIATE_56", # W

523]

524

525

526def _NTLMStrField(name, default):

527 return MultipleTypeField(

528 [

529 (

530 StrFieldUtf16(name, default),

531 lambda pkt: pkt.NegotiateFlags.NEGOTIATE_UNICODE,

532 )

533 ],

534 StrField(name, default),

535 )

538# Sect 2.2.2.10

541class _NTLM_Version(Packet):

542 fields_desc = [

543 ByteField("ProductMajorVersion", 0),

544 ByteField("ProductMinorVersion", 0),

545 LEShortField("ProductBuild", 0),

546 LEThreeBytesField("res_ver", 0),

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

548 ]

551# Sect 2.2.1.1

554class NTLM_NEGOTIATE(_NTLM_VARIANT_Packet, NTLM_Header):

555 name = "NTLM Negotiate"

556 __slots__ = ["VARIANT"]

557 MessageType = 1

558 OFFSET = lambda pkt: (

559 32

560 if (

561 pkt.VARIANT == NTLM_VARIANT.NT_OR_2000

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

563 )

564 else 40

565 )

566 fields_desc = (

567 [

568 NTLM_Header,

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

570 # DomainNameFields

571 LEShortField("DomainNameLen", None),

572 LEShortField("DomainNameMaxLen", None),

573 LEIntField("DomainNameBufferOffset", None),

574 # WorkstationFields

575 LEShortField("WorkstationNameLen", None),

576 LEShortField("WorkstationNameMaxLen", None),

577 LEIntField("WorkstationNameBufferOffset", None),

578 ]

579 + [

580 # VERSION

581 ConditionalField(

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

583 x,

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

585 and (

586 (

587 (

588 40

589 if pkt.DomainNameBufferOffset is None

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

591 )

592 > 32

593 )

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

595 ),

596 )

597 for x in _NTLM_Version.fields_desc

598 ]

599 + [

600 # Payload

601 _NTLMPayloadField(

602 "Payload",

603 OFFSET,

604 [

605 # "MUST be encoded using the OEM character set"

606 StrField("DomainName", b""),

607 StrField("WorkstationName", b""),

608 ],

609 ),

610 ]

611 )

612

613 def post_build(self, pkt, pay):

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

615 return (

616 _NTLM_post_build(

617 self,

618 pkt,

619 self.OFFSET(),

620 {

621 "DomainName": 16,

622 "WorkstationName": 24,

623 },

624 )

625 + pay

626 )

629# Challenge

632class Single_Host_Data(Packet):

633 fields_desc = [

634 LEIntField("Size", None),

635 LEIntField("Z4", 0),

636 # "CustomData" guessed using LSAP_TOKEN_INFO_INTEGRITY.

637 FlagsField(

638 "Flags",

639 0,

640 -32,

641 {

642 0x00000001: "UAC-Restricted",

643 },

644 ),

645 LEIntEnumField(

646 "TokenIL",

647 0x00002000,

648 {

649 0x00000000: "Untrusted",

650 0x00001000: "Low",

651 0x00002000: "Medium",

652 0x00003000: "High",

653 0x00004000: "System",

654 0x00005000: "Protected process",

655 },

656 ),

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

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

659 ConditionalField(

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

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

662 ),

663 ]

664

665 def post_build(self, pkt, pay):

666 if self.Size is None:

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

668 return pkt + pay

669

670 def default_payload_class(self, payload):

671 return conf.padding_layer

672

673

674class AV_PAIR(Packet):

675 name = "NTLM AV Pair"

676 fields_desc = [

677 LEShortEnumField(

678 "AvId",

679 0,

680 {

681 0x0000: "MsvAvEOL",

682 0x0001: "MsvAvNbComputerName",

683 0x0002: "MsvAvNbDomainName",

684 0x0003: "MsvAvDnsComputerName",

685 0x0004: "MsvAvDnsDomainName",

686 0x0005: "MsvAvDnsTreeName",

687 0x0006: "MsvAvFlags",

688 0x0007: "MsvAvTimestamp",

689 0x0008: "MsvAvSingleHost",

690 0x0009: "MsvAvTargetName",

691 0x000A: "MsvAvChannelBindings",

692 },

693 ),

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

695 MultipleTypeField(

696 [

697 (

698 LEIntEnumField(

699 "Value",

700 1,

701 {

702 0x0001: "constrained",

703 0x0002: "MIC integrity",

704 0x0004: "SPN from untrusted source",

705 },

706 ),

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

708 ),

709 (

710 UTCTimeField(

711 "Value",

712 None,

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

714 custom_scaling=1e7,

715 fmt="<Q",

716 ),

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

718 ),

719 (

720 PacketField("Value", Single_Host_Data(), Single_Host_Data),

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

722 ),

723 (

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

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

726 ),

727 ],

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

729 ),

730 ]

731

732 def default_payload_class(self, payload):

733 return conf.padding_layer

734

735

736class NTLM_CHALLENGE(_NTLM_VARIANT_Packet, NTLM_Header):

737 name = "NTLM Challenge"

738 __slots__ = ["VARIANT"]

739 MessageType = 2

740 OFFSET = lambda pkt: (

741 48

742 if (

743 pkt.VARIANT == NTLM_VARIANT.NT_OR_2000

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

745 )

746 else 56

747 )

748 fields_desc = (

749 [

750 NTLM_Header,

751 # TargetNameFields

752 LEShortField("TargetNameLen", None),

753 LEShortField("TargetNameMaxLen", None),

754 LEIntField("TargetNameBufferOffset", None),

755 #

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

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

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

759 # TargetInfoFields

760 LEShortField("TargetInfoLen", None),

761 LEShortField("TargetInfoMaxLen", None),

762 LEIntField("TargetInfoBufferOffset", None),

763 ]

764 + [

765 # VERSION

766 ConditionalField(

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

768 x,

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

770 and (

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

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

773 ),

774 )

775 for x in _NTLM_Version.fields_desc

776 ]

777 + [

778 # Payload

779 _NTLMPayloadField(

780 "Payload",

781 OFFSET,

782 [

783 _NTLMStrField("TargetName", b""),

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

785 ],

786 ),

787 ]

788 )

789

790 def getAv(self, AvId):

791 try:

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

793 except (StopIteration, AttributeError):

794 raise IndexError

795

796 def post_build(self, pkt, pay):

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

798 return (

799 _NTLM_post_build(

800 self,

801 pkt,

802 self.OFFSET(),

803 {

804 "TargetName": 12,

805 "TargetInfo": 40,

806 },

807 )

808 + pay

809 )

812# Authenticate

815class LM_RESPONSE(Packet):

816 fields_desc = [

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

818 ]

819

820

821class LMv2_RESPONSE(Packet):

822 fields_desc = [

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

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

825 ]

826

827

828class NTLM_RESPONSE(Packet):

829 fields_desc = [

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

831 ]

832

833

834class NTLMv2_CLIENT_CHALLENGE(Packet):

835 fields_desc = [

836 ByteField("RespType", 1),

837 ByteField("HiRespType", 1),

838 LEShortField("Reserved1", 0),

839 LEIntField("Reserved2", 0),

840 UTCTimeField(

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

842 ),

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

844 LEIntField("Reserved3", 0),

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

846 ]

847

848 def getAv(self, AvId):

849 try:

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

851 except StopIteration:

852 raise IndexError

853

854

855class NTLMv2_RESPONSE(NTLMv2_CLIENT_CHALLENGE):

856 fields_desc = [

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

858 NTLMv2_CLIENT_CHALLENGE,

859 ]

860

861 def computeNTProofStr(self, ResponseKeyNT, ServerChallenge):

862 """

863 Set temp to ConcatenationOf(Responserversion, HiResponserversion,

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

865 Set NTProofStr to HMAC_MD5(ResponseKeyNT,

866 ConcatenationOf(CHALLENGE_MESSAGE.ServerChallenge,temp))

867

868 Remember ServerName = AvPairs

869 """

870 Responserversion = b"\x01"

871 HiResponserversion = b"\x01"

872

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

874 temp = b"".join(

875 [

876 Responserversion,

877 HiResponserversion,

878 b"\x00" * 6,

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

880 self.ChallengeFromClient,

881 b"\x00" * 4,

882 ServerName,

883 # Final Z(4) is the EOL AvPair

884 ]

885 )

886 return HMAC_MD5(ResponseKeyNT, ServerChallenge + temp)

887

888

889class NTLM_AUTHENTICATE(_NTLM_VARIANT_Packet, NTLM_Header):

890 name = "NTLM Authenticate"

891 __slots__ = ["VARIANT"]

892 MessageType = 3

893 NTLM_VERSION = 1

894 OFFSET = lambda pkt: (

895 64

896 if (

897 pkt.VARIANT == NTLM_VARIANT.NT_OR_2000

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

899 )

900 else (

901 72

902 if pkt.VARIANT == NTLM_VARIANT.XP_OR_2003

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

904 else 88

905 )

906 )

907 fields_desc = (

908 [

909 NTLM_Header,

910 # LmChallengeResponseFields

911 LEShortField("LmChallengeResponseLen", None),

912 LEShortField("LmChallengeResponseMaxLen", None),

913 LEIntField("LmChallengeResponseBufferOffset", None),

914 # NtChallengeResponseFields

915 LEShortField("NtChallengeResponseLen", None),

916 LEShortField("NtChallengeResponseMaxLen", None),

917 LEIntField("NtChallengeResponseBufferOffset", None),

918 # DomainNameFields

919 LEShortField("DomainNameLen", None),

920 LEShortField("DomainNameMaxLen", None),

921 LEIntField("DomainNameBufferOffset", None),

922 # UserNameFields

923 LEShortField("UserNameLen", None),

924 LEShortField("UserNameMaxLen", None),

925 LEIntField("UserNameBufferOffset", None),

926 # WorkstationFields

927 LEShortField("WorkstationLen", None),

928 LEShortField("WorkstationMaxLen", None),

929 LEIntField("WorkstationBufferOffset", None),

930 # EncryptedRandomSessionKeyFields

931 LEShortField("EncryptedRandomSessionKeyLen", None),

932 LEShortField("EncryptedRandomSessionKeyMaxLen", None),

933 LEIntField("EncryptedRandomSessionKeyBufferOffset", None),

934 # NegotiateFlags

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

936 # VERSION

937 ]

938 + [

939 ConditionalField(

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

941 x,

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

943 and (

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

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

946 ),

947 )

948 for x in _NTLM_Version.fields_desc

949 ]

950 + [

951 # MIC

952 ConditionalField(

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

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

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

956 and (

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

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

959 ),

960 ),

961 # Payload

962 _NTLMPayloadField(

963 "Payload",

964 OFFSET,

965 [

966 MultipleTypeField(

967 [

968 (

969 PacketField(

970 "LmChallengeResponse",

971 LMv2_RESPONSE(),

972 LMv2_RESPONSE,

973 ),

974 lambda pkt: pkt.NTLM_VERSION == 2,

975 )

976 ],

977 PacketField("LmChallengeResponse", LM_RESPONSE(), LM_RESPONSE),

978 ),

979 MultipleTypeField(

980 [

981 (

982 PacketField(

983 "NtChallengeResponse",

984 NTLMv2_RESPONSE(),

985 NTLMv2_RESPONSE,

986 ),

987 lambda pkt: pkt.NTLM_VERSION == 2,

988 )

989 ],

990 PacketField(

991 "NtChallengeResponse", NTLM_RESPONSE(), NTLM_RESPONSE

992 ),

993 ),

994 _NTLMStrField("DomainName", b""),

995 _NTLMStrField("UserName", b""),

996 _NTLMStrField("Workstation", b""),

997 XStrField("EncryptedRandomSessionKey", b""),

998 ],

999 ),

1000 ]

1001 )

1002

1003 def post_build(self, pkt, pay):

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

1005 return (

1006 _NTLM_post_build(

1007 self,

1008 pkt,

1009 self.OFFSET(),

1010 {

1011 "LmChallengeResponse": 12,

1012 "NtChallengeResponse": 20,

1013 "DomainName": 28,

1014 "UserName": 36,

1015 "Workstation": 44,

1016 "EncryptedRandomSessionKey": 52,

1017 },

1018 )

1019 + pay

1020 )

1021

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

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

1024 self.MIC = HMAC_MD5(

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

1026 )

1027

1028 def verify_mic(self, ExportedSessionKey, negotiate, challenge):

1029 auth = self.copy()

1030 auth.MIC = b"\x00" * 16

1031 return self.MIC == HMAC_MD5(

1032 ExportedSessionKey, bytes(negotiate) + bytes(challenge) + bytes(auth)

1033 )

1034

1035

1036class NTLM_AUTHENTICATE_V2(NTLM_AUTHENTICATE):

1037 NTLM_VERSION = 2

1038

1039

1040def HTTP_ntlm_negotiate(ntlm_negotiate):

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

1042 assert isinstance(ntlm_negotiate, NTLM_NEGOTIATE)

1043 from scapy.layers.http import HTTP, HTTPRequest

1044

1045 return HTTP() / HTTPRequest(

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

1047 )

1048

1049

1050# Experimental - Reversed stuff

1051

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

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

1054# placeholder.

1055

1056

1057class NEGOEX_EXCHANGE_NTLM_ITEM(ASN1_Packet):

1058 ASN1_codec = ASN1_Codecs.BER

1059 ASN1_root = ASN1F_SEQUENCE(

1060 ASN1F_SEQUENCE(

1061 ASN1F_SEQUENCE(

1062 ASN1F_OID("oid", ""),

1063 ASN1F_PRINTABLE_STRING("token", ""),

1064 explicit_tag=0x31,

1065 ),

1066 explicit_tag=0x80,

1067 )

1068 )

1069

1070

1071class NEGOEX_EXCHANGE_NTLM(ASN1_Packet):

1072 """

1073 GSSAPI NegoEX Exchange metadata blob

1074 This was reversed and may be meaningless

1075 """

1076

1077 ASN1_codec = ASN1_Codecs.BER

1078 ASN1_root = ASN1F_SEQUENCE(

1079 ASN1F_SEQUENCE(

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

1081 ),

1082 )

1083

1084

1085# Crypto - [MS-NLMP]

1086

1087

1088def HMAC_MD5(key, data):

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

1090

1091

1092def MD4le(x):

1093 """

1094 MD4 over a string encoded as utf-16le

1095 """

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

1097

1098

1099def RC4Init(key):

1100 """Alleged RC4"""

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

1102

1103 try:

1104 # cryptography > 43.0

1105 from cryptography.hazmat.decrepit.ciphers import (

1106 algorithms as decrepit_algorithms,

1107 )

1108 except ImportError:

1109 decrepit_algorithms = algorithms

1110

1111 algorithm = decrepit_algorithms.ARC4(key)

1112 cipher = Cipher(algorithm, mode=None)

1113 encryptor = cipher.encryptor()

1114 return encryptor

1115

1116

1117def RC4(handle, data):

1118 """The RC4 Encryption Algorithm"""

1119 return handle.update(data)

1120

1121

1122def RC4K(key, data):

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

1124 RC4 algorithm.

1125 """

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

1127

1128 try:

1129 # cryptography > 43.0

1130 from cryptography.hazmat.decrepit.ciphers import (

1131 algorithms as decrepit_algorithms,

1132 )

1133 except ImportError:

1134 decrepit_algorithms = algorithms

1135

1136 algorithm = decrepit_algorithms.ARC4(key)

1137 cipher = Cipher(algorithm, mode=None)

1138 encryptor = cipher.encryptor()

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

1140

1141

1142# sect 2.2.2.9 - With Extended Session Security

1143

1144

1145class NTLMSSP_MESSAGE_SIGNATURE(Packet):

1146 # [MS-RPCE] sect 2.2.2.9.1/2.2.2.9.2

1147 fields_desc = [

1148 LEIntField("Version", 0x00000001),

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

1150 LEIntField("SeqNum", 0x00000000),

1151 ]

1152

1153 def default_payload_class(self, payload):

1154 return conf.padding_layer

1155

1156

1157_GSSAPI_OIDS["1.3.6.1.4.1.311.2.2.10"] = NTLM_Header

1158_GSSAPI_SIGNATURE_OIDS["1.3.6.1.4.1.311.2.2.10"] = NTLMSSP_MESSAGE_SIGNATURE

1161# sect 3.3.2

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

1165 """

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

1167

1168 :param Passwd: the plain password

1169 :param User: the username

1170 :param UserDom: the domain name

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

1172 Passwd to None

1173 """

1174 if HashNt is None:

1175 HashNt = MD4le(Passwd)

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

1177

1178

1179def NTLMv2_ComputeSessionBaseKey(ResponseKeyNT, NTProofStr):

1180 return HMAC_MD5(ResponseKeyNT, NTProofStr)

1181

1182

1183# sect 3.4.4.2 - With Extended Session Security

1184

1185

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

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

1188 if Handle:

1189 chksum = RC4(Handle, chksum)

1190 return NTLMSSP_MESSAGE_SIGNATURE(

1191 Version=0x00000001,

1192 Checksum=chksum,

1193 SeqNum=SeqNum,

1194 )

1197# sect 3.4.2

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

1201 # append? where is this used?!

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

1205# sect 3.4.3

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

1209 """

1210 SEAL() according to [MS-NLMP]

1211 """

1212 # this is unused. Use GSS_WrapEx

1213 sealed_message = RC4(Handle, Message)

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

1215 return sealed_message, signature

1216

1217

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

1219 """

1220 UNSEAL() according to [MS-NLMP]

1221 """

1222 # this is unused. Use GSS_UnwrapEx

1223 unsealed_message = RC4(Handle, Message)

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

1225 return unsealed_message, signature

1228# sect 3.4.5.2

1231def SIGNKEY(NegFlg, ExportedSessionKey, Mode):

1232 if NegFlg.NEGOTIATE_EXTENDED_SESSIONSECURITY:

1233 if Mode == "Client":

1234 return Hash_MD5().digest(

1235 ExportedSessionKey

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

1237 )

1238 elif Mode == "Server":

1239 return Hash_MD5().digest(

1240 ExportedSessionKey

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

1242 )

1243 else:

1244 raise ValueError("Unknown Mode")

1245 else:

1246 return None

1249# sect 3.4.5.3

1252def SEALKEY(NegFlg, ExportedSessionKey, Mode):

1253 if NegFlg.NEGOTIATE_EXTENDED_SESSIONSECURITY:

1254 if NegFlg.NEGOTIATE_128:

1255 SealKey = ExportedSessionKey

1256 elif NegFlg.NEGOTIATE_56:

1257 SealKey = ExportedSessionKey[:7]

1258 else:

1259 SealKey = ExportedSessionKey[:5]

1260 if Mode == "Client":

1261 return Hash_MD5().digest(

1262 SealKey

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

1264 )

1265 elif Mode == "Server":

1266 return Hash_MD5().digest(

1267 SealKey

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

1269 )

1270 else:

1271 raise ValueError("Unknown Mode")

1272 elif NegFlg.NEGOTIATE_LM_KEY:

1273 if NegFlg.NEGOTIATE_56:

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

1275 else:

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

1277 else:

1278 return ExportedSessionKey

1281# --- SSP

1284class NTLMSSP(SSP):

1285 """

1286 The NTLM SSP

1287

1288 Common arguments:

1289

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

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

1292

1293 In case of a client::

1294

1295 - NegotiateFlags

1296 - ProductMajorVersion

1297 - ProductMinorVersion

1298 - ProductBuild

1299

1300 In case of a server::

1301

1302 - NetbiosDomainName

1303 - NetbiosComputerName

1304 - DnsComputerName

1305 - DnsDomainName (defaults to DOMAIN)

1306 - DnsTreeName (defaults to DOMAIN)

1307 - Flags

1308 - Timestamp

1309

1310 Client-only arguments:

1311

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

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

1314 if without domain)

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

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

1317 :param LOCAL: use local authentication (must be running locally on Windows)

1318

1319 Server-only arguments:

1320

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

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

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

1324 :param COMPUTER_FQDN: the server FQDN

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

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

1327 Setting this value enables signature computation and

1328 authenticates inbound users.

1329 """

1330

1331 auth_type = 0x0A

1332

1333 class STATE(SSP.STATE):

1334 INIT = 1

1335 CLI_SENT_NEGO = 2

1336 CLI_SENT_AUTH = 3

1337 SRV_SENT_CHAL = 4

1338

1339 class CONTEXT(SSP.CONTEXT):

1340 __slots__ = [

1341 "SessionKey",

1342 "ExportedSessionKey",

1343 "IsAcceptor",

1344 "SendSignKey",

1345 "SendSealKey",

1346 "RecvSignKey",

1347 "RecvSealKey",

1348 "SendSealHandle",

1349 "RecvSealHandle",

1350 "SendSeqNum",

1351 "RecvSeqNum",

1352 "neg_tok",

1353 "chall_tok",

1354 "ServerHostname",

1355 "ServerDomain",

1356 ]

1357

1358 @crypto_validator

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

1360 self.state = NTLMSSP.STATE.INIT

1361 self.SessionKey = None

1362 self.ExportedSessionKey = None

1363 self.SendSignKey = None

1364 self.SendSealKey = None

1365 self.SendSealHandle = None

1366 self.RecvSignKey = None

1367 self.RecvSealKey = None

1368 self.RecvSealHandle = None

1369 self.SendSeqNum = 0

1370 self.RecvSeqNum = 0

1371 self.neg_tok = None

1372 self.chall_tok = None

1373 self.ServerHostname = None

1374 self.ServerDomain = None

1375 self.IsAcceptor = IsAcceptor

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

1377

1378 def clifailure(self):

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

1380

1381 def __repr__(self):

1382 return "NTLMSSP"

1383

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

1385 NTLM_MaxLifetime = 36 * 3600

1386

1387 def __init__(

1388 self,

1389 UPN=None,

1390 HASHNT=None,

1391 PASSWORD=None,

1392 USE_MIC=True,

1393 VARIANT: NTLM_VARIANT = NTLM_VARIANT.RECENT,

1394 NTLM_VALUES={},

1395 DOMAIN_FQDN=None,

1396 DOMAIN_NB_NAME=None,

1397 COMPUTER_NB_NAME=None,

1398 COMPUTER_FQDN=None,

1399 IDENTITIES=None,

1400 DO_NOT_CHECK_LOGIN=False,

1401 SERVER_CHALLENGE=None,

1402 **kwargs,

1403 ):

1404 self.UPN = UPN

1405 if HASHNT is None and PASSWORD is not None:

1406 HASHNT = MD4le(PASSWORD)

1407 self.HASHNT = HASHNT

1408 self.VARIANT = VARIANT

1409 if self.VARIANT != NTLM_VARIANT.RECENT:

1410 log_runtime.warning(

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

1412 )

1413 self.USE_MIC = False

1414 else:

1415 self.USE_MIC = USE_MIC

1416

1417 if UPN is not None:

1418 # Populate values used only in server mode.

1419 from scapy.layers.kerberos import _parse_upn

1420

1421 try:

1422 user, realm = _parse_upn(UPN)

1423 if DOMAIN_FQDN is None:

1424 DOMAIN_FQDN = realm

1425 if COMPUTER_NB_NAME is None:

1426 COMPUTER_NB_NAME = user

1427 except ValueError:

1428 pass

1429

1430 # Compute various netbios/fqdn names

1431 self.DOMAIN_FQDN = DOMAIN_FQDN or "WORKGROUP"

1432 self.DOMAIN_NB_NAME = (

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

1434 )

1435 self.COMPUTER_NB_NAME = COMPUTER_NB_NAME or "WIN10"

1436 self.COMPUTER_FQDN = COMPUTER_FQDN or (

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

1438 )

1439 self.NTLM_VALUES = NTLM_VALUES

1440

1441 if IDENTITIES:

1442 self.IDENTITIES = {

1443 # Windows usernames are case insensitive

1444 user.upper(): hashnt

1445 for user, hashnt in IDENTITIES.items()

1446 }

1447 else:

1448 self.IDENTITIES = IDENTITIES

1449

1450 self.DO_NOT_CHECK_LOGIN = DO_NOT_CHECK_LOGIN

1451 self.SERVER_CHALLENGE = SERVER_CHALLENGE

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

1453

1454 def LegsAmount(self, Context: CONTEXT):

1455 return 3

1456

1457 def GSS_Inquire_names_for_mech(self):

1458 return ["1.3.6.1.4.1.311.2.2.10"]

1459

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

1461 """

1462 [MS-NLMP] sect 3.4.8

1463 """

1464 # Concatenate the ToSign

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

1466 sig = MAC(

1467 Context.SendSealHandle,

1468 Context.SendSignKey,

1469 Context.SendSeqNum,

1470 ToSign,

1471 )

1472 Context.SendSeqNum += 1

1473 return sig

1474

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

1476 """

1477 [MS-NLMP] sect 3.4.9

1478 """

1479 Context.RecvSeqNum = signature.SeqNum

1480 # Concatenate the ToSign

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

1482 sig = MAC(

1483 Context.RecvSealHandle,

1484 Context.RecvSignKey,

1485 Context.RecvSeqNum,

1486 ToSign,

1487 )

1488 if sig.Checksum != signature.Checksum:

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

1490

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

1492 """

1493 [MS-NLMP] sect 3.4.6

1494 """

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

1496 # Encrypt

1497 for msg in msgs:

1498 if msg.conf_req_flag:

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

1500 # Sign

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

1502 return (

1503 msgs,

1504 sig,

1505 )

1506

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

1508 """

1509 [MS-NLMP] sect 3.4.7

1510 """

1511 # Decrypt

1512 for msg in msgs:

1513 if msg.conf_req_flag:

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

1515 # Check signature

1516 self.GSS_VerifyMICEx(Context, msgs, signature)

1517 return msgs

1518

1519 def SupportsMechListMIC(self):

1520 if not self.USE_MIC:

1521 # RFC 4178

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

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

1524 return False

1525 if self.DO_NOT_CHECK_LOGIN:

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

1527 return False

1528 return True

1529

1530 def GetMechListMIC(self, Context, input):

1531 # [MS-SPNG]

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

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

1534 # OriginalHandle after generating the mechListMIC."

1535 OriginalHandle = Context.SendSealHandle

1536 Context.SendSealHandle = RC4Init(Context.SendSealKey)

1537 try:

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

1539 finally:

1540 Context.SendSealHandle = OriginalHandle

1541

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

1543 # [MS-SPNG]

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

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

1546 # validating the mechListMIC."

1547 OriginalHandle = Context.RecvSealHandle

1548 Context.RecvSealHandle = RC4Init(Context.RecvSealKey)

1549 try:

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

1551 finally:

1552 Context.RecvSealHandle = OriginalHandle

1553

1554 def GSS_Init_sec_context(

1555 self,

1556 Context: CONTEXT,

1557 input_token=None,

1558 target_name: Optional[str] = None,

1559 req_flags: Optional[GSS_C_FLAGS] = None,

1560 chan_bindings: GssChannelBindings = GSS_C_NO_CHANNEL_BINDINGS,

1561 ):

1562 if Context is None:

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

1564

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

1566 # Client: negotiate

1567

1568 # Create a default token

1569 tok = NTLM_NEGOTIATE(

1570 VARIANT=self.VARIANT,

1571 NegotiateFlags="+".join(

1572 [

1573 "NEGOTIATE_UNICODE",

1574 "REQUEST_TARGET",

1575 "NEGOTIATE_NTLM",

1576 "NEGOTIATE_ALWAYS_SIGN",

1577 "TARGET_TYPE_DOMAIN",

1578 "NEGOTIATE_EXTENDED_SESSIONSECURITY",

1579 "NEGOTIATE_TARGET_INFO",

1580 "NEGOTIATE_128",

1581 "NEGOTIATE_56",

1582 ]

1583 + (

1584 ["NEGOTIATE_VERSION"]

1585 if self.VARIANT >= NTLM_VARIANT.XP_OR_2003

1586 else []

1587 )

1588 + (

1589 [

1590 "NEGOTIATE_KEY_EXCH",

1591 ]

1592 if Context.flags

1593 & (GSS_C_FLAGS.GSS_C_INTEG_FLAG | GSS_C_FLAGS.GSS_C_CONF_FLAG)

1594 else []

1595 )

1596 + (

1597 [

1598 "NEGOTIATE_SIGN",

1599 ]

1600 if Context.flags & GSS_C_FLAGS.GSS_C_INTEG_FLAG

1601 else []

1602 )

1603 + (

1604 [

1605 "NEGOTIATE_SEAL",

1606 ]

1607 if Context.flags & GSS_C_FLAGS.GSS_C_CONF_FLAG

1608 else []

1609 )

1610 + (

1611 [

1612 "NEGOTIATE_IDENTIFY",

1613 ]

1614 if Context.flags & GSS_C_FLAGS.GSS_C_IDENTIFY_FLAG

1615 else []

1616 )

1617 ),

1618 ProductMajorVersion=10,

1619 ProductMinorVersion=0,

1620 ProductBuild=26100,

1621 )

1622

1623 # Update that token with the customs one

1624 if self.NTLM_VALUES:

1625 for key in [

1626 "NegotiateFlags",

1627 "ProductMajorVersion",

1628 "ProductMinorVersion",

1629 "ProductBuild",

1630 "DomainName",

1631 "WorkstationName",

1632 ]:

1633 if key in self.NTLM_VALUES:

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

1635 Context.neg_tok = tok

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

1637 Context.state = self.STATE.CLI_SENT_NEGO

1638 return Context, tok, GSS_S_CONTINUE_NEEDED

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

1640 # Client: auth (token=challenge)

1641 chall_tok = input_token

1642

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

1644 raise ValueError(

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

1646 "running in standalone !"

1647 )

1648

1649 from scapy.layers.kerberos import _parse_upn

1650

1651 # Check token sanity

1652 if not chall_tok or NTLM_CHALLENGE not in chall_tok:

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

1654 return Context, None, GSS_S_DEFECTIVE_TOKEN

1655

1656 # Some information from the CHALLENGE are stored

1657 try:

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

1659 except IndexError:

1660 pass

1661 try:

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

1663 except IndexError:

1664 pass

1665 try:

1666 # the server SHOULD set the timestamp in the CHALLENGE_MESSAGE

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

1668 ServerTime = (ServerTimestamp / 1e7) - 11644473600

1669

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

1671 log_runtime.warning(

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

1673 )

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

1675 except IndexError:

1676 pass

1677

1678 # Initialize a default token

1679 tok = NTLM_AUTHENTICATE_V2(

1680 VARIANT=self.VARIANT,

1681 NegotiateFlags=chall_tok.NegotiateFlags,

1682 ProductMajorVersion=10,

1683 ProductMinorVersion=0,

1684 ProductBuild=26100,

1685 )

1686

1687 # Populate the token

1688 tok.LmChallengeResponse = LMv2_RESPONSE()

1689

1690 # 1. Set username

1691 try:

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

1693 except ValueError:

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

1695

1696 # 2. Set domain name

1697 if realm is None:

1698 log_runtime.warning(

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

1700 )

1701 tok.DomainName = self.DOMAIN_FQDN

1702 else:

1703 tok.DomainName = realm

1704

1705 # 3. Set workstation name

1706 tok.Workstation = self.COMPUTER_NB_NAME

1707

1708 # 4. Create and calculate the ChallengeResponse

1709 # 4.1 Build the payload

1710 cr = tok.NtChallengeResponse = NTLMv2_RESPONSE(

1711 ChallengeFromClient=os.urandom(8),

1712 )

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

1714 cr.AvPairs = (

1715 # Repeat AvPairs from the server

1716 chall_tok.TargetInfo[:-1]

1717 + (

1718 [

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

1720 ]

1721 if self.USE_MIC

1722 else []

1723 )

1724 + [

1725 AV_PAIR(

1726 AvId="MsvAvSingleHost",

1727 Value=Single_Host_Data(

1728 MachineID=os.urandom(32),

1729 PermanentMachineID=os.urandom(32),

1730 ),

1731 ),

1732 ]

1733 + (

1734 [

1735 AV_PAIR(

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

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

1738 AvId="MsvAvChannelBindings",

1739 Value=chan_bindings.digestMD5(),

1740 ),

1741 ]

1742 if chan_bindings != GSS_C_NO_CHANNEL_BINDINGS

1743 else []

1744 )

1745 + [

1746 AV_PAIR(

1747 AvId="MsvAvTargetName",

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

1749 ),

1750 AV_PAIR(AvId="MsvAvEOL"),

1751 ]

1752 )

1753 if self.NTLM_VALUES:

1754 # Update that token with the customs one

1755 for key in [

1756 "NegotiateFlags",

1757 "ProductMajorVersion",

1758 "ProductMinorVersion",

1759 "ProductBuild",

1760 ]:

1761 if key in self.NTLM_VALUES:

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

1763

1764 # 4.2 Compute the ResponseKeyNT

1765 ResponseKeyNT = NTOWFv2(

1766 None,

1767 tok.UserName,

1768 tok.DomainName,

1769 HashNt=self.HASHNT,

1770 )

1771

1772 # 4.3 Compute the NTProofStr

1773 cr.NTProofStr = cr.computeNTProofStr(

1774 ResponseKeyNT,

1775 chall_tok.ServerChallenge,

1776 )

1777

1778 # 4.4 Compute the Session Key

1779 SessionBaseKey = NTLMv2_ComputeSessionBaseKey(ResponseKeyNT, cr.NTProofStr)

1780 KeyExchangeKey = SessionBaseKey # Only true for NTLMv2

1781 if chall_tok.NegotiateFlags.NEGOTIATE_KEY_EXCH:

1782 ExportedSessionKey = os.urandom(16)

1783 tok.EncryptedRandomSessionKey = RC4K(

1784 KeyExchangeKey,

1785 ExportedSessionKey,

1786 )

1787 else:

1788 ExportedSessionKey = KeyExchangeKey

1789

1790 # 4.5 Compute the MIC

1791 if self.USE_MIC:

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

1793

1794 # 5. Perform key computations

1795 Context.ExportedSessionKey = ExportedSessionKey

1796 # [MS-SMB] 3.2.5.3

1797 Context.SessionKey = Context.ExportedSessionKey

1798 # Compute NTLM keys

1799 Context.SendSignKey = SIGNKEY(

1800 tok.NegotiateFlags, ExportedSessionKey, "Client"

1801 )

1802 Context.SendSealKey = SEALKEY(

1803 tok.NegotiateFlags, ExportedSessionKey, "Client"

1804 )

1805 Context.SendSealHandle = RC4Init(Context.SendSealKey)

1806 Context.RecvSignKey = SIGNKEY(

1807 tok.NegotiateFlags, ExportedSessionKey, "Server"

1808 )

1809 Context.RecvSealKey = SEALKEY(

1810 tok.NegotiateFlags, ExportedSessionKey, "Server"

1811 )

1812 Context.RecvSealHandle = RC4Init(Context.RecvSealKey)

1813

1814 # Update the state

1815 Context.state = self.STATE.CLI_SENT_AUTH

1816

1817 return Context, tok, GSS_S_COMPLETE

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

1819 if input_token:

1820 # what is that?

1821 status = GSS_S_DEFECTIVE_TOKEN

1822 else:

1823 status = GSS_S_COMPLETE

1824 return Context, None, status

1825 else:

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

1827

1828 def GSS_Accept_sec_context(

1829 self,

1830 Context: CONTEXT,

1831 input_token=None,

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

1833 chan_bindings: GssChannelBindings = GSS_C_NO_CHANNEL_BINDINGS,

1834 ):

1835 if Context is None:

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

1837

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

1839 # Server: challenge (input_token=negotiate)

1840 neg_tok = input_token

1841 if not neg_tok or NTLM_NEGOTIATE not in neg_tok:

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

1843 return Context, None, GSS_S_DEFECTIVE_TOKEN

1844

1845 Context.neg_tok = neg_tok

1846

1847 # Build the challenge token

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

1849 tok = NTLM_CHALLENGE(

1850 VARIANT=self.VARIANT,

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

1852 NegotiateFlags="+".join(

1853 [

1854 "NEGOTIATE_UNICODE",

1855 "REQUEST_TARGET",

1856 "NEGOTIATE_NTLM",

1857 "NEGOTIATE_ALWAYS_SIGN",

1858 "NEGOTIATE_EXTENDED_SESSIONSECURITY",

1859 "NEGOTIATE_TARGET_INFO",

1860 "TARGET_TYPE_DOMAIN",

1861 "NEGOTIATE_128",

1862 "NEGOTIATE_KEY_EXCH",

1863 "NEGOTIATE_56",

1864 ]

1865 + (

1866 ["NEGOTIATE_VERSION"]

1867 if self.VARIANT >= NTLM_VARIANT.XP_OR_2003

1868 else []

1869 )

1870 + (

1871 ["NEGOTIATE_SIGN"]

1872 if neg_tok.NegotiateFlags.NEGOTIATE_SIGN

1873 else []

1874 )

1875 + (

1876 ["NEGOTIATE_SEAL"]

1877 if neg_tok.NegotiateFlags.NEGOTIATE_SEAL

1878 else []

1879 )

1880 ),

1881 ProductMajorVersion=10,

1882 ProductMinorVersion=0,

1883 ProductBuild=26100,

1884 Payload=[

1885 ("TargetName", ""),

1886 (

1887 "TargetInfo",

1888 [

1889 # MsvAvNbComputerName

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

1891 # MsvAvNbDomainName

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

1893 # MsvAvDnsComputerName

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

1895 # MsvAvDnsDomainName

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

1897 # MsvAvDnsTreeName

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

1899 # MsvAvTimestamp

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

1901 # MsvAvEOL

1902 AV_PAIR(AvId=0),

1903 ],

1904 ),

1905 ],

1906 )

1907 if self.NTLM_VALUES:

1908 # Update that token with the customs one

1909 for key in [

1910 "ServerChallenge",

1911 "NegotiateFlags",

1912 "ProductMajorVersion",

1913 "ProductMinorVersion",

1914 "TargetName",

1915 ]:

1916 if key in self.NTLM_VALUES:

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

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

1919 tok.TargetInfo = [

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

1921 for (i, x) in [

1922 (2, "NetbiosDomainName"),

1923 (1, "NetbiosComputerName"),

1924 (4, "DnsDomainName"),

1925 (3, "DnsComputerName"),

1926 (5, "DnsTreeName"),

1927 (6, "Flags"),

1928 (7, "Timestamp"),

1929 (0, None),

1930 ]

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

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

1933 ]

1934

1935 # Store for next step

1936 Context.chall_tok = tok

1937

1938 # Update the state

1939 Context.state = self.STATE.SRV_SENT_CHAL

1940

1941 return Context, tok, GSS_S_CONTINUE_NEEDED

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

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

1944 auth_tok = input_token

1945

1946 if not auth_tok or NTLM_AUTHENTICATE_V2 not in auth_tok:

1947 log_runtime.debug(

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

1949 )

1950 return Context, None, GSS_S_DEFECTIVE_TOKEN

1951

1952 if self.DO_NOT_CHECK_LOGIN:

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

1954 return Context, None, GSS_S_COMPLETE

1955

1956 # Compute the session key

1957 SessionBaseKey = self._getSessionBaseKey(Context, auth_tok)

1958 if SessionBaseKey:

1959 # [MS-NLMP] sect 3.2.5.1.2

1960 KeyExchangeKey = SessionBaseKey # Only true for NTLMv2

1961 if auth_tok.NegotiateFlags.NEGOTIATE_KEY_EXCH:

1962 try:

1963 EncryptedRandomSessionKey = auth_tok.EncryptedRandomSessionKey

1964 except AttributeError:

1965 # No EncryptedRandomSessionKey. libcurl for instance

1966 # hmm. this looks bad

1967 EncryptedRandomSessionKey = b"\x00" * 16

1968 ExportedSessionKey = RC4K(KeyExchangeKey, EncryptedRandomSessionKey)

1969 else:

1970 ExportedSessionKey = KeyExchangeKey

1971 Context.ExportedSessionKey = ExportedSessionKey

1972 # [MS-SMB] 3.2.5.3

1973 Context.SessionKey = Context.ExportedSessionKey

1974 else:

1975 # Bad NTProofStr or unknown user

1976 Context.SessionKey = None

1977 Context.state = self.STATE.INIT

1978 return Context, None, GSS_S_DEFECTIVE_CREDENTIAL

1979

1980 # Verify MIC

1981 try:

1982 Flags = auth_tok.NtChallengeResponse.getAv(0x0006).Value

1983 if Flags & 2:

1984 # MIC is required

1985 if not auth_tok.verify_mic(

1986 ExportedSessionKey, Context.neg_tok, Context.chall_tok

1987 ):

1988 log_runtime.warning("Client MIC is invalid !")

1989 return Context, None, GSS_S_BAD_MIC

1990 except IndexError:

1991 pass

1992

1993 # Check the timestamp

1994 try:

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

1996 ClientTime = (ClientTimestamp / 1e7) - 11644473600

1997

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

1999 log_runtime.warning(

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

2001 )

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

2003 except IndexError:

2004 pass

2005

2006 # Check the channel bindings

2007 if chan_bindings == GSS_C_NO_CHANNEL_BINDINGS:

2008 if (

2009 req_flags is not None

2010 and GSS_S_FLAGS.GSS_S_ALLOW_MISSING_BINDINGS

2011 not in req_flags

2012 ):

2013 return Context, None, GSS_S_BAD_BINDINGS

2014 else:

2015 try:

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

2017 if Bnd != chan_bindings.digestMD5():

2018 # Bad channel bindings

2019 return Context, None, GSS_S_BAD_BINDINGS

2020 except IndexError:

2021 if GSS_S_FLAGS.GSS_S_ALLOW_MISSING_BINDINGS not in req_flags:

2022 # Uhoh, we required channel bindings

2023 return Context, None, GSS_S_BAD_BINDINGS

2024

2025 # Compute NTLM keys

2026 Context.SendSignKey = SIGNKEY(

2027 auth_tok.NegotiateFlags, ExportedSessionKey, "Server"

2028 )

2029 Context.SendSealKey = SEALKEY(

2030 auth_tok.NegotiateFlags, ExportedSessionKey, "Server"

2031 )

2032 Context.SendSealHandle = RC4Init(Context.SendSealKey)

2033 Context.RecvSignKey = SIGNKEY(

2034 auth_tok.NegotiateFlags, ExportedSessionKey, "Client"

2035 )

2036 Context.RecvSealKey = SEALKEY(

2037 auth_tok.NegotiateFlags, ExportedSessionKey, "Client"

2038 )

2039 Context.RecvSealHandle = RC4Init(Context.RecvSealKey)

2040

2041 # Check the NTProofStr

2042 if self._checkLogin(Context, auth_tok):

2043 # Set negotiated flags

2044 if auth_tok.NegotiateFlags.NEGOTIATE_SIGN:

2045 Context.flags |= GSS_C_FLAGS.GSS_C_INTEG_FLAG

2046 if auth_tok.NegotiateFlags.NEGOTIATE_SEAL:

2047 Context.flags |= GSS_C_FLAGS.GSS_C_CONF_FLAG

2048 return Context, None, GSS_S_COMPLETE

2049

2050 else:

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

2052

2053 def MaximumSignatureLength(self, Context: CONTEXT):

2054 """

2055 Returns the Maximum Signature length.

2056

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

2058 PFC_SUPPORT_HEADER_SIGN to work properly.

2059 """

2060 return 16 # len(NTLMSSP_MESSAGE_SIGNATURE())

2061

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

2063 if Context is None:

2064 Context = self.CONTEXT(True)

2065 Context.passive = True

2066

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

2068 # and discard the output.

2069

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

2071 if not token or NTLM_NEGOTIATE not in token:

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

2073 return None, GSS_S_DEFECTIVE_TOKEN

2074 Context.neg_tok = token

2075 Context.state = self.STATE.CLI_SENT_NEGO

2076 return Context, GSS_S_CONTINUE_NEEDED

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

2078 if not token or NTLM_CHALLENGE not in token:

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

2080 return None, GSS_S_DEFECTIVE_TOKEN

2081 Context.chall_tok = token

2082 Context.state = self.STATE.SRV_SENT_CHAL

2083 return Context, GSS_S_CONTINUE_NEEDED

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

2085 if not token or NTLM_AUTHENTICATE_V2 not in token:

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

2087 return None, GSS_S_DEFECTIVE_TOKEN

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

2089 if status != GSS_S_COMPLETE:

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

2091 Context.state = self.STATE.INIT

2092 return Context, status

2093 else:

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

2095

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

2097 if Context.IsAcceptor is not IsAcceptor:

2098 return

2099 # Swap everything

2100 Context.SendSignKey, Context.RecvSignKey = (

2101 Context.RecvSignKey,

2102 Context.SendSignKey,

2103 )

2104 Context.SendSealKey, Context.RecvSealKey = (

2105 Context.RecvSealKey,

2106 Context.SendSealKey,

2107 )

2108 Context.SendSealHandle, Context.RecvSealHandle = (

2109 Context.RecvSealHandle,

2110 Context.SendSealHandle,

2111 )

2112 Context.SendSeqNum, Context.RecvSeqNum = Context.RecvSeqNum, Context.SendSeqNum

2113 Context.IsAcceptor = not Context.IsAcceptor

2114

2115 def _getSessionBaseKey(self, Context, auth_tok):

2116 """

2117 Function that returns the SessionBaseKey from the ntlm Authenticate.

2118 """

2119 try:

2120 # Windows usernames are case insensitive

2121 username = auth_tok.UserName.upper()

2122 except AttributeError:

2123 username = None

2124 try:

2125 domain = auth_tok.DomainName

2126 except AttributeError:

2127 domain = ""

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

2129 ResponseKeyNT = NTOWFv2(

2130 None,

2131 username,

2132 domain,

2133 HashNt=self.IDENTITIES[username],

2134 )

2135 return NTLMv2_ComputeSessionBaseKey(

2136 ResponseKeyNT,

2137 auth_tok.NtChallengeResponse.NTProofStr,

2138 )

2139 elif self.IDENTITIES:

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

2141 return None

2142

2143 def _checkLogin(self, Context, auth_tok):

2144 """

2145 Function that checks the validity of an authentication.

2146

2147 Overwrite and return True to bypass.

2148 """

2149 try:

2150 # Windows usernames are case insensitive

2151 username = auth_tok.UserName.upper()

2152 except AttributeError:

2153 username = None

2154 try:

2155 domain = auth_tok.DomainName

2156 except AttributeError:

2157 domain = ""

2158 if username in self.IDENTITIES:

2159 ResponseKeyNT = NTOWFv2(

2160 None,

2161 username,

2162 domain,

2163 HashNt=self.IDENTITIES[username],

2164 )

2165 NTProofStr = auth_tok.NtChallengeResponse.computeNTProofStr(

2166 ResponseKeyNT,

2167 Context.chall_tok.ServerChallenge,

2168 )

2169 if NTProofStr == auth_tok.NtChallengeResponse.NTProofStr:

2170 return True

2171 return False

2172

2173

2174class NTLMSSP_DOMAIN(NTLMSSP):

2175 """

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

2177 keys from the domain using a Netlogon channel.

2178

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

2180 mode:

2181

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

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

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

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

2186 :param DC_FQDN: (optional) specify the FQDN of the DC.

2187

2188 Netlogon example::

2189

2190 >>> mySSP = NTLMSSP_DOMAIN(

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

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

2193 ... )

2194

2195 Kerberos example::

2196

2197 >>> mySSP = NTLMSSP_DOMAIN(

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

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

2200 ... key=bytes.fromhex(

2201 ... "85abb9b61dc2fa49d4cc04317bbd108f8f79df28"

2202 ... "239155ed7b144c5d2ebcf016"

2203 ... )

2204 ... ),

2205 ... )

2206 """

2207

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

2209 from scapy.layers.kerberos import KerberosSSP

2210

2211 # Either PASSWORD or HASHNT or ssp

2212 if (

2213 "HASHNT" not in kwargs

2214 and "PASSWORD" not in kwargs

2215 and "KEY" not in kwargs

2216 and ssp is None

2217 ):

2218 raise ValueError(

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

2220 "provide a ssp=KerberosSSP()"

2221 )

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

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

2224

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

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

2227

2228 # UPN is mandatory

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

2230 UPN = ssp.UPN

2231 elif UPN is None:

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

2233 kwargs["UPN"] = UPN

2234

2235 # Call parent

2236 super(NTLMSSP_DOMAIN, self).__init__(

2237 *args,

2238 **kwargs,

2239 )

2240

2241 # Treat specific parameters

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

2243 if self.DC_FQDN is None:

2244 # Get DC_FQDN from dclocator

2245 from scapy.layers.ldap import dclocator

2246

2247 dc = dclocator(

2248 self.DOMAIN_FQDN,

2249 timeout=timeout,

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

2251 )

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

2253

2254 # If logging in via Kerberos

2255 self.ssp = ssp

2256

2257 def _getSessionBaseKey(self, Context, ntlm):

2258 """

2259 Return the Session Key by asking the DC.

2260 """

2261 # No user / no domain: skip.

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

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

2264

2265 # Import RPC stuff

2266 from scapy.layers.dcerpc import NDRUnion

2267 from scapy.layers.msrpce.msnrpc import (

2268 NETLOGON_SECURE_CHANNEL_METHOD,

2269 NetlogonClient,

2270 )

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

2272 NETLOGON_LOGON_IDENTITY_INFO,

2273 NetrLogonSamLogonWithFlags_Request,

2274 PNETLOGON_AUTHENTICATOR,

2275 PNETLOGON_NETWORK_INFO,

2276 STRING,

2277 UNICODE_STRING,

2278 )

2279

2280 # Create NetlogonClient with PRIVACY

2281 client = NetlogonClient()

2282 client.connect(self.DC_FQDN)

2283

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

2285 try:

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

2287 # Login via classic NetlogonSSP

2288 client.establish_secure_channel(

2289 mode=NETLOGON_SECURE_CHANNEL_METHOD.NetrServerAuthenticate3,

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

2291 DC_FQDN=self.DC_FQDN,

2292 HASHNT=self.HASHNT,

2293 )

2294 else:

2295 # Login via KerberosSSP (Windows 2025)

2296 client.establish_secure_channel(

2297 mode=NETLOGON_SECURE_CHANNEL_METHOD.NetrServerAuthenticateKerberos,

2298 UPN=self.UPN,

2299 DC_FQDN=self.DC_FQDN,

2300 PASSWORD=self.PASSWORD,

2301 KEY=self.KEY,

2302 ssp=self.ssp,

2303 )

2304 except ValueError:

2305 log_runtime.warning(

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

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

2308 )

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

2310

2311 # Request validation of the NTLM request

2312 req = NetrLogonSamLogonWithFlags_Request(

2313 LogonServer="",

2314 ComputerName=self.COMPUTER_NB_NAME,

2315 Authenticator=client.create_authenticator(),

2316 ReturnAuthenticator=PNETLOGON_AUTHENTICATOR(),

2317 LogonLevel=6, # NetlogonNetworkTransitiveInformation

2318 LogonInformation=NDRUnion(

2319 tag=6,

2320 value=PNETLOGON_NETWORK_INFO(

2321 Identity=NETLOGON_LOGON_IDENTITY_INFO(

2322 LogonDomainName=UNICODE_STRING(

2323 Buffer=ntlm.DomainName,

2324 ),

2325 ParameterControl=0x00002AE0,

2326 UserName=UNICODE_STRING(

2327 Buffer=ntlm.UserName,

2328 ),

2329 Workstation=UNICODE_STRING(

2330 Buffer=ntlm.Workstation,

2331 ),

2332 ),

2333 LmChallenge=Context.chall_tok.ServerChallenge,

2334 NtChallengeResponse=STRING(

2335 Buffer=bytes(ntlm.NtChallengeResponse),

2336 ),

2337 LmChallengeResponse=STRING(

2338 Buffer=bytes(ntlm.LmChallengeResponse),

2339 ),

2340 ),

2341 ),

2342 ValidationLevel=6,

2343 ExtraFlags=0,

2344 ndr64=client.ndr64,

2345 )

2346

2347 # Get response

2348 resp = client.sr1_req(req)

2349 if resp and resp.status == 0:

2350 # Success

2351

2352 # Validate DC authenticator

2353 client.validate_authenticator(resp.ReturnAuthenticator.value)

2354

2355 # Get and return the SessionKey

2356 UserSessionKey = resp.ValidationInformation.value.value.UserSessionKey

2357 return bytes(UserSessionKey)

2358 else:

2359 # Failed

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

2361

2362 def _checkLogin(self, Context, auth_tok):

2363 # Always OK if we got the session key

2364 return True