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

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 <gabriel[]potter[]fr> 

5 

6""" 

7SPNEGO 

8 

9Implements parts of: 

10 

11- GSSAPI SPNEGO: RFC4178 > RFC2478 

12- GSSAPI SPNEGO NEGOEX: [MS-NEGOEX] 

13 

14.. note:: 

15 You will find more complete documentation for this layer over at 

16 `GSSAPI <https://scapy.readthedocs.io/en/latest/layers/gssapi.html#spnego>`_ 

17""" 

18 

19import os 

20import struct 

21from uuid import UUID 

22 

23from scapy.asn1.asn1 import ( 

24 ASN1_Codecs, 

25 ASN1_OID, 

26 ASN1_GENERAL_STRING, 

27) 

28from scapy.asn1.mib import conf # loads conf.mib 

29from scapy.asn1fields import ( 

30 ASN1F_CHOICE, 

31 ASN1F_ENUMERATED, 

32 ASN1F_FLAGS, 

33 ASN1F_GENERAL_STRING, 

34 ASN1F_OID, 

35 ASN1F_optional, 

36 ASN1F_PACKET, 

37 ASN1F_SEQUENCE_OF, 

38 ASN1F_SEQUENCE, 

39 ASN1F_STRING_ENCAPS, 

40 ASN1F_STRING, 

41) 

42from scapy.asn1packet import ASN1_Packet 

43from scapy.fields import ( 

44 FieldListField, 

45 LEIntEnumField, 

46 LEIntField, 

47 LELongEnumField, 

48 LELongField, 

49 LEShortField, 

50 MultipleTypeField, 

51 PacketField, 

52 PacketListField, 

53 StrField, 

54 StrFixedLenField, 

55 UUIDEnumField, 

56 UUIDField, 

57 XStrFixedLenField, 

58 XStrLenField, 

59) 

60from scapy.error import log_runtime 

61from scapy.packet import Packet, bind_layers 

62from scapy.utils import ( 

63 valid_ip, 

64 valid_ip6, 

65) 

66 

67from scapy.layers.gssapi import ( 

68 _GSSAPI_OIDS, 

69 _GSSAPI_SIGNATURE_OIDS, 

70 GSS_C_FLAGS, 

71 GSS_C_NO_CHANNEL_BINDINGS, 

72 GSS_S_BAD_MECH, 

73 GSS_S_COMPLETE, 

74 GSS_S_CONTINUE_NEEDED, 

75 GSS_S_FAILURE, 

76 GSS_S_FLAGS, 

77 GSSAPI_BLOB_SIGNATURE, 

78 GSSAPI_BLOB, 

79 GssChannelBindings, 

80 SSP, 

81) 

82 

83# SSP Providers 

84from scapy.layers.kerberos import ( 

85 Kerberos, 

86 KerberosSSP, 

87 _parse_spn, 

88 _parse_upn, 

89) 

90from scapy.layers.ntlm import ( 

91 NTLMSSP, 

92 MD4le, 

93 NEGOEX_EXCHANGE_NTLM, 

94 NTLM_Header, 

95 _NTLMPayloadField, 

96 _NTLMPayloadPacket, 

97) 

98 

99# Typing imports 

100from typing import ( 

101 Dict, 

102 List, 

103 Optional, 

104 Tuple, 

105) 

106 

107# SPNEGO negTokenInit 

108# https://datatracker.ietf.org/doc/html/rfc4178#section-4.2.1 

109 

110 

111class SPNEGO_MechType(ASN1_Packet): 

112 ASN1_codec = ASN1_Codecs.BER 

113 ASN1_root = ASN1F_OID("oid", None) 

114 

115 

116class SPNEGO_MechTypes(ASN1_Packet): 

117 ASN1_codec = ASN1_Codecs.BER 

118 ASN1_root = ASN1F_SEQUENCE_OF("mechTypes", None, SPNEGO_MechType) 

119 

120 

121class SPNEGO_MechListMIC(ASN1_Packet): 

122 ASN1_codec = ASN1_Codecs.BER 

123 ASN1_root = ASN1F_STRING_ENCAPS("value", "", GSSAPI_BLOB_SIGNATURE) 

124 

125 

126_mechDissector = { 

127 "1.3.6.1.4.1.311.2.2.10": NTLM_Header, # NTLM 

128 "1.2.840.48018.1.2.2": Kerberos, # MS KRB5 - Microsoft Kerberos 5 

129 "1.2.840.113554.1.2.2": Kerberos, # Kerberos 5 

130 "1.2.840.113554.1.2.2.3": Kerberos, # Kerberos 5 - User to User 

131} 

132 

133 

134class _SPNEGO_Token_Field(ASN1F_STRING): 

135 def i2m(self, pkt, x): 

136 if x is None: 

137 x = b"" 

138 return super(_SPNEGO_Token_Field, self).i2m(pkt, bytes(x)) 

139 

140 def m2i(self, pkt, s): 

141 dat, r = super(_SPNEGO_Token_Field, self).m2i(pkt, s) 

142 types = None 

143 if isinstance(pkt.underlayer, SPNEGO_negTokenInit): 

144 types = pkt.underlayer.mechTypes 

145 elif isinstance(pkt.underlayer, SPNEGO_negTokenResp): 

146 types = [pkt.underlayer.supportedMech] 

147 if types and types[0] and types[0].oid.val in _mechDissector: 

148 return _mechDissector[types[0].oid.val](dat.val), r 

149 else: 

150 # Use heuristics 

151 return GSSAPI_BLOB(dat.val), r 

152 

153 

154class SPNEGO_Token(ASN1_Packet): 

155 ASN1_codec = ASN1_Codecs.BER 

156 ASN1_root = _SPNEGO_Token_Field("value", None) 

157 

158 

159_ContextFlags = [ 

160 "delegFlag", 

161 "mutualFlag", 

162 "replayFlag", 

163 "sequenceFlag", 

164 "superseded", 

165 "anonFlag", 

166 "confFlag", 

167 "integFlag", 

168] 

169 

170 

171class SPNEGO_negHints(ASN1_Packet): 

172 # [MS-SPNG] 2.2.1 

173 ASN1_codec = ASN1_Codecs.BER 

174 ASN1_root = ASN1F_SEQUENCE( 

175 ASN1F_optional( 

176 ASN1F_GENERAL_STRING( 

177 "hintName", "not_defined_in_RFC4178@please_ignore", explicit_tag=0xA0 

178 ), 

179 ), 

180 ASN1F_optional( 

181 ASN1F_GENERAL_STRING("hintAddress", None, explicit_tag=0xA1), 

182 ), 

183 ) 

184 

185 

186class SPNEGO_negTokenInit(ASN1_Packet): 

187 ASN1_codec = ASN1_Codecs.BER 

188 ASN1_root = ASN1F_SEQUENCE( 

189 ASN1F_optional( 

190 ASN1F_SEQUENCE_OF("mechTypes", None, SPNEGO_MechType, explicit_tag=0xA0) 

191 ), 

192 ASN1F_optional(ASN1F_FLAGS("reqFlags", None, _ContextFlags, implicit_tag=0x81)), 

193 ASN1F_optional( 

194 ASN1F_PACKET("mechToken", None, SPNEGO_Token, explicit_tag=0xA2) 

195 ), 

196 # [MS-SPNG] flavor ! 

197 ASN1F_optional( 

198 ASN1F_PACKET("negHints", None, SPNEGO_negHints, explicit_tag=0xA3) 

199 ), 

200 ASN1F_optional( 

201 ASN1F_PACKET("mechListMIC", None, SPNEGO_MechListMIC, explicit_tag=0xA4) 

202 ), 

203 # Compat with RFC 4178's SPNEGO_negTokenInit 

204 ASN1F_optional( 

205 ASN1F_PACKET("_mechListMIC", None, SPNEGO_MechListMIC, explicit_tag=0xA3) 

206 ), 

207 ) 

208 

209 

210# SPNEGO negTokenTarg 

211# https://datatracker.ietf.org/doc/html/rfc4178#section-4.2.2 

212 

213 

214class SPNEGO_negTokenResp(ASN1_Packet): 

215 ASN1_codec = ASN1_Codecs.BER 

216 ASN1_root = ASN1F_SEQUENCE( 

217 ASN1F_optional( 

218 ASN1F_ENUMERATED( 

219 "negState", 

220 0, 

221 { 

222 0: "accept-completed", 

223 1: "accept-incomplete", 

224 2: "reject", 

225 3: "request-mic", 

226 }, 

227 explicit_tag=0xA0, 

228 ), 

229 ), 

230 ASN1F_optional( 

231 ASN1F_PACKET( 

232 "supportedMech", SPNEGO_MechType(), SPNEGO_MechType, explicit_tag=0xA1 

233 ), 

234 ), 

235 ASN1F_optional( 

236 ASN1F_PACKET("responseToken", None, SPNEGO_Token, explicit_tag=0xA2) 

237 ), 

238 ASN1F_optional( 

239 ASN1F_PACKET("mechListMIC", None, SPNEGO_MechListMIC, explicit_tag=0xA3) 

240 ), 

241 ) 

242 

243 

244class SPNEGO_negToken(ASN1_Packet): 

245 ASN1_codec = ASN1_Codecs.BER 

246 ASN1_root = ASN1F_CHOICE( 

247 "token", 

248 SPNEGO_negTokenInit(), 

249 ASN1F_PACKET( 

250 "negTokenInit", 

251 SPNEGO_negTokenInit(), 

252 SPNEGO_negTokenInit, 

253 explicit_tag=0xA0, 

254 ), 

255 ASN1F_PACKET( 

256 "negTokenResp", 

257 SPNEGO_negTokenResp(), 

258 SPNEGO_negTokenResp, 

259 explicit_tag=0xA1, 

260 ), 

261 ) 

262 

263 

264# Register for the GSS API Blob 

265 

266_GSSAPI_OIDS["1.3.6.1.5.5.2"] = SPNEGO_negToken 

267_GSSAPI_SIGNATURE_OIDS["1.3.6.1.5.5.2"] = SPNEGO_negToken 

268 

269 

270def mechListMIC(oids): 

271 """ 

272 Implementation of RFC 4178 - Appendix D. mechListMIC Computation 

273 

274 NOTE: The documentation on mechListMIC isn't super clear, so note that: 

275 

276 - The mechListMIC that the client sends is computed over the 

277 list of mechanisms that it requests. 

278 - the mechListMIC that the server sends is computed over the 

279 list of mechanisms that the client requested. 

280 

281 This also means that NegTokenInit2 added by [MS-SPNG] is NOT protected. 

282 That's not necessarily an issue, since it was optional in most cases, 

283 but it's something to keep in mind. 

284 """ 

285 return bytes(SPNEGO_MechTypes(mechTypes=oids)) 

286 

287 

288# NEGOEX 

289# https://docs.microsoft.com/en-us/openspecs/windows_protocols/ms-negoex/0ad7a003-ab56-4839-a204-b555ca6759a2 

290 

291 

292_NEGOEX_AUTH_SCHEMES = { 

293 # Reversed. Is there any doc related to this? 

294 # The NEGOEX doc is very ellusive 

295 UUID("5c33530d-eaf9-0d4d-b2ec-4ae3786ec308"): "UUID('[NTLM-UUID]')", 

296} 

297 

298 

299class NEGOEX_MESSAGE_HEADER(Packet): 

300 fields_desc = [ 

301 StrFixedLenField("Signature", "NEGOEXTS", length=8), 

302 LEIntEnumField( 

303 "MessageType", 

304 0, 

305 { 

306 0x0: "INITIATOR_NEGO", 

307 0x01: "ACCEPTOR_NEGO", 

308 0x02: "INITIATOR_META_DATA", 

309 0x03: "ACCEPTOR_META_DATA", 

310 0x04: "CHALLENGE", 

311 0x05: "AP_REQUEST", 

312 0x06: "VERIFY", 

313 0x07: "ALERT", 

314 }, 

315 ), 

316 LEIntField("SequenceNum", 0), 

317 LEIntField("cbHeaderLength", None), 

318 LEIntField("cbMessageLength", None), 

319 UUIDField("ConversationId", None), 

320 ] 

321 

322 def post_build(self, pkt, pay): 

323 if self.cbHeaderLength is None: 

324 pkt = pkt[16:] + struct.pack("<I", len(pkt)) + pkt[20:] 

325 if self.cbMessageLength is None: 

326 pkt = pkt[20:] + struct.pack("<I", len(pkt) + len(pay)) + pkt[24:] 

327 return pkt + pay 

328 

329 

330def _NEGOEX_post_build(self, p, pay_offset, fields): 

331 # type: (Packet, bytes, int, Dict[str, Tuple[str, int]]) -> bytes 

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

333 for field_name, value in self.fields["Payload"]: 

334 length = self.get_field("Payload").fields_map[field_name].i2len(self, value) 

335 count = self.get_field("Payload").fields_map[field_name].i2count(self, value) 

336 offset = fields[field_name] 

337 # Offset 

338 if self.getfieldval(field_name + "BufferOffset") is None: 

339 p = p[:offset] + struct.pack("<I", pay_offset) + p[offset + 4 :] 

340 # Count 

341 if self.getfieldval(field_name + "Count") is None: 

342 p = p[: offset + 4] + struct.pack("<H", count) + p[offset + 6 :] 

343 pay_offset += length 

344 return p 

345 

346 

347class NEGOEX_BYTE_VECTOR(Packet): 

348 fields_desc = [ 

349 LEIntField("ByteArrayBufferOffset", 0), 

350 LEIntField("ByteArrayLength", 0), 

351 ] 

352 

353 def guess_payload_class(self, payload): 

354 return conf.padding_layer 

355 

356 

357class NEGOEX_EXTENSION_VECTOR(Packet): 

358 fields_desc = [ 

359 LELongField("ExtensionArrayOffset", 0), 

360 LEShortField("ExtensionCount", 0), 

361 ] 

362 

363 

364class NEGOEX_NEGO_MESSAGE(_NTLMPayloadPacket): 

365 OFFSET = 92 

366 show_indent = 0 

367 fields_desc = [ 

368 NEGOEX_MESSAGE_HEADER, 

369 XStrFixedLenField("Random", b"", length=32), 

370 LELongField("ProtocolVersion", 0), 

371 LEIntField("AuthSchemeBufferOffset", None), 

372 LEShortField("AuthSchemeCount", None), 

373 LEIntField("ExtensionBufferOffset", None), 

374 LEShortField("ExtensionCount", None), 

375 # Payload 

376 _NTLMPayloadField( 

377 "Payload", 

378 OFFSET, 

379 [ 

380 FieldListField( 

381 "AuthScheme", 

382 [], 

383 UUIDEnumField("", None, _NEGOEX_AUTH_SCHEMES), 

384 count_from=lambda pkt: pkt.AuthSchemeCount, 

385 ), 

386 PacketListField( 

387 "Extension", 

388 [], 

389 NEGOEX_EXTENSION_VECTOR, 

390 count_from=lambda pkt: pkt.ExtensionCount, 

391 ), 

392 ], 

393 length_from=lambda pkt: pkt.cbMessageLength - 92, 

394 ), 

395 # TODO: dissect extensions 

396 ] 

397 

398 def post_build(self, pkt, pay): 

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

400 return ( 

401 _NEGOEX_post_build( 

402 self, 

403 pkt, 

404 self.OFFSET, 

405 { 

406 "AuthScheme": 96, 

407 "Extension": 102, 

408 }, 

409 ) 

410 + pay 

411 ) 

412 

413 @classmethod 

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

415 if _pkt and len(_pkt) >= 12: 

416 MessageType = struct.unpack("<I", _pkt[8:12])[0] 

417 if MessageType in [0, 1]: 

418 return NEGOEX_NEGO_MESSAGE 

419 elif MessageType in [2, 3]: 

420 return NEGOEX_EXCHANGE_MESSAGE 

421 return cls 

422 

423 

424# RFC3961 

425_checksum_types = { 

426 1: "CRC32", 

427 2: "RSA-MD4", 

428 3: "RSA-MD4-DES", 

429 4: "DES-MAC", 

430 5: "DES-MAC-K", 

431 6: "RSA-MDA-DES-K", 

432 7: "RSA-MD5", 

433 8: "RSA-MD5-DES", 

434 9: "RSA-MD5-DES3", 

435 10: "SHA1", 

436 12: "HMAC-SHA1-DES3-KD", 

437 13: "HMAC-SHA1-DES3", 

438 14: "SHA1", 

439 15: "HMAC-SHA1-96-AES128", 

440 16: "HMAC-SHA1-96-AES256", 

441} 

442 

443 

444def _checksum_size(pkt): 

445 if pkt.ChecksumType == 1: 

446 return 4 

447 elif pkt.ChecksumType in [2, 4, 6, 7]: 

448 return 16 

449 elif pkt.ChecksumType in [3, 8, 9]: 

450 return 24 

451 elif pkt.ChecksumType == 5: 

452 return 8 

453 elif pkt.ChecksumType in [10, 12, 13, 14, 15, 16]: 

454 return 20 

455 return 0 

456 

457 

458class NEGOEX_CHECKSUM(Packet): 

459 fields_desc = [ 

460 LELongField("cbHeaderLength", 20), 

461 LELongEnumField("ChecksumScheme", 1, {1: "CHECKSUM_SCHEME_RFC3961"}), 

462 LELongEnumField("ChecksumType", None, _checksum_types), 

463 XStrLenField("ChecksumValue", b"", length_from=_checksum_size), 

464 ] 

465 

466 

467class NEGOEX_EXCHANGE_MESSAGE(_NTLMPayloadPacket): 

468 OFFSET = 64 

469 show_indent = 0 

470 fields_desc = [ 

471 NEGOEX_MESSAGE_HEADER, 

472 UUIDEnumField("AuthScheme", None, _NEGOEX_AUTH_SCHEMES), 

473 LEIntField("ExchangeBufferOffset", 0), 

474 LEIntField("ExchangeLen", 0), 

475 _NTLMPayloadField( 

476 "Payload", 

477 OFFSET, 

478 [ 

479 # The NEGOEX doc mentions the following blob as as an 

480 # "opaque handshake for the client authentication scheme". 

481 # NEGOEX_EXCHANGE_NTLM is a reversed interpretation, and is 

482 # probably not accurate. 

483 MultipleTypeField( 

484 [ 

485 ( 

486 PacketField("Exchange", None, NEGOEX_EXCHANGE_NTLM), 

487 lambda pkt: pkt.AuthScheme 

488 == UUID("5c33530d-eaf9-0d4d-b2ec-4ae3786ec308"), 

489 ), 

490 ], 

491 StrField("Exchange", b""), 

492 ) 

493 ], 

494 length_from=lambda pkt: pkt.cbMessageLength - pkt.cbHeaderLength, 

495 ), 

496 ] 

497 

498 

499class NEGOEX_VERIFY_MESSAGE(Packet): 

500 show_indent = 0 

501 fields_desc = [ 

502 NEGOEX_MESSAGE_HEADER, 

503 UUIDEnumField("AuthScheme", None, _NEGOEX_AUTH_SCHEMES), 

504 PacketField("Checksum", NEGOEX_CHECKSUM(), NEGOEX_CHECKSUM), 

505 ] 

506 

507 

508bind_layers(NEGOEX_NEGO_MESSAGE, NEGOEX_NEGO_MESSAGE) 

509 

510 

511_mechDissector["1.3.6.1.4.1.311.2.2.30"] = NEGOEX_NEGO_MESSAGE 

512 

513# -- SSP 

514 

515 

516class SPNEGOSSP(SSP): 

517 """ 

518 The SPNEGO SSP 

519 

520 :param ssps: a dict with keys being the SSP class, and the value being a 

521 dictionary of the keyword arguments to pass it on init. 

522 

523 Example:: 

524 

525 from scapy.layers.ntlm import NTLMSSP 

526 from scapy.layers.kerberos import KerberosSSP 

527 from scapy.layers.spnego import SPNEGOSSP 

528 from scapy.layers.smbserver import smbserver 

529 from scapy.libs.rfc3961 import Encryption, Key 

530 

531 ssp = SPNEGOSSP([ 

532 NTLMSSP( 

533 IDENTITIES={ 

534 "User1": MD4le("Password1"), 

535 "Administrator": MD4le("Password123!"), 

536 } 

537 ), 

538 KerberosSSP( 

539 SPN="cifs/server2.domain.local", 

540 KEY=Key( 

541 Encryption.AES256, 

542 key=hex_bytes("5e9255c907b2f7e969ddad816eabbec8f1f7a387c7194ecc98b827bdc9421c2b") 

543 ) 

544 ) 

545 ]) 

546 smbserver(ssp=ssp) 

547 """ 

548 

549 __slots__ = [ 

550 "ssps", 

551 ] 

552 

553 auth_type = 0x09 

554 

555 class STATE(SSP.STATE): 

556 FIRST = 1 

557 SUBSEQUENT = 2 

558 

559 class CONTEXT(SSP.CONTEXT): 

560 __slots__ = [ 

561 "req_flags", 

562 "ssps", 

563 "other_mechtypes", 

564 "sent_mechtypes", 

565 "first_choice", 

566 "require_mic", 

567 "verified_mic", 

568 "ssp", 

569 "ssp_context", 

570 "ssp_mechtype", 

571 "raw", 

572 ] 

573 

574 def __init__( 

575 self, 

576 ssps: List[SSP], 

577 req_flags=None, 

578 ): 

579 self.state = SPNEGOSSP.STATE.FIRST 

580 self.req_flags = req_flags 

581 # Information used during negotiation 

582 self.ssps = ssps 

583 self.other_mechtypes = None # the mechtypes our peer requested 

584 self.sent_mechtypes = None # the mechtypes we sent when acting as a client 

585 self.first_choice = True # whether the SSP was the peer's first choice 

586 self.require_mic = False # whether the mechListMIC is required or not 

587 self.verified_mic = False # whether mechListMIC has been verified 

588 # Information about the currently selected SSP 

589 self.ssp = None 

590 self.ssp_context = None 

591 self.ssp_mechtype = None 

592 self.raw = False # fallback to raw SSP 

593 super(SPNEGOSSP.CONTEXT, self).__init__() 

594 

595 # This is the order Windows chooses 

596 _PREF_ORDER = [ 

597 "1.2.840.113554.1.2.2.3", # Kerberos 5 - User to User 

598 "1.2.840.48018.1.2.2", # MS KRB5 

599 "1.2.840.113554.1.2.2", # Kerberos 5 

600 "1.3.6.1.4.1.311.2.2.30", # NEGOEX 

601 "1.3.6.1.4.1.311.2.2.10", # NTLM 

602 ] 

603 

604 def get_supported_mechtypes(self): 

605 """ 

606 Return an ordered list of mechtypes that are still available. 

607 """ 

608 # 1. Build mech list 

609 mechs = [] 

610 for ssp in self.ssps: 

611 mechs.extend(ssp.GSS_Inquire_names_for_mech()) 

612 

613 # 2. Sort according to the preference order. 

614 mechs.sort(key=lambda x: self._PREF_ORDER.index(x)) 

615 

616 # 3. Return wrapped in MechType 

617 return [SPNEGO_MechType(oid=ASN1_OID(oid)) for oid in mechs] 

618 

619 def negotiate_ssp(self) -> None: 

620 """ 

621 Perform SSP negotiation. 

622 

623 This updates our context and sets it with the first SSP that is 

624 common to both client and server. This also applies rules from 

625 [MS-SPNG] and RFC4178 to determine if mechListMIC is required. 

626 """ 

627 if self.other_mechtypes is None: 

628 # We don't have any information about the peer's preferred SSPs. 

629 # This typically happens on client side, when NegTokenInit2 isn't used. 

630 self.ssp = self.ssps[0] 

631 ssp_oid = self.ssp.GSS_Inquire_names_for_mech()[0] 

632 else: 

633 # Get first common SSP between us and our peer 

634 other_oids = [x.oid.val for x in self.other_mechtypes] 

635 try: 

636 self.ssp, ssp_oid = next( 

637 (ssp, requested_oid) 

638 for requested_oid in other_oids 

639 for ssp in self.ssps 

640 if requested_oid in ssp.GSS_Inquire_names_for_mech() 

641 ) 

642 except StopIteration: 

643 raise ValueError( 

644 "Could not find a common SSP with the remote peer !" 

645 ) 

646 

647 # Check whether the selected SSP was the one preferred by the client 

648 self.first_choice = ssp_oid == other_oids[0] 

649 

650 # Check whether mechListMIC is mandatory for this exchange 

651 if not self.first_choice: 

652 # RFC4178 rules for mechListMIC: mandatory if not the first choice. 

653 self.require_mic = True 

654 elif ssp_oid == "1.3.6.1.4.1.311.2.2.10" and self.ssp.SupportsMechListMIC(): 

655 # [MS-SPNG] note 8: "If NTLM authentication is most preferred by 

656 # the client and the server, and the client includes a MIC in 

657 # AUTHENTICATE_MESSAGE, then the mechListMIC field becomes 

658 # mandatory" 

659 self.require_mic = True 

660 

661 # Get the associated ssp dissection class and mechtype 

662 self.ssp_mechtype = SPNEGO_MechType(oid=ASN1_OID(ssp_oid)) 

663 

664 # Reset the ssp context 

665 self.ssp_context = None 

666 

667 # Passthrough attributes and functions 

668 

669 def clifailure(self): 

670 if self.ssp_context is not None: 

671 self.ssp_context.clifailure() 

672 

673 def __getattr__(self, attr): 

674 try: 

675 return object.__getattribute__(self, attr) 

676 except AttributeError: 

677 return getattr(self.ssp_context, attr) 

678 

679 def __setattr__(self, attr, val): 

680 try: 

681 return object.__setattr__(self, attr, val) 

682 except AttributeError: 

683 return setattr(self.ssp_context, attr, val) 

684 

685 # Passthrough the flags property 

686 

687 @property 

688 def flags(self): 

689 if self.ssp_context: 

690 return self.ssp_context.flags 

691 return GSS_C_FLAGS(0) 

692 

693 @flags.setter 

694 def flags(self, x): 

695 if not self.ssp_context: 

696 return 

697 self.ssp_context.flags = x 

698 

699 def __repr__(self): 

700 return "SPNEGOSSP[%s]" % repr(self.ssp_context) 

701 

702 def __init__(self, ssps: List[SSP], **kwargs): 

703 self.ssps = ssps 

704 super(SPNEGOSSP, self).__init__(**kwargs) 

705 

706 @classmethod 

707 def from_cli_arguments( 

708 cls, 

709 UPN: str, 

710 target: str, 

711 password: str = None, 

712 HashNt: bytes = None, 

713 HashAes256Sha96: bytes = None, 

714 HashAes128Sha96: bytes = None, 

715 kerberos_required: bool = False, 

716 ST=None, 

717 TGT=None, 

718 KEY=None, 

719 ccache: str = None, 

720 debug: int = 0, 

721 use_krb5ccname: bool = False, 

722 ): 

723 """ 

724 Initialize a SPNEGOSSP from a list of many arguments. 

725 This is useful in a CLI, with NTLM and Kerberos supported by default. 

726 

727 :param UPN: the UPN of the user to use. 

728 :param target: the target IP/hostname entered by the user. 

729 :param kerberos_required: require kerberos 

730 :param password: (string) if provided, used for auth 

731 :param HashNt: (bytes) if provided, used for auth (NTLM) 

732 :param HashAes256Sha96: (bytes) if provided, used for auth (Kerberos) 

733 :param HashAes128Sha96: (bytes) if provided, used for auth (Kerberos) 

734 :param ST: if provided, the service ticket to use (Kerberos) 

735 :param TGT: if provided, the TGT to use (Kerberos) 

736 :param KEY: if ST provided, the session key associated to the ticket (Kerberos). 

737 This can be either for the ST or TGT. Else, the user secret key. 

738 :param ccache: (str) if provided, a path to a CCACHE (Kerberos) 

739 :param use_krb5ccname: (bool) if true, the KRB5CCNAME environment variable will 

740 be used if available. 

741 """ 

742 kerberos = True 

743 hostname = None 

744 # Check if target is a hostname / Check IP 

745 if ":" in target: 

746 if not valid_ip6(target): 

747 hostname = target 

748 else: 

749 if not valid_ip(target): 

750 hostname = target 

751 

752 # Check UPN 

753 try: 

754 _, realm = _parse_upn(UPN) 

755 if realm == ".": 

756 # Local 

757 kerberos = False 

758 except ValueError: 

759 # not a UPN: NTLM only 

760 kerberos = False 

761 

762 # If we're asked, check the environment for KRB5CCNAME 

763 if use_krb5ccname and ccache is None and "KRB5CCNAME" in os.environ: 

764 ccache = os.environ["KRB5CCNAME"] 

765 

766 # Do we need to ask the password? 

767 if all( 

768 x is None 

769 for x in [ 

770 ST, 

771 password, 

772 HashNt, 

773 HashAes256Sha96, 

774 HashAes128Sha96, 

775 ccache, 

776 ] 

777 ): 

778 # yes. 

779 from prompt_toolkit import prompt 

780 

781 password = prompt("Password: ", is_password=True) 

782 

783 ssps = [] 

784 # Kerberos 

785 if kerberos and hostname: 

786 # Get ticket if we don't already have one. 

787 if ST is None and TGT is None and ccache is not None: 

788 # In this case, load the KerberosSSP from ccache 

789 from scapy.modules.ticketer import Ticketer 

790 

791 # Import into a Ticketer object 

792 t = Ticketer() 

793 t.open_ccache(ccache) 

794 

795 # Look for the ticket that we'll use. We chose: 

796 # - either a ST if the SPN matches our target 

797 # - else a TGT if we got nothing better 

798 tgts = [] 

799 for i, (tkt, key, upn, spn) in enumerate(t.iter_tickets()): 

800 spn, _ = _parse_spn(spn) 

801 spn_host = spn.split("/")[-1] 

802 # Check that it's for the correct user 

803 if upn.lower() == UPN.lower(): 

804 # Check that it's either a TGT or a ST to the correct service 

805 if spn.lower().startswith("krbtgt/"): 

806 # TGT. Keep it, and see if we don't have a better ST. 

807 tgts.append(t.ssp(i)) 

808 elif hostname.lower() == spn_host.lower(): 

809 # ST. We're done ! 

810 ssps.append(t.ssp(i)) 

811 break 

812 else: 

813 # No ST found 

814 if tgts: 

815 # Using a TGT ! 

816 ssps.append(tgts[0]) 

817 else: 

818 # Nothing found 

819 t.show() 

820 raise ValueError( 

821 f"Could not find a ticket for {upn}, either a " 

822 f"TGT or towards {hostname}" 

823 ) 

824 elif ST is None and TGT is None: 

825 # In this case, KEY is supposed to be the user's key. 

826 from scapy.libs.rfc3961 import Key, EncryptionType 

827 

828 if KEY is None and HashAes256Sha96: 

829 KEY = Key( 

830 EncryptionType.AES256_CTS_HMAC_SHA1_96, 

831 HashAes256Sha96, 

832 ) 

833 elif KEY is None and HashAes128Sha96: 

834 KEY = Key( 

835 EncryptionType.AES128_CTS_HMAC_SHA1_96, 

836 HashAes128Sha96, 

837 ) 

838 elif KEY is None and HashNt: 

839 KEY = Key( 

840 EncryptionType.RC4_HMAC, 

841 HashNt, 

842 ) 

843 # Make a SSP that only has a UPN and secret. 

844 ssps.append( 

845 KerberosSSP( 

846 UPN=UPN, 

847 PASSWORD=password, 

848 KEY=KEY, 

849 debug=debug, 

850 ) 

851 ) 

852 else: 

853 # We have a ST, use it with the key. 

854 ssps.append( 

855 KerberosSSP( 

856 UPN=UPN, 

857 ST=ST, 

858 TGT=TGT, 

859 KEY=KEY, 

860 debug=debug, 

861 ) 

862 ) 

863 elif kerberos_required: 

864 raise ValueError( 

865 "Kerberos required but domain not specified in the UPN, " 

866 "or target isn't a hostname !" 

867 ) 

868 

869 # NTLM 

870 if not kerberos_required: 

871 if HashNt is None and password is not None: 

872 HashNt = MD4le(password) 

873 if HashNt is not None: 

874 ssps.append(NTLMSSP(UPN=UPN, HASHNT=HashNt)) 

875 

876 if not ssps: 

877 raise ValueError("Unexpected case ! Please report.") 

878 

879 # Build the SSP 

880 return cls(ssps) 

881 

882 def NegTokenInit2(self): 

883 """ 

884 Server-Initiation of GSSAPI/SPNEGO. 

885 See [MS-SPNG] sect 3.2.5.2 

886 """ 

887 Context = SPNEGOSSP.CONTEXT(list(self.ssps)) 

888 return ( 

889 Context, 

890 GSSAPI_BLOB( 

891 innerToken=SPNEGO_negToken( 

892 token=SPNEGO_negTokenInit( 

893 mechTypes=Context.get_supported_mechtypes(), 

894 negHints=SPNEGO_negHints( 

895 hintName=ASN1_GENERAL_STRING( 

896 "not_defined_in_RFC4178@please_ignore" 

897 ), 

898 ), 

899 ) 

900 ) 

901 ), 

902 ) 

903 

904 # NOTE: NegoEX has an effect on how the SecurityContext is 

905 # initialized, as detailed in [MS-AUTHSOD] sect 3.3.2 

906 # But the format that the Exchange token uses appears not to 

907 # be documented :/ 

908 

909 # resp.SecurityBlob.innerToken.token.mechTypes.insert( 

910 # 0, 

911 # # NEGOEX 

912 # SPNEGO_MechType(oid="1.3.6.1.4.1.311.2.2.30"), 

913 # ) 

914 # resp.SecurityBlob.innerToken.token.mechToken = SPNEGO_Token( 

915 # value=negoex_token 

916 # ) # noqa: E501 

917 

918 def GSS_WrapEx(self, Context, *args, **kwargs): 

919 # Passthrough 

920 return Context.ssp.GSS_WrapEx(Context.ssp_context, *args, **kwargs) 

921 

922 def GSS_UnwrapEx(self, Context, *args, **kwargs): 

923 # Passthrough 

924 return Context.ssp.GSS_UnwrapEx(Context.ssp_context, *args, **kwargs) 

925 

926 def GSS_GetMICEx(self, Context, *args, **kwargs): 

927 # Passthrough 

928 return Context.ssp.GSS_GetMICEx(Context.ssp_context, *args, **kwargs) 

929 

930 def GSS_VerifyMICEx(self, Context, *args, **kwargs): 

931 # Passthrough 

932 return Context.ssp.GSS_VerifyMICEx(Context.ssp_context, *args, **kwargs) 

933 

934 def LegsAmount(self, Context: CONTEXT): 

935 return 4